fix(webapp): explicit opt-in for mollifier stale-sweep + SYSTEM_FAILURE fallback on cancel-bifurcation

Two fixes from CodeRabbit on PR #3754:

1. `TRIGGER_MOLLIFIER_STALE_SWEEP_ENABLED` was inheriting
   `TRIGGER_MOLLIFIER_ENABLED`, so any deployment already running the
   mollifier would auto-start the sweep worker on upgrade — defeating
   the point of having a separate kill switch. Hard-default to "0" so
   the sweep is an explicit ops decision.

2. In the cancel-bifurcation path, a non-conflict + non-retryable
   error from `createCancelledRun` rethrew out of the handler. The
   drainer's `onTerminalFailure` handler gates on
   `cause === "max-attempts-exhausted"` and skips "non-retryable", so
   `buffer.fail()` deleted the entry without ever writing a PG row —
   the cancelled run disappeared silently. Mirror the SYSTEM_FAILURE
   fallback the non-cancel trigger path already uses: write a terminal
   row via the shared helper, falling back to the original throw only
   when the write also fails non-retryably (and re-throw retryable
   write errors so the drainer requeues).

UX trade: the customer initiated a cancel but sees SYSTEM_FAILURE if
the cancel write itself was structurally rejected. Terminal-but-
mislabelled beats terminal silence — the alternative was the run
disappearing from the dashboard entirely. The path is narrow (only
non-conflict, non-retryable createCancelledRun failures).

Two new unit tests cover the cancel branch: non-retryable creates the
SYSTEM_FAILURE row + ACKs the entry; retryable rethrows so the
drainer requeues.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: d-cs

apps/webapp/app/env.server.ts

@@ -1104,13 +1104,19 @@ const EnvironmentSchema = z
     // Periodic sweep that scans buffer queue LISTs for entries whose
     // dwell exceeds the stale threshold. Independent of the drainer —
     // its job is exactly to make a stuck/offline drainer visible to
-    // ops. Defaults: enabled when the mollifier is enabled, run every
-    // 5 minutes, alert on anything that's been dwelling for 5+ minutes
-    // (matches the sweep interval — "anything still here when we
-    // check" is the simplest threshold that converges).
-    TRIGGER_MOLLIFIER_STALE_SWEEP_ENABLED: z
-      .string()
-      .default(process.env.TRIGGER_MOLLIFIER_ENABLED ?? "0"),
+    // ops. Defaults: explicitly opt-in (a separate kill switch from
+    // the mollifier itself), run every 5 minutes, alert on anything
+    // that's been dwelling for 5+ minutes (matches the sweep interval
+    // — "anything still here when we check" is the simplest threshold
+    // that converges).
+    //
+    // The sweep was previously defaulting to inherit
+    // `TRIGGER_MOLLIFIER_ENABLED`, which meant any deployment already
+    // running with the mollifier on would auto-start the sweep worker
+    // on upgrade — turning on new background load with no explicit
+    // rollout step. Hard-defaulting to "0" preserves the intent of
+    // exposing the sweep as a separate switch.
+    TRIGGER_MOLLIFIER_STALE_SWEEP_ENABLED: z.string().default("0"),
     TRIGGER_MOLLIFIER_STALE_SWEEP_INTERVAL_MS: z.coerce
       .number()
       .int()

apps/webapp/app/v3/mollifier/mollifierDrainerHandler.server.ts

@@ -103,6 +103,39 @@ export function createDrainerHandler(deps: {
             const isConflict =
               err instanceof Error && err.message.startsWith("createCancelledRun conflict");
             if (!isConflict) {
+              // Mirror the SYSTEM_FAILURE fallback the non-cancelled
+              // trigger path uses below. Without this branch, a
+              // non-retryable createCancelledRun failure rethrows, the
+              // drainer's onTerminalFailure handler skips because it
+              // gates on `cause === "max-attempts-exhausted"` (and the
+              // outer drainer classifies non-retryable failures with
+              // `cause: "non-retryable"`), and buffer.fail() deletes
+              // the entry — leaving NO PG row. The cancellation
+              // disappears silently from the customer's dashboard.
+              // Writing a SYSTEM_FAILURE row gives the run a terminal,
+              // visible state.
+              if (isRetryablePgError(err)) {
+                throw err;
+              }
+              span.setAttribute("mollifier.cancel_terminal_failure_reason",
+                err instanceof Error ? err.message : String(err));
+              try {
+                const wrote = await writeMollifierTerminalFailureRow(deps, {
+                  friendlyId: input.runId,
+                  snapshot: input.payload as Record<string, unknown>,
+                  reason: err instanceof Error ? err.message : String(err),
+                });
+                if (wrote) return;
+              } catch (writeErr) {
+                if (isRetryablePgError(writeErr)) {
+                  span.setAttribute("mollifier.cancel_terminal_write_retryable", true);
+                  throw writeErr;
+                }
+                span.setAttribute(
+                  "mollifier.cancel_terminal_write_error",
+                  writeErr instanceof Error ? writeErr.message : String(writeErr)
+                );
+              }
               throw err;
             }
             span.setAttribute("mollifier.cancel_conflict", true);

apps/webapp/test/mollifierDrainerHandler.test.ts

@@ -348,6 +348,85 @@ describe("createDrainerHandler", () => {
     ).rejects.toThrow("Can't reach database server");
   });
 
+  it("writes a SYSTEM_FAILURE row when createCancelledRun fails non-retryably (cancel bifurcation)", async () => {
+    // Without this guard a non-conflict, non-retryable failure from
+    // createCancelledRun rethrows out of the handler. The drainer's
+    // onTerminalFailure gates on cause==="max-attempts-exhausted" and
+    // skips "non-retryable", so buffer.fail() deletes the entry with
+    // no PG row written — the cancellation disappears silently.
+    // Mirror the non-cancel path's SYSTEM_FAILURE fallback so the
+    // customer always sees a terminal row.
+    const friendlyId = RunId.generate().friendlyId;
+    const cancelErr = new Error("validation failed: bad cancel snapshot");
+    const createCancelledRun = vi.fn(async () => {
+      throw cancelErr;
+    });
+    const createFailedTaskRun = vi.fn(async () => ({ id: "internal_x" }));
+    const handler = createDrainerHandler({
+      engine: { createCancelledRun, createFailedTaskRun } as any,
+      prisma: {} as any,
+    });
+
+    await handler({
+      runId: friendlyId,
+      envId: "env_a",
+      orgId: "org_1",
+      payload: {
+        friendlyId,
+        taskIdentifier: "t",
+        environment: envFixture,
+        cancelledAt: new Date().toISOString(),
+        cancelReason: "Canceled by user",
+      },
+      attempts: 0,
+      createdAt: new Date(),
+    } as any);
+
+    // SYSTEM_FAILURE row was written via the shared helper. Handler
+    // returns cleanly so the drainer ACKs the entry instead of
+    // buffer.fail()ing it.
+    expect(createFailedTaskRun).toHaveBeenCalledOnce();
+    expect(createFailedTaskRun.mock.calls[0][0].friendlyId).toBe(friendlyId);
+    expect(createFailedTaskRun.mock.calls[0][0].error.raw).toContain(
+      "validation failed: bad cancel snapshot"
+    );
+  });
+
+  it("requeues when createCancelledRun fails with a retryable PG error (cancel bifurcation)", async () => {
+    // Retryable PG failures must rethrow so the drainer requeues the
+    // entry — writing a SYSTEM_FAILURE row when PG is transiently
+    // unreachable would still fail. The drainer's existing retry loop
+    // handles the requeue.
+    const friendlyId = RunId.generate().friendlyId;
+    const cancelErr = new Error("Can't reach database server");
+    const createCancelledRun = vi.fn(async () => {
+      throw cancelErr;
+    });
+    const createFailedTaskRun = vi.fn();
+    const handler = createDrainerHandler({
+      engine: { createCancelledRun, createFailedTaskRun } as any,
+      prisma: {} as any,
+    });
+
+    await expect(
+      handler({
+        runId: friendlyId,
+        envId: "env_a",
+        orgId: "org_1",
+        payload: {
+          friendlyId,
+          taskIdentifier: "t",
+          environment: envFixture,
+          cancelledAt: new Date().toISOString(),
+          cancelReason: "Canceled by user",
+        },
+        attempts: 0,
+        createdAt: new Date(),
+      } as any)
+    ).rejects.toThrow("Can't reach database server");
+    expect(createFailedTaskRun).not.toHaveBeenCalled();
+  });
+
   it("rethrows the original error when the snapshot lacks an environment block", async () => {
     const triggerErr = new Error("engine rejected the snapshot");
     const trigger = vi.fn(async () => {