From 2657593bb0f0ba6d1a4ee152997e635fb71914f4 Mon Sep 17 00:00:00 2001 From: tt-a1i Date: Mon, 7 Sep 2026 08:12:23 +0800 Subject: [PATCH 1/2] fix(subagents): own cancelled startup and retain current progress --- ...ILD_ACQUISITION_AND_PROGRESS_2026-09-07.md | 57 +++++ docs/research/README.md | 2 + extensions/subagents/index.ts | 16 +- extensions/subagents/src/backends/pi.ts | 129 +++++----- .../subagents/src/backends/tool-preview.ts | 29 +++ extensions/subagents/src/runtime.ts | 13 +- extensions/workflows/index.ts | 11 +- extensions/workflows/progress-projection.ts | 8 +- .../subagents/pi-backend-lifecycle.test.ts | 56 ++++ .../subagents/startup-worktree.test.ts | 239 ++++++++++++++++++ .../extensions/subagents/tool-preview.test.ts | 58 +++++ .../extensions/workflows/execute.e2e.test.ts | 70 +++++ .../workflows/progress-projection.test.ts | 46 ++++ 13 files changed, 664 insertions(+), 70 deletions(-) create mode 100644 docs/research/CHILD_ACQUISITION_AND_PROGRESS_2026-09-07.md create mode 100644 extensions/subagents/src/backends/tool-preview.ts create mode 100644 tests/extensions/subagents/startup-worktree.test.ts create mode 100644 tests/extensions/subagents/tool-preview.test.ts diff --git a/docs/research/CHILD_ACQUISITION_AND_PROGRESS_2026-09-07.md b/docs/research/CHILD_ACQUISITION_AND_PROGRESS_2026-09-07.md new file mode 100644 index 00000000..cceba566 --- /dev/null +++ b/docs/research/CHILD_ACQUISITION_AND_PROGRESS_2026-09-07.md @@ -0,0 +1,57 @@ +# Child acquisition and progress reliability audit + +- Status: validated for deterministic defect reproduction; repair acceptance is recorded in the linked PR +- Created / verified: 2026-09-07 +- Source boundary: `eaf470bab4ac2dda607d16c6ddee66dc35367527` (main after PR #426) +- Issue: [#428](https://github.com/openpi-dev/openpi/issues/428) +- Supersedes: none + +## Scope and provenance + +This audit checks child startup ownership and progress projection. The repair uses an isolated source worktree; `pi list` still identifies the user's separate `openpi-main-runtime` checkout at `c8f2c13`. Source validation does not imply that the user's already-running Pi loaded the fixes. No user role files, settings, private Sessions, or ignored evidence were changed. + +Existing viewport, retained-history, headless-shell and terminal-artifact PRs (#327, #319, #423 and #386) remain separate. The previous Cursor/tool/cwd repair was merged as [#426](https://github.com/openpi-dev/openpi/pull/426). + +## Confirmed mechanisms + +### Cancelled ordinary child acquisition loses its owner + +The production Pi backend used an interruptible `Effect.tryPromise` acquisition but ignored its abort signal. Its cleanup finalizer was registered only after resource loading, session creation and extension binding. A deterministic probe paused the session factory, interrupted the scoped spawn, then released the factory. The late child bound extensions despite cancellation and received zero abort, shutdown or dispose calls. + +The repair gives startup provisional cleanup ownership before the first asynchronous acquisition, observes cancellation at acquisition boundaries and reuses the bounded child shutdown helper. A child arriving after cancellation is disposed without beginning extension binding. Ownership transfers to the existing backend finalizer after registration. The tool boundary preserves interruption as a typed error, including combined interruption and cleanup failure, instead of treating it as a known quiescent startup failure. For isolated startup, it retains the checkout and reports its path/branch while quiescence is unknown; a paused binding hook can therefore finish writing without its cwd disappearing. Known non-interrupted startup failures retain the existing empty-worktree cleanup. + +Already-running extension hooks cannot be forcibly stopped by an SDK API that offers no cancellation; bounded cleanup is not a claim that arbitrary extension code becomes cancellable. + +### Cancelled Workflow worktree acquisition skips cleanup + +A disposable Git repository's post-checkout hook held `git worktree add` in progress. Cancelling the Workflow, releasing the hook and awaiting run completion left two worktrees instead of one. The abort check threw after successful creation but before the owning `try/finally`. + +The repair moves that check inside the existing cleanup region. Existing policies remain authoritative: an empty worktree may be reclaimed, while user-produced dirty work and uncertain cleanup retain their evidence. + +### Byte pressure hides the newest failure + +With an initial task, twenty 16-KiB tool results and a final error, the production progress projection omitted the final error and last showed `call-15`. Its entry-count policy retained the first and newest entries, but its byte-budget pass consumed the budget from oldest to newest. + +The repair reserves the initial entry and allocates the remaining text budget from newest to oldest, then restores chronological display. Tool identity, error state and timing travel with retained entries. Existing per-entry truncation and omission markers remain explicit. The 256-KiB limit is the source-text budget; markers and metadata are not a claim of a strict serialized-object byte limit. Pi's canonical messages and tool results are unchanged. + +### Tool preview scans content it never displays + +`toolPreview` split the entire tool output into lines for every update even though the UI needed only its first nonempty line. An exact-source synthetic probe returned the same five characters (`ready`) while ten updates over an 8-MiB dense-line payload took approximately 442 ms. This is a stress case, not a claim that ordinary Pi Bash updates routinely contain 8 MiB; native tools may already truncate their partial results. + +The repair finds the first meaningful text and examines only the first-line preview window, using the existing consumer's 64-KiB character limit. Necessary leading-whitespace search remains linear; trailing multiline output no longer creates a whole-output array. Full tool results remain canonical and available independently of the preview. + +One local Node 26.3.0 diagnostic comparison reused identical text payloads, extracted the before function from the frozen source and imported the repaired helper. Both returned `ready`; timing excluded construction of the payload. This isolates preview work rather than end-to-end latency: + +| Payload and repeated calls | Before | After | +| --- | --- | --- | +| 50 KiB, 80-character lines, 1000 calls | 8.882 ms | 0.310 ms | +| Synthetic 8 MiB, 80-character lines, 100 calls | 165.259 ms | 0.013 ms | +| Synthetic 8 MiB, short lines, 10 calls | 440.523 ms | 0.003 ms | + +Small after-values approach timer/JIT noise. The useful result is removal of whole-tail splitting, enforced by a deterministic regression, not a precise whole-application speedup multiplier. + +## Evidence and limits + +Regression tests exercise production Effect cancellation and the real Git worktree lifecycle, plus byte-pressure error retention and first-line preview behavior. `bun run check` passed. Standard `bun run test` passed 1446 Node tests with one platform skip and 30 Vitest tests. Two independent reviews are clean after closing the interrupted-startup worktree classification gap. Remote CI status is recorded in the PR. No paid model calls are needed to reproduce these deterministic runtime defects. + +A separate synthetic Cursor transport probe identified repeated copying while accumulating a large fragmented Connect frame. It is outside this bounded child acquisition/progress repair, as are speculative Graph recomputation and renderer-retention concerns without a completed failure proof. This record is a diagnostic investigation, not a formal throughput Benchmark or an assertion that every possible performance issue was resolved. diff --git a/docs/research/README.md b/docs/research/README.md index 6840d3d6..302bb726 100644 --- a/docs/research/README.md +++ b/docs/research/README.md @@ -4,6 +4,8 @@ Research records preserve sourced investigation and distinguish observations, in ## Validated investigations +- [`CHILD_ACQUISITION_AND_PROGRESS_2026-09-07.md`](CHILD_ACQUISITION_AND_PROGRESS_2026-09-07.md) — cancelled child/worktree acquisitions and stale progress evidence under output pressure ([#428](https://github.com/openpi-dev/openpi/issues/428)). + - [`WORKFLOW_CHILD_FAILURES_2026-09-07.md`](WORKFLOW_CHILD_FAILURES_2026-09-07.md) — child tool transport, cwd and timeout failure mechanisms, intended capability inheritance, and acceptance limits ([#424](https://github.com/openpi-dev/openpi/issues/424)). - [`WORKFLOW_DASHBOARD_REFRESH_2026-09-07.md`](WORKFLOW_DASHBOARD_REFRESH_2026-09-07.md) — repeated synchronous history loading on dashboard animation ticks, its regression boundary, and measurement limits ([#420](https://github.com/openpi-dev/openpi/issues/420)). diff --git a/extensions/subagents/index.ts b/extensions/subagents/index.ts index 06819fef..f753c77b 100644 --- a/extensions/subagents/index.ts +++ b/extensions/subagents/index.ts @@ -145,6 +145,7 @@ import { createSubagentResultDelivery } from "./src/result-delivery.ts"; import { createSubagentRuntime, runTool, + SubagentToolInterruptedError, type SubagentRuntime, } from "./src/runtime.ts"; import { openSubagentPicker, openSubagentTakeover } from "./src/ui/takeover.ts"; @@ -985,11 +986,22 @@ export default function ( interruptMessage: "Subagent spawn aborted.", }); } catch (error) { - // The session scope owns reclamation, but it never opened, so this - // worktree would otherwise be orphaned on disk. + // Known startup failures can reclaim their empty checkout. Interrupted + // startup must preserve it while asynchronous acquisition may continue. if (worktree) { const spawnError = error instanceof Error ? error.message : String(error); + // Cancelling Effect acquisition does not prove an asynchronous + // factory or extension hook has quiesced. It may still use this cwd. + if ( + signal?.aborted || + error instanceof SubagentToolInterruptedError + ) { + throw new Error( + `${spawnError}; startup quiescence is unknown; checkout preserved at ${worktree.path} (branch ${worktree.branch})`, + { cause: error }, + ); + } let cleanupWarning: string | undefined; let cleanupError: unknown; try { diff --git a/extensions/subagents/src/backends/pi.ts b/extensions/subagents/src/backends/pi.ts index ab968406..098f4317 100644 --- a/extensions/subagents/src/backends/pi.ts +++ b/extensions/subagents/src/backends/pi.ts @@ -25,6 +25,7 @@ import { import type { Cause, Scope } from "effect"; import { Effect, Queue, Stream } from "effect"; import { resolveAgentModel } from "../agent-types.ts"; +import { toolPreview } from "./tool-preview.ts"; import type { SubagentBackend, SubagentCleanupReceipt, @@ -118,27 +119,6 @@ function safeJson(value: unknown): string | undefined { } } -/** First non-empty line of a tool result-ish value (v1 liveToolPreview). */ -function toolPreview(value: unknown): string | undefined { - if (typeof value === "string") { - return value - .split("\n") - .find((line) => line.trim()) - ?.trim(); - } - if (!value || typeof value !== "object") return undefined; - const content = (value as { content?: unknown }).content; - if (!Array.isArray(content)) return undefined; - for (const part of content) { - if (!part || typeof part !== "object") continue; - const record = part as { type?: unknown; text?: unknown }; - if (record.type !== "text" || typeof record.text !== "string") continue; - const firstLine = record.text.split("\n").find((line) => line.trim()); - if (firstLine) return firstLine.trim(); - } - return undefined; -} - function assistantParts(msg: AssistantMessage): TranscriptPart[] { const parts: TranscriptPart[] = []; for (const part of msg.content) { @@ -215,42 +195,70 @@ const makePiSession = ( capturedStructured = encodeStructuredResult(value); }); + // Own the session before asynchronous startup. Interruption can happen + // before the normal backend finalizer has been installed. + let acquiringSession: AgentSession | undefined; + let startupOwned = true; + const cleanupStartup = () => + acquiringSession + ? shutdownAndDisposeChildSession(acquiringSession, { + abort: true, + timeoutMs: options.shutdownTimeoutMs, + }) + : Promise.resolve(); + yield* Effect.addFinalizer(() => + Effect.promise(async () => { + if (startupOwned) await cleanupStartup(); + }), + ); + const session = yield* Effect.tryPromise({ - try: async () => { - const appendSystemPrompt = [ - ...(task.appendSystemPrompt ?? []), - ...(structuredOutputTool - ? [STRUCTURED_OUTPUT_SYSTEM_INSTRUCTION] - : []), - ]; - const { loader, settingsManager } = await createChildResources({ - cwd: task.cwd, - projectTrusted: task.parent.projectTrusted, - ...(appendSystemPrompt.length > 0 ? { appendSystemPrompt } : {}), - }); - const { session } = await ( - options.sessionFactory ?? createAgentSession - )({ - cwd: task.cwd, - sessionManager: SessionManager.create(task.cwd), - settingsManager, - resourceLoader: loader, - model, - thinkingLevel, - ...(structuredOutputTool - ? { customTools: [structuredOutputTool] } - : {}), - ...childToolPolicy( - childToolsWithStructuredOutput( - task.tools, - structuredOutputTool !== undefined, - ), - ), - }); - // Start child extension session hooks/resources in headless mode. - // A rejection here would otherwise leak the freshly created session: - // the scope finalizer that owns cleanup is only registered later. + try: async (signal) => { + const checkCancelled = () => { + if (signal.aborted) + throw signal.reason ?? new Error("Subagent startup cancelled"); + }; + const onCancelled = () => { + void cleanupStartup().catch(() => {}); + }; + signal.addEventListener("abort", onCancelled, { once: true }); try { + checkCancelled(); + const appendSystemPrompt = [ + ...(task.appendSystemPrompt ?? []), + ...(structuredOutputTool + ? [STRUCTURED_OUTPUT_SYSTEM_INSTRUCTION] + : []), + ]; + const { loader, settingsManager } = await createChildResources({ + cwd: task.cwd, + projectTrusted: task.parent.projectTrusted, + ...(appendSystemPrompt.length > 0 ? { appendSystemPrompt } : {}), + }); + checkCancelled(); + const { session } = await ( + options.sessionFactory ?? createAgentSession + )({ + cwd: task.cwd, + sessionManager: SessionManager.create(task.cwd), + settingsManager, + resourceLoader: loader, + model, + thinkingLevel, + ...(structuredOutputTool + ? { customTools: [structuredOutputTool] } + : {}), + ...childToolPolicy( + childToolsWithStructuredOutput( + task.tools, + structuredOutputTool !== undefined, + ), + ), + }); + acquiringSession = session; + checkCancelled(); + // Never start extension binding for a factory that completed after + // cancellation. Already-running hooks retain bounded cleanup ownership. await bindChildSessionExtensions( session, childToolsWithStructuredOutput( @@ -258,13 +266,14 @@ const makePiSession = ( structuredOutputTool !== undefined, ), ); + checkCancelled(); + return session; } catch (error) { - await shutdownAndDisposeChildSession(session, { - timeoutMs: options.shutdownTimeoutMs, - }); + await cleanupStartup(); throw error; + } finally { + signal.removeEventListener("abort", onCancelled); } - return session; }, catch: (error) => new SpawnError({ message: boundedError(error) }), }); @@ -650,6 +659,8 @@ const makePiSession = ( }), ); + startupOwned = false; + /** Start a fresh run (v1 manager.run): fire-and-forget, errors -> events. */ const startRun = (text: string) => { if (state.activePrompt) { diff --git a/extensions/subagents/src/backends/tool-preview.ts b/extensions/subagents/src/backends/tool-preview.ts new file mode 100644 index 00000000..f0b09f93 --- /dev/null +++ b/extensions/subagents/src/backends/tool-preview.ts @@ -0,0 +1,29 @@ +// Match the manager's existing transcript text limit; canonical Pi tool results +// remain intact. Only the normalized event's single-line preview is bounded. +const TOOL_PREVIEW_MAX_LENGTH = 64 * 1_024; + +function firstMeaningfulLine(text: string) { + // Search only until the first non-whitespace character. Unlike splitting the + // entire log, this preserves blank-line behavior without visiting its tail. + const start = text.search(/\S/); + if (start < 0) return undefined; + const prefix = text.slice(start, start + TOOL_PREVIEW_MAX_LENGTH); + const newline = prefix.indexOf("\n"); + return (newline < 0 ? prefix : prefix.slice(0, newline)).trimEnd(); +} + +/** First meaningful line of a tool result, without splitting accumulated logs. */ +export function toolPreview(value: unknown) { + if (typeof value === "string") return firstMeaningfulLine(value); + if (!value || typeof value !== "object") return undefined; + const content = (value as { content?: unknown }).content; + if (!Array.isArray(content)) return undefined; + for (const part of content) { + if (!part || typeof part !== "object") continue; + const record = part as { type?: unknown; text?: unknown }; + if (record.type !== "text" || typeof record.text !== "string") continue; + const firstLine = firstMeaningfulLine(record.text); + if (firstLine) return firstLine; + } + return undefined; +} diff --git a/extensions/subagents/src/runtime.ts b/extensions/subagents/src/runtime.ts index 7587e129..a480e428 100644 --- a/extensions/subagents/src/runtime.ts +++ b/extensions/subagents/src/runtime.ts @@ -45,6 +45,9 @@ export function createSubagentRuntime(config: SubagentManagerConfig = {}) { export type SubagentRuntime = ReturnType; +/** Canonical interruption, distinct from a known startup failure. */ +export class SubagentToolInterruptedError extends Error {} + /** * Run an effect from an async tool handler. Typed failures and defects are * converted to thrown Errors (what pi's tool contract expects); interruption @@ -60,9 +63,13 @@ export async function runTool( options.signal ? { signal: options.signal } : undefined, ); if (Exit.isSuccess(exit)) return exit.value; - if (Cause.hasInterruptsOnly(exit.cause)) { - throw new Error(options.interruptMessage ?? "Operation was aborted."); - } const [first] = Cause.prettyErrors(exit.cause); + if (Cause.hasInterrupts(exit.cause)) { + const interrupted = options.interruptMessage ?? "Operation was aborted."; + const detail = Cause.hasInterruptsOnly(exit.cause) + ? "" + : ` ${first?.message ?? Cause.pretty(exit.cause)}`; + throw new SubagentToolInterruptedError(`${interrupted}${detail}`); + } throw new Error(first?.message ?? Cause.pretty(exit.cause)); } diff --git a/extensions/workflows/index.ts b/extensions/workflows/index.ts index c928dc98..ff0f5a9c 100644 --- a/extensions/workflows/index.ts +++ b/extensions/workflows/index.ts @@ -1977,11 +1977,6 @@ export default function workflows( worktree = created.worktree; if (!runSettled) record.worktreeBranch = worktree.branch; } - if (runSignal.aborted || runSettled) { - throw runSignal.reason instanceof Error - ? runSignal.reason - : new Error("Workflow was aborted"); - } const agentCwd = worktree?.path ?? requestedCwd; // Inside the try, not before it: building resources can throw @@ -1989,6 +1984,12 @@ export default function workflows( // would skip the finally and leak the worktree permanently — // nothing sweeps `.git/pi-worktrees/` afterwards. try { + if (runSignal.aborted || runSettled) { + throw runSignal.reason instanceof Error + ? runSignal.reason + : new Error("Workflow was aborted"); + } + let rejectResourceLoad: (() => void) | undefined; const resourceAbort = new Promise((_resolve, reject) => { rejectResourceLoad = () => diff --git a/extensions/workflows/progress-projection.ts b/extensions/workflows/progress-projection.ts index 59303f61..a957b5c6 100644 --- a/extensions/workflows/progress-projection.ts +++ b/extensions/workflows/progress-projection.ts @@ -250,8 +250,11 @@ export class AgentProgressProjection { snapshot( toolTimings: ReadonlyMap = new Map(), ): AgentProgressProjectionSnapshot { + // Reserve the initial task, then spend the remaining byte budget on the + // newest evidence. Forward selection would silently discard final errors + // after enough large tool results, even below the entry-count limit. const selected = this.firstEntry - ? [this.firstEntry, ...this.tailEntries] + ? [this.firstEntry, ...this.tailEntries.slice().reverse()] : []; const transcript: TranscriptEntry[] = []; let totalBytes = 0; @@ -277,6 +280,9 @@ export class AgentProgressProjection { : { timestamp: entry.timestamp }), }); } + // Budgeting order is not display order: keep the retained tail chronological. + const newestFirstTail = transcript.splice(1); + transcript.push(...newestFirstTail.reverse()); if (transcript.length < this.totalEntries) { transcript.push({ role: "toolResult", diff --git a/tests/extensions/subagents/pi-backend-lifecycle.test.ts b/tests/extensions/subagents/pi-backend-lifecycle.test.ts index bd7c5d0b..f4a5f0d5 100644 --- a/tests/extensions/subagents/pi-backend-lifecycle.test.ts +++ b/tests/extensions/subagents/pi-backend-lifecycle.test.ts @@ -1294,3 +1294,59 @@ test("adapter scope cleanup is bounded across shutdown timeouts and failures", a assert.equal(failedHarness.calls.shutdowns, 1); assert.equal(failedHarness.calls.disposals, 1); }); + +for (const stage of ["factory", "binding"] as const) { + test(`cancelling during child ${stage} disposes late acquisition exactly once`, async () => { + const entered = deferred(); + const gate = deferred(); + const harness = createPiAgentSessionHarness({ + model: FIXTURE_MODEL as AgentSession["model"], + activeTools: ["read"], + shutdown: async () => {}, + bind: + stage === "binding" + ? async () => { + entered.resolve(); + await gate.promise; + } + : undefined, + }); + const backend = makePiBackend({ + sessionFactory: async () => { + if (stage === "factory") { + entered.resolve(); + await gate.promise; + } + return { session: harness.session }; + }, + shutdownTimeoutMs: 50, + }); + const abort = new AbortController(); + const result = Effect.runPromise( + Effect.scoped(backend.spawn(task("cancel acquisition"))), + { signal: abort.signal }, + ).catch(() => undefined); + try { + await entered.promise; + abort.abort(); + await result; + if (stage === "binding") { + await waitFor( + () => harness.calls.disposals === 1, + "cancelled binding cleanup", + ); + } + gate.resolve(); + await waitFor(() => harness.calls.disposals === 1, "late child cleanup"); + assert.equal(harness.calls.bindings.length, stage === "factory" ? 0 : 1); + assert.equal(harness.calls.aborts, 1); + assert.equal(harness.calls.shutdowns, 1); + assert.equal(harness.calls.disposals, 1); + assert.deepEqual(harness.calls.prompts, []); + } finally { + abort.abort(); + gate.resolve(); + await result; + } + }); +} diff --git a/tests/extensions/subagents/startup-worktree.test.ts b/tests/extensions/subagents/startup-worktree.test.ts new file mode 100644 index 00000000..cf08fe03 --- /dev/null +++ b/tests/extensions/subagents/startup-worktree.test.ts @@ -0,0 +1,239 @@ +import assert from "node:assert/strict"; +import { execFileSync } from "node:child_process"; +import { + existsSync, + mkdtempSync, + mkdirSync, + writeFileSync, + readFileSync, + rmSync, +} from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import test from "node:test"; +import { Cause, Effect } from "effect"; +import type { + ExtensionAPI, + ExtensionContext, + AgentSession, +} from "@earendil-works/pi-coding-agent"; +import subagents from "../../../extensions/subagents/index.ts"; +import { SpawnError } from "../../../extensions/subagents/src/domain.ts"; +import { makePiBackend } from "../../../extensions/subagents/src/backends/pi.ts"; +import { + __setSubagentTestBackends, + createSubagentRuntime, + runTool, + SubagentToolInterruptedError, +} from "../../../extensions/subagents/src/runtime.ts"; +import { createPiAgentSessionHarness } from "../../support/pi-agent-session-harness.ts"; + +function deferred() { + let resolve!: () => void; + const promise = new Promise((done) => { + resolve = done; + }); + return { promise, resolve }; +} + +for (const interrupted of [true, false]) { + test(`isolated startup ${interrupted ? "interruption preserves a pending hook cwd" : "known failure reclaims the empty checkout"}`, async () => { + const root = mkdtempSync(path.join(tmpdir(), "openpi-startup-worktree-")); + const cwd = path.join(root, "repo"); + mkdirSync(cwd); + const previous = process.env.PI_CODING_AGENT_DIR; + process.env.PI_CODING_AGENT_DIR = path.join(root, "agent"); + const git = (args: string[]) => + execFileSync("git", args, { + cwd, + encoding: "utf8", + stdio: ["ignore", "pipe", "pipe"], + }); + git(["init", "-q"]); + git(["config", "user.email", "fixture@example.com"]); + git(["config", "user.name", "fixture"]); + writeFileSync(path.join(cwd, "file.txt"), "fixture"); + git(["add", "."]); + git(["commit", "-qm", "fixture"]); + const entered = deferred(); + const release = deferred(); + const resumed = deferred(); + let childCwd = ""; + const model = { + provider: "fixture", + id: "model", + name: "fixture", + api: "openai-completions", + baseUrl: "http://127.0.0.1:1", + reasoning: false, + input: ["text"], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 8192, + maxTokens: 100, + }; + __setSubagentTestBackends([ + makePiBackend({ + sessionFactory: async (options) => { + assert.ok(options?.cwd); + childCwd = options.cwd; + const harness = createPiAgentSessionHarness({ + model: model as AgentSession["model"], + activeTools: ["read"], + bind: async () => { + entered.resolve(); + await release.promise; + try { + writeFileSync( + path.join(childCwd, "late-hook.txt"), + "late evidence", + ); + } finally { + resumed.resolve(); + } + }, + }); + return { session: harness.session }; + }, + shutdownTimeoutMs: 30, + }), + ]); + const hooks = new Map unknown>(); + let spawn: ((...args: unknown[]) => Promise) | undefined; + const pi = { + events: { on() {}, emit() {} }, + on(name: string, handler: (...args: unknown[]) => unknown) { + hooks.set(name, handler); + }, + registerTool(tool: { + name: string; + execute: (...args: unknown[]) => Promise; + }) { + if (tool.name === "subagent_spawn") spawn = tool.execute; + }, + registerCommand() {}, + registerMessageRenderer() {}, + registerEntryRenderer() {}, + getActiveTools: () => ["read"], + setActiveTools() {}, + getThinkingLevel: () => "off", + } as unknown as ExtensionAPI; + const ctx = { + cwd, + hasUI: false, + isProjectTrusted: () => false, + model, + modelRegistry: { + find: (_provider: string, id: string) => + id === "model" ? model : undefined, + getAll: () => [model], + }, + } as unknown as ExtensionContext; + subagents(pi); + const abort = new AbortController(); + try { + assert.ok(spawn); + const result = spawn( + "spawn", + { + prompt: "probe", + name: "probe", + isolation: "worktree", + ...(interrupted ? {} : { model: "fixture/missing" }), + }, + abort.signal, + undefined, + ctx, + ).catch((error: unknown) => error); + if (interrupted) { + await entered.promise; + abort.abort(); + } + const error = await result; + assert.ok(error instanceof Error); + if (interrupted) { + assert.ok( + existsSync(childCwd), + "startup hook still owns its working directory", + ); + assert.ok( + error.message.includes(childCwd), + "failure identifies preserved checkout", + ); + const branch = execFileSync("git", ["branch", "--show-current"], { + cwd: childCwd, + encoding: "utf8", + }).trim(); + assert.ok( + error.message.includes(branch), + "failure identifies preserved branch", + ); + release.resolve(); + await resumed.promise; + assert.equal( + readFileSync(path.join(childCwd, "late-hook.txt"), "utf8"), + "late evidence", + ); + assert.equal( + (git(["worktree", "list", "--porcelain"]).match(/^worktree /gm) ?? []) + .length, + 2, + ); + } else { + assert.match(error.message, /model/i); + assert.equal( + (git(["worktree", "list", "--porcelain"]).match(/^worktree /gm) ?? []) + .length, + 1, + ); + } + } finally { + release.resolve(); + if (childCwd) await resumed.promise; + await hooks.get("session_shutdown")?.({}, ctx); + __setSubagentTestBackends(undefined); + if (previous === undefined) delete process.env.PI_CODING_AGENT_DIR; + else process.env.PI_CODING_AGENT_DIR = previous; + rmSync(root, { recursive: true, force: true }); + } + }); +} + +test("runTool preserves canonical runtime interruption without a caller signal", async () => { + const runtime = createSubagentRuntime(); + try { + await assert.rejects( + runTool(runtime, Effect.interrupt, { + interruptMessage: "spawn interrupted", + }), + (error: unknown) => + error instanceof SubagentToolInterruptedError && + error.message === "spawn interrupted", + ); + await assert.rejects( + runTool( + runtime, + Effect.failCause( + Cause.combine( + Cause.interrupt(1), + Cause.die(new Error("startup cleanup failed")), + ), + ), + ), + (error: unknown) => + error instanceof SubagentToolInterruptedError && + error.message.includes("startup cleanup failed"), + ); + await assert.rejects( + runTool( + runtime, + Effect.fail(new SpawnError({ message: "known failure" })), + ), + (error: unknown) => + error instanceof Error && + !(error instanceof SubagentToolInterruptedError) && + error.message === "known failure", + ); + } finally { + await runtime.dispose(); + } +}); diff --git a/tests/extensions/subagents/tool-preview.test.ts b/tests/extensions/subagents/tool-preview.test.ts new file mode 100644 index 00000000..493b4ae0 --- /dev/null +++ b/tests/extensions/subagents/tool-preview.test.ts @@ -0,0 +1,58 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { toolPreview } from "../../../extensions/subagents/src/backends/tool-preview.ts"; + +test("tool preview retains the first meaningful line across blank lines and text parts", () => { + assert.equal(toolPreview("\n\r\n \t\n ready \r\nignored"), "ready"); + assert.equal(toolPreview("\uFEFF\u00A0ready\u00A0"), "ready"); + assert.equal(toolPreview(" \n\t\r\n"), undefined); + assert.equal(toolPreview(null), undefined); + assert.equal( + toolPreview({ + content: [ + null, + { type: "image", data: "x" }, + { type: "text", text: "\n " }, + { type: "text", text: "\n done\nrest" }, + ], + }), + "done", + ); + assert.equal( + toolPreview("\n".repeat(128 * 1024) + "late line\nignored"), + "late line", + ); +}); + +test("tool preview bounds a long first line to the existing consumer's 64KiB character limit", () => { + const text = "z".repeat(1024 * 1024); + assert.equal(toolPreview(text)?.length, 64 * 1024); + assert.equal( + toolPreview({ content: [{ type: "text", text }] })?.length, + 64 * 1024, + ); +}); + +test("tool preview does not split a large trailing log to return its short first line", () => { + const text = "ready\n" + "x\n".repeat(4 * 1024 * 1024); + const originalSplit = String.prototype.split; + String.prototype.split = function ( + separator: + | string + | RegExp + | { [Symbol.split](text: string, limit?: number): string[] }, + limit?: number, + ) { + assert.ok( + this.length <= 64 * 1024, + "preview split the full accumulated log", + ); + return Reflect.apply(originalSplit, this, [separator, limit]); + }; + try { + assert.equal(toolPreview(text), "ready"); + assert.equal(toolPreview({ content: [{ type: "text", text }] }), "ready"); + } finally { + String.prototype.split = originalSplit; + } +}); diff --git a/tests/extensions/workflows/execute.e2e.test.ts b/tests/extensions/workflows/execute.e2e.test.ts index ac587aed..c0aad385 100644 --- a/tests/extensions/workflows/execute.e2e.test.ts +++ b/tests/extensions/workflows/execute.e2e.test.ts @@ -1803,3 +1803,73 @@ test("built-in Workflow children inherit active shell/network tools in the selec __setWorkflowTestAgentSessionFactory(undefined); } }); + +for (const dirty of [false, true]) + test(`cancellation during worktree creation ${dirty ? "preserves dirty work" : "reclaims the checkout"}`, { + skip: process.platform === "win32", + }, async () => { + const marker = join(agentDir, `checkout-hook-entered-${dirty}`); + const release = join(agentDir, `checkout-hook-release-${dirty}`); + const hook = join(repoDir, ".git", "hooks", "post-checkout"); + writeFileSync( + hook, + `#!/bin/sh\n${dirty ? "printf evidence > cancellation-evidence.txt\n" : ""}touch '${marker}'\ni=0\nwhile [ ! -f '${release}' ] && [ "$i" -lt 100 ]; do sleep 0.05; i=$((i+1)); done\n`, + ); + (await import("node:fs")).chmodSync(hook, 0o755); + let run: + | Parameters[0][number] + | undefined; + __setWorkflowTestLifecycleHooks({ + onRunStarted(value) { + run = value; + }, + }); + try { + const launch = (await workflow.execute( + "cancel-checkout", + { + script: 'return await agent("probe", { isolation: "worktree" });', + background: true, + }, + undefined, + undefined, + ctx, + )) as { details: { runId: string } }; + await waitFor(() => existsSync(marker), "checkout hook"); + await workflowStop.execute("cancel-checkout-stop", { + runId: launch.details.runId, + }); + writeFileSync(release, ""); + assert.ok(run); + await run.completion; + const listing = execFileSync("git", ["worktree", "list", "--porcelain"], { + cwd: repoDir, + encoding: "utf8", + }); + assert.equal((listing.match(/^worktree /gm) ?? []).length, dirty ? 2 : 1); + const agent = ( + readWorkflowJson(launch.details.runId).agents as Array< + Record + > + )[0]; + assert.equal( + (agent?.worktreeCleanup as { removed?: boolean })?.removed, + !dirty, + ); + if (dirty) { + assert.equal(typeof agent?.worktreePath, "string"); + assert.equal( + readFileSync( + join(String(agent?.worktreePath), "cancellation-evidence.txt"), + "utf8", + ), + "evidence", + ); + } else assert.equal(agent?.worktreeBranch, undefined); + } finally { + writeFileSync(release, ""); + if (run) await run.completion; + rmSync(hook, { force: true }); + __setWorkflowTestLifecycleHooks(undefined); + } + }); diff --git a/tests/extensions/workflows/progress-projection.test.ts b/tests/extensions/workflows/progress-projection.test.ts index 3e53fa7c..d7f864d7 100644 --- a/tests/extensions/workflows/progress-projection.test.ts +++ b/tests/extensions/workflows/progress-projection.test.ts @@ -163,3 +163,49 @@ test("transcript byte limits preserve UTF-8 boundaries and explicit markers", () ); assert.match(bounded.at(-1)?.text ?? "", /retained 16 of 20 entries/); }); + +test("byte pressure retains the initial task and newest failure evidence", () => { + const messages: AgentMessage[] = [ + { role: "user", content: "initial task", timestamp: 0 }, + ...Array.from({ length: 20 }, (_, index) => ({ + role: "toolResult" as const, + toolName: "read", + toolCallId: `read-${index}`, + content: [{ type: "text" as const, text: "x".repeat(16 * 1024) }], + isError: false, + timestamp: index + 1, + })), + { + role: "toolResult", + toolName: "bash", + toolCallId: "latest-failure", + content: [{ type: "text", text: "LATEST_FAILURE: build failed" }], + isError: true, + timestamp: 99, + }, + ]; + const projection = new AgentProgressProjection(messages); + const transcript = projection.snapshot( + new Map([ + ["latest-failure", { startedAt: 97, finishedAt: 99, durationMs: 2 }], + ]), + ).transcript; + assert.equal(transcript[0]?.text, "initial task"); + const latest = transcript.at(-2); + assert.equal(latest?.text, "LATEST_FAILURE: build failed"); + assert.equal(latest?.isError, true); + assert.equal(latest?.toolCallId, "latest-failure"); + assert.equal(latest?.durationMs, 2); + assert.equal( + transcript.some((entry) => entry.toolCallId === "read-0"), + false, + ); + const timestamps = transcript.flatMap((entry) => + entry.timestamp === undefined ? [] : [entry.timestamp], + ); + assert.deepEqual( + timestamps, + [...timestamps].sort((a, b) => a - b), + ); + assert.match(transcript.at(-1)?.text ?? "", /transcript truncated/); +}); From 342ac927736b64d12c8953e0c4c932eedecb3451 Mon Sep 17 00:00:00 2001 From: tt-a1i Date: Mon, 7 Sep 2026 08:13:19 +0800 Subject: [PATCH 2/2] docs: link child startup and progress repair PR --- docs/research/CHILD_ACQUISITION_AND_PROGRESS_2026-09-07.md | 1 + 1 file changed, 1 insertion(+) diff --git a/docs/research/CHILD_ACQUISITION_AND_PROGRESS_2026-09-07.md b/docs/research/CHILD_ACQUISITION_AND_PROGRESS_2026-09-07.md index cceba566..54f3e212 100644 --- a/docs/research/CHILD_ACQUISITION_AND_PROGRESS_2026-09-07.md +++ b/docs/research/CHILD_ACQUISITION_AND_PROGRESS_2026-09-07.md @@ -4,6 +4,7 @@ - Created / verified: 2026-09-07 - Source boundary: `eaf470bab4ac2dda607d16c6ddee66dc35367527` (main after PR #426) - Issue: [#428](https://github.com/openpi-dev/openpi/issues/428) +- Repair PR: [#429](https://github.com/openpi-dev/openpi/pull/429) - Supersedes: none ## Scope and provenance