From 5ee93329a4ca5f9cd8800259590facde160d9af5 Mon Sep 17 00:00:00 2001 From: Huarch Date: Mon, 9 Nov 2026 18:49:32 +0800 Subject: [PATCH] refactor: streamline query requests and session storage --- src/product/mcp-query-client.ts | 6 ++-- src/routes/chatInteractionRoutes.ts | 5 +-- src/runtime/opencode.ts | 13 ++++---- src/server.ts | 6 ++-- src/sessions/transcriptStore.ts | 32 +++++------------- src/sessions/uiStateStore.ts | 30 ++++------------- tests/chat/answer-ui.test.ts | 4 +-- tests/integration/mcp-query-smoke.ts | 1 + tests/product/mcpQueryClient.test.ts | 18 ++++++++++ tests/routes/answerViewStateRoutes.test.ts | 2 +- tests/routes/chat-topics.test.ts | 2 +- tests/routes/credentialRefreshRoute.test.ts | 3 +- tests/routes/hostQuestionStream.test.ts | 2 +- tests/routes/legacyQuestionRoutes.test.ts | 33 +++++++++++++++++-- tests/routes/permissionPersistence.test.ts | 2 +- tests/routes/question-persistence.test.ts | 4 +-- tests/routes/responses.test.ts | 2 +- tests/routes/sessionIsolation.test.ts | 2 +- .../task-collaboration-sessions.test.ts | 2 +- tests/runtime/hostQuestion.test.ts | 6 ++-- tests/sessions/answerViewState.test.ts | 2 +- tests/sessions/learningLifecycle.test.ts | 2 +- tests/sessions/transcriptStore.test.ts | 4 +-- tests/sessions/uiStateStore.test.ts | 6 ++-- tests/sessions/workspaceStore.test.ts | 22 ++++++------- 25 files changed, 111 insertions(+), 100 deletions(-) diff --git a/src/product/mcp-query-client.ts b/src/product/mcp-query-client.ts index 5f1b460..be4c697 100644 --- a/src/product/mcp-query-client.ts +++ b/src/product/mcp-query-client.ts @@ -203,7 +203,8 @@ export class McpQueryClient { if (!owner.context.projectId || owner.context.projectKey.startsWith("personal-") || !owner.context.purpose || !["ordinary_chat", "task_execution", "task_collaboration"].includes(owner.context.purpose)) throw error(403, "Project query context required"); // No model-supplied identity, scope, bearer or Server URL can pass this boundary. const schema = this.catalog().find((tool) => tool.name === name)!.inputSchema; - const budget = queryTimeouts(this.tools.find((tool) => tool.name === name)!, args); + const definition = this.tools.find((tool) => tool.name === name)!; + const budget = queryTimeouts(definition, args); const reserved = new Set(["context_handle", "authorization", "headers", "project_id", "projectId", "network", "url", "server_url", "user_id", "subject", "client_id", "purpose", "native_session_id", "parent_session_id", "task_id", "scope", "scopes"]); if (Object.keys(args).some((key) => reserved.has(key) || !Object.hasOwn(schema.properties ?? {}, key))) throw error(400, "Unknown query parameter"); const token = await this.userToken(owner.runtimeSessionId, owner.context.projectId, signal); @@ -223,8 +224,7 @@ export class McpQueryClient { const handle = body.context_handle; try { return await this.withClient(token.token, project, signal, async (client) => { - await client.listTools(undefined, { signal, timeout: 5000 }); - return client.callTool({ name, arguments: { ...args, ...(this.tools.find((tool) => tool.name === name)!.inputSchema.properties?.timeout_seconds ? { timeout_seconds: budget.serverMs / 1000 } : {}), context_handle: handle }, _meta: { "io.tjwater/callId": callId } }, { signal, timeout: budget.callMs }); + return client.callTool({ name, arguments: { ...args, ...(definition.inputSchema.properties?.timeout_seconds ? { timeout_seconds: budget.serverMs / 1000 } : {}), context_handle: handle }, _meta: { "io.tjwater/callId": callId } }, { signal, timeout: budget.callMs, toolDefinition: definition }); }, budget.callMs); } finally { // Expiry remains authoritative if transport cancellation prevents revocation. diff --git a/src/routes/chatInteractionRoutes.ts b/src/routes/chatInteractionRoutes.ts index cae279d..06fb9d0 100644 --- a/src/routes/chatInteractionRoutes.ts +++ b/src/routes/chatInteractionRoutes.ts @@ -196,7 +196,8 @@ export const registerChatInteractionRoutes = ( } } // A host-confirmed reply remains authoritative after native forms are pruned. - const nativeStatus = confirmedStatus ? undefined : await runtime.getQuestionStatus({ sessionId: session.sessionId, requestId: canonicalId }); + const form = confirmedStatus ? undefined : await runtime.getQuestion({ sessionId: session.sessionId, requestId: canonicalId }); + const nativeStatus = form?.state.status ?? "missing"; const status = confirmedStatus ?? (nativeStatus === "cancelled" ? "rejected" : nativeStatus === "missing" ? "invalidated" : nativeStatus!); const persist = async (nextStatus: string, answers?: string[][]) => { const response = { @@ -215,7 +216,7 @@ export const registerChatInteractionRoutes = ( return; } const answers = action === "reply" ? questionReplyPayloadSchema.parse(req.body).answers : undefined; - if (action === "reply") await runtime.replyQuestion({ sessionId: session.sessionId, requestId: canonicalId, answers: answers ?? [] }); + if (action === "reply") await runtime.replyQuestion({ sessionId: session.sessionId, requestId: canonicalId, answers: answers ?? [], form: form! }); else await runtime.rejectQuestion({ sessionId: session.sessionId, requestId: canonicalId }); await persist(action === "reply" ? "answered" : "rejected", answers); if (run && run.status !== "running" && !run.pendingPermissions.size && !run.pendingQuestions.size) activeRuns.delete(session.sessionId); diff --git a/src/runtime/opencode.ts b/src/runtime/opencode.ts index 43f7f63..5ff550b 100644 --- a/src/runtime/opencode.ts +++ b/src/runtime/opencode.ts @@ -2,7 +2,7 @@ import { LOCAL_QUERY_TOOL_IDS, PRODUCT_MCP_TOOL_IDS, PRODUCT_MCP_WRITE_TOOL_IDS, import { assertWebSearchReady } from "./web-search-check.js"; import { allowedQueryTools } from "./query-mcp-policy.js"; import { setTimeout as delayAbortable } from "node:timers/promises"; -import { isMessageNotFoundError, isSessionNotFoundError, type OpenCodeClient as OpencodeClient } from "@opencode/client"; +import { isMessageNotFoundError, isSessionNotFoundError, type FormDetail, type OpenCodeClient as OpencodeClient } from "@opencode/client"; import { createRuntimeClient, startOpencode2 } from "./opencode2-server.js"; import { formAnswers, normalizeRuntimeMessage, Opencode2EventNormalizer } from "./opencode2-events.js"; import { existsSync, readFileSync } from "node:fs"; @@ -726,19 +726,18 @@ export class OpencodeRuntimeAdapter { return client.permission.reply({ sessionID: options.sessionId, requestID: options.requestId, decision: options.reply, message: options.message }); } - async replyQuestion(options: { requestId: string; sessionId?: string; answers: QuestionAnswers }) { + async replyQuestion(options: { requestId: string; sessionId: string; answers: QuestionAnswers; form: FormDetail }) { if (!options.sessionId) throw new Error("OpenCode 2 form replies require sessionId"); const client = await this.ensureClient(); - const form = await client.session.form.get({ sessionID: options.sessionId, formID: options.requestId }); - return client.session.form.reply({ sessionID: options.sessionId, formID: options.requestId, answer: formAnswers(form, options.answers) }); + return client.session.form.reply({ sessionID: options.sessionId, formID: options.requestId, answer: formAnswers(options.form, options.answers) }); } - async getQuestionStatus(options: { requestId: string; sessionId: string }): Promise<"pending" | "answered" | "cancelled" | "missing"> { + async getQuestion(options: { requestId: string; sessionId: string }): Promise { const client = await this.ensureClient(); - try { return (await client.session.form.get({ sessionID: options.sessionId, formID: options.requestId })).state.status; } + try { return await client.session.form.get({ sessionID: options.sessionId, formID: options.requestId }); } catch (error) { const tag = error && typeof error === "object" && "_tag" in error ? error._tag : undefined; - if (tag === "FormNotFoundError") return "missing"; + if (tag === "FormNotFoundError") return null; throw error; } } diff --git a/src/server.ts b/src/server.ts index f58b532..7af2fe4 100644 --- a/src/server.ts +++ b/src/server.ts @@ -48,9 +48,9 @@ const sessionWorkspaces = new SessionWorkspaceStore(config.CONVERSATION_WORKSPAC const sessionMetadataStore = new SessionMetadataStore(undefined, sessionWorkspaces); const answerViewStateStore = new AnswerViewStateStore(sessionWorkspaces); const interactionRequestStore = new InteractionRequestStore(sessionWorkspaces); -const sessionUiStateStore = new SessionUiStateStore(undefined, sessionWorkspaces); +const sessionUiStateStore = new SessionUiStateStore(sessionWorkspaces); const memoryStore = new MemoryStore(); -const sessionTranscriptStore = new SessionTranscriptStore(undefined, sessionWorkspaces); +const sessionTranscriptStore = new SessionTranscriptStore(sessionWorkspaces); const learningOrchestrator = new LearningOrchestrator( opencodeRuntime, memoryStore, @@ -169,10 +169,8 @@ const bootstrap = async () => { logger.info(migrated, "session records migrated to owned workspaces; legacy sources archived"); } await Promise.all([ - sessionUiStateStore.initialize(), learningOrchestrator.initialize(), memoryStore.initialize(), - sessionTranscriptStore.initialize(), ]); }; diff --git a/src/sessions/transcriptStore.ts b/src/sessions/transcriptStore.ts index e7e9f39..d473e73 100644 --- a/src/sessions/transcriptStore.ts +++ b/src/sessions/transcriptStore.ts @@ -1,10 +1,5 @@ -import { join } from "node:path"; - import { config } from "../config.js"; import { - atomicWriteJson, - ensureDirectory, - listJsonFiles, readJsonFile, toStableId, } from "../utils/fileStore.js"; @@ -50,11 +45,7 @@ const DEFAULT_SEARCH_MAX_QUERY_CHARS = 240; export class SessionTranscriptStore { private readonly writeQueues = new Map>(); - constructor(private readonly baseDir = config.SESSION_TRANSCRIPT_STORAGE_DIR, private readonly workspaces?: SessionWorkspaceStore) {} - - async initialize() { - if (!this.workspaces) await ensureDirectory(this.baseDir); - } + constructor(private readonly workspaces: SessionWorkspaceStore) {} async appendTurn( context: SessionTranscriptContext, @@ -66,7 +57,7 @@ export class SessionTranscriptStore { ) { const key = this.filePath(context); return this.serializeWrite(key, async () => { - if (this.workspaces) await this.workspaces.ensure(this.workspaceContext(context)); + await this.workspaces.ensure(this.workspaceContext(context)); // 同一会话的多次写入串行化，防止流式结束和 UI 同步同时写同一个 transcript。 const transcript = (await this.readTranscript(context)) ?? { actorKey: context.actorKey, @@ -148,10 +139,8 @@ export class SessionTranscriptStore { turns: projectTurnsForFork(sourceTranscript?.turns ?? [], keepMessageCount), updatedAt: timestamp, }; - if (this.workspaces) { - await this.workspaces.ensure(this.workspaceContext(targetContext)); - await this.workspaces.validate(this.workspaceContext(targetContext), "transcript.json"); - } + await this.workspaces.ensure(this.workspaceContext(targetContext)); + await this.workspaces.validate(this.workspaceContext(targetContext), "transcript.json"); await this.persist(targetContext, nextTranscript); return nextTranscript; } @@ -168,7 +157,7 @@ export class SessionTranscriptStore { } const queryTokens = normalizedQuery.split(/\s+/).filter(Boolean); const hits: SessionSearchHit[] = []; - const files = this.workspaces ? await this.workspaces.list(context, "transcript.json") : await listJsonFiles(this.baseDir); + const files = await this.workspaces.list(context, "transcript.json"); for (const file of files) { const transcript = await readJsonFile(file); if (!transcript) { @@ -205,7 +194,7 @@ export class SessionTranscriptStore { } private async readTranscript(context: SessionTranscriptContext) { - if (this.workspaces && !await this.workspaces.validate(this.workspaceContext(context), "transcript.json")) return null; + if (!await this.workspaces.validate(this.workspaceContext(context), "transcript.json")) return null; const record = await readJsonFile(this.filePath(context)); if (record && (record.actorKey !== context.actorKey || record.projectKey !== context.projectKey || (record.clientSessionId || record.sessionId) !== (context.clientSessionId || context.sessionId))) throw new Error("chat transcript ownership mismatch"); @@ -213,16 +202,11 @@ export class SessionTranscriptStore { } private filePath(context: SessionTranscriptContext) { - if (this.workspaces) return this.workspaces.path(this.workspaceContext(context), "transcript.json"); - return join( - this.baseDir, - `${context.actorKey}__${context.projectKey}__${context.sessionId}.json`, - ); + return this.workspaces.path(this.workspaceContext(context), "transcript.json"); } private async persist(context: SessionTranscriptContext, transcript: SessionTranscriptRecord) { - if (this.workspaces) await this.workspaces.writeRecord(this.workspaceContext(context), "transcript.json", transcript); - else await atomicWriteJson(this.filePath(context), transcript); + await this.workspaces.writeRecord(this.workspaceContext(context), "transcript.json", transcript); } private workspaceContext(context: SessionTranscriptContext) { diff --git a/src/sessions/uiStateStore.ts b/src/sessions/uiStateStore.ts index 7fce1b1..2dbabd3 100644 --- a/src/sessions/uiStateStore.ts +++ b/src/sessions/uiStateStore.ts @@ -1,14 +1,9 @@ import type { SessionWorkspaceContext } from "../runtime/conversationWorkspace.js"; -import { join } from "node:path"; -import { config } from "../config.js"; import { type SessionWorkspaceStore } from "./workspace-store.js"; import { - atomicWriteJson, - ensureDirectory, readJsonFile, removeFileIfExists, - slugify, } from "../utils/fileStore.js"; export type SessionUiStateRecord = { @@ -21,49 +16,36 @@ export type SessionUiStateRecord = { type SessionUiStateContext = SessionWorkspaceContext; export class SessionUiStateStore { - constructor(private readonly baseDir = config.SESSION_UI_STATE_STORAGE_DIR, private readonly workspaces?: SessionWorkspaceStore) {} - - async initialize() { - if (!this.workspaces) await ensureDirectory(this.baseDir); - } + constructor(private readonly workspaces: SessionWorkspaceStore) {} async read(context: SessionUiStateContext) { - if (this.workspaces && !await this.workspaces.validate(this.workspaceContext(context), "messages.json")) return null; + if (!await this.workspaces.validate(this.workspaceContext(context), "messages.json")) return null; return await readJsonFile(this.filePath(context)); } async write(context: SessionUiStateContext, state: SessionUiStateRecord) { if (state.sessionId !== context.sessionId) throw new Error("chat state session mismatch"); - if (this.workspaces) { - await this.workspaces.ensure(this.workspaceContext(context)); - await this.workspaces.validate(this.workspaceContext(context), "messages.json"); - } - if (this.workspaces) await this.workspaces.writeRecord(this.workspaceContext(context), "messages.json", state); - else await atomicWriteJson(this.filePath(context), state); + await this.workspaces.writeRecord(this.workspaceContext(context), "messages.json", state); return state; } async update(context: SessionUiStateContext, change: (state: SessionUiStateRecord) => SessionUiStateRecord) { - if (this.workspaces) return this.workspaces.updateRecord(this.workspaceContext(context), "messages.json", async ({ current }) => { + return this.workspaces.updateRecord(this.workspaceContext(context), "messages.json", async ({ current }) => { const state = current as SessionUiStateRecord | null; if (!state || state.sessionId !== context.sessionId || !Array.isArray(state.messages)) throw new Error("chat state unavailable"); const next = change(state); if (next.sessionId !== context.sessionId) throw new Error("chat state session mismatch"); return { result: next, value: next }; }); - const state = await this.read(context); - if (!state) throw new Error("chat state unavailable"); - return this.write(context, change(state)); } async remove(context: SessionUiStateContext) { - if (this.workspaces && !await this.workspaces.validate(this.workspaceContext(context), "messages.json")) return; + if (!await this.workspaces.validate(this.workspaceContext(context), "messages.json")) return; await removeFileIfExists(this.filePath(context)); } private filePath(context: SessionUiStateContext) { - if (this.workspaces) return this.workspaces.path(this.workspaceContext(context), "messages.json"); - return join(this.baseDir, `${slugify(context.sessionId)}.json`); + return this.workspaces.path(this.workspaceContext(context), "messages.json"); } private workspaceContext(context: SessionUiStateContext) { diff --git a/tests/chat/answer-ui.test.ts b/tests/chat/answer-ui.test.ts index 5c825cd..4177f7a 100644 --- a/tests/chat/answer-ui.test.ts +++ b/tests/chat/answer-ui.test.ts @@ -99,11 +99,11 @@ describe("locally rendered answers", () => { try { const owner = { actorKey: "actor-answer", projectKey: "project-answer", sessionId: "answer-session" }; const workspaces = new SessionWorkspaceStore(join(root, "workspaces")); - const store = new SessionUiStateStore(join(root, "legacy"), workspaces); + const store = new SessionUiStateStore(workspaces); const parts: UIMessage["parts"] = [{ type: "text", text: "演练：" }, { type: "data-answer_ui", id: "pressure", data: parseAnswerUiInput(input) }, { type: "text", text: "仍待核验。" }]; const messages = [{ id: "a", role: "assistant", parts }]; await store.write(owner, { sessionId: owner.sessionId, messages }); - const restored = new SessionUiStateStore(join(root, "legacy"), new SessionWorkspaceStore(workspaces.root)); + const restored = new SessionUiStateStore(new SessionWorkspaceStore(workspaces.root)); expect((await restored.read(owner))?.messages).toEqual(messages); expect(await restored.read({ ...owner, actorKey: "actor-other" })).toBeNull(); expect(await restored.read({ ...owner, projectKey: "project-other" })).toBeNull(); diff --git a/tests/integration/mcp-query-smoke.ts b/tests/integration/mcp-query-smoke.ts index 207bab0..0d154ca 100644 --- a/tests/integration/mcp-query-smoke.ts +++ b/tests/integration/mcp-query-smoke.ts @@ -68,6 +68,7 @@ try { if (response.isError) throw new Error(`Extended tool failed through merged MCP: ${name}`); } if (!requests.includes("server/discover") || !requests.includes("tools/call") || requests.includes("initialize") || requests.includes("GET")) throw new Error("Transport used unexpected protocol requests"); + if (requests.filter((method) => method === "tools/list").length !== 1) throw new Error("Business queries fetched the catalog again after startup"); if (grants.join(",") !== "client_credentials,urn:ietf:params:oauth:grant-type:token-exchange") throw new Error("OAuth grants were not validated"); console.log(`PASS: local MCP ${QUERY_MCP_TOOL_IDS.length}-tool discovery, real Server schemas, OAuth discovery, resource-bound grants, host binding, extended query and revocation over MCP HTTP`); } finally { diff --git a/tests/product/mcpQueryClient.test.ts b/tests/product/mcpQueryClient.test.ts index 1496d22..5a05cb6 100644 --- a/tests/product/mcpQueryClient.test.ts +++ b/tests/product/mcpQueryClient.test.ts @@ -26,6 +26,24 @@ const resourceToken = jwt("resource"); const token = () => Response.json({ access_token: resourceToken, token_type: "Bearer", expires_in: 300 }); function credential(id: string) { sessions.push(id); bindSessionCredential(id, { authorization: `Bearer ${jwt(id)}`, projectId: "project" }); } +it("uses cached tool definitions without fetching the catalog on each query", async () => { + const id = "cached-tool-definition"; credential(id); + const client = fixture((async (url, init) => { + if (String(url).includes("/token")) return token(); + return init?.method === "POST" + ? Response.json({ context_handle: "x".repeat(43) }) + : new Response(null, { status: 204 }); + }) as McpHttpFetch); + const definition = { ...catalog[0], outputSchema: { type: "object", properties: { ok: { type: "boolean" } } } }; + Object.assign(client, { tools: [definition], withClient: async (_token: string, _project: string, _signal: AbortSignal, run: (client: unknown) => Promise) => run({ + callTool: async (_params: unknown, options: { toolDefinition: unknown }) => { + expect(options.toolDefinition).toBe(definition); + return { content: [], structuredContent: { ok: true } }; + }, + }) }); + expect((await client.query(id, owner(id), definition.name, {}, "call")).structuredContent).toEqual({ ok: true }); +}); + it("owns simulation HTTP budgets and rejects model overrides before OAuth", async () => { const id = "mcp-calculation-budget"; credential(id); let requests = 0, transportTimeout = 0, sdkTimeout = 0; diff --git a/tests/routes/answerViewStateRoutes.test.ts b/tests/routes/answerViewStateRoutes.test.ts index 78f46d7..51a8a4e 100644 --- a/tests/routes/answerViewStateRoutes.test.ts +++ b/tests/routes/answerViewStateRoutes.test.ts @@ -31,7 +31,7 @@ it("authenticates state reads and writes against the persisted session owner and const context = getLocalAgentContext(ownerRequest); const sessionId = "state-session"; await sessions.ensure({ ...context, sessionId }); - await new SessionUiStateStore(join(root, "unused"), workspaces).write({ ...context, sessionId }, { sessionId, messages: [{ id: "assistant", role: "assistant", parts: [{ type: "data-answer_ui", data: { version: "answer-ui@1", block_id: "block", fallback_text: "fallback", spec: { root: "evidence", elements: { evidence: { type: "Evidence", props: { title: "Evidence", summary: "Summary", source: "source", evidence_id: null, observed_at: null, kind: "observed" }, children: [] } } } } }] }] }); + await new SessionUiStateStore(workspaces).write({ ...context, sessionId }, { sessionId, messages: [{ id: "assistant", role: "assistant", parts: [{ type: "data-answer_ui", data: { version: "answer-ui@1", block_id: "block", fallback_text: "fallback", spec: { root: "evidence", elements: { evidence: { type: "Evidence", props: { title: "Evidence", summary: "Summary", source: "source", evidence_id: null, observed_at: null, kind: "observed" }, children: [] } } } } }] }] }); const app = express(); app.use(express.json({ limit: "2mb" })); app.use(async (req, res, next) => { diff --git a/tests/routes/chat-topics.test.ts b/tests/routes/chat-topics.test.ts index 5204795..78adba5 100644 --- a/tests/routes/chat-topics.test.ts +++ b/tests/routes/chat-topics.test.ts @@ -22,7 +22,7 @@ it("persists owned topics, searches old text across themes and preserves chats w globalThis.fetch = (async (_input, init) => Response.json({ user_id: new Headers(init?.headers).get("authorization") === "Bearer owner" ? "owner" : "other", project_id: new Headers(init?.headers).get("x-project-id"), network: "supply" })) as typeof fetch; const workspaces = new SessionWorkspaceStore(join(root, "workspaces")); const sessions = new SessionMetadataStore(join(root, "metadata"), workspaces); - const states = new SessionUiStateStore(join(root, "ui"), workspaces); + const states = new SessionUiStateStore(workspaces); for (const id of ["a", "b", "unclassified"]) { await sessions.ensure({ ...context, sessionId: id }); await states.write({ ...context, sessionId: id }, { sessionId: id, messages: [ diff --git a/tests/routes/credentialRefreshRoute.test.ts b/tests/routes/credentialRefreshRoute.test.ts index cd16ae2..2168e48 100644 --- a/tests/routes/credentialRefreshRoute.test.ts +++ b/tests/routes/credentialRefreshRoute.test.ts @@ -7,6 +7,7 @@ import { attachTrustedAgentContext } from "../../src/context/localContext.js"; import { registerChatInteractionRoutes } from "../../src/routes/chatInteractionRoutes.js"; import { bindSessionCredential, credentialRefreshCoordinator, releaseSessionCredential } from "../../src/product/credential-refresh.js"; import { SessionMetadataStore } from "../../src/sessions/metadataStore.js"; +import { SessionWorkspaceStore } from "../../src/sessions/workspace-store.js"; import { SessionUiStateStore } from "../../src/sessions/uiStateStore.js"; import { toActorKey, toProjectKey } from "../../src/utils/fileStore.js"; import type { ActiveRun } from "../../src/routes/chatUiState.js"; @@ -28,7 +29,7 @@ describe("credential refresh HTTP route", () => { const router = express.Router(); router.use((req, res, next) => { void attachTrustedAgentContext(req).then(() => next()).catch(() => res.sendStatus(401)); }); const activeRuns = new Map([["credential-route-session", { clientSessionId: "credential-route-session", controller: new AbortController(), messages: [], pendingPermissions: new Map(), pendingQuestions: new Map(), subscribers: new Set(), status: "running" }]]); - registerChatInteractionRoutes(router, { activeRuns, runtime: {} as OpencodeRuntimeAdapter, sessionMetadataStore: metadata, sessionUiStateStore: new SessionUiStateStore(join(directory, "ui")) }); + registerChatInteractionRoutes(router, { activeRuns, runtime: {} as OpencodeRuntimeAdapter, sessionMetadataStore: metadata, sessionUiStateStore: new SessionUiStateStore(new SessionWorkspaceStore(join(directory, "workspaces"))) }); app.use(router); const server = app.listen(0, "127.0.0.1"); await new Promise((resolve) => server.once("listening", resolve)); diff --git a/tests/routes/hostQuestionStream.test.ts b/tests/routes/hostQuestionStream.test.ts index 160b699..37eda5e 100644 --- a/tests/routes/hostQuestionStream.test.ts +++ b/tests/routes/hostQuestionStream.test.ts @@ -115,7 +115,7 @@ it("uses authenticated /responses for saved answers and rejects bypass and forei globalThis.fetch = (async (_input, init) => Response.json({ user_id: new Headers(init?.headers).get("Authorization") === "Bearer owner" ? "owner" : "other", project_id: project, network: "fixture" })) as typeof fetch; const workspaces = new SessionWorkspaceStore(join(root, "workspaces")); const sessions = new SessionMetadataStore(join(root, "metadata"), workspaces); - const states = new SessionUiStateStore(join(root, "unused"), workspaces); + const states = new SessionUiStateStore(workspaces); const interactions = new InteractionRequestStore(workspaces); const context = { actorKey: toActorKey("product:owner"), projectKey: toProjectKey(`product:${project}`), projectId: project, userId: "owner", sessionId: "owned" }; await sessions.ensure(context); diff --git a/tests/routes/legacyQuestionRoutes.test.ts b/tests/routes/legacyQuestionRoutes.test.ts index b31ad96..0aebb3b 100644 --- a/tests/routes/legacyQuestionRoutes.test.ts +++ b/tests/routes/legacyQuestionRoutes.test.ts @@ -5,16 +5,43 @@ 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 { OpencodeRuntimeAdapter } from "../../src/runtime/opencode.js"; +import { 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("reads a native form once when checking and answering a persisted question", async () => { + const root = await mkdtemp(join(tmpdir(), "question-single-read-")); + const workspaces = new SessionWorkspaceStore(join(root, "workspaces")); + const sessions = new SessionMetadataStore(join(root, "metadata"), workspaces); + const states = new SessionUiStateStore(workspaces); + const context = { ...getLocalAgentContext(), sessionId: "session" }; + await sessions.ensure(context); + await states.write(context, { sessionId: context.sessionId, messages: [{ role: "assistant", questions: [{ requestId: "form", status: "pending" }] }] }); + const calls: string[] = []; + const runtime = Object.assign(Object.create(OpencodeRuntimeAdapter.prototype), { + ensureClient: async () => ({ session: { form: { + get: async () => { calls.push("get"); return { fields: [{ key: "answer", type: "text" }], state: { status: "pending" } }; }, + reply: async (input: { answer: unknown }) => { calls.push("reply"); expect(input.answer).toEqual({ answer: "A" }); }, + } } }), + }); + const routes = new Map Promise>(); + const router = { get() {}, post(path: string, handler: (req: unknown, res: unknown) => Promise) { routes.set(path, handler); } }; + registerChatInteractionRoutes(router as unknown as express.Router, { activeRuns: new Map(), runtime, sessionMetadataStore: sessions, sessionUiStateStore: states }); + const req = { params: { request_id: "form" }, body: { session_id: "session", answers: [["A"]] }, header() { return undefined; } }; + const response = { code: 200, setHeader() {}, status(code: number) { this.code = code; return this; }, json() { return this; } }; + try { + await routes.get("/question/:request_id/reply")!(req, response); + expect(response.code).toBe(202); + expect(calls).toEqual(["get", "reply"]); + } finally { await rm(root, { recursive: true, force: true }); } +}); + it("reconciles persisted legacy forms and replies after the active run was lost", async () => { const root = await mkdtemp(join(tmpdir(), "legacy-question-routes-")); const workspaces = new SessionWorkspaceStore(join(root, "workspaces")); const sessions = new SessionMetadataStore(join(root, "metadata"), workspaces); - const states = new SessionUiStateStore(join(root, "unused"), workspaces); + const states = new SessionUiStateStore(workspaces); const context = { ...getLocalAgentContext(), sessionId: "session" }; await sessions.ensure(context); await states.write(context, { sessionId: "session", messages: [{ id: "old", role: "assistant", questions: [{ requestId: "form", status: "pending", tool: { callID: "call" } }] }, { id: "new", role: "assistant", text: "later message", parts: [{ type: "data-question_request", data: { request_id: "form2", session_id: "session", questions: [], tool: { callID: "call2" } } }, { type: "data-question_request", data: { request_id: "form3", session_id: "session", questions: [] } }, { type: "data-question_request", data: { request_id: "form4", session_id: "session", questions: [] } }] }] }); @@ -22,7 +49,7 @@ it("reconciles persisted legacy forms and replies after the active run was lost" let unavailable = false; let reads = 0; let replies = 0; - const runtime = { getQuestionStatus: async () => { reads++; if (unavailable) throw new Error("private-native-path"); return status; }, replyQuestion: async () => { replies++; status = "answered"; }, rejectQuestion: async () => { status = "cancelled"; } } as unknown as OpencodeRuntimeAdapter; + const runtime = { getQuestion: async () => { reads++; if (unavailable) throw new Error("private-native-path"); return status === "missing" ? null : { state: { status } }; }, replyQuestion: async () => { replies++; status = "answered"; }, rejectQuestion: async () => { status = "cancelled"; } } as unknown as OpencodeRuntimeAdapter; const app = express(); app.use(express.json()); registerChatInteractionRoutes(app, { activeRuns: new Map(), runtime, sessionMetadataStore: sessions, sessionUiStateStore: states }); const server = app.listen(0, "127.0.0.1"); diff --git a/tests/routes/permissionPersistence.test.ts b/tests/routes/permissionPersistence.test.ts index 226b13b..96829ad 100644 --- a/tests/routes/permissionPersistence.test.ts +++ b/tests/routes/permissionPersistence.test.ts @@ -16,7 +16,7 @@ it.each([false, true])("persists permission replies in the owner's workspace (na 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 states = new SessionUiStateStore(workspaces); const context = { ...getLocalAgentContext(), sessionId: "permission-session" }; await sessions.ensure(context); const messages = [{ id: "answer", role: "assistant", permissions: [] }]; diff --git a/tests/routes/question-persistence.test.ts b/tests/routes/question-persistence.test.ts index 2cddacf..caa6540 100644 --- a/tests/routes/question-persistence.test.ts +++ b/tests/routes/question-persistence.test.ts @@ -17,7 +17,7 @@ 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 states = new SessionUiStateStore(workspaces); const context = { ...getLocalAgentContext(), sessionId: "history-session" }; await sessions.ensure(context); const history: UIMessage[] = [{ id: "old-answer", role: "assistant", parts: [ @@ -35,7 +35,7 @@ it("keeps historical question invalidation across the next live snapshot", async 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 runtime = { getQuestion: async () => { nativeReads++; return null; } } as unknown as OpencodeRuntimeAdapter; const app = express(); registerChatInteractionRoutes(app, { activeRuns: new Map([[context.sessionId, run]]), runtime, sessionMetadataStore: sessions, sessionUiStateStore: states }); diff --git a/tests/routes/responses.test.ts b/tests/routes/responses.test.ts index 2f6893e..9fce7bc 100644 --- a/tests/routes/responses.test.ts +++ b/tests/routes/responses.test.ts @@ -44,7 +44,7 @@ beforeEach(async () => { })) as typeof fetch; const workspaces = new SessionWorkspaceStore(join(root, "workspaces")); const sessions = new SessionMetadataStore(join(root, "metadata"), workspaces); - states = new SessionUiStateStore(join(root, "ui"), workspaces); + states = new SessionUiStateStore(workspaces); interactions = new InteractionRequestStore(workspaces); await sessions.initialize(); const context = { actorKey: toActorKey("product:owner"), projectKey: toProjectKey(`product:${project}`), projectId: project, userId: "owner", sessionId: "owned" }; diff --git a/tests/routes/sessionIsolation.test.ts b/tests/routes/sessionIsolation.test.ts index e4a034c..bef0d16 100644 --- a/tests/routes/sessionIsolation.test.ts +++ b/tests/routes/sessionIsolation.test.ts @@ -47,7 +47,7 @@ describe("chat session HTTP isolation", () => { const workspaces = new SessionWorkspaceStore(join(metadataDir, "workspaces")); const sessionMetadataStore = new SessionMetadataStore(metadataDir, workspaces); - const sessionUiStateStore = new SessionUiStateStore(join(metadataDir, "old-ui"), workspaces); + const sessionUiStateStore = new SessionUiStateStore(workspaces); await sessionMetadataStore.initialize(); const ownerContext = { actorKey: toActorKey("product:user-a"), diff --git a/tests/routes/task-collaboration-sessions.test.ts b/tests/routes/task-collaboration-sessions.test.ts index f280ba6..d5b2ff1 100644 --- a/tests/routes/task-collaboration-sessions.test.ts +++ b/tests/routes/task-collaboration-sessions.test.ts @@ -33,7 +33,7 @@ async function fixture() { const context = { actorKey: toActorKey("product:owner"), projectKey: toProjectKey(`product:${projectId}`), projectId, userId: "owner" }; const workspaces = new SessionWorkspaceStore(join(root, "workspaces")); const sessions = new SessionMetadataStore(join(root, "metadata"), workspaces); - const states = new SessionUiStateStore(join(root, "states"), workspaces); + const states = new SessionUiStateStore(workspaces); const taskNames = new Map([[firstTask, "南桥管道爆管调查"], [secondTask, "城南供水压力复核"]]); const taskReads: string[] = []; let available = true; diff --git a/tests/runtime/hostQuestion.test.ts b/tests/runtime/hostQuestion.test.ts index 551c7ed..2d8434c 100644 --- a/tests/runtime/hostQuestion.test.ts +++ b/tests/runtime/hostQuestion.test.ts @@ -33,11 +33,11 @@ test("native question lookup distinguishes missing form from unavailable service const adapter = Object.assign(Object.create(OpencodeRuntimeAdapter.prototype), { ensureClient: async () => ({ session: { form: { get: async () => { if (failure) throw failure; return { state: { status: "answered" } }; } } } }), }) as OpencodeRuntimeAdapter; - expect(await adapter.getQuestionStatus({ sessionId: "s", requestId: "form" })).toBe("missing"); + expect(await adapter.getQuestion({ sessionId: "s", requestId: "form" })).toBeNull(); failure = new Error("network unavailable"); - await expect(adapter.getQuestionStatus({ sessionId: "s", requestId: "form" })).rejects.toThrow("network unavailable"); + await expect(adapter.getQuestion({ sessionId: "s", requestId: "form" })).rejects.toThrow("network unavailable"); failure = undefined; - expect(await adapter.getQuestionStatus({ sessionId: "s", requestId: "form" })).toBe("answered"); + expect(await adapter.getQuestion({ sessionId: "s", requestId: "form" })).toMatchObject({ state: { status: "answered" } }); }); test("message admission lookup fails closed on inbox and message transport errors", async () => { let inboxFailure = true; diff --git a/tests/sessions/answerViewState.test.ts b/tests/sessions/answerViewState.test.ts index 48a185f..df6fb5e 100644 --- a/tests/sessions/answerViewState.test.ts +++ b/tests/sessions/answerViewState.test.ts @@ -26,7 +26,7 @@ describe("answer view state", () => { beforeEach(async () => { base = await mkdtemp(join(tmpdir(), "answer-state-")); workspaces = new SessionWorkspaceStore(join(base, "workspaces")); - messages = new SessionUiStateStore(join(base, "unused"), workspaces); + messages = new SessionUiStateStore(workspaces); states = new AnswerViewStateStore(workspaces); currentSpec = spec(); await saveMessages(); diff --git a/tests/sessions/learningLifecycle.test.ts b/tests/sessions/learningLifecycle.test.ts index b6a6eb1..9193bce 100644 --- a/tests/sessions/learningLifecycle.test.ts +++ b/tests/sessions/learningLifecycle.test.ts @@ -44,7 +44,7 @@ describe("learning session lifecycle", () => { }); const orchestrator = (runtime: object) => new LearningOrchestrator( runtime as OpencodeRuntimeAdapter, new MemoryStore(join(base, "memory"), join(base, "memory-backup")), - new SessionTranscriptStore(join(base, "legacy-transcripts"), records), state, + new SessionTranscriptStore(records), state, ); it("releases both native helper sessions after a completed review and persists progress in chat", async () => { diff --git a/tests/sessions/transcriptStore.test.ts b/tests/sessions/transcriptStore.test.ts index 066eb1d..b3b9f1f 100644 --- a/tests/sessions/transcriptStore.test.ts +++ b/tests/sessions/transcriptStore.test.ts @@ -3,6 +3,7 @@ import { mkdtemp, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import { SessionWorkspaceStore } from "../../src/sessions/workspace-store.js"; import { SessionTranscriptStore } from "../../src/sessions/transcriptStore.js"; describe("SessionTranscriptStore", () => { @@ -11,8 +12,7 @@ describe("SessionTranscriptStore", () => { beforeEach(async () => { tempDir = await mkdtemp(join(tmpdir(), "tjwater-transcript-")); - store = new SessionTranscriptStore(tempDir); - await store.initialize(); + store = new SessionTranscriptStore(new SessionWorkspaceStore(tempDir)); }); afterEach(async () => { diff --git a/tests/sessions/uiStateStore.test.ts b/tests/sessions/uiStateStore.test.ts index 666872c..bb56d20 100644 --- a/tests/sessions/uiStateStore.test.ts +++ b/tests/sessions/uiStateStore.test.ts @@ -2,16 +2,16 @@ import { expect, it } from "bun:test"; import { mkdtemp, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import { SessionWorkspaceStore } from "../../src/sessions/workspace-store.js"; import { SessionUiStateStore } from "../../src/sessions/uiStateStore.js"; it("restores plan history independently of chat messages after reconnecting", async () => { const directory = await mkdtemp(join(tmpdir(), "tjwater-plan-state-")); try { - const store = new SessionUiStateStore(directory); - await store.initialize(); + const store = new SessionUiStateStore(new SessionWorkspaceStore(directory)); const executionPlans = [{ revision: 1, nodes: [{ id: "sampling", status: "waiting" }] }]; 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" }); + const reloaded = await new SessionUiStateStore(new SessionWorkspaceStore(directory)).read({ sessionId: "session-1", actorKey: "actor-test", projectKey: "project-test" }); expect(reloaded?.executionPlans).toEqual(executionPlans); expect(await store.read({ sessionId: "other-session", actorKey: "actor-test", projectKey: "project-test" })).toBeNull(); } finally { diff --git a/tests/sessions/workspaceStore.test.ts b/tests/sessions/workspaceStore.test.ts index 833f354..c3cde85 100644 --- a/tests/sessions/workspaceStore.test.ts +++ b/tests/sessions/workspaceStore.test.ts @@ -20,11 +20,11 @@ describe("owned session workspaces", () => { afterEach(async () => { await rm(base, { recursive: true, force: true }); }); it("keeps messages and searchable transcripts private to their user and scope", async () => { - const ui = new SessionUiStateStore(join(base, "old-ui"), records); - const transcript = new SessionTranscriptStore(join(base, "old-transcripts"), records); + const ui = new SessionUiStateStore(records); + const transcript = new SessionTranscriptStore(records); await ui.write(owner, { sessionId: owner.sessionId, messages: [{ text: "private answer" }] }); await transcript.appendTurn(owner, { userMessage: "private question", assistantMessage: "private answer", toolCallCount: 0 }); - expect((await new SessionUiStateStore(join(base, "old-ui"), records).read(owner))?.messages).toHaveLength(1); + expect((await new SessionUiStateStore(records).read(owner))?.messages).toHaveLength(1); expect(await ui.read({ ...owner, actorKey: "actor-b" })).toBeNull(); expect(await ui.read({ ...owner, projectKey: "project-b" })).toBeNull(); expect(await transcript.search(owner, "private")).toHaveLength(2); @@ -36,7 +36,7 @@ describe("owned session workspaces", () => { }); it("has a separate personal namespace for each user and separates it from projects", async () => { - const ui = new SessionUiStateStore(join(base, "old-ui"), records); + const ui = new SessionUiStateStore(records); const personal = { ...owner, projectKey: "personal-actor-a" }; await ui.write(personal, { sessionId: owner.sessionId, messages: [{ text: "personal answer" }] }); expect(records.path(personal, "messages.json")).toContain("/actor-a/personal/"); @@ -60,13 +60,13 @@ describe("owned session workspaces", () => { expect(await records.migrateLegacyRecords(options)).toEqual({ uiStates: 1, transcripts: 1, learningStates: 0, retainedUnownedUiStates: 1 }); expect(await readFile(records.path(owner, "messages.json"), "utf8")).toBe(messages); expect(await readFile(records.path(owner, "transcript.json"), "utf8")).toBe(turns); - await new SessionUiStateStore(uiStateDir, records).write(owner, { sessionId: owner.sessionId, messages: [{ text: "new answer" }] }); + await new SessionUiStateStore(records).write(owner, { sessionId: owner.sessionId, messages: [{ text: "new answer" }] }); expect((await records.migrateLegacyRecords(options)).uiStates).toBe(0); const archive = join(sessionWorkspaceDirectory(records.backupRoot, owner), "chat"); const backup = (await readdir(archive)).find((name) => name.startsWith("messages-"))!; expect(await readFile(join(archive, backup), "utf8")).toBe(messages); expect(await readdir(uiStateDir)).toEqual(["orphan.json"]); - expect((await new SessionUiStateStore(uiStateDir, records).read(owner))?.messages).toEqual([{ text: "new answer" }]); + expect((await new SessionUiStateStore(records).read(owner))?.messages).toEqual([{ text: "new answer" }]); }); it("rejects traversal, linked records and mismatched ownership", async () => { @@ -75,7 +75,7 @@ describe("owned session workspaces", () => { const foreign = join(base, "foreign.json"); await writeFile(foreign, "private"); await symlink(foreign, records.path(owner, "messages.json")); - await expect(new SessionUiStateStore(join(base, "old-ui"), records).read(owner)).rejects.toThrow("symbolic link"); + await expect(new SessionUiStateStore(records).read(owner)).rejects.toThrow("symbolic link"); await writeFile(join(sessionWorkspaceDirectory(records.root, owner), "chat", "owner.json"), JSON.stringify({ ...owner, actorKey: "actor-b" })); await expect(records.ensure(owner)).rejects.toThrow("ownership mismatch"); }); @@ -114,8 +114,8 @@ describe("owned session workspaces", () => { for (const path of [metadataDir, uiStateDir, transcriptDir, learningStateDir]) await mkdir(path); const metadata = new SessionMetadataStore(metadataDir, records); const { record } = await metadata.ensure(owner); - const ui = new SessionUiStateStore(uiStateDir, records); - const transcript = new SessionTranscriptStore(transcriptDir, records); + const ui = new SessionUiStateStore(records); + const transcript = new SessionTranscriptStore(records); const learning = new SessionLearningStateStore(learningStateDir, records); const legacyUi = JSON.stringify({ sessionId: owner.sessionId, messages: [{ text: "private content" }] }); await writeFile(join(uiStateDir, "chat.json"), legacyUi); @@ -153,7 +153,7 @@ describe("owned session workspaces", () => { const metadataDir = join(base, "metadata"); const metadata = new SessionMetadataStore(metadataDir, records); const { record } = await metadata.ensure(owner); - const transcript = new SessionTranscriptStore(join(base, "legacy"), records); + const transcript = new SessionTranscriptStore(records); await transcript.appendTurn(owner, { userMessage: "private question", assistantMessage: "private answer", toolCallCount: 0 }); const markDeleted = records.markDeleted.bind(records); records.markDeleted = async () => { throw new Error("disk write interrupted"); }; @@ -163,7 +163,7 @@ describe("owned session workspaces", () => { await expect(records.writeRecord(owner, "messages.json", { sessionId: owner.sessionId, messages: [] })).rejects.toThrow("session not found"); records = new SessionWorkspaceStore(records.root, records.backupRoot); const restarted = new SessionMetadataStore(metadataDir, records); - expect(await new SessionTranscriptStore(join(base, "legacy"), records).search(owner, "private")).toEqual([]); + expect(await new SessionTranscriptStore(records).search(owner, "private")).toEqual([]); await expect(restarted.ensure(owner)).rejects.toThrow("session not found"); await markDeleted(owner); await restarted.remove(record);