feat(webapp): dashboard parity for mollifier-buffered runs

Synthesise the SpanRun shape from buffer snapshots so the run-detail
page's inspector panel renders identically to a PG-resident run. SSE
log stream, realtime stream resources, logs-download and debug resource
fall back to the buffer instead of 404-ing. Short-URL redirects resolve
buffered runs to the canonical dashboard URL. Bulk-cancel scans the
buffer alongside the ClickHouse selection so runs queued mid-burst are
included. Trigger response now carries the snapshot's spanId so the
dashboard's Run Test redirect opens the details panel without an extra
click.

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

Author: d-cs

.server-changes/mollifier-dashboard-parity.md

@@ -0,0 +1,16 @@
+---
+area: webapp
+type: feature
+---
+
+Dashboard parity for runs that live in the mollifier buffer. Synthesises
+the SpanRun shape from the buffer snapshot so the run-detail page's
+inspector panel renders identically to a PG-resident run. SSE log
+stream, realtime stream resources, logs-download and debug resources
+fall back to the buffer instead of 404-ing. Short-URL redirects
+(`/runs/{id}`, `/@/runs/{id}`, `/projects/v3/{ref}/runs/{id}`) resolve
+buffered runs to the canonical dashboard URL. Bulk-cancel scans the
+buffer alongside the ClickHouse selection so runs queued mid-burst are
+included in the action. Trigger response now carries the snapshot's
+spanId so the dashboard's Run Test redirect opens the details panel
+without an extra click.

apps/webapp/app/components/runs/MollifierBanner.tsx

@@ -1090,7 +1090,7 @@ const EnvironmentSchema = z
       .transform((v) => v ?? process.env.REDIS_PASSWORD),
     TRIGGER_MOLLIFIER_REDIS_TLS_DISABLED: z.string().default(process.env.REDIS_TLS_DISABLED ?? "false"),
     TRIGGER_MOLLIFIER_TRIP_WINDOW_MS: z.coerce.number().int().positive().default(200),
-    TRIGGER_MOLLIFIER_TRIP_THRESHOLD: z.coerce.number().int().positive().default(100),
+    TRIGGER_MOLLIFIER_TRIP_THRESHOLD: z.coerce.number().int().nonnegative().default(100),
     TRIGGER_MOLLIFIER_HOLD_MS: z.coerce.number().int().positive().default(500),
     TRIGGER_MOLLIFIER_DRAIN_CONCURRENCY: z.coerce.number().int().positive().default(50),
     TRIGGER_MOLLIFIER_ENTRY_TTL_S: z.coerce.number().int().positive().default(600),

apps/webapp/app/env.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 { deserialiseSnapshot } from "@trigger.dev/redis-worker";
 import { tracePubSub } from "~/v3/services/tracePubSub.server";
 
 const PING_INTERVAL = 5_000;
@@ -37,17 +39,45 @@ 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) {
+                const snapshot = deserialiseSnapshot<{ traceId?: string }>(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 +135,7 @@ export class RunStreamPresenter {
           cleanup: () => {
             logger.info("RunStreamPresenter.cleanup", {
               runFriendlyId,
-              traceId: run.traceId,
+              traceId: resolvedRun.traceId,
             });
 
             // Remove message listener
@@ -119,13 +149,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/presenters/v3/RunStreamPresenter.server.ts

@@ -4,6 +4,7 @@ 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 { findBufferedRunRedirectInfo } from "~/v3/mollifier/syntheticRedirectInfo.server";
 
 const ParamsSchema = z.object({
   runParam: z.string(),
@@ -51,6 +52,26 @@ 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) {
+      return redirect(
+        impersonate(
+          v3RunPath(
+            { slug: buffered.organizationSlug },
+            { slug: buffered.projectSlug },
+            { slug: buffered.environmentSlug },
+            { friendlyId: runParam }
+          )
+        )
+      );
+    }
     return redirectWithErrorMessage(rootPath(), request, "Run doesn't exist", {
       ephemeral: false,
     });

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

@@ -67,7 +67,6 @@ import {
   useTree,
 } from "~/components/primitives/TreeView/TreeView";
 import { type NodesState } from "~/components/primitives/TreeView/reducer";
-import { MollifierBanner } from "~/components/runs/MollifierBanner";
 import { CancelRunDialog } from "~/components/runs/v3/CancelRunDialog";
 import { ReplayRunDialog } from "~/components/runs/v3/ReplayRunDialog";
 import { getRunFiltersFromSearchParams } from "~/components/runs/v3/RunFilters";
@@ -95,6 +94,7 @@ 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 { buildSyntheticTraceForBufferedRun } from "~/v3/mollifier/syntheticTrace.server";
 import { clickhouseClient } from "~/services/clickhouseInstance.server";
 import { getImpersonationId } from "~/services/impersonation.server";
 import { logger } from "~/services/logger.server";
@@ -297,11 +297,10 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
 
       return json({
         run: buffered.run,
-        trace: undefined,
+        trace: buffered.trace,
         maximumLiveReloadingSetting: env.MAXIMUM_LIVE_RELOADING_EVENTS,
         resizable: { parent, tree },
         runsList: null,
-        isMollified: true,
       });
     }
 
@@ -330,7 +329,6 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
       tree,
     },
     runsList,
-    isMollified: false,
   });
 };
 
@@ -376,13 +374,14 @@ async function tryMollifiedRunFallback(args: {
         userName: undefined,
       },
     },
+    trace: buildSyntheticTraceForBufferedRun(buffered),
   };
 }
 
 type LoaderData = SerializeFrom<typeof loader>;
 
 export default function Page() {
-  const { run, trace, maximumLiveReloadingSetting, runsList, resizable, isMollified } =
+  const { run, trace, maximumLiveReloadingSetting, runsList, resizable } =
     useLoaderData<typeof loader>();
   const organization = useOrganization();
   const project = useProject();
@@ -502,7 +501,6 @@ export default function Page() {
         </PageAccessories>
       </NavBar>
       <PageBody scrollable={false}>
-        {isMollified ? <MollifierBanner className="mx-3 mt-3" /> : null}
         {trace ? (
           <TraceView
             run={run}
@@ -691,9 +689,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/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam/route.tsx

@@ -2,7 +2,8 @@ import { type LoaderFunctionArgs, redirect } from "@remix-run/server-runtime";
 import { z } from "zod";
 import { prisma } from "~/db.server";
 import { requireUserId } from "~/services/session.server";
-import { v3RunSpanPath } from "~/utils/pathBuilder";
+import { v3RunPath, v3RunSpanPath } from "~/utils/pathBuilder";
+import { findBufferedRunRedirectInfo } from "~/v3/mollifier/syntheticRedirectInfo.server";
 
 const ParamsSchema = z.object({
   projectRef: z.string(),
@@ -44,6 +45,28 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
   });
 
   if (!run) {
+    // Fall back to the mollifier buffer so a /projects/v3/{ref}/runs/{id}
+    // share link works during the buffered window.
+    const buffered = await findBufferedRunRedirectInfo({
+      runFriendlyId: validatedParams.runParam,
+      userId,
+    });
+    if (buffered) {
+      const url = new URL(request.url);
+      const searchParams = url.searchParams;
+      if (!searchParams.has("span") && buffered.spanId) {
+        searchParams.set("span", buffered.spanId);
+      }
+      return redirect(
+        v3RunPath(
+          { slug: buffered.organizationSlug },
+          { slug: buffered.projectSlug },
+          { slug: buffered.environmentSlug },
+          { friendlyId: validatedParams.runParam },
+          searchParams
+        )
+      );
+    }
     throw new Response("Not found", { status: 404 });
   }
 

apps/webapp/app/routes/projects.v3.$projectRef.runs.$runParam.ts

@@ -12,6 +12,7 @@ import {
 import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
 import { requireUserId } from "~/services/session.server";
 import { EnvironmentParamSchema } from "~/utils/pathBuilder";
+import { findRunByIdWithMollifierFallback } from "~/v3/mollifier/readFallback.server";
 
 const ParamsSchema = z.object({
   runParam: z.string(),
@@ -59,6 +60,20 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
   });
 
   if (!run) {
+    // Buffered run has no Session linkage yet. Return 204 so the SDK's
+    // SSE client treats this as "channel not yet active" and retries
+    // naturally once the drainer materialises the row.
+    const buffered = await findRunByIdWithMollifierFallback({
+      runId: runParam,
+      environmentId: environment.id,
+      organizationId: project.organizationId,
+    });
+    if (buffered) {
+      return new Response(null, {
+        status: 204,
+        headers: { "content-type": "text/event-stream; charset=utf-8" },
+      });
+    }
     return new Response("Run not found", { status: 404 });
   }