diff --git a/.opencode/agent-plugin.json b/.opencode/agent-plugin.json index 4442e6c..4d21cd1 100644 --- a/.opencode/agent-plugin.json +++ b/.opencode/agent-plugin.json @@ -32,7 +32,7 @@ "plugins": [ { "id": "tjwater.tools", - "description": "注册 20 个回答、展示、调查写入、通信、记忆/技能管理及待办工具；安装会话 Shell 沙箱、原生 MCP 连接和权限策略。查询由 MCP 提供，公开网页搜索使用原生 websearch。", + "description": "注册 21 个回答、展示、调查写入、通信、记忆/技能管理及待办工具；安装会话 Shell 沙箱、原生 MCP 连接和权限策略。查询由 MCP 提供，公开网页搜索使用原生 websearch。", "package": "tjwater-plugin", "entrypoint": "tjwater-plugin/index.ts", "implementation": "plugins/tjwater-tools.ts", @@ -41,6 +41,7 @@ "description": "本地交互式回答与完整最终答复；不代表后台任务完成。", "tools": [ "present_answer", + "ask_user", "final_answer" ] }, diff --git a/.opencode/agents/instruction.md b/.opencode/agents/instruction.md index f541d05..37bde0e 100644 --- a/.opencode/agents/instruction.md +++ b/.opencode/agents/instruction.md @@ -86,7 +86,7 @@ temperature: 0.2 一个界面块发布后可以继续查询或解释，再发布下一块；相同 block_id 更新本轮同一块。不要重复在文字中抄写整张表或图。每个 Chart 和 Metric 保留单位或来源；查询事实、用户事实、推断和演练数据分别标注。MapLocation 仅提供用户主动点击后的定位入口。FollowUpForm 只收集输入，用户提交后形成新的对话消息，不能在未提交时当作已确认事实。后台 task_execution 使用原任务进度与展示渠道，不调用 present_answer。 -当前步骤确实依赖用户才能决定的范围、参数或取舍时，普通聊天与任务协作主动使用 `question`，简要说明该选择如何影响后续执行，并合并相关问题。可查询的事实、已确认的参数和工具 schema 已有的默认值先自行处理；工具权限审批仍走宿主流程。等待真实回答后接着执行，不先提交 `final_answer`；取消提问不表示确认任何选项。`question` 不可用时用简短文字提问，说明哪一步等待回答。FollowUpForm 用于最终报告后可选的资料补充，不代替执行中阻塞的提问；同一问题只使用一种收集方式。 +当前步骤确实依赖用户才能决定的范围、参数或取舍时，普通聊天与任务协作优先使用当前目录提供的 `ask_user`。先完成其他工具，再单独提出最多三个问题，简要说明选择的影响；定义稳定的问题与选项 ID。一般澄清问题优先设置 allow_custom=true，允许用户直接填写回答或补充已选内容；只有确实必须限定为已有选项时才关闭。宿主确认问题已保存后，本轮结束，不继续工具、不猜测默认答案、不调用 `final_answer`。用户回答会以原会话新的用户输入送达，再继续执行。旧客户端没有 `ask_user` 时才使用可用的原生 `question`；二者都不可用时用简短文字提问。可查询的事实、已确认的参数和工具 schema 已有的默认值先自行处理；工具权限审批仍走宿主流程。取消或跳过不表示确认任何选项。FollowUpForm 用于最终报告后可选的资料补充，不代替执行中阻塞的提问；同一问题只使用一种收集方式。 核对目标完成情况或真实阻塞，并完成所有已启动工具、必要写回和界面更新后，普通聊天与任务协作调用一次 `final_answer` 结束本轮。阶段说明、等待回答和可继续执行的下一步留在过程里；最终答复后不继续查询、重复改写报告或调用 `present_answer`。后台调查执行按追加的调查指令和原有任务运行接口收尾。 diff --git a/.opencode/host-question-policy.ts b/.opencode/host-question-policy.ts new file mode 100644 index 0000000..946fbf5 --- /dev/null +++ b/.opencode/host-question-policy.ts @@ -0,0 +1,35 @@ +import type { Plugin } from "@opencode/plugin"; +import { getInternalBridgeConfig } from "./internalBridge.js"; + +export async function hostQuestionBridge(operation: string, body: Record, signal?: AbortSignal): Promise { + const { baseUrl, internalToken } = getInternalBridgeConfig(); + const response = await fetch(`${baseUrl}/internal/chat-interactions/${operation}`, { + method: "POST", headers: { "content-type": "application/json", "x-agent-internal-token": internalToken }, + body: JSON.stringify(body), signal: signal ? AbortSignal.any([signal, AbortSignal.timeout(10000)]) : AbortSignal.timeout(10000), + }); + if (!response.ok) throw new Error("Interaction gate rejected the operation. Finish existing tools before asking, then wait for the user."); + return response.json(); +} + +export async function registerHostQuestionPolicy(ctx: Parameters[0]) { + await ctx.session.hook("context", async (event) => { + const policy = await hostQuestionBridge("policy", { session_id: event.sessionID }) as { enabled: boolean; waiting: boolean }; + if (!policy.enabled) delete event.tools.ask_user; + else { + delete event.tools.question; + if (policy.waiting) for (const name of Object.keys(event.tools)) delete event.tools[name]; + } + }); + // Wrap native, MCP and custom executors, including calls via Code Mode. + await ctx.tool.transform((editor) => { + for (const definition of editor.list()) editor.update(definition.id, (draft) => { + const execute = draft.execute; + draft.execute = async (input, context) => { + const identity = { session_id: context.sessionID, call_id: context.id, tool: definition.id }; + await hostQuestionBridge("enter", identity, context.signal); + try { return await execute(input, context); } + finally { await hostQuestionBridge("leave", identity).catch(() => undefined); } + }; + }); + }); +} diff --git a/.opencode/plugins/tjwater-tools.ts b/.opencode/plugins/tjwater-tools.ts index 49046c4..d64c82d 100644 --- a/.opencode/plugins/tjwater-tools.ts +++ b/.opencode/plugins/tjwater-tools.ts @@ -1,3 +1,5 @@ +import ask_user from "../tools/ask_user.js"; +import { registerHostQuestionPolicy } from "../host-question-policy.js"; import final_answer from "../tools/final_answer.js"; import present_answer from "../tools/present_answer.js"; import locate_features from "../tools/locate_features.js"; @@ -29,6 +31,7 @@ import send_task_message from "../tools/send_task_message.js"; import reply_agent_message from "../tools/reply_agent_message.js"; const businessTools = { + ask_user, final_answer, present_answer, locate_features, @@ -57,7 +60,7 @@ export default Plugin.define({ await ctx.tool.transform((editor) => { for (const [name, definition] of Object.entries(businessTools)) { const object = z.object(definition.args); - const input = (AGENT_MESSAGE_TOOLS as readonly string[]).includes(name) ? object.strict() : object; + const input = name === "ask_user" || (AGENT_MESSAGE_TOOLS as readonly string[]).includes(name) ? object.strict() : object; editor.add({ name, description: definition.description, @@ -99,6 +102,7 @@ export default Plugin.define({ const mcpReport = await registerMcpPolicy(ctx); await registerTaskAgentPolicy(ctx); await registerAgentMessagePolicy(ctx); + await registerHostQuestionPolicy(ctx); await ctx.rpc.register(Rpc.define({ id: "tjwater.tools", methods: { diff --git a/.opencode/tools/ask_user.ts b/.opencode/tools/ask_user.ts new file mode 100644 index 0000000..c175d67 --- /dev/null +++ b/.opencode/tools/ask_user.ts @@ -0,0 +1,18 @@ +import { tool } from "../tool.js"; +import { interactionQuestionsSchema } from "../../src/chat/interaction-request-contract.js"; +import { hostQuestionBridge } from "../host-question-policy.js"; + +export default tool({ + description: "向用户提出最多三个澄清问题。先完成其他工具，再单独调用；问题由宿主持久保存，调用成功后本轮结束，等待用户回答。不要猜测答案，不要继续调用工具，不代替业务审批。选项使用稳定 id；支持单选、多选和自定义文字。", + args: { questions: tool.schema.array(tool.schema.object({ + question_id: tool.schema.string().min(1).max(128), title: tool.schema.string().min(1).max(200), body: tool.schema.string().min(1).max(2000), + options: tool.schema.array(tool.schema.object({ id: tool.schema.string().min(1).max(128), label: tool.schema.string().min(1).max(200) }).strict()).max(12), + mode: tool.schema.enum(["single", "multiple"]), allow_custom: tool.schema.boolean(), required: tool.schema.boolean(), + }).strict()).min(1).max(3) }, + async execute(args, context) { + if (!context.callID) throw new Error("Question requires a trusted tool call ID"); + return JSON.stringify(await hostQuestionBridge("create", { + session_id: context.sessionID, call_id: context.callID, tool: "ask_user", questions: interactionQuestionsSchema.parse(args.questions), + }, context.signal)); + }, +}); diff --git a/README.md b/README.md index 7d0265f..afcd6af 100644 --- a/README.md +++ b/README.md @@ -546,3 +546,13 @@ UI 分页只影响读取窗口，服务端继续保存完整消息并从完整 主题是当前用户在个人/业务项目范围内的会话组织单元，与业务 `project_id` 独立。`GET/POST /api/v1/topics` 读取/创建主题，`PATCH/DELETE /api/v1/topics/:topic_id` 重命名/移除主题，`PATCH /api/v1/session/:session_id/topic` 以 `topic_id`（UUID 或 null）调整会话归属。主题名 1–80 字符，当前主题列表存于 `SESSION_METADATA_STORAGE_DIR/topics/.json`，与元数据目录一起持久化。删除主题保留会话，遗留成员 ID 在读取时视为未分类；不会改写业务项目、原生会话或授权。旧会话默认未分类，无需迁移。普通会话和任务协作可归类，后台任务执行会话不进入用户主题。 `POST /session` 或首次 `/responses` 的 `tjwater.topic_id` 可指定已授权范围内的主题；后续续聊沿用原会话，移动使用上述独立接口。主题只组织记录，本轮不自动拼接其他会话或引入主题共享指令/附件。正文搜索在通过所有权校验的记录内读取完整消息 JSON（活动会话使用当前运行消息），匹配可见文本而非工具输入或推理；分页仍按更新时间及 ID 排序，不另建搜索服务。 + +## 可恢复提问与回答展示 + +支持 `host-question@1` 的客户端使用自有 `ask_user` 提问。宿主在会话 `chat/interaction-requests.json` 保存问题、选项草稿、已提交答案和固定消息 ID 的续跑记录；切换会话不取消提问。文件最多 2 MiB、1000 个请求，不保存登录凭据。接口位于 `/api/v1/session/:session_id/interactions`，提供列表、单项、草稿、reply/skip/cancel/resume；保存答案与模型继续分别显示状态。续跑通过现有 `/responses` 的 `tjwater.interaction_id` 使用服务端答案，不接受调用者替换答案文字。 + +续跑的用户消息保留同一个 `delivery.native_message_id`，保存可读文本及 `data-interaction_reply` 已提交回答快照；给模型的协议前缀仅用于推理输入。会话详情、分页和恢复流首次状态按当前会话的投递 ID 投影旧记录，提供相同的回答表示，不猜测文本前缀、不改写历史文件。客户端可在独立提问状态接口返回前还原回答卡片和关联过程，随后按修订更新投递状态。 + +原生 question 保留兼容入口，新增 `GET /api/v1/session/:session_id/questions/:request_id` 核对状态；原生请求缺失时显示失效，已持久化的回答终态保留。服务重启后，在有效用户授权下核对已保存的续跑记录；固定 ID 已被消费时不重发。无法核实对应的已完成回复时显示继续失败，不能把上一轮回答当作恢复结果。模型服务与插件需要同步更新，现有实例不会因源码修改自动完成发布。 + +聚焦验证：`bun test tests/sessions/interactionRequest.test.ts tests/routes/hostQuestionStream.test.ts tests/routes/interactionRequestRoutes.test.ts tests/routes/chatInteractionBridge.test.ts tests/runtime/hostQuestion.test.ts`。 diff --git a/opencode.json b/opencode.json index ab09dbf..a2d4fac 100644 --- a/opencode.json +++ b/opencode.json @@ -220,6 +220,11 @@ "resource": "*", "effect": "allow" }, + { + "action": "ask_user", + "resource": "*", + "effect": "allow" + }, { "action": "websearch", "resource": "*", diff --git a/src/chat/interaction-message-presentation.ts b/src/chat/interaction-message-presentation.ts new file mode 100644 index 0000000..34400b7 --- /dev/null +++ b/src/chat/interaction-message-presentation.ts @@ -0,0 +1,28 @@ +import type { UIMessage } from "ai"; +import { renderInteractionReplyText, type InteractionRequest } from "./interaction-request-contract.js"; + +/** Stable user presentation, including accepted-answer data for a cold reload. */ +export function createInteractionReplyMessage(record: InteractionRequest): UIMessage { + if (!record.delivery) throw new Error("Submitted interaction has no delivery identity"); + return { id: record.delivery.native_message_id, role: "user", parts: [ + { type: "text", text: renderInteractionReplyText(record) }, + { type: "data-interaction_reply", data: structuredClone(record) }, + ] }; +} + +/** Only verified delivery IDs are projected; unrelated user text is never prefix-matched. */ +export function projectInteractionReplyMessages(messages: unknown[], records: InteractionRequest[]): unknown[] { + const replies = new Map(records.filter((record) => record.delivery && ["answered", "skipped"].includes(record.status)) + .map((record) => [record.delivery!.native_message_id, record])); + return messages.map((message) => { + if (!message || typeof message !== "object") return message; + const value = message as Record; + const record = value.role === "user" && typeof value.id === "string" ? replies.get(value.id) : undefined; + if (!record) return message; + const reply = createInteractionReplyMessage(record); + const otherParts = Array.isArray(value.parts) ? value.parts.filter((part) => part && typeof part === "object" + && part.type !== "text" && part.type !== "data-interaction_reply") : []; + return { ...value, ...reply, ...(typeof value.content === "string" ? { content: renderInteractionReplyText(record) } : {}), + parts: [...reply.parts, ...otherParts] }; + }); +} diff --git a/src/chat/interaction-request-contract.ts b/src/chat/interaction-request-contract.ts new file mode 100644 index 0000000..d2ed45c --- /dev/null +++ b/src/chat/interaction-request-contract.ts @@ -0,0 +1,68 @@ +import { z } from "zod"; + +export const INTERACTION_VERSION = "agent-interaction-request@1" as const; +const id = z.string().min(1).max(128); +const unique = (values: string[]) => new Set(values).size === values.length; +export const interactionQuestionSchema = z.object({ + question_id: id, title: z.string().min(1).max(200), body: z.string().min(1).max(2000), + options: z.array(z.object({ id, label: z.string().min(1).max(200) }).strict()).max(12), + mode: z.enum(["single", "multiple"]), allow_custom: z.boolean(), required: z.boolean(), +}).strict().refine((q) => unique(q.options.map((o) => o.id)), "duplicate option ids") + .refine((q) => q.options.length > 0 || q.allow_custom, "question must offer an answer input"); +const byteLimit = (value: unknown, max: number) => Buffer.byteLength(JSON.stringify(value), "utf8") <= max; +export const interactionQuestionsSchema = z.array(interactionQuestionSchema).min(1).max(3) + .refine((qs) => unique(qs.map((q) => q.question_id)), "duplicate question ids") + .refine((qs) => qs.reduce((n, q) => n + q.title.length + q.body.length + q.options.reduce((m, o) => m + o.label.length, 0), 0) <= 10000, "definition text too long") + .refine((qs) => byteLimit(qs, 32768), "definition too large"); +export const interactionAnswerSchema = z.object({ question_id: id, selected_option_ids: z.array(id).max(12), custom_text: z.string().max(2000) }).strict() + .refine((a) => unique(a.selected_option_ids), "duplicate selected option ids"); +export const interactionAnswersSchema = z.array(interactionAnswerSchema).max(3) + .refine((as) => unique(as.map((a) => a.question_id)), "duplicate question ids") + .refine((as) => as.reduce((n, a) => n + a.custom_text.length, 0) <= 10000, "answer text too long") + .refine((as) => byteLimit(as, 16384), "answers too large"); +const write = { version: z.literal(INTERACTION_VERSION), request_id: id }; +export const interactionDraftInputSchema = z.object({ ...write, expected_draft_revision: z.number().int().nonnegative().max(Number.MAX_SAFE_INTEGER), answers: interactionAnswersSchema }).strict(); +export const interactionReplyInputSchema = z.object({ ...write, expected_revision: z.number().int().nonnegative().max(Number.MAX_SAFE_INTEGER), answers: interactionAnswersSchema }).strict(); +export const interactionMutationInputSchema = z.object({ ...write, expected_revision: z.number().int().nonnegative().max(Number.MAX_SAFE_INTEGER) }).strict(); +export type InteractionQuestion = z.infer; +export type InteractionAnswer = z.infer; +export type InteractionResolvedAnswer = InteractionAnswer & { selected_options: Array<{ id: string; label: string }> }; +export type InteractionStatus = "pending" | "answered" | "skipped" | "cancelled" | "invalidated"; +export type InteractionDelivery = { native_message_id: string; revision: number; status: "queued" | "dispatching" | "delivered" | "blocked" | "cancelled"; error_code?: string; response_id?: string }; +export type InteractionRequest = { + version: typeof INTERACTION_VERSION; interaction_id: string; message_id: string; turn_id: string; call_id: string; + questions: InteractionQuestion[]; status: InteractionStatus; revision: number; draft_revision: number; + draft: InteractionAnswer[]; answers?: InteractionResolvedAnswer[]; created_at: string; resolved_at?: string; + delivery?: InteractionDelivery; +}; +export const interactionDeliverySchema = z.object({ + native_message_id: z.string().regex(/^msg_[a-f0-9]{32}$/), revision: z.number().int().nonnegative().max(Number.MAX_SAFE_INTEGER), + status: z.enum(["queued", "dispatching", "delivered", "blocked", "cancelled"]), error_code: id.optional(), response_id: id.optional(), +}).strict(); +export const interactionRequestSchema = z.object({ + version: z.literal(INTERACTION_VERSION), interaction_id: id, message_id: id, turn_id: id, call_id: id, + questions: interactionQuestionsSchema, status: z.enum(["pending", "answered", "skipped", "cancelled", "invalidated"]), + revision: z.number().int().nonnegative().max(Number.MAX_SAFE_INTEGER), draft_revision: z.number().int().nonnegative().max(Number.MAX_SAFE_INTEGER), + draft: interactionAnswersSchema, + answers: z.array(z.object({ question_id: id, selected_option_ids: z.array(id).max(12), custom_text: z.string().max(2000), selected_options: z.array(z.object({ id, label: z.string().min(1).max(200) }).strict()).max(12) }).strict()).max(3).optional(), + created_at: z.string().datetime(), resolved_at: z.string().datetime().optional(), delivery: interactionDeliverySchema.optional(), +}).strict(); +export const interactionBlocksTurn = (r: InteractionRequest) => r.status === "pending" || Boolean(r.delivery && !["delivered", "cancelled"].includes(r.delivery.status)); + +export const renderInteractionReplyText = (record: InteractionRequest): string => { + if (record.status === "skipped") return "已跳过本组可选问题。"; + if (record.status !== "answered") throw new Error("interaction has no submitted answer"); + const lines = (record.answers ?? []).map((answer) => { + const question = record.questions.find((q) => q.question_id === answer.question_id); + return `${question?.title ?? answer.question_id}: ${[...answer.selected_options.map((o) => o.label), answer.custom_text].filter(Boolean).join("；") || "未填写"}`; + }); + return lines.join("\n"); +}; + +/** Model-facing protocol stays separate from the persisted human reply. */ +export const renderInteractionAnswer = (record: InteractionRequest): string => { + if (record.status === "skipped") return "用户明确跳过了本组非必填问题。"; + const text = `用户提交的提问回复：\n${renderInteractionReplyText(record)}`; + if (text.length > 10000) throw new Error("rendered interaction answer exceeds 10000 characters"); + return text; +}; diff --git a/src/chat/response-inputs.ts b/src/chat/response-inputs.ts index 4a6beff..06ee352 100644 --- a/src/chat/response-inputs.ts +++ b/src/chat/response-inputs.ts @@ -1,5 +1,34 @@ export type InputDelivery = "steer" | "queue"; export type ResponseInput = { id: string; text: string; delivery: InputDelivery; status: "pending" | "delivered" | "cancelled" }; +type WaitingInboxRuntime = { + listPendingUserInputs: (sessionId: string) => Promise; + agentMessageState: (sessionId: string, id: string) => Promise<"missing" | "enqueued" | "delivered">; + enqueueUserInput: (sessionId: string, input: ResponseInput, options?: { resume?: boolean }) => Promise; +}; + +/** A question boundary admits input without starting an unowned model loop. */ +export class WaitingResponseInputQueue { + private readonly operations = new Map>(); + constructor(private readonly runtime: WaitingInboxRuntime) {} + + async submit(sessionId: string, input: ResponseInput): Promise { + const operation = (this.operations.get(sessionId) ?? Promise.resolve()).catch(() => undefined).then(async () => { + const pending = await this.runtime.listPendingUserInputs(sessionId); + const previous = pending.find((item) => item.id === input.id); + if (previous) { + if (previous.text !== input.text) throw Object.assign(new Error("Input ID already belongs to another message"), { status: 409 }); + return previous; + } + if (await this.runtime.agentMessageState(sessionId, input.id) === "delivered") return { ...input, status: "delivered" as const }; + if (pending.length >= 20) throw Object.assign(new Error("At most 20 user messages may wait in this session"), { status: 429 }); + await this.runtime.enqueueUserInput(sessionId, input, { resume: false }); + return { ...input }; + }); + this.operations.set(sessionId, operation); + try { return await operation; } + finally { if (this.operations.get(sessionId) === operation) this.operations.delete(sessionId); } + } +} type InboxRuntime = { enqueueUserInput: (sessionId: string, input: ResponseInput) => Promise; listPendingUserInputs: (sessionId: string) => Promise; diff --git a/src/chat/responses-protocol.ts b/src/chat/responses-protocol.ts index 7e7cfd1..3431f7d 100644 --- a/src/chat/responses-protocol.ts +++ b/src/chat/responses-protocol.ts @@ -22,6 +22,7 @@ export const responseRequestSchema = z.object({ client_message_id: z.string().min(1).max(128).optional(), delivery: z.enum(["steer", "queue"]).optional(), input_id: z.string().regex(/^msg_[a-f0-9]{32}$/).optional(), + interaction_id: z.string().min(1).max(128).optional(), approval_mode: z.enum(["request", "always"]).default("request"), capabilities: z.array(z.string().max(64)).max(20).default([]), trigger_context: agentTriggerContextSchema.optional(), diff --git a/src/chat/sessionBridge.ts b/src/chat/sessionBridge.ts index 1484610..d8bddca 100644 --- a/src/chat/sessionBridge.ts +++ b/src/chat/sessionBridge.ts @@ -35,6 +35,7 @@ export class ChatSessionBridge { constructor(private readonly runtime: OpencodeRuntimeAdapter) {} async resolve(context: { + preserveExecution?: boolean; sessionId?: string; network?: string; projectId?: string; @@ -49,7 +50,7 @@ export class ChatSessionBridge { }> { let requestContext = this.buildRequestContext(context); const existingSessionId = context.sessionId?.trim(); - await this.abortActiveRuntime(requestContext.clientSessionId, existingSessionId); + if (!context.preserveExecution) await this.abortActiveRuntime(requestContext.clientSessionId, existingSessionId); let sessionId = existingSessionId; let created = false; diff --git a/src/routes/chat-interaction-bridge.ts b/src/routes/chat-interaction-bridge.ts new file mode 100644 index 0000000..4bde8e7 --- /dev/null +++ b/src/routes/chat-interaction-bridge.ts @@ -0,0 +1,64 @@ +import type { Express } from "express"; +import { z } from "zod"; +import { isTrustedInternalBridgeRequest } from "../product/internal-bridge-auth.js"; +import { getRuntimeSessionContext } from "../runtime/sessionContext.js"; +import { interactionQuestionsSchema } from "../chat/interaction-request-contract.js"; +import type { InteractionRequestStore } from "../sessions/interaction-request-store.js"; +import type { OpencodeRuntimeAdapter } from "../runtime/opencode.js"; + +const identity = { session_id: z.string().min(1).max(128) }; +const call = z.object({ ...identity, call_id: z.string().min(1).max(128), tool: z.string().min(1).max(200) }).strict(); + +/** Only trusted executors enter this gate; model input never selects an owner or turn. */ +export function registerChatInteractionBridge(app: Express, options: { internalToken: string; interactions: InteractionRequestStore; runtime?: OpencodeRuntimeAdapter }) { + const calls = new Map }>(); + for (const operation of ["policy", "enter", "leave", "create"] as const) { + app.post(`/internal/chat-interactions/${operation}`, async (req, res) => { + if (!isTrustedInternalBridgeRequest(req, options.internalToken)) { res.status(403).json({ message: "forbidden" }); return; } + try { + const body = operation === "policy" ? z.object(identity).strict().parse(req.body) + : operation === "create" ? call.extend({ questions: interactionQuestionsSchema }).parse(req.body) : call.parse(req.body); + const context = getRuntimeSessionContext(body.session_id); + const enabled = Boolean(context?.hostTurnId && context.hostMessageId && context.capabilities?.includes("host-question@1") + && context.purpose !== "task_execution" && context.learningMode !== "review"); + if (operation === "policy") { + const requested = context && enabled ? await options.interactions.getForTurn(context, context.hostTurnId!) : null; + res.json({ enabled, waiting: Boolean(requested) }); return; + } + const input = call.parse({ session_id: body.session_id, call_id: "call_id" in body ? body.call_id : "", tool: "tool" in body ? body.tool : "" }); + if (!context) { res.status(403).json({ message: "No authorized host turn is bound to this session" }); return; } + if (!enabled || !context) { + if (input.tool === "ask_user") { res.status(403).json({ message: "Host questions are not enabled for this turn" }); return; } + res.json({ allowed: true }); return; + } + let state = calls.get(context.sessionId); + if (!state || state.turn !== context.hostTurnId) { state = { turn: context.hostTurnId!, active: new Map() }; calls.set(context.sessionId, state); } + if (operation === "leave") { state.active.delete(input.call_id); res.json({ allowed: true }); return; } + if (context.hostQuestionClosed) { res.status(409).json({ message: "The owning turn has been stopped" }); return; } + if (operation === "create") { + if (input.tool !== "ask_user" || state.active.get(input.call_id) !== "ask_user" || !("questions" in body)) { + res.status(409).json({ message: "Question tool has not acquired the turn" }); return; + } + const previous = await options.interactions.getBlocking(context); + if (previous?.delivery && options.runtime && await options.runtime.agentMessageState(context.sessionId, previous.delivery.native_message_id) === "delivered") { + await options.interactions.updateDelivery(context, previous.interaction_id, { expected_revision: previous.delivery.revision, status: "delivered", response_id: previous.delivery.response_id }); + } + const record = await options.interactions.create(context, { message_id: context.hostMessageId!, turn_id: context.hostTurnId!, call_id: input.call_id, questions: body.questions }); + res.json(record); return; + } + const blocking = await options.interactions.getBlocking(context); + const requested = await options.interactions.getForTurn(context, context.hostTurnId!); + if (requested || blocking?.status === "pending" || input.tool === "question" || [...state.active.values()].includes("ask_user") + || (input.tool === "ask_user" && state.active.size > 0)) { + res.status(409).json({ message: "Finish other tools before asking; after asking, wait for the user" }); return; + } + state.active.set(input.call_id, input.tool); + res.json({ allowed: true }); + } catch (error) { + if (error instanceof z.ZodError) { res.status(400).json({ message: "invalid interaction bridge request" }); return; } + const status = error && typeof error === "object" && "status" in error && typeof error.status === "number" ? error.status : 503; + res.status(status).json({ message: "Interaction state is unavailable; no question was acknowledged" }); + } + }); + } +} diff --git a/src/routes/chat.ts b/src/routes/chat.ts index 99f965b..7b38157 100644 --- a/src/routes/chat.ts +++ b/src/routes/chat.ts @@ -1,3 +1,5 @@ +import { registerInteractionRequestRoutes } from "./interaction-request-routes.js"; +import type { InteractionRequestStore } from "../sessions/interaction-request-store.js"; import { Router, type NextFunction, type Request, type Response } from "express"; import { agentModelOptions, resolveDefaultModel } from "../chat/models.js"; @@ -31,6 +33,7 @@ import { registerArtifactRoutes } from "./artifact-routes.js"; import type { AgentMessageCoordinator } from "../chat/agent-message-coordinator.js"; export type BuildChatRouterOptions = { + interactionRequestStore?: InteractionRequestStore; agentMessages?: AgentMessageCoordinator; frontendActionCoordinator: FrontendActionCoordinator; learningOrchestrator: LearningOrchestrator; @@ -77,6 +80,8 @@ export const buildChatRouter = (options: BuildChatRouterOptions) => { models: agentModelOptions, }); }); + if (options.interactionRequestStore) registerInteractionRequestRoutes(chatRouter, { sessions: sessionMetadataStore, interactions: options.interactionRequestStore, + runtime, isRunning: (sessionId) => activeRuns.get(sessionId)?.status === "running" }); registerArtifactRoutes(chatRouter, { sessionMetadataStore, runtime }); chatRouter.get("/ui-registry", (_req, res) => { @@ -131,6 +136,7 @@ export const buildChatRouter = (options: BuildChatRouterOptions) => { }); registerChatSessionRoutes(chatRouter, { + interactionRequestStore: options.interactionRequestStore, learningOrchestrator, frontendActionCoordinator, runState, @@ -145,6 +151,8 @@ export const buildChatRouter = (options: BuildChatRouterOptions) => { if (agentMessages) registerAgentMessageRoutes(chatRouter, { sessions: sessionMetadataStore, messages: agentMessages }); registerChatAuxiliaryRoutes(chatRouter, { + interactionRequestStore: options.interactionRequestStore, + runtime, activeRuns, lastRunStatuses, sessionBridge, @@ -162,6 +170,7 @@ export const buildChatRouter = (options: BuildChatRouterOptions) => { registerResponseInputRoutes(chatRouter, { runtime, sessionMetadataStore, runState }); registerChatStreamRoute(chatRouter, { + interactionRequestStore: options.interactionRequestStore, agentMessages, frontendActionCoordinator, learningOrchestrator, diff --git a/src/routes/chatAuxiliaryRoutes.ts b/src/routes/chatAuxiliaryRoutes.ts index be4ca9b..f4bad00 100644 --- a/src/routes/chatAuxiliaryRoutes.ts +++ b/src/routes/chatAuxiliaryRoutes.ts @@ -1,5 +1,8 @@ import { type Router } from "express"; import { z } from "zod"; +import type { InteractionRequestStore } from "../sessions/interaction-request-store.js"; +import type { OpencodeRuntimeAdapter } from "../runtime/opencode.js"; +import { getRuntimeSessionContext, setRuntimeSessionContext } from "../runtime/sessionContext.js"; import { getLocalAgentContext } from "../context/localContext.js"; import { type ChatSessionBridge } from "../chat/sessionBridge.js"; @@ -20,6 +23,8 @@ const abortPayloadSchema = z.object({ }); type RegisterAuxiliaryRoutesOptions = { + interactionRequestStore?: InteractionRequestStore; + runtime?: OpencodeRuntimeAdapter; activeRuns: Map; lastRunStatuses: Map; sessionBridge: ChatSessionBridge; @@ -37,6 +42,8 @@ const toSessionUiStateContext = (record: { sessionId: string; actorKey: string; export const registerChatAuxiliaryRoutes = ( chatRouter: Router, { + interactionRequestStore, + runtime, activeRuns, lastRunStatuses, sessionBridge, @@ -70,6 +77,18 @@ export const registerChatAuxiliaryRoutes = ( return; } frontendActionCoordinator.cancelSession(sessionRecord.sessionId); + 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)); + 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, { + expected_revision: latest.delivery.revision, status, + }); + } activeRuns.get(sessionRecord.sessionId)?.inputs?.close(); const binding = await sessionBridge.abort({ clientSessionId: sessionRecord.sessionId, diff --git a/src/routes/chatInteractionRoutes.ts b/src/routes/chatInteractionRoutes.ts index b9dad34..ff7b89b 100644 --- a/src/routes/chatInteractionRoutes.ts +++ b/src/routes/chatInteractionRoutes.ts @@ -5,14 +5,12 @@ import { z } from "zod"; import { getLocalAgentContext, getTrustedProductCredential } from "../context/localContext.js"; import { logger } from "../logger.js"; import { type OpencodeRuntimeAdapter } from "../runtime/opencode.js"; -import { type SessionMetadataStore } from "../sessions/metadataStore.js"; +import { SessionContextMismatchError, type SessionMetadataStore } from "../sessions/metadataStore.js"; import { type SessionUiStateStore } from "../sessions/uiStateStore.js"; -import { toActorKey, toProjectKey } from "../utils/fileStore.js"; import { type ActiveRun, toPermissionStatus, updateLastAssistantPermission, - updateLastAssistantQuestion, } from "./chatUiState.js"; const permissionReplyPayloadSchema = z.object({ @@ -37,32 +35,7 @@ type RegisterInteractionRoutesOptions = { sessionUiStateStore: SessionUiStateStore; }; -type ResolvedPendingQuestion = { - requestId: string; - question: NonNullable>; -}; - -const toSessionUiStateContext = (sessionId: string) => ({ - sessionId, -}); - -const resolvePendingQuestion = ( - run: ActiveRun, - requestId: string, -): ResolvedPendingQuestion | null => { - const exactQuestion = run.pendingQuestions.get(requestId); - if (exactQuestion) { - return { requestId, question: exactQuestion }; - } - - for (const [pendingRequestId, question] of run.pendingQuestions.entries()) { - if (question.tool?.callID === requestId) { - return { requestId: pendingRequestId, question }; - } - } - - return null; -}; +const toSessionUiStateContext = (sessionId: string) => ({ sessionId }); export const registerChatInteractionRoutes = ( chatRouter: Router, @@ -213,265 +186,80 @@ export const registerChatInteractionRoutes = ( } }); - chatRouter.post("/question/:request_id/reply", async (req, res) => { - const requestId = req.params.request_id?.trim(); - const parsed = questionReplyPayloadSchema.safeParse(req.body); - if (!requestId) { - res.status(400).json({ message: "request_id is required" }); - return; - } - if (!parsed.success) { - res.status(400).json({ - message: "invalid request payload", - detail: parsed.error.flatten(), - }); - return; - } - + const isRecord = (value: unknown): value is Record => Boolean(value && typeof value === "object" && !Array.isArray(value)); + const questionHandler = (action: "status" | "reply" | "reject") => async (req: import("express").Request, res: import("express").Response) => { + res.setHeader("Cache-Control", "private, no-store"); + const requestId = typeof req.params.request_id === "string" ? req.params.request_id.trim() : undefined; + const parsed = action === "status" + ? questionRejectPayloadSchema.strict().safeParse({ session_id: req.params.session_id }) + : (action === "reply" ? questionReplyPayloadSchema.strict() : questionRejectPayloadSchema.strict()).safeParse(req.body); + if (!requestId || requestId.length > 128 || !parsed.success) { res.status(400).json({ code: "QUESTION_INVALID", message: "invalid question request" }); return; } try { - const authContext = getLocalAgentContext(req); - const projectId = authContext.projectId; - const userId = authContext.userId; - const actorKey = authContext.actorKey; - const projectKey = authContext.projectKey; - const sessionRecord = await sessionMetadataStore.get( - { actorKey, projectId, projectKey, userId }, - parsed.data.session_id, - ); - if (!sessionRecord) { - res.status(404).json({ message: "session not found" }); - return; + const session = await sessionMetadataStore.get(getLocalAgentContext(req), parsed.data.session_id); + if (!session) { res.status(404).json({ code: "SESSION_NOT_FOUND", message: "session not found" }); return; } + const context = { sessionId: session.sessionId, actorKey: session.actorKey, projectKey: session.projectKey }; + const saved = await sessionUiStateStore.read(context); + const run = activeRuns.get(session.sessionId); + // Only host-persisted questions grant access to the native form API. + const candidates: Record[] = []; + for (const message of saved?.messages ?? []) { + if (!isRecord(message)) continue; + const historical = Array.isArray(message.questions) ? message.questions : []; + const canonical = Array.isArray(message.parts) ? message.parts.flatMap((part: unknown) => isRecord(part) && part.type === "data-question_request" && isRecord(part.data) ? [{ ...part.data, requestId: part.data.request_id }] : []) : []; + for (const q of [...historical, ...canonical]) if (isRecord(q) && (q.requestId === requestId || (isRecord(q.tool) && q.tool.callID === requestId))) candidates.push(q); } - - const run = activeRuns.get(sessionRecord.sessionId); - if (!run) { - res.status(409).json({ message: "session is not waiting for questions" }); - return; + const stored = candidates.find((q) => !isRecord(q.tool) || q.requestId !== q.tool.callID) ?? candidates[0]; + if (!stored || typeof stored.requestId !== "string") { res.status(404).json({ code: "QUESTION_NOT_FOUND", message: "question request not found" }); return; } + const canonicalId = stored.requestId; + let confirmedStatus = ["answered", "rejected", "invalidated"].includes(String(stored.status)) ? String(stored.status) : undefined; + for (const message of saved?.messages ?? []) { + if (!isRecord(message) || !Array.isArray(message.parts)) continue; + for (const part of message.parts) { + if (isRecord(part) && part.type === "data-question_response" && isRecord(part.data) && part.data.request_id === canonicalId) confirmedStatus = typeof part.data.status === "string" ? part.data.status : part.data.rejected === true ? "rejected" : "answered"; + } } - - const pendingQuestion = resolvePendingQuestion(run, requestId); - if (!pendingQuestion) { - res.status(404).json({ message: "question request not found" }); - return; - } - const persistQuestionState = async () => { - const currentState = await sessionUiStateStore.read( - toSessionUiStateContext(sessionRecord.sessionId), - ); - await sessionUiStateStore.write(toSessionUiStateContext(sessionRecord.sessionId), { - sessionId: sessionRecord.sessionId, - isTitleManuallyEdited: currentState?.isTitleManuallyEdited ?? false, - messages: run.messages, + // A host-confirmed reply remains authoritative after native forms are pruned. + 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 } : {}) }; }); + // 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) })); }; - - try { - await runtime.replyQuestion({ - requestId: pendingQuestion.requestId, - sessionId: sessionRecord.sessionId, - answers: parsed.data.answers, - }); - } catch (error) { - run.messages = updateLastAssistantQuestion( - run.messages, - pendingQuestion.requestId, - (question) => ({ - ...question, - status: "error", - error: - error instanceof Error - ? error.message - : "failed to reply question", - }), - ); - await persistQuestionState().catch((persistError) => { - logger.warn( - { err: persistError, sessionId: sessionRecord.sessionId }, - "failed to persist question error state", - ); - }); - res.status(502).json({ - message: "question reply failed", - detail: error instanceof Error ? error.message : String(error), - }); + if (action === "status" || status !== "pending") { + await persist(status); + res.status(action === "status" ? 200 : 409).json({ session_id: session.sessionId, request_id: canonicalId, status, ...(action !== "status" ? { code: "QUESTION_RESOLVED" } : {}) }); return; } - - run.pendingQuestions.delete(pendingQuestion.requestId); - run.messages = updateLastAssistantQuestion( - run.messages, - pendingQuestion.requestId, - (question) => ({ - ...question, - status: "answered", - answers: parsed.data.answers, - repliedAt: Date.now(), - error: undefined, - }), - ); - await persistQuestionState().catch((persistError) => { - logger.warn( - { err: persistError, sessionId: sessionRecord.sessionId }, - "failed to persist question reply state", - ); - }); - for (const subscriber of run.subscribers) { - subscriber.write("question_response", { - session_id: pendingQuestion.question.session_id, - request_id: pendingQuestion.requestId, - answers: parsed.data.answers, - }); - } - if ( - run.status !== "running" && - run.pendingPermissions.size === 0 && - run.pendingQuestions.size === 0 - ) { - activeRuns.delete(sessionRecord.sessionId); - } - - res.status(202).json({ - session_id: pendingQuestion.question.session_id, - request_id: pendingQuestion.requestId, - answers: parsed.data.answers, - }); + const answers = action === "reply" ? questionReplyPayloadSchema.parse(req.body).answers : undefined; + 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) { - const detail = error instanceof Error ? error.message : String(error); - logger.error({ err: error }, "question reply route failed"); - res.status(500).json({ - message: "question reply route failed", - detail, - }); + // Native connectivity failure is not proof that a historical question expired. + if (error instanceof SessionContextMismatchError) { res.status(404).json({ code: "SESSION_NOT_FOUND", message: "session not found" }); return; } + const tag = isRecord(error) ? error._tag : undefined; + if (tag === "FormInvalidAnswerError") { res.status(400).json({ code: "QUESTION_INVALID_ANSWER", message: "invalid question answer" }); return; } + if (tag === "FormAlreadySettledError" || tag === "FormNotFoundError") { res.status(409).json({ code: "QUESTION_RESOLVED", message: "question is no longer pending" }); return; } + logger.warn({ err: error }, "legacy question reconciliation unavailable"); + res.status(503).json({ code: "QUESTION_UNAVAILABLE", message: "question state temporarily unavailable" }); } - }); - - chatRouter.post("/question/:request_id/reject", async (req, res) => { - const requestId = req.params.request_id?.trim(); - const parsed = questionRejectPayloadSchema.safeParse(req.body); - if (!requestId) { - res.status(400).json({ message: "request_id is required" }); - return; - } - if (!parsed.success) { - res.status(400).json({ - message: "invalid request payload", - detail: parsed.error.flatten(), - }); - return; - } - - try { - const authContext = getLocalAgentContext(req); - const projectId = authContext.projectId; - const userId = authContext.userId; - const actorKey = authContext.actorKey; - const projectKey = authContext.projectKey; - const sessionRecord = await sessionMetadataStore.get( - { actorKey, projectId, projectKey, userId }, - parsed.data.session_id, - ); - if (!sessionRecord) { - res.status(404).json({ message: "session not found" }); - return; - } - - const run = activeRuns.get(sessionRecord.sessionId); - if (!run) { - res.status(409).json({ message: "session is not waiting for questions" }); - return; - } - - const pendingQuestion = resolvePendingQuestion(run, requestId); - if (!pendingQuestion) { - res.status(404).json({ message: "question request not found" }); - return; - } - const persistQuestionState = async () => { - const currentState = await sessionUiStateStore.read( - toSessionUiStateContext(sessionRecord.sessionId), - ); - await sessionUiStateStore.write(toSessionUiStateContext(sessionRecord.sessionId), { - sessionId: sessionRecord.sessionId, - isTitleManuallyEdited: currentState?.isTitleManuallyEdited ?? false, - messages: run.messages, - }); - }; - - try { - await runtime.rejectQuestion({ - requestId: pendingQuestion.requestId, - sessionId: sessionRecord.sessionId, - }); - } catch (error) { - run.messages = updateLastAssistantQuestion( - run.messages, - pendingQuestion.requestId, - (question) => ({ - ...question, - status: "error", - error: - error instanceof Error - ? error.message - : "failed to reject question", - }), - ); - await persistQuestionState().catch((persistError) => { - logger.warn( - { err: persistError, sessionId: sessionRecord.sessionId }, - "failed to persist question error state", - ); - }); - res.status(502).json({ - message: "question reject failed", - detail: error instanceof Error ? error.message : String(error), - }); - return; - } - - run.pendingQuestions.delete(pendingQuestion.requestId); - run.messages = updateLastAssistantQuestion( - run.messages, - pendingQuestion.requestId, - (question) => ({ - ...question, - status: "rejected", - repliedAt: Date.now(), - error: undefined, - }), - ); - await persistQuestionState().catch((persistError) => { - logger.warn( - { err: persistError, sessionId: sessionRecord.sessionId }, - "failed to persist question reject state", - ); - }); - for (const subscriber of run.subscribers) { - subscriber.write("question_response", { - session_id: pendingQuestion.question.session_id, - request_id: pendingQuestion.requestId, - rejected: true, - }); - } - if ( - run.status !== "running" && - run.pendingPermissions.size === 0 && - run.pendingQuestions.size === 0 - ) { - activeRuns.delete(sessionRecord.sessionId); - } - - res.status(202).json({ - session_id: pendingQuestion.question.session_id, - request_id: pendingQuestion.requestId, - rejected: true, - }); - } catch (error) { - const detail = error instanceof Error ? error.message : String(error); - logger.error({ err: error }, "question reject route failed"); - res.status(500).json({ - message: "question reject route failed", - detail, - }); - } - }); + }; + chatRouter.get("/session/:session_id/questions/:request_id", questionHandler("status")); + chatRouter.post("/question/:request_id/reply", questionHandler("reply")); + chatRouter.post("/question/:request_id/reject", questionHandler("reject")); }; diff --git a/src/routes/chatSessionRoutes.ts b/src/routes/chatSessionRoutes.ts index f0faf3c..d7ba002 100644 --- a/src/routes/chatSessionRoutes.ts +++ b/src/routes/chatSessionRoutes.ts @@ -5,6 +5,8 @@ import { type Router } from "express"; import { z } from "zod"; import { type ChatSessionBridge } from "../chat/sessionBridge.js"; +import { projectInteractionReplyMessages } from "../chat/interaction-message-presentation.js"; +import type { InteractionRequestStore } from "../sessions/interaction-request-store.js"; import { agentModelOptions } from "../chat/models.js"; import { config } from "../config.js"; import { @@ -76,6 +78,7 @@ const forkPayloadSchema = z.object({ }); type RegisterChatSessionRoutesOptions = { + interactionRequestStore?: InteractionRequestStore; learningOrchestrator: LearningOrchestrator; frontendActionCoordinator: FrontendActionCoordinator; runState: ChatRunState; @@ -100,6 +103,7 @@ export const registerChatSessionRoutes = ( options: RegisterChatSessionRoutesOptions, ) => { const { + interactionRequestStore, frontendActionCoordinator, learningOrchestrator, runState, @@ -113,6 +117,17 @@ export const registerChatSessionRoutes = ( const { activeRuns, lastRunStatuses } = runState; const getSessionRunStatus = runState.getStatus; + const withInteractionReplies = async (record: SessionRecord, messages: unknown[]) => { + if (!interactionRequestStore) return messages; + const ids = new Set(messages.flatMap((message) => { + if (!message || typeof message !== "object") return []; + const value = message as Record; + return value.role === "user" && typeof value.id === "string" ? [value.id] : []; + })); + const replies = await interactionRequestStore.forMessages(toSessionUiStateContext(record), ids); + return projectInteractionReplyMessages(messages, replies); + }; + const refreshCollaborationTitle = async (record: SessionRecord, credential: ProductCredential | null, knownTaskTitle?: string, signal?: AbortSignal) => { if (readPurpose(record) !== "task_collaboration" || !record.taskId || !isGenericTaskCollaborationTitle(record.title)) return record; @@ -603,8 +618,8 @@ export const registerChatSessionRoutes = ( const messages = readPurpose(record) === "task_collaboration" ? state?.messages ?? [] : activeRuns.get(record.sessionId)?.messages ?? state?.messages ?? []; try { - res.json(sessionMessagePage(messages, { scope: historyScope([record.actorKey, record.projectKey, record.sessionId]), - ...query.data })); + const page = sessionMessagePage(messages, { scope: historyScope([record.actorKey, record.projectKey, record.sessionId]), ...query.data }); + res.json({ ...page, messages: await withInteractionReplies(record, page.messages) }); } catch (error) { if (!(error instanceof HistoryCursorError)) throw error; res.status(400).json({ message: error.message }); @@ -659,6 +674,7 @@ export const registerChatSessionRoutes = ( status: sessionRecord.status, session_id: sessionRecord.sessionId, ...messagePayload, + messages: await withInteractionReplies(sessionRecord, messagePayload.messages), execution_plans: state?.executionPlans ?? [], parent_session_id: sessionRecord.parentSessionId, is_streaming: activeRuns.get(sessionRecord.sessionId)?.status === "running", @@ -709,6 +725,7 @@ export const registerChatSessionRoutes = ( toSse("state", { session_id: sessionRecord.sessionId, ...messagePayload, + messages: await withInteractionReplies(sessionRecord, messagePayload.messages), execution_plans: run ? undefined : state?.executionPlans, is_streaming: run?.status === "running", run_status: getSessionRunStatus(sessionRecord.sessionId), diff --git a/src/routes/chatStream.ts b/src/routes/chatStream.ts index d5d3912..34bed14 100644 --- a/src/routes/chatStream.ts +++ b/src/routes/chatStream.ts @@ -47,6 +47,7 @@ import { normalizeTodoPriority, normalizeTodoStatus, normalizeToolParams, + sanitizeToolProgressInput, normalizeToolStatus, type PermissionRequestPayload, type TodoUpdatePayload, @@ -70,6 +71,8 @@ export { export type ApprovalMode = "request" | "always"; type StreamPromptOptions = { + interactionMessageId?: string; + onHostQuestion?: (interactionId: string) => Promise>; onPromptDispatched?: () => void; onInboxEvent?: (type: string, data: { inboxID?: string; delivery?: "steer" | "queue" }) => void; waitForUserInput?: () => Promise; @@ -98,6 +101,7 @@ type ProgressPayload = { status: ProgressStatus; title: string; detail?: string; + tool?: { name: string; call_id?: string; input?: Record }; }; const getPermissionTarget = (metadata: unknown) => { @@ -158,10 +162,13 @@ export const streamPromptResponse = async ({ onPromptDispatched, onInboxEvent, waitForUserInput, + interactionMessageId, + onHostQuestion, }: StreamPromptOptions): Promise<{ aborted: boolean; failed: boolean; toolCallCount: number; + waitingInput?: boolean; }> => { const eventStream = await runtime.subscribeEvents(); const iterator = eventStream[Symbol.asyncIterator](); @@ -191,6 +198,7 @@ export const streamPromptResponse = async ({ let promptSettled = false; let aborted = signal?.aborted ?? false; let failed = false; + let waitingInput = false; let finalAnswer: string | undefined; const presentedAnswerIds = new Set(); let finalAnswerUiBlockIds: string[] = []; @@ -227,7 +235,7 @@ export const streamPromptResponse = async ({ }) : null; - const emitProgress = ({ id, phase, status, title, detail }: ProgressPayload) => { + const emitProgress = ({ id, phase, status, title, detail, tool }: ProgressPayload) => { if (status === "running" && finalizedProgressIds.has(id)) { return; } @@ -246,6 +254,7 @@ export const streamPromptResponse = async ({ status, title, detail, + tool, started_at: startedAt, elapsed_ms: Math.max(0, now - startedAt), }); @@ -262,6 +271,7 @@ export const streamPromptResponse = async ({ status, title, detail, + tool, started_at: startedAt, ended_at: now, duration_ms: durationMs, @@ -278,7 +288,15 @@ export const streamPromptResponse = async ({ const promptPromise = Promise.resolve().then(async () => { await beforePrompt?.(); - await runtime.prompt(sessionId, message, toRuntimeModel(model)); + if (interactionMessageId) { + const result = await runtime.continueInteraction(sessionId, interactionMessageId, message, toRuntimeModel(model), signal); + if (result === "completed") { + const recovered = await runtime.readInteractionResult(sessionId, interactionMessageId); + tokenSmoother.writeToken(recovered); + emittedText = true; + done = true; + } + } else await runtime.prompt(sessionId, message, toRuntimeModel(model)); onPromptDispatched?.(); }) .then(() => { @@ -749,6 +767,30 @@ export const streamPromptResponse = async ({ }); } if (part.type === "tool") { + if (part.tool === "ask_user") { + // Reserve every lifecycle state before generic tool bookkeeping: + // a running call must not consume its completed-question identity. + if (part.state.status === "completed" && !emittedToolParts.has(part.id)) { + const output = parseToolOutput(part.state.output); + if (!onHostQuestion || !isObjectRecord(output) || typeof output.interaction_id !== "string") { + throw new Error("Host question completed without a valid persisted interaction"); + } + emittedToolParts.add(part.id); + const record = await onHostQuestion(output.interaction_id); + write("interaction_request", { ...record, session_id: clientSessionId }); + await runtime.abortSession(sessionId); + await runtime.waitForSessionIdle(sessionId, undefined, true); + waitingInput = true; + done = true; + } else if (part.state.status === "error") { + emitProgress({ + id: part.id, phase: "tool", status: "error", + title: getToolProgressTitle(part.tool, part.state.status), + detail: part.state.error, + }); + } + continue; + } if (part.tool === "present_answer") { if (suppressAnswerUi) continue; if (part.state.status === "completed" && !emittedToolParts.has(part.id)) { @@ -863,6 +905,7 @@ export const streamPromptResponse = async ({ phase: "tool", status: normalizeToolStatus(part.state.status), title: getToolProgressTitle(part.tool, part.state.status), + tool: { name: part.tool, call_id: part.callID, input: sanitizeToolProgressInput(toolParams) }, detail: buildToolProgressDetail( part.tool, part.state.status, @@ -1035,6 +1078,12 @@ export const streamPromptResponse = async ({ } await promptPromise; + if (waitingInput) { + await tokenSmoother.flush(); + emitProgress({ id: "request-completed", phase: "question", status: "completed", title: "等待你的回答", detail: "问题已保存，可以切换会话后继续回答。" }); + write("done", { session_id: clientSessionId, reason: "waiting_input" }); + return { aborted: false, failed: false, toolCallCount, waitingInput: true }; + } if (!emittedText && !finalAnswer) { logDevelopmentDebug("no streamed text emitted, falling back to messages()", { ...debugContext, diff --git a/src/routes/chatStreamEvents.ts b/src/routes/chatStreamEvents.ts index 20d3703..1a0d9a3 100644 --- a/src/routes/chatStreamEvents.ts +++ b/src/routes/chatStreamEvents.ts @@ -1,5 +1,6 @@ import type { Event as OpencodeEvent, Part } from "@opencode-ai/sdk/v2"; +import { containsForbiddenPersistentContent } from "../utils/persistencePolicy.js"; import { logger } from "../logger.js"; import { type QuestionAnswers } from "../runtime/opencode.js"; @@ -61,8 +62,8 @@ const isDevelopmentDebugLoggingEnabled = process.env.NODE_ENV === "development"; const toolLabels: Record = { clear_scada_analysis: "清除 SCADA 地图分析", - memory_manager: "记忆写入", render_scada_analysis: "渲染 SCADA 地图分析", + memory_manager: "记忆写入", session_search: "历史会话检索", skill_manager: "流程沉淀", }; @@ -356,6 +357,40 @@ export const normalizeToolStatus = (status: string) => { return "running"; }; +/** Browser-visible tool inputs: retain useful values without credentials or large payloads. */ +export const sanitizeToolProgressInput = (params: Record): Record => { + let remaining = 16_000; + let nodes = 512; + const visit = (value: unknown, depth: number): unknown => { + if (remaining <= 0 || nodes-- <= 0) return "[truncated]"; + if (typeof value === "string") { + if (containsForbiddenPersistentContent(value)) return "[redacted]"; + const limit = Math.min(remaining, 4_000); + remaining -= Math.min(value.length, limit); + return value.length > limit ? `${value.slice(0, limit)}… [truncated]` : value; + } + if (value === null || typeof value === "boolean" || typeof value === "number") return value; + if (depth >= 5) return "[truncated]"; + if (Array.isArray(value)) { + const items = value.slice(0, 64).map((item) => visit(item, depth + 1)); + if (value.length > 64) items.push("[truncated]"); + return items; + } + if (value && typeof value === "object") { + const entries = Object.entries(value).slice(0, 64).map(([rawKey, item]) => { + const key = rawKey.slice(0, 100); + remaining -= key.length; + return [key, /authorization|cookie|password|secret|token|api[_-]?key|credential|headers?/i.test(key) + ? "[redacted]" : visit(item, depth + 1)]; + }); + if (Object.keys(value).length > 64) entries.push(["…", "[truncated]"]); + return Object.fromEntries(entries); + } + return "[unsupported]"; + }; + return visit(params, 0) as Record; +}; + const formatProgressValue = (value: unknown): string => { if (typeof value === "string") { return value.length > 120 ? `${value.slice(0, 117)}...` : value; @@ -435,11 +470,11 @@ export const buildToolProgressDetail = ( error?: string, ) => { const toolName = toolLabels[tool] ?? tool; - const reasonText = reason ? `；说明：${reason}` : ""; - const paramsText = `；关键参数：${summarizeToolParams(params)}`; + const reasonText = reason ? `；说明：${containsForbiddenPersistentContent(reason) ? "[redacted]" : reason}` : ""; + const paramsText = `；关键参数：${summarizeToolParams(sanitizeToolProgressInput(params))}`; if (status === "error") { - const errorText = error ? `；错误：${error}` : ""; + const errorText = error ? `；错误：${containsForbiddenPersistentContent(error) ? "[redacted]" : error}` : ""; return `${toolName} 调用失败${reasonText}${paramsText}${errorText}`; } if (status === "completed") { diff --git a/src/routes/chatStreamRoute.ts b/src/routes/chatStreamRoute.ts index 7e8dda4..7543b58 100644 --- a/src/routes/chatStreamRoute.ts +++ b/src/routes/chatStreamRoute.ts @@ -1,5 +1,9 @@ import { createResponsesProjection, responseRequestSchema, responsePrompt } from "../chat/responses-protocol.js"; -import { ResponseInputTracker } from "../chat/response-inputs.js"; +import { ResponseInputTracker, WaitingResponseInputQueue } from "../chat/response-inputs.js"; +import { reconcileInteractionDelivery } from "../sessions/interaction-delivery-reconciliation.js"; +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 type { AgentMessageCoordinator } from "../chat/agent-message-coordinator.js"; import { credentialRefreshCoordinator } from "../product/credential-refresh.js"; @@ -158,9 +162,9 @@ export const toCollaborationMessages = (messages: UIMessage[]) => messages.flatMap((message) => { if (message.role !== "user" && message.role !== "assistant") return []; const content = readUiMessageText(message); - const reference = message.parts.filter((part) => part.type === "data-task_context"); + const reference = message.parts.filter((part) => part.type === "data-task_context" || part.type === "data-interaction_reply"); const rich = message.parts.some((part) => part.type === "data-answer_ui" || part.type === "data-final_answer"); - const parts = rich ? message.parts.filter((part) => ["text", "data-answer_ui", "data-final_answer", "data-task_context"].includes(part.type)) : [{ type: "text" as const, text: content }, ...reference]; + const parts = rich ? message.parts.filter((part) => ["text", "data-answer_ui", "data-final_answer", "data-task_context", "data-interaction_reply"].includes(part.type)) : [{ type: "text" as const, text: content }, ...reference]; return content || rich ? [{ ...message, parts } as UIMessage] : []; }); @@ -248,6 +252,7 @@ export const buildFailedTaskExecutionPlan = (history: Record[]) }; type RegisterChatStreamRouteOptions = { + interactionRequestStore?: InteractionRequestStore; agentMessages?: AgentMessageCoordinator; frontendActionCoordinator: FrontendActionCoordinator; learningOrchestrator: LearningOrchestrator; @@ -263,6 +268,7 @@ type RegisterChatStreamRouteOptions = { export const registerChatStreamRoute = ( chatRouter: Router, { + interactionRequestStore, frontendActionCoordinator, learningOrchestrator, memoryStore, @@ -276,6 +282,7 @@ export const registerChatStreamRoute = ( }: RegisterChatStreamRouteOptions, ) => { const { activeRuns, lastRunStatuses } = runState; + const waitingInputs = new WaitingResponseInputQueue(runtime); const startingSessions = new Set(); chatRouter.post("/responses", async (req, res) => { const request = responseRequestSchema.safeParse(req.body); @@ -300,6 +307,9 @@ export const registerChatStreamRoute = ( let wire: ReturnType | undefined; let boundOperationalSessionId: string | undefined; let startingSessionId: string | undefined; + let continuation: InteractionRequest | null = null; + let continuationScope: ReturnType | undefined; + let waitingInteraction: InteractionRequest | null = null; try { const authContext = getLocalAgentContext(req); const projectId = authContext.projectId; @@ -311,7 +321,7 @@ export const registerChatStreamRoute = ( const requestUiMessages = toUiMessages(parsed.data.messages); const requestPromptText = parsed.data.message?.trim() || readLatestUserText(requestUiMessages); - const promptText = trigger + let promptText = trigger ? buildAgentTriggerPrompt(trigger) : requestPromptText; if (!promptText) { @@ -348,8 +358,49 @@ export const registerChatStreamRoute = ( res.status(409).json({ message: "trigger does not match session" }); return; } + if (input.tjwater.interaction_id && (!existingSessionRecord || !interactionRequestStore || trigger || input.tjwater.delivery || input.tjwater.input_id)) { + res.status(400).json({ message: "Interaction continuation requires an existing conversation" }); return; + } + if (existingSessionRecord && interactionRequestStore) { + continuationScope = toSessionUiStateContext(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") { + res.status(409).json({ code: "INTERACTION_NOT_READY", message: "Question has no answer waiting to continue" }); return; + } + promptText = renderInteractionAnswer(continuation); + } + const blocking = await interactionRequestStore.getBlocking(continuationScope); + waitingInteraction = blocking ? await reconcileInteractionDelivery(continuationScope, blocking, interactionRequestStore, runtime, + activeRuns.get(existingSessionRecord.sessionId)?.status === "running") : null; + if (continuation && waitingInteraction?.interaction_id === continuation.interaction_id) continuation = waitingInteraction; + if (waitingInteraction?.delivery?.status === "delivered") waitingInteraction = null; + if (waitingInteraction && waitingInteraction.interaction_id !== continuation?.interaction_id + && !(input.tjwater.delivery && input.tjwater.input_id && !input.stream && !trigger && readPurpose(existingSessionRecord) !== "task_execution")) { + res.status(409).json({ code: "INTERACTION_PENDING", message: "请先回答或取消当前提问。" }); return; + } + if (continuation?.delivery?.status === "delivered") { + // An acknowledged retry must not append or reorder persisted conversation messages. + const text = await runtime.readInteractionResult(existingSessionRecord.sessionId, continuation.delivery.native_message_id); + if (input.stream) { + res.setHeader("Content-Type", "text/event-stream; charset=utf-8"); + res.setHeader("Cache-Control", "private, no-store"); + } + const replay = createResponsesProjection({ model: parsed.data.model ?? config.OPENCODE_MODEL, conversationId: existingSessionRecord.sessionId, + emit: (event) => { if (input.stream) res.write(`event: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`); }, + }); + replay.write({ type: "start", messageId: `answer_${continuation.delivery.native_message_id}` }); + replay.write({ type: "text-start", id: "recovered-answer" }); + replay.write({ type: "text-delta", id: "recovered-answer", delta: text }); + replay.write({ type: "text-end", id: "recovered-answer" }); + replay.finish(); + if (input.stream) res.end(); else res.json(replay.response); + return; + } + } const running = existingSessionRecord ? activeRuns.get(existingSessionRecord.sessionId) : undefined; if (running?.status === "running") { + if (continuation) { res.status(409).json({ code: "INTERACTION_TURN_BUSY", message: "答案已保存，请等待当前轮次结束后继续。" }); return; } if (!input.tjwater.delivery || !input.tjwater.input_id || input.stream || trigger || readPurpose(existingSessionRecord!) === "task_execution") { res.status(409).json({ message: "Running conversations require a non-streaming steer/queue input with input_id" }); return; @@ -362,8 +413,16 @@ export const registerChatStreamRoute = ( res.status(202).json({ object: "tjwater.response_input", session_id: existingSessionRecord!.sessionId, ...receipt }); return; } + if (waitingInteraction && !continuation && existingSessionRecord && input.tjwater.delivery && input.tjwater.input_id && !input.stream && !trigger + && readPurpose(existingSessionRecord) !== "task_execution") { + const receipt = await waitingInputs.submit(existingSessionRecord.sessionId, { + id: input.tjwater.input_id, text: promptText, delivery: input.tjwater.delivery, status: "pending" + }); + res.status(202).json({ object: "tjwater.response_input", session_id: existingSessionRecord.sessionId, ...receipt }); + return; + } if (input.tjwater.delivery || input.tjwater.input_id) { - res.status(409).json({ message: "The response has finished; send a new response instead" }); + res.status(409).json({ code: "RESPONSE_FINISHED", message: "The response has finished; send a new response instead" }); return; } if ( @@ -386,6 +445,7 @@ export const registerChatStreamRoute = ( } const { binding, requestContext, created } = await sessionBridge.resolve({ + preserveExecution: Boolean(continuation), sessionId: requestedSessionId, network: authContext.network, projectId, @@ -458,7 +518,7 @@ export const registerChatStreamRoute = ( toSessionUiStateContext(activeSessionRecord), ); const persistedUiMessages = toUiMessages(initialSessionState?.messages); - if (input.tjwater.client_message_id && persistedUiMessages.some((message) => message.id === input.tjwater.client_message_id)) { + if (!continuation && input.tjwater.client_message_id && persistedUiMessages.some((message) => message.id === input.tjwater.client_message_id)) { res.status(409).json({ message: "client_message_id already exists in this conversation" }); return; } @@ -469,7 +529,14 @@ export const registerChatStreamRoute = ( const visibleRequestMessages = sessionPurpose === "task_collaboration" ? toCollaborationMessages(requestUiMessages) : requestUiMessages; - const requestMessages = + const continuationAssistantId = continuation ? `answer_${continuation.delivery!.native_message_id}` : undefined; + const continuationUserExists = continuation && visiblePersistedMessages.some((message) => message.id === continuation!.delivery!.native_message_id); + const continuationReply = continuation ? createInteractionReplyMessage(continuation) : undefined; + const requestMessages = continuationReply + ? [...visiblePersistedMessages.filter((message) => message.id !== continuationAssistantId) + .map((message) => message.id === continuationReply.id ? { ...message, ...continuationReply } : message), + ...(continuationUserExists ? [] : [continuationReply])] + : visibleRequestMessages.length > 0 ? visibleRequestMessages : [ @@ -531,7 +598,8 @@ export const registerChatStreamRoute = ( startingSessionId = undefined; lastRunStatuses.set(clientSessionId, "running"); const sessionUiStateContext = toSessionUiStateContext(activeSessionRecord); - const assistantUiMessageId = `msg_${randomUUID().replaceAll("-", "")}`; + const assistantUiMessageId = continuationAssistantId ?? `msg_${randomUUID().replaceAll("-", "")}`; + setRuntimeSessionContext({ ...getRuntimeSessionContext(binding.sessionId)!, hostTurnId: wire.response.id, hostMessageId: assistantUiMessageId }); const syncLatestUiMessages = () => { latestUiMessages = [ ...requestMessages, @@ -637,14 +705,14 @@ export const registerChatStreamRoute = ( progress: upsertBackendProgress(message.progress, data), })); } else if (event === "done") { - activeRun.status = "completed"; - lastRunStatuses.set(clientSessionId, "completed"); + 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 - : "Agent 已完成处理，但没有生成文本回答。请查看过程记录，或换个更具体的问题重试。", + : data.reason === "waiting_input" ? "等待你的回答。" : "Agent 已完成处理，但没有生成文本回答。请查看过程记录，或换个更具体的问题重试。", progress: completeBackendProgress(message.progress), })); } else if (event === "error") { @@ -903,7 +971,18 @@ export const registerChatStreamRoute = ( publish("user_input", { session_id: clientSessionId, ...receipt }); }); activeRun.inputs = inputTracker; + if (continuation && interactionRequestStore && continuationScope && continuation.delivery?.status !== "delivered") { + continuation = await interactionRequestStore.claimDelivery(continuationScope, continuation.interaction_id, { + expected_revision: continuation.delivery!.revision, response_id: wire.response.id, + }); + } const streamResult = await streamPromptResponse({ + interactionMessageId: continuation?.delivery?.native_message_id, + onHostQuestion: interactionRequestStore ? async (id) => { + const record = await interactionRequestStore.get(sessionUiStateContext, id); + if (!record || record.turn_id !== wire!.response.id || record.message_id !== assistantUiMessageId) throw new Error("Question does not belong to this turn"); + return { ...record }; + } : undefined, runtime, sessionId: binding.sessionId, clientSessionId, @@ -1021,7 +1100,16 @@ export const registerChatStreamRoute = ( logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state"); }); - if (!streamResult.aborted && !streamResult.failed) { + if (continuation && interactionRequestStore && continuationScope) continuation = await interactionRequestStore.get(continuationScope, continuation.interaction_id); + if (continuation && interactionRequestStore && continuationScope && continuation.delivery && !["delivered", "cancelled"].includes(continuation.delivery.status)) { + const nativeState = await runtime.agentMessageState(binding.sessionId, continuation.delivery!.native_message_id); + continuation = await interactionRequestStore.updateDelivery(continuationScope, continuation.interaction_id, { + expected_revision: continuation.delivery!.revision, status: nativeState === "delivered" ? "delivered" : "blocked", + response_id: wire.response.id, ...(nativeState !== "delivered" ? { error_code: "CONTINUATION_NOT_CONSUMED" } : {}), + }); + publish("interaction_request", { ...continuation, session_id: clientSessionId }); + } + if (!streamResult.aborted && !streamResult.failed && !streamResult.waitingInput) { const messages = await runtime.messages(binding.sessionId, 60); const assistantMessage = [...messages] .reverse() @@ -1188,6 +1276,14 @@ export const registerChatStreamRoute = ( if (boundOperationalSessionId) { releaseOperationalTaskCredential(boundOperationalSessionId); } + if (continuation && continuationScope && interactionRequestStore) { + try { + const latest = await interactionRequestStore.get(continuationScope, continuation.interaction_id); + if (latest?.delivery && !["delivered", "cancelled"].includes(latest.delivery.status)) await interactionRequestStore.updateDelivery(continuationScope, latest.interaction_id, { + expected_revision: latest.delivery.revision, status: "blocked", error_code: "CONTINUATION_UNCONFIRMED", response_id: latest.delivery.response_id, + }); + } catch { /* Keep the durable input identity even when state storage is unavailable. */ } + } logger.error({ err: error }, "chat stream failed"); if (wire) { wire.write({ type: "error", errorText: "Agent response failed" }); diff --git a/src/routes/chatUiState.ts b/src/routes/chatUiState.ts index c945f6b..48aa43b 100644 --- a/src/routes/chatUiState.ts +++ b/src/routes/chatUiState.ts @@ -7,7 +7,7 @@ import { type TodoUpdatePayload, } from "./chatStream.js"; -export type RunStatus = "running" | "completed" | "error" | "aborted"; +export type RunStatus = "running" | "waiting_input" | "completed" | "error" | "aborted"; export type StreamSubscriber = { write: (event: string, data: Record) => void; @@ -82,8 +82,13 @@ export const upsertBackendProgress = ( const next = Array.isArray(progress) ? [...progress] : []; const id = typeof payload.id === "string" ? payload.id : `progress-${Date.now()}`; const index = next.findIndex((item) => isObjectRecord(item) && item.id === id); + const existing = index >= 0 && isObjectRecord(next[index]) ? next[index] : undefined; + const tool = isObjectRecord(payload.tool) && typeof payload.tool.name === "string" + ? { name: payload.tool.name, callId: typeof payload.tool.call_id === "string" ? payload.tool.call_id : undefined, input: isObjectRecord(payload.tool.input) ? payload.tool.input : undefined } + : existing?.tool; const nextItem = { id, + tool, phase: typeof payload.phase === "string" ? payload.phase : "progress", status: payload.status === "completed" || payload.status === "error" diff --git a/src/routes/interaction-request-routes.ts b/src/routes/interaction-request-routes.ts new file mode 100644 index 0000000..cb3feef --- /dev/null +++ b/src/routes/interaction-request-routes.ts @@ -0,0 +1,60 @@ +import type { Request, Response, Router } from "express"; +import { z } from "zod"; +import { getLocalAgentContext } from "../context/localContext.js"; +import { InteractionRequestError, type InteractionRequestStore } from "../sessions/interaction-request-store.js"; +import { SessionContextMismatchError, type SessionMetadataStore } from "../sessions/metadataStore.js"; +import type { OpencodeRuntimeAdapter } from "../runtime/opencode.js"; +import { reconcileInteractionDelivery } from "../sessions/interaction-delivery-reconciliation.js"; + +const identifier = z.string().min(1).max(128); +const listQuerySchema = z.object({ + limit: z.string().regex(/^[1-9]\d?$/).transform(Number).pipe(z.number().max(50)).optional(), + cursor: identifier.optional(), +}).strict(); + +type Options = { sessions: SessionMetadataStore; interactions: InteractionRequestStore; + runtime?: Pick; + isRunning?: (sessionId: string) => boolean }; + +/** These routes only persist user intent. Authorized /responses owns model continuation. */ +export function registerInteractionRequestRoutes(router: Router, options: Options) { + const handle = (operation: "list" | "get" | "draft" | "reply" | "skip" | "cancel" | "resume") => async (req: Request, res: Response) => { + res.setHeader("Cache-Control", "private, no-store"); + try { + const sessionId = identifier.parse(req.params.session_id); + const session = await options.sessions.get(getLocalAgentContext(req), sessionId); + if (!session) { res.status(404).json({ code: "SESSION_NOT_FOUND", detail: "session not found" }); return; } + const context = { sessionId: session.sessionId, actorKey: session.actorKey, projectKey: session.projectKey }; + if (operation === "list") { + const page = await options.interactions.list(context, listQuerySchema.parse(req.query)); + const items = await Promise.all(page.items.map((record) => reconcileInteractionDelivery(context, record, options.interactions, options.runtime, options.isRunning?.(sessionId)))); + res.json({ ...page, items }); + return; + } + const interactionId = identifier.parse(req.params.interaction_id); + if (operation === "get") { + if (Object.keys(req.query).length) throw new InteractionRequestError("invalid_query", 400); + const record = await options.interactions.get(context, interactionId); + if (!record) { res.status(404).json({ code: "interaction_not_found", detail: "interaction not found" }); return; } + res.json(await reconcileInteractionDelivery(context, record, options.interactions, options.runtime, options.isRunning?.(sessionId))); + return; + } + if (Object.keys(req.query).length) throw new InteractionRequestError("invalid_query", 400); + const record = operation === "draft" + ? await options.interactions.updateDraft(context, interactionId, req.body) + : await options.interactions[operation](context, interactionId, req.body); + // A queued or blocked delivery is a successful saved answer, never an inference error. + res.json(record); + } catch (error) { + if (error instanceof z.ZodError) { res.status(400).json({ code: "INTERACTION_INVALID", detail: "invalid interaction request" }); return; } + if (error instanceof InteractionRequestError) { res.status(error.status).json({ code: error.code, detail: error.message, ...(error.current ? { current: error.current } : {}) }); return; } + if (error instanceof SessionContextMismatchError) { res.status(404).json({ code: "SESSION_NOT_FOUND", detail: "session not found" }); return; } + res.status(503).json({ code: "INTERACTION_UNAVAILABLE", detail: "interaction unavailable" }); + } + }; + const base = "/session/:session_id/interactions"; + router.get(base, handle("list")); + router.get(`${base}/:interaction_id`, handle("get")); + router.put(`${base}/:interaction_id/draft`, handle("draft")); + for (const operation of ["reply", "skip", "cancel", "resume"] as const) router.post(`${base}/:interaction_id/${operation}`, handle(operation)); +} diff --git a/src/runtime/opencode.ts b/src/runtime/opencode.ts index 35c34ff..bbf7ba6 100644 --- a/src/runtime/opencode.ts +++ b/src/runtime/opencode.ts @@ -164,7 +164,7 @@ export class OpencodeRuntimeAdapter { return response.output as { id: string; direct: boolean; sandboxed: boolean; workspaceRoot?: string }[] | undefined; }; let tools = await readTools(); - const required = [...LOCAL_QUERY_TOOL_IDS, ...(config.TJWATER_MCP_ENABLED ? PRODUCT_MCP_TOOL_IDS : []), "shell", "final_answer", "present_answer", "locate_features", "zoom_to_map", "show_chart", "view_history", "memory_manager", "skill_manager", "render_scada_analysis", "clear_scada_analysis", "publish_execution_plan", "publish_task_report", "publish_task_progress", "publish_task_presentation", "propose_task_action", "add_task_human_evidence", "request_task_data_access", "todowrite", "send_task_message", "reply_agent_message"]; + const required = [...LOCAL_QUERY_TOOL_IDS, ...(config.TJWATER_MCP_ENABLED ? PRODUCT_MCP_TOOL_IDS : []), "shell", "ask_user", "final_answer", "present_answer", "locate_features", "zoom_to_map", "show_chart", "view_history", "memory_manager", "skill_manager", "render_scada_analysis", "clear_scada_analysis", "publish_execution_plan", "publish_task_report", "publish_task_progress", "publish_task_presentation", "propose_task_action", "add_task_human_evidence", "request_task_data_access", "todowrite", "send_task_message", "reply_agent_message"]; let missing = required.filter((name) => !tools?.some((tool) => tool.id === name && tool.direct)); const catalogDeadline = Date.now() + 10_000; // Native MCP connection/catalog publication can finish after plugin setup. @@ -362,10 +362,10 @@ export class OpencodeRuntimeAdapter { description: "调查协作消息（非用户直接指令）", metadata, delivery: "steer", resume: false }); } - async enqueueUserInput(sessionId: string, input: ResponseInput) { + async enqueueUserInput(sessionId: string, input: ResponseInput, options: { resume?: boolean } = {}) { const client = await this.ensureClient(); await client.session.prompt({ sessionID: sessionId, id: input.id, text: input.text, - delivery: input.delivery, resume: true, metadata: { source: "tjwater_user_input" } }); + delivery: input.delivery, resume: options.resume ?? true, metadata: { source: "tjwater_user_input" } }); } async listPendingUserInputs(sessionId: string): Promise { @@ -416,7 +416,7 @@ export class OpencodeRuntimeAdapter { return { ...item, status: "cancelled" }; } - async prompt(sessionId: string, text: string, model?: RuntimeModelOverride) { + async prompt(sessionId: string, text: string, model?: RuntimeModelOverride, messageId?: string) { const client = await this.ensureClient(); const startedAt = Date.now(); logDevelopmentDebug( @@ -433,7 +433,7 @@ export class OpencodeRuntimeAdapter { const agent = context?.purpose === "task_execution" && context.taskId && context.projectId ? TASK_AGENT : "instruction"; await client.session.switchAgent({ sessionID: sessionId, agent }); if (model) await client.session.switchModel({ sessionID: sessionId, model: { providerID: model.providerID, id: model.modelID } }); - await client.session.prompt({ sessionID: sessionId, text }); + await client.session.prompt({ sessionID: sessionId, text, ...(messageId ? { id: messageId, delivery: "steer" as const, metadata: { source: "tjwater_host_answer" } } : {}) }); logDevelopmentDebug( "opencode session.prompt returned", { @@ -443,12 +443,79 @@ export class OpencodeRuntimeAdapter { ); } + private interactionOperations?: Map>; + private async serializeInteraction(sessionId: string, operation: () => Promise): Promise { + const operations = this.interactionOperations ??= new Map(); + const pending = (operations.get(sessionId) ?? Promise.resolve()).catch(() => undefined).then(operation); + operations.set(sessionId, pending); + try { return await pending; } + finally { if (operations.get(sessionId) === pending) operations.delete(sessionId); } + } + + async stopInteractionInput(sessionId: string, messageId: string): Promise<"delivered" | "cancelled"> { + return this.serializeInteraction(sessionId, async () => { + const client = await this.ensureClient(); + const state = await this.agentMessageState(sessionId, messageId); + if (state === "delivered") return "delivered"; + if (state === "enqueued") await client.session.inbox.cancel({ sessionID: sessionId, inboxID: messageId }); + return await this.agentMessageState(sessionId, messageId) === "delivered" ? "delivered" : "cancelled"; + }); + } + + async continueInteraction(sessionId: string, messageId: string, text: string, model?: RuntimeModelOverride, signal?: AbortSignal): Promise<"started" | "observing" | "completed"> { + return this.serializeInteraction(sessionId, async () => { + signal?.throwIfAborted(); + const client = await this.ensureClient(); + const state = await this.agentMessageState(sessionId, messageId); + const active = Boolean((await client.session.active())[sessionId]); + signal?.throwIfAborted(); + if (state === "missing") { + if (active) throw new Error("Previous native turn is still active"); + await this.prompt(sessionId, text, model, messageId); + return "started"; + } + if (active) return "observing"; + if (state === "enqueued") { + // Resume an admitted steering input without constructing another message. + await client.session.interrupt({ sessionID: sessionId, resume: true }); + return "observing"; + } + return "completed"; + }); + } + async messages(sessionId: string, limit = 20) { const client = await this.ensureClient(); const messages = await client.message.list({ sessionID: sessionId, limit, order: "desc" }); return messages.data.reverse().map((message) => normalizeRuntimeMessage(message, sessionId)); } + async readInteractionResult(sessionId: string, messageId: string): Promise { + const client = await this.ensureClient(); + const messages = (await client.message.list({ sessionID: sessionId, limit: 100, order: "desc" })).data.slice().reverse(); + const start = messages.findIndex((message) => message.id === messageId && message.type === "user"); + if (start < 0) throw new Error("Saved answer result is outside the recovery window"); + const tail = messages.slice(start + 1); + const nextUser = tail.findIndex((message) => message.type === "user"); + const related = nextUser < 0 ? tail : tail.slice(0, nextUser); + for (const message of related.slice().reverse()) { + if (message.type !== "assistant" || message.error) continue; + for (const content of message.content.slice().reverse()) { + if (content.type === "tool" && content.name === "final_answer" && content.state.status === "completed") { + for (const item of content.state.content) { + if (item.type !== "text") continue; + try { const parsed: unknown = JSON.parse(item.text); if (parsed && typeof parsed === "object" && "answer" in parsed && typeof parsed.answer === "string") return parsed.answer; } catch { /* A non-JSON tool result is not a final answer. */ } + } + } + } + if (message.time.completed && message.finish === "stop") { + const text = message.content.filter((part) => part.type === "text").map((part) => part.text).join("\n"); + if (text) return text; + } + } + throw new Error("Answer was consumed but no completed response could be verified; do not resend it"); + } + async revertMessage(sessionId: string, messageId: string) { const client = await this.ensureClient(); return client.session.revert.stage({ sessionID: sessionId, messageID: messageId, files: false }); @@ -488,7 +555,7 @@ export class OpencodeRuntimeAdapter { return client.session.interrupt({ sessionID: sessionId }); } - async waitForSessionIdle(sessionId: string, timeoutMs = config.OPENCODE_TIMEOUT_MS) { + async waitForSessionIdle(sessionId: string, timeoutMs = config.OPENCODE_TIMEOUT_MS, strict = false) { const client = await this.ensureClient(); const startedAt = Date.now(); @@ -500,6 +567,7 @@ export class OpencodeRuntimeAdapter { await delay(100); } + if (strict) throw new Error("Native session did not reach the waiting boundary"); logger.warn( { sessionId, timeoutMs }, "timed out waiting for opencode session to become idle", @@ -554,6 +622,16 @@ export class OpencodeRuntimeAdapter { return client.session.form.reply({ sessionID: options.sessionId, formID: options.requestId, answer: formAnswers(form, options.answers) }); } + async getQuestionStatus(options: { requestId: string; sessionId: string }): Promise<"pending" | "answered" | "cancelled" | "missing"> { + const client = await this.ensureClient(); + try { return (await client.session.form.get({ sessionID: options.sessionId, formID: options.requestId })).state.status; } + catch (error) { + const tag = error && typeof error === "object" && "_tag" in error ? error._tag : undefined; + if (tag === "FormNotFoundError") return "missing"; + throw error; + } + } + async rejectQuestion(options: { requestId: string; sessionId?: string }) { if (!options.sessionId) throw new Error("OpenCode 2 form cancellation requires sessionId"); const client = await this.ensureClient(); diff --git a/src/runtime/sessionContext.ts b/src/runtime/sessionContext.ts index 445cd55..0a4df8e 100644 --- a/src/runtime/sessionContext.ts +++ b/src/runtime/sessionContext.ts @@ -14,6 +14,9 @@ export type RuntimeSessionContext = { taskId?: string; reportRevision?: number; workspaceDirectory?: string; + hostTurnId?: string; + hostMessageId?: string; + hostQuestionClosed?: boolean; }; const contexts = new Map(); diff --git a/src/server.ts b/src/server.ts index 7afaaee..3bae091 100644 --- a/src/server.ts +++ b/src/server.ts @@ -1,3 +1,5 @@ +import { InteractionRequestStore } from "./sessions/interaction-request-store.js"; +import { registerChatInteractionBridge } from "./routes/chat-interaction-bridge.js"; import { writeQueryMcpConnection } from "./runtime/query-mcp-connection.js"; import { startQueryMcpServer } from "./mcp/query-server.js"; import { createLocalQueryExecutor } from "./mcp/local-query-tools.js"; @@ -47,6 +49,7 @@ const app = express(); const sessionBridge = new ChatSessionBridge(opencodeRuntime); const sessionWorkspaces = new SessionWorkspaceStore(config.CONVERSATION_WORKSPACE_ROOT, join(config.PERSISTENCE_BACKUP_DIR, "legacy-chat")); const sessionMetadataStore = new SessionMetadataStore(undefined, sessionWorkspaces); +const interactionRequestStore = new InteractionRequestStore(sessionWorkspaces); const sessionUiStateStore = new SessionUiStateStore(undefined, sessionWorkspaces); const memoryStore = new MemoryStore(); const sessionTranscriptStore = new SessionTranscriptStore(undefined, sessionWorkspaces); @@ -68,6 +71,7 @@ registerSandboxToolBridge(app, { internalToken, runtime: opencodeRuntime }); const skillStore = new SkillStore(); registerLearningToolBridge(app, { internalToken, memoryStore, skillStore }); registerTaskPublicationBridge(app, { internalToken }); +registerChatInteractionBridge(app, { internalToken, interactions: interactionRequestStore, runtime: opencodeRuntime }); registerAgentMessageBridge(app, { internalToken, sessions: sessionMetadataStore, messages: agentMessages }); const resolveTaskReader = registerTaskAgentBridge(app, { internalToken, @@ -221,6 +225,7 @@ app.get("/health", async (_req, res) => { app.use( "/api/v1", buildChatRouter({ + interactionRequestStore, agentMessages, frontendActionCoordinator, learningOrchestrator, diff --git a/src/sessions/interaction-delivery-reconciliation.ts b/src/sessions/interaction-delivery-reconciliation.ts new file mode 100644 index 0000000..3b16f34 --- /dev/null +++ b/src/sessions/interaction-delivery-reconciliation.ts @@ -0,0 +1,30 @@ +import type { InteractionRequest } from "../chat/interaction-request-contract.js"; +import type { OpencodeRuntimeAdapter } from "../runtime/opencode.js"; +import type { SessionWorkspaceContext } from "../runtime/conversationWorkspace.js"; +import { InteractionRequestError, type InteractionRequestStore } from "./interaction-request-store.js"; + +export async function reconcileInteractionDelivery( + context: SessionWorkspaceContext, + record: InteractionRequest, + store: InteractionRequestStore, + runtime?: Pick, + running = false +): Promise { + if (running || !runtime || !record.delivery || !["answered", "skipped"].includes(record.status) + || ["delivered", "cancelled"].includes(record.delivery.status)) return record; + try { + if (await runtime.agentMessageState(context.sessionId, record.delivery.native_message_id) !== "delivered") return record; + // Receipt recovery requires completed output, not merely a consumed input or partial text. + if (!(await runtime.readInteractionResult(context.sessionId, record.delivery.native_message_id)).trim()) return record; + } catch { return record; } + try { + return await store.updateDelivery(context, record.interaction_id, { + expected_revision: record.delivery.revision, status: "delivered", response_id: record.delivery.response_id + }); + } catch (error) { + if (error instanceof InteractionRequestError && ["revision_conflict", "delivery_resolved"].includes(error.code)) { + return await store.get(context, record.interaction_id) ?? record; + } + throw error; + } +} diff --git a/src/sessions/interaction-request-store.ts b/src/sessions/interaction-request-store.ts new file mode 100644 index 0000000..c679395 --- /dev/null +++ b/src/sessions/interaction-request-store.ts @@ -0,0 +1,154 @@ +import { createHash, randomUUID } from "node:crypto"; +import { z } from "zod"; +import { readJsonFile } from "../utils/fileStore.js"; +import type { SessionWorkspaceContext } from "../runtime/conversationWorkspace.js"; +import { SessionContextMismatchError } from "./metadataStore.js"; +import { SessionWorkspaceStore } from "./workspace-store.js"; +import { INTERACTION_VERSION, renderInteractionAnswer, interactionRequestSchema, interactionAnswersSchema, interactionBlocksTurn, interactionDraftInputSchema, interactionMutationInputSchema, interactionQuestionsSchema, interactionReplyInputSchema, type InteractionAnswer, type InteractionDelivery, type InteractionRequest, type InteractionResolvedAnswer } from "../chat/interaction-request-contract.js"; + +type Receipt = { operation: string; request_id: string; digest: string; result: InteractionRequest }; +type State = { version: typeof INTERACTION_VERSION; requests: InteractionRequest[]; receipts: Receipt[] }; +const filename = "interaction-requests.json"; +export class InteractionRequestError extends Error { + constructor(readonly code: string, readonly status: number = 409, readonly current?: InteractionRequest) { super(code); } +} +const initial = (): State => ({ version: INTERACTION_VERSION, requests: [], receipts: [] }); +const parseState = (value: unknown): State => { + if (value === undefined || value === null) return initial(); + return z.object({ version: z.literal(INTERACTION_VERSION), requests: z.array(interactionRequestSchema).max(1000), receipts: z.array(z.object({ operation: z.string(), request_id: z.string().min(1).max(128), digest: z.string(), result: interactionRequestSchema }).strict()) }).strict().parse(value); +}; +const copy = (value: T): T => structuredClone(value); +const validateAnswers = (record: InteractionRequest, answers: InteractionAnswer[], complete: boolean): InteractionResolvedAnswer[] => { + const parsed = interactionAnswersSchema.parse(answers); + const result = parsed.map((answer) => { + const q = record.questions.find((q) => q.question_id === answer.question_id); + if (!q || (!q.allow_custom && answer.custom_text.length > 0) || (q.mode === "single" && answer.selected_option_ids.length > 1)) throw new InteractionRequestError("invalid_answer", 400); + const selected_options = answer.selected_option_ids.map((id) => { + const option = q.options.find((o) => o.id === id); + if (!option) throw new InteractionRequestError("invalid_option", 400); + return option; + }); + return { ...answer, selected_options }; + }); + if (complete && record.questions.some((q) => { + const a = result.find((a) => a.question_id === q.question_id); + return !a || (q.required && !a.selected_option_ids.length && !a.custom_text.trim()); + })) throw new InteractionRequestError("incomplete_answer", 400); + if (Buffer.byteLength(JSON.stringify(result)) > 16384 || result.reduce((n, a) => n + a.custom_text.length + a.selected_options.reduce((m, o) => m + o.label.length, 0), 0) > 10000) throw new InteractionRequestError("answer_too_large", 400); + return result; +}; +export class InteractionRequestStore { + constructor(readonly workspaceStore: SessionWorkspaceStore) {} + private async read(context: SessionWorkspaceContext) { + if (!await this.workspaceStore.validate(context, filename)) throw new SessionContextMismatchError(); + return parseState(await readJsonFile(this.workspaceStore.path(context, filename))); + } + private async mutate(context: SessionWorkspaceContext, operation: (state: State) => T): Promise { + return this.workspaceStore.updateRecord(context, filename, async ({ current }) => { + const state = parseState(current); + const result = operation(state); + state.receipts = state.receipts.slice(-32); + while (state.receipts.length && Buffer.byteLength(JSON.stringify(state.receipts)) > 128 * 1024) state.receipts.shift(); + if (state.requests.length > 1000 || Buffer.byteLength(JSON.stringify(state, null, 2)) > 2 * 1024 * 1024) throw new InteractionRequestError("interaction_capacity_exceeded", 413); + return { result: copy(result), value: state }; + }); + } + async get(context: SessionWorkspaceContext, id: string) { return copy((await this.read(context)).requests.find((r) => r.interaction_id === id) ?? null); } + async forMessages(context: SessionWorkspaceContext, messageIds: ReadonlySet) { + if (!messageIds.size) return []; + return copy((await this.read(context)).requests.filter((record) => record.delivery && messageIds.has(record.delivery.native_message_id))); + } + async list(context: SessionWorkspaceContext, options: { limit?: number; cursor?: string } = {}) { + const limit = z.number().int().min(1).max(50).parse(options.limit ?? 20); + const requests = [...(await this.read(context)).requests].reverse(); + const start = options.cursor ? requests.findIndex((r) => r.interaction_id === options.cursor) + 1 : 0; + if (options.cursor && !start) throw new InteractionRequestError("invalid_cursor", 400); + const items = requests.slice(start, start + limit); + return copy({ items, next_cursor: start + limit < requests.length ? items.at(-1)?.interaction_id : undefined }); + } + async getBlocking(context: SessionWorkspaceContext) { return copy((await this.read(context)).requests.find(interactionBlocksTurn) ?? null); } + async getForTurn(context: SessionWorkspaceContext, turnId: string) { return copy((await this.read(context)).requests.find((record) => record.turn_id === turnId) ?? null); } + async create(context: SessionWorkspaceContext, input: { message_id: string; turn_id: string; call_id: string; questions: unknown }) { + const parsed = z.object({ message_id: z.string().min(1).max(128), turn_id: z.string().min(1).max(128), call_id: z.string().min(1).max(128), questions: interactionQuestionsSchema }).strict().parse(input); + return this.mutate(context, (state) => { + const existing = state.requests.find((r) => r.turn_id === parsed.turn_id && r.call_id === parsed.call_id); + if (existing) { + if (JSON.stringify(existing.questions) !== JSON.stringify(parsed.questions) || existing.message_id !== parsed.message_id) throw new InteractionRequestError("idempotency_mismatch"); + return existing; + } + if (state.requests.some(interactionBlocksTurn)) throw new InteractionRequestError("interaction_unresolved"); + const record: InteractionRequest = { version: INTERACTION_VERSION, interaction_id: randomUUID(), ...parsed, status: "pending", revision: 0, draft_revision: 0, draft: [], created_at: new Date().toISOString() }; + state.requests.push(record); + return record; + }); + } + private async apply(context: SessionWorkspaceContext, id: string, operation: string, input: unknown, draft: boolean, change: (record: InteractionRequest, input: Record) => void) { + const parsed = (draft ? interactionDraftInputSchema : operation === "reply" ? interactionReplyInputSchema : interactionMutationInputSchema).parse(input); + return this.mutate(context, (state) => { + const digest = createHash("sha256").update(JSON.stringify(parsed)).digest("hex"); + const previous = state.receipts.find((r) => r.request_id === parsed.request_id); + if (previous) { + if (previous.operation !== `${id}:${operation}` || previous.digest !== digest) throw new InteractionRequestError("idempotency_mismatch"); + return state.requests.find((r) => r.interaction_id === id) ?? previous.result; + } + const record = state.requests.find((r) => r.interaction_id === id); + if (!record) throw new InteractionRequestError("interaction_not_found", 404); + if (record.status !== "pending") throw new InteractionRequestError("interaction_resolved", 409, copy(record)); + const expected = "expected_draft_revision" in parsed ? parsed.expected_draft_revision : parsed.expected_revision; + if (expected !== (draft ? record.draft_revision : record.revision)) throw new InteractionRequestError("revision_conflict", 409, copy(record)); + change(record, parsed); + if (draft) record.draft_revision++; else { record.revision++; record.resolved_at = new Date().toISOString(); } + state.receipts.push({ operation: `${id}:${operation}`, request_id: parsed.request_id, digest, result: copy(record) }); + return record; + }); + } + updateDraft(context: SessionWorkspaceContext, id: string, input: unknown) { return this.apply(context, id, "draft", input, true, (r, p) => { validateAnswers(r, p.answers as InteractionAnswer[], false); r.draft = p.answers as InteractionAnswer[]; }); } + reply(context: SessionWorkspaceContext, id: string, input: unknown) { return this.apply(context, id, "reply", input, false, (r, p) => { r.answers = validateAnswers(r, p.answers as InteractionAnswer[], true); r.status = "answered"; + try { renderInteractionAnswer(r); } catch { throw new InteractionRequestError("answer_too_large", 400); } + r.delivery = this.delivery(); }); } + skip(context: SessionWorkspaceContext, id: string, input: unknown) { return this.apply(context, id, "skip", input, false, (r) => { if (r.questions.some((q) => q.required)) throw new InteractionRequestError("required_question", 400); r.status = "skipped"; r.answers = []; r.delivery = this.delivery(); }); } + cancel(context: SessionWorkspaceContext, id: string, input: unknown) { return this.apply(context, id, "cancel", input, false, (r) => { r.status = "cancelled"; }); } + private delivery(): InteractionDelivery { return { native_message_id: `msg_${randomUUID().replaceAll("-", "")}`, revision: 0, status: "queued" }; } + async updateDelivery(context: SessionWorkspaceContext, id: string, input: { expected_revision: number; status: InteractionDelivery["status"]; response_id?: string; error_code?: string }) { + const parsed = z.object({ expected_revision: z.number().int().nonnegative().max(Number.MAX_SAFE_INTEGER), status: z.enum(["queued", "dispatching", "delivered", "blocked", "cancelled"]), response_id: z.string().min(1).max(128).optional(), error_code: z.string().min(1).max(128).optional() }).strict().parse(input); + return this.mutate(context, (state) => { + const r = state.requests.find((r) => r.interaction_id === id); + if (!r?.delivery) throw new InteractionRequestError("delivery_not_found", 404); + if (r.delivery.revision !== parsed.expected_revision) throw new InteractionRequestError("revision_conflict", 409, copy(r)); + if (["delivered", "cancelled"].includes(r.delivery.status)) throw new InteractionRequestError("delivery_resolved"); + r.delivery = { native_message_id: r.delivery.native_message_id, revision: r.delivery.revision + 1, status: parsed.status, ...(parsed.response_id ? { response_id: parsed.response_id } : {}), ...(parsed.error_code ? { error_code: parsed.error_code } : {}) }; + return r; + }); + } + async resume(context: SessionWorkspaceContext, id: string, input: unknown) { + const parsed = interactionMutationInputSchema.parse(input); + return this.mutate(context, (state) => { + const digest = createHash("sha256").update(JSON.stringify(parsed)).digest("hex"); + const previous = state.receipts.find((r) => r.request_id === parsed.request_id); + if (previous) { + if (previous.operation !== `${id}:resume` || previous.digest !== digest) throw new InteractionRequestError("idempotency_mismatch"); + return state.requests.find((r) => r.interaction_id === id) ?? previous.result; + } + const record = state.requests.find((r) => r.interaction_id === id); + if (!record?.delivery) throw new InteractionRequestError("delivery_not_found", 404); + if (record.delivery.revision !== parsed.expected_revision) throw new InteractionRequestError("revision_conflict", 409, copy(record)); + state.receipts.push({ operation: `${id}:resume`, request_id: parsed.request_id, digest, result: copy(record) }); + return record; + }); + } + claimDelivery(context: SessionWorkspaceContext, id: string, input: { expected_revision: number; response_id: string }) { + return this.updateDelivery(context, id, { ...input, status: "dispatching" }); + } + async invalidate(context: SessionWorkspaceContext, id: string, input: unknown) { + return this.apply(context, id, "invalidate", input, false, (r) => { r.status = "invalidated"; }); + } + async cancelPending(context: SessionWorkspaceContext) { + return this.mutate(context, (state) => { + for (const r of state.requests) { + if (r.status === "pending") { r.status = "cancelled"; r.revision++; r.resolved_at = new Date().toISOString(); } + if (r.delivery && ["queued", "blocked"].includes(r.delivery.status)) { r.delivery.status = "cancelled"; r.delivery.revision++; } + } + return state.requests; + }); + } +} diff --git a/src/sessions/uiStateStore.ts b/src/sessions/uiStateStore.ts index b8ffd51..86265fd 100644 --- a/src/sessions/uiStateStore.ts +++ b/src/sessions/uiStateStore.ts @@ -46,6 +46,19 @@ export class SessionUiStateStore { return state; } + async update(context: SessionUiStateContext, change: (state: SessionUiStateRecord) => SessionUiStateRecord) { + if (this.workspaces) 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; await removeFileIfExists(this.filePath(context)); diff --git a/src/sessions/workspace-store.ts b/src/sessions/workspace-store.ts index 43a8ca4..1c5d6f7 100644 --- a/src/sessions/workspace-store.ts +++ b/src/sessions/workspace-store.ts @@ -9,7 +9,7 @@ import { import { atomicWriteJson, listJsonFiles, readJsonFile } from "../utils/fileStore.js"; import { SessionContextMismatchError, type SessionRecord } from "./metadataStore.js"; -type ChatRecordName = "messages.json" | "transcript.json" | "learning.json"; +type ChatRecordName = "messages.json" | "transcript.json" | "learning.json" | "interaction-requests.json"; type LegacySources = { metadataDir: string; uiStateDir: string; transcriptDir: string; learningStateDir?: string }; const exists = async (path: string) => { try { await lstat(path); return true; } @@ -104,6 +104,23 @@ export class SessionWorkspaceStore { }); } + /** Read, compare and replace under the same queue as message writes and deletion. */ + async updateRecord( + context: SessionWorkspaceContext, + name: ChatRecordName, + operation: (records: { messages: unknown; current: unknown }) => Promise<{ result: T; value?: object }>, + ): Promise { + return this.serialize(context, async () => { + if (!await this.validate(context, "messages.json") || !await this.validate(context, name)) throw new SessionContextMismatchError(); + const messages = await readJsonFile(this.path(context, "messages.json")); + const current = await readJsonFile(this.path(context, name)); + const update = await operation({ messages, current }); + if (!await this.validate(context, "messages.json") || !await this.validate(context, name)) throw new SessionContextMismatchError(); + if (update.value !== undefined) await atomicWriteJson(this.path(context, name), update.value as Record, 0o600); + return update.result; + }); + } + async markDeleted(context: SessionWorkspaceContext) { return this.serialize(context, async () => { const directory = await this.ensure(context, true); diff --git a/tests/chat/responseInputs.test.ts b/tests/chat/responseInputs.test.ts index b4043ce..6b116c9 100644 --- a/tests/chat/responseInputs.test.ts +++ b/tests/chat/responseInputs.test.ts @@ -1,8 +1,35 @@ import { expect, it, mock } from "bun:test"; -import { ResponseInputTracker, type ResponseInput } from "../../src/chat/response-inputs.js"; +import { ResponseInputTracker, WaitingResponseInputQueue, type ResponseInput } from "../../src/chat/response-inputs.js"; const input = (id: number): ResponseInput => ({ id: `msg_${String(id).padStart(32, "0")}`, text: `Input ${id}`, delivery: "steer", status: "pending" }); +it("serializes waiting input admission, deduplicates retries and enforces capacity without resuming", async () => { + const pending: ResponseInput[] = []; + const enqueue = mock(async (_session: string, item: ResponseInput, options?: { resume?: boolean }) => { + expect(options?.resume).toBe(false); + pending.push(item); + }); + const queue = new WaitingResponseInputQueue({ listPendingUserInputs: async () => pending, + agentMessageState: async () => "missing", enqueueUserInput: enqueue }); + await Promise.all([queue.submit("s", input(1)), queue.submit("s", input(1))]); + expect(enqueue).toHaveBeenCalledTimes(1); + await expect(queue.submit("s", { ...input(1), text: "different" })).rejects.toMatchObject({ status: 409 }); + const results = await Promise.allSettled(Array.from({ length: 20 }, (_, i) => queue.submit("s", input(i + 2)))); + expect(results.filter((result) => result.status === "fulfilled")).toHaveLength(19); + expect(pending).toHaveLength(20); + expect(enqueue).toHaveBeenCalledTimes(20); +}); + +it("does not reinject a consumed waiting input or infer absence after a native read failure", async () => { + const enqueue = mock(async () => {}); + const runtime = { listPendingUserInputs: async () => [], agentMessageState: async () => "delivered" as const, enqueueUserInput: enqueue }; + await expect(new WaitingResponseInputQueue(runtime).submit("s", input(1))).resolves.toMatchObject({ status: "delivered" }); + expect(enqueue).not.toHaveBeenCalled(); + const unavailable = { ...runtime, agentMessageState: async () => { throw new Error("native unavailable"); } }; + await expect(new WaitingResponseInputQueue(unavailable).submit("s", input(2))).rejects.toThrow("native unavailable"); + expect(enqueue).not.toHaveBeenCalled(); +}); + it("waits for admission at completion, deduplicates concurrent retries and closes only after native consumption", async () => { let release = () => {}; const gate = new Promise((resolve) => { release = resolve; }); diff --git a/tests/routes/chatInteractionBridge.test.ts b/tests/routes/chatInteractionBridge.test.ts new file mode 100644 index 0000000..17b3eff --- /dev/null +++ b/tests/routes/chatInteractionBridge.test.ts @@ -0,0 +1,52 @@ +import { expect, test } from "bun:test"; +import express from "express"; +import { mkdtemp, rm } from "node:fs/promises"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { registerChatInteractionBridge } from "../../src/routes/chat-interaction-bridge.js"; +import { setRuntimeSessionContext, removeRuntimeSessionContext } from "../../src/runtime/sessionContext.js"; +import { InteractionRequestStore } from "../../src/sessions/interaction-request-store.js"; +import { SessionWorkspaceStore } from "../../src/sessions/workspace-store.js"; + +test("trusted gate binds host identity, rejects parallel tools and persists waiting before tool leave/idle", async () => { + const root = await mkdtemp(join(tmpdir(), "interaction-bridge-")); + const context = { sessionId: "bridge-session", clientSessionId: "bridge-session", actorKey: "trusted-actor", projectKey: "trusted-project", traceId: "test", hostTurnId: "host-turn", hostMessageId: "host-message", capabilities: ["host-question@1"], purpose: "ordinary_chat" as const }; + const workspace = new SessionWorkspaceStore(root); await workspace.ensure(context); + const interactions = new InteractionRequestStore(workspace); + setRuntimeSessionContext(context); + const app = express(); app.use(express.json()); registerChatInteractionBridge(app, { internalToken: "token", interactions }); + const server = app.listen(0, "127.0.0.1"); await new Promise((resolve) => server.once("listening", resolve)); + const base = `http://127.0.0.1:${(server.address() as { port: number }).port}`; + const post = (operation: string, body: unknown, headers: Record = {}) => fetch(`${base}/internal/chat-interactions/${operation}`, { method: "POST", headers: { "Content-Type": "application/json", "x-agent-internal-token": "token", ...headers }, body: JSON.stringify({ session_id: context.sessionId, ...body as object }) }); + const ask = { tool: "ask_user", call_id: "ask" }; + const query = { tool: "mcp_query", call_id: "query" }; + const questions = [{ question_id: "q", title: "Title", body: "Question", options: [], mode: "single", allow_custom: true, required: true }]; + try { + expect((await post("enter", ask, { "x-agent-internal-token": "wrong" })).status).toBe(403); + expect((await post("enter", ask, { Origin: "https://browser.test" })).status).toBe(403); + expect((await post("create", { ...ask, questions, actorKey: "forged" })).status).toBe(400); + expect((await post("create", { ...ask, questions })).status).toBe(409); + expect((await post("enter", query)).status).toBe(200); + expect((await post("enter", ask)).status).toBe(409); + await post("leave", query); + expect((await post("enter", ask)).status).toBe(200); + expect((await post("enter", query)).status).toBe(409); + const response = await post("create", { ...ask, questions }); + expect(response.status).toBe(200); + const record = await response.json(); + expect(record).toMatchObject({ message_id: "host-message", turn_id: "host-turn", call_id: "ask", status: "pending" }); + expect(await interactions.get(context, record.interaction_id)).toEqual(record); + expect(await (await post("policy", {})).json()).toEqual({ enabled: true, waiting: true }); + await post("leave", ask); + expect((await post("enter", query)).status).toBe(409); + expect((await post("enter", { tool: "question", call_id: "native" })).status).toBe(409); + await interactions.cancelPending(context); + expect((await post("enter", query)).status).toBe(409); + setRuntimeSessionContext({ ...context, hostTurnId: "next-turn" }); + expect((await post("enter", query)).status).toBe(200); + } finally { + removeRuntimeSessionContext(context.sessionId); + await new Promise((resolve) => server.close(() => resolve())); + await rm(root, { recursive: true, force: true }); + } +}); diff --git a/tests/routes/hostQuestionStream.test.ts b/tests/routes/hostQuestionStream.test.ts new file mode 100644 index 0000000..160b699 --- /dev/null +++ b/tests/routes/hostQuestionStream.test.ts @@ -0,0 +1,207 @@ +import { expect, it } from "bun:test"; +import { streamPromptResponse } from "../../src/routes/chatStream.js"; +import type { OpencodeRuntimeAdapter } from "../../src/runtime/opencode.js"; + +const questionEvent = { type: "message.part.updated", properties: { sessionID: "native", part: { id: "part", type: "tool", tool: "ask_user", sessionID: "native", messageID: "assistant", callID: "call", state: { status: "completed", input: {}, output: JSON.stringify({ interaction_id: "question" }), time: { start: 1, end: 2 } } } } }; +const stream = (events: unknown[]) => ({ async *[Symbol.asyncIterator]() { yield* events; } }); + +it("reserves ask_user running events so completed questions stop the turn exactly once", async () => { + const events: Array<{ event: string; data: Record }> = []; + const calls: string[] = []; + const running = structuredClone(questionEvent); + Object.assign(running.properties.part.state, { status: "running", input: { questions: [{ question_id: "q" }] }, output: undefined }); + const runtime = { + subscribeEvents: async () => ({ async *[Symbol.asyncIterator]() { + yield running; + yield questionEvent; + yield questionEvent; + throw new Error("must stop observing after the question"); + } }), + prompt: async () => {}, + abortSession: async () => { calls.push("interrupt"); }, + waitForSessionIdle: async () => { calls.push("idle"); }, + messages: async () => [], + } as unknown as OpencodeRuntimeAdapter; + const result = await streamPromptResponse({ runtime, sessionId: "native", clientSessionId: "client", message: "question", onHostQuestion: async (id) => { calls.push("question"); return { interaction_id: id, status: "pending" }; }, write: (event, data) => events.push({ event, data }) }); + expect(result).toMatchObject({ failed: false, aborted: false, waitingInput: true }); + expect(calls).toEqual(["question", "interrupt", "idle"]); + expect(events.filter((event) => event.event === "interaction_request")).toHaveLength(1); + expect(events.filter((event) => event.event === "done").map((event) => event.data.reason)).toEqual(["waiting_input"]); + expect(events.some((event) => ["tool_call", "final_answer", "error"].includes(event.event))).toBeFalse(); +}); + +it("ends ask_user at idle as waiting_input, without fabricating a final answer or failed run", async () => { + const order: string[] = []; + const events: Array<{ event: string; data: Record }> = []; + const runtime = { + subscribeEvents: async () => { order.push("subscribe"); return stream([questionEvent, questionEvent]); }, + prompt: async () => { order.push("prompt"); }, + abortSession: async () => { order.push("interrupt"); }, + waitForSessionIdle: async () => { order.push("idle"); }, + messages: async () => { throw new Error("waiting must not read a fallback final answer"); }, + } as unknown as OpencodeRuntimeAdapter; + const result = await streamPromptResponse({ runtime, sessionId: "native", clientSessionId: "client", message: "question", onHostQuestion: async (id) => { expect(id).toBe("question"); order.push("persisted"); return { interaction_id: id, status: "pending" }; }, write: (event, data) => events.push({ event, data }) }); + expect(result).toMatchObject({ failed: false, aborted: false, waitingInput: true }); + expect(order).toEqual(["subscribe", "prompt", "persisted", "interrupt", "idle"]); + expect(events.filter((e) => e.event === "interaction_request")).toHaveLength(1); + expect(events.filter((e) => e.event === "done")).toEqual([{ event: "done", data: { session_id: "client", reason: "waiting_input" } }]); + expect(events.some((e) => e.event === "final_answer" || e.event === "error")).toBeFalse(); +}); + +it("refuses an unowned or unpersisted question without publishing an actionable card", async () => { + let interrupts = 0; + const events: string[] = []; + const runtime = { subscribeEvents: async () => stream([questionEvent]), prompt: async () => {}, abortSession: async () => { interrupts++; } } as unknown as OpencodeRuntimeAdapter; + await expect(streamPromptResponse({ runtime, sessionId: "native", clientSessionId: "client", message: "question", onHostQuestion: async () => { throw new Error("Question does not belong to this turn"); }, write: (event) => events.push(event) })).rejects.toThrow("Question does not belong"); + expect(events).not.toContain("interaction_request"); + expect(events).not.toContain("done"); + expect(interrupts).toBe(0); +}); + +it("rejects a completed question with malformed output instead of silently finishing", async () => { + const malformed = structuredClone(questionEvent); + malformed.properties.part.state.output = "invalid JSON"; + const events: string[] = []; + const runtime = { subscribeEvents: async () => stream([malformed]), prompt: async () => {} } as unknown as OpencodeRuntimeAdapter; + await expect(streamPromptResponse({ runtime, sessionId: "native", clientSessionId: "client", message: "question", onHostQuestion: async () => ({}), write: (event) => events.push(event) })).rejects.toThrow("valid persisted interaction"); + expect(events).not.toContain("done"); + expect(events).not.toContain("interaction_request"); +}); + +it("recovers an already completed fixed-ID answer without another prompt or waiting for new native events", async () => { + const calls: string[] = []; + const events: Array<{ event: string; data: Record }> = []; + const runtime = { + subscribeEvents: async () => { calls.push("subscribe"); return { [Symbol.asyncIterator]: () => ({ next: () => new Promise>(() => {}), return: async () => ({ done: true as const, value: undefined }) }) }; }, + continueInteraction: async (...args: unknown[]) => { calls.push(`continue:${args[0]}:${args[1]}`); return "completed"; }, + prompt: async () => { throw new Error("must not reinject a consumed answer"); }, + readInteractionResult: async (sessionId: string, messageId: string) => { expect([sessionId, messageId]).toEqual(["native", "fixed"]); return "Recovered answer"; }, + messages: async () => { throw new Error("generic history may contain an unrelated old answer"); }, + } as unknown as OpencodeRuntimeAdapter; + const result = await streamPromptResponse({ runtime, sessionId: "native", clientSessionId: "client", interactionMessageId: "fixed", message: "saved answer", write: (event, data) => events.push({ event, data }) }); + expect(result).toMatchObject({ failed: false, aborted: false }); + expect(calls).toEqual(["subscribe", "continue:native:fixed"]); + expect(events.filter((e) => e.event === "token").map((e) => e.data.content).join("")).toBe("Recovered answer"); + expect(events.filter((e) => e.event === "done")).toHaveLength(1); +}); + +it("observes an admitted fixed-ID answer without issuing an ordinary prompt", async () => { + const runtime = { + subscribeEvents: async () => stream([{ type: "message.part.updated", properties: { sessionID: "native", part: { id: "text", type: "text", text: "", sessionID: "native", messageID: "assistant" } } }, { type: "message.part.delta", properties: { sessionID: "native", partID: "text", field: "text", delta: "Continued" } }, { type: "session.idle", properties: { sessionID: "native" } }]), + continueInteraction: async () => "observing", + prompt: async () => { throw new Error("duplicate prompt"); }, + } as unknown as OpencodeRuntimeAdapter; + const tokens: unknown[] = []; + const result = await streamPromptResponse({ runtime, sessionId: "native", clientSessionId: "client", interactionMessageId: "fixed", message: "saved answer", write: (event, data) => { if (event === "token") tokens.push(data.content); } }); + expect(result.failed).toBeFalse(); expect(tokens.join("")).toBe("Continued"); +}); + +it("uses authenticated /responses for saved answers and rejects bypass and foreign actors while retaining the fixed native ID", async () => { + const { mkdtemp, rm } = await import("node:fs/promises"); + const { tmpdir } = await import("node:os"); + const { join } = await import("node:path"); + const { default: express } = await import("express"); + const { buildChatRouter } = await import("../../src/routes/chat.js"); + const { ChatSessionBridge } = await import("../../src/chat/sessionBridge.js"); + const { SessionWorkspaceStore } = await import("../../src/sessions/workspace-store.js"); + const { SessionMetadataStore } = await import("../../src/sessions/metadataStore.js"); + const { SessionUiStateStore } = await import("../../src/sessions/uiStateStore.js"); + const { InteractionRequestStore } = await import("../../src/sessions/interaction-request-store.js"); + const { toActorKey, toProjectKey } = await import("../../src/utils/fileStore.js"); + const { removeRuntimeSessionContext } = await import("../../src/runtime/sessionContext.js"); + const root = await mkdtemp(join(tmpdir(), "host-stream-")); + const http = globalThis.fetch; + const project = "10000000-0000-4000-8000-000000000001"; + 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 interactions = new InteractionRequestStore(workspaces); + const context = { actorKey: toActorKey("product:owner"), projectKey: toProjectKey(`product:${project}`), projectId: project, userId: "owner", sessionId: "owned" }; + await sessions.ensure(context); + await states.write(context, { sessionId: "owned", isTitleManuallyEdited: true, messages: [] }); + const question = await interactions.create(context, { message_id: "assistant", turn_id: "old", call_id: "call", questions: [{ question_id: "q", title: "Choose", body: "Which?", options: [{ id: "a", label: "Accepted option" }], mode: "single", allow_custom: false, required: true }] }); + const nativeIds: string[] = []; + let admissionCount = 0; + let releaseContinuation: () => void = () => {}; + const continuationGate = new Promise((resolve) => { releaseContinuation = resolve; }); + const runtime = { + messages: async () => [], prompt: async () => { throw new Error("ordinary prompt must not bypass a question"); }, + listPendingUserInputs: async () => [], cancelPendingUserInputs: async () => {}, + continueInteraction: async (_session: string, id: string, text: string) => { nativeIds.push(id); expect(text).toContain("Accepted option"); expect(text).not.toContain("must not replace saved answer"); admissionCount++; await continuationGate; return "observing"; }, + agentMessageState: async () => "delivered", + subscribeEvents: async () => stream([{ type: "message.part.updated", properties: { sessionID: "owned", part: { id: "text", type: "text", text: "", sessionID: "owned", messageID: "assistant" } } }, { type: "message.part.delta", properties: { sessionID: "owned", partID: "text", field: "text", delta: "Done" } }, { type: "session.idle", properties: { sessionID: "owned" } }]), + } as unknown as OpencodeRuntimeAdapter; + const bridge = new ChatSessionBridge(runtime); + const app = express(); app.use(express.json()); + app.use("/api/v1", buildChatRouter({ runtime, sessionBridge: bridge, sessionMetadataStore: sessions, sessionUiStateStore: states, interactionRequestStore: interactions, sessionTranscriptStore: { getRecentTurns: async () => [] } as never, memoryStore: { buildPromptSnapshot: async () => "" } as never, learningOrchestrator: { onTurnCompleted: async () => {} } as never, frontendActionCoordinator: { cancelSession: () => {} } as never })); + const server = app.listen(0, "127.0.0.1"); + await new Promise((resolve) => server.once("listening", resolve)); + const base = `http://127.0.0.1:${(server.address() as { port: number }).port}`; + const call = (body: unknown, token = "owner") => http(`${base}/api/v1/responses`, { method: "POST", headers: { Authorization: `Bearer ${token}`, "X-Project-Id": project, "Content-Type": "application/json" }, body: JSON.stringify(body), signal: AbortSignal.timeout(5000) }); + try { + expect((await call({ input: "bypass", conversation: "owned" })).status).toBe(409); + const saved = await interactions.reply(context, question.interaction_id, { version: "agent-interaction-request@1", expected_revision: 0, request_id: "answer", answers: [{ question_id: "q", selected_option_ids: ["a"], custom_text: "" }] }); + expect((await call({ input: "bypass saved delivery", conversation: "owned" })).status).toBe(409); + const request = { input: "must not replace saved answer", conversation: "owned", tjwater: { interaction_id: question.interaction_id } }; + expect((await call(request, "other")).status).toBe(404); + const firstRequest = call(request); + for (let i = 0; i < 100 && admissionCount === 0; i++) await Bun.sleep(5); + expect(admissionCount).toBe(1); + expect((await call(request)).status).toBe(409); + releaseContinuation(); + const response = await firstRequest; + const body = await response.json(); + expect(response.status).toBe(200); expect(body).toMatchObject({ status: "completed" }); + expect(nativeIds).toEqual([saved.delivery!.native_message_id]); + expect((await interactions.get(context, question.interaction_id))?.delivery?.status).toBe("delivered"); + expect(admissionCount).toBe(1); + const restored = await states.read(context); + const replyMessage = (restored?.messages as Array<{ id: string; parts: unknown[] }>).find((message) => message.id === saved.delivery!.native_message_id); + expect(replyMessage?.parts).toContainEqual({ type: "text", text: "Choose: Accepted option" }); + expect(replyMessage?.parts).toContainEqual(expect.objectContaining({ type: "data-interaction_reply" })); + // Legacy persisted turns contain the model-facing prefix. All read paths project + // the same accepted answer, even before the separate interactions request returns. + const legacy = [{ id: saved.delivery!.native_message_id, role: "user", parts: [{ type: "text", text: "用户提交的提问回复：\nChoose: Accepted option" }] }]; + await states.write(context, { sessionId: "owned", isTitleManuallyEdited: true, messages: legacy }); + const read = (path: string, token = "owner") => http(`${base}/api/v1/session/owned${path}`, { headers: { Authorization: `Bearer ${token}`, "X-Project-Id": project } }); + for (const path of ["", "/messages?limit=1"]) { + const history = await (await read(path)).json(); + expect(history.messages[0].parts[0]).toEqual({ type: "text", text: "Choose: Accepted option" }); + expect(history.messages[0].parts[1]).toMatchObject({ type: "data-interaction_reply", data: { status: "answered", delivery: { status: "delivered" } } }); + } + const streamState = await (await read("/stream")).text(); + expect(streamState).toContain("data-interaction_reply"); + expect(streamState).not.toContain("用户提交的提问回复"); + expect((await read("", "other")).status).toBe(404); + expect((await states.read(context))?.messages).toEqual(legacy); + } finally { + releaseContinuation(); globalThis.fetch = http; removeRuntimeSessionContext("owned"); server.closeAllConnections(); + await new Promise((resolve) => server.close(() => resolve())); + await rm(root, { recursive: true, force: true }); + } +}); + + +it("fails consumed-answer recovery when no completed matching answer exists instead of showing old history", async () => { + const events: string[] = []; + const runtime = { + subscribeEvents: async () => ({ [Symbol.asyncIterator]: () => ({ next: () => new Promise>(() => {}), return: async () => ({ done: true as const, value: undefined }) }) }), + continueInteraction: async () => "completed", + readInteractionResult: async () => { throw new Error("Consumed answer has no completed matching response"); }, + messages: async () => [{ info: { role: "assistant" }, parts: [{ type: "text", text: "OLD UNRELATED ANSWER" }] }], + prompt: async () => { throw new Error("must not reinject consumed answer"); }, + } as unknown as OpencodeRuntimeAdapter; + await expect(streamPromptResponse({ runtime, sessionId: "native", clientSessionId: "client", interactionMessageId: "fixed", message: "saved answer", write: (event) => events.push(event) })).rejects.toThrow("no completed matching response"); + expect(events).not.toContain("token"); expect(events).not.toContain("done"); +}); + +it("does not report waiting_input until the interrupted question turn is confirmed idle", async () => { + const events: string[] = []; + const runtime = { + subscribeEvents: async () => stream([questionEvent]), prompt: async () => {}, abortSession: async () => {}, + waitForSessionIdle: async () => { throw new Error("native turn still active"); }, + } as unknown as OpencodeRuntimeAdapter; + await expect(streamPromptResponse({ runtime, sessionId: "native", clientSessionId: "client", message: "question", onHostQuestion: async () => ({ interaction_id: "question" }), write: (event) => events.push(event) })).rejects.toThrow("native turn still active"); + expect(events).not.toContain("done"); +}); diff --git a/tests/routes/interactionRequestRoutes.test.ts b/tests/routes/interactionRequestRoutes.test.ts new file mode 100644 index 0000000..cfe51ff --- /dev/null +++ b/tests/routes/interactionRequestRoutes.test.ts @@ -0,0 +1,69 @@ +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 { INTERACTION_VERSION } from "../../src/chat/interaction-request-contract.js"; +import { registerInteractionRequestRoutes } from "../../src/routes/interaction-request-routes.js"; +import { InteractionRequestStore } from "../../src/sessions/interaction-request-store.js"; +import { SessionMetadataStore } from "../../src/sessions/metadataStore.js"; +import { SessionWorkspaceStore } from "../../src/sessions/workspace-store.js"; + +it("authorizes session ownership, validates CAS, persists drafts and answers without inference", async () => { + const root = await mkdtemp(join(tmpdir(), "interaction-routes-")); + const workspaces = new SessionWorkspaceStore(join(root, "workspaces")); + const sessions = new SessionMetadataStore(join(root, "metadata"), workspaces); + const interactions = new InteractionRequestStore(workspaces); + const context = { ...getLocalAgentContext(), sessionId: "session" }; + await sessions.ensure(context); + const record = await interactions.create(context, { message_id: "message", turn_id: "turn", call_id: "call", questions: [{ question_id: "q", title: "Choose", body: "Which one?", options: [{ id: "a", label: "A" }], mode: "single", allow_custom: false, required: true }] }); + const app = express(); + app.use(express.json()); + registerInteractionRequestRoutes(app, { sessions, interactions }); + const server = app.listen(0, "127.0.0.1"); + await new Promise((resolve) => server.once("listening", resolve)); + const base = `http://127.0.0.1:${(server.address() as { port: number }).port}/session/session/interactions`; + const headers = { "Content-Type": "application/json" }; + const send = (path: string, body: unknown, method = "POST", override = headers) => fetch(`${base}/${record.interaction_id}${path}`, { method, headers: override, body: JSON.stringify(body) }); + const answers = [{ question_id: "q", selected_option_ids: ["a"], custom_text: "" }]; + const mutation = { version: INTERACTION_VERSION, request_id: "reply", expected_revision: 0, answers }; + try { + for (const bad of [{ ...headers, "X-Agent-Local-User": "other" }, { ...headers, "X-Agent-Local-Project": "other" }]) { + expect((await fetch(base, { headers: bad })).status).toBe(404); + expect((await send("/reply", mutation, "POST", bad)).status).toBe(404); + } + const list = await fetch(`${base}?limit=1`); + expect(list.headers.get("Cache-Control")).toBe("private, no-store"); + expect(await list.json()).toMatchObject({ items: [{ interaction_id: record.interaction_id }] }); + for (const query of ["limit=51", "limit=1.5", "limit=01", "unexpected=1", "cursor=unknown"]) expect((await fetch(`${base}?${query}`)).status).toBe(400); + expect((await send("/draft", { version: INTERACTION_VERSION, request_id: "draft", expected_draft_revision: 0, answers }, "PUT")).status).toBe(200); + expect((await send("/reply", { ...mutation, expected_revision: Number.MAX_SAFE_INTEGER + 1 })).status).toBe(400); + expect((await send("/reply", { ...mutation, unexpected: true })).status).toBe(400); + expect((await send("/skip", { version: INTERACTION_VERSION, request_id: "skip-required", expected_revision: 0 })).status).toBe(400); + const reply = await send("/reply", mutation); + expect(reply.status).toBe(200); + const saved = await reply.json() as { delivery: { native_message_id: string; revision: number } }; + expect(saved).toMatchObject({ status: "answered", delivery: { status: "queued" } }); + expect(await (await send("/reply", mutation)).json()).toEqual(saved); + expect((await send("/reply", { ...mutation, request_id: "competing" })).status).toBe(409); + expect((await send("/draft", { version: INTERACTION_VERSION, request_id: "late", expected_draft_revision: 1, answers: [] }, "PUT")).status).toBe(409); + const resume = await send("/resume", { version: INTERACTION_VERSION, request_id: "resume", expected_revision: saved.delivery.revision }); + expect(resume.status).toBe(200); + expect(await resume.json()).toMatchObject({ delivery: { native_message_id: saved.delivery.native_message_id } }); + const second = await interactions.create(context, { message_id: "second", turn_id: "second", call_id: "second", questions: record.questions.map((q) => ({ ...q, required: false })) }).catch(() => null); + // Saved queued replies block creation until their delivery is settled. + expect(second).toBeNull(); + const originalGet = interactions.get.bind(interactions); + interactions.get = async () => { throw new Error(`${root}/private-answer`); }; + const unavailable = await fetch(`${base}/${record.interaction_id}`); + expect(unavailable.status).toBe(503); + expect(await unavailable.json()).toEqual({ code: "INTERACTION_UNAVAILABLE", detail: "interaction unavailable" }); + interactions.get = originalGet; + await sessions.remove((await sessions.get(context, "session"))!); + expect((await send("/reply", mutation)).status).toBe(404); + } finally { + await new Promise((resolve) => server.close(() => resolve())); + await rm(root, { recursive: true, force: true }); + } +}); diff --git a/tests/routes/legacyQuestionRoutes.test.ts b/tests/routes/legacyQuestionRoutes.test.ts new file mode 100644 index 0000000..b31ad96 --- /dev/null +++ b/tests/routes/legacyQuestionRoutes.test.ts @@ -0,0 +1,61 @@ +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 { 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("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 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: [] } }] }] }); + let status: "pending" | "answered" | "cancelled" | "missing" = "pending"; + 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 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"); + await new Promise((resolve) => server.once("listening", resolve)); + const base = `http://127.0.0.1:${(server.address() as { port: number }).port}`; + const get = (id = "form", headers = {}) => fetch(`${base}/session/session/questions/${id}`, { headers }); + const reply = (id = "form") => fetch(`${base}/question/${id}/reply`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ session_id: "session", answers: [["A"]] }) }); + try { + expect((await get("unknown")).status).toBe(404); expect(reads).toBe(0); + expect((await get("form", { "X-Agent-Local-User": "other" })).status).toBe(404); expect(reads).toBe(0); + expect(await (await get("call")).json()).toMatchObject({ request_id: "form", status: "pending" }); + unavailable = true; + const failed = await get(); expect(failed.status).toBe(503); expect(await failed.text()).not.toContain("private-native-path"); + expect((await states.read(context))?.messages[0]).toMatchObject({ questions: [{ status: "pending" }] }); + unavailable = false; + expect((await reply()).status).toBe(202); expect(replies).toBe(1); + expect((await states.read(context))?.messages[0]).toMatchObject({ questions: [{ status: "answered", answers: [["A"]] }] }); + expect((await states.read(context))?.messages[1]).toMatchObject({ text: "later message" }); + expect((await reply()).status).toBe(409); expect(replies).toBe(1); + status = "missing"; expect(await (await get()).json()).toMatchObject({ status: "answered" }); + status = "cancelled"; expect(await (await get("call2")).json()).toMatchObject({ request_id: "form2", status: "rejected" }); + expect((await states.read(context))?.messages[1]).toMatchObject({ parts: [{ type: "data-question_request" }, { type: "data-question_request" }, { type: "data-question_request" }, { type: "data-question_response", data: { request_id: "form2", status: "rejected" } }] }); + status = "missing"; expect(await (await get("form2")).json()).toMatchObject({ status: "rejected" }); + expect(await (await get("form3")).json()).toMatchObject({ status: "invalidated" }); + status = "pending"; + expect((await reply("form4")).status).toBe(202); + const canonical = (await states.read(context))?.messages[1] as { parts: Array<{ type: string; data?: unknown }> }; + expect(canonical.parts.find((part) => part.type === "data-question_response" && (part.data as { request_id: string }).request_id === "form4")).toMatchObject({ data: { request_id: "form4", status: "answered", answers: [["A"]] } }); + status = "missing"; expect(await (await get("form4")).json()).toMatchObject({ status: "answered" }); + await sessions.remove((await sessions.get(context, "session"))!); + expect((await get()).status).toBe(404); + } finally { + await new Promise((resolve) => server.close(() => resolve())); + await rm(root, { recursive: true, force: true }); + } +}); diff --git a/tests/routes/taskCollaboration.test.ts b/tests/routes/taskCollaboration.test.ts index c272dff..27397a1 100644 --- a/tests/routes/taskCollaboration.test.ts +++ b/tests/routes/taskCollaboration.test.ts @@ -7,6 +7,15 @@ import { } from "../../src/routes/chatStreamRoute.js"; describe("task collaboration history", () => { + it("keeps accepted reply snapshots with both text and rich history", () => { + const reply = { type: "data-interaction_reply" as const, data: { interaction_id: "accepted-question" } }; + for (const rich of [false, true]) { + const parts: UIMessage["parts"] = [{ type: "text", text: "核查范围: 北侧管网" }, reply, + ...(rich ? [{ type: "data-final_answer" as const, data: { content: "核查范围: 北侧管网" } }] : [])]; + const messages = [{ id: "reply", role: "user", parts: [...parts, { type: "data-progress", data: { title: "执行记录" } }] }] as unknown as UIMessage[]; + expect(toCollaborationMessages(messages)).toEqual([{ id: "reply", role: "user", parts }]); + } + }); it("persists only user and assistant text instead of execution trajectory", () => { const messages = [ { id: "user-1", role: "user", parts: [{ type: "text", text: "为什么这样判断？" }] }, diff --git a/tests/runtime/hostQuestion.test.ts b/tests/runtime/hostQuestion.test.ts new file mode 100644 index 0000000..18a5406 --- /dev/null +++ b/tests/runtime/hostQuestion.test.ts @@ -0,0 +1,96 @@ +import { expect, test } from "bun:test"; +import { OpencodeRuntimeAdapter } from "../../src/runtime/opencode.js"; + +function runtime(state: "missing" | "enqueued" | "delivered", active = false) { + const calls: unknown[] = []; + const adapter = Object.assign(Object.create(OpencodeRuntimeAdapter.prototype), { + ensureClient: async () => ({ session: { active: async () => active ? { s: {} } : {}, interrupt: async (input: unknown) => { calls.push(input); } } }), + agentMessageState: async () => state, + prompt: async (...args: unknown[]) => { calls.push(args); }, + }) as OpencodeRuntimeAdapter; + return { adapter, calls }; +} +test("host reply continuation injects fixed ID only when absent; queued and consumed IDs never reinject", async () => { + const missing = runtime("missing"); + expect(await missing.adapter.continueInteraction("s", "msg_fixed", "answer")).toBe("started"); + expect(missing.calls).toEqual([["s", "answer", undefined, "msg_fixed"]]); + const busy = runtime("missing", true); + await expect(busy.adapter.continueInteraction("s", "msg_fixed", "answer")).rejects.toThrow("still active"); + expect(busy.calls).toEqual([]); + const enqueued = runtime("enqueued"); + expect(await enqueued.adapter.continueInteraction("s", "msg_fixed", "answer")).toBe("observing"); + expect(enqueued.calls).toEqual([{ sessionID: "s", resume: true }]); + const consumed = runtime("delivered"); + expect(await consumed.adapter.continueInteraction("s", "msg_fixed", "answer")).toBe("completed"); + expect(consumed.calls).toEqual([]); + const observing = runtime("delivered", true); + expect(await observing.adapter.continueInteraction("s", "msg_fixed", "answer")).toBe("observing"); + expect(observing.calls).toEqual([]); +}); +test("native question lookup distinguishes missing form from unavailable service", async () => { + let failure: unknown = { _tag: "FormNotFoundError" }; + 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"); + failure = new Error("network unavailable"); + await expect(adapter.getQuestionStatus({ sessionId: "s", requestId: "form" })).rejects.toThrow("network unavailable"); + failure = undefined; + expect(await adapter.getQuestionStatus({ sessionId: "s", requestId: "form" })).toBe("answered"); +}); +test("message admission lookup fails closed on inbox and message transport errors", async () => { + let inboxFailure = true; + let messageFailure: unknown = { status: 404 }; + const adapter = Object.assign(Object.create(OpencodeRuntimeAdapter.prototype), { + ensureClient: async () => ({ session: { inbox: { list: async () => { if (inboxFailure) throw new Error("inbox unavailable"); return []; } }, message: { get: async () => { throw messageFailure; } } } }), + }) as OpencodeRuntimeAdapter; + await expect(adapter.agentMessageState("s", "msg_fixed")).rejects.toThrow("inbox unavailable"); + inboxFailure = false; + expect(await adapter.agentMessageState("s", "msg_fixed")).toBe("missing"); + messageFailure = new Error("message unavailable"); + await expect(adapter.agentMessageState("s", "msg_fixed")).rejects.toThrow("message unavailable"); +}); +test("stop serializes behind in-flight answer admission and cancels its exact inbox identity", async () => { + let release!: () => void; + let admitted!: () => void; + const entered = new Promise((resolve) => { admitted = resolve; }); + const barrier = new Promise((resolve) => { release = resolve; }); + let state: "missing" | "enqueued" = "missing"; + const calls: string[] = []; + const adapter = Object.assign(Object.create(OpencodeRuntimeAdapter.prototype), { + ensureClient: async () => ({ session: { active: async () => ({}), inbox: { cancel: async (input: { inboxID: string }) => { calls.push(`cancel:${input.inboxID}`); state = "missing"; } } } }), + agentMessageState: async () => state, + prompt: async (_sessionId: string, _text: string, _model: unknown, messageId: string) => { calls.push(`admit:${messageId}`); admitted(); await barrier; state = "enqueued"; }, + }) as OpencodeRuntimeAdapter; + const continuation = adapter.continueInteraction("s", "msg_fixed", "answer"); + await entered; + const stopping = adapter.stopInteractionInput("s", "msg_fixed"); + await Promise.resolve(); + expect(calls).toEqual(["admit:msg_fixed"]); + release(); + expect(await continuation).toBe("started"); + expect(await stopping).toBe("cancelled"); + expect(calls).toEqual(["admit:msg_fixed", "cancel:msg_fixed"]); +}); +test("completed-answer recovery uses the matching user boundary and refuses missing final evidence", async () => { + let messages: unknown[] = [ + { id: "later-answer", type: "assistant", time: { completed: 1 }, finish: "stop", content: [{ type: "text", text: "Unrelated answer" }] }, + { id: "later-user", type: "user" }, + { id: "answer", type: "assistant", time: { completed: 1 }, finish: "stop", content: [{ type: "text", text: "Saved answer result" }] }, + { id: "msg_fixed", type: "user" }, + ]; + const adapter = Object.assign(Object.create(OpencodeRuntimeAdapter.prototype), { + ensureClient: async () => ({ message: { list: async () => ({ data: messages }) } }), + }) as OpencodeRuntimeAdapter; + expect(await adapter.readInteractionResult("s", "msg_fixed")).toBe("Saved answer result"); + messages = [{ id: "partial", type: "assistant", time: {}, content: [{ type: "text", text: "Partial output" }] }, { id: "msg_fixed", type: "user" }]; + await expect(adapter.readInteractionResult("s", "msg_fixed")).rejects.toThrow("do not resend"); + await expect(adapter.readInteractionResult("s", "outside-window")).rejects.toThrow("outside the recovery window"); +}); +test("strict waiting boundary requires verified idle and rejects an exhausted deadline", async () => { + const adapter = Object.assign(Object.create(OpencodeRuntimeAdapter.prototype), { + ensureClient: async () => ({ session: { active: async () => ({}) } }), + }) as OpencodeRuntimeAdapter; + await expect(adapter.waitForSessionIdle("s", 100, true)).resolves.toBeUndefined(); + await expect(adapter.waitForSessionIdle("s", 0, true)).rejects.toThrow("waiting boundary"); +}); diff --git a/tests/sessions/interactionDeliveryReconciliation.test.ts b/tests/sessions/interactionDeliveryReconciliation.test.ts new file mode 100644 index 0000000..4ab9a9c --- /dev/null +++ b/tests/sessions/interactionDeliveryReconciliation.test.ts @@ -0,0 +1,36 @@ +import { expect, it, mock } from "bun:test"; +import type { InteractionRequest } from "../../src/chat/interaction-request-contract.js"; +import { reconcileInteractionDelivery } from "../../src/sessions/interaction-delivery-reconciliation.js"; +import { InteractionRequestError } from "../../src/sessions/interaction-request-store.js"; + +const context = { actorKey: "actor", projectKey: "project", sessionId: "session" }; +const record = { interaction_id: "question", status: "answered", delivery: { native_message_id: "native", revision: 2, status: "blocked", response_id: "response" } } as InteractionRequest; + +it("recovers a lost acknowledgement only from consumed input and completed output without inference", async () => { + const recovered = { ...record, delivery: { ...record.delivery!, revision: 3, status: "delivered" as const } }; + const updateDelivery = mock(async () => recovered); + const store = { updateDelivery } as never; + const runtime = { agentMessageState: mock(async () => "delivered" as const), readInteractionResult: mock(async () => "complete result") }; + expect(await reconcileInteractionDelivery(context, record, store, runtime)).toEqual(recovered); + expect(updateDelivery).toHaveBeenCalledWith(context, "question", { expected_revision: 2, status: "delivered", response_id: "response" }); +}); + +it("retains waiting, running, incomplete and unavailable native state", async () => { + const updateDelivery = mock(async () => record); + const store = { updateDelivery } as never; + const complete = { agentMessageState: async () => "delivered" as const, readInteractionResult: async () => "result" }; + for (const runtime of [ + { ...complete, agentMessageState: async () => "enqueued" as const }, + { ...complete, agentMessageState: async () => "missing" as const }, + { ...complete, readInteractionResult: async () => { throw new Error("partial output"); } }, + { ...complete, agentMessageState: async () => { throw new Error("offline"); } } + ]) expect(await reconcileInteractionDelivery(context, record, store, runtime)).toEqual(record); + expect(await reconcileInteractionDelivery(context, record, store, complete, true)).toEqual(record); + expect(updateDelivery).not.toHaveBeenCalled(); +}); + +it("keeps concurrent cancellation instead of overwriting a newer delivery revision", async () => { + const cancelled = { ...record, delivery: { ...record.delivery!, revision: 3, status: "cancelled" as const } }; + const store = { updateDelivery: async () => { throw new InteractionRequestError("revision_conflict"); }, get: async () => cancelled } as never; + expect(await reconcileInteractionDelivery(context, record, store, { agentMessageState: async () => "delivered", readInteractionResult: async () => "result" })).toEqual(cancelled); +}); diff --git a/tests/sessions/interactionRequest.test.ts b/tests/sessions/interactionRequest.test.ts new file mode 100644 index 0000000..c648775 --- /dev/null +++ b/tests/sessions/interactionRequest.test.ts @@ -0,0 +1,110 @@ +import { afterEach, expect, test } 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 { InteractionRequestStore } from "../../src/sessions/interaction-request-store.js"; +import { INTERACTION_VERSION } from "../../src/chat/interaction-request-contract.js"; +import { createInteractionReplyMessage, projectInteractionReplyMessages } from "../../src/chat/interaction-message-presentation.js"; +import { renderInteractionAnswer } from "../../src/chat/interaction-request-contract.js"; +const roots: string[] = []; +afterEach(async () => { await Promise.all(roots.splice(0).map((p) => rm(p, { recursive: true, force: true }))); }); +const context = { sessionId: "session-a", actorKey: "actor-a", projectKey: "project-a" }; +const questions = [{ question_id: "q", title: "Title", body: "Question", options: [{ id: "a", label: "Answer" }], mode: "single" as const, allow_custom: true, required: true }]; +async function setup() { + const root = await mkdtemp(join(tmpdir(), "interaction-test-")); roots.push(root); + const workspace = new SessionWorkspaceStore(root); await workspace.ensure(context); + const store = new InteractionRequestStore(workspace); + const request = await store.create(context, { message_id: "message", turn_id: "turn", call_id: "call", questions }); + return { store, request, workspace }; +} +test("draft has independent CAS and atomic reply keeps one stable outbox across retries and reload", async () => { + const { store, request, workspace } = await setup(); + const answers = [{ question_id: "q", selected_option_ids: ["a"], custom_text: "" }]; + await store.updateDraft(context, request.interaction_id, { version: INTERACTION_VERSION, request_id: "draft", expected_draft_revision: 0, answers }); + const input = { version: INTERACTION_VERSION, request_id: "reply", expected_revision: 0, answers }; + const results = await Promise.allSettled([store.reply(context, request.interaction_id, input), store.reply(context, request.interaction_id, { ...input, request_id: "competing" })]); + expect(results.filter((r) => r.status === "fulfilled")).toHaveLength(1); + const record = await store.reply(context, request.interaction_id, input); + expect(record.delivery?.native_message_id).toMatch(/^msg_[a-f0-9]{32}$/); + expect(record.answers?.[0]?.selected_options[0]?.label).toBe("Answer"); + expect(await new InteractionRequestStore(workspace).get(context, request.interaction_id)).toEqual(record); + await expect(store.updateDraft(context, request.interaction_id, { version: INTERACTION_VERSION, request_id: "late", expected_draft_revision: 1, answers })).rejects.toMatchObject({ code: "interaction_resolved" }); + await expect(store.create(context, { message_id: "new", turn_id: "new", call_id: "new", questions })).rejects.toMatchObject({ code: "interaction_unresolved" }); +}); +test("restores accepted replies from scoped delivery IDs without rewriting ordinary text or stored history", async () => { + const { store, request, workspace } = await setup(); + const record = await store.reply(context, request.interaction_id, { version: INTERACTION_VERSION, request_id: "reply", expected_revision: 0, + answers: [{ question_id: "q", selected_option_ids: ["a"], custom_text: "现场确认" }] }); + const reply = createInteractionReplyMessage(record); + expect(reply.parts[0]).toEqual({ type: "text", text: "Title: Answer；现场确认" }); + expect(renderInteractionAnswer(record)).toBe("用户提交的提问回复：\nTitle: Answer；现场确认"); + const ordinary = { id: "ordinary", role: "user", parts: [{ type: "text", text: "用户提交的提问回复：手动输入" }] }; + const original = [ordinary, { id: reply.id, role: "user", content: renderInteractionAnswer(record), parts: [ + { type: "text", text: renderInteractionAnswer(record) }, { type: "data-task_context", data: { task_id: "task" } } + ] }, { id: reply.id, role: "assistant", parts: [{ type: "text", text: "原回复" }] }]; + const before = JSON.stringify(original); + const records = await store.forMessages(context, new Set([reply.id])); + const projected = projectInteractionReplyMessages(original, records) as typeof original; + expect(projected[0]).toBe(ordinary); + expect(projected[1]).toMatchObject({ content: "Title: Answer；现场确认", parts: [...reply.parts, original[1]!.parts[1]!] }); + expect(projected[2]).toBe(original[2]); + expect(JSON.stringify(original)).toBe(before); + expect(await store.forMessages(context, new Set(["other-message"]))).toEqual([]); + const otherContext = { ...context, sessionId: "other-session" }; + await workspace.ensure(otherContext); + expect(await store.forMessages(otherContext, new Set([reply.id]))).toEqual([]); + await expect(store.forMessages({ ...context, actorKey: "other-actor" }, new Set([reply.id]))).rejects.toThrow(); +}); +test("strict definitions, answers, deduplication, and deletion boundary", async () => { + const { store, request, workspace } = await setup(); + await expect(store.create(context, { message_id: "unanswerable", turn_id: "bad", call_id: "bad", questions: [{ ...questions[0], options: [], allow_custom: false }] })).rejects.toThrow("question must offer an answer input"); + expect(await store.create(context, { message_id: "message", turn_id: "turn", call_id: "call", questions })).toEqual(request); + await expect(store.reply(context, request.interaction_id, { version: INTERACTION_VERSION, request_id: "bad", expected_revision: 0, answers: [{ question_id: "q", selected_option_ids: ["unknown"], custom_text: "" }] })).rejects.toMatchObject({ code: "invalid_option" }); + await expect(store.skip(context, request.interaction_id, { version: INTERACTION_VERSION, request_id: "skip", expected_revision: 0 })).rejects.toMatchObject({ code: "required_question" }); + await workspace.markDeleted(context); + await expect(store.get(context, request.interaction_id)).rejects.toThrow(); + await expect(store.cancelPending(context)).rejects.toThrow(); +}); +test("delivery CAS preserves native identity and terminal delivery cannot be overwritten", async () => { + const { store, request } = await setup(); + const record = await store.reply(context, request.interaction_id, { version: INTERACTION_VERSION, request_id: "reply", expected_revision: 0, answers: [{ question_id: "q", selected_option_ids: [], custom_text: "custom" }] }); + const claimed = await store.claimDelivery(context, request.interaction_id, { expected_revision: 0, response_id: "response" }); + expect(claimed.delivery?.native_message_id).toBe(record.delivery?.native_message_id); + await expect(store.claimDelivery(context, request.interaction_id, { expected_revision: 0, response_id: "second" })).rejects.toMatchObject({ code: "revision_conflict" }); + await store.updateDelivery(context, request.interaction_id, { expected_revision: 1, status: "delivered", response_id: "response" }); + expect(await store.getBlocking(context)).toBeNull(); + await expect(store.updateDelivery(context, request.interaction_id, { expected_revision: 2, status: "queued" })).rejects.toMatchObject({ code: "delivery_resolved" }); + const resume = { version: INTERACTION_VERSION, request_id: "resume", expected_revision: 2 }; + expect(await store.resume(context, request.interaction_id, resume)).toEqual(await store.resume(context, request.interaction_id, resume)); +}); +test("bounded draft receipts do not resurrect pending state and newest page restores active request", async () => { + const { store, request, workspace } = await setup(); + const draft = { version: INTERACTION_VERSION, request_id: "draft-0", expected_draft_revision: 0, answers: [] }; + await store.updateDraft(context, request.interaction_id, draft); + const cancel = { version: INTERACTION_VERSION, request_id: "cancel", expected_revision: 0 }; + await store.cancel(context, request.interaction_id, cancel); + expect((await store.updateDraft(context, request.interaction_id, draft)).status).toBe("cancelled"); + let newest = request; + for (let i = 0; i < 22; i++) { + newest = await store.create(context, { message_id: `m${i}`, turn_id: `t${i}`, call_id: `c${i}`, questions }); + if (i < 21) await store.cancel(context, newest.interaction_id, { ...cancel, request_id: `cancel-${i}` }); + } + const page = await store.list(context); + expect(page.items[0]?.interaction_id).toBe(newest.interaction_id); + expect((await store.list(context, { cursor: page.next_cursor })).items).toHaveLength(3); + for (let i = 0; i < 40; i++) await store.updateDraft(context, newest.interaction_id, { ...draft, request_id: `d${i}`, expected_draft_revision: i, answers: [{ question_id: "q", selected_option_ids: [], custom_text: "x".repeat(2000) }] }); + const state = await Bun.file(workspace.path(context, "interaction-requests.json")).json(); + expect(state.receipts.length).toBeLessThanOrEqual(32); + expect(Buffer.byteLength(JSON.stringify(state.receipts))).toBeLessThanOrEqual(128 * 1024); +}, 20000); +test("complete oversized reply projection is rejected atomically instead of truncated", async () => { + const { store } = await setup(); + await store.cancelPending(context); + const qs = Array.from({ length: 3 }, (_, i) => ({ question_id: `q${i}`, title: "T".repeat(200), body: "?", options: Array.from({ length: 12 }, (_, j) => ({ id: `o${j}`, label: "L".repeat(200) })), mode: "multiple" as const, allow_custom: true, required: true })); + const record = await store.create(context, { message_id: "large", turn_id: "large", call_id: "large", questions: qs }); + await expect(store.reply(context, record.interaction_id, { version: INTERACTION_VERSION, request_id: "large", expected_revision: 0, answers: qs.map((q) => ({ question_id: q.question_id, selected_option_ids: q.options.map((o) => o.id), custom_text: "C".repeat(900) })) })).rejects.toMatchObject({ code: "answer_too_large" }); + const unchanged = await store.get(context, record.interaction_id); + expect(unchanged?.status).toBe("pending"); + expect(unchanged?.delivery).toBeUndefined(); +});