feat(webapp): reschedule + replay APIs handle buffered runs (Phase C4 + C5)

Reschedule (C4): switches to mutateWithFallback. PG hits go through
the existing RescheduleTaskRunService (which enforces status ===
"DELAYED"). Buffered hits land a set_delay patch on the snapshot;
the drainer materialises the PG row with the new delayUntil. Synth-
esised response returns { id, delayUntil }.

Replay (C5): adds a read-fallback after the PG miss. The B4-extended
SyntheticRun carries every field ReplayTaskRunService reads from a
TaskRun, so the buffered case casts through and uses the existing
service unchanged. Replay creates a fresh trigger that itself
re-enters the mollifier gate — no special surge handling needed
beyond what the gate already does. Also tightens the PG lookup to
findFirst with runtimeEnvironmentId scoping (was findUnique on
friendlyId only).

Author: d-cs

.server-changes/mollifier-reschedule-replay-buffered.md

@@ -0,0 +1,10 @@
+---
+area: webapp
+type: feature
+---
+
+Reschedule and replay APIs now handle buffered runs.
+
+`POST /api/v1/runs/{id}/reschedule` switches to `mutateWithFallback`. PG hits go through the existing `RescheduleTaskRunService` (which enforces `status === "DELAYED"`). Buffered-QUEUED hits land a `set_delay` patch on the snapshot; the drainer materialises the PG row with the new `delayUntil`. `busy` snapshots wait for drainer resolution then route through PG. Synthesised response returns `{ id, delayUntil }` for the SDK to confirm.
+
+`POST /api/v1/runs/{id}/replay` adds a read-fallback after the PG miss: when the original run is still in the buffer, the synthesised TaskRun (extended in Phase B4 with all `ReplayTaskRunService`-relevant fields) is passed straight to the existing replay service. Replay creates a fresh trigger that itself re-enters the mollifier gate — no special surge handling needed. Also tightens the PG lookup to `findFirst` with `runtimeEnvironmentId` scoping; the prior `findUnique` left auth boundary checks to the upper layer.

apps/webapp/app/routes/api.v1.runs.$runParam.replay.ts

@@ -1,10 +1,12 @@
 import type { ActionFunctionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
+import type { TaskRun } from "@trigger.dev/database";
 import { z } from "zod";
 import { prisma } from "~/db.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { ReplayTaskRunService } from "~/v3/services/replayTaskRun.server";
+import { findRunByIdWithMollifierFallback } from "~/v3/mollifier/readFallback.server";
 import { sanitizeTriggerSource } from "~/utils/triggerSource";
 
 const ParamsSchema = z.object({
@@ -32,12 +34,34 @@ export async function action({ request, params }: ActionFunctionArgs) {
   const { runParam } = parsed.data;
 
   try {
-    const taskRun = await prisma.taskRun.findUnique({
+    const env = authenticationResult.environment;
+    // PG-first. Replay works on any status per audit (Q2 design) — no
+    // filter beyond friendlyId is the existing semantic; findFirst with
+    // env scoping tightens it minimally without changing behaviour for
+    // a correctly-authed caller.
+    let taskRun: TaskRun | null = await prisma.taskRun.findFirst({
       where: {
         friendlyId: runParam,
+        runtimeEnvironmentId: env.id,
       },
     });
 
+    if (!taskRun) {
+      // Buffered fallback (Q2). The SyntheticRun shape was extended in
+      // Phase B4 to carry every field ReplayTaskRunService reads from a
+      // TaskRun. Cast through unknown — the synthesised object has the
+      // same field surface as a real PG row from the service's
+      // perspective.
+      const buffered = await findRunByIdWithMollifierFallback({
+        runId: runParam,
+        environmentId: env.id,
+        organizationId: env.organizationId,
+      });
+      if (buffered) {
+        taskRun = buffered as unknown as TaskRun;
+      }
+    }
+
     if (!taskRun) {
       return json({ error: "Run not found" }, { status: 404 });
     }

apps/webapp/app/routes/api.v1.runs.$runParam.reschedule.ts

@@ -3,90 +3,107 @@ import { json } from "@remix-run/server-runtime";
 import { RescheduleRunRequestBody } from "@trigger.dev/core/v3/schemas";
 import { z } from "zod";
 import { getApiVersion } from "~/api/versions";
-import { prisma } from "~/db.server";
 import { ApiRetrieveRunPresenter } from "~/presenters/v3/ApiRetrieveRunPresenter.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
 import { logger } from "~/services/logger.server";
 import { ServiceValidationError } from "~/v3/services/baseService.server";
 import { RescheduleTaskRunService } from "~/v3/services/rescheduleTaskRun.server";
+import { mutateWithFallback } from "~/v3/mollifier/mutateWithFallback.server";
+import { parseDelay } from "~/utils/delays";
 
 const ParamsSchema = z.object({
   runParam: z.string(),
 });
 
 export async function action({ request, params }: ActionFunctionArgs) {
-  // Ensure this is a POST request
   if (request.method.toUpperCase() !== "POST") {
     return { status: 405, body: "Method Not Allowed" };
   }
 
-  // Authenticate the request
   const authenticationResult = await authenticateApiRequest(request);
-
   if (!authenticationResult) {
     return json({ error: "Invalid or missing API Key" }, { status: 401 });
   }
 
   const parsed = ParamsSchema.safeParse(params);
-
   if (!parsed.success) {
     return json({ error: "Invalid or missing run ID" }, { status: 400 });
   }
 
-  const { runParam } = parsed.data;
-
-  const taskRun = await prisma.taskRun.findUnique({
-    where: {
-      friendlyId: runParam,
-      runtimeEnvironmentId: authenticationResult.environment.id,
-    },
-  });
-
-  if (!taskRun) {
-    return json({ error: "Run not found" }, { status: 404 });
-  }
-
   const anyBody = await request.json();
-
   const body = RescheduleRunRequestBody.safeParse(anyBody);
-
   if (!body.success) {
     return json({ error: "Invalid request body" }, { status: 400 });
   }
 
-  const service = new RescheduleTaskRunService();
+  const env = authenticationResult.environment;
+  // Pre-resolve the absolute Date the buffer snapshot should encode.
+  // RescheduleTaskRunService expects this to be present on the body for
+  // its PG-side flow; for the buffer-side patch we encode the same
+  // wall-clock value so the drainer's engine.trigger sees the intended
+  // delayUntil after materialisation.
+  const delayUntil = await parseDelay(body.data.delay);
+  if (!delayUntil) {
+    return json({ error: "Invalid delay value" }, { status: 400 });
+  }
 
   try {
-    const updatedRun = await service.call(taskRun, body.data);
-
-    if (!updatedRun) {
-      return json({ error: "An unknown error occurred" }, { status: 500 });
-    }
-
-    const run = await ApiRetrieveRunPresenter.findRun(
-      updatedRun.friendlyId,
-      authenticationResult.environment
-    );
-
-    if (!run) {
+    const outcome = await mutateWithFallback<Response>({
+      runId: parsed.data.runParam,
+      environmentId: env.id,
+      organizationId: env.organizationId,
+      bufferPatch: {
+        type: "set_delay",
+        delayUntil: delayUntil.toISOString(),
+      },
+      pgMutation: async (taskRun) => {
+        const service = new RescheduleTaskRunService();
+        const updatedRun = await service.call(taskRun, body.data);
+        if (!updatedRun) {
+          return json({ error: "An unknown error occurred" }, { status: 500 });
+        }
+
+        const run = await ApiRetrieveRunPresenter.findRun(updatedRun.friendlyId, env);
+        if (!run) {
+          return json({ error: "Run not found" }, { status: 404 });
+        }
+        const apiVersion = getApiVersion(request);
+        const presenter = new ApiRetrieveRunPresenter(apiVersion);
+        const result = await presenter.call(run, env);
+        if (!result) {
+          return json({ error: "Run not found" }, { status: 404 });
+        }
+        return json(result);
+      },
+      // Buffered snapshot has been patched. Synthesise a minimal
+      // retrieve-shape response — the run hasn't materialised yet, so
+      // the presenter's full pass would synthesise mostly defaults
+      // anyway. Returning the friendlyId + the new delay is sufficient
+      // for SDK confirmation; subsequent retrieve calls go through the
+      // existing presenter with read-fallback (Phase A).
+      synthesisedResponse: () =>
+        json(
+          {
+            id: parsed.data.runParam,
+            delayUntil: delayUntil.toISOString(),
+          },
+          { status: 200 }
+        ),
+      abortSignal: getRequestAbortSignal(),
+    });
+
+    if (outcome.kind === "not_found") {
       return json({ error: "Run not found" }, { status: 404 });
     }
-
-    const apiVersion = getApiVersion(request);
-
-    const presenter = new ApiRetrieveRunPresenter(apiVersion);
-    const result = await presenter.call(run, authenticationResult.environment);
-
-    if (!result) {
-      return json({ error: "Run not found" }, { status: 404 });
+    if (outcome.kind === "timed_out") {
+      return json({ error: "Run materialisation timed out" }, { status: 503 });
     }
-
-    return json(result);
+    return outcome.response;
   } catch (error) {
     if (error instanceof ServiceValidationError) {
       return json({ error: error.message }, { status: 400 });
     }
-
     logger.error("Failed to reschedule run", { error });
     return json({ error: "Something went wrong, please try again." }, { status: 500 });
   }