fix(webapp): align buffer-path tags success count with PG (post-dedup)

Devin follow-ups on PR #3756.

* Tags route synthesisedResponse reported `nonEmptyTags.length` (pre-
  dedup input count) while the PG path reports `newTags.length` (post-
  dedup against existing tags). Same API call → different message
  count across the buffered/materialised boundary.

  mutateWithFallback now forwards the pre-mutation BufferEntry from its
  env-auth pre-check into synthesisedResponse + rejectedResponse, so
  the tags route can dedup against the snapshot's existing tags
  without an extra Redis round-trip. Same TOCTOU semantics as the PG
  path; the Lua keeps the actual write atomically deduped regardless.

* Replay route's `as unknown as TaskRun` cast bypasses TS for the
  buffered branch; if ReplayTaskRunService later reads an extra
  TaskRun field without that field being added to
  BufferedReplayInputSchema, the buffered branch silently feeds the
  service `undefined`. Run-time mitigation (safeParse + warn log +
  404) already exists; added an inline comment near the cast spelling
  out the manual sync requirement so future edits notice.

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

Author: d-cs

apps/webapp/app/routes/api.v1.runs.$runId.tags.ts

@@ -1,5 +1,6 @@
 import { type ActionFunctionArgs, json } from "@remix-run/server-runtime";
 import { AddTagsRequestBody } from "@trigger.dev/core/v3";
+import type { BufferEntry } from "@trigger.dev/redis-worker";
 import { z } from "zod";
 import { prisma } from "~/db.server";
 import { MAX_TAGS_PER_RUN } from "~/models/taskRunTag.server";
@@ -8,6 +9,22 @@ import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
 import { logger } from "~/services/logger.server";
 import { mutateWithFallback } from "~/v3/mollifier/mutateWithFallback.server";
 
+// Pull the existing tags out of a buffer entry's serialised payload so
+// the buffer-path response can dedup against them, matching the
+// PG-path's `newTags.length` count rather than the pre-dedup input
+// count. Returns null on any parse failure / shape mismatch so the
+// caller can fall back gracefully.
+function parseSnapshotTags(entry: BufferEntry | null): string[] | null {
+  if (!entry) return null;
+  try {
+    const snapshot = JSON.parse(entry.payload) as { tags?: unknown };
+    if (!Array.isArray(snapshot.tags)) return null;
+    return snapshot.tags.filter((t): t is string => typeof t === "string");
+  } catch {
+    return null;
+  }
+}
+
 const ParamsSchema = z.object({
   runId: z.string(),
 });
@@ -80,8 +97,23 @@ export async function action({ request, params }: ActionFunctionArgs) {
       // MAX_TAGS_PER_RUN via the `maxTags` we pass in `bufferPatch` —
       // matching the PG-path cap above so a buffered run can't exceed the
       // limit the trigger validator applies at creation.
-      synthesisedResponse: () =>
-        json({ message: `Successfully set ${nonEmptyTags.length} new tags.` }, { status: 200 }),
+      //
+      // Dedup the success-count off the pre-mutation entry (already
+      // fetched by mutateWithFallback's env-auth pre-check, so no extra
+      // Redis read) so the message reports the same `newTags.length` the
+      // PG path reports — not the pre-dedup request count, which would
+      // give an inconsistent number across the buffered/materialised
+      // boundary for the same input.
+      synthesisedResponse: ({ bufferEntry }) => {
+        const existing = parseSnapshotTags(bufferEntry);
+        const newTagsCount = existing
+          ? nonEmptyTags.filter((t) => !existing.includes(t)).length
+          : nonEmptyTags.length;
+        return json(
+          { message: `Successfully set ${newTagsCount} new tags.` },
+          { status: 200 }
+        );
+      },
       // Buffer rejected the append because it would exceed the cap. We
       // don't know the exact deduped overflow count here (the Lua does),
       // so report the limit rather than a precise "trying to set N".

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

@@ -93,6 +93,18 @@ export async function action({ request, params }: ActionFunctionArgs) {
       if (buffered) {
         const parsed = BufferedReplayInputSchema.safeParse(buffered);
         if (parsed.success) {
+          // Manual sync point: `BufferedReplayInputSchema` covers only
+          // the subset of `TaskRun` fields `ReplayTaskRunService.call`
+          // currently reads from `existingTaskRun`. The cast is `as
+          // unknown as TaskRun` because the full `TaskRun` type carries
+          // ~40 fields the service never touches; mirroring all of them
+          // on a synthetic snapshot would be misleading. If a future
+          // change to `ReplayTaskRunService` reads an additional
+          // `existingTaskRun` field, **add it to the schema above** —
+          // otherwise the buffered path will silently feed the service
+          // `undefined` for that field while the PG-source replay
+          // works. The `safeParse` + warn-log + 404 below is the
+          // run-time fail-safe; this comment is the design fail-safe.
           taskRun = parsed.data as unknown as TaskRun;
         } else {
           logger.warn("replay: buffered fallback failed schema validation", {

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

@@ -1,4 +1,5 @@
 import type {
+  BufferEntry,
   MollifierBuffer,
   MutateSnapshotResult,
   SnapshotPatch,
@@ -26,13 +27,26 @@ export type MutateWithFallbackInput<TResponse> = {
   // Receives the full TaskRun shape and returns the customer-visible body.
   pgMutation: (pgRow: TaskRun) => Promise<TResponse>;
   // Called when the patch landed cleanly on the buffer snapshot. The
-  // drainer will see the patched payload on its next pop.
-  synthesisedResponse: () => TResponse | Promise<TResponse>;
+  // drainer will see the patched payload on its next pop. Receives the
+  // pre-mutation snapshot entry (the one fetched for the env auth
+  // check above) so the caller can compute response details that
+  // depend on the prior state — e.g. the tags route needs to dedup
+  // against the existing tags to report an accurate `newTags` count
+  // matching the PG path, without an extra Redis round-trip.
+  // `bufferEntry` is `null` in the rare race where the entry didn't
+  // exist at pre-check time but appeared before `mutateSnapshot`.
+  synthesisedResponse: (ctx: {
+    bufferEntry: BufferEntry | null;
+  }) => TResponse | Promise<TResponse>;
   // Called when the buffer rejected the patch as invalid (e.g. an
   // `append_tags` patch carrying `maxTags` would exceed the cap). Required
   // only by callers that send a rejectable patch; the helper throws if the
-  // buffer reports a rejection and no builder was supplied.
-  rejectedResponse?: () => TResponse | Promise<TResponse>;
+  // buffer reports a rejection and no builder was supplied. Receives the
+  // same `bufferEntry` context as `synthesisedResponse` so a rejection
+  // message can reference the prior state if useful.
+  rejectedResponse?: (ctx: {
+    bufferEntry: BufferEntry | null;
+  }) => TResponse | Promise<TResponse>;
   abortSignal?: AbortSignal;
   // Override defaults for tests.
   safetyNetMs?: number;
@@ -110,7 +124,10 @@ export async function mutateWithFallback<TResponse>(
   );
 
   if (result === "applied_to_snapshot") {
-    return { kind: "snapshot", response: await input.synthesisedResponse() };
+    return {
+      kind: "snapshot",
+      response: await input.synthesisedResponse({ bufferEntry: entryForAuth }),
+    };
   }
 
   if (result === "limit_exceeded") {
@@ -122,7 +139,10 @@ export async function mutateWithFallback<TResponse>(
         "mutateWithFallback: buffer returned 'limit_exceeded' but no rejectedResponse was provided",
       );
     }
-    return { kind: "rejected", response: await input.rejectedResponse() };
+    return {
+      kind: "rejected",
+      response: await input.rejectedResponse({ bufferEntry: entryForAuth }),
+    };
   }
 
   if (result === "not_found") {

apps/webapp/test/mollifierMutateWithFallback.test.ts

@@ -133,6 +133,32 @@ describe("mutateWithFallback", () => {
     expect(pgMutation).not.toHaveBeenCalled();
   });
 
+  it("applied_to_snapshot forwards the pre-mutation entry to synthesisedResponse (lets callers dedup)", async () => {
+    // The tags route uses this to compute the same post-dedup count
+    // the PG path reports, without an extra Redis round-trip.
+    const synthesised = vi.fn(({ bufferEntry }: { bufferEntry: BufferEntry | null }) => {
+      // Caller can inspect bufferEntry.payload (or other fields) to
+      // produce a response that depends on the prior snapshot state.
+      return bufferEntry ? "snap-with-entry" : "snap-without-entry";
+    });
+    const result = await mutateWithFallback({
+      ...baseInput,
+      pgMutation: async () => "pg",
+      synthesisedResponse: synthesised,
+      prismaReplica: fakePrisma([null]) as unknown as typeof import("~/db.server").$replica,
+      prismaWriter: fakePrisma([]) as unknown as typeof import("~/db.server").prisma,
+      getBuffer: () => bufferReturning("applied_to_snapshot"),
+    });
+    expect(result).toEqual({ kind: "snapshot", response: "snap-with-entry" });
+    expect(synthesised).toHaveBeenCalledTimes(1);
+    const ctx = synthesised.mock.calls[0]?.[0];
+    expect(ctx?.bufferEntry).not.toBeNull();
+    // The pre-check entry has the env-matching shape set up by
+    // bufferReturning() / preCheckEntry().
+    expect(ctx?.bufferEntry?.envId).toBe("env_a");
+    expect(ctx?.bufferEntry?.orgId).toBe("org_1");
+  });
+
   it("replica miss + buffer not_found + writer miss → not_found", async () => {
     const result = await mutateWithFallback({
       ...baseInput,