From 4ec010455b2e3ead28a1ac4a9b3eed97fcfef25b Mon Sep 17 00:00:00 2001 From: Huarch Date: Wed, 26 Aug 2026 12:26:29 +0800 Subject: [PATCH] fix(agent): stabilize sandboxed streaming workflows --- .opencode/agents/instruction.md | 2 + .../hydraulic-bottleneck-analysis/SKILL.md | 3 + .../workflow/service-area-analysis/SKILL.md | 3 + .../scripts/service_area_partition.py | 16 +- Dockerfile | 29 ++ README.md | 12 + patches/opencode-1.18.13-message-phase.patch | 186 +++++++++++ src/routes/chatStream.ts | 191 +++++++++++- src/routes/chatStreamEvents.ts | 18 ++ src/runtime/opencode.ts | 60 +++- tests/routes/chatStream.test.ts | 293 +++++++++++++++++- tests/runtime/opencode.test.ts | 137 +++++++- 12 files changed, 916 insertions(+), 34 deletions(-) create mode 100644 patches/opencode-1.18.13-message-phase.patch diff --git a/.opencode/agents/instruction.md b/.opencode/agents/instruction.md index fd8e1b4..95e457a 100644 --- a/.opencode/agents/instruction.md +++ b/.opencode/agents/instruction.md @@ -10,6 +10,8 @@ model: deepseek/deepseek-v4-flash - 工具执行期间不输出过程说明，全部完成后只回复最终结果 - 直接给出结论、关键数据和可执行建议，默认仅展示最重要的 Top 5；数据不足或任务失败时简要说明影响和下一步 - 多步骤或预计超过 30 秒的任务，开始时使用 `todowrite` 给用户展示计划，并在每个里程碑更新状态；简单问答不创建计划 +- `todowrite` 是面向用户的业务任务摘要：每项只描述目标或可验证结果，不出现函数名、脚本/文件名、命令、工具名、参数、内部目录或具体修复实现；这些技术细节仅保留在工具过程信息中 +- 任务标题使用简洁的业务语言，例如“准备供水分区所需数据”“计算供水服务范围”“生成并展示分析结果”“整理可复用分析经验” ## 工作流生命周期 diff --git a/.opencode/skills/workflow/hydraulic-bottleneck-analysis/SKILL.md b/.opencode/skills/workflow/hydraulic-bottleneck-analysis/SKILL.md index 52529b7..186d251 100644 --- a/.opencode/skills/workflow/hydraulic-bottleneck-analysis/SKILL.md +++ b/.opencode/skills/workflow/hydraulic-bottleneck-analysis/SKILL.md @@ -148,3 +148,6 @@ if length < 0.01 and headloss > 0.5 → "短管高水损，检查是否存在模 - 水头损失百分位阈值（P80/P90）基于**全管网**统计，如果管网上游存在极端水损（如 400m+），会拉高整体 P 值，导致部分中高水损管段被漏判。极端场景下可考虑对水损做分位数裁剪（如排除 >P99.9 的离群值）后再计算 P80/P90。 - **setting 字段不可用**：`data timeseries realtime links` 返回的 `setting` 值为无效数据，本工作流已移除所有基于 setting 的阀门节流 / 水泵出口判定。若需要此类判定，应通过 `network get-link-properties` 逐条获取属性中的 setting 作为替代。 - 脚本读取全量 JSON 入内存，峰值内存约 200-300MB，需确保执行环境有足够内存。 + +## Learned Patterns +- [5ba58cd24c9cea84b6ab5861] **数据 schema 实测适配（2026-04 验证）**：`data timeseries realtime links` 返回记录的管道主键为 `link_id`（不是 `id`），`data timeseries realtime nodes` 返回记录的节点主键为 `node_id`（不是 `id`），且 `time` 字段为 UTC 格式（如 `2026-04-01T00:00:00+00:00`）。运行 `bottleneck_analysis.py` 前需：① 脚本内将 `r['id']` 改为 `r['link_id']`、`n['id']` 改为 `n['node_id']`；② `--target-tim... diff --git a/.opencode/skills/workflow/service-area-analysis/SKILL.md b/.opencode/skills/workflow/service-area-analysis/SKILL.md index 790cf74..2c1cfa4 100644 --- a/.opencode/skills/workflow/service-area-analysis/SKILL.md +++ b/.opencode/skills/workflow/service-area-analysis/SKILL.md @@ -144,3 +144,6 @@ show_chart(title="各水源分区节点数/压力对比", chart_type="bar", ...) - **水库顺序敏感**：多源 BFS 中先遍历到的水库优先分配，不同水库启动顺序可能影响边界区域分配结果 - **单时刻快照**：分区仅反映目标时刻的水力工况，不同时段的泵站启停、阀门切换可能导致分区边界变化 - **零流量阈值**：`1e-6` 阈值过滤极低流量管段，若管网有长期小流量管段可能漏判方向 + +## Learned Patterns +- [6614a914a8f7dcc2fc34c1ba] 实时数据时间匹配必须用 norm_time 归一化为 UTC 再比较（数据 time 为 UTC 格式如 2026-06-03T00:00:00+00:00，target-time 传 +08:00 会因字符串不等而筛出 0 条）；links/nodes 记录主键字段为 link_id/node_id（不是 id）。执行分区前先确认目标时刻存在实时数据，可用 `data timeseries realtime simulation-by-id-time` 探测；数据可能只覆盖某几天（本模型覆盖 2026-06-03 附近，2026-04 与 2026-08 均无数据）。 diff --git a/.opencode/skills/workflow/service-area-analysis/scripts/service_area_partition.py b/.opencode/skills/workflow/service-area-analysis/scripts/service_area_partition.py index 1a96a6a..fb8f538 100644 --- a/.opencode/skills/workflow/service-area-analysis/scripts/service_area_partition.py +++ b/.opencode/skills/workflow/service-area-analysis/scripts/service_area_partition.py @@ -29,6 +29,13 @@ COLORS = [ "rgba(204,153,204,0.7)", "rgba(153,153,153,0.7)" ] + +def norm_time(t): + """把 ISO8601 字符串归一化为 UTC 的 ISO 字符串，用于跨时区比较""" + from datetime import datetime + return datetime.fromisoformat(t.replace("Z", "+00:00")).astimezone( + __import__("datetime").timezone.utc).isoformat() + def load_json(path, label): print(f"Loading {label}...", file=sys.stderr) with open(path) as f: @@ -63,11 +70,12 @@ def main(): # --- Step 3: Load link flow at target time --- ldata = load_json(args.links, "link flows")["data"] - target_links = [l for l in ldata if l["time"] == args.target_time] + target_ts = norm_time(args.target_time) + target_links = [l for l in ldata if norm_time(l["time"]) == target_ts] flow_direction = {} pipe_flow = {} for l in target_links: - lid = l["id"] + lid = l.get("link_id") or l.get("id") flow_val = l["flow"] pipe_flow[lid] = abs(flow_val) if lid in pipe_topology: @@ -81,11 +89,11 @@ def main(): # --- Step 4: Load node data at target time --- ndata = load_json(args.nodes, "node data")["data"] - target_nodes = [n for n in ndata if n["time"] == args.target_time] + target_nodes = [n for n in ndata if norm_time(n["time"]) == target_ts] node_pressure = {} node_demand = {} for n in target_nodes: - nid = n["id"] + nid = n.get("node_id") or n.get("id") node_pressure[nid] = n.get("pressure", 0) node_demand[nid] = n.get("actual_demand", 0) print(f" {len(target_nodes)} nodes", file=sys.stderr) diff --git a/Dockerfile b/Dockerfile index d28d03b..85b8ce7 100644 --- a/Dockerfile +++ b/Dockerfile @@ -48,6 +48,31 @@ RUN if [ -n "${UBUNTU_APT_MIRROR}" ]; then \ (getent passwd 10001 >/dev/null || useradd --uid 10001 --gid 10001 --no-create-home --shell /usr/sbin/nologin tjwater-sandbox) && \ rm -rf /var/lib/apt/lists/* +FROM base AS opencode-builder + +ARG OPENCODE_SOURCE_COMMIT=a105350812f05f914c768e468559dbd6bd508d8e +ARG OPENCODE_PATCH_VERSION=1.18.13-tjwater.1 +WORKDIR /tmp/opencode-src + +RUN git init . && \ + git remote add origin https://github.com/anomalyco/opencode.git && \ + git fetch --depth 1 origin "$OPENCODE_SOURCE_COMMIT" && \ + git checkout --detach FETCH_HEAD + +COPY patches/opencode-1.18.13-message-phase.patch /tmp/opencode-message-phase.patch + +RUN git apply --check /tmp/opencode-message-phase.patch && \ + git apply /tmp/opencode-message-phase.patch && \ + bun install --frozen-lockfile --ignore-scripts && \ + bun test --cwd packages/llm test/provider/openai-responses.test.ts && \ + OPENCODE_VERSION="$OPENCODE_PATCH_VERSION" bun run --cwd packages/opencode build --single --skip-install --skip-embed-web-ui && \ + case "$(uname -m)" in \ + x86_64) binary=packages/opencode/dist/opencode-linux-x64/bin/opencode ;; \ + aarch64|arm64) binary=packages/opencode/dist/opencode-linux-arm64/bin/opencode ;; \ + *) echo "unsupported OpenCode build architecture: $(uname -m)" >&2; exit 1 ;; \ + esac && \ + install -D -m 0755 "$binary" /out/opencode + FROM base AS deps WORKDIR /app @@ -70,6 +95,7 @@ COPY .opencode ./.opencode RUN bun run check FROM build AS test +COPY --from=opencode-builder /out/opencode /usr/local/bin/opencode RUN apt-get update && apt-get install -y --no-install-recommends nodejs && \ rm -rf /var/lib/apt/lists/* COPY contracts ./contracts @@ -81,6 +107,9 @@ RUN bun run test:ci FROM build AS runner WORKDIR /app +COPY --from=opencode-builder /out/opencode /usr/local/bin/opencode +RUN opencode --version + ENV NODE_ENV=production ENV HOST=0.0.0.0 ENV PORT=8787 diff --git a/README.md b/README.md index 2f4e921..0d6b465 100644 --- a/README.md +++ b/README.md @@ -76,6 +76,18 @@ TJWATER_API_BASE_URL=http://127.0.0.1:8000 当前仅支持 Embedded 模式，不支持连接外部 OpenCode server。 +生产镜像会从固定的 OpenCode `v1.18.13` 源码提交构建 CLI，并应用仓库内的 +`patches/opencode-1.18.13-message-phase.patch`。该补丁只透传 OpenAI Responses +输出项已有的 `commentary` / `final_answer` phase，不改变模型行为：过程文本继续写入 +可折叠的 Agent 过程卡，`final_answer` 到达后立即按增量写入正式回答。未提供 phase 的 +DeepSeek 模型启用 OpenCode 1.18.13 内置的 JSON Schema 最终回答工具 +`StructuredOutput`：模型必须先完成全部分析和工具调用，再把完整回答写入 `answer`； +该工具成功后 OpenCode 会直接结束运行循环，不再进入下一轮模型或工具调用。Agent 将 +`answer` 映射为正式文本推送；若模型未按协议调用该工具，仍保留会话 idle 后提取最终 +文本的兼容兜底。 +本地直接运行 `bun --watch src/server.ts` 时，`PATH` 中也需要放置应用了同一补丁的 +`opencode` CLI，才能启用 phase 驱动的正式文本流式输出。 + ## 认证续期与学习工具 后端工具调用遇到即将过期的 access token 或首次 `401` 时，Agent 会通过当前 SSE 流发送 `credential_refresh_required`。前端使用服务端保存的 Keycloak refresh token 强制换取新 access token，再调用 `POST /api/v1/agent/sessions/{session_id}/credential-refreshes` 唤醒原工具调用。等待上限为 30 秒，同一会话的并发请求合并为一次续期，原调用最多重试一次；`403` 不触发续期。 diff --git a/patches/opencode-1.18.13-message-phase.patch b/patches/opencode-1.18.13-message-phase.patch new file mode 100644 index 0000000..148e9c4 --- /dev/null +++ b/patches/opencode-1.18.13-message-phase.patch @@ -0,0 +1,186 @@ +diff --git a/package.json b/package.json +index 15725c8..4378766 100644 +--- a/package.json ++++ b/package.json +@@ -4,7 +4,7 @@ + "description": "AI-powered development tool", + "private": true, + "type": "module", +- "packageManager": "bun@1.3.14", ++ "packageManager": "bun@1.3.13", + "scripts": { + "dev": "bun run --cwd packages/opencode --conditions=browser src/index.ts", + "dev:desktop": "bun --cwd packages/desktop dev", +diff --git a/packages/llm/src/protocols/openai-responses.ts b/packages/llm/src/protocols/openai-responses.ts +index 4936d31..53138af 100644 +--- a/packages/llm/src/protocols/openai-responses.ts ++++ b/packages/llm/src/protocols/openai-responses.ts +@@ -177,6 +177,7 @@ type OpenAIResponsesUsage = Schema.Schema.Type + const OpenAIResponsesStreamItem = Schema.Struct({ + type: Schema.String, + id: Schema.optional(Schema.String), ++ phase: optionalNull(Schema.String), + call_id: Schema.optional(Schema.String), + name: Schema.optional(Schema.String), + arguments: Schema.optional(Schema.String), +@@ -238,6 +239,7 @@ interface ParserState { + readonly hasFunctionCall: boolean + readonly lifecycle: Lifecycle.State + readonly reasoningItems: Readonly> ++ readonly textMetadata: Readonly> + readonly store: boolean | undefined + } + +@@ -614,9 +616,19 @@ const TERMINAL_TYPES = new Set(["response.completed", "response.incomplete", "re + + const onOutputTextDelta = (state: ParserState, event: OpenAIResponsesEvent): StepResult => { + if (!event.delta) return [state, NO_EVENTS] ++ const itemID = event.item_id ?? "text-0" + const events: LLMEvent[] = [] + return [ +- { ...state, lifecycle: Lifecycle.textDelta(state.lifecycle, events, event.item_id ?? "text-0", event.delta) }, ++ { ++ ...state, ++ lifecycle: Lifecycle.textDelta( ++ state.lifecycle, ++ events, ++ itemID, ++ event.delta, ++ state.textMetadata[itemID], ++ ), ++ }, + events, + ] + } +@@ -655,6 +667,22 @@ const reasoningMetadata = (item: OpenAIResponsesStreamItem & { id: string }) => + // best-effort, not guaranteed. + const onOutputItemAdded = (state: ParserState, event: OpenAIResponsesEvent): StepResult => { + const item = event.item ++ if (item?.type === "message" && item.id) { ++ const phase = item.phase === "commentary" || item.phase === "final_answer" ? item.phase : undefined ++ return [ ++ { ++ ...state, ++ textMetadata: { ++ ...state.textMetadata, ++ [item.id]: openaiMetadata({ ++ itemId: item.id, ++ ...(phase ? { phase } : {}), ++ }), ++ }, ++ }, ++ NO_EVENTS, ++ ] ++ } + if (item && isReasoningItem(item)) { + const events: LLMEvent[] = [] + return [ +@@ -812,6 +840,20 @@ const onOutputItemDone = Effect.fn("OpenAIResponses.onOutputItemDone")(function* + const item = event.item + if (!item) return [state, NO_EVENTS] satisfies StepResult + ++ if (item.type === "message" && item.id) { ++ const events: LLMEvent[] = [] ++ const lifecycle = Lifecycle.textEnd(state.lifecycle, events, item.id, state.textMetadata[item.id]) ++ const { [item.id]: _removed, ...textMetadata } = state.textMetadata ++ return [ ++ { ++ ...state, ++ lifecycle, ++ textMetadata, ++ }, ++ events, ++ ] satisfies StepResult ++ } ++ + if (item.type === "function_call") { + if (!item.id || !item.call_id || !item.name) return [state, NO_EVENTS] satisfies StepResult + const tools = state.tools[item.id] +@@ -969,6 +1011,7 @@ export const protocol = Protocol.make({ + tools: ToolStream.empty(), + lifecycle: Lifecycle.initial(), + reasoningItems: {}, ++ textMetadata: {}, + store: OpenAIOptions.store(request), + }), + step, +diff --git a/packages/llm/src/protocols/utils/lifecycle.ts b/packages/llm/src/protocols/utils/lifecycle.ts +index eb6c95d..64248be 100644 +--- a/packages/llm/src/protocols/utils/lifecycle.ts ++++ b/packages/llm/src/protocols/utils/lifecycle.ts +@@ -14,13 +14,22 @@ export const stepStart = (state: State, events: LLMEvent[]): State => { + return { ...state, stepStarted: true } + } + +-export const textDelta = (state: State, events: LLMEvent[], id: string, text: string): State => { ++export const textDelta = ( ++ state: State, ++ events: LLMEvent[], ++ id: string, ++ text: string, ++ providerMetadata?: ProviderMetadata, ++): State => { + const stepped = stepStart(state, events) + if (stepped.text.has(id)) { +- events.push(LLMEvent.textDelta({ id, text })) ++ events.push(LLMEvent.textDelta({ id, text, providerMetadata })) + return stepped + } +- events.push(LLMEvent.textStart({ id }), LLMEvent.textDelta({ id, text })) ++ events.push( ++ LLMEvent.textStart({ id, providerMetadata }), ++ LLMEvent.textDelta({ id, text, providerMetadata }), ++ ) + return { ...stepped, text: new Set([...stepped.text, id]) } + } + +diff --git a/packages/llm/test/provider/openai-responses.test.ts b/packages/llm/test/provider/openai-responses.test.ts +index cd8bad5..89ae576 100644 +--- a/packages/llm/test/provider/openai-responses.test.ts ++++ b/packages/llm/test/provider/openai-responses.test.ts +@@ -754,6 +754,45 @@ describe("OpenAI Responses route", () => { + }), + ) + ++ it.effect("preserves assistant message phase in text provider metadata", () => ++ Effect.gen(function* () { ++ const message = { ++ type: "message", ++ id: "msg_final", ++ phase: "final_answer", ++ } ++ const body = sseEvents( ++ { type: "response.output_item.added", item: message }, ++ { type: "response.output_text.delta", item_id: "msg_final", delta: "Final" }, ++ { type: "response.output_item.done", item: message }, ++ { type: "response.completed", response: { id: "resp_1" } }, ++ ) ++ ++ const response = yield* LLMClient.generate(request).pipe(Effect.provide(fixedResponse(body))) ++ const metadata = { ++ openai: { ++ itemId: "msg_final", ++ phase: "final_answer", ++ }, ++ } ++ ++ expect(response.text).toBe("Final") ++ expect(response.events).toMatchObject([ ++ { type: "step-start", index: 0 }, ++ { type: "text-start", id: "msg_final", providerMetadata: metadata }, ++ { ++ type: "text-delta", ++ id: "msg_final", ++ text: "Final", ++ providerMetadata: metadata, ++ }, ++ { type: "text-end", id: "msg_final", providerMetadata: metadata }, ++ { type: "step-finish", index: 0, reason: "stop" }, ++ { type: "finish", reason: "stop" }, ++ ]) ++ }), ++ ) ++ + it.effect("parses reasoning summary stream fixtures", () => + Effect.gen(function* () { + const body = sseEvents( diff --git a/src/routes/chatStream.ts b/src/routes/chatStream.ts index 51a5634..b9a1871 100644 --- a/src/routes/chatStream.ts +++ b/src/routes/chatStream.ts @@ -2,6 +2,7 @@ import type { Event as OpencodeEvent, Part } from "@opencode-ai/sdk/v2"; import { writeLlmRequestAuditLog } from "../audit/llmRequestAudit.js"; import { type SupportedModel } from "../chat/models.js"; +import { config } from "../config.js"; import { logger } from "../logger.js"; import { type PermissionReply, @@ -17,6 +18,7 @@ import { extractRequestReason, extractSkillAuditInfo, getErrorMessage, + getAssistantMessagePhase, getToolProgressTitle, getUnknownErrorMessage, hasToolParams, @@ -43,6 +45,7 @@ import { normalizeToolStatus, type PermissionRequestPayload, type QuestionRequestPayload, + type AssistantMessagePhase, type TodoItemPayload, type TodoUpdatePayload, } from "./chatStreamEvents.js"; @@ -112,17 +115,59 @@ const toRuntimeModel = (model?: SupportedModel) => { }; }; +const STRUCTURED_OUTPUT_TOOL_NAME = "StructuredOutput"; +const STRUCTURED_FINAL_ANSWER_FORMAT = { + type: "json_schema" as const, + schema: { + type: "object", + properties: { + answer: { + type: "string", + description: "直接展示给用户的完整最终回答，使用简体中文和 Markdown。", + }, + }, + required: ["answer"], + additionalProperties: false, + }, +}; + +const requiresStructuredFinalAnswer = (model?: SupportedModel) => + (model ?? config.OPENCODE_MODEL).split("/", 1)[0] === "deepseek"; + +const extractStructuredAnswer = (value: unknown) => + isObjectRecord(value) && typeof value.answer === "string" + ? value.answer.trim() + : ""; + const emitFinalMessage = async ( runtime: OpencodeRuntimeAdapter, sessionId: string, clientSessionId: string, currentAssistantMessageIds: Set, assistantTextParts: Map>, + assistantTextPartPhases: Map, write: (event: string, data: Record) => void, ) => { let text = [...currentAssistantMessageIds] .reverse() - .map((messageId) => [...(assistantTextParts.get(messageId)?.values() ?? [])].join("")) + .map((messageId) => { + const parts = assistantTextParts.get(messageId); + if (!parts) { + return ""; + } + const candidates = [...parts.entries()]; + const finalText = candidates + .filter(([partId]) => assistantTextPartPhases.get(partId) === "final_answer") + .map(([, content]) => content) + .join(""); + if (finalText) { + return finalText; + } + return candidates + .filter(([partId]) => assistantTextPartPhases.get(partId) !== "commentary") + .map(([, content]) => content) + .join(""); + }) .find((content) => content.length > 0) ?? ""; if (!text) { @@ -135,7 +180,20 @@ const emitFinalMessage = async ( (currentAssistantMessageIds.size === 0 || currentAssistantMessageIds.has(message.info.id)), ); - text = collectTextContent(assistantMessage?.parts ?? []); + const structured = + assistantMessage && "structured" in assistantMessage.info + ? assistantMessage.info.structured + : undefined; + const structuredAnswer = extractStructuredAnswer(structured); + text = + structuredAnswer || + collectTextContent( + (assistantMessage?.parts ?? []).filter( + (part) => + part.type !== "text" || + getAssistantMessagePhase(part.metadata) !== "commentary", + ), + ); } if (text) { @@ -165,7 +223,7 @@ export const streamPromptResponse = async ({ failed: boolean; toolCallCount: number; }> => { - const eventStream = await runtime.subscribeEvents(); + const eventStream = await runtime.subscribeEvents(workspaceRoot); const iterator = eventStream[Symbol.asyncIterator](); const requestStartedAt = Date.now(); const promptStartedAt = Date.now(); @@ -176,6 +234,8 @@ export const streamPromptResponse = async ({ const emittedQuestionRequestIds = new Set(); const currentAssistantMessageIds = new Set(); const assistantTextParts = new Map>(); + const assistantTextPartPhases = new Map(); + const emittedFinalTextLengths = new Map(); const partTypes = new Map(); const pendingTextDeltas = new Map(); const reasoningStatuses = new Map(); @@ -192,6 +252,7 @@ export const streamPromptResponse = async ({ let promptSettled = false; let aborted = signal?.aborted ?? false; let failed = false; + let structuredFinalAnswerEmitted = false; const debugContext = { sessionId, clientSessionId, @@ -258,6 +319,37 @@ export const streamPromptResponse = async ({ }); }; + const emitFinalText = (partId: string, text: string) => { + const emittedLength = emittedFinalTextLengths.get(partId) ?? 0; + if (text.length <= emittedLength) { + return; + } + const content = text.slice(emittedLength); + emittedFinalTextLengths.set(partId, text.length); + emittedText = true; + write("token", { + session_id: clientSessionId, + content, + }); + }; + + const emitCommentaryProgress = ( + partId: string, + text: string, + completed: boolean, + ) => { + if (!text.trim()) { + return; + } + emitProgress({ + id: `commentary-${partId}`, + phase: "commentary", + status: completed ? "completed" : "running", + title: completed ? "Agent 过程已更新" : "Agent 正在处理", + detail: text, + }); + }; + emitProgress({ id: "request-received", phase: "start", @@ -267,7 +359,14 @@ export const streamPromptResponse = async ({ }); const promptPromise = runtime - .prompt(sessionId, message, toRuntimeModel(model)) + .prompt( + sessionId, + message, + toRuntimeModel(model), + requiresStructuredFinalAnswer(model) + ? { format: STRUCTURED_FINAL_ANSWER_FORMAT } + : undefined, + ) .then(() => { promptSettled = true; logDevelopmentDebug("runtime.prompt resolved", { @@ -425,6 +524,7 @@ export const streamPromptResponse = async ({ await runtime.replyPermission({ requestId: event.properties.id, sessionId, + directory: workspaceRoot, reply, }); write("permission_response", { @@ -481,6 +581,7 @@ export const streamPromptResponse = async ({ await runtime.replyPermission({ requestId: event.properties.id, sessionId, + directory: workspaceRoot, reply, }); write("permission_response", { @@ -674,11 +775,15 @@ export const streamPromptResponse = async ({ const partType = partTypes.get(event.properties.partID); if (partType === "text") { const messageParts = assistantTextParts.get(event.properties.messageID) ?? new Map(); - messageParts.set( - event.properties.partID, - `${messageParts.get(event.properties.partID) ?? ""}${event.properties.delta}`, - ); + const text = `${messageParts.get(event.properties.partID) ?? ""}${event.properties.delta}`; + messageParts.set(event.properties.partID, text); assistantTextParts.set(event.properties.messageID, messageParts); + const phase = assistantTextPartPhases.get(event.properties.partID) ?? "unknown"; + if (phase === "final_answer") { + emitFinalText(event.properties.partID, text); + } else if (phase === "commentary") { + emitCommentaryProgress(event.properties.partID, text, false); + } } else if (!partType) { const pending = pendingTextDeltas.get(event.properties.partID) ?? []; pending.push(event.properties.delta); @@ -695,11 +800,19 @@ export const streamPromptResponse = async ({ currentAssistantMessageIds.add(part.messageID); } if (part.type === "text") { + const phase = getAssistantMessagePhase(part.metadata); + assistantTextPartPhases.set(part.id, phase); const pendingText = (pendingTextDeltas.get(part.id) ?? []).join(""); pendingTextDeltas.delete(part.id); const messageParts = assistantTextParts.get(part.messageID) ?? new Map(); - messageParts.set(part.id, part.text || pendingText); + const text = part.text || pendingText; + messageParts.set(part.id, text); assistantTextParts.set(part.messageID, messageParts); + if (phase === "final_answer") { + emitFinalText(part.id, text); + } else if (phase === "commentary") { + emitCommentaryProgress(part.id, text, Boolean(part.time?.end)); + } } else { pendingTextDeltas.delete(part.id); } @@ -756,6 +869,47 @@ export const streamPromptResponse = async ({ }); } + if (part.tool === STRUCTURED_OUTPUT_TOOL_NAME) { + if (part.state.status === "error") { + logger.warn( + { + sessionId, + clientSessionId, + partId: part.id, + error: part.state.error, + }, + "structured final answer tool failed", + ); + continue; + } + if ( + (part.state.status !== "running" && + part.state.status !== "completed") || + structuredFinalAnswerEmitted + ) { + continue; + } + + const answer = extractStructuredAnswer(toolParams); + if (!answer) { + logger.warn( + { sessionId, clientSessionId, partId: part.id }, + "structured final answer tool received without an answer", + ); + continue; + } + + structuredFinalAnswerEmitted = true; + emitFinalText(part.id, answer); + logDevelopmentDebug("final answer submitted through StructuredOutput", { + ...debugContext, + partId: part.id, + answerChars: answer.length, + elapsedMs: Math.max(0, Date.now() - requestStartedAt), + }); + continue; + } + const questionToolPayload = normalizeQuestionToolPayload( part, toolParams, @@ -942,14 +1096,17 @@ export const streamPromptResponse = async ({ } await promptPromise; - emittedText = await emitFinalMessage( - runtime, - sessionId, - clientSessionId, - currentAssistantMessageIds, - assistantTextParts, - write, - ); + if (!emittedText) { + emittedText = await emitFinalMessage( + runtime, + sessionId, + clientSessionId, + currentAssistantMessageIds, + assistantTextParts, + assistantTextPartPhases, + write, + ); + } emitProgress({ id: "request-received", phase: "start", diff --git a/src/routes/chatStreamEvents.ts b/src/routes/chatStreamEvents.ts index 10cf841..67cf8de 100644 --- a/src/routes/chatStreamEvents.ts +++ b/src/routes/chatStreamEvents.ts @@ -57,6 +57,11 @@ export type TodoUpdatePayload = { created_at: number; }; +export type AssistantMessagePhase = + | "commentary" + | "final_answer" + | "unknown"; + const isDevelopmentDebugLoggingEnabled = process.env.NODE_ENV === "development"; const toolLabels: Record = { @@ -110,6 +115,19 @@ export const getUnknownErrorMessage = (error: unknown) => { export const isObjectRecord = (value: unknown): value is Record => typeof value === "object" && value !== null && !Array.isArray(value); +export const getAssistantMessagePhase = ( + metadata: unknown, +): AssistantMessagePhase => { + if (!isObjectRecord(metadata)) { + return "unknown"; + } + const openai = isObjectRecord(metadata.openai) ? metadata.openai : undefined; + const phase = openai?.phase ?? metadata.phase; + return phase === "commentary" || phase === "final_answer" + ? phase + : "unknown"; +}; + export const normalizeToolParams = (value: unknown): Record => { if (isObjectRecord(value)) { return value; diff --git a/src/runtime/opencode.ts b/src/runtime/opencode.ts index 0b01421..7b39981 100644 --- a/src/runtime/opencode.ts +++ b/src/runtime/opencode.ts @@ -1,6 +1,7 @@ import { createOpencode, type OpencodeClient, + type OutputFormat, } from "@opencode-ai/sdk/v2"; import { randomUUID } from "node:crypto"; import { existsSync, readFileSync } from "node:fs"; @@ -38,6 +39,10 @@ type RuntimeModelOverride = { modelID: string; }; +type RuntimePromptOptions = { + format?: OutputFormat; +}; + export type PermissionReply = "once" | "always" | "reject"; export type QuestionAnswers = string[][]; @@ -180,7 +185,12 @@ export class OpencodeRuntimeAdapter { return this.messages(sessionId); } - async prompt(sessionId: string, text: string, model?: RuntimeModelOverride) { + async prompt( + sessionId: string, + text: string, + model?: RuntimeModelOverride, + options?: RuntimePromptOptions, + ) { const client = await this.ensureClient(); const startedAt = Date.now(); logDevelopmentDebug( @@ -194,6 +204,7 @@ export class OpencodeRuntimeAdapter { await client.session.prompt({ sessionID: sessionId, model, + ...(options?.format ? { format: options.format } : {}), parts: [{ type: "text", text }], }); logDevelopmentDebug( @@ -299,35 +310,41 @@ export class OpencodeRuntimeAdapter { ); } - async subscribeEvents() { + async subscribeEvents(directory?: string) { const client = await this.ensureClient(); - const response = await client.event.subscribe(); + const response = await client.event.subscribe( + directory ? { directory } : undefined, + ); return response.stream; } async replyPermission(options: { requestId: string; sessionId?: string; + directory?: string; reply: PermissionReply; message?: string; }) { const client = await this.ensureClient(); + const directory = await this.resolveInteractionDirectory(options); if ("permission" in client && client.permission?.reply) { const response = await client.permission.reply({ requestID: options.requestId, + directory, reply: options.reply, message: options.message, }); - return response.data; + return requireData(response.data, "permission.reply"); } if ("permission" in client && client.permission?.respond && options.sessionId) { const response = await client.permission.respond({ sessionID: options.sessionId, permissionID: options.requestId, + directory, response: options.reply, }); - return response.data; + return requireData(response.data, "permission.respond"); } throw new Error("opencode permission reply API is unavailable"); @@ -336,16 +353,19 @@ export class OpencodeRuntimeAdapter { async replyQuestion(options: { requestId: string; sessionId?: string; + directory?: string; answers: QuestionAnswers; }) { const client = await this.ensureClient(); + const directory = await this.resolveInteractionDirectory(options); if ("question" in client && client.question?.reply) { try { const response = await client.question.reply({ requestID: options.requestId, + directory, answers: options.answers, }); - return response.data; + return requireData(response.data, "question.reply"); } catch (error) { if (!options.sessionId) { throw error; @@ -360,6 +380,7 @@ export class OpencodeRuntimeAdapter { reply?: (parameters: { sessionID: string; requestID: string; + directory?: string; questionV2Reply: { answers: QuestionAnswers }; }) => Promise<{ data: unknown }>; }; @@ -371,11 +392,12 @@ export class OpencodeRuntimeAdapter { const response = await v2Question.reply({ sessionID: options.sessionId, requestID: options.requestId, + directory, questionV2Reply: { answers: options.answers, }, }); - return response.data; + return requireData(response.data, "question.v2.reply"); } throw new Error("opencode question reply API is unavailable"); @@ -384,14 +406,17 @@ export class OpencodeRuntimeAdapter { async rejectQuestion(options: { requestId: string; sessionId?: string; + directory?: string; }) { const client = await this.ensureClient(); + const directory = await this.resolveInteractionDirectory(options); if ("question" in client && client.question?.reject) { try { const response = await client.question.reject({ requestID: options.requestId, + directory, }); - return response.data; + return requireData(response.data, "question.reject"); } catch (error) { if (!options.sessionId) { throw error; @@ -406,6 +431,7 @@ export class OpencodeRuntimeAdapter { reject?: (parameters: { sessionID: string; requestID: string; + directory?: string; }) => Promise<{ data: unknown }>; }; }; @@ -416,13 +442,29 @@ export class OpencodeRuntimeAdapter { const response = await v2Question.reject({ sessionID: options.sessionId, requestID: options.requestId, + directory, }); - return response.data; + return requireData(response.data, "question.v2.reject"); } throw new Error("opencode question reject API is unavailable"); } + private async resolveInteractionDirectory(options: { + sessionId?: string; + directory?: string; + }): Promise { + const directory = options.directory?.trim(); + if (directory) { + return directory; + } + if (!options.sessionId) { + return undefined; + } + const session = await this.getSession(options.sessionId); + return session.directory; + } + async dispose(): Promise { if (this.toolOutputCleanupTimer) { clearInterval(this.toolOutputCleanupTimer); diff --git a/tests/routes/chatStream.test.ts b/tests/routes/chatStream.test.ts index 11e15f5..511d3ad 100644 --- a/tests/routes/chatStream.test.ts +++ b/tests/routes/chatStream.test.ts @@ -16,9 +16,11 @@ const createEventStream = (events: unknown[]) => ({ describe("streamPromptResponse", () => { it("emits only the final assistant text after tool-driven intermediate messages", async () => { + let subscribedDirectory: string | undefined; const runtime = { - subscribeEvents: async () => - createEventStream([ + subscribeEvents: async (directory?: string) => { + subscribedDirectory = directory; + return createEventStream([ { type: "message.part.delta", properties: { @@ -73,7 +75,8 @@ describe("streamPromptResponse", () => { type: "session.idle", properties: { sessionID: "runtime-session-1" }, }, - ]), + ]); + }, prompt: async () => undefined, messages: async () => [ { @@ -109,9 +112,11 @@ describe("streamPromptResponse", () => { sessionId: "runtime-session-1", clientSessionId: "client-session-1", message: "分析管网瓶颈", + workspaceRoot: "/tmp/conversation-workspace-1", write: (event, data) => events.push({ event, data }), }); + expect(subscribedDirectory).toBe("/tmp/conversation-workspace-1"); expect( events .filter((item) => item.event === "token") @@ -120,6 +125,143 @@ describe("streamPromptResponse", () => { ).toBe("共识别 56 条瓶颈管段，建议优先改造 Top 5。"); }); + it("streams final_answer deltas while routing commentary to progress", async () => { + let messagesCalls = 0; + const runtime = { + subscribeEvents: async () => + createEventStream([ + { + type: "message.part.updated", + properties: { + sessionID: "runtime-session-1", + part: { + id: "commentary-part", + sessionID: "runtime-session-1", + messageID: "assistant-commentary", + type: "text", + text: "", + metadata: { openai: { phase: "commentary" } }, + time: { start: 1 }, + }, + }, + }, + { + type: "message.part.delta", + properties: { + sessionID: "runtime-session-1", + messageID: "assistant-commentary", + partID: "commentary-part", + field: "text", + delta: "我先检查相关数据。", + }, + }, + { + type: "message.part.updated", + properties: { + sessionID: "runtime-session-1", + part: { + id: "commentary-part", + sessionID: "runtime-session-1", + messageID: "assistant-commentary", + type: "text", + text: "我先检查相关数据。", + metadata: { openai: { phase: "commentary" } }, + time: { start: 1, end: 2 }, + }, + }, + }, + { + type: "message.part.updated", + properties: { + sessionID: "runtime-session-1", + part: { + id: "final-part", + sessionID: "runtime-session-1", + messageID: "assistant-final", + type: "text", + text: "", + metadata: { openai: { phase: "final_answer" } }, + time: { start: 3 }, + }, + }, + }, + { + type: "message.part.delta", + properties: { + sessionID: "runtime-session-1", + messageID: "assistant-final", + partID: "final-part", + field: "text", + delta: "分析完成，", + }, + }, + { + type: "message.part.delta", + properties: { + sessionID: "runtime-session-1", + messageID: "assistant-final", + partID: "final-part", + field: "text", + delta: "结果正常。", + }, + }, + { + type: "message.part.updated", + properties: { + sessionID: "runtime-session-1", + part: { + id: "final-part", + sessionID: "runtime-session-1", + messageID: "assistant-final", + type: "text", + text: "分析完成，结果正常。", + metadata: { openai: { phase: "final_answer" } }, + time: { start: 3, end: 4 }, + }, + }, + }, + { + type: "session.idle", + properties: { sessionID: "runtime-session-1" }, + }, + ]), + prompt: async () => undefined, + messages: async () => { + messagesCalls += 1; + return []; + }, + } as unknown as OpencodeRuntimeAdapter; + const events: Array<{ event: string; data: Record }> = []; + + await streamPromptResponse({ + runtime, + sessionId: "runtime-session-1", + clientSessionId: "client-session-1", + message: "分析管网", + write: (event, data) => events.push({ event, data }), + }); + + expect( + events.filter((item) => item.event === "token").map((item) => item.data.content), + ).toEqual(["分析完成，", "结果正常。"]); + expect(messagesCalls).toBe(0); + expect( + events.find( + (item) => + item.event === "progress" && item.data.id === "commentary-commentary-part", + )?.data, + ).toMatchObject({ + phase: "commentary", + status: "running", + detail: "我先检查相关数据。", + }); + expect( + events.some( + (item) => item.event === "token" && item.data.content === "我先检查相关数据。", + ), + ).toBe(false); + }); + it("uses the final text event cache when the messages lookup fails", async () => { const runtime = { subscribeEvents: async () => @@ -334,6 +476,7 @@ describe("streamPromptResponse", () => { clientSessionId: "client-session-1", message: "run tests", approvalMode: "auto", + workspaceRoot: "/tmp/conversation-workspace-1", write: (event, data) => events.push({ event, data }), }); @@ -341,6 +484,7 @@ describe("streamPromptResponse", () => { { requestId: "perm-1", sessionId: "runtime-session-1", + directory: "/tmp/conversation-workspace-1", reply: "once", }, ]); @@ -422,6 +566,7 @@ describe("streamPromptResponse", () => { clientSessionId: "client-session-1", message: "run tests", approvalMode: "always", + workspaceRoot: "/tmp/conversation-workspace-1", write: (event, data) => events.push({ event, data }), }); @@ -429,6 +574,7 @@ describe("streamPromptResponse", () => { { requestId: "perm-always-bash", sessionId: "runtime-session-1", + directory: "/tmp/conversation-workspace-1", reply: "once", }, ]); @@ -465,6 +611,7 @@ describe("streamPromptResponse", () => { clientSessionId: "client-session-1", message: "delete recursively", approvalMode: "always", + workspaceRoot: "/tmp/conversation-workspace-1", write: (event, data) => events.push({ event, data }), }); @@ -472,6 +619,7 @@ describe("streamPromptResponse", () => { { requestId: "perm-always-rm-rf", sessionId: "runtime-session-1", + directory: "/tmp/conversation-workspace-1", reply: "reject", }, ]); @@ -741,4 +889,143 @@ describe("streamPromptResponse", () => { }); }); + it("uses StructuredOutput as the terminal action for models without reliable phases", async () => { + const promptCalls: unknown[][] = []; + const runtime = { + subscribeEvents: async () => + createEventStream([ + { + type: "message.part.updated", + properties: { + sessionID: "runtime-session-1", + part: { + id: "final-answer-part", + sessionID: "runtime-session-1", + messageID: "assistant-final", + type: "tool", + callID: "final-answer-call", + tool: "StructuredOutput", + state: { + status: "running", + input: { answer: "供水服务分区分析已完成。" }, + time: { start: 1 }, + }, + }, + }, + }, + { + type: "message.part.updated", + properties: { + sessionID: "runtime-session-1", + part: { + id: "final-answer-part", + sessionID: "runtime-session-1", + messageID: "assistant-final", + type: "tool", + callID: "final-answer-call", + tool: "StructuredOutput", + state: { + status: "completed", + input: { answer: "供水服务分区分析已完成。" }, + output: "最终回答已提交。", + title: "StructuredOutput", + metadata: {}, + time: { start: 1, end: 2 }, + }, + }, + }, + }, + { + type: "session.idle", + properties: { sessionID: "runtime-session-1" }, + }, + ]), + prompt: async (...args: unknown[]) => { + promptCalls.push(args); + }, + messages: async () => { + throw new Error("StructuredOutput should avoid the message fallback"); + }, + } as unknown as OpencodeRuntimeAdapter; + const events: Array<{ event: string; data: Record }> = []; + + const result = await streamPromptResponse({ + runtime, + sessionId: "runtime-session-1", + clientSessionId: "client-session-1", + message: "分析供水服务分区", + model: "deepseek/deepseek-v4-flash", + write: (event, data) => events.push({ event, data }), + }); + + expect(promptCalls[0]?.[3]).toMatchObject({ + format: { + type: "json_schema", + schema: { + required: ["answer"], + }, + }, + }); + expect(result).toEqual({ aborted: false, failed: false, toolCallCount: 0 }); + expect( + events + .filter((item) => item.event === "token") + .map((item) => item.data.content) + .join(""), + ).toBe("供水服务分区分析已完成。"); + expect(events.some((item) => item.event === "tool_call")).toBe(false); + expect(events.some((item) => item.event === "done")).toBe(true); + }); + + it("recovers the final answer from structured message data when tool events are missed", async () => { + const runtime = { + subscribeEvents: async () => + createEventStream([ + { + type: "message.updated", + properties: { + info: { + id: "assistant-final", + sessionID: "runtime-session-1", + role: "assistant", + }, + }, + }, + { + type: "session.idle", + properties: { sessionID: "runtime-session-1" }, + }, + ]), + prompt: async () => undefined, + messages: async () => [ + { + info: { + id: "assistant-final", + sessionID: "runtime-session-1", + role: "assistant", + structured: { answer: "已从结构化结果恢复最终回答。" }, + }, + parts: [], + }, + ], + } as unknown as OpencodeRuntimeAdapter; + const events: Array<{ event: string; data: Record }> = []; + + await streamPromptResponse({ + runtime, + sessionId: "runtime-session-1", + clientSessionId: "client-session-1", + message: "恢复最终回答", + model: "deepseek/deepseek-v4-flash", + write: (event, data) => events.push({ event, data }), + }); + + expect( + events + .filter((item) => item.event === "token") + .map((item) => item.data.content) + .join(""), + ).toBe("已从结构化结果恢复最终回答。"); + }); + }); diff --git a/tests/runtime/opencode.test.ts b/tests/runtime/opencode.test.ts index 4fc82bb..26a6553 100644 --- a/tests/runtime/opencode.test.ts +++ b/tests/runtime/opencode.test.ts @@ -90,6 +90,136 @@ describe("OpencodeRuntimeAdapter.ensureClient", () => { }); }); +describe("OpencodeRuntimeAdapter.subscribeEvents", () => { + it("subscribes to the conversation workspace directory", async () => { + const calls: Array | undefined> = []; + const stream = (async function* () { + return; + })(); + const client = { + event: { + subscribe: async (input?: Record) => { + calls.push(input); + return { stream }; + }, + }, + } as unknown as OpencodeClient; + const runtime = Object.assign(Object.create(OpencodeRuntimeAdapter.prototype), { + ensureClient: async () => client, + }) as OpencodeRuntimeAdapter; + + await expect( + runtime.subscribeEvents("/tmp/conversation-workspace-1"), + ).resolves.toBe(stream); + expect(calls).toEqual([ + { directory: "/tmp/conversation-workspace-1" }, + ]); + }); +}); + +describe("OpencodeRuntimeAdapter.prompt", () => { + it("forwards the final answer schema", async () => { + const calls: unknown[] = []; + const client = { + session: { + prompt: async (input: unknown) => { + calls.push(input); + return { data: {} }; + }, + }, + } as unknown as OpencodeClient; + const runtime = Object.assign(Object.create(OpencodeRuntimeAdapter.prototype), { + ensureClient: async () => client, + }) as OpencodeRuntimeAdapter; + + await runtime.prompt( + "session-1", + "分析供水分区", + { providerID: "deepseek", modelID: "deepseek-v4-flash" }, + { + format: { + type: "json_schema", + schema: { + type: "object", + properties: { answer: { type: "string" } }, + required: ["answer"], + }, + }, + }, + ); + + expect(calls).toEqual([ + { + sessionID: "session-1", + model: { + providerID: "deepseek", + modelID: "deepseek-v4-flash", + }, + format: { + type: "json_schema", + schema: { + type: "object", + properties: { answer: { type: "string" } }, + required: ["answer"], + }, + }, + parts: [{ type: "text", text: "分析供水分区" }], + }, + ]); + }); +}); + +describe("OpencodeRuntimeAdapter interaction replies", () => { + it("replies to permissions in the conversation workspace directory", async () => { + const calls: unknown[] = []; + const client = { + permission: { + reply: async (input: unknown) => { + calls.push(input); + return { data: true }; + }, + }, + } as unknown as OpencodeClient; + const runtime = Object.assign(Object.create(OpencodeRuntimeAdapter.prototype), { + ensureClient: async () => client, + }) as OpencodeRuntimeAdapter; + + await runtime.replyPermission({ + requestId: "permission-1", + sessionId: "session-1", + directory: "/tmp/conversation-workspace-1", + reply: "once", + }); + + expect(calls).toEqual([ + { + requestID: "permission-1", + directory: "/tmp/conversation-workspace-1", + reply: "once", + }, + ]); + }); + + it("fails when OpenCode returns no permission reply data", async () => { + const client = { + permission: { + reply: async () => ({ data: undefined }), + }, + } as unknown as OpencodeClient; + const runtime = Object.assign(Object.create(OpencodeRuntimeAdapter.prototype), { + ensureClient: async () => client, + }) as OpencodeRuntimeAdapter; + + await expect( + runtime.replyPermission({ + requestId: "permission-1", + directory: "/tmp/conversation-workspace-1", + reply: "once", + }), + ).rejects.toThrow("permission.reply returned no data"); + }); +}); + describe("OpencodeRuntimeAdapter.createSession", () => { it("creates a real chat session inside a dedicated conversation workspace", async () => { const workspaceRoot = await mkdtemp(join(tmpdir(), "tjwater-conversations-")); @@ -225,11 +355,16 @@ describe("OpencodeRuntimeAdapter.warmup", () => { await runtime.replyQuestion({ requestId: "question-1", sessionId: "session-1", + directory: "/tmp/conversation-workspace-1", answers: [["继续"]], }); expect(calls).toEqual([ - { requestID: "question-1", answers: [["继续"]] }, + { + requestID: "question-1", + directory: "/tmp/conversation-workspace-1", + answers: [["继续"]], + }, ]); }); });