fix: expose retryable projection gaps

internal/cloud/server-http-utils.ts

@@ -26,14 +26,23 @@ export function sendJson(response, statusCode, body) {
   if (response.headersSent || response.writableEnded) return false;
   const payload = decorateErrorPayloadForHttp(body, { response, statusCode });
   logErrorEnvelope(payload, { statusCode });
-  response.writeHead(statusCode, {
+  const headers: Record<string, string> = {
     "content-type": "application/json; charset=utf-8",
     "cache-control": "no-store"
-  });
+  };
+  const retryAfterSeconds = retryAfterHeaderSeconds(payload);
+  if (retryAfterSeconds !== null) headers["retry-after"] = String(retryAfterSeconds);
+  response.writeHead(statusCode, headers);
   response.end(JSON.stringify(payload) + "\n");
   return true;
 }
 
+function retryAfterHeaderSeconds(payload) {
+  const retryAfterMs = Number(payload?.error?.retryAfterMs ?? payload?.retryAfterMs);
+  if (!Number.isFinite(retryAfterMs) || retryAfterMs <= 0) return null;
+  return Math.max(1, Math.ceil(retryAfterMs / 1000));
+}
+
 export function sendRedirect(response, location, body = {}) {
   response.writeHead(302, {
     "content-type": "application/json; charset=utf-8",

internal/cloud/server-workbench-http.ts

@@ -825,7 +825,17 @@ function factCheckpointForTrace(facts = {}, traceId) {
 }
 
 function workbenchProjectionStoreError(error) {
-  return workbenchError("projection_store_unavailable", "Workbench durable projection store is unavailable.", { causeCode: error?.code ?? null });
+  const data = objectValue(error?.data) ?? {};
+  const retryAfterNumber = Number(data.retryAfterMs);
+  const retryAfterMs = Number.isFinite(retryAfterNumber) && retryAfterNumber >= 0 ? Math.trunc(retryAfterNumber) : null;
+  return workbenchError("projection_store_unavailable", "Workbench durable projection store is unavailable.", {
+    causeCode: error?.code ?? null,
+    queryResult: textValue(data.queryResult) || null,
+    blockedLayer: textValue(data.blockedLayer) || null,
+    retryable: data.retryable !== false,
+    transient: data.transient === true || data.retryable === true,
+    retryAfterMs: retryAfterMs ?? null
+  });
 }
 
 function factSessionId(session) {
@@ -1090,9 +1100,10 @@ async function handleWorkbenchTraceEventPage(response, url, options, actor, rawT
   const pageResult = await readModel.queryFacts({ traceId, families: WORKBENCH_TRACE_EVENT_PAGE_FAMILIES, afterProjectedSeq: pageOptions.afterProjectedSeq, limit: pageOptions.limit + 1 });
   if (pageResult.error) return sendJson(response, 503, workbenchProjectionStoreError(pageResult.error));
   const page = traceEventPageFromFacts(pageResult.facts.traceEvents, pageOptions, { total: projection.lastProjectedSeq, traceStatus: session.status });
-  if (traceEventPageMissing(page, projection, pageOptions)) {
+  const missingTraceEvents = traceEventPageMissing(page, projection, pageOptions);
+  if (missingTraceEvents) {
     const blocker = traceEventsReadModelBlocker("workbench_trace_events_missing", "Workbench trace events are missing from the durable read model.", { traceId, projection, session, route: url.pathname });
-    return sendJson(response, 404, workbenchTraceEventsReadModelError(blocker, { traceId, projection, session }));
+    page.blocker = blocker;
   }
   const responseProjection = page.blocker ? blockedTraceEventProjection(projection, page.blocker) : projection;
   sendJson(response, 200, {
@@ -1229,6 +1240,7 @@ function traceEventsReadModelBlocker(code, message, { traceId, projection = {},
     lastProjectedSeq: projection.lastProjectedSeq ?? null,
     sourceRunId: projection.sourceRunId ?? null,
     sourceCommandId: projection.sourceCommandId ?? null,
+    retryAfterMs: 5000,
     valuesPrinted: false
   };
 }

internal/cloud/workbench-projection-writer.ts

@@ -99,8 +99,9 @@ export async function writeWorkbenchProjectionEvent({ runtimeStore = null, event
   if (typeof runtimeStore?.allocateWorkbenchProjectedSeq !== "function") throw new Error("Workbench projection event writer requires a durable projectedSeq allocator.");
   const traceId = safeTraceId(event.traceId ?? requestMeta.traceId);
   if (!traceId) return null;
+  const projectedAt = new Date().toISOString();
   const sourceSeq = nonNegativeInteger(event.sourceSeq ?? event.seq);
-  const occurredAt = timestampValue(event.createdAt ?? event.occurredAt ?? event.updatedAt);
+  const occurredAt = timestampValue(event.createdAt ?? event.occurredAt ?? event.updatedAt ?? projectedAt);
   const sessionId = textValue(event.sessionId ?? requestMeta.sessionId) || null;
   const turnId = textValue(event.turnId) || traceId;
   const terminal = event.terminal === true || TERMINAL_STATUSES.has(normalizeWorkbenchStatus(event.status));
@@ -109,8 +110,9 @@ export async function writeWorkbenchProjectionEvent({ runtimeStore = null, event
   const previousTiming = normalizeTimingProjection(previousCheckpoint?.timing) ?? normalizeTimingProjection(previousCheckpoint);
   const explicitStartedAt = optionalTimestampValue(event.startedAt ?? event.traceStartedAt ?? event.runnerStartedAt ?? event.agentRun?.startedAt);
   const startedAt = explicitStartedAt ?? previousTiming?.startedAt ?? (sourceSeq <= 1 ? occurredAt : null);
-  const finishedAt = terminal ? optionalTimestampValue(event.finishedAt) ?? occurredAt : null;
-  const timing = eventTimingProjection({ startedAt, lastEventAt: occurredAt, finishedAt, terminal });
+  const lastEventAt = latestTimestamp(previousTiming?.lastEventAt, occurredAt, projectedAt);
+  const finishedAt = terminal ? latestTimestamp(previousTiming?.lastEventAt, optionalTimestampValue(event.finishedAt), occurredAt, projectedAt) : null;
+  const timing = eventTimingProjection({ startedAt, lastEventAt, finishedAt, terminal });
   const sourceEventId = workbenchSourceEventId(event, { traceId, sourceSeq, occurredAt });
   const allocation = await runtimeStore.allocateWorkbenchProjectedSeq({
     traceId,
@@ -122,7 +124,7 @@ export async function writeWorkbenchProjectionEvent({ runtimeStore = null, event
     sourceEventId,
     eventType: textValue(event.eventType ?? event.type ?? event.label) || "event",
     occurredAt,
-    updatedAt: timestampValue(event.updatedAt ?? occurredAt)
+    updatedAt: projectedAt
   }, {
     traceId,
     sessionId,
@@ -147,7 +149,7 @@ export async function writeWorkbenchProjectionEvent({ runtimeStore = null, event
       lastEventAt: timing.lastEventAt,
       finishedAt: timing.finishedAt,
       durationMs: timing.durationMs,
-      updatedAt: occurredAt
+      updatedAt: projectedAt
     }] : [],
     traceEvents: [{
       ...event,
@@ -168,7 +170,7 @@ export async function writeWorkbenchProjectionEvent({ runtimeStore = null, event
       finishedAt: timing.finishedAt,
       durationMs: timing.durationMs,
       occurredAt,
-      updatedAt: timestampValue(event.updatedAt ?? occurredAt)
+      updatedAt: projectedAt
     }],
     checkpoints: [{
       traceId,
@@ -193,7 +195,7 @@ export async function writeWorkbenchProjectionEvent({ runtimeStore = null, event
         lastEventStatus: textValue(event.status) || null,
         valuesRedacted: true
       },
-      updatedAt: timestampValue(event.updatedAt ?? occurredAt)
+      updatedAt: projectedAt
     }]
   };
   return runtimeStore.writeWorkbenchFacts({ facts }, {
@@ -434,6 +436,19 @@ function elapsedBetween(startedAt, endedAt) {
   return Math.trunc(end - start);
 }
 
+function latestTimestamp(...values) {
+  let latest = null;
+  let latestMs = Number.NEGATIVE_INFINITY;
+  for (const value of values) {
+    const timestamp = optionalTimestampValue(value);
+    const ms = Date.parse(String(timestamp ?? ""));
+    if (!Number.isFinite(ms) || ms < latestMs) continue;
+    latest = timestamp;
+    latestMs = ms;
+  }
+  return latest;
+}
+
 function isEmptyFacts(facts) {
   return Object.values(facts).every((items) => !Array.isArray(items) || items.length === 0);
 }

internal/db/runtime-store.ts

@@ -2789,6 +2789,9 @@ function durableRuntimeReadBlockedData(method, readiness = {}) {
     blockedLayer,
     requiredEvidence: durability.requiredEvidence ?? RUNTIME_DURABILITY_REQUIRED_EVIDENCE,
     dbLiveEvidenceIsDurabilityEvidence: false,
+    retryable: readiness.retryable === true || connection.queryResult === "connect_timeout",
+    transient: readiness.transient === true || connection.queryResult === "connect_timeout",
+    retryAfterMs: Number.isFinite(Number(readiness.retryAfterMs)) ? Math.max(0, Math.trunc(Number(readiness.retryAfterMs))) : undefined,
     secretMaterialRead: false,
     valueRedacted: true,
     endpointRedacted: true,