diff --git a/.scratch/durable-multiview/HANDOFF.md b/.scratch/durable-multiview/HANDOFF.md index a7f99996..23b6d467 100644 --- a/.scratch/durable-multiview/HANDOFF.md +++ b/.scratch/durable-multiview/HANDOFF.md @@ -1,7 +1,7 @@ # Handoff:Pi Durable 多端同屏 spike - 日期:2026-10-02 -- 状态:P0–P3 完成(见 §10–§13),下一步 P4(演示扩展与剧本)。PR 栈:P0 [#426](https://github.com/bubbleptr/pace/pull/426) ← P1 [#427](https://github.com/bubbleptr/pace/pull/427) ← P2 [#429](https://github.com/bubbleptr/pace/pull/429) ← P3 `feat/durable-multiview-pace`。前置的 Pi 1.0 升级 [PR #425](https://github.com/bubbleptr/pace/pull/425) 已合入,分支 `feat/durable-multiview-spike` 从 `origin/main` 开出。 +- 状态:P0–P3 完成(见 §10–§13),P4 拆成两段:P4a 宿主侧演示扩展与协议已完成(§14),下一步 P4b(三端界面与剧本验证)。PR 栈:P0 [#426](https://github.com/bubbleptr/pace/pull/426) ← P1 [#427](https://github.com/bubbleptr/pace/pull/427) ← P2 [#429](https://github.com/bubbleptr/pace/pull/429) ← P3 [#430](https://github.com/bubbleptr/pace/pull/430) ← P4a `feat/durable-multiview-demo`。前置的 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) - 目的:把上一个会话里已核实的事实、已定的决策和待决问题交给新会话,避免重查。 @@ -267,3 +267,42 @@ Pace 页面放在 `apps/desktop/src/pages/` 下的 dev-only 路由,backend 侧 - 渲染进程:一个传输适配器,加上把新事件类型列入忽略列表。 - 视图状态由渲染进程持有,ops 只应用一次,没有经过 Session Projection 和 Journal。这一点和 ADR-0044「渲染进程单写者投影」一致,但**和 ADR-0021 的双持久化轨道是并列的另一条路**:Durable 自己的 SQLite 才是真相,Pace 不再需要 Journal。结论文档要讨论这一点。 - 后端只做透传,所以 Durable 的背压(`watch()` 积压 100 帧后折叠)在宿主到后端这一段生效。后端到渲染进程这一段的 IPC 没有背压:慢渲染进程会在 Electron IPC 队列里积压。spike 里看不到问题,但真实接入要考虑。 + +## 14. P4a 进展(2026-10-02):宿主侧演示扩展与协议 + +P4 拆成两个 PR:P4a 是宿主扩展、网关和客户端协议,三端界面暂时不变;P4b 做三端界面(审批、待办、fork、思考块、窄屏、用量)和完整剧本的验证脚本。 + +### 14.1 落地的内容 + +- `host/demo/`(`node host/main.ts --demo` 走真模型,`--faux-demo` 走剧本模型;`bun run demo` / `bun run demo:faux`): + - `oncall.ts`:只给 main 和它的 fork 用的扩展。`playbook` section 把 §6 的流程写死给模型;`plan` section 把待办渲染进系统提示。工具有 `update_plan`(写 `demo.plan` 文档)、`subagent`(照示例 22,`replay: "safe"`,按 `area` 只给子会话对应的那一个调查工具,并去掉 `oncall` 扩展)、`rollback`、`schedule_check`。 + - **审批(§8.1 选了 a)**:`beforeTool` 钩子只拦 `rollback`。先读 memo,再读 `demo.approvals` 文档,都没有才去内存里的 `ApprovalBoard` 登记等待;拿到决定后写 memo。客户端点「批准」走网关的 `approve`,宿主在一次提交里 `decisions[id] ??= 决定`,先写者胜,后点的人收到「已被 X 批准」。等待随钩子的 `context` 取消(Esc)。 + - `tasks.ts`:`demo.rollback` 一次建 3 个区域子任务,`waiting` + `failFast`。`eu-west` 的健康检查必失败,另外两个区域正在切换时被中止,在 abort 处理里「恢复流量」作为补偿(自下而上,补偿完才落 `aborted`)。进度写进 `demo.rollout` 文档,`rollback` 工具轮询它作为运行中输出。`rollback` 也是 `replay: "safe"`:在同一次提交里先按 kind 扫描任务、按 `owner` 过滤,找到就复用,所以崩溃后不会再建第二个回滚。 + - `demo.reminder`:`schedule_check` 建的 background 任务,`until` 是绝对时间存在输入里,`runtime.sleep` 到点后以 `requestId: reminder:` 发 follow-up。 + - `investigation.ts`:三个模拟调查工具,逐行慢速 `api.output`。按路径加载,`Demo.reload()` 用带 `?v=N` 的 URL 重新 import,`registry.install` 同名替换。`main.ts` 在 `--demo` 下监听这个文件,保存后自动热替换并通知三端。改 `const FORMAT = "plain"` 为 `"table"` 就能看到下一次输出变成表格。 + - `faux-script.ts`:剧本模型。主会话和三个并行子会话的请求交错到达,所以每次都只按请求本身决定下一步:请求里提供了哪些工具,以及最新一条用户输入之后的消息。 +- 网关与协议: + - 新流 `approvals`(待审批列表)和 `doc::`(`hello.docs` 列出可订阅的文档 kind)。 + - 新调用 `approve`、`fork`(`removeTools` 按名字从父会话当前提供的工具里去掉)。 + - 会话标签区分 `subagent N` / `fork N`。fork 的标题取它自己的第一条用户消息,不取继承来的。 + - `DurableView` 多了 `docs`(当前会话的扩展文档)和 `approvals`;`DurableController` 多了 `approve(approval, approved)`、`fork(entryId, prompt, removeTools?)`。客户端有 `clientName`,审批记录里的 `by` 就是它。 +- `openHost` 多了 `approvals`、`docs`、`onOpen`,返回值多了 `notify`。演示的 `onOpen` 给 main 去掉三个调查工具,逼它委派给子代理。 + +### 14.2 验收 + +- `test/demo.test.ts`(6 条,剧本模型 + 进程内宿主): + - 待办同步;3 个子代理同时在跑(任务图里 3 个调用同时挂着子会话);子会话只提供自己那个工具。 + - 两端同时看到审批;A 批准、B 拒绝,B 收到「已被 web 批准」;回滚结果是 `eu-west` 失败、另外两个区域补偿。 + - 批准后、回滚进行中关掉宿主再开:不再弹审批,也只有一次回滚。 + - `schedule_check` 之后 Esc,再重启宿主:提醒照样以 follow-up 到达,只到一次。 + - 在答案处 fork 并去掉 `rollback`:会话列表出现 `fork N`,fork 里的待办是 fork 那一刻的(`asOf`),工具里没有 `rollback`。 + - 热替换后,下一次子代理的工具输出变成表格。 +- `test/crash-recovery.test.ts` 新增一条:宿主作为子进程,3 个子代理都在跑调查工具时 `kill -9`,重启后会话列表里仍是同样的 3 个子会话,调查完成并给出建议。 +- 真模型冒烟(`radius/deepseek-v4.1-flash`,`--demo`):模型按 playbook 写了待办,一次派出 3 个子代理(任务图里同时 3 个),汇总后建议回滚到 v2.2。说「回滚」后弹出审批,脚本批准;`eu-west` 失败,另外两个区域补偿。第一次跑时模型因为回滚失败没有安排复查,playbook 已改成「不论成败都 `schedule_check`」。 + +### 14.3 新发现 + +- **section 的变化会以 system 消息插在工具结果后面。** `update_plan` 改了待办,下一次请求里,`toolResult` 之后会多一条只带 `sections` 的 system 消息。剧本模型取「最后一条消息」时要跳过它;Pace 渲染转录时也要决定这类 `pi.system` 条目显示不显示。 +- **`ConversationView.docs` 只有 4 个内置文档**,扩展文档要单独 `watchDoc`。`watchDoc` 不会创建文档,文档不存在时返回 `undefined`;网关先发 `null`,再等创建它的那次提交(`document` 或 fork 时的 `document.copy` 变更)到来后挂上 watch。 +- **钩子不能提交,所以「谁在等审批」不是持久状态。** 决定是持久的(文档 + memo),等待只在宿主内存里;宿主重启后钩子重跑,会重新登记。这一点对 Pace 有影响:待审批列表要么走 spike 这样的旁路流,要么改成 §8.1 的 b)方案,用工具加任务把审批变成持久状态。 +- `TaskQuery` 没有按 owner 过滤的字段,只能按 `conversationId` + `kind` 扫,再按 `owner` 过滤。 diff --git a/spikes/durable-multiview/.gitignore b/spikes/durable-multiview/.gitignore new file mode 100644 index 00000000..3018b3a6 --- /dev/null +++ b/spikes/durable-multiview/.gitignore @@ -0,0 +1 @@ +.tmp/ diff --git a/spikes/durable-multiview/host/demo/approvals.ts b/spikes/durable-multiview/host/demo/approvals.ts new file mode 100644 index 00000000..df30462a --- /dev/null +++ b/spikes/durable-multiview/host/demo/approvals.ts @@ -0,0 +1,92 @@ +import type { Context } from "@earendil-works/chord"; +import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; +import type { ConversationId, Harness } from "@earendil-works/pi-durable"; +import type { ApprovalDecision, PendingApproval } from "../../protocol/demo.ts"; +import { ApprovalsDoc } from "./state.ts"; + +type Waiter = { readonly request: PendingApproval; resolve(decision: ApprovalDecision): void }; + +/** + * Who is waiting for a decision. A hook can neither commit nor sleep durably, so the waiting + * itself lives here, in memory; a restarted host's hook runs again and asks again. Decisions + * are committed to `ApprovalsDoc`, which is what makes them stand. + */ +export class ApprovalBoard { + readonly #waiters = new Map(); + readonly #decided = new Map(); + readonly #listeners = new Set<(pending: readonly PendingApproval[]) => void>(); + #harness: Harness | undefined; + + attach(harness: Harness): void { + this.#harness = harness; + } + + get value(): readonly PendingApproval[] { + return [...this.#waiters.values()].map((waiters) => waiters[0]!.request); + } + + subscribe(listener: (pending: readonly PendingApproval[]) => void): () => void { + this.#listeners.add(listener); + return () => this.#listeners.delete(listener); + } + + /** Resolves with the decision; rejects when `context` is cancelled, for example by Esc. */ + wait(request: PendingApproval, context: Context): Promise { + const decided = this.#decided.get(request.id); + if (decided !== undefined) return Promise.resolve(decided); + return new Promise((resolve, reject) => { + const signal = context.abortSignal; + const waiter: Waiter = { + request, + resolve: (decision) => { + signal?.removeEventListener("abort", cancel); + resolve(decision); + }, + }; + const cancel = (): void => { + this.#remove(waiter); + reject(signal?.reason ?? new Error("Approval wait cancelled")); + }; + if (signal?.aborted) return cancel(); + signal?.addEventListener("abort", cancel, { once: true }); + this.#waiters.set(request.id, [...(this.#waiters.get(request.id) ?? []), waiter]); + this.#changed(); + }); + } + + /** Commit a decision unless one exists; returns the one that stands and whether it is this one. */ + async decide( + conversationId: ConversationId, + id: string, + approved: boolean, + by: string, + ): Promise<{ decision: ApprovalDecision; first: boolean }> { + const harness = this.#harness; + if (harness === undefined) throw new Error("Approvals are not attached to a Harness"); + const candidate: ApprovalDecision = { approved, by, at: Date.now() }; + const decision = await harness.commit(async (tx) => { + const doc = await tx.doc(ApprovalsDoc, conversationId); + if (doc.decisions[id] === undefined && !this.#waiters.has(id)) throw new Error(`No pending approval ${id}`); + doc.decisions[id] ??= { ...candidate }; + return { ...doc.decisions[id] } as ApprovalDecision; + }, BACKGROUND_CONTEXT); + this.#decided.set(id, decision); + const waiters = this.#waiters.get(id) ?? []; + this.#waiters.delete(id); + for (const waiter of waiters) waiter.resolve(decision); + if (waiters.length > 0) this.#changed(); + return { decision, first: decision.by === candidate.by && decision.at === candidate.at }; + } + + #remove(waiter: Waiter): void { + const rest = (this.#waiters.get(waiter.request.id) ?? []).filter((candidate) => candidate !== waiter); + if (rest.length > 0) this.#waiters.set(waiter.request.id, rest); + else this.#waiters.delete(waiter.request.id); + this.#changed(); + } + + #changed(): void { + const pending = this.value; + for (const listener of this.#listeners) listener(pending); + } +} diff --git a/spikes/durable-multiview/host/demo/faux-script.ts b/spikes/durable-multiview/host/demo/faux-script.ts new file mode 100644 index 00000000..e6315823 --- /dev/null +++ b/spikes/durable-multiview/host/demo/faux-script.ts @@ -0,0 +1,86 @@ +// A scripted stand-in for a real model that walks the on-call playbook. Requests of the main +// conversation and its parallel subagents arrive interleaved, so every answer is decided from +// the request itself: the tools it offers and the messages since the newest user input. +import type { Message } from "@earendil-works/pi-ai"; +import { fauxAssistantMessage, type FauxResponseFactory, fauxText, fauxToolCall } from "@earendil-works/pi-ai/providers/faux"; +import { AREAS } from "./oncall.ts"; + +const INVESTIGATION: readonly string[] = Object.values(AREAS).map((area) => area.tool); + +function textOf(message: Message | undefined): string { + if (message === undefined || message.role === "system") return ""; + if (typeof message.content === "string") return message.content; + return message.content.flatMap((block) => (block.type === "text" ? [block.text] : [])).join(""); +} + +/** The tools a request offers, from its system messages' additions and removals. */ +function offeredTools(messages: readonly Message[]): string[] { + const tools = new Set(); + for (const message of messages) { + if (message.role !== "system") continue; + for (const tool of message.toolsAdded ?? []) tools.add(tool.name); + for (const tool of message.toolsRemoved ?? []) tools.delete(tool.name); + } + return [...tools]; +} + +const PLAN = ["Search the deploy logs", "Compare error metrics", "Review the commits in v2.3"]; + +export function oncallScript({ reminderSeconds = 60 }: { reminderSeconds?: number } = {}): FauxResponseFactory { + return (context, _options, state) => { + const { messages } = context; + const call = (name: string, args: Parameters[1], index = 0) => + fauxToolCall(name, args, { id: `call-${state.callCount}-${index}` }); + const calls = (...blocks: ReturnType[]) => fauxAssistantMessage(blocks, { stopReason: "toolUse" }); + const tools = offeredTools(messages); + const userAt = messages.findLastIndex((message) => message.role === "user"); + const input = textOf(messages[userAt]); + // A changed section, such as the plan after update_plan, lands as a system message after the results. + const last = messages.slice(userAt + 1).findLast((message) => message.role !== "system"); + + // A subagent: one investigation tool, called once, then reported on. + const own = tools.length === 1 && INVESTIGATION.includes(tools[0]!) ? tools[0]! : undefined; + if (own !== undefined) { + if (last?.role !== "toolResult") return calls(call(own, { query: input })); + const first = textOf(last).split("\n").find((line) => /ERROR|->|pg driver/.test(line)) ?? textOf(last).split("\n")[0]; + return fauxAssistantMessage(`Found in ${own}: ${first}`); + } + + if (last?.role === "toolResult") { + if (last.isError) return fauxAssistantMessage(`${last.toolName} did not run: ${textOf(last)} Standing by.`); + switch (last.toolName) { + case "update_plan": + if (messages.some((message) => message.role === "toolResult" && message.toolName === "subagent")) { + return fauxAssistantMessage( + "The pg driver bump in a1b2c3d cut the pool from 100 to 20; eu-west exhausted it at 09:14 and errors went to 14.8%. I recommend rolling back to v2.2.", + ); + } + return calls( + call("subagent", { area: "logs", task: "find the errors around the v2.3 deploy in the logs" }, 0), + call("subagent", { area: "metrics", task: "compare error metrics before and after v2.3" }, 1), + call("subagent", { area: "commits", task: "review the commits in v2.3 for risky changes" }, 2), + ); + case "subagent": + return calls(call("update_plan", { items: PLAN.map((text) => ({ text, status: "done" })) })); + case "rollback": + return calls(call("schedule_check", { seconds: reminderSeconds, note: "verify the error rate after the rollback" })); + case "schedule_check": + return fauxAssistantMessage("Check scheduled; I will re-check when the reminder arrives."); + default: + return fauxAssistantMessage("Done."); + } + } + + if (input.startsWith("Reminder:")) return fauxAssistantMessage("Re-checked: the error rate is back to baseline (0.2%) in every region."); + if (/postmortem/i.test(input)) { + return fauxAssistantMessage([fauxText("Postmortem: v2.3 shipped a pg driver bump that shrank the connection pool; eu-west saturated first.")]); + } + if (/roll ?back/i.test(input)) return calls(call("rollback", { version: "v2.2", reason: "pg driver pool regression in v2.3" })); + const delay = /check again in (\d+) seconds?/i.exec(input); + if (delay !== null) return calls(call("schedule_check", { seconds: Number(delay[1]), note: "re-check the error rate" })); + if (tools.includes("update_plan") && !messages.some((message) => message.role === "toolResult" && message.toolName === "update_plan")) { + return calls(call("update_plan", { items: PLAN.map((text) => ({ text, status: "todo" })) })); + } + return fauxAssistantMessage(`Noted: ${input}`); + }; +} diff --git a/spikes/durable-multiview/host/demo/index.ts b/spikes/durable-multiview/host/demo/index.ts new file mode 100644 index 00000000..2f7bad88 --- /dev/null +++ b/spikes/durable-multiview/host/demo/index.ts @@ -0,0 +1,63 @@ +import { fileURLToPath, pathToFileURL } from "node:url"; +import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; +import { createRegistry, type Extension, type Registry } from "@earendil-works/pi-durable"; +import type { OpenHostOptions } from "../host.ts"; +import { ApprovalBoard } from "./approvals.ts"; +import { AREAS, createOncall } from "./oncall.ts"; +import { PlanDoc, RolloutDoc } from "./state.ts"; +import { DemoTasks } from "./tasks.ts"; + +export interface DemoOptions { + /** Delay between streamed lines of an investigation tool's output. */ + readonly paceMs?: number; + /** Length of one rollback step in a region. */ + readonly stepMs?: number; + /** The investigation tools' module; `reload()` imports it again. */ + readonly investigationModule?: string; +} + +export interface Demo { + readonly registry: Registry; + readonly investigationModule: string; + /** What `openHost` needs to run the demo: its registry, approvals, documents, and root setup. */ + readonly hostOptions: Pick; + /** Reinstall the investigation tools from their module; returns the module's load count. */ + reload(): Promise; +} + +const defaultModule = fileURLToPath(new URL("./investigation.ts", import.meta.url)); + +export async function createDemo(options: DemoOptions = {}): Promise { + const { paceMs = 400, stepMs = 2000, investigationModule = defaultModule } = options; + const registry = createRegistry(); + const approvals = new ApprovalBoard(); + let loads = 0; + const load = async (): Promise => { + // A fresh URL is a fresh module instance; the registry replaces the extension of the same name in place. + const module = (await import(`${pathToFileURL(investigationModule).href}?v=${++loads}`)) as typeof import("./investigation.ts"); + return module.createInvestigation({ paceMs }); + }; + registry.install(DemoTasks); + registry.install(await load()); + registry.install(createOncall({ approvals, stepMs })); + return { + registry, + investigationModule, + hostOptions: { + registry, + approvals, + docs: [PlanDoc, RolloutDoc], + async onOpen(harness, root) { + approvals.attach(harness); + // The main agent delegates instead of investigating itself; tools are stored by name, so this survives reloads. + const investigation = registry.snapshot().extension("investigation")!.tools ?? []; + const own = investigation.filter((tool) => Object.values(AREAS).some((area) => area.tool === tool.name)); + await root.configure({ tools: { remove: own } }, BACKGROUND_CONTEXT); + }, + }, + async reload() { + registry.install(await load()); + return loads; + }, + }; +} diff --git a/spikes/durable-multiview/host/demo/investigation.ts b/spikes/durable-multiview/host/demo/investigation.ts new file mode 100644 index 00000000..29bde4d5 --- /dev/null +++ b/spikes/durable-multiview/host/demo/investigation.ts @@ -0,0 +1,68 @@ +// The investigation tools, loaded by path so the host can reinstall them while it runs +// (`Demo.reload()`): edit FORMAT below and the next call prints the other way. Imports +// packages only, so a copy of this file anywhere in the spike loads the same. +import { setTimeout as sleep } from "node:timers/promises"; +import type { Context } from "@earendil-works/chord"; +import { Type } from "@earendil-works/pi-ai"; +import { defineExtension, defineTool, type Extension, type ToolExecutionApi } from "@earendil-works/pi-durable"; + +const FORMAT = "plain" as "plain" | "table"; + +type Row = readonly [string, string, string]; + +const LOGS: readonly Row[] = [ + ["09:12:41", "deploy", "v2.3.0 rolled out to us-east, eu-west, ap-south"], + ["09:14:03", "eu-west", "ERROR api: connection pool exhausted (max=20, waiting=183)"], + ["09:14:05", "eu-west", "ERROR checkout: upstream timeout after 30s"], + ["09:14:20", "us-east", "WARN api: pool wait p99 2.1s"], + ["09:15:02", "deploy", "health check failed, release v2.3.0 marked FAILED"], +]; + +const METRICS: readonly Row[] = [ + ["error rate", "0.2% -> 14.8%", "since 09:14 in eu-west"], + ["db connections", "20/20", "saturated in every region"], + ["p99 latency", "180ms -> 30s", "checkout path"], + ["cpu", "41%", "unchanged"], +]; + +const COMMITS: readonly Row[] = [ + ["a1b2c3d", "deps", "bump pg driver 8.11 -> 9.0 (pool default 100 -> 20)"], + ["e4f5a6b", "checkout", "retry upstream calls 3x"], + ["c7d8e9f", "docs", "update release notes"], +]; + +function render(header: Row, rows: readonly Row[]): string[] { + if (FORMAT === "table") return [`| ${header.join(" | ")} |`, ...rows.map((row) => `| ${row.join(" | ")} |`)]; + return rows.map((row) => row.join(" ")); +} + +/** Streams the rows one by one, so a crash or a reload can land mid-call. */ +async function stream(api: ToolExecutionApi, lines: readonly string[], paceMs: number, context: Context): Promise { + for (const line of lines) { + await sleep(paceMs, undefined, context.abortSignal === undefined ? {} : { signal: context.abortSignal }); + api.output(`${line}\n`); + } +} + +export function createInvestigation(options: { readonly paceMs: number }): Extension { + const tool = (name: string, description: string, header: Row, rows: readonly Row[]) => + defineTool({ + name, + description, + parameters: Type.Object({ query: Type.String({ description: "What to look for" }) }), + // Read-only, so a rerun after a crash just reads again. + replay: "safe", + execute: async (_args, api, context) => { + await stream(api, render(header, rows), options.paceMs, context); + return {}; + }, + }); + return defineExtension({ + name: "investigation", + tools: [ + tool("search_logs", "Search the production logs of the v2.3 release.", ["time", "source", "message"], LOGS), + tool("query_metrics", "Query production metrics around the v2.3 release.", ["metric", "change", "where"], METRICS), + tool("list_commits", "List the commits that went into v2.3.", ["sha", "area", "message"], COMMITS), + ], + }); +} diff --git a/spikes/durable-multiview/host/demo/oncall.ts b/spikes/durable-multiview/host/demo/oncall.ts new file mode 100644 index 00000000..3bf423c2 --- /dev/null +++ b/spikes/durable-multiview/host/demo/oncall.ts @@ -0,0 +1,193 @@ +// The on-call agent's own extension, selected by the main conversation and its forks, not by +// subagents. The subagent tool follows pi-durable's test/examples/22-subagent-foreground.ts (MIT). +import { setTimeout as sleep } from "node:timers/promises"; +import type { Context } from "@earendil-works/chord"; +import { type AssistantMessage, Type } from "@earendil-works/pi-ai"; +import { + AssistantEntry, + configure, + defineExtension, + defineTool, + type EntryId, + type Extension, + hook, + section, + type TaskId, + type ToolExecutionApi, + ToolTask, +} from "@earendil-works/pi-durable"; +import type { ApprovalDecision } from "../../protocol/demo.ts"; +import type { ApprovalBoard } from "./approvals.ts"; +import { ApprovalsDoc, PlanDoc, RolloutDoc } from "./state.ts"; +import { Reminder, Rollback, type RollbackReport } from "./tasks.ts"; + +export const ONCALL = "oncall"; + +export const AREAS = { + logs: { tool: "search_logs", focus: "the production logs" }, + metrics: { tool: "query_metrics", focus: "the production metrics" }, + commits: { tool: "list_commits", focus: "the commits in the release" }, +} as const; +type Area = keyof typeof AREAS; + +const PLAYBOOK = `You are the on-call agent investigating the failed v2.3 release. Work in this order: +1. Call update_plan with three items: "Search the deploy logs", "Compare error metrics", "Review the commits in v2.3". +2. Call the subagent tool three times in one response, once per area (logs, metrics, commits), so the three run in parallel. Do not investigate yourself. +3. Call update_plan to mark the items done, then summarize the findings in a few sentences and end with a recommendation. +4. Only when the user asks for a rollback, call rollback. A human must approve it; if it is denied, say so and stop. +5. After any rollback attempt, whether it succeeded or failed, call schedule_check with seconds=60 to re-check the error rate, then report. +When a message starts with "Reminder:", re-check briefly (you may delegate to the metrics subagent) and report.`; + +const childInstructions = (area: Area): string => + `You investigate ${AREAS[area].focus} for the failed v2.3 release. Call ${AREAS[area].tool} once, then report what matters in two sentences.`; + +async function answerText(api: ToolExecutionApi, answer: EntryId, context: Context): Promise { + const entry = await api.commit((tx) => tx.entry(AssistantEntry, answer), context); + const message = entry?.model?.[0] as AssistantMessage | undefined; + return message?.content.flatMap((block) => (block.type === "text" ? [block.text] : [])).join("") ?? ""; +} + +const updatePlan = defineTool({ + name: "update_plan", + description: "Replace the investigation plan shown to everyone watching.", + parameters: Type.Object({ + items: Type.Array( + Type.Object({ text: Type.String(), status: Type.Union([Type.Literal("todo"), Type.Literal("doing"), Type.Literal("done")]) }), + ), + }), + execute: async (args, api, context) => { + // The plan changes in the same commit as anything else the call writes. + await api.commit(async (tx) => { + (await tx.doc(PlanDoc, api.conversationId)).items = args.items.map((item) => ({ ...item })); + }, context); + return { content: [{ type: "text", text: `Plan updated: ${args.items.length} items.` }] }; + }, +}); + +const subagent = defineTool({ + name: "subagent", + description: "Delegate one area of the investigation to a subagent and get its findings back. Call it once per area, in one response.", + parameters: Type.Object({ + area: Type.Union([Type.Literal("logs"), Type.Literal("metrics"), Type.Literal("commits")]), + task: Type.String({ description: "What the subagent should find out" }), + }), + // A rerun after a crash finds the child it created and the submission it made. + replay: "safe", + execute: async (args, api, context) => { + const area = args.area as Area; + const child = await api.commit(async (tx) => { + const existing = (await tx.scanConversations({ ownerTaskId: api.taskId }, 1)).items[0]; + if (existing !== undefined) return existing.id; + const created = await tx.createConversation({ ownership: { kind: "task", taskId: api.taskId } }); + const tool = api.registry.tools().find((entry) => entry.tool.name === AREAS[area].tool)?.tool; + if (tool === undefined) throw new Error(`No ${AREAS[area].tool} tool installed`); + await configure(tx, created.id, { + extensions: { remove: [api.registry.extension(ONCALL)!] }, + tools: [tool], + instructions: childInstructions(area), + }); + return created.id; + }, context); + await api.details({ conversationId: child }, context); + const handle = (await api.conversation(child, context))!; + const request = { type: "input", content: `${args.area}: ${args.task}`, requestId: `subagent:${api.taskId}` } as const; + const settled = await (await handle.submit(request, context)).wait(context); + if (settled.status !== "done" || settled.type !== "input") throw new Error(`Subagent failed: ${settled.status}`); + return { content: [{ type: "text", text: await answerText(api, settled.answer, context) }], details: { conversationId: child } }; + }, +}); + +function rollbackTool(stepMs: number) { + return defineTool({ + name: "rollback", + description: "Roll production back to an earlier version in every region. Needs a human approval.", + parameters: Type.Object({ version: Type.String(), reason: Type.String() }), + // A rerun after a crash waits for the rollback it started instead of starting another. + replay: "safe", + execute: async (args, api, context) => { + const id = await api.commit(async (tx) => { + const page = await tx.scanTasks({ conversationId: api.conversationId, kind: Rollback.definition.name }, 256); + const existing = page.items.find((task) => task.owner === api.taskId); + if (existing !== undefined) return existing.id as TaskId; + return tx.createTask(Rollback, { version: args.version, stepMs }, { ownership: { kind: "task", taskId: api.taskId } }); + }, context); + const settled = api.waitForTask(id, context); + // Region progress, as the region tasks commit it, becomes the call's running output. + let shown: Record = {}; + let done = false; + void settled.finally(() => (done = true)).catch(() => {}); + while (!done) { + const regions = (await api.snapshot(RolloutDoc, api.conversationId, context))?.runs[String(id)] ?? {}; + for (const [region, status] of Object.entries(regions)) if (shown[region] !== status) api.output(`${region}: ${status}\n`); + shown = { ...regions }; + await Promise.race([settled.catch(() => {}), sleep(Math.max(50, stepMs / 4))]); + } + const { state } = await settled; + if (state.outcome.status !== "completed") throw new Error(`Rollback ${state.outcome.status}`); + const report = state.outcome.result; + const lines = report.regions.map((region) => `${region.region}: ${region.outcome}${region.error === undefined ? "" : ` (${region.error})`}`); + const ok = report.regions.every((region) => region.outcome === "completed"); + return { + content: [{ type: "text", text: `Rollback to ${report.version} ${ok ? "succeeded" : "failed; drained regions restored"}.\n${lines.join("\n")}` }], + details: { rollbackTaskId: String(id) }, + }; + }, + }); +} + +const scheduleCheck = defineTool({ + name: "schedule_check", + description: "Schedule a re-check that arrives as a message starting with \"Reminder:\" after the given delay, even if the run is stopped.", + parameters: Type.Object({ seconds: Type.Number({ minimum: 1 }), note: Type.String() }), + execute: async (args, api, context) => { + // Background: Esc on this conversation aborts its ordinary work, not this. + await api.createTask( + Reminder, + { until: Date.now() + args.seconds * 1000, note: args.note }, + { ownership: { kind: "conversation" }, background: true }, + context, + ); + return { content: [{ type: "text", text: `Check scheduled in ${args.seconds}s: ${args.note}` }] }; + }, +}); + +export function createOncall(options: { readonly approvals: ApprovalBoard; readonly stepMs: number }): Extension { + return defineExtension({ + name: ONCALL, + tools: [updatePlan, subagent, rollbackTool(options.stepMs), scheduleCheck], + sections: [ + section("playbook", () => PLAYBOOK), + section("plan", async (input, context) => { + const plan = await input.read.snapshot(PlanDoc, input.conversationId, context); + return plan === undefined || plan.items.length === 0 ? undefined : plan.items.map((item) => `- [${item.status}] ${item.text}`).join("\n"); + }), + ], + hooks: [ + hook(ToolTask, { + // Hooks cannot sleep durably, so the decision is memoized: a rerun after a crash reads it instead of asking again. + beforeTool: async (call, api, context) => { + if (call.name !== "rollback") return undefined; + const id = String(api.taskId); + const known = + (await api.memo("approval", context)) ?? + (await api.snapshot(ApprovalsDoc, api.conversationId, context))?.decisions[id]; + const args = call.arguments as { version?: string; reason?: string }; + const decided = + known ?? + (await options.approvals.wait( + { + id, + conversationId: api.conversationId, + tool: call.name, + summary: `Roll back to ${args.version ?? "?"}: ${args.reason ?? ""}`, + requestedAt: Date.now(), + }, + context, + )); + const decision = await api.memo("approval", { ...decided }, context); + return decision.approved ? undefined : { block: `Rollback denied by ${decision.by}.` }; + }, + }), + ], + }); +} diff --git a/spikes/durable-multiview/host/demo/state.ts b/spikes/durable-multiview/host/demo/state.ts new file mode 100644 index 00000000..28e3b815 --- /dev/null +++ b/spikes/durable-multiview/host/demo/state.ts @@ -0,0 +1,33 @@ +import { defineDoc } from "@earendil-works/pi-durable"; +import type { ApprovalDecision, PlanState, RolloutState } from "../../protocol/demo.ts"; + +type Mutable = { -readonly [K in keyof T]: T[K] extends object ? Mutable : T[K] }; + +/** Rewindable, so a fork starts with the plan its parent had at the fork entry. */ +export const PlanDoc = defineDoc>({ + kind: "demo.plan", + version: 1, + scope: "conversation", + history: "rewindable", + fork: "asOf", + initial: () => ({ items: [] }), +}); + +export const RolloutDoc = defineDoc>({ + kind: "demo.rollout", + version: 1, + scope: "conversation", + history: "latest", + fork: "initial", + initial: () => ({ runs: {} }), +}); + +/** Decisions keyed by the held call's task ID. The first one committed stands. */ +export const ApprovalsDoc = defineDoc<{ decisions: Record> }>({ + kind: "demo.approvals", + version: 1, + scope: "conversation", + history: "latest", + fork: "initial", + initial: () => ({ decisions: {} }), +}); diff --git a/spikes/durable-multiview/host/demo/tasks.ts b/spikes/durable-multiview/host/demo/tasks.ts new file mode 100644 index 00000000..6acded13 --- /dev/null +++ b/spikes/durable-multiview/host/demo/tasks.ts @@ -0,0 +1,132 @@ +import { type ConversationId, defineExtension, defineTask, type TaskId, type Tx } from "@earendil-works/pi-durable"; +import { RolloutDoc } from "./state.ts"; + +/** The region whose health check always fails, so every rollback shows failFast and compensation. */ +export const FAILING_REGION = "eu-west"; +export const REGIONS = ["us-east", FAILING_REGION, "ap-south"] as const; + +async function setRegion(tx: Tx, conversationId: ConversationId, rollout: string, region: string, status: string): Promise { + const doc = await tx.doc(RolloutDoc, conversationId); + doc.runs[rollout] = { ...doc.runs[rollout], [region]: status }; +} + +type RegionInput = { rollout: string; region: string; version: string; stepMs: number }; +type RegionState = { phase: "drain" } | { phase: "verify"; until: number } | { phase: "finish"; until: number }; + +/** Drain, health-check, and switch one region. Aborted, it restores the traffic it drained. */ +export const RegionRollback = defineTask({ + name: "demo.region-rollback", + version: 1, + initial: () => ({ phase: "drain" }), + phases: { + drain: async (task, runtime, context) => { + const { rollout, region, stepMs } = task.input; + // The failing check takes longer, so its siblings are mid-switch when it fails. + const until = runtime.now() + (region === FAILING_REGION ? stepMs * 1.5 : stepMs); + await runtime.commit(async (tx) => { + await setRegion(tx, task.conversationId, rollout, region, "draining"); + return { status: "running", checkpoint: { phase: "verify", until } }; + }, context); + }, + verify: async (task, runtime, context) => { + const { rollout, region, version, stepMs } = task.input; + await runtime.sleep(task.state.checkpoint.until, context); + if (region === FAILING_REGION) { + await runtime.commit(async (tx) => { + await setRegion(tx, task.conversationId, rollout, region, "failed: health check"); + return { status: "terminal", outcome: { status: "failed", error: { message: `${region}: health check failed on ${version}` } } }; + }, context); + return; + } + await runtime.commit(async (tx) => { + await setRegion(tx, task.conversationId, rollout, region, `switching to ${version}`); + return { status: "running", checkpoint: { phase: "finish", until: runtime.now() + stepMs * 2 } }; + }, context); + }, + finish: async (task, runtime, context) => { + const { rollout, region, version } = task.input; + await runtime.sleep(task.state.checkpoint.until, context); + await runtime.commit(async (tx) => { + await setRegion(tx, task.conversationId, rollout, region, `rolled back to ${version}`); + return { status: "terminal", outcome: { status: "completed", result: { region } } }; + }, context); + }, + }, + // Runs once the region's own work is done; the compensation is this region's to undo. + abort: async (task, runtime, context) => { + const { rollout, region, stepMs } = task.input; + await runtime.commit(async (tx) => { + await setRegion(tx, task.conversationId, rollout, region, "compensating: restoring traffic"); + return undefined; + }, context); + await runtime.sleep(runtime.now() + stepMs, context); + await runtime.commit(async (tx) => { + await setRegion(tx, task.conversationId, rollout, region, "compensated: traffic restored"); + return { status: "terminal", outcome: { status: "aborted" } }; + }, context); + }, +}); + +export type RollbackReport = { version: string; regions: { region: string; outcome: string; error?: string }[] }; +type RollbackState = { phase: "start" } | { phase: "report"; regions: TaskId<{ region: string }>[] }; + +/** Rolls every region back in parallel; the first failed region aborts the others. */ +export const Rollback = defineTask<{ version: string; stepMs: number }, RollbackState, RollbackReport>({ + name: "demo.rollback", + version: 1, + initial: () => ({ phase: "start" }), + phases: { + start: async (task, runtime, context) => { + await runtime.commit(async (tx) => { + const regions: TaskId<{ region: string }>[] = []; + for (const region of REGIONS) { + const input = { rollout: String(task.id), region, version: task.input.version, stepMs: task.input.stepMs }; + regions.push(await tx.createTask(RegionRollback, input, { ownership: { kind: "task", taskId: task.id } })); + } + return { status: "waiting", checkpoint: { phase: "report", regions }, on: regions, policy: "failFast" }; + }, context); + }, + report: async (task, runtime, context) => { + const outcomes = await runtime.outcomes(task.state.checkpoint.regions, context); + const regions = outcomes.map((outcome, index) => ({ + region: REGIONS[index]!, + outcome: outcome.status, + ...(outcome.status === "failed" ? { error: outcome.error.message } : {}), + })); + await runtime.commit(() => ({ status: "terminal", outcome: { status: "completed", result: { version: task.input.version, regions } } }), context); + }, + }, + abort: async (_task, runtime, context) => { + await runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted" } }), context); + }, +}); + +type ReminderInput = { until: number; note: string }; + +/** + * A background check: Esc on its conversation does not reach it. It sleeps to an absolute time + * kept in its input, so a restart neither loses nor restarts the wait. + */ +export const Reminder = defineTask({ + name: "demo.reminder", + version: 1, + initial: () => ({ phase: "wait" }), + phases: { + wait: async (task, runtime, context) => { + await runtime.sleep(task.input.until, context); + const conversation = (await runtime.conversation(task.conversationId, context))!; + // The request ID makes a rerun after a crash find the follow-up it already posted. + await conversation.submit( + { type: "input", content: `Reminder: ${task.input.note}`, whenBusy: "followUp", requestId: `reminder:${task.id}` }, + context, + ); + await runtime.commit(() => ({ status: "terminal", outcome: { status: "completed", result: null } }), context); + }, + }, + abort: async (_task, runtime, context) => { + await runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted" } }), context); + }, +}); + +/** Task code is looked up by name in the registry, whichever conversation selects what. */ +export const DemoTasks = defineExtension({ name: "demo-tasks", tasks: [RegionRollback, Rollback, Reminder] }); diff --git a/spikes/durable-multiview/host/gateway.ts b/spikes/durable-multiview/host/gateway.ts index e45cc206..e2b0dbc1 100644 --- a/spikes/durable-multiview/host/gateway.ts +++ b/spikes/durable-multiview/host/gateway.ts @@ -8,14 +8,19 @@ import { AgentDoc, type AgentState, type Conversation, + type ConversationDocToken, type ConversationId, + type ConversationRecord, type Cursor, + type EntryId, type EntryRecord, type Harness, + type JsonObject, ROOT_CONVERSATION_ID, type WatchHandle, } from "@earendil-works/pi-durable"; import { WebSocket, WebSocketServer } from "ws"; +import type { ApprovalDecision, PendingApproval } from "../protocol/demo.ts"; import { type CallMethod, type CallMethods, @@ -28,6 +33,13 @@ import type { ConversationSummary, ModelSummary, Notice, SessionInfo } from "../ const context: Context = BACKGROUND_CONTEXT; +/** Tool calls held for a human decision, and the way to decide them. */ +export interface ApprovalSource { + readonly value: readonly PendingApproval[]; + subscribe(listener: (pending: readonly PendingApproval[]) => void): () => void; + decide(conversationId: ConversationId, id: string, approved: boolean, by: string): Promise<{ decision: ApprovalDecision; first: boolean }>; +} + export interface GatewayOptions { readonly harness: Harness; readonly models: Models; @@ -35,6 +47,9 @@ export interface GatewayOptions { readonly session: SessionInfo; readonly token: string; readonly port: number; + readonly approvals?: ApprovalSource; + /** Conversation documents offered as `doc::` streams. */ + readonly docs?: readonly ConversationDocToken[]; } export interface Gateway { @@ -96,7 +111,7 @@ class ConversationList { 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)) }); + for (const record of page.items) summaries.push({ id: record.id, label: labelOf(record), ...(await firstInput(harness, record.id)) }); cursor = page.next; } while (cursor !== undefined); const list = new ConversationList(summaries); @@ -105,7 +120,7 @@ class ConversationList { 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) }]; + next = [...next, { id: change.value.id, label: labelOf(change.value) }]; } 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)); @@ -141,7 +156,11 @@ class ConversationList { } } -const labelOf = (id: ConversationId): string => (id === ROOT_CONVERSATION_ID ? "main" : `subagent ${id}`); +function labelOf(record: ConversationRecord): string { + if (record.id === ROOT_CONVERSATION_ID) return "main"; + if (record.owner !== undefined) return `subagent ${record.id}`; + return record.parent === undefined ? `conversation ${record.id}` : `fork ${record.id}`; +} function titleOf(entry: EntryRecord | undefined): { title?: string } { const message = entry?.model?.[0]; @@ -156,6 +175,7 @@ function titleOf(entry: EntryRecord | undefined): { title?: string } { /** * The oldest user message of a conversation, a subagent's task. Unlike upstream, main gets * one too: the commit listener titles it live, and a restarted host must list the same. + * A fork's own first message, not one it inherits, as the listener sees it. */ async function firstInput(harness: Harness, id: ConversationId): Promise<{ title?: string }> { const conversation = (await harness.conversation(id, context))!; @@ -163,7 +183,7 @@ async function firstInput(harness: Harness, id: ConversationId): Promise<{ title let cursor: Cursor | undefined; do { const page = await conversation.entries({}, 256, cursor, context); - first = page.items.findLast((entry) => entry.kind === "pi.user") ?? first; + first = page.items.findLast((entry) => entry.kind === "pi.user" && entry.conversationId === id) ?? first; cursor = page.next; } while (cursor !== undefined); return titleOf(first); @@ -188,6 +208,7 @@ class GatewayClient { session: options.session, root: ROOT_CONVERSATION_ID, models: options.modelSummaries(), + docs: (options.docs ?? []).map((token) => token.definition.kind), }); } @@ -229,11 +250,56 @@ class GatewayClient { const unsubscribe = this.#conversations.subscribe((value) => void this.send({ type: "snapshot", stream, value })); return { stop: unsubscribe }; } + if (stream === "approvals") { + const approvals = this.#options.approvals; + void this.send({ type: "snapshot", stream, value: approvals?.value ?? [] }); + const unsubscribe = approvals?.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)); + if (stream.startsWith("doc:")) { + const at = stream.lastIndexOf(":"); + return this.#openDoc(stream, stream.slice("doc:".length, at), Number(stream.slice(at + 1)) as ConversationId); + } const conversation = await this.#conversation(Number(stream.slice("conversation:".length)) as ConversationId); return this.#forward(stream, await conversation.watch(context)); } + /** A document that does not exist yet streams `null`, then its value from the commit that creates it. */ + async #openDoc(stream: StreamName, kind: string, id: ConversationId): Promise { + const token = this.#options.docs?.find((candidate) => candidate.definition.kind === kind); + if (token === undefined) throw new Error(`Unknown document ${kind}`); + await this.#conversation(id); + const { harness } = this.#options; + const existing = await harness.watchDoc(token, id, context); + if (existing !== undefined) return this.#forward(stream, existing); + await this.send({ type: "snapshot", stream, value: null }); + let inner: Subscription | undefined; + let stopped = false; + const unsubscribe = harness.subscribeCommits((publication) => { + const created = publication.changes.some( + (change) => (change.type === "document" || change.type === "document.copy") && change.record.kind === kind && change.conversationId === id, + ); + if (!created) return; + unsubscribe(); + // A commit listener must not call Session APIs; attach after it returns. + setImmediate(() => { + void harness.watchDoc(token, id, context).then(async (watch) => { + if (watch === undefined) return; + if (stopped) await watch.stop(); + else inner = await this.#forward(stream, watch); + }); + }); + }); + return { + stop: async () => { + stopped = true; + unsubscribe(); + await inner?.stop(); + }, + }; + } + 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 })); @@ -307,6 +373,22 @@ class GatewayClient { await conversation.configure({ thinkingLevel: levels[(levels.indexOf(level) + 1) % levels.length] ?? "off" }, context); return null; } + case "approve": { + const { approvalId, approved, by } = args as CallMethods["approve"]["args"]; + if (this.#options.approvals === undefined) throw new Error("This host has no approvals"); + return this.#options.approvals.decide(conversation.id, approvalId, approved, by); + } + case "fork": { + const { entryId, removeTools = [] } = args as CallMethods["fork"]["args"]; + // Tools are stored by name; the fork's agent drops these from what its parent offered. + const remove = (await conversation.agent(context)).tools.filter((tool) => removeTools.includes(tool.name)); + const fork = await conversation.fork( + Number(entryId) as EntryId, + { ownership: { kind: "ownerless" }, ...(remove.length === 0 ? {} : { agent: { tools: { remove } } }) }, + context, + ); + return { conversationId: fork.id }; + } } } diff --git a/spikes/durable-multiview/host/host.ts b/spikes/durable-multiview/host/host.ts index 2e749ba8..ee449ce0 100644 --- a/spikes/durable-multiview/host/host.ts +++ b/spikes/durable-multiview/host/host.ts @@ -5,6 +5,7 @@ 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 { + type Conversation, createRegistry, Harness, type HarnessSettings, @@ -13,8 +14,8 @@ import { } 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"; +import type { ModelSummary, Notice } from "../protocol/view.ts"; +import { type GatewayOptions, startGateway } from "./gateway.ts"; const context = BACKGROUND_CONTEXT; @@ -32,12 +33,20 @@ export interface OpenHostOptions { 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; + /** Tool calls held for a human decision, offered to every client. */ + readonly approvals?: GatewayOptions["approvals"]; + /** Conversation documents clients may subscribe to besides the built-in ones. */ + readonly docs?: GatewayOptions["docs"]; + /** Runs after the root exists and before recovered work resumes. */ + readonly onOpen?: (harness: Harness, root: Conversation) => Promise; } export interface OpenedHost { readonly url: string; readonly token: string; readonly harness: Harness; + /** Tell every connected client. */ + notify(level: Notice["level"], message: string): void; close(): Promise; } @@ -83,13 +92,14 @@ export async function openHost(options: OpenHostOptions): Promise { context, ); const { initialModel } = options; - await harness.root(context, { + const root = await harness.root(context, { agent: { cwd: options.cwd, ...(initialModel === undefined ? {} : { model: { provider: initialModel.provider, modelId: initialModel.modelId } }), ...(initialModel?.thinkingLevel === undefined ? {} : { thinkingLevel: initialModel.thinkingLevel }), }, }); + await options.onOpen?.(harness, root); const gateway = await startGateway({ harness, models: options.models, @@ -97,6 +107,8 @@ export async function openHost(options: OpenHostOptions): Promise { session: { id: basename(options.dataDir), directory: options.dataDir, cwd: options.cwd }, token, port: options.port, + ...(options.approvals === undefined ? {} : { approvals: options.approvals }), + ...(options.docs === undefined ? {} : { docs: options.docs }), }); report = (error) => gateway.broadcast("warning", error instanceof Error ? error.message : String(error)); for (const error of reports.splice(0)) report(error); @@ -109,6 +121,7 @@ export async function openHost(options: OpenHostOptions): Promise { url: gateway.url, token, harness, + notify: (level, message) => gateway.broadcast(level, message), close() { closing ??= (async () => { try { diff --git a/spikes/durable-multiview/host/main.ts b/spikes/durable-multiview/host/main.ts index b72d8981..b43983d7 100644 --- a/spikes/durable-multiview/host/main.ts +++ b/spikes/durable-multiview/host/main.ts @@ -3,12 +3,17 @@ // 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 --demo # the on-call demo on the real model; edits to host/demo/investigation.ts reload live +// node host/main.ts --faux-demo # the same demo driven by a scripted model, for tests and rehearsals // node host/main.ts --faux "scripted answer" [--faux-tps 40] # no real model, for tests +import { watch } from "node:fs"; 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 { type FauxResponseStep, fauxAssistantMessage, fauxProvider } from "@earendil-works/pi-ai/providers/faux"; import { getAgentDir, ModelRuntime, SettingsManager } from "@earendil-works/pi-coding-agent"; +import { oncallScript } from "./demo/faux-script.ts"; +import { createDemo, type Demo } from "./demo/index.ts"; import { type OpenHostOptions, openHost } from "./host.ts"; import { configureHarnessHttp, createHarnessSettings, defaultModel, modelSummaries } from "./pi-setup.ts"; @@ -20,24 +25,38 @@ const { values } = parseArgs({ faux: { type: "string" }, "faux-tps": { type: "string", default: "40" }, "lock-stale-ms": { type: "string" }, + demo: { type: "boolean", default: false }, + "faux-demo": { type: "boolean", default: false }, + "pace-ms": { type: "string" }, + "step-ms": { type: "string" }, + "reminder-seconds": { type: "string" }, }, }); +const number = (value: string | undefined): number | undefined => (value === undefined ? undefined : Number(value)); const cwd = resolve(values.cwd ?? process.cwd()); const dataDir = resolve(values["data-dir"] ?? join(getAgentDir(), "experimental", "durable-multiview", "default")); +const demo: Demo | undefined = + values.demo || values["faux-demo"] + ? await createDemo({ + ...(values["pace-ms"] === undefined ? {} : { paceMs: Number(values["pace-ms"]) }), + ...(values["step-ms"] === undefined ? {} : { stepMs: Number(values["step-ms"]) }), + }) + : undefined; const common = { dataDir, cwd, port: Number(values.port), ...(values["lock-stale-ms"] === undefined ? {} : { lockStaleMs: Number(values["lock-stale-ms"]) }), + ...demo?.hostOptions, }; -function fauxOptions(answer: string): OpenHostOptions { +function fauxOptions(responses: () => FauxResponseStep): 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))); + // Every request gets a fresh step, so a request rerun after a crash streams it again. + faux.setResponses(Array.from({ length: 1000 }, responses)); 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 } }; @@ -57,12 +76,33 @@ async function piOptions(): Promise { }; } -const options = values.faux === undefined ? await piOptions() : fauxOptions(values.faux); +const reminderSeconds = number(values["reminder-seconds"]); +const options = + values["faux-demo"] + ? fauxOptions(() => oncallScript(reminderSeconds === undefined ? {} : { reminderSeconds })) + : values.faux !== undefined + ? fauxOptions(() => () => fauxAssistantMessage(values.faux!)) + : await piOptions(); const host = await openHost(options); const model = options.initialModel === undefined ? null : `${options.initialModel.provider}/${options.initialModel.modelId}`; // The web client (`bun run web`) reads the host from the fragment, which never leaves the browser. const web = `http://127.0.0.1:5199/#token=${encodeURIComponent(host.token)}&url=${encodeURIComponent(host.url)}`; -console.log(JSON.stringify({ event: "ready", url: host.url, token: host.token, dataDir, model, web })); +console.log(JSON.stringify({ event: "ready", url: host.url, token: host.token, dataDir, model, web, demo: demo !== undefined })); + +if (demo !== undefined) { + // Editors save in bursts; reload once the burst settles. + let timer: NodeJS.Timeout | undefined; + const watcher = watch(demo.investigationModule, () => { + clearTimeout(timer); + timer = setTimeout(() => { + demo.reload().then( + (loads) => host.notify("info", `Reloaded the investigation tools (load ${loads}).`), + (error: unknown) => host.notify("error", `Reload failed: ${error instanceof Error ? error.message : String(error)}`), + ); + }, 200); + }); + process.once("exit", () => watcher.close()); +} let stopping = false; const stop = (): void => { diff --git a/spikes/durable-multiview/package.json b/spikes/durable-multiview/package.json index 1d404f50..8c909eaf 100644 --- a/spikes/durable-multiview/package.json +++ b/spikes/durable-multiview/package.json @@ -6,6 +6,8 @@ "type": "module", "scripts": { "host": "node host/main.ts", + "demo": "node host/main.ts --demo", + "demo:faux": "node host/main.ts --faux-demo --faux-tps 30 --reminder-seconds 20", "observe": "node cli/observe.ts", "test": "vitest run --config vitest.config.ts", "typecheck": "tsc --noEmit -p tsconfig.json && tsc --noEmit -p web/tsconfig.json", diff --git a/spikes/durable-multiview/protocol/demo.ts b/spikes/durable-multiview/protocol/demo.ts new file mode 100644 index 00000000..379ec73f --- /dev/null +++ b/spikes/durable-multiview/protocol/demo.ts @@ -0,0 +1,34 @@ +// Shapes of the on-call demo's own state, as clients receive it. Type-only, so browsers can import it. +import type { ConversationId } from "@earendil-works/pi-durable"; + +export type PlanItem = { + readonly text: string; + readonly status: "todo" | "doing" | "done"; +} + +/** `demo.plan`: the investigation plan the main agent keeps. */ +export type PlanState = { + readonly items: readonly PlanItem[]; +} + +/** `demo.rollout`: each rollback's regions and where each one got to, keyed by rollback task ID. */ +export type RolloutState = { + readonly runs: Readonly>>>; +} + +export type ApprovalDecision = { + readonly approved: boolean; + /** The client that decided. */ + readonly by: string; + readonly at: number; +} + +/** A tool call held by the approval hook until some client decides. */ +export type PendingApproval = { + /** The held tool call's task ID. */ + readonly id: string; + readonly conversationId: ConversationId; + readonly tool: string; + readonly summary: string; + readonly requestedAt: number; +} diff --git a/spikes/durable-multiview/protocol/frames.ts b/spikes/durable-multiview/protocol/frames.ts index c65d7cef..0d963772 100644 --- a/spikes/durable-multiview/protocol/frames.ts +++ b/spikes/durable-multiview/protocol/frames.ts @@ -1,15 +1,23 @@ import type { Op } from "@earendil-works/chord/delta"; import type { ConversationId, ModelRef } from "@earendil-works/pi-durable"; +import type { ApprovalDecision } from "./demo.ts"; 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`. + * A subscribable value: one conversation's view, one of its documents (`null` + * while absent), the task graph, the conversation list, or the pending + * approvals. A subscription starts with a snapshot; ops follow and apply to it + * with chord's `applyImmutable`. */ -export type StreamName = `conversation:${ConversationId}` | "tasks" | "conversations"; +export type StreamName = + | `conversation:${ConversationId}` + | `doc:${string}:${ConversationId}` + | "tasks" + | "conversations" + | "approvals"; export const conversationStream = (id: ConversationId): StreamName => `conversation:${id}`; +export const docStream = (kind: string, id: ConversationId): StreamName => `doc:${kind}:${id}`; /** Close code for a rejected token; clients must not reconnect after it. */ export const UNAUTHORIZED_CLOSE_CODE = 4401; @@ -29,6 +37,16 @@ export interface CallMethods { compact: { args: { conversationId: ConversationId; instructions?: string }; result: { taskId: string } }; setModel: { args: { conversationId: ConversationId; model: ModelRef }; result: null }; cycleThinking: { args: { conversationId: ConversationId }; result: null }; + /** Decide a pending approval; the first decision committed stands. */ + approve: { + args: { conversationId: ConversationId; approvalId: string; approved: boolean; by: string }; + result: { decision: ApprovalDecision; first: boolean }; + }; + /** A new ownerless conversation that sees this one's entries through `entryId`. */ + fork: { + args: { conversationId: ConversationId; entryId: string; removeTools?: readonly string[] }; + result: { conversationId: ConversationId }; + }; } export type CallMethod = keyof CallMethods; @@ -39,6 +57,8 @@ export type ServerFrame = readonly session: SessionInfo; readonly root: ConversationId; readonly models: readonly ModelSummary[]; + /** Kinds of the conversation documents offered as `doc:` streams. */ + readonly docs: readonly string[]; } | { readonly type: "snapshot"; readonly stream: StreamName; readonly value: unknown } | { readonly type: "ops"; readonly stream: StreamName; readonly ops: readonly Op[] } diff --git a/spikes/durable-multiview/protocol/remote-durable.ts b/spikes/durable-multiview/protocol/remote-durable.ts index 096519b6..b4a9c292 100644 --- a/spikes/durable-multiview/protocol/remote-durable.ts +++ b/spikes/durable-multiview/protocol/remote-durable.ts @@ -1,10 +1,12 @@ import { applyImmutable } from "@earendil-works/chord/delta"; -import type { ConversationId, ConversationView, TaskGraph } from "@earendil-works/pi-durable"; +import type { ConversationId, ConversationView, JsonObject, TaskGraph } from "@earendil-works/pi-durable"; +import type { PendingApproval } from "./demo.ts"; import { type CallMethod, type CallMethods, type ClientFrame, conversationStream, + docStream, type ServerFrame, type StreamName, UNAUTHORIZED_CLOSE_CODE, @@ -25,7 +27,11 @@ export type RemoteDurableOptions = ( readonly token: string; } | { readonly transport: FrameTransport } -) & { readonly reconnectDelayMs?: { readonly min: number; readonly max: number } }; +) & { + readonly reconnectDelayMs?: { readonly min: number; readonly max: number }; + /** Who this client is in approval decisions, such as "web" or "tui". */ + readonly clientName?: string; +}; export interface RemoteDurable { readonly view: DurableViewSource; @@ -49,7 +55,8 @@ class RemoteClient { readonly #options: RemoteDurableOptions; readonly #listeners = new Set<() => void>(); readonly #values = new Map(); - readonly #wanted = new Set(["conversations"]); + readonly #wanted = new Set(["conversations", "approvals"]); + #docKinds: readonly string[] = []; readonly #snapshotWaiters = new Map(); readonly #pending = new Map(); readonly #transport: FrameTransport; @@ -123,7 +130,8 @@ class RemoteClient { case "hello": { this.#attempt = 0; this.#current ??= frame.root; - this.#wanted.add(conversationStream(this.#current)); + this.#docKinds = frame.docs; + for (const stream of this.#conversationStreams(this.#current)) this.#wanted.add(stream); 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); @@ -171,7 +179,10 @@ class RemoteClient { models: hello.models, notices: [], connection: "connected", + docs: {}, + approvals: [], }; + this.#refresh(); this.#resolveReady({ view: this.#viewSource(), controller: this.#controller(), close: () => this.#close() }); }); } @@ -185,15 +196,30 @@ class RemoteClient { }); } + /** The streams that show one conversation: its view and its documents. */ + #conversationStreams(id: ConversationId): StreamName[] { + return [conversationStream(id), ...this.#docKinds.map((kind) => docStream(kind, id))]; + } + #refresh(): void { if (this.#state === undefined || this.#current === undefined) return; - const conversation = this.#values.get(conversationStream(this.#current)) as ConversationView | undefined; + const current = this.#current; + const conversation = this.#values.get(conversationStream(current)) as ConversationView | undefined; const conversations = this.#values.get("conversations") as ConversationSummary[] | undefined; + const approvals = this.#values.get("approvals") as PendingApproval[] | undefined; const tasks = this.#wanted.has("tasks") ? (this.#values.get("tasks") as TaskGraph | undefined) : undefined; + const docs = Object.fromEntries( + this.#docKinds.flatMap((kind) => { + const stream = docStream(kind, current); + return this.#values.has(stream) ? [[kind, this.#values.get(stream) as JsonObject | null]] : []; + }), + ); this.#update({ ...(conversation === undefined ? {} : { conversation }), ...(conversations === undefined ? {} : { conversations }), + ...(approvals === undefined ? {} : { approvals }), tasks, + docs, }); } @@ -276,24 +302,46 @@ class RemoteClient { this.#send({ type: "subscribe", stream: "tasks" }); await shown; }), - switchConversation: (id) => + switchConversation: (id) => this.#command(() => this.#switch(id)), + approve: (approval, approved) => 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(); + const by = this.#options.clientName ?? "client"; + const args = { conversationId: approval.conversationId, approvalId: approval.id, approved, by }; + const { decision, first } = await this.#call("approve", args); + const verdict = decision.approved ? "approved" : "denied"; + this.#notice("info", first ? `You ${verdict} the ${approval.tool}.` : `Already ${verdict} by ${decision.by}.`); + }), + fork: (entryId, prompt, removeTools) => + this.#command(async () => { + const { conversationId: forked } = await this.#call("fork", { + conversationId: conversationId(), + entryId, + ...(removeTools === undefined ? {} : { removeTools }), + }); + // The fork is listed once its creating commit reaches the conversation list; switching does not need that. + await this.#switch(forked); + await this.#call("submit", { conversationId: forked, text: prompt, whenBusy: "followUp", requestId: crypto.randomUUID() }); }), }; } + async #switch(id: ConversationId): Promise { + const previous = this.#current!; + if (id === previous) return; + const next = this.#conversationStreams(id); + for (const stream of next) this.#wanted.add(stream); + const shown = Promise.all(next.map((stream) => this.#snapshot(stream))); + for (const stream of next) this.#send({ type: "subscribe", stream }); + await shown; + this.#current = id; + for (const stream of this.#conversationStreams(previous)) { + this.#wanted.delete(stream); + this.#values.delete(stream); + this.#send({ type: "unsubscribe", stream }); + } + this.#refresh(); + } + #close(): void { if (this.#closed) return; this.#closed = true; diff --git a/spikes/durable-multiview/protocol/view.ts b/spikes/durable-multiview/protocol/view.ts index b324aba3..d7c0e23d 100644 --- a/spikes/durable-multiview/protocol/view.ts +++ b/spikes/durable-multiview/protocol/view.ts @@ -2,7 +2,8 @@ // 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"; +import type { ConversationId, ConversationView, JsonObject, ModelRef, TaskGraph } from "@earendil-works/pi-durable"; +import type { PendingApproval } from "./demo.ts"; export interface ModelSummary extends ModelRef { readonly name: string; @@ -42,6 +43,10 @@ export interface DurableView { /** The live task graph while the task panel is open. */ readonly tasks?: TaskGraph; readonly connection: ConnectionState; + /** The shown conversation's extension documents, by kind; `null` while one does not exist. */ + readonly docs: Readonly>; + /** Tool calls of any conversation waiting for a human decision. */ + readonly approvals: readonly PendingApproval[]; } export interface DurableViewSource { @@ -60,4 +65,8 @@ export interface DurableController { toggleTasks(): Promise; /** Show and talk to another conversation. */ switchConversation(id: ConversationId): Promise; + /** Decide a pending approval as this client; a later decision than the first is reported, not applied. */ + approve(approval: PendingApproval, approved: boolean): Promise; + /** Fork the shown conversation at an entry, switch to the fork, and send it `prompt`. */ + fork(entryId: string, prompt: string, removeTools?: readonly string[]): Promise; } diff --git a/spikes/durable-multiview/test/crash-recovery.test.ts b/spikes/durable-multiview/test/crash-recovery.test.ts index 00e950cb..0a364985 100644 --- a/spikes/durable-multiview/test/crash-recovery.test.ts +++ b/spikes/durable-multiview/test/crash-recovery.test.ts @@ -19,10 +19,10 @@ interface HostProcess { readonly token: string; } -async function spawnHost(dataDir: string, port: number): Promise { +async function spawnHost(dataDir: string, port: number, model: readonly string[] = ["--faux", LONG_ANSWER, "--faux-tps", "40"]): 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"], + ["host/main.ts", "--data-dir", dataDir, "--cwd", dataDir, "--port", String(port), ...model, "--lock-stale-ms", "2000"], { cwd: spikeDir, stdio: ["ignore", "pipe", "inherit"] }, ); cleanups.push(() => { @@ -80,3 +80,35 @@ it("resumes both clients after the host is killed mid-stream and restarted", asy expect(rest).toEqual([]); expect(b.view.current().conversation).toEqual(a.view.current().conversation); }); + +it("finds the same three subagents after the host is killed while they investigate", async () => { + const dir = await tempDir(); + cleanups.push(dir.remove); + const port = await freePort(); + const demo = ["--faux-demo", "--pace-ms", "150"]; + const first = await spawnHost(dir.path, port, demo); + const client = await connectRemoteDurable({ url: first.url, token: first.token, reconnectDelayMs: { min: 100, max: 500 } }); + cleanups.push(() => client.close()); + const subagents = () => client.view.current().conversations.filter((summary) => summary.label.startsWith("subagent")); + + await client.controller.toggleTasks(); + await client.controller.submit("v2.3 release failed, find out why", "followUp"); + // Kill once all three children run their investigation tools, before any finished. + await waitForView(client.view, (view) => { + const children = new Set(subagents().map((summary) => summary.id)); + return Object.values(view.tasks?.tasks ?? {}).filter((node) => node.kind === "pi.tool" && children.has(node.conversationId)).length === 3; + }); + const before = subagents().map((summary) => summary.id); + expect(before).toHaveLength(3); + first.child.kill("SIGKILL"); + await once(first.child, "exit"); + await waitForView(client.view, (view) => view.connection === "reconnecting"); + + await spawnHost(dir.path, port, demo); + await waitForView( + client.view, + (view) => view.connection === "connected" && !isBusy(view.conversation) && /recommend rolling back/i.test(transcript(view.conversation).at(-1)?.text ?? ""), + 30_000, + ); + expect(subagents().map((summary) => summary.id)).toEqual(before); +}); diff --git a/spikes/durable-multiview/test/demo.test.ts b/spikes/durable-multiview/test/demo.test.ts new file mode 100644 index 00000000..fee63474 --- /dev/null +++ b/spikes/durable-multiview/test/demo.test.ts @@ -0,0 +1,157 @@ +import { copyFile, mkdir, readFile, rm, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; +import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; +import type { ConversationId, TaskGraph } from "@earendil-works/pi-durable"; +import { describe, expect, it } from "vitest"; +import type { PlanState, RolloutState } from "../protocol/demo.ts"; +import { isBusy, transcript } from "../protocol/transcript.ts"; +import type { DurableView } from "../protocol/view.ts"; +import { connectTo, freePort, startDemoHost, useCleanups, waitForView } from "./support.ts"; + +const defer = useCleanups(); +const context = BACKGROUND_CONTEXT; +const spikeDir = fileURLToPath(new URL("..", import.meta.url)); + +const idle = (view: DurableView): boolean => !isBusy(view.conversation) && view.conversation.entries.length > 0; +const lastText = (view: DurableView): string | undefined => transcript(view.conversation).at(-1)?.text; +const subagents = (view: DurableView) => view.conversations.filter((summary) => summary.label.startsWith("subagent")); +const rollout = (view: DurableView) => view.docs["demo.rollout"] as RolloutState | null | undefined; +const plan = (view: DurableView) => view.docs["demo.plan"] as PlanState | null | undefined; + +describe("on-call demo", () => { + it("plans, then delegates to three subagents in parallel; each child is a listed conversation", async () => { + const { host } = await startDemoHost(defer); + const client = await connectTo(defer, host); + await client.controller.toggleTasks(); + let parallelCalls = 0; + client.view.subscribe(() => { + const graph: TaskGraph | undefined = client.view.current().tasks; + const calls = Object.values(graph?.tasks ?? {}).filter((node) => node.kind === "pi.tool" && node.conversations.length > 0); + parallelCalls = Math.max(parallelCalls, calls.length); + }); + + await client.controller.submit("v2.3 release failed, find out why", "followUp"); + await waitForView(client.view, (view) => idle(view) && /recommend rolling back/i.test(lastText(view) ?? "")); + + const view = client.view.current(); + expect(plan(view)!.items.map((item) => item.text)).toEqual([ + "Search the deploy logs", + "Compare error metrics", + "Review the commits in v2.3", + ]); + // Created in parallel, so in no particular order. + expect(subagents(view).map((summary) => summary.title?.split(":")[0]).sort()).toEqual(["commits", "logs", "metrics"]); + expect(parallelCalls).toBe(3); + // A child is offered only its own investigation tool. + const logs = subagents(view).find((summary) => summary.title?.startsWith("logs"))!; + const child = (await host.harness.conversation(logs.id, context))!; + const childAgent = await child.agent(context); + expect(childAgent.tools.map((tool) => tool.name)).toEqual(["search_logs"]); + }); + + it("asks every client to approve a rollback; the first decision wins and the later click is told", async () => { + const { host } = await startDemoHost(defer); + const a = await connectTo(defer, host, host.token, "web"); + const b = await connectTo(defer, host, host.token, "tui"); + + await a.controller.submit("roll back to v2.2", "followUp"); + await Promise.all([a, b].map((client) => waitForView(client.view, (view) => view.approvals.length === 1))); + const [approval] = b.view.current().approvals; + expect(approval).toMatchObject({ tool: "rollback", conversationId: a.view.current().conversation.conversation.id }); + + await a.controller.approve(approval!, true); + await b.controller.approve(approval!, false); + expect(b.view.current().notices.at(-1)?.message).toMatch(/already approved by web/i); + await waitForView(b.view, (view) => view.approvals.length === 0); + + await waitForView(a.view, (view) => idle(view) && /check scheduled/i.test(lastText(view) ?? "")); + const regions = Object.values(rollout(a.view.current())!.runs)[0]!; + expect(regions).toEqual({ + "us-east": "compensated: traffic restored", + "eu-west": "failed: health check", + "ap-south": "compensated: traffic restored", + }); + }); + + it("does not ask again after a restart once the rollback was approved", async () => { + const port = await freePort(); + const first = await startDemoHost(defer, { port, stepMs: 600 }); + const client = await connectTo(defer, first.host, first.host.token, "web"); + await client.controller.submit("roll back to v2.2", "followUp"); + await waitForView(client.view, (view) => view.approvals.length === 1); + await client.controller.approve(client.view.current().approvals[0]!, true); + await waitForView(client.view, (view) => Object.keys(Object.values(rollout(view)?.runs ?? {})[0] ?? {}).length === 3); + + const dataDir = client.view.current().session.directory; + await first.host.close(); + await waitForView(client.view, (view) => view.connection === "reconnecting"); + let askedAgain = false; + client.view.subscribe(() => { + if (client.view.current().approvals.length > 0) askedAgain = true; + }); + await startDemoHost(defer, { port, dataDir, stepMs: 600 }); + + await waitForView(client.view, (view) => view.connection === "connected" && idle(view) && /check scheduled/i.test(lastText(view) ?? ""), 30_000); + expect(askedAgain).toBe(false); + // The replay-safe tool found the rollback it had started instead of starting a second one. + expect(Object.keys(rollout(client.view.current())!.runs)).toHaveLength(1); + }); + + it("delivers a scheduled check as a follow-up after Esc and across a restart", async () => { + const port = await freePort(); + const first = await startDemoHost(defer, { port }); + const client = await connectTo(defer, first.host); + await client.controller.submit("check again in 2 seconds", "followUp"); + await waitForView(client.view, (view) => idle(view) && /check scheduled/i.test(lastText(view) ?? "")); + await client.controller.abort(); + + const dataDir = client.view.current().session.directory; + await first.host.close(); + await startDemoHost(defer, { port, dataDir }); + + await waitForView(client.view, (view) => idle(view) && /back to baseline/i.test(lastText(view) ?? ""), 30_000); + const reminders = transcript(client.view.current().conversation).filter((line) => line.text.startsWith("Reminder:")); + expect(reminders).toHaveLength(1); + }); + + it("forks at an answer into a conversation without the rollback tool", async () => { + const { host } = await startDemoHost(defer); + const client = await connectTo(defer, host); + await client.controller.submit("v2.3 release failed, find out why", "followUp"); + await waitForView(client.view, (view) => idle(view) && /recommend rolling back/i.test(lastText(view) ?? "")); + const answer = client.view.current().conversation.entries.findLast((entry) => entry.kind === "pi.assistant")!; + const parent = client.view.current().conversation.conversation.id; + + await client.controller.fork(String(answer.id), "write the postmortem", ["rollback"]); + + await waitForView(client.view, (view) => view.conversation.conversation.id !== parent && idle(view) && /postmortem/i.test(lastText(view) ?? "")); + const view = client.view.current(); + expect(view.conversations.find((summary) => summary.id === view.conversation.conversation.id)).toMatchObject({ label: expect.stringMatching(/^fork/) }); + expect(plan(view)!.items).toHaveLength(3); + const fork = (await host.harness.conversation(view.conversation.conversation.id as ConversationId, context))!; + expect((await fork.agent(context)).tools.map((tool) => tool.name)).not.toContain("rollback"); + expect(transcript(view.conversation).at(-2)).toEqual({ role: "user", text: "write the postmortem" }); + }); + + it("hot-reloads the investigation tools; the next call uses the new code", async () => { + const dir = join(spikeDir, ".tmp", `reload-${process.pid}-${Date.now()}`); + await mkdir(dir, { recursive: true }); + defer(() => rm(dir, { recursive: true, force: true })); + const module = join(dir, "investigation.ts"); + await copyFile(join(spikeDir, "host", "demo", "investigation.ts"), module); + const { host, demo } = await startDemoHost(defer, { investigationModule: module }); + const client = await connectTo(defer, host); + + await writeFile(module, (await readFile(module, "utf8")).replace(`const FORMAT = "plain"`, `const FORMAT = "table"`)); + expect(await demo.reload()).toBe(2); + + await client.controller.submit("v2.3 release failed, find out why", "followUp"); + await waitForView(client.view, (view) => idle(view) && /recommend rolling back/i.test(lastText(view) ?? "")); + const logs = subagents(client.view.current()).find((summary) => summary.title?.startsWith("logs"))!; + const child = await host.harness.conversation(logs.id, context); + const { messages } = await child!.context(context); + const output = messages.flatMap((message) => (message.role === "toolResult" ? message.content : [])); + expect(output.map((block) => (block.type === "text" ? block.text : "")).join("")).toMatch(/^\| /m); + }); +}); diff --git a/spikes/durable-multiview/test/support.ts b/spikes/durable-multiview/test/support.ts index 9fd2aa23..5d93061d 100644 --- a/spikes/durable-multiview/test/support.ts +++ b/spikes/durable-multiview/test/support.ts @@ -5,6 +5,8 @@ import { join } from "node:path"; import { createModels } from "@earendil-works/pi-ai/models"; import { fauxAssistantMessage, fauxProvider } from "@earendil-works/pi-ai/providers/faux"; import { afterEach } from "vitest"; +import { createDemo, type Demo } from "../host/demo/index.ts"; +import { oncallScript } from "../host/demo/faux-script.ts"; import { openHost, type OpenedHost } from "../host/host.ts"; import { connectRemoteDurable, type RemoteDurable } from "../protocol/remote-durable.ts"; import type { DurableViewSource } from "../protocol/view.ts"; @@ -54,12 +56,49 @@ export async function startFauxHost( return host; } +/** An in-process host with the on-call demo installed, driven by the scripted faux model. */ +export async function startDemoHost( + defer: (cleanup: () => Promise | void) => void, + { + dataDir, + port = 0, + stepMs = 150, + investigationModule, + }: { dataDir?: string; port?: number; stepMs?: number; investigationModule?: string } = {}, +): Promise<{ host: OpenedHost; demo: Demo }> { + const dir = dataDir === undefined ? await tempDir() : { path: dataDir, remove: () => {} }; + defer(dir.remove); + const faux = fauxProvider({ tokensPerSecond: 400, tokenSize: { min: 2, max: 4 } }); + const models = createModels(); + models.setProvider(faux.provider); + faux.setResponses(Array.from({ length: 200 }, () => oncallScript())); + const model = faux.getModel(); + const demo = await createDemo({ paceMs: 10, stepMs, ...(investigationModule === undefined ? {} : { investigationModule }) }); + 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, + ...demo.hostOptions, + }); + defer(() => host.close()); + return { host, demo }; +} + export async function connectTo( defer: (cleanup: () => Promise | void) => void, host: OpenedHost, token = host.token, + clientName?: string, ): Promise { - const client = await connectRemoteDurable({ url: host.url, token }); + const client = await connectRemoteDurable({ + url: host.url, + token, + reconnectDelayMs: { min: 100, max: 500 }, + ...(clientName === undefined ? {} : { clientName }), + }); defer(() => client.close()); return client; }