feat(webapp): mollifier read-fallback for spans + attempts + metadata-get

d-cs · claude · d-cs · commit e21dbee5e9b8 · 2026-05-20T13:48:03.000+01:00
Phase A2/A5/A6 of the mollifier API parity work — three more read
endpoints get the buffer fallback, plus two route-level bug fixes for
endpoints that had no GET handler.

A2 spans/{spanId}: discriminated PG vs buffered findResource (mirrors
the trace endpoint pattern from A1). For buffered runs, the only valid
spanId is the snapshot's queued spanId (recorded at gate time, reused
as the run's root spanId on materialise). That spanId returns a minimal
"span exists, no execution data yet" shape; any other spanId is a
deterministic 404.

A5 attempts: pre-existing route-bug fix. The route only had `action`
(POST creates attempt); GET hit Remix's "no loader" 400 with an internal
error message. New loader returns 200 `{ attempts: [] }` for both PG
and buffered runs. The detailed attempt list belongs on the v3 retrieve
endpoint, not here.

A6 metadata GET: same pre-existing route-bug. The route only had PUT;
GET had no handler. New loader returns
`{ metadata, metadataType }` from either the PG row or the buffer
snapshot. PG-side reads only the two fields it needs.

A3 events and A4 result need no code change — events already works via
`ApiRetrieveRunPresenter.findRun`'s existing buffer fallback (querying
events for a buffered traceId naturally returns `{ events: [] }`), and
result's 404 message "Run either doesn't exist or is not finished"
already covers both buffered-not-in-PG and PG-delayed-not-finished
cases.

Co-Authored-By: Claude Opus 4.7 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/apps/webapp/app/routes/api.v1.runs.$runId.metadata.ts b/apps/webapp/app/routes/api.v1.runs.$runId.metadata.ts
@@ -1,15 +1,66 @@
+import type { LoaderFunctionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
 import { tryCatch } from "@trigger.dev/core/utils";
 import { UpdateMetadataRequestBody } from "@trigger.dev/core/v3";
 import { z } from "zod";
+import { $replica } from "~/db.server";
+import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { updateMetadataService } from "~/services/metadata/updateMetadataInstance.server";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 import { ServiceValidationError } from "~/v3/services/common.server";
+import { findRunByIdWithMollifierFallback } from "~/v3/mollifier/readFallback.server";
 
 const ParamsSchema = z.object({
   runId: z.string(),
 });
 
+// Phase A6 — fixes the pre-existing route bug where GET on this URL
+// returned a Remix "no loader" 400. The route only exposed PUT (update);
+// GET had no handler. Returns `{ metadata, metadataType }` from either
+// the Postgres row or the mollifier buffer snapshot.
+export async function loader({ request, params }: LoaderFunctionArgs) {
+  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 env = authenticationResult.environment;
+
+  const pgRun = await $replica.taskRun.findFirst({
+    where: { friendlyId: parsed.data.runId, runtimeEnvironmentId: env.id },
+    select: { metadata: true, metadataType: true },
+  });
+  if (pgRun) {
+    return json({ metadata: pgRun.metadata, metadataType: pgRun.metadataType }, { status: 200 });
+  }
+
+  const buffered = await findRunByIdWithMollifierFallback({
+    runId: parsed.data.runId,
+    environmentId: env.id,
+    organizationId: env.organizationId,
+  });
+  if (buffered) {
+    // Buffered snapshot stores metadata as the original packet shape
+    // (could be a string for application/json payloads). Pass through
+    // without re-encoding — the consumer expects the same shape PG would
+    // return.
+    return json(
+      {
+        metadata: buffered.metadata ?? null,
+        metadataType: buffered.metadataType ?? "application/json",
+      },
+      { status: 200 }
+    );
+  }
+
+  return json({ error: "Run not found" }, { status: 404 });
+}
+
 const { action } = createActionApiRoute(
   {
     params: ParamsSchema,
diff --git a/apps/webapp/app/routes/api.v1.runs.$runId.spans.$spanId.ts b/apps/webapp/app/routes/api.v1.runs.$runId.spans.$spanId.ts
@@ -9,42 +9,106 @@ import {
 } from "~/services/routeBuilders/apiBuilder.server";
 import { resolveEventRepositoryForStore } from "~/v3/eventRepository/index.server";
 import { getTaskEventStoreTableForRun } from "~/v3/taskEventStore.server";
+import { findRunByIdWithMollifierFallback } from "~/v3/mollifier/readFallback.server";
 
 const ParamsSchema = z.object({
   runId: z.string(),
   spanId: z.string(),
 });
 
+// Phase A2 — discriminated union for PG vs buffered runs. Buffered runs
+// only have one valid spanId (the queued span recorded at gate time and
+// reused as the run's root spanId when the drainer materialises). Any
+// other spanId returns a deterministic 404; the queued span returns a
+// minimal synthesised shape so the customer's SDK sees the same 200
+// contract they'd get for a freshly-triggered run.
+type ResolvedRun =
+  | { source: "pg"; run: Awaited<ReturnType<typeof findPgRun>> & {} }
+  | { source: "buffer"; run: NonNullable<Awaited<ReturnType<typeof findRunByIdWithMollifierFallback>>> };
+
+async function findPgRun(runId: string, environmentId: string) {
+  return $replica.taskRun.findFirst({
+    where: { friendlyId: runId, runtimeEnvironmentId: environmentId },
+  });
+}
+
 export const loader = createLoaderApiRoute(
   {
     params: ParamsSchema,
     allowJWT: true,
     corsStrategy: "all",
-    findResource: (params, auth) => {
-      return $replica.taskRun.findFirst({
-        where: {
-          friendlyId: params.runId,
-          runtimeEnvironmentId: auth.environment.id,
-        },
+    findResource: async (params, auth): Promise<ResolvedRun | null> => {
+      const pgRun = await findPgRun(params.runId, auth.environment.id);
+      if (pgRun) return { source: "pg", run: pgRun };
+
+      const buffered = await findRunByIdWithMollifierFallback({
+        runId: params.runId,
+        environmentId: auth.environment.id,
+        organizationId: auth.environment.organizationId,
       });
+      if (buffered) return { source: "buffer", run: buffered };
+
+      return null;
     },
     shouldRetryNotFound: true,
     authorization: {
       action: "read",
-      resource: (run) => {
+      resource: (resolved) => {
+        if (resolved.source === "pg") {
+          const run = resolved.run;
+          const resources = [
+            { type: "runs", id: run.friendlyId },
+            { type: "tasks", id: run.taskIdentifier },
+            ...run.runTags.map((tag) => ({ type: "tags", id: tag })),
+          ];
+          if (run.batchId) {
+            resources.push({ type: "batch", id: BatchId.toFriendlyId(run.batchId) });
+          }
+          return anyResource(resources);
+        }
+        const run = resolved.run;
         const resources = [
           { type: "runs", id: run.friendlyId },
-          { type: "tasks", id: run.taskIdentifier },
-          ...run.runTags.map((tag) => ({ type: "tags", id: tag })),
+          ...(run.taskIdentifier ? [{ type: "tasks", id: run.taskIdentifier }] : []),
+          ...run.tags.map((tag) => ({ type: "tags", id: tag })),
         ];
-        if (run.batchId) {
-          resources.push({ type: "batch", id: BatchId.toFriendlyId(run.batchId) });
-        }
         return anyResource(resources);
       },
     },
   },
-  async ({ params, resource: run, authentication }) => {
+  async ({ params, resource: resolved, authentication }) => {
+    if (resolved.source === "buffer") {
+      // Buffered runs have exactly one valid spanId — the queued span the
+      // mollifier gate recorded at trigger time, which becomes the run's
+      // root spanId once the drainer materialises. Any other spanId is a
+      // deterministic 404. The matching spanId returns a minimal shape
+      // representing "span exists, no execution data yet."
+      if (resolved.run.spanId !== params.spanId) {
+        return json({ error: "Span not found" }, { status: 404 });
+      }
+      return json(
+        {
+          spanId: resolved.run.spanId,
+          parentId: resolved.run.parentSpanId,
+          runId: resolved.run.friendlyId,
+          message: resolved.run.taskIdentifier ?? "",
+          isError: false,
+          isPartial: true,
+          isCancelled: false,
+          level: "TRACE",
+          startTime: resolved.run.createdAt,
+          durationMs: 0,
+          properties: undefined,
+          events: undefined,
+          entityType: undefined,
+          ai: undefined,
+          triggeredRuns: undefined,
+        },
+        { status: 200 }
+      );
+    }
+
+    const run = resolved.run;
     const eventRepository = resolveEventRepositoryForStore(run.taskEventStore);
     const eventStore = getTaskEventStoreTableForRun(run);
 
diff --git a/apps/webapp/app/routes/api.v1.runs.$runParam.attempts.ts b/apps/webapp/app/routes/api.v1.runs.$runParam.attempts.ts
@@ -1,4 +1,4 @@
-import type { ActionFunctionArgs } from "@remix-run/server-runtime";
+import type { ActionFunctionArgs, LoaderFunctionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
 import { z } from "zod";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
@@ -11,6 +11,30 @@ const ParamsSchema = z.object({
   runParam: z.string(),
 });
 
+// Phase A5 — fixes the pre-existing route bug where GET on this URL
+// returned a Remix "no loader" 400 with an internal error message. The
+// route only exposed `action` (POST creates a new attempt); GET had no
+// handler, so any well-intentioned SDK probe hit the framework error
+// instead of a proper API response.
+//
+// Returns `{ attempts: [] }` for both PG and buffered runs. The detailed
+// attempt list belongs on the v3 retrieve endpoint, not here — this is
+// the dual of the POST that creates attempts, and the empty-list shape
+// gives the parity script a stable contract to assert against.
+export async function loader({ request, params }: LoaderFunctionArgs) {
+  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 });
+  }
+
+  return json({ attempts: [] }, { status: 200 });
+}
+
 export async function action({ request, params }: ActionFunctionArgs) {
   // Authenticate the request
   const authenticationResult = await authenticateApiRequest(request);
