From 251312190a141388d9481fdc936507c19319a511 Mon Sep 17 00:00:00 2001 From: Kieran Zhang Date: Fri, 2 Oct 2026 15:25:05 +0800 Subject: [PATCH 1/2] docs(spike): add durable multiview spike handoff --- .scratch/durable-multiview/HANDOFF.md | 162 ++++++++++++++++++++++++++ 1 file changed, 162 insertions(+) create mode 100644 .scratch/durable-multiview/HANDOFF.md diff --git a/.scratch/durable-multiview/HANDOFF.md b/.scratch/durable-multiview/HANDOFF.md new file mode 100644 index 00000000..cd660bdd --- /dev/null +++ b/.scratch/durable-multiview/HANDOFF.md @@ -0,0 +1,162 @@ +# Handoff：Pi Durable 多端同屏 spike + +- 日期：2026-10-02 +- 状态：设计已定，未动手。前置的 Pi 1.0 升级在 [PR #425](https://github.com/bubbleptr/pace/pull/425)，合入后从 `origin/main` 开 `feat/durable-multiview-spike`。 +- 关联：[docs/research/pi-durable-analysis.md](../../docs/research/pi-durable-analysis.md)（框架分析，用户的写作素材，只在本地工作区，不提交）、[ADR-0040](../../docs/adr/0040-root-session-process-isolation.md)、[ADR-0042](../../docs/adr/0042-pace-as-chord-presentation-host.md)、[ADR-0044](../../docs/adr/0044-single-writer-session-projection-in-renderer.md) +- 目的：把上一个会话里已核实的事实、已定的决策和待决问题交给新会话，避免重查。 + +## 1. 目标 + +一个 Durable 宿主进程上跑一个演示任务（包含子代理），三个界面**同时**显示并能驱动它： + +1. 服务器上的 TUI（先在本机模拟服务器）； +2. 浏览器里的 Web UI； +3. 本机的 Pace（一个 dev-only 页面）。 + +演示要尽量把 Durable 的特性都用上（见 §6）。spike 的另一半价值是回答「Pace 将来接入 Durable 需要什么」，最后写成结论文档。 + +## 2. 已定的决策（用户已确认） + +| 问题 | 决定 | +|---|---| +| 本机 GUI | Pace 里加一个 dev-only 页面，连到宿主（不是另做 Electron 壳） | +| 代码位置 | Pace 仓库内（先升级到 Pi 1.0 消除版本冲突，见 PR #425） | +| 模型 | 真模型，走 Pi 的 `ModelRuntime` 与 `~/.pi/agent` 凭据（与 `pi` 共用登录） | +| 服务器 | 先全部在本机跑通，宿主只监听 `127.0.0.1`，以后搬到服务器加 SSH 隧道 | +| TUI | 移植上游 durable 编码 Agent 的 `tui.ts`，接远程版 `DurableView` / `DurableController`，TUI 是独立客户端进程 | + +架构上的关键选择：**宿主是一个无界面进程**，独占 Harness 与 SQLite；TUI、Web、Pace 三端地位相同，都是客户端。这样「`kill -9` 宿主 → 三端显示重连 → 重启宿主后三端续上」这一幕才演得出来。 + +## 3. 已核实的事实 + +### 3.1 Durable 本身 + +- 包：`@earendil-works/pi-durable@1.0.0`，依赖 `chord ^1.0.0`、`pi-ai ^1.0.0`、`typebox 1.3.27`、`diff 8.0.4`；`engines.node >= 22.19.0`。本机 Node 是 v25.9.0。**它还不是 Pace 的依赖，spike 要自己加，建议锁精确版本**（README 第一行：API 会无预告变化）。 +- 一份存储同一时间只能被一个进程打开，没有跨进程锁（README「Storage」）。上游 durable 编码 Agent 用 `proper-lockfile` 加了锁文件，崩溃留下的锁 10 秒后失效。宿主应照做。 +- `Conversation.watch()` 返回 `WatchHandle`：`value` 是当前值，`start((value, ops, ctx) => Promise)` 每次提交给出 `ops`。慢消费者积压超过 100 帧会被替换为一帧完整视图；重连从当前视图开始，不回放。 +- `ops` 的类型是 `@earendil-works/chord/delta` 的 `Op`，纯 JSON 元组（源码注释：「Tuples are the form — in memory, on the wire, on disk」）。客户端用同一模块的 `applyImmutable(target, ops)` 还原，**客户端不需要加载 Harness**。delta 模块对 `__proto__` 路径有防护，客户端务必用它而不是自己写 applier。 +- 其他可订阅状态：`harness.taskGraph()` / `watchTaskGraph()`，`harness.documentState(Doc, id)` / `watchDoc()`，`harness.subscribeCommits()`（发现新会话，比如新子代理）。 +- 定时器：`TaskRuntime.sleep(until)`，`until` 是绝对时间，要存在输入或检查点里才能跨重启。 +- `TaskOptions.background` 只对 `ownership: { kind: "conversation" }` 的任务有效。 +- 钩子的 `api`（`HookApi`）只有 `taskId`、`conversationId`、`memo()` 和文档读取（`snapshot`）。**钩子没有 sleep，也拿不到会话句柄**（影响 §8 的审批设计）。 + +### 3.2 上游参考代码（Pi 仓库 main，MIT） + +- 两种子代理示例：[`22-subagent-foreground.ts`](https://github.com/earendil-works/pi/blob/main/packages/durable/test/examples/22-subagent-foreground.ts)（子会话归工具调用所有，`replay: "safe"`，用 `scanConversations({ ownerTaskId })` 找回）和 [`23-subagent-background.ts`](https://github.com/earendil-works/pi/blob/main/packages/durable/test/examples/23-subagent-background.ts)（Anchor + Reporter 两个 background 任务、`app.subagents` 文档、工具 `replay: "unsafe"`）。 +- durable 编码 Agent：[`packages/coding-agent/src/experimental/durable/`](https://github.com/earendil-works/pi/tree/main/packages/coding-agent/src/experimental/durable)，文件为 `main.ts`、`runtime.ts`、`tui.ts`（约 25KB）、`subagent.ts`、`harness-setup.ts`、`prompt.ts`、`sessions.ts`。**不在 npm 发布物里**。 +- **`runtime.ts` 的接缝就是我们要的客户端契约**：`DurableViewSource { current(); subscribe(listener) }`，`DurableView`（「Plain values; no Harness objects cross this boundary」：`session`、`conversation: ConversationView`、`conversations`、`models`、`notices`、`tasks?: TaskGraph`），`DurableController`（`submit(text, whenBusy)`、`compact`、`abort`、`cycleThinking`、`setModel`、`toggleTasks`、`switchConversation`）。`tui.ts` 只依赖这三样加 `SettingsManager`。README 原话：「Nothing in the TUI handles recovery; it only renders the conversation view.」 +- `tui.ts` 用到的 Pi 内部模块，1.0 npm 包（`@earendil-works/pi-coding-agent` 根入口）导出情况： + +| 已导出 | 未导出（需要替代） | +|---|---| +| `AssistantMessageComponent`、`ToolExecutionComponent`、`UserMessageComponent`、`CustomEditor`、`DynamicBorder`、`initTheme`、`getMarkdownTheme`、`KeybindingsManager`、`keyText`、`getAgentDir`、`SettingsManager`、`ModelRuntime`、`resolveCliModel` | `createAllToolRenderers`、`WorkingStatusIndicator`、`getEditorTheme`、`InteractiveThemeController`、`formatTokens`、`findInitialModel`、`configureHttpDispatcher`、`applyHttpProxySettings` | + + 未导出的可以按文件路径引 `dist/...`（Pace 对 mcp 就是这么做的，见根 `vite.config.ts` 的 `@pace/pi-mcp` 别名），或者拷一份进 spike 并注明出处。 + +- **HTTP dispatcher 必须配置。** 上游 `harness-setup.ts` 的注释：「pi's HTTP setup: proxy, idle timeouts, and one undici for fetch. Without it, some provider streams break off.」实现在 Pi 的 `src/core/http-dispatcher.ts`（undici 全局 dispatcher、默认空闲超时 300 s、`autoSelectFamilyAttemptTimeout` 2 s、吞掉 undici 的内部 `error` 事件）。Pace 后端里没有对应代码（搜不到 `setGlobalDispatcher`），宿主要自己补。 + +### 3.3 Pace 侧约束 + +- **Pace renderer 的 CSP 是 `connect-src 'self'`**（`apps/desktop/index.html`），renderer 不能直连 `ws://127.0.0.1`。两条路： + - a）（推荐）由 Pace backend（utilityProcess）当 WebSocket 客户端，经现有 IPC 把快照和 ops 转给 renderer，renderer 只做 `applyImmutable` 和渲染。这和 ADR-0042 的「transport 复用进程 IPC」、ADR-0044 的单写者投影一致，也是 spike 要验证的路径。 + - b）只在 dev 下放宽 CSP。改动小，但验证不到 Pace 真实的接入路径。 +- 根 `tsconfig.json` 只 include `apps` 和 `packages`；`bun` workspaces 是 `apps/*`、`packages/*`；根 `vitest run` 会收集**整个仓库**的测试文件，默认 jsdom 环境（`vite.config.ts`）。spike 放在 `spikes/durable-multiview/` 时要：把 `spikes/*` 加进 workspaces（依赖走根 lockfile），在根 vitest 的 `exclude` 加 `spikes/**`，spike 用自己的 `tsconfig.json` 和 `vitest.config.ts`（node 环境）。 +- AGENTS.md：新增 `shared/ui/` 组件要上 `/design`；dev 页面优先用 Astryx 组件（`bunx astryx build ""`），不手写 div/hex。 +- 宿主跑在 Node 上（Durable 的 SQLite 适配器与 engines 要求），不要用 Bun 运行宿主。 + +## 4. 架构 + +```mermaid +flowchart LR + subgraph Host["spikes/durable-multiview/host（Node）"] + H["Harness + SQLite + 锁文件"] + X["演示扩展
子代理 · 审批 · 回滚任务 · 提醒"] + G["WebSocket 网关
127.0.0.1 + token"] + H --- X + H --- G + end + TUI["tui（独立进程）
移植的 tui.ts + RemoteDurable"] -->|ws| G + Web["web（Vite + React）
RemoteDurable"] -->|ws| G + subgraph Pace["Pace（dev-only 页面）"] + BE["backend：ws 客户端"] -->|IPC| R["renderer：applyImmutable + 渲染"] + end + BE -->|ws| G +``` + +建议的目录（新会话可调整）： + +```text +spikes/durable-multiview/ + host/ Harness、演示扩展、网关、HTTP dispatcher、锁 + protocol/ 帧类型、RemoteDurable（DurableViewSource + DurableController 的远程实现） + tui/ 移植的 tui.ts（保留 MIT 出处）+ main + web/ Vite + React + cli/ P0 用的命令行观察器 +``` + +Pace 页面放在 `apps/desktop/src/pages/` 下的 dev-only 路由，backend 侧桥接放 `packages/backend/src/`，用开关隔离，spike 结束可整体删除。 + +## 5. 网关协议草案 + +- 流（stream）：`conversation:`（`watch()`）、`tasks`（`watchTaskGraph()`）、`doc::`（`watchDoc()`）、`conversations`（由 `subscribeCommits()` 维护的会话列表）、`notices`。 +- 服务端 → 客户端：`{ type: "snapshot", stream, value }`、`{ type: "ops", stream, ops }`、`{ type: "result", id, value | error }`。 +- 客户端 → 服务端：`{ type: "subscribe" | "unsubscribe", stream }`、`{ type: "call", id, method, args }`。`method` 对齐 `DurableController`，另加 `fork`、`configure`、`approve`、`reset`。 +- 重连：重新 `subscribe`，拿新快照；客户端不做历史回放。 +- 鉴权：启动时生成 token，客户端连接时带上；只监听 `127.0.0.1`。 +- 背压：服务端每个客户端一个发送队列；依赖 `watch()` 自带的 100 帧折叠。 + +## 6. 演示剧本与特性覆盖 + +场景：值班 Agent 调查「v2.3 发布失败」。**所有工具都是模拟的，不装 `CodingTools`/bash**：任何客户端都能驱动 Agent，工具又在宿主机上执行。 + +| 步骤 | Durable 特性 | 三端看到 | +|---|---|---| +| 1. 提问，主 Agent 写调查计划到待办 | 文档（同提交） | 待办实时同步 | +| 2. 并行派 3 个前台子代理（日志 / 指标 / 提交历史） | 所有权树、并行工具、`api.output`、`details` 挂载 | 任务图，子代理挂在调用下 | +| 3. 调查中途 `kill -9` 宿主，再重启 | 检查点恢复、`replay: "safe"` 只重跑未完成、`requestId` 找回子代理 | 三端断线 → 续上 | +| 4. TUI steer，网页排 follow-up | inbox 语义 | 队列三端可见 | +| 5. Pace 里切到某个子代理并直接 steer | 子会话即普通会话 | 任一端驱动任一会话 | +| 6. 提出回滚，触发审批钩子 | `beforeTool` + memo（先写者胜） | 三端同时弹审批，先点的生效；审批后崩溃不再问 | +| 7. 回滚为自定义任务，3 个区域并行，一个失败 | `defineTask`、`waiting` + `failFast`、自下而上 abort 补偿 | 任务图里兄弟节点被中止、补偿 | +| 8. Esc；「60 秒后复查」提醒照样触发 | background 边界、`runtime.sleep` | 提醒以 follow-up 出现 | +| 9. 网页在某条消息处 fork「写复盘」并去掉回滚工具 | fork、按会话配置工具、文档 fork 语义 | 会话列表多一个分支 | +| 10. Pace 发起手动压缩 | 压缩是任务，不打断对话 | 压缩状态三端可见 | +| 11. 改扩展文件，宿主热替换 | Registry 原地替换 | 下一次工具输出格式变化 | +| 12. 用量面板 | `pi.usage` | 用量 | + +覆盖不到：自定义存储后端、Bun / Durable Object 运行、提示缓存差量、任务版本迁移。 + +真模型的代价：子代理不一定按剧本派出，崩溃时机也要看着任务图手动掐。用子代理工具的描述和会话 `instructions` 把流程写死一些；工具输出用 `api.output` 慢速流出，给崩溃演示留时间窗。 + +## 7. 分阶段与验收 + +| 阶段 | 内容 | 验收 | +|---|---|---| +| P0 | 宿主（锁、HTTP dispatcher、ModelRuntime、最小扩展）、网关、`protocol` 的 RemoteDurable、命令行观察器 | 两个观察器看到同一段流式输出；`kill -9` 宿主再重启，两端自动续上且转录一致 | +| P1 | 移植 TUI | TUI 作为远程客户端完成提问、steer、abort、切会话、任务面板 | +| P2 | Web UI | 与 TUI 同时连接，状态一致 | +| P3 | Pace dev 页面（backend ws 客户端 + IPC + renderer 投影） | 三端同屏 | +| P4 | 演示扩展与剧本（§6） | 剧本能完整走一遍，崩溃与热替换各演示一次 | +| P5 | `docs/research/` 结论文档 | 回答：Pace 接入 Durable 的传输、投影、子代理呈现、进程模型各需要什么，与 ADR-0040/0042/0044 的冲突点 | + +每个阶段先写失败测试（宿主与协议用 vitest node 环境；网关可以用真 WebSocket 起本地端口测）。 + +## 8. 待决问题 + +1. **审批钩子怎么等人点。** `HookApi` 没有 sleep，也没有会话句柄。候选： + - a）钩子里用普通 `setTimeout` 轮询 `snapshot(ApprovalsDoc)`，拿到结果立刻写 memo。轮询本身不持久，但重启后钩子重跑，会先读 memo，所以语义正确。 + - b）不用钩子，做成工具：`request_approval` 创建一个 Approval 任务，任务用 `runtime.sleep` 轮询文档并持久等待，`rollback` 工具检查审批结果。 + + 建议先试 a），不顺再换 b）。客户端点击走网关的 `approve` 调用，由宿主提交到 `ApprovalsDoc`。 +2. **Pace 页面走 §3.3 的 a）还是 b）。** 推荐 a）。 +3. **`tui.ts` 缺失导出的替代**：按路径引 `dist` 还是拷贝。前者随 Pi 升级自动跟进但可能断，后者稳定但要注明出处。 +4. **子代理用前台还是后台写法**：剧本步骤 2 用前台（示例 22）；如果想演示「主 Agent 不被阻塞、子代理跨轮常驻」，再加一个后台子代理（示例 23）。 +5. 真模型用哪个 provider：读 `~/.pi/agent/settings.json` 的默认模型即可，宿主启动时打印出来。 + +## 9. 新会话的起步步骤 + +1. 确认 PR #425 已合入；在主检出上 `git checkout main && git pull`，删掉本地 `chore/upgrade-pi-1.0`。 +2. `git fetch && git checkout -b feat/durable-multiview-spike origin/main`。 +3. 只把本文件提交进该分支。`docs/research/` 下的 `pi-durable-analysis.md`、`pi-durable-evidence.md`、`pi-durable-visual-guide.md` 是用户写文章的素材，留在工作区、**不提交**，`git add` 时不要用 `-A` 或 `.`。 +4. 读上游 `runtime.ts`、`tui.ts`、`harness-setup.ts`、示例 22/23（链接见 §3.2）。本机有 1.0.0 的 tarball 解包可直接对照：`/private/tmp/pi1/`（`pi-durable`、`pi-coding-agent`、`chord`）。 +5. 按 §7 从 P0 开始，先定 workspace / vitest / tsconfig 的接入方式（§3.3）。 From bece77f92b5c3a45d76a0468f5cd25e0850ada06 Mon Sep 17 00:00:00 2001 From: Kieran Zhang Date: Fri, 2 Oct 2026 15:38:49 +0800 Subject: [PATCH 2/2] feat(spike): durable multiview host, gateway, and remote client (P0) A headless Node host owns the Durable Harness, SQLite store, and lock, and serves conversation views, the task graph, and the conversation list over a 127.0.0.1 WebSocket gateway as snapshots plus chord ops. RemoteDurable implements the upstream DurableViewSource/DurableController over it and reconnects from fresh snapshots after a host restart. Includes a line-mode observer and tests for two-client streaming and kill -9 recovery. --- .scratch/durable-multiview/HANDOFF.md | 27 +- bun.lock | 37 +- package.json | 3 +- spikes/durable-multiview/cli/observe.ts | 58 ++++ spikes/durable-multiview/cli/printer.ts | 47 +++ spikes/durable-multiview/host/gateway.ts | 322 ++++++++++++++++++ spikes/durable-multiview/host/host.ts | 130 +++++++ .../durable-multiview/host/http-dispatcher.ts | 81 +++++ spikes/durable-multiview/host/main.ts | 72 ++++ spikes/durable-multiview/host/pi-setup.ts | 64 ++++ spikes/durable-multiview/package.json | 26 ++ spikes/durable-multiview/protocol/frames.ts | 61 ++++ .../protocol/remote-durable.ts | 293 ++++++++++++++++ .../durable-multiview/protocol/transcript.ts | 43 +++ spikes/durable-multiview/protocol/view.ts | 63 ++++ .../test/crash-recovery.test.ts | 82 +++++ spikes/durable-multiview/test/gateway.test.ts | 86 +++++ spikes/durable-multiview/test/printer.test.ts | 51 +++ spikes/durable-multiview/test/support.ts | 47 +++ spikes/durable-multiview/tsconfig.json | 16 + spikes/durable-multiview/vitest.config.ts | 11 + vite.config.ts | 2 + 22 files changed, 1617 insertions(+), 5 deletions(-) create mode 100644 spikes/durable-multiview/cli/observe.ts create mode 100644 spikes/durable-multiview/cli/printer.ts create mode 100644 spikes/durable-multiview/host/gateway.ts create mode 100644 spikes/durable-multiview/host/host.ts create mode 100644 spikes/durable-multiview/host/http-dispatcher.ts create mode 100644 spikes/durable-multiview/host/main.ts create mode 100644 spikes/durable-multiview/host/pi-setup.ts create mode 100644 spikes/durable-multiview/package.json create mode 100644 spikes/durable-multiview/protocol/frames.ts create mode 100644 spikes/durable-multiview/protocol/remote-durable.ts create mode 100644 spikes/durable-multiview/protocol/transcript.ts create mode 100644 spikes/durable-multiview/protocol/view.ts create mode 100644 spikes/durable-multiview/test/crash-recovery.test.ts create mode 100644 spikes/durable-multiview/test/gateway.test.ts create mode 100644 spikes/durable-multiview/test/printer.test.ts create mode 100644 spikes/durable-multiview/test/support.ts create mode 100644 spikes/durable-multiview/tsconfig.json create mode 100644 spikes/durable-multiview/vitest.config.ts diff --git a/.scratch/durable-multiview/HANDOFF.md b/.scratch/durable-multiview/HANDOFF.md index cd660bdd..7cc3992e 100644 --- a/.scratch/durable-multiview/HANDOFF.md +++ b/.scratch/durable-multiview/HANDOFF.md @@ -1,7 +1,7 @@ # Handoff：Pi Durable 多端同屏 spike - 日期：2026-10-02 -- 状态：设计已定，未动手。前置的 Pi 1.0 升级在 [PR #425](https://github.com/bubbleptr/pace/pull/425)，合入后从 `origin/main` 开 `feat/durable-multiview-spike`。 +- 状态：P0 完成（见 §10），下一步 P1。前置的 Pi 1.0 升级 [PR #425](https://github.com/bubbleptr/pace/pull/425) 已合入，分支 `feat/durable-multiview-spike` 从 `origin/main` 开出。 - 关联：[docs/research/pi-durable-analysis.md](../../docs/research/pi-durable-analysis.md)（框架分析，用户的写作素材，只在本地工作区，不提交）、[ADR-0040](../../docs/adr/0040-root-session-process-isolation.md)、[ADR-0042](../../docs/adr/0042-pace-as-chord-presentation-host.md)、[ADR-0044](../../docs/adr/0044-single-writer-session-projection-in-renderer.md) - 目的：把上一个会话里已核实的事实、已定的决策和待决问题交给新会话，避免重查。 @@ -160,3 +160,28 @@ Pace 页面放在 `apps/desktop/src/pages/` 下的 dev-only 路由，backend 侧 3. 只把本文件提交进该分支。`docs/research/` 下的 `pi-durable-analysis.md`、`pi-durable-evidence.md`、`pi-durable-visual-guide.md` 是用户写文章的素材，留在工作区、**不提交**，`git add` 时不要用 `-A` 或 `.`。 4. 读上游 `runtime.ts`、`tui.ts`、`harness-setup.ts`、示例 22/23（链接见 §3.2）。本机有 1.0.0 的 tarball 解包可直接对照：`/private/tmp/pi1/`（`pi-durable`、`pi-coding-agent`、`chord`）。 5. 按 §7 从 P0 开始，先定 workspace / vitest / tsconfig 的接入方式（§3.3）。 + +## 10. P0 进展（2026-10-02） + +### 10.1 落地的内容 + +- `spikes/durable-multiview/` 是 bun workspace 包 `@pace/durable-multiview-spike`，依赖锁精确版本（pi-durable / chord / pi-ai / pi-coding-agent 均为 1.0.0，`ws`、`proper-lockfile`、`undici`）。根 `package.json` 的 workspaces 加了 `spikes/*`，根 vitest `exclude` 加了 `spikes/**`；spike 有自己的 `tsconfig.json`（NodeNext、`allowImportingTsExtensions`、`erasableSyntaxOnly`）和 `vitest.config.ts`（node 环境）。源码直接用 Node 25 的类型剥离运行，不需要构建。 +- `host/`：`host.ts`（`openHost`：锁、持久 token、SQLite、Harness、网关、`resume()`）、`gateway.ts`（WebSocket 网关）、`main.ts`（入口，`--faux` 用假模型，否则走 `ModelRuntime` 与 `~/.pi/agent`）、`pi-setup.ts`（上游 `harness-setup.ts` 的设置读取）、`http-dispatcher.ts`（从 Pi 拷贝，注明出处）。 +- `protocol/`：`frames.ts`（帧类型）、`view.ts`（与上游 `DurableView` / `DurableController` 结构一致，只多一个 `connection` 字段）、`remote-durable.ts`（`connectRemoteDurable`）、`transcript.ts`（转录与流式文本提取）。 +- `cli/observe.ts`：命令行观察器，能看也能发（普通行为提问或 follow-up，`/steer`、`/abort`、`/tasks`）。 +- 运行：`node host/main.ts` 起宿主（默认端口 7420，数据目录 `~/.pi/agent/experimental/durable-multiview/default`），另开终端 `node cli/observe.ts`。spike 测试：在 spike 目录下 `bun run test`。 + +### 10.2 验收 + +- `test/gateway.test.ts`：宿主在测试进程内，两个远程客户端看到同一段流式输出（B 端收到的每个片段都是最终答案的前缀），最终视图逐字段相等；错误 token 被拒；会话列表与任务图订阅可用。 +- `test/crash-recovery.test.ts`：宿主作为子进程，流式到一半 `kill -9`，两端进入 `reconnecting`，同端口重启后两端自动续上，转录一致。连跑 5 次稳定。 +- 真模型冒烟：`radius/deepseek-v4.1-flash`（读自 `settings.json`）经宿主答复正常。 + +### 10.3 新发现（P1 之后要用） + +- **崩溃后转录里会多一条 aborted 助手消息。** 规格 §8.1 / §8.3：被打断的已提交片段会变成一条 `stopReason: "aborted"` 的 `pi.assistant` 条目，留在转录里但不进入模型上下文，然后用同样的上下文重发请求。所以崩溃后的转录是「用户 → aborted 片段 → 完整答案」。三端 UI 都要把 aborted 条目区分渲染（观察器标 `[aborted]`），剧本 §6 第 3 步的讲解要提这一点。 +- **`ConversationId` 是带品牌的 number**，不是字符串。流名 `conversation:` 解析时要 `Number(...)`。 +- **`watch()` 与 `viewState()` 的区别**：上游 TUI 用的是进程内的 `viewState()`（Chord 已挂载状态，只给值）；网关用 `watch()`，它给 `ops`，并且 listener 返回的 Promise 起背压作用。网关在 socket 写入回调后才 resolve，慢客户端的积压由 `watch()` 自己折叠。 +- **token 要持久化**（数据目录下 `token`，0600）。如果每次启动都重新生成，`kill -9` 重启后客户端就连不上了。鉴权失败时用关闭码 4401，这样浏览器也能读到原因（浏览器 WebSocket 拿不到 HTTP 状态码）。 +- **`submit` 带 `requestId`**（客户端生成 UUID），为「回复丢了再重发不重复提交」做准备；目前客户端还不会自动重发。 +- §3.1 里与 P4 相关的事实（`TaskOptions.background`、`HookApi` 的能力）本阶段没有复核，P4 用到时再核。 diff --git a/bun.lock b/bun.lock index 7d99d55a..32917495 100644 --- a/bun.lock +++ b/bun.lock @@ -90,6 +90,23 @@ "name": "@pace/core", "version": "0.1.0", }, + "spikes/durable-multiview": { + "name": "@pace/durable-multiview-spike", + "version": "0.0.0", + "dependencies": { + "@earendil-works/chord": "1.0.0", + "@earendil-works/pi-ai": "1.0.0", + "@earendil-works/pi-coding-agent": "1.0.0", + "@earendil-works/pi-durable": "1.0.0", + "proper-lockfile": "4.1.2", + "undici": "8.10.2", + "ws": "8.22.0", + }, + "devDependencies": { + "@types/proper-lockfile": "4.1.4", + "@types/ws": "8.18.2", + }, + }, }, "packages": { "@adobe/css-tools": ["@adobe/css-tools@4.5.0", "", {}, "sha512-6OzddxPio9UiWTCemp4N8cYLV2ZN1ncRnV1cVGtve7dhPOtRkleRyx32GQCYSwDYgaHU3USMm84tNsvKzRCa1Q=="], @@ -288,6 +305,8 @@ "@earendil-works/pi-coding-agent": ["@earendil-works/pi-coding-agent@1.0.0", "", { "dependencies": { "@earendil-works/chord": "^1.0.0", "@earendil-works/pi-agent-core": "^1.0.0", "@earendil-works/pi-ai": "^1.0.0", "@earendil-works/pi-codemode": "^1.0.0", "@earendil-works/pi-mcp": "^1.0.0", "@earendil-works/pi-tui": "^1.0.0", "@silvia-odwyer/photon-node": "0.3.4", "chalk": "6.0.0", "cross-spawn": "7.0.6", "diff": "8.0.4", "grok-mermaid": "0.2.3", "highlight.js": "10.7.3", "hosted-git-info": "9.0.3", "ignore": "7.0.8", "jiti": "2.7.0", "minimatch": "10.2.6", "proper-lockfile": "4.1.2", "quickjs-wasi": "3.6.2", "semver": "7.8.5", "typebox": "1.3.27", "undici": "8.10.2", "yaml": "2.9.0" }, "bin": { "pi": "dist/bundle/cli.js" } }, "sha512-/FtbxoSQU/mEv1QnichJjRjqteqaIaMWxmhB4G367+MwZfX7/DI5B9YAg5lqbN7nztFskBEtUSZ+FlmMBECtMw=="], + "@earendil-works/pi-durable": ["@earendil-works/pi-durable@1.0.0", "", { "dependencies": { "@earendil-works/chord": "^1.0.0", "@earendil-works/pi-ai": "^1.0.0", "diff": "8.0.4", "typebox": "1.3.27" } }, "sha512-PvjCIdy5jRsBoJDRpCCI0RfwBax2ZPnMT5aw/tRQxcQOqbGGj2mq1z0WuVxoyimWWi+WYSYT2VySg3RO1qzySw=="], + "@earendil-works/pi-mcp": ["@earendil-works/pi-mcp@1.0.0", "", { "dependencies": { "cross-spawn": "7.0.6" } }, "sha512-rYra0aF5iPmJd+fsB+VBuqxWNABGJ3Iiyhg0CqIc/91ta2n7IvQmCp0Cac/YkUTNnvRsVZnjgAiBkRi6+ouWuw=="], "@earendil-works/pi-telemetry": ["@earendil-works/pi-telemetry@1.0.0", "", {}, "sha512-WjNBj5TYIiPZFQEz2WlULcDwPLaKwIlmsKjVeYM+LJSbnSp38kWsHUJysIDGnu23IcLbKoPtywsvYfUJCeZePA=="], @@ -558,6 +577,8 @@ "@pace/desktop": ["@pace/desktop@workspace:apps/desktop"], + "@pace/durable-multiview-spike": ["@pace/durable-multiview-spike@workspace:spikes/durable-multiview"], + "@pace/server": ["@pace/server@workspace:apps/server"], "@pace/web": ["@pace/web@workspace:apps/web"], @@ -988,6 +1009,8 @@ "@types/parse-json": ["@types/parse-json@4.0.2", "", {}, "sha512-dISoDXWWQwUquiKsyZ4Ng+HX2KsPL7LyHKHQwgGFEA3IaKac4Obd+h2a/a6waisAoepJlBcx9paWqjA8/HVjCw=="], + "@types/proper-lockfile": ["@types/proper-lockfile@4.1.4", "", { "dependencies": { "@types/retry": "*" } }, "sha512-uo2ABllncSqg9F1D4nugVl9v93RmjxF6LJzQLMLDdPaXCUIDPeOJ21Gbqi43xNKzBi/WQ0Q0dICqufzQbMjipQ=="], + "@types/react": ["@types/react@19.2.17", "", { "dependencies": { "csstype": "^3.2.2" } }, "sha512-MXfmqaVPEVgkBT/aY0aGCkRWWtByiYQXo3xdQ8r5RzuFrPiRn8Gar2tQdXSUQ2GKV3bkXckek89V8wQBY2Q/Aw=="], "@types/react-dom": ["@types/react-dom@19.2.3", "", { "peerDependencies": { "@types/react": "^19.2.0" } }, "sha512-jp2L/eY6fn+KgVVQAOqYItbF0VY/YApe5Mz2F0aykSO8gx31bYCZyvSeYxCHKvzHG5eZjc+zyaS5BrBWya2+kQ=="], @@ -1004,6 +1027,8 @@ "@types/use-sync-external-store": ["@types/use-sync-external-store@0.0.6", "", {}, "sha512-zFDAD+tlpf2r4asuHEj0XH6pY6i0g5NeAHPn+15wk3BV6JA69eERFXC1gyGThDkVa1zCyKr5jox1+2LbV/AMLg=="], + "@types/ws": ["@types/ws@8.18.2", "", { "dependencies": { "@types/node": "*" } }, "sha512-67MQl+fpWKVTT1NYdnmo3U4sc/xPo/zQBncVnI74qmQa0z/b+1g6iYqNmGCPbxO+zz2aklb08a0oHfegiVd0/w=="], + "@typescript-eslint/eslint-plugin": ["@typescript-eslint/eslint-plugin@8.70.1", "", { "dependencies": { "@eslint-community/regexpp": "^4.12.2", "@typescript-eslint/scope-manager": "8.70.1", "@typescript-eslint/type-utils": "8.70.1", "@typescript-eslint/utils": "8.70.1", "@typescript-eslint/visitor-keys": "8.70.1", "ignore": "^7.0.5", "natural-compare": "^1.4.0", "ts-api-utils": "^2.5.0" }, "peerDependencies": { "@typescript-eslint/parser": "^8.70.1", "eslint": "^8.57.0 || ^9.0.0 || ^10.0.0", "typescript": ">=4.8.4 <6.1.0" } }, "sha512-nDNrUQ/4ruSNYbu749TRY7cfrzPtoLHEXSNBI8aaNY32LlZCajixqRf3FqcKC4p5Cam4VOHYx/t+i5+nKXvrqA=="], "@typescript-eslint/parser": ["@typescript-eslint/parser@8.70.1", "", { "dependencies": { "@typescript-eslint/scope-manager": "8.70.1", "@typescript-eslint/types": "8.70.1", "@typescript-eslint/typescript-estree": "8.70.1", "@typescript-eslint/visitor-keys": "8.70.1", "debug": "^4.4.3" }, "peerDependencies": { "eslint": "^8.57.0 || ^9.0.0 || ^10.0.0", "typescript": ">=4.8.4 <6.1.0" } }, "sha512-nO974WLllwhSFWQXnMLj6nDGa8f0khKEz1JzpPJ1u7Vm/4X1X6ZHajpoknU4bb41vJyMB0HHVyS2GqdhWfIXZw=="], @@ -2482,7 +2507,7 @@ "typescript-eslint": ["typescript-eslint@8.70.1", "", { "dependencies": { "@typescript-eslint/eslint-plugin": "8.70.1", "@typescript-eslint/parser": "8.70.1", "@typescript-eslint/typescript-estree": "8.70.1", "@typescript-eslint/utils": "8.70.1" }, "peerDependencies": { "eslint": "^8.57.0 || ^9.0.0 || ^10.0.0", "typescript": ">=4.8.4 <6.1.0" } }, "sha512-AcWG7KDjZ2THNXsgwttMaGmzVi0VFRlFYfqFHYQRbDpF3owuYbuiL8c7UUrd2k8s3PoSfIQrWfrGXfcElrWLYA=="], - "undici": ["undici@7.28.0", "", {}, "sha512-cRZYrTDwWznlnRiPjggAGxZXanty6M8RV1ff8Wm4LWXBp7/IG8v5DnOm74DtUBp9OONpK75YlPnIjQqX0dBDtA=="], + "undici": ["undici@8.10.2", "", {}, "sha512-/y4/bH9YNU5hi9NIrpOuvGXFcxrj3CMrV+/AYpowAYTpHn8gX/XPFjNy766FPoYY0miQhdW977JFWKGNhBdwyQ=="], "undici-types": ["undici-types@8.3.0", "", {}, "sha512-j375ScV60dom+YkPFIfTLcOiPxkN/buHz5GobjLhixFuANaNs3C9l4GmrWqejgXWJ7BbJcFYpTEUkS1Ge8bpZQ=="], @@ -2584,7 +2609,7 @@ "write-file-atomic": ["write-file-atomic@5.0.1", "", { "dependencies": { "imurmurhash": "^0.1.4", "signal-exit": "^4.0.1" } }, "sha512-+QU2zd6OTD8XWIJCbffaiQeH9U73qIqafo1x6V1snCWYGJf6cVE0cDR4D8xRzcEnfI21IFrUPzPGtcPf8AC+Rw=="], - "ws": ["ws@8.21.0", "", { "peerDependencies": { "bufferutil": "^4.0.1", "utf-8-validate": ">=5.0.2" }, "optionalPeers": ["bufferutil", "utf-8-validate"] }, "sha512-Vsp28b7DRcimFQvrqu2Wek3z1iYxDCWqHYB8Qsnk/S4RfaCQzPGPyBNuVjJV3cd6UiKtUtp6sNM77gWvzcCH+g=="], + "ws": ["ws@8.22.0", "", { "peerDependencies": { "bufferutil": "^4.0.1", "utf-8-validate": ">=5.0.2" }, "optionalPeers": ["bufferutil", "utf-8-validate"] }, "sha512-Ydggc987+RO0AnWtZ/7Wq9FtNvcrL1b/RO0ud9mWjUPgDrsAAwQSF51sm2hm1XofbU/4jkpGEsLFsZZxU+1DOg=="], "xml-name-validator": ["xml-name-validator@5.0.0", "", {}, "sha512-EvGK8EJ3DhaHfbRlETOWAS5pO9MZITeauHKJyb8wyajUfQUenkIg2MvLDTZ4T/TgIcm3HU0TFBgWWboAZ30UHg=="], @@ -2630,7 +2655,7 @@ "@earendil-works/pi-coding-agent/ignore": ["ignore@7.0.8", "", {}, "sha512-YYNsSlXBjMk92SKnkwvB5LOVSa6OznlFUGcsvrFgNJbJCd0M1XKeFVRc8ZByeCqz32FivYNHJVooLmdqrmvp/Q=="], - "@earendil-works/pi-coding-agent/undici": ["undici@8.10.2", "", {}, "sha512-/y4/bH9YNU5hi9NIrpOuvGXFcxrj3CMrV+/AYpowAYTpHn8gX/XPFjNy766FPoYY0miQhdW977JFWKGNhBdwyQ=="], + "@earendil-works/pi-durable/diff": ["diff@8.0.4", "", {}, "sha512-DPi0FmjiSU5EvQV0++GFDOJ9ASQUVFh5kD+OzOnYdi7n3Wpm9hWWGfB/O2blfHcMVTL5WkQXSnRiK9makhrcnw=="], "@earendil-works/pi-tui/marked": ["marked@18.0.11", "", { "bin": { "marked": "bin/marked.js" } }, "sha512-HnslJfsZkRPBDJRHvVtAaWlZHEpSu7u8LgQuJCELjRKuWR+hpq4A7sLq3p8HaI9ypVoXDXxV34CsQJEe1+J5Aw=="], @@ -2640,6 +2665,8 @@ "@electron/fuses/fs-extra": ["fs-extra@9.1.0", "", { "dependencies": { "at-least-node": "^1.0.0", "graceful-fs": "^4.2.0", "jsonfile": "^6.0.1", "universalify": "^2.0.0" } }, "sha512-hcg3ZmepS30/7BSFqRvoo3DOMQu7IjqxO5nCDt+zM9XWjb33Wg7ziNT+Qvqbuc3+gWpzO02JubVyk2G4Zvo1OQ=="], + "@electron/get/undici": ["undici@7.28.0", "", {}, "sha512-cRZYrTDwWznlnRiPjggAGxZXanty6M8RV1ff8Wm4LWXBp7/IG8v5DnOm74DtUBp9OONpK75YlPnIjQqX0dBDtA=="], + "@electron/notarize/fs-extra": ["fs-extra@9.1.0", "", { "dependencies": { "at-least-node": "^1.0.0", "graceful-fs": "^4.2.0", "jsonfile": "^6.0.1", "universalify": "^2.0.0" } }, "sha512-hcg3ZmepS30/7BSFqRvoo3DOMQu7IjqxO5nCDt+zM9XWjb33Wg7ziNT+Qvqbuc3+gWpzO02JubVyk2G4Zvo1OQ=="], "@electron/osx-sign/isbinaryfile": ["isbinaryfile@4.0.10", "", {}, "sha512-iHrqe5shvBUcFbmZq9zOQHBoeOhZJu6RQGrDpBgenUm/Am+F3JM2MgQj+rK3Z601fzrL5gLZWtAPH2OBaSVcyw=="], @@ -2670,6 +2697,8 @@ "@eslint/config-array/minimatch": ["minimatch@10.2.5", "", { "dependencies": { "brace-expansion": "^5.0.5" } }, "sha512-MULkVLfKGYDFYejP07QOurDLLQpcjk7Fw+7jXS2R2czRQzR56yHRveU5NDJEOviH+hETZKSkIk5c+T23GjFUMg=="], + "@google/genai/ws": ["ws@8.21.0", "", { "peerDependencies": { "bufferutil": "^4.0.1", "utf-8-validate": ">=5.0.2" }, "optionalPeers": ["bufferutil", "utf-8-validate"] }, "sha512-Vsp28b7DRcimFQvrqu2Wek3z1iYxDCWqHYB8Qsnk/S4RfaCQzPGPyBNuVjJV3cd6UiKtUtp6sNM77gWvzcCH+g=="], + "@hyperse/eslint-config-hyperse/typescript-eslint": ["typescript-eslint@8.62.0", "", { "dependencies": { "@typescript-eslint/eslint-plugin": "8.62.0", "@typescript-eslint/parser": "8.62.0", "@typescript-eslint/typescript-estree": "8.62.0", "@typescript-eslint/utils": "8.62.0" }, "peerDependencies": { "eslint": "^8.57.0 || ^9.0.0 || ^10.0.0", "typescript": ">=4.8.4 <6.1.0" } }, "sha512-8QxXi+ZACKX0kaqO4gY8kn0RSD9gFfaHDWwjqtEN48aWCBkX4MJaufWN+c3BzlrXLOxfywDL8CaoqUwcRq4j4Q=="], "@isaacs/cliui/string-width": ["string-width@5.1.2", "", { "dependencies": { "eastasianwidth": "^0.2.0", "emoji-regex": "^9.2.2", "strip-ansi": "^7.0.1" } }, "sha512-HnLOCR3vjcY8beoNLtcjZ5/nxn2afmME6lhrDrebokqMap+XbeW8n9TXpPDOqdGK5qcI3oT0GKTW6wC7EMiVqA=="], @@ -2856,6 +2885,8 @@ "hoist-non-react-statics/react-is": ["react-is@16.13.1", "", {}, "sha512-24e6ynE2H+OKt4kqsOvNd8kBpV65zoxbA4BVsEOB3ARVWQki/DHzaUoC5KuON/BiccDaCCTZBuOcfZs70kR8bQ=="], + "jsdom/undici": ["undici@7.28.0", "", {}, "sha512-cRZYrTDwWznlnRiPjggAGxZXanty6M8RV1ff8Wm4LWXBp7/IG8v5DnOm74DtUBp9OONpK75YlPnIjQqX0dBDtA=="], + "katex/commander": ["commander@8.3.0", "", {}, "sha512-OkTL9umf+He2DZkUq8f8J9of7yL6RJKI24dVITBmNfZBmri9zYZQrKkuXiKhyfPSu8tUhnVBB1iKXevvnlR4Ww=="], "make-dir/semver": ["semver@5.7.2", "", { "bin": { "semver": "bin/semver" } }, "sha512-cBznnQ9KjJqU67B52RMC65CMarK2600WFnbkcaiwWq3xy/5haFJlshgnpjovMVJ+Hff49d8GEn0b87C5pDQ10g=="], diff --git a/package.json b/package.json index 3e32bc63..dfaed161 100644 --- a/package.json +++ b/package.json @@ -6,7 +6,8 @@ "type": "module", "workspaces": [ "apps/*", - "packages/*" + "packages/*", + "spikes/*" ], "scripts": { "postinstall": "install-electron", diff --git a/spikes/durable-multiview/cli/observe.ts b/spikes/durable-multiview/cli/observe.ts new file mode 100644 index 00000000..9e8a9c6e --- /dev/null +++ b/spikes/durable-multiview/cli/observe.ts @@ -0,0 +1,58 @@ +#!/usr/bin/env node +// A line-mode client of a running host: prints the shown conversation as it +// streams, and sends each typed line. Run several to watch one host from many terminals. +// +// node cli/observe.ts [--data-dir DIR] [--url ws://127.0.0.1:7420] +// typed lines: text (prompt; follow-up while busy), /steer text, /abort, /tasks, /quit +import { readFile } from "node:fs/promises"; +import { join, resolve } from "node:path"; +import { createInterface } from "node:readline"; +import { parseArgs } from "node:util"; +import { getAgentDir } from "@earendil-works/pi-coding-agent"; +import { connectRemoteDurable } from "../protocol/remote-durable.ts"; +import type { DurableView } from "../protocol/view.ts"; +import { initialPrinter, printIncrement } from "./printer.ts"; + +const { values } = parseArgs({ + options: { + "data-dir": { type: "string" }, + url: { type: "string", default: "ws://127.0.0.1:7420" }, + token: { type: "string" }, + }, +}); +const dataDir = resolve(values["data-dir"] ?? join(getAgentDir(), "experimental", "durable-multiview", "default")); +const token = values.token ?? (await readFile(join(dataDir, "token"), "utf8")).trim(); + +const remote = await connectRemoteDurable({ url: values.url, token }); +let printer = initialPrinter; +let shown: DurableView | undefined; +const render = (): void => { + const view = remote.view.current(); + if (shown !== undefined && view.connection !== shown.connection) process.stdout.write(`\n[${view.connection}]\n`); + for (const notice of view.notices.slice(shown === undefined ? 0 : shown.notices.length)) { + process.stdout.write(`\n[${notice.level}] ${notice.message}\n`); + } + if (view.tasks !== shown?.tasks && view.tasks !== undefined) { + const nodes = Object.values(view.tasks.tasks).map((task) => `${task.id}:${task.kind}(${task.state.status})`); + process.stdout.write(`\n[tasks] ${nodes.length === 0 ? "none" : nodes.join(" ")}\n`); + } + shown = view; + const step = printIncrement(printer, view.conversation); + printer = step.state; + process.stdout.write(step.output); +}; +remote.view.subscribe(render); +process.stdout.write(`[connected to ${values.url}, session ${remote.view.current().session.id}]\n`); +render(); + +const input = createInterface({ input: process.stdin }); +for await (const line of input) { + const text = line.trim(); + if (text === "") continue; + if (text === "/quit") break; + if (text === "/abort") await remote.controller.abort(); + else if (text === "/tasks") await remote.controller.toggleTasks(); + else if (text.startsWith("/steer ")) await remote.controller.submit(text.slice("/steer ".length), "steer"); + else await remote.controller.submit(text, "followUp"); +} +remote.close(); diff --git a/spikes/durable-multiview/cli/printer.ts b/spikes/durable-multiview/cli/printer.ts new file mode 100644 index 00000000..940044c8 --- /dev/null +++ b/spikes/durable-multiview/cli/printer.ts @@ -0,0 +1,47 @@ +import type { ConversationView } from "@earendil-works/pi-durable"; +import { streamingText, transcript, type TranscriptLine } from "../protocol/transcript.ts"; + +/** What is already on the terminal: finished lines, and a partial left open on the last line. */ +export interface PrinterState { + readonly printed: readonly TranscriptLine[]; + readonly streamed: string | undefined; +} + +export const initialPrinter: PrinterState = { printed: [], streamed: undefined }; + +const sameLine = (a: TranscriptLine, b: TranscriptLine): boolean => JSON.stringify(a) === JSON.stringify(b); + +const suffixOf = (line: TranscriptLine): string => + line.role === "assistant" && line.stopReason !== "stop" ? ` [${line.stopReason}]` : ""; + +const format = (line: TranscriptLine): string => `${line.role}: ${line.text}${suffixOf(line)}`; + +/** Append-only terminal output for the next view; a view that rewrites printed history is printed afresh. */ +export function printIncrement(state: PrinterState, view: ConversationView): { state: PrinterState; output: string } { + const lines = transcript(view); + let { printed, streamed } = state; + let output = ""; + + if (printed.length > lines.length || printed.some((line, i) => !sameLine(line, lines[i]!))) { + output += `${streamed === undefined ? "" : "\n"}--- view replaced ---\n`; + printed = []; + streamed = undefined; + } + + for (const line of lines.slice(printed.length)) { + if (streamed === undefined) output += `${format(line)}\n`; + else if (line.role === "assistant" && line.text.startsWith(streamed)) output += `${line.text.slice(streamed.length)}${suffixOf(line)}\n`; + else output += `\n${format(line)}\n`; + streamed = undefined; + } + printed = lines; + + const partial = streamingText(view); + if (partial !== undefined && partial !== "") { + if (streamed === undefined) output += `assistant: ${partial}`; + else if (partial.startsWith(streamed)) output += partial.slice(streamed.length); + else output += `\nassistant: ${partial}`; + streamed = partial; + } + return { state: { printed, streamed }, output }; +} diff --git a/spikes/durable-multiview/host/gateway.ts b/spikes/durable-multiview/host/gateway.ts new file mode 100644 index 00000000..267482f6 --- /dev/null +++ b/spikes/durable-multiview/host/gateway.ts @@ -0,0 +1,322 @@ +import type { IncomingMessage } from "node:http"; +import type { AddressInfo } from "node:net"; +import type { Context } from "@earendil-works/chord"; +import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; +import { clampThinkingLevel, getSupportedThinkingLevels, type ModelThinkingLevel } from "@earendil-works/pi-ai"; +import type { Models } from "@earendil-works/pi-ai/models"; +import { + AgentDoc, + type AgentState, + type Conversation, + type ConversationId, + type Cursor, + type EntryRecord, + type Harness, + ROOT_CONVERSATION_ID, + type WatchHandle, +} from "@earendil-works/pi-durable"; +import { WebSocket, WebSocketServer } from "ws"; +import { + type CallMethod, + type CallMethods, + type ClientFrame, + type ServerFrame, + type StreamName, + UNAUTHORIZED_CLOSE_CODE, +} from "../protocol/frames.ts"; +import type { ConversationSummary, ModelSummary, Notice, SessionInfo } from "../protocol/view.ts"; + +const context: Context = BACKGROUND_CONTEXT; + +export interface GatewayOptions { + readonly harness: Harness; + readonly models: Models; + readonly modelSummaries: () => readonly ModelSummary[]; + readonly session: SessionInfo; + readonly token: string; + readonly port: number; +} + +export interface Gateway { + readonly url: string; + /** Tell every client, for example a Harness report. */ + broadcast(level: Notice["level"], message: string): void; + close(): Promise; +} + +export async function startGateway(options: GatewayOptions): Promise { + const conversations = await ConversationList.open(options.harness); + const server = new WebSocketServer({ host: "127.0.0.1", port: options.port }); + await new Promise((resolve, reject) => { + server.once("listening", resolve); + server.once("error", reject); + }); + const clients = new Set(); + server.on("connection", (socket: WebSocket, request: IncomingMessage) => { + const token = new URL(request.url ?? "/", "ws://127.0.0.1").searchParams.get("token"); + if (token !== options.token) { + socket.close(UNAUTHORIZED_CLOSE_CODE, "unauthorized"); + return; + } + const client = new GatewayClient(socket, options, conversations); + clients.add(client); + socket.once("close", () => { + clients.delete(client); + void client.dispose(); + }); + }); + const { port } = server.address() as AddressInfo; + return { + url: `ws://127.0.0.1:${port}`, + broadcast: (level, message) => { + for (const client of clients) client.send({ type: "notice", level, message }); + }, + async close() { + conversations.dispose(); + await Promise.all([...clients].map((client) => client.dispose())); + for (const socket of server.clients) socket.terminate(); + await new Promise((resolve) => server.close(() => resolve())); + }, + }; +} + +/** The conversation list, kept from commit publications as the in-process TUI keeps it. */ +class ConversationList { + readonly #listeners = new Set<(list: readonly ConversationSummary[]) => void>(); + #list: readonly ConversationSummary[]; + #unsubscribe: () => void = () => {}; + #notifying = false; + + private constructor(list: readonly ConversationSummary[]) { + this.#list = list; + } + + static async open(harness: Harness): Promise { + const summaries: ConversationSummary[] = []; + let cursor: Cursor | undefined; + do { + const page = await harness.commit((tx) => tx.scanConversations({}, 256, cursor), context); + for (const { id } of page.items) summaries.push({ id, label: labelOf(id), ...(await firstInput(harness, id)) }); + cursor = page.next; + } while (cursor !== undefined); + const list = new ConversationList(summaries); + // A commit listener only records; it calls no Session API. + list.#unsubscribe = harness.subscribeCommits((publication) => { + let next = list.#list; + for (const change of publication.changes) { + if (change.type === "conversation") { + next = [...next, { id: change.value.id, label: labelOf(change.value.id) }]; + } else if (change.type === "entry" && change.value.kind === "pi.user") { + const id = change.value.conversationId; + next = next.map((summary) => (summary.id === id && summary.title === undefined ? { ...summary, ...titleOf(change.value) } : summary)); + } + } + if (next !== list.#list) list.#set(next); + }); + return list; + } + + get value(): readonly ConversationSummary[] { + return this.#list; + } + + subscribe(listener: (list: readonly ConversationSummary[]) => void): () => void { + this.#listeners.add(listener); + return () => this.#listeners.delete(listener); + } + + #set(list: readonly ConversationSummary[]): void { + this.#list = list; + if (this.#notifying) return; + this.#notifying = true; + setImmediate(() => { + this.#notifying = false; + for (const listener of this.#listeners) listener(this.#list); + }); + } + + dispose(): void { + this.#unsubscribe(); + this.#listeners.clear(); + } +} + +const labelOf = (id: ConversationId): string => (id === ROOT_CONVERSATION_ID ? "main" : `subagent ${id}`); + +function titleOf(entry: EntryRecord | undefined): { title?: string } { + const message = entry?.model?.[0]; + if (message?.role !== "user") return {}; + const text = + typeof message.content === "string" + ? message.content + : message.content.flatMap((block) => (block.type === "text" ? [block.text] : [])).join(" "); + return { title: text.replace(/\s+/g, " ").trim() }; +} + +/** A subagent's task: the oldest user message of its conversation. */ +async function firstInput(harness: Harness, id: ConversationId): Promise<{ title?: string }> { + if (id === ROOT_CONVERSATION_ID) return {}; + const conversation = (await harness.conversation(id, context))!; + let first: EntryRecord | undefined; + let cursor: Cursor | undefined; + do { + const page = await conversation.entries({}, 256, cursor, context); + first = page.items.findLast((entry) => entry.kind === "pi.user") ?? first; + cursor = page.next; + } while (cursor !== undefined); + return titleOf(first); +} + +type Subscription = { stop(): Promise | void }; + +class GatewayClient { + readonly #socket: WebSocket; + readonly #options: GatewayOptions; + readonly #conversations: ConversationList; + readonly #subscriptions = new Map(); + #disposed = false; + + constructor(socket: WebSocket, options: GatewayOptions, conversations: ConversationList) { + this.#socket = socket; + this.#options = options; + this.#conversations = conversations; + socket.on("message", (data) => void this.#receive(JSON.parse(String(data)) as ClientFrame)); + this.send({ + type: "hello", + session: options.session, + root: ROOT_CONVERSATION_ID, + models: options.modelSummaries(), + }); + } + + /** Resolves once the frame is handed to the socket, so a watch callback holds its next frame until then. */ + send(frame: ServerFrame): Promise { + if (this.#socket.readyState !== WebSocket.OPEN) return Promise.resolve(); + return new Promise((resolve) => this.#socket.send(JSON.stringify(frame), () => resolve())); + } + + async #receive(frame: ClientFrame): Promise { + if (frame.type === "subscribe") await this.#subscribe(frame.stream); + else if (frame.type === "unsubscribe") await this.#unsubscribe(frame.stream); + else await this.#answer(frame.id, frame.method, frame.args); + } + + async #subscribe(stream: StreamName): Promise { + // A resubscribe restarts the stream from a fresh snapshot. + await this.#unsubscribe(stream); + this.#subscriptions.set(stream, "pending"); + let subscription: Subscription | undefined; + try { + subscription = await this.#open(stream); + } catch (error) { + this.#subscriptions.delete(stream); + await this.send({ type: "ended", stream, reason: error instanceof Error ? error.message : String(error) }); + return; + } + // Unsubscribed, or the socket went away, while the watch was being acquired. + if (this.#disposed || this.#subscriptions.get(stream) !== "pending") { + await subscription.stop(); + return; + } + this.#subscriptions.set(stream, subscription); + } + + async #open(stream: StreamName): Promise { + if (stream === "conversations") { + void this.send({ type: "snapshot", stream, value: this.#conversations.value }); + const unsubscribe = this.#conversations.subscribe((value) => void this.send({ type: "snapshot", stream, value })); + return { stop: unsubscribe }; + } + if (stream === "tasks") return this.#forward(stream, await this.#options.harness.watchTaskGraph(context)); + const conversation = await this.#conversation(Number(stream.slice("conversation:".length)) as ConversationId); + return this.#forward(stream, await conversation.watch(context)); + } + + async #forward(stream: StreamName, watch: WatchHandle): Promise { + await this.send({ type: "snapshot", stream, value: watch.value }); + watch.start((_value, ops) => this.send({ type: "ops", stream, ops })); + return watch; + } + + async #unsubscribe(stream: StreamName): Promise { + const subscription = this.#subscriptions.get(stream); + this.#subscriptions.delete(stream); + if (subscription !== undefined && subscription !== "pending") await subscription.stop(); + } + + async #conversation(id: ConversationId): Promise { + const conversation = await this.#options.harness.conversation(id, context); + if (conversation === undefined) throw new Error(`Conversation ${id} does not exist`); + return conversation; + } + + async #answer(id: number, method: CallMethod, args: CallMethods[CallMethod]["args"]): Promise { + try { + const value = await this.#call(method, args); + await this.send({ type: "result", id, ok: true, value: value ?? null }); + } catch (error) { + await this.send({ type: "result", id, ok: false, error: error instanceof Error ? error.message : String(error) }); + } + } + + async #call(method: CallMethod, args: CallMethods[CallMethod]["args"]): Promise { + const conversation = await this.#conversation(args.conversationId); + switch (method) { + case "submit": { + const { text, whenBusy, requestId } = args as CallMethods["submit"]["args"]; + const submission = await conversation.submit({ type: "input", content: text, whenBusy, requestId }, context); + void submission.wait(context).then( + (settled) => { + if (settled.status === "unanswered" && settled.reason !== "aborted") { + void this.send({ type: "notice", level: "error", message: `No answer: ${settled.reason}` }); + } + }, + () => {}, + ); + return { submissionId: submission.id }; + } + case "abort": + await conversation.abort(context); + return null; + case "compact": { + const { instructions } = args as CallMethods["compact"]["args"]; + const taskId = await conversation.compact(instructions, context); + void this.#options.harness.waitForTask(taskId, context).then( + (receipt) => void this.send({ type: "notice", level: "info", message: `Compaction ${receipt.state.outcome.status}.` }), + () => {}, + ); + return { taskId }; + } + case "setModel": { + const { model: ref } = args as CallMethods["setModel"]["args"]; + const model = this.#options.models.getModel(ref.provider, ref.modelId); + if (model === undefined) throw new Error(`Unknown model: ${ref.provider}/${ref.modelId}`); + const thinking: ModelThinkingLevel = (await this.#agent(conversation.id)).thinkingLevel ?? "off"; + await conversation.configure({ model: ref, thinkingLevel: clampThinkingLevel(model, thinking) }, context); + return null; + } + case "cycleThinking": { + const agent = await this.#agent(conversation.id); + const model = agent.model === undefined ? undefined : this.#options.models.getModel(agent.model.provider, agent.model.modelId); + if (model === undefined) throw new Error("No model selected"); + if (!model.reasoning) throw new Error("Current model does not support thinking"); + const levels = getSupportedThinkingLevels(model); + const level = agent.thinkingLevel ?? "off"; + await conversation.configure({ thinkingLevel: levels[(levels.indexOf(level) + 1) % levels.length] ?? "off" }, context); + return null; + } + } + } + + async #agent(id: ConversationId): Promise> { + return (await this.#options.harness.snapshot(AgentDoc, id, context)) ?? {}; + } + + async dispose(): Promise { + if (this.#disposed) return; + this.#disposed = true; + const subscriptions = [...this.#subscriptions.values()]; + this.#subscriptions.clear(); + for (const subscription of subscriptions) if (subscription !== "pending") await subscription.stop(); + } +} diff --git a/spikes/durable-multiview/host/host.ts b/spikes/durable-multiview/host/host.ts new file mode 100644 index 00000000..2e749ba8 --- /dev/null +++ b/spikes/durable-multiview/host/host.ts @@ -0,0 +1,130 @@ +import { randomBytes } from "node:crypto"; +import { mkdir, readFile, writeFile } from "node:fs/promises"; +import { basename, join } from "node:path"; +import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; +import type { ModelThinkingLevel } from "@earendil-works/pi-ai"; +import type { Models } from "@earendil-works/pi-ai/models"; +import { + createRegistry, + Harness, + type HarnessSettings, + type ModelRef, + type Registry, +} from "@earendil-works/pi-durable"; +import { openNodeSqliteStorage } from "@earendil-works/pi-durable/storage/sqlite/node"; +import lockfile from "proper-lockfile"; +import type { ModelSummary } from "../protocol/view.ts"; +import { startGateway } from "./gateway.ts"; + +const context = BACKGROUND_CONTEXT; + +export interface OpenHostOptions { + /** Holds `session.sqlite`, the token, and the lock; one host process at a time. */ + readonly dataDir: string; + readonly cwd: string; + readonly models: Models; + readonly modelSummaries: () => readonly ModelSummary[]; + readonly registry?: Registry; + readonly settings?: HarnessSettings; + /** Applied only when the root conversation is created. */ + readonly initialModel?: ModelRef & { readonly thinkingLevel?: ModelThinkingLevel }; + /** 0 picks a free port. A fixed port lets clients reconnect to a restarted host. */ + readonly port: number; + /** How long a lock left by a killed host blocks the next one. proper-lockfile's minimum is 2000. */ + readonly lockStaleMs?: number; +} + +export interface OpenedHost { + readonly url: string; + readonly token: string; + readonly harness: Harness; + close(): Promise; +} + +/** Persisted so a restarted host accepts the clients that were connected before it died. */ +async function hostToken(dataDir: string): Promise { + const path = join(dataDir, "token"); + try { + return (await readFile(path, "utf8")).trim(); + } catch { + const token = randomBytes(24).toString("base64url"); + await writeFile(path, `${token}\n`, { mode: 0o600 }); + return token; + } +} + +export async function openHost(options: OpenHostOptions): Promise { + await mkdir(options.dataDir, { recursive: true }); + // Durable has no cross-process lock of its own; a second opener would corrupt the store. + const stale = Math.max(2000, options.lockStaleMs ?? 10_000); + const release = await lockfile + .lock(options.dataDir, { + realpath: false, + stale, + retries: { retries: Math.ceil(stale / 1000) + 2, minTimeout: 1000, maxTimeout: 1000 }, + }) + .catch((error: unknown) => { + throw new Error(`Session is already open in another process: ${options.dataDir}`, { cause: error }); + }); + + let harness: Harness | undefined; + try { + const token = await hostToken(options.dataDir); + const reports: unknown[] = []; + let report = (error: unknown): void => void reports.push(error); + harness = await Harness.open( + await openNodeSqliteStorage(join(options.dataDir, "session.sqlite")), + { + models: options.models, + registry: options.registry ?? createRegistry(), + ...(options.settings === undefined ? {} : { settings: options.settings }), + onReport: (error) => report(error), + }, + context, + ); + const { initialModel } = options; + await harness.root(context, { + agent: { + cwd: options.cwd, + ...(initialModel === undefined ? {} : { model: { provider: initialModel.provider, modelId: initialModel.modelId } }), + ...(initialModel?.thinkingLevel === undefined ? {} : { thinkingLevel: initialModel.thinkingLevel }), + }, + }); + const gateway = await startGateway({ + harness, + models: options.models, + modelSummaries: options.modelSummaries, + session: { id: basename(options.dataDir), directory: options.dataDir, cwd: options.cwd }, + token, + port: options.port, + }); + report = (error) => gateway.broadcast("warning", error instanceof Error ? error.message : String(error)); + for (const error of reports.splice(0)) report(error); + // Recovered work from an interrupted turn continues now. + harness.resume(); + + const opened = harness; + let closing: Promise | undefined; + return { + url: gateway.url, + token, + harness, + close() { + closing ??= (async () => { + try { + await gateway.close(); + // Close writes no outcome: a running turn resumes at the next open. + await opened.close(context); + } finally { + await release(); + } + })(); + return closing; + }, + }; + } catch (error) { + await harness?.close(context).catch(() => {}); + await release().catch(() => {}); + throw error; + } +} diff --git a/spikes/durable-multiview/host/http-dispatcher.ts b/spikes/durable-multiview/host/http-dispatcher.ts new file mode 100644 index 00000000..566e87d5 --- /dev/null +++ b/spikes/durable-multiview/host/http-dispatcher.ts @@ -0,0 +1,81 @@ +// Copied from pi's packages/coding-agent/src/core/http-dispatcher.ts (MIT, Earendil Works) +// at the 1.0.0 release, trimmed to what the host calls. It is outside pi-coding-agent's +// package exports, and the upstream durable harness notes that without it some provider +// streams break off. +import { EventEmitter } from "node:events"; +import * as undici from "undici"; + +export const DEFAULT_HTTP_IDLE_TIMEOUT_MS = 300_000; +// Node's 250ms default can terminate valid connection attempts on high-latency routes. +const DEFAULT_AUTO_SELECT_FAMILY_ATTEMPT_TIMEOUT_MS = 2_000; + +const originalGlobalFetch = globalThis.fetch; +let installedGlobalFetch: typeof globalThis.fetch | undefined; + +export function applyHttpProxySettings(httpProxy: string | undefined): void { + const proxy = httpProxy?.trim(); + if (!proxy) return; + process.env.HTTP_PROXY ??= proxy; + process.env.HTTPS_PROXY ??= proxy; +} + +const ignoreUndiciDispatcherError = (_error: unknown): void => {}; + +// Undici can emit an internal Client "error" while terminating a mid-stream +// fetch body. The body stream still rejects through reader.read(); this listener +// only prevents EventEmitter's unhandled "error" special case from crashing the host. +function withUndiciErrorListener(dispatcher: T): T { + if (dispatcher instanceof EventEmitter) { + EventEmitter.prototype.on.call(dispatcher, "error", ignoreUndiciDispatcherError); + } + return dispatcher; +} + +function createUndiciClient(origin: string | URL, options: object): undici.Dispatcher { + return withUndiciErrorListener(new undici.Client(origin, options as undici.Client.Options)); +} + +function createUndiciOriginDispatcher(origin: string | URL, options: object): undici.Dispatcher { + const dispatcherOptions = options as undici.Pool.Options; + if (dispatcherOptions.connections === 1) { + return createUndiciClient(origin, dispatcherOptions); + } + return withUndiciErrorListener( + new undici.Pool(origin, { + ...dispatcherOptions, + factory: createUndiciClient, + }), + ); +} + +export function configureHttpDispatcher(timeoutMs: number = DEFAULT_HTTP_IDLE_TIMEOUT_MS): void { + if (!Number.isFinite(timeoutMs) || timeoutMs < 0) { + throw new Error(`Invalid HTTP idle timeout: ${String(timeoutMs)}`); + } + const normalizedTimeoutMs = Math.floor(timeoutMs); + const dispatcher = withUndiciErrorListener( + new undici.EnvHttpProxyAgent({ + allowH2: false, + // Keep HTTP origins on CONNECT tunnels as they were before Undici 8.7. + proxyTunnel: true, + bodyTimeout: normalizedTimeoutMs, + connect: { + autoSelectFamilyAttemptTimeout: DEFAULT_AUTO_SELECT_FAMILY_ATTEMPT_TIMEOUT_MS, + }, + headersTimeout: normalizedTimeoutMs, + clientFactory: createUndiciClient, + factory: createUndiciOriginDispatcher, + }), + ); + undici.setGlobalDispatcher(dispatcher); + // Keep fetch and the dispatcher on the same undici implementation. Node 26.0's + // bundled fetch can otherwise consume compressed responses through npm undici's + // dispatcher without decompressing them, causing response.json() failures. + // If a caller replaced fetch after module load, preserve that deliberate override. + const shouldInstallGlobals = + installedGlobalFetch === undefined ? globalThis.fetch === originalGlobalFetch : globalThis.fetch === installedGlobalFetch; + if (shouldInstallGlobals) { + undici.install?.(); + installedGlobalFetch = globalThis.fetch; + } +} diff --git a/spikes/durable-multiview/host/main.ts b/spikes/durable-multiview/host/main.ts new file mode 100644 index 00000000..76a13a8a --- /dev/null +++ b/spikes/durable-multiview/host/main.ts @@ -0,0 +1,72 @@ +#!/usr/bin/env node +// The headless host: owns the Harness, its SQLite store, and the lock; every UI +// is a gateway client. Run on Node, not Bun (node:sqlite, pi-durable's engines). +// +// node host/main.ts [--data-dir DIR] [--port 7420] [--cwd DIR] +// node host/main.ts --faux "scripted answer" [--faux-tps 40] # no real model, for tests +import { join, resolve } from "node:path"; +import { parseArgs } from "node:util"; +import { createModels } from "@earendil-works/pi-ai/models"; +import { fauxAssistantMessage, fauxProvider } from "@earendil-works/pi-ai/providers/faux"; +import { getAgentDir, ModelRuntime, SettingsManager } from "@earendil-works/pi-coding-agent"; +import { type OpenHostOptions, openHost } from "./host.ts"; +import { configureHarnessHttp, createHarnessSettings, defaultModel, modelSummaries } from "./pi-setup.ts"; + +const { values } = parseArgs({ + options: { + "data-dir": { type: "string" }, + port: { type: "string", default: "7420" }, + cwd: { type: "string" }, + faux: { type: "string" }, + "faux-tps": { type: "string", default: "40" }, + "lock-stale-ms": { type: "string" }, + }, +}); + +const cwd = resolve(values.cwd ?? process.cwd()); +const dataDir = resolve(values["data-dir"] ?? join(getAgentDir(), "experimental", "durable-multiview", "default")); +const common = { + dataDir, + cwd, + port: Number(values.port), + ...(values["lock-stale-ms"] === undefined ? {} : { lockStaleMs: Number(values["lock-stale-ms"]) }), +}; + +function fauxOptions(answer: string): OpenHostOptions { + const faux = fauxProvider({ tokensPerSecond: Number(values["faux-tps"]), tokenSize: { min: 1, max: 1 } }); + const models = createModels(); + models.setProvider(faux.provider); + // Every request gets the same answer, so a request rerun after a crash streams it again. + faux.setResponses(Array.from({ length: 100 }, () => () => fauxAssistantMessage(answer))); + const model = faux.getModel(); + const summary = { provider: model.provider, modelId: model.id, name: model.name, contextWindow: model.contextWindow }; + return { ...common, models, modelSummaries: () => [summary], initialModel: { provider: model.provider, modelId: model.id } }; +} + +async function piOptions(): Promise { + const modelRuntime = await ModelRuntime.create(); + const settingsManager = SettingsManager.create(cwd); + configureHarnessHttp(settingsManager); + const initialModel = defaultModel(settingsManager, modelRuntime); + return { + ...common, + models: modelRuntime, + modelSummaries: () => modelSummaries(modelRuntime), + settings: createHarnessSettings(settingsManager), + ...(initialModel === undefined ? {} : { initialModel }), + }; +} + +const options = values.faux === undefined ? await piOptions() : fauxOptions(values.faux); +const host = await openHost(options); +const model = options.initialModel === undefined ? null : `${options.initialModel.provider}/${options.initialModel.modelId}`; +console.log(JSON.stringify({ event: "ready", url: host.url, token: host.token, dataDir, model })); + +let stopping = false; +const stop = (): void => { + if (stopping) return; + stopping = true; + void host.close().finally(() => process.exit(0)); +}; +process.on("SIGINT", stop); +process.on("SIGTERM", stop); diff --git a/spikes/durable-multiview/host/pi-setup.ts b/spikes/durable-multiview/host/pi-setup.ts new file mode 100644 index 00000000..72f7faa3 --- /dev/null +++ b/spikes/durable-multiview/host/pi-setup.ts @@ -0,0 +1,64 @@ +// After createHarnessSettings / configureHarnessHttp / findInitialAgentModel in pi's +// packages/coding-agent/src/experimental/durable/harness-setup.ts (MIT, Earendil Works). +import type { ModelThinkingLevel } from "@earendil-works/pi-ai"; +import type { ModelRef, HarnessSettings } from "@earendil-works/pi-durable"; +import type { ModelRuntime, SettingsManager } from "@earendil-works/pi-coding-agent"; +import type { ModelSummary } from "../protocol/view.ts"; +import { applyHttpProxySettings, configureHttpDispatcher } from "./http-dispatcher.ts"; + +export function configureHarnessHttp(settingsManager: SettingsManager): void { + applyHttpProxySettings(settingsManager.getGlobalSettings().httpProxy); + configureHttpDispatcher(settingsManager.getHttpIdleTimeoutMs()); +} + +/** Harness settings read at every use from pi's settings as loaded at startup. */ +export function createHarnessSettings(settingsManager: SettingsManager): HarnessSettings { + return { + get stream() { + const provider = settingsManager.getProviderRetrySettings(); + const idle = settingsManager.getHttpIdleTimeoutMs(); + return { + timeoutMs: provider.timeoutMs ?? (idle === 0 ? 2147483647 : idle), + maxRetryDelayMs: provider.maxRetryDelayMs, + ...(provider.maxRetries === undefined ? {} : { maxRetries: provider.maxRetries }), + }; + }, + get compaction() { + return settingsManager.getCompactionSettings(); + }, + get retry() { + return settingsManager.getRetrySettings(); + }, + get steeringMode() { + return settingsManager.getSteeringMode(); + }, + get followUpMode() { + return settingsManager.getFollowUpMode(); + }, + }; +} + +/** + * pi's default model from settings.json when it is available. Simpler than pi's + * findInitialModel, which is not exported: no scoped models, no provider fallback. + */ +export function defaultModel( + settingsManager: SettingsManager, + modelRuntime: ModelRuntime, +): (ModelRef & { readonly thinkingLevel?: ModelThinkingLevel }) | undefined { + const provider = settingsManager.getDefaultProvider(); + const modelId = settingsManager.getDefaultModel(); + if (provider === undefined || modelId === undefined) return undefined; + if (modelRuntime.getModel(provider, modelId) === undefined) return undefined; + const thinkingLevel = settingsManager.getDefaultThinkingLevel(); + return { provider, modelId, ...(thinkingLevel === undefined ? {} : { thinkingLevel }) }; +} + +export function modelSummaries(modelRuntime: ModelRuntime): ModelSummary[] { + return modelRuntime.getAvailableSnapshot().map((model) => ({ + provider: model.provider, + modelId: model.id, + name: model.name, + contextWindow: model.contextWindow, + })); +} diff --git a/spikes/durable-multiview/package.json b/spikes/durable-multiview/package.json new file mode 100644 index 00000000..39de8ba6 --- /dev/null +++ b/spikes/durable-multiview/package.json @@ -0,0 +1,26 @@ +{ + "name": "@pace/durable-multiview-spike", + "version": "0.0.0", + "private": true, + "license": "Apache-2.0", + "type": "module", + "scripts": { + "host": "node host/main.ts", + "observe": "node cli/observe.ts", + "test": "vitest run --config vitest.config.ts", + "typecheck": "tsc --noEmit -p tsconfig.json" + }, + "dependencies": { + "@earendil-works/chord": "1.0.0", + "@earendil-works/pi-ai": "1.0.0", + "@earendil-works/pi-coding-agent": "1.0.0", + "@earendil-works/pi-durable": "1.0.0", + "proper-lockfile": "4.1.2", + "undici": "8.10.2", + "ws": "8.22.0" + }, + "devDependencies": { + "@types/proper-lockfile": "4.1.4", + "@types/ws": "8.18.2" + } +} diff --git a/spikes/durable-multiview/protocol/frames.ts b/spikes/durable-multiview/protocol/frames.ts new file mode 100644 index 00000000..c65d7cef --- /dev/null +++ b/spikes/durable-multiview/protocol/frames.ts @@ -0,0 +1,61 @@ +import type { Op } from "@earendil-works/chord/delta"; +import type { ConversationId, ModelRef } from "@earendil-works/pi-durable"; +import type { ModelSummary, Notice, SessionInfo } from "./view.ts"; + +/** + * A subscribable value: one conversation's view, the task graph, or the + * conversation list. A subscription starts with a snapshot; ops follow and + * apply to it with chord's `applyImmutable`. + */ +export type StreamName = `conversation:${ConversationId}` | "tasks" | "conversations"; + +export const conversationStream = (id: ConversationId): StreamName => `conversation:${id}`; + +/** Close code for a rejected token; clients must not reconnect after it. */ +export const UNAUTHORIZED_CLOSE_CODE = 4401; + +export interface CallMethods { + submit: { + args: { + conversationId: ConversationId; + text: string; + whenBusy: "steer" | "followUp"; + /** Makes a resend after a lost reply admit nothing new. */ + requestId: string; + }; + result: { submissionId: string }; + }; + abort: { args: { conversationId: ConversationId }; result: null }; + compact: { args: { conversationId: ConversationId; instructions?: string }; result: { taskId: string } }; + setModel: { args: { conversationId: ConversationId; model: ModelRef }; result: null }; + cycleThinking: { args: { conversationId: ConversationId }; result: null }; +} + +export type CallMethod = keyof CallMethods; + +export type ServerFrame = + | { + readonly type: "hello"; + readonly session: SessionInfo; + readonly root: ConversationId; + readonly models: readonly ModelSummary[]; + } + | { readonly type: "snapshot"; readonly stream: StreamName; readonly value: unknown } + | { readonly type: "ops"; readonly stream: StreamName; readonly ops: readonly Op[] } + /** The stream will send nothing more, for example an unknown conversation. */ + | { readonly type: "ended"; readonly stream: StreamName; readonly reason: string } + | { readonly type: "result"; readonly id: number; readonly ok: true; readonly value: unknown } + | { readonly type: "result"; readonly id: number; readonly ok: false; readonly error: string } + | { readonly type: "notice"; readonly level: Notice["level"]; readonly message: string }; + +export type ClientFrame = + | { readonly type: "subscribe"; readonly stream: StreamName } + | { readonly type: "unsubscribe"; readonly stream: StreamName } + | { + [M in CallMethod]: { + readonly type: "call"; + readonly id: number; + readonly method: M; + readonly args: CallMethods[M]["args"]; + }; + }[CallMethod]; diff --git a/spikes/durable-multiview/protocol/remote-durable.ts b/spikes/durable-multiview/protocol/remote-durable.ts new file mode 100644 index 00000000..b9f02e86 --- /dev/null +++ b/spikes/durable-multiview/protocol/remote-durable.ts @@ -0,0 +1,293 @@ +import { applyImmutable } from "@earendil-works/chord/delta"; +import type { ConversationId, ConversationView, TaskGraph } from "@earendil-works/pi-durable"; +import { + type CallMethod, + type CallMethods, + type ClientFrame, + conversationStream, + type ServerFrame, + type StreamName, + UNAUTHORIZED_CLOSE_CODE, +} from "./frames.ts"; +import type { + ConversationSummary, + DurableController, + DurableView, + DurableViewSource, + Notice, +} from "./view.ts"; + +export interface RemoteDurableOptions { + /** `ws://127.0.0.1:` of a host's gateway. */ + readonly url: string; + readonly token: string; + readonly reconnectDelayMs?: { readonly min: number; readonly max: number }; +} + +export interface RemoteDurable { + readonly view: DurableViewSource; + readonly controller: DurableController; + close(): void; +} + +/** + * A DurableViewSource and DurableController over a host's gateway. Holds no + * Harness: each subscribed stream is a snapshot plus chord ops, and a lost + * host is reconnected to and resubscribed from fresh snapshots, never replayed. + */ +export function connectRemoteDurable(options: RemoteDurableOptions): Promise { + return new RemoteClient(options).ready; +} + +type Pending = { resolve(value: unknown): void; reject(error: Error): void }; + +class RemoteClient { + readonly ready: Promise; + readonly #options: RemoteDurableOptions; + readonly #listeners = new Set<() => void>(); + readonly #values = new Map(); + readonly #wanted = new Set(["conversations"]); + readonly #snapshotWaiters = new Map(); + readonly #pending = new Map(); + #socket: WebSocket | undefined; + #state: DurableView | undefined; + #current: ConversationId | undefined; + #nextCall = 1; + #nextNotice = 1; + #attempt = 0; + #closed = false; + #notifying = false; + #commands = Promise.resolve(); + #resolveReady!: (value: RemoteDurable) => void; + #rejectReady!: (error: Error) => void; + + constructor(options: RemoteDurableOptions) { + this.#options = options; + this.ready = new Promise((resolve, reject) => { + this.#resolveReady = resolve; + this.#rejectReady = reject; + }); + this.#open(); + } + + #open(): void { + const url = new URL(this.#options.url); + url.searchParams.set("token", this.#options.token); + const socket = new WebSocket(url); + this.#socket = socket; + socket.onmessage = (event) => this.#receive(JSON.parse(String(event.data)) as ServerFrame); + socket.onclose = (event) => this.#lost(socket, event.code); + } + + #lost(socket: WebSocket, code: number): void { + if (this.#socket !== socket) return; + this.#socket = undefined; + for (const pending of this.#pending.values()) pending.reject(new Error("Disconnected from host")); + this.#pending.clear(); + if (this.#state === undefined) { + this.#closed = true; + this.#rejectReady(new Error(code === UNAUTHORIZED_CLOSE_CODE ? "Unauthorized: wrong host token" : `Could not connect to ${this.#options.url}`)); + return; + } + if (this.#closed) return; + if (code === UNAUTHORIZED_CLOSE_CODE) { + this.#closed = true; + this.#update({ connection: "closed" }); + this.#notice("error", "Host rejected the token."); + return; + } + this.#update({ connection: "reconnecting" }); + const { min, max } = this.#options.reconnectDelayMs ?? { min: 200, max: 2000 }; + const delay = Math.min(max, min * 2 ** this.#attempt++); + setTimeout(() => { + if (!this.#closed) this.#open(); + }, delay); + } + + #send(frame: ClientFrame): void { + this.#socket?.send(JSON.stringify(frame)); + } + + #receive(frame: ServerFrame): void { + switch (frame.type) { + case "hello": { + this.#attempt = 0; + this.#current ??= frame.root; + this.#wanted.add(conversationStream(this.#current)); + for (const stream of this.#wanted) this.#send({ type: "subscribe", stream }); + if (this.#state !== undefined) this.#update({ session: frame.session, models: frame.models, connection: "connected" }); + else this.#awaitFirstView(frame); + return; + } + case "snapshot": + if (!this.#wanted.has(frame.stream)) return; + this.#values.set(frame.stream, frame.value); + this.#refresh(); + for (const waiter of this.#snapshotWaiters.get(frame.stream)?.splice(0) ?? []) waiter.resolve(); + return; + case "ops": + if (!this.#wanted.has(frame.stream) || !this.#values.has(frame.stream)) return; + this.#values.set(frame.stream, applyImmutable(this.#values.get(frame.stream), frame.ops)); + this.#refresh(); + return; + case "ended": + this.#wanted.delete(frame.stream); + this.#values.delete(frame.stream); + for (const waiter of this.#snapshotWaiters.get(frame.stream)?.splice(0) ?? []) { + waiter.reject(new Error(`Stream ${frame.stream} ended: ${frame.reason}`)); + } + this.#refresh(); + return; + case "result": { + const pending = this.#pending.get(frame.id); + this.#pending.delete(frame.id); + if (frame.ok) pending?.resolve(frame.value); + else pending?.reject(new Error(frame.error)); + return; + } + case "notice": + this.#notice(frame.level, frame.message); + return; + } + } + + #awaitFirstView(hello: Extract): void { + const conversation = conversationStream(hello.root); + void Promise.all([this.#snapshot(conversation), this.#snapshot("conversations")]).then(() => { + this.#state = { + session: hello.session, + conversation: this.#values.get(conversation) as ConversationView, + conversations: this.#values.get("conversations") as ConversationSummary[], + models: hello.models, + notices: [], + connection: "connected", + }; + this.#resolveReady({ view: this.#viewSource(), controller: this.#controller(), close: () => this.#close() }); + }); + } + + /** Resolve at the stream's next snapshot; reject if the host ends it instead. */ + #snapshot(stream: StreamName): Promise { + return new Promise((resolve, reject) => { + const waiters = this.#snapshotWaiters.get(stream) ?? []; + waiters.push({ resolve, reject }); + this.#snapshotWaiters.set(stream, waiters); + }); + } + + #refresh(): void { + if (this.#state === undefined || this.#current === undefined) return; + const conversation = this.#values.get(conversationStream(this.#current)) as ConversationView | undefined; + const conversations = this.#values.get("conversations") as ConversationSummary[] | undefined; + const tasks = this.#wanted.has("tasks") ? (this.#values.get("tasks") as TaskGraph | undefined) : undefined; + this.#update({ + ...(conversation === undefined ? {} : { conversation }), + ...(conversations === undefined ? {} : { conversations }), + tasks, + }); + } + + #update(patch: Partial): void { + if (this.#state === undefined) return; + this.#state = { ...this.#state, ...patch }; + if (this.#notifying) return; + this.#notifying = true; + // One render per burst of frames, as the in-process view source does. + queueMicrotask(() => { + this.#notifying = false; + for (const listener of this.#listeners) listener(); + }); + } + + #notice(level: Notice["level"], message: string): void { + if (this.#state === undefined) return; + this.#update({ notices: [...this.#state.notices, { id: this.#nextNotice++, level, message }].slice(-20) }); + } + + #call(method: M, args: CallMethods[M]["args"]): Promise { + if (this.#socket === undefined || this.#state?.connection !== "connected") { + return Promise.reject(new Error("Not connected to the host")); + } + const id = this.#nextCall++; + return new Promise((resolve, reject) => { + this.#pending.set(id, { resolve: resolve as (value: unknown) => void, reject }); + this.#send({ type: "call", id, method, args } as ClientFrame); + }); + } + + /** Serialized like the in-process controller, so toggles and switches apply in order; failures become notices. */ + #command(operation: () => Promise): Promise { + this.#commands = this.#commands + .then(operation) + .then( + () => {}, + (error: unknown) => this.#notice("error", error instanceof Error ? error.message : String(error)), + ); + return this.#commands; + } + + #viewSource(): DurableViewSource { + return { + current: () => this.#state!, + subscribe: (listener) => { + this.#listeners.add(listener); + return () => this.#listeners.delete(listener); + }, + }; + } + + #controller(): DurableController { + const conversationId = (): ConversationId => this.#current!; + return { + submit: (text, whenBusy) => + this.#command(() => this.#call("submit", { conversationId: conversationId(), text, whenBusy, requestId: crypto.randomUUID() })), + compact: (instructions) => + this.#command(() => + this.#call("compact", { conversationId: conversationId(), ...(instructions === undefined ? {} : { instructions }) }), + ), + // Not queued: it resolves once the conversation is idle. + abort: () => + this.#call("abort", { conversationId: conversationId() }).then( + () => {}, + (error: unknown) => this.#notice("error", error instanceof Error ? error.message : String(error)), + ), + cycleThinking: () => this.#command(() => this.#call("cycleThinking", { conversationId: conversationId() })), + setModel: (model) => this.#command(() => this.#call("setModel", { conversationId: conversationId(), model })), + toggleTasks: () => + this.#command(async () => { + if (this.#wanted.delete("tasks")) { + this.#send({ type: "unsubscribe", stream: "tasks" }); + this.#values.delete("tasks"); + this.#refresh(); + return; + } + this.#wanted.add("tasks"); + const shown = this.#snapshot("tasks"); + this.#send({ type: "subscribe", stream: "tasks" }); + await shown; + }), + switchConversation: (id) => + this.#command(async () => { + const next = conversationStream(id); + const previous = conversationStream(conversationId()); + if (next === previous) return; + this.#wanted.add(next); + const shown = this.#snapshot(next); + this.#send({ type: "subscribe", stream: next }); + await shown; + this.#current = id; + this.#wanted.delete(previous); + this.#values.delete(previous); + this.#send({ type: "unsubscribe", stream: previous }); + this.#refresh(); + }), + }; + } + + #close(): void { + if (this.#closed) return; + this.#closed = true; + this.#update({ connection: "closed" }); + this.#socket?.close(); + } +} diff --git a/spikes/durable-multiview/protocol/transcript.ts b/spikes/durable-multiview/protocol/transcript.ts new file mode 100644 index 00000000..8dd905d9 --- /dev/null +++ b/spikes/durable-multiview/protocol/transcript.ts @@ -0,0 +1,43 @@ +import type { ConversationView, EntryRecord } from "@earendil-works/pi-durable"; + +export type TranscriptLine = + | { readonly role: "user"; readonly text: string } + /** `aborted` marks a partial an interrupted attempt left behind; it stays in the transcript but out of model context. */ + | { readonly role: "assistant"; readonly text: string; readonly stopReason: string }; + +type ContentBlock = { readonly type: string; readonly text?: string }; +type MessageLike = { readonly role: string; readonly content: string | readonly ContentBlock[]; readonly stopReason?: string }; + +function textOf(message: MessageLike): string { + if (typeof message.content === "string") return message.content; + return message.content.flatMap((block) => (block.type === "text" && block.text !== undefined ? [block.text] : [])).join(""); +} + +function lineOf(entry: EntryRecord): TranscriptLine | undefined { + if (entry.kind !== "pi.user" && entry.kind !== "pi.assistant") return undefined; + const message = entry.model?.[0] as MessageLike | undefined; + if (message === undefined) return undefined; + if (entry.kind === "pi.user") return { role: "user", text: textOf(message) }; + return { role: "assistant", text: textOf(message), stopReason: message.stopReason ?? "stop" }; +} + +/** The user and assistant text of the active transcript, oldest first. */ +export function transcript(view: ConversationView): TranscriptLine[] { + return view.entries.flatMap((entry) => { + const line = lineOf(entry); + return line === undefined ? [] : [line]; + }); +} + +/** Text of the in-flight response, from the committed `pi.live` partial; absent when nothing streams. */ +export function streamingText(view: ConversationView): string | undefined { + const live = view.docs["pi.live"] as { generation?: { message?: MessageLike } } | undefined; + const message = live?.generation?.message; + return message === undefined ? undefined : textOf(message); +} + +/** Whether the conversation has a run in flight. */ +export function isBusy(view: ConversationView): boolean { + const live = view.docs["pi.live"] as { run?: unknown } | undefined; + return live?.run !== undefined; +} diff --git a/spikes/durable-multiview/protocol/view.ts b/spikes/durable-multiview/protocol/view.ts new file mode 100644 index 00000000..b324aba3 --- /dev/null +++ b/spikes/durable-multiview/protocol/view.ts @@ -0,0 +1,63 @@ +// The client contract, after DurableView / DurableViewSource / DurableController in +// pi's packages/coding-agent/src/experimental/durable/runtime.ts (MIT, Earendil Works). +// Kept structurally identical so the ported TUI renders a remote host unchanged; +// `connection` is the one addition, since a remote view can lose its host. +import type { ConversationId, ConversationView, ModelRef, TaskGraph } from "@earendil-works/pi-durable"; + +export interface ModelSummary extends ModelRef { + readonly name: string; + readonly contextWindow: number; +} + +export interface Notice { + readonly id: number; + readonly level: "info" | "warning" | "error"; + readonly message: string; +} + +/** A conversation the user can switch to: the main one, or a subagent's. */ +export interface ConversationSummary { + readonly id: ConversationId; + readonly label: string; + /** The first user message, for a subagent its task. */ + readonly title?: string; +} + +export interface SessionInfo { + readonly id: string; + readonly directory: string; + readonly cwd: string; +} + +export type ConnectionState = "connected" | "reconnecting" | "closed"; + +/** Everything a client renders. Plain values; no Harness objects cross this boundary. */ +export interface DurableView { + readonly session: SessionInfo; + /** The conversation shown and talked to. */ + readonly conversation: ConversationView; + readonly conversations: readonly ConversationSummary[]; + readonly models: readonly ModelSummary[]; + readonly notices: readonly Notice[]; + /** The live task graph while the task panel is open. */ + readonly tasks?: TaskGraph; + readonly connection: ConnectionState; +} + +export interface DurableViewSource { + current(): DurableView; + subscribe(listener: () => void): () => void; +} + +/** What a client may ask for. */ +export interface DurableController { + /** Prompt when idle; otherwise steer or queue a follow-up. */ + submit(text: string, whenBusy: "steer" | "followUp"): Promise; + compact(instructions: string | undefined): Promise; + abort(): Promise; + cycleThinking(): Promise; + setModel(model: ModelRef): Promise; + toggleTasks(): Promise; + /** Show and talk to another conversation. */ + switchConversation(id: ConversationId): Promise; +} diff --git a/spikes/durable-multiview/test/crash-recovery.test.ts b/spikes/durable-multiview/test/crash-recovery.test.ts new file mode 100644 index 00000000..00e950cb --- /dev/null +++ b/spikes/durable-multiview/test/crash-recovery.test.ts @@ -0,0 +1,82 @@ +import { type ChildProcess, spawn } from "node:child_process"; +import { once } from "node:events"; +import { createInterface } from "node:readline"; +import { fileURLToPath } from "node:url"; +import { afterEach, expect, it } from "vitest"; +import { connectRemoteDurable, type RemoteDurable } from "../protocol/remote-durable.ts"; +import { isBusy, streamingText, transcript } from "../protocol/transcript.ts"; +import { freePort, LONG_ANSWER, tempDir, waitForView } from "./support.ts"; + +const spikeDir = fileURLToPath(new URL("..", import.meta.url)); +const cleanups: (() => Promise | void)[] = []; +afterEach(async () => { + for (const cleanup of cleanups.splice(0).reverse()) await cleanup(); +}); + +interface HostProcess { + readonly child: ChildProcess; + readonly url: string; + readonly token: string; +} + +async function spawnHost(dataDir: string, port: number): Promise { + const child = spawn( + process.execPath, + ["host/main.ts", "--data-dir", dataDir, "--cwd", dataDir, "--port", String(port), "--faux", LONG_ANSWER, "--faux-tps", "40", "--lock-stale-ms", "2000"], + { cwd: spikeDir, stdio: ["ignore", "pipe", "inherit"] }, + ); + cleanups.push(() => { + if (child.exitCode === null && child.signalCode === null) child.kill("SIGKILL"); + }); + const lines = createInterface({ input: child.stdout! }); + for await (const line of lines) { + const event = JSON.parse(line) as { event: string; url: string; token: string }; + if (event.event === "ready") return { child, url: event.url, token: event.token }; + } + throw new Error("host exited before it was ready"); +} + +it("resumes both clients after the host is killed mid-stream and restarted", async () => { + const dir = await tempDir(); + cleanups.push(dir.remove); + const port = await freePort(); + const first = await spawnHost(dir.path, port); + + const connect = async (): Promise => { + const client = await connectRemoteDurable({ url: first.url, token: first.token, reconnectDelayMs: { min: 100, max: 500 } }); + cleanups.push(() => client.close()); + return client; + }; + const a = await connect(); + const b = await connect(); + + await a.controller.submit("investigate", "followUp"); + // Kill while the answer is still streaming, partway through. + await waitForView(b.view, (view) => (streamingText(view.conversation)?.length ?? 0) > 20); + const seenBeforeKill = streamingText(b.view.current().conversation)!; + expect(seenBeforeKill.length).toBeLessThan(LONG_ANSWER.length); + first.child.kill("SIGKILL"); + await once(first.child, "exit"); + await Promise.all([a, b].map((client) => waitForView(client.view, (view) => view.connection === "reconnecting"))); + + const second = await spawnHost(dir.path, port); + expect(second.token).toBe(first.token); + + const settled = (client: RemoteDurable) => + waitForView( + client.view, + (view) => view.connection === "connected" && !isBusy(view.conversation) && transcript(view.conversation).at(-1)?.role === "assistant", + 30_000, + ); + await Promise.all([settled(a), settled(b)]); + + // The committed partial survives the crash as an aborted entry; recovery then resends the request. + const [user, interrupted, answer, ...rest] = transcript(a.view.current().conversation); + expect(user).toEqual({ role: "user", text: "investigate" }); + expect(interrupted).toMatchObject({ role: "assistant", stopReason: "aborted" }); + expect(LONG_ANSWER.startsWith(interrupted!.text)).toBe(true); + expect(interrupted!.text.length).toBeGreaterThanOrEqual(seenBeforeKill.length); + expect(answer).toEqual({ role: "assistant", text: LONG_ANSWER, stopReason: "stop" }); + expect(rest).toEqual([]); + expect(b.view.current().conversation).toEqual(a.view.current().conversation); +}); diff --git a/spikes/durable-multiview/test/gateway.test.ts b/spikes/durable-multiview/test/gateway.test.ts new file mode 100644 index 00000000..844ab661 --- /dev/null +++ b/spikes/durable-multiview/test/gateway.test.ts @@ -0,0 +1,86 @@ +import { createModels } from "@earendil-works/pi-ai/models"; +import { fauxAssistantMessage, fauxProvider } from "@earendil-works/pi-ai/providers/faux"; +import { afterEach, describe, expect, it } from "vitest"; +import { openHost, type OpenedHost } from "../host/host.ts"; +import { connectRemoteDurable, type RemoteDurable } from "../protocol/remote-durable.ts"; +import { streamingText, transcript } from "../protocol/transcript.ts"; +import { LONG_ANSWER, tempDir, waitForView } from "./support.ts"; + +const cleanups: (() => Promise | void)[] = []; +afterEach(async () => { + for (const cleanup of cleanups.splice(0).reverse()) await cleanup(); +}); + +async function startFauxHost(): Promise { + const dir = await tempDir(); + cleanups.push(dir.remove); + const faux = fauxProvider({ tokensPerSecond: 200 }); + const models = createModels(); + models.setProvider(faux.provider); + faux.setResponses([fauxAssistantMessage(LONG_ANSWER)]); + const model = faux.getModel(); + const host = await openHost({ + dataDir: dir.path, + cwd: dir.path, + models, + modelSummaries: () => [{ provider: model.provider, modelId: model.id, name: model.name, contextWindow: model.contextWindow }], + initialModel: { provider: model.provider, modelId: model.id }, + port: 0, + }); + cleanups.push(() => host.close()); + return host; +} + +async function connect(host: OpenedHost, token = host.token): Promise { + const client = await connectRemoteDurable({ url: host.url, token }); + cleanups.push(() => client.close()); + return client; +} + +describe("gateway", () => { + it("shows one streamed answer to two clients, driven by either", async () => { + const host = await startFauxHost(); + const a = await connect(host); + const b = await connect(host); + expect(b.view.current().conversation.conversation.id).toBe(a.view.current().conversation.conversation.id); + + const partialsSeenByB: string[] = []; + b.view.subscribe(() => { + const text = streamingText(b.view.current().conversation); + if (text !== undefined && text !== "") partialsSeenByB.push(text); + }); + + await a.controller.submit("investigate", "followUp"); + + const answered = (client: RemoteDurable) => + waitForView(client.view, (view) => transcript(view.conversation).at(-1)?.text === LONG_ANSWER); + await Promise.all([answered(a), answered(b)]); + + expect(transcript(b.view.current().conversation)).toEqual([ + { role: "user", text: "investigate" }, + { role: "assistant", text: LONG_ANSWER, stopReason: "stop" }, + ]); + expect(b.view.current().conversation).toEqual(a.view.current().conversation); + expect(partialsSeenByB.length).toBeGreaterThan(0); + for (const partial of partialsSeenByB) expect(LONG_ANSWER.startsWith(partial)).toBe(true); + expect(partialsSeenByB.some((partial) => partial.length < LONG_ANSWER.length)).toBe(true); + }); + + it("rejects a client with the wrong token", async () => { + const host = await startFauxHost(); + await expect(connect(host, "not-the-token")).rejects.toThrow(/unauthorized/i); + }); + + it("lists the conversations and opens the task graph on request", async () => { + const host = await startFauxHost(); + const client = await connect(host); + const rootId = client.view.current().conversation.conversation.id; + expect(client.view.current().conversations.map((summary) => summary.id)).toEqual([rootId]); + + expect(client.view.current().tasks).toBeUndefined(); + await client.controller.toggleTasks(); + await waitForView(client.view, (view) => view.tasks !== undefined); + await client.controller.toggleTasks(); + expect(client.view.current().tasks).toBeUndefined(); + }); +}); diff --git a/spikes/durable-multiview/test/printer.test.ts b/spikes/durable-multiview/test/printer.test.ts new file mode 100644 index 00000000..8479aba8 --- /dev/null +++ b/spikes/durable-multiview/test/printer.test.ts @@ -0,0 +1,51 @@ +import type { ConversationView, EntryRecord } from "@earendil-works/pi-durable"; +import { describe, expect, it } from "vitest"; +import { initialPrinter, printIncrement } from "../cli/printer.ts"; + +let nextEntry = 1; +const user = (text: string) => + ({ id: nextEntry++, conversationId: 0, kind: "pi.user", model: [{ role: "user", content: text }] }) as unknown as EntryRecord; +const assistant = (text: string, stopReason = "stop") => + ({ + id: nextEntry++, + conversationId: 0, + kind: "pi.assistant", + model: [{ role: "assistant", content: [{ type: "text", text }], stopReason }], + }) as unknown as EntryRecord; +const view = (entries: EntryRecord[], partial?: string): ConversationView => + ({ + conversation: { id: 0 }, + entries, + docs: partial === undefined ? {} : { "pi.live": { generation: { attempt: 1, message: { role: "assistant", content: [{ type: "text", text: partial }] } } } }, + }) as unknown as ConversationView; + +function run(views: ConversationView[]): string { + let state = initialPrinter; + let out = ""; + for (const next of views) { + const step = printIncrement(state, next); + state = step.state; + out += step.output; + } + return out; +} + +describe("printIncrement", () => { + it("streams a partial inline, then ends the line when the answer lands", () => { + const u = user("hi"); + const a = assistant("hello there"); + expect(run([view([u]), view([u], "hel"), view([u], "hello th"), view([u, a])])).toBe("user: hi\nassistant: hello there\n"); + }); + + it("marks an interrupted partial and prints the resent answer in full", () => { + const u = user("hi"); + const out = run([view([u], "hel"), view([u, assistant("hel", "aborted")], "hello"), view([u, assistant("hel", "aborted"), assistant("hello")])]); + expect(out).toBe("user: hi\nassistant: hel [aborted]\nassistant: hello\n"); + }); + + it("reprints everything when the transcript is not an extension of what was printed", () => { + const before = [user("one"), assistant("first")]; + const after = [user("two")]; + expect(run([view(before), view(after)])).toBe("user: one\nassistant: first\n--- view replaced ---\nuser: two\n"); + }); +}); diff --git a/spikes/durable-multiview/test/support.ts b/spikes/durable-multiview/test/support.ts new file mode 100644 index 00000000..37e19286 --- /dev/null +++ b/spikes/durable-multiview/test/support.ts @@ -0,0 +1,47 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { createServer } from "node:net"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { DurableViewSource } from "../protocol/view.ts"; + +export async function tempDir(): Promise<{ path: string; remove(): Promise }> { + const path = await mkdtemp(join(tmpdir(), "durable-multiview-")); + return { path, remove: () => rm(path, { recursive: true, force: true }) }; +} + +/** Resolve once `predicate` holds for the source's view, checked on every update. */ +export function waitForView( + source: DurableViewSource, + predicate: (view: ReturnType) => boolean, + timeoutMs = 20_000, +): Promise { + return new Promise((resolve, reject) => { + const check = (): boolean => { + if (!predicate(source.current())) return false; + clearTimeout(timer); + unsubscribe(); + resolve(); + return true; + }; + const timer = setTimeout(() => { + unsubscribe(); + reject(new Error(`view condition not met within ${timeoutMs} ms`)); + }, timeoutMs); + const unsubscribe = source.subscribe(() => void check()); + check(); + }); +} + +export function freePort(): Promise { + return new Promise((resolve, reject) => { + const server = createServer(); + server.once("error", reject); + server.listen(0, "127.0.0.1", () => { + const address = server.address(); + server.close(() => (typeof address === "object" && address ? resolve(address.port) : reject(new Error("no port")))); + }); + }); +} + +/** A reply long enough at the faux provider's pace to leave a window mid-stream. */ +export const LONG_ANSWER = Array.from({ length: 40 }, (_, i) => `step-${i}`).join(" "); diff --git a/spikes/durable-multiview/tsconfig.json b/spikes/durable-multiview/tsconfig.json new file mode 100644 index 00000000..f287e780 --- /dev/null +++ b/spikes/durable-multiview/tsconfig.json @@ -0,0 +1,16 @@ +{ + "compilerOptions": { + "target": "ES2023", + "lib": ["ES2023", "DOM"], + "module": "NodeNext", + "moduleResolution": "NodeNext", + "allowImportingTsExtensions": true, + "erasableSyntaxOnly": true, + "verbatimModuleSyntax": true, + "strict": true, + "skipLibCheck": true, + "noEmit": true, + "types": ["node"] + }, + "include": ["host", "protocol", "cli", "test", "vitest.config.ts"] +} diff --git a/spikes/durable-multiview/vitest.config.ts b/spikes/durable-multiview/vitest.config.ts new file mode 100644 index 00000000..90164608 --- /dev/null +++ b/spikes/durable-multiview/vitest.config.ts @@ -0,0 +1,11 @@ +import { defineConfig } from "vitest/config"; + +export default defineConfig({ + test: { + environment: "node", + include: ["test/**/*.test.ts"], + pool: "forks", + testTimeout: 60_000, + hookTimeout: 30_000, + }, +}); diff --git a/vite.config.ts b/vite.config.ts index f5302e37..28fb7040 100644 --- a/vite.config.ts +++ b/vite.config.ts @@ -33,6 +33,8 @@ export default defineConfig({ "**/.claude/worktrees/**", // Playwright drive scripts, not Vitest suites. ".cursor/**", + // Spikes run their own Node-environment suites. + "spikes/**", ], pool: "forks", // @lobehub/icons ships extensionless directory imports that Node's ESM