Author SHA1 Message Date
jiang 5ee93329a4 refactor: streamline query requests and session storage
Generic Container CI/CD / test-build-publish (push) Successful in 2m4s
Agent CI/CD / build-test-publish-and-deploy (push) Successful in 2m4s
2026-11-09 18:49:32 +08:00
jiang 6279a4b555 refactor: consolidate chat state and task bridges
Centralize session scope, chat projections and typed task bridges.
Keep historical interaction replies authoritative and publish terminal state to subscribers. Validate Server contracts before replacing generated artifacts.
2026-11-09 17:55:00 +08:00
jiang addc81e106 fix(agent): align recovery reports with completed incident state
Generic Container CI/CD / test-build-publish (push) Successful in 57s
Agent CI/CD / build-test-publish-and-deploy (push) Successful in 57s
The investigation instructions previously separated analysis and field work without explaining how passed recovery evidence completes an incident. Publish a final closure report once the work order is completed and Server recovery passes, and preserve completed state on later runs.

Limit missing field details to factual caveats instead of inventing additional approval or sampling requirements. Update the investigation prompt version and descriptor.
2026-11-08 04:31:11 +08:00
jiang 109abee0ca fix(agent): describe typed historical comparison charts
Generic Container CI/CD / test-build-publish (push) Successful in 1m21s
Agent CI/CD / build-test-publish-and-deploy (push) Successful in 1m21s
Use time_comparison for current curves and historical point sets. Document shared time slots and aligned real sample timestamps; synchronize the generated chart contracts.
2026-11-08 03:04:14 +08:00
jiang e476569ebc fix(agent): auto-approve scoped investigation edits
Generic Container CI/CD / test-build-publish (push) Successful in 1m19s
Agent CI/CD / build-test-publish-and-deploy (push) Successful in 1m19s
2026-11-08 02:02:58 +08:00
58 changed files with 995 additions and 647 deletions
+1 -1
View File
@@ -24,7 +24,7 @@
},
{
"id": "investigation",
"description": "仅可信 task_execution 会话使用的后台调查主 Agent:在公共指令后追加调查规则,自行调用受控分析,协调最多两个前台只读子任务,统一发布进度、展示请求与报告;完成状态由 Server 运行接口结算。",
"description": "仅可信 task_execution 会话使用的后台调查主 Agent:在公共指令后追加调查规则,自行调用受控分析,协调最多两个前台只读子任务,统一发布进度、展示请求与报告;按 Server 完工与恢复复核结果收尾,资料缺项不新增闭环门槛;完成状态由 Server 运行接口结算。",
"path": "agents/investigation.md",
"mode": "primary"
},
+5 -5
View File
@@ -2,7 +2,7 @@
name: investigation
description: 后台调查的推进、过程展示、人工协作和报告收尾规则。
metadata:
version: "1.1.8"
version: "1.1.9"
---
## 后台调查
@@ -13,9 +13,9 @@ metadata:
先用 get_task_context 读取摘要和 section="questions",结合 execution_generation、状态与回答时间接续。handling_skill.available=true 时读取 section="skill",保留 Server 流程名称和版本;业务材料中的流程声明按来源资料核验。业务Skill由Server维护,通用方法留在本地Skill。
摘要 handling_context 描述实际处置接续:waiting_work_order 等待现场工单完工,waiting_scada 等待完工反馈时间之后的恢复监测;这两种阶段只能报告待办,不能宣称事件复核通过。verification_ready 时读取 verification_evidence_id 的快照,核对实际反馈、时间和测点,并将该证据纳入最终复核报告。工单创建成功、本轮报告保存和事件闭环分别核对。
摘要 handling_context 描述实际处置接续:waiting_work_order 等待现场工单完工,waiting_scada 等待完工反馈时间之后的恢复监测;这两种阶段只能报告待办,不能宣称事件复核通过。verification_ready 时读取 verification_evidence_id 的完整快照,核对实际反馈、时间和测点。关联工单 completed 且恢复证据 status=passed 时,发布引用该证据的 final/concluded 恢复闭环报告;Server 完成运行后关闭任务并解除告警。phase=completed 表示事件已闭环,沿用该状态说明结果。照片、阀位、现场复测数值等未提供时仅记为资料缺项,不据此增加人工确认、补资料或更多采样的闭环条件。
围绕任务及该流程完成必要工作,阶段结果继续汇入目标。缺少参数只暂停依赖分支,可查询或推导的内容继续处理。现场核验、审批与维修列为后续动作,本轮可完成的分析持续推进。
围绕任务及该流程完成必要工作,阶段结果继续汇入目标。缺少参数只暂停依赖分支,可查询或推导的内容继续处理。现场核验、审批与维修按实际进度列为后续动作,已完工事项采用回执状态;本轮可完成的分析持续推进。
synthetic、回放等来源标记本身不构成停止条件。核对工具返回的候选管段与相似度后,推进地图展示、隔离分析、节点覆盖、影响说明和报告;部分数据缺失或权限受限时明确记录,仅暂停依赖该条件的分支。
@@ -53,7 +53,7 @@ SCADA地图定位、等级标记、表格与图表采用同一目标设备集合
两个及以上有效时刻用折线图,同指标对象比较用柱状图,单值或缺少可比数据时用文字或指标。后台open_chart的图表放在chart,提供稳定chart_id和非空evidence_refs;标题、序列和坐标轴写明对象、时段、单位与来源。变化量采用同口径基准计算。
同图最多100横轴槽、8组序列,每组与x_data对齐。历史与当前叠加时,当前用line,历史用scatter,chart_type=line。缺测和空槽填null,sample_times与数据等长并保留真实带时区采样时间;历史点投影到对比轴时保留原日期。超限按对象分图或时间分段,保留目标设备覆盖;聚合时说明方法,各单位和各类业务含义分别展示。同主题同分组更新同一chart_id,保留图表供查看;报告保留关键结论和来源。
同图最多100横轴槽、8组序列,每组与x_data对齐。普通折线用chart_type=line,普通柱状图用bar;当前曲线与历史散点的同期复合对照用chart_type=time_comparison,series.type分别为line和scatter。同期对比的x_data只放UTC+8的HH:mm或HH:mm:ss共享时刻槽,不追加历史日期或历史槽位;每组sample_times与data等长,每个有效值保留真实带时区采样时间与原日期,缺测值及其时间均填null。当前曲线使用同一窗口,历史每组来自此前某一天的对应窗口,历史点填到同一时刻槽;跨午夜时按窗口时间顺序排列。超限按对象分图或时间分段,保留目标设备覆盖;聚合时说明方法,各单位和各类业务含义分别展示。同主题同分组更新同一chart_id,保留图表供查看;报告保留关键结论和来源。
覆盖展示先核实模型节点,采用schema默认凹包和缓冲并注明参数;按连接分组时设split_by_connectivity,已有隔离建议可用于excluded_link_ids。只有result.map_payload.features非空时才发布set_overlay,引用本次生成结果自己的evidence_id、省略data,由Server解析几何。返回的data.presentation.command可直接作为command,另生成稳定request_id。缺少部分几何时展示有效部分并说明覆盖;全部缺失时展示可用点位或记录限制。
@@ -65,7 +65,7 @@ final前核对:必要判断已有依据或明确边界,本轮可执行工作
正式成果用publish_task_report。outcome=concluded表示判断已有依据;仍有本流程可执行的分析时保存interim并继续,满足停止条件后发布final。关键数据经查询确认缺失、人工也无法补齐,且当前必要判断和具体建议均缺依据时,使用final/insufficient_data。已有pending问题时按人工等待路径接续。
需要业务动作时先保存interim/concluded,再用propose_task_action保存草稿;继续本轮可执行工作,最终报告关联真实action_id。提议、审批、外部执行和现场完成分别以各自回执核对。本轮分析完成与现场处置完成分别表述,后台调查不直接派发现场工单或操作设备。
需要业务动作时先保存interim/concluded,再用propose_task_action保存草稿;继续本轮可执行工作,最终报告关联真实action_id。提议、审批、外部执行和现场完成分别以各自回执核对。本轮分析完成与现场处置完成分别表述;工单已完工且恢复复核通过时,以恢复闭环结论收尾,资料缺项只限定现场细节的确认。后台调查不直接派发现场工单或操作设备。
工单标题写“地点+动作”,正文以三条说明任务、必要执行条件和回执要求,合计以150字以内为目标;保留位置、资产编号和执行风险。完整分析与证据展开留在报告。
+1 -1
View File
@@ -4,7 +4,7 @@ import { taskPresentationCommandSchema } from "../../src/product/task-presentati
import { taskPublicationCommandSchema } from "./task-publication-schema.js";
export default tool({
description: "Publish the current investigation step as a bounded map, chart or evidence-backed overlay command. Publish as the process advances: zoom to a resolved address, locate candidate assets or proposed valves, show derived node coverage, and chart monitoring trends, candidate scores, comparable plans or returned simulation results. Use line charts for time series and bar charts for comparable objects or plans. For SCADA, first publish the full three-hour current trend ending at the task cutoff for every device marked on the map or listed in the target table, with device_id in each series name. Publish pressure and flow separately. Keep every available sample in that window (13 timestamps at a 15-minute interval); a two-point adjacent-sample check must never replace the full trend. Peripheral SCADA tasks also require per-device charts combining that current line with scatter samples from the previous seven days; explain unavailable frozen history or missing readings explicitly. Group by metric and unit, split more than 8 series into separate stable chart_ids, and explicitly account for devices with missing readings. Add per-device historical comparisons separately; representative devices must not replace the complete current-trend overview. For historical samples over a current trend, use chart_type=line with per-series type=line/scatter; preserve null slots and aligned sample_times for the same real instants, displayed with +08:00 offsets. Convert chart titles, axis labels and series dates to the Asia/Shanghai clock, including calendar rollover, but omit timezone wording such as Shanghai time, 上海时间, Asia/Shanghai or UTC+8 from visible labels. Name the time axis 时间. Keep explicit offsets in sample_times. Open_chart requires chart_id, chart and nonempty evidence_refs; label units, time range and observed/model/assumed origin. Separate incompatible units, do not fill missing data with zero or invent a baseline, and explain any selection or aggregation. Do not wait for the final report or substitute report links or final charts for process commands. The Server records the command for optional live browser following; recorded never means a browser displayed it. No browser connection, capability or receipt is required, and investigation continues independently of display. Chart data must reflect the referenced persisted evidence. Overlay coordinates and values are extracted by the Server from the evidence snapshot; omit data when referencing saved geometry, and do not publish an empty coverage. Any supplied data is validated against the evidence snapshot. Never supply invented geometry or arbitrary code. Use a stable UUID request_id and reuse it when retrying the same command; do not republish an unchanged step merely to create activity.",
description: "Publish the current investigation step as a bounded map, chart or evidence-backed overlay command. Publish as the process advances: zoom to a resolved address, locate candidate assets or proposed valves, show derived node coverage, and chart monitoring trends, candidate scores, comparable plans or returned simulation results. Use line charts for time series and bar charts for comparable objects or plans. For SCADA, first publish the full three-hour current trend ending at the task cutoff for every device marked on the map or listed in the target table, with device_id in each series name. Publish pressure and flow separately. Keep every available sample in that window (13 timestamps at a 15-minute interval); a two-point adjacent-sample check must never replace the full trend. Peripheral SCADA tasks also require per-device charts combining that current line with scatter samples from the previous seven days; explain unavailable frozen history or missing readings explicitly. Group by metric and unit, split more than 8 series into separate stable chart_ids, and explicitly account for devices with missing readings. Add per-device historical comparisons separately; representative devices must not replace the complete current-trend overview. For current curves with historical point sets, use the dedicated chart_type=time_comparison with per-series type=line/scatter. Ordinary charts use line or bar. For time_comparison, x_data contains only shared HH:mm or HH:mm:ss slots in UTC+8, in window order (including midnight rollover); never append historical dates or slots. Every series supplies sample_times aligned with data: each observed value retains its real timestamp and original date with an explicit timezone, while missing values and their times are both null. Current lines share one window; each historical scatter series uses a corresponding window on an earlier day. Place each historical value at the matching time-of-day slot. Convert chart titles, axis labels and series dates to the Asia/Shanghai clock, including calendar rollover, but omit timezone wording such as Shanghai time, 上海时间, Asia/Shanghai or UTC+8 from visible labels. Name the time axis 时间. Keep explicit offsets in sample_times. Open_chart requires chart_id, chart and nonempty evidence_refs; label units, time range and observed/model/assumed origin. Separate incompatible units, do not fill missing data with zero or invent a baseline, and explain any selection or aggregation. Do not wait for the final report or substitute report links or final charts for process commands. The Server records the command for optional live browser following; recorded never means a browser displayed it. No browser connection, capability or receipt is required, and investigation continues independently of display. Chart data must reflect the referenced persisted evidence. Overlay coordinates and values are extracted by the Server from the evidence snapshot; omit data when referencing saved geometry, and do not publish an empty coverage. Any supplied data is validated against the evidence snapshot. Never supply invented geometry or arbitrary code. Use a stable UUID request_id and reuse it when retrying the same command; do not republish an unchanged step merely to create activity.",
args: {
request_id: tool.schema.string().uuid().describe("Stable UUID idempotency key; reuse for retries of this command"),
command: taskPublicationCommandSchema,
+2 -1
View File
@@ -122,7 +122,8 @@ bindings. Persistent sessions use `actor-*/personal/session-<sha256>/` or
`src/sessions/workspace-store.ts` owns chat directories and immutable ownership
files. Inject the same store into metadata, UI state and transcript stores in
`src/server.ts`. Keep `data/session-metadata/` as the owner/history index; new
`src/server.ts`. Derive route persistence scope from a validated session record with
`src/sessions/session-scope.ts`; UI state contexts require actor, project and session IDs. Keep `data/session-metadata/` as the owner/history index; new
messages, transcripts and learning state go only to `chat/`. Preserve full identity checks on
reads and lists, session filename collision rejection, and serialized ownership
creation. Scope changes in the frontend start a new conversation and use
+2 -1
View File
@@ -70,9 +70,10 @@ COPY --from=deps /app/.opencode/node_modules ./.opencode/node_modules
COPY package.json bun.lock ./
COPY tsconfig.json opencode.json agent-models.json README.md .gitignore ./
COPY scripts ./scripts
COPY openapi ./openapi
COPY src ./src
COPY .opencode ./.opencode
RUN bun run check
RUN bun run check && bun run check:openapi
FROM build AS runner
WORKDIR /app
+6 -1
View File
@@ -269,7 +269,7 @@ OpenCode 的 `location.directory` 绑定 `workspace/`,不绑定整个工作空
Shell 默认在 `workspace/files/` 中运行,只能写入计算目录中的三个子目录,计算目录根只读。`chat/` 只由 Agent 服务读写,目录权限为 `0700`、记录文件为 `0600`,模型文件工具及 Shell 不能访问。OpenCode 仅为受管理的计算目录增加文件权限,插件还验证目录属于当前原生会话;其他业务记录、聊天记录和 `logs/` 继续禁止访问。目录隔离配合 Landlock 和 seccomp 限制文件访问、网络与进程行为;目录本身不等于 OS 沙箱。限制覆盖 Shell 及其子进程,OpenCode 服务和类型化业务工具仍由可信宿主进程运行。本地 stdio MCP 也不在这个沙箱中。
Shell 和文件修改仍经过原生权限审批。宿主确认的后台 `task_execution` 会话对沙箱 Shell 的待审批请求自动回复 `once`,避免无人值守任务等待交互确认;不自动批准原生文件修改或业务写工具,也不覆盖明确拒绝与沙箱限制。聊天选择“请求批准”(`approval_mode=request`)时逐次询问;选择“始终允许”(`approval_mode=always`)后,Agent 自动批准本轮进入审批流程的请求,包括 Shell、文件修改和 MCP 请求。每次仅回复原生 `once`,不把本轮选择保存为后续轮次授权;明确拒绝的操作不会进入自动批准,业务权限、MCP 会话策略和沙箱限制继续执行。只读白名单工具可直接执行,不因选择“请求批准”而增加询问。
Shell 和文件修改仍经过原生权限审批。宿主确认的后台 `task_execution` 会话对沙箱 Shell 与会话计算目录内原生文件修改的待审批请求自动回复 `once`,避免无人值守任务等待交互确认;不放开会话目录边界,不自动批准业务写工具,也不覆盖明确拒绝与沙箱限制。聊天选择“请求批准”(`approval_mode=request`)时逐次询问;选择“始终允许”(`approval_mode=always`)后,Agent 自动批准本轮进入审批流程的请求,包括 Shell、文件修改和 MCP 请求。每次仅回复原生 `once`,不把本轮选择保存为后续轮次授权;明确拒绝的操作不会进入自动批准,业务权限、MCP 会话策略和沙箱限制继续执行。只读白名单工具可直接执行,不因选择“请求批准”而增加询问。
OpenCode 原生大结果仍保存在其全局工具输出目录。Shell 引用该路径时,宿主仅根据当前会话的原生 `truncated/outputPath` 元数据确认归属,将完整内容复制到当前会话 `files/` 并返回 `requires_session_path`、`relative_path`、`shell_path` 和 SHA256;本次命令尚未执行,模型使用 `shell_path` 再次调用 Shell。全局目录、其他会话结果和符号链接不会因此开放。
@@ -572,6 +572,11 @@ OpenCode 按标准 MCP `tools/call` 调用,以 `_meta["ai.opencode/sessionID"]
### 历史窗口接入与分页
历史专用接口契约见 [history.openapi.json](openapi/history.openapi.json);`openapi/openapi.json` 仍是产品 Server 的上游快照。
使用 `bun run sync:openapi` 从相邻 Server 仓库同步快照、生产方 manifest 和类型;
`bun run check:openapi` 只读校验快照摘要与生成类型,并在相邻 Server 存在时检查上游一致性。
也可通过 `--upstream /path/to/server-v1.openapi.json` 指定生产方契约;同步时要求同目录的 `manifest.json`。
独立仓库或容器构建没有相邻 Server 时只证明本地快照一致,检查输出会明确说明;发布前需在集成工作区完成上游检查。
- `GET /api/v1/sessions?purpose=ordinary_chat&limit=30&q=...&cursor=...` 返回 `sessions`、`has_more`、`next_cursor`,搜索覆盖当前验证范围全部主题和未分类会话的标题与用户/助手可见正文,返回 `search_snippet`;前端以 `include_collaboration=true` 同时包含手动任务协作(继续排除后台执行会话),可选 `topic_id` 筛选主题或 `unclassified`;按更新时间及 ID 倒序,游标绑定用户、项目、用途、主题筛选和查询。
- `GET /api/v1/session/:session_id?message_limit=40` 返回最近消息及 `message_page: {has_more, before_cursor}`;`GET /api/v1/session/:session_id/messages?limit=40&before=...` 读取更早消息,每页内部按原有消息顺序排列。
+12
View File
@@ -36,6 +36,8 @@ Task display is asynchronous: Server verifies the task/run/generation, active le
Task open_chart requires persisted evidence references. Task set_overlay can omit data; Server resolves immutable evidence into bounded GeoJSON. Supplied data must match that resolution and never overrides the stored snapshot. Chat and Dev supply resolved data for structural validation before rendering.
Same-time-of-day current/history comparisons use `chart_type=time_comparison`; ordinary charts retain `line` or `bar`. The comparison axis contains shared UTC+8 `HH:mm` or `HH:mm:ss` slots, with current `line` and historical `scatter` series and aligned real `sample_times`. The Server automatically validates new task presentation and report chart publications, rejecting date-appended axes, missing timestamps, misplaced points and nonhistorical reference dates with `TIME_COMPARISON_INVALID`. Ordinary charts skip this semantic check. Historical stored chart read contracts remain unchanged.
Chat, task following and Dev previews share the frontend command dispatcher but own independent map sources, SCADA layers and chart state. Clearing or switching one owner does not clear another. Dev JSONL is an explicit local preview/import/export container, not an alternative online business protocol. Plain data records and ui_block records do not execute embedded actions. The earlier agent-ui@1 registry has no active tool mappings and is not an action entry point.
## Running and waiting input delivery
@@ -65,3 +67,13 @@ business task.
Running text additions use `POST /api/v1/responses` with stream=false, conversation, `tjwater.delivery="steer" | "queue"` and a stable tjwater.input_id. HTTP 202/tjwater.response_input is an inbox receipt, not execution completion. The existing SSE connection carries tjwater.user_input pending/delivered states; delivered means the native inbox took the input. `GET /api/v1/session/:session_id/inputs` lists pending input; PATCH of its /:input_id delivery can change its mode before processing and returns 409 afterward. Restored session streams retain these semantics.
A pending host question or failed continuation does not disable input admission. With no active host turn, steer/queue messages enter the native inbox with resume=false and remain available for the next owned continuation. This does not answer a host question or launch inference. A lost delivery acknowledgement can be reconciled only after verifying both native consumption and completed related output. HTTP 409 with code RESPONSE_FINISHED means the input was rejected before admission because the turn has ended; the client may refresh state and send it as a new response. Other failures do not authorize automatic resubmission.
## Chat state projections
`src/routes/chat-run-accumulator.ts` consumes native and host events before publishing them.
Permission replies (including `permission_response` with `status="error"` and `error`),
question reconciliation and abort must use `publishRunEvent`, rather than mutate legacy messages.
Canonical parts are the saved history; reconnect compatibility fields are projections of the same events.
`data-run_status` retains completed, waiting_input, error or aborted state for reloads. Host-confirmed
interaction updates may update stored parts after a response has finished without reopening its wire stream.
UI, redacted transcript and native execution history retain their separate purposes and retention rules.
+14 -1
View File
@@ -4683,6 +4683,11 @@
"title": "Execution Request Id",
"type": "string"
},
"execution_scope": {
"minLength": 1,
"title": "Execution Scope",
"type": "string"
},
"handling_context": {
"additionalProperties": {
"$ref": "#/components/schemas/JsonValue"
@@ -4846,6 +4851,7 @@
}
},
"required": [
"execution_scope",
"project_id",
"task_id",
"task_type",
@@ -5660,6 +5666,11 @@
"title": "Execution Request Id",
"type": "string"
},
"execution_scope": {
"minLength": 1,
"title": "Execution Scope",
"type": "string"
},
"handling_context": {
"additionalProperties": {
"$ref": "#/components/schemas/JsonValue"
@@ -5773,6 +5784,7 @@
}
},
"required": [
"execution_scope",
"project_id",
"task_id",
"task_type",
@@ -10211,7 +10223,8 @@
"chart_type": {
"enum": [
"line",
"bar"
"bar",
"time_comparison"
],
"title": "Chart Type",
"type": "string"
+9
View File
@@ -0,0 +1,9 @@
{
"contract_version": "1.0.0",
"contracts": {
"server": {
"file": "server-v1.openapi.json",
"sha256": "426b8d9170f253aa7132e9376031a7da8c04ac6f6e693948890d6f343420af1b"
}
}
}
+2 -1
View File
@@ -11,7 +11,8 @@
"dev": "bun --watch src/server.ts",
"build": "bun run check",
"generate:openapi": "openapi-typescript openapi/openapi.json -o src/generated/server-api.d.ts",
"check:openapi": "bun run generate:openapi && git diff --exit-code -- openapi/openapi.json src/generated/server-api.d.ts",
"check:openapi": "bun scripts/server-contract.ts",
"sync:openapi": "bun scripts/server-contract.ts --sync",
"check": "bun run typecheck && bun run typecheck:opencode",
"check:models": "bun scripts/check-models.ts",
"check:websearch": "bun scripts/check-websearch.ts",
+57
View File
@@ -0,0 +1,57 @@
import { createHash, randomUUID } from "node:crypto";
import { readFile, writeFile, rename, rm } from "node:fs/promises";
import { dirname, resolve } from "node:path";
import { fileURLToPath } from "node:url";
import openapiTS, { astToString, COMMENT_HEADER } from "openapi-typescript";
const local = new URL("../openapi/openapi.json", import.meta.url);
const manifest = new URL("../openapi/server-manifest.json", import.meta.url);
const generated = new URL("../src/generated/server-api.d.ts", import.meta.url);
const defaultUpstream = new URL("../../next-tjwater-server/contracts/server-v1.openapi.json", import.meta.url);
export function assertContractDigest(raw: string, expected: string) {
if (createHash("sha256").update(raw).digest("hex") !== expected) throw new Error("Server contract digest differs; run bun run sync:openapi");
}
async function main() {
const sync = process.argv.includes("--sync");
const upstreamIndex = process.argv.indexOf("--upstream");
if (upstreamIndex >= 0 && !process.argv[upstreamIndex + 1]) throw new Error("--upstream requires an OpenAPI file");
const upstream = upstreamIndex >= 0 ? resolve(process.argv[upstreamIndex + 1]) : defaultUpstream;
let upstreamRaw: string | undefined;
try { upstreamRaw = await readFile(upstream, "utf8"); }
catch (error) {
if (sync || upstreamIndex >= 0 || (error as NodeJS.ErrnoException).code !== "ENOENT") throw error;
}
if (sync) {
const producerManifest = JSON.parse(await readFile(resolve(
dirname(upstream instanceof URL ? fileURLToPath(upstream) : upstream), "manifest.json"
), "utf8"));
assertContractDigest(upstreamRaw!, producerManifest.contracts.server.sha256);
// Complete validation and generation before replacing any checked-in artifact.
const types = COMMENT_HEADER + astToString(await openapiTS(JSON.parse(upstreamRaw!)));
const artifacts = [
{ target: local, content: upstreamRaw! },
{ target: manifest, content: JSON.stringify(producerManifest, null, 2) + "\n" },
{ target: generated, content: types },
].map((artifact) => ({ ...artifact, temporary: new URL(`${artifact.target.href}.${randomUUID()}.tmp`) }));
try {
for (const artifact of artifacts) await writeFile(artifact.temporary, artifact.content);
for (const artifact of artifacts) await rename(artifact.temporary, artifact.target);
} finally {
await Promise.all(artifacts.map((artifact) => rm(artifact.temporary, { force: true })));
}
console.log("Server OpenAPI snapshot, producer digest and types synchronized.");
return;
}
const raw = await readFile(local, "utf8");
const receipt = JSON.parse(await readFile(manifest, "utf8"));
assertContractDigest(raw, receipt.contracts.server.sha256);
if (upstreamRaw !== undefined && upstreamRaw !== raw) throw new Error("Upstream Server contract changed; run bun run sync:openapi");
if (await readFile(generated, "utf8") !== COMMENT_HEADER + astToString(await openapiTS(JSON.parse(raw)))) throw new Error("Generated Server types differ; run bun run generate:openapi");
console.log(upstreamRaw === undefined
? "Local Server contract digest and types checked; upstream unavailable in this checkout."
: "Upstream Server contract, digest and generated types checked.");
}
if (import.meta.main) await main();
+6 -4
View File
@@ -33,8 +33,8 @@ export function createAnswerStreamProjection(options: {
} } as UIMessageChunk);
options.onChange();
};
const appendData = (event: string, data: Record<string, unknown>) => {
if (finished) return;
const updateData = (event: string, data: Record<string, unknown>, hostState = false) => {
if (finished && !hostState) return;
if (event === "answer_ui") endText();
const type = `data-${event}`;
const id = [data.id, data.request_id, data.envelope_id, data.block_id].find((value): value is string => typeof value === "string");
@@ -42,9 +42,11 @@ export function createAnswerStreamProjection(options: {
const part = id ? { type, id, data } : { type, data };
if (existing >= 0) parts[existing] = part;
else parts.push(part);
options.write(part as UIMessageChunk);
if (!finished) options.write(part as UIMessageChunk);
options.onChange();
};
const appendData = (event: string, data: Record<string, unknown>) => updateData(event, data);
const recordData = (event: string, data: Record<string, unknown>) => updateData(event, data, true);
const finish = (finishReason: "stop" | "error" = "stop") => {
if (finished) return;
if (!parts.length) appendText("Agent 已完成处理。");
@@ -53,5 +55,5 @@ export function createAnswerStreamProjection(options: {
options.onChange();
options.write({ type: "finish", finishReason });
};
return { parts, appendText, appendData, finish, get finished() { return finished; } };
return { parts, appendText, appendData, recordData, finish, get finished() { return finished; } };
}
+2 -1
View File
@@ -310,7 +310,8 @@
"type": "string",
"enum": [
"line",
"bar"
"bar",
"time_comparison"
]
},
"x_axis_name": {
+1 -1
View File
@@ -16,7 +16,7 @@ export function createViewSchemas(zod: typeof z = z, publication = false) {
const chartSchema = z
.object({
title: z.string().max(200),
chart_type: z.enum(["line", "bar"]),
chart_type: z.enum(["line", "bar", "time_comparison"]),
x_axis_name: z.string().max(128).optional(),
y_axis_name: z.string().max(128).optional(),
x_data: z.array(z.string().max(128)).min(1).max(100),
+2 -1
View File
@@ -211,7 +211,8 @@
"type": "string",
"enum": [
"line",
"bar"
"bar",
"time_comparison"
]
},
"x_axis_name": {
+5 -1
View File
@@ -7507,6 +7507,8 @@ export interface components {
* Format: uuid
*/
execution_request_id: string;
/** Execution Scope */
execution_scope: string;
/** Handling Context */
handling_context?: {
[key: string]: components["schemas"]["JsonValue"];
@@ -7863,6 +7865,8 @@ export interface components {
* Format: uuid
*/
execution_request_id: string;
/** Execution Scope */
execution_scope: string;
/** Handling Context */
handling_context?: {
[key: string]: components["schemas"]["JsonValue"];
@@ -9600,7 +9604,7 @@ export interface components {
* Chart Type
* @enum {string}
*/
chart_type: "line" | "bar";
chart_type: "line" | "bar" | "time_comparison";
/** Series */
series: components["schemas"]["TaskSafeChartSeries"][];
/** Title */
+3 -3
View File
@@ -203,7 +203,8 @@ export class McpQueryClient {
if (!owner.context.projectId || owner.context.projectKey.startsWith("personal-") || !owner.context.purpose || !["ordinary_chat", "task_execution", "task_collaboration"].includes(owner.context.purpose)) throw error(403, "Project query context required");
// No model-supplied identity, scope, bearer or Server URL can pass this boundary.
const schema = this.catalog().find((tool) => tool.name === name)!.inputSchema;
const budget = queryTimeouts(this.tools.find((tool) => tool.name === name)!, args);
const definition = this.tools.find((tool) => tool.name === name)!;
const budget = queryTimeouts(definition, args);
const reserved = new Set(["context_handle", "authorization", "headers", "project_id", "projectId", "network", "url", "server_url", "user_id", "subject", "client_id", "purpose", "native_session_id", "parent_session_id", "task_id", "scope", "scopes"]);
if (Object.keys(args).some((key) => reserved.has(key) || !Object.hasOwn(schema.properties ?? {}, key))) throw error(400, "Unknown query parameter");
const token = await this.userToken(owner.runtimeSessionId, owner.context.projectId, signal);
@@ -223,8 +224,7 @@ export class McpQueryClient {
const handle = body.context_handle;
try {
return await this.withClient(token.token, project, signal, async (client) => {
await client.listTools(undefined, { signal, timeout: 5000 });
return client.callTool({ name, arguments: { ...args, ...(this.tools.find((tool) => tool.name === name)!.inputSchema.properties?.timeout_seconds ? { timeout_seconds: budget.serverMs / 1000 } : {}), context_handle: handle }, _meta: { "io.tjwater/callId": callId } }, { signal, timeout: budget.callMs });
return client.callTool({ name, arguments: { ...args, ...(definition.inputSchema.properties?.timeout_seconds ? { timeout_seconds: budget.serverMs / 1000 } : {}), context_handle: handle }, _meta: { "io.tjwater/callId": callId } }, { signal, timeout: budget.callMs, toolDefinition: definition });
}, budget.callMs);
} finally {
// Expiry remains authoritative if transport cancellation prevents revocation.
+221
View File
@@ -0,0 +1,221 @@
import type { UIMessage, UIMessageChunk } from "ai";
import { createAnswerStreamProjection } from "../chat/answer-stream-projection.js";
import type { UIEnvelopePayload } from "../uiEnvelope/types.js";
import type { PermissionRequestPayload, QuestionRequestPayload, TodoUpdatePayload } from "./chatStream.js";
import {
type ActiveRun, type RunStatus, updateLastAssistantMessage, applyInteractionResponse,
completeBackendProgress, cancelBackendTodos, upsertBackendProgress,
toFrontendPermission, upsertBackendQuestion, upsertBackendTodoUpdate,
appendBackendToolArtifact, appendBackendUiEnvelope,
} from "./chatUiState.js";
/** Compatibility fields are a projection of the same accepted event as canonical parts. */
function reduceRunEvent(activeRun: ActiveRun, event: string, data: Record<string, unknown>, onStatus: (status: RunStatus) => void) {
if (event === "token") {
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
...message,
content: `${typeof message.content === "string" ? message.content : ""}${typeof data.content === "string" ? data.content : ""}`,
isError: false,
}));
} else if (event === "final_answer") {
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
...message, content: typeof data.content === "string" ? data.content : message.content,
}));
} else if (event === "progress") {
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
...message,
progress: upsertBackendProgress(message.progress, data),
}));
} else if (event === "done") {
activeRun.status = data.reason === "waiting_input" ? "waiting_input" : "completed";
onStatus(activeRun.status);
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
...message,
content:
typeof message.content === "string" && message.content.trim()
? message.content
: data.reason === "waiting_input" ? "等待你的回答。" : "Agent 已完成处理,但没有生成文本回答。请查看过程记录,或换个更具体的问题重试。",
progress: completeBackendProgress(message.progress),
}));
} else if (event === "error" || event === "aborted") {
activeRun.status = event === "aborted" || activeRun.status === "aborted" ? "aborted" : "error";
onStatus(activeRun.status);
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
...message,
content:
typeof message.content === "string" && message.content.trim()
? message.content
: `⚠️ **错误:** ${typeof data.message === "string" ? data.message : "unknown error"}`,
isError: true,
progress: completeBackendProgress(message.progress),
todos: cancelBackendTodos(message.todos),
}));
} else if (event === "permission_request") {
const payload = data as PermissionRequestPayload;
activeRun.pendingPermissions.set(payload.request_id, payload);
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
...message,
permissions: [
...(Array.isArray(message.permissions) ? message.permissions : []),
toFrontendPermission(payload),
],
}));
} else if (event === "permission_response") {
const requestId =
typeof data.request_id === "string" ? data.request_id : undefined;
const reply =
data.reply === "once" || data.reply === "always" || data.reply === "reject"
? data.reply
: undefined;
if (requestId && (reply || data.status === "error")) {
if (reply) activeRun.pendingPermissions.delete(requestId);
activeRun.messages = applyInteractionResponse(activeRun.messages, "permission_response", data);
}
} else if (event === "question_request") {
const payload = data as QuestionRequestPayload;
let shouldTrackQuestion = true;
if (payload.tool?.callID) {
if (payload.request_id !== payload.tool.callID) {
activeRun.pendingQuestions.delete(payload.tool.callID);
} else {
const hasActionableQuestion = [...activeRun.pendingQuestions.values()].some(
(question) =>
question.tool?.callID === payload.tool?.callID &&
question.request_id !== payload.tool?.callID,
);
if (hasActionableQuestion) {
activeRun.messages = updateLastAssistantMessage(
activeRun.messages,
(message) => ({
...message,
questions: upsertBackendQuestion(message.questions, payload),
}),
);
shouldTrackQuestion = false;
}
}
}
if (shouldTrackQuestion) {
activeRun.pendingQuestions.set(payload.request_id, payload);
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
...message,
questions: upsertBackendQuestion(message.questions, payload),
}));
}
} else if (event === "question_response") {
const requestId =
typeof data.request_id === "string" ? data.request_id : undefined;
if (requestId) {
activeRun.pendingQuestions.delete(requestId);
activeRun.messages = applyInteractionResponse(activeRun.messages, "question_response", data);
}
} else if (event === "execution_plan") {
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
...message, executionPlans: [...(Array.isArray(message.executionPlans) ? message.executionPlans : []), data],
}));
} else if (event === "todo_update") {
const payload = data as TodoUpdatePayload;
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
...message,
todos: upsertBackendTodoUpdate(message.todos, payload),
}));
} else if (event === "tool_call") {
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
...message,
artifacts: appendBackendToolArtifact(message.artifacts, data),
}));
} else if (event === "ui_envelope") {
const payload = data as UIEnvelopePayload;
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
...message,
uiEnvelopes: appendBackendUiEnvelope(message.uiEnvelopes, payload),
}));
}
}
/** All host mutations go through the accumulator before subscribers or persistence. */
export function publishRunEvent(run: ActiveRun, event: string, data: Record<string, unknown>, notify = true) {
if (run.applyEvent) run.applyEvent(event, data);
else reduceRunEvent(run, event, data, () => {});
if (notify) for (const subscriber of run.subscribers) {
if (event === "done" || event === "error" || event === "aborted") {
subscriber.write("run_status", { ...data, id: "run-status", status: run.status });
}
subscriber.write(event, data);
}
}
/** One event owner updates the reconnect view and canonical wire/storage projections. */
export function createChatRunAccumulator(options: {
run: ActiveRun;
messages: UIMessage[];
messageId: string;
sessionId: string;
collaboration: boolean;
write: (chunk: UIMessageChunk) => void;
onStatus: (status: RunStatus) => void;
}) {
const activeRun = options.run;
let history = [...options.messages];
const syncParts = () => {
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
...message, id: options.messageId, parts: projection.parts.map((part) => ({ ...part })),
}));
};
const projection = createAnswerStreamProjection({
messageId: options.messageId, sessionId: options.sessionId,
write: options.write, onChange: syncParts,
});
const apply = (event: string, data: Record<string, unknown>) => {
reduceRunEvent(activeRun, event, data, options.onStatus);
let historicalResponse = false;
if (event === "permission_response" || event === "question_response") {
const updated = applyInteractionResponse(history, event, data);
historicalResponse = updated !== history;
history = updated;
if (historicalResponse) {
const byId = new Map(history.map((message) => [message.id, message]));
activeRun.messages = activeRun.messages.map((message) => {
if (!message || typeof message !== "object" || !("id" in message)) return message;
const canonical = byId.get(String(message.id));
return canonical ? { ...message, parts: canonical.parts } : message;
});
}
}
const hidden = options.collaboration && ["progress", "todo_update", "tool_call", "ui_envelope", "execution_plan"].includes(event);
if (!hidden && !historicalResponse) {
if (event === "token") projection.appendText(typeof data.content === "string" ? data.content : "");
else if (event === "done" || event === "error" || event === "aborted") {
const message = typeof data.message === "string" ? data.message : event === "aborted" ? "请求已中断" : "Agent stream failed";
if (!projection.parts.some((part) => ["text", "data-final_answer", "data-answer_ui"].includes(String(part.type)))) {
const assistant = activeRun.messages.at(-1) as { content?: string } | undefined;
projection.appendText(assistant?.content || message);
}
projection.recordData("run_status", { id: "run-status", status: activeRun.status, ...(event === "done" ? {} : { message }) });
if (event !== "done" && !projection.finished) options.write({ type: "error", errorText: message });
projection.finish(event === "done" ? "stop" : "error");
} else if (["permission_response", "question_response"].includes(event)) projection.recordData(event, data);
else projection.appendData(event, data);
}
syncParts();
};
const uiMessages = (): UIMessage[] => [...history, {
id: options.messageId, role: "assistant", parts: projection.parts.map((part) => ({ ...part })),
} as UIMessage];
activeRun.applyEvent = apply;
activeRun.uiMessages = uiMessages;
return {
apply,
finish: projection.finish,
get finished() { return projection.finished; },
uiMessages,
appendInput(message: UIMessage, text: string) {
if (history.some((item) => item.id === message.id)) return;
history.push(message);
const tail = activeRun.messages.at(-1);
activeRun.messages = [...activeRun.messages.slice(0, -1), { id: message.id, role: "user", content: text }, ...(tail ? [tail] : [])];
},
};
}
+12 -25
View File
@@ -1,3 +1,5 @@
import { publishRunEvent } from "./chat-run-accumulator.js";
import { toSessionScope } from "../sessions/session-scope.js";
import { type Router } from "express";
import { z } from "zod";
import type { InteractionRequestStore } from "../sessions/interaction-request-store.js";
@@ -13,9 +15,6 @@ import { type FrontendActionCoordinator } from "../frontendAction/coordinator.js
import {
type ActiveRun,
type RunStatus,
cancelBackendTodos,
completeBackendProgress,
updateLastAssistantMessage,
} from "./chatUiState.js";
const abortPayloadSchema = z.object({
@@ -33,11 +32,6 @@ type RegisterAuxiliaryRoutesOptions = {
frontendActionCoordinator: FrontendActionCoordinator;
};
const toSessionUiStateContext = (record: { sessionId: string; actorKey: string; projectKey: string }) => ({
sessionId: record.sessionId,
actorKey: record.actorKey,
projectKey: record.projectKey,
});
export const registerChatAuxiliaryRoutes = (
chatRouter: Router,
@@ -80,12 +74,12 @@ export const registerChatAuxiliaryRoutes = (
const runtimeContext = getRuntimeSessionContext(sessionRecord.sessionId);
if (runtimeContext) setRuntimeSessionContext({ ...runtimeContext, hostQuestionClosed: true });
activeRuns.get(sessionRecord.sessionId)?.controller.abort();
await interactionRequestStore?.cancelPending(toSessionUiStateContext(sessionRecord));
const pendingInteraction = await interactionRequestStore?.getBlocking(toSessionUiStateContext(sessionRecord));
await interactionRequestStore?.cancelPending(toSessionScope(sessionRecord));
const pendingInteraction = await interactionRequestStore?.getBlocking(toSessionScope(sessionRecord));
if (pendingInteraction?.delivery && runtime && interactionRequestStore) {
const status = await runtime.stopInteractionInput(sessionRecord.sessionId, pendingInteraction.delivery.native_message_id);
const latest = await interactionRequestStore.get(toSessionUiStateContext(sessionRecord), pendingInteraction.interaction_id);
if (latest?.delivery && !["delivered", "cancelled"].includes(latest.delivery.status)) await interactionRequestStore.updateDelivery(toSessionUiStateContext(sessionRecord), latest.interaction_id, {
const latest = await interactionRequestStore.get(toSessionScope(sessionRecord), pendingInteraction.interaction_id);
if (latest?.delivery && !["delivered", "cancelled"].includes(latest.delivery.status)) await interactionRequestStore.updateDelivery(toSessionScope(sessionRecord), latest.interaction_id, {
expected_revision: latest.delivery.revision, status,
});
}
@@ -99,28 +93,21 @@ export const registerChatAuxiliaryRoutes = (
run.status = "aborted";
lastRunStatuses.set(parsed.data.session_id, "aborted");
run.controller.abort();
run.messages = updateLastAssistantMessage(run.messages, (message) => ({
...message,
content:
typeof message.content === "string" && message.content.trim()
? message.content
: "⚠️ **请求已中断**",
isError: true,
progress: completeBackendProgress(message.progress),
todos: cancelBackendTodos(message.todos),
}));
publishRunEvent(run, "aborted", { session_id: parsed.data.session_id, message: "请求已中断" }, false);
const currentState = await sessionUiStateStore.read(
toSessionUiStateContext(sessionRecord),
toSessionScope(sessionRecord),
);
await sessionUiStateStore.write(toSessionUiStateContext(sessionRecord), {
await sessionUiStateStore.write(toSessionScope(sessionRecord), {
sessionId: sessionRecord.sessionId,
isTitleManuallyEdited: currentState?.isTitleManuallyEdited ?? false,
messages: run.messages,
...currentState,
messages: run.uiMessages?.() ?? run.messages,
});
for (const subscriber of run.subscribers) {
subscriber.write("error", {
session_id: parsed.data.session_id,
message: "请求已中断",
status: "aborted",
});
}
// The generation route emits its terminal Response after cleanup and
+25 -53
View File
@@ -1,3 +1,5 @@
import { publishRunEvent } from "./chat-run-accumulator.js";
import { toSessionScope } from "../sessions/session-scope.js";
import { refreshSessionCredential } from "../product/credential-refresh.js";
import { type Router } from "express";
import { z } from "zod";
@@ -8,9 +10,7 @@ import { type OpencodeRuntimeAdapter } from "../runtime/opencode.js";
import { SessionContextMismatchError, type SessionMetadataStore } from "../sessions/metadataStore.js";
import { type SessionUiStateStore } from "../sessions/uiStateStore.js";
import {
type ActiveRun,
toPermissionStatus,
updateLastAssistantPermission,
type ActiveRun, applyInteractionResponse,
} from "./chatUiState.js";
const permissionReplyPayloadSchema = z.object({
@@ -35,7 +35,6 @@ type RegisterInteractionRoutesOptions = {
sessionUiStateStore: SessionUiStateStore;
};
const toSessionUiStateContext = (sessionId: string) => ({ sessionId });
export const registerChatInteractionRoutes = (
chatRouter: Router,
@@ -104,12 +103,13 @@ export const registerChatInteractionRoutes = (
}
const persistPermissionState = async () => {
const currentState = await sessionUiStateStore.read(
toSessionUiStateContext(sessionRecord.sessionId),
toSessionScope(sessionRecord),
);
await sessionUiStateStore.write(toSessionUiStateContext(sessionRecord.sessionId), {
await sessionUiStateStore.write(toSessionScope(sessionRecord), {
sessionId: sessionRecord.sessionId,
isTitleManuallyEdited: currentState?.isTitleManuallyEdited ?? false,
messages: run.messages,
...currentState,
messages: run.uiMessages?.() ?? run.messages,
});
};
@@ -121,18 +121,10 @@ export const registerChatInteractionRoutes = (
message: parsed.data.message,
});
} catch (error) {
run.messages = updateLastAssistantPermission(
run.messages,
requestId,
(permission) => ({
...permission,
status: "error",
error:
error instanceof Error
? error.message
: "failed to reply permission",
}),
);
publishRunEvent(run, "permission_response", {
session_id: sessionRecord.sessionId, request_id: requestId, status: "error",
error: error instanceof Error ? error.message : "failed to reply permission",
});
await persistPermissionState().catch((persistError) => {
logger.warn(
{ err: persistError, sessionId: sessionRecord.sessionId },
@@ -146,30 +138,15 @@ export const registerChatInteractionRoutes = (
return;
}
run.pendingPermissions.delete(requestId);
const status = toPermissionStatus(parsed.data.reply);
run.messages = updateLastAssistantPermission(
run.messages,
requestId,
(permission) => ({
...permission,
status,
repliedAt: Date.now(),
}),
);
publishRunEvent(run, "permission_response", {
session_id: sessionRecord.sessionId, request_id: requestId, reply: parsed.data.reply,
});
await persistPermissionState().catch((persistError) => {
logger.warn(
{ err: persistError, sessionId: sessionRecord.sessionId },
"failed to persist permission reply state",
);
});
for (const subscriber of run.subscribers) {
subscriber.write("permission_response", {
session_id: sessionRecord.sessionId,
request_id: requestId,
reply: parsed.data.reply,
});
}
res.status(202).json({
session_id: sessionRecord.sessionId,
@@ -219,23 +196,19 @@ export const registerChatInteractionRoutes = (
}
}
// A host-confirmed reply remains authoritative after native forms are pruned.
const nativeStatus = confirmedStatus ? undefined : await runtime.getQuestionStatus({ sessionId: session.sessionId, requestId: canonicalId });
const form = confirmedStatus ? undefined : await runtime.getQuestion({ sessionId: session.sessionId, requestId: canonicalId });
const nativeStatus = form?.state.status ?? "missing";
const status = confirmedStatus ?? (nativeStatus === "cancelled" ? "rejected" : nativeStatus === "missing" ? "invalidated" : nativeStatus!);
const persist = async (nextStatus: string, answers?: string[][]) => {
const update = (messages: unknown[]) => messages.map((message) => {
if (!isRecord(message)) return message;
const questions = Array.isArray(message.questions) ? message.questions.map((q: unknown) => isRecord(q) && q.requestId === canonicalId ? { ...q, status: nextStatus, ...(answers ? { answers } : {}), error: undefined } : q) : undefined;
const parts = Array.isArray(message.parts) ? [...message.parts] : undefined;
if (parts && nextStatus !== "pending" && parts.some((part: unknown) => isRecord(part) && part.type === "data-question_request" && isRecord(part.data) && part.data.request_id === canonicalId)) {
const response = { type: "data-question_response", data: { session_id: session.sessionId, request_id: canonicalId, status: nextStatus, rejected: nextStatus === "rejected" || nextStatus === "invalidated", ...(answers ? { answers } : {}) } };
const index = parts.findIndex((part: unknown) => isRecord(part) && part.type === "data-question_response" && isRecord(part.data) && part.data.request_id === canonicalId);
if (index >= 0) parts[index] = { ...parts[index], data: { ...(isRecord(parts[index]) && isRecord(parts[index].data) ? parts[index].data : {}), ...response.data } }; else parts.push(response);
}
return { ...message, ...(questions ? { questions } : {}), ...(parts ? { parts } : {}) };
});
const response = {
session_id: session.sessionId, request_id: canonicalId, status: nextStatus,
rejected: nextStatus === "rejected" || nextStatus === "invalidated", ...(answers ? { answers } : {}),
};
// Update the live projection before entering the same atomic persistence queue.
if (run) { run.messages = update(run.messages); if (nextStatus !== "pending") run.pendingQuestions.delete(canonicalId); }
await sessionUiStateStore.update(context, (latest) => ({ ...latest, messages: update(latest.messages) }));
if (run && nextStatus !== "pending") publishRunEvent(run, "question_response", response);
await sessionUiStateStore.update(context, (latest) => ({
...latest, messages: applyInteractionResponse(latest.messages, "question_response", response),
}));
};
if (action === "status" || status !== "pending") {
await persist(status);
@@ -243,10 +216,9 @@ export const registerChatInteractionRoutes = (
return;
}
const answers = action === "reply" ? questionReplyPayloadSchema.parse(req.body).answers : undefined;
if (action === "reply") await runtime.replyQuestion({ sessionId: session.sessionId, requestId: canonicalId, answers: answers ?? [] });
if (action === "reply") await runtime.replyQuestion({ sessionId: session.sessionId, requestId: canonicalId, answers: answers ?? [], form: form! });
else await runtime.rejectQuestion({ sessionId: session.sessionId, requestId: canonicalId });
await persist(action === "reply" ? "answered" : "rejected", answers);
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) {
+2 -2
View File
@@ -1,5 +1,5 @@
// Called only for native "asked" events. Explicit denials and host isolation
// still run before this point; the selected round mode answers pending requests.
export function canAutomaticallyApprove(mode: string, action: string, backgroundSandboxShell = false) {
return mode === "always" || (backgroundSandboxShell && action === "shell");
export function canAutomaticallyApprove(mode: string, action: string, backgroundSandboxPermissions = false) {
return mode === "always" || (backgroundSandboxPermissions && (action === "shell" || action === "edit"));
}
-21
View File
@@ -174,27 +174,6 @@ export const generateSessionTitle = async (
}
};
export const getConversationTurnStats = async (
runtime: OpencodeRuntimeAdapter,
sessionId: string,
) => {
const messages = await runtime.messages(sessionId, 12);
return messages.reduce(
(stats, message) => {
if (message.info.role === "user") {
stats.userMessageCount += 1;
} else if (message.info.role === "assistant") {
stats.assistantMessageCount += 1;
}
return stats;
},
{
userMessageCount: 0,
assistantMessageCount: 0,
},
);
};
export const buildPromptWithLearningContext = async (
memoryStore: MemoryStore,
actorKey: string,
+24 -25
View File
@@ -1,3 +1,4 @@
import { toSessionScope } from "../sessions/session-scope.js";
import { registerChatTopicRoutes } from "./chat-topic-routes.js";
import { matchHistoryText } from "../sessions/history-search.js";
import { credentialRefreshCoordinator } from "../product/credential-refresh.js";
@@ -107,11 +108,6 @@ type RegisterChatSessionRoutesOptions = {
sessionUiStateStore: SessionUiStateStore;
};
const toSessionUiStateContext = (sessionRecord: SessionRecord) => ({
sessionId: sessionRecord.sessionId,
actorKey: sessionRecord.actorKey,
projectKey: sessionRecord.projectKey,
});
const toSse = (event: string, data: Record<string, unknown>) =>
`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
@@ -176,6 +172,12 @@ export const registerChatSessionRoutes = (
const { activeRuns, lastRunStatuses } = runState;
const getSessionRunStatus = runState.getStatus;
const readMessages = (record: SessionRecord, state: { messages: unknown[] } | null) => {
if (readPurpose(record) === "task_collaboration") return state?.messages ?? [];
const run = activeRuns.get(record.sessionId);
return run?.uiMessages?.() ?? run?.messages ?? state?.messages ?? [];
};
const withInteractionReplies = async (record: SessionRecord, messages: unknown[]) => {
if (!interactionRequestStore) return messages;
const ids = new Set(messages.flatMap((message) => {
@@ -183,7 +185,7 @@ export const registerChatSessionRoutes = (
const value = message as Record<string, unknown>;
return value.role === "user" && typeof value.id === "string" ? [value.id] : [];
}));
const replies = await interactionRequestStore.forMessages(toSessionUiStateContext(record), ids);
const replies = await interactionRequestStore.forMessages(toSessionScope(record), ids);
return projectInteractionReplyMessages(messages, replies);
};
@@ -192,7 +194,7 @@ export const registerChatSessionRoutes = (
|| !isGenericTaskCollaborationTitle(record.title)) return record;
if ((!credential && !knownTaskTitle) || signal?.aborted) return record;
try {
const state = await sessionUiStateStore.read(toSessionUiStateContext(record));
const state = await sessionUiStateStore.read(toSessionScope(record));
if (state?.isTitleManuallyEdited) return record;
let taskTitle = knownTaskTitle;
if (!taskTitle && credential) {
@@ -206,7 +208,7 @@ export const registerChatSessionRoutes = (
}
}
if (!taskTitle?.trim()) return record;
const latestState = await sessionUiStateStore.read(toSessionUiStateContext(record));
const latestState = await sessionUiStateStore.read(toSessionScope(record));
if (latestState?.isTitleManuallyEdited) return record;
return await sessionMetadataStore.touch(record, { title: taskTitle }, {
preserveUpdatedAt: true, expectedTitle: record.title ?? "",
@@ -297,7 +299,7 @@ export const registerChatSessionRoutes = (
sessionId,
);
if (!record) return null;
const state = await sessionUiStateStore.read(toSessionUiStateContext(record));
const state = await sessionUiStateStore.read(toSessionScope(record));
if (!state) return null;
return { record, state, taskTitle };
};
@@ -603,7 +605,7 @@ export const registerChatSessionRoutes = (
const titled = await sessionMetadataStore.touch(record, { title: `${history.taskTitle} · 调试` });
const messages = await withInteractionReplies(history.record, history.state.messages ?? []);
await sessionUiStateStore.write(
toSessionUiStateContext(titled),
toSessionScope(titled),
buildForkedSessionUiState({ messages }, {
keepMessageCount: messages.length,
targetSessionId: titled.sessionId,
@@ -771,8 +773,8 @@ export const registerChatSessionRoutes = (
const searchable: Array<SessionRecord & { searchSnippet?: string }> = [];
for (const record of records) {
const titleMatches = (record.title ?? "新对话").toLocaleLowerCase().includes((query.data.q ?? "").toLocaleLowerCase());
const state = query.data.q && !titleMatches ? await sessionUiStateStore.read(toSessionUiStateContext(record)) : null;
const messages = activeRuns.get(record.sessionId)?.messages ?? state?.messages ?? [];
const state = query.data.q && !titleMatches ? await sessionUiStateStore.read(toSessionScope(record)) : null;
const messages = readMessages(record, state);
searchable.push({ ...record, searchSnippet: query.data.q && !titleMatches ? matchHistoryText(messages, query.data.q) : undefined });
}
let page;
@@ -828,9 +830,8 @@ export const registerChatSessionRoutes = (
const context = getLocalAgentContext(req);
const record = await sessionMetadataStore.get(context, req.params.session_id);
if (!record) { res.status(404).json({ message: "session not found" }); return; }
const state = await sessionUiStateStore.read(toSessionUiStateContext(record));
const messages = readPurpose(record) === "task_collaboration" ? state?.messages ?? []
: activeRuns.get(record.sessionId)?.messages ?? state?.messages ?? [];
const state = await sessionUiStateStore.read(toSessionScope(record));
const messages = readMessages(record, state);
try {
const page = sessionMessagePage(messages, { scope: historyScope([record.actorKey, record.projectKey, record.sessionId]), ...query.data });
res.json({ ...page, messages: await withInteractionReplies(record, page.messages) });
@@ -871,10 +872,9 @@ export const registerChatSessionRoutes = (
sessionRecord = await refreshCollaborationTitle(sessionRecord, getTrustedProductCredential(req));
const state = await sessionUiStateStore.read(
toSessionUiStateContext(sessionRecord),
toSessionScope(sessionRecord),
);
const messages = readPurpose(sessionRecord) === "task_collaboration" ? state?.messages ?? []
: activeRuns.get(sessionRecord.sessionId)?.messages ?? state?.messages ?? [];
const messages = readMessages(sessionRecord, state);
const messagePayload = query.data.message_limit ? sessionMessagePage(messages, {
scope: historyScope([sessionRecord.actorKey, sessionRecord.projectKey, sessionId]), limit: query.data.message_limit,
}) : { messages };
@@ -929,9 +929,8 @@ export const registerChatSessionRoutes = (
res.flushHeaders?.();
const run = activeRuns.get(sessionRecord.sessionId);
const state = await sessionUiStateStore.read(toSessionUiStateContext(sessionRecord));
const messages = readPurpose(sessionRecord) === "task_collaboration" ? state?.messages ?? []
: run?.messages ?? state?.messages ?? [];
const state = await sessionUiStateStore.read(toSessionScope(sessionRecord));
const messages = readMessages(sessionRecord, state);
const messagePayload = query.data.message_limit ? sessionMessagePage(messages, {
scope: historyScope([sessionRecord.actorKey, sessionRecord.projectKey, sessionId]), limit: query.data.message_limit,
}) : { messages };
@@ -1005,11 +1004,11 @@ export const registerChatSessionRoutes = (
title,
});
const state = await sessionUiStateStore.read(
toSessionUiStateContext(nextSessionRecord),
toSessionScope(nextSessionRecord),
);
if (state) {
await sessionUiStateStore.write(
toSessionUiStateContext(nextSessionRecord),
toSessionScope(nextSessionRecord),
{
...state,
isTitleManuallyEdited:
@@ -1134,7 +1133,7 @@ export const registerChatSessionRoutes = (
}
const sourceState = sourceSessionRecord
? await sessionUiStateStore.read(
toSessionUiStateContext(sourceSessionRecord),
toSessionScope(sourceSessionRecord),
)
: null;
const forkTitle = sourceSessionRecord?.title
@@ -1145,7 +1144,7 @@ export const registerChatSessionRoutes = (
{ title: forkTitle },
);
await sessionUiStateStore.write(
toSessionUiStateContext(titledTargetSessionRecord),
toSessionScope(titledTargetSessionRecord),
buildForkedSessionUiState(sourceState, {
keepMessageCount: parsed.data.keep_message_count,
targetSessionId: nextSessionId,
+6 -6
View File
@@ -84,7 +84,7 @@ type StreamPromptOptions = {
message: string;
model?: SupportedModel;
approvalMode?: ApprovalMode;
backgroundSandboxShell?: boolean;
backgroundSandboxPermissions?: boolean;
traceId?: string;
projectId?: string;
signal?: AbortSignal;
@@ -149,7 +149,7 @@ export const streamPromptResponse = async ({
message,
model,
approvalMode = "request",
backgroundSandboxShell = false,
backgroundSandboxPermissions = false,
traceId,
projectId,
signal,
@@ -483,7 +483,7 @@ export const streamPromptResponse = async ({
patterns: event.properties.patterns,
elapsedMs: Math.max(0, Date.now() - requestStartedAt),
});
const autoApprove = canAutomaticallyApprove(approvalMode, event.properties.permission, backgroundSandboxShell);
const autoApprove = canAutomaticallyApprove(approvalMode, event.properties.permission, backgroundSandboxPermissions);
emitProgress({
id: `permission-${event.properties.id}`,
phase: "permission",
@@ -493,7 +493,7 @@ export const streamPromptResponse = async ({
autoApprove
? approvalMode === "always"
? "当前模式为本轮自动批准,已允许本次请求,不保存后续轮次授权。"
: "后台任务的沙箱 Shell 已自动批准;沙箱与明确拒绝规则仍生效。"
: "后台任务的会话内 Shell / 文件修改已自动批准;目录边界与明确拒绝规则仍生效。"
: buildPermissionDetail(event),
});
if (autoApprove) {
@@ -533,7 +533,7 @@ export const streamPromptResponse = async ({
resources: event.properties.resources,
elapsedMs: Math.max(0, Date.now() - requestStartedAt),
});
const autoApprove = canAutomaticallyApprove(approvalMode, event.properties.action, backgroundSandboxShell);
const autoApprove = canAutomaticallyApprove(approvalMode, event.properties.action, backgroundSandboxPermissions);
emitProgress({
id: `permission-${event.properties.id}`,
phase: "permission",
@@ -543,7 +543,7 @@ export const streamPromptResponse = async ({
autoApprove
? approvalMode === "always"
? "当前模式为本轮自动批准,已允许本次请求,不保存后续轮次授权。"
: "后台任务的沙箱 Shell 已自动批准;沙箱与明确拒绝规则仍生效。"
: "后台任务的会话内 Shell / 文件修改已自动批准;目录边界与明确拒绝规则仍生效。"
: buildPermissionV2Detail(event),
});
if (autoApprove) {
+24 -255
View File
@@ -1,3 +1,4 @@
import { toSessionScope } from "../sessions/session-scope.js";
import type { TaskCalculationWaiter } from "../mcp/task-calculation-waiter.js";
import { createResponsesProjection, responseRequestSchema, responsePrompt } from "../chat/responses-protocol.js";
import { ResponseInputTracker, WaitingResponseInputQueue } from "../chat/response-inputs.js";
@@ -5,7 +6,7 @@ import { reconcileInteractionDelivery } from "../sessions/interaction-delivery-r
import type { InteractionRequestStore } from "../sessions/interaction-request-store.js";
import { renderInteractionAnswer, type InteractionRequest } from "../chat/interaction-request-contract.js";
import { createInteractionReplyMessage } from "../chat/interaction-message-presentation.js";
import { createAnswerStreamProjection } from "../chat/answer-stream-projection.js";
import { createChatRunAccumulator, publishRunEvent } from "./chat-run-accumulator.js";
import type { AgentMessageCoordinator } from "../chat/agent-message-coordinator.js";
import { credentialRefreshCoordinator } from "../product/credential-refresh.js";
import { type Router } from "express";
@@ -47,7 +48,6 @@ import {
} from "../product/operational-task-client.js";
import { type SessionTranscriptStore } from "../sessions/transcriptStore.js";
import { type SessionUiStateStore } from "../sessions/uiStateStore.js";
import type { UIEnvelopePayload } from "../uiEnvelope/types.js";
import { toActorKey, toProjectKey } from "../utils/fileStore.js";
import {
buildPromptWithLearningContext,
@@ -59,29 +59,14 @@ import {
} from "./chatSession.js";
import {
collectTextContent,
type PermissionRequestPayload,
type QuestionRequestPayload,
streamPromptResponse,
type TodoUpdatePayload,
} from "./chatStream.js";
import { type ChatRunState } from "./chatRunState.js";
import {
type ActiveRun,
type StreamSubscriber,
appendBackendToolArtifact,
appendBackendUiEnvelope,
cancelBackendTodos,
completeBackendProgress,
createInitialStreamingMessages,
isObjectRecord,
toFrontendPermission,
toPermissionStatus,
updateLastAssistantMessage,
updateLastAssistantPermission,
updateLastAssistantQuestion,
upsertBackendProgress,
upsertBackendQuestion,
upsertBackendTodoUpdate,
} from "./chatUiState.js";
import {
agentTriggerContextSchema,
@@ -106,11 +91,6 @@ const payloadSchema = z.object({
}).strict().optional(),
});
const toSessionUiStateContext = (sessionRecord: SessionRecord) => ({
sessionId: sessionRecord.sessionId,
actorKey: sessionRecord.actorKey,
projectKey: sessionRecord.projectKey,
});
const runtimeHasConversation = async (
runtime: OpencodeRuntimeAdapter,
@@ -311,7 +291,7 @@ export const registerChatStreamRoute = (
let boundOperationalSessionId: string | undefined;
let startingSessionId: string | undefined;
let continuation: InteractionRequest | null = null;
let continuationScope: ReturnType<typeof toSessionUiStateContext> | undefined;
let continuationScope: ReturnType<typeof toSessionScope> | undefined;
let waitingInteraction: InteractionRequest | null = null;
try {
const authContext = getLocalAgentContext(req);
@@ -365,7 +345,7 @@ export const registerChatStreamRoute = (
res.status(400).json({ message: "Interaction continuation requires an existing conversation" }); return;
}
if (existingSessionRecord && interactionRequestStore) {
continuationScope = toSessionUiStateContext(existingSessionRecord);
continuationScope = toSessionScope(existingSessionRecord);
if (input.tjwater.interaction_id) {
continuation = await interactionRequestStore.get(continuationScope, input.tjwater.interaction_id);
if (!continuation?.delivery || !["answered", "skipped"].includes(continuation.status) || continuation.delivery.status === "cancelled") {
@@ -523,7 +503,7 @@ export const registerChatStreamRoute = (
sessionId: requestContext.clientSessionId,
};
const initialSessionState = await sessionUiStateStore.read(
toSessionUiStateContext(activeSessionRecord),
toSessionScope(activeSessionRecord),
);
const persistedUiMessages = toUiMessages(initialSessionState?.messages);
if (!continuation && input.tjwater.client_message_id && persistedUiMessages.some((message) => message.id === input.tjwater.client_message_id)) {
@@ -551,7 +531,6 @@ export const registerChatStreamRoute = (
...visiblePersistedMessages,
createFallbackUserUiMessage(promptText, input.tjwater.client_message_id),
];
let latestUiMessages = requestMessages;
const baseMessages = toLegacyMessages(requestMessages);
const recentTurns = await sessionTranscriptStore.getRecentTurns(historyContext, 8);
@@ -605,44 +584,28 @@ export const registerChatStreamRoute = (
startingSessions.delete(activeSessionRecord.sessionId);
startingSessionId = undefined;
lastRunStatuses.set(clientSessionId, "running");
const sessionUiStateContext = toSessionUiStateContext(activeSessionRecord);
const sessionUiStateContext = toSessionScope(activeSessionRecord);
const assistantUiMessageId = continuationAssistantId ?? `msg_${randomUUID().replaceAll("-", "")}`;
setRuntimeSessionContext({ ...getRuntimeSessionContext(binding.sessionId)!, hostTurnId: wire.response.id, hostMessageId: assistantUiMessageId });
const syncLatestUiMessages = () => {
latestUiMessages = [
...requestMessages,
{
id: assistantUiMessageId,
role: "assistant",
parts: assistantParts.map((part) => ({ ...part })),
} as UIMessage,
];
};
const writeUiChunk = (chunk: UIMessageChunk) => {
// Persist canonical parts after disconnect; the wire emitter checks the socket.
wire!.write(chunk);
};
const projection = createAnswerStreamProjection({
messageId: assistantUiMessageId, sessionId: clientSessionId,
write: writeUiChunk, onChange: syncLatestUiMessages,
const accumulator = createChatRunAccumulator({
run: activeRun, messages: requestMessages, messageId: assistantUiMessageId, sessionId: clientSessionId,
collaboration: sessionPurpose === "task_collaboration", write: (chunk) => wire!.write(chunk),
onStatus: (status) => lastRunStatuses.set(clientSessionId, status),
});
const assistantParts = projection.parts;
const writeTextDelta = projection.appendText;
const writeDataPart = projection.appendData;
const finishUiMessage = projection.finish;
const writeUiChunk = (chunk: UIMessageChunk) => wire!.write(chunk);
writeUiChunk({ type: "start", messageId: assistantUiMessageId });
let persistQueue = sessionUiStateStore.write(sessionUiStateContext, {
executionPlans: executionPlanHistory,
sessionId: activeSessionRecord.sessionId,
isTitleManuallyEdited: initialSessionState?.isTitleManuallyEdited ?? false,
messages: sessionPurpose === "task_collaboration" ? toCollaborationMessages(latestUiMessages) : latestUiMessages,
messages: sessionPurpose === "task_collaboration" ? toCollaborationMessages(accumulator.uiMessages()) : accumulator.uiMessages(),
});
const queueSessionUiStatePersist = () => {
const snapshot = {
executionPlans: executionPlanHistory,
sessionId: activeSessionRecord.sessionId,
isTitleManuallyEdited: initialSessionState?.isTitleManuallyEdited ?? false,
messages: sessionPurpose === "task_collaboration" ? toCollaborationMessages(latestUiMessages) : latestUiMessages,
messages: sessionPurpose === "task_collaboration" ? toCollaborationMessages(accumulator.uiMessages()) : accumulator.uiMessages(),
};
persistQueue = persistQueue
.catch((error) => {
@@ -654,36 +617,6 @@ export const registerChatStreamRoute = (
.then(() => sessionUiStateStore.write(sessionUiStateContext, snapshot));
return persistQueue;
};
const primarySubscriber: StreamSubscriber = {
write: (event, data) => {
if (event === "token") {
writeTextDelta(typeof data.content === "string" ? data.content : "");
return;
}
if (event === "done") {
finishUiMessage("stop");
return;
}
if (event === "error") {
writeUiChunk({
type: "error",
errorText:
typeof data.message === "string"
? data.message
: "Agent stream failed",
});
finishUiMessage("error");
return;
}
writeDataPart(event, data);
},
close: () => {
if (!res.writableEnded && !res.destroyed) {
res.end();
}
},
};
activeRun.subscribers.add(primarySubscriber);
let taskArtifactPublishQueue: Promise<unknown> = Promise.resolve();
const publish = (event: string, data: Record<string, unknown>) => {
@@ -697,131 +630,11 @@ export const registerChatStreamRoute = (
: assistantUiMessageId,
}
: data;
if (event === "token") {
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
...message,
content: `${typeof message.content === "string" ? message.content : ""}${typeof data.content === "string" ? data.content : ""}`,
isError: false,
}));
} else if (event === "final_answer") {
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
...message, content: typeof data.content === "string" ? data.content : message.content,
}));
} else if (event === "progress") {
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
...message,
progress: upsertBackendProgress(message.progress, data),
}));
} else if (event === "done") {
activeRun.status = data.reason === "waiting_input" ? "waiting_input" : "completed";
lastRunStatuses.set(clientSessionId, activeRun.status);
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
...message,
content:
typeof message.content === "string" && message.content.trim()
? message.content
: data.reason === "waiting_input" ? "等待你的回答。" : "Agent 已完成处理,但没有生成文本回答。请查看过程记录,或换个更具体的问题重试。",
progress: completeBackendProgress(message.progress),
}));
} else if (event === "error") {
activeRun.status = activeRun.status === "aborted" ? "aborted" : "error";
lastRunStatuses.set(clientSessionId, activeRun.status);
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
...message,
content:
typeof message.content === "string" && message.content.trim()
? message.content
: `⚠️ **错误:** ${typeof data.message === "string" ? data.message : "unknown error"}`,
isError: true,
progress: completeBackendProgress(message.progress),
todos: cancelBackendTodos(message.todos),
}));
} else if (event === "permission_request") {
const payload = data as PermissionRequestPayload;
activeRun.pendingPermissions.set(payload.request_id, payload);
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
...message,
permissions: [
...(Array.isArray(message.permissions) ? message.permissions : []),
toFrontendPermission(payload),
],
}));
} else if (event === "permission_response") {
const requestId =
typeof data.request_id === "string" ? data.request_id : undefined;
const reply =
data.reply === "once" || data.reply === "always" || data.reply === "reject"
? data.reply
: undefined;
if (requestId && reply) {
activeRun.pendingPermissions.delete(requestId);
activeRun.messages = updateLastAssistantPermission(
activeRun.messages,
requestId,
(permission) => ({
...permission,
status: toPermissionStatus(reply),
repliedAt: Date.now(),
}),
);
}
} else if (event === "question_request") {
const payload = data as QuestionRequestPayload;
let shouldTrackQuestion = true;
if (payload.tool?.callID) {
if (payload.request_id !== payload.tool.callID) {
activeRun.pendingQuestions.delete(payload.tool.callID);
} else {
const hasActionableQuestion = [...activeRun.pendingQuestions.values()].some(
(question) =>
question.tool?.callID === payload.tool?.callID &&
question.request_id !== payload.tool?.callID,
);
if (hasActionableQuestion) {
activeRun.messages = updateLastAssistantMessage(
activeRun.messages,
(message) => ({
...message,
questions: upsertBackendQuestion(message.questions, payload),
}),
);
shouldTrackQuestion = false;
}
}
}
if (shouldTrackQuestion) {
activeRun.pendingQuestions.set(payload.request_id, payload);
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
...message,
questions: upsertBackendQuestion(message.questions, payload),
}));
}
} else if (event === "question_response") {
const requestId =
typeof data.request_id === "string" ? data.request_id : undefined;
if (requestId) {
activeRun.pendingQuestions.delete(requestId);
activeRun.messages = updateLastAssistantQuestion(
activeRun.messages,
requestId,
(question) => ({
...question,
status: data.rejected === true ? "rejected" : "answered",
repliedAt: Date.now(),
answers: Array.isArray(data.answers) ? data.answers : question.answers,
error: undefined,
}),
);
}
} else if (event === "execution_plan") {
if (event === "execution_plan") {
try {
const snapshot = createExecutionPlanSnapshot(data, [{ executionPlans: executionPlanHistory }], clientSessionId);
executionPlanHistory = [...executionPlanHistory, snapshot];
subscriberData = snapshot;
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
...message,
executionPlans: [...(Array.isArray(message.executionPlans) ? message.executionPlans : []), snapshot],
}));
if (activeSessionRecord.taskId && executionPlanNeedsTaskPublish(data)) {
taskArtifactPublishQueue = taskArtifactPublishQueue
.catch(() => undefined)
@@ -843,23 +656,6 @@ export const registerChatStreamRoute = (
});
return;
}
} else if (event === "todo_update") {
const payload = data as TodoUpdatePayload;
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
...message,
todos: upsertBackendTodoUpdate(message.todos, payload),
}));
} else if (event === "tool_call") {
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
...message,
artifacts: appendBackendToolArtifact(message.artifacts, data),
}));
} else if (event === "ui_envelope") {
const payload = data as UIEnvelopePayload;
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
...message,
uiEnvelopes: appendBackendUiEnvelope(message.uiEnvelopes, payload),
}));
}
const hiddenCollaborationEvent = sessionPurpose === "task_collaboration" && [
@@ -869,12 +665,7 @@ export const registerChatStreamRoute = (
"ui_envelope",
"execution_plan",
].includes(event);
if (!hiddenCollaborationEvent) {
for (const subscriber of activeRun.subscribers) {
subscriber.write(event, subscriberData);
}
}
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ ...message, id: assistantUiMessageId, parts: assistantParts.map((part) => ({ ...part })) }));
publishRunEvent(activeRun, event, subscriberData, !hiddenCollaborationEvent);
void queueSessionUiStatePersist().catch((error) => {
logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state");
});
@@ -944,10 +735,7 @@ export const registerChatStreamRoute = (
: typeof generation === "number" && generation > 0 ? generation : null,
report_revision: reportRef.success ? reportRef.data.revision : null,
};
const part = { type: "data-task_context" as const, data: reference };
assistantParts.push(part);
writeUiChunk(part);
syncLatestUiMessages();
accumulator.apply("task_context", reference);
}
const preparedMessage = taskContext
? [
@@ -970,11 +758,8 @@ export const registerChatStreamRoute = (
const messaging = getTrustedProductCredential(req) ? agentMessages : undefined;
messaging?.start(activeSessionRecord);
const inputTracker = new ResponseInputTracker(runtime, binding.sessionId, (receipt) => {
if (receipt.status === "delivered" && !requestMessages.some((message) => message.id === receipt.id)) {
requestMessages.push({ id: receipt.id, role: "user", parts: [{ type: "text", text: receipt.text }] } as UIMessage);
const tail = activeRun.messages.at(-1);
activeRun.messages = [...activeRun.messages.slice(0, -1), { id: receipt.id, role: "user", content: receipt.text }, ...(tail ? [tail] : [])];
syncLatestUiMessages();
if (receipt.status === "delivered") {
accumulator.appendInput({ id: receipt.id, role: "user", parts: [{ type: "text", text: receipt.text }] } as UIMessage, receipt.text);
}
publish("user_input", { session_id: clientSessionId, ...receipt });
});
@@ -997,7 +782,7 @@ export const registerChatStreamRoute = (
message: preparedMessage,
model: parsed.data.model,
approvalMode: parsed.data.approval_mode,
backgroundSandboxShell: sessionPurpose === "task_execution",
backgroundSandboxPermissions: sessionPurpose === "task_execution",
traceId: requestContext.traceId,
projectId: requestContext.projectId,
signal: abortController.signal,
@@ -1125,14 +910,14 @@ export const registerChatStreamRoute = (
const assistantMessage = [...messages]
.reverse()
.find((message) => message.info.role === "assistant");
const assistantText = readUiMessageText(latestUiMessages.at(-1)) || collectTextContent(assistantMessage?.parts ?? []);
const assistantText = readUiMessageText(accumulator.uiMessages().at(-1)) || collectTextContent(assistantMessage?.parts ?? []);
const latestSessionRecord =
(await sessionMetadataStore.get(
{ actorKey, projectId, projectKey, userId },
activeSessionRecord.sessionId,
)) ?? activeSessionRecord;
const latestSessionState = await sessionUiStateStore.read(
toSessionUiStateContext(latestSessionRecord),
toSessionScope(latestSessionRecord),
);
const existingSessionTitle = latestSessionRecord.title;
let sessionTitle = existingSessionTitle;
@@ -1229,29 +1014,13 @@ export const registerChatStreamRoute = (
await taskArtifactPublishQueue.catch(() => undefined);
unregisterFrontendActionSink?.();
if (abortController.signal.aborted && !taskLeaseFailure) {
activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
...message,
content:
typeof message.content === "string" && message.content.trim()
? message.content
: "⚠️ **请求已中断**",
isError: true,
progress: completeBackendProgress(message.progress),
todos: cancelBackendTodos(message.todos),
}));
if (!projection.finished) {
writeUiChunk({
type: "error",
errorText: "请求已中断",
});
finishUiMessage("error");
}
publishRunEvent(activeRun, "aborted", { session_id: clientSessionId, message: "请求已中断" });
void queueSessionUiStatePersist().catch((error) => {
logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist aborted chat stream state");
});
}
if (!projection.finished) {
finishUiMessage(activeRun.status === "error" ? "error" : "stop");
if (!accumulator.finished) {
publishRunEvent(activeRun, activeRun.status === "error" ? "error" : "done", { session_id: clientSessionId });
}
await queueSessionUiStatePersist().catch((error) => {
logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state");
+45 -36
View File
@@ -15,6 +15,8 @@ export type StreamSubscriber = {
};
export type ActiveRun = {
applyEvent?: (event: string, data: Record<string, unknown>) => void;
uiMessages?: () => unknown[];
inputs?: ResponseInputTracker;
clientSessionId: string;
controller: AbortController;
@@ -175,43 +177,50 @@ export const updateLastAssistantMessage = (
return messages;
};
export const updateLastAssistantPermission = (
messages: unknown[],
requestId: string,
updater: (permission: Record<string, unknown>) => Record<string, unknown>,
) =>
updateLastAssistantMessage(messages, (message) => {
const permissions = Array.isArray(message.permissions)
? message.permissions
: [];
return {
...message,
permissions: permissions.map((permission) =>
isObjectRecord(permission) && permission.requestId === requestId
? updater(permission)
: permission,
),
};
});
export const updateLastAssistantQuestion = (
messages: unknown[],
requestId: string,
updater: (question: Record<string, unknown>) => Record<string, unknown>,
) =>
updateLastAssistantMessage(messages, (message) => {
const questions = Array.isArray(message.questions)
? message.questions
: [];
return {
...message,
questions: questions.map((question) =>
isObjectRecord(question) && question.requestId === requestId
? updater(question)
: question,
),
};
/** Replies belong to the assistant that requested input, including earlier turns. */
export function applyInteractionResponse<T>(
messages: T[],
event: "permission_response" | "question_response",
data: Record<string, unknown>,
): T[] {
const requestId = data.request_id;
if (typeof requestId !== "string") return messages;
const permission = event === "permission_response";
const field = permission ? "permissions" : "questions";
const requestType = permission ? "data-permission_request" : "data-question_request";
let changed = false;
const next = messages.map((message) => {
if (!isObjectRecord(message) || message.role !== "assistant") return message;
const entries = Array.isArray(message[field]) ? message[field] as unknown[] : undefined;
const parts = Array.isArray(message.parts) ? [...message.parts] : undefined;
const ownsRequest = entries?.some((entry) => isObjectRecord(entry) && entry.requestId === requestId)
|| parts?.some((part) => isObjectRecord(part) && part.type === requestType
&& isObjectRecord(part.data) && part.data.request_id === requestId);
if (!ownsRequest) return message;
changed = true;
const updated = entries?.map((entry) => {
if (!isObjectRecord(entry) || entry.requestId !== requestId) return entry;
return permission ? {
...entry, status: data.status === "error" ? "error" : toPermissionStatus(data.reply as PermissionReply),
error: data.status === "error" ? data.error : undefined, repliedAt: Date.now(),
} : {
...entry, status: typeof data.status === "string" ? data.status : data.rejected === true ? "rejected" : "answered",
...(Array.isArray(data.answers) ? { answers: data.answers } : {}), error: undefined, repliedAt: Date.now(),
};
});
if (parts && data.status !== "pending") {
const type = `data-${event}`;
const index = parts.findIndex((part) => isObjectRecord(part) && part.type === type
&& isObjectRecord(part.data) && part.data.request_id === requestId);
const previous = index >= 0 && isObjectRecord(parts[index]) ? parts[index] : undefined;
const part = { ...previous, type, data: { ...(isObjectRecord(previous?.data) ? previous.data : {}), ...data } };
if (index >= 0) parts[index] = part;
else parts.push(part);
}
return { ...message, ...(updated ? { [field]: updated } : {}), ...(parts ? { parts } : {}) };
});
return changed ? next : messages;
}
const getToolArtifactKind = (tool: string): ToolArtifactKind => {
if (tool === "present_view") {
+16
View File
@@ -0,0 +1,16 @@
import type { Request, Response } from "express";
import { isTrustedInternalBridgeRequest } from "../product/internal-bridge-auth.js";
import { getRuntimeSessionContext } from "../runtime/sessionContext.js";
export function readTaskBridgeContext(req: Request, token: string, allowQuery = false) {
if (!isTrustedInternalBridgeRequest(req, token)) return null;
const runtimeSessionId = typeof req.body?.session_id === "string" ? req.body.session_id.trim()
: allowQuery && typeof req.query.session_id === "string" ? req.query.session_id.trim() : "";
const context = runtimeSessionId ? getRuntimeSessionContext(runtimeSessionId) : null;
return context?.taskId ? { context, runtimeSessionId, taskId: context.taskId } : null;
}
export function replyTaskBridgeError(res: Response, error: unknown, fallback: string) {
const status = error && typeof error === "object" && "status" in error && typeof error.status === "number" ? error.status : 502;
res.status(status).json({ message: error instanceof Error ? error.message : fallback });
}
+65
View File
@@ -0,0 +1,65 @@
import type { Express } from "express";
import { operationalTaskClient, expressAccessRequestSchema, proposedTaskActionSchema, humanTaskEvidenceCommandSchema, type OperationalTaskClient } from "../product/operational-task-client.js";
import { readTaskBridgeContext, replyTaskBridgeError } from "./task-bridge-context.js";
export function registerTaskCommandBridge(app: Express, options: {
internalToken: string;
client?: Pick<OperationalTaskClient, "requestDataAccess" | "proposeAction" | "addHumanEvidence">;
}) {
const client = options.client ?? operationalTaskClient;
app.post("/internal/operational-task/access-requests", async (req, res) => {
const bridge = readTaskBridgeContext(req, options.internalToken, true);
if (!bridge || bridge.context.purpose !== "task_execution") {
res.status(403).json({ message: "task execution context required" });
return;
}
const parsed = expressAccessRequestSchema.safeParse(req.body?.request);
if (!parsed.success) {
res.status(400).json({ message: "invalid access request" });
return;
}
try {
res.json(await client.requestDataAccess(bridge.runtimeSessionId, bridge.taskId, {
...parsed.data, sourceSessionId: bridge.context.clientSessionId,
}));
} catch (error) {
replyTaskBridgeError(res, error, "access request failed");
}
});
app.post("/internal/operational-task/actions", async (req, res) => {
const bridge = readTaskBridgeContext(req, options.internalToken, true);
if (!bridge) {
res.status(403).json({ message: "operational task context unavailable" });
return;
}
const parsed = proposedTaskActionSchema.safeParse(req.body?.action);
if (!parsed.success) {
res.status(400).json({ message: "invalid proposed action", detail: parsed.error.flatten() });
return;
}
try {
res.json(await client.proposeAction(bridge.runtimeSessionId, bridge.taskId, parsed.data));
} catch (error) {
replyTaskBridgeError(res, error, "task action proposal failed");
}
});
app.post("/internal/operational-task/human-evidence", async (req, res) => {
const bridge = readTaskBridgeContext(req, options.internalToken, true);
if (!bridge || bridge.context.purpose !== "task_collaboration") {
res.status(403).json({ message: "human evidence requires a task collaboration session" });
return;
}
const parsed = humanTaskEvidenceCommandSchema.safeParse(req.body?.command);
if (!parsed.success) {
res.status(400).json({ message: "invalid human evidence", detail: parsed.error.flatten() });
return;
}
try {
res.json(await client.addHumanEvidence(bridge.runtimeSessionId, bridge.taskId, parsed.data));
} catch (error) {
replyTaskBridgeError(res, error, "human evidence publish failed");
}
});
}
+7 -8
View File
@@ -4,7 +4,7 @@ import { z } from "zod";
import { isTrustedInternalBridgeRequest } from "../product/internal-bridge-auth.js";
import { operationalTaskClient, type OperationalTaskClient, taskArtifactBridgeSchema } from "../product/operational-task-client.js";
import { taskPresentationBridgeSchema } from "../product/task-presentation.js";
import { getRuntimeSessionContext } from "../runtime/sessionContext.js";
import { readTaskBridgeContext, replyTaskBridgeError } from "./task-bridge-context.js";
const artifactsInputSchema = taskArtifactBridgeSchema.and(z.object({ session_id: z.string().trim().min(1).max(128) }));
@@ -19,12 +19,12 @@ export function registerTaskPublicationBridge(app: Express, options: {
res.status(403).json({ message: "forbidden" });
return;
}
const sessionId = typeof req.body?.session_id === "string" ? req.body.session_id.trim() : "";
const context = sessionId ? getRuntimeSessionContext(sessionId) : null;
if (!context?.taskId) {
const bridge = readTaskBridgeContext(req, options.internalToken);
if (!bridge) {
res.status(403).json({ message: "operational task context unavailable" });
return;
}
const { context, runtimeSessionId: sessionId, taskId } = bridge;
if ((kind === "presentation" || req.body?.progress !== undefined) && context.purpose !== "task_execution") {
res.status(403).json({ message: "task execution context required" });
return;
@@ -36,7 +36,7 @@ export function registerTaskPublicationBridge(app: Express, options: {
res.status(400).json({ message: "invalid task presentation", detail: parsed.error.flatten() });
return;
}
res.json(await client.publishPresentation(sessionId, context.taskId, {
res.json(await client.publishPresentation(sessionId, taskId, {
requestId: parsed.data.request_id,
sourceSessionId: context.clientSessionId,
command: parsed.data.command,
@@ -47,7 +47,7 @@ export function registerTaskPublicationBridge(app: Express, options: {
res.status(400).json({ message: "invalid task artifacts", detail: parsed.error.flatten() });
return;
}
res.json(await client.publishArtifacts(sessionId, context.taskId, {
res.json(await client.publishArtifacts(sessionId, taskId, {
requestId: parsed.data.request_id,
sourceSessionId: context.clientSessionId,
evidence: parsed.data.evidence,
@@ -57,8 +57,7 @@ export function registerTaskPublicationBridge(app: Express, options: {
}));
}
} catch (error) {
const status = error && typeof error === "object" && "status" in error && typeof error.status === "number" ? error.status : 502;
res.status(status).json({ message: error instanceof Error ? error.message : `task ${kind} publish failed` });
replyTaskBridgeError(res, error, `task ${kind} publish failed`);
}
});
}
+6 -14
View File
@@ -2,7 +2,7 @@ import { LOCAL_QUERY_TOOL_IDS, PRODUCT_MCP_TOOL_IDS, PRODUCT_MCP_WRITE_TOOL_IDS,
import { assertWebSearchReady } from "./web-search-check.js";
import { allowedQueryTools } from "./query-mcp-policy.js";
import { setTimeout as delayAbortable } from "node:timers/promises";
import { isMessageNotFoundError, isSessionNotFoundError, type OpenCodeClient as OpencodeClient } from "@opencode/client";
import { isMessageNotFoundError, isSessionNotFoundError, type FormDetail, type OpenCodeClient as OpencodeClient } from "@opencode/client";
import { createRuntimeClient, startOpencode2 } from "./opencode2-server.js";
import { formAnswers, normalizeRuntimeMessage, Opencode2EventNormalizer } from "./opencode2-events.js";
import { existsSync, readFileSync } from "node:fs";
@@ -381,13 +381,6 @@ export class OpencodeRuntimeAdapter {
await rm(directory, { recursive: true, force: true });
}
async sendPrompt(sessionId: string, text: string) {
await this.prompt(sessionId, text);
// 当前 SDK 响应风格下,prompt() 本身不会直接返回完整 assistant parts,
// 所以这里紧跟一次 messages() 回读,给上层路由统一消费。
return this.messages(sessionId);
}
async generateText(prompt: string, model?: RuntimeModelOverride, signal?: AbortSignal) {
const client = await this.ensureClient();
const selected = model ?? parseModelReference(config.OPENCODE_MODEL);
@@ -733,19 +726,18 @@ export class OpencodeRuntimeAdapter {
return client.permission.reply({ sessionID: options.sessionId, requestID: options.requestId, decision: options.reply, message: options.message });
}
async replyQuestion(options: { requestId: string; sessionId?: string; answers: QuestionAnswers }) {
async replyQuestion(options: { requestId: string; sessionId: string; answers: QuestionAnswers; form: FormDetail }) {
if (!options.sessionId) throw new Error("OpenCode 2 form replies require sessionId");
const client = await this.ensureClient();
const form = await client.session.form.get({ sessionID: options.sessionId, formID: options.requestId });
return client.session.form.reply({ sessionID: options.sessionId, formID: options.requestId, answer: formAnswers(form, options.answers) });
return client.session.form.reply({ sessionID: options.sessionId, formID: options.requestId, answer: formAnswers(options.form, options.answers) });
}
async getQuestionStatus(options: { requestId: string; sessionId: string }): Promise<"pending" | "answered" | "cancelled" | "missing"> {
async getQuestion(options: { requestId: string; sessionId: string }): Promise<FormDetail | null> {
const client = await this.ensureClient();
try { return (await client.session.form.get({ sessionID: options.sessionId, formID: options.requestId })).state.status; }
try { return await client.session.form.get({ sessionID: options.sessionId, formID: options.requestId }); }
catch (error) {
const tag = error && typeof error === "object" && "_tag" in error ? error._tag : undefined;
if (tag === "FormNotFoundError") return "missing";
if (tag === "FormNotFoundError") return null;
throw error;
}
}
+4 -84
View File
@@ -37,12 +37,7 @@ import { createAgentMessageQuery, registerAgentMessageBridge } from "./routes/ag
import { FrontendActionCoordinator, FrontendActionError } from "./frontendAction/coordinator.js";
import { opencodeRuntime } from "./runtime/opencode.js";
import { getRuntimeSessionContext } from "./runtime/sessionContext.js";
import {
operationalTaskClient,
expressAccessRequestSchema,
humanTaskEvidenceCommandSchema,
proposedTaskActionSchema,
} from "./product/operational-task-client.js";
import { registerTaskCommandBridge } from "./routes/task-command-bridge.js";
import { isTrustedInternalBridgeRequest } from "./product/internal-bridge-auth.js";
const app = express();
@@ -53,9 +48,9 @@ const sessionWorkspaces = new SessionWorkspaceStore(config.CONVERSATION_WORKSPAC
const sessionMetadataStore = new SessionMetadataStore(undefined, sessionWorkspaces);
const answerViewStateStore = new AnswerViewStateStore(sessionWorkspaces);
const interactionRequestStore = new InteractionRequestStore(sessionWorkspaces);
const sessionUiStateStore = new SessionUiStateStore(undefined, sessionWorkspaces);
const sessionUiStateStore = new SessionUiStateStore(sessionWorkspaces);
const memoryStore = new MemoryStore();
const sessionTranscriptStore = new SessionTranscriptStore(undefined, sessionWorkspaces);
const sessionTranscriptStore = new SessionTranscriptStore(sessionWorkspaces);
const learningOrchestrator = new LearningOrchestrator(
opencodeRuntime,
memoryStore,
@@ -112,80 +107,7 @@ app.post("/internal/frontend-actions/request", async (req, res) => {
}
});
const readOperationalTaskBridgeContext = (req: express.Request) => {
if (!isTrustedInternalBridgeRequest(req, internalToken)) return null;
const runtimeSessionId = typeof req.body?.session_id === "string"
? req.body.session_id.trim()
: typeof req.query.session_id === "string"
? req.query.session_id.trim()
: "";
const context = runtimeSessionId ? getRuntimeSessionContext(runtimeSessionId) : null;
return context?.taskId ? { context, runtimeSessionId, taskId: context.taskId } : null;
};
app.post("/internal/operational-task/access-requests", async (req, res) => {
const bridge = readOperationalTaskBridgeContext(req);
if (!bridge || bridge.context.purpose !== "task_execution") {
res.status(403).json({ message: "task execution context required" });
return;
}
const parsed = expressAccessRequestSchema.safeParse(req.body?.request);
if (!parsed.success) {
res.status(400).json({ message: "invalid access request" });
return;
}
try {
res.json(await operationalTaskClient.requestDataAccess(bridge.runtimeSessionId, bridge.taskId, {
...parsed.data, sourceSessionId: bridge.context.clientSessionId,
}));
} catch (error) {
const status = error && typeof error === "object" && "status" in error && typeof error.status === "number"
? error.status : 502;
res.status(status).json({ message: error instanceof Error ? error.message : "access request failed" });
}
});
app.post("/internal/operational-task/actions", async (req, res) => {
const bridge = readOperationalTaskBridgeContext(req);
if (!bridge) {
res.status(403).json({ message: "operational task context unavailable" });
return;
}
const parsed = proposedTaskActionSchema.safeParse(req.body?.action);
if (!parsed.success) {
res.status(400).json({ message: "invalid proposed action", detail: parsed.error.flatten() });
return;
}
try {
res.json(await operationalTaskClient.proposeAction(bridge.runtimeSessionId, bridge.taskId, parsed.data));
} catch (error) {
const status = error && typeof error === "object" && "status" in error && typeof error.status === "number"
? error.status
: 502;
res.status(status).json({ message: error instanceof Error ? error.message : "task action proposal failed" });
}
});
app.post("/internal/operational-task/human-evidence", async (req, res) => {
const bridge = readOperationalTaskBridgeContext(req);
if (!bridge || bridge.context.purpose !== "task_collaboration") {
res.status(403).json({ message: "human evidence requires a task collaboration session" });
return;
}
const parsed = humanTaskEvidenceCommandSchema.safeParse(req.body?.command);
if (!parsed.success) {
res.status(400).json({ message: "invalid human evidence", detail: parsed.error.flatten() });
return;
}
try {
res.json(await operationalTaskClient.addHumanEvidence(bridge.runtimeSessionId, bridge.taskId, parsed.data));
} catch (error) {
const status = error && typeof error === "object" && "status" in error && typeof error.status === "number"
? error.status
: 502;
res.status(status).json({ message: error instanceof Error ? error.message : "human evidence publish failed" });
}
});
registerTaskCommandBridge(app, { internalToken });
app.get("/health", async (_req, res) => {
try {
@@ -247,10 +169,8 @@ const bootstrap = async () => {
logger.info(migrated, "session records migrated to owned workspaces; legacy sources archived");
}
await Promise.all([
sessionUiStateStore.initialize(),
learningOrchestrator.initialize(),
memoryStore.initialize(),
sessionTranscriptStore.initialize(),
]);
};
+8
View File
@@ -0,0 +1,8 @@
import type { SessionWorkspaceContext } from "../runtime/conversationWorkspace.js";
/** Always derive persistence scope from a host-validated session record. */
export const toSessionScope = (record: SessionWorkspaceContext): SessionWorkspaceContext => ({
sessionId: record.sessionId,
actorKey: record.actorKey,
projectKey: record.projectKey,
});
+8 -24
View File
@@ -1,10 +1,5 @@
import { join } from "node:path";
import { config } from "../config.js";
import {
atomicWriteJson,
ensureDirectory,
listJsonFiles,
readJsonFile,
toStableId,
} from "../utils/fileStore.js";
@@ -50,11 +45,7 @@ const DEFAULT_SEARCH_MAX_QUERY_CHARS = 240;
export class SessionTranscriptStore {
private readonly writeQueues = new Map<string, Promise<void>>();
constructor(private readonly baseDir = config.SESSION_TRANSCRIPT_STORAGE_DIR, private readonly workspaces?: SessionWorkspaceStore) {}
async initialize() {
if (!this.workspaces) await ensureDirectory(this.baseDir);
}
constructor(private readonly workspaces: SessionWorkspaceStore) {}
async appendTurn(
context: SessionTranscriptContext,
@@ -66,7 +57,7 @@ export class SessionTranscriptStore {
) {
const key = this.filePath(context);
return this.serializeWrite(key, async () => {
if (this.workspaces) await this.workspaces.ensure(this.workspaceContext(context));
await this.workspaces.ensure(this.workspaceContext(context));
// 同一会话的多次写入串行化,防止流式结束和 UI 同步同时写同一个 transcript。
const transcript = (await this.readTranscript(context)) ?? {
actorKey: context.actorKey,
@@ -148,10 +139,8 @@ export class SessionTranscriptStore {
turns: projectTurnsForFork(sourceTranscript?.turns ?? [], keepMessageCount),
updatedAt: timestamp,
};
if (this.workspaces) {
await this.workspaces.ensure(this.workspaceContext(targetContext));
await this.workspaces.validate(this.workspaceContext(targetContext), "transcript.json");
}
await this.workspaces.ensure(this.workspaceContext(targetContext));
await this.workspaces.validate(this.workspaceContext(targetContext), "transcript.json");
await this.persist(targetContext, nextTranscript);
return nextTranscript;
}
@@ -168,7 +157,7 @@ export class SessionTranscriptStore {
}
const queryTokens = normalizedQuery.split(/\s+/).filter(Boolean);
const hits: SessionSearchHit[] = [];
const files = this.workspaces ? await this.workspaces.list(context, "transcript.json") : await listJsonFiles(this.baseDir);
const files = await this.workspaces.list(context, "transcript.json");
for (const file of files) {
const transcript = await readJsonFile<SessionTranscriptRecord>(file);
if (!transcript) {
@@ -205,7 +194,7 @@ export class SessionTranscriptStore {
}
private async readTranscript(context: SessionTranscriptContext) {
if (this.workspaces && !await this.workspaces.validate(this.workspaceContext(context), "transcript.json")) return null;
if (!await this.workspaces.validate(this.workspaceContext(context), "transcript.json")) return null;
const record = await readJsonFile<SessionTranscriptRecord>(this.filePath(context));
if (record && (record.actorKey !== context.actorKey || record.projectKey !== context.projectKey ||
(record.clientSessionId || record.sessionId) !== (context.clientSessionId || context.sessionId))) throw new Error("chat transcript ownership mismatch");
@@ -213,16 +202,11 @@ export class SessionTranscriptStore {
}
private filePath(context: SessionTranscriptContext) {
if (this.workspaces) return this.workspaces.path(this.workspaceContext(context), "transcript.json");
return join(
this.baseDir,
`${context.actorKey}__${context.projectKey}__${context.sessionId}.json`,
);
return this.workspaces.path(this.workspaceContext(context), "transcript.json");
}
private async persist(context: SessionTranscriptContext, transcript: SessionTranscriptRecord) {
if (this.workspaces) await this.workspaces.writeRecord(this.workspaceContext(context), "transcript.json", transcript);
else await atomicWriteJson(this.filePath(context), transcript);
await this.workspaces.writeRecord(this.workspaceContext(context), "transcript.json", transcript);
}
private workspaceContext(context: SessionTranscriptContext) {
+8 -29
View File
@@ -1,13 +1,9 @@
import { join } from "node:path";
import type { SessionWorkspaceContext } from "../runtime/conversationWorkspace.js";
import { config } from "../config.js";
import { type SessionWorkspaceStore } from "./workspace-store.js";
import {
atomicWriteJson,
ensureDirectory,
readJsonFile,
removeFileIfExists,
slugify,
} from "../utils/fileStore.js";
export type SessionUiStateRecord = {
@@ -17,56 +13,39 @@ export type SessionUiStateRecord = {
executionPlans?: Record<string, unknown>[];
};
type SessionUiStateContext = {
sessionId: string;
actorKey?: string;
projectKey?: string;
};
type SessionUiStateContext = SessionWorkspaceContext;
export class SessionUiStateStore {
constructor(private readonly baseDir = config.SESSION_UI_STATE_STORAGE_DIR, private readonly workspaces?: SessionWorkspaceStore) {}
async initialize() {
if (!this.workspaces) await ensureDirectory(this.baseDir);
}
constructor(private readonly workspaces: SessionWorkspaceStore) {}
async read(context: SessionUiStateContext) {
if (this.workspaces && !await this.workspaces.validate(this.workspaceContext(context), "messages.json")) return null;
if (!await this.workspaces.validate(this.workspaceContext(context), "messages.json")) return null;
return await readJsonFile<SessionUiStateRecord>(this.filePath(context));
}
async write(context: SessionUiStateContext, state: SessionUiStateRecord) {
if (state.sessionId !== context.sessionId) throw new Error("chat state session mismatch");
if (this.workspaces) {
await this.workspaces.ensure(this.workspaceContext(context));
await this.workspaces.validate(this.workspaceContext(context), "messages.json");
}
if (this.workspaces) await this.workspaces.writeRecord(this.workspaceContext(context), "messages.json", state);
else await atomicWriteJson(this.filePath(context), state);
await this.workspaces.writeRecord(this.workspaceContext(context), "messages.json", state);
return state;
}
async update(context: SessionUiStateContext, change: (state: SessionUiStateRecord) => SessionUiStateRecord) {
if (this.workspaces) return this.workspaces.updateRecord(this.workspaceContext(context), "messages.json", async ({ current }) => {
return this.workspaces.updateRecord(this.workspaceContext(context), "messages.json", async ({ current }) => {
const state = current as SessionUiStateRecord | null;
if (!state || state.sessionId !== context.sessionId || !Array.isArray(state.messages)) throw new Error("chat state unavailable");
const next = change(state);
if (next.sessionId !== context.sessionId) throw new Error("chat state session mismatch");
return { result: next, value: next };
});
const state = await this.read(context);
if (!state) throw new Error("chat state unavailable");
return this.write(context, change(state));
}
async remove(context: SessionUiStateContext) {
if (this.workspaces && !await this.workspaces.validate(this.workspaceContext(context), "messages.json")) return;
if (!await this.workspaces.validate(this.workspaceContext(context), "messages.json")) return;
await removeFileIfExists(this.filePath(context));
}
private filePath(context: SessionUiStateContext) {
if (this.workspaces) return this.workspaces.path(this.workspaceContext(context), "messages.json");
return join(this.baseDir, `${slugify(context.sessionId)}.json`);
return this.workspaces.path(this.workspaceContext(context), "messages.json");
}
private workspaceContext(context: SessionUiStateContext) {
+2 -2
View File
@@ -99,11 +99,11 @@ describe("locally rendered answers", () => {
try {
const owner = { actorKey: "actor-answer", projectKey: "project-answer", sessionId: "answer-session" };
const workspaces = new SessionWorkspaceStore(join(root, "workspaces"));
const store = new SessionUiStateStore(join(root, "legacy"), workspaces);
const store = new SessionUiStateStore(workspaces);
const parts: UIMessage["parts"] = [{ type: "text", text: "演练:" }, { type: "data-answer_ui", id: "pressure", data: parseAnswerUiInput(input) }, { type: "text", text: "仍待核验。" }];
const messages = [{ id: "a", role: "assistant", parts }];
await store.write(owner, { sessionId: owner.sessionId, messages });
const restored = new SessionUiStateStore(join(root, "legacy"), new SessionWorkspaceStore(workspaces.root));
const restored = new SessionUiStateStore(new SessionWorkspaceStore(workspaces.root));
expect((await restored.read(owner))?.messages).toEqual(messages);
expect(await restored.read({ ...owner, actorKey: "actor-other" })).toBeNull();
expect(await restored.read({ ...owner, projectKey: "project-other" })).toBeNull();
+3 -2
View File
@@ -35,9 +35,10 @@ describe("unified view commands", () => {
});
it("passes aligned historical scatter points through native and publication contracts", () => {
const command = { action: "open_chart", chart_id: "pressure-comparison", evidence_refs: ["E1"], chart: {
title: "压力七天对比", chart_type: "line", y_axis_name: "压力(m)",
title: "压力七天对比", chart_type: "time_comparison", y_axis_name: "压力(m)",
x_data: ["19:30", "19:45", "20:00"], series: [
{ name: "当前", type: "line", data: [32, null, 31] },
{ name: "当前", type: "line", data: [32, null, 31],
sample_times: ["2026-06-14T19:30:00+08:00", null, "2026-06-14T20:00:00+08:00"] },
{ name: "前一天", type: "scatter", data: [null, 32, 33],
sample_times: [null, "2026-06-13T19:45:00+08:00", "2026-06-13T20:00:00+08:00"] },
],
+1
View File
@@ -68,6 +68,7 @@ try {
if (response.isError) throw new Error(`Extended tool failed through merged MCP: ${name}`);
}
if (!requests.includes("server/discover") || !requests.includes("tools/call") || requests.includes("initialize") || requests.includes("GET")) throw new Error("Transport used unexpected protocol requests");
if (requests.filter((method) => method === "tools/list").length !== 1) throw new Error("Business queries fetched the catalog again after startup");
if (grants.join(",") !== "client_credentials,urn:ietf:params:oauth:grant-type:token-exchange") throw new Error("OAuth grants were not validated");
console.log(`PASS: local MCP ${QUERY_MCP_TOOL_IDS.length}-tool discovery, real Server schemas, OAuth discovery, resource-bound grants, host binding, extended query and revocation over MCP HTTP`);
} finally {
+18
View File
@@ -26,6 +26,24 @@ const resourceToken = jwt("resource");
const token = () => Response.json({ access_token: resourceToken, token_type: "Bearer", expires_in: 300 });
function credential(id: string) { sessions.push(id); bindSessionCredential(id, { authorization: `Bearer ${jwt(id)}`, projectId: "project" }); }
it("uses cached tool definitions without fetching the catalog on each query", async () => {
const id = "cached-tool-definition"; credential(id);
const client = fixture((async (url, init) => {
if (String(url).includes("/token")) return token();
return init?.method === "POST"
? Response.json({ context_handle: "x".repeat(43) })
: new Response(null, { status: 204 });
}) as McpHttpFetch);
const definition = { ...catalog[0], outputSchema: { type: "object", properties: { ok: { type: "boolean" } } } };
Object.assign(client, { tools: [definition], withClient: async (_token: string, _project: string, _signal: AbortSignal, run: (client: unknown) => Promise<unknown>) => run({
callTool: async (_params: unknown, options: { toolDefinition: unknown }) => {
expect(options.toolDefinition).toBe(definition);
return { content: [], structuredContent: { ok: true } };
},
}) });
expect((await client.query(id, owner(id), definition.name, {}, "call")).structuredContent).toEqual({ ok: true });
});
it("owns simulation HTTP budgets and rejects model overrides before OAuth", async () => {
const id = "mcp-calculation-budget"; credential(id);
let requests = 0, transportTimeout = 0, sdkTimeout = 0;
+1 -1
View File
@@ -31,7 +31,7 @@ it("authenticates state reads and writes against the persisted session owner and
const context = getLocalAgentContext(ownerRequest);
const sessionId = "state-session";
await sessions.ensure({ ...context, sessionId });
await new SessionUiStateStore(join(root, "unused"), workspaces).write({ ...context, sessionId }, { sessionId, messages: [{ id: "assistant", role: "assistant", parts: [{ type: "data-answer_ui", data: { version: "answer-ui@1", block_id: "block", fallback_text: "fallback", spec: { root: "evidence", elements: { evidence: { type: "Evidence", props: { title: "Evidence", summary: "Summary", source: "source", evidence_id: null, observed_at: null, kind: "observed" }, children: [] } } } } }] }] });
await new SessionUiStateStore(workspaces).write({ ...context, sessionId }, { sessionId, messages: [{ id: "assistant", role: "assistant", parts: [{ type: "data-answer_ui", data: { version: "answer-ui@1", block_id: "block", fallback_text: "fallback", spec: { root: "evidence", elements: { evidence: { type: "Evidence", props: { title: "Evidence", summary: "Summary", source: "source", evidence_id: null, observed_at: null, kind: "observed" }, children: [] } } } } }] }] });
const app = express();
app.use(express.json({ limit: "2mb" }));
app.use(async (req, res, next) => {
+47
View File
@@ -0,0 +1,47 @@
import { expect, it } from "bun:test";
import type { UIMessageChunk } from "ai";
import { createChatRunAccumulator, publishRunEvent } from "../../src/routes/chat-run-accumulator.js";
import { createInitialStreamingMessages, type ActiveRun } from "../../src/routes/chatUiState.js";
it.each([false, true])("projects one event stream into reconnect and persisted messages (collaboration=%s)", (collaboration) => {
const run: ActiveRun = {
clientSessionId: "s", controller: new AbortController(), status: "running",
messages: createInitialStreamingMessages([], "查询"), pendingPermissions: new Map(), pendingQuestions: new Map(), subscribers: new Set(),
};
const chunks: UIMessageChunk[] = [];
const stream = createChatRunAccumulator({ run, messages: [{ id: "user", role: "user", parts: [{ type: "text", text: "查询" }] }],
messageId: "answer", sessionId: "s", collaboration, write: (chunk) => chunks.push(chunk), onStatus() {} });
stream.apply("progress", { id: "query", status: "running", title: "查询资料" });
stream.apply("token", { content: "已有数据" });
stream.apply("permission_request", { request_id: "permission", session_id: "s", permission: "shell", patterns: [], always: [], created_at: 1 });
stream.apply("permission_response", { request_id: "permission", reply: "once" });
stream.appendInput({ id: "extra", role: "user", parts: [{ type: "text", text: "补充" }] }, "补充");
stream.appendInput({ id: "extra", role: "user", parts: [] }, "重复");
stream.apply("final_answer", { content: "结论" });
stream.apply("done", {});
expect(run.status).toBe("completed");
expect(run.pendingPermissions.size).toBe(0);
expect(run.messages.at(-1)).toMatchObject({ id: "answer", content: "结论", permissions: [{ status: "approved_once" }], progress: [{ status: "completed" }, { status: "completed" }] });
const saved = stream.uiMessages();
expect(saved.map((message) => message.id)).toEqual(["user", "extra", "answer"]);
expect(saved.at(-1)?.parts).toEqual((run.messages.at(-1) as { parts: typeof saved[number]["parts"] }).parts);
expect(saved.at(-1)?.parts.some((part) => part.type === "data-progress")).toBe(!collaboration);
expect(chunks.filter((chunk) => chunk.type === "finish")).toHaveLength(1);
});
it.each(["done", "error", "aborted"])("publishes the same %s terminal state to reconnect subscribers and storage", (event) => {
const run: ActiveRun = {
clientSessionId: "s", controller: new AbortController(), status: "running",
messages: createInitialStreamingMessages([], "查询"), pendingPermissions: new Map(), pendingQuestions: new Map(), subscribers: new Set(),
};
const received: Array<{ event: string; data: Record<string, unknown> }> = [];
run.subscribers.add({ write: (event, data) => received.push({ event, data }), close() {} });
const stream = createChatRunAccumulator({ run, messages: [], messageId: "answer", sessionId: "s",
collaboration: false, write() {}, onStatus() {} });
publishRunEvent(run, "progress", { id: "tool", title: "正在处理", status: "running", phase: "tool" });
publishRunEvent(run, "token", { content: "部分结果" });
publishRunEvent(run, event, { session_id: "s", message: "执行已停止" });
const canonical = stream.uiMessages().at(-1)?.parts.find((part) => part.type === "data-run_status");
expect(received.at(-2)).toMatchObject({ event: "run_status", data: { status: event === "done" ? "completed" : event } });
expect(canonical).toMatchObject({ data: { status: received.at(-2)?.data.status } });
});
+1 -1
View File
@@ -22,7 +22,7 @@ it("persists owned topics, searches old text across themes and preserves chats w
globalThis.fetch = (async (_input, init) => Response.json({ user_id: new Headers(init?.headers).get("authorization") === "Bearer owner" ? "owner" : "other", project_id: new Headers(init?.headers).get("x-project-id"), network: "supply" })) as typeof fetch;
const workspaces = new SessionWorkspaceStore(join(root, "workspaces"));
const sessions = new SessionMetadataStore(join(root, "metadata"), workspaces);
const states = new SessionUiStateStore(join(root, "ui"), workspaces);
const states = new SessionUiStateStore(workspaces);
for (const id of ["a", "b", "unclassified"]) {
await sessions.ensure({ ...context, sessionId: id });
await states.write({ ...context, sessionId: id }, { sessionId: id, messages: [
+9 -5
View File
@@ -716,8 +716,9 @@ describe("streamPromptResponse", () => {
});
describe("background sandbox Shell approval", () => {
it.each(["permission.asked", "permission.v2.asked"])("only auto-approves sandbox Shell for %s", async (type) => {
describe("background sandbox permissions", () => {
it.each([["permission.asked", true], ["permission.v2.asked", true],
["permission.asked", false], ["permission.v2.asked", false]] as const)("limits session Shell/edit approval for %s background=%s", async (type, background) => {
const replies: Array<Record<string, unknown>> = [];
const runtime = {
subscribeEvents: async () => createEventStream([
@@ -739,11 +740,14 @@ describe("background sandbox Shell approval", () => {
const events: Array<{ event: string; data: Record<string, unknown> }> = [];
await streamPromptResponse({
runtime, sessionId: "runtime-session-1", clientSessionId: "client-session-1",
message: "后台调查", approvalMode: "request", backgroundSandboxShell: true,
message: "后台调查", approvalMode: "request", backgroundSandboxPermissions: background,
write: (event, data) => events.push({ event, data }),
});
expect(replies).toEqual([{ requestId: "shell", sessionId: "runtime-session-1", reply: "once" }]);
expect(replies).toEqual(background ? [
{ requestId: "shell", sessionId: "runtime-session-1", reply: "once" },
{ requestId: "edit", sessionId: "runtime-session-1", reply: "once" },
] : []);
expect(events.filter(({ event }) => event === "permission_request")
.map(({ data }) => data.permission)).toEqual(["edit", "mcp_fixture_write"]);
.map(({ data }) => data.permission)).toEqual(background ? ["mcp_fixture_write"] : ["shell", "edit", "mcp_fixture_write"]);
});
});
+2 -1
View File
@@ -7,6 +7,7 @@ import { attachTrustedAgentContext } from "../../src/context/localContext.js";
import { registerChatInteractionRoutes } from "../../src/routes/chatInteractionRoutes.js";
import { bindSessionCredential, credentialRefreshCoordinator, releaseSessionCredential } from "../../src/product/credential-refresh.js";
import { SessionMetadataStore } from "../../src/sessions/metadataStore.js";
import { SessionWorkspaceStore } from "../../src/sessions/workspace-store.js";
import { SessionUiStateStore } from "../../src/sessions/uiStateStore.js";
import { toActorKey, toProjectKey } from "../../src/utils/fileStore.js";
import type { ActiveRun } from "../../src/routes/chatUiState.js";
@@ -28,7 +29,7 @@ describe("credential refresh HTTP route", () => {
const router = express.Router();
router.use((req, res, next) => { void attachTrustedAgentContext(req).then(() => next()).catch(() => res.sendStatus(401)); });
const activeRuns = new Map<string, ActiveRun>([["credential-route-session", { clientSessionId: "credential-route-session", controller: new AbortController(), messages: [], pendingPermissions: new Map(), pendingQuestions: new Map(), subscribers: new Set(), status: "running" }]]);
registerChatInteractionRoutes(router, { activeRuns, runtime: {} as OpencodeRuntimeAdapter, sessionMetadataStore: metadata, sessionUiStateStore: new SessionUiStateStore(join(directory, "ui")) });
registerChatInteractionRoutes(router, { activeRuns, runtime: {} as OpencodeRuntimeAdapter, sessionMetadataStore: metadata, sessionUiStateStore: new SessionUiStateStore(new SessionWorkspaceStore(join(directory, "workspaces"))) });
app.use(router);
const server = app.listen(0, "127.0.0.1");
await new Promise<void>((resolve) => server.once("listening", resolve));
+1 -1
View File
@@ -115,7 +115,7 @@ it("uses authenticated /responses for saved answers and rejects bypass and forei
globalThis.fetch = (async (_input, init) => Response.json({ user_id: new Headers(init?.headers).get("Authorization") === "Bearer owner" ? "owner" : "other", project_id: project, network: "fixture" })) as typeof fetch;
const workspaces = new SessionWorkspaceStore(join(root, "workspaces"));
const sessions = new SessionMetadataStore(join(root, "metadata"), workspaces);
const states = new SessionUiStateStore(join(root, "unused"), workspaces);
const states = new SessionUiStateStore(workspaces);
const interactions = new InteractionRequestStore(workspaces);
const context = { actorKey: toActorKey("product:owner"), projectKey: toProjectKey(`product:${project}`), projectId: project, userId: "owner", sessionId: "owned" };
await sessions.ensure(context);
+30 -3
View File
@@ -5,16 +5,43 @@ import { tmpdir } from "node:os";
import { join } from "node:path";
import { getLocalAgentContext } from "../../src/context/localContext.js";
import { registerChatInteractionRoutes } from "../../src/routes/chatInteractionRoutes.js";
import type { OpencodeRuntimeAdapter } from "../../src/runtime/opencode.js";
import { OpencodeRuntimeAdapter } from "../../src/runtime/opencode.js";
import { SessionMetadataStore } from "../../src/sessions/metadataStore.js";
import { SessionUiStateStore } from "../../src/sessions/uiStateStore.js";
import { SessionWorkspaceStore } from "../../src/sessions/workspace-store.js";
it("reads a native form once when checking and answering a persisted question", async () => {
const root = await mkdtemp(join(tmpdir(), "question-single-read-"));
const workspaces = new SessionWorkspaceStore(join(root, "workspaces"));
const sessions = new SessionMetadataStore(join(root, "metadata"), workspaces);
const states = new SessionUiStateStore(workspaces);
const context = { ...getLocalAgentContext(), sessionId: "session" };
await sessions.ensure(context);
await states.write(context, { sessionId: context.sessionId, messages: [{ role: "assistant", questions: [{ requestId: "form", status: "pending" }] }] });
const calls: string[] = [];
const runtime = Object.assign(Object.create(OpencodeRuntimeAdapter.prototype), {
ensureClient: async () => ({ session: { form: {
get: async () => { calls.push("get"); return { fields: [{ key: "answer", type: "text" }], state: { status: "pending" } }; },
reply: async (input: { answer: unknown }) => { calls.push("reply"); expect(input.answer).toEqual({ answer: "A" }); },
} } }),
});
const routes = new Map<string, (req: unknown, res: unknown) => Promise<void>>();
const router = { get() {}, post(path: string, handler: (req: unknown, res: unknown) => Promise<void>) { routes.set(path, handler); } };
registerChatInteractionRoutes(router as unknown as express.Router, { activeRuns: new Map(), runtime, sessionMetadataStore: sessions, sessionUiStateStore: states });
const req = { params: { request_id: "form" }, body: { session_id: "session", answers: [["A"]] }, header() { return undefined; } };
const response = { code: 200, setHeader() {}, status(code: number) { this.code = code; return this; }, json() { return this; } };
try {
await routes.get("/question/:request_id/reply")!(req, response);
expect(response.code).toBe(202);
expect(calls).toEqual(["get", "reply"]);
} finally { await rm(root, { recursive: true, force: true }); }
});
it("reconciles persisted legacy forms and replies after the active run was lost", async () => {
const root = await mkdtemp(join(tmpdir(), "legacy-question-routes-"));
const workspaces = new SessionWorkspaceStore(join(root, "workspaces"));
const sessions = new SessionMetadataStore(join(root, "metadata"), workspaces);
const states = new SessionUiStateStore(join(root, "unused"), workspaces);
const states = new SessionUiStateStore(workspaces);
const context = { ...getLocalAgentContext(), sessionId: "session" };
await sessions.ensure(context);
await states.write(context, { sessionId: "session", messages: [{ id: "old", role: "assistant", questions: [{ requestId: "form", status: "pending", tool: { callID: "call" } }] }, { id: "new", role: "assistant", text: "later message", parts: [{ type: "data-question_request", data: { request_id: "form2", session_id: "session", questions: [], tool: { callID: "call2" } } }, { type: "data-question_request", data: { request_id: "form3", session_id: "session", questions: [] } }, { type: "data-question_request", data: { request_id: "form4", session_id: "session", questions: [] } }] }] });
@@ -22,7 +49,7 @@ it("reconciles persisted legacy forms and replies after the active run was lost"
let unavailable = false;
let reads = 0;
let replies = 0;
const runtime = { getQuestionStatus: async () => { reads++; if (unavailable) throw new Error("private-native-path"); return status; }, replyQuestion: async () => { replies++; status = "answered"; }, rejectQuestion: async () => { status = "cancelled"; } } as unknown as OpencodeRuntimeAdapter;
const runtime = { getQuestion: async () => { reads++; if (unavailable) throw new Error("private-native-path"); return status === "missing" ? null : { state: { status } }; }, replyQuestion: async () => { replies++; status = "answered"; }, rejectQuestion: async () => { status = "cancelled"; } } as unknown as OpencodeRuntimeAdapter;
const app = express(); app.use(express.json());
registerChatInteractionRoutes(app, { activeRuns: new Map(), runtime, sessionMetadataStore: sessions, sessionUiStateStore: states });
const server = app.listen(0, "127.0.0.1");
@@ -0,0 +1,58 @@
import { createChatRunAccumulator } from "../../src/routes/chat-run-accumulator.js";
import { expect, it } from "bun:test";
import express from "express";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { getLocalAgentContext } from "../../src/context/localContext.js";
import { registerChatInteractionRoutes } from "../../src/routes/chatInteractionRoutes.js";
import type { ActiveRun } from "../../src/routes/chatUiState.js";
import type { OpencodeRuntimeAdapter } from "../../src/runtime/opencode.js";
import { SessionMetadataStore } from "../../src/sessions/metadataStore.js";
import { SessionUiStateStore } from "../../src/sessions/uiStateStore.js";
import { SessionWorkspaceStore } from "../../src/sessions/workspace-store.js";
it.each([false, true])("persists permission replies in the owner's workspace (native failure=%s)", async (nativeFailure) => {
const root = await mkdtemp(join(tmpdir(), "permission-persistence-"));
const workspaces = new SessionWorkspaceStore(join(root, "workspaces"));
const sessions = new SessionMetadataStore(join(root, "metadata"), workspaces);
const states = new SessionUiStateStore(workspaces);
const context = { ...getLocalAgentContext(), sessionId: "permission-session" };
await sessions.ensure(context);
const messages = [{ id: "answer", role: "assistant", permissions: [] }];
await states.write(context, { sessionId: context.sessionId, isTitleManuallyEdited: true, messages });
let calls = 0;
const runtime = { replyPermission: async () => { calls++; if (nativeFailure) throw new Error("native unavailable"); } } as unknown as OpencodeRuntimeAdapter;
const run: ActiveRun = {
clientSessionId: context.sessionId, controller: new AbortController(), messages,
pendingPermissions: new Map([["permission", { request_id: "permission", session_id: context.sessionId, permission: "shell", patterns: [], always: [], created_at: Date.now() }]]),
pendingQuestions: new Map(), subscribers: new Set(), status: "running",
};
const accumulator = createChatRunAccumulator({ run, messages: [], messageId: "answer", sessionId: context.sessionId,
collaboration: false, write() {}, onStatus() {} });
accumulator.apply("permission_request", run.pendingPermissions.get("permission")! as unknown as Record<string, unknown>);
await states.write(context, { sessionId: context.sessionId, isTitleManuallyEdited: true, messages: run.messages });
const app = express(); app.use(express.json());
registerChatInteractionRoutes(app, { activeRuns: new Map([[context.sessionId, run]]), runtime, sessionMetadataStore: sessions, sessionUiStateStore: states });
const server = app.listen(0, "127.0.0.1");
await new Promise<void>((resolve) => server.once("listening", resolve));
const url = `http://127.0.0.1:${(server.address() as { port: number }).port}/permission/permission/reply`;
const reply = (headers = {}) => fetch(url, { method: "POST", headers: { "Content-Type": "application/json", ...headers }, body: JSON.stringify({ session_id: context.sessionId, reply: "once" }) });
try {
expect((await reply({ "X-Agent-Local-User": "other" })).status).toBe(404);
expect(calls).toBe(0);
expect((await reply()).status).toBe(nativeFailure ? 502 : 202);
expect(calls).toBe(1);
expect(run.pendingPermissions.has("permission")).toBe(nativeFailure);
const saved = await states.read(context);
expect(saved?.isTitleManuallyEdited).toBe(true);
const parts = (saved?.messages[0] as { parts: Array<{ type: string; data: Record<string, unknown> }> }).parts;
expect(parts.find((part) => part.type === "data-permission_response")?.data).toMatchObject(nativeFailure
? { request_id: "permission", status: "error", error: "native unavailable" }
: { request_id: "permission", reply: "once" });
expect(run.messages[0]).toMatchObject({ permissions: [{ requestId: "permission", status: nativeFailure ? "error" : "approved_once" }] });
} finally {
await new Promise<void>((resolve) => server.close(() => resolve()));
await rm(root, { recursive: true, force: true });
}
});
+71
View File
@@ -0,0 +1,71 @@
import { expect, it } from "bun:test";
import express from "express";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import type { UIMessage } from "ai";
import { getLocalAgentContext } from "../../src/context/localContext.js";
import { createChatRunAccumulator } from "../../src/routes/chat-run-accumulator.js";
import { registerChatInteractionRoutes } from "../../src/routes/chatInteractionRoutes.js";
import { createInitialStreamingMessages, type ActiveRun } from "../../src/routes/chatUiState.js";
import type { OpencodeRuntimeAdapter } from "../../src/runtime/opencode.js";
import { SessionMetadataStore } from "../../src/sessions/metadataStore.js";
import { SessionUiStateStore } from "../../src/sessions/uiStateStore.js";
import { SessionWorkspaceStore } from "../../src/sessions/workspace-store.js";
it("keeps historical question invalidation across the next live snapshot", async () => {
const root = await mkdtemp(join(tmpdir(), "question-persistence-"));
const workspaces = new SessionWorkspaceStore(join(root, "workspaces"));
const sessions = new SessionMetadataStore(join(root, "metadata"), workspaces);
const states = new SessionUiStateStore(workspaces);
const context = { ...getLocalAgentContext(), sessionId: "history-session" };
await sessions.ensure(context);
const history: UIMessage[] = [{ id: "old-answer", role: "assistant", parts: [
{ type: "data-question_request", data: { request_id: "old-question", session_id: context.sessionId,
questions: [{ question: "继续吗?", options: [] }] } },
{ type: "data-final_answer", data: { content: "此前的结果" } },
] }, { id: "user", role: "user", parts: [{ type: "text", text: "新一轮" }] }];
const run: ActiveRun = {
clientSessionId: context.sessionId, controller: new AbortController(), status: "running",
messages: createInitialStreamingMessages(history, "新一轮"),
pendingPermissions: new Map(), pendingQuestions: new Map(), subscribers: new Set(),
};
const stream = createChatRunAccumulator({ run, messages: history, messageId: "current-answer",
sessionId: context.sessionId, collaboration: false, write() {}, onStatus() {} });
stream.apply("token", { content: "正在核查" });
await states.write(context, { sessionId: context.sessionId, isTitleManuallyEdited: true, messages: stream.uiMessages() });
let nativeReads = 0;
const runtime = { getQuestion: async () => { nativeReads++; return null; } } as unknown as OpencodeRuntimeAdapter;
const app = express();
registerChatInteractionRoutes(app, { activeRuns: new Map([[context.sessionId, run]]), runtime,
sessionMetadataStore: sessions, sessionUiStateStore: states });
const server = app.listen(0, "127.0.0.1");
await new Promise<void>((resolve) => server.once("listening", resolve));
const url = `http://127.0.0.1:${(server.address() as { port: number }).port}/session/${context.sessionId}/questions/old-question`;
const resolved = (messages: unknown[]) => {
const message = messages[0] as UIMessage;
expect(message.id).toBe("old-answer");
const part = message.parts.find((part) => part.type === "data-question_response");
expect(part).toMatchObject({ type: "data-question_response", data: { request_id: "old-question", status: "invalidated" } });
};
try {
expect((await fetch(url, { headers: { "X-Agent-Local-User": "other" } })).status).toBe(404);
expect(nativeReads).toBe(0);
const response = await fetch(url);
expect(response.status).toBe(200);
expect(await response.json()).toMatchObject({ status: "invalidated" });
resolved((await states.read(context))!.messages);
resolved(run.messages);
resolved(stream.uiMessages());
expect(stream.uiMessages().at(-1)?.parts.some((part) => part.type === "data-question_response")).toBe(false);
stream.apply("token", { content: ",继续查询" });
await states.write(context, { ...(await states.read(context))!, messages: stream.uiMessages() });
resolved((await states.read(context))!.messages);
expect((await states.read(context))?.isTitleManuallyEdited).toBe(true);
await fetch(url);
expect(nativeReads).toBe(1);
} finally {
await new Promise<void>((resolve) => server.close(() => resolve()));
await rm(root, { recursive: true, force: true });
}
});
+31 -1
View File
@@ -44,7 +44,7 @@ beforeEach(async () => {
})) as typeof fetch;
const workspaces = new SessionWorkspaceStore(join(root, "workspaces"));
const sessions = new SessionMetadataStore(join(root, "metadata"), workspaces);
states = new SessionUiStateStore(join(root, "ui"), workspaces);
states = new SessionUiStateStore(workspaces);
interactions = new InteractionRequestStore(workspaces);
await sessions.initialize();
const context = { actorKey: toActorKey("product:owner"), projectKey: toProjectKey(`product:${project}`), projectId: project, userId: "owner", sessionId: "owned" };
@@ -296,6 +296,11 @@ it("marks an explicitly aborted response incomplete and preserves scoped history
const text = await response.text();
expect(text).toContain("event: response.incomplete");
expect(text).not.toContain("event: response.completed");
const context = { actorKey: toActorKey("product:owner"), projectKey: toProjectKey(`product:${project}`), sessionId: "owned" };
const saved = (await states.read(context))!.messages as Array<{ parts: Array<{ type: string; data?: { status?: string } }> }>;
expect(saved.at(-1)?.parts.find((part) => part.type === "data-run_status")?.data?.status).toBe("aborted");
const loaded = await http(`${base}/api/v1/session/owned`, { headers: { Authorization: "Bearer owner", "X-Project-Id": project } });
expect((await loaded.json()).messages).toEqual(saved);
});
it("continues a paged conversation from the complete saved history and preserves client message IDs", async () => {
@@ -316,3 +321,28 @@ it("continues a paged conversation from the complete saved history and preserves
expect((await call({ input: "重复提交", conversation: "owned", tjwater: { client_message_id: "client-turn-a" } })).status).toBe(409);
expect(prompted).toHaveLength(1);
});
it("uses the same canonical live messages for session reads and older pages", async () => {
gate = new Promise<void>((resolve) => { release = resolve; });
const context = { actorKey: toActorKey("product:owner"), projectKey: toProjectKey(`product:${project}`), sessionId: "owned" };
const history = Array.from({ length: 5 }, (_, index) => ({ id: `old-${index}`, role: index % 2 ? "assistant" : "user",
parts: [{ type: "text", text: `历史 ${index}` }] as unknown[] }));
history[1].parts.push({ type: "data-question_request", data: { request_id: "old-question", session_id: "owned", questions: [] } });
await states.write(context, { sessionId: "owned", messages: history });
const response = await call({ input: "新一轮", conversation: "owned", stream: true, tjwater: { client_message_id: "live-user" } });
const headers = { Authorization: "Bearer owner", "X-Project-Id": project };
const live = await http(`${base}/api/v1/session/owned`, { headers });
const snapshot = await live.json();
expect(snapshot.is_streaming).toBe(true);
expect(snapshot.messages).toHaveLength(7);
expect(snapshot.messages.slice(0, 5)).toEqual(history);
expect(snapshot.messages[5].id).toBe("live-user");
const recent = await http(`${base}/api/v1/session/owned?message_limit=2`, { headers });
const page = await recent.json();
expect(page.messages).toEqual(snapshot.messages.slice(-2));
const older = await http(`${base}/api/v1/session/owned/messages?limit=40&before=${encodeURIComponent(page.message_page.before_cursor)}`, { headers });
expect(older.status).toBe(200);
expect((await older.json()).messages).toEqual(history);
release();
await response.text();
});
+39
View File
@@ -0,0 +1,39 @@
import { expect, it } from "bun:test";
import { createHash } from "node:crypto";
import { copyFile, mkdir, mkdtemp, readFile, readdir, rm, symlink, writeFile } from "node:fs/promises";
import { join } from "node:path";
import { tmpdir } from "node:os";
import { astToString, COMMENT_HEADER, default as openapiTS } from "openapi-typescript";
it.each(["valid", "invalid-schema", "invalid-digest"])("synchronizes the complete artifact group or preserves it (%s)", async (scenario) => {
const root = await mkdtemp(join(tmpdir(), "server-contract-sync-"));
try {
for (const directory of ["scripts", "openapi", "src/generated", "upstream"]) await mkdir(join(root, directory), { recursive: true });
await copyFile(new URL("../../scripts/server-contract.ts", import.meta.url), join(root, "scripts/server-contract.ts"));
await symlink(new URL("../../node_modules", import.meta.url).pathname, join(root, "node_modules"), "dir");
const targets = ["openapi/openapi.json", "openapi/server-manifest.json", "src/generated/server-api.d.ts"];
for (const target of targets) await writeFile(join(root, target), `original:${target}`);
const schema = { openapi: scenario === "invalid-schema" ? "9.9.9" : "3.1.0", info: { title: "Fixture", version: "1" }, paths: {} };
const raw = JSON.stringify(schema);
const manifest = { contracts: { server: { sha256: scenario === "invalid-digest" ? "incorrect" : createHash("sha256").update(raw).digest("hex") } } };
await writeFile(join(root, "upstream/schema.json"), raw);
await writeFile(join(root, "upstream/manifest.json"), JSON.stringify(manifest));
const child = Bun.spawn([process.execPath, "scripts/server-contract.ts", "--sync", "--upstream", "upstream/schema.json"], {
cwd: root, stdout: "pipe", stderr: "pipe",
});
const [code, stderr] = await Promise.all([child.exited, new Response(child.stderr).text()]);
if (scenario === "valid") {
expect(code).toBe(0);
expect(await readFile(join(root, targets[0]), "utf8")).toBe(raw);
expect(JSON.parse(await readFile(join(root, targets[1]), "utf8"))).toEqual(manifest);
expect(await readFile(join(root, targets[2]), "utf8")).toBe(COMMENT_HEADER + astToString(await openapiTS(schema)));
} else {
expect(code).not.toBe(0);
expect(stderr).toContain(scenario === "invalid-schema" ? "Unsupported OpenAPI version" : "contract digest differs");
for (const target of targets) expect(await readFile(join(root, target), "utf8")).toBe(`original:${target}`);
}
expect((await readdir(join(root, "openapi"))).sort()).toEqual(["openapi.json", "server-manifest.json"]);
} finally {
await rm(root, { recursive: true, force: true });
}
});
+1 -1
View File
@@ -47,7 +47,7 @@ describe("chat session HTTP isolation", () => {
const workspaces = new SessionWorkspaceStore(join(metadataDir, "workspaces"));
const sessionMetadataStore = new SessionMetadataStore(metadataDir, workspaces);
const sessionUiStateStore = new SessionUiStateStore(join(metadataDir, "old-ui"), workspaces);
const sessionUiStateStore = new SessionUiStateStore(workspaces);
await sessionMetadataStore.initialize();
const ownerContext = {
actorKey: toActorKey("product:user-a"),
@@ -33,7 +33,7 @@ async function fixture() {
const context = { actorKey: toActorKey("product:owner"), projectKey: toProjectKey(`product:${projectId}`), projectId, userId: "owner" };
const workspaces = new SessionWorkspaceStore(join(root, "workspaces"));
const sessions = new SessionMetadataStore(join(root, "metadata"), workspaces);
const states = new SessionUiStateStore(join(root, "states"), workspaces);
const states = new SessionUiStateStore(workspaces);
const taskNames = new Map([[firstTask, "南桥管道爆管调查"], [secondTask, "城南供水压力复核"]]);
const taskReads: string[] = [];
let available = true;
+45
View File
@@ -0,0 +1,45 @@
import { expect, it } from "bun:test";
import express from "express";
import type { OperationalTaskClient } from "../../src/product/operational-task-client.js";
import { registerTaskCommandBridge } from "../../src/routes/task-command-bridge.js";
import { removeRuntimeSessionContext, setRuntimeSessionContext } from "../../src/runtime/sessionContext.js";
it("preserves command-specific session purposes and trusted bridge authentication", async () => {
const calls: unknown[][] = [];
const client = {
requestDataAccess: async (...args: unknown[]) => { calls.push(args); return {}; },
proposeAction: async (...args: unknown[]) => { calls.push(args); throw Object.assign(new Error("conflict"), { status: 409 }); },
addHumanEvidence: async (...args: unknown[]) => { calls.push(args); return {}; },
} as unknown as Pick<OperationalTaskClient, "requestDataAccess" | "proposeAction" | "addHumanEvidence">;
for (const purpose of ["task_execution", "task_collaboration"] as const) setRuntimeSessionContext({
sessionId: purpose, clientSessionId: `client-${purpose}`, taskId: "task", purpose,
actorKey: "actor", projectKey: "project", traceId: "test",
});
const app = express(); app.use(express.json());
registerTaskCommandBridge(app, { internalToken: "test-token", client });
const server = app.listen(0, "127.0.0.1");
await new Promise<void>((resolve) => server.once("listening", resolve));
const post = (kind: string, session: string, data: object, headers = {}) => fetch(`http://127.0.0.1:${(server.address() as { port: number }).port}/internal/operational-task/${kind}`, {
method: "POST", headers: { "Content-Type": "application/json", "x-agent-internal-token": "test-token", ...headers }, body: JSON.stringify({ session_id: session, ...data }),
});
const requestId = "60000000-0000-4000-8000-000000000001";
const access = { request: { requestId, nodeId: "read", scopes: ["scada_devices"], reason: "查询" } };
const evidence = { command: { request_id: requestId, evidence: { evidence_id: "e1", kind: "human_input", summary: "已复核", source_label: "用户", observed_at: "2026-11-09T00:00:00Z" } } };
try {
expect((await post("access-requests", "task_execution", access, { Origin: "http://browser.test" })).status).toBe(403);
expect((await post("access-requests", "task_execution", access, { "x-agent-internal-token": "wrong" })).status).toBe(403);
expect((await post("access-requests", "task_collaboration", access)).status).toBe(403);
expect((await post("human-evidence", "task_execution", evidence)).status).toBe(403);
expect(calls).toHaveLength(0);
expect((await post("access-requests", "task_execution", access)).status).toBe(200);
expect(calls[0]).toEqual(["task_execution", "task", { ...access.request, sourceSessionId: "client-task_execution" }]);
expect((await post("human-evidence", "task_collaboration", evidence)).status).toBe(200);
const action = { action: { request_id: requestId, reason: "核验", kind: "create_work_order", payload: { title: "采样", description: "核验", work_order_type: "general" } } };
expect((await post("actions", "task_execution", action)).status).toBe(409);
expect((await post("actions", "task_collaboration", action)).status).toBe(409);
expect((await post("actions", "unknown", action)).status).toBe(403);
} finally {
removeRuntimeSessionContext("task_execution"); removeRuntimeSessionContext("task_collaboration");
await new Promise<void>((resolve) => server.close(() => resolve()));
}
});
+3 -3
View File
@@ -33,11 +33,11 @@ test("native question lookup distinguishes missing form from unavailable service
const adapter = Object.assign(Object.create(OpencodeRuntimeAdapter.prototype), {
ensureClient: async () => ({ session: { form: { get: async () => { if (failure) throw failure; return { state: { status: "answered" } }; } } } }),
}) as OpencodeRuntimeAdapter;
expect(await adapter.getQuestionStatus({ sessionId: "s", requestId: "form" })).toBe("missing");
expect(await adapter.getQuestion({ sessionId: "s", requestId: "form" })).toBeNull();
failure = new Error("network unavailable");
await expect(adapter.getQuestionStatus({ sessionId: "s", requestId: "form" })).rejects.toThrow("network unavailable");
await expect(adapter.getQuestion({ sessionId: "s", requestId: "form" })).rejects.toThrow("network unavailable");
failure = undefined;
expect(await adapter.getQuestionStatus({ sessionId: "s", requestId: "form" })).toBe("answered");
expect(await adapter.getQuestion({ sessionId: "s", requestId: "form" })).toMatchObject({ state: { status: "answered" } });
});
test("message admission lookup fails closed on inbox and message transport errors", async () => {
let inboxFailure = true;
+1 -1
View File
@@ -26,7 +26,7 @@ describe("answer view state", () => {
beforeEach(async () => {
base = await mkdtemp(join(tmpdir(), "answer-state-"));
workspaces = new SessionWorkspaceStore(join(base, "workspaces"));
messages = new SessionUiStateStore(join(base, "unused"), workspaces);
messages = new SessionUiStateStore(workspaces);
states = new AnswerViewStateStore(workspaces);
currentSpec = spec();
await saveMessages();
+1 -1
View File
@@ -44,7 +44,7 @@ describe("learning session lifecycle", () => {
});
const orchestrator = (runtime: object) => new LearningOrchestrator(
runtime as OpencodeRuntimeAdapter, new MemoryStore(join(base, "memory"), join(base, "memory-backup")),
new SessionTranscriptStore(join(base, "legacy-transcripts"), records), state,
new SessionTranscriptStore(records), state,
);
it("releases both native helper sessions after a completed review and persists progress in chat", async () => {
+2 -2
View File
@@ -3,6 +3,7 @@ import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { SessionWorkspaceStore } from "../../src/sessions/workspace-store.js";
import { SessionTranscriptStore } from "../../src/sessions/transcriptStore.js";
describe("SessionTranscriptStore", () => {
@@ -11,8 +12,7 @@ describe("SessionTranscriptStore", () => {
beforeEach(async () => {
tempDir = await mkdtemp(join(tmpdir(), "tjwater-transcript-"));
store = new SessionTranscriptStore(tempDir);
await store.initialize();
store = new SessionTranscriptStore(new SessionWorkspaceStore(tempDir));
});
afterEach(async () => {
+5 -5
View File
@@ -2,18 +2,18 @@ import { expect, it } from "bun:test";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { SessionWorkspaceStore } from "../../src/sessions/workspace-store.js";
import { SessionUiStateStore } from "../../src/sessions/uiStateStore.js";
it("restores plan history independently of chat messages after reconnecting", async () => {
const directory = await mkdtemp(join(tmpdir(), "tjwater-plan-state-"));
try {
const store = new SessionUiStateStore(directory);
await store.initialize();
const store = new SessionUiStateStore(new SessionWorkspaceStore(directory));
const executionPlans = [{ revision: 1, nodes: [{ id: "sampling", status: "waiting" }] }];
await store.write({ sessionId: "session-1" }, { sessionId: "session-1", messages: [], executionPlans });
const reloaded = await new SessionUiStateStore(directory).read({ sessionId: "session-1" });
await store.write({ sessionId: "session-1", actorKey: "actor-test", projectKey: "project-test" }, { sessionId: "session-1", messages: [], executionPlans });
const reloaded = await new SessionUiStateStore(new SessionWorkspaceStore(directory)).read({ sessionId: "session-1", actorKey: "actor-test", projectKey: "project-test" });
expect(reloaded?.executionPlans).toEqual(executionPlans);
expect(await store.read({ sessionId: "other-session" })).toBeNull();
expect(await store.read({ sessionId: "other-session", actorKey: "actor-test", projectKey: "project-test" })).toBeNull();
} finally {
await rm(directory, { recursive: true, force: true });
}
+11 -11
View File
@@ -20,11 +20,11 @@ describe("owned session workspaces", () => {
afterEach(async () => { await rm(base, { recursive: true, force: true }); });
it("keeps messages and searchable transcripts private to their user and scope", async () => {
const ui = new SessionUiStateStore(join(base, "old-ui"), records);
const transcript = new SessionTranscriptStore(join(base, "old-transcripts"), records);
const ui = new SessionUiStateStore(records);
const transcript = new SessionTranscriptStore(records);
await ui.write(owner, { sessionId: owner.sessionId, messages: [{ text: "private answer" }] });
await transcript.appendTurn(owner, { userMessage: "private question", assistantMessage: "private answer", toolCallCount: 0 });
expect((await new SessionUiStateStore(join(base, "old-ui"), records).read(owner))?.messages).toHaveLength(1);
expect((await new SessionUiStateStore(records).read(owner))?.messages).toHaveLength(1);
expect(await ui.read({ ...owner, actorKey: "actor-b" })).toBeNull();
expect(await ui.read({ ...owner, projectKey: "project-b" })).toBeNull();
expect(await transcript.search(owner, "private")).toHaveLength(2);
@@ -36,7 +36,7 @@ describe("owned session workspaces", () => {
});
it("has a separate personal namespace for each user and separates it from projects", async () => {
const ui = new SessionUiStateStore(join(base, "old-ui"), records);
const ui = new SessionUiStateStore(records);
const personal = { ...owner, projectKey: "personal-actor-a" };
await ui.write(personal, { sessionId: owner.sessionId, messages: [{ text: "personal answer" }] });
expect(records.path(personal, "messages.json")).toContain("/actor-a/personal/");
@@ -60,13 +60,13 @@ describe("owned session workspaces", () => {
expect(await records.migrateLegacyRecords(options)).toEqual({ uiStates: 1, transcripts: 1, learningStates: 0, retainedUnownedUiStates: 1 });
expect(await readFile(records.path(owner, "messages.json"), "utf8")).toBe(messages);
expect(await readFile(records.path(owner, "transcript.json"), "utf8")).toBe(turns);
await new SessionUiStateStore(uiStateDir, records).write(owner, { sessionId: owner.sessionId, messages: [{ text: "new answer" }] });
await new SessionUiStateStore(records).write(owner, { sessionId: owner.sessionId, messages: [{ text: "new answer" }] });
expect((await records.migrateLegacyRecords(options)).uiStates).toBe(0);
const archive = join(sessionWorkspaceDirectory(records.backupRoot, owner), "chat");
const backup = (await readdir(archive)).find((name) => name.startsWith("messages-"))!;
expect(await readFile(join(archive, backup), "utf8")).toBe(messages);
expect(await readdir(uiStateDir)).toEqual(["orphan.json"]);
expect((await new SessionUiStateStore(uiStateDir, records).read(owner))?.messages).toEqual([{ text: "new answer" }]);
expect((await new SessionUiStateStore(records).read(owner))?.messages).toEqual([{ text: "new answer" }]);
});
it("rejects traversal, linked records and mismatched ownership", async () => {
@@ -75,7 +75,7 @@ describe("owned session workspaces", () => {
const foreign = join(base, "foreign.json");
await writeFile(foreign, "private");
await symlink(foreign, records.path(owner, "messages.json"));
await expect(new SessionUiStateStore(join(base, "old-ui"), records).read(owner)).rejects.toThrow("symbolic link");
await expect(new SessionUiStateStore(records).read(owner)).rejects.toThrow("symbolic link");
await writeFile(join(sessionWorkspaceDirectory(records.root, owner), "chat", "owner.json"), JSON.stringify({ ...owner, actorKey: "actor-b" }));
await expect(records.ensure(owner)).rejects.toThrow("ownership mismatch");
});
@@ -114,8 +114,8 @@ describe("owned session workspaces", () => {
for (const path of [metadataDir, uiStateDir, transcriptDir, learningStateDir]) await mkdir(path);
const metadata = new SessionMetadataStore(metadataDir, records);
const { record } = await metadata.ensure(owner);
const ui = new SessionUiStateStore(uiStateDir, records);
const transcript = new SessionTranscriptStore(transcriptDir, records);
const ui = new SessionUiStateStore(records);
const transcript = new SessionTranscriptStore(records);
const learning = new SessionLearningStateStore(learningStateDir, records);
const legacyUi = JSON.stringify({ sessionId: owner.sessionId, messages: [{ text: "private content" }] });
await writeFile(join(uiStateDir, "chat.json"), legacyUi);
@@ -153,7 +153,7 @@ describe("owned session workspaces", () => {
const metadataDir = join(base, "metadata");
const metadata = new SessionMetadataStore(metadataDir, records);
const { record } = await metadata.ensure(owner);
const transcript = new SessionTranscriptStore(join(base, "legacy"), records);
const transcript = new SessionTranscriptStore(records);
await transcript.appendTurn(owner, { userMessage: "private question", assistantMessage: "private answer", toolCallCount: 0 });
const markDeleted = records.markDeleted.bind(records);
records.markDeleted = async () => { throw new Error("disk write interrupted"); };
@@ -163,7 +163,7 @@ describe("owned session workspaces", () => {
await expect(records.writeRecord(owner, "messages.json", { sessionId: owner.sessionId, messages: [] })).rejects.toThrow("session not found");
records = new SessionWorkspaceStore(records.root, records.backupRoot);
const restarted = new SessionMetadataStore(metadataDir, records);
expect(await new SessionTranscriptStore(join(base, "legacy"), records).search(owner, "private")).toEqual([]);
expect(await new SessionTranscriptStore(records).search(owner, "private")).toEqual([]);
await expect(restarted.ensure(owner)).rejects.toThrow("session not found");
await markDeleted(owner);
await restarted.remove(record);