feat(webapp): enforce watch plan limits (#4556)

**What & why.** Watches now honour a plan's watch limits. A watch whose
window exceeds the plan's `agentWatchMaxHours`, or that would push the
org past its `agentWatchers` count, is refused with a new
`watch_limit_reached` result and an upgrade hint (a chat line on the
card, HTTP 409 on the API).

**Key decisions.**
- Plan limits are a floor *below* the existing code ceilings: `min(plan,
WATCH_MAX_HOURS=24)` for the window, and the per-chat cap of 3 still
applies independently. Plans only tighten, never loosen.
- Watcher count is org-wide and checked only after the immediate check
declines, so a one-shot consumes no slot.
- Fails open: an absent limit resolves to the unlimited sentinel, so
self-hosted is unaffected; the upgrade nudge is gated on
`isBillingConfigured()`.
- Follow-up: quiet Pro-mark on the card's long-window options.

TRI-12863

Author: kathiekiwi

.server-changes/agent-watch-plan-limits.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: feature
+---
+
+Watches now respect your plan's limits: free plans can run a limited number of watches at once and for a shorter window, with a prompt to upgrade for more.

apps/webapp/app/services/dashboardAgentInvestigationSweep.server.ts

@@ -5,8 +5,11 @@
 
 import {
   listStaleOpenInvestigations,
+  recordInvestigationSweepAttempt,
   settleInvestigationAndCloseCard,
+  settleInvestigationAsInconclusive,
   type Investigation,
+  type SettledInvestigation,
   type SettledInvestigationCard,
 } from "@internal/dashboard-agent-db";
 import { UNSETTLED_INVESTIGATION_NOTE } from "@internal/dashboard-agent-contracts";
@@ -22,6 +25,13 @@ export const INVESTIGATION_STALE_MS = 30 * 60 * 1000;
 /** Per-run cap. Oldest first, so the rest land next run. */
 const SWEEP_BATCH_LIMIT = 100;
 
+/**
+ * After this many failed settle attempts a row is force-abandoned: settled `inconclusive`
+ * WITHOUT the closing card, so a card that never renders leaves the queue instead of
+ * looping forever. The rare stuck spinner is the price of not starving every other row.
+ */
+export const MAX_SWEEP_ATTEMPTS = 5;
+
 export type InvestigationSweepResult = {
   /** Stale `in_progress` rows seen. */
   stale: number;
@@ -30,6 +40,8 @@ export type InvestigationSweepResult = {
   closed: number;
   /** A turn (or another sweep) settled it first. */
   alreadySettled: number;
+  /** Rows past the attempt cap, force-settled without a card so they leave the queue. */
+  abandoned: number;
   failed: number;
 };
 
@@ -46,6 +58,10 @@ export type InvestigationSweepDeps = {
     chatId: string;
     note: string;
   }) => Promise<SettledInvestigationCard | null>;
+  /** Record a failed settle out-of-band; returns the new attempt count, or null if gone. */
+  recordAttempt?: (params: { id: string }) => Promise<number | null>;
+  /** Force a poison row terminal without the failing render path. */
+  forceAbandon?: (params: { id: string; note: string }) => Promise<SettledInvestigation | null>;
 };
 
 /**
@@ -61,12 +77,17 @@ export async function sweepDashboardAgentInvestigations(
     deps.listStale ?? ((params) => listStaleOpenInvestigations(dashboardAgentDb, params));
   const settleAndClose =
     deps.settleAndClose ?? ((params) => settleInvestigationAndCloseCard(dashboardAgentDb, params));
+  const recordAttempt =
+    deps.recordAttempt ?? ((params) => recordInvestigationSweepAttempt(dashboardAgentDb, params));
+  const forceAbandon =
+    deps.forceAbandon ?? ((params) => settleInvestigationAsInconclusive(dashboardAgentDb, params));
 
   const result: InvestigationSweepResult = {
     stale: 0,
     settled: 0,
     closed: 0,
     alreadySettled: 0,
+    abandoned: 0,
     failed: 0,
   };
 
@@ -93,10 +114,49 @@ export async function sweepDashboardAgentInvestigations(
       result.settled++;
       if (outcome.closed) result.closed++;
     } catch (error) {
+      // The settle rolled back, so the row is still `in_progress`. Record the attempt in
+      // its own write — this rotates the row to the back of the sweep order (see
+      // `listStaleOpenInvestigations`) so it can't pin the head and starve newer rows.
+      let attempts: number | null = null;
+      try {
+        attempts = await recordAttempt({ id: investigation.id });
+      } catch (recordError) {
+        logger.error("Dashboard agent investigation sweep: failed to record a sweep attempt", {
+          investigationId: investigation.id,
+          chatId: investigation.chatId,
+          error: recordError,
+        });
+      }
+
+      // Past the cap the card will never render; force it terminal without the render
+      // path so it leaves the queue instead of looping forever.
+      if (attempts !== null && attempts >= MAX_SWEEP_ATTEMPTS) {
+        try {
+          await forceAbandon({ id: investigation.id, note: UNSETTLED_INVESTIGATION_NOTE });
+          result.abandoned++;
+          logger.warn(
+            "Dashboard agent investigation sweep: abandoned a card past the attempt cap",
+            {
+              investigationId: investigation.id,
+              chatId: investigation.chatId,
+              attempts,
+            }
+          );
+          continue;
+        } catch (abandonError) {
+          logger.error("Dashboard agent investigation sweep: failed to abandon a poison card", {
+            investigationId: investigation.id,
+            chatId: investigation.chatId,
+            error: abandonError,
+          });
+        }
+      }
+
       result.failed++;
       logger.error("Dashboard agent investigation sweep: failed to settle an investigation", {
         investigationId: investigation.id,
         chatId: investigation.chatId,
+        attempts,
         error,
       });
     }

apps/webapp/app/services/dashboardAgentWatchErrorStatus.server.ts

@@ -6,6 +6,7 @@ import type { SubmitWatchErrorCode } from "./dashboardAgentWatches.server";
  */
 const STATUS_BY_CODE: Record<SubmitWatchErrorCode, number> = {
   limit_reached: 409,
+  watch_limit_reached: 409,
   duplicate: 409,
   request_conflict: 409,
   invalid_target: 404,

apps/webapp/app/services/dashboardAgentWatchLimits.server.ts

@@ -0,0 +1,59 @@
+import type { Limits } from "@trigger.dev/platform";
+import { WATCH_MAX_HOURS } from "@internal/dashboard-agent-contracts";
+import { getCachedLimitAllowingZero, isBillingConfigured } from "./platform.v3.server";
+
+// The unlimited sentinel, matching the message quota (TRI-12863 P1). Never Infinity: it
+// serializes to null in the limit cache.
+export const UNLIMITED_WATCH_LIMIT = 100_000_000;
+
+// Filled by cloud billing (TRI-12863 P0). Absent until then, and always on self-hosted, so
+// the fallback applies and the plan floor is off.
+const WATCH_MAX_HOURS_LIMIT_KEY = "agentWatchMaxHours" as keyof Limits;
+const WATCH_COUNT_LIMIT_KEY = "agentWatchers" as keyof Limits;
+
+export type WatchPlanLimits = {
+  /** Longest window one watch may run for, in hours. */
+  maxHours: number;
+  /** How many active watches the org may run at once. */
+  watchers: number;
+};
+
+async function readLimit(organizationId: string, key: keyof Limits): Promise<number> {
+  // A plan of 0 means zero, not absent: an org with watches switched off must not read as
+  // unlimited. Only a missing limit falls open.
+  const cached = await getCachedLimitAllowingZero(organizationId, key, UNLIMITED_WATCH_LIMIT);
+  // A cache error leaves `val` empty; fall open to unlimited.
+  return cached.val ?? UNLIMITED_WATCH_LIMIT;
+}
+
+/**
+ * The org's plan floors for watches. Fails open: an absent limit (self-hosted, or before the
+ * cloud side ships) resolves to the unlimited sentinel, so neither floor bites. `read` is the
+ * plan-limit seam: tests pass their own reader instead of the cached platform one.
+ */
+export async function resolveWatchPlanLimits(
+  organizationId: string,
+  read: (organizationId: string, key: keyof Limits) => Promise<number> = readLimit
+): Promise<WatchPlanLimits> {
+  const [maxHours, watchers] = await Promise.all([
+    read(organizationId, WATCH_MAX_HOURS_LIMIT_KEY),
+    read(organizationId, WATCH_COUNT_LIMIT_KEY),
+  ]);
+  return { maxHours, watchers };
+}
+
+/**
+ * The window ceiling actually in force: the plan floor under the code ceiling. A plan that
+ * allows 100 hours still caps at {@link WATCH_MAX_HOURS}.
+ */
+export function effectiveWatchMaxHours(planMaxHours: number): number {
+  return Math.min(planMaxHours, WATCH_MAX_HOURS);
+}
+
+/**
+ * A watch-limit refusal, plus an upgrade nudge when billing is present. Self-hosted never
+ * hits this (fails open above), and the nudge is gated so a stray refusal stays quiet there.
+ */
+export function watchLimitHint(base: string, billingConfigured = isBillingConfigured()): string {
+  return billingConfigured ? `${base} Upgrade your plan for more.` : base;
+}

apps/webapp/app/services/dashboardAgentWatches.server.ts

@@ -10,6 +10,7 @@ import {
   cancelWatch,
   chatExists,
   claimWatchSubmission,
+  countActiveWatchesForOrg,
   createChat,
   createWatch,
   generateWatchId,
@@ -68,6 +69,12 @@ import {
 import { watchCreationCheckDeps } from "~/services/dashboardAgentWatchChecks.server";
 import { normalizeErrorFingerprint } from "~/services/dashboardAgentWatchErrorChecks";
 import { subscribeUserToWatchAlerts } from "~/services/dashboardAgentWatchAlerts.server";
+import {
+  effectiveWatchMaxHours,
+  resolveWatchPlanLimits,
+  watchLimitHint,
+  type WatchPlanLimits,
+} from "~/services/dashboardAgentWatchLimits.server";
 import {
   mintDashboardAgentWatchBatchToken,
   mintDashboardAgentWatchToken,
@@ -165,6 +172,7 @@ export async function authorizeWatchEnvironmentById(params: {
 
 export type CreateWatchErrorCode =
   | "limit_reached"
+  | "watch_limit_reached"
   | "duplicate"
   | "invalid_target"
   | "chat_not_found"
@@ -273,6 +281,12 @@ export async function createDashboardAgentWatch(params: {
     scheduleTick?: typeof scheduleWatchTick;
     /** Skip the real trigger-config gate when a tick scheduler is injected. */
     configured?: () => boolean;
+    /** Plan floors on window and count. Fails open to unlimited when absent. */
+    resolveLimits?: (organizationId: string) => Promise<WatchPlanLimits>;
+    /** Org-wide active-watch count, for the watcher-count floor. */
+    countActiveWatches?: (organizationId: string) => Promise<number>;
+    /** Gates the upgrade nudge, so self-hosted stays quiet. */
+    billingConfigured?: () => boolean;
   };
 }): Promise<CreateDashboardAgentWatchResult> {
   const { environment, userId, chatId } = params;
@@ -284,6 +298,11 @@ export async function createDashboardAgentWatch(params: {
   const buildCheckDeps = params.deps?.checkDeps ?? watchCreationCheckDeps;
   const scheduleTick = params.deps?.scheduleTick ?? scheduleWatchTick;
   const isDashboardAgentConfigured = params.deps?.configured ?? isDashboardAgentConfiguredDefault;
+  const resolveLimits = params.deps?.resolveLimits ?? resolveWatchPlanLimits;
+  const countActiveWatches =
+    params.deps?.countActiveWatches ??
+    ((organizationId: string) => countActiveWatchesForOrg(dashboardAgentDb, { organizationId }));
+  const hint = (base: string) => watchLimitHint(base, params.deps?.billingConfigured?.());
   const checkDeps = buildCheckDeps(environment, now);
 
   if (!isDashboardAgentConfigured()) {
@@ -331,6 +350,29 @@ export async function createDashboardAgentWatch(params: {
     return { ok: true, watching: false, identity, immediate };
   }
 
+  // Both floors are read only now the immediate check didn't answer: a one-shot creates no
+  // row, so a plan floor must not turn an answerable question into an upgrade nudge. Plan
+  // floors sit below the code ceilings (min(plan, ceiling)) and fail open: an absent limit
+  // resolves to unlimited, so neither bites on self-hosted.
+  const planLimits = await resolveLimits(environment.organizationId);
+  if (spec.maxHours > effectiveWatchMaxHours(planLimits.maxHours)) {
+    return {
+      ok: false,
+      code: "watch_limit_reached",
+      error: hint("That watch window is longer than your plan allows."),
+    };
+  }
+
+  // The per-chat cap of 3 still applies independently, in `createWatch`.
+  const activeCount = await countActiveWatches(environment.organizationId);
+  if (activeCount >= planLimits.watchers) {
+    return {
+      ok: false,
+      code: "watch_limit_reached",
+      error: hint("You've reached the number of active watches your plan allows."),
+    };
+  }
+
   const expiresAt = new Date(now.getTime() + spec.maxHours * 60 * 60 * 1000);
 
   const created = await createWatch(dashboardAgentDb, {

apps/webapp/app/services/platform.v3.server.ts

@@ -482,6 +482,37 @@ export async function getCachedLimit(orgId: string, limit: keyof Limits, fallbac
   });
 }
 
+/**
+ * Reads one plan limit, treating 0 as zero rather than absent: only a missing limit falls back.
+ * {@link getLimit} keeps its `!result` fallback, which its callers depend on.
+ */
+export function limitValueAllowingZero(
+  limits: Limits | undefined,
+  limit: keyof Limits,
+  fallback: number
+): number {
+  const result = limits?.[limit];
+
+  if (result === undefined || result === null) return fallback;
+  if (typeof result === "number") return result;
+  if (typeof result === "object" && "number" in result) return result.number;
+  return fallback;
+}
+
+/**
+ * Like {@link getCachedLimit}, but a plan value of 0 means zero. Cached under its own key so it
+ * never crosses with {@link getCachedLimit}.
+ */
+export async function getCachedLimitAllowingZero(
+  orgId: string,
+  limit: keyof Limits,
+  fallback: number
+) {
+  return platformCache.limits.swr(`${orgId}:${limit}:allow-zero`, async () =>
+    limitValueAllowingZero(await getLimits(orgId), limit, fallback)
+  );
+}
+
 export async function customerPortalUrl(orgId: string, orgSlug: string) {
   if (!client) return undefined;