fix(webapp): review fixes for mollifier dashboard parity

Address review findings on PR #3757:

- replay: guard `synthetic.traceId` / `spanId` before adapting a
  buffered snapshot into the TaskRun shape ReplayTaskRunService expects.
  Without the guard, an older snapshot missing those fields produces
  `00-undefined-undefined-01` as the W3C traceparent, which OTel
  silently drops — severing the replayed run's trace linkage. Extract
  the adapter into `buildSyntheticReplayTaskRun` so the guard is unit
  tested.
- run-detail: reflect the buffered snapshot's CANCELED state into the
  NavBar status + Cancel-button gate. Previously `tryMollifiedRunFallback`
  hardcoded `status: PENDING` / `isFinished: false` regardless of
  `buffered.status` / `cancelledAt`, so cancelling a buffered run left
  the NavBar showing Pending with a Cancel button until the drainer
  materialised the row. Extract the mapping into `buildSyntheticRunHeader`
  and unit test PENDING vs CANCELED.
- admin runs redirect (@.runs.$runParam): preselect the root span on
  the buffered redirect via `v3RunSpanPath`, matching the sibling
  redirect routes. Without this the trace tree opened with nothing
  selected for buffered runs reached via the admin impersonate path.
- RunStreamPresenter: switch from `deserialiseSnapshot` (redis-worker)
  to the webapp wrapper `deserialiseMollifierSnapshot` so both read-side
  modules share one deserialisation path, per the contract comment in
  `syntheticRedirectInfo.server.ts`.
- cancel route: collapse the `mutateSnapshot` "not_found" branch into
  the same retry message as "busy". Both indicate the drainer raced the
  mutation between `getEntry` and `mutateSnapshot`; the run is in PG by
  then and a retry hits the regular cancel path.

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

Author: d-cs

apps/webapp/app/presenters/v3/RunStreamPresenter.server.ts

@@ -3,6 +3,8 @@ import { logger } from "~/services/logger.server";
 import { singleton } from "~/utils/singleton";
 import { ABORT_REASON_SEND_ERROR, createSSELoader, SendFunction } from "~/utils/sse";
 import { throttle } from "~/utils/throttle";
+import { getMollifierBuffer } from "~/v3/mollifier/mollifierBuffer.server";
+import { deserialiseMollifierSnapshot } from "~/v3/mollifier/mollifierSnapshot.server";
 import { tracePubSub } from "~/v3/services/tracePubSub.server";
 
 const PING_INTERVAL = 5_000;
@@ -37,17 +39,48 @@ export class RunStreamPresenter {
           },
         });
 
-        if (!run) {
+        // Fall back to the mollifier buffer when the run isn't in PG yet.
+        // The buffered run has no execution events to stream, but we still
+        // attach a trace-pubsub subscription using the snapshot's traceId
+        // so that the moment the drainer materialises the row and execution
+        // begins, those events flow to this open SSE connection. Closing
+        // with 404 would force the dashboard to keep retrying.
+        let traceId: string | null = run?.traceId ?? null;
+        if (!traceId) {
+          const buffer = getMollifierBuffer();
+          if (buffer) {
+            try {
+              const entry = await buffer.getEntry(runFriendlyId);
+              if (entry) {
+                // Go through the webapp wrapper so this read-side module
+                // shares a single deserialisation path with readFallback —
+                // see the contract comment in syntheticRedirectInfo.server.ts.
+                const snapshot = deserialiseMollifierSnapshot(entry.payload);
+                if (typeof snapshot.traceId === "string") {
+                  traceId = snapshot.traceId;
+                }
+              }
+            } catch (err) {
+              logger.warn("RunStreamPresenter buffer fallback failed", {
+                runFriendlyId,
+                err: err instanceof Error ? err.message : String(err),
+              });
+            }
+          }
+        }
+
+        if (!traceId) {
           throw new Response("Not found", { status: 404 });
         }
+        const resolvedRun = { traceId };
 
         logger.info("RunStreamPresenter.start", {
           runFriendlyId,
-          traceId: run.traceId,
+          traceId: resolvedRun.traceId,
         });
 
         // Subscribe to trace updates
-        const { unsubscribe, eventEmitter } = await tracePubSub.subscribeToTrace(run.traceId);
+        const { unsubscribe, eventEmitter } = await tracePubSub.subscribeToTrace(resolvedRun.traceId);
 
         // Only send max every 1 second
         const throttledSend = throttle(
@@ -105,7 +138,7 @@ export class RunStreamPresenter {
           cleanup: () => {
             logger.info("RunStreamPresenter.cleanup", {
               runFriendlyId,
-              traceId: run.traceId,
+              traceId: resolvedRun.traceId,
             });
 
             // Remove message listener
@@ -119,13 +152,13 @@ export class RunStreamPresenter {
               .then(() => {
                 logger.info("RunStreamPresenter.cleanup.unsubscribe succeeded", {
                   runFriendlyId,
-                  traceId: run.traceId,
+                  traceId: resolvedRun.traceId,
                 });
               })
               .catch((error) => {
                 logger.error("RunStreamPresenter.cleanup.unsubscribe failed", {
                   runFriendlyId,
-                  traceId: run.traceId,
+                  traceId: resolvedRun.traceId,
                   error: {
                     name: error.name,
                     message: error.message,

apps/webapp/app/routes/@.runs.$runParam.ts

@@ -3,7 +3,8 @@ import { z } from "zod";
 import { prisma } from "~/db.server";
 import { redirectWithErrorMessage } from "~/models/message.server";
 import { requireUser } from "~/services/session.server";
-import { impersonate, rootPath, v3RunPath } from "~/utils/pathBuilder";
+import { impersonate, rootPath, v3RunPath, v3RunSpanPath } from "~/utils/pathBuilder";
+import { findBufferedRunRedirectInfo } from "~/v3/mollifier/syntheticRedirectInfo.server";
 
 const ParamsSchema = z.object({
   runParam: z.string(),
@@ -32,6 +33,7 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
       friendlyId: runParam,
     },
     select: {
+      spanId: true,
       runtimeEnvironment: {
         select: {
           slug: true,
@@ -51,16 +53,45 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
   });
 
   if (!run) {
+    // Admin impersonation route — bypass org membership so admins can
+    // open any buffered run by friendlyId, mirroring the existing PG
+    // behaviour above (no membership filter on the find).
+    const buffered = await findBufferedRunRedirectInfo({
+      runFriendlyId: runParam,
+      userId: user.id,
+      skipOrgMembershipCheck: true,
+    });
+    if (buffered) {
+      // Preselect the root span so the run-detail trace tree opens with
+      // the buffered run's span highlighted, matching the sibling
+      // redirect routes (runs.$runParam.ts, projects.v3.$projectRef…).
+      const path = buffered.spanId
+        ? v3RunSpanPath(
+            { slug: buffered.organizationSlug },
+            { slug: buffered.projectSlug },
+            { slug: buffered.environmentSlug },
+            { friendlyId: runParam },
+            { spanId: buffered.spanId }
+          )
+        : v3RunPath(
+            { slug: buffered.organizationSlug },
+            { slug: buffered.projectSlug },
+            { slug: buffered.environmentSlug },
+            { friendlyId: runParam }
+          );
+      return redirect(impersonate(path));
+    }
     return redirectWithErrorMessage(rootPath(), request, "Run doesn't exist", {
       ephemeral: false,
     });
   }
 
-  const path = v3RunPath(
+  const path = v3RunSpanPath(
     { slug: run.project.organization.slug },
     { slug: run.project.slug },
     { slug: run.runtimeEnvironment.slug },
-    { friendlyId: runParam }
+    { friendlyId: runParam },
+    { spanId: run.spanId }
   );
 
   return redirect(impersonate(path));

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam/route.tsx

@@ -88,10 +88,14 @@ import { useReplaceSearchParams } from "~/hooks/useReplaceSearchParams";
 import { useSearchParams } from "~/hooks/useSearchParam";
 import { type Shortcut, useShortcutKeys } from "~/hooks/useShortcutKeys";
 import { useHasAdminAccess } from "~/hooks/useUser";
+import { env } from "~/env.server";
 import { findProjectBySlug } from "~/models/project.server";
 import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
 import { NextRunListPresenter } from "~/presenters/v3/NextRunListPresenter.server";
 import { RunEnvironmentMismatchError, RunPresenter } from "~/presenters/v3/RunPresenter.server";
+import { findRunByIdWithMollifierFallback } from "~/v3/mollifier/readFallback.server";
+import { buildSyntheticRunHeader } from "~/v3/mollifier/syntheticRunHeader.server";
+import { buildSyntheticTraceForBufferedRun } from "~/v3/mollifier/syntheticTrace.server";
 import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server";
 import { getImpersonationId } from "~/services/impersonation.server";
 import { logger } from "~/services/logger.server";
@@ -277,6 +281,31 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
       );
     }
 
+    // PG miss → try the mollifier buffer. When the gate diverts a trigger
+    // the run sits in Redis until the drainer materialises it; without
+    // this fallback the run-detail page 404s for the brief buffered window
+    // even though the API has accepted the trigger and returned an id.
+    const buffered = await tryMollifiedRunFallback({
+      runFriendlyId: runParam,
+      organizationSlug,
+      projectSlug: projectParam,
+      envSlug: envParam,
+      userId,
+    });
+
+    if (buffered) {
+      const parent = await getResizableSnapshot(request, resizableSettings.parent.autosaveId);
+      const tree = await getResizableSnapshot(request, resizableSettings.tree.autosaveId);
+
+      return json({
+        run: buffered.run,
+        trace: buffered.trace,
+        maximumLiveReloadingSetting: env.MAXIMUM_LIVE_RELOADING_EVENTS,
+        resizable: { parent, tree },
+        runsList: null,
+      });
+    }
+
     throw error;
   }
 
@@ -305,6 +334,39 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
   });
 };
 
+async function tryMollifiedRunFallback(args: {
+  runFriendlyId: string;
+  organizationSlug: string;
+  projectSlug: string;
+  envSlug: string;
+  userId: string;
+}) {
+  const project = await findProjectBySlug(args.organizationSlug, args.projectSlug, args.userId);
+  if (!project) return null;
+  const environment = await findEnvironmentBySlug(project.id, args.envSlug, args.userId);
+  if (!environment) return null;
+
+  const buffered = await findRunByIdWithMollifierFallback({
+    runId: args.runFriendlyId,
+    environmentId: environment.id,
+    organizationId: project.organizationId,
+  });
+  if (!buffered) return null;
+
+  return {
+    run: buildSyntheticRunHeader({
+      run: buffered,
+      environment: {
+        id: environment.id,
+        organizationId: project.organizationId,
+        type: environment.type,
+        slug: environment.slug,
+      },
+    }),
+    trace: buildSyntheticTraceForBufferedRun(buffered),
+  };
+}
+
 type LoaderData = SerializeFrom<typeof loader>;
 
 export default function Page() {
@@ -407,23 +469,17 @@ export default function Page() {
             />
           </Dialog>
           {run.isFinished ? null : (
-            <Dialog key={`cancel-${run.friendlyId}`}>
-              <DialogTrigger asChild>
-                <Button variant="danger/small" LeadingIcon={StopCircleIcon} shortcut={{ key: "C" }}>
-                  Cancel run…
-                </Button>
-              </DialogTrigger>
-              <CancelRunDialog
-                runFriendlyId={run.friendlyId}
-                redirectPath={v3RunSpanPath(
-                  organization,
-                  project,
-                  environment,
-                  { friendlyId: run.friendlyId },
-                  { spanId: run.spanId }
-                )}
-              />
-            </Dialog>
+            <ControlledCancelRunDialog
+              key={`cancel-${run.friendlyId}`}
+              runFriendlyId={run.friendlyId}
+              redirectPath={v3RunSpanPath(
+                organization,
+                project,
+                environment,
+                { friendlyId: run.friendlyId },
+                { spanId: run.spanId }
+              )}
+            />
           )}
         </PageAccessories>
       </NavBar>
@@ -587,6 +643,35 @@ function TraceView({
   );
 }
 
+// Controlled wrapper around the cancel dialog. Owns the Radix open state
+// so the dialog closes itself once the cancel action transitions through
+// submission. We can't `<DialogClose asChild>`-wrap the submit button
+// because Radix's onClick handler swallows the button's name=value pair
+// that the form action depends on for `redirectUrl`.
+function ControlledCancelRunDialog({
+  runFriendlyId,
+  redirectPath,
+}: {
+  runFriendlyId: string;
+  redirectPath: string;
+}) {
+  const [open, setOpen] = useState(false);
+  return (
+    <Dialog open={open} onOpenChange={setOpen}>
+      <DialogTrigger asChild>
+        <Button variant="danger/small" LeadingIcon={StopCircleIcon} shortcut={{ key: "C" }}>
+          Cancel run…
+        </Button>
+      </DialogTrigger>
+      <CancelRunDialog
+        runFriendlyId={runFriendlyId}
+        redirectPath={redirectPath}
+        onCancelSubmitted={() => setOpen(false)}
+      />
+    </Dialog>
+  );
+}
+
 function NoLogsView({ run, resizable }: Pick<LoaderData, "run" | "resizable">) {
   const plan = useCurrentPlan();
   const organization = useOrganization();
@@ -616,9 +701,13 @@ function NoLogsView({ run, resizable }: Pick<LoaderData, "run" | "resizable">) {
         >
           <div className="grid h-full place-items-center">
             {daysSinceCompleted === undefined ? (
-              <InfoPanel variant="info" icon={InformationCircleIcon} title="We delete old logs">
+              <InfoPanel
+                variant="info"
+                icon={InformationCircleIcon}
+                title="Waiting to start"
+              >
                 <Paragraph variant="small">
-                  We tidy up older logs to keep things running smoothly.
+                  This run is queued. Logs will appear here once it begins executing.
                 </Paragraph>
               </InfoPanel>
             ) : isWithinLogRetention ? (

apps/webapp/app/routes/resources.taskruns.$runParam.cancel.ts

@@ -6,6 +6,7 @@ import { redirectWithErrorMessage, redirectWithSuccessMessage } from "~/models/m
 import { logger } from "~/services/logger.server";
 import { requireUserId } from "~/services/session.server";
 import { CancelTaskRunService } from "~/v3/services/cancelTaskRun.server";
+import { getMollifierBuffer } from "~/v3/mollifier/mollifierBuffer.server";
 
 export const cancelSchema = z.object({
   redirectUrl: z.string(),
@@ -42,15 +43,56 @@ export const action: ActionFunction = async ({ request, params }) => {
       },
     });
 
-    if (!taskRun) {
+    if (taskRun) {
+      const cancelRunService = new CancelTaskRunService();
+      await cancelRunService.call(taskRun);
+      return redirectWithSuccessMessage(submission.value.redirectUrl, request, `Canceled run`);
+    }
+
+    // PG miss — try the mollifier buffer. The customer can hit cancel
+    // on a buffered run from the dashboard during the burst window.
+    // Snapshot a `mark_cancelled` patch; the drainer's
+    // bifurcation routes the run to `engine.createCancelledRun` on
+    // next pop.
+    const buffer = getMollifierBuffer();
+    const entry = buffer ? await buffer.getEntry(runParam) : null;
+    if (!entry) {
       submission.error = { runParam: ["Run not found"] };
       return json(submission);
     }
 
-    const cancelRunService = new CancelTaskRunService();
-    await cancelRunService.call(taskRun);
+    // Dashboard auth: verify the requesting user is a member of the
+    // buffered run's org. The API path scopes by env id from the
+    // authenticated request; the dashboard route uses org-membership
+    // because the URL doesn't carry an envId.
+    const member = await prisma.orgMember.findFirst({
+      where: { userId, organizationId: entry.orgId },
+      select: { id: true },
+    });
+    if (!member) {
+      submission.error = { runParam: ["Run not found"] };
+      return json(submission);
+    }
 
-    return redirectWithSuccessMessage(submission.value.redirectUrl, request, `Canceled run`);
+    const result = await buffer!.mutateSnapshot(runParam, {
+      type: "mark_cancelled",
+      cancelledAt: new Date().toISOString(),
+      cancelReason: "Canceled by user",
+    });
+    if (result === "applied_to_snapshot") {
+      return redirectWithSuccessMessage(submission.value.redirectUrl, request, `Canceled run`);
+    }
+    // "not_found" or "busy" — both indicate the drainer raced us between
+    // the getEntry check above and mutateSnapshot. On "not_found" the
+    // entry was just popped and the PG row is in flight; on "busy" the
+    // drainer is mid-materialisation. Either way the customer should
+    // retry — by then the PG row exists and the regular cancel path at
+    // the top of this action takes over.
+    return redirectWithErrorMessage(
+      submission.value.redirectUrl,
+      request,
+      "Run is materialising — retry in a moment"
+    );
   } catch (error) {
     if (error instanceof Error) {
       logger.error("Failed to cancel run", {