diff --git a/AGENTS.md b/AGENTS.md index 78e9569..2699e74 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -122,7 +122,8 @@ bindings. Persistent sessions use `actor-*/personal/session-/` or `src/sessions/workspace-store.ts` owns chat directories and immutable ownership files. Inject the same store into metadata, UI state and transcript stores in -`src/server.ts`. Keep `data/session-metadata/` as the owner/history index; new +`src/server.ts`. Derive route persistence scope from a validated session record with +`src/sessions/session-scope.ts`; UI state contexts require actor, project and session IDs. Keep `data/session-metadata/` as the owner/history index; new messages, transcripts and learning state go only to `chat/`. Preserve full identity checks on reads and lists, session filename collision rejection, and serialized ownership creation. Scope changes in the frontend start a new conversation and use diff --git a/Dockerfile b/Dockerfile index 9211c81..2f80363 100644 --- a/Dockerfile +++ b/Dockerfile @@ -70,9 +70,10 @@ COPY --from=deps /app/.opencode/node_modules ./.opencode/node_modules COPY package.json bun.lock ./ COPY tsconfig.json opencode.json agent-models.json README.md .gitignore ./ COPY scripts ./scripts +COPY openapi ./openapi COPY src ./src COPY .opencode ./.opencode -RUN bun run check +RUN bun run check && bun run check:openapi FROM build AS runner WORKDIR /app diff --git a/README.md b/README.md index fc2ae18..dab5d5c 100644 --- a/README.md +++ b/README.md @@ -572,6 +572,11 @@ OpenCode 按标准 MCP `tools/call` 调用，以 `_meta["ai.opencode/sessionID"] ### 历史窗口接入与分页 历史专用接口契约见 [history.openapi.json](openapi/history.openapi.json)；`openapi/openapi.json` 仍是产品 Server 的上游快照。 +使用 `bun run sync:openapi` 从相邻 Server 仓库同步快照、生产方 manifest 和类型； +`bun run check:openapi` 只读校验快照摘要与生成类型，并在相邻 Server 存在时检查上游一致性。 +也可通过 `--upstream /path/to/server-v1.openapi.json` 指定生产方契约；同步时要求同目录的 `manifest.json`。 +独立仓库或容器构建没有相邻 Server 时只证明本地快照一致，检查输出会明确说明；发布前需在集成工作区完成上游检查。 + - `GET /api/v1/sessions?purpose=ordinary_chat&limit=30&q=...&cursor=...` 返回 `sessions`、`has_more`、`next_cursor`，搜索覆盖当前验证范围全部主题和未分类会话的标题与用户/助手可见正文，返回 `search_snippet`；前端以 `include_collaboration=true` 同时包含手动任务协作（继续排除后台执行会话），可选 `topic_id` 筛选主题或 `unclassified`；按更新时间及 ID 倒序，游标绑定用户、项目、用途、主题筛选和查询。 - `GET /api/v1/session/:session_id?message_limit=40` 返回最近消息及 `message_page: {has_more, before_cursor}`；`GET /api/v1/session/:session_id/messages?limit=40&before=...` 读取更早消息，每页内部按原有消息顺序排列。 diff --git a/docs/agent-ui-protocol.md b/docs/agent-ui-protocol.md index 9444a09..c6e33e7 100644 --- a/docs/agent-ui-protocol.md +++ b/docs/agent-ui-protocol.md @@ -67,3 +67,13 @@ business task. Running text additions use `POST /api/v1/responses` with stream=false, conversation, `tjwater.delivery="steer" | "queue"` and a stable tjwater.input_id. HTTP 202/tjwater.response_input is an inbox receipt, not execution completion. The existing SSE connection carries tjwater.user_input pending/delivered states; delivered means the native inbox took the input. `GET /api/v1/session/:session_id/inputs` lists pending input; PATCH of its /:input_id delivery can change its mode before processing and returns 409 afterward. Restored session streams retain these semantics. A pending host question or failed continuation does not disable input admission. With no active host turn, steer/queue messages enter the native inbox with resume=false and remain available for the next owned continuation. This does not answer a host question or launch inference. A lost delivery acknowledgement can be reconciled only after verifying both native consumption and completed related output. HTTP 409 with code RESPONSE_FINISHED means the input was rejected before admission because the turn has ended; the client may refresh state and send it as a new response. Other failures do not authorize automatic resubmission. + +## Chat state projections + +`src/routes/chat-run-accumulator.ts` consumes native and host events before publishing them. +Permission replies (including `permission_response` with `status="error"` and `error`), +question reconciliation and abort must use `publishRunEvent`, rather than mutate legacy messages. +Canonical parts are the saved history; reconnect compatibility fields are projections of the same events. +`data-run_status` retains completed, waiting_input, error or aborted state for reloads. Host-confirmed +interaction updates may update stored parts after a response has finished without reopening its wire stream. +UI, redacted transcript and native execution history retain their separate purposes and retention rules. diff --git a/openapi/openapi.json b/openapi/openapi.json index 3b7afc4..43bbbb4 100644 --- a/openapi/openapi.json +++ b/openapi/openapi.json @@ -4683,6 +4683,11 @@ "title": "Execution Request Id", "type": "string" }, + "execution_scope": { + "minLength": 1, + "title": "Execution Scope", + "type": "string" + }, "handling_context": { "additionalProperties": { "$ref": "#/components/schemas/JsonValue" @@ -4846,6 +4851,7 @@ } }, "required": [ + "execution_scope", "project_id", "task_id", "task_type", @@ -5660,6 +5666,11 @@ "title": "Execution Request Id", "type": "string" }, + "execution_scope": { + "minLength": 1, + "title": "Execution Scope", + "type": "string" + }, "handling_context": { "additionalProperties": { "$ref": "#/components/schemas/JsonValue" @@ -5773,6 +5784,7 @@ } }, "required": [ + "execution_scope", "project_id", "task_id", "task_type", diff --git a/openapi/server-manifest.json b/openapi/server-manifest.json new file mode 100644 index 0000000..50226fc --- /dev/null +++ b/openapi/server-manifest.json @@ -0,0 +1,9 @@ +{ + "contract_version": "1.0.0", + "contracts": { + "server": { + "file": "server-v1.openapi.json", + "sha256": "426b8d9170f253aa7132e9376031a7da8c04ac6f6e693948890d6f343420af1b" + } + } +} diff --git a/package.json b/package.json index 03eaaf3..23dd15c 100644 --- a/package.json +++ b/package.json @@ -11,7 +11,8 @@ "dev": "bun --watch src/server.ts", "build": "bun run check", "generate:openapi": "openapi-typescript openapi/openapi.json -o src/generated/server-api.d.ts", - "check:openapi": "bun run generate:openapi && git diff --exit-code -- openapi/openapi.json src/generated/server-api.d.ts", + "check:openapi": "bun scripts/server-contract.ts", + "sync:openapi": "bun scripts/server-contract.ts --sync", "check": "bun run typecheck && bun run typecheck:opencode", "check:models": "bun scripts/check-models.ts", "check:websearch": "bun scripts/check-websearch.ts", diff --git a/scripts/server-contract.ts b/scripts/server-contract.ts new file mode 100644 index 0000000..102bd8d --- /dev/null +++ b/scripts/server-contract.ts @@ -0,0 +1,57 @@ +import { createHash, randomUUID } from "node:crypto"; +import { readFile, writeFile, rename, rm } from "node:fs/promises"; +import { dirname, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; +import openapiTS, { astToString, COMMENT_HEADER } from "openapi-typescript"; + +const local = new URL("../openapi/openapi.json", import.meta.url); +const manifest = new URL("../openapi/server-manifest.json", import.meta.url); +const generated = new URL("../src/generated/server-api.d.ts", import.meta.url); +const defaultUpstream = new URL("../../next-tjwater-server/contracts/server-v1.openapi.json", import.meta.url); + +export function assertContractDigest(raw: string, expected: string) { + if (createHash("sha256").update(raw).digest("hex") !== expected) throw new Error("Server contract digest differs; run bun run sync:openapi"); +} + +async function main() { + const sync = process.argv.includes("--sync"); + const upstreamIndex = process.argv.indexOf("--upstream"); + if (upstreamIndex >= 0 && !process.argv[upstreamIndex + 1]) throw new Error("--upstream requires an OpenAPI file"); + const upstream = upstreamIndex >= 0 ? resolve(process.argv[upstreamIndex + 1]) : defaultUpstream; + let upstreamRaw: string | undefined; + try { upstreamRaw = await readFile(upstream, "utf8"); } + catch (error) { + if (sync || upstreamIndex >= 0 || (error as NodeJS.ErrnoException).code !== "ENOENT") throw error; + } + if (sync) { + const producerManifest = JSON.parse(await readFile(resolve( + dirname(upstream instanceof URL ? fileURLToPath(upstream) : upstream), "manifest.json" + ), "utf8")); + assertContractDigest(upstreamRaw!, producerManifest.contracts.server.sha256); + // Complete validation and generation before replacing any checked-in artifact. + const types = COMMENT_HEADER + astToString(await openapiTS(JSON.parse(upstreamRaw!))); + const artifacts = [ + { target: local, content: upstreamRaw! }, + { target: manifest, content: JSON.stringify(producerManifest, null, 2) + "\n" }, + { target: generated, content: types }, + ].map((artifact) => ({ ...artifact, temporary: new URL(`${artifact.target.href}.${randomUUID()}.tmp`) })); + try { + for (const artifact of artifacts) await writeFile(artifact.temporary, artifact.content); + for (const artifact of artifacts) await rename(artifact.temporary, artifact.target); + } finally { + await Promise.all(artifacts.map((artifact) => rm(artifact.temporary, { force: true }))); + } + console.log("Server OpenAPI snapshot, producer digest and types synchronized."); + return; + } + const raw = await readFile(local, "utf8"); + const receipt = JSON.parse(await readFile(manifest, "utf8")); + assertContractDigest(raw, receipt.contracts.server.sha256); + if (upstreamRaw !== undefined && upstreamRaw !== raw) throw new Error("Upstream Server contract changed; run bun run sync:openapi"); + if (await readFile(generated, "utf8") !== COMMENT_HEADER + astToString(await openapiTS(JSON.parse(raw)))) throw new Error("Generated Server types differ; run bun run generate:openapi"); + console.log(upstreamRaw === undefined + ? "Local Server contract digest and types checked; upstream unavailable in this checkout." + : "Upstream Server contract, digest and generated types checked."); +} + +if (import.meta.main) await main(); diff --git a/src/chat/answer-stream-projection.ts b/src/chat/answer-stream-projection.ts index 73d13bd..f038aed 100644 --- a/src/chat/answer-stream-projection.ts +++ b/src/chat/answer-stream-projection.ts @@ -33,8 +33,8 @@ export function createAnswerStreamProjection(options: { } } as UIMessageChunk); options.onChange(); }; - const appendData = (event: string, data: Record) => { - if (finished) return; + const updateData = (event: string, data: Record, hostState = false) => { + if (finished && !hostState) return; if (event === "answer_ui") endText(); const type = `data-${event}`; const id = [data.id, data.request_id, data.envelope_id, data.block_id].find((value): value is string => typeof value === "string"); @@ -42,9 +42,11 @@ export function createAnswerStreamProjection(options: { const part = id ? { type, id, data } : { type, data }; if (existing >= 0) parts[existing] = part; else parts.push(part); - options.write(part as UIMessageChunk); + if (!finished) options.write(part as UIMessageChunk); options.onChange(); }; + const appendData = (event: string, data: Record) => updateData(event, data); + const recordData = (event: string, data: Record) => updateData(event, data, true); const finish = (finishReason: "stop" | "error" = "stop") => { if (finished) return; if (!parts.length) appendText("Agent 已完成处理。"); @@ -53,5 +55,5 @@ export function createAnswerStreamProjection(options: { options.onChange(); options.write({ type: "finish", finishReason }); }; - return { parts, appendText, appendData, finish, get finished() { return finished; } }; + return { parts, appendText, appendData, recordData, finish, get finished() { return finished; } }; } diff --git a/src/generated/server-api.d.ts b/src/generated/server-api.d.ts index a075ca2..7ad6931 100644 --- a/src/generated/server-api.d.ts +++ b/src/generated/server-api.d.ts @@ -7507,6 +7507,8 @@ export interface components { * Format: uuid */ execution_request_id: string; + /** Execution Scope */ + execution_scope: string; /** Handling Context */ handling_context?: { [key: string]: components["schemas"]["JsonValue"]; @@ -7863,6 +7865,8 @@ export interface components { * Format: uuid */ execution_request_id: string; + /** Execution Scope */ + execution_scope: string; /** Handling Context */ handling_context?: { [key: string]: components["schemas"]["JsonValue"]; diff --git a/src/routes/chat-run-accumulator.ts b/src/routes/chat-run-accumulator.ts new file mode 100644 index 0000000..4148095 --- /dev/null +++ b/src/routes/chat-run-accumulator.ts @@ -0,0 +1,221 @@ +import type { UIMessage, UIMessageChunk } from "ai"; +import { createAnswerStreamProjection } from "../chat/answer-stream-projection.js"; +import type { UIEnvelopePayload } from "../uiEnvelope/types.js"; +import type { PermissionRequestPayload, QuestionRequestPayload, TodoUpdatePayload } from "./chatStream.js"; +import { + type ActiveRun, type RunStatus, updateLastAssistantMessage, applyInteractionResponse, + completeBackendProgress, cancelBackendTodos, upsertBackendProgress, + toFrontendPermission, upsertBackendQuestion, upsertBackendTodoUpdate, + appendBackendToolArtifact, appendBackendUiEnvelope, +} from "./chatUiState.js"; + +/** Compatibility fields are a projection of the same accepted event as canonical parts. */ +function reduceRunEvent(activeRun: ActiveRun, event: string, data: Record, onStatus: (status: RunStatus) => void) { + if (event === "token") { + activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ + ...message, + content: `${typeof message.content === "string" ? message.content : ""}${typeof data.content === "string" ? data.content : ""}`, + isError: false, + })); + } else if (event === "final_answer") { + activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ + ...message, content: typeof data.content === "string" ? data.content : message.content, + })); + } else if (event === "progress") { + activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ + ...message, + progress: upsertBackendProgress(message.progress, data), + })); + } else if (event === "done") { + activeRun.status = data.reason === "waiting_input" ? "waiting_input" : "completed"; + onStatus(activeRun.status); + activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ + ...message, + content: + typeof message.content === "string" && message.content.trim() + ? message.content + : data.reason === "waiting_input" ? "等待你的回答。" : "Agent 已完成处理，但没有生成文本回答。请查看过程记录，或换个更具体的问题重试。", + progress: completeBackendProgress(message.progress), + })); + } else if (event === "error" || event === "aborted") { + activeRun.status = event === "aborted" || activeRun.status === "aborted" ? "aborted" : "error"; + onStatus(activeRun.status); + activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ + ...message, + content: + typeof message.content === "string" && message.content.trim() + ? message.content + : `⚠️ **错误：** ${typeof data.message === "string" ? data.message : "unknown error"}`, + isError: true, + progress: completeBackendProgress(message.progress), + todos: cancelBackendTodos(message.todos), + })); + } else if (event === "permission_request") { + const payload = data as PermissionRequestPayload; + activeRun.pendingPermissions.set(payload.request_id, payload); + activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ + ...message, + permissions: [ + ...(Array.isArray(message.permissions) ? message.permissions : []), + toFrontendPermission(payload), + ], + })); + } else if (event === "permission_response") { + const requestId = + typeof data.request_id === "string" ? data.request_id : undefined; + const reply = + data.reply === "once" || data.reply === "always" || data.reply === "reject" + ? data.reply + : undefined; + if (requestId && (reply || data.status === "error")) { + if (reply) activeRun.pendingPermissions.delete(requestId); + activeRun.messages = applyInteractionResponse(activeRun.messages, "permission_response", data); + } + } else if (event === "question_request") { + const payload = data as QuestionRequestPayload; + let shouldTrackQuestion = true; + if (payload.tool?.callID) { + if (payload.request_id !== payload.tool.callID) { + activeRun.pendingQuestions.delete(payload.tool.callID); + } else { + const hasActionableQuestion = [...activeRun.pendingQuestions.values()].some( + (question) => + question.tool?.callID === payload.tool?.callID && + question.request_id !== payload.tool?.callID, + ); + if (hasActionableQuestion) { + activeRun.messages = updateLastAssistantMessage( + activeRun.messages, + (message) => ({ + ...message, + questions: upsertBackendQuestion(message.questions, payload), + }), + ); + shouldTrackQuestion = false; + } + } + } + if (shouldTrackQuestion) { + activeRun.pendingQuestions.set(payload.request_id, payload); + activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ + ...message, + questions: upsertBackendQuestion(message.questions, payload), + })); + } + } else if (event === "question_response") { + const requestId = + typeof data.request_id === "string" ? data.request_id : undefined; + if (requestId) { + activeRun.pendingQuestions.delete(requestId); + activeRun.messages = applyInteractionResponse(activeRun.messages, "question_response", data); + } + } else if (event === "execution_plan") { + activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ + ...message, executionPlans: [...(Array.isArray(message.executionPlans) ? message.executionPlans : []), data], + })); + } else if (event === "todo_update") { + const payload = data as TodoUpdatePayload; + activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ + ...message, + todos: upsertBackendTodoUpdate(message.todos, payload), + })); + } else if (event === "tool_call") { + activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ + ...message, + artifacts: appendBackendToolArtifact(message.artifacts, data), + })); + } else if (event === "ui_envelope") { + const payload = data as UIEnvelopePayload; + activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ + ...message, + uiEnvelopes: appendBackendUiEnvelope(message.uiEnvelopes, payload), + })); + } + +} + +/** All host mutations go through the accumulator before subscribers or persistence. */ +export function publishRunEvent(run: ActiveRun, event: string, data: Record, notify = true) { + if (run.applyEvent) run.applyEvent(event, data); + else reduceRunEvent(run, event, data, () => {}); + if (notify) for (const subscriber of run.subscribers) { + if (event === "done" || event === "error" || event === "aborted") { + subscriber.write("run_status", { ...data, id: "run-status", status: run.status }); + } + subscriber.write(event, data); + } +} + +/** One event owner updates the reconnect view and canonical wire/storage projections. */ +export function createChatRunAccumulator(options: { + run: ActiveRun; + messages: UIMessage[]; + messageId: string; + sessionId: string; + collaboration: boolean; + write: (chunk: UIMessageChunk) => void; + onStatus: (status: RunStatus) => void; +}) { + const activeRun = options.run; + let history = [...options.messages]; + const syncParts = () => { + activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ + ...message, id: options.messageId, parts: projection.parts.map((part) => ({ ...part })), + })); + }; + const projection = createAnswerStreamProjection({ + messageId: options.messageId, sessionId: options.sessionId, + write: options.write, onChange: syncParts, + }); + const apply = (event: string, data: Record) => { + reduceRunEvent(activeRun, event, data, options.onStatus); + let historicalResponse = false; + if (event === "permission_response" || event === "question_response") { + const updated = applyInteractionResponse(history, event, data); + historicalResponse = updated !== history; + history = updated; + if (historicalResponse) { + const byId = new Map(history.map((message) => [message.id, message])); + activeRun.messages = activeRun.messages.map((message) => { + if (!message || typeof message !== "object" || !("id" in message)) return message; + const canonical = byId.get(String(message.id)); + return canonical ? { ...message, parts: canonical.parts } : message; + }); + } + } + + const hidden = options.collaboration && ["progress", "todo_update", "tool_call", "ui_envelope", "execution_plan"].includes(event); + if (!hidden && !historicalResponse) { + if (event === "token") projection.appendText(typeof data.content === "string" ? data.content : ""); + else if (event === "done" || event === "error" || event === "aborted") { + const message = typeof data.message === "string" ? data.message : event === "aborted" ? "请求已中断" : "Agent stream failed"; + if (!projection.parts.some((part) => ["text", "data-final_answer", "data-answer_ui"].includes(String(part.type)))) { + const assistant = activeRun.messages.at(-1) as { content?: string } | undefined; + projection.appendText(assistant?.content || message); + } + projection.recordData("run_status", { id: "run-status", status: activeRun.status, ...(event === "done" ? {} : { message }) }); + if (event !== "done" && !projection.finished) options.write({ type: "error", errorText: message }); + projection.finish(event === "done" ? "stop" : "error"); + } else if (["permission_response", "question_response"].includes(event)) projection.recordData(event, data); + else projection.appendData(event, data); + } + syncParts(); + }; + const uiMessages = (): UIMessage[] => [...history, { + id: options.messageId, role: "assistant", parts: projection.parts.map((part) => ({ ...part })), + } as UIMessage]; + activeRun.applyEvent = apply; + activeRun.uiMessages = uiMessages; + return { + apply, + finish: projection.finish, + get finished() { return projection.finished; }, + uiMessages, + appendInput(message: UIMessage, text: string) { + if (history.some((item) => item.id === message.id)) return; + history.push(message); + const tail = activeRun.messages.at(-1); + activeRun.messages = [...activeRun.messages.slice(0, -1), { id: message.id, role: "user", content: text }, ...(tail ? [tail] : [])]; + }, + }; +} diff --git a/src/routes/chatAuxiliaryRoutes.ts b/src/routes/chatAuxiliaryRoutes.ts index f4bad00..ee3eff8 100644 --- a/src/routes/chatAuxiliaryRoutes.ts +++ b/src/routes/chatAuxiliaryRoutes.ts @@ -1,3 +1,5 @@ +import { publishRunEvent } from "./chat-run-accumulator.js"; +import { toSessionScope } from "../sessions/session-scope.js"; import { type Router } from "express"; import { z } from "zod"; import type { InteractionRequestStore } from "../sessions/interaction-request-store.js"; @@ -13,9 +15,6 @@ import { type FrontendActionCoordinator } from "../frontendAction/coordinator.js import { type ActiveRun, type RunStatus, - cancelBackendTodos, - completeBackendProgress, - updateLastAssistantMessage, } from "./chatUiState.js"; const abortPayloadSchema = z.object({ @@ -33,11 +32,6 @@ type RegisterAuxiliaryRoutesOptions = { frontendActionCoordinator: FrontendActionCoordinator; }; -const toSessionUiStateContext = (record: { sessionId: string; actorKey: string; projectKey: string }) => ({ - sessionId: record.sessionId, - actorKey: record.actorKey, - projectKey: record.projectKey, -}); export const registerChatAuxiliaryRoutes = ( chatRouter: Router, @@ -80,12 +74,12 @@ export const registerChatAuxiliaryRoutes = ( const runtimeContext = getRuntimeSessionContext(sessionRecord.sessionId); if (runtimeContext) setRuntimeSessionContext({ ...runtimeContext, hostQuestionClosed: true }); activeRuns.get(sessionRecord.sessionId)?.controller.abort(); - await interactionRequestStore?.cancelPending(toSessionUiStateContext(sessionRecord)); - const pendingInteraction = await interactionRequestStore?.getBlocking(toSessionUiStateContext(sessionRecord)); + await interactionRequestStore?.cancelPending(toSessionScope(sessionRecord)); + const pendingInteraction = await interactionRequestStore?.getBlocking(toSessionScope(sessionRecord)); if (pendingInteraction?.delivery && runtime && interactionRequestStore) { const status = await runtime.stopInteractionInput(sessionRecord.sessionId, pendingInteraction.delivery.native_message_id); - const latest = await interactionRequestStore.get(toSessionUiStateContext(sessionRecord), pendingInteraction.interaction_id); - if (latest?.delivery && !["delivered", "cancelled"].includes(latest.delivery.status)) await interactionRequestStore.updateDelivery(toSessionUiStateContext(sessionRecord), latest.interaction_id, { + const latest = await interactionRequestStore.get(toSessionScope(sessionRecord), pendingInteraction.interaction_id); + if (latest?.delivery && !["delivered", "cancelled"].includes(latest.delivery.status)) await interactionRequestStore.updateDelivery(toSessionScope(sessionRecord), latest.interaction_id, { expected_revision: latest.delivery.revision, status, }); } @@ -99,28 +93,21 @@ export const registerChatAuxiliaryRoutes = ( run.status = "aborted"; lastRunStatuses.set(parsed.data.session_id, "aborted"); run.controller.abort(); - run.messages = updateLastAssistantMessage(run.messages, (message) => ({ - ...message, - content: - typeof message.content === "string" && message.content.trim() - ? message.content - : "⚠️ **请求已中断**", - isError: true, - progress: completeBackendProgress(message.progress), - todos: cancelBackendTodos(message.todos), - })); + publishRunEvent(run, "aborted", { session_id: parsed.data.session_id, message: "请求已中断" }, false); const currentState = await sessionUiStateStore.read( - toSessionUiStateContext(sessionRecord), + toSessionScope(sessionRecord), ); - await sessionUiStateStore.write(toSessionUiStateContext(sessionRecord), { + await sessionUiStateStore.write(toSessionScope(sessionRecord), { sessionId: sessionRecord.sessionId, isTitleManuallyEdited: currentState?.isTitleManuallyEdited ?? false, - messages: run.messages, + ...currentState, + messages: run.uiMessages?.() ?? run.messages, }); for (const subscriber of run.subscribers) { subscriber.write("error", { session_id: parsed.data.session_id, message: "请求已中断", + status: "aborted", }); } // The generation route emits its terminal Response after cleanup and diff --git a/src/routes/chatInteractionRoutes.ts b/src/routes/chatInteractionRoutes.ts index ff7b89b..cae279d 100644 --- a/src/routes/chatInteractionRoutes.ts +++ b/src/routes/chatInteractionRoutes.ts @@ -1,3 +1,5 @@ +import { publishRunEvent } from "./chat-run-accumulator.js"; +import { toSessionScope } from "../sessions/session-scope.js"; import { refreshSessionCredential } from "../product/credential-refresh.js"; import { type Router } from "express"; import { z } from "zod"; @@ -8,9 +10,7 @@ import { type OpencodeRuntimeAdapter } from "../runtime/opencode.js"; import { SessionContextMismatchError, type SessionMetadataStore } from "../sessions/metadataStore.js"; import { type SessionUiStateStore } from "../sessions/uiStateStore.js"; import { - type ActiveRun, - toPermissionStatus, - updateLastAssistantPermission, + type ActiveRun, applyInteractionResponse, } from "./chatUiState.js"; const permissionReplyPayloadSchema = z.object({ @@ -35,7 +35,6 @@ type RegisterInteractionRoutesOptions = { sessionUiStateStore: SessionUiStateStore; }; -const toSessionUiStateContext = (sessionId: string) => ({ sessionId }); export const registerChatInteractionRoutes = ( chatRouter: Router, @@ -104,12 +103,13 @@ export const registerChatInteractionRoutes = ( } const persistPermissionState = async () => { const currentState = await sessionUiStateStore.read( - toSessionUiStateContext(sessionRecord.sessionId), + toSessionScope(sessionRecord), ); - await sessionUiStateStore.write(toSessionUiStateContext(sessionRecord.sessionId), { + await sessionUiStateStore.write(toSessionScope(sessionRecord), { sessionId: sessionRecord.sessionId, isTitleManuallyEdited: currentState?.isTitleManuallyEdited ?? false, - messages: run.messages, + ...currentState, + messages: run.uiMessages?.() ?? run.messages, }); }; @@ -121,18 +121,10 @@ export const registerChatInteractionRoutes = ( message: parsed.data.message, }); } catch (error) { - run.messages = updateLastAssistantPermission( - run.messages, - requestId, - (permission) => ({ - ...permission, - status: "error", - error: - error instanceof Error - ? error.message - : "failed to reply permission", - }), - ); + publishRunEvent(run, "permission_response", { + session_id: sessionRecord.sessionId, request_id: requestId, status: "error", + error: error instanceof Error ? error.message : "failed to reply permission", + }); await persistPermissionState().catch((persistError) => { logger.warn( { err: persistError, sessionId: sessionRecord.sessionId }, @@ -146,30 +138,15 @@ export const registerChatInteractionRoutes = ( return; } - run.pendingPermissions.delete(requestId); - const status = toPermissionStatus(parsed.data.reply); - run.messages = updateLastAssistantPermission( - run.messages, - requestId, - (permission) => ({ - ...permission, - status, - repliedAt: Date.now(), - }), - ); + publishRunEvent(run, "permission_response", { + session_id: sessionRecord.sessionId, request_id: requestId, reply: parsed.data.reply, + }); await persistPermissionState().catch((persistError) => { logger.warn( { err: persistError, sessionId: sessionRecord.sessionId }, "failed to persist permission reply state", ); }); - for (const subscriber of run.subscribers) { - subscriber.write("permission_response", { - session_id: sessionRecord.sessionId, - request_id: requestId, - reply: parsed.data.reply, - }); - } res.status(202).json({ session_id: sessionRecord.sessionId, @@ -222,20 +199,15 @@ export const registerChatInteractionRoutes = ( const nativeStatus = confirmedStatus ? undefined : await runtime.getQuestionStatus({ sessionId: session.sessionId, requestId: canonicalId }); const status = confirmedStatus ?? (nativeStatus === "cancelled" ? "rejected" : nativeStatus === "missing" ? "invalidated" : nativeStatus!); const persist = async (nextStatus: string, answers?: string[][]) => { - const update = (messages: unknown[]) => messages.map((message) => { - if (!isRecord(message)) return message; - const questions = Array.isArray(message.questions) ? message.questions.map((q: unknown) => isRecord(q) && q.requestId === canonicalId ? { ...q, status: nextStatus, ...(answers ? { answers } : {}), error: undefined } : q) : undefined; - const parts = Array.isArray(message.parts) ? [...message.parts] : undefined; - if (parts && nextStatus !== "pending" && parts.some((part: unknown) => isRecord(part) && part.type === "data-question_request" && isRecord(part.data) && part.data.request_id === canonicalId)) { - const response = { type: "data-question_response", data: { session_id: session.sessionId, request_id: canonicalId, status: nextStatus, rejected: nextStatus === "rejected" || nextStatus === "invalidated", ...(answers ? { answers } : {}) } }; - const index = parts.findIndex((part: unknown) => isRecord(part) && part.type === "data-question_response" && isRecord(part.data) && part.data.request_id === canonicalId); - if (index >= 0) parts[index] = { ...parts[index], data: { ...(isRecord(parts[index]) && isRecord(parts[index].data) ? parts[index].data : {}), ...response.data } }; else parts.push(response); - } - return { ...message, ...(questions ? { questions } : {}), ...(parts ? { parts } : {}) }; - }); + const response = { + session_id: session.sessionId, request_id: canonicalId, status: nextStatus, + rejected: nextStatus === "rejected" || nextStatus === "invalidated", ...(answers ? { answers } : {}), + }; // Update the live projection before entering the same atomic persistence queue. - if (run) { run.messages = update(run.messages); if (nextStatus !== "pending") run.pendingQuestions.delete(canonicalId); } - await sessionUiStateStore.update(context, (latest) => ({ ...latest, messages: update(latest.messages) })); + if (run && nextStatus !== "pending") publishRunEvent(run, "question_response", response); + await sessionUiStateStore.update(context, (latest) => ({ + ...latest, messages: applyInteractionResponse(latest.messages, "question_response", response), + })); }; if (action === "status" || status !== "pending") { await persist(status); @@ -246,7 +218,6 @@ export const registerChatInteractionRoutes = ( if (action === "reply") await runtime.replyQuestion({ sessionId: session.sessionId, requestId: canonicalId, answers: answers ?? [] }); else await runtime.rejectQuestion({ sessionId: session.sessionId, requestId: canonicalId }); await persist(action === "reply" ? "answered" : "rejected", answers); - for (const subscriber of run?.subscribers ?? []) subscriber.write("question_response", { session_id: session.sessionId, request_id: canonicalId, ...(action === "reply" ? { answers } : { rejected: true }) }); if (run && run.status !== "running" && !run.pendingPermissions.size && !run.pendingQuestions.size) activeRuns.delete(session.sessionId); res.status(202).json({ session_id: session.sessionId, request_id: canonicalId, ...(action === "reply" ? { answers } : { rejected: true }) }); } catch (error) { diff --git a/src/routes/chatSession.ts b/src/routes/chatSession.ts index e0ac354..86b4457 100644 --- a/src/routes/chatSession.ts +++ b/src/routes/chatSession.ts @@ -174,27 +174,6 @@ export const generateSessionTitle = async ( } }; -export const getConversationTurnStats = async ( - runtime: OpencodeRuntimeAdapter, - sessionId: string, -) => { - const messages = await runtime.messages(sessionId, 12); - return messages.reduce( - (stats, message) => { - if (message.info.role === "user") { - stats.userMessageCount += 1; - } else if (message.info.role === "assistant") { - stats.assistantMessageCount += 1; - } - return stats; - }, - { - userMessageCount: 0, - assistantMessageCount: 0, - }, - ); -}; - export const buildPromptWithLearningContext = async ( memoryStore: MemoryStore, actorKey: string, diff --git a/src/routes/chatSessionRoutes.ts b/src/routes/chatSessionRoutes.ts index ae7f492..6a71596 100644 --- a/src/routes/chatSessionRoutes.ts +++ b/src/routes/chatSessionRoutes.ts @@ -1,3 +1,4 @@ +import { toSessionScope } from "../sessions/session-scope.js"; import { registerChatTopicRoutes } from "./chat-topic-routes.js"; import { matchHistoryText } from "../sessions/history-search.js"; import { credentialRefreshCoordinator } from "../product/credential-refresh.js"; @@ -107,11 +108,6 @@ type RegisterChatSessionRoutesOptions = { sessionUiStateStore: SessionUiStateStore; }; -const toSessionUiStateContext = (sessionRecord: SessionRecord) => ({ - sessionId: sessionRecord.sessionId, - actorKey: sessionRecord.actorKey, - projectKey: sessionRecord.projectKey, -}); const toSse = (event: string, data: Record) => `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`; @@ -176,6 +172,12 @@ export const registerChatSessionRoutes = ( const { activeRuns, lastRunStatuses } = runState; const getSessionRunStatus = runState.getStatus; + const readMessages = (record: SessionRecord, state: { messages: unknown[] } | null) => { + if (readPurpose(record) === "task_collaboration") return state?.messages ?? []; + const run = activeRuns.get(record.sessionId); + return run?.uiMessages?.() ?? run?.messages ?? state?.messages ?? []; + }; + const withInteractionReplies = async (record: SessionRecord, messages: unknown[]) => { if (!interactionRequestStore) return messages; const ids = new Set(messages.flatMap((message) => { @@ -183,7 +185,7 @@ export const registerChatSessionRoutes = ( const value = message as Record; return value.role === "user" && typeof value.id === "string" ? [value.id] : []; })); - const replies = await interactionRequestStore.forMessages(toSessionUiStateContext(record), ids); + const replies = await interactionRequestStore.forMessages(toSessionScope(record), ids); return projectInteractionReplyMessages(messages, replies); }; @@ -192,7 +194,7 @@ export const registerChatSessionRoutes = ( || !isGenericTaskCollaborationTitle(record.title)) return record; if ((!credential && !knownTaskTitle) || signal?.aborted) return record; try { - const state = await sessionUiStateStore.read(toSessionUiStateContext(record)); + const state = await sessionUiStateStore.read(toSessionScope(record)); if (state?.isTitleManuallyEdited) return record; let taskTitle = knownTaskTitle; if (!taskTitle && credential) { @@ -206,7 +208,7 @@ export const registerChatSessionRoutes = ( } } if (!taskTitle?.trim()) return record; - const latestState = await sessionUiStateStore.read(toSessionUiStateContext(record)); + const latestState = await sessionUiStateStore.read(toSessionScope(record)); if (latestState?.isTitleManuallyEdited) return record; return await sessionMetadataStore.touch(record, { title: taskTitle }, { preserveUpdatedAt: true, expectedTitle: record.title ?? "", @@ -297,7 +299,7 @@ export const registerChatSessionRoutes = ( sessionId, ); if (!record) return null; - const state = await sessionUiStateStore.read(toSessionUiStateContext(record)); + const state = await sessionUiStateStore.read(toSessionScope(record)); if (!state) return null; return { record, state, taskTitle }; }; @@ -603,7 +605,7 @@ export const registerChatSessionRoutes = ( const titled = await sessionMetadataStore.touch(record, { title: `${history.taskTitle} · 调试` }); const messages = await withInteractionReplies(history.record, history.state.messages ?? []); await sessionUiStateStore.write( - toSessionUiStateContext(titled), + toSessionScope(titled), buildForkedSessionUiState({ messages }, { keepMessageCount: messages.length, targetSessionId: titled.sessionId, @@ -771,8 +773,8 @@ export const registerChatSessionRoutes = ( const searchable: Array = []; for (const record of records) { const titleMatches = (record.title ?? "新对话").toLocaleLowerCase().includes((query.data.q ?? "").toLocaleLowerCase()); - const state = query.data.q && !titleMatches ? await sessionUiStateStore.read(toSessionUiStateContext(record)) : null; - const messages = activeRuns.get(record.sessionId)?.messages ?? state?.messages ?? []; + const state = query.data.q && !titleMatches ? await sessionUiStateStore.read(toSessionScope(record)) : null; + const messages = readMessages(record, state); searchable.push({ ...record, searchSnippet: query.data.q && !titleMatches ? matchHistoryText(messages, query.data.q) : undefined }); } let page; @@ -828,9 +830,8 @@ export const registerChatSessionRoutes = ( const context = getLocalAgentContext(req); const record = await sessionMetadataStore.get(context, req.params.session_id); if (!record) { res.status(404).json({ message: "session not found" }); return; } - const state = await sessionUiStateStore.read(toSessionUiStateContext(record)); - const messages = readPurpose(record) === "task_collaboration" ? state?.messages ?? [] - : activeRuns.get(record.sessionId)?.messages ?? state?.messages ?? []; + const state = await sessionUiStateStore.read(toSessionScope(record)); + const messages = readMessages(record, state); try { const page = sessionMessagePage(messages, { scope: historyScope([record.actorKey, record.projectKey, record.sessionId]), ...query.data }); res.json({ ...page, messages: await withInteractionReplies(record, page.messages) }); @@ -871,10 +872,9 @@ export const registerChatSessionRoutes = ( sessionRecord = await refreshCollaborationTitle(sessionRecord, getTrustedProductCredential(req)); const state = await sessionUiStateStore.read( - toSessionUiStateContext(sessionRecord), + toSessionScope(sessionRecord), ); - const messages = readPurpose(sessionRecord) === "task_collaboration" ? state?.messages ?? [] - : activeRuns.get(sessionRecord.sessionId)?.messages ?? state?.messages ?? []; + const messages = readMessages(sessionRecord, state); const messagePayload = query.data.message_limit ? sessionMessagePage(messages, { scope: historyScope([sessionRecord.actorKey, sessionRecord.projectKey, sessionId]), limit: query.data.message_limit, }) : { messages }; @@ -929,9 +929,8 @@ export const registerChatSessionRoutes = ( res.flushHeaders?.(); const run = activeRuns.get(sessionRecord.sessionId); - const state = await sessionUiStateStore.read(toSessionUiStateContext(sessionRecord)); - const messages = readPurpose(sessionRecord) === "task_collaboration" ? state?.messages ?? [] - : run?.messages ?? state?.messages ?? []; + const state = await sessionUiStateStore.read(toSessionScope(sessionRecord)); + const messages = readMessages(sessionRecord, state); const messagePayload = query.data.message_limit ? sessionMessagePage(messages, { scope: historyScope([sessionRecord.actorKey, sessionRecord.projectKey, sessionId]), limit: query.data.message_limit, }) : { messages }; @@ -1005,11 +1004,11 @@ export const registerChatSessionRoutes = ( title, }); const state = await sessionUiStateStore.read( - toSessionUiStateContext(nextSessionRecord), + toSessionScope(nextSessionRecord), ); if (state) { await sessionUiStateStore.write( - toSessionUiStateContext(nextSessionRecord), + toSessionScope(nextSessionRecord), { ...state, isTitleManuallyEdited: @@ -1134,7 +1133,7 @@ export const registerChatSessionRoutes = ( } const sourceState = sourceSessionRecord ? await sessionUiStateStore.read( - toSessionUiStateContext(sourceSessionRecord), + toSessionScope(sourceSessionRecord), ) : null; const forkTitle = sourceSessionRecord?.title @@ -1145,7 +1144,7 @@ export const registerChatSessionRoutes = ( { title: forkTitle }, ); await sessionUiStateStore.write( - toSessionUiStateContext(titledTargetSessionRecord), + toSessionScope(titledTargetSessionRecord), buildForkedSessionUiState(sourceState, { keepMessageCount: parsed.data.keep_message_count, targetSessionId: nextSessionId, diff --git a/src/routes/chatStreamRoute.ts b/src/routes/chatStreamRoute.ts index 546d1b5..dff92e2 100644 --- a/src/routes/chatStreamRoute.ts +++ b/src/routes/chatStreamRoute.ts @@ -1,3 +1,4 @@ +import { toSessionScope } from "../sessions/session-scope.js"; import type { TaskCalculationWaiter } from "../mcp/task-calculation-waiter.js"; import { createResponsesProjection, responseRequestSchema, responsePrompt } from "../chat/responses-protocol.js"; import { ResponseInputTracker, WaitingResponseInputQueue } from "../chat/response-inputs.js"; @@ -5,7 +6,7 @@ import { reconcileInteractionDelivery } from "../sessions/interaction-delivery-r import type { InteractionRequestStore } from "../sessions/interaction-request-store.js"; import { renderInteractionAnswer, type InteractionRequest } from "../chat/interaction-request-contract.js"; import { createInteractionReplyMessage } from "../chat/interaction-message-presentation.js"; -import { createAnswerStreamProjection } from "../chat/answer-stream-projection.js"; +import { createChatRunAccumulator, publishRunEvent } from "./chat-run-accumulator.js"; import type { AgentMessageCoordinator } from "../chat/agent-message-coordinator.js"; import { credentialRefreshCoordinator } from "../product/credential-refresh.js"; import { type Router } from "express"; @@ -47,7 +48,6 @@ import { } from "../product/operational-task-client.js"; import { type SessionTranscriptStore } from "../sessions/transcriptStore.js"; import { type SessionUiStateStore } from "../sessions/uiStateStore.js"; -import type { UIEnvelopePayload } from "../uiEnvelope/types.js"; import { toActorKey, toProjectKey } from "../utils/fileStore.js"; import { buildPromptWithLearningContext, @@ -59,29 +59,14 @@ import { } from "./chatSession.js"; import { collectTextContent, - type PermissionRequestPayload, - type QuestionRequestPayload, streamPromptResponse, - type TodoUpdatePayload, } from "./chatStream.js"; import { type ChatRunState } from "./chatRunState.js"; import { type ActiveRun, - type StreamSubscriber, - appendBackendToolArtifact, - appendBackendUiEnvelope, - cancelBackendTodos, - completeBackendProgress, createInitialStreamingMessages, isObjectRecord, - toFrontendPermission, - toPermissionStatus, updateLastAssistantMessage, - updateLastAssistantPermission, - updateLastAssistantQuestion, - upsertBackendProgress, - upsertBackendQuestion, - upsertBackendTodoUpdate, } from "./chatUiState.js"; import { agentTriggerContextSchema, @@ -106,11 +91,6 @@ const payloadSchema = z.object({ }).strict().optional(), }); -const toSessionUiStateContext = (sessionRecord: SessionRecord) => ({ - sessionId: sessionRecord.sessionId, - actorKey: sessionRecord.actorKey, - projectKey: sessionRecord.projectKey, -}); const runtimeHasConversation = async ( runtime: OpencodeRuntimeAdapter, @@ -311,7 +291,7 @@ export const registerChatStreamRoute = ( let boundOperationalSessionId: string | undefined; let startingSessionId: string | undefined; let continuation: InteractionRequest | null = null; - let continuationScope: ReturnType | undefined; + let continuationScope: ReturnType | undefined; let waitingInteraction: InteractionRequest | null = null; try { const authContext = getLocalAgentContext(req); @@ -365,7 +345,7 @@ export const registerChatStreamRoute = ( res.status(400).json({ message: "Interaction continuation requires an existing conversation" }); return; } if (existingSessionRecord && interactionRequestStore) { - continuationScope = toSessionUiStateContext(existingSessionRecord); + continuationScope = toSessionScope(existingSessionRecord); if (input.tjwater.interaction_id) { continuation = await interactionRequestStore.get(continuationScope, input.tjwater.interaction_id); if (!continuation?.delivery || !["answered", "skipped"].includes(continuation.status) || continuation.delivery.status === "cancelled") { @@ -523,7 +503,7 @@ export const registerChatStreamRoute = ( sessionId: requestContext.clientSessionId, }; const initialSessionState = await sessionUiStateStore.read( - toSessionUiStateContext(activeSessionRecord), + toSessionScope(activeSessionRecord), ); const persistedUiMessages = toUiMessages(initialSessionState?.messages); if (!continuation && input.tjwater.client_message_id && persistedUiMessages.some((message) => message.id === input.tjwater.client_message_id)) { @@ -551,7 +531,6 @@ export const registerChatStreamRoute = ( ...visiblePersistedMessages, createFallbackUserUiMessage(promptText, input.tjwater.client_message_id), ]; - let latestUiMessages = requestMessages; const baseMessages = toLegacyMessages(requestMessages); const recentTurns = await sessionTranscriptStore.getRecentTurns(historyContext, 8); @@ -605,44 +584,28 @@ export const registerChatStreamRoute = ( startingSessions.delete(activeSessionRecord.sessionId); startingSessionId = undefined; lastRunStatuses.set(clientSessionId, "running"); - const sessionUiStateContext = toSessionUiStateContext(activeSessionRecord); + const sessionUiStateContext = toSessionScope(activeSessionRecord); const assistantUiMessageId = continuationAssistantId ?? `msg_${randomUUID().replaceAll("-", "")}`; setRuntimeSessionContext({ ...getRuntimeSessionContext(binding.sessionId)!, hostTurnId: wire.response.id, hostMessageId: assistantUiMessageId }); - const syncLatestUiMessages = () => { - latestUiMessages = [ - ...requestMessages, - { - id: assistantUiMessageId, - role: "assistant", - parts: assistantParts.map((part) => ({ ...part })), - } as UIMessage, - ]; - }; - const writeUiChunk = (chunk: UIMessageChunk) => { - // Persist canonical parts after disconnect; the wire emitter checks the socket. - wire!.write(chunk); - }; - const projection = createAnswerStreamProjection({ - messageId: assistantUiMessageId, sessionId: clientSessionId, - write: writeUiChunk, onChange: syncLatestUiMessages, + const accumulator = createChatRunAccumulator({ + run: activeRun, messages: requestMessages, messageId: assistantUiMessageId, sessionId: clientSessionId, + collaboration: sessionPurpose === "task_collaboration", write: (chunk) => wire!.write(chunk), + onStatus: (status) => lastRunStatuses.set(clientSessionId, status), }); - const assistantParts = projection.parts; - const writeTextDelta = projection.appendText; - const writeDataPart = projection.appendData; - const finishUiMessage = projection.finish; + const writeUiChunk = (chunk: UIMessageChunk) => wire!.write(chunk); writeUiChunk({ type: "start", messageId: assistantUiMessageId }); let persistQueue = sessionUiStateStore.write(sessionUiStateContext, { executionPlans: executionPlanHistory, sessionId: activeSessionRecord.sessionId, isTitleManuallyEdited: initialSessionState?.isTitleManuallyEdited ?? false, - messages: sessionPurpose === "task_collaboration" ? toCollaborationMessages(latestUiMessages) : latestUiMessages, + messages: sessionPurpose === "task_collaboration" ? toCollaborationMessages(accumulator.uiMessages()) : accumulator.uiMessages(), }); const queueSessionUiStatePersist = () => { const snapshot = { executionPlans: executionPlanHistory, sessionId: activeSessionRecord.sessionId, isTitleManuallyEdited: initialSessionState?.isTitleManuallyEdited ?? false, - messages: sessionPurpose === "task_collaboration" ? toCollaborationMessages(latestUiMessages) : latestUiMessages, + messages: sessionPurpose === "task_collaboration" ? toCollaborationMessages(accumulator.uiMessages()) : accumulator.uiMessages(), }; persistQueue = persistQueue .catch((error) => { @@ -654,36 +617,6 @@ export const registerChatStreamRoute = ( .then(() => sessionUiStateStore.write(sessionUiStateContext, snapshot)); return persistQueue; }; - const primarySubscriber: StreamSubscriber = { - write: (event, data) => { - if (event === "token") { - writeTextDelta(typeof data.content === "string" ? data.content : ""); - return; - } - if (event === "done") { - finishUiMessage("stop"); - return; - } - if (event === "error") { - writeUiChunk({ - type: "error", - errorText: - typeof data.message === "string" - ? data.message - : "Agent stream failed", - }); - finishUiMessage("error"); - return; - } - writeDataPart(event, data); - }, - close: () => { - if (!res.writableEnded && !res.destroyed) { - res.end(); - } - }, - }; - activeRun.subscribers.add(primarySubscriber); let taskArtifactPublishQueue: Promise = Promise.resolve(); const publish = (event: string, data: Record) => { @@ -697,131 +630,11 @@ export const registerChatStreamRoute = ( : assistantUiMessageId, } : data; - if (event === "token") { - activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ - ...message, - content: `${typeof message.content === "string" ? message.content : ""}${typeof data.content === "string" ? data.content : ""}`, - isError: false, - })); - } else if (event === "final_answer") { - activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ - ...message, content: typeof data.content === "string" ? data.content : message.content, - })); - } else if (event === "progress") { - activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ - ...message, - progress: upsertBackendProgress(message.progress, data), - })); - } else if (event === "done") { - activeRun.status = data.reason === "waiting_input" ? "waiting_input" : "completed"; - lastRunStatuses.set(clientSessionId, activeRun.status); - activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ - ...message, - content: - typeof message.content === "string" && message.content.trim() - ? message.content - : data.reason === "waiting_input" ? "等待你的回答。" : "Agent 已完成处理，但没有生成文本回答。请查看过程记录，或换个更具体的问题重试。", - progress: completeBackendProgress(message.progress), - })); - } else if (event === "error") { - activeRun.status = activeRun.status === "aborted" ? "aborted" : "error"; - lastRunStatuses.set(clientSessionId, activeRun.status); - activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ - ...message, - content: - typeof message.content === "string" && message.content.trim() - ? message.content - : `⚠️ **错误：** ${typeof data.message === "string" ? data.message : "unknown error"}`, - isError: true, - progress: completeBackendProgress(message.progress), - todos: cancelBackendTodos(message.todos), - })); - } else if (event === "permission_request") { - const payload = data as PermissionRequestPayload; - activeRun.pendingPermissions.set(payload.request_id, payload); - activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ - ...message, - permissions: [ - ...(Array.isArray(message.permissions) ? message.permissions : []), - toFrontendPermission(payload), - ], - })); - } else if (event === "permission_response") { - const requestId = - typeof data.request_id === "string" ? data.request_id : undefined; - const reply = - data.reply === "once" || data.reply === "always" || data.reply === "reject" - ? data.reply - : undefined; - if (requestId && reply) { - activeRun.pendingPermissions.delete(requestId); - activeRun.messages = updateLastAssistantPermission( - activeRun.messages, - requestId, - (permission) => ({ - ...permission, - status: toPermissionStatus(reply), - repliedAt: Date.now(), - }), - ); - } - } else if (event === "question_request") { - const payload = data as QuestionRequestPayload; - let shouldTrackQuestion = true; - if (payload.tool?.callID) { - if (payload.request_id !== payload.tool.callID) { - activeRun.pendingQuestions.delete(payload.tool.callID); - } else { - const hasActionableQuestion = [...activeRun.pendingQuestions.values()].some( - (question) => - question.tool?.callID === payload.tool?.callID && - question.request_id !== payload.tool?.callID, - ); - if (hasActionableQuestion) { - activeRun.messages = updateLastAssistantMessage( - activeRun.messages, - (message) => ({ - ...message, - questions: upsertBackendQuestion(message.questions, payload), - }), - ); - shouldTrackQuestion = false; - } - } - } - if (shouldTrackQuestion) { - activeRun.pendingQuestions.set(payload.request_id, payload); - activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ - ...message, - questions: upsertBackendQuestion(message.questions, payload), - })); - } - } else if (event === "question_response") { - const requestId = - typeof data.request_id === "string" ? data.request_id : undefined; - if (requestId) { - activeRun.pendingQuestions.delete(requestId); - activeRun.messages = updateLastAssistantQuestion( - activeRun.messages, - requestId, - (question) => ({ - ...question, - status: data.rejected === true ? "rejected" : "answered", - repliedAt: Date.now(), - answers: Array.isArray(data.answers) ? data.answers : question.answers, - error: undefined, - }), - ); - } - } else if (event === "execution_plan") { + if (event === "execution_plan") { try { const snapshot = createExecutionPlanSnapshot(data, [{ executionPlans: executionPlanHistory }], clientSessionId); executionPlanHistory = [...executionPlanHistory, snapshot]; subscriberData = snapshot; - activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ - ...message, - executionPlans: [...(Array.isArray(message.executionPlans) ? message.executionPlans : []), snapshot], - })); if (activeSessionRecord.taskId && executionPlanNeedsTaskPublish(data)) { taskArtifactPublishQueue = taskArtifactPublishQueue .catch(() => undefined) @@ -843,23 +656,6 @@ export const registerChatStreamRoute = ( }); return; } - } else if (event === "todo_update") { - const payload = data as TodoUpdatePayload; - activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ - ...message, - todos: upsertBackendTodoUpdate(message.todos, payload), - })); - } else if (event === "tool_call") { - activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ - ...message, - artifacts: appendBackendToolArtifact(message.artifacts, data), - })); - } else if (event === "ui_envelope") { - const payload = data as UIEnvelopePayload; - activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ - ...message, - uiEnvelopes: appendBackendUiEnvelope(message.uiEnvelopes, payload), - })); } const hiddenCollaborationEvent = sessionPurpose === "task_collaboration" && [ @@ -869,12 +665,7 @@ export const registerChatStreamRoute = ( "ui_envelope", "execution_plan", ].includes(event); - if (!hiddenCollaborationEvent) { - for (const subscriber of activeRun.subscribers) { - subscriber.write(event, subscriberData); - } - } - activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ ...message, id: assistantUiMessageId, parts: assistantParts.map((part) => ({ ...part })) })); + publishRunEvent(activeRun, event, subscriberData, !hiddenCollaborationEvent); void queueSessionUiStatePersist().catch((error) => { logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state"); }); @@ -944,10 +735,7 @@ export const registerChatStreamRoute = ( : typeof generation === "number" && generation > 0 ? generation : null, report_revision: reportRef.success ? reportRef.data.revision : null, }; - const part = { type: "data-task_context" as const, data: reference }; - assistantParts.push(part); - writeUiChunk(part); - syncLatestUiMessages(); + accumulator.apply("task_context", reference); } const preparedMessage = taskContext ? [ @@ -970,11 +758,8 @@ export const registerChatStreamRoute = ( const messaging = getTrustedProductCredential(req) ? agentMessages : undefined; messaging?.start(activeSessionRecord); const inputTracker = new ResponseInputTracker(runtime, binding.sessionId, (receipt) => { - if (receipt.status === "delivered" && !requestMessages.some((message) => message.id === receipt.id)) { - requestMessages.push({ id: receipt.id, role: "user", parts: [{ type: "text", text: receipt.text }] } as UIMessage); - const tail = activeRun.messages.at(-1); - activeRun.messages = [...activeRun.messages.slice(0, -1), { id: receipt.id, role: "user", content: receipt.text }, ...(tail ? [tail] : [])]; - syncLatestUiMessages(); + if (receipt.status === "delivered") { + accumulator.appendInput({ id: receipt.id, role: "user", parts: [{ type: "text", text: receipt.text }] } as UIMessage, receipt.text); } publish("user_input", { session_id: clientSessionId, ...receipt }); }); @@ -1125,14 +910,14 @@ export const registerChatStreamRoute = ( const assistantMessage = [...messages] .reverse() .find((message) => message.info.role === "assistant"); - const assistantText = readUiMessageText(latestUiMessages.at(-1)) || collectTextContent(assistantMessage?.parts ?? []); + const assistantText = readUiMessageText(accumulator.uiMessages().at(-1)) || collectTextContent(assistantMessage?.parts ?? []); const latestSessionRecord = (await sessionMetadataStore.get( { actorKey, projectId, projectKey, userId }, activeSessionRecord.sessionId, )) ?? activeSessionRecord; const latestSessionState = await sessionUiStateStore.read( - toSessionUiStateContext(latestSessionRecord), + toSessionScope(latestSessionRecord), ); const existingSessionTitle = latestSessionRecord.title; let sessionTitle = existingSessionTitle; @@ -1229,29 +1014,13 @@ export const registerChatStreamRoute = ( await taskArtifactPublishQueue.catch(() => undefined); unregisterFrontendActionSink?.(); if (abortController.signal.aborted && !taskLeaseFailure) { - activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ - ...message, - content: - typeof message.content === "string" && message.content.trim() - ? message.content - : "⚠️ **请求已中断**", - isError: true, - progress: completeBackendProgress(message.progress), - todos: cancelBackendTodos(message.todos), - })); - if (!projection.finished) { - writeUiChunk({ - type: "error", - errorText: "请求已中断", - }); - finishUiMessage("error"); - } + publishRunEvent(activeRun, "aborted", { session_id: clientSessionId, message: "请求已中断" }); void queueSessionUiStatePersist().catch((error) => { logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist aborted chat stream state"); }); } - if (!projection.finished) { - finishUiMessage(activeRun.status === "error" ? "error" : "stop"); + if (!accumulator.finished) { + publishRunEvent(activeRun, activeRun.status === "error" ? "error" : "done", { session_id: clientSessionId }); } await queueSessionUiStatePersist().catch((error) => { logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state"); diff --git a/src/routes/chatUiState.ts b/src/routes/chatUiState.ts index 29276c0..09c93ad 100644 --- a/src/routes/chatUiState.ts +++ b/src/routes/chatUiState.ts @@ -15,6 +15,8 @@ export type StreamSubscriber = { }; export type ActiveRun = { + applyEvent?: (event: string, data: Record) => void; + uiMessages?: () => unknown[]; inputs?: ResponseInputTracker; clientSessionId: string; controller: AbortController; @@ -175,43 +177,50 @@ export const updateLastAssistantMessage = ( return messages; }; -export const updateLastAssistantPermission = ( - messages: unknown[], - requestId: string, - updater: (permission: Record) => Record, -) => - updateLastAssistantMessage(messages, (message) => { - const permissions = Array.isArray(message.permissions) - ? message.permissions - : []; - return { - ...message, - permissions: permissions.map((permission) => - isObjectRecord(permission) && permission.requestId === requestId - ? updater(permission) - : permission, - ), - }; - }); - -export const updateLastAssistantQuestion = ( - messages: unknown[], - requestId: string, - updater: (question: Record) => Record, -) => - updateLastAssistantMessage(messages, (message) => { - const questions = Array.isArray(message.questions) - ? message.questions - : []; - return { - ...message, - questions: questions.map((question) => - isObjectRecord(question) && question.requestId === requestId - ? updater(question) - : question, - ), - }; +/** Replies belong to the assistant that requested input, including earlier turns. */ +export function applyInteractionResponse( + messages: T[], + event: "permission_response" | "question_response", + data: Record, +): T[] { + const requestId = data.request_id; + if (typeof requestId !== "string") return messages; + const permission = event === "permission_response"; + const field = permission ? "permissions" : "questions"; + const requestType = permission ? "data-permission_request" : "data-question_request"; + let changed = false; + const next = messages.map((message) => { + if (!isObjectRecord(message) || message.role !== "assistant") return message; + const entries = Array.isArray(message[field]) ? message[field] as unknown[] : undefined; + const parts = Array.isArray(message.parts) ? [...message.parts] : undefined; + const ownsRequest = entries?.some((entry) => isObjectRecord(entry) && entry.requestId === requestId) + || parts?.some((part) => isObjectRecord(part) && part.type === requestType + && isObjectRecord(part.data) && part.data.request_id === requestId); + if (!ownsRequest) return message; + changed = true; + const updated = entries?.map((entry) => { + if (!isObjectRecord(entry) || entry.requestId !== requestId) return entry; + return permission ? { + ...entry, status: data.status === "error" ? "error" : toPermissionStatus(data.reply as PermissionReply), + error: data.status === "error" ? data.error : undefined, repliedAt: Date.now(), + } : { + ...entry, status: typeof data.status === "string" ? data.status : data.rejected === true ? "rejected" : "answered", + ...(Array.isArray(data.answers) ? { answers: data.answers } : {}), error: undefined, repliedAt: Date.now(), + }; + }); + if (parts && data.status !== "pending") { + const type = `data-${event}`; + const index = parts.findIndex((part) => isObjectRecord(part) && part.type === type + && isObjectRecord(part.data) && part.data.request_id === requestId); + const previous = index >= 0 && isObjectRecord(parts[index]) ? parts[index] : undefined; + const part = { ...previous, type, data: { ...(isObjectRecord(previous?.data) ? previous.data : {}), ...data } }; + if (index >= 0) parts[index] = part; + else parts.push(part); + } + return { ...message, ...(updated ? { [field]: updated } : {}), ...(parts ? { parts } : {}) }; }); + return changed ? next : messages; +} const getToolArtifactKind = (tool: string): ToolArtifactKind => { if (tool === "present_view") { diff --git a/src/routes/task-bridge-context.ts b/src/routes/task-bridge-context.ts new file mode 100644 index 0000000..b4a3de4 --- /dev/null +++ b/src/routes/task-bridge-context.ts @@ -0,0 +1,16 @@ +import type { Request, Response } from "express"; +import { isTrustedInternalBridgeRequest } from "../product/internal-bridge-auth.js"; +import { getRuntimeSessionContext } from "../runtime/sessionContext.js"; + +export function readTaskBridgeContext(req: Request, token: string, allowQuery = false) { + if (!isTrustedInternalBridgeRequest(req, token)) return null; + const runtimeSessionId = typeof req.body?.session_id === "string" ? req.body.session_id.trim() + : allowQuery && typeof req.query.session_id === "string" ? req.query.session_id.trim() : ""; + const context = runtimeSessionId ? getRuntimeSessionContext(runtimeSessionId) : null; + return context?.taskId ? { context, runtimeSessionId, taskId: context.taskId } : null; +} + +export function replyTaskBridgeError(res: Response, error: unknown, fallback: string) { + const status = error && typeof error === "object" && "status" in error && typeof error.status === "number" ? error.status : 502; + res.status(status).json({ message: error instanceof Error ? error.message : fallback }); +} diff --git a/src/routes/task-command-bridge.ts b/src/routes/task-command-bridge.ts new file mode 100644 index 0000000..10cee88 --- /dev/null +++ b/src/routes/task-command-bridge.ts @@ -0,0 +1,65 @@ +import type { Express } from "express"; +import { operationalTaskClient, expressAccessRequestSchema, proposedTaskActionSchema, humanTaskEvidenceCommandSchema, type OperationalTaskClient } from "../product/operational-task-client.js"; +import { readTaskBridgeContext, replyTaskBridgeError } from "./task-bridge-context.js"; + +export function registerTaskCommandBridge(app: Express, options: { + internalToken: string; + client?: Pick; +}) { + const client = options.client ?? operationalTaskClient; + app.post("/internal/operational-task/access-requests", async (req, res) => { + const bridge = readTaskBridgeContext(req, options.internalToken, true); + if (!bridge || bridge.context.purpose !== "task_execution") { + res.status(403).json({ message: "task execution context required" }); + return; + } + const parsed = expressAccessRequestSchema.safeParse(req.body?.request); + if (!parsed.success) { + res.status(400).json({ message: "invalid access request" }); + return; + } + try { + res.json(await client.requestDataAccess(bridge.runtimeSessionId, bridge.taskId, { + ...parsed.data, sourceSessionId: bridge.context.clientSessionId, + })); + } catch (error) { + replyTaskBridgeError(res, error, "access request failed"); + } + }); + + app.post("/internal/operational-task/actions", async (req, res) => { + const bridge = readTaskBridgeContext(req, options.internalToken, true); + if (!bridge) { + res.status(403).json({ message: "operational task context unavailable" }); + return; + } + const parsed = proposedTaskActionSchema.safeParse(req.body?.action); + if (!parsed.success) { + res.status(400).json({ message: "invalid proposed action", detail: parsed.error.flatten() }); + return; + } + try { + res.json(await client.proposeAction(bridge.runtimeSessionId, bridge.taskId, parsed.data)); + } catch (error) { + replyTaskBridgeError(res, error, "task action proposal failed"); + } + }); + + app.post("/internal/operational-task/human-evidence", async (req, res) => { + const bridge = readTaskBridgeContext(req, options.internalToken, true); + if (!bridge || bridge.context.purpose !== "task_collaboration") { + res.status(403).json({ message: "human evidence requires a task collaboration session" }); + return; + } + const parsed = humanTaskEvidenceCommandSchema.safeParse(req.body?.command); + if (!parsed.success) { + res.status(400).json({ message: "invalid human evidence", detail: parsed.error.flatten() }); + return; + } + try { + res.json(await client.addHumanEvidence(bridge.runtimeSessionId, bridge.taskId, parsed.data)); + } catch (error) { + replyTaskBridgeError(res, error, "human evidence publish failed"); + } + }); +} diff --git a/src/routes/task-publication-bridge.ts b/src/routes/task-publication-bridge.ts index 8279185..66ff446 100644 --- a/src/routes/task-publication-bridge.ts +++ b/src/routes/task-publication-bridge.ts @@ -4,7 +4,7 @@ import { z } from "zod"; import { isTrustedInternalBridgeRequest } from "../product/internal-bridge-auth.js"; import { operationalTaskClient, type OperationalTaskClient, taskArtifactBridgeSchema } from "../product/operational-task-client.js"; import { taskPresentationBridgeSchema } from "../product/task-presentation.js"; -import { getRuntimeSessionContext } from "../runtime/sessionContext.js"; +import { readTaskBridgeContext, replyTaskBridgeError } from "./task-bridge-context.js"; const artifactsInputSchema = taskArtifactBridgeSchema.and(z.object({ session_id: z.string().trim().min(1).max(128) })); @@ -19,12 +19,12 @@ export function registerTaskPublicationBridge(app: Express, options: { res.status(403).json({ message: "forbidden" }); return; } - const sessionId = typeof req.body?.session_id === "string" ? req.body.session_id.trim() : ""; - const context = sessionId ? getRuntimeSessionContext(sessionId) : null; - if (!context?.taskId) { + const bridge = readTaskBridgeContext(req, options.internalToken); + if (!bridge) { res.status(403).json({ message: "operational task context unavailable" }); return; } + const { context, runtimeSessionId: sessionId, taskId } = bridge; if ((kind === "presentation" || req.body?.progress !== undefined) && context.purpose !== "task_execution") { res.status(403).json({ message: "task execution context required" }); return; @@ -36,7 +36,7 @@ export function registerTaskPublicationBridge(app: Express, options: { res.status(400).json({ message: "invalid task presentation", detail: parsed.error.flatten() }); return; } - res.json(await client.publishPresentation(sessionId, context.taskId, { + res.json(await client.publishPresentation(sessionId, taskId, { requestId: parsed.data.request_id, sourceSessionId: context.clientSessionId, command: parsed.data.command, @@ -47,7 +47,7 @@ export function registerTaskPublicationBridge(app: Express, options: { res.status(400).json({ message: "invalid task artifacts", detail: parsed.error.flatten() }); return; } - res.json(await client.publishArtifacts(sessionId, context.taskId, { + res.json(await client.publishArtifacts(sessionId, taskId, { requestId: parsed.data.request_id, sourceSessionId: context.clientSessionId, evidence: parsed.data.evidence, @@ -57,8 +57,7 @@ export function registerTaskPublicationBridge(app: Express, options: { })); } } catch (error) { - const status = error && typeof error === "object" && "status" in error && typeof error.status === "number" ? error.status : 502; - res.status(status).json({ message: error instanceof Error ? error.message : `task ${kind} publish failed` }); + replyTaskBridgeError(res, error, `task ${kind} publish failed`); } }); } diff --git a/src/runtime/opencode.ts b/src/runtime/opencode.ts index 83380fc..43f7f63 100644 --- a/src/runtime/opencode.ts +++ b/src/runtime/opencode.ts @@ -381,13 +381,6 @@ export class OpencodeRuntimeAdapter { await rm(directory, { recursive: true, force: true }); } - async sendPrompt(sessionId: string, text: string) { - await this.prompt(sessionId, text); - // 当前 SDK 响应风格下，prompt() 本身不会直接返回完整 assistant parts， - // 所以这里紧跟一次 messages() 回读，给上层路由统一消费。 - return this.messages(sessionId); - } - async generateText(prompt: string, model?: RuntimeModelOverride, signal?: AbortSignal) { const client = await this.ensureClient(); const selected = model ?? parseModelReference(config.OPENCODE_MODEL); diff --git a/src/server.ts b/src/server.ts index b2a577b..f58b532 100644 --- a/src/server.ts +++ b/src/server.ts @@ -37,12 +37,7 @@ import { createAgentMessageQuery, registerAgentMessageBridge } from "./routes/ag import { FrontendActionCoordinator, FrontendActionError } from "./frontendAction/coordinator.js"; import { opencodeRuntime } from "./runtime/opencode.js"; import { getRuntimeSessionContext } from "./runtime/sessionContext.js"; -import { - operationalTaskClient, - expressAccessRequestSchema, - humanTaskEvidenceCommandSchema, - proposedTaskActionSchema, -} from "./product/operational-task-client.js"; +import { registerTaskCommandBridge } from "./routes/task-command-bridge.js"; import { isTrustedInternalBridgeRequest } from "./product/internal-bridge-auth.js"; const app = express(); @@ -112,80 +107,7 @@ app.post("/internal/frontend-actions/request", async (req, res) => { } }); -const readOperationalTaskBridgeContext = (req: express.Request) => { - if (!isTrustedInternalBridgeRequest(req, internalToken)) return null; - const runtimeSessionId = typeof req.body?.session_id === "string" - ? req.body.session_id.trim() - : typeof req.query.session_id === "string" - ? req.query.session_id.trim() - : ""; - const context = runtimeSessionId ? getRuntimeSessionContext(runtimeSessionId) : null; - return context?.taskId ? { context, runtimeSessionId, taskId: context.taskId } : null; -}; - -app.post("/internal/operational-task/access-requests", async (req, res) => { - const bridge = readOperationalTaskBridgeContext(req); - if (!bridge || bridge.context.purpose !== "task_execution") { - res.status(403).json({ message: "task execution context required" }); - return; - } - const parsed = expressAccessRequestSchema.safeParse(req.body?.request); - if (!parsed.success) { - res.status(400).json({ message: "invalid access request" }); - return; - } - try { - res.json(await operationalTaskClient.requestDataAccess(bridge.runtimeSessionId, bridge.taskId, { - ...parsed.data, sourceSessionId: bridge.context.clientSessionId, - })); - } catch (error) { - const status = error && typeof error === "object" && "status" in error && typeof error.status === "number" - ? error.status : 502; - res.status(status).json({ message: error instanceof Error ? error.message : "access request failed" }); - } -}); - -app.post("/internal/operational-task/actions", async (req, res) => { - const bridge = readOperationalTaskBridgeContext(req); - if (!bridge) { - res.status(403).json({ message: "operational task context unavailable" }); - return; - } - const parsed = proposedTaskActionSchema.safeParse(req.body?.action); - if (!parsed.success) { - res.status(400).json({ message: "invalid proposed action", detail: parsed.error.flatten() }); - return; - } - try { - res.json(await operationalTaskClient.proposeAction(bridge.runtimeSessionId, bridge.taskId, parsed.data)); - } catch (error) { - const status = error && typeof error === "object" && "status" in error && typeof error.status === "number" - ? error.status - : 502; - res.status(status).json({ message: error instanceof Error ? error.message : "task action proposal failed" }); - } -}); - -app.post("/internal/operational-task/human-evidence", async (req, res) => { - const bridge = readOperationalTaskBridgeContext(req); - if (!bridge || bridge.context.purpose !== "task_collaboration") { - res.status(403).json({ message: "human evidence requires a task collaboration session" }); - return; - } - const parsed = humanTaskEvidenceCommandSchema.safeParse(req.body?.command); - if (!parsed.success) { - res.status(400).json({ message: "invalid human evidence", detail: parsed.error.flatten() }); - return; - } - try { - res.json(await operationalTaskClient.addHumanEvidence(bridge.runtimeSessionId, bridge.taskId, parsed.data)); - } catch (error) { - const status = error && typeof error === "object" && "status" in error && typeof error.status === "number" - ? error.status - : 502; - res.status(status).json({ message: error instanceof Error ? error.message : "human evidence publish failed" }); - } -}); +registerTaskCommandBridge(app, { internalToken }); app.get("/health", async (_req, res) => { try { diff --git a/src/sessions/session-scope.ts b/src/sessions/session-scope.ts new file mode 100644 index 0000000..517e477 --- /dev/null +++ b/src/sessions/session-scope.ts @@ -0,0 +1,8 @@ +import type { SessionWorkspaceContext } from "../runtime/conversationWorkspace.js"; + +/** Always derive persistence scope from a host-validated session record. */ +export const toSessionScope = (record: SessionWorkspaceContext): SessionWorkspaceContext => ({ + sessionId: record.sessionId, + actorKey: record.actorKey, + projectKey: record.projectKey, +}); diff --git a/src/sessions/uiStateStore.ts b/src/sessions/uiStateStore.ts index 86265fd..7fce1b1 100644 --- a/src/sessions/uiStateStore.ts +++ b/src/sessions/uiStateStore.ts @@ -1,3 +1,4 @@ +import type { SessionWorkspaceContext } from "../runtime/conversationWorkspace.js"; import { join } from "node:path"; import { config } from "../config.js"; @@ -17,11 +18,7 @@ export type SessionUiStateRecord = { executionPlans?: Record[]; }; -type SessionUiStateContext = { - sessionId: string; - actorKey?: string; - projectKey?: string; -}; +type SessionUiStateContext = SessionWorkspaceContext; export class SessionUiStateStore { constructor(private readonly baseDir = config.SESSION_UI_STATE_STORAGE_DIR, private readonly workspaces?: SessionWorkspaceStore) {} diff --git a/tests/routes/chat-run-accumulator.test.ts b/tests/routes/chat-run-accumulator.test.ts new file mode 100644 index 0000000..eadccfa --- /dev/null +++ b/tests/routes/chat-run-accumulator.test.ts @@ -0,0 +1,47 @@ +import { expect, it } from "bun:test"; +import type { UIMessageChunk } from "ai"; +import { createChatRunAccumulator, publishRunEvent } from "../../src/routes/chat-run-accumulator.js"; +import { createInitialStreamingMessages, type ActiveRun } from "../../src/routes/chatUiState.js"; + +it.each([false, true])("projects one event stream into reconnect and persisted messages (collaboration=%s)", (collaboration) => { + const run: ActiveRun = { + clientSessionId: "s", controller: new AbortController(), status: "running", + messages: createInitialStreamingMessages([], "查询"), pendingPermissions: new Map(), pendingQuestions: new Map(), subscribers: new Set(), + }; + const chunks: UIMessageChunk[] = []; + const stream = createChatRunAccumulator({ run, messages: [{ id: "user", role: "user", parts: [{ type: "text", text: "查询" }] }], + messageId: "answer", sessionId: "s", collaboration, write: (chunk) => chunks.push(chunk), onStatus() {} }); + stream.apply("progress", { id: "query", status: "running", title: "查询资料" }); + stream.apply("token", { content: "已有数据" }); + stream.apply("permission_request", { request_id: "permission", session_id: "s", permission: "shell", patterns: [], always: [], created_at: 1 }); + stream.apply("permission_response", { request_id: "permission", reply: "once" }); + stream.appendInput({ id: "extra", role: "user", parts: [{ type: "text", text: "补充" }] }, "补充"); + stream.appendInput({ id: "extra", role: "user", parts: [] }, "重复"); + stream.apply("final_answer", { content: "结论" }); + stream.apply("done", {}); + expect(run.status).toBe("completed"); + expect(run.pendingPermissions.size).toBe(0); + expect(run.messages.at(-1)).toMatchObject({ id: "answer", content: "结论", permissions: [{ status: "approved_once" }], progress: [{ status: "completed" }, { status: "completed" }] }); + const saved = stream.uiMessages(); + expect(saved.map((message) => message.id)).toEqual(["user", "extra", "answer"]); + expect(saved.at(-1)?.parts).toEqual((run.messages.at(-1) as { parts: typeof saved[number]["parts"] }).parts); + expect(saved.at(-1)?.parts.some((part) => part.type === "data-progress")).toBe(!collaboration); + expect(chunks.filter((chunk) => chunk.type === "finish")).toHaveLength(1); +}); + +it.each(["done", "error", "aborted"])("publishes the same %s terminal state to reconnect subscribers and storage", (event) => { + const run: ActiveRun = { + clientSessionId: "s", controller: new AbortController(), status: "running", + messages: createInitialStreamingMessages([], "查询"), pendingPermissions: new Map(), pendingQuestions: new Map(), subscribers: new Set(), + }; + const received: Array<{ event: string; data: Record }> = []; + run.subscribers.add({ write: (event, data) => received.push({ event, data }), close() {} }); + const stream = createChatRunAccumulator({ run, messages: [], messageId: "answer", sessionId: "s", + collaboration: false, write() {}, onStatus() {} }); + publishRunEvent(run, "progress", { id: "tool", title: "正在处理", status: "running", phase: "tool" }); + publishRunEvent(run, "token", { content: "部分结果" }); + publishRunEvent(run, event, { session_id: "s", message: "执行已停止" }); + const canonical = stream.uiMessages().at(-1)?.parts.find((part) => part.type === "data-run_status"); + expect(received.at(-2)).toMatchObject({ event: "run_status", data: { status: event === "done" ? "completed" : event } }); + expect(canonical).toMatchObject({ data: { status: received.at(-2)?.data.status } }); +}); diff --git a/tests/routes/permissionPersistence.test.ts b/tests/routes/permissionPersistence.test.ts new file mode 100644 index 0000000..226b13b --- /dev/null +++ b/tests/routes/permissionPersistence.test.ts @@ -0,0 +1,58 @@ +import { createChatRunAccumulator } from "../../src/routes/chat-run-accumulator.js"; +import { expect, it } from "bun:test"; +import express from "express"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { getLocalAgentContext } from "../../src/context/localContext.js"; +import { registerChatInteractionRoutes } from "../../src/routes/chatInteractionRoutes.js"; +import type { ActiveRun } from "../../src/routes/chatUiState.js"; +import type { OpencodeRuntimeAdapter } from "../../src/runtime/opencode.js"; +import { SessionMetadataStore } from "../../src/sessions/metadataStore.js"; +import { SessionUiStateStore } from "../../src/sessions/uiStateStore.js"; +import { SessionWorkspaceStore } from "../../src/sessions/workspace-store.js"; + +it.each([false, true])("persists permission replies in the owner's workspace (native failure=%s)", async (nativeFailure) => { + const root = await mkdtemp(join(tmpdir(), "permission-persistence-")); + const workspaces = new SessionWorkspaceStore(join(root, "workspaces")); + const sessions = new SessionMetadataStore(join(root, "metadata"), workspaces); + const states = new SessionUiStateStore(join(root, "unused"), workspaces); + const context = { ...getLocalAgentContext(), sessionId: "permission-session" }; + await sessions.ensure(context); + const messages = [{ id: "answer", role: "assistant", permissions: [] }]; + await states.write(context, { sessionId: context.sessionId, isTitleManuallyEdited: true, messages }); + let calls = 0; + const runtime = { replyPermission: async () => { calls++; if (nativeFailure) throw new Error("native unavailable"); } } as unknown as OpencodeRuntimeAdapter; + const run: ActiveRun = { + clientSessionId: context.sessionId, controller: new AbortController(), messages, + pendingPermissions: new Map([["permission", { request_id: "permission", session_id: context.sessionId, permission: "shell", patterns: [], always: [], created_at: Date.now() }]]), + pendingQuestions: new Map(), subscribers: new Set(), status: "running", + }; + const accumulator = createChatRunAccumulator({ run, messages: [], messageId: "answer", sessionId: context.sessionId, + collaboration: false, write() {}, onStatus() {} }); + accumulator.apply("permission_request", run.pendingPermissions.get("permission")! as unknown as Record); + await states.write(context, { sessionId: context.sessionId, isTitleManuallyEdited: true, messages: run.messages }); + const app = express(); app.use(express.json()); + registerChatInteractionRoutes(app, { activeRuns: new Map([[context.sessionId, run]]), runtime, sessionMetadataStore: sessions, sessionUiStateStore: states }); + const server = app.listen(0, "127.0.0.1"); + await new Promise((resolve) => server.once("listening", resolve)); + const url = `http://127.0.0.1:${(server.address() as { port: number }).port}/permission/permission/reply`; + const reply = (headers = {}) => fetch(url, { method: "POST", headers: { "Content-Type": "application/json", ...headers }, body: JSON.stringify({ session_id: context.sessionId, reply: "once" }) }); + try { + expect((await reply({ "X-Agent-Local-User": "other" })).status).toBe(404); + expect(calls).toBe(0); + expect((await reply()).status).toBe(nativeFailure ? 502 : 202); + expect(calls).toBe(1); + expect(run.pendingPermissions.has("permission")).toBe(nativeFailure); + const saved = await states.read(context); + expect(saved?.isTitleManuallyEdited).toBe(true); + const parts = (saved?.messages[0] as { parts: Array<{ type: string; data: Record }> }).parts; + expect(parts.find((part) => part.type === "data-permission_response")?.data).toMatchObject(nativeFailure + ? { request_id: "permission", status: "error", error: "native unavailable" } + : { request_id: "permission", reply: "once" }); + expect(run.messages[0]).toMatchObject({ permissions: [{ requestId: "permission", status: nativeFailure ? "error" : "approved_once" }] }); + } finally { + await new Promise((resolve) => server.close(() => resolve())); + await rm(root, { recursive: true, force: true }); + } +}); diff --git a/tests/routes/question-persistence.test.ts b/tests/routes/question-persistence.test.ts new file mode 100644 index 0000000..2cddacf --- /dev/null +++ b/tests/routes/question-persistence.test.ts @@ -0,0 +1,71 @@ +import { expect, it } from "bun:test"; +import express from "express"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { UIMessage } from "ai"; +import { getLocalAgentContext } from "../../src/context/localContext.js"; +import { createChatRunAccumulator } from "../../src/routes/chat-run-accumulator.js"; +import { registerChatInteractionRoutes } from "../../src/routes/chatInteractionRoutes.js"; +import { createInitialStreamingMessages, type ActiveRun } from "../../src/routes/chatUiState.js"; +import type { OpencodeRuntimeAdapter } from "../../src/runtime/opencode.js"; +import { SessionMetadataStore } from "../../src/sessions/metadataStore.js"; +import { SessionUiStateStore } from "../../src/sessions/uiStateStore.js"; +import { SessionWorkspaceStore } from "../../src/sessions/workspace-store.js"; + +it("keeps historical question invalidation across the next live snapshot", async () => { + const root = await mkdtemp(join(tmpdir(), "question-persistence-")); + const workspaces = new SessionWorkspaceStore(join(root, "workspaces")); + const sessions = new SessionMetadataStore(join(root, "metadata"), workspaces); + const states = new SessionUiStateStore(join(root, "unused"), workspaces); + const context = { ...getLocalAgentContext(), sessionId: "history-session" }; + await sessions.ensure(context); + const history: UIMessage[] = [{ id: "old-answer", role: "assistant", parts: [ + { type: "data-question_request", data: { request_id: "old-question", session_id: context.sessionId, + questions: [{ question: "继续吗？", options: [] }] } }, + { type: "data-final_answer", data: { content: "此前的结果" } }, + ] }, { id: "user", role: "user", parts: [{ type: "text", text: "新一轮" }] }]; + const run: ActiveRun = { + clientSessionId: context.sessionId, controller: new AbortController(), status: "running", + messages: createInitialStreamingMessages(history, "新一轮"), + pendingPermissions: new Map(), pendingQuestions: new Map(), subscribers: new Set(), + }; + const stream = createChatRunAccumulator({ run, messages: history, messageId: "current-answer", + sessionId: context.sessionId, collaboration: false, write() {}, onStatus() {} }); + stream.apply("token", { content: "正在核查" }); + await states.write(context, { sessionId: context.sessionId, isTitleManuallyEdited: true, messages: stream.uiMessages() }); + let nativeReads = 0; + const runtime = { getQuestionStatus: async () => { nativeReads++; return "missing"; } } as unknown as OpencodeRuntimeAdapter; + const app = express(); + registerChatInteractionRoutes(app, { activeRuns: new Map([[context.sessionId, run]]), runtime, + sessionMetadataStore: sessions, sessionUiStateStore: states }); + const server = app.listen(0, "127.0.0.1"); + await new Promise((resolve) => server.once("listening", resolve)); + const url = `http://127.0.0.1:${(server.address() as { port: number }).port}/session/${context.sessionId}/questions/old-question`; + const resolved = (messages: unknown[]) => { + const message = messages[0] as UIMessage; + expect(message.id).toBe("old-answer"); + const part = message.parts.find((part) => part.type === "data-question_response"); + expect(part).toMatchObject({ type: "data-question_response", data: { request_id: "old-question", status: "invalidated" } }); + }; + try { + expect((await fetch(url, { headers: { "X-Agent-Local-User": "other" } })).status).toBe(404); + expect(nativeReads).toBe(0); + const response = await fetch(url); + expect(response.status).toBe(200); + expect(await response.json()).toMatchObject({ status: "invalidated" }); + resolved((await states.read(context))!.messages); + resolved(run.messages); + resolved(stream.uiMessages()); + expect(stream.uiMessages().at(-1)?.parts.some((part) => part.type === "data-question_response")).toBe(false); + stream.apply("token", { content: "，继续查询" }); + await states.write(context, { ...(await states.read(context))!, messages: stream.uiMessages() }); + resolved((await states.read(context))!.messages); + expect((await states.read(context))?.isTitleManuallyEdited).toBe(true); + await fetch(url); + expect(nativeReads).toBe(1); + } finally { + await new Promise((resolve) => server.close(() => resolve())); + await rm(root, { recursive: true, force: true }); + } +}); diff --git a/tests/routes/responses.test.ts b/tests/routes/responses.test.ts index 6427f1e..2f6893e 100644 --- a/tests/routes/responses.test.ts +++ b/tests/routes/responses.test.ts @@ -296,6 +296,11 @@ it("marks an explicitly aborted response incomplete and preserves scoped history const text = await response.text(); expect(text).toContain("event: response.incomplete"); expect(text).not.toContain("event: response.completed"); + const context = { actorKey: toActorKey("product:owner"), projectKey: toProjectKey(`product:${project}`), sessionId: "owned" }; + const saved = (await states.read(context))!.messages as Array<{ parts: Array<{ type: string; data?: { status?: string } }> }>; + expect(saved.at(-1)?.parts.find((part) => part.type === "data-run_status")?.data?.status).toBe("aborted"); + const loaded = await http(`${base}/api/v1/session/owned`, { headers: { Authorization: "Bearer owner", "X-Project-Id": project } }); + expect((await loaded.json()).messages).toEqual(saved); }); it("continues a paged conversation from the complete saved history and preserves client message IDs", async () => { @@ -316,3 +321,28 @@ it("continues a paged conversation from the complete saved history and preserves expect((await call({ input: "重复提交", conversation: "owned", tjwater: { client_message_id: "client-turn-a" } })).status).toBe(409); expect(prompted).toHaveLength(1); }); + +it("uses the same canonical live messages for session reads and older pages", async () => { + gate = new Promise((resolve) => { release = resolve; }); + const context = { actorKey: toActorKey("product:owner"), projectKey: toProjectKey(`product:${project}`), sessionId: "owned" }; + const history = Array.from({ length: 5 }, (_, index) => ({ id: `old-${index}`, role: index % 2 ? "assistant" : "user", + parts: [{ type: "text", text: `历史 ${index}` }] as unknown[] })); + history[1].parts.push({ type: "data-question_request", data: { request_id: "old-question", session_id: "owned", questions: [] } }); + await states.write(context, { sessionId: "owned", messages: history }); + const response = await call({ input: "新一轮", conversation: "owned", stream: true, tjwater: { client_message_id: "live-user" } }); + const headers = { Authorization: "Bearer owner", "X-Project-Id": project }; + const live = await http(`${base}/api/v1/session/owned`, { headers }); + const snapshot = await live.json(); + expect(snapshot.is_streaming).toBe(true); + expect(snapshot.messages).toHaveLength(7); + expect(snapshot.messages.slice(0, 5)).toEqual(history); + expect(snapshot.messages[5].id).toBe("live-user"); + const recent = await http(`${base}/api/v1/session/owned?message_limit=2`, { headers }); + const page = await recent.json(); + expect(page.messages).toEqual(snapshot.messages.slice(-2)); + const older = await http(`${base}/api/v1/session/owned/messages?limit=40&before=${encodeURIComponent(page.message_page.before_cursor)}`, { headers }); + expect(older.status).toBe(200); + expect((await older.json()).messages).toEqual(history); + release(); + await response.text(); +}); diff --git a/tests/routes/server-contract-sync.test.ts b/tests/routes/server-contract-sync.test.ts new file mode 100644 index 0000000..d898950 --- /dev/null +++ b/tests/routes/server-contract-sync.test.ts @@ -0,0 +1,39 @@ +import { expect, it } from "bun:test"; +import { createHash } from "node:crypto"; +import { copyFile, mkdir, mkdtemp, readFile, readdir, rm, symlink, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { astToString, COMMENT_HEADER, default as openapiTS } from "openapi-typescript"; + +it.each(["valid", "invalid-schema", "invalid-digest"])("synchronizes the complete artifact group or preserves it (%s)", async (scenario) => { + const root = await mkdtemp(join(tmpdir(), "server-contract-sync-")); + try { + for (const directory of ["scripts", "openapi", "src/generated", "upstream"]) await mkdir(join(root, directory), { recursive: true }); + await copyFile(new URL("../../scripts/server-contract.ts", import.meta.url), join(root, "scripts/server-contract.ts")); + await symlink(new URL("../../node_modules", import.meta.url).pathname, join(root, "node_modules"), "dir"); + const targets = ["openapi/openapi.json", "openapi/server-manifest.json", "src/generated/server-api.d.ts"]; + for (const target of targets) await writeFile(join(root, target), `original:${target}`); + const schema = { openapi: scenario === "invalid-schema" ? "9.9.9" : "3.1.0", info: { title: "Fixture", version: "1" }, paths: {} }; + const raw = JSON.stringify(schema); + const manifest = { contracts: { server: { sha256: scenario === "invalid-digest" ? "incorrect" : createHash("sha256").update(raw).digest("hex") } } }; + await writeFile(join(root, "upstream/schema.json"), raw); + await writeFile(join(root, "upstream/manifest.json"), JSON.stringify(manifest)); + const child = Bun.spawn([process.execPath, "scripts/server-contract.ts", "--sync", "--upstream", "upstream/schema.json"], { + cwd: root, stdout: "pipe", stderr: "pipe", + }); + const [code, stderr] = await Promise.all([child.exited, new Response(child.stderr).text()]); + if (scenario === "valid") { + expect(code).toBe(0); + expect(await readFile(join(root, targets[0]), "utf8")).toBe(raw); + expect(JSON.parse(await readFile(join(root, targets[1]), "utf8"))).toEqual(manifest); + expect(await readFile(join(root, targets[2]), "utf8")).toBe(COMMENT_HEADER + astToString(await openapiTS(schema))); + } else { + expect(code).not.toBe(0); + expect(stderr).toContain(scenario === "invalid-schema" ? "Unsupported OpenAPI version" : "contract digest differs"); + for (const target of targets) expect(await readFile(join(root, target), "utf8")).toBe(`original:${target}`); + } + expect((await readdir(join(root, "openapi"))).sort()).toEqual(["openapi.json", "server-manifest.json"]); + } finally { + await rm(root, { recursive: true, force: true }); + } +}); diff --git a/tests/routes/task-command-bridge.test.ts b/tests/routes/task-command-bridge.test.ts new file mode 100644 index 0000000..aa2d812 --- /dev/null +++ b/tests/routes/task-command-bridge.test.ts @@ -0,0 +1,45 @@ +import { expect, it } from "bun:test"; +import express from "express"; +import type { OperationalTaskClient } from "../../src/product/operational-task-client.js"; +import { registerTaskCommandBridge } from "../../src/routes/task-command-bridge.js"; +import { removeRuntimeSessionContext, setRuntimeSessionContext } from "../../src/runtime/sessionContext.js"; + +it("preserves command-specific session purposes and trusted bridge authentication", async () => { + const calls: unknown[][] = []; + const client = { + requestDataAccess: async (...args: unknown[]) => { calls.push(args); return {}; }, + proposeAction: async (...args: unknown[]) => { calls.push(args); throw Object.assign(new Error("conflict"), { status: 409 }); }, + addHumanEvidence: async (...args: unknown[]) => { calls.push(args); return {}; }, + } as unknown as Pick; + for (const purpose of ["task_execution", "task_collaboration"] as const) setRuntimeSessionContext({ + sessionId: purpose, clientSessionId: `client-${purpose}`, taskId: "task", purpose, + actorKey: "actor", projectKey: "project", traceId: "test", + }); + const app = express(); app.use(express.json()); + registerTaskCommandBridge(app, { internalToken: "test-token", client }); + const server = app.listen(0, "127.0.0.1"); + await new Promise((resolve) => server.once("listening", resolve)); + const post = (kind: string, session: string, data: object, headers = {}) => fetch(`http://127.0.0.1:${(server.address() as { port: number }).port}/internal/operational-task/${kind}`, { + method: "POST", headers: { "Content-Type": "application/json", "x-agent-internal-token": "test-token", ...headers }, body: JSON.stringify({ session_id: session, ...data }), + }); + const requestId = "60000000-0000-4000-8000-000000000001"; + const access = { request: { requestId, nodeId: "read", scopes: ["scada_devices"], reason: "查询" } }; + const evidence = { command: { request_id: requestId, evidence: { evidence_id: "e1", kind: "human_input", summary: "已复核", source_label: "用户", observed_at: "2026-11-09T00:00:00Z" } } }; + try { + expect((await post("access-requests", "task_execution", access, { Origin: "http://browser.test" })).status).toBe(403); + expect((await post("access-requests", "task_execution", access, { "x-agent-internal-token": "wrong" })).status).toBe(403); + expect((await post("access-requests", "task_collaboration", access)).status).toBe(403); + expect((await post("human-evidence", "task_execution", evidence)).status).toBe(403); + expect(calls).toHaveLength(0); + expect((await post("access-requests", "task_execution", access)).status).toBe(200); + expect(calls[0]).toEqual(["task_execution", "task", { ...access.request, sourceSessionId: "client-task_execution" }]); + expect((await post("human-evidence", "task_collaboration", evidence)).status).toBe(200); + const action = { action: { request_id: requestId, reason: "核验", kind: "create_work_order", payload: { title: "采样", description: "核验", work_order_type: "general" } } }; + expect((await post("actions", "task_execution", action)).status).toBe(409); + expect((await post("actions", "task_collaboration", action)).status).toBe(409); + expect((await post("actions", "unknown", action)).status).toBe(403); + } finally { + removeRuntimeSessionContext("task_execution"); removeRuntimeSessionContext("task_collaboration"); + await new Promise((resolve) => server.close(() => resolve())); + } +}); diff --git a/tests/sessions/uiStateStore.test.ts b/tests/sessions/uiStateStore.test.ts index 70efdf7..666872c 100644 --- a/tests/sessions/uiStateStore.test.ts +++ b/tests/sessions/uiStateStore.test.ts @@ -10,10 +10,10 @@ it("restores plan history independently of chat messages after reconnecting", as const store = new SessionUiStateStore(directory); await store.initialize(); const executionPlans = [{ revision: 1, nodes: [{ id: "sampling", status: "waiting" }] }]; - await store.write({ sessionId: "session-1" }, { sessionId: "session-1", messages: [], executionPlans }); - const reloaded = await new SessionUiStateStore(directory).read({ sessionId: "session-1" }); + await store.write({ sessionId: "session-1", actorKey: "actor-test", projectKey: "project-test" }, { sessionId: "session-1", messages: [], executionPlans }); + const reloaded = await new SessionUiStateStore(directory).read({ sessionId: "session-1", actorKey: "actor-test", projectKey: "project-test" }); expect(reloaded?.executionPlans).toEqual(executionPlans); - expect(await store.read({ sessionId: "other-session" })).toBeNull(); + expect(await store.read({ sessionId: "other-session", actorKey: "actor-test", projectKey: "project-test" })).toBeNull(); } finally { await rm(directory, { recursive: true, force: true }); }