diff --git a/internal/cloud/server-agent-chat.test.ts b/internal/cloud/server-agent-chat.test.ts index 812c0904..38ff0cce 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-19-p0-projector-resume; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0. +// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-20-p2-terminal-outbox-recovery; 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"; @@ -2813,7 +2813,19 @@ test("cloud api resumes AgentRun projection from durable state after restart", a ] }); } if (request.method === "GET" && url.pathname === `/api/v1/runs/${runId}/commands/${commandId}/result`) { - return send({ ok: false, message: "resume projection must not block on result" }, 500); + return send({ + runId, + commandId, + status: "completed", + runStatus: "completed", + 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); }); @@ -2946,10 +2958,10 @@ test("cloud api resumes AgentRun projection from durable state after restart", a try { 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.equal(calls.some((call) => call.path === `/api/v1/runs/${runId}/commands/${commandId}/result`), false); + assert.equal(calls.some((call) => call.path === `/api/v1/runs/${runId}/commands/${commandId}/result`), true); assert.equal(projectionStates.get(traceId)?.lastSourceSeq, 23); - assert.equal(projectionStates.get(traceId)?.resultSyncState, "pending"); - assert.equal(ownerWrites.some((write) => write.status === "completed"), false); + assert.equal(projectionStates.get(traceId)?.resultSyncState, "synced"); + assert.equal(ownerWrites.some((write) => write.status === "completed"), true); assert.ok(durableEvents.some((event) => event.sourceSeq === 22)); } finally { await new Promise((resolve, reject) => { @@ -4158,6 +4170,74 @@ test("cloud api /v1/agent/chat/cancel rejects AgentRun trace scope mismatch", as } }); +test("cloud api /v1/agent/chat/cancel keeps already terminal AgentRun result instead of overwriting", async () => { + const traceId = "trc_cancel_already_terminal"; + const codeAgentChatResults = new Map(); + const traceStore = createCodeAgentTraceStore(); + codeAgentChatResults.set(traceId, { + traceId, + status: "completed", + finalResponse: "OK", + conversationId: "cnv_cancel_already_terminal", + sessionId: "ses_cancel_already_terminal", + threadId: "thread-cancel-already-terminal", + agentRun: { + runId: "run_cancel_already_terminal", + commandId: "cmd_cancel_already_terminal", + sessionId: "ses_agentrun_cancel_already_terminal", + conversationId: "cnv_cancel_already_terminal", + threadId: "thread-cancel-already-terminal", + status: "completed", + commandState: "completed", + terminalStatus: "completed" + } + }); + const accessController = { + required: false, + async authenticate() { return { ok: true, actor: TEST_AGENT_ACTOR, session: TEST_AUTH_SESSION }; } + }; + const server = createCloudApiServer({ + traceStore, + codeAgentChatResults, + accessController, + env: { + PATH: process.env.PATH, + HWLAB_CODE_AGENT_ADAPTER: "agentrun-v01", + HWLAB_CODE_AGENT_PROVIDER: "agentrun-v01" + } + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + + try { + const { port } = server.address(); + const response = await fetch(`http://127.0.0.1:${port}/v1/agent/chat/cancel`, { + method: "POST", + headers: { + "content-type": "application/json", + "x-trace-id": traceId + }, + body: JSON.stringify({ + conversationId: "cnv_cancel_already_terminal", + sessionId: "ses_cancel_already_terminal", + threadId: "thread-cancel-already-terminal", + traceId + }) + }); + assert.equal(response.status, 200); + const payload = await response.json(); + assert.equal(payload.status, "completed"); + assert.equal(payload.finalResponse, "OK"); + assert.equal(payload.cancelDisposition.status, "already_terminal"); + assert.equal(payload.cancelDisposition.forwarded, false); + assert.equal(codeAgentChatResults.get(traceId).status, "completed"); + assert.ok(traceStore.snapshot(traceId)?.events?.some((event) => event.label === "agentrun:cancel:already-terminal")); + } finally { + await new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + }); + } +}); + test("cloud api /v1/agent/chat reports Codex stdio blocker instead of structured skills fallback", async () => { const root = await mkdtemp(path.join(os.tmpdir(), "hwlab-agent-no-skills-")); const skillsDir = path.join(root, "missing-skills"); diff --git a/internal/cloud/server-code-agent-http.ts b/internal/cloud/server-code-agent-http.ts index a67d3de5..347caf2a 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-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. +// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-20-p2-terminal-outbox-recovery; PJ2026-010403 API契约 draft-2026-06-20-p2-terminal-outbox-recovery; 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"; @@ -1426,13 +1426,13 @@ function agentRunProjectionResumeCandidateFromState(state = {}, { minAgeMs = DEF async function resumeAgentRunProjectionCandidate({ candidate, options = {}, traceStore = defaultCodeAgentTraceStore } = {}) { const currentResult = await agentRunProjectionResumeResultWithCursor(candidate, options); - const resumeOptions = { ...options, deferAgentRunResultSync: true }; + const resumeOptions = { ...options, deferAgentRunResultSync: false }; const synced = await syncAgentRunChatResult({ traceId: candidate.traceId, currentResult, options: resumeOptions, traceStore, - forceResultSync: false, + forceResultSync: true, refreshEvents: true }); const payload = synced?.result ?? currentResult; @@ -2276,6 +2276,11 @@ export async function handleCodeAgentCancelHttp(request, response, options) { })); return; } + const alreadyTerminal = await syncAlreadyTerminalAgentRunBeforeCancel({ traceId, currentResult, params, options, traceStore }); + if (alreadyTerminal) { + sendJson(response, 200, alreadyTerminal); + return; + } const payload = await cancelAgentRunChatTurn({ traceId, currentResult, options, traceStore }); if (payload) { await recordCodeAgentSessionOwner({ payload, params: { ...params, traceId, ownerUserId: options.actor?.id, ownerRole: options.actor?.role }, options, status: "canceled" }); @@ -3154,6 +3159,70 @@ function cancelScopeFieldMismatch(field, requested, expectedValues) { return { field, requested, expected: expectedValues[0] }; } +async function syncAlreadyTerminalAgentRunBeforeCancel({ traceId, currentResult = null, params = {}, options = {}, traceStore = defaultCodeAgentTraceStore } = {}) { + if (!safeTraceId(traceId) || !currentResult?.agentRun?.runId || !currentResult?.agentRun?.commandId) return null; + const currentTerminalStatus = currentResult.status ?? currentResult.agentRun?.terminalStatus ?? currentResult.agentRun?.commandState ?? currentResult.agentRun?.status; + if (isTraceCommandTerminalStatus(currentTerminalStatus)) { + return alreadyTerminalCancelPayload({ traceId, payload: currentResult, params, options, traceStore }); + } + try { + const synced = await syncAgentRunChatResult({ + traceId, + currentResult, + options: { ...options, deferAgentRunResultSync: false }, + traceStore, + forceResultSync: true, + refreshEvents: true + }); + const payload = synced?.result ?? null; + const payloadTerminalStatus = payload?.status ?? payload?.agentRun?.terminalStatus ?? payload?.agentRun?.commandState ?? payload?.agentRun?.status; + if (!isTraceCommandTerminalStatus(payloadTerminalStatus)) return null; + return alreadyTerminalCancelPayload({ traceId, payload, params, options, traceStore }); + } catch (error) { + traceStore.append(traceId, { + type: "cancel", + status: "degraded", + label: "agentrun:cancel:terminal-check-degraded", + errorCode: error?.code ?? "terminal_check_failed", + message: error?.message ?? "Failed to check AgentRun terminal state before forwarding cancel.", + waitingFor: "cancel-terminal-check", + valuesPrinted: false + }); + return null; + } +} + +async function alreadyTerminalCancelPayload({ traceId, payload = {}, params = {}, options = {}, traceStore = defaultCodeAgentTraceStore } = {}) { + traceStore.append(traceId, { + type: "cancel", + status: "blocked", + label: "agentrun:cancel:already-terminal", + message: "AgentRun command was already terminal; HWLAB did not forward cancel and kept the sealed terminal projection.", + runId: payload.agentRun?.runId ?? null, + commandId: payload.agentRun?.commandId ?? null, + terminalStatus: payload.agentRun?.terminalStatus ?? payload.status, + waitingFor: "already-terminal-projection", + valuesPrinted: false + }); + await recordCodeAgentTerminalTurnStatusEffects({ + payload, + params: { ...params, traceId, ownerUserId: options.actor?.id, ownerRole: options.actor?.role }, + options, + preserveLastTraceId: true + }); + return { + ...payload, + cancelDisposition: { + status: "already_terminal", + forwarded: false, + route: "/v1/agent/chat/cancel", + message: "AgentRun command was already terminal; cancel was not forwarded.", + valuesPrinted: false + }, + runnerTrace: traceStore.snapshot(traceId) + }; +} + function cancelBlockedPayload({ code, message, diff --git a/internal/cloud/server-workbench-http.test.ts b/internal/cloud/server-workbench-http.test.ts index ff34f15f..ce18b6fe 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-19-p0-projector-resume; PJ2026-010403 API契约 draft-2026-06-18-r1. +// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-20-p2-terminal-outbox-recovery; PJ2026-010403 API契约 draft-2026-06-20-p2-terminal-outbox-recovery. // Responsibility: Workbench read model, durable projection diagnostics, and pure-GET regression tests. import assert from "node:assert/strict"; @@ -284,6 +284,65 @@ test("workbench trace event page pushes projectedSeq cursor into durable trace f } }); +test("workbench trace events stay visible after session lastTraceId advances", async () => { + const firstTraceId = "trc_workbench_historical_visible_first"; + const secondTraceId = "trc_workbench_historical_visible_second"; + const session = { + id: "ses_workbench_historical_visible", + projectId: "prj_hwpod_workbench", + agentId: "hwlab-code-agent", + status: "completed", + ownerUserId: ACTOR.id, + conversationId: "cnv_workbench_historical_visible", + threadId: "thread-workbench-historical-visible", + lastTraceId: firstTraceId, + updatedAt: "2026-06-20T10:00:00.000Z", + session: { sessionStatus: "completed", lastTraceId: firstTraceId, messages: [{ role: "user", text: "first", traceId: firstTraceId, status: "sent" }, { role: "agent", text: "OK", traceId: firstTraceId, status: "completed" }] } + }; + const facts = buildDurableFactsForSession({ + session, + status: "completed", + finalText: "OK", + events: [ + { projectedSeq: 1, sourceSeq: 1, type: "request", status: "accepted", label: "first:accepted", createdAt: "2026-06-20T09:59:58.000Z" }, + { projectedSeq: 2, sourceSeq: 2, type: "result", status: "completed", label: "first:completed", terminal: true, createdAt: "2026-06-20T10:00:00.000Z" } + ] + }); + facts.sessions[0].lastTraceId = secondTraceId; + facts.sessions[0].status = "running"; + const queries = []; + const runtimeStore = { + async queryWorkbenchFacts(params = {}) { + queries.push(params); + const filtered = filterFacts(facts, params); + return { + facts: filtered, + count: Object.values(filtered).reduce((sum, rows) => sum + rows.length, 0), + persistence: { adapter: "test-durable-workbench-facts", durable: true } + }; + } + }; + const accessController = { + store: {}, + async ensureBootstrap() {}, + async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; } + }; + const server = createCloudApiServer({ accessController, runtimeStore, codeAgentChatResults: createCodeAgentChatResultStore() }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + + try { + const { port } = server.address(); + const trace = await getJson(port, `/v1/workbench/traces/${encodeURIComponent(firstTraceId)}/events?limit=10`); + assert.equal(trace.status, 200); + assert.equal(trace.body.sessionId, session.id); + assert.deepEqual(trace.body.events.map((event) => event.label), ["first:accepted", "first:completed"]); + assert.ok(queries.some((query) => query.traceId === firstTraceId && query.families?.includes("messages") && query.families?.includes("turns"))); + assert.ok(queries.some((query) => query.sessionId === session.id && query.families?.includes("sessions") && !query.traceId)); + } finally { + await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve())); + } +}); + test("workbench read model exposes session, messages, turn, and trace without write repair", async () => { const traceStore = createCodeAgentTraceStore(); const results = createCodeAgentChatResultStore(); diff --git a/internal/cloud/server-workbench-http.ts b/internal/cloud/server-workbench-http.ts index bc5d5c07..8a7c6757 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-19-p0-projector-resume; draft-2026-06-20-p1-zero-split-durable-realtime; draft-2026-06-20-p3-read-side-zero; 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-20-p2-terminal-outbox-recovery; PJ2026-010403 API契约 draft-2026-06-20-p2-terminal-outbox-recovery; PJ2026-010401 Web工作台 draft-2026-06-20-p2-terminal-outbox-recovery; 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"; @@ -22,7 +22,7 @@ const MAX_PAGE_LIMIT = 100; const DEFAULT_WORKBENCH_SSE_HEARTBEAT_MS = 15000; const WORKBENCH_SESSION_LIST_PAGE_FAMILIES = Object.freeze(["sessions"]); const WORKBENCH_SESSION_DETAIL_FAMILIES = Object.freeze(["sessions", "messages", "parts", "turns", "checkpoints"]); -const WORKBENCH_TRACE_EVENT_METADATA_FAMILIES = Object.freeze(["sessions", "checkpoints"]); +const WORKBENCH_TRACE_EVENT_METADATA_FAMILIES = Object.freeze(["sessions", "messages", "turns", "checkpoints"]); const WORKBENCH_TRACE_EVENT_PAGE_FAMILIES = Object.freeze(["traceEvents"]); export async function handleWorkbenchReadModelHttp(request, response, url, options = {}) { try { @@ -467,6 +467,26 @@ function canActorReadFactSession(session, actor) { return session.ownerUserId === actor.id; } +async function visibleFactSessionForTrace(readModel, facts, actor, traceId) { + const visible = visibleFactSessions(facts, actor); + const byLastTrace = visible.find((item) => item.lastTraceId === traceId) ?? null; + if (byLastTrace) return { session: byLastTrace, facts }; + const relatedSessionIds = uniqueText([ + factTurnForTrace(facts, traceId)?.sessionId, + factArray(facts.messages).find((message) => message.traceId === traceId)?.sessionId, + factCheckpointForTrace(facts, traceId)?.sessionId + ]); + const alreadyLoaded = visible.find((item) => relatedSessionIds.includes(factSessionId(item))) ?? null; + if (alreadyLoaded) return { session: alreadyLoaded, facts }; + for (const sessionId of relatedSessionIds) { + const result = await queryFactsByRouteId(readModel, sessionId, { families: ["sessions"], limit: 1 }); + if (result.error) return { error: result.error }; + const session = visibleFactSessions(result.facts, actor)[0] ?? null; + if (session) return { session, facts: mergeFactSets(facts, result.facts) }; + } + return { session: null, facts }; +} + function factSessionMatchesRouteId(session, routeId) { const sessionId = safeSessionId(routeId); if (sessionId && factSessionId(session) === sessionId) return true; @@ -1034,8 +1054,10 @@ async function handleWorkbenchTraceEventPage(response, url, options, actor, rawT const pageOptions = tracePageOptions(url); const metadata = await readModel.queryFacts({ traceId, families: WORKBENCH_TRACE_EVENT_METADATA_FAMILIES, limit: 1 }); if (metadata.error) return sendJson(response, 503, workbenchProjectionStoreError(metadata.error)); - const metadataSessions = visibleFactSessions(metadata.facts, actor); - const session = metadataSessions.find((item) => item.lastTraceId === traceId) ?? metadataSessions[0] ?? null; + const resolved = await visibleFactSessionForTrace(readModel, metadata.facts, actor, traceId); + if (resolved.error) return sendJson(response, 503, workbenchProjectionStoreError(resolved.error)); + const session = resolved.session; + const metadataFacts = resolved.facts ?? metadata.facts; if (!session) { const context = await readModel.queryFacts({ traceId, families: ["sessions", "turns", "checkpoints"], limit: MAX_PAGE_LIMIT }); if (context.error) return sendJson(response, 503, workbenchProjectionStoreError(context.error)); @@ -1049,7 +1071,7 @@ async function handleWorkbenchTraceEventPage(response, url, options, actor, rawT } return sendJson(response, 404, workbenchError("workbench_trace_not_found", "Workbench trace is not visible to the current actor.", { traceId })); } - const projection = factProjectionForTrace(metadata.facts, traceId); + const projection = factProjectionForTrace(metadataFacts, traceId); const pageResult = await readModel.queryFacts({ traceId, families: WORKBENCH_TRACE_EVENT_PAGE_FAMILIES, afterProjectedSeq: pageOptions.afterProjectedSeq, limit: pageOptions.limit + 1 }); if (pageResult.error) return sendJson(response, 503, workbenchProjectionStoreError(pageResult.error)); const page = traceEventPageFromFacts(pageResult.facts.traceEvents, pageOptions, { total: projection.lastProjectedSeq, traceStatus: session.status }); diff --git a/internal/db/runtime-store.test.ts b/internal/db/runtime-store.test.ts index e543c253..4409056b 100644 --- a/internal/db/runtime-store.test.ts +++ b/internal/db/runtime-store.test.ts @@ -195,6 +195,57 @@ test("configured postgres runtime passes normalized pool config to pg", async () assert.equal(store.summary().durabilityContract.blockedLayer, "schema"); }); +test("configured postgres runtime consumes pg pool idle errors as degraded readiness", async () => { + const pools = []; + const warnings = []; + const store = createConfiguredCloudRuntimeStore({ + env: { + HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres", + HWLAB_CLOUD_RUNTIME_DURABLE: "true", + HWLAB_CLOUD_DB_PROBE_TIMEOUT_MS: "1800" + }, + dbUrl: "postgres://hwlab_user:fixture-pass@db.example.test:5432/hwlab?sslmode=require", + logger: { warn(entry) { warnings.push(entry); } }, + pgModuleLoader: async () => ({ + Pool: class FakePool { + constructor(config) { + this.config = config; + this.handlers = new Map(); + pools.push(this); + } + + on(event, handler) { + this.handlers.set(event, handler); + return this; + } + + async query() { + return { rows: [] }; + } + + emit(event, error) { + this.handlers.get(event)?.(error); + } + } + }) + }); + + await store.readiness(); + const error = new Error("connection timed out after fixture secret host"); + error.code = "ETIMEDOUT"; + pools[0].emit("error", error); + + const summary = store.summary(); + assert.equal(summary.blocker, RUNTIME_DURABLE_ADAPTER_QUERY_BLOCKED); + assert.equal(summary.connection.queryResult, "query_blocked"); + assert.equal(summary.connection.errorCode, "ETIMEDOUT"); + assert.equal(summary.durabilityContract.blockedLayer, "durability_query"); + assert.equal(warnings.length, 1); + assert.equal(warnings[0].event, "postgres_runtime_pool_error"); + assert.equal(warnings[0].valuesRedacted, true); + assert.equal(JSON.stringify(warnings).includes("fixture secret host"), false); +}); + test("configured postgres runtime classifies pg_hba no-encryption rejection as SSL before auth", async () => { const store = createConfiguredCloudRuntimeStore({ env: { diff --git a/internal/db/runtime-store.ts b/internal/db/runtime-store.ts index 3d32b60a..00a9fb5d 100644 --- a/internal/db/runtime-store.ts +++ b/internal/db/runtime-store.ts @@ -773,7 +773,8 @@ export class PostgresCloudRuntimeStore { sslMode = "require", now = () => new Date().toISOString(), queryClient, - pgModuleLoader + pgModuleLoader, + logger = console } = {}) { this.env = env; this.dbUrl = dbUrl; @@ -781,6 +782,7 @@ export class PostgresCloudRuntimeStore { this.now = now; this.queryClient = queryClient; this.pgModuleLoader = pgModuleLoader; + this.logger = logger; this.pool = null; this.runtimeReadIndexesReady = false; this.runtimeReadIndexesReadyPromise = null; @@ -1839,9 +1841,25 @@ export class PostgresCloudRuntimeStore { timeoutMs: this.env?.HWLAB_CLOUD_DB_PROBE_TIMEOUT_MS, poolMax: this.env?.[RUNTIME_DB_POOL_MAX_ENV] })); + if (typeof this.pool.on === "function") { + this.pool.on("error", (error) => this.handlePostgresPoolError(error)); + } return this.pool; } + handlePostgresPoolError(error) { + const classified = classifyRuntimeDbError(error); + this.lastReadiness = this.blockedSummary(classified); + this.logger?.warn?.({ + event: "postgres_runtime_pool_error", + blocker: classified.blocker, + queryResult: classified.connection?.queryResult ?? "query_blocked", + errorCode: classified.connection?.errorCode ?? "UNKNOWN", + valuesRedacted: true, + endpointRedacted: true + }); + } + blockedSummary({ blocker, reason, schema, migration, connection, gates }) { const blockedSchema = schema ?? { ready: false,