From bc9ee41d83666bd81a35c79c2cf0f0aa2fe674c0 Mon Sep 17 00:00:00 2001 From: lyon Date: Wed, 17 Jun 2026 11:12:04 +0800 Subject: [PATCH] feat(workbench): add read-only API read model --- internal/cloud/access-control.test.ts | 69 ++- internal/cloud/access-control.ts | 12 +- internal/cloud/server-workbench-http.test.ts | 175 +++++++ internal/cloud/server-workbench-http.ts | 497 +++++++++++++++++++ internal/cloud/server.ts | 10 + 5 files changed, 720 insertions(+), 43 deletions(-) create mode 100644 internal/cloud/server-workbench-http.test.ts create mode 100644 internal/cloud/server-workbench-http.ts diff --git a/internal/cloud/access-control.test.ts b/internal/cloud/access-control.test.ts index 0b14ae4a..8b74c133 100644 --- a/internal/cloud/access-control.test.ts +++ b/internal/cloud/access-control.test.ts @@ -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 { diff --git a/internal/cloud/access-control.ts b/internal/cloud/access-control.ts index 15727402..3f3aa10b 100644 --- a/internal/cloud/access-control.ts +++ b/internal/cloud/access-control.ts @@ -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 }); } diff --git a/internal/cloud/server-workbench-http.test.ts b/internal/cloud/server-workbench-http.test.ts new file mode 100644 index 00000000..d0a9f5c9 --- /dev/null +++ b/internal/cloud/server-workbench-http.test.ts @@ -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() + }; +} diff --git a/internal/cloud/server-workbench-http.ts b/internal/cloud/server-workbench-http.ts new file mode 100644 index 00000000..7b226e8b --- /dev/null +++ b/internal/cloud/server-workbench-http.ts @@ -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 sync、billing 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"); +} diff --git a/internal/cloud/server.ts b/internal/cloud/server.ts index 9ccc1a4f..c8160962 100644 --- a/internal/cloud/server.ts +++ b/internal/cloud/server.ts @@ -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 dispatcher。Workbench 新资源路由应委托 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]));