fix: recover Workbench terminal outbox projection
This commit is contained in:
lyon
@@ -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;
|
||||
|
||||
@@ -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");
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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"}'
|
||||
)
|
||||
|
||||
@@ -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<string, unknown> = {}): Promise<ApiResult<{ session?: Record<string, unknown> }>> => fetchJson("/v1/agent/sessions", { method: "POST", body: JSON.stringify(payload), timeoutMs: 30000, timeoutName: "Code Agent session create" }),
|
||||
getAgentSession: (sessionId: string): Promise<ApiResult<{ session?: Record<string, unknown> }>> => fetchJson(`/v1/agent/sessions/${encodeURIComponent(sessionId)}`, { timeoutMs: 8000, timeoutName: "Code Agent session" }),
|
||||
deleteAgentSession: (sessionId: string): Promise<ApiResult<{ ok?: boolean; status?: string; session?: Record<string, unknown>; archivedCount?: number }>> => fetchJson(`/v1/agent/sessions/${encodeURIComponent(sessionId)}`, { method: "DELETE", timeoutMs: 30000, timeoutName: "Code Agent session delete" }),
|
||||
cancelAgentMessage: (payload: Record<string, unknown>): Promise<ApiResult<{ ok?: boolean; canceled?: boolean; error?: { code?: string; message?: string; userMessage?: string } }>> => fetchJson("/v1/agent/chat/cancel", { method: "POST", body: JSON.stringify(payload), timeoutMs: 30000, timeoutName: "Code Agent cancel" })
|
||||
cancelAgentMessage: (payload: Record<string, unknown>): Promise<ApiResult<AgentChatResponse & { canceled?: boolean; alreadyTerminal?: boolean; cancelStatus?: string; error?: { code?: string; message?: string; userMessage?: string } }>> => fetchJson("/v1/agent/chat/cancel", { method: "POST", body: JSON.stringify(payload), timeoutMs: 30000, timeoutName: "Code Agent cancel" })
|
||||
};
|
||||
|
||||
@@ -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<string, unknown> | 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);
|
||||
|
||||
Reference in New Issue
Block a user