mirrors/pikasTech-HWLAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #1537 from pikasTech/fix/1519-p5-final

Browse Source 
fix: 清零 Workbench GET read-through projection repair
This commit is contained in:
Lyon
2026-06-19 00:11:22 +08:00
committed by GitHub
parent 62308c8170 98d7b1c6f0
commit 8f11fef91e
9 changed files with 294 additions and 488 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cloud/server-agent-chat.test.ts
+70 -103
View File
@@ -1,4 +1,4 @@
// SPEC: PJ2026-01060505 Workbench Performance draft-2026-06-17-p0.
// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-18-p0-unique-projection; 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";
@@ -1153,8 +1153,8 @@ test("cloud api AgentRun adapter exposes invalid tool-call attribution in result
assert.match(payload.finalResponse.text, /invalid function arguments json string/u);
assert.equal(payload.traceSummary.terminalStatus, "failed");
assert.match(payload.traceSummary.finalAssistantRow.textPreview, /invalid function arguments json string/u);
assert.equal(payload.blocker.category, "provider_invalid_tool_call");
assert.match(payload.blocker.summary, /invalid function arguments json string/u);
assert.equal(payload.blocker?.category ?? payload.error.category, "provider_invalid_tool_call");
assert.match(payload.blocker?.summary ?? payload.error.userMessage, /invalid function arguments json string/u);
assert.equal(payload.session.status, "failed");
assert.equal(payload.session.lifecycle.requiresNewSession, false);
assert.equal(payload.sessionReuse.threadId, "thread_invalid_tool");
@@ -1881,7 +1881,7 @@ test("cloud api /v1/agent/chat/inspect returns not_found for empty conversation
}
});
test("cloud api trace uses AgentRun command result evidence when live trace store has expired", async () => {
test("cloud api trace reports expired projection without AgentRun read-through when live trace store has expired (#1519)", async () => {
const ownerSessions = new Map();
const agentRunCalls = [];
const commandFinalText = "历史 trace 已过期，但 final response 来自 AgentRun command result。";
@@ -1994,21 +1994,14 @@ test("cloud api trace uses AgentRun command result evidence when live trace stor
});
assert.equal(response.status, 200);
const body = await response.json();
assert.equal(body.ok, true);
assert.equal(body.status, "expired");
assert.equal(body.traceStatus, "expired");
assert.equal(body.ok, false);
assert.equal(body.status, "missing");
assert.equal(body.traceStatus, "missing");
assert.equal(body.eventCount, 0);
assert.equal(body.terminalEvidence.available, true);
assert.equal(body.terminalEvidence.source, "agentrun-command-result");
assert.equal(body.finalResponse.text, commandFinalText);
assert.equal(body.traceSummary.source, "agentrun-command-result");
assert.equal(body.traceSummary.sourceEventCount, 0);
assert.equal(body.agentRun.runId, "run_issue842_expired");
assert.equal(body.agentRun.commandId, "cmd_issue842_expired");
assert.equal(body.retention.liveTraceStore, "expired-or-evicted");
assert.equal(body.projectionStatus, "unknown");
assert.equal(body.retention.liveTraceStore, "missing");
assert.equal(JSON.stringify(body).includes("旧 session snapshot final"), false);
assert.ok(agentRunCalls.some((call) => call.path === "/api/v1/runs/run_issue842_expired/commands"));
assert.ok(agentRunCalls.some((call) => call.path === "/api/v1/runs/run_issue842_expired/commands/cmd_issue842_expired/result"));
assert.deepEqual(agentRunCalls, []);
assert.equal(JSON.stringify(body).includes("password"), false);
} finally {
await new Promise((resolve, reject) => {
@@ -2020,7 +2013,7 @@ test("cloud api trace uses AgentRun command result evidence when live trace stor
}
});
test("cloud api trace replays an earlier AgentRun command after same-run lastSeq advances (#955)", async () => {
test("cloud api trace does not replay an earlier AgentRun command from GET after same-run lastSeq advances (#1519)", async () => {
const calls = [];
const firstTraceId = "trc_issue955_first_trace";
const codeAgentChatResults = new Map();
@@ -2110,11 +2103,11 @@ test("cloud api trace replays an earlier AgentRun command after same-run lastSeq
const body = await response.json();
const text = JSON.stringify(body);
assert.equal(body.traceId, firstTraceId);
assert.equal(body.status, "completed");
assert.ok(calls.some((call) => call.path === `/api/v1/runs/${runId}/events` && call.afterSeq === "0"));
assert.ok(body.events.some((event) => event.commandId === firstCommandId && event.label === "item/commandExecution:completed"));
assert.ok(body.events.some((event) => event.commandId === firstCommandId && event.label === "agentrun:assistant:message"));
assert.equal(body.events.some((event) => event.commandId === secondCommandId), false);
assert.equal(body.status, "expired");
assert.equal(body.projectionStatus, "projecting");
assert.deepEqual(calls, []);
assert.equal(body.eventCount, 0);
assert.deepEqual(body.events, []);
assert.equal(body.finalResponse.text, firstFinalText);
assert.match(text, /d601-f103-v2/u);
assert.doesNotMatch(text, /编译成功/u);
@@ -2128,7 +2121,7 @@ test("cloud api trace replays an earlier AgentRun command after same-run lastSeq
}
});
test("cloud api running trace refreshes AgentRun events without waiting for command result (#1000)", async () => {
test("cloud api running trace reports current projection without AgentRun event refresh (#1519)", async () => {
const calls = [];
const traceId = "trc_issue1000_running_trace";
const runId = "run_issue1000_running_trace";
@@ -2197,11 +2190,11 @@ test("cloud api running trace refreshes AgentRun events without waiting for comm
});
assert.equal(response.status, 200);
const body = await response.json();
assert.equal(body.status, "running");
assert.equal(body.eventCount, 4);
assert.ok(body.events.some((event) => event.label === "item/commandExecution:completed"));
assert.ok(body.events.some((event) => event.label === "agentrun:assistant:message"));
assert.ok(calls.some((call) => call.path === `/api/v1/runs/${runId}/events`));
assert.equal(body.status, "missing");
assert.equal(body.projectionStatus, "projecting");
assert.equal(body.eventCount, 0);
assert.deepEqual(body.events, []);
assert.deepEqual(calls, []);
assert.equal(calls.some((call) => call.path === `/api/v1/runs/${runId}/commands/${commandId}/result`), false);
} finally {
await new Promise((resolve, reject) => {
@@ -2213,7 +2206,7 @@ test("cloud api running trace refreshes AgentRun events without waiting for comm
}
});
test("cloud api turn status reuses AgentRun result-sync trace refresh (#1422)", async () => {
test("cloud api turn status reads Workbench projection without AgentRun refresh (#1519)", async () => {
const calls = [];
const traceId = "trc_issue1422_turn_status_fast_path";
const runId = "run_issue1422_turn_status_fast_path";
@@ -2280,10 +2273,10 @@ test("cloud api turn status reuses AgentRun result-sync trace refresh (#1422)",
assert.equal(response.status, 200);
const body = await response.json();
assert.equal(body.status, "running");
assert.ok(body.eventCount >= 1);
assert.ok(Array.isArray(body.runnerTrace.events));
assert.ok(body.runnerTrace.events.length >= 1);
assert.equal(calls.filter((call) => call.path === `/api/v1/runs/${runId}/events`).length, 1);
assert.equal(body.projectionStatus, "projecting");
assert.equal(body.eventCount, 0);
assert.equal(body.runnerTrace, null);
assert.equal(calls.filter((call) => call.path === `/api/v1/runs/${runId}/events`).length, 0);
assert.equal(calls.some((call) => call.path === `/api/v1/runs/${runId}/commands/${commandId}/result`), false);
} finally {
await new Promise((resolve, reject) => {
@@ -2295,7 +2288,7 @@ test("cloud api turn status reuses AgentRun result-sync trace refresh (#1422)",
}
});
test("cloud api turn status reuses request session lookup for project check and persisted AgentRun result (#1422)", async () => {
test("cloud api turn status reuses request session lookup without loading persisted AgentRun result (#1519)", async () => {
const calls = [];
const traceId = "trc_issue1422_turn_status_session_cache";
const runId = "run_issue1422_turn_status_session_cache";
@@ -2372,12 +2365,12 @@ test("cloud api turn status reuses request session lookup for project check and
const response = await fetch(`http://127.0.0.1:${port}/v1/agent/turns/${traceId}?projectId=prj_hwpod_workbench`, {
headers: { cookie: "hwlab_session=test-stub-session" }
});
assert.equal(response.status, 200);
assert.equal(response.status, 404);
const body = await response.json();
assert.equal(body.status, "running");
assert.equal(body.sessionId, "ses_issue1422_turn_status_session_cache");
assert.equal(sessionLookupCount, 1);
assert.equal(calls.filter((call) => call.path === `/api/v1/runs/${runId}/events`).length, 1);
assert.equal(body.status, "unknown");
assert.equal(body.projectionStatus, "unknown");
assert.equal(sessionLookupCount, 2);
assert.equal(calls.filter((call) => call.path === `/api/v1/runs/${runId}/events`).length, 0);
} finally {
await new Promise((resolve, reject) => {
server.close((error) => (error ? reject(error) : resolve()));
@@ -2502,7 +2495,7 @@ test("cloud api trace skips AgentRun refresh when terminal snapshot is already c
}
});
test("cloud api trace schedules terminal side effects without blocking reads (#1422)", async () => {
test("cloud api trace does not schedule terminal side effects from GET reads (#1519)", async () => {
const ownerCalls = [];
const factCalls = [];
let releaseOwnerWrite;
@@ -2625,12 +2618,11 @@ test("cloud api trace schedules terminal side effects without blocking reads (#1
await fetchTrace();
}
for (let i = 0; i < 20 && ownerCalls.length < 1; i += 1) await delay(10);
assert.equal(ownerCalls.length, 1);
assert.equal(factCalls.length, 1);
assert.equal(codeAgentChatResults.get(traceId).turnStatusTerminalEffects.pending, true);
assert.equal(ownerCalls.length, 0);
assert.equal(factCalls.length, 0);
assert.equal(codeAgentChatResults.get(traceId).turnStatusTerminalEffects, undefined);
releaseOwnerWrite();
for (let i = 0; i < 20 && codeAgentChatResults.get(traceId).turnStatusTerminalEffects?.recorded !== true; i += 1) await delay(10);
assert.equal(codeAgentChatResults.get(traceId).turnStatusTerminalEffects.recorded, true);
assert.equal(codeAgentChatResults.get(traceId).turnStatusTerminalEffects, undefined);
} finally {
releaseOwnerWrite?.();
await new Promise((resolve, reject) => {
@@ -2732,7 +2724,7 @@ test("cloud api turn status skips AgentRun refresh for complete terminal evidenc
}
});
test("cloud api turn status records terminal side effects once (#1422)", async () => {
test("cloud api turn status does not record terminal side effects from GET reads (#1519)", async () => {
const ownerCalls = [];
const factCalls = [];
let releaseOwnerWrite;
@@ -2832,12 +2824,11 @@ test("cloud api turn status records terminal side effects once (#1422)", async (
await fetchTurnStatus();
}
for (let i = 0; i < 20 && ownerCalls.length < 1; i += 1) await delay(10);
assert.equal(ownerCalls.length, 1);
assert.equal(factCalls.length, 1);
assert.equal(codeAgentChatResults.get(traceId).turnStatusTerminalEffects.pending, true);
assert.equal(ownerCalls.length, 0);
assert.equal(factCalls.length, 0);
assert.equal(codeAgentChatResults.get(traceId).turnStatusTerminalEffects, undefined);
releaseOwnerWrite();
for (let i = 0; i < 20 && codeAgentChatResults.get(traceId).turnStatusTerminalEffects?.recorded !== true; i += 1) await delay(10);
assert.equal(codeAgentChatResults.get(traceId).turnStatusTerminalEffects.recorded, true);
assert.equal(codeAgentChatResults.get(traceId).turnStatusTerminalEffects, undefined);
} finally {
releaseOwnerWrite?.();
await new Promise((resolve, reject) => {
@@ -2846,7 +2837,7 @@ test("cloud api turn status records terminal side effects once (#1422)", async (
}
});
test("cloud api repairs historical same-session AgentRun trace after lastTraceId advances (#955)", async () => {
test("cloud api legacy trace read does not repair historical AgentRun trace after lastTraceId advances (#1519)", async () => {
const calls = [];
const ownerSessions = new Map();
const runId = "run_issue955_historical";
@@ -2963,46 +2954,23 @@ test("cloud api repairs historical same-session AgentRun trace after lastTraceId
});
assert.equal(response.status, 200);
const body = await response.json();
const text = JSON.stringify(body);
assert.equal(body.traceId, firstTraceId);
assert.equal(body.status, "completed");
assert.equal(body.finalResponse.text, firstFinalText);
assert.equal(body.agentRun.commandId, firstCommandId);
assert.equal(body.eventCount, 4);
assert.equal(body.traceSummary.sourceEventCount, body.eventCount);
assert.ok(body.events.some((event) => event.commandId === firstCommandId));
assert.equal(body.events.some((event) => event.commandId === secondCommandId), false);
assert.match(text, /d601-f103-v2/u);
assert.doesNotMatch(text, /编译成功/u);
assert.ok(calls.some((call) => call.path === `/api/v1/runs/${runId}/commands`));
assert.ok(calls.some((call) => call.path === `/api/v1/runs/${runId}/commands/${firstCommandId}/result`));
assert.equal(calls.some((call) => call.path === `/api/v1/runs/${runId}/commands/${secondCommandId}/result`), false);
for (let index = 0; index < 20 && !ownerSessions.get("ses_issue955_historical")?.session?.traceResults?.[firstTraceId]; index += 1) await delay(10);
assert.equal(ownerSessions.get("ses_issue955_historical").session.traceResults[firstTraceId].finalResponse.text, firstFinalText);
const repairedSession = ownerSessions.get("ses_issue955_historical");
assert.equal(repairedSession.lastTraceId, secondTraceId);
assert.equal(repairedSession.session.traceResults[firstTraceId].agentRun.commandId, firstCommandId);
assert.equal(body.status, "missing");
assert.equal(body.projectionStatus, "unknown");
assert.equal(body.eventCount, 0);
assert.deepEqual(calls, []);
assert.equal(ownerSessions.get("ses_issue955_historical").session.traceResults[firstTraceId], undefined);
const secondResultResponse = await fetch(`http://127.0.0.1:${port}/v1/agent/chat/result/${secondTraceId}`, {
headers: { cookie: "hwlab_session=test-stub-session" }
});
assert.equal(secondResultResponse.status, 200);
const secondResult = await secondResultResponse.json();
const secondText = JSON.stringify(secondResult);
assert.equal(secondResult.traceId, secondTraceId);
assert.equal(secondResult.status, "completed");
assert.equal(secondResult.agentRun.commandId, secondCommandId);
assert.equal(secondResult.agentRun.traceId, secondTraceId);
assert.equal(secondResult.terminalEvidence.available, true);
assert.equal(secondResult.terminalEvidence.agentRun.commandId, secondCommandId);
assert.equal(secondResult.terminalEvidence.finalResponse.traceId, secondTraceId);
assert.equal(secondResult.finalResponse.traceId, secondTraceId);
assert.match(secondText, /编译成功/u);
assert.doesNotMatch(secondText, /目前只有一个 HWPOD 可用/u);
assert.equal(secondText.includes('"fallback"'), false);
assert.ok(calls.some((call) => call.path === `/api/v1/runs/${runId}/commands/${secondCommandId}/result`));
assert.equal(ownerSessions.get("ses_issue955_historical").session.traceResults[secondTraceId].agentRun.commandId, secondCommandId);
assert.equal(ownerSessions.get("ses_issue955_historical").session.traceResults[secondTraceId].finalResponse.text, secondFinalText);
assert.equal(secondResult.projectionStatus, "projecting");
assert.equal(secondResult.agentRun.commandId, firstCommandId);
assert.deepEqual(calls, []);
assert.equal(ownerSessions.get("ses_issue955_historical").session.traceResults[secondTraceId].agentRun.commandId, firstCommandId);
} finally {
await new Promise((resolve, reject) => {
server.close((error) => (error ? reject(error) : resolve()));
@@ -3013,7 +2981,7 @@ test("cloud api repairs historical same-session AgentRun trace after lastTraceId
}
});
test("cloud api result polling repairs polluted completed AgentRun memory cache (#955)", async () => {
test("cloud api result polling does not repair polluted completed AgentRun memory cache (#1519)", async () => {
const calls = [];
const runId = "run_issue955_live_cache";
const firstTraceId = "trc_issue955_live_cache_first";
@@ -3115,20 +3083,18 @@ test("cloud api result polling repairs polluted completed AgentRun memory cache
const text = JSON.stringify(body);
assert.equal(body.traceId, secondTraceId);
assert.equal(body.status, "completed");
assert.equal(body.agentRun.commandId, secondCommandId);
assert.equal(body.projectionStatus, "projecting");
assert.equal(body.agentRun.commandId, firstCommandId);
assert.equal(body.agentRun.traceId, secondTraceId);
assert.equal(body.terminalEvidence.available, true);
assert.equal(body.terminalEvidence.agentRun.commandId, secondCommandId);
assert.equal(body.terminalEvidence.agentRun.commandId, firstCommandId);
assert.equal(body.terminalEvidence.finalResponse.traceId, secondTraceId);
assert.equal(body.reply.content, secondFinalText);
assert.match(text, /编译成功/u);
assert.match(text, /Keil MDK/u);
assert.doesNotMatch(text, /目前只有一个 HWPOD 可用/u);
assert.equal(body.reply.content, firstFinalText);
assert.match(text, /目前只有一个 HWPOD 可用/u);
assert.doesNotMatch(text, /编译成功/u);
assert.equal(text.includes('"fallback"'), false);
assert.ok(calls.some((call) => call.path === `/api/v1/runs/${runId}/commands`));
assert.ok(calls.some((call) => call.path === `/api/v1/runs/${runId}/commands/${secondCommandId}/result`));
assert.equal(calls.some((call) => call.path === `/api/v1/runs/${runId}/commands/${firstCommandId}/result`), false);
assert.equal(codeAgentChatResults.get(secondTraceId).agentRun.commandId, secondCommandId);
assert.deepEqual(calls, []);
assert.equal(codeAgentChatResults.get(secondTraceId).agentRun.commandId, firstCommandId);
} finally {
await new Promise((resolve, reject) => {
server.close((error) => (error ? reject(error) : resolve()));
@@ -3139,7 +3105,7 @@ test("cloud api result polling repairs polluted completed AgentRun memory cache
}
});
test("cloud api result polling fails closed when AgentRun command registry misses trace (#955)", async () => {
test("cloud api result polling does not query AgentRun registry when projection is stale (#1519)", async () => {
const calls = [];
const runId = "run_issue955_registry_missing";
const traceId = "trc_issue955_registry_missing";
@@ -3201,14 +3167,15 @@ test("cloud api result polling fails closed when AgentRun command registry misse
const response = await fetch(`http://127.0.0.1:${port}/v1/agent/chat/result/${traceId}`, {
headers: { cookie: "hwlab_session=test-stub-session" }
});
assert.equal(response.status, 404);
assert.equal(response.status, 200);
const body = await response.json();
const text = JSON.stringify(body);
assert.equal(body.error.code, "agentrun_trace_command_not_found");
assert.equal(body.error.traceId, traceId);
assert.equal(body.error.runId, runId);
assert.doesNotMatch(text, /目前只有一个 HWPOD 可用/u);
assert.ok(calls.some((call) => call.path === `/api/v1/runs/${runId}/commands`));
assert.equal(body.traceId, traceId);
assert.equal(body.status, "completed");
assert.equal(body.projectionStatus, "projecting");
assert.equal(body.agentRun.commandId, staleCommandId);
assert.match(text, /目前只有一个 HWPOD 可用/u);
assert.deepEqual(calls, []);
assert.equal(calls.some((call) => call.path.includes("/result")), false);
} finally {
await new Promise((resolve, reject) => {
internal/cloud/server-code-agent-http.ts
+83 -264
View File
@@ -10,7 +10,6 @@ import {
codeAgentAgentRunAdapterEnabled,
initialAgentRunChatResult,
loadPersistedAgentRunResult,
refreshAgentRunTrace,
steerAgentRunChatTurn,
submitAgentRunChatTurn,
syncAgentRunChatResult
@@ -20,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 { createWorkbenchReadModel } from "./workbench-read-model.ts";
import {
firstHeaderValue,
getHeader,
@@ -1071,111 +1071,28 @@ export async function handleCodeAgentChatResultHttp(request, response, url, opti
});
return;
}
const results = options.codeAgentChatResults;
const result = results?.get(traceId) ?? null;
if (result && !canAccessOwnedResult(result, options.actor)) {
sendJson(response, 403, {
error: {
code: "agent_session_owner_required",
message: "Only the session owner or admin can read this Code Agent result"
}
});
const context = await measureCodeAgentHttpPhase(options, "result_projection_read", () => readCodeAgentCompatProjection(traceId, options));
if (context.statusCode) {
sendJson(response, context.statusCode, context.body);
return;
}
const adapterEnabled = codeAgentAgentRunAdapterEnabled(options.env ?? process.env);
if ((result?.agentRun?.runId && adapterEnabled) || (!result && adapterEnabled)) {
try {
const synced = await syncAgentRunChatResult({ traceId, currentResult: result, options, traceStore: options.traceStore ?? defaultCodeAgentTraceStore });
if (synced.result && !canAccessOwnedResult(synced.result, options.actor)) {
sendJson(response, 403, {
error: {
code: "agent_session_owner_required",
message: "Only the session owner or admin can read this Code Agent result"
}
});
return;
}
if (synced.result && synced.result.status !== "running") {
await finalizeCodeAgentBillingUsage({ payload: synced.result, params: synced.result, options });
recordCodeAgentConversationFact(synced.result, options);
await recordCodeAgentSessionOwner({ payload: synced.result, params: synced.result, options, status: codeAgentOwnerStatusForResult(synced.result) });
sendJson(response, 200, compactCodeAgentChatResultPayload(synced.result, options));
return;
}
if (synced.result) {
sendJson(response, 202, {
accepted: true,
status: "running",
shortConnection: true,
traceId,
agentRun: synced.result.agentRun,
runnerTrace: compactRunnerTraceForResult(synced.runnerTrace, resultTraceEventLimit(options)),
waitingFor: synced.runnerTrace?.waitingFor ?? "agentrun-result"
});
return;
}
} catch (error) {
const traceStore = options.traceStore ?? defaultCodeAgentTraceStore;
traceStore.append(traceId, {
type: "result",
status: "degraded",
label: "agentrun:result:poll-failed",
errorCode: error?.code ?? "agentrun_result_poll_failed",
message: error?.message ?? "AgentRun result polling failed",
runId: result?.agentRun?.runId ?? null,
commandId: result?.agentRun?.commandId ?? null,
timeoutMs: numberOrNull(error?.timeoutMs),
timeoutStage: textValue(error?.timeoutStage) || null,
upstreamRoute: textValue(error?.route ?? error?.path) || null,
waitingFor: "agentrun-result",
valuesPrinted: false
});
if (result?.agentRun && result.status === "running") {
sendJson(response, 202, {
accepted: true,
status: "running",
shortConnection: true,
traceId,
agentRun: result.agentRun,
runnerTrace: compactRunnerTraceForResult(traceStore.snapshot(traceId), resultTraceEventLimit(options)),
waitingFor: "agentrun-result",
degraded: true,
error: codeAgentRefreshErrorPayload(error, traceId, result?.agentRun, "agentrun_result_poll_failed")
});
return;
}
if (result?.agentRun) {
sendJson(response, error?.statusCode === 404 ? 404 : 502, {
error: {
code: error?.code ?? "agentrun_result_poll_failed",
message: error?.message ?? "AgentRun result polling failed",
traceId,
runId: result.agentRun.runId ?? null,
commandId: result.agentRun.commandId ?? null
}
});
return;
}
}
}
const result = context.result;
if (result && result.status !== "running") {
sendJson(response, 200, compactCodeAgentChatResultPayload(result, options));
sendJson(response, 200, codeAgentCompatProjectionPayload(compactCodeAgentChatResultPayload(result, options), context));
return;
}
const traceStore = options.traceStore ?? defaultCodeAgentTraceStore;
const runnerTrace = traceStore.snapshot(traceId);
if (result || runnerTrace.status !== "missing") {
sendJson(response, 202, {
accepted: true,
status: "running",
if (context.found) {
const body = codeAgentTurnStatusPayload({ traceId, result, snapshot: context.trace, resultPollError: null, refreshError: null, options });
sendJson(response, body.terminal === true ? 200 : 202, codeAgentCompatProjectionPayload({
accepted: body.ok,
shortConnection: true,
traceId,
runnerTrace: compactRunnerTraceForResult(runnerTrace, resultTraceEventLimit(options)),
waitingFor: runnerTrace.waitingFor ?? "codex-stdio"
});
waitingFor: body.waitingFor ?? context.trace?.waitingFor ?? "workbench-projection",
...body
}, context));
return;
}
sendJson(response, 404, {
...codeAgentCompatProjectionPayload({ traceId, status: "unknown" }, context),
error: {
code: "code_agent_result_not_found",
message: `No Code Agent result is registered for ${traceId}`
@@ -1210,74 +1127,17 @@ export async function handleCodeAgentTurnHttp(request, response, url, options) {
}
async function resolveCodeAgentTurnStatusSnapshot(traceId, options) {
const traceStore = options.traceStore ?? defaultCodeAgentTraceStore;
const refreshOptions = codeAgentTurnStatusRefreshOptions(options);
let result = options.codeAgentChatResults?.get(traceId) ?? null;
if (result && !canAccessOwnedResult(result, options.actor)) return forbiddenTurnSnapshot(traceId);
const adapterEnabled = codeAgentAgentRunAdapterEnabled(options.env ?? process.env);
let resultPollError = null;
let turnRefreshSatisfiedByResultSync = false;
if (adapterEnabled && (result?.agentRun?.runId || !result)) {
try {
const synced = await measureCodeAgentHttpPhase(options, "turn_result_sync", () => syncAgentRunChatResult({ traceId, currentResult: result, options: refreshOptions, traceStore }));
result = synced.result ?? result;
turnRefreshSatisfiedByResultSync = synced.eventsRefreshed === true || synced.resultSynced === true || synced.terminalRefreshSkipped === true;
if (result && !canAccessOwnedResult(result, options.actor)) return forbiddenTurnSnapshot(traceId);
if (result && isTraceCommandTerminalStatus(result.status)) {
scheduleCodeAgentTerminalTurnStatusEffects({ payload: result, params: result, options });
}
} catch (error) {
resultPollError = error;
traceStore.append(traceId, {
type: "turn-status",
status: "degraded",
label: "turn-status:result-sync-failed",
errorCode: error?.code ?? "agentrun_result_poll_failed",
message: error?.message ?? "AgentRun result polling failed",
runId: result?.agentRun?.runId ?? null,
commandId: result?.agentRun?.commandId ?? null,
timeoutMs: numberOrNull(error?.timeoutMs),
timeoutStage: textValue(error?.timeoutStage) || null,
upstreamRoute: textValue(error?.route ?? error?.path) || null,
waitingFor: "agentrun-result",
valuesPrinted: false
});
}
}
let agentRunResult = result?.agentRun ? result : null;
if (!agentRunResult && adapterEnabled) {
try {
agentRunResult = await measureCodeAgentHttpPhase(options, "turn_load_persisted_result", () => loadPersistedAgentRunResult(traceId, options));
} catch {
agentRunResult = null;
}
}
if (agentRunResult && !canAccessOwnedResult(agentRunResult, options.actor)) return forbiddenTurnSnapshot(traceId);
let refreshError = null;
if (agentRunResult?.agentRun && !turnRefreshSatisfiedByResultSync && !resultPollError) {
try {
const refreshedTrace = await measureCodeAgentHttpPhase(options, "turn_trace_refresh", () => refreshAgentRunTrace({ traceId, result: agentRunResult, options: refreshOptions, traceStore }));
agentRunResult = options.codeAgentChatResults?.get?.(traceId) ?? agentRunResult;
if (traceNeedsCommandResultSync(agentRunResult, refreshedTrace)) {
const synced = await measureCodeAgentHttpPhase(options, "turn_force_result_sync", () => syncAgentRunChatResult({ traceId, currentResult: agentRunResult, options: refreshOptions, traceStore, appendResultEvent: false, refreshEvents: false, forceResultSync: true }));
agentRunResult = synced.result ?? agentRunResult;
}
if (isTraceCommandTerminalStatus(agentRunResult?.status)) {
scheduleCodeAgentTerminalTurnStatusEffects({ payload: agentRunResult, params: agentRunResult, options, preserveLastTraceId: true });
}
} catch (error) {
refreshError = error;
}
}
const body = await measureCodeAgentHttpPhase(options, "turn_payload", () => {
const snapshot = traceSnapshotWithTerminalEvidence(traceStore.snapshot(traceId), agentRunResult ?? result, traceId, refreshError);
return codeAgentTurnStatusPayload({ traceId, result: agentRunResult ?? result, snapshot, resultPollError, refreshError, options });
});
return { statusCode: body.ok ? 200 : 404, body };
const context = await readCodeAgentCompatProjection(traceId, options);
if (context.statusCode) return context;
const body = await measureCodeAgentHttpPhase(options, "turn_payload", () => codeAgentTurnStatusPayload({
traceId,
result: context.result,
snapshot: context.trace,
resultPollError: null,
refreshError: null,
options
}));
return { statusCode: body.ok ? 200 : 404, body: codeAgentCompatProjectionPayload(body, context) };
}
function codeAgentTurnStatusRefreshOptions(options = {}) {
@@ -1325,6 +1185,59 @@ function forbiddenTurnSnapshot(traceId) {
};
}
async function readCodeAgentCompatProjection(traceId, options = {}) {
const readModel = createWorkbenchReadModel(options, options.actor ?? null);
const session = await readModel.getSessionByTraceId(traceId);
const result = readModel.resultForTrace(traceId) ?? projectedSessionTraceResult(session, traceId);
if (result && !canAccessOwnedResult(result, options.actor)) return forbiddenTurnSnapshot(traceId);
if (result?.ownerUserId && !readModel.canReadOwner(result.ownerUserId)) return forbiddenTurnSnapshot(traceId);
const projectionTrace = await readModel.traceSnapshot(traceId);
const trace = traceSnapshotWithTerminalEvidence(projectionTrace, result, traceId, null);
const projection = readModel.projectionDiagnostics({ traceId, result, trace });
const found = Boolean(result || session || (trace && trace.status !== "missing") || trace?.persisted === true);
return { result, session, trace, projection, found };
}
function projectedSessionTraceResult(session, traceId) {
const safeId = safeTraceId(traceId);
if (!session || !safeId) return null;
const snapshot = session.session && typeof session.session === "object" ? session.session : null;
const traceResults = snapshot?.traceResults && typeof snapshot.traceResults === "object" ? snapshot.traceResults : null;
const stored = traceResults?.[safeId] && typeof traceResults[safeId] === "object" ? traceResults[safeId] : null;
if (!stored) return null;
return {
...stored,
traceId: safeId,
conversationId: safeConversationId(stored.conversationId ?? session.conversationId) || null,
sessionId: safeSessionId(stored.sessionId ?? session.id) || null,
threadId: safeOpaqueId(stored.threadId ?? session.threadId) || null,
ownerUserId: session.ownerUserId ?? stored.ownerUserId ?? null,
ownerRole: session.ownerRole ?? stored.ownerRole ?? null,
status: stored.status ?? session.status ?? null,
valuesRedacted: true
};
}
function codeAgentCompatProjectionPayload(payload = {}, context = {}) {
const traceId = safeTraceId(payload.traceId ?? context.trace?.traceId) ?? null;
const projection = context.projection && typeof context.projection === "object" ? context.projection : {};
return {
...payload,
projection,
projectionStatus: projection.projectionStatus ?? null,
lastProjectedSeq: projection.lastProjectedSeq ?? null,
sourceRunId: projection.sourceRunId ?? null,
sourceCommandId: projection.sourceCommandId ?? null,
blocker: projection.blocker ?? null,
workbench: traceId ? {
turnUrl: `/v1/workbench/turns/${encodeURIComponent(traceId)}`,
traceEventsUrl: `/v1/workbench/traces/${encodeURIComponent(traceId)}/events`
} : null,
valuesRedacted: true,
secretMaterialStored: false
};
}
async function traceProjectMismatchSnapshot(traceId, url, options, resourceKind) {
const requestedProjectId = textValue(url.searchParams.get("projectId"));
if (!requestedProjectId || !options.accessController?.getAgentSessionByTraceId) return null;
@@ -2066,29 +1979,6 @@ function codeAgentOwnerStatusForResult(result = {}) {
return result?.status ?? "active";
}
async function resolveWorkbenchActiveTraceResult(traceId, options = {}) {
if (!safeTraceId(traceId)) return null;
const cached = options.codeAgentChatResults?.get?.(traceId) ?? null;
if (cached?.agentRun?.runId) {
try {
const synced = await syncAgentRunChatResult({ traceId, currentResult: cached, options, traceStore: options.traceStore ?? defaultCodeAgentTraceStore });
return synced.result ?? null;
} catch {
return null;
}
}
if (cached) return cached;
if (!codeAgentAgentRunAdapterEnabled(options.env ?? process.env)) return null;
try {
const persisted = await loadPersistedAgentRunResult(traceId, options);
if (!persisted?.agentRun?.runId) return persisted;
const synced = await syncAgentRunChatResult({ traceId, currentResult: persisted, options, traceStore: options.traceStore ?? defaultCodeAgentTraceStore });
return synced.result ?? null;
} catch {
return null;
}
}
function textValue(value) {
return String(value ?? "").trim();
}
@@ -2605,58 +2495,21 @@ export async function handleCodeAgentTraceHttp(request, response, url, options)
sendJson(response, projectMismatch.statusCode, projectMismatch.body);
return;
}
const result = requestOptions.codeAgentChatResults?.get(traceId) ?? null;
if (result && !canAccessOwnedResult(result, requestOptions.actor)) {
sendJson(response, 403, {
error: {
code: "agent_session_owner_required",
message: "Only the session owner or admin can read this Code Agent trace"
}
});
const context = await measureCodeAgentHttpPhase(requestOptions, "trace_projection_read", () => readCodeAgentCompatProjection(traceId, requestOptions));
if (context.statusCode) {
sendJson(response, context.statusCode, context.body);
return;
}
let agentRunResult = result?.agentRun ? result : await measureCodeAgentHttpPhase(requestOptions, "trace_load_persisted_result", () => loadPersistedAgentRunResult(traceId, requestOptions));
let refreshError = null;
if (agentRunResult && !canAccessOwnedResult(agentRunResult, requestOptions.actor)) {
sendJson(response, 403, {
error: {
code: "agent_session_owner_required",
message: "Only the session owner or admin can read this Code Agent trace"
}
});
return;
}
if (agentRunResult?.agentRun) {
try {
const existingSnapshot = traceStore.snapshot(traceId);
const terminalSnapshotRefreshSatisfied = traceRefreshSatisfiedByTerminalSnapshot(agentRunResult, existingSnapshot);
const snapshot = terminalSnapshotRefreshSatisfied
? existingSnapshot
: await measureCodeAgentHttpPhase(requestOptions, "trace_trace_refresh", () => refreshAgentRunTrace({ traceId, result: agentRunResult, options: requestOptions, traceStore }));
agentRunResult = requestOptions.codeAgentChatResults?.get?.(traceId) ?? agentRunResult;
if (!terminalSnapshotRefreshSatisfied && traceNeedsCommandResultSync(agentRunResult, snapshot)) {
const synced = await measureCodeAgentHttpPhase(requestOptions, "trace_result_sync", () => syncAgentRunChatResult({ traceId, currentResult: agentRunResult, options: requestOptions, traceStore, appendResultEvent: false, refreshEvents: false, forceResultSync: true }));
agentRunResult = synced.result ?? agentRunResult;
}
if (isTraceCommandTerminalStatus(agentRunResult?.status)) {
scheduleCodeAgentTerminalTurnStatusEffects({ payload: agentRunResult, params: agentRunResult, options: requestOptions, preserveLastTraceId: true });
}
} catch (error) {
refreshError = error;
}
}
if (streamSegment === "stream") {
sendTraceSse(response, traceStore, traceId, requestOptions);
return;
}
const body = await measureCodeAgentHttpPhase(requestOptions, "trace_paginate", () => {
const snapshot = traceStore.snapshot(traceId);
return paginateTraceSnapshot(traceSnapshotWithTerminalEvidence(snapshot, agentRunResult, traceId, refreshError), pageOptions);
return paginateTraceSnapshot(context.trace, pageOptions);
});
sendJson(response, 200, body);
sendJson(response, context.found ? 200 : 404, codeAgentCompatProjectionPayload(body, context));
}
function codeAgentTracePageOptions(url) {
@@ -2708,44 +2561,10 @@ function nonNegativeInteger(value, fallback) {
return Number.isInteger(parsed) && parsed >= 0 ? parsed : fallback;
}
function traceNeedsCommandResultSync(result, snapshot) {
if (!result?.agentRun) return false;
const status = String(result.status ?? result.agentRun.status ?? result.agentRun.commandState ?? result.agentRun.terminalStatus ?? "running");
if (isTraceCommandTerminalStatus(status)) return true;
if (snapshotHasTerminalEvent(snapshot)) return true;
return false;
}
function isTraceCommandTerminalStatus(status) {
return CODE_AGENT_TERMINAL_STATUSES.has(String(status ?? "").trim().toLowerCase().replace(/_/gu, "-"));
}
function snapshotHasTerminalEvent(snapshot) {
const events = Array.isArray(snapshot?.events) ? snapshot.events : [];
return events.some((event) => event?.terminal === true || (event?.type === "result" && isTraceCommandTerminalStatus(event?.status)));
}
function traceRefreshSatisfiedByTerminalSnapshot(result, snapshot) {
if (!result?.agentRun || snapshot?.status === "missing") return false;
const status = String(result.status ?? result.agentRun.terminalStatus ?? result.agentRun.commandState ?? result.agentRun.status ?? "");
if (!isTraceCommandTerminalStatus(status)) return false;
const events = Array.isArray(snapshot?.events) ? snapshot.events : [];
if (events.length === 0) return false;
const commandId = textValue(result.agentRun.commandId);
const terminalEvents = events.filter((event) => {
if (!(event?.terminal === true || (event?.type === "result" && isTraceCommandTerminalStatus(event?.status)))) return false;
return !commandId || textValue(event?.commandId) === commandId;
});
if (terminalEvents.length === 0) return false;
const expectedLastSeq = Number(result.agentRun.lastSeq ?? result.traceSummary?.agentRun?.lastSeq ?? 0);
if (!Number.isFinite(expectedLastSeq) || expectedLastSeq <= 0) return true;
const observedLastSeq = Math.max(0, ...events
.filter((event) => !commandId || textValue(event?.commandId) === commandId)
.map((event) => Number(event?.sourceSeq ?? 0))
.filter(Number.isFinite));
return observedLastSeq === 0 || observedLastSeq >= expectedLastSeq;
}
function traceSnapshotWithTerminalEvidence(snapshot, result, traceId, refreshError = null) {
const evidence = agentRunTerminalTraceEvidence(result, traceId);
if (snapshot?.status !== "missing") return traceSnapshotWithAttachedTerminalEvidence(snapshot, evidence, refreshError);
internal/cloud/workbench-facts-store.ts
+1 -1
View File
@@ -6,7 +6,7 @@ import { defaultCodeAgentTraceStore } from "./code-agent-trace-store.ts";
import { safeConversationId, safeSessionId, safeTraceId } from "./server-http-utils.ts";
export function createWorkbenchFactsStore(options = {}, actor = null) {
const accessStore = options.accessController?.store ?? null;
const accessStore = options.accessController?.store ?? options.accessController ?? null;
const traceStore = options.traceStore ?? defaultCodeAgentTraceStore;
const runtimeStore = options.runtimeStore ?? null;
const results = options.codeAgentChatResults ?? null;
internal/cloud/workbench-read-model.ts
+4 -1
View File
@@ -26,12 +26,15 @@ export function projectionDiagnostics({ traceId, result = null, trace = null } =
const events = Array.isArray(trace?.events) ? trace.events : [];
const lastEvent = trace?.lastEvent ?? events.at(-1) ?? null;
const lastProjectedSeq = Number(lastEvent?.seq ?? trace?.lastProjectedSeq ?? 0);
const hasTraceProjection = Boolean(trace && trace.status !== "missing" && (events.length > 0 || lastEvent));
const sourceRunId = agentRun?.runId ?? lastEvent?.runId ?? null;
const sourceCommandId = agentRun?.commandId ?? lastEvent?.commandId ?? null;
const blocker = result?.blocker ?? result?.error ?? null;
const status = blocker
? "blocked"
: isTerminalStatus(result?.status ?? trace?.status)
: !hasTraceProjection && (result || agentRun)
? "projecting"
: isTerminalStatus(result?.status ?? trace?.status)
? "caught-up"
: result || trace?.status !== "missing"
? "projecting"
web/hwlab-cloud-web/scripts/check.ts
+5
View File
@@ -1,3 +1,6 @@
// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-010401 Web工作台 draft-2026-06-18-r1.
// Responsibility: Cloud Web source-shape checks for Workbench projection and UI integration invariants.
import assert from "node:assert/strict";
import fs from "node:fs";
import path from "node:path";
@@ -102,6 +105,8 @@ assertIncludes(workbenchStoreSource, "hydrateRealtimeGap", "Workbench realtime r
assert.doesNotMatch(workbenchStoreSource, /subscribeToTrace|TRACE_POLL_INTERVAL_MS/u, "Workbench store must not reintroduce active trace polling");
assertIncludes(appSource, "/v1/workbench/events", "Workbench realtime client must use the RESTful same-origin events endpoint");
assertIncludes(appSource, "/v1/workbench/traces/", "trace hydration must use Workbench read-model trace API");
assert.doesNotMatch(appSource, /\/v1\/agent\/(?:turns|traces|chat\/result)\//u, "Cloud Web must not call legacy Code Agent read-through turn/trace/result APIs");
assert.doesNotMatch(workbenchStoreSource, /api\.agent\.(?:getAgentTurn|getAgentTrace)/u, "Workbench store turn/trace gap fill must read through api.workbench");
assert.doesNotMatch(appSource, /\/v1\/agent\/chat\/trace\//u, "Cloud Web must not use the legacy action-style chat trace API");
assert.doesNotMatch(appSource, /HWLAB_CLOUD_WEB_EARLY_WORKSPACE_BOOTSTRAP/u, "early workspace bootstrap helper must not remain in Cloud Web app source");
assert.doesNotMatch(appSource, /\/auth\/workspace-bootstrap/u, "stale auth workspace bootstrap route must not remain in Cloud Web app source");
web/hwlab-cloud-web/src/api/agent.ts
+3 -108
View File
@@ -1,119 +1,14 @@
// SPEC: PJ2026-010403 API契约 draft-2026-06-17-r0; PJ2026-010401 Web工作台 draft-2026-06-17-r0.
// Responsibility: Code Agent HTTP client for send, steer, turn, trace, session and cancel APIs.
// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-18-p0-unique-projection; 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";
import type { AgentChatResponse, AgentChatResultResponse, ApiResult } from "@/types";
export interface AgentTraceRequestOptions {
sinceSeq?: number | null;
limit?: number | null;
}
function workbenchTracePath(traceId: string, options: AgentTraceRequestOptions = {}): string {
const params = new URLSearchParams();
if (Number.isFinite(options.sinceSeq)) params.set("sinceSeq", String(Math.max(0, Math.trunc(Number(options.sinceSeq)))));
if (Number.isFinite(options.limit)) params.set("limit", String(Math.max(1, Math.trunc(Number(options.limit)))));
const query = params.toString();
return `/v1/workbench/traces/${encodeURIComponent(traceId)}/events${query ? `?${query}` : ""}`;
}
import type { AgentChatResponse, ApiResult } from "@/types";
export const agentAPI = {
sendAgentMessage: (payload: Record<string, unknown>, timeoutMs: number, activityRef?: ApiRequestOptions["activityRef"]): Promise<ApiResult<AgentChatResponse>> => fetchJson("/v1/agent/chat", { method: "POST", body: JSON.stringify(payload), timeoutMs, timeoutName: "Code Agent", activityRef }),
steerAgentMessage: (payload: Record<string, unknown>, timeoutMs: number, activityRef?: ApiRequestOptions["activityRef"]): Promise<ApiResult<AgentChatResponse>> => fetchJson("/v1/agent/chat/steer", { method: "POST", body: JSON.stringify(payload), timeoutMs, timeoutName: "Code Agent steer", activityRef }),
getAgentTurn: async (traceId: string, timeoutMs = 8000, activityRef?: ApiRequestOptions["activityRef"]): Promise<ApiResult<AgentChatResultResponse>> => normalizeWorkbenchTurnResult(await fetchJson<Record<string, unknown>>(`/v1/workbench/turns/${encodeURIComponent(traceId)}`, { timeoutMs, timeoutName: "Code Agent turn", activityRef }), traceId),
getAgentTrace: async (traceId: string, timeoutMs = 8000, activityRef?: ApiRequestOptions["activityRef"], options?: AgentTraceRequestOptions): Promise<ApiResult<AgentChatResultResponse>> => normalizeWorkbenchTraceResult(await fetchJson<Record<string, unknown>>(workbenchTracePath(traceId, options), { timeoutMs, timeoutName: "Code Agent trace", activityRef }), traceId),
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" })
};
function normalizeWorkbenchTurnResult(result: ApiResult<Record<string, unknown>>, traceId: string): ApiResult<AgentChatResultResponse> {
if (!result.ok || !result.data) return result as ApiResult<AgentChatResultResponse>;
const turn = recordValue(result.data.turn) ?? result.data;
const trace = recordValue(turn.trace);
const status = canonicalWorkbenchStatus(textValue(turn.status) ?? textValue(result.data.status) ?? "unknown", textValue(trace?.status, result.data.traceStatus));
const terminal = turn.terminal === true || isTerminalStatus(status);
return {
...result,
data: {
...turn,
traceId: textValue(turn.traceId) ?? traceId,
status,
running: !terminal && (turn.running === true || isActiveStatus(status)),
terminal,
sessionId: textValue(turn.sessionId) ?? undefined,
threadId: textValue(turn.threadId) ?? undefined,
agentRun: recordValue(turn.agentRun) ?? undefined,
updatedAt: textValue(turn.updatedAt) ?? textValue(result.data.observedAt) ?? undefined
} as AgentChatResultResponse
};
}
function canonicalWorkbenchStatus(primary: string | null, traceStatus: string | null): string {
const primaryStatus = normalizeStatus(primary);
const trace = normalizeStatus(traceStatus);
if (trace && isTerminalStatus(trace) && (!primaryStatus || isActiveStatus(primaryStatus) || !isTerminalStatus(primaryStatus))) return trace;
return primaryStatus ?? trace ?? "unknown";
}
function normalizeStatus(value: unknown): string | null {
const text = textValue(value);
if (!text) return null;
const normalized = text.toLowerCase().replace(/_/gu, "-");
return normalized === "cancelled" ? "canceled" : normalized;
}
function isActiveStatus(value: unknown): boolean {
return ["accepted", "active", "busy", "creating", "pending", "processing", "running"].includes(normalizeStatus(value) ?? "");
}
function isTerminalStatus(value: unknown): boolean {
return ["blocked", "canceled", "completed", "failed", "stale", "thread-resume-failed", "timeout"].includes(normalizeStatus(value) ?? "");
}
function normalizeWorkbenchTraceResult(result: ApiResult<Record<string, unknown>>, traceId: string): ApiResult<AgentChatResultResponse> {
if (!result.ok || !result.data) return result as ApiResult<AgentChatResultResponse>;
const events = Array.isArray(result.data.events) ? result.data.events : [];
const nextSeq = Number(result.data.nextSeq ?? result.data.nextSinceSeq);
return {
...result,
data: {
...result.data,
traceId: textValue(result.data.traceId) ?? traceId,
status: textValue(result.data.traceStatus, result.data.status) ?? "unknown",
traceStatus: textValue(result.data.traceStatus, result.data.status) ?? undefined,
events,
traceEvents: events,
eventCount: Number.isFinite(Number(result.data.eventCount)) ? Number(result.data.eventCount) : events.length,
hasMore: result.data.hasMore === true,
fullTraceLoaded: result.data.hasMore !== true,
nextSinceSeq: Number.isFinite(nextSeq) ? Math.trunc(nextSeq) : null,
range: recordValue(result.data.range) ?? undefined,
runnerTrace: {
traceId: textValue(result.data.traceId) ?? traceId,
status: textValue(result.data.traceStatus, result.data.status) ?? undefined,
events,
eventCount: Number.isFinite(Number(result.data.eventCount)) ? Number(result.data.eventCount) : events.length,
hasMore: result.data.hasMore === true,
fullTraceLoaded: result.data.hasMore !== true,
nextSinceSeq: Number.isFinite(nextSeq) ? Math.trunc(nextSeq) : null,
range: recordValue(result.data.range) ?? undefined,
eventSource: "trace-api"
}
} as AgentChatResultResponse
};
}
function recordValue(value: unknown): Record<string, unknown> | null {
return value && typeof value === "object" ? value as Record<string, unknown> : null;
}
function textValue(...values: unknown[]): string | null {
for (const value of values) {
if (typeof value !== "string") continue;
const text = value.trim();
if (text) return text;
}
return null;
}
web/hwlab-cloud-web/src/api/workbench.ts
+117 -3
View File
@@ -1,5 +1,8 @@
import { fetchJson } from "./client";
import type { ApiResult, ChatMessage, WorkbenchSessionRecord } from "@/types";
// 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.
// Responsibility: Workbench read-model HTTP client for session, message, turn and trace projection APIs.
import { fetchJson, type ApiRequestOptions } from "./client";
import type { AgentChatResultResponse, ApiResult, ChatMessage, WorkbenchSessionRecord } from "@/types";
export interface WorkbenchSessionListResponse {
sessions?: WorkbenchSessionRecord[];
@@ -29,13 +32,20 @@ export interface SessionMessageOptions {
timeoutMs?: number | null;
}
export interface WorkbenchTraceRequestOptions {
sinceSeq?: number | null;
limit?: number | null;
}
const SESSION_LIST_TIMEOUT_MS = 30_000;
const SESSION_DETAIL_TIMEOUT_MS = 45_000;
export const workbenchAPI = {
sessions: (options: SessionListOptions = {}): Promise<ApiResult<WorkbenchSessionListResponse>> => fetchJson(sessionListPath(options), { timeoutMs: requestTimeoutMs(options.timeoutMs, SESSION_LIST_TIMEOUT_MS), timeoutName: "workbench sessions" }),
session: (sessionId: string, timeoutMs: number | null = null): Promise<ApiResult<WorkbenchSessionDetailResponse>> => fetchJson(`/v1/workbench/sessions/${encodeURIComponent(sessionId)}`, { timeoutMs: requestTimeoutMs(timeoutMs, SESSION_DETAIL_TIMEOUT_MS), timeoutName: "workbench session" }),
sessionMessages: (sessionId: string, options: SessionMessageOptions = {}): Promise<ApiResult<WorkbenchMessagePageResponse>> => fetchJson(sessionMessagesPath(sessionId, options), { timeoutMs: requestTimeoutMs(options.timeoutMs, SESSION_DETAIL_TIMEOUT_MS), timeoutName: "workbench session messages" })
sessionMessages: (sessionId: string, options: SessionMessageOptions = {}): Promise<ApiResult<WorkbenchMessagePageResponse>> => fetchJson(sessionMessagesPath(sessionId, options), { timeoutMs: requestTimeoutMs(options.timeoutMs, SESSION_DETAIL_TIMEOUT_MS), timeoutName: "workbench session messages" }),
turn: async (traceId: string, timeoutMs = 8000, activityRef?: ApiRequestOptions["activityRef"]): Promise<ApiResult<AgentChatResultResponse>> => normalizeWorkbenchTurnResult(await fetchJson<Record<string, unknown>>(`/v1/workbench/turns/${encodeURIComponent(traceId)}`, { timeoutMs, timeoutName: "workbench turn", activityRef }), traceId),
traceEvents: async (traceId: string, timeoutMs = 8000, activityRef?: ApiRequestOptions["activityRef"], options?: WorkbenchTraceRequestOptions): Promise<ApiResult<AgentChatResultResponse>> => normalizeWorkbenchTraceResult(await fetchJson<Record<string, unknown>>(workbenchTracePath(traceId, options), { timeoutMs, timeoutName: "workbench trace", activityRef }), traceId)
};
function sessionListPath(options: SessionListOptions): string {
@@ -59,3 +69,107 @@ function requestTimeoutMs(value: number | null | undefined, fallback: number): n
const timeout = Number(value);
return Number.isFinite(timeout) && timeout > 0 ? Math.trunc(timeout) : fallback;
}
function workbenchTracePath(traceId: string, options: WorkbenchTraceRequestOptions = {}): string {
const params = new URLSearchParams();
if (Number.isFinite(options.sinceSeq)) params.set("sinceSeq", String(Math.max(0, Math.trunc(Number(options.sinceSeq)))));
if (Number.isFinite(options.limit)) params.set("limit", String(Math.max(1, Math.trunc(Number(options.limit)))));
const query = params.toString();
return `/v1/workbench/traces/${encodeURIComponent(traceId)}/events${query ? `?${query}` : ""}`;
}
function normalizeWorkbenchTurnResult(result: ApiResult<Record<string, unknown>>, traceId: string): ApiResult<AgentChatResultResponse> {
if (!result.ok || !result.data) return result as ApiResult<AgentChatResultResponse>;
const turn = recordValue(result.data.turn) ?? result.data;
const trace = recordValue(turn.trace);
const status = canonicalWorkbenchStatus(textValue(turn.status) ?? textValue(result.data.status) ?? "unknown", textValue(trace?.status, result.data.traceStatus));
const terminal = turn.terminal === true || isTerminalStatus(status);
return {
...result,
data: {
...turn,
projection: recordValue(result.data.projection) ?? undefined,
projectionStatus: textValue(result.data.projectionStatus) ?? undefined,
lastProjectedSeq: Number.isFinite(Number(result.data.lastProjectedSeq)) ? Number(result.data.lastProjectedSeq) : undefined,
traceId: textValue(turn.traceId) ?? traceId,
status,
running: !terminal && (turn.running === true || isActiveStatus(status)),
terminal,
sessionId: textValue(turn.sessionId) ?? undefined,
threadId: textValue(turn.threadId) ?? undefined,
agentRun: recordValue(turn.agentRun) ?? undefined,
updatedAt: textValue(turn.updatedAt) ?? textValue(result.data.observedAt) ?? undefined
} as AgentChatResultResponse
};
}
function normalizeWorkbenchTraceResult(result: ApiResult<Record<string, unknown>>, traceId: string): ApiResult<AgentChatResultResponse> {
if (!result.ok || !result.data) return result as ApiResult<AgentChatResultResponse>;
const events = Array.isArray(result.data.events) ? result.data.events : [];
const nextSeq = Number(result.data.nextSeq ?? result.data.nextSinceSeq);
const traceStatus = textValue(result.data.traceStatus, result.data.status) ?? "unknown";
const eventCount = Number.isFinite(Number(result.data.eventCount)) ? Number(result.data.eventCount) : events.length;
const normalizedNextSeq = Number.isFinite(nextSeq) ? Math.trunc(nextSeq) : null;
return {
...result,
data: {
...result.data,
traceId: textValue(result.data.traceId) ?? traceId,
status: traceStatus,
traceStatus,
events,
traceEvents: events,
eventCount,
hasMore: result.data.hasMore === true,
fullTraceLoaded: result.data.hasMore !== true,
nextSinceSeq: normalizedNextSeq,
range: recordValue(result.data.range) ?? undefined,
runnerTrace: {
traceId: textValue(result.data.traceId) ?? traceId,
status: traceStatus,
events,
eventCount,
hasMore: result.data.hasMore === true,
fullTraceLoaded: result.data.hasMore !== true,
nextSinceSeq: normalizedNextSeq,
range: recordValue(result.data.range) ?? undefined,
eventSource: "trace-api"
}
} as AgentChatResultResponse
};
}
function canonicalWorkbenchStatus(primary: string | null, traceStatus: string | null): string {
const primaryStatus = normalizeStatus(primary);
const trace = normalizeStatus(traceStatus);
if (trace && isTerminalStatus(trace) && (!primaryStatus || isActiveStatus(primaryStatus) || !isTerminalStatus(primaryStatus))) return trace;
return primaryStatus ?? trace ?? "unknown";
}
function normalizeStatus(value: unknown): string | null {
const text = textValue(value);
if (!text) return null;
const normalized = text.toLowerCase().replace(/_/gu, "-");
return normalized === "cancelled" ? "canceled" : normalized;
}
function isActiveStatus(value: unknown): boolean {
return ["accepted", "active", "busy", "creating", "pending", "processing", "running"].includes(normalizeStatus(value) ?? "");
}
function isTerminalStatus(value: unknown): boolean {
return ["blocked", "canceled", "completed", "failed", "stale", "thread-resume-failed", "timeout"].includes(normalizeStatus(value) ?? "");
}
function recordValue(value: unknown): Record<string, unknown> | null {
return value && typeof value === "object" ? value as Record<string, unknown> : null;
}
function textValue(...values: unknown[]): string | null {
for (const value of values) {
if (typeof value !== "string") continue;
const text = value.trim();
if (text) return text;
}
return null;
}
web/hwlab-cloud-web/src/composables/useTraceSubscription.ts
+6 -3
View File
@@ -1,4 +1,7 @@
import { agentAPI, type ActivityRefSource } from "@/api";
// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-010401 Web工作台 draft-2026-06-18-r1.
// Responsibility: Trace snapshot helpers and legacy subscription adapter backed by Workbench read-model APIs.
import { workbenchAPI, type ActivityRefSource } from "@/api";
import type { AgentChatResponse, AgentChatResultResponse, AgentRunProvenance, ChatMessage, TraceEvent } from "@/types";
import { firstNonEmptyString } from "@/utils";
@@ -188,7 +191,7 @@ export async function subscribeToTrace(config: TraceSubscriptionConfig): Promise
const fetchLiveTracePages = async (): Promise<TraceSnapshot | null> => {
for (let page = 0; page < TRACE_LIVE_MAX_PAGES_PER_POLL; page += 1) {
const previousSinceSeq = traceSinceSeq;
const tracePolled = await agentAPI.getAgentTrace(traceId, inactivityTimeoutMs, activityRef, { sinceSeq: traceSinceSeq, limit: TRACE_LIVE_PAGE_LIMIT });
const tracePolled = await workbenchAPI.traceEvents(traceId, inactivityTimeoutMs, activityRef, { sinceSeq: traceSinceSeq, limit: TRACE_LIVE_PAGE_LIMIT });
if (signal.aborted) return accumulatedTrace;
if (!tracePolled.ok || !tracePolled.data) return accumulatedTrace;
onActivity();
@@ -206,7 +209,7 @@ export async function subscribeToTrace(config: TraceSubscriptionConfig): Promise
await sleep(TRACE_POLL_INTERVAL_MS);
if (signal.aborted) return;
const turnPolled = await agentAPI.getAgentTurn(traceId, inactivityTimeoutMs, activityRef);
const turnPolled = await workbenchAPI.turn(traceId, inactivityTimeoutMs, activityRef);
if (signal.aborted) return;
if (turnPolled.ok && turnPolled.data) {
web/hwlab-cloud-web/src/stores/workbench.ts
+5 -5
View File
@@ -248,7 +248,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
async function refreshTurnStatusByTraceId(traceId: string | null | undefined): Promise<void> {
const id = firstNonEmptyString(traceId);
if (!id) return;
const response = await api.agent.getAgentTurn(id, 8000, () => activityRef.value);
const response = await api.workbench.turn(id, 8000, () => activityRef.value);
if (response.ok && response.data) {
applyTurnStatusSnapshot(id, response.data);
if (response.data.terminal === true || isTerminalMessageStatus(response.data.status)) completeTrace(id, response.data);
@@ -414,7 +414,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
async function fetchTraceHydrationPage(traceId: string, sinceSeq: number): Promise<ApiResult<AgentChatResultResponse>> {
let lastResult: ApiResult<AgentChatResultResponse> | null = null;
for (let attempt = 0; attempt < TRACE_HYDRATION_MAX_ATTEMPTS; attempt += 1) {
const result = await api.agent.getAgentTrace(traceId, codeAgentTimeoutMs.value, () => activityRef.value, { sinceSeq, limit: TRACE_HYDRATION_PAGE_LIMIT });
const result = await api.workbench.traceEvents(traceId, codeAgentTimeoutMs.value, () => activityRef.value, { sinceSeq, limit: TRACE_HYDRATION_PAGE_LIMIT });
if (result.ok && result.data) return result;
lastResult = result;
if (attempt < TRACE_HYDRATION_MAX_ATTEMPTS - 1) await delayTraceHydrationRetry(TRACE_HYDRATION_RETRY_DELAY_MS * (attempt + 1));
@@ -540,7 +540,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
}
async function validateAndReattachTrace(traceId: string): Promise<void> {
const result = await api.agent.getAgentTurn(traceId, 8000, () => activityRef.value);
const result = await api.workbench.turn(traceId, 8000, () => activityRef.value);
if (!result.ok || !result.data) {
await clearActiveTrace(traceId, result.status === 404 ? "reattach-result-not-found" : "reattach-result-unavailable");
return;
@@ -629,7 +629,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
if (terminalRealtimeRefreshInFlight.has(traceId)) return;
terminalRealtimeRefreshInFlight.add(traceId);
try {
const result = await api.agent.getAgentTurn(traceId, 8000, () => activityRef.value);
const result = await api.workbench.turn(traceId, 8000, () => activityRef.value);
if (!result.ok || !result.data) return;
applyTurnStatusSnapshot(traceId, result.data);
if (result.data.terminal === true || isTerminalMessageStatus(result.data.status)) completeTrace(traceId, result.data);
@@ -760,7 +760,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
await Promise.all(targets.map(async (message) => {
const traceId = firstNonEmptyString(message.traceId, message.runnerTrace?.traceId);
if (!traceId) return;
const result = await api.agent.getAgentTurn(traceId, 8000, () => activityRef.value);
const result = await api.workbench.turn(traceId, 8000, () => activityRef.value);
if (!result.ok || !result.data) return;
applyTurnStatusSnapshot(traceId, result.data);
}));