chore(mollifier): address review follow-ups for phase-1 PR

- Gate drainer init on WORKER_ENABLED so only worker replicas run the polling loop.
- Update the enqueueSystem TTL comment now that delayed/pending-version are first enqueues.
- Correct the mollifier gate docstring to describe the fixed-window counter and tripped-key rearm.
- Swap findUnique for findFirst in the trigger task test to match the webapp Prisma rule.

Author: d-cs

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

@@ -131,6 +131,14 @@ export async function init() {
     await workerQueue.initialize();
   }
 
+  // Only the worker role drains the mollifier buffer. API-only replicas
+  // still produce into the buffer via the trigger hot path, but the
+  // polling loop + Redis consumer connection only belongs on workers —
+  // otherwise every webapp replica races for the same entries.
+  if (env.WORKER_ENABLED !== "true") {
+    return;
+  }
+
   try {
     const drainer = getMollifierDrainer();
     if (drainer && !global.__mollifierShutdownRegistered__) {

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

@@ -10,10 +10,15 @@ import {
   type DecisionReason,
 } from "./mollifierTelemetry.server";
 
-// `count` is the *single-instance* sliding-window counter, not a fleet-wide
-// aggregate. Each webapp instance maintains its own Redis key, so the fleet
-// effective ceiling is `instance_count * threshold`. Phase 2 consumers must
-// not treat `count` as a global rate.
+// `count` is the *single-instance* fixed-window counter (INCR with a PEXPIRE
+// armed on the first tick of each window — see `mollifierEvaluateTrip` in
+// `packages/redis-worker/src/mollifier/buffer.ts`). It is not a fleet-wide
+// aggregate: each webapp instance maintains its own Redis key, so the fleet
+// effective ceiling is `instance_count * threshold`, and at a window boundary
+// the instance can briefly admit up to ~2x threshold before tripping. The
+// tripped marker is refreshed on every overage call, so a sustained burst
+// holds the divert state until the rate falls below threshold within a
+// window. Phase 2 consumers must not treat `count` as a global rate.
 export type TripDecision =
   | { divert: false }
   | {

apps/webapp/test/engine/triggerTask.test.ts

@@ -1342,7 +1342,7 @@ describe("RunEngineTriggerTaskService", () => {
       // engine.trigger ran — Postgres has the run
       expect(result).toBeDefined();
       expect(result?.run.friendlyId).toBeDefined();
-      const pgRun = await prisma.taskRun.findUnique({ where: { id: result!.run.id } });
+      const pgRun = await prisma.taskRun.findFirst({ where: { id: result!.run.id } });
       expect(pgRun).not.toBeNull();
       expect(pgRun!.friendlyId).toBe(result!.run.friendlyId);
 

internal-packages/run-engine/src/engine/systems/enqueueSystem.ts

@@ -88,8 +88,11 @@ export class EnqueueSystem {
 
       const timestamp = (run.queueTimestamp ?? run.createdAt).getTime() - run.priorityMs;
 
-      // Include TTL only when explicitly requested (first enqueue from trigger).
-      // Re-enqueues (waitpoint, checkpoint, delayed, pending version) must not add TTL.
+      // Include TTL only when explicitly requested. Callers pass `includeTtl: true`
+      // on the first enqueue that puts the run into the run queue — the initial
+      // trigger path, the delayed run release, and the pending-version release.
+      // Waitpoint/checkpoint re-enqueues must NOT pass it: the run has already
+      // started, so the queued-but-never-started TTL no longer applies.
       let ttlExpiresAt: number | undefined;
       if (includeTtl && run.ttl) {
         const expireAt = parseNaturalLanguageDuration(run.ttl);