diff --git a/internal/cloud/code-agent-agentrun-adapter.ts b/internal/cloud/code-agent-agentrun-adapter.ts index 4e58e99b..6b4ca805 100644 --- a/internal/cloud/code-agent-agentrun-adapter.ts +++ b/internal/cloud/code-agent-agentrun-adapter.ts @@ -1,4 +1,4 @@ -// 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. +// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p1-agentrun-incremental-cursor; draft-2026-06-20-p2-terminal-outbox-recovery; 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"; @@ -737,6 +737,33 @@ export async function refreshAgentRunTrace({ traceId, result = null, options = { export async function cancelAgentRunChatTurn({ traceId, currentResult = null, options = {}, traceStore = defaultCodeAgentTraceStore }) { const mapped = currentResult ?? await loadPersistedAgentRunResult(traceId, options); if (!mapped?.agentRun?.commandId) return null; + const localTerminal = agentRunCancelTerminalStatus(mapped); + if (localTerminal) return agentRunAlreadyTerminalCancelPayload({ traceId, payload: mapped, terminalStatus: localTerminal, options, traceStore }); + try { + const guard = await syncAgentRunChatResult({ + traceId, + currentResult: mapped, + options: { ...options, deferAgentRunResultSync: false }, + traceStore, + forceResultSync: true, + refreshEvents: true + }); + const guarded = guard?.result ?? null; + const guardedTerminal = agentRunCancelTerminalStatus(guarded); + if (guardedTerminal) return agentRunAlreadyTerminalCancelPayload({ traceId, payload: guarded, terminalStatus: guardedTerminal, options, traceStore }); + } catch (error) { + traceStore.append(traceId, agentRunTraceEvent({ + type: "cancel", + status: "running", + label: "agentrun:cancel:terminal_guard_failed", + errorCode: error?.code ?? "agentrun_cancel_terminal_guard_failed", + message: "HWLAB could not confirm AgentRun terminal state before forwarding cancel; continuing with cancel because no terminal authority was observed.", + runId: mapped.agentRun.runId, + commandId: mapped.agentRun.commandId, + waitingFor: "cancel-terminal-guard", + valuesPrinted: false + }, mapped.agentRun)); + } const env = options.env ?? process.env; const fetchImpl = options.fetchImpl ?? globalThis.fetch; const managerUrl = resolveAgentRunManagerUrl(env, mapped.agentRun.managerUrl); @@ -783,6 +810,45 @@ export async function cancelAgentRunChatTurn({ traceId, currentResult = null, op return payload; } +function agentRunCancelTerminalStatus(payload = null) { + if (!payload || typeof payload !== "object") return null; + for (const value of [payload.status, payload.agentRun?.terminalStatus, payload.agentRun?.commandState, payload.agentRun?.status, payload.agentRun?.runStatus]) { + const status = normalizeAgentRunStatus(value); + if (isAgentRunTerminalStatus(status)) return status; + } + return null; +} + +function agentRunAlreadyTerminalCancelPayload({ traceId, payload, terminalStatus, options = {}, traceStore = defaultCodeAgentTraceStore } = {}) { + const normalizedTerminalStatus = normalizeAgentRunStatus(terminalStatus); + const agentRun = payload?.agentRun && typeof payload.agentRun === "object" ? payload.agentRun : {}; + traceStore.append(traceId, agentRunTraceEvent({ + type: "cancel", + status: normalizedTerminalStatus || "completed", + label: "agentrun:cancel:already_terminal", + message: "AgentRun command is already terminal; HWLAB did not forward cancel or overwrite the terminal Workbench projection.", + runId: agentRun.runId, + commandId: agentRun.commandId, + terminal: true, + valuesPrinted: false + }, agentRun)); + const payloadStatus = normalizeAgentRunStatus(payload?.status); + const status = isAgentRunTerminalStatus(payloadStatus) ? payloadStatus : normalizedTerminalStatus || payloadStatus || "completed"; + const result = { + ...payload, + status, + canceled: false, + alreadyTerminal: true, + cancelStatus: "already_terminal", + updatedAt: nowIso(options.now), + runnerTrace: traceStore.snapshot(traceId), + agentRun: { ...agentRun, terminalStatus: agentRun.terminalStatus ?? normalizedTerminalStatus, valuesPrinted: false }, + valuesPrinted: false + }; + options.codeAgentChatResults?.set?.(traceId, result); + return result; +} + export async function steerAgentRunChatTurn({ traceId, currentResult = null, params = {}, options = {}, traceStore = defaultCodeAgentTraceStore }) { const mapped = currentResult ?? await loadPersistedAgentRunResult(traceId, options); if (!safeTraceId(traceId) || !mapped?.agentRun?.runId || !mapped?.agentRun?.commandId) return null; diff --git a/internal/cloud/server-agent-chat.test.ts b/internal/cloud/server-agent-chat.test.ts index 38ff0cce..865f4d01 100644 --- a/internal/cloud/server-agent-chat.test.ts +++ b/internal/cloud/server-agent-chat.test.ts @@ -2956,7 +2956,7 @@ test("cloud api resumes AgentRun projection from durable state after restart", a await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); try { - for (let index = 0; index < 80 && (projectionStates.get(traceId)?.lastSourceSeq ?? 0) < 23; index += 1) await delay(10); + for (let index = 0; index < 80 && projectionStates.get(traceId)?.resultSyncState !== "synced"; 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`), true); assert.equal(projectionStates.get(traceId)?.lastSourceSeq, 23); @@ -4238,6 +4238,83 @@ test("cloud api /v1/agent/chat/cancel keeps already terminal AgentRun result ins } }); +test("cloud api /v1/agent/chat/cancel returns already-terminal when AgentRun result completed", async () => { + const calls = []; + const traceId = "trc_cancel_terminal_result_authority"; + const runId = "run_cancel_terminal_result_authority"; + const commandId = "cmd_cancel_terminal_result_authority"; + const finalText = "Cancel did not overwrite the completed AgentRun result."; + 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 }); + 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`) return send({ items: [] }); + if (request.method === "GET" && url.pathname === `/api/v1/runs/${runId}/commands/${commandId}/result`) { + return send({ runId, commandId, status: "completed", runStatus: "completed", commandState: "completed", terminalStatus: "completed", completed: true, reply: finalText, lastSeq: 7, eventCount: 7 }); + } + if (request.method === "POST" && url.pathname === `/api/v1/commands/${commandId}/cancel`) return send({ ok: true, canceled: true }); + 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 traceStore = createCodeAgentTraceStore(); + const codeAgentChatResults = new Map(); + codeAgentChatResults.set(traceId, { + traceId, + status: "running", + conversationId: "cnv_cancel_terminal_result_authority", + sessionId: "ses_cancel_terminal_result_authority", + threadId: "thread-cancel-terminal-result-authority", + agentRun: { + adapter: "agentrun-v01", + managerUrl: `http://127.0.0.1:${agentRunPort}`, + runId, + commandId, + traceId, + providerTrace: { traceId, runId, commandId, valuesPrinted: false }, + lastSeq: 6, + status: "running", + commandState: "running", + valuesPrinted: false + }, + valuesRedacted: true + }); + const server = createCloudApiServer({ + traceStore, + codeAgentChatResults, + env: { + PATH: process.env.PATH, + HWLAB_CODE_AGENT_ADAPTER: "agentrun-v01", + HWLAB_CODE_AGENT_PROVIDER: "agentrun-v01", + HWLAB_CODE_AGENT_AGENTRUN_ALLOW_NON_K3S_URL: "1" + } + }); + 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({ traceId, conversationId: "cnv_cancel_terminal_result_authority", sessionId: "ses_cancel_terminal_result_authority", threadId: "thread-cancel-terminal-result-authority" }) + }); + assert.equal(response.status, 200); + const payload = await response.json(); + assert.equal(payload.status, "completed"); + assert.equal(payload.cancelDisposition.status, "already_terminal"); + assert.equal(payload.cancelDisposition.forwarded, false); + assert.equal(payload.finalResponse.text, finalText); + assert.equal(calls.some((call) => call.method === "POST" && call.path === `/api/v1/commands/${commandId}/cancel`), false); + assert.equal(codeAgentChatResults.get(traceId)?.status, "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 /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 347caf2a..1591659c 100644 --- a/internal/cloud/server-code-agent-http.ts +++ b/internal/cloud/server-code-agent-http.ts @@ -1425,7 +1425,7 @@ function agentRunProjectionResumeCandidateFromState(state = {}, { minAgeMs = DEF } async function resumeAgentRunProjectionCandidate({ candidate, options = {}, traceStore = defaultCodeAgentTraceStore } = {}) { - const currentResult = await agentRunProjectionResumeResultWithCursor(candidate, options); + const { currentResult } = await agentRunProjectionResumeResultWithCursor(candidate, options); const resumeOptions = { ...options, deferAgentRunResultSync: false }; const synced = await syncAgentRunChatResult({ traceId: candidate.traceId, @@ -1450,12 +1450,15 @@ async function agentRunProjectionResumeResultWithCursor(candidate, options = {}) pageLimit: options.env?.HWLAB_CODE_AGENT_AGENTRUN_EVENTS_PAGE_LIMIT }); return { - ...candidate.result, - traceSummary: null, - agentRun: { - ...candidate.result.agentRun, - lastSeq: plan.afterSeq, - valuesPrinted: false + projectionState, + currentResult: { + ...candidate.result, + traceSummary: null, + agentRun: { + ...candidate.result.agentRun, + lastSeq: plan.afterSeq, + valuesPrinted: false + } } }; } @@ -2283,7 +2286,11 @@ export async function handleCodeAgentCancelHttp(request, response, options) { } 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" }); + if (payload.alreadyTerminal === true || isTraceCommandTerminalStatus(payload.status)) { + await recordCodeAgentTerminalTurnStatusEffects({ payload, params: { ...params, traceId, ownerUserId: options.actor?.id, ownerRole: options.actor?.role }, options, preserveLastTraceId: true }); + } else { + await recordCodeAgentSessionOwner({ payload, params: { ...params, traceId, ownerUserId: options.actor?.id, ownerRole: options.actor?.role }, options, status: "canceled" }); + } sendJson(response, 200, payload); return; } diff --git a/internal/cloud/server-workbench-http.test.ts b/internal/cloud/server-workbench-http.test.ts index ce18b6fe..98bb19ec 100644 --- a/internal/cloud/server-workbench-http.test.ts +++ b/internal/cloud/server-workbench-http.test.ts @@ -1551,6 +1551,86 @@ test("workbench read model exposes runtime trace projection query failures as pr } }); +test("workbench trace events remain visible after session lastTraceId moves", async () => { + const oldTraceId = "trc_workbench_history_old_trace"; + const newTraceId = "trc_workbench_history_new_trace"; + const session = { + id: "ses_workbench_history_trace", + projectId: "prj_hwpod_workbench", + agentId: "hwlab-code-agent", + status: "running", + ownerUserId: ACTOR.id, + conversationId: "cnv_workbench_history_trace", + threadId: "thread-workbench-history-trace", + lastTraceId: newTraceId, + updatedAt: "2026-06-20T10:10:00.000Z", + session: { + messages: [ + { role: "user", text: "old turn", traceId: oldTraceId, status: "sent", createdAt: "2026-06-20T10:00:00.000Z" }, + { role: "agent", text: "old final", traceId: oldTraceId, status: "completed", createdAt: "2026-06-20T10:00:01.000Z" } + ], + valuesRedacted: true + } + }; + const oldFacts = buildDurableFactsForSession({ + session: { ...session, status: "completed", lastTraceId: oldTraceId, updatedAt: "2026-06-20T10:01:00.000Z" }, + status: "completed", + finalText: "old final", + runId: "run_workbench_history_old_trace", + commandId: "cmd_workbench_history_old_trace", + events: [ + { projectedSeq: 1, sourceSeq: 1, type: "backend", status: "running", label: "runner:started", createdAt: "2026-06-20T10:00:00.000Z" }, + { projectedSeq: 2, sourceSeq: 2, type: "result", status: "completed", label: "runner:completed", terminal: true, createdAt: "2026-06-20T10:01:00.000Z" } + ], + lastProjectedSeq: 2 + }); + const newFacts = buildDurableFactsForSession({ + session: { ...session, session: { messages: [{ role: "user", text: "new turn", traceId: newTraceId, status: "sent" }] } }, + status: "running", + runId: "run_workbench_history_new_trace", + commandId: "cmd_workbench_history_new_trace", + events: [{ projectedSeq: 1, sourceSeq: 1, type: "backend", status: "running", label: "runner:started", createdAt: "2026-06-20T10:10:00.000Z" }], + lastProjectedSeq: 1 + }); + const facts = emptyFacts(); + mergeFacts(facts, oldFacts); + mergeFacts(facts, newFacts); + facts.sessions = newFacts.sessions; + 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 = { + async ensureBootstrap() {}, + async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; } + }; + const server = createCloudApiServer({ accessController, runtimeStore }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + + try { + const { port } = server.address(); + const response = await getJson(port, `/v1/workbench/traces/${encodeURIComponent(oldTraceId)}/events?limit=10`); + assert.equal(response.status, 200); + assert.equal(response.body.status, "succeeded"); + assert.equal(response.body.sessionId, session.id); + assert.equal(response.body.traceId, oldTraceId); + assert.equal(response.body.events.length, 2); + assert.equal(response.body.events.at(-1).status, "completed"); + assert.equal(response.body.error, undefined); + assert.equal(queries.some((query) => query.sessionId === session.id && Array.isArray(query.families) && query.families.includes("sessions")), true); + } finally { + await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve())); + } +}); + async function getJson(port, path) { const response = await fetch(`http://127.0.0.1:${port}${path}`); return { diff --git a/internal/db/migrations/0001_cloud_core_skeleton.sql b/internal/db/migrations/0001_cloud_core_skeleton.sql index 6b4efac5..9f74543e 100644 --- a/internal/db/migrations/0001_cloud_core_skeleton.sql +++ b/internal/db/migrations/0001_cloud_core_skeleton.sql @@ -505,7 +505,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-v4', + 'runtime-durable-postgres-v5', CURRENT_TIMESTAMP, '{"path":"internal/db/migrations/0001_cloud_core_skeleton.sql","runtime":"cloud-api"}' ) diff --git a/web/hwlab-cloud-web/src/api/agent.ts b/web/hwlab-cloud-web/src/api/agent.ts index 89120851..c8c18fac 100644 --- a/web/hwlab-cloud-web/src/api/agent.ts +++ b/web/hwlab-cloud-web/src/api/agent.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-18-p0-unique-projection; draft-2026-06-20-p2-terminal-outbox-recovery; PJ2026-010403 API契约 draft-2026-06-18-r1. // Responsibility: Code Agent mutation/action HTTP client for send, steer, session and cancel APIs. import { fetchJson, type ApiRequestOptions } from "./client"; @@ -10,5 +10,5 @@ export const agentAPI = { createAgentSession: (payload: Record = {}): Promise }>> => fetchJson("/v1/agent/sessions", { method: "POST", body: JSON.stringify(payload), timeoutMs: 30000, timeoutName: "Code Agent session create" }), getAgentSession: (sessionId: string): Promise }>> => fetchJson(`/v1/agent/sessions/${encodeURIComponent(sessionId)}`, { timeoutMs: 8000, timeoutName: "Code Agent session" }), deleteAgentSession: (sessionId: string): Promise; archivedCount?: number }>> => fetchJson(`/v1/agent/sessions/${encodeURIComponent(sessionId)}`, { method: "DELETE", timeoutMs: 30000, timeoutName: "Code Agent session delete" }), - cancelAgentMessage: (payload: Record): Promise> => fetchJson("/v1/agent/chat/cancel", { method: "POST", body: JSON.stringify(payload), timeoutMs: 30000, timeoutName: "Code Agent cancel" }) + cancelAgentMessage: (payload: Record): Promise> => fetchJson("/v1/agent/chat/cancel", { method: "POST", body: JSON.stringify(payload), timeoutMs: 30000, timeoutName: "Code Agent cancel" }) }; diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index 00aa7099..7198dd62 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -580,9 +580,19 @@ export const useWorkbenchStore = defineStore("workbench", () => { const traceId = message.traceId ?? message.runnerTrace?.traceId; if (!traceId) return; const response = await api.agent.cancelAgentMessage({ traceId, sessionId: message.sessionId ?? selectedSessionId.value, threadId: message.threadId ?? selectedThreadId.value }); - applyTurnStatusSnapshot(traceId, { traceId, status: firstNonEmptyString((response.data as Record | null)?.status, "canceled"), running: false, terminal: true, sessionId: message.sessionId ?? selectedSessionId.value ?? undefined, threadId: message.threadId ?? selectedThreadId.value ?? undefined } as AgentChatResultResponse); - markMessage(traceId, { status: "canceled", text: "用户已取消该 turn。" }); - void clearActiveTrace(traceId, "cancel-agent-message"); + const payload = response.data as AgentChatResultResponse | null; + const status = normalizedStatusText(payload?.status) ?? "canceled"; + const terminal = payload?.terminal === true || isTerminalMessageStatus(status); + if (response.ok && payload) { + applyTurnStatusSnapshot(traceId, payload); + if (terminal) completeTrace(traceId, payload); + } else { + applyTurnStatusSnapshot(traceId, { traceId, status: "canceled", running: false, terminal: true, sessionId: message.sessionId ?? selectedSessionId.value ?? undefined, threadId: message.threadId ?? selectedThreadId.value ?? undefined } as AgentChatResultResponse); + } + if (status === "canceled" || status === "cancelled") { + markMessage(traceId, { status: "canceled", text: "用户已取消该 turn。" }); + void clearActiveTrace(traceId, "cancel-agent-message"); + } if (message.status === "running") chatPending.value = false; currentRequest.value = null; void refreshSessions(message.sessionId ?? selectedSessionId.value);