From 4f578243578ea33dc2b708e64ede68766ec24326 Mon Sep 17 00:00:00 2001 From: tt-a1i Date: Sat, 22 Aug 2026 19:53:24 +0800 Subject: [PATCH 1/7] fix: doc/example intent words, MIT notice, no-repeat hint, drop format.ts copy MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - README quick-start example now uses explicit capability intent words (在后台/子代理) so the default explicit router actually loads them - remove stale 'no project license' claim contradicting MIT - background terminal and workflow follow-up results now carry the 'already shown to the user, do not repeat' hint (subagents had it) - delete subagents/src/format.ts, a self-confessed copy of shared/context-utilization.ts; point 3 imports at the shared helper --- README.md | 8 +++- extensions/background-terminals/src/prompt.ts | 2 +- extensions/subagents/index.ts | 2 +- extensions/subagents/navigation.ts | 2 +- extensions/subagents/src/format.ts | 48 ------------------- extensions/subagents/src/ui/takeover.ts | 2 +- extensions/workflows/prompt.ts | 2 +- 7 files changed, 11 insertions(+), 55 deletions(-) delete mode 100644 extensions/subagents/src/format.ts diff --git a/README.md b/README.md index 0b47aa2c..8e341dc2 100644 --- a/README.md +++ b/README.md @@ -69,12 +69,16 @@ pi install npm:@tt-a1i/openpi 重启 Pi，或在当前 Session 运行 `/reload`。然后直接描述真实任务： ```text -启动前端 dev server；并行检查 API 主链路和测试覆盖； +在后台启动前端 dev server；用子代理并行检查 API 主链路和测试覆盖； 结果回来后汇总风险，主会话不要原地等待。 ``` OpenPI 会把长期进程放到后台，把独立任务交给隔离 Context 的 Pi Subagent，把多阶段依赖组织成 Workflow。状态会持续显示；完整运行可从 `/ps`、`/subagents` 和 `/workflows` 检查或终止。 +> [!TIP] +> Capability discovery 默认 `explicit`：表述里需要出现明确的能力词才会加载对应组—— +> 「在后台 / 后台运行」→ 后台终端；「子代理 / 委派」→ Subagent；「Workflow / 多阶段编排」→ Workflow；「搜索」→ `fd`/`rg`。像上面示例那样把意图说清楚即可，不需要记住任何工具名。 + > [!IMPORTANT] > 默认安装是安静的：不改主题、不绑定 Provider 或模型、不开启下一步预测，也不执行 post-edit 命令。Capability discovery 默认 `explicit`；只有用户通过 `/openpi-setup` 选择 `adaptive` 后，模型才会常驻看到一个小型发现网关并可自主加载额外能力。 @@ -570,4 +574,4 @@ npm 仍用于发布包的 `pack` / clean-install 验证，因为用户通过 npm `extensions/sessions/` 改编自 [jayshah5696/pi-agent-extensions](https://github.com/jayshah5696/pi-agent-extensions)。可选的顶层 Session 通信由 [pi-intercom](https://github.com/nicobailon/pi-intercom) 提供。完整第三方说明见 [`THIRD_PARTY_NOTICES.md`](THIRD_PARTY_NOTICES.md)。 -本仓库目前没有项目级开源许可证；`THIRD_PARTY_NOTICES.md` 只记录第三方来源与各自许可，不等同于授予本项目使用许可。 +本项目以 MIT 许可证发布（见 [`LICENSE`](LICENSE)）；`THIRD_PARTY_NOTICES.md` 记录第三方来源与各自许可。 diff --git a/extensions/background-terminals/src/prompt.ts b/extensions/background-terminals/src/prompt.ts index 538ac415..4eb65cfe 100644 --- a/extensions/background-terminals/src/prompt.ts +++ b/extensions/background-terminals/src/prompt.ts @@ -178,7 +178,7 @@ export function buildTerminalResultMessage(snap: TerminalSnapshot) { if (snap.stderr.totalBytes > 0) { text += `\n\n${outputSection("stderr", snap.stderr, RESULT_STDERR_MAX, RESULT_STDERR_MAX_LINES)}`; } - return text; + return `${text}\n\n(This result is already shown to the user. Act on it and relay only the decisions or next steps — do not repeat it verbatim.)`; } /** Preserve every retained terminal identity while globally bounding batch logs. */ diff --git a/extensions/subagents/index.ts b/extensions/subagents/index.ts index 9cabfc9f..e2ba73bd 100644 --- a/extensions/subagents/index.ts +++ b/extensions/subagents/index.ts @@ -73,7 +73,7 @@ import { registerEditorLayer, removeEditorLayer, } from "../shared/editor-layers.ts"; -import { formatContextUtilization } from "./src/format.ts"; +import { formatContextUtilization } from "../shared/context-utilization.ts"; import { SubagentManager, type SubagentManagerShape } from "./src/manager.ts"; import { buildSubagentResultMessage, diff --git a/extensions/subagents/navigation.ts b/extensions/subagents/navigation.ts index becf7b45..6b9fbfd5 100644 --- a/extensions/subagents/navigation.ts +++ b/extensions/subagents/navigation.ts @@ -12,7 +12,7 @@ import { import { spinnerFrame } from "../shared/spinner.ts"; import { sanitizeTerminalText } from "../shared/terminal-text.ts"; import { formatElapsed, type SubagentSnapshot } from "./src/domain.ts"; -import { contextPercent } from "./src/format.ts"; +import { contextPercent } from "../shared/context-utilization.ts"; export interface SubagentStripEntry { snapshot: SubagentSnapshot; diff --git a/extensions/subagents/src/format.ts b/extensions/subagents/src/format.ts deleted file mode 100644 index 7c2b219b..00000000 --- a/extensions/subagents/src/format.ts +++ /dev/null @@ -1,48 +0,0 @@ -/** - * Formatting helpers (self-contained copy of the v1 shared - * context-utilization helper). - */ - -export interface ContextUtilization { - /** Current conversation context occupancy; undefined while unknown. */ - tokens?: number | null; - /** Capacity of the model currently serving the conversation. */ - contextWindow?: number | null; -} - -function usableTokens(value: number | null | undefined) { - return typeof value === "number" && Number.isFinite(value) && value >= 0 - ? value - : undefined; -} - -function usableCapacity(value: number | null | undefined) { - return typeof value === "number" && Number.isFinite(value) && value > 0 - ? value - : undefined; -} - -export function contextPercent(usage: ContextUtilization) { - const tokens = usableTokens(usage.tokens); - const capacity = usableCapacity(usage.contextWindow); - if (tokens === undefined || capacity === undefined) return undefined; - return Math.round(Math.min(100, Math.max(0, (tokens / capacity) * 100))); -} - -export function formatCompactTokens(count: number) { - if (count < 1000) return Math.round(count).toString(); - if (count < 10000) return `${(count / 1000).toFixed(1)}k`; - if (count < 1000000) return `${Math.round(count / 1000)}k`; - return `${(count / 1000000).toFixed(1)}M`; -} - -/** - * Render `%/capacity`. If occupancy is unknown, retain the useful capacity - * as `?%/capacity`; with no valid capacity, omit the statistic entirely. - */ -export function formatContextUtilization(usage: ContextUtilization) { - const capacity = usableCapacity(usage.contextWindow); - if (capacity === undefined) return ""; - const percent = contextPercent(usage); - return `${percent === undefined ? "?" : percent}%/${formatCompactTokens(capacity)}`; -} diff --git a/extensions/subagents/src/ui/takeover.ts b/extensions/subagents/src/ui/takeover.ts index 43768ba2..2b53a52e 100644 --- a/extensions/subagents/src/ui/takeover.ts +++ b/extensions/subagents/src/ui/takeover.ts @@ -20,7 +20,7 @@ import { } from "../../../shared/screen-chrome.ts"; import { sanitizeTerminalText } from "../../../shared/terminal-text.ts"; import { formatElapsed, type SubagentSnapshot } from "../domain.ts"; -import { formatContextUtilization } from "../format.ts"; +import { formatContextUtilization } from "../../../shared/context-utilization.ts"; import type { SubagentReadModel } from "../manager.ts"; import { buildTranscriptLines, diff --git a/extensions/workflows/prompt.ts b/extensions/workflows/prompt.ts index 8ab2c281..606837e6 100644 --- a/extensions/workflows/prompt.ts +++ b/extensions/workflows/prompt.ts @@ -167,7 +167,7 @@ export function buildBackgroundWorkflowFollowUp(options: { // Sentence lead-in matching the subagent/terminal completion messages. const label = options.name ? `"${options.name}"` : options.runId; const verb = options.status === "completed" ? "finished" : options.status; - return `Background workflow ${label} (${options.runId}) ${verb}.\n\n${options.result}`; + return `Background workflow ${label} (${options.runId}) ${verb}.\n\n${options.result}\n\n(This result is already shown to the user. Act on it and relay only the decisions or next steps — do not repeat it verbatim.)`; } /** Builds the background-launch result and tells the parent model how to inspect or stop the run. */ From d7163763702f4306cb26454103d604b8a25b247f Mon Sep 17 00:00:00 2001 From: tt-a1i Date: Sat, 22 Aug 2026 20:02:45 +0800 Subject: [PATCH 2/7] fix(file-search): cache downloaded fd/rg binaries in ~/.pi/agent/bin The download destination resolved to a directory inside the installed npm package (node_modules/@tt-a1i/openpi/bin), which is wiped on every package update, forcing a fresh ~10MB download each time. The "repositoryBinDir" comment also wrongly claimed it resolved ~/.pi/agent, and the "bundled" resolution layer can never hit first in a published install - it is purely a download cache. Resolve binaries from the Pi agent's private managed bin directory (~/.pi/agent/bin, via getAgentDir(), the same pattern used by the setup and child-session extensions) so cached binaries survive package updates. Rename the misleading "bundled" source to "cached" and fix the comments to describe the actual semantics. SHA-256 pinning, resolution order, and all other layers are unchanged. --- SETUP.md | 2 +- extensions/file-search/index.spec.ts | 33 ++++++++++++++++------- extensions/file-search/index.ts | 15 ++++++----- extensions/file-search/src/binaries.ts | 36 +++++++++++++------------- 4 files changed, 51 insertions(+), 35 deletions(-) diff --git a/SETUP.md b/SETUP.md index ab5b420e..d1a5ee8e 100644 --- a/SETUP.md +++ b/SETUP.md @@ -16,7 +16,7 @@ Pi installs the package dependencies automatically. Restart Pi or run `/reload` ## fd and rg tools -The `file-search` extension registers `fd` and `rg` as model tools. They stay outside an ordinary parent turn until the user explicitly asks to use `fd`/`rg` or structured file search, or the model loads the `search` group through `openpi_load_tools`. The gateway is shown after an explicit OpenPI-capability request, or remains visible when the user opts into adaptive discovery; child sessions may still receive `fd` and `rg` through the reviewed child-safe allowlist. No setup is normally needed: at startup it silently uses a system-installed `fd` (or `fdfind` on Debian/Ubuntu) and `rg` when available, or an existing fallback in this package's private `bin/` directory. Only when neither exists does it download an official release binary (macOS/Linux, arm64/x64, over HTTPS) into that package-local directory and show a one-time notification. If your platform is unsupported, install `fd` and `rg` with your package manager and restart Pi. +The `file-search` extension registers `fd` and `rg` as model tools. They stay outside an ordinary parent turn until the user explicitly asks to use `fd`/`rg` or structured file search, or the model loads the `search` group through `openpi_load_tools`. The gateway is shown after an explicit OpenPI-capability request, or remains visible when the user opts into adaptive discovery; child sessions may still receive `fd` and `rg` through the reviewed child-safe allowlist. No setup is normally needed: at startup it silently uses a system-installed `fd` (or `fdfind` on Debian/Ubuntu) and `rg` when available, or an existing binary in the agent's private managed bin directory (`~/.pi/agent/bin`). Only when neither exists does it download an official release binary (macOS/Linux, arm64/x64, over HTTPS) into that directory — a persistent cache that survives package updates — and show a one-time notification. If your platform is unsupported, install `fd` and `rg` with your package manager and restart Pi. ## Theme diff --git a/extensions/file-search/index.spec.ts b/extensions/file-search/index.spec.ts index 774bb69d..78b57cd5 100644 --- a/extensions/file-search/index.spec.ts +++ b/extensions/file-search/index.spec.ts @@ -1,6 +1,7 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; import { randomUUID } from "node:crypto"; +import { mkdtempSync, rmSync } from "node:fs"; import { homedir, tmpdir } from "node:os"; import { dirname, join } from "node:path"; import { Writable } from "node:stream"; @@ -15,6 +16,7 @@ import { import { FD_INTEL_DARWIN_VERSION, InstallError, + managedBinDir, readBoundedResponse, releaseAsset, resolveBinary, @@ -228,20 +230,20 @@ it.effect("binary resolution: fdfind is accepted as a system fd", () => }), ); -it.effect("binary resolution: existing bin fallback is used silently", () => +it.effect("binary resolution: an existing cached binary is used silently", () => Effect.gen(function* () { - const env = makeEnv({ available: ["/repo/bin/rg"] }); + const env = makeEnv({ available: ["/cache/bin/rg"] }); const resolved = yield* resolveBinary( TOOL_SPECS.rg, - "/repo/bin", + "/cache/bin", darwinArm, env, ); assert.deepEqual(resolved, { tool: "rg", - command: "/repo/bin/rg", - source: "bundled", + command: "/cache/bin/rg", + source: "cached", }); assert.equal(env.installs.length, 0); }), @@ -316,6 +318,19 @@ it.effect("binary resolution: one failed tool does not disable the other", () => }), ); +it("managed bin directory lives inside the agent directory", () => { + const agentDir = mkdtempSync(join(tmpdir(), "my-pi-setup-bins-")); + const previous = process.env.PI_CODING_AGENT_DIR; + process.env.PI_CODING_AGENT_DIR = agentDir; + try { + assert.equal(managedBinDir(), join(agentDir, "bin")); + } finally { + if (previous === undefined) delete process.env.PI_CODING_AGENT_DIR; + else process.env.PI_CODING_AGENT_DIR = previous; + rmSync(agentDir, { recursive: true, force: true }); + } +}); + it("release assets cover macOS and Linux on arm64 and x64 over HTTPS", () => { for (const os of ["darwin", "linux"] as const) { for (const arch of ["arm64", "x64"] as const) { @@ -377,10 +392,10 @@ it("notifications: only fresh installs notify", () => { command: "fd", source: "system", }; - const bundled: ResolvedBinary = { + const cached: ResolvedBinary = { tool: "rg", - command: "/repo/bin/rg", - source: "bundled", + command: "/cache/bin/rg", + source: "cached", }; const installed: ResolvedBinary = { tool: "rg", @@ -389,7 +404,7 @@ it("notifications: only fresh installs notify", () => { version: "15.2.0", }; - assert.deepEqual(installNotifications([system, bundled]), []); + assert.deepEqual(installNotifications([system, cached]), []); const messages = installNotifications([system, installed]); assert.equal(messages.length, 1); assert.match(messages[0], /downloaded rg 15\.2\.0/); diff --git a/extensions/file-search/index.ts b/extensions/file-search/index.ts index ceacfa78..9b86ee96 100644 --- a/extensions/file-search/index.ts +++ b/extensions/file-search/index.ts @@ -3,10 +3,11 @@ * * On session start the extension resolves a usable binary for each tool: * a normally installed system binary is preferred (silently), then an - * existing fallback in this repo's `bin/` directory (silently), and only - * when neither exists is an official release downloaded into `bin/` — the - * single case that shows a UI notification. Tools await that initialization - * before executing, and report a clear error if it failed. + * existing binary in the agent's managed bin directory (`~/.pi/agent/bin`, + * silently — e.g. cached by an earlier download), and only when neither + * exists is an official release downloaded into that directory — the single + * case that shows a UI notification. Tools await that initialization before + * executing, and report a clear error if it failed. */ import * as NodeServices from "@effect/platform-node/NodeServices"; @@ -33,7 +34,7 @@ import { import { currentTarget, liveBinaryEnv, - repositoryBinDir, + managedBinDir, resolveBinary, TOOL_SPECS, type BinaryEnv, @@ -80,7 +81,7 @@ export function installNotifications(binaries: readonly ResolvedBinary[]) { .map( (binary) => `file-search: no system ${binary.tool} found — downloaded ${binary.tool} ${binary.version ?? ""}`.trimEnd() + - ` to ${repositoryBinDir()}`, + ` to ${managedBinDir()}`, ); } @@ -136,7 +137,7 @@ export default function fileSearchTools(pi: ExtensionAPI) { } }; - const binDir = repositoryBinDir(); + const binDir = managedBinDir(); const target = currentTarget(); const initializers = makeBinaryInitializers(binDir, target, liveBinaryEnv); diff --git a/extensions/file-search/src/binaries.ts b/extensions/file-search/src/binaries.ts index 405935d2..88db195d 100644 --- a/extensions/file-search/src/binaries.ts +++ b/extensions/file-search/src/binaries.ts @@ -3,21 +3,22 @@ * * Resolution order (per tool, first usable wins): * 1. A normally installed system binary (`fd`/`fdfind`, `rg`) — used silently. - * 2. An existing fallback in this repository's `bin/` directory — used silently. - * 3. A fresh download of an official release into `bin/` — the only case that - * should surface a UI notification. + * 2. An existing binary in the agent's managed bin directory + * (`~/.pi/agent/bin`), cached by an earlier download — used silently. + * 3. A fresh download of an official release into that directory — the only + * case that should surface a UI notification. * * The decision logic is an Effect over a small injectable environment * (`BinaryEnv`) so tests can drive it without touching the filesystem or the * network. `liveBinaryEnv` is the real implementation. */ +import { getAgentDir } from "@earendil-works/pi-coding-agent"; import * as NodeHttpClient from "@effect/platform-node/NodeHttpClient"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { execFile } from "node:child_process"; import { tmpdir } from "node:os"; import { dirname, join } from "node:path"; -import { fileURLToPath } from "node:url"; import { promisify } from "node:util"; import { Crypto, Data, Effect, Encoding, FileSystem, Stream } from "effect"; import { FetchHttpClient, HttpClient } from "effect/unstable/http"; @@ -56,13 +57,13 @@ const RG_SHA256: Readonly> = { }; export type ToolName = "fd" | "rg"; -export type BinarySource = "system" | "bundled" | "installed"; +export type BinarySource = "system" | "cached" | "installed"; export interface ToolSpec { readonly tool: ToolName; /** Commands probed on PATH, in order. Debian/Ubuntu install fd as `fdfind`. */ readonly systemCommands: readonly string[]; - /** Executable name used inside release archives and the repo bin directory. */ + /** Executable name used inside release archives and the managed bin directory. */ readonly binaryName: string; } @@ -144,10 +145,9 @@ export function currentTarget(): PlatformTarget { return { os: process.platform, arch: process.arch }; } -/** Repository root (`~/.pi/agent`) resolved from this module's location. */ -export function repositoryBinDir() { - const moduleDir = dirname(fileURLToPath(import.meta.url)); - return join(moduleDir, "..", "..", "..", "bin"); +/** Pi agent's managed binaries directory (`~/.pi/agent/bin`). */ +export function managedBinDir() { + return join(getAgentDir(), "bin"); } export class UnsupportedPlatformError extends Data.TaggedError( @@ -179,7 +179,7 @@ export interface ResolvedBinary { readonly version?: string; } -/** Resolve one tool: system binary, existing bin fallback, or fresh install. */ +/** Resolve one tool: system binary, cached fallback, or fresh install. */ export function resolveBinary( spec: ToolSpec, binDir: string, @@ -193,9 +193,9 @@ export function resolveBinary( } } - const bundled = join(binDir, spec.binaryName); - if (yield* env.probe(bundled, spec.tool)) { - return { tool: spec.tool, command: bundled, source: "bundled" as const }; + const cached = join(binDir, spec.binaryName); + if (yield* env.probe(cached, spec.tool)) { + return { tool: spec.tool, command: cached, source: "cached" as const }; } const asset = releaseAsset(spec.tool, target); @@ -205,17 +205,17 @@ export function resolveBinary( }); } - yield* env.install(asset, bundled); + yield* env.install(asset, cached); - if (!(yield* env.probe(bundled, spec.tool))) { + if (!(yield* env.probe(cached, spec.tool))) { return yield* new InstallError({ - message: `${spec.tool} ${asset.version} was installed to ${bundled} but failed to run.`, + message: `${spec.tool} ${asset.version} was installed to ${cached} but failed to run.`, }); } return { tool: spec.tool, - command: bundled, + command: cached, source: "installed" as const, version: asset.version, }; From 636b1abd74d414b8f709c455c79fea5259cf4709 Mon Sep 17 00:00:00 2001 From: tt-a1i Date: Sat, 22 Aug 2026 20:03:39 +0800 Subject: [PATCH 3/7] feat(subagents): add first-response watchdog to the subagent manager MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Direct subagents had no guard against a provider that accepts a request but never emits its first assistant event: the run stayed "running" forever and occupied one of the model pool's concurrency slots, while the workflow runner already settles such runs via its 45s first-response watchdog (the asymmetry flagged in the peer runtime audit). Arm a per-run watchdog in the manager, mirroring the workflow runner's semantics: it starts when a run becomes live (spawn or RunStarted) and clears on the first assistant event (AssistantDelta/AssistantMessage). On expiry the run settles as an explicit failure through the normal settle path — releasing its slot and letting the result flow back — and the stalled session is force-disposed like the abort-deadline path so it cannot revive into a zombie. The timeout defaults to FIRST_RESPONSE_TIMEOUT_MS (45s, matching the workflow runner) and is injectable via makeSubagentManagerLayer for tests. The stub backend gains a "HANG:" prompt mode standing in for a stalled provider request; manager.test.ts covers both the timeout-settles-and-frees-slot and first-response-clears cases. --- extensions/subagents/manager.test.ts | 94 +- extensions/subagents/src/backends/stub.ts | 7 + extensions/subagents/src/manager.ts | 1150 +++++++++++---------- 3 files changed, 710 insertions(+), 541 deletions(-) diff --git a/extensions/subagents/manager.test.ts b/extensions/subagents/manager.test.ts index b3bca7ff..50372091 100644 --- a/extensions/subagents/manager.test.ts +++ b/extensions/subagents/manager.test.ts @@ -12,10 +12,11 @@ import { BackendRegistry, type SubagentBackend } from "./src/backend.ts"; import { makeStubBackend } from "./src/backends/stub.ts"; import type { BackendName, ParentContext, SpawnTask } from "./src/domain.ts"; import { + makeSubagentManagerLayer, MAX_RUNNING, MAX_RUNNING_BTW, SubagentManager, - SubagentManagerLive, + type SubagentManagerConfig, type SubagentManagerShape, } from "./src/manager.ts"; import { runTool } from "./src/runtime.ts"; @@ -35,9 +36,11 @@ const TestRegistryLive = Layer.sync(BackendRegistry, () => { ); }); -const createTestRuntime = () => +const createTestRuntime = (managerConfig?: SubagentManagerConfig) => ManagedRuntime.make( - SubagentManagerLive.pipe(Layer.provide(TestRegistryLive)), + makeSubagentManagerLayer(managerConfig).pipe( + Layer.provide(TestRegistryLive), + ), ); const parent: ParentContext = { @@ -54,8 +57,9 @@ async function withManager( manager: SubagentManagerShape, runtime: ReturnType, ) => Promise, + managerConfig?: SubagentManagerConfig, ) { - const runtime = createTestRuntime(); + const runtime = createTestRuntime(managerConfig); try { const manager = await runtime.runPromise(SubagentManager); await run(manager, runtime); @@ -348,3 +352,85 @@ test("a wait issued alongside a restart does not return the stale run", () => { assert.match(after?.finalText ?? "", /Second/); }); }); + +test("a run with no first response is settled by the watchdog and frees its slot", async () => { + await withManager( + async (manager, runtime) => { + const settled: Array<{ id: string; status: string; consumed: boolean }> = + []; + manager.view.setOnSettled((snap, consumed) => + settled.push({ id: snap.id, status: snap.status, consumed }), + ); + + // Fill every model slot with runs that accept the prompt but never + // emit a first assistant event (stalled provider requests). + const hung = await runTool( + runtime, + Effect.forEach( + [1, 2, 3, 4], + (n) => manager.spawn("pi", task(`HANG: stall ${n}`)), + { concurrency: "unbounded" }, + ), + ); + await assert.rejects( + runTool(runtime, manager.spawn("pi", task("Task 5"))), + /Max 4 subagent sessions/, + ); + + // The watchdog settles each hung run as an explicit error through the + // normal settle path (so waits and result delivery observe it). + await runTool(runtime, manager.waitFor(hung.map((snap) => snap.id))); + for (const snap of hung) { + const failed = manager.view.get(snap.id); + assert.equal(failed?.status, "error"); + assert.match( + failed?.errorText ?? "", + /no assistant response event.*provider request may be stalled/, + ); + } + assert.deepEqual( + settled.sort((a, b) => a.id.localeCompare(b.id)), + hung + .map((snap) => ({ + id: snap.id, + status: "error", + consumed: true, + })) + .sort((a, b) => a.id.localeCompare(b.id)), + ); + + // The freed slots accept new spawns again. + const fresh = await runTool( + runtime, + manager.spawn("pi", task("ok after the stall")), + ); + assert.equal(fresh.status, "running"); + await runTool(runtime, manager.waitFor([fresh.id])); + assert.equal(manager.view.get(fresh.id)?.status, "done"); + }, + { firstResponseTimeoutMs: 150 }, + ); +}); + +test("a first response clears the watchdog so slower runs are not killed", async () => { + await withManager( + async (manager, runtime) => { + // The stub streams its first assistant delta within one cadence + // (~40ms) but needs well over the watchdog budget to finish the whole + // turn; without clearing on first response it would be killed mid-run. + const snap = await runTool( + runtime, + manager.spawn("pi", task("Slow but responsive")), + ); + await runTool(runtime, manager.waitFor([snap.id])); + const done = manager.view.get(snap.id); + assert.equal(done?.status, "done"); + assert.match( + done?.finalText ?? "", + /\[stub:pi\] completed: Slow but responsive/, + ); + assert.equal(done?.errorText, undefined); + }, + { firstResponseTimeoutMs: 250 }, + ); +}); diff --git a/extensions/subagents/src/backends/stub.ts b/extensions/subagents/src/backends/stub.ts index fc87f65c..5b77417b 100644 --- a/extensions/subagents/src/backends/stub.ts +++ b/extensions/subagents/src/backends/stub.ts @@ -10,6 +10,8 @@ * - supports interrupt (RunSettled Interrupted -> status "error", matching v1); * - fails the run when the prompt starts with "FAIL:", and refuses to spawn * at all when it starts with "SPAWNFAIL:" (error-path testing); + * - hangs the run after RunStarted without any assistant event when the + * prompt starts with "HANG:" (first-response watchdog testing); * - appends every event to a JSONL "session file" in tmpdir so the * "full transcript in session file" pointers resolve. */ @@ -114,6 +116,11 @@ const makeStubSession = ( Effect.gen(function* () { yield* emit({ _tag: "RunStarted" }); const failing = userText.trimStart().startsWith("FAIL:"); + // Stand in for a provider that accepts the request but never emits + // its first assistant event. + if (userText.trimStart().startsWith("HANG:")) { + return yield* Effect.never; + } const thinking = "Looking at the task and planning an approach..."; for (const delta of chunked(thinking, 16)) { diff --git a/extensions/subagents/src/manager.ts b/extensions/subagents/src/manager.ts index 9941a295..995e0385 100644 --- a/extensions/subagents/src/manager.ts +++ b/extensions/subagents/src/manager.ts @@ -9,6 +9,11 @@ * The manager also exposes a synchronous `SubagentReadModel` so the * imperative TUI components (which render synchronously) can read snapshots * and issue fire-and-forget commands without touching the Effect runtime. + * + * Every run is guarded by a first-response watchdog: a provider that accepts + * the request but never emits its first assistant event is settled as a + * failure (releasing its concurrency slot) instead of hanging forever, + * mirroring the workflow runner's watchdog. */ import { @@ -54,6 +59,13 @@ export const MAX_TRACKED = 64; const STOP_TIMEOUT_MS = 5_000; /** Session abort/shutdown (5s) plus bounded direct-worktree cleanup (4s). */ const ENTRY_CLOSE_TIMEOUT_MS = 10_000; +/** + * First-response watchdog: a run whose provider accepts the request but + * never emits an assistant event is settled as a failure so it cannot + * occupy a concurrency slot forever. Matches the workflow runner's + * FIRST_RESPONSE_TIMEOUT_MS (extensions/workflows/runner.ts). + */ +export const FIRST_RESPONSE_TIMEOUT_MS = 45_000; const ERROR_TEXT_MAX_LENGTH = 4_096; const TRANSCRIPT_TEXT_MAX_LENGTH = 64 * 1_024; const LIVE_ASSISTANT_MAX_LENGTH = 128 * 1_024; @@ -64,6 +76,10 @@ function bounded(text: string) { return text.slice(0, ERROR_TEXT_MAX_LENGTH); } +function formatWatchdogTimeout(ms: number) { + return ms % 1_000 === 0 ? `${ms / 1_000} seconds` : `${ms} ms`; +} + function boundedTranscriptText(text: string) { return text.slice(0, TRANSCRIPT_TEXT_MAX_LENGTH); } @@ -108,6 +124,8 @@ interface Entry { scope: Scope.Closeable; pump?: Fiber.Fiber; liveToolMap: Map; + /** First-response watchdog timer for the active (or just-armed) run. */ + watchdogTimer?: ReturnType; /** Idle restart dispatched but RunStarted not folded yet; counts as running * so concurrent restarts cannot race past the cap. */ restarting?: boolean; @@ -183,589 +201,647 @@ export class SubagentManager extends Context.Service< // --- Implementation -------------------------------------------------------------- -const makeManager = Effect.gen(function* () { - const registry = yield* BackendRegistry; - // Detached forker for sync contexts (read-model commands, pruning) that - // preserves the manager's services instead of using the global runtime. - const runDetached = Effect.runForkWith(yield* Effect.context()); - - const entries = new Map(); - const waitInterest = new Map(); - const listeners = new Set<() => void>(); - /** One-shot nextChange waiters, swapped out before invocation so waiters - * re-registering during notification are not visited in the same sweep. */ - let changeWaiters: Array<() => void> = []; - const idListeners = new Map void>>(); - const cleanups = new Set>(); - let modelCounter = 0; - let btwCounter = 0; - // Reservations are tracked per pool so the model and user "by the way" asides - // never contend for the same slots. - let reservedModel = 0; - let reservedBtw = 0; - let disposed = false; - let onSettled: - | ((snap: SubagentSnapshot, consumed: boolean) => void) - | undefined; - - const notify = (id?: string) => { - const waiters = changeWaiters; - changeWaiters = []; - for (const waiter of waiters) waiter(); - for (const listener of [...listeners]) { - try { - listener(); - } catch { - // A failed status/render listener must not corrupt lifecycle state. - } - } - if (id) { - for (const listener of idListeners.get(id) ?? []) { +const makeManager = (config: SubagentManagerConfig = {}) => + Effect.gen(function* () { + const firstResponseTimeoutMs = + config.firstResponseTimeoutMs ?? FIRST_RESPONSE_TIMEOUT_MS; + const registry = yield* BackendRegistry; + // Detached forker for sync contexts (read-model commands, pruning) that + // preserves the manager's services instead of using the global runtime. + const runDetached = Effect.runForkWith(yield* Effect.context()); + + const entries = new Map(); + const waitInterest = new Map(); + const listeners = new Set<() => void>(); + /** One-shot nextChange waiters, swapped out before invocation so waiters + * re-registering during notification are not visited in the same sweep. */ + let changeWaiters: Array<() => void> = []; + const idListeners = new Map void>>(); + const cleanups = new Set>(); + let modelCounter = 0; + let btwCounter = 0; + // Reservations are tracked per pool so the model and user "by the way" asides + // never contend for the same slots. + let reservedModel = 0; + let reservedBtw = 0; + let disposed = false; + let onSettled: + | ((snap: SubagentSnapshot, consumed: boolean) => void) + | undefined; + + const notify = (id?: string) => { + const waiters = changeWaiters; + changeWaiters = []; + for (const waiter of waiters) waiter(); + for (const listener of [...listeners]) { try { listener(); } catch { - // Same. + // A failed status/render listener must not corrupt lifecycle state. + } + } + if (id) { + for (const listener of idListeners.get(id) ?? []) { + try { + listener(); + } catch { + // Same. + } } } - } - }; - - /** Resolves on the next state change. Interruption unregisters the waiter. */ - const nextChange = Effect.callback((resume) => { - const waiter = () => resume(Effect.void); - changeWaiters.push(waiter); - return Effect.sync(() => { - const index = changeWaiters.indexOf(waiter); - if (index >= 0) changeWaiters.splice(index, 1); + }; + + /** Resolves on the next state change. Interruption unregisters the waiter. */ + const nextChange = Effect.callback((resume) => { + const waiter = () => resume(Effect.void); + changeWaiters.push(waiter); + return Effect.sync(() => { + const index = changeWaiters.indexOf(waiter); + if (index >= 0) changeWaiters.splice(index, 1); + }); }); - }); - /** - * A restart dispatched by `send` occupies a slot immediately, but the - * `RunStarted` that flips `snapshot.status` only arrives on the async pump. - * Every caller that asks "is this busy?" must honor that window, or a - * wait/cancel issued in the same turn as the restart would observe the old - * settled run and return (or cancel) the wrong thing. - */ - const isBusy = (entry: Entry | undefined) => - entry !== undefined && - (entry.snapshot.status === "running" || entry.restarting === true); - - const runningCount = (origin?: SubagentOrigin) => - [...entries.values()].filter( - (e) => - isBusy(e) && (origin === undefined || e.snapshot.origin === origin), - ).length; - - /** Per-pool capacity: model asides and user "by the way" asides never mix. */ - const poolLimit = (origin: SubagentOrigin) => - origin === "btw" ? MAX_RUNNING_BTW : MAX_RUNNING; - const poolReserved = (origin: SubagentOrigin) => - origin === "btw" ? reservedBtw : reservedModel; - const atPoolCapacity = (origin: SubagentOrigin) => - runningCount(origin) + poolReserved(origin) >= poolLimit(origin); - - const addInterest = (ids: ReadonlyArray) => { - for (const id of ids) waitInterest.set(id, (waitInterest.get(id) ?? 0) + 1); - }; - const releaseInterest = (ids: ReadonlyArray) => { - for (const id of ids) { - const count = (waitInterest.get(id) ?? 1) - 1; - if (count <= 0) waitInterest.delete(id); - else waitInterest.set(id, count); - } - }; - - const closeEntryScope = (entry: Entry) => - Scope.close(entry.scope, Exit.void).pipe(Effect.ignore); - - const pruneSettled = () => { - if (entries.size <= MAX_TRACKED) return; - const candidates = [...entries.values()] - .filter((e) => !isBusy(e) && !waitInterest.has(e.snapshot.id)) - .sort( - (a, b) => - (a.snapshot.settledAt ?? a.snapshot.createdAt) - - (b.snapshot.settledAt ?? b.snapshot.createdAt), - ); - for (const entry of candidates) { - if (entries.size <= MAX_TRACKED) break; - entries.delete(entry.snapshot.id); - const fiber = runDetached(closeEntryScope(entry)); - cleanups.add(fiber); - fiber.addObserver(() => cleanups.delete(fiber)); - } - }; - - const settle = (entry: Entry, outcome: RunOutcome) => { - const s = entry.snapshot; - const wasRestarting = entry.restarting === true; - entry.restarting = false; - if (s.status !== "running") { - if (!wasRestarting) return; - // A cancel can clear a queued restart before RunStarted reaches the - // manager. Its RunSettled still belongs to the new run, not the old - // settled snapshot, so promote the lifecycle before applying it. - s.status = "running"; - s.settledAt = undefined; - s.errorText = undefined; - } - s.settledAt = Date.now(); - switch (outcome._tag) { - case "Completed": - s.status = "done"; - s.errorText = undefined; - s.finalText = outcome.finalText.slice(0, FINAL_TEXT_MAX_LENGTH); - break; - case "Failed": - s.status = "error"; - s.errorText = bounded(outcome.errorText); - // Never let a failed run report the previous run's successful output. - s.finalText = (outcome.partialText ?? "").slice( - 0, - FINAL_TEXT_MAX_LENGTH, - ); - break; - case "Interrupted": - s.status = "error"; - s.errorText = "Run was aborted"; - s.finalText = (outcome.partialText ?? "").slice( - 0, - FINAL_TEXT_MAX_LENGTH, + /** + * A restart dispatched by `send` occupies a slot immediately, but the + * `RunStarted` that flips `snapshot.status` only arrives on the async pump. + * Every caller that asks "is this busy?" must honor that window, or a + * wait/cancel issued in the same turn as the restart would observe the old + * settled run and return (or cancel) the wrong thing. + */ + const isBusy = (entry: Entry | undefined) => + entry !== undefined && + (entry.snapshot.status === "running" || entry.restarting === true); + + const runningCount = (origin?: SubagentOrigin) => + [...entries.values()].filter( + (e) => + isBusy(e) && (origin === undefined || e.snapshot.origin === origin), + ).length; + + /** Per-pool capacity: model asides and user "by the way" asides never mix. */ + const poolLimit = (origin: SubagentOrigin) => + origin === "btw" ? MAX_RUNNING_BTW : MAX_RUNNING; + const poolReserved = (origin: SubagentOrigin) => + origin === "btw" ? reservedBtw : reservedModel; + const atPoolCapacity = (origin: SubagentOrigin) => + runningCount(origin) + poolReserved(origin) >= poolLimit(origin); + + const addInterest = (ids: ReadonlyArray) => { + for (const id of ids) + waitInterest.set(id, (waitInterest.get(id) ?? 0) + 1); + }; + const releaseInterest = (ids: ReadonlyArray) => { + for (const id of ids) { + const count = (waitInterest.get(id) ?? 1) - 1; + if (count <= 0) waitInterest.delete(id); + else waitInterest.set(id, count); + } + }; + + const closeEntryScope = (entry: Entry) => + Scope.close(entry.scope, Exit.void).pipe(Effect.ignore); + + const pruneSettled = () => { + if (entries.size <= MAX_TRACKED) return; + const candidates = [...entries.values()] + .filter((e) => !isBusy(e) && !waitInterest.has(e.snapshot.id)) + .sort( + (a, b) => + (a.snapshot.settledAt ?? a.snapshot.createdAt) - + (b.snapshot.settledAt ?? b.snapshot.createdAt), ); - break; - } - s.liveAssistant = undefined; - entry.liveToolMap.clear(); - s.liveTools = []; - s.queued = []; - const consumed = (waitInterest.get(s.id) ?? 0) > 0; - notify(s.id); - try { - // During teardown, don't queue results into a shutting-down session. - if (!disposed) onSettled?.(s, consumed); - } catch { - // The parent session may be unavailable; settlement stays final. - } - pruneSettled(); - }; - - const foldEvent = (entry: Entry, event: SubagentEvent) => { - const s = entry.snapshot; - switch (event._tag) { - case "RunStarted": - entry.restarting = false; + for (const entry of candidates) { + if (entries.size <= MAX_TRACKED) break; + entries.delete(entry.snapshot.id); + const fiber = runDetached(closeEntryScope(entry)); + cleanups.add(fiber); + fiber.addObserver(() => cleanups.delete(fiber)); + } + }; + + const settle = (entry: Entry, outcome: RunOutcome) => { + clearWatchdog(entry); + const s = entry.snapshot; + const wasRestarting = entry.restarting === true; + entry.restarting = false; + if (s.status !== "running") { + if (!wasRestarting) return; + // A cancel can clear a queued restart before RunStarted reaches the + // manager. Its RunSettled still belongs to the new run, not the old + // settled snapshot, so promote the lifecycle before applying it. s.status = "running"; s.settledAt = undefined; s.errorText = undefined; - break; - case "RunSettled": - settle(entry, event.outcome); - return; // settle() already notified - case "UserMessage": - appendTranscript(s, { - kind: "user", - text: boundedTranscriptText(event.text), - }); - break; - case "AssistantDelta": { - const live = s.liveAssistant ?? { text: "", thinking: "" }; - s.liveAssistant = - event.kind === "text" - ? { - ...live, - text: (live.text + event.delta).slice( - -LIVE_ASSISTANT_MAX_LENGTH, - ), - } - : { - ...live, - thinking: (live.thinking + event.delta).slice( - -LIVE_ASSISTANT_MAX_LENGTH, - ), - }; - break; } - case "AssistantMessage": - appendTranscript(s, { - kind: "assistant", - parts: event.parts.map((part) => - part.type === "toolCall" + s.settledAt = Date.now(); + switch (outcome._tag) { + case "Completed": + s.status = "done"; + s.errorText = undefined; + s.finalText = outcome.finalText.slice(0, FINAL_TEXT_MAX_LENGTH); + break; + case "Failed": + s.status = "error"; + s.errorText = bounded(outcome.errorText); + // Never let a failed run report the previous run's successful output. + s.finalText = (outcome.partialText ?? "").slice( + 0, + FINAL_TEXT_MAX_LENGTH, + ); + break; + case "Interrupted": + s.status = "error"; + s.errorText = "Run was aborted"; + s.finalText = (outcome.partialText ?? "").slice( + 0, + FINAL_TEXT_MAX_LENGTH, + ); + break; + } + s.liveAssistant = undefined; + entry.liveToolMap.clear(); + s.liveTools = []; + s.queued = []; + const consumed = (waitInterest.get(s.id) ?? 0) > 0; + notify(s.id); + try { + // During teardown, don't queue results into a shutting-down session. + if (!disposed) onSettled?.(s, consumed); + } catch { + // The parent session may be unavailable; settlement stays final. + } + pruneSettled(); + }; + + /** Stop the first-response watchdog (first response arrived / run settled). */ + const clearWatchdog = (entry: Entry) => { + if (entry.watchdogTimer !== undefined) { + clearTimeout(entry.watchdogTimer); + entry.watchdogTimer = undefined; + } + }; + + /** Settle a run whose provider never emitted a first assistant response. */ + const watchdogExpired = (entry: Entry) => { + entry.watchdogTimer = undefined; + if (!isBusy(entry)) return; + const model = entry.snapshot.meta.modelLabel; + settle(entry, { + _tag: "Failed", + errorText: `Agent received no assistant response event${model ? ` for ${model}` : ""} within ${formatWatchdogTimeout(firstResponseTimeoutMs)}; the provider request may be stalled. Retry the subagent.`, + }); + // The stalled session cannot be trusted to abort cooperatively; dispose + // it like the abort-deadline path so it cannot revive into a zombie run. + const fiber = runDetached( + closeEntryScope(entry).pipe( + Effect.timeout(ENTRY_CLOSE_TIMEOUT_MS), + Effect.ignore, + ), + ); + cleanups.add(fiber); + fiber.addObserver(() => cleanups.delete(fiber)); + }; + + /** Arm the first-response watchdog for the entry's current run. */ + const armWatchdog = (entry: Entry) => { + clearWatchdog(entry); + entry.watchdogTimer = setTimeout( + () => watchdogExpired(entry), + firstResponseTimeoutMs, + ); + }; + + const foldEvent = (entry: Entry, event: SubagentEvent) => { + const s = entry.snapshot; + switch (event._tag) { + case "RunStarted": + entry.restarting = false; + s.status = "running"; + s.settledAt = undefined; + s.errorText = undefined; + armWatchdog(entry); + break; + case "RunSettled": + settle(entry, event.outcome); + return; // settle() already notified + case "UserMessage": + appendTranscript(s, { + kind: "user", + text: boundedTranscriptText(event.text), + }); + break; + case "AssistantDelta": { + clearWatchdog(entry); + const live = s.liveAssistant ?? { text: "", thinking: "" }; + s.liveAssistant = + event.kind === "text" ? { - ...part, - argsPreview: part.argsPreview - ? boundedTranscriptText(part.argsPreview) - : undefined, + ...live, + text: (live.text + event.delta).slice( + -LIVE_ASSISTANT_MAX_LENGTH, + ), } - : { ...part, text: boundedTranscriptText(part.text) }, - ), - }); - s.liveAssistant = undefined; - s.turns++; - break; - case "ToolStart": - entry.liveToolMap.set(event.toolId, { - toolId: event.toolId, - name: event.name, - argsPreview: event.argsPreview - ? boundedTranscriptText(event.argsPreview) - : undefined, - }); - s.liveTools = [...entry.liveToolMap.values()]; - break; - case "ToolUpdate": { - const current = entry.liveToolMap.get(event.toolId); - if (current) { + : { + ...live, + thinking: (live.thinking + event.delta).slice( + -LIVE_ASSISTANT_MAX_LENGTH, + ), + }; + break; + } + case "AssistantMessage": + clearWatchdog(entry); + appendTranscript(s, { + kind: "assistant", + parts: event.parts.map((part) => + part.type === "toolCall" + ? { + ...part, + argsPreview: part.argsPreview + ? boundedTranscriptText(part.argsPreview) + : undefined, + } + : { ...part, text: boundedTranscriptText(part.text) }, + ), + }); + s.liveAssistant = undefined; + s.turns++; + break; + case "ToolStart": entry.liveToolMap.set(event.toolId, { - ...current, - outputPreview: event.outputPreview - ? boundedTranscriptText(event.outputPreview) - : current.outputPreview, + toolId: event.toolId, + name: event.name, + argsPreview: event.argsPreview + ? boundedTranscriptText(event.argsPreview) + : undefined, }); s.liveTools = [...entry.liveToolMap.values()]; + break; + case "ToolUpdate": { + const current = entry.liveToolMap.get(event.toolId); + if (current) { + entry.liveToolMap.set(event.toolId, { + ...current, + outputPreview: event.outputPreview + ? boundedTranscriptText(event.outputPreview) + : current.outputPreview, + }); + s.liveTools = [...entry.liveToolMap.values()]; + } + break; } - break; + case "ToolEnd": + entry.liveToolMap.delete(event.toolId); + s.liveTools = [...entry.liveToolMap.values()]; + appendTranscript(s, { + kind: "toolResult", + toolId: event.toolId, + name: event.name, + isError: event.isError, + outputPreview: event.outputPreview + ? boundedTranscriptText(event.outputPreview) + : undefined, + }); + break; + case "QueueChanged": + s.queued = event.queued; + break; + case "UsageChanged": + s.usage = { + tokens: event.tokens ?? s.usage.tokens, + contextWindow: event.contextWindow ?? s.usage.contextWindow, + }; + break; + case "MetaChanged": + s.meta = { ...s.meta, ...event.meta }; + break; + case "BackendError": + s.errorText = bounded(event.message); + break; } - case "ToolEnd": - entry.liveToolMap.delete(event.toolId); - s.liveTools = [...entry.liveToolMap.values()]; - appendTranscript(s, { - kind: "toolResult", - toolId: event.toolId, - name: event.name, - isError: event.isError, - outputPreview: event.outputPreview - ? boundedTranscriptText(event.outputPreview) - : undefined, - }); - break; - case "QueueChanged": - s.queued = event.queued; - break; - case "UsageChanged": - s.usage = { - tokens: event.tokens ?? s.usage.tokens, - contextWindow: event.contextWindow ?? s.usage.contextWindow, - }; - break; - case "MetaChanged": - s.meta = { ...s.meta, ...event.meta }; - break; - case "BackendError": - s.errorText = bounded(event.message); - break; - } - notify(s.id); - }; - - const spawn = (backendName: BackendName, task: SpawnTask) => - Effect.gen(function* () { - const origin: SubagentOrigin = task.origin ?? "model"; - // Reserve synchronously (before the first yield inside doSpawn) so - // parallel tool calls cannot race past the pool cap. - yield* Effect.suspend( - (): Effect.Effect => { - if (disposed) { - return new SpawnError({ - message: "Subagent manager is shutting down.", + notify(s.id); + }; + + const spawn = (backendName: BackendName, task: SpawnTask) => + Effect.gen(function* () { + const origin: SubagentOrigin = task.origin ?? "model"; + // Reserve synchronously (before the first yield inside doSpawn) so + // parallel tool calls cannot race past the pool cap. + yield* Effect.suspend( + (): Effect.Effect => { + if (disposed) { + return new SpawnError({ + message: "Subagent manager is shutting down.", + }); + } + if (atPoolCapacity(origin)) { + return new ConcurrencyLimitError({ + message: `Max ${poolLimit(origin)} ${ + origin === "btw" ? "by-the-way" : "subagent" + } sessions can run concurrently. Wait for one to finish before spawning another.`, + }); + } + if (origin === "btw") reservedBtw++; + else reservedModel++; + return Effect.void; + }, + ); + + const doSpawn = Effect.gen(function* () { + const backend: SubagentBackend | undefined = + registry.get(backendName); + if (!backend) { + return yield* new BackendUnavailableError({ + message: `Unknown backend "${backendName}".`, }); } - if (atPoolCapacity(origin)) { - return new ConcurrencyLimitError({ - message: `Max ${poolLimit(origin)} ${ - origin === "btw" ? "by-the-way" : "subagent" - } sessions can run concurrently. Wait for one to finish before spawning another.`, + const scope = yield* Scope.make(); + const session = yield* Scope.provide(backend.spawn(task), scope).pipe( + Effect.onError(() => Scope.close(scope, Exit.void)), + ); + if (disposed) { + yield* Scope.close(scope, Exit.void); + return yield* new SpawnError({ + message: "Subagent manager shut down while spawning.", }); } - if (origin === "btw") reservedBtw++; - else reservedModel++; - return Effect.void; - }, - ); - const doSpawn = Effect.gen(function* () { - const backend: SubagentBackend | undefined = registry.get(backendName); - if (!backend) { - return yield* new BackendUnavailableError({ - message: `Unknown backend "${backendName}".`, - }); - } - const scope = yield* Scope.make(); - const session = yield* Scope.provide(backend.spawn(task), scope).pipe( - Effect.onError(() => Scope.close(scope, Exit.void)), - ); - if (disposed) { - yield* Scope.close(scope, Exit.void); - return yield* new SpawnError({ - message: "Subagent manager shut down while spawning.", - }); - } + const id = + origin === "btw" ? `btw-${++btwCounter}` : `sa-${++modelCounter}`; + const meta = yield* session.meta; + const entry: Entry = { + snapshot: { + id, + origin, + backend: backendName, + title: task.title, + prompt: task.prompt, + cwd: task.cwd, + status: "running", + createdAt: Date.now(), + meta, + usage: { contextWindow: meta.contextWindow }, + transcript: [], + liveTools: [], + queued: [], + finalText: "", + turns: 0, + }, + session, + scope, + liveToolMap: new Map(), + }; + entries.set(id, entry); + // The run is live from the caller's perspective before RunStarted + // reaches the pump; guard that window too. + armWatchdog(entry); + + // Pump: fold the event stream into the snapshot. Tied to the entry + // scope, so closing the scope stops it. If the stream ends while the + // subagent still looks running, the backend died out from under us. + const pump = Stream.runForEach(session.events, (event) => + Effect.sync(() => foldEvent(entry, event)), + ).pipe( + Effect.ensuring( + Effect.sync(() => { + if (entry.snapshot.status === "running") { + settle(entry, { + _tag: "Failed", + errorText: "Backend event stream ended unexpectedly", + }); + } + }), + ), + ); + entry.pump = yield* Scope.provide(Effect.forkScoped(pump), scope); - const id = - origin === "btw" ? `btw-${++btwCounter}` : `sa-${++modelCounter}`; - const meta = yield* session.meta; - const entry: Entry = { - snapshot: { - id, - origin, - backend: backendName, - title: task.title, - prompt: task.prompt, - cwd: task.cwd, - status: "running", - createdAt: Date.now(), - meta, - usage: { contextWindow: meta.contextWindow }, - transcript: [], - liveTools: [], - queued: [], - finalText: "", - turns: 0, - }, - session, - scope, - liveToolMap: new Map(), - }; - entries.set(id, entry); - - // Pump: fold the event stream into the snapshot. Tied to the entry - // scope, so closing the scope stops it. If the stream ends while the - // subagent still looks running, the backend died out from under us. - const pump = Stream.runForEach(session.events, (event) => - Effect.sync(() => foldEvent(entry, event)), - ).pipe( + notify(id); + return entry.snapshot as SubagentSnapshot; + }); + + return yield* doSpawn.pipe( Effect.ensuring( Effect.sync(() => { - if (entry.snapshot.status === "running") { - settle(entry, { - _tag: "Failed", - errorText: "Backend event stream ended unexpectedly", - }); - } + if (origin === "btw") reservedBtw--; + else reservedModel--; + notify(); }), ), ); - entry.pump = yield* Scope.provide(Effect.forkScoped(pump), scope); - - notify(id); - return entry.snapshot as SubagentSnapshot; }); - return yield* doSpawn.pipe( - Effect.ensuring( - Effect.sync(() => { - if (origin === "btw") reservedBtw--; - else reservedModel--; - notify(); - }), - ), - ); - }); + const waitFor = ( + ids: ReadonlyArray, + onPending?: (pending: string[]) => void, + ) => + Effect.suspend(() => { + const unique = [...new Set(ids)]; + addInterest(unique); + const loop = Effect.gen(function* () { + while (true) { + const pending = unique.filter((id) => isBusy(entries.get(id))); + if (pending.length === 0) return; + onPending?.(pending); + yield* nextChange; + } + }); + return loop.pipe( + Effect.ensuring( + Effect.sync(() => { + releaseInterest(unique); + pruneSettled(); + }), + ), + ); + }); - const waitFor = ( - ids: ReadonlyArray, - onPending?: (pending: string[]) => void, - ) => - Effect.suspend(() => { - const unique = [...new Set(ids)]; - addInterest(unique); - const loop = Effect.gen(function* () { - while (true) { - const pending = unique.filter((id) => isBusy(entries.get(id))); - if (pending.length === 0) return; - onPending?.(pending); - yield* nextChange; + /** Interrupt one busy entry, including the pre-RunStarted restart window. */ + const abortEntry = (entry: Entry) => + Effect.gen(function* () { + if (!isBusy(entry)) return; + const graceful = yield* entry.session.interrupt.pipe( + Effect.timeout(STOP_TIMEOUT_MS), + Effect.result, + ); + if (Result.isFailure(graceful)) { + // Settle before closing the scope so the pump's stream-ended + // fallback ("Backend event stream ended unexpectedly") cannot win + // the race and report the wrong terminal reason. + yield* Effect.sync(() => { + settle(entry, { _tag: "Interrupted" }); + entry.snapshot.errorText = + "Abort deadline exceeded; session was force-disposed"; + notify(entry.snapshot.id); + }); + // Bound the close like disposeAll does: a stuck backend finalizer + // must not hang cancel after the run is already settled. + yield* closeEntryScope(entry).pipe( + Effect.timeout(ENTRY_CLOSE_TIMEOUT_MS), + Effect.ignore, + ); } }); - return loop.pipe( - Effect.ensuring( - Effect.sync(() => { - releaseInterest(unique); - pruneSettled(); - }), - ), - ); - }); - /** Interrupt one busy entry, including the pre-RunStarted restart window. */ - const abortEntry = (entry: Entry) => - Effect.gen(function* () { - if (!isBusy(entry)) return; - const graceful = yield* entry.session.interrupt.pipe( - Effect.timeout(STOP_TIMEOUT_MS), - Effect.result, - ); - if (Result.isFailure(graceful)) { - // Settle before closing the scope so the pump's stream-ended - // fallback ("Backend event stream ended unexpectedly") cannot win - // the race and report the wrong terminal reason. - yield* Effect.sync(() => { - settle(entry, { _tag: "Interrupted" }); - entry.snapshot.errorText = - "Abort deadline exceeded; session was force-disposed"; - notify(entry.snapshot.id); + const cancel = (ids: ReadonlyArray) => + Effect.suspend(() => { + const unique = [...new Set(ids)]; + const running = unique + .map((id) => entries.get(id)) + .filter((entry): entry is Entry => isBusy(entry)); + const runningIds = running.map((entry) => entry.snapshot.id); + // Mark consumed before interrupting so cancellation does not also + // enqueue duplicate automatic result messages into the parent. + addInterest(runningIds); + const work = Effect.gen(function* () { + yield* Effect.forEach(running, abortEntry, { + concurrency: "unbounded", + }); + while (running.some(isBusy)) yield* nextChange; }); - // Bound the close like disposeAll does: a stuck backend finalizer - // must not hang cancel after the run is already settled. - yield* closeEntryScope(entry).pipe( - Effect.timeout(ENTRY_CLOSE_TIMEOUT_MS), - Effect.ignore, + return work.pipe( + Effect.ensuring( + Effect.sync(() => { + releaseInterest(runningIds); + pruneSettled(); + }), + ), + Effect.map( + (): ReadonlyArray => + unique.map((id) => { + const snapshot = entries.get(id)?.snapshot; + return { + id, + title: snapshot?.title ?? "?", + status: snapshot?.status ?? "error", + cancelled: runningIds.includes(id), + }; + }), + ), ); - } - }); - - const cancel = (ids: ReadonlyArray) => - Effect.suspend(() => { - const unique = [...new Set(ids)]; - const running = unique - .map((id) => entries.get(id)) - .filter((entry): entry is Entry => isBusy(entry)); - const runningIds = running.map((entry) => entry.snapshot.id); - // Mark consumed before interrupting so cancellation does not also - // enqueue duplicate automatic result messages into the parent. - addInterest(runningIds); - const work = Effect.gen(function* () { - yield* Effect.forEach(running, abortEntry, { - concurrency: "unbounded", - }); - while (running.some(isBusy)) yield* nextChange; }); - return work.pipe( - Effect.ensuring( - Effect.sync(() => { - releaseInterest(runningIds); - pruneSettled(); - }), - ), - Effect.map( - (): ReadonlyArray => - unique.map((id) => { - const snapshot = entries.get(id)?.snapshot; - return { - id, - title: snapshot?.title ?? "?", - status: snapshot?.status ?? "error", - cancelled: runningIds.includes(id), - }; - }), - ), - ); - }); - const send = (id: string, text: string) => - Effect.suspend((): Effect.Effect => { - const entry = entries.get(id); - if (!entry || disposed) { - return new SendError({ - message: `Subagent "${id}" is no longer tracked.`, - }); - } - // Restarting a settled subagent occupies a running slot again, so it - // must respect the same cap as spawn. Steering an already-running one - // does not consume additional capacity. - if (!isBusy(entry)) { - const origin = entry.snapshot.origin; - if (atPoolCapacity(origin)) { + const send = (id: string, text: string) => + Effect.suspend((): Effect.Effect => { + const entry = entries.get(id); + if (!entry || disposed) { return new SendError({ - message: `Max ${poolLimit(origin)} ${ - origin === "btw" ? "by-the-way" : "subagent" - } sessions can run concurrently; restarting "${id}" would exceed that.`, + message: `Subagent "${id}" is no longer tracked.`, }); } - // Occupy the slot synchronously: the RunStarted that flips status - // arrives via the async pump, and two concurrent restarts must not - // both pass the check in that window. Cleared by RunStarted/settle, - // or here when the backend rejects the send. - entry.restarting = true; - return entry.session.send(text).pipe( - Effect.onError(() => - Effect.sync(() => { - entry.restarting = false; - notify(entry.snapshot.id); - }), + // Restarting a settled subagent occupies a running slot again, so it + // must respect the same cap as spawn. Steering an already-running one + // does not consume additional capacity. + if (!isBusy(entry)) { + const origin = entry.snapshot.origin; + if (atPoolCapacity(origin)) { + return new SendError({ + message: `Max ${poolLimit(origin)} ${ + origin === "btw" ? "by-the-way" : "subagent" + } sessions can run concurrently; restarting "${id}" would exceed that.`, + }); + } + // Occupy the slot synchronously: the RunStarted that flips status + // arrives via the async pump, and two concurrent restarts must not + // both pass the check in that window. Cleared by RunStarted/settle, + // or here when the backend rejects the send. + entry.restarting = true; + return entry.session.send(text).pipe( + Effect.onError(() => + Effect.sync(() => { + entry.restarting = false; + notify(entry.snapshot.id); + }), + ), + ); + } + return entry.session.send(text); + }); + + const disposeAll = Effect.gen(function* () { + disposed = true; + const all = [...entries.values()]; + for (const entry of all) clearWatchdog(entry); + entries.clear(); + yield* Effect.forEach( + all, + (entry) => + closeEntryScope(entry).pipe( + Effect.timeout(ENTRY_CLOSE_TIMEOUT_MS), + Effect.ignore, ), - ); - } - return entry.session.send(text); + { concurrency: "unbounded" }, + ); + // Pruning cleanups are detached; bound them like everything else so a + // stuck backend finalizer cannot block runtime shutdown indefinitely. + yield* Effect.forEach( + [...cleanups], + (fiber) => + Fiber.await(fiber).pipe( + Effect.timeout(STOP_TIMEOUT_MS), + Effect.ignore, + ), + { concurrency: "unbounded" }, + ).pipe(Effect.ignore); + yield* Effect.sync(() => notify()); }); - const disposeAll = Effect.gen(function* () { - disposed = true; - const all = [...entries.values()]; - entries.clear(); - yield* Effect.forEach( - all, - (entry) => - closeEntryScope(entry).pipe( - Effect.timeout(ENTRY_CLOSE_TIMEOUT_MS), - Effect.ignore, - ), - { concurrency: "unbounded" }, - ); - // Pruning cleanups are detached; bound them like everything else so a - // stuck backend finalizer cannot block runtime shutdown indefinitely. - yield* Effect.forEach( - [...cleanups], - (fiber) => - Fiber.await(fiber).pipe(Effect.timeout(STOP_TIMEOUT_MS), Effect.ignore), - { concurrency: "unbounded" }, - ).pipe(Effect.ignore); - yield* Effect.sync(() => notify()); + const view: SubagentReadModel = { + list: () => [...entries.values()].map((entry) => entry.snapshot), + get: (id) => entries.get(id)?.snapshot, + size: () => entries.size, + subscribe: (listener) => { + listeners.add(listener); + return () => listeners.delete(listener); + }, + subscribeTo: (id, listener) => { + let set = idListeners.get(id); + if (!set) { + set = new Set(); + idListeners.set(id, set); + } + set.add(listener); + return () => { + set.delete(listener); + if (set.size === 0) idListeners.delete(id); + }; + }, + requestSend: (id, text) => { + runDetached(send(id, text).pipe(Effect.ignore)); + }, + requestAbort: (id) => { + const entry = entries.get(id); + if (!entry) return; + // UI-initiated aborts are not "consumed": the failed result still + // flows back to the parent as a follow-up message, matching v1. + runDetached(abortEntry(entry).pipe(Effect.ignore)); + }, + setOnSettled: (hook) => { + onSettled = hook; + }, + }; + + // Safety net: disposing the ManagedRuntime tears everything down even if + // the extension forgot to call disposeAll explicitly. + yield* Effect.addFinalizer(() => disposeAll); + + return SubagentManager.of({ + spawn, + waitFor, + cancel, + send, + get: (id) => Effect.sync(() => entries.get(id)?.snapshot), + list: Effect.sync(() => [...entries.values()].map((e) => e.snapshot)), + disposeAll, + view, + }); }); - const view: SubagentReadModel = { - list: () => [...entries.values()].map((entry) => entry.snapshot), - get: (id) => entries.get(id)?.snapshot, - size: () => entries.size, - subscribe: (listener) => { - listeners.add(listener); - return () => listeners.delete(listener); - }, - subscribeTo: (id, listener) => { - let set = idListeners.get(id); - if (!set) { - set = new Set(); - idListeners.set(id, set); - } - set.add(listener); - return () => { - set.delete(listener); - if (set.size === 0) idListeners.delete(id); - }; - }, - requestSend: (id, text) => { - runDetached(send(id, text).pipe(Effect.ignore)); - }, - requestAbort: (id) => { - const entry = entries.get(id); - if (!entry) return; - // UI-initiated aborts are not "consumed": the failed result still - // flows back to the parent as a follow-up message, matching v1. - runDetached(abortEntry(entry).pipe(Effect.ignore)); - }, - setOnSettled: (hook) => { - onSettled = hook; - }, - }; - - // Safety net: disposing the ManagedRuntime tears everything down even if - // the extension forgot to call disposeAll explicitly. - yield* Effect.addFinalizer(() => disposeAll); - - return SubagentManager.of({ - spawn, - waitFor, - cancel, - send, - get: (id) => Effect.sync(() => entries.get(id)?.snapshot), - list: Effect.sync(() => [...entries.values()].map((e) => e.snapshot)), - disposeAll, - view, - }); -}); +export interface SubagentManagerConfig { + /** Test-only override for the first-response watchdog timeout. */ + firstResponseTimeoutMs?: number; +} -export const SubagentManagerLive: Layer.Layer< - SubagentManager, - never, - BackendRegistry -> = Layer.effect(SubagentManager, makeManager); +export const makeSubagentManagerLayer = (config: SubagentManagerConfig = {}) => + Layer.effect(SubagentManager, makeManager(config)); + +export const SubagentManagerLive = makeSubagentManagerLayer(); From 18231cbb265ac8d83e16153ea1b0d4536294cb7d Mon Sep 17 00:00:00 2001 From: tt-a1i Date: Sat, 22 Aug 2026 20:05:36 +0800 Subject: [PATCH 4/7] refactor(workflows): unify persisted run reads through one normalized reader listRuns (/workflows non-TUI listing) and runDetailText parsed workflow.json raw while the dashboard and resolveRunDetails went through normalizePersistedWorkflowDetails + recoverStaleWorkflowDetails, so the same old-format run could show different statuses per surface. Add listPersistedRunIds() and readPersistedWorkflowDetails() beside the existing normalizer in dashboard.ts as the single persisted-run read entry (parse + normalize + optional result/transcript hydration), and make listRuns, runDetailText, resolveRunDetails, and loadRunEntries thin consumers of it. Dashboard output is unchanged; listRuns now reports old-format runs with the same normalized status as every other surface. --- extensions/workflows/dashboard.ts | 143 ++++++++++++++++++------------ 1 file changed, 87 insertions(+), 56 deletions(-) diff --git a/extensions/workflows/dashboard.ts b/extensions/workflows/dashboard.ts index 9893f40b..0e2032ce 100644 --- a/extensions/workflows/dashboard.ts +++ b/extensions/workflows/dashboard.ts @@ -86,6 +86,80 @@ function runsDir(): string { return path.join(getAgentDir(), "workflows"); } +/** Every persisted run id on disk; empty when no runs directory exists. */ +export function listPersistedRunIds(): string[] { + try { + return fs.readdirSync(runsDir()).filter(isWorkflowRunId); + } catch { + // No runs yet. + } + return []; +} + +/** Hydrate the result/transcript side artifacts referenced by workflow.json. */ +function hydrateRunArtifacts(runId: string, details: WorkflowDetails) { + const runDir = path.join(runsDir(), runId); + if (details.resultArtifact) { + try { + details.result = JSON.parse( + fs.readFileSync( + path.join(runDir, path.basename(details.resultArtifact)), + "utf8", + ), + ); + } catch { + // Keep the compact compatibility marker from workflow.json. + } + } + if (details.transcriptArtifact) { + try { + const transcripts = JSON.parse( + fs.readFileSync( + path.join(runDir, path.basename(details.transcriptArtifact)), + "utf8", + ), + ) as Record; + for (const agent of details.agents) { + agent.transcript = normalizeTranscript( + transcripts[String(agent.index)], + ); + } + } catch { + // Older or partially written artifacts simply lack transcripts. + } + } +} + +export interface ReadPersistedRunOptions { + /** Hydrate result and transcript side artifacts referenced by workflow.json. */ + hydrateArtifacts?: boolean; +} + +/** + * The single read entry for a persisted workflow.json: parse, normalize + * (including runs written by older tooling), and optionally hydrate the side + * artifacts. Unreadable or invalid runs read as undefined. Stale "running" + * reconciliation stays with callers so selection filters can run on the + * recorded timestamps first (`recoverStaleWorkflowDetails`). + */ +export function readPersistedWorkflowDetails( + runId: string, + options: ReadPersistedRunOptions = {}, +): WorkflowDetails | undefined { + let details: WorkflowDetails | undefined; + try { + const raw: unknown = JSON.parse( + fs.readFileSync(path.join(runsDir(), runId, "workflow.json"), "utf8"), + ); + details = normalizePersistedWorkflowDetails(runId, raw); + } catch { + return undefined; + } + if (!details) return undefined; + if (options.hydrateArtifacts) hydrateRunArtifacts(runId, details); + return details; +} + function isWorktreeCleanup( value: unknown, ): value is NonNullable { @@ -395,69 +469,26 @@ export function loadRunEntries( /** Hide runs untouched by the current request; live runs always show. */ startedSince = 0, ): RunEntry[] { - let names: string[] = []; - try { - names = fs.readdirSync(runsDir()).filter(isWorkflowRunId); - } catch { - // No runs yet. - } const entries: RunEntry[] = []; - for (const runId of names) { + for (const runId of listPersistedRunIds()) { const live = active.get(runId); if (live) { entries.push({ runId, details: live, live: true }); continue; } - try { - const raw = JSON.parse( - fs.readFileSync(path.join(runsDir(), runId, "workflow.json"), "utf8"), - ); - const details = normalizePersistedWorkflowDetails(runId, raw); - const touchedAt = Math.max( - details?.startedAt ?? 0, - details?.finishedAt ?? 0, - ); - if ( - details && - touchedAt >= startedSince && - (details.sessionId === sessionId || referencedRunIds.has(runId)) - ) { - const runDir = path.join(runsDir(), runId); - if (details.resultArtifact) { - try { - details.result = JSON.parse( - fs.readFileSync( - path.join(runDir, path.basename(details.resultArtifact)), - "utf8", - ), - ); - } catch { - // Keep the compact compatibility marker from workflow.json. - } - } - if (details.transcriptArtifact) { - try { - const transcripts = JSON.parse( - fs.readFileSync( - path.join(runDir, path.basename(details.transcriptArtifact)), - "utf8", - ), - ) as Record; - for (const agent of details.agents) { - agent.transcript = normalizeTranscript( - transcripts[String(agent.index)], - ); - } - } catch { - // Older or partially written artifacts simply lack transcripts. - } - } - recoverStaleWorkflowDetails(details); - entries.push({ runId, details, live: false }); - } - } catch { - // Skip unreadable runs. + const details = readPersistedWorkflowDetails(runId, { + hydrateArtifacts: true, + }); + if (!details) continue; + const touchedAt = Math.max(details.startedAt, details.finishedAt ?? 0); + if ( + touchedAt < startedSince || + (details.sessionId !== sessionId && !referencedRunIds.has(runId)) + ) { + continue; } + recoverStaleWorkflowDetails(details); + entries.push({ runId, details, live: false }); } return entries.sort((a, b) => b.details.startedAt - a.details.startedAt); } From 8ac44fc0f3e38a6dcfb4dd10387702439520a071 Mon Sep 17 00:00:00 2001 From: tt-a1i Date: Sat, 22 Aug 2026 20:05:47 +0800 Subject: [PATCH 5/7] test(workflows): agent-session injection seam and execute-level e2e tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a module-level test-only seam in index.ts — __setWorkflowTestAgentSessionFactory() — that runChild forwards to runAgent's existing sessionFactory option, so the replay/acceptance/ worktree glue in the production path is reachable from tests without spawning real provider sessions. Add execute.e2e.test.ts on the target-resolution.test.ts harness (temp agent dir, fake pi, real sandbox child process): a foreground agent-less run with artifact assertions and activeRuns cleanup, idle and busy background follow-up delivery (followUp+triggerTurn vs nextTurn), a failing script recorded as failed, and a full reviewer-agent run through the injected fake session whose resume replays the journal without touching the session factory. --- extensions/workflows/execute.e2e.test.ts | 417 +++++++++++++++++++++++ extensions/workflows/index.ts | 127 +++---- 2 files changed, 472 insertions(+), 72 deletions(-) create mode 100644 extensions/workflows/execute.e2e.test.ts diff --git a/extensions/workflows/execute.e2e.test.ts b/extensions/workflows/execute.e2e.test.ts new file mode 100644 index 00000000..0e192e93 --- /dev/null +++ b/extensions/workflows/execute.e2e.test.ts @@ -0,0 +1,417 @@ +/** + * Execute-level workflow tests: real tool registration through index.ts, the + * sandbox child process, artifact persistence, and the background follow-up + * message — with agent sessions faked through the test-only + * `__setWorkflowTestAgentSessionFactory` injection seam. + */ + +import assert from "node:assert/strict"; +import { execFileSync } from "node:child_process"; +import { + existsSync, + mkdtempSync, + readFileSync, + rmSync, + writeFileSync, +} from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; +import type { + AgentSession, + AgentSessionEventListener, + ExtensionAPI, + ExtensionContext, +} from "@earendil-works/pi-coding-agent"; +import type { WorkflowAgentSessionFactory } from "./runner.ts"; + +const agentDir = mkdtempSync(join(tmpdir(), "my-pi-setup-wf-e2e-")); +process.env.PI_CODING_AGENT_DIR = agentDir; + +// A clean committed git checkout: replay identity fingerprints the repository +// (HEAD, diff, untracked files), so resume tests need a real repo without +// untracked or ignored content. +const repoDir = mkdtempSync(join(tmpdir(), "my-pi-setup-wf-e2e-repo-")); +function git(args: readonly string[]) { + execFileSync("git", args, { + cwd: repoDir, + stdio: ["ignore", "ignore", "ignore"], + }); +} +git(["init", "-q"]); +git(["config", "user.email", "workflow-e2e@example.com"]); +git(["config", "user.name", "Workflow E2E"]); +writeFileSync(join(repoDir, "fixture.txt"), "fixture\n"); +git(["add", "."]); +git(["commit", "-q", "-m", "fixture"]); + +const { default: workflows, __setWorkflowTestAgentSessionFactory } = + await import("./index.ts"); + +type CapturedTool = { + name: string; + execute: ( + id: string, + params: Record, + signal?: AbortSignal, + onUpdate?: unknown, + ctx?: ExtensionContext, + ) => unknown; +}; + +type SentMessage = { + message: Record & { + customType?: string; + details?: { runId?: unknown }; + }; + options: unknown; +}; + +const tools = new Map(); +let activeTools: string[] = []; +const handlers = new Map< + string, + Array<(event: unknown, ctx: ExtensionContext) => unknown> +>(); +const sentMessages: SentMessage[] = []; +let modelIdle = true; + +const pi = { + registerTool(tool: CapturedTool) { + tools.set(tool.name, tool); + activeTools = [ + ...activeTools.filter((name) => name !== tool.name), + tool.name, + ]; + }, + registerCommand() {}, + registerMessageRenderer() {}, + on(event: string, handler: unknown) { + handlers.set(event, [ + ...(handlers.get(event) ?? []), + handler as (event: unknown, ctx: ExtensionContext) => unknown, + ]); + }, + getThinkingLevel: () => "off", + getActiveTools: () => [...activeTools], + setActiveTools(names: string[]) { + activeTools = [...names]; + }, + sendMessage(message: SentMessage["message"], options: unknown) { + sentMessages.push({ message, options }); + }, +} as unknown as ExtensionAPI; + +const ctx = { + cwd: repoDir, + mode: "tui", + hasUI: true, + isIdle: () => modelIdle, + isProjectTrusted: () => false, + sessionManager: { + getSessionId: () => "wf-e2e-session", + getEntries: () => [], + }, + model: undefined, + modelRegistry: { find: () => undefined }, + ui: { + theme: { fg: (_color: string, text: string) => text }, + setStatus() {}, + setWidget() {}, + }, +} as unknown as ExtensionContext; + +workflows(pi); +for (const handler of handlers.get("session_start") ?? []) { + await handler({}, { + ...ctx, + hasUI: false, + mode: "print", + } as unknown as ExtensionContext); +} + +const workflow = tools.get("workflow")!; +const status = tools.get("workflow_status")!; +assert.ok(workflow && status); + +function runDirFor(runId: unknown) { + assert.equal(typeof runId, "string"); + return join(agentDir, "workflows", runId as string); +} + +function readWorkflowJson(runId: unknown) { + return JSON.parse( + readFileSync(join(runDirFor(runId), "workflow.json"), "utf8"), + ) as Record; +} + +async function waitFor( + predicate: () => boolean, + label: string, + timeoutMs = 20_000, +) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (predicate()) return; + await new Promise((resolve) => setTimeout(resolve, 25)); + } + assert.ok(predicate(), `timed out waiting for ${label}`); +} + +/** A minimal AgentSession stand-in for one successful child agent call. */ +function fakeAgentSession(output: string) { + const listeners = new Set(); + // The reviewer agent type requests the read-only tool surface; the child + // preflight in bindChildSessionExtensions requires all of them active. + const toolNames = ["read", "grep", "find", "ls", "fd", "rg"]; + const messages = [ + { role: "user", content: "fixture prompt", timestamp: 1 }, + { + role: "assistant", + content: [{ type: "text", text: output }], + provider: "fixture", + model: "fixture", + usage: { + input: 3, + output: 5, + cacheRead: 0, + cacheWrite: 0, + cost: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + total: 0, + }, + }, + stopReason: "stop", + timestamp: 2, + }, + ]; + return { + messages, + model: undefined, + extensionRunner: { hasHandlers: () => false, emit: async () => {} }, + async bindExtensions() {}, + subscribe(listener: AgentSessionEventListener) { + listeners.add(listener); + return () => listeners.delete(listener); + }, + async prompt() {}, + async abort() {}, + dispose() {}, + getContextUsage: () => undefined, + getAllTools: () => toolNames.map((name) => ({ name })), + getToolDefinition: () => undefined, + getActiveToolNames: () => [...toolNames], + setActiveToolsByName() {}, + } as unknown as AgentSession; +} + +test("foreground run without agents returns the result and persists artifacts", async () => { + const result = (await workflow.execute( + "e2e-foreground", + { + script: + 'export const meta = { name: "plain-run", description: "no agents" };\nlog("hi");\nreturn { x: 1 };', + }, + undefined, + undefined, + ctx, + )) as { + content: Array<{ type: string; text: string }>; + details: { runId?: unknown; status?: unknown }; + }; + + assert.equal(result.details.status, "completed"); + const text = result.content[0]!.text; + assert.match(text, /"plain-run" completed/); + assert.match(text, /"x":\s*1/); + + const runId = result.details.runId; + const runDir = runDirFor(runId); + assert.ok(existsSync(join(runDir, "script.js"))); + const persisted = readWorkflowJson(runId); + assert.equal(persisted.status, "completed"); + assert.equal(persisted.resultArtifact, "result.json"); + assert.deepEqual( + JSON.parse(readFileSync(join(runDir, "result.json"), "utf8")), + { x: 1 }, + ); + + // The run left activeRuns: the in-memory status listing holds it exactly + // once (active and settled would both list it), settled and completed. + const listing = (await status.execute("e2e-foreground-status", {})) as { + details: { runs: Array<{ runId: unknown; status: unknown }> }; + }; + const mine = listing.details.runs.filter((run) => run.runId === runId); + assert.equal(mine.length, 1); + assert.equal(mine[0]!.status, "completed"); +}); + +test("background runs deliver a follow-up that triggers a turn only when idle", async () => { + sentMessages.length = 0; + + modelIdle = true; + const idleRun = (await workflow.execute( + "e2e-bg-idle", + { + script: 'export const meta = { name: "bg-idle" };\nlog("bg");\nreturn 7;', + background: true, + }, + undefined, + undefined, + ctx, + )) as { details: { runId?: unknown } }; + assert.equal(typeof idleRun.details.runId, "string"); + + await waitFor( + () => + sentMessages.some( + (sent) => sent.message.details?.runId === idleRun.details.runId, + ), + "idle background follow-up", + ); + const idleFollowUp = sentMessages.find( + (sent) => sent.message.details?.runId === idleRun.details.runId, + )!; + assert.equal(idleFollowUp.message.customType, "workflow-result"); + assert.equal(idleFollowUp.message.display, true); + assert.match(String(idleFollowUp.message.content), /bg-idle/); + assert.deepEqual(idleFollowUp.options, { + deliverAs: "followUp", + triggerTurn: true, + }); + + modelIdle = false; + const busyRun = (await workflow.execute( + "e2e-bg-busy", + { + script: 'export const meta = { name: "bg-busy" };\nreturn 8;', + background: true, + }, + undefined, + undefined, + ctx, + )) as { details: { runId?: unknown } }; + assert.equal(typeof busyRun.details.runId, "string"); + + await waitFor( + () => + sentMessages.some( + (sent) => sent.message.details?.runId === busyRun.details.runId, + ), + "busy background follow-up", + ); + const busyFollowUp = sentMessages.find( + (sent) => sent.message.details?.runId === busyRun.details.runId, + )!; + // A busy model is not woken: the result rides along with the next turn. + assert.deepEqual(busyFollowUp.options, { deliverAs: "nextTurn" }); +}); + +test("a failing script reports the error and records the run as failed", async () => { + await assert.rejects( + Promise.resolve( + workflow.execute( + "e2e-failing", + { + script: + 'export const meta = { name: "boom-run" };\nthrow new Error("kaboom");', + }, + undefined, + undefined, + ctx, + ), + ), + /kaboom/, + ); + + const runs = (await status.execute("e2e-failing-status", {})) as { + details: { + runs: Array<{ runId: unknown; name: unknown; status: unknown }>; + }; + }; + const failed = runs.details.runs.find((run) => run.name === "boom-run"); + assert.ok(failed); + assert.equal(failed.status, "failed"); + const persisted = readWorkflowJson(failed.runId); + assert.equal(persisted.status, "failed"); + assert.match(String(persisted.error), /kaboom/); +}); + +test("agent calls run through the injected session factory and resume replays the journal", async () => { + let sessionCreations = 0; + const factory: WorkflowAgentSessionFactory = async () => { + sessionCreations += 1; + return { session: fakeAgentSession("injected agent output") }; + }; + __setWorkflowTestAgentSessionFactory(factory); + + const agentScript = + 'export const meta = { name: "agent-run" };\n' + + 'const r = await agent("say something", { agent_type: "reviewer", label: "speaker" });\n' + + 'log("agent said: " + r.output);\n' + + "return { ok: r.ok, output: r.output };"; + + try { + const first = (await workflow.execute( + "e2e-agent-first", + { script: agentScript }, + undefined, + undefined, + ctx, + )) as { + content: Array<{ type: string; text: string }>; + details: { runId?: unknown }; + }; + const firstText = first.content[0]!.text; + assert.match(firstText, /"agent-run" completed/); + assert.match(firstText, /injected agent output/); + assert.equal(sessionCreations, 1); + + const firstRunId = first.details.runId; + const firstDir = runDirFor(firstRunId); + // A replay-safe read-only agent call is journaled on success. + const journal = JSON.parse( + readFileSync(join(firstDir, "journal.json"), "utf8"), + ) as { entries: unknown[] }; + assert.equal(journal.entries.length, 1); + + // Resume with identical script and call content: the journal hit means no + // new child session is created. + const resumed = (await workflow.execute( + "e2e-agent-resume", + { script: agentScript, resume_from_run_id: String(firstRunId) }, + undefined, + undefined, + ctx, + )) as { + content: Array<{ type: string; text: string }>; + details: { runId?: unknown }; + }; + const resumedText = resumed.content[0]!.text; + assert.match(resumedText, /injected agent output/); + assert.match(resumedText, /Resumed from .*replayed 1\/1 agent call/); + assert.equal(sessionCreations, 1); + + const persisted = readWorkflowJson(resumed.details.runId); + const agents = persisted.agents as Array<{ + state: unknown; + replayed?: unknown; + }>; + assert.equal(agents.length, 1); + assert.equal(agents[0]!.state, "done"); + assert.equal(agents[0]!.replayed, true); + } finally { + __setWorkflowTestAgentSessionFactory(undefined); + } +}); + +test.after(() => { + for (const handler of handlers.get("session_shutdown") ?? []) { + void handler({}, ctx); + } + rmSync(agentDir, { recursive: true, force: true }); + rmSync(repoDir, { recursive: true, force: true }); +}); diff --git a/extensions/workflows/index.ts b/extensions/workflows/index.ts index f91e8012..fb209149 100644 --- a/extensions/workflows/index.ts +++ b/extensions/workflows/index.ts @@ -89,7 +89,8 @@ import { } from "./journal.ts"; import { RunController } from "./controller.ts"; import { - normalizePersistedWorkflowDetails, + listPersistedRunIds, + readPersistedWorkflowDetails, recoverStaleWorkflowDetails, sessionWorkflowRunIds, showWorkflowDashboard, @@ -148,6 +149,7 @@ import { createWorkflowResources, runAgent, type ThinkingLevel, + type WorkflowAgentSessionFactory, type WorkflowModel, } from "./runner.ts"; import { @@ -172,6 +174,20 @@ import { safeStringify, writeFileAtomic } from "./serialization.ts"; const PREVIEW_LENGTH = 200; const EMIT_INTERVAL_MS = 120; +/** + * Test-only injection seam for execute-level tests: production never sets it. + * The underscore-prefixed setter name makes any accidental production use + * self-evidently wrong. + */ +let testAgentSessionFactory: WorkflowAgentSessionFactory | undefined; + +/** Test-only: override how workflow children create their agent sessions. */ +export function __setWorkflowTestAgentSessionFactory( + factory: WorkflowAgentSessionFactory | undefined, +) { + testAgentSessionFactory = factory; +} + const THINKING_LEVELS = [ "off", "minimal", @@ -348,15 +364,8 @@ function listRuns( referencedRunIds: ReadonlySet, startedSince = 0, ): RunSummary[] { - const base = path.join(getAgentDir(), "workflows"); - let names: string[] = []; - try { - names = fs.readdirSync(base).filter(isWorkflowRunId); - } catch { - // No runs yet. - } const summaries: RunSummary[] = []; - for (const runId of names) { + for (const runId of listPersistedRunIds()) { const live = activeRuns.get(runId); if (live) { const { done, failed } = countStates(live); @@ -371,34 +380,28 @@ function listRuns( }); continue; } - try { - const parsed = JSON.parse( - fs.readFileSync(path.join(base, runId, "workflow.json"), "utf8"), - ) as Partial; - const startedAt = parsed.startedAt ?? 0; - const touchedAt = Math.max(startedAt, parsed.finishedAt ?? 0); - if ( - touchedAt < startedSince || - (parsed.sessionId !== sessionId && !referencedRunIds.has(runId)) - ) { - continue; - } - const agents = parsed.agents ?? []; - summaries.push({ - runId, - name: parsed.name, - status: - parsed.status === "running" - ? "aborted" - : (parsed.status ?? "unknown"), - done: agents.filter((agent) => agent.state !== "running").length, - total: agents.length, - startedAt: parsed.startedAt ?? 0, - active: false, - }); - } catch { - // Ignore unreadable artifacts because their session cannot be verified. + // Same reader as the dashboard and workflow_status: old-format runs are + // normalized here too, so every surface reports one status per run. + const details = readPersistedWorkflowDetails(runId); + if (!details) continue; + const touchedAt = Math.max(details.startedAt, details.finishedAt ?? 0); + if ( + touchedAt < startedSince || + (details.sessionId !== sessionId && !referencedRunIds.has(runId)) + ) { + continue; } + recoverStaleWorkflowDetails(details); + const { done, failed } = countStates(details); + summaries.push({ + runId, + name: details.name, + status: details.status, + done: done + failed, + total: details.agents.length, + startedAt: details.startedAt, + active: false, + }); } return summaries.sort((a, b) => b.startedAt - a.startedAt); } @@ -410,14 +413,9 @@ function runDetailText( const runDir = path.join(getAgentDir(), "workflows", run.runId); const live = activeRuns.get(run.runId); if (live) return buildWorkflowResultMessage(live, runDir); - try { - const parsed = JSON.parse( - fs.readFileSync(path.join(runDir, "workflow.json"), "utf8"), - ) as WorkflowDetails; - return buildWorkflowResultMessage(parsed, runDir); - } catch { - return `Run ${run.runId} — ${run.status}`; - } + const details = readPersistedWorkflowDetails(run.runId); + if (details) return buildWorkflowResultMessage(details, runDir); + return `Run ${run.runId} — ${run.status}`; } export default function workflows(pi: ExtensionAPI) { @@ -1365,6 +1363,9 @@ export default function workflows(pi: ExtensionAPI) { ...(sessionManager ? { sessionManager } : {}), modelRegistry: ctx.modelRegistry, ...(agentType?.tools ? { tools: agentType.tools } : {}), + ...(testAgentSessionFactory + ? { sessionFactory: testAgentSessionFactory } + : {}), ...(replayIdentity ? { replayFilesystemBoundary: { @@ -1921,17 +1922,10 @@ export default function workflows(pi: ExtensionAPI) { /** Resolve one run from live, settled, or persisted state. */ const resolveRunDetails = (target: string) => { - const base = path.join(getAgentDir(), "workflows"); - let persistedIds: string[] = []; - try { - persistedIds = fs.readdirSync(base).filter(isWorkflowRunId); - } catch { - // In-memory runs remain inspectable without the artifact directory. - } const resolution = resolveWorkflowRunTarget(target, [ ...activeRuns.keys(), ...settledRuns.keys(), - ...persistedIds, + ...listPersistedRunIds(), ]); if (!resolution.ok) return resolution; @@ -1940,31 +1934,20 @@ export default function workflows(pi: ExtensionAPI) { const settled = settledRuns.get(resolution.runId); if (settled) return { ok: true, details: settled } as const; - try { - const parsed: unknown = JSON.parse( - fs.readFileSync( - path.join(base, resolution.runId, "workflow.json"), - "utf8", - ), - ); - const details = normalizePersistedWorkflowDetails( - resolution.runId, - parsed, - ); - if (!details) throw new Error("invalid workflow details"); - // A run absent from activeRuns cannot still be running this session; a - // persisted "running" is a run that was hard-killed or missed the - // shutdown settle deadline. - return { - ok: true, - details: recoverStaleWorkflowDetails(details), - } as const; - } catch { + const details = readPersistedWorkflowDetails(resolution.runId); + if (!details) { return { ok: false, error: `Workflow run ${resolution.runId} could not be read.`, } as const; } + // A run absent from activeRuns cannot still be running this session; a + // persisted "running" is a run that was hard-killed or missed the + // shutdown settle deadline. + return { + ok: true, + details: recoverStaleWorkflowDetails(details), + } as const; }; pi.registerTool({ From b1e36d6baadcd42dbcbea7b3e23c41e2728d6162 Mon Sep 17 00:00:00 2001 From: tt-a1i Date: Sat, 22 Aug 2026 21:33:35 +0800 Subject: [PATCH 6/7] fix(workflows): recover stale status and hydrate artifacts in detail reads Follow-ups from review of the unified persisted-run reader: - runDetailText now applies recoverStaleWorkflowDetails, so /workflows no longer disagrees with the list it follows (aborted vs running) - runDetailText and workflow_status's resolveRunDetails hydrate result/ transcript side artifacts, so cross-session status no longer surfaces the internal '[stored in result.json]' marker to the model --- extensions/workflows/index.ts | 102 +++++++++++++++++++--------------- 1 file changed, 56 insertions(+), 46 deletions(-) diff --git a/extensions/workflows/index.ts b/extensions/workflows/index.ts index fb209149..1688c7c5 100644 --- a/extensions/workflows/index.ts +++ b/extensions/workflows/index.ts @@ -33,15 +33,15 @@ import { randomBytes } from "node:crypto"; import * as fs from "node:fs"; import * as path from "node:path"; import { + type ExtensionAPI, + type ExtensionContext, getAgentDir, getMarkdownTheme, keyHint, - type ExtensionAPI, type SessionManager, - type ExtensionContext, } from "@earendil-works/pi-coding-agent"; import { Container, Markdown, Spacer, Text } from "@earendil-works/pi-tui"; -import { Type, type Static } from "typebox"; +import { type Static, Type } from "typebox"; import { formatActivityStatus } from "../shared/activity-status.ts"; import { waitBounded } from "../shared/child-session.ts"; import { @@ -53,6 +53,12 @@ import { OPENPI_TOOL_SURFACE, patchOwnedTools, } from "../shared/tool-surface.ts"; +import { + createWorktree, + reclaimWorktree, + type Worktree, + type WorktreeCleanup, +} from "../shared/worktree.ts"; import { loadAgentTypes, resolveAgentModel, @@ -60,16 +66,25 @@ import { selectSubagentModel, } from "../subagents/src/agent-types.ts"; import { - createWorktree, - reclaimWorktree, - type Worktree, - type WorktreeCleanup, -} from "../shared/worktree.ts"; + acceptanceInstruction, + acceptanceSchema, + applyAcceptance, + evaluateAcceptance, + parseAcceptanceContract, +} from "./acceptance.ts"; import { createWorkflowPersistence, loadJournal, persistWorkflowJson, } from "./artifacts.ts"; +import { RunController } from "./controller.ts"; +import { + listPersistedRunIds, + readPersistedWorkflowDetails, + recoverStaleWorkflowDetails, + sessionWorkflowRunIds, + showWorkflowDashboard, +} from "./dashboard.ts"; import { createWorkflowHandoffRegistry } from "./handoff.ts"; import { classifyInterruptedInvocation, @@ -77,53 +92,51 @@ import { requestInvocation, transitionInvocation, } from "./invocation-ledger.ts"; -import { - normalizeWorkflowOperatorKey, - WorkflowOperatorRegistry, -} from "./operator.ts"; import { agentCallKey, createReplayCache, type JournalEntry, type ReplayCache, } from "./journal.ts"; -import { RunController } from "./controller.ts"; -import { - listPersistedRunIds, - readPersistedWorkflowDetails, - recoverStaleWorkflowDetails, - sessionWorkflowRunIds, - showWorkflowDashboard, -} from "./dashboard.ts"; import { extractMeta, prepareWorkflowScript, type WorkflowMeta, } from "./meta.ts"; import { + type AgentRecord, agentContext, aggregateUsage, appendLog, countStates, + createUsageReader, emptyUsage, formatElapsed, formatUsage, isWorkflowRunId, phaseGroups, + refreshWorkflowGraph, resolveWorkflowRunTarget, resultJson, + SQUARE, sanitizeLine, sanitizeWorkflowDisplayLine, sanitizeWorkflowDisplayText, stateSquare, statusColor, statusWord, - createUsageReader, - refreshWorkflowGraph, - SQUARE, - type AgentRecord, type WorkflowDetails, } from "./model.ts"; +import { + WorkflowNavigationEditor, + type WorkflowStripEntry, + WorkflowStripState, + WorkflowStripWidget, +} from "./navigation.ts"; +import { + normalizeWorkflowOperatorKey, + WorkflowOperatorRegistry, +} from "./operator.ts"; import { buildBackgroundWorkflowFollowUp, buildBackgroundWorkflowLaunchResult, @@ -140,11 +153,10 @@ import { WORKFLOW_TOOL_DESCRIPTION, } from "./prompt.ts"; import { - WorkflowNavigationEditor, - WorkflowStripState, - WorkflowStripWidget, - type WorkflowStripEntry, -} from "./navigation.ts"; + beginProcessReplayWorkspaceLease, + createReplayIdentity, + isReplaySafeAgentCall, +} from "./replay-safety.ts"; import { createWorkflowResources, runAgent, @@ -152,24 +164,12 @@ import { type WorkflowAgentSessionFactory, type WorkflowModel, } from "./runner.ts"; -import { - beginProcessReplayWorkspaceLease, - createReplayIdentity, - isReplaySafeAgentCall, -} from "./replay-safety.ts"; import { runWorkflowSandbox } from "./sandbox.ts"; -import { - acceptanceInstruction, - acceptanceSchema, - applyAcceptance, - evaluateAcceptance, - parseAcceptanceContract, -} from "./acceptance.ts"; +import { safeStringify, writeFileAtomic } from "./serialization.ts"; import { finalizeWorktreeHandoff, prepareWorktreeHandoff, } from "./worktree-handoff.ts"; -import { safeStringify, writeFileAtomic } from "./serialization.ts"; const PREVIEW_LENGTH = 200; const EMIT_INTERVAL_MS = 120; @@ -382,7 +382,9 @@ function listRuns( } // Same reader as the dashboard and workflow_status: old-format runs are // normalized here too, so every surface reports one status per run. - const details = readPersistedWorkflowDetails(runId); + const details = readPersistedWorkflowDetails(runId, { + hydrateArtifacts: false, + }); if (!details) continue; const touchedAt = Math.max(details.startedAt, details.finishedAt ?? 0); if ( @@ -413,8 +415,14 @@ function runDetailText( const runDir = path.join(getAgentDir(), "workflows", run.runId); const live = activeRuns.get(run.runId); if (live) return buildWorkflowResultMessage(live, runDir); - const details = readPersistedWorkflowDetails(run.runId); - if (details) return buildWorkflowResultMessage(details, runDir); + const details = readPersistedWorkflowDetails(run.runId, { + hydrateArtifacts: true, + }); + if (details) + return buildWorkflowResultMessage( + recoverStaleWorkflowDetails(details), + runDir, + ); return `Run ${run.runId} — ${run.status}`; } @@ -1934,7 +1942,9 @@ export default function workflows(pi: ExtensionAPI) { const settled = settledRuns.get(resolution.runId); if (settled) return { ok: true, details: settled } as const; - const details = readPersistedWorkflowDetails(resolution.runId); + const details = readPersistedWorkflowDetails(resolution.runId, { + hydrateArtifacts: true, + }); if (!details) { return { ok: false, From fa00a1961b0618c5ce008168d1c23e07d82f7fbe Mon Sep 17 00:00:00 2001 From: tt-a1i Date: Sat, 22 Aug 2026 21:42:21 +0800 Subject: [PATCH 7/7] fix(capabilities,subagents): make advertised intent phrases actually match MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-ups on the explicit-intent router and watchdog: - CAPABILITY_INTENT now accepts the word forms the README advertises: background adds 启动 (在后台启动 dev server), delegate/workflow/search accept 用 as a verb prefix, delegate accepts 委派给, workflow accepts bare '用 Workflow' - the README TIP lists only phrases that are tested to match - new test pins the README quick-start example (loads background + delegate) and every advertised trigger phrase, so docs and regex cannot drift apart again - the test harness now replicates production hideLifecycleTools wiring (deferred tools disabled on session_start) - arm the first-response watchdog in send()'s restart window too: a backend that accepts the send but never emits RunStarted would hold the slot forever, the same hang class the watchdog exists to kill; regression test included (762 node:test + 30 vitest green) --- README.md | 5 +-- extensions/capabilities/index.test.ts | 49 +++++++++++++++++++++++++++ extensions/capabilities/index.ts | 8 ++--- extensions/subagents/manager.test.ts | 30 ++++++++++++++++ extensions/subagents/src/manager.ts | 4 +++ 5 files changed, 90 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 8e341dc2..5bb169e9 100644 --- a/README.md +++ b/README.md @@ -76,8 +76,9 @@ pi install npm:@tt-a1i/openpi OpenPI 会把长期进程放到后台，把独立任务交给隔离 Context 的 Pi Subagent，把多阶段依赖组织成 Workflow。状态会持续显示；完整运行可从 `/ps`、`/subagents` 和 `/workflows` 检查或终止。 > [!TIP] -> Capability discovery 默认 `explicit`：表述里需要出现明确的能力词才会加载对应组—— -> 「在后台 / 后台运行」→ 后台终端；「子代理 / 委派」→ Subagent；「Workflow / 多阶段编排」→ Workflow；「搜索」→ `fd`/`rg`。像上面示例那样把意图说清楚即可，不需要记住任何工具名。 +> Capability discovery 默认 `explicit`：明确说出能力意图才会加载对应组。 +> 例如「在后台运行 dev server」→ 后台终端；「用/使用子代理检查」→ Subagent；「用工作流编排」→ Workflow；「用 fd/rg 搜索」→ 搜索工具。 +> 关键是把意图说清楚（说「用子代理」「后台运行」这类带动作的短语），不需要记住任何工具名。 > [!IMPORTANT] > 默认安装是安静的：不改主题、不绑定 Provider 或模型、不开启下一步预测，也不执行 post-edit 命令。Capability discovery 默认 `explicit`；只有用户通过 `/openpi-setup` 选择 `adaptive` 后，模型才会常驻看到一个小型发现网关并可自主加载额外能力。 diff --git a/extensions/capabilities/index.test.ts b/extensions/capabilities/index.test.ts index d05d8384..3c8b932c 100644 --- a/extensions/capabilities/index.test.ts +++ b/extensions/capabilities/index.test.ts @@ -42,6 +42,14 @@ function harness(options: { discovery?: "explicit" | "adaptive" } = {}) { "subagent_send", "subagent_check", "subagent_list", + "workflow", + "workflow_stop", + "workflow_status", + "bg_start", + "bg_status", + "bg_list", + "bg_kill", + "bg_watch", ]; let active = [...available]; const tools = new Map(); @@ -89,6 +97,15 @@ function harness(options: { discovery?: "explicit" | "adaptive" } = {}) { patchOwnedTools(pi, "subagents", { enable: OPENPI_TOOL_SURFACE.subagents.entry, }); + // The real extensions disable their deferred lifecycle tools on + // session_start (hideLifecycleTools); replicate that so the harness + // matches production wiring and tools only appear via capability loads. + patchOwnedTools(pi, "workflows", { + disable: OPENPI_TOOL_SURFACE.workflows.deferred, + }); + patchOwnedTools(pi, "background", { + disable: OPENPI_TOOL_SURFACE.background.deferred, + }); }); return { active: () => [...active], @@ -223,6 +240,38 @@ test("common Chinese and multi-agent delegation requests are explicit intent", ( } }); +test("the README quick-start example and its advertised phrases load capabilities", () => { + // The flagship README example: both clauses must be explicit intent. + const example = harness(); + example.start(); + example.before( + "在后台启动前端 dev server；用子代理并行检查 API 主链路和测试覆盖；\n结果回来后汇总风险，主会话不要原地等待。", + ); + const active = example.active(); + assert.ok(active.includes("bg_start"), "example loads background terminals"); + assert.ok(active.includes("subagent_spawn"), "example loads subagents"); + + // Every phrase the README TIP advertises as a trigger must actually match. + const advertised: Array<[string, string]> = [ + ["在后台运行 dev server", "bg_start"], + ["用子代理检查", "subagent_spawn"], + ["使用子代理检查", "subagent_spawn"], + ["用工作流编排", "workflow"], + ["使用工作流编排", "workflow"], + ["用 fd 搜索", "fd"], + ["使用 rg 搜索", "rg"], + ]; + for (const [phrase, tool] of advertised) { + const h = harness(); + h.start(); + h.before(phrase); + assert.ok( + h.active().includes(tool), + `README-advertised phrase ${JSON.stringify(phrase)} should load ${tool}, got: ${h.active().join(",")}`, + ); + } +}); + test("the gateway returns progressive skill guidance for a loaded group", async () => { const h = harness(); h.start(); diff --git a/extensions/capabilities/index.ts b/extensions/capabilities/index.ts index 891bb4af..8af9fd26 100644 --- a/extensions/capabilities/index.ts +++ b/extensions/capabilities/index.ts @@ -37,13 +37,13 @@ type OpenPiLoadToolsInput = Static; const CAPABILITY_INTENT = { search: - /\b(?:use|run)\s+(?:fd|rg)\b|\buse\s+(?:structured\s+)?(?:(?:file|code|content)\s+)?search\b|\b(?:structured|fast)\s+(?:file|code|content)\s+search\b|(?:使用|运行).{0,8}(?:fd|rg)|结构化(?:文件|代码|内容)搜索/iu, + /\b(?:use|run)\s+(?:fd|rg)\b|\buse\s+(?:structured\s+)?(?:(?:file|code|content)\s+)?search\b|\b(?:structured|fast)\s+(?:file|code|content)\s+search\b|(?:使用|用|运行).{0,8}(?:fd|rg)|结构化(?:文件|代码|内容)搜索/iu, delegate: - /\b(?:use|spawn|run)\s+(?:an?\s+|multiple\s+|several\s+|two\s+)?(?:pi\s+)?subagents?\b|(?:^|[.!?]\s+)(?:please\s+)?(?:delegate|parallelize)\s+(?:this|the)\s+(?:task|work)\b|\b(?:can|could|would)\s+you\s+(?:please\s+)?(?:delegate|parallelize)\s+(?:this|the)\s+(?:task|work)\b|\bparallel\s+agents?\b|(?:使用|启动|调用|来|开).{0,8}子代理|(?:多个?|多路)子代理|并行.{0,8}(?:代理|agent)|委派.{0,6}任务/iu, + /\b(?:use|spawn|run)\s+(?:an?\s+|multiple\s+|several\s+|two\s+)?(?:pi\s+)?subagents?\b|(?:^|[.!?]\s+)(?:please\s+)?(?:delegate|parallelize)\s+(?:this|the)\s+(?:task|work)\b|\b(?:can|could|would)\s+you\s+(?:please\s+)?(?:delegate|parallelize)\s+(?:this|the)\s+(?:task|work)\b|\bparallel\s+agents?\b|(?:使用|用|启动|调用|来|开).{0,8}子代理|(?:多个?|多路)子代理|并行.{0,8}(?:代理|agent)|委派.{0,6}(?:任务|给|出去)/iu, workflow: - /\b(?:use|run|create|build)\s+(?:(?:an?|the)\s+)?(?:openpi\s+)?workflow\b|(?:使用|运行|创建|构建).{0,8}工作流/iu, + /\b(?:use|run|create|build)\s+(?:(?:an?|the)\s+)?(?:openpi\s+)?workflow\b|(?:使用|用|运行|创建|构建).{0,8}工作流|用.{0,4}workflow/iu, background: - /\b(?:run|start|keep)\b.{0,40}\b(?:in the background|background\s+(?:process|terminal|job))\b|后台.{0,8}(?:运行|进程|终端|任务)/iu, + /\b(?:run|start|keep)\b.{0,40}\b(?:in the background|background\s+(?:process|terminal|job))\b|后台.{0,8}(?:运行|启动|进程|终端|任务)/iu, session: /\b(?:create|set|update|track)\s+(?:an?\s+)?(?:session\s+)?(?:goal|task list|tasks)\b|(?:设置|创建|更新|跟踪|追踪).{0,8}(?:目标|任务)/iu, } as const satisfies Record; diff --git a/extensions/subagents/manager.test.ts b/extensions/subagents/manager.test.ts index 50372091..8dd45c3c 100644 --- a/extensions/subagents/manager.test.ts +++ b/extensions/subagents/manager.test.ts @@ -434,3 +434,33 @@ test("a first response clears the watchdog so slower runs are not killed", async { firstResponseTimeoutMs: 250 }, ); }); + +test("a restart whose run never starts is settled by the watchdog", async () => { + await withManager( + async (manager, runtime) => { + const snap = await runTool(runtime, manager.spawn("pi", task("First"))); + await runTool(runtime, manager.waitFor([snap.id])); + assert.equal(manager.view.get(snap.id)?.status, "done"); + + // The backend accepts the send but the new run never emits RunStarted, + // so the restarting entry holds its slot with nothing to clear it. + await runTool(runtime, manager.send(snap.id, "HANG: stalled restart")); + await runTool(runtime, manager.waitFor([snap.id])); + const after = manager.view.get(snap.id); + assert.equal(after?.status, "error"); + assert.match( + after?.errorText ?? "", + /no assistant response event.*provider request may be stalled/, + ); + + // The freed slot accepts a fresh spawn again. + const fresh = await runTool( + runtime, + manager.spawn("pi", task("ok after the stalled restart")), + ); + await runTool(runtime, manager.waitFor([fresh.id])); + assert.equal(manager.view.get(fresh.id)?.status, "done"); + }, + { firstResponseTimeoutMs: 150 }, + ); +}); diff --git a/extensions/subagents/src/manager.ts b/extensions/subagents/src/manager.ts index 995e0385..763a3f7e 100644 --- a/extensions/subagents/src/manager.ts +++ b/extensions/subagents/src/manager.ts @@ -745,6 +745,10 @@ const makeManager = (config: SubagentManagerConfig = {}) => // both pass the check in that window. Cleared by RunStarted/settle, // or here when the backend rejects the send. entry.restarting = true; + // A backend that accepts the send but never starts the run would + // hold the slot forever; guard the restart window the same way the + // spawn path guards its pre-RunStarted window. + armWatchdog(entry); return entry.session.send(text).pipe( Effect.onError(() => Effect.sync(() => {