feat(workbench): add read-only API read model

This commit is contained in:
lyon
2026-06-17 11:12:04 +08:00
parent a3132be5af
commit bc9ee41d83
5 changed files with 720 additions and 43 deletions
+31 -38
View File
@@ -438,7 +438,7 @@ test("workbench workspace permits continuation when stale active trace has idle
}
});
test("workbench workspace status clears completed AgentRun active trace on read", async () => {
test("workbench workspace read does not clear completed AgentRun active trace", async () => {
const agentRunCalls = [];
const agentSessions = new Map();
const agentRunServer = createServer(async (request, response) => {
@@ -557,23 +557,21 @@ test("workbench workspace status clears completed AgentRun active trace on read"
const restored = await getJson(port, "/v1/workbench/workspace?projectId=prj_hwpod_workbench", aliceLogin.cookie);
assert.equal(restored.status, 200);
assert.equal(restored.body.workspace.activeTraceId, null);
assert.equal(restored.body.workspace.workspace.lastTraceId, "trc_issue664_status_done");
assert.equal(restored.body.workspace.workspace.sessionStatus, "idle");
assert.equal(restored.body.workspace.selectedConversation.status, "idle");
assert.equal(restored.body.workspace.selectedConversation.lastTraceId, "trc_issue664_status_done");
assert.equal(restored.body.workspace.activeTraceId, "trc_issue664_status_done");
assert.equal(restored.body.workspace.workspace.sessionStatus, "running");
assert.equal(restored.body.workspace.selectedConversation.lastTraceId, "trc_issue664_previous");
assert.equal(restored.body.workspace.selectedConversation.messages.length, 1);
assert.equal(restored.body.workspace.selectedConversation.messages[0].status, "idle");
assert.equal(restored.body.workspace.selectedConversation.messages[0].text, "previous turn completed");
assert.equal(restored.body.workspace.revision, 3);
assert.ok(agentRunCalls.some((call) => call.path === "/api/v1/runs/run_workspace_status/commands/cmd_workspace_status/result"));
assert.equal(restored.body.workspace.selectedConversation.messages[0].status, "running");
assert.equal(restored.body.workspace.selectedConversation.messages[0].text, "previous turn still running");
assert.equal(restored.body.workspace.revision, 2);
assert.equal(agentRunCalls.length, 0);
} 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("workbench workspace terminal status sync preserves a newer selected conversation", async () => {
test("workbench workspace read preserves a newer selected conversation without terminal sync", async () => {
const agentRunCalls = [];
const agentSessions = new Map();
const agentRunServer = createServer(async (request, response) => {
@@ -720,7 +718,7 @@ test("workbench workspace terminal status sync preserves a newer selected conver
assert.equal(restored.body.workspace.workspace.selectedConversationId, "cnv_issue808_new");
assert.equal(restored.body.workspace.workspace.selectedAgentSessionId, "ses_issue808_new");
assert.equal(restored.body.workspace.workspace.lastTraceId, "trc_issue808_old_done");
assert.ok(agentRunCalls.some((call) => call.path === "/api/v1/runs/run_issue808_old/commands/cmd_issue808_old/result"));
assert.equal(agentRunCalls.length, 0);
const selectedOld = await postJson(port, `/v1/workbench/workspace/${workspace.body.workspace.workspaceId}/select-conversation`, {
projectId: "prj_hwpod_workbench",
@@ -736,11 +734,11 @@ test("workbench workspace terminal status sync preserves a newer selected conver
assert.equal(repaired.body.workspace.activeTraceId, null);
assert.equal(repaired.body.workspace.selectedConversationId, "cnv_issue808_old");
assert.equal(repaired.body.workspace.selectedAgentSessionId, "ses_issue808_old");
assert.equal(repaired.body.workspace.selectedConversation.status, "idle");
assert.equal(repaired.body.workspace.selectedConversation.status, "running");
assert.equal(repaired.body.workspace.selectedConversation.lastTraceId, "trc_issue808_old_done");
assert.equal(repaired.body.workspace.selectedConversation.messages[0].status, "idle");
assert.equal(repaired.body.workspace.selectedConversation.messages[0].text, "old turn completed");
assert.ok(agentRunCalls.some((call) => call.path === "/api/v1/runs/run_issue808_old/commands/cmd_issue808_old/result"));
assert.equal(repaired.body.workspace.selectedConversation.messages[0].status, "running");
assert.equal(repaired.body.workspace.selectedConversation.messages[0].text, "old turn running");
assert.equal(agentRunCalls.length, 0);
const repairedOld = await getJson(port, "/v1/agent/conversations/cnv_issue808_old", aliceLogin.cookie);
assert.equal(repairedOld.status, 200);
@@ -870,7 +868,7 @@ test("manual Code Agent session select restores running active trace", async ()
}
});
test("workbench workspace status repairs terminal selected conversation after active trace was cleared", async () => {
test("workbench workspace read does not repair terminal selected conversation after active trace was cleared", async () => {
const agentRunCalls = [];
const agentSessions = new Map();
const agentRunServer = createServer(async (request, response) => {
@@ -984,22 +982,21 @@ test("workbench workspace status repairs terminal selected conversation after ac
}, aliceLogin.cookie);
assert.equal(update.status, 200);
assert.equal(update.body.workspace.activeTraceId, null);
assert.equal(update.body.workspace.selectedConversation.status, "idle");
assert.equal(update.body.workspace.selectedConversation.messages.length, 1);
assert.equal(update.body.workspace.selectedConversation.messages[0].status, "idle");
assert.equal(update.body.workspace.selectedConversation.messages[0].text, "repair completed");
assert.equal(update.body.workspace.workspace.lastTraceId, "trc_issue664_repair_done");
assert.equal(update.body.workspace.workspace.sessionStatus, "idle");
assert.equal(update.body.workspace.selectedConversation.status, "running");
assert.equal(update.body.workspace.selectedConversation.messages.some((message) => message.text === "repair completed"), false);
const restored = await getJson(port, "/v1/workbench/workspace?projectId=prj_hwpod_workbench", aliceLogin.cookie);
assert.equal(restored.status, 200);
assert.equal(restored.body.workspace.activeTraceId, null);
assert.equal(restored.body.workspace.workspace.lastTraceId, "trc_issue664_repair_done");
assert.equal(restored.body.workspace.workspace.sessionStatus, "idle");
assert.equal(restored.body.workspace.selectedConversation.status, "idle");
assert.equal(restored.body.workspace.selectedConversation.status, "running");
assert.equal(restored.body.workspace.selectedConversation.lastTraceId, "trc_issue664_repair_done");
assert.equal(restored.body.workspace.selectedConversation.messages.length, 1);
assert.equal(restored.body.workspace.selectedConversation.messages[0].status, "idle");
assert.equal(restored.body.workspace.selectedConversation.messages[0].text, "repair completed");
assert.ok(agentRunCalls.some((call) => call.path === "/api/v1/runs/run_workspace_repair/commands/cmd_workspace_repair/result"));
assert.equal(restored.body.workspace.selectedConversation.messages.some((message) => message.text === "repair completed"), false);
assert.equal(restored.body.workspace.revision, 2);
assert.equal(agentRunCalls.length, 0);
} 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())));
@@ -1317,7 +1314,7 @@ test("cloud api repairs conversation snapshot when persisted result advances las
}
});
test("cloud api terminal workspace sync preserves saved messages and stores runner trace", async () => {
test("cloud api workspace read preserves saved messages without terminal sync", async () => {
const codeAgentChatResults = new Map();
const traceId = "trc_issue853_fast_fail_trace";
const conversationId = "cnv_issue853_fast_fail";
@@ -1420,18 +1417,15 @@ test("cloud api terminal workspace sync preserves saved messages and stores runn
const restored = await getJson(port, "/v1/workbench/workspace?projectId=prj_hwpod_workbench", aliceLogin.cookie);
assert.equal(restored.status, 200);
assert.equal(restored.body.workspace.activeTraceId, null);
assert.equal(restored.body.workspace.activeTraceId, traceId);
const messages = restored.body.workspace.selectedConversation.messages;
assert.equal(messages.length, 2);
assert.equal(messages[0].role, "user");
assert.equal(messages[0].text, "请执行 hwpod inspect");
assert.equal(messages[1].role, "agent");
assert.equal(messages[1].status, "failed");
assert.equal(messages[1].text, "HyueAPI 403 INSUFFICIENT_BALANCE");
assert.equal(messages[1].runnerTrace.traceId, traceId);
assert.equal(messages[1].runnerTrace.eventCount, 35);
assert.equal(messages[1].runnerTrace.eventsCompacted, true);
assert.equal(messages[1].runnerTrace.fullTraceLoaded, false);
assert.equal(messages[1].status, "running");
assert.equal(messages[1].text, "Code Agent 仍在处理,可以继续 steer 或等待 trace 完成。");
assert.equal(Object.hasOwn(messages[1], "runnerTrace"), false);
assert.equal(restored.body.workspace.selectedConversation.messageCount, 2);
assert.equal(restored.body.workspace.selectedConversation.firstUserMessagePreview, "请执行 hwpod inspect");
} finally {
@@ -1441,7 +1435,7 @@ test("cloud api terminal workspace sync preserves saved messages and stores runn
}
});
test("cloud api terminal workspace sync rebuilds missing user message from saved preview", async () => {
test("cloud api workspace read preserves preview-only conversation without terminal sync", async () => {
const codeAgentChatResults = new Map();
const traceId = "trc_issue853_preview_only_trace";
const conversationId = "cnv_issue853_preview_only";
@@ -1525,7 +1519,7 @@ test("cloud api terminal workspace sync rebuilds missing user message from saved
const restored = await getJson(port, "/v1/workbench/workspace?projectId=prj_hwpod_workbench", aliceLogin.cookie);
assert.equal(restored.status, 200);
assert.equal(restored.body.workspace.activeTraceId, null);
assert.equal(restored.body.workspace.activeTraceId, traceId);
const messages = restored.body.workspace.selectedConversation.messages;
assert.equal(messages.length, 2);
assert.equal(messages[0].role, "user");
@@ -1534,8 +1528,7 @@ test("cloud api terminal workspace sync rebuilds missing user message from saved
assert.equal(messages[1].role, "agent");
assert.equal(messages[1].status, "failed");
assert.equal(messages[1].text, "HyueAPI 403 INSUFFICIENT_BALANCE");
assert.equal(messages[1].runnerTrace.traceId, traceId);
assert.equal(messages[1].runnerTrace.eventCount, 35);
assert.equal(Object.hasOwn(messages[1], "runnerTrace"), false);
assert.equal(restored.body.workspace.selectedConversation.messageCount, 2);
assert.equal(restored.body.workspace.selectedConversation.firstUserMessagePreview, userText);
} finally {
+7 -5
View File
@@ -1,3 +1,7 @@
/*
* SPEC: PJ2026-010403 API契约 draft-2026-06-17-r0; PJ2026-010401 Web工作台 draft-2026-06-17-r0
* 职责: 访问控制、用户/session/workspace mutation 与 legacy compat read。GET workspace projection 不得隐式执行 terminal repair。
*/
import { createHash, createPublicKey, randomBytes, randomUUID, verify as verifySignature } from "node:crypto";
import { CLOUD_API_SERVICE_ID } from "../audit/index.mjs";
@@ -1180,7 +1184,6 @@ class AccessController {
projectId,
now: this.now()
});
workspace = await this.syncTerminalWorkbenchWorkspace(workspace, auth.actor);
return sendJson(response, 200, await this.workbenchWorkspacePayload(workspace, auth.actor));
}
@@ -1206,7 +1209,7 @@ class AccessController {
}
const selectedConversationId = textOr(body.selectedConversationId ?? body.conversationId, current.selectedConversationId ?? "");
if (selectedConversationId) {
const visible = await this.visibleConversationForActor(auth.actor, selectedConversationId, body.projectId ?? current.projectId);
const visible = await this.visibleConversationForActor(auth.actor, selectedConversationId, body.projectId ?? current.projectId, { skipTerminalRepair: true });
if (!visible) return sendJson(response, 403, errorPayload("workspace_conversation_forbidden", "Selected conversation is not visible to the current actor", 403));
}
const workspace = await this.store.updateWorkspace?.({
@@ -1238,7 +1241,7 @@ class AccessController {
const body = await jsonBody(request);
const conversationId = textOr(body.conversationId, "") || `cnv_${randomUUID()}`;
if (!safeConversationIdLocal(conversationId)) return sendJson(response, 400, errorPayload("invalid_conversation_id", "conversationId must start with cnv_", 400));
const existing = await this.visibleConversationForActor(auth.actor, conversationId, body.projectId ?? current.projectId);
const existing = await this.visibleConversationForActor(auth.actor, conversationId, body.projectId ?? current.projectId, { skipTerminalRepair: true });
if (!existing && body.create !== true) return sendJson(response, 404, errorPayload("agent_conversation_not_found", "Agent conversation is not visible to the current actor", 404));
const sessionId = safeAgentSessionId(body.sessionId ?? existing?.sessionId) || `ses_${conversationId.slice(4)}`;
if (!existing) {
@@ -1303,7 +1306,6 @@ class AccessController {
if (!auth.ok) return sendJson(response, auth.status, auth);
let workspace = await this.store.getWorkspaceForUser?.({ workspaceId, ownerUserId: auth.actor.id, actorRole: auth.actor.role });
if (!workspace) return sendJson(response, 404, errorPayload("workbench_workspace_not_found", "Workbench workspace is not visible to the current actor", 404));
workspace = await this.syncTerminalWorkbenchWorkspace(workspace, auth.actor);
const afterRevision = Number.parseInt(String(url.searchParams.get("afterRevision") ?? ""), 10) || 0;
const changed = workspace.revision > afterRevision;
return sendJson(response, 200, {
@@ -1420,7 +1422,7 @@ class AccessController {
async publicWorkbenchWorkspace(workspace, actor) {
const conversation = workspace?.selectedConversationId
? await this.visibleConversationForActor(actor, workspace.selectedConversationId, workspace.projectId)
? await this.visibleConversationForActor(actor, workspace.selectedConversationId, workspace.projectId, { skipTerminalRepair: true })
: null;
return publicWorkbenchWorkspace(workspace, { conversation });
}
@@ -0,0 +1,175 @@
import assert from "node:assert/strict";
import { test } from "bun:test";
import { createCloudApiServer } from "./server.ts";
import { createCodeAgentTraceStore } from "./code-agent-trace-store.ts";
import { createCodeAgentChatResultStore } from "./server-code-agent-http.ts";
const ACTOR = { id: "usr_workbench_reader", username: "reader", displayName: "Reader", role: "user", status: "active" };
test("workbench read model exposes session, messages, turn, and trace without write repair", async () => {
const traceStore = createCodeAgentTraceStore();
const results = createCodeAgentChatResultStore();
const writes = { recordAgentSessionOwner: 0, updateWorkspace: 0 };
const traceId = "trc_workbench_read_model";
const session = {
id: "ses_workbench_read_model",
projectId: "prj_hwpod_workbench",
agentId: "hwlab-code-agent",
status: "completed",
startedAt: "2026-06-17T00:00:00.000Z",
endedAt: "2026-06-17T00:00:03.000Z",
ownerUserId: ACTOR.id,
conversationId: "cnv_workbench_read_model",
threadId: "thread-workbench-read-model",
lastTraceId: traceId,
updatedAt: "2026-06-17T00:00:03.000Z",
session: {
sessionStatus: "completed",
providerProfile: "codex-api",
lastTraceId: traceId,
messages: [
{ role: "user", text: "ping", traceId, createdAt: "2026-06-17T00:00:00.000Z" },
{ role: "assistant", text: "pong", traceId, createdAt: "2026-06-17T00:00:03.000Z" }
],
valuesRedacted: true,
secretMaterialStored: false
}
};
traceStore.append(traceId, { type: "request", status: "accepted", label: "request:accepted" });
traceStore.append(traceId, { type: "result", status: "completed", label: "result:completed", terminal: true });
results.set(traceId, {
status: "completed",
traceId,
ownerUserId: ACTOR.id,
conversationId: session.conversationId,
sessionId: session.id,
threadId: session.threadId,
finalResponse: "pong",
agentRun: { runId: "run_workbench_read_model", commandId: "cmd_workbench_read_model", status: "completed" }
});
const accessController = {
store: {
async listAgentSessionsForUser() { return [session]; },
async getAgentSession(sessionId) { return sessionId === session.id ? session : null; },
async getAgentSessionByTraceId(requestTraceId) { return requestTraceId === traceId ? session : null; },
async recordAgentSessionOwner() { writes.recordAgentSessionOwner += 1; throw new Error("read model must not write session owner"); },
async updateWorkspace() { writes.updateWorkspace += 1; throw new Error("read model must not update workspace"); }
},
async ensureBootstrap() {},
async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; }
};
const server = createCloudApiServer({ accessController, traceStore, codeAgentChatResults: results });
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
try {
const { port } = server.address();
const sessions = await getJson(port, "/v1/workbench/sessions?projectId=prj_hwpod_workbench");
assert.equal(sessions.status, 200);
assert.equal(sessions.body.contractVersion, "workbench-sessions-v1");
assert.equal(sessions.body.sessions[0].sessionId, session.id);
assert.equal(sessions.body.sessions[0].turnSummary.traceId, traceId);
const detail = await getJson(port, `/v1/workbench/sessions/${encodeURIComponent(session.id)}`);
assert.equal(detail.status, 200);
assert.equal(detail.body.session.conversationId, session.conversationId);
assert.equal(detail.body.session.messageCount, 2);
const messages = await getJson(port, `/v1/workbench/sessions/${encodeURIComponent(session.id)}/messages?limit=1`);
assert.equal(messages.status, 200);
assert.equal(messages.body.contractVersion, "workbench-message-page-v1");
assert.equal(messages.body.messages.length, 1);
assert.equal(messages.body.messages[0].role, "user");
assert.equal(messages.body.hasMore, true);
const turn = await getJson(port, `/v1/workbench/turns/${encodeURIComponent(traceId)}`);
assert.equal(turn.status, 200);
assert.equal(turn.body.contractVersion, "workbench-turn-snapshot-v1");
assert.equal(turn.body.turn.traceId, traceId);
assert.equal(turn.body.turn.userMessageId, messages.body.messages[0].messageId);
assert.equal(turn.body.turn.terminal, true);
const trace = await getJson(port, `/v1/workbench/traces/${encodeURIComponent(traceId)}/events?limit=1`);
assert.equal(trace.status, 200);
assert.equal(trace.body.contractVersion, "workbench-trace-events-v1");
assert.equal(trace.body.events.length, 1);
assert.equal(trace.body.hasMore, true);
assert.deepEqual(writes, { recordAgentSessionOwner: 0, updateWorkspace: 0 });
} finally {
await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
}
});
test("workbench read model does not expose trace-only memory without visible session or result owner", async () => {
const traceStore = createCodeAgentTraceStore();
const traceId = "trc_workbench_trace_only";
traceStore.append(traceId, { type: "request", status: "accepted", label: "request:accepted" });
const accessController = {
store: {
async getAgentSessionByTraceId() { return null; }
},
async ensureBootstrap() {},
async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; }
};
const server = createCloudApiServer({ accessController, traceStore, codeAgentChatResults: createCodeAgentChatResultStore() });
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
try {
const { port } = server.address();
const turn = await getJson(port, `/v1/workbench/turns/${encodeURIComponent(traceId)}`);
assert.equal(turn.status, 404);
assert.equal(turn.body.error.code, "workbench_turn_not_found");
const trace = await getJson(port, `/v1/workbench/traces/${encodeURIComponent(traceId)}/events`);
assert.equal(trace.status, 404);
assert.equal(trace.body.error.code, "workbench_trace_not_found");
} finally {
await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
}
});
test("workbench read model treats active session state as running", async () => {
const traceId = "trc_workbench_active_state";
const session = {
id: "ses_workbench_active_state",
projectId: "prj_hwpod_workbench",
agentId: "hwlab-code-agent",
status: "active",
ownerUserId: ACTOR.id,
conversationId: "cnv_workbench_active_state",
threadId: "thread-workbench-active-state",
lastTraceId: traceId,
updatedAt: "2026-06-17T00:01:00.000Z",
session: { sessionStatus: "active", lastTraceId: traceId, messages: [{ role: "user", text: "still running", traceId }] }
};
const accessController = {
store: {
async listAgentSessionsForUser() { return [session]; }
},
async ensureBootstrap() {},
async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; }
};
const server = createCloudApiServer({ accessController, codeAgentChatResults: createCodeAgentChatResultStore() });
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
try {
const { port } = server.address();
const sessions = await getJson(port, "/v1/workbench/sessions?projectId=prj_hwpod_workbench");
assert.equal(sessions.status, 200);
assert.equal(sessions.body.sessions[0].status, "active");
assert.equal(sessions.body.sessions[0].running, true);
assert.equal(sessions.body.sessions[0].terminal, false);
} 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 {
status: response.status,
body: await response.json()
};
}
+497
View File
@@ -0,0 +1,497 @@
/*
* SPEC: PJ2026-010403 API契约 draft-2026-06-17-r0; PJ2026-010401 Web工作台 draft-2026-06-17-r0
* 职责: Workbench REST read model session/message/turn/trace facts AgentRun syncbilling finalize workspace repair
*/
import { createHash } from "node:crypto";
import { defaultCodeAgentTraceStore } from "./code-agent-trace-store.ts";
import {
parsePositiveInteger,
safeConversationId,
safeOpaqueId,
safeSessionId,
safeTraceId,
sendJson
} from "./server-http-utils.ts";
const DEFAULT_PROJECT_ID = "prj_hwpod_workbench";
const DEFAULT_PAGE_LIMIT = 50;
const MAX_PAGE_LIMIT = 100;
const TERMINAL_STATUSES = new Set(["completed", "failed", "blocked", "timeout", "cancelled", "canceled", "idle"]);
const RUNNING_STATUSES = new Set(["running", "pending", "queued", "accepted", "dispatching", "streaming", "active"]);
export async function handleWorkbenchReadModelHttp(request, response, url, options = {}) {
const auth = await authenticateWorkbenchRead(request, response, options);
if (!auth) return;
if (url.pathname === "/v1/workbench/sessions") {
if (request.method !== "GET") return methodNotAllowed(response, "GET");
await handleWorkbenchSessionList(response, url, options, auth.actor);
return;
}
const sessionMessagesMatch = url.pathname.match(/^\/v1\/workbench\/sessions\/([^/]+)\/messages$/u);
if (sessionMessagesMatch) {
if (request.method !== "GET") return methodNotAllowed(response, "GET");
await handleWorkbenchMessagePage(response, url, options, auth.actor, decodeURIComponent(sessionMessagesMatch[1]));
return;
}
const sessionMatch = url.pathname.match(/^\/v1\/workbench\/sessions\/([^/]+)$/u);
if (sessionMatch) {
if (request.method !== "GET") return methodNotAllowed(response, "GET");
await handleWorkbenchSessionDetail(response, options, auth.actor, decodeURIComponent(sessionMatch[1]));
return;
}
const turnMatch = url.pathname.match(/^\/v1\/workbench\/turns\/([^/]+)$/u);
if (turnMatch) {
if (request.method !== "GET") return methodNotAllowed(response, "GET");
await handleWorkbenchTurnSnapshot(response, url, options, auth.actor, decodeURIComponent(turnMatch[1]));
return;
}
const traceEventsMatch = url.pathname.match(/^\/v1\/workbench\/traces\/([^/]+)\/events$/u);
if (traceEventsMatch) {
if (request.method !== "GET") return methodNotAllowed(response, "GET");
await handleWorkbenchTraceEventPage(response, url, options, auth.actor, decodeURIComponent(traceEventsMatch[1]));
return;
}
sendJson(response, 404, workbenchError("workbench_route_not_found", "Workbench read model route is not implemented.", { route: url.pathname }));
}
async function authenticateWorkbenchRead(request, response, options) {
const access = options.accessController;
if (!access?.authenticate) {
sendJson(response, 503, workbenchError("workbench_access_controller_missing", "Workbench read model requires access controller."));
return null;
}
await access.ensureBootstrap?.();
const auth = await access.authenticate(request, { required: true });
if (!auth?.ok) {
sendJson(response, auth?.status ?? 401, auth ?? workbenchError("auth_required", "Authentication is required."));
return null;
}
return auth;
}
async function handleWorkbenchSessionList(response, url, options, actor) {
const projectId = textValue(url.searchParams.get("projectId")) || DEFAULT_PROJECT_ID;
const limit = boundedLimit(url.searchParams.get("limit"));
const includeSessionId = safeSessionId(url.searchParams.get("includeSessionId"));
const store = options.accessController?.store;
let sessions = await store?.listAgentSessionsForUser?.({
ownerUserId: actor.id,
actorRole: actor.role,
ownerScoped: true,
projectId,
limit,
includeArchived: false
}) ?? [];
if (includeSessionId && !sessions.some((session) => session.id === includeSessionId)) {
const included = await visibleSessionById(store, includeSessionId, actor);
if (included && (!projectId || included.projectId === projectId)) sessions = [included, ...sessions];
}
const summaries = sessions.map((session) => sessionSummary(session, options)).filter(Boolean);
sendJson(response, 200, {
ok: true,
status: "succeeded",
contractVersion: "workbench-sessions-v1",
projectId,
sessions: summaries,
conversations: summaries.map(sessionConversationSeed),
count: summaries.length,
nextCursor: null,
valuesRedacted: true,
secretMaterialStored: false
});
}
async function handleWorkbenchSessionDetail(response, options, actor, sessionId) {
const session = await visibleSessionById(options.accessController?.store, sessionId, actor);
if (!session) return sendJson(response, 404, workbenchError("workbench_session_not_found", "Workbench session is not visible to the current actor.", { sessionId }));
sendJson(response, 200, {
ok: true,
status: "found",
contractVersion: "workbench-session-detail-v1",
session: sessionDetail(session, options),
valuesRedacted: true,
secretMaterialStored: false
});
}
async function handleWorkbenchMessagePage(response, url, options, actor, sessionId) {
const session = await visibleSessionById(options.accessController?.store, sessionId, actor);
if (!session) return sendJson(response, 404, workbenchError("workbench_session_not_found", "Workbench session is not visible to the current actor.", { sessionId }));
const messages = sessionMessages(session);
const limit = boundedLimit(url.searchParams.get("limit"));
const offset = cursorOffset(url.searchParams.get("cursor") ?? url.searchParams.get("after"));
const page = messages.slice(offset, offset + limit);
const nextOffset = offset + page.length;
sendJson(response, 200, {
ok: true,
status: "succeeded",
contractVersion: "workbench-message-page-v1",
sessionId: session.id,
conversationId: safeConversationId(session.conversationId) ?? null,
messages: page,
count: page.length,
total: messages.length,
cursor: offset > 0 ? cursorFromOffset(offset) : null,
nextCursor: nextOffset < messages.length ? cursorFromOffset(nextOffset) : null,
hasMore: nextOffset < messages.length,
valuesRedacted: true,
secretMaterialStored: false
});
}
async function handleWorkbenchTurnSnapshot(response, url, options, actor, rawTurnId) {
const queryTraceId = safeTraceId(url.searchParams.get("traceId"));
const traceId = safeTraceId(rawTurnId) ?? queryTraceId;
const turnId = safeTurnId(rawTurnId) || traceId || rawTurnId;
if (!traceId) return sendJson(response, 400, workbenchError("invalid_turn_id", "turnId must currently be a trace id or include traceId query.", { turnId }));
const result = options.codeAgentChatResults?.get?.(traceId) ?? null;
const session = await visibleSessionByTrace(options.accessController?.store, traceId, actor);
if (result?.ownerUserId && !canActorReadOwner(result.ownerUserId, actor)) {
return sendJson(response, 403, workbenchError("agent_session_owner_required", "Only the session owner or admin can read this Workbench turn.", { traceId }));
}
const resultVisible = result && (session || actor.role === "admin" || Boolean(result.ownerUserId));
const trace = traceSnapshot(options, traceId);
const status = turnStatus(result, session, trace);
const found = Boolean(resultVisible || session);
if (!found) return sendJson(response, 404, workbenchError("workbench_turn_not_found", "Workbench turn is not visible to the current actor.", { turnId, traceId }));
sendJson(response, 200, {
ok: true,
status,
contractVersion: "workbench-turn-snapshot-v1",
turn: turnSnapshot({ turnId, traceId, status, result, session, trace }),
valuesRedacted: true,
secretMaterialStored: false
});
}
async function handleWorkbenchTraceEventPage(response, url, options, actor, rawTraceId) {
const traceId = safeTraceId(rawTraceId);
if (!traceId) return sendJson(response, 400, workbenchError("invalid_trace_id", "traceId must start with trc_.", { traceId: rawTraceId }));
const result = options.codeAgentChatResults?.get?.(traceId) ?? null;
const session = await visibleSessionByTrace(options.accessController?.store, traceId, actor);
if (result?.ownerUserId && !canActorReadOwner(result.ownerUserId, actor)) {
return sendJson(response, 403, workbenchError("agent_session_owner_required", "Only the session owner or admin can read this Workbench trace.", { traceId }));
}
const resultVisible = result && (session || actor.role === "admin" || Boolean(result.ownerUserId));
if (!resultVisible && !session) {
return sendJson(response, 404, workbenchError("workbench_trace_not_found", "Workbench trace is not visible to the current actor.", { traceId }));
}
const page = traceEventPage(traceSnapshot(options, traceId), tracePageOptions(url));
sendJson(response, 200, {
ok: true,
status: "succeeded",
contractVersion: "workbench-trace-events-v1",
traceId,
...page,
valuesRedacted: true,
secretMaterialStored: false
});
}
async function visibleSessionById(store, sessionId, actor) {
const safeId = safeSessionId(sessionId);
if (!safeId) return null;
const session = await store?.getAgentSession?.(safeId) ?? null;
return canActorReadSession(session, actor) ? session : null;
}
async function visibleSessionByTrace(store, traceId, actor) {
const safeId = safeTraceId(traceId);
if (!safeId) return null;
const session = await store?.getAgentSessionByTraceId?.(safeId) ?? null;
return canActorReadSession(session, actor) ? session : null;
}
function sessionSummary(session, options) {
if (!session?.id) return null;
const snapshot = objectValue(session.session);
const traceId = safeTraceId(session.lastTraceId ?? snapshot.lastTraceId ?? snapshot.currentTraceId) ?? null;
const trace = traceId ? traceSnapshot(options, traceId) : null;
const messages = sessionMessages(session);
const status = normalizeStatus(snapshot.sessionStatus ?? session.status ?? trace?.status);
return {
sessionId: session.id,
conversationId: safeConversationId(session.conversationId) ?? null,
threadId: safeOpaqueId(session.threadId) ?? (textValue(session.threadId) || null),
projectId: session.projectId ?? null,
agentId: session.agentId ?? "hwlab-code-agent",
status,
running: RUNNING_STATUSES.has(status),
terminal: TERMINAL_STATUSES.has(status) && !RUNNING_STATUSES.has(status),
lastTraceId: traceId,
providerProfile: textValue(snapshot.providerProfile) || null,
messageCount: messages.length,
firstUserMessagePreview: firstUserPreview(messages),
updatedAt: session.updatedAt ?? snapshot.updatedAt ?? null,
turnSummary: traceId ? {
turnId: traceId,
traceId,
status,
running: RUNNING_STATUSES.has(status),
terminal: TERMINAL_STATUSES.has(status) && !RUNNING_STATUSES.has(status),
eventCount: trace?.eventCount ?? null,
updatedAt: trace?.updatedAt ?? session.updatedAt ?? null
} : null,
valuesRedacted: true
};
}
function sessionDetail(session, options) {
return {
...sessionSummary(session, options),
metadata: {
startedAt: session.startedAt ?? null,
endedAt: session.endedAt ?? null,
ownerUserId: session.ownerUserId ?? null
},
messagePageUrl: `/v1/workbench/sessions/${encodeURIComponent(session.id)}/messages`,
valuesRedacted: true,
secretMaterialStored: false
};
}
function sessionConversationSeed(summary) {
return {
conversationId: summary.conversationId,
sessionId: summary.sessionId,
threadId: summary.threadId,
status: summary.status,
lastTraceId: summary.lastTraceId,
messageCount: summary.messageCount,
firstUserMessagePreview: summary.firstUserMessagePreview,
updatedAt: summary.updatedAt,
valuesRedacted: true
};
}
function sessionMessages(session) {
const snapshot = objectValue(session.session);
const raw = Array.isArray(snapshot.messages) ? snapshot.messages : Array.isArray(snapshot.chatMessages) ? snapshot.chatMessages : [];
const messages = raw.map((message, index) => messageFact(message, index, session, snapshot)).filter(Boolean);
if (!messages.some((message) => message.role === "assistant") && textValue(snapshot.finalResponse)) {
messages.push(messageFact({ role: "assistant", text: snapshot.finalResponse, traceId: session.lastTraceId ?? snapshot.lastTraceId, source: "finalResponse" }, messages.length, session, snapshot));
}
return messages;
}
function messageFact(message, index, session, snapshot) {
const role = textValue(message?.role) || (index % 2 === 0 ? "user" : "assistant");
const traceId = safeTraceId(message?.traceId ?? message?.turnTraceId ?? session.lastTraceId ?? snapshot.lastTraceId) ?? null;
const messageId = safeMessageId(message?.messageId ?? message?.id) || `msg_${hash(`${session.id}:${index}:${role}:${traceId ?? "none"}`).slice(0, 24)}`;
const text = textValue(message?.text ?? message?.content ?? message?.message ?? message?.finalResponse);
const rawParts = Array.isArray(message?.parts) ? message.parts : [];
const parts = rawParts.length > 0
? rawParts.map((part, partIndex) => partFact(part, partIndex, messageId, traceId)).filter(Boolean)
: text ? [partFact({ type: "text", text }, 0, messageId, traceId)] : [];
return {
messageId,
role,
sessionId: session.id,
conversationId: safeConversationId(session.conversationId) ?? null,
traceId,
turnId: safeTurnId(message?.turnId) || traceId,
status: normalizeStatus(message?.status ?? (role === "assistant" ? session.status : "completed")),
parts,
textPreview: text ? text.slice(0, 240) : null,
createdAt: textValue(message?.createdAt ?? message?.timestamp ?? message?.at) || session.startedAt || session.updatedAt || null,
updatedAt: textValue(message?.updatedAt) || session.updatedAt || null,
valuesRedacted: message?.valuesRedacted !== false
};
}
function partFact(part, index, messageId, traceId) {
const type = textValue(part?.type) || "text";
const text = textValue(part?.text ?? part?.content ?? part?.message);
return {
partId: safePartId(part?.partId ?? part?.id) || `prt_${hash(`${messageId}:${index}:${type}`).slice(0, 24)}`,
messageId,
traceId,
type,
text: text || null,
status: normalizeStatus(part?.status ?? "completed"),
toolName: textValue(part?.toolName ?? part?.name) || null,
createdAt: textValue(part?.createdAt ?? part?.timestamp) || null,
valuesRedacted: part?.valuesRedacted !== false
};
}
function turnSnapshot({ turnId, traceId, status, result, session, trace }) {
const messages = session ? sessionMessages(session) : [];
const userMessage = messages.find((message) => message.role === "user") ?? null;
const assistantMessage = [...messages].reverse().find((message) => message.role === "assistant") ?? null;
return {
turnId,
traceId,
status,
running: RUNNING_STATUSES.has(status),
terminal: TERMINAL_STATUSES.has(status) && !RUNNING_STATUSES.has(status),
sessionId: safeSessionId(result?.sessionId ?? result?.session?.sessionId ?? session?.id) ?? null,
conversationId: safeConversationId(result?.conversationId ?? session?.conversationId) ?? null,
threadId: safeOpaqueId(result?.threadId ?? result?.session?.threadId ?? session?.threadId) ?? (textValue(session?.threadId) || null),
userMessageId: userMessage?.messageId ?? null,
assistantMessageId: assistantMessage?.messageId ?? null,
agentRun: result?.agentRun ?? null,
trace: {
traceId,
status: trace?.status ?? "missing",
eventCount: trace?.eventCount ?? 0,
updatedAt: trace?.updatedAt ?? null
},
urls: {
self: `/v1/workbench/turns/${encodeURIComponent(turnId)}`,
traceEvents: `/v1/workbench/traces/${encodeURIComponent(traceId)}/events`,
compatResult: `/v1/agent/chat/result/${encodeURIComponent(traceId)}`,
compatTurn: `/v1/agent/turns/${encodeURIComponent(traceId)}`,
compatTrace: `/v1/agent/traces/${encodeURIComponent(traceId)}`
}
};
}
function traceEventPage(snapshot, options) {
const sourceEvents = Array.isArray(snapshot?.events) ? snapshot.events : [];
const indexed = sourceEvents.map((event, index) => ({ event, seq: eventSeq(event, index) }));
const startIndex = indexed.findIndex((item) => item.seq > options.afterSeq);
const offset = startIndex >= 0 ? startIndex : indexed.length;
const page = indexed.slice(offset, offset + options.limit);
const events = page.map((item) => ({ ...item.event, seq: item.seq }));
const toSeq = page.length ? page[page.length - 1].seq : options.afterSeq;
const hasMore = offset + page.length < indexed.length;
return {
events,
eventCount: indexed.length,
range: {
afterSeq: options.afterSeq,
fromSeq: page.length ? page[0].seq : null,
toSeq,
limit: options.limit,
returned: events.length,
total: indexed.length
},
hasMore,
nextSeq: toSeq,
nextCursor: hasMore ? `seq:${toSeq}` : null,
traceStatus: snapshot?.status ?? "missing",
updatedAt: snapshot?.updatedAt ?? null
};
}
function tracePageOptions(url) {
const cursor = textValue(url.searchParams.get("cursor"));
const cursorSeq = cursor.startsWith("seq:") ? Number.parseInt(cursor.slice(4), 10) : NaN;
const afterSeq = Number.isInteger(cursorSeq)
? cursorSeq
: nonNegativeInteger(url.searchParams.get("afterSeq") ?? url.searchParams.get("sinceSeq") ?? url.searchParams.get("since"), 0);
return { afterSeq, limit: boundedLimit(url.searchParams.get("limit")) };
}
function traceSnapshot(options, traceId) {
return (options.traceStore ?? defaultCodeAgentTraceStore).snapshot(traceId);
}
function turnStatus(result, session, trace) {
return normalizeStatus(
result?.status ??
result?.agentRun?.terminalStatus ??
result?.agentRun?.commandState ??
result?.agentRun?.status ??
session?.status ??
objectValue(session?.session).sessionStatus ??
trace?.status ??
"unknown"
);
}
function canActorReadSession(session, actor) {
if (!session || !actor) return false;
if (actor.role === "admin") return true;
return session.ownerUserId === actor.id;
}
function canActorReadOwner(ownerUserId, actor) {
if (!ownerUserId || !actor) return true;
if (actor.role === "admin") return true;
return ownerUserId === actor.id;
}
function methodNotAllowed(response, allowed) {
sendJson(response, 405, workbenchError("method_not_allowed", `Use ${allowed} for this Workbench read model route.`));
}
function workbenchError(code, message, extra = {}) {
return {
ok: false,
status: "failed",
error: { code, message, ...extra },
valuesRedacted: true,
secretMaterialStored: false
};
}
function boundedLimit(value) {
return Math.min(MAX_PAGE_LIMIT, parsePositiveInteger(value, DEFAULT_PAGE_LIMIT));
}
function cursorOffset(value) {
const text = textValue(value);
if (!text) return 0;
if (text.startsWith("idx:")) return nonNegativeInteger(text.slice(4), 0);
return nonNegativeInteger(text, 0);
}
function cursorFromOffset(offset) {
return `idx:${Math.max(0, Math.trunc(Number(offset) || 0))}`;
}
function nonNegativeInteger(value, fallback) {
const parsed = Number.parseInt(value ?? "", 10);
return Number.isInteger(parsed) && parsed >= 0 ? parsed : fallback;
}
function eventSeq(event, index) {
const seq = Number(event?.seq);
return Number.isFinite(seq) && seq > 0 ? Math.trunc(seq) : index + 1;
}
function normalizeStatus(value) {
const text = textValue(value).toLowerCase().replace(/_/gu, "-");
if (text === "cancelled") return "canceled";
return text || "unknown";
}
function firstUserPreview(messages) {
return messages.find((message) => message.role === "user")?.textPreview ?? null;
}
function objectValue(value) {
return value && typeof value === "object" && !Array.isArray(value) ? value : {};
}
function textValue(value) {
return String(value ?? "").trim();
}
function safeTurnId(value) {
const text = textValue(value);
return /^turn_[A-Za-z0-9_.:-]+$/u.test(text) ? text : null;
}
function safeMessageId(value) {
const text = textValue(value);
return /^msg_[A-Za-z0-9_.:-]+$/u.test(text) ? text : null;
}
function safePartId(value) {
const text = textValue(value);
return /^prt_[A-Za-z0-9_.:-]+$/u.test(text) ? text : null;
}
function hash(value) {
return createHash("sha256").update(String(value)).digest("hex");
}
+10
View File
@@ -1,3 +1,7 @@
/*
* SPEC: PJ2026-010403 API契约 draft-2026-06-17-r0; PJ2026-010401 Web工作台 draft-2026-06-17-r0
* 职责: Cloud API REST/SSE route dispatcherWorkbench read model / compat wrapper dispatcher
*/
import { createServer } from "node:http";
import { randomUUID } from "node:crypto";
import { copyFileSync, existsSync, lstatSync, mkdirSync, symlinkSync } from "node:fs";
@@ -62,6 +66,7 @@ import {
handleCodeAgentTurnHttp,
handleCodeAgentTraceHttp
} from "./server-code-agent-http.ts";
import { handleWorkbenchReadModelHttp } from "./server-workbench-http.ts";
import { handleM3IoControlHttp } from "./server-m3-http.ts";
import { handleSkillsHttp } from "./server-skills-http.ts";
import { handleProviderProfileCatalogHttp, handleProviderProfilesHttp } from "./provider-profile-management.ts";
@@ -478,6 +483,11 @@ async function handleRestAdapter(request, response, url, options) {
return;
}
if (url.pathname === "/v1/workbench/sessions" || url.pathname.startsWith("/v1/workbench/sessions/") || url.pathname.startsWith("/v1/workbench/turns/") || url.pathname.startsWith("/v1/workbench/traces/")) {
await handleWorkbenchReadModelHttp(request, response, url, options);
return;
}
const agentConversationMatch = url.pathname.match(/^\/v1\/agent\/conversations\/([^/]+)$/u);
if (agentConversationMatch && request.method === "GET") {
await options.accessController.getAgentConversation(request, response, decodeURIComponent(agentConversationMatch[1]));