diff --git a/internal/cloud/code-agent-agentrun-adapter.ts b/internal/cloud/code-agent-agentrun-adapter.ts index a33ecd19..e5ecfba9 100644 --- a/internal/cloud/code-agent-agentrun-adapter.ts +++ b/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 })); diff --git a/internal/cloud/server-agent-chat.test.ts b/internal/cloud/server-agent-chat.test.ts index 490395c6..cef68326 100644 --- a/internal/cloud/server-agent-chat.test.ts +++ b/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())); diff --git a/internal/cloud/server-code-agent-http.ts b/internal/cloud/server-code-agent-http.ts index d0b3b482..82e8545d 100644 --- a/internal/cloud/server-code-agent-http.ts +++ b/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) { diff --git a/internal/cloud/workbench-projection-cursor.test.ts b/internal/cloud/workbench-projection-cursor.test.ts new file mode 100644 index 00000000..08633544 --- /dev/null +++ b/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"); +}); diff --git a/internal/cloud/workbench-projection-cursor.ts b/internal/cloud/workbench-projection-cursor.ts new file mode 100644 index 00000000..8ca74894 --- /dev/null +++ b/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(); +} diff --git a/internal/db/migrations/0001_cloud_core_skeleton.sql b/internal/db/migrations/0001_cloud_core_skeleton.sql index f7b8ab5b..dcc7cf0a 100644 --- a/internal/db/migrations/0001_cloud_core_skeleton.sql +++ b/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"}' ) diff --git a/internal/db/runtime-store.test.ts b/internal/db/runtime-store.test.ts index 09f8caec..a5b9738c 100644 --- a/internal/db/runtime-store.test.ts +++ b/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}`); } }; diff --git a/internal/db/runtime-store.ts b/internal/db/runtime-store.ts index d5d4e26c..66da7420 100644 --- a/internal/db/runtime-store.ts +++ b/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; } diff --git a/internal/db/schema.ts b/internal/db/schema.ts index 7c019f32..5743ed79 100644 --- a/internal/db/schema.ts +++ b/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" ]) }); diff --git a/scripts/src/dev-runtime-migration.test.mjs b/scripts/src/dev-runtime-migration.test.mjs index 567d7cee..73db813f 100644 --- a/scripts/src/dev-runtime-migration.test.mjs +++ b/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}`); } };