Merge pull request #1618 from pikasTech/fix/1616-agentrun-incremental-cursor

fix: 持久化 Workbench projection cursor

internal/cloud/code-agent-agentrun-adapter.ts

@@ -1,5 +1,5 @@
-// SPEC: PJ2026-01060505 Workbench Performance draft-2026-06-17-p0.
-// Responsibility: AgentRun v0.1 adapter and upstream timing projection for Cloud API observability.
+// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p1-agentrun-incremental-cursor; PJ2026-010205 HWLAB接入 draft-2026-06-17-r0; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0.
+// Responsibility: AgentRun v0.1 adapter, incremental Workbench projection cursor sync, and upstream timing projection for Cloud API observability.
 
 import { createHash, randomUUID } from "node:crypto";
 
@@ -15,6 +15,7 @@ import {
   truthyFlag
 } from "./server-http-utils.ts";
 import { backendPerformanceRouteTemplate, currentBackendPerformanceContext } from "./backend-performance.ts";
+import { agentRunResultSyncDeferred, buildAgentRunProjectionEventsFetchPlan, buildWorkbenchProjectionStateUpdate } from "./workbench-projection-cursor.ts";
 
 const ADAPTER_ID = "agentrun-v01";
 const DEFAULT_AGENTRUN_MGR_URL = "http://agentrun-mgr.agentrun-v01.svc.cluster.local:8080";
@@ -451,12 +452,29 @@ export async function syncAgentRunChatResult({ traceId, currentResult = null, op
   const fetchImpl = options.fetchImpl ?? globalThis.fetch;
   const managerUrl = resolveAgentRunManagerUrl(env, mapped.agentRun.managerUrl);
   const timeoutMs = parsePositiveInteger(env.HWLAB_CODE_AGENT_AGENTRUN_HTTP_TIMEOUT_MS, 20_000);
-  const previousLastSeq = Number(mapped.agentRun.lastSeq ?? 0);
-  const eventsResponse = refreshEvents ? await measuredAgentRunEventsFetch({ performanceStore, fetchImpl, managerUrl, timeoutMs, env, mapping: { ...mapped.agentRun, traceSummary: mapped.traceSummary } }) : null;
+  const projectionState = await loadWorkbenchProjectionStateForAgentRun(options.runtimeStore, { traceId, agentRun: mapped.agentRun });
+  const fetchPlan = buildAgentRunProjectionEventsFetchPlan({
+    projectionState,
+    agentRun: mapped.agentRun,
+    pageLimit: env.HWLAB_CODE_AGENT_AGENTRUN_EVENTS_PAGE_LIMIT
+  });
+  const previousLastSeq = Number(fetchPlan.afterSeq ?? mapped.agentRun.lastSeq ?? 0);
+  const eventsResponse = refreshEvents ? await measuredAgentRunEventsFetch({ performanceStore, fetchImpl, managerUrl, timeoutMs, env, mapping: { ...mapped.agentRun, lastSeq: fetchPlan.afterSeq, afterSeqOverride: fetchPlan.afterSeq, eventsPageLimit: fetchPlan.limit, traceSummary: mapped.traceSummary } }) : null;
   if (eventsResponse) appendAgentRunEventsToTrace(traceStore, traceId, eventsResponse.events, mapped.agentRun);
   const nextLastSeq = eventsResponse ? agentRunTraceCursorSeq(eventsResponse, mapped.agentRun.lastSeq) : mapped.agentRun.lastSeq;
   recordAgentRunEventsProjectionMetrics(performanceStore, eventsResponse, previousLastSeq, nextLastSeq);
-  if (!forceResultSync && !agentRunCommandResultSyncRequired(mapped, eventsResponse)) {
+  const terminalFromEvents = agentRunTerminalStatusFromEvents(eventsResponse?.events);
+  await writeWorkbenchProjectionStateForAgentRun(options.runtimeStore, {
+    currentState: projectionState,
+    result: mapped,
+    agentRun: { ...mapped.agentRun, lastSeq: nextLastSeq },
+    eventsResponse,
+    resultSyncState: terminalFromEvents ? "pending" : null,
+    projectionStatus: terminalFromEvents ? "terminal" : null,
+    now: options.now
+  });
+  const deferResultSync = agentRunResultSyncDeferred({ forceResultSync, options, env });
+  if (!forceResultSync && (deferResultSync || !agentRunCommandResultSyncRequired(mapped, eventsResponse))) {
     const nextMapping = withAgentRunRunnerJobCount({
       ...mapped.agentRun,
       lastSeq: nextLastSeq,
@@ -464,6 +482,7 @@ export async function syncAgentRunChatResult({ traceId, currentResult = null, op
       runStatus: mapped.agentRun.runStatus ?? null,
       commandState: mapped.agentRun.commandState ?? null,
       terminalStatus: null,
+      resultSyncState: terminalFromEvents ? "pending" : mapped.agentRun.resultSyncState ?? null,
       updatedAt: nowIso(options.now),
       valuesPrinted: false
     });
@@ -471,7 +490,22 @@ export async function syncAgentRunChatResult({ traceId, currentResult = null, op
     options.codeAgentChatResults?.set?.(traceId, payload);
     return { result: payload, runnerTrace: payload.runnerTrace ?? traceStore.snapshot(traceId), found: true, eventsRefreshed: Boolean(eventsResponse), resultSynced: false };
   }
-  const result = await measuredAgentRunResultFetch({ performanceStore, fetchImpl, managerUrl, timeoutMs, env, mapped, eventsResponse });
+  let result;
+  try {
+    result = await measuredAgentRunResultFetch({ performanceStore, fetchImpl, managerUrl, timeoutMs, env, mapped, eventsResponse });
+  } catch (error) {
+    await writeWorkbenchProjectionStateForAgentRun(options.runtimeStore, {
+      currentState: projectionState,
+      result: mapped,
+      agentRun: { ...mapped.agentRun, lastSeq: nextLastSeq },
+      eventsResponse,
+      resultSyncState: agentRunMetricStatusForError(error) === "timeout" ? "timed_out" : "failed",
+      projectionStatus: terminalFromEvents ? "terminal" : null,
+      error,
+      now: options.now
+    });
+    throw error;
+  }
   const nextMapping = withAgentRunRunnerJobCount({
     ...mapped.agentRun,
     ...agentRunResultRefs(result),
@@ -486,9 +520,39 @@ export async function syncAgentRunChatResult({ traceId, currentResult = null, op
   const base = { ...mapped, agentRun: nextMapping, updatedAt: nowIso(options.now) };
   const payload = agentRunResultToCodeAgentPayload({ base, result, traceStore, traceId, appendResultEvent });
   options.codeAgentChatResults?.set?.(traceId, payload);
+  await writeWorkbenchProjectionStateForAgentRun(options.runtimeStore, {
+    currentState: projectionState,
+    result: payload,
+    agentRun: nextMapping,
+    eventsResponse,
+    resultSyncState: "synced",
+    projectionStatus: isAgentRunTerminalStatus(payload?.status) ? "terminal" : null,
+    now: options.now
+  });
   return { result: payload, runnerTrace: payload.runnerTrace ?? traceStore.snapshot(traceId), found: true, eventsRefreshed: Boolean(eventsResponse), resultSynced: true };
 }
 
+async function loadWorkbenchProjectionStateForAgentRun(runtimeStore, { traceId, agentRun = {} } = {}) {
+  if (!runtimeStore || typeof runtimeStore.getWorkbenchProjectionState !== "function") return null;
+  try {
+    const result = await runtimeStore.getWorkbenchProjectionState({ traceId });
+    const state = result?.projectionState ?? null;
+    if (!state) return null;
+    if (text(state.runId ?? state.sourceRunId) !== text(agentRun.runId)) return null;
+    if (text(state.commandId ?? state.sourceCommandId) !== text(agentRun.commandId)) return null;
+    return state;
+  } catch {
+    return null;
+  }
+}
+
+async function writeWorkbenchProjectionStateForAgentRun(runtimeStore, params = {}) {
+  if (!runtimeStore || typeof runtimeStore.writeWorkbenchProjectionState !== "function") return null;
+  const projectionState = buildWorkbenchProjectionStateUpdate(params);
+  if (!projectionState.traceId || !projectionState.runId || !projectionState.commandId) return null;
+  return await runtimeStore.writeWorkbenchProjectionState({ projectionState });
+}
+
 function agentRunCommandResultSyncRequired(mapped = {}, eventsResponse = null) {
   const agentRun = mapped?.agentRun && typeof mapped.agentRun === "object" ? mapped.agentRun : {};
   if (isAgentRunTerminalStatus(mapped?.status)) return true;
@@ -1875,8 +1939,12 @@ function appendAgentRunEventsToTrace(traceStore, traceId, events, mapping = {})
 async function fetchAgentRunEventsForTrace({ fetchImpl, managerUrl, timeoutMs, env, mapping = {} }) {
   const runId = requiredString(mapping.runId, "runId");
   const currentCommandId = typeof mapping.commandId === "string" ? mapping.commandId : "";
-  const { afterSeq, endSeq } = agentRunTraceReplayWindow(mapping);
-  const path = `/api/v1/runs/${encodeURIComponent(runId)}/events?afterSeq=${encodeURIComponent(String(afterSeq))}&limit=500`;
+  const replayWindow = agentRunTraceReplayWindow(mapping);
+  const afterSeqOverride = Number(mapping.afterSeqOverride);
+  const afterSeq = Number.isFinite(afterSeqOverride) && afterSeqOverride >= 0 ? Math.floor(afterSeqOverride) : replayWindow.afterSeq;
+  const endSeq = Number.isFinite(afterSeqOverride) && afterSeqOverride >= 0 ? 0 : replayWindow.endSeq;
+  const limit = parsePositiveInteger(mapping.eventsPageLimit ?? env?.HWLAB_CODE_AGENT_AGENTRUN_EVENTS_PAGE_LIMIT, 500);
+  const path = `/api/v1/runs/${encodeURIComponent(runId)}/events?afterSeq=${encodeURIComponent(String(afterSeq))}&limit=${encodeURIComponent(String(limit))}`;
   const response = await agentRunJson(fetchImpl, managerUrl, path, { method: "GET", timeoutMs, env });
   const rawEvents = Array.isArray(response?.items) ? response.items : [];
   const events = rawEvents.filter((event) => agentRunEventBelongsToTrace(event, { currentCommandId, afterSeq, endSeq }));

internal/cloud/server-agent-chat.test.ts

@@ -2514,9 +2514,10 @@ test("cloud api turn status reuses request session lookup without loading persis
   }
 });
 
-test("cloud api resumes AgentRun projection from durable session after restart", async () => {
+test("cloud api resumes AgentRun projection from durable state after restart", async () => {
   const calls = [];
   const durableEvents = [];
+  const projectionStates = new Map();
   const ownerWrites = [];
   const traceId = "trc_projection_resume_restart";
   const runId = "run_projection_resume_restart";
@@ -2540,23 +2541,7 @@ test("cloud api resumes AgentRun projection from durable session after restart",
       ] });
     }
     if (request.method === "GET" && url.pathname === `/api/v1/runs/${runId}/commands/${commandId}/result`) {
-      return send({
-        runId,
-        commandId,
-        attemptId: "attempt_projection_resume_restart",
-        runnerId: "runner_projection_resume_restart",
-        jobName: "agentrun-v02-runner-projection-resume",
-        namespace: "agentrun-v02",
-        status: "completed",
-        runStatus: "claimed",
-        commandState: "completed",
-        terminalStatus: "completed",
-        completed: true,
-        reply: finalText,
-        lastSeq: 23,
-        eventCount: 23,
-        sessionRef: { sessionId, conversationId, threadId: "thread-projection-resume-restart" }
-      });
+      return send({ ok: false, message: "resume projection must not block on result" }, 500);
     }
     return send({ ok: false, message: `unexpected ${request.method} ${url.pathname}` }, 404);
   });
@@ -2609,6 +2594,31 @@ test("cloud api resumes AgentRun projection from durable session after restart",
     createdAt: "2026-06-18T16:31:04.200Z",
     valuesPrinted: false
   });
+  projectionStates.set(traceId, {
+    traceId,
+    sessionId,
+    conversationId,
+    threadId: "thread-projection-resume-restart",
+    ownerUserId: TEST_AGENT_ACTOR.id,
+    ownerRole: TEST_AGENT_ACTOR.role,
+    runId,
+    commandId,
+    sourceRunId: runId,
+    sourceCommandId: commandId,
+    managerUrl: `http://127.0.0.1:${agentRunPort}`,
+    backendProfile: "codex",
+    lastSourceSeq: 21,
+    lastAgentRunSeq: 21,
+    lastProjectedSeq: 27,
+    sourceLatestSeq: 21,
+    upstreamLatestSeq: 21,
+    projectionStatus: "projecting",
+    projectionHealth: "healthy",
+    resultSyncState: "not_started",
+    createdAt: "2026-06-18T16:31:04.000Z",
+    updatedAt: "2026-06-18T16:31:04.000Z",
+    valuesPrinted: false
+  });
 
   const runtimeStore = {
     async queryAgentTraceEvents(params = {}) {
@@ -2618,6 +2628,18 @@ test("cloud api resumes AgentRun projection from durable session after restart",
       const event = params.event ?? params.traceEvent ?? params;
       durableEvents.push(event);
       return { written: true, traceEvent: event };
+    },
+    async getWorkbenchProjectionState(params = {}) {
+      const projectionState = projectionStates.get(params.traceId) ?? null;
+      return { projectionState, found: Boolean(projectionState) };
+    },
+    async queryWorkbenchProjectionStates() {
+      return { states: [...projectionStates.values()], count: projectionStates.size };
+    },
+    async writeWorkbenchProjectionState(params = {}) {
+      const projectionState = params.projectionState ?? params.state ?? params;
+      projectionStates.set(projectionState.traceId, projectionState);
+      return { written: true, projectionState };
     }
   };
   const accessController = {
@@ -2650,13 +2672,13 @@ test("cloud api resumes AgentRun projection from durable session after restart",
   await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
 
   try {
-    for (let index = 0; index < 80 && !ownerWrites.some((write) => write.status === "completed"); index += 1) await delay(10);
+    for (let index = 0; index < 80 && (projectionStates.get(traceId)?.lastSourceSeq ?? 0) < 23; index += 1) await delay(10);
     assert.ok(calls.some((call) => call.path === `/api/v1/runs/${runId}/events` && call.afterSeq === "21"));
-    assert.ok(calls.some((call) => call.path === `/api/v1/runs/${runId}/commands/${commandId}/result`));
-    const completedWrite = ownerWrites.find((write) => write.status === "completed");
-    assert.ok(completedWrite);
-    assert.equal(completedWrite.session?.traceResults?.[traceId]?.finalResponse?.text, finalText);
-    assert.ok(durableEvents.some((event) => event.label === "agentrun:result:completed"));
+    assert.equal(calls.some((call) => call.path === `/api/v1/runs/${runId}/commands/${commandId}/result`), false);
+    assert.equal(projectionStates.get(traceId)?.lastSourceSeq, 23);
+    assert.equal(projectionStates.get(traceId)?.resultSyncState, "pending");
+    assert.equal(ownerWrites.some((write) => write.status === "completed"), false);
+    assert.ok(durableEvents.some((event) => event.sourceSeq === 22));
   } finally {
     await new Promise((resolve, reject) => {
       server.close((error) => (error ? reject(error) : resolve()));

internal/cloud/server-code-agent-http.ts

@@ -1,4 +1,4 @@
-// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p0-projector-resume; PJ2026-010403 API契约 draft-2026-06-18-r1; PJ2026-010205 HWLAB接入 draft-2026-06-17-r0; PJ2026-0102 Agent编排 draft-2026-06-17-r0; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0.
+// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p1-agentrun-incremental-cursor; PJ2026-010403 API契约 draft-2026-06-18-r1; PJ2026-010205 HWLAB接入 draft-2026-06-17-r0; PJ2026-0102 Agent编排 draft-2026-06-17-r0; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0.
 // Responsibility: Code Agent HTTP turn admission, lifecycle compatibility wrappers, and Workbench projection writer/finalizer integration.
 
 import { createHash, randomUUID } from "node:crypto";
@@ -19,6 +19,7 @@ import { defaultCodeAgentTraceStore } from "./code-agent-trace-store.ts";
 import { codeAgentSessionLifecycleSummary } from "./code-agent-session-lifecycle.ts";
 import { scheduleWorkbenchProjectionFinalizer } from "./workbench-projection-finalizer.ts";
 import { writeWorkbenchProjectionSession } from "./workbench-projection-writer.ts";
+import { buildAgentRunProjectionEventsFetchPlan } from "./workbench-projection-cursor.ts";
 import { createWorkbenchReadModel } from "./workbench-read-model.ts";
 import {
   firstHeaderValue,
@@ -942,8 +943,9 @@ export function startAgentRunProjectionResume({ options = {}, traceStore = defau
 }
 
 async function runAgentRunProjectionResumePass({ options = {}, traceStore = defaultCodeAgentTraceStore, logger = console, batchSize, minAgeMs, offset = 0, reason = "manual" } = {}) {
-  const sessions = await listAgentRunProjectionResumeSessions(options, batchSize, offset);
-  const candidates = sessions.flatMap((session) => agentRunProjectionResumeCandidates(session, { minAgeMs }));
+  const stateCandidates = await listAgentRunProjectionResumeStateCandidates(options, batchSize, offset, { minAgeMs });
+  const sessions = stateCandidates.length > 0 ? [] : await listAgentRunProjectionResumeSessions(options, batchSize, offset);
+  const candidates = stateCandidates.length > 0 ? stateCandidates : sessions.flatMap((session) => agentRunProjectionResumeCandidates(session, { minAgeMs }));
   let resumed = 0;
   let terminal = 0;
   let failed = 0;
@@ -963,6 +965,7 @@ async function runAgentRunProjectionResumePass({ options = {}, traceStore = defa
     status: failed === 0 ? "completed" : "degraded",
     reason,
     offset,
+    scannedProjectionStates: stateCandidates.length,
     scannedSessions: sessions.length,
     candidates: candidates.length,
     resumed,
@@ -973,6 +976,25 @@ async function runAgentRunProjectionResumePass({ options = {}, traceStore = defa
   return { offset, scannedSessions: sessions.length, candidates: candidates.length, resumed, terminal, failed };
 }
 
+async function listAgentRunProjectionResumeStateCandidates(options = {}, batchSize = DEFAULT_AGENTRUN_PROJECTION_RESUME_BATCH_SIZE, offset = 0, { minAgeMs = DEFAULT_AGENTRUN_PROJECTION_RESUME_MIN_AGE_MS } = {}) {
+  const runtimeStore = options.runtimeStore;
+  if (typeof runtimeStore?.queryWorkbenchProjectionStates !== "function") return [];
+  let result;
+  try {
+    result = await runtimeStore.queryWorkbenchProjectionStates({
+      projectionStatuses: ["projecting", "degraded"],
+      dueAt: new Date().toISOString(),
+      limit: batchSize,
+      offset
+    });
+  } catch {
+    return [];
+  }
+  return (Array.isArray(result?.states) ? result.states : [])
+    .map((state) => agentRunProjectionResumeCandidateFromState(state, { minAgeMs }))
+    .filter(Boolean);
+}
+
 async function listAgentRunProjectionResumeSessions(options = {}, batchSize = DEFAULT_AGENTRUN_PROJECTION_RESUME_BATCH_SIZE, offset = 0) {
   const accessStore = options.accessController?.store ?? options.accessController ?? null;
   if (typeof accessStore?.listAgentSessionsForUser !== "function") return [];
@@ -1050,14 +1072,70 @@ function agentRunProjectionResumeCandidate(session = {}, record = {}, { minAgeMs
   };
 }
 
+function agentRunProjectionResumeCandidateFromState(state = {}, { minAgeMs = DEFAULT_AGENTRUN_PROJECTION_RESUME_MIN_AGE_MS, nowMs = Date.now() } = {}) {
+  const traceId = safeTraceId(state.traceId);
+  const runId = textValue(state.runId ?? state.sourceRunId);
+  const commandId = textValue(state.commandId ?? state.sourceCommandId);
+  if (!traceId || !runId || !commandId) return null;
+  if (isTraceCommandTerminalStatus(state.projectionStatus)) return null;
+  const updatedAtMs = Date.parse(state.updatedAt ?? "");
+  if (minAgeMs > 0 && Number.isFinite(updatedAtMs) && nowMs - updatedAtMs < minAgeMs) return null;
+  const sessionId = safeSessionId(state.sessionId) || null;
+  const conversationId = safeConversationId(state.conversationId) || null;
+  const threadId = safeOpaqueId(state.threadId) || null;
+  const ownerUserId = textValue(state.ownerUserId);
+  const ownerRole = textValue(state.ownerRole) || "user";
+  const lastSeq = Number.isFinite(Number(state.lastSourceSeq ?? state.lastAgentRunSeq)) ? Number(state.lastSourceSeq ?? state.lastAgentRunSeq) : 0;
+  return {
+    traceId,
+    params: {
+      traceId,
+      sessionId,
+      conversationId,
+      threadId,
+      ownerUserId,
+      ownerRole
+    },
+    result: {
+      accepted: true,
+      status: "running",
+      shortConnection: true,
+      traceId,
+      sessionId,
+      conversationId,
+      threadId,
+      ownerUserId,
+      ownerRole,
+      finalResponse: null,
+      traceSummary: null,
+      agentRun: {
+        traceId,
+        runId,
+        commandId,
+        lastSeq,
+        managerUrl: state.managerUrl ?? null,
+        backendProfile: state.backendProfile ?? null,
+        providerId: state.providerId ?? null,
+        status: "running",
+        commandState: "running",
+        resultSyncState: state.resultSyncState ?? null,
+        providerTrace: { traceId, runId, commandId, valuesPrinted: false },
+        valuesPrinted: false
+      },
+      valuesRedacted: true
+    }
+  };
+}
+
 async function resumeAgentRunProjectionCandidate({ candidate, options = {}, traceStore = defaultCodeAgentTraceStore } = {}) {
   const currentResult = await agentRunProjectionResumeResultWithCursor(candidate, options);
+  const resumeOptions = { ...options, deferAgentRunResultSync: true };
   const synced = await syncAgentRunChatResult({
     traceId: candidate.traceId,
     currentResult,
-    options,
+    options: resumeOptions,
     traceStore,
-    forceResultSync: true,
+    forceResultSync: false,
     refreshEvents: true
   });
   const payload = synced?.result ?? currentResult;
@@ -1068,29 +1146,35 @@ async function resumeAgentRunProjectionCandidate({ candidate, options = {}, trac
 }
 
 async function agentRunProjectionResumeResultWithCursor(candidate, options = {}) {
-  const readModel = createWorkbenchReadModel(options, { id: candidate.params.ownerUserId || "projection-resume", role: "admin" });
-  const trace = await readModel.traceSnapshot(candidate.traceId);
-  const durableSourceSeq = agentRunProjectionLastSourceSeq(trace, candidate.result.agentRun.commandId);
-  const existingLastSeq = Number(candidate.result.agentRun.lastSeq ?? 0);
-  const lastSeq = Math.max(Number.isFinite(existingLastSeq) ? existingLastSeq : 0, durableSourceSeq);
+  const projectionState = await loadAgentRunProjectionState(options.runtimeStore, candidate);
+  const plan = buildAgentRunProjectionEventsFetchPlan({
+    projectionState,
+    agentRun: candidate.result.agentRun,
+    pageLimit: options.env?.HWLAB_CODE_AGENT_AGENTRUN_EVENTS_PAGE_LIMIT
+  });
   return {
     ...candidate.result,
     traceSummary: null,
     agentRun: {
       ...candidate.result.agentRun,
-      lastSeq,
+      lastSeq: plan.afterSeq,
       valuesPrinted: false
     }
   };
 }
 
-function agentRunProjectionLastSourceSeq(trace = {}, commandId = null) {
-  const events = Array.isArray(trace?.events) ? trace.events : [];
-  return events.reduce((max, event) => {
-    if (commandId && event?.commandId && event.commandId !== commandId) return max;
-    const sourceSeq = Number(event?.sourceSeq ?? 0);
-    return Number.isFinite(sourceSeq) && sourceSeq > max ? Math.trunc(sourceSeq) : max;
-  }, 0);
+async function loadAgentRunProjectionState(runtimeStore, candidate) {
+  if (!runtimeStore || typeof runtimeStore.getWorkbenchProjectionState !== "function") return null;
+  try {
+    const result = await runtimeStore.getWorkbenchProjectionState({ traceId: candidate.traceId });
+    const state = result?.projectionState ?? null;
+    if (!state) return null;
+    if (String(state.runId ?? state.sourceRunId ?? "") !== String(candidate.result.agentRun.runId ?? "")) return null;
+    if (String(state.commandId ?? state.sourceCommandId ?? "") !== String(candidate.result.agentRun.commandId ?? "")) return null;
+    return state;
+  } catch {
+    return null;
+  }
 }
 
 function appendProjectionResumeError(traceStore, candidate, error) {

internal/cloud/workbench-projection-cursor.test.ts

@@ -0,0 +1,82 @@
+// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p1-agentrun-incremental-cursor.
+// Responsibility: Independent backend unit tests for AgentRun projection cursor planning.
+
+import assert from "node:assert/strict";
+import { test } from "bun:test";
+
+import { buildAgentRunProjectionEventsFetchPlan, buildWorkbenchProjectionStateUpdate } from "./workbench-projection-cursor.ts";
+
+test("AgentRun projection fetch plan uses durable cursor without scanning long trace", () => {
+  const poisonTrace = {};
+  Object.defineProperty(poisonTrace, "events", {
+    get() {
+      throw new Error("trace events must not be read to compute AgentRun afterSeq");
+    }
+  });
+  const poisonResult = {};
+  Object.defineProperty(poisonResult, "events", {
+    get() {
+      throw new Error("result events must not be read to compute AgentRun afterSeq");
+    }
+  });
+
+  const plan = buildAgentRunProjectionEventsFetchPlan({
+    projectionState: {
+      traceId: "trc_cursor_o1",
+      sourceRunId: "run_cursor_o1",
+      sourceCommandId: "cmd_cursor_o1",
+      lastSourceSeq: 3700,
+      lastProjectedSeq: 129
+    },
+    agentRun: {
+      runId: "run_cursor_o1",
+      commandId: "cmd_cursor_o1",
+      lastSeq: 12
+    },
+    trace: poisonTrace,
+    result: poisonResult,
+    pageLimit: 500
+  });
+
+  assert.equal(plan.runId, "run_cursor_o1");
+  assert.equal(plan.commandId, "cmd_cursor_o1");
+  assert.equal(plan.afterSeq, 3700);
+  assert.equal(plan.limit, 500);
+  assert.equal(plan.cursorSource, "projection-state");
+});
+
+test("AgentRun projection state advances from incremental events response", () => {
+  const state = buildWorkbenchProjectionStateUpdate({
+    currentState: {
+      traceId: "trc_cursor_update",
+      sourceRunId: "run_cursor_update",
+      sourceCommandId: "cmd_cursor_update",
+      lastSourceSeq: 3700,
+      lastProjectedSeq: 3700,
+      sourceLatestSeq: 3700,
+      projectionStatus: "projecting",
+      resultSyncState: "not_started",
+      createdAt: "2026-06-19T12:00:00.000Z"
+    },
+    agentRun: {
+      runId: "run_cursor_update",
+      commandId: "cmd_cursor_update",
+      lastSeq: 3700
+    },
+    eventsResponse: {
+      traceLastSeq: 3720,
+      maxSeq: 3721,
+      rawEventCount: 20
+    },
+    now: () => "2026-06-19T12:01:00.000Z"
+  });
+
+  assert.equal(state.lastSourceSeq, 3720);
+  assert.equal(state.lastAgentRunSeq, 3720);
+  assert.equal(state.lastProjectedSeq, 3720);
+  assert.equal(state.sourceLatestSeq, 3721);
+  assert.equal(state.projectionStatus, "projecting");
+  assert.equal(state.resultSyncState, "not_started");
+  assert.equal(state.createdAt, "2026-06-19T12:00:00.000Z");
+  assert.equal(state.updatedAt, "2026-06-19T12:01:00.000Z");
+});

internal/cloud/workbench-projection-cursor.ts

@@ -0,0 +1,136 @@
+/*
+ * SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p1-agentrun-incremental-cursor; PJ2026-010205 HWLAB接入 draft-2026-06-17-r0.
+ * Responsibility: Pure cursor planning for AgentRun events projection; never derives cursor by scanning trace events or result payloads.
+ */
+
+const DEFAULT_AGENTRUN_EVENTS_PAGE_LIMIT = 500;
+const MAX_AGENTRUN_EVENTS_PAGE_LIMIT = 1000;
+
+export function buildAgentRunProjectionEventsFetchPlan({ projectionState = null, agentRun = null, pageLimit = DEFAULT_AGENTRUN_EVENTS_PAGE_LIMIT } = {}) {
+  const state = normalizeProjectionState(projectionState);
+  const mapping = normalizeAgentRunMapping(agentRun);
+  const runId = textValue(state?.sourceRunId ?? state?.runId ?? mapping.runId);
+  const commandId = textValue(state?.sourceCommandId ?? state?.commandId ?? mapping.commandId);
+  const stateSeq = maxNonNegativeInteger(state?.lastSourceSeq, state?.lastAgentRunSeq);
+  const mappingSeq = maxNonNegativeInteger(mapping.lastSeq);
+  return {
+    runId,
+    commandId,
+    afterSeq: Math.max(stateSeq, mappingSeq),
+    limit: boundedPageLimit(pageLimit),
+    cursorSource: state ? "projection-state" : "agent-run-result",
+    valuesPrinted: false
+  };
+}
+
+export function buildWorkbenchProjectionStateUpdate({
+  currentState = null,
+  result = null,
+  agentRun = null,
+  eventsResponse = null,
+  resultSyncState = null,
+  projectionStatus = null,
+  projectionHealth = null,
+  error = null,
+  now = () => new Date().toISOString()
+} = {}) {
+  const previous = normalizeProjectionState(currentState) ?? {};
+  const mapping = normalizeAgentRunMapping(agentRun ?? result?.agentRun);
+  const timestamp = timestampValue(now);
+  const traceId = textValue(previous.traceId ?? result?.traceId ?? mapping.traceId);
+  const runId = textValue(previous.sourceRunId ?? previous.runId ?? mapping.runId);
+  const commandId = textValue(previous.sourceCommandId ?? previous.commandId ?? mapping.commandId);
+  const lastSourceSeq = Math.max(
+    maxNonNegativeInteger(previous.lastSourceSeq, previous.lastAgentRunSeq),
+    maxNonNegativeInteger(mapping.lastSeq),
+    maxNonNegativeInteger(eventsResponse?.traceLastSeq)
+  );
+  const sourceLatestSeq = Math.max(
+    maxNonNegativeInteger(previous.sourceLatestSeq, previous.upstreamLatestSeq),
+    maxNonNegativeInteger(eventsResponse?.maxSeq),
+    lastSourceSeq
+  );
+  const nextResultSyncState = textValue(resultSyncState ?? previous.resultSyncState) || "not_started";
+  const nextProjectionStatus = textValue(projectionStatus) || inferProjectionStatus({ lastSourceSeq, sourceLatestSeq, resultSyncState: nextResultSyncState, previous });
+  const nextProjectionHealth = textValue(projectionHealth) || (error ? "degraded" : "healthy");
+  return {
+    ...previous,
+    traceId,
+    sessionId: textValue(previous.sessionId ?? result?.sessionId ?? mapping.sessionId) || null,
+    conversationId: textValue(previous.conversationId ?? result?.conversationId ?? mapping.conversationId) || null,
+    threadId: textValue(previous.threadId ?? result?.threadId ?? mapping.threadId) || null,
+    ownerUserId: textValue(previous.ownerUserId ?? result?.ownerUserId) || null,
+    ownerRole: textValue(previous.ownerRole ?? result?.ownerRole) || null,
+    managerUrl: textValue(previous.managerUrl ?? mapping.managerUrl) || null,
+    backendProfile: textValue(previous.backendProfile ?? mapping.backendProfile) || null,
+    providerId: textValue(previous.providerId ?? mapping.providerId) || null,
+    sourceRunId: runId,
+    sourceCommandId: commandId,
+    runId,
+    commandId,
+    lastSourceSeq,
+    lastAgentRunSeq: lastSourceSeq,
+    lastProjectedSeq: Math.max(maxNonNegativeInteger(previous.lastProjectedSeq), lastSourceSeq),
+    sourceLatestSeq,
+    upstreamLatestSeq: sourceLatestSeq,
+    projectionStatus: nextProjectionStatus,
+    projectionHealth: nextProjectionHealth,
+    resultSyncState: nextResultSyncState,
+    lastProjectedAt: lastSourceSeq > maxNonNegativeInteger(previous.lastSourceSeq, previous.lastAgentRunSeq) ? timestamp : previous.lastProjectedAt ?? timestamp,
+    lastResultSyncAt: nextResultSyncState === "synced" ? timestamp : previous.lastResultSyncAt ?? null,
+    lastErrorCode: error ? textValue(error.code ?? error.errorCode ?? "projection_sync_failed") : null,
+    lastErrorMessage: error ? textValue(error.message ?? "Projection sync failed.") : null,
+    failureCount: error ? maxNonNegativeInteger(previous.failureCount) + 1 : 0,
+    nextRetryAt: error ? previous.nextRetryAt ?? null : null,
+    createdAt: previous.createdAt ?? timestamp,
+    updatedAt: timestamp,
+    valuesPrinted: false
+  };
+}
+
+export function agentRunResultSyncDeferred({ forceResultSync = false, options = {}, env = process.env } = {}) {
+  if (forceResultSync) return false;
+  if (options.deferAgentRunResultSync === true) return true;
+  const configured = String(env?.HWLAB_CODE_AGENT_AGENTRUN_RESULT_SYNC_IN_RUNNING ?? "").trim().toLowerCase();
+  return ["0", "false", "off", "no"].includes(configured);
+}
+
+function normalizeProjectionState(value) {
+  if (!value || typeof value !== "object" || Array.isArray(value)) return null;
+  return value;
+}
+
+function normalizeAgentRunMapping(value) {
+  return value && typeof value === "object" && !Array.isArray(value) ? value : {};
+}
+
+function inferProjectionStatus({ lastSourceSeq, sourceLatestSeq, resultSyncState, previous }) {
+  if (["pending", "timed_out", "failed"].includes(resultSyncState)) return "terminal";
+  if (sourceLatestSeq > lastSourceSeq) return "projecting";
+  return textValue(previous?.projectionStatus) || "projecting";
+}
+
+function boundedPageLimit(value) {
+  const parsed = Number.parseInt(String(value ?? ""), 10);
+  if (!Number.isInteger(parsed) || parsed <= 0) return DEFAULT_AGENTRUN_EVENTS_PAGE_LIMIT;
+  return Math.min(parsed, MAX_AGENTRUN_EVENTS_PAGE_LIMIT);
+}
+
+function maxNonNegativeInteger(...values) {
+  let max = 0;
+  for (const value of values) {
+    const parsed = Number.parseInt(String(value ?? ""), 10);
+    if (Number.isInteger(parsed) && parsed >= 0 && parsed > max) max = parsed;
+  }
+  return max;
+}
+
+function timestampValue(now) {
+  const value = typeof now === "function" ? now() : now;
+  const parsed = Date.parse(String(value ?? ""));
+  return Number.isFinite(parsed) ? new Date(parsed).toISOString() : new Date().toISOString();
+}
+
+function textValue(value) {
+  return String(value ?? "").trim();
+}

internal/db/migrations/0001_cloud_core_skeleton.sql

@@ -168,6 +168,33 @@ CREATE INDEX IF NOT EXISTS idx_agent_trace_events_trace_order ON agent_trace_eve
 CREATE INDEX IF NOT EXISTS idx_agent_trace_events_session_order ON agent_trace_events(agent_session_id, occurred_at, id);
 CREATE INDEX IF NOT EXISTS idx_agent_trace_events_worker_order ON agent_trace_events(worker_session_id, occurred_at, id);
 
+CREATE TABLE IF NOT EXISTS workbench_projection_state (
+  trace_id TEXT PRIMARY KEY,
+  session_id TEXT,
+  conversation_id TEXT,
+  thread_id TEXT,
+  run_id TEXT NOT NULL,
+  command_id TEXT NOT NULL,
+  last_agentrun_seq INTEGER NOT NULL DEFAULT 0,
+  last_projected_seq INTEGER NOT NULL DEFAULT 0,
+  upstream_latest_seq INTEGER,
+  projection_status TEXT NOT NULL DEFAULT 'projecting',
+  projection_health TEXT NOT NULL DEFAULT 'healthy',
+  result_sync_state TEXT NOT NULL DEFAULT 'not_started',
+  last_projected_at TEXT,
+  last_result_sync_at TEXT,
+  last_error_code TEXT,
+  last_error_message TEXT,
+  failure_count INTEGER NOT NULL DEFAULT 0,
+  next_retry_at TEXT,
+  projection_json TEXT NOT NULL DEFAULT '{}',
+  created_at TEXT NOT NULL,
+  updated_at TEXT NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_workbench_projection_state_status_retry_updated ON workbench_projection_state(projection_status, next_retry_at, updated_at);
+CREATE INDEX IF NOT EXISTS idx_workbench_projection_state_run_command ON workbench_projection_state(run_id, command_id);
+CREATE INDEX IF NOT EXISTS idx_workbench_projection_state_session_updated ON workbench_projection_state(session_id, updated_at DESC);
+
 CREATE TABLE IF NOT EXISTS evidence_records (
   id TEXT PRIMARY KEY,
   project_id TEXT,
@@ -188,7 +215,7 @@ CREATE TABLE IF NOT EXISTS hwlab_schema_migrations (
 INSERT INTO hwlab_schema_migrations (id, schema_version, applied_at, migration_json)
 VALUES (
   '0001_cloud_core_skeleton',
-  'runtime-durable-postgres-v2',
+  'runtime-durable-postgres-v3',
   CURRENT_TIMESTAMP,
   '{"path":"internal/db/migrations/0001_cloud_core_skeleton.sql","runtime":"cloud-api"}'
 )

internal/db/runtime-store.test.ts

@@ -646,6 +646,50 @@ test("configured postgres runtime persists and queries Code Agent trace events",
   assert.ok(queryClient.calls.some((call) => call.sql.startsWith("INSERT INTO agent_trace_events")));
 });
 
+test("configured postgres runtime persists and queries Workbench projection state", async () => {
+  const queryClient = createFakePostgresClient({ migrationReady: true });
+  const store = createConfiguredCloudRuntimeStore({
+    env: {
+      HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres",
+      HWLAB_CLOUD_RUNTIME_DURABLE: "true"
+    },
+    dbUrl: "postgres://hwlab_redacted@db.example.invalid:5432/hwlab",
+    queryClient,
+    now: () => "2026-06-19T13:05:00.000Z"
+  });
+
+  const write = await store.writeWorkbenchProjectionState({
+    projectionState: {
+      traceId: "trc_projection_state",
+      sessionId: "ses_projection_state",
+      conversationId: "cnv_projection_state",
+      threadId: "thread-projection-state",
+      runId: "run_projection_state",
+      commandId: "cmd_projection_state",
+      lastSourceSeq: 3700,
+      lastProjectedSeq: 129,
+      sourceLatestSeq: 3701,
+      projectionStatus: "projecting",
+      resultSyncState: "pending",
+      updatedAt: "2026-06-19T13:05:00.000Z"
+    }
+  });
+  const loaded = await store.getWorkbenchProjectionState({ traceId: "trc_projection_state" });
+  const due = await store.queryWorkbenchProjectionStates({ projectionStatuses: ["projecting"], dueAt: "2026-06-19T13:06:00.000Z", limit: 10 });
+  const readiness = await store.readiness();
+
+  assert.equal(write.written, true);
+  assert.equal(write.projectionState.lastSourceSeq, 3700);
+  assert.equal(write.projectionState.lastAgentRunSeq, 3700);
+  assert.equal(loaded.found, true);
+  assert.equal(loaded.projectionState.commandId, "cmd_projection_state");
+  assert.equal(due.count, 1);
+  assert.equal(due.states[0].resultSyncState, "pending");
+  assert.equal(readiness.counts.workbenchProjectionStates, 1);
+  assert.ok(queryClient.calls.some((call) => call.sql.startsWith("CREATE INDEX IF NOT EXISTS idx_workbench_projection_state_status_retry_updated")));
+  assert.ok(queryClient.calls.some((call) => call.sql.startsWith("INSERT INTO workbench_projection_state")));
+});
+
 test("configured postgres runtime pushes trace event query filters into SQL", async () => {
   const queryClient = createFakePostgresClient({ migrationReady: true });
   const store = createConfiguredCloudRuntimeStore({
@@ -685,6 +729,7 @@ function createFakePostgresClient({
     evidence_records: new Map(),
     worker_sessions: new Map(),
     agent_trace_events: new Map(),
+    workbench_projection_state: new Map(),
     hwlab_schema_migrations: new Map()
   };
   if (migrationReady) {
@@ -723,6 +768,9 @@ function createFakePostgresClient({
       if (sql.startsWith("CREATE INDEX IF NOT EXISTS idx_agent_trace_events_")) {
         return { rows: [] };
       }
+      if (sql.startsWith("CREATE INDEX IF NOT EXISTS idx_workbench_projection_state_")) {
+        return { rows: [] };
+      }
       if (sql.startsWith("SELECT gateway_session_json FROM gateway_sessions")) {
         return jsonSelect(state.gateway_sessions, params[0], "gateway_session_json");
       }
@@ -759,6 +807,25 @@ function createFakePostgresClient({
           .map((record) => ({ event_json: record.event_json }));
         return { rows };
       }
+      if (sql.startsWith("SELECT projection_json FROM workbench_projection_state")) {
+        if (readErrorCode) {
+          const error = new Error("workbench projection state read query failed");
+          error.code = readErrorCode;
+          throw error;
+        }
+        let rows = [...state.workbench_projection_state.values()];
+        if (sql.includes("trace_id = $1")) rows = rows.filter((record) => record.trace_id === params[0]);
+        if (sql.includes("projection_status = ANY")) {
+          const statuses = params.find((param) => Array.isArray(param)) ?? [];
+          rows = rows.filter((record) => statuses.includes(record.projection_status));
+        }
+        if (sql.includes("next_retry_at IS NULL")) {
+          const dueAt = params.find((param) => typeof param === "string" && /^\d{4}-\d{2}-\d{2}T/u.test(param));
+          rows = rows.filter((record) => !record.next_retry_at || !dueAt || record.next_retry_at <= dueAt);
+        }
+        rows.sort((left, right) => String(left.updated_at).localeCompare(String(right.updated_at)) || String(left.trace_id).localeCompare(String(right.trace_id)));
+        return { rows: rows.map((record) => ({ projection_json: record.projection_json })) };
+      }
       if (sql.startsWith("SELECT metadata_json FROM evidence_records")) {
         if (readErrorCode) {
           const error = new Error("evidence read query failed");
@@ -804,6 +871,27 @@ function createFakePostgresClient({
         });
         return { rows: [] };
       }
+      if (sql.startsWith("INSERT INTO workbench_projection_state")) {
+        state.workbench_projection_state.set(params[0], {
+          trace_id: params[0],
+          session_id: params[1],
+          conversation_id: params[2],
+          thread_id: params[3],
+          run_id: params[4],
+          command_id: params[5],
+          last_agentrun_seq: params[6],
+          last_projected_seq: params[7],
+          upstream_latest_seq: params[8],
+          projection_status: params[9],
+          projection_health: params[10],
+          result_sync_state: params[11],
+          next_retry_at: params[17],
+          projection_json: params[18],
+          created_at: params[19],
+          updated_at: params[20]
+        });
+        return { rows: [] };
+      }
       throw new Error(`unexpected sql: ${sql}`);
     }
   };

internal/db/runtime-store.ts

@@ -1,6 +1,6 @@
 /*
- * SPEC: PJ2026-010403 API契约 draft-2026-06-17-r0; PJ2026-0102 Agent编排 draft-2026-06-17-r0
- * 职责: Cloud runtime durable store。Code Agent trace/session 投影事实必须从同一持久化适配器写入和恢复。
+ * SPEC: PJ2026-010403 API契约 draft-2026-06-17-r0; PJ2026-0102 Agent编排 draft-2026-06-17-r0; PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p1-agentrun-incremental-cursor.
+ * 职责: Cloud runtime durable store。Code Agent trace/session/projection cursor 投影事实必须从同一持久化适配器写入和恢复。
  */
 import { createHash, randomUUID } from "node:crypto";
 
@@ -51,13 +51,17 @@ const postgresCountTables = Object.freeze([
   "agent_sessions",
   "account_workspaces",
   "worker_sessions",
-  "agent_trace_events"
+  "agent_trace_events",
+  "workbench_projection_state"
 ]);
 
 const postgresRuntimeReadIndexes = Object.freeze([
   "CREATE INDEX IF NOT EXISTS idx_agent_trace_events_trace_order ON agent_trace_events(trace_id, occurred_at, id)",
   "CREATE INDEX IF NOT EXISTS idx_agent_trace_events_session_order ON agent_trace_events(agent_session_id, occurred_at, id)",
-  "CREATE INDEX IF NOT EXISTS idx_agent_trace_events_worker_order ON agent_trace_events(worker_session_id, occurred_at, id)"
+  "CREATE INDEX IF NOT EXISTS idx_agent_trace_events_worker_order ON agent_trace_events(worker_session_id, occurred_at, id)",
+  "CREATE INDEX IF NOT EXISTS idx_workbench_projection_state_status_retry_updated ON workbench_projection_state(projection_status, next_retry_at, updated_at)",
+  "CREATE INDEX IF NOT EXISTS idx_workbench_projection_state_run_command ON workbench_projection_state(run_id, command_id)",
+  "CREATE INDEX IF NOT EXISTS idx_workbench_projection_state_session_updated ON workbench_projection_state(session_id, updated_at DESC)"
 ]);
 
 export function createCloudRuntimeStore(options = {}) {
@@ -103,6 +107,7 @@ export class CloudRuntimeStore {
     this.auditEvents = new Map();
     this.evidenceRecords = new Map();
     this.agentTraceEvents = new Map();
+    this.workbenchProjectionStates = new Map();
   }
 
   summary() {
@@ -124,7 +129,8 @@ export class CloudRuntimeStore {
         hardwareOperations: this.hardwareOperations.size,
         auditEvents: this.auditEvents.size,
         evidenceRecords: this.evidenceRecords.size,
-        agentTraceEvents: this.agentTraceEvents.size
+        agentTraceEvents: this.agentTraceEvents.size,
+        workbenchProjectionStates: this.workbenchProjectionStates.size
       }
     });
   }
@@ -513,6 +519,41 @@ export class CloudRuntimeStore {
     };
   }
 
+  writeWorkbenchProjectionState(params = {}, requestMeta = {}) {
+    const state = normalizeWorkbenchProjectionState(params.projectionState ?? params.state ?? params, requestMeta, this.now());
+    this.workbenchProjectionStates.set(state.traceId, state);
+    return {
+      written: true,
+      projectionState: state,
+      persistence: this.summary()
+    };
+  }
+
+  getWorkbenchProjectionState(params = {}) {
+    const traceId = textOr(params.traceId, "");
+    const state = traceId ? this.workbenchProjectionStates.get(traceId) ?? null : null;
+    return {
+      projectionState: state,
+      found: Boolean(state),
+      persistence: this.summary()
+    };
+  }
+
+  queryWorkbenchProjectionStates(params = {}) {
+    const dueAt = textOr(params.dueAt, "");
+    const statuses = Array.isArray(params.projectionStatuses) ? new Set(params.projectionStatuses.map((item) => textOr(item, "")).filter(Boolean)) : null;
+    const records = [...this.workbenchProjectionStates.values()]
+      .filter((record) => matchesQuery(record, params, ["traceId", "sessionId", "runId", "commandId", "sourceRunId", "sourceCommandId"]))
+      .filter((record) => !statuses || statuses.has(record.projectionStatus))
+      .filter((record) => !dueAt || !record.nextRetryAt || String(record.nextRetryAt) <= dueAt)
+      .sort(compareProjectionStateRecords);
+    return {
+      states: limitResults(records, params.limit),
+      count: records.length,
+      persistence: this.summary()
+    };
+  }
+
   recordAuditEvent(input) {
     return this.writeAuditEvent(this.buildAuditEventFromInput(input, input.requestMeta), input.requestMeta).auditEvent;
   }
@@ -1021,6 +1062,68 @@ export class PostgresCloudRuntimeStore {
     };
   }
 
+  async writeWorkbenchProjectionState(params = {}, requestMeta = {}) {
+    const readiness = this.summary();
+    if (readiness.ready !== true || readiness.durable !== true || readiness.liveRuntimeEvidence !== true) {
+      await this.assertReadyForWrites();
+    }
+    const result = this.memory.writeWorkbenchProjectionState(params, requestMeta);
+    await this.persistWorkbenchProjectionState(result.projectionState);
+    return withPersistence(result, this.summary());
+  }
+
+  async getWorkbenchProjectionState(params = {}) {
+    const traceId = textOr(params.traceId, "");
+    if (!traceId) return withPersistence({ projectionState: null, found: false }, this.summary());
+    const result = await this.queryWorkbenchProjectionStates({ traceId, limit: 1 });
+    const projectionState = result.states[0] ?? null;
+    return withPersistence({ projectionState, found: Boolean(projectionState) }, result.persistence);
+  }
+
+  async queryWorkbenchProjectionStates(params = {}) {
+    await this.assertReadyForDurableReads("workbench.projection-state.query");
+    const clauses = [];
+    const queryParams = [];
+    const addTextClause = (field, column) => {
+      const value = textOr(params[field], "");
+      if (!value) return;
+      queryParams.push(value);
+      clauses.push(`${column} = $${queryParams.length}`);
+    };
+    addTextClause("traceId", "trace_id");
+    addTextClause("sessionId", "session_id");
+    addTextClause("runId", "run_id");
+    addTextClause("commandId", "command_id");
+    const statuses = Array.isArray(params.projectionStatuses) ? params.projectionStatuses.map((item) => textOr(item, "")).filter(Boolean) : [];
+    if (statuses.length > 0) {
+      queryParams.push(statuses);
+      clauses.push(`projection_status = ANY($${queryParams.length})`);
+    }
+    const dueAt = textOr(params.dueAt, "");
+    if (dueAt) {
+      queryParams.push(dueAt);
+      clauses.push(`(next_retry_at IS NULL OR next_retry_at <= $${queryParams.length})`);
+    }
+    const limit = positiveIntegerOrNull(params.limit);
+    const offset = positiveIntegerOrNull(params.offset);
+    let sql = `SELECT projection_json FROM workbench_projection_state${clauses.length ? ` WHERE ${clauses.join(" AND ")}` : ""} ORDER BY updated_at ASC, trace_id ASC`;
+    if (limit) {
+      queryParams.push(limit);
+      sql += ` LIMIT $${queryParams.length}`;
+    }
+    if (offset) {
+      queryParams.push(offset);
+      sql += ` OFFSET $${queryParams.length}`;
+    }
+    const result = await this.queryDurableReadRows("workbench.projection-state.query", sql, queryParams);
+    const states = result.rows.map((row) => parseJsonColumn(row.projection_json, null)).filter(Boolean);
+    return {
+      states,
+      count: states.length,
+      persistence: this.summary()
+    };
+  }
+
   async hydrateOperationRefs(params = {}) {
     await this.hydrateGatewaySession(params.gatewaySessionId);
     await this.hydrateBoxResource(params.resourceId);
@@ -1082,6 +1185,9 @@ export class PostgresCloudRuntimeStore {
     for (const record of newRecords(this.memory.agentTraceEvents, before.agentTraceEvents)) {
       await this.persistAgentTraceEvent(record);
     }
+    for (const record of changedRecords(this.memory.workbenchProjectionStates, before.workbenchProjectionStates)) {
+      await this.persistWorkbenchProjectionState(record);
+    }
   }
 
   async assertReadyForWrites() {
@@ -1228,6 +1334,35 @@ export class PostgresCloudRuntimeStore {
     );
   }
 
+  async persistWorkbenchProjectionState(record) {
+    await this.query(
+      "INSERT INTO workbench_projection_state (trace_id, session_id, conversation_id, thread_id, run_id, command_id, last_agentrun_seq, last_projected_seq, upstream_latest_seq, projection_status, projection_health, result_sync_state, last_projected_at, last_result_sync_at, last_error_code, last_error_message, failure_count, next_retry_at, projection_json, created_at, updated_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21) ON CONFLICT (trace_id) DO UPDATE SET session_id = EXCLUDED.session_id, conversation_id = EXCLUDED.conversation_id, thread_id = EXCLUDED.thread_id, run_id = EXCLUDED.run_id, command_id = EXCLUDED.command_id, last_agentrun_seq = EXCLUDED.last_agentrun_seq, last_projected_seq = EXCLUDED.last_projected_seq, upstream_latest_seq = EXCLUDED.upstream_latest_seq, projection_status = EXCLUDED.projection_status, projection_health = EXCLUDED.projection_health, result_sync_state = EXCLUDED.result_sync_state, last_projected_at = EXCLUDED.last_projected_at, last_result_sync_at = EXCLUDED.last_result_sync_at, last_error_code = EXCLUDED.last_error_code, last_error_message = EXCLUDED.last_error_message, failure_count = EXCLUDED.failure_count, next_retry_at = EXCLUDED.next_retry_at, projection_json = EXCLUDED.projection_json, updated_at = EXCLUDED.updated_at",
+      [
+        record.traceId,
+        record.sessionId,
+        record.conversationId,
+        record.threadId,
+        record.runId,
+        record.commandId,
+        record.lastAgentRunSeq,
+        record.lastProjectedSeq,
+        record.upstreamLatestSeq,
+        record.projectionStatus,
+        record.projectionHealth,
+        record.resultSyncState,
+        record.lastProjectedAt,
+        record.lastResultSyncAt,
+        record.lastErrorCode,
+        record.lastErrorMessage,
+        record.failureCount,
+        record.nextRetryAt,
+        stableJson(record),
+        record.createdAt,
+        record.updatedAt
+      ]
+    );
+  }
+
   async readCounts() {
     const counts = {};
     for (const table of postgresCountTables) {
@@ -1444,6 +1579,53 @@ function normalizeAgentTraceEvent(value, requestMeta = {}, now) {
   };
 }
 
+function normalizeWorkbenchProjectionState(value, requestMeta = {}, now) {
+  const input = normalizeJsonObject(value);
+  const traceId = textOr(input.traceId ?? requestMeta.traceId, "");
+  const runId = textOr(input.runId ?? input.sourceRunId, "");
+  const commandId = textOr(input.commandId ?? input.sourceCommandId, "");
+  if (!traceId || !runId || !commandId) {
+    throw new HwlabProtocolError("workbench projection state requires traceId, runId, and commandId", {
+      code: ERROR_CODES.invalidParams,
+      data: { required: ["traceId", "runId", "commandId"] }
+    });
+  }
+  const timestamp = timestampOr(input.updatedAt, now);
+  const lastAgentRunSeq = nonNegativeInteger(input.lastAgentRunSeq ?? input.lastSourceSeq);
+  const lastProjectedSeq = nonNegativeInteger(input.lastProjectedSeq ?? lastAgentRunSeq);
+  const upstreamLatestSeq = nonNegativeIntegerOrNull(input.upstreamLatestSeq ?? input.sourceLatestSeq);
+  return pruneUndefined({
+    ...input,
+    traceId,
+    sessionId: textOr(input.sessionId ?? requestMeta.sessionId, "") || null,
+    conversationId: textOr(input.conversationId, "") || null,
+    threadId: textOr(input.threadId, "") || null,
+    ownerUserId: textOr(input.ownerUserId, "") || null,
+    ownerRole: textOr(input.ownerRole, "") || null,
+    sourceRunId: runId,
+    sourceCommandId: commandId,
+    runId,
+    commandId,
+    lastSourceSeq: lastAgentRunSeq,
+    lastAgentRunSeq,
+    lastProjectedSeq,
+    sourceLatestSeq: upstreamLatestSeq ?? lastAgentRunSeq,
+    upstreamLatestSeq: upstreamLatestSeq ?? lastAgentRunSeq,
+    projectionStatus: projectionStatusOr(input.projectionStatus, "projecting"),
+    projectionHealth: projectionHealthOr(input.projectionHealth, "healthy"),
+    resultSyncState: resultSyncStateOr(input.resultSyncState, "not_started"),
+    lastProjectedAt: timestampOr(input.lastProjectedAt, timestamp),
+    lastResultSyncAt: input.lastResultSyncAt ? timestampOr(input.lastResultSyncAt, timestamp) : null,
+    lastErrorCode: textOr(input.lastErrorCode, "") || null,
+    lastErrorMessage: textOr(input.lastErrorMessage, "") || null,
+    failureCount: nonNegativeInteger(input.failureCount),
+    nextRetryAt: input.nextRetryAt ? timestampOr(input.nextRetryAt, timestamp) : null,
+    createdAt: timestampOr(input.createdAt, timestamp),
+    updatedAt: timestamp,
+    valuesPrinted: false
+  });
+}
+
 function compareTraceEventRecords(left, right) {
   const leftSeq = positiveIntegerOrNull(left?.event?.seq);
   const rightSeq = positiveIntegerOrNull(right?.event?.seq);
@@ -1452,6 +1634,11 @@ function compareTraceEventRecords(left, right) {
     String(left?.id ?? "").localeCompare(String(right?.id ?? ""));
 }
 
+function compareProjectionStateRecords(left, right) {
+  return String(left?.updatedAt ?? "").localeCompare(String(right?.updatedAt ?? "")) ||
+    String(left?.traceId ?? "").localeCompare(String(right?.traceId ?? ""));
+}
+
 function traceEventLevel(value) {
   const text = textOr(value, "info").toLowerCase();
   if (["failed", "failure", "error", "timeout"].includes(text)) return "error";
@@ -1465,6 +1652,31 @@ function positiveIntegerOrNull(value) {
   return Number.isInteger(parsed) && parsed > 0 ? parsed : null;
 }
 
+function nonNegativeInteger(value) {
+  const parsed = Number.parseInt(value ?? "", 10);
+  return Number.isInteger(parsed) && parsed >= 0 ? parsed : 0;
+}
+
+function nonNegativeIntegerOrNull(value) {
+  const parsed = Number.parseInt(value ?? "", 10);
+  return Number.isInteger(parsed) && parsed >= 0 ? parsed : null;
+}
+
+function projectionStatusOr(value, fallback) {
+  const text = textOr(value, fallback).toLowerCase();
+  return ["projecting", "caught_up", "terminal", "degraded", "blocked", "stalled"].includes(text) ? text : fallback;
+}
+
+function projectionHealthOr(value, fallback) {
+  const text = textOr(value, fallback).toLowerCase();
+  return ["healthy", "degraded", "unavailable", "stalled"].includes(text) ? text : fallback;
+}
+
+function resultSyncStateOr(value, fallback) {
+  const text = textOr(value, fallback).toLowerCase();
+  return ["not_started", "pending", "synced", "failed", "timed_out"].includes(text) ? text : fallback;
+}
+
 function timestampOr(value, fallback) {
   const text = textOr(value, "");
   if (text) {
@@ -1962,7 +2174,8 @@ function snapshotMemory(memory) {
     hardwareOperations: new Map(memory.hardwareOperations),
     auditEvents: new Map(memory.auditEvents),
     evidenceRecords: new Map(memory.evidenceRecords),
-    agentTraceEvents: new Map(memory.agentTraceEvents)
+    agentTraceEvents: new Map(memory.agentTraceEvents),
+    workbenchProjectionStates: new Map(memory.workbenchProjectionStates)
   };
 }
 
@@ -2008,6 +2221,7 @@ function toCountKey(table) {
   if (table === "account_workspaces") return "accountWorkspaces";
   if (table === "worker_sessions") return "workerSessions";
   if (table === "agent_trace_events") return "agentTraceEvents";
+  if (table === "workbench_projection_state") return "workbenchProjectionStates";
   return table;
 }
 

internal/db/schema.ts

@@ -5,7 +5,7 @@
 import { TABLES } from "../protocol/index.mjs";
 
 export const CLOUD_CORE_MIGRATION_ID = "0001_cloud_core_skeleton";
-export const CLOUD_RUNTIME_DURABLE_ADAPTER_SCHEMA_VERSION = "runtime-durable-postgres-v2";
+export const CLOUD_RUNTIME_DURABLE_ADAPTER_SCHEMA_VERSION = "runtime-durable-postgres-v3";
 export const CLOUD_RUNTIME_DURABLE_MIGRATION_TABLE = "hwlab_schema_migrations";
 
 export const FROZEN_CLOUD_CORE_TABLES = Object.freeze([...TABLES]);
@@ -141,6 +141,29 @@ export const CLOUD_RUNTIME_DURABLE_TABLE_COLUMNS = Object.freeze({
     "message",
     "event_json",
     "occurred_at"
+  ]),
+  workbench_projection_state: Object.freeze([
+    "trace_id",
+    "session_id",
+    "conversation_id",
+    "thread_id",
+    "run_id",
+    "command_id",
+    "last_agentrun_seq",
+    "last_projected_seq",
+    "upstream_latest_seq",
+    "projection_status",
+    "projection_health",
+    "result_sync_state",
+    "last_projected_at",
+    "last_result_sync_at",
+    "last_error_code",
+    "last_error_message",
+    "failure_count",
+    "next_retry_at",
+    "projection_json",
+    "created_at",
+    "updated_at"
   ])
 });
 

scripts/src/dev-runtime-migration.test.mjs

@@ -429,6 +429,9 @@ function createFakeQueryClient({ migrationReady, countErrorCode = null }) {
         }
         return { rows: [{ count: 0 }] };
       }
+      if (sql.startsWith("CREATE INDEX IF NOT EXISTS ")) {
+        return { rows: [] };
+      }
       throw new Error(`unexpected query: ${sql}`);
     }
   };