diff --git a/deploy/deploy.yaml b/deploy/deploy.yaml index a8050f5a..c2a8d6c8 100644 --- a/deploy/deploy.yaml +++ b/deploy/deploy.yaml @@ -310,6 +310,11 @@ lanes: HWLAB_CODE_AGENT_TURN_STATUS_REFRESH_TIMEOUT_MS: "30000" HWLAB_CODE_AGENT_AGENTRUN_NO_RESPONSE_TIMEOUT_MS: "600000" HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_SYNC_POLL_INTERVAL_MS: "1000" + HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_ENABLED: "1" + HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_INITIAL_DELAY_MS: "5000" + HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_INTERVAL_MS: "60000" + HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_MIN_AGE_MS: "30000" + HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_BATCH_SIZE: "100" observable: true hwlab-user-billing: runtimeKind: go-service diff --git a/internal/cloud/server-agent-chat.test.ts b/internal/cloud/server-agent-chat.test.ts index 6bf0b5ce..3f46999f 100644 --- a/internal/cloud/server-agent-chat.test.ts +++ b/internal/cloud/server-agent-chat.test.ts @@ -1,4 +1,4 @@ -// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0. +// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p0-projector-resume; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0. // Responsibility: Cloud API AgentRun adapter and trace observability regression tests. import assert from "node:assert/strict"; @@ -2493,6 +2493,159 @@ test("cloud api turn status reuses request session lookup without loading persis } }); +test("cloud api resumes AgentRun projection from durable session after restart", async () => { + const calls = []; + const durableEvents = []; + const ownerWrites = []; + const traceId = "trc_projection_resume_restart"; + const runId = "run_projection_resume_restart"; + const commandId = "cmd_projection_resume_restart"; + const sessionId = "ses_projection_resume_restart"; + const conversationId = "cnv_projection_resume_restart"; + const finalText = "OK from resumed AgentRun projection."; + + const agentRunServer = createHttpServer(async (request, response) => { + const url = new URL(request.url || "/", "http://127.0.0.1"); + calls.push({ method: request.method, path: url.pathname, search: url.search, afterSeq: url.searchParams.get("afterSeq") }); + const send = (body, status = 200) => { + response.writeHead(status, { "content-type": "application/json" }); + response.end(`${JSON.stringify(body)}\n`); + }; + if (request.method === "GET" && url.pathname === `/api/v1/runs/${runId}/events`) { + assert.equal(url.searchParams.get("afterSeq"), "21"); + return send({ items: [ + { id: "evt_projection_resume_22", runId, seq: 22, type: "assistant_message", payload: { commandId, text: finalText }, createdAt: "2026-06-18T16:31:11.000Z" }, + { id: "evt_projection_resume_23", runId, seq: 23, type: "terminal_status", payload: { commandId, terminalStatus: "completed" }, createdAt: "2026-06-18T16:31:12.000Z" } + ] }); + } + 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: `unexpected ${request.method} ${url.pathname}` }, 404); + }); + await new Promise((resolve) => agentRunServer.listen(0, "127.0.0.1", resolve)); + const agentRunPort = agentRunServer.address().port; + + const session = testAgentSessionRecord({ + sessionId, + conversationId, + projectId: "prj_hwpod_workbench", + status: "running", + traceId, + updatedAt: "2026-06-18T16:31:04.000Z", + session: { + traceResults: { + [traceId]: { + traceId, + status: "running", + sessionId, + conversationId, + updatedAt: "2026-06-18T16:31:04.000Z", + agentRun: { + adapter: "agentrun-v01", + managerUrl: `http://127.0.0.1:${agentRunPort}`, + runId, + commandId, + traceId, + lastSeq: 0, + status: "running", + commandState: "running", + backendProfile: "codex", + valuesPrinted: false + }, + valuesRedacted: true + } + } + } + }); + + durableEvents.push({ + traceId, + seq: 27, + source: "agentrun", + sourceSeq: 21, + type: "backend", + status: "running", + label: "agentrun:backend:mcpServer/startupStatus/updated", + commandId, + runId, + createdAt: "2026-06-18T16:31:04.200Z", + valuesPrinted: false + }); + + const runtimeStore = { + async queryAgentTraceEvents(params = {}) { + return { events: durableEvents.filter((event) => event.traceId === params.traceId), count: durableEvents.length }; + }, + async writeAgentTraceEvent(params = {}) { + const event = params.event ?? params.traceEvent ?? params; + durableEvents.push(event); + return { written: true, traceEvent: event }; + } + }; + const accessController = { + required: false, + store: { + async listAgentSessionsForUser() { return [session]; }, + async getAgentSessionByTraceId(requestTraceId) { return requestTraceId === traceId ? session : null; } + }, + async authenticate() { return { ok: true, actor: TEST_AGENT_ACTOR, session: TEST_AUTH_SESSION }; }, + async recordAgentSessionOwner(input = {}) { + ownerWrites.push(input); + return { ...session, status: input.status ?? session.status, session: input.session ?? session.session }; + } + }; + + const server = createCloudApiServer({ + accessController, + runtimeStore, + codeAgentChatResults: new Map(), + env: { + HWLAB_CODE_AGENT_ADAPTER: "agentrun-v01", + HWLAB_CODE_AGENT_AGENTRUN_ALLOW_NON_K3S_URL: "1", + HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_ENABLED: "1", + HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_INITIAL_DELAY_MS: "1", + HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_INTERVAL_MS: "0", + HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_MIN_AGE_MS: "0", + HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_BATCH_SIZE: "10" + } + }); + 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); + 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")); + } finally { + await new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + }); + await new Promise((resolve, reject) => { + agentRunServer.close((error) => (error ? reject(error) : resolve())); + }); + } +}); + test("cloud api trace skips AgentRun refresh when terminal snapshot is already complete (#1422)", async () => { const calls = []; const traceId = "trc_issue1422_trace_terminal_snapshot"; diff --git a/internal/cloud/server-code-agent-http.ts b/internal/cloud/server-code-agent-http.ts index 7f4572a6..fdb8e6f3 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-18-p0-unique-projection; 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-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. // Responsibility: Code Agent HTTP turn admission, lifecycle compatibility wrappers, and Workbench projection writer/finalizer integration. import { createHash, randomUUID } from "node:crypto"; @@ -40,6 +40,10 @@ const MAX_CODE_AGENT_TRACE_PAGE_LIMIT = 100; const DEFAULT_CODE_AGENT_TURN_STATUS_REFRESH_TIMEOUT_MS = 2500; const DEFAULT_CODE_AGENT_PROJECT_ID = "prj_hwpod_workbench"; const DEFAULT_CODE_AGENT_PROVIDER_PROFILE = "deepseek"; +const DEFAULT_AGENTRUN_PROJECTION_RESUME_INITIAL_DELAY_MS = 5000; +const DEFAULT_AGENTRUN_PROJECTION_RESUME_INTERVAL_MS = 60000; +const DEFAULT_AGENTRUN_PROJECTION_RESUME_MIN_AGE_MS = 30000; +const DEFAULT_AGENTRUN_PROJECTION_RESUME_BATCH_SIZE = 100; const CODE_AGENT_TERMINAL_STATUSES = new Set(["completed", "failed", "blocked", "timeout", "cancelled", "canceled"]); const CODE_AGENT_PROVIDER_PROFILE_ALIASES = Object.freeze({ codex: "codex-api" }); const CODE_AGENT_PROVIDER_PROFILE_ID_PATTERN = /^[a-z0-9][a-z0-9-]{0,63}$/u; @@ -866,6 +870,242 @@ function scheduleAgentRunProjectionSync({ traceId, params = {}, options = {}, tr }); } +export function startAgentRunProjectionResume({ options = {}, traceStore = defaultCodeAgentTraceStore, logger = console } = {}) { + const env = options.env ?? process.env; + if (!truthyFlag(env.HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_ENABLED)) { + return { started: false, reason: "disabled", stop() {}, valuesPrinted: false }; + } + const initialDelayMs = parseNonNegativeInteger(env.HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_INITIAL_DELAY_MS, DEFAULT_AGENTRUN_PROJECTION_RESUME_INITIAL_DELAY_MS); + const intervalMs = parseNonNegativeInteger(env.HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_INTERVAL_MS, DEFAULT_AGENTRUN_PROJECTION_RESUME_INTERVAL_MS); + const minAgeMs = parseNonNegativeInteger(env.HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_MIN_AGE_MS, DEFAULT_AGENTRUN_PROJECTION_RESUME_MIN_AGE_MS); + const batchSize = parsePositiveInteger(env.HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_BATCH_SIZE, DEFAULT_AGENTRUN_PROJECTION_RESUME_BATCH_SIZE); + let stopped = false; + let active = false; + let timer = null; + + const schedule = (delayMs, reason) => { + if (stopped) return; + timer = setTimeout(() => { + timer = null; + void runAgentRunProjectionResumePass({ options, traceStore, logger, batchSize, minAgeMs, reason }) + .catch((error) => logAgentRunProjectionResume(logger, { + event: "workbench_projection_resume", + ok: false, + status: "failed", + reason, + errorCode: error?.code ?? "projection_resume_pass_failed", + message: error?.message ?? "AgentRun projection resume pass failed.", + valuesRedacted: true + })) + .finally(() => { + active = false; + if (!stopped && intervalMs > 0) schedule(intervalMs, "interval"); + }); + active = true; + }, Math.max(0, Number(delayMs) || 0)); + timer.unref?.(); + }; + + schedule(initialDelayMs, "startup"); + return { + started: true, + initialDelayMs, + intervalMs, + minAgeMs, + batchSize, + valuesPrinted: false, + stop() { + stopped = true; + if (timer) clearTimeout(timer); + timer = null; + }, + active: () => active + }; +} + +async function runAgentRunProjectionResumePass({ options = {}, traceStore = defaultCodeAgentTraceStore, logger = console, batchSize, minAgeMs, reason = "manual" } = {}) { + const sessions = await listAgentRunProjectionResumeSessions(options, batchSize); + const candidates = sessions.flatMap((session) => agentRunProjectionResumeCandidates(session, { minAgeMs })); + let resumed = 0; + let terminal = 0; + let failed = 0; + for (const candidate of candidates) { + try { + const result = await resumeAgentRunProjectionCandidate({ candidate, options, traceStore }); + resumed += 1; + if (isTraceCommandTerminalStatus(result?.status)) terminal += 1; + } catch (error) { + failed += 1; + appendProjectionResumeError(traceStore, candidate, error); + } + } + logAgentRunProjectionResume(logger, { + event: "workbench_projection_resume", + ok: failed === 0, + status: failed === 0 ? "completed" : "degraded", + reason, + scannedSessions: sessions.length, + candidates: candidates.length, + resumed, + terminal, + failed, + valuesRedacted: true + }); + return { scannedSessions: sessions.length, candidates: candidates.length, resumed, terminal, failed }; +} + +async function listAgentRunProjectionResumeSessions(options = {}, batchSize = DEFAULT_AGENTRUN_PROJECTION_RESUME_BATCH_SIZE) { + const accessStore = options.accessController?.store ?? options.accessController ?? null; + if (typeof accessStore?.listAgentSessionsForUser !== "function") return []; + return await accessStore.listAgentSessionsForUser({ + actorRole: "admin", + ownerScoped: false, + includeArchived: false, + limit: batchSize, + offset: 0 + }) ?? []; +} + +function agentRunProjectionResumeCandidates(session = {}, { minAgeMs = DEFAULT_AGENTRUN_PROJECTION_RESUME_MIN_AGE_MS, nowMs = Date.now() } = {}) { + const snapshot = session?.session && typeof session.session === "object" ? session.session : {}; + const records = new Map(); + const traceResults = snapshot.traceResults && typeof snapshot.traceResults === "object" ? snapshot.traceResults : null; + if (traceResults) { + for (const [traceId, record] of Object.entries(traceResults)) { + const safeId = safeTraceId(traceId); + if (safeId && record && typeof record === "object") records.set(safeId, { ...record, traceId: safeId }); + } + } + const directTraceId = safeTraceId(snapshot.traceId ?? session.lastTraceId); + if (directTraceId && snapshot.agentRun && typeof snapshot.agentRun === "object") { + records.set(directTraceId, { ...snapshot, traceId: directTraceId }); + } + return [...records.values()] + .map((record) => agentRunProjectionResumeCandidate(session, record, { minAgeMs, nowMs })) + .filter(Boolean); +} + +function agentRunProjectionResumeCandidate(session = {}, record = {}, { minAgeMs = DEFAULT_AGENTRUN_PROJECTION_RESUME_MIN_AGE_MS, nowMs = Date.now() } = {}) { + const traceId = safeTraceId(record.traceId ?? session.lastTraceId); + const agentRun = record.agentRun && typeof record.agentRun === "object" ? record.agentRun : null; + if (!traceId || !agentRun?.runId || !agentRun?.commandId) return null; + if (isTraceCommandTerminalStatus(record.status ?? agentRun.terminalStatus ?? agentRun.commandState ?? agentRun.status)) return null; + const updatedAtMs = Date.parse(record.updatedAt ?? session.updatedAt ?? ""); + if (minAgeMs > 0 && Number.isFinite(updatedAtMs) && nowMs - updatedAtMs < minAgeMs) return null; + const sessionId = safeSessionId(record.sessionId ?? session.id) || null; + const conversationId = safeConversationId(record.conversationId ?? session.conversationId) || null; + const threadId = safeOpaqueId(record.threadId ?? session.threadId) || null; + const ownerUserId = textValue(record.ownerUserId ?? session.ownerUserId); + const ownerRole = textValue(record.ownerRole ?? session.ownerRole) || "user"; + return { + traceId, + params: { + traceId, + sessionId, + conversationId, + threadId, + projectId: record.projectId ?? session.projectId ?? null, + ownerUserId, + ownerRole + }, + result: { + ...record, + traceId, + sessionId, + conversationId, + threadId, + ownerUserId, + ownerRole, + status: record.status || session.status || "running", + finalResponse: null, + traceSummary: null, + agentRun: { + ...agentRun, + traceId, + providerTrace: agentRun.providerTrace ?? { traceId, runId: agentRun.runId, commandId: agentRun.commandId, valuesPrinted: false }, + lastSeq: Number.isFinite(Number(agentRun.lastSeq)) ? Number(agentRun.lastSeq) : 0, + valuesPrinted: false + }, + valuesRedacted: true + } + }; +} + +async function resumeAgentRunProjectionCandidate({ candidate, options = {}, traceStore = defaultCodeAgentTraceStore } = {}) { + const currentResult = await agentRunProjectionResumeResultWithCursor(candidate, options); + const synced = await syncAgentRunChatResult({ + traceId: candidate.traceId, + currentResult, + options, + traceStore, + forceResultSync: true, + refreshEvents: true + }); + const payload = synced?.result ?? currentResult; + if (isTraceCommandTerminalStatus(payload?.status)) { + await recordCodeAgentTerminalTurnStatusEffects({ payload, params: candidate.params, options, preserveLastTraceId: true }); + } + return payload; +} + +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); + return { + ...candidate.result, + traceSummary: null, + agentRun: { + ...candidate.result.agentRun, + lastSeq, + 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); +} + +function appendProjectionResumeError(traceStore, candidate, error) { + const traceId = safeTraceId(candidate?.traceId); + if (!traceId) return; + traceStore.append(traceId, { + type: "turn-status", + status: "degraded", + label: "projection-resume:sync-failed", + errorCode: error?.code ?? "projection_resume_sync_failed", + message: error?.message ?? "AgentRun projection resume failed.", + runId: candidate?.result?.agentRun?.runId ?? null, + commandId: candidate?.result?.agentRun?.commandId ?? null, + waitingFor: "agentrun-result-resume", + terminal: false, + valuesPrinted: false + }); +} + +function logAgentRunProjectionResume(logger, payload) { + try { + const line = JSON.stringify(payload); + if (payload.ok === false && typeof logger?.warn === "function") logger.warn(line); + else if (typeof logger?.log === "function") logger.log(line); + } catch { + // Projection resume logging must not affect the projector. + } +} + +function parseNonNegativeInteger(value, fallback) { + const parsed = Number.parseInt(value ?? "", 10); + return Number.isInteger(parsed) && parsed >= 0 ? parsed : fallback; +} + function agentRunProjectionActivitySignature(result, runnerTrace) { const agentRun = result?.agentRun && typeof result.agentRun === "object" ? result.agentRun : {}; const trace = runnerTrace && typeof runnerTrace === "object" ? runnerTrace : result?.runnerTrace; @@ -2617,7 +2857,7 @@ function paginateTraceSnapshot(snapshot, { sinceSeq, limit }) { const pageItems = indexedEvents.slice(startIndex, startIndex + limit); const events = pageItems.map((item) => item.event); const fromSeq = pageItems.length > 0 ? pageItems[0].seq : null; - const toSeq = pageItems.length > 0 ? pageItems[pageItems.length - 1].seq : sinceSeq; + const toSeq = pageItems.length > 0 ? pageItems[pageItems.length - 1].seq : null; const hasMore = startIndex + pageItems.length < indexedEvents.length; const totalEvents = snapshot?.eventCount ?? indexedEvents.length; return { @@ -2634,7 +2874,7 @@ function paginateTraceSnapshot(snapshot, { sinceSeq, limit }) { }, truncated: hasMore, hasMore, - nextSinceSeq: toSeq, + nextSinceSeq: pageItems.length > 0 ? toSeq : sinceSeq, fullTraceLoaded: !hasMore }; } diff --git a/internal/cloud/server-workbench-http.test.ts b/internal/cloud/server-workbench-http.test.ts index 3588037d..1f6cca0e 100644 --- a/internal/cloud/server-workbench-http.test.ts +++ b/internal/cloud/server-workbench-http.test.ts @@ -1,4 +1,4 @@ -// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-010403 API契约 draft-2026-06-18-r1. +// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p0-projector-resume; PJ2026-010403 API契约 draft-2026-06-18-r1. // Responsibility: Workbench read model, durable projection diagnostics, and pure-GET regression tests. import assert from "node:assert/strict"; @@ -36,6 +36,27 @@ test("workbench projection diagnostics exposes AgentRun refresh errors without t assert.equal(diagnostics.sourceCommandId, "cmd_projection_refresh_degraded"); }); +test("workbench projection diagnostics keeps projecting health distinct from caught-up", () => { + const traceId = "trc_projection_health_projecting"; + const result = { + traceId, + status: "running", + agentRun: { runId: "run_projection_health_projecting", commandId: "cmd_projection_health_projecting", status: "running" } + }; + const trace = { + traceId, + status: "running", + events: [{ traceId, seq: 7, sourceSeq: 7, type: "backend_status", status: "running", createdAt: "2026-06-19T04:40:00.000Z" }], + eventCount: 1, + updatedAt: "2026-06-19T04:40:00.000Z" + }; + const projection = createWorkbenchTurnProjection({ traceId, result, trace }); + const diagnostics = projectionDiagnostics({ traceId, result, trace, projection }); + assert.equal(projection.status, "running"); + assert.equal(diagnostics.projectionStatus, "projecting"); + assert.equal(diagnostics.projectionHealth, "projecting"); +}); + test("workbench read model exposes session, messages, turn, and trace without write repair", async () => { const traceStore = createCodeAgentTraceStore(); const results = createCodeAgentChatResultStore(); @@ -138,6 +159,13 @@ test("workbench read model exposes session, messages, turn, and trace without wr assert.equal(trace.body.events.length, 1); assert.equal(trace.body.hasMore, true); + const emptyTracePage = await getJson(port, `/v1/workbench/traces/${encodeURIComponent(traceId)}/events?cursor=seq:99`); + assert.equal(emptyTracePage.status, 200); + assert.equal(emptyTracePage.body.events.length, 0); + assert.equal(emptyTracePage.body.range.fromSeq, null); + assert.equal(emptyTracePage.body.range.toSeq, null); + assert.equal(emptyTracePage.body.nextSeq, 99); + assert.deepEqual(writes, { recordAgentSessionOwner: 0, updateWorkspace: 0 }); } finally { await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve())); diff --git a/internal/cloud/server-workbench-http.ts b/internal/cloud/server-workbench-http.ts index ed077c05..59384046 100644 --- a/internal/cloud/server-workbench-http.ts +++ b/internal/cloud/server-workbench-http.ts @@ -1,5 +1,5 @@ /* - * SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-010403 API契约 draft-2026-06-18-r1; PJ2026-010401 Web工作台 draft-2026-06-18-r1; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0 + * SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p0-projector-resume; PJ2026-010403 API契约 draft-2026-06-18-r1; PJ2026-010401 Web工作台 draft-2026-06-18-r1; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0 * 职责: Workbench REST read model。只读投影 session/message/turn/trace facts，不执行 AgentRun sync、billing finalize 或 workspace repair。 */ import { createHash } from "node:crypto"; @@ -663,7 +663,7 @@ function traceEventPage(snapshot, options) { const offset = startIndex >= 0 ? startIndex : indexed.length; const page = indexed.slice(offset, offset + options.limit); const events = page.map((item) => ({ ...item.event, seq: item.seq })); - const toSeq = page.length ? page[page.length - 1].seq : options.afterSeq; + const toSeq = page.length ? page[page.length - 1].seq : null; const hasMore = offset + page.length < indexed.length; return { events, @@ -677,7 +677,7 @@ function traceEventPage(snapshot, options) { total: indexed.length }, hasMore, - nextSeq: toSeq, + nextSeq: page.length ? toSeq : options.afterSeq, nextCursor: hasMore ? `seq:${toSeq}` : null, traceStatus: snapshot?.status ?? "missing", updatedAt: snapshot?.updatedAt ?? null diff --git a/internal/cloud/server.ts b/internal/cloud/server.ts index f5abc693..5c356a46 100644 --- a/internal/cloud/server.ts +++ b/internal/cloud/server.ts @@ -1,5 +1,5 @@ /* - * SPEC: PJ2026-010403 API契约 draft-2026-06-17-r0; PJ2026-010401 Web工作台 draft-2026-06-17-r0; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0 + * SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p0-projector-resume; PJ2026-010403 API契约 draft-2026-06-17-r0; PJ2026-010401 Web工作台 draft-2026-06-17-r0; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0 * 职责: Cloud API REST/SSE route dispatcher。Workbench 新资源路由应委托 read model / compat wrapper，不在 dispatcher 内写业务事实。 */ import { createServer } from "node:http"; @@ -65,7 +65,8 @@ import { handleCodeAgentSessionsHttp, handleCodeAgentSteerHttp, handleCodeAgentTurnHttp, - handleCodeAgentTraceHttp + handleCodeAgentTraceHttp, + startAgentRunProjectionResume } from "./server-code-agent-http.ts"; import { handleWorkbenchReadModelHttp, handleWorkbenchRealtimeHttp } from "./server-workbench-http.ts"; import { startWorkbenchEmptySessionGc } from "./workbench-empty-session-gc.ts"; @@ -117,6 +118,11 @@ export function createCloudApiServer(options = {}) { accessController.configureCodeAgentWorkspaceContext({ env, fetchImpl: options.fetchImpl, traceStore, codeAgentChatResults }); } const workbenchEmptySessionGc = startWorkbenchEmptySessionGc({ env, accessController, logger: console }); + const agentRunProjectionResume = startAgentRunProjectionResume({ + options: { ...options, env, runtimeStore, accessController, traceStore, codeAgentChatResults }, + traceStore, + logger: console + }); const server = createServer(async (request, response) => { const backendPerformance = backendPerformanceStore.beginHttpRequest(request, response); await withBackendPerformanceContext(backendPerformance, async () => { @@ -135,7 +141,10 @@ export function createCloudApiServer(options = {}) { } }); }); - server.on("close", () => workbenchEmptySessionGc.stop()); + server.on("close", () => { + workbenchEmptySessionGc.stop(); + agentRunProjectionResume.stop(); + }); return server; } diff --git a/internal/cloud/workbench-turn-projection.ts b/internal/cloud/workbench-turn-projection.ts index baa65cb4..7c9e62cd 100644 --- a/internal/cloud/workbench-turn-projection.ts +++ b/internal/cloud/workbench-turn-projection.ts @@ -1,5 +1,5 @@ /* - * SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-010403 API契约 draft-2026-06-18-r1. + * SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p0-projector-resume; PJ2026-010403 API契约 draft-2026-06-18-r1. * 职责: 生成唯一 Workbench turn projection；trace event 状态只作为输入证据，不直接成为 turn lifecycle。 */ @@ -188,13 +188,15 @@ function normalizeProjectionStatus(value) { function normalizeProjectionHealth(value) { const text = String(value ?? "").trim().toLowerCase().replace(/_/gu, "-"); - return ["caught-up", "degraded", "stalled", "unavailable", "unknown"].includes(text) ? text : null; + return ["caught-up", "projecting", "degraded", "stalled", "unavailable", "unknown"].includes(text) ? text : null; } function projectionHealthFor({ status, turn, hasProjectionInput, blocker }) { if (blocker?.code === "projection_store_unavailable") return "unavailable"; if (blocker) return status === "stalled" ? "stalled" : "degraded"; - if (turn.terminal || status === "caught-up" || hasProjectionInput) return "caught-up"; + if (turn.terminal || status === "caught-up") return "caught-up"; + if (status === "stalled") return "stalled"; + if (status === "projecting" || hasProjectionInput) return "projecting"; return "unknown"; }