feat(webapp): wire mollifier idempotency into trigger hot path (Phase B6b)

Three integration points that connect B6a's buffer-side primitives to
the customer-facing flow per Q5:

- IdempotencyKeyConcern.handleTriggerRequest falls through to
  buffer.lookupIdempotency after a PG miss. Buffered hits return
  isCached:true with a synthesised TaskRun via the existing
  findRunByIdWithMollifierFallback. Skipped when
  resumeParentOnCompletion is set: waitpoint blocking requires a PG
  row that doesn't exist yet; the follow-up accept SETNX still
  dedupes the trigger itself. Buffer outages fail open to "no cache
  hit" so the trigger hot path is never wedged by a transient Redis
  issue.

- mollifyTrigger passes idempotencyKey + taskIdentifier through to
  buffer.accept. The SETNX race loser receives duplicate_idempotency
  with the winner's runId; the API response echoes it with
  isCached:true, matching PG-side cache-hit shape.

- ResetIdempotencyKeyService calls buffer.resetIdempotency alongside
  the existing PG updateMany. 404 only fires when both stores report
  nothing bound. Buffer outage during reset is logged and treated as
  a miss; PG-side reset still works.

Author: d-cs

.server-changes/mollifier-idempotency-integration.md

@@ -0,0 +1,12 @@
+---
+area: webapp
+type: improvement
+---
+
+Wire the mollifier buffer's idempotency surface into the trigger hot path per Q5. Three connected changes:
+
+- `IdempotencyKeyConcern.handleTriggerRequest` now falls through to `buffer.lookupIdempotency` after a PG miss. A buffered cache hit synthesises a TaskRun via the existing `findRunByIdWithMollifierFallback` and returns `{ isCached: true, run }`. Skipped when `resumeParentOnCompletion` is set: blocking a parent on a buffered child via waitpoint requires a PG row that doesn't exist yet, and the follow-up accept's SETNX still catches the duplicate trigger itself. Buffer outages fail open to "no cache hit" so the trigger hot path can't be wedged by a transient Redis issue.
+
+- `mollifyTrigger` passes `idempotencyKey` + `taskIdentifier` through to `MollifierBuffer.accept`. When the buffer's SETNX races with another concurrent buffered trigger using the same key, the race loser receives `{ kind: "duplicate_idempotency", existingRunId }` and the API response echoes the winner's runId with `isCached: true`, matching PG-side cache-hit shape.
+
+- `ResetIdempotencyKeyService` calls `buffer.resetIdempotency` alongside the existing PG `updateMany`. The 404 only fires when both stores report nothing was bound. A buffer outage during reset is logged and treated as a miss — the PG side still works.

apps/webapp/app/runEngine/concerns/idempotencyKeys.server.ts

@@ -4,6 +4,8 @@ import { logger } from "~/services/logger.server";
 import { resolveIdempotencyKeyTTL } from "~/utils/idempotencyKeys.server";
 import type { RunEngine } from "~/v3/runEngine.server";
 import { shouldIdempotencyKeyBeCleared } from "~/v3/taskStatus";
+import { getMollifierBuffer } from "~/v3/mollifier/mollifierBuffer.server";
+import { findRunByIdWithMollifierFallback } from "~/v3/mollifier/readFallback.server";
 import type { TraceEventConcern, TriggerTaskRequest } from "../types";
 
 export type IdempotencyKeyConcernResult =
@@ -17,6 +19,47 @@ export class IdempotencyKeyConcern {
     private readonly traceEventConcern: TraceEventConcern
   ) {}
 
+  // Q5 buffer-side dedup. Resolves an idempotency key against the
+  // mollifier buffer when PG missed. Returns a SyntheticRun cast to
+  // TaskRun so the route handler (which only reads run.id / run.friendlyId)
+  // can echo the buffered run's friendlyId as a cached hit. Returns null
+  // for any failure or miss — buffer outages must not 500 the trigger
+  // hot path; we fail open to "no cache hit" and let the request through.
+  private async findBufferedRunWithIdempotency(
+    environmentId: string,
+    organizationId: string,
+    taskIdentifier: string,
+    idempotencyKey: string,
+  ): Promise<TaskRun | null> {
+    const buffer = getMollifierBuffer();
+    if (!buffer) return null;
+
+    let bufferedRunId: string | null;
+    try {
+      bufferedRunId = await buffer.lookupIdempotency({
+        envId: environmentId,
+        taskIdentifier,
+        idempotencyKey,
+      });
+    } catch (err) {
+      logger.error("IdempotencyKeyConcern: buffer lookupIdempotency failed", {
+        environmentId,
+        taskIdentifier,
+        err: err instanceof Error ? err.message : String(err),
+      });
+      return null;
+    }
+    if (!bufferedRunId) return null;
+
+    const synthetic = await findRunByIdWithMollifierFallback({
+      runId: bufferedRunId,
+      environmentId,
+      organizationId,
+    });
+    if (!synthetic) return null;
+    return synthetic as unknown as TaskRun;
+  }
+
   async handleTriggerRequest(
     request: TriggerTaskRequest,
     parentStore: string | undefined
@@ -44,6 +87,25 @@ export class IdempotencyKeyConcern {
         })
       : undefined;
 
+    // Buffer fallback per Q5 mollifier-idempotency design. PG missed —
+    // the same key may belong to a buffered run that hasn't materialised
+    // yet. Skipped when `resumeParentOnCompletion` is set: blocking a
+    // parent on a buffered child via waitpoint requires a PG row that
+    // doesn't exist yet. The follow-up accept's SETNX in mollifyTrigger
+    // still dedupes the trigger itself; the waitpoint just doesn't fire
+    // for this rare race window.
+    if (!existingRun && idempotencyKey && !request.body.options?.resumeParentOnCompletion) {
+      const buffered = await this.findBufferedRunWithIdempotency(
+        request.environment.id,
+        request.environment.organizationId,
+        request.taskId,
+        idempotencyKey,
+      );
+      if (buffered) {
+        return { isCached: true, run: buffered };
+      }
+    }
+
     if (existingRun) {
       // The idempotency key has expired
       if (existingRun.idempotencyKeyExpiresAt && existingRun.idempotencyKeyExpiresAt < new Date()) {

apps/webapp/app/runEngine/services/triggerTask.server.ts

@@ -445,6 +445,10 @@ export class RunEngineTriggerTaskService {
                 engineTriggerInput,
                 decision: mollifierOutcome.decision,
                 buffer: mollifierBuffer,
+                // Idempotency-key triple wires the buffer's SETNX into
+                // the trigger-time dedup symmetric with PG (Q5).
+                idempotencyKey,
+                taskIdentifier: taskId,
               });
 
               logger.info("mollifier.buffered", {

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

@@ -11,8 +11,12 @@ export type MollifyNotice = {
 export type MollifySyntheticResult = {
   run: { friendlyId: string };
   error: undefined;
-  isCached: false;
-  notice: MollifyNotice;
+  // The race-loser path (Q5): if accept's SETNX hit an existing
+  // buffered run with the same (env, task, idempotencyKey), the
+  // response echoes the winner's runId with isCached=true. The
+  // mollifier-queued notice is only attached for the happy accept.
+  isCached: boolean;
+  notice?: MollifyNotice;
 };
 
 const NOTICE: MollifyNotice = {
@@ -29,14 +33,35 @@ export async function mollifyTrigger(args: {
   engineTriggerInput: MollifierSnapshot;
   decision: Extract<TripDecision, { divert: true }>;
   buffer: MollifierBuffer;
+  // Optional idempotency context. When both are passed, accept SETNXes
+  // the lookup so the buffered window participates in trigger-time
+  // dedup symmetrically with PG (Q5).
+  idempotencyKey?: string;
+  taskIdentifier?: string;
 }): Promise<MollifySyntheticResult> {
-  await args.buffer.accept({
+  const result = await args.buffer.accept({
     runId: args.runFriendlyId,
     envId: args.environmentId,
     orgId: args.organizationId,
     payload: serialiseMollifierSnapshot(args.engineTriggerInput),
+    idempotencyKey: args.idempotencyKey,
+    taskIdentifier: args.taskIdentifier,
   });
 
+  if (result.kind === "duplicate_idempotency") {
+    // Race loser. Echo the winner's runId so the SDK's response shape
+    // matches PG-side idempotency cache hits.
+    return {
+      run: { friendlyId: result.existingRunId },
+      error: undefined,
+      isCached: true,
+    };
+  }
+
+  // Both "accepted" and "duplicate_run_id" produce the same customer-
+  // visible response: a buffered-trigger acknowledgement. The duplicate
+  // runId case is unreachable in practice (runIds are server-generated
+  // and unique) but is silently idempotent at the buffer layer either way.
   return {
     run: { friendlyId: args.runFriendlyId },
     error: undefined,

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

@@ -1,14 +1,15 @@
 import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { BaseService, ServiceValidationError } from "./baseService.server";
 import { logger } from "~/services/logger.server";
+import { getMollifierBuffer } from "~/v3/mollifier/mollifierBuffer.server";
 
 export class ResetIdempotencyKeyService extends BaseService {
   public async call(
     idempotencyKey: string,
     taskIdentifier: string,
     authenticatedEnv: AuthenticatedEnvironment
   ): Promise<{ id: string }> {
-    const { count } = await this._prisma.taskRun.updateMany({
+    const { count: pgCount } = await this._prisma.taskRun.updateMany({
       where: {
         idempotencyKey,
         taskIdentifier,
@@ -20,15 +21,41 @@ export class ResetIdempotencyKeyService extends BaseService {
       },
     });
 
-    if (count === 0) {
+    // Buffer-side reset (Q5): the key may belong to a buffered run that
+    // hasn't materialised yet. The PG updateMany above can't see it.
+    // resetIdempotency clears both the snapshot fields and the Redis
+    // lookup atomically. Returns null when nothing was bound there.
+    const buffer = getMollifierBuffer();
+    const bufferResult = buffer
+      ? await buffer
+          .resetIdempotency({
+            envId: authenticatedEnv.id,
+            taskIdentifier,
+            idempotencyKey,
+          })
+          .catch((err) => {
+            // Buffer outage shouldn't 500 the reset endpoint if PG
+            // already cleared something. Log and treat as a miss.
+            logger.error("ResetIdempotencyKeyService: buffer reset failed", {
+              idempotencyKey,
+              taskIdentifier,
+              err: err instanceof Error ? err.message : String(err),
+            });
+            return { clearedRunId: null };
+          })
+      : { clearedRunId: null };
+
+    const totalCount = pgCount + (bufferResult.clearedRunId ? 1 : 0);
+
+    if (totalCount === 0) {
       throw new ServiceValidationError(
         `No runs found with idempotency key: ${idempotencyKey} and task: ${taskIdentifier}`,
         404
       );
     }
 
     logger.info(
-      `Reset idempotency key: ${idempotencyKey} for task: ${taskIdentifier} in env: ${authenticatedEnv.id}, affected ${count} run(s)`
+      `Reset idempotency key: ${idempotencyKey} for task: ${taskIdentifier} in env: ${authenticatedEnv.id}, affected ${totalCount} run(s) (pg=${pgCount}, buffered=${bufferResult.clearedRunId ? 1 : 0})`
     );
 
     return { id: idempotencyKey };