From 7a1043e5e0bccf9b1884311b477a997e9c5eef9b Mon Sep 17 00:00:00 2001 From: JB Date: Thu, 3 Sep 2026 23:39:33 -0400 Subject: [PATCH] fix(interactive-subagents): harden parallel lifecycle delivery --- docs/todos.md | 18 +- extensions/interactive-subagents/README.md | 12 +- .../pi-extension/subagents/herdr.ts | 26 +- .../pi-extension/subagents/index.ts | 396 ++++++++++++++---- .../pi-extension/subagents/orca.ts | 19 +- .../pi-extension/subagents/subagent-done.ts | 7 +- .../pi-extension/subagents/surface.ts | 42 +- .../pi-extension/subagents/tmux.ts | 21 +- .../agents/test-default-autonomous.md | 9 + .../integration/subagent-lifecycle.test.ts | 109 ++++- extensions/interactive-subagents/test/test.ts | 320 +++++++++++++- 11 files changed, 848 insertions(+), 131 deletions(-) create mode 100644 extensions/interactive-subagents/test/integration/agents/test-default-autonomous.md diff --git a/docs/todos.md b/docs/todos.md index 6260246..6abe459 100644 --- a/docs/todos.md +++ b/docs/todos.md @@ -20,23 +20,9 @@ Slice 1 found two Nono 0.74 limitations while building behavioral policy proofs. The strict `workflow` extension repeatedly failed before implementation because planner output and persisted state-machine constraints were too brittle for adaptive subagent orchestration. In a separate session, design a repository-managed skill where the main Pi agent owns the plan, task tree, isolated worktree, explicit user approval gates, verification, and independent review while launching standard `scout` and `worker` subagents directly. Preserve file-ownership boundaries, no inferred approvals, review independence, and resumable repository evidence, but prefer native Pi primitives over a custom orchestration engine. Evaluate a small committed state manifest for crash recovery before retiring or archiving the extension. Use the abandoned `.worktrees/workflow-plan-seed` implementation only as reference; do not integrate it by default. -## Support subagent launches through `multi_tool_use.parallel` +## Support finished-session resume for Claude CLI subagents -A user reported that launching subagents through Pi's `multi_tool_use.parallel` wrapper does not work, although equivalent launches succeed outside the wrapper in another session. The tool surface in the incident below was the harness-provided `functions.subagent` tool wrapped by `multi_tool_use.parallel`; it has not been proven to be the `subagent_spawn`/`subagent_wait` surface from this repository's `@jbcode/pi-subagents` package. Keep those implementations separate until tracing identifies the owner. - -A second incident occurred while planning the skill-led development workflow on 03SEP2026. The complete sanitized wrapper payload, prompts, expected lifecycle, observed lifecycle, and direct-launch control are retained in [`docs/fixtures/subagent-parallel-idle-delivery-03SEP2026.json`](./fixtures/subagent-parallel-idle-delivery-03SEP2026.json). Environment evidence captured at `2026-09-03T15:54:57-04:00`: - -- Pi `0.84.4`; repository subagent package `0.1.0` was present but is not yet identified as the failing tool implementation. -- Parent session `01a06885-7f00-7cff-bc2f-334984151941`, file `/home/jb/.pi/agent/sessions/--srv-data-code-jb-pi-harness-tools--/2026-09-03T18-26-14-400Z_01a06885-7f00-7cff-bc2f-334984151941.jsonl`. -- Parent backend `openai-codex/gpt-5.6-sol`, reasoning `high`. -- One `multi_tool_use.parallel` call launched `workflow-architecture` (`codebase-analyzer`), `workflow-history` (`precedent-locator`), and `workflow-patterns` (`codebase-pattern-finder`) with cwd `/srv/data/code/jb-pi/harness-tools`. Each wrapped result returned only a successful fire-and-forget acknowledgement; no child task or session ID was exposed. -- The children later appeared completed in the UI, but their automatic result payloads did not wake the idle orchestrator. The orchestrator remained parked until the user intervened. -- Follow-up `subagent_message` calls for all three returned acknowledgements saying the message was delivered to a running subagent, inconsistent with the completed UI state. -- A later direct, unwrapped `subagent` launch named `skill-workflow-plan-review` in the same parent session completed and delivered its result automatically after 14 minutes 9 seconds. - -Use the retained three-agent JSON payload as the captured failing fixture. Also reproduce with two minimal independent children, comparing direct calls with one wrapped call. Include a case where the parent performs other work and then becomes idle before both children finish. Preserve exact tool-call payloads, acknowledgement timestamps, child IDs when available, terminal-state timestamps, parent idle transitions, callback queue entries, and delivered steer messages. Capture whether failure occurs during schema validation, dispatch, asynchronous acknowledgement, task registration, terminal-state publication, idle-parent wake-up, or completion delivery. - -The wrapper must return promptly for fire-and-forget launches, preserve each subagent's name and working directory, expose enough identity to correlate acknowledgements and terminal results, transition UI and tool-visible status consistently, and deliver each result exactly once without polling or user intervention. Add focused integration coverage for successful parallel launch, staggered completion while the parent is idle, one-child failure, cancellation, late steering of an apparently completed child, and duplicate-delivery prevention before changing either tool. +`subagent_message` can steer a running Claude CLI pane, but a completed Claude launch has no Pi session file and cannot use the existing finished-session resume path. Store the Claude session identifier and original sandbox metadata in the parent registry, then resume with Claude's native session mechanism while retaining the original name and fire-and-forget result delivery. Until implemented, keep the README's finished-session contract explicitly limited to Pi-backed subagents. ## Investigate transient concurrent NeMo benchmark failures diff --git a/extensions/interactive-subagents/README.md b/extensions/interactive-subagents/README.md index 3abe993..96e951b 100644 --- a/extensions/interactive-subagents/README.md +++ b/extensions/interactive-subagents/README.md @@ -2,7 +2,7 @@ # pi-interactive-subagents -Async subagents for [pi](https://github.com/badlogic/pi-mono), running in Herdr tabs, Orca terminal tabs, or tmux panes. Spawn a sub-agent, keep working in the main session, and get the result steered back when it finishes. Fully non-blocking. +Async subagents for [pi](https://github.com/badlogic/pi-mono), running in Herdr tabs, Orca terminal tabs, or tmux panes. Spawn a sub-agent, keep working in the main session, and get the result as an automatic follow-up when it finishes. Fully non-blocking. The backend is selected automatically: Herdr is used when Pi runs in a Herdr-managed pane (`HERDR_ENV=1`), then Orca when Pi runs in an Orca-managed terminal (`ORCA_TERMINAL_HANDLE`), otherwise tmux is used when available. @@ -16,7 +16,7 @@ Subagents 2 running ○ 00:45 scout-2 waiting 2m ``` -Spawn several in parallel — they run concurrently and steer results back independently as each finishes. +Spawn several in parallel — they run concurrently and deliver results independently as each finishes. Running subagents survive a parent `/reload`. Their terminal sessions and completion watchers stay active, then rebind result delivery and the status widget to the fresh Pi extension runtime. Replacing or closing the parent session still cancels its running subagents. This lifecycle behavior requires Pi 0.68 or newer. @@ -62,8 +62,8 @@ subagent({ agent: "worker", name: "dark-mode", task: "Implement the dark mode to subagent_message({ name: "scout", message: "Also check the auth middleware" }); ``` -- **Running** — the message is typed into the live pane (newlines flattened) and picked up at the next turn boundary. The call returns immediately; the eventual completion still arrives as a steer message. -- **Finished** — the session is resumed with the message as the follow-up task, like a fresh spawn: fire-and-forget, always autonomous, result steered back later. The resumed run reclaims its original name. +- **Running** — the message is typed into the live pane (newlines flattened) and picked up at the next turn boundary. The call returns immediately; the eventual completion still arrives as an automatic follow-up. +- **Finished** — a Pi-backed session is resumed with the message as the follow-up task, like a fresh spawn: fire-and-forget, always autonomous, result delivered automatically later. The resumed run reclaims its original name. Claude CLI sessions can be steered while running but cannot currently be resumed after completion. Every spawn records name → session file in `artifacts//subagent-registry.json`, so names stay addressable across pi restarts. A nested sub-agent that spawns children gets its own registry keyed by its own session id. Resume is refused with a clear error (listing known names) if the name isn't registered, the session file is gone, or the session predates sandboxed resume. @@ -116,7 +116,7 @@ You are a specialized agent that does X... | `skills` | string | Comma-separated skill names to auto-load | | `session-mode` | string | `standalone` (default), `lineage-only`, or `fork` — see below | | `system-prompt` | string | `append` or `replace`: pass the body as the child's `--append-system-prompt` / `--system-prompt`. Omit and the body is prepended to the task prompt instead | -| `auto-exit` | boolean | Auto-shutdown when the agent finishes (see below) | +| `auto-exit` | boolean | Auto-shutdown when the agent finishes. Defaults to `true` (see below) | | `interactive` | boolean | Whether stall/recovery transitions wake the parent (see below) | | `cwd` | string | Default working directory | | `disable-model-invocation` | boolean | Hide from `subagents_list`; still spawnable by explicit name | @@ -130,7 +130,7 @@ You are a specialized agent that does X... ### auto-exit -With `auto-exit: true`, the session shuts down when the agent's turn ends — the agent just writes its final message and stops (there is no "done" tool). The last assistant message becomes the summary returned to the parent. Recommended for all autonomous agents. +`auto-exit` defaults to `true`, matching the fire-and-forget tool contract: the session shuts down when the agent's turn ends, and the parent receives the last assistant message automatically (there is no "done" tool). Set `auto-exit: false` only for a user-driven session that should remain open until the user exits its pane. Notes: diff --git a/extensions/interactive-subagents/pi-extension/subagents/herdr.ts b/extensions/interactive-subagents/pi-extension/subagents/herdr.ts index 8f2da59..996211c 100644 --- a/extensions/interactive-subagents/pi-extension/subagents/herdr.ts +++ b/extensions/interactive-subagents/pi-extension/subagents/herdr.ts @@ -166,13 +166,35 @@ export async function readScreenAsync(surface: string, lines = 50): Promise { + try { + requireHerdr(); + await execFileAsync( + "herdr", + ["pane", "read", surface, "--source", "recent-unwrapped", "--lines", "1"], + { encoding: "utf8", timeout: 5_000, killSignal: "SIGKILL", signal }, + ); + return true; + } catch { + return false; + } +} + export function closeSurface(surface: string): void { requireHerdr(); const tabId = tabByPane.get(surface); if (tabId) { - execFileSync("herdr", ["tab", "close", tabId], { encoding: "utf8" }); + try { + execFileSync("herdr", ["tab", "close", tabId], { encoding: "utf8" }); + } catch { + // Replaced-shell launches may close their own tab before watcher cleanup. + } tabByPane.delete(surface); return; } - execFileSync("herdr", ["pane", "close", surface], { encoding: "utf8" }); + try { + execFileSync("herdr", ["pane", "close", surface], { encoding: "utf8" }); + } catch { + // Closing an already-ended pane is idempotent cleanup. + } } diff --git a/extensions/interactive-subagents/pi-extension/subagents/index.ts b/extensions/interactive-subagents/pi-extension/subagents/index.ts index 191435a..c94e5e7 100644 --- a/extensions/interactive-subagents/pi-extension/subagents/index.ts +++ b/extensions/interactive-subagents/pi-extension/subagents/index.ts @@ -12,6 +12,7 @@ import { mkdirSync, copyFileSync, unlinkSync, + rmSync, } from "node:fs"; import { homedir } from "node:os"; import { @@ -22,6 +23,7 @@ import { sendLongCommand, pollForExit, closeSurface, + isSurfaceAlive, shellEscape, readScreen, getMuxBackend, @@ -350,10 +352,11 @@ function discoverAgentDefinitions(): ListedAgentDefinition[] { function resolveSubagentPaths( params: Static, agentDefs: AgentDefaults | null, + parentCwd = process.cwd(), ): { effectiveCwd: string | null; localAgentDir: string | null; effectiveAgentDir: string } { const rawCwd = params.cwd ?? agentDefs?.cwd ?? null; const cwdIsFromAgent = !params.cwd && agentDefs?.cwd != null; - const cwdBase = cwdIsFromAgent ? getAgentConfigDir() : process.cwd(); + const cwdBase = cwdIsFromAgent ? getAgentConfigDir() : parentCwd; const effectiveCwd = rawCwd ? rawCwd.startsWith("/") ? rawCwd @@ -400,22 +403,37 @@ function resolveLaunchBehavior( }; } +/** + * Fire-and-forget launches must terminate without a human closing their pane. + * Profiles can opt into a user-driven session with `auto-exit: false`. + */ +function resolveEffectiveAutoExit(agentDefs: AgentDefaults | null): boolean { + return agentDefs?.autoExit ?? true; +} + +function autoExitEnvAssignment(autoExit: boolean): string { + return `PI_SUBAGENT_AUTO_EXIT=${autoExit ? "1" : "0"}`; +} + +function resolveClaudeSentinelFile(id: string, autoExit: boolean): string | undefined { + return autoExit ? `/tmp/pi-claude-${id}-done` : undefined; +} + /** * Decide whether a subagent is interactive (user-driven, long-running). * * Resolution order: * 1. Explicit `interactive` frontmatter field on the agent. - * 2. Default: the inverse of `auto-exit`. Agents that auto-exit are - * autonomous (scout, researcher) and the parent session should be - * woken on stall/recovery transitions. Agents that don't auto-exit are - * driven by the user in their own pane (worker) and stall pings are noise. + * 2. Default: the inverse of the effective `auto-exit` behavior. Autonomous + * agents wake the parent on stall/recovery transitions; explicitly + * user-driven agents stay quiet because the user is already in their pane. */ function resolveEffectiveInteractive( _params: Static, agentDefs: AgentDefaults | null, ): boolean { if (agentDefs?.interactive != null) return agentDefs.interactive; - return !(agentDefs?.autoExit ?? false); + return !resolveEffectiveAutoExit(agentDefs); } function loadAgentDefaults(agentName: string): AgentDefaults | null { @@ -687,7 +705,10 @@ interface RunningSubagent { cli?: string; sentinelFile?: string; completionFile?: string; + completionObserved?: boolean; statusState: SubagentStatusState; + /** Parent-session generation. Reload preserves it; replacement increments it. */ + parentGeneration: number; /** * When true, status transitions (stalled/recovered) do not wake the parent * session via a steer message. The widget still updates locally. Used for @@ -731,13 +752,16 @@ function isStatusWidgetClaim(value: unknown): value is StatusWidgetClaim { interface SharedSubagentRuntime { runningSubagents: Map; + resumingNames: Set; latestCtx: ExtensionContext | null; latestPi: ExtensionAPI | null; widgetInterval: ReturnType | null; statusInterval: ReturnType | null; standaloneWidgetSyncScheduled: boolean; pendingDeliveries: PendingDelivery[]; + resultDeliveryStates: Map; deliveryFlushScheduled: boolean; + parentGeneration: number; activeRuntimeToken: symbol | null; registeredRuntimeToken: symbol | null; statusWidgetClaim: StatusWidgetClaim | null; @@ -748,13 +772,16 @@ interface SharedSubagentRuntime { function createSharedRuntime(): SharedSubagentRuntime { return { runningSubagents: new Map(), + resumingNames: new Set(), latestCtx: null, latestPi: null, widgetInterval: null, statusInterval: null, standaloneWidgetSyncScheduled: false, pendingDeliveries: [], + resultDeliveryStates: new Map(), deliveryFlushScheduled: false, + parentGeneration: 0, activeRuntimeToken: null, registeredRuntimeToken: null, statusWidgetClaim: null, @@ -783,9 +810,36 @@ if (sharedRuntime.activeRuntimeToken !== null && typeof sharedRuntime.activeRunt if (typeof sharedRuntime.standaloneWidgetSyncScheduled !== "boolean") { sharedRuntime.standaloneWidgetSyncScheduled = false; } +if (!(sharedRuntime.resultDeliveryStates instanceof Map)) { + sharedRuntime.resultDeliveryStates = new Map(); +} +if (!(sharedRuntime.resumingNames instanceof Set)) { + sharedRuntime.resumingNames = new Set(); +} +if (!Number.isSafeInteger(sharedRuntime.parentGeneration)) { + sharedRuntime.parentGeneration = 0; +} /** All currently running subagents, preserved across /reload. */ const runningSubagents = sharedRuntime.runningSubagents; +const resumingNames = sharedRuntime.resumingNames; + +function reserveResumeName(name: string): boolean { + if (resumingNames.has(name)) return false; + resumingNames.add(name); + return true; +} + +function clearStaleSessionSignals(sessionFile: string): void { + rmSync(`${sessionFile}.exit`, { force: true }); + rmSync(`${sessionFile}.ask`, { force: true }); +} + +for (const running of runningSubagents.values()) { + if (!Number.isSafeInteger(running.parentGeneration)) { + running.parentGeneration = sharedRuntime.parentGeneration; + } +} // When this extension is loaded inside a subagent that itself spawns children // (e.g. a worker delegating to scout/researcher), `subagent-done.ts` runs in the @@ -798,22 +852,83 @@ const RUNNING_CHILDREN_COUNT_KEY = Symbol.for("pi-subagents/running-children-cou // ── Widget and runtime delivery management ── +const MAX_RESULT_DELIVERY_TOMBSTONES = 1_024; +const AUTOMATIC_DELIVERY_OPTIONS = { triggerTurn: true, deliverAs: "followUp" as const }; + +function getReadyRuntimePi(): ExtensionAPI | null { + if ( + sharedRuntime.latestPi + && sharedRuntime.activeRuntimeToken !== null + && sharedRuntime.activeRuntimeToken === sharedRuntime.registeredRuntimeToken + ) { + return sharedRuntime.latestPi; + } + return null; +} + function sendWhenRuntimeReady(delivery: PendingDelivery): void { - const pi = sharedRuntime.latestPi; + const pi = getReadyRuntimePi(); if (pi) { - delivery(pi); - return; + try { + delivery(pi); + return; + } catch { + // Retry against the next bound runtime. A failed notification must not + // prevent sibling callbacks from being processed. + } } sharedRuntime.pendingDeliveries.push(delivery); } +function pruneResultDeliveryTombstones(): void { + if (sharedRuntime.resultDeliveryStates.size <= MAX_RESULT_DELIVERY_TOMBSTONES) return; + for (const [id, state] of sharedRuntime.resultDeliveryStates) { + if (state !== "delivered") continue; + sharedRuntime.resultDeliveryStates.delete(id); + if (sharedRuntime.resultDeliveryStates.size <= MAX_RESULT_DELIVERY_TOMBSTONES) return; + } +} + +/** Queue or deliver one terminal result for a child id, rejecting duplicates. */ +function sendResultWhenRuntimeReady(id: string, delivery: PendingDelivery): boolean { + if (sharedRuntime.resultDeliveryStates.has(id)) return false; + sharedRuntime.resultDeliveryStates.set(id, "queued"); + + const deliverOnce = (pi: ExtensionAPI) => { + if (sharedRuntime.resultDeliveryStates.get(id) === "delivered") return; + delivery(pi); + // Mark only after sendMessage succeeds so a stale runtime can be retried. + sharedRuntime.resultDeliveryStates.delete(id); + sharedRuntime.resultDeliveryStates.set(id, "delivered"); + pruneResultDeliveryTombstones(); + }; + + const pi = getReadyRuntimePi(); + if (pi) { + try { + deliverOnce(pi); + return true; + } catch { + // Keep the queued state and retry after a fresh runtime binds. + } + } + sharedRuntime.pendingDeliveries.push(deliverOnce); + return true; +} + function flushPendingDeliveries(): void { sharedRuntime.deliveryFlushScheduled = false; - const pi = sharedRuntime.latestPi; + const pi = getReadyRuntimePi(); if (!pi || sharedRuntime.pendingDeliveries.length === 0) return; const pending = sharedRuntime.pendingDeliveries.splice(0); - for (const delivery of pending) delivery(pi); + for (const delivery of pending) { + try { + delivery(pi); + } catch { + sharedRuntime.pendingDeliveries.push(delivery); + } + } } function schedulePendingDeliveryFlush(): void { @@ -1366,6 +1481,28 @@ function steerSubagent( } } +function hasTerminalCompletionSignal(running: RunningSubagent): boolean { + return !!( + running.completionObserved + || (running.completionFile && existsSync(running.completionFile)) + || existsSync(`${running.sessionFile}.exit`) + || (running.sentinelFile && existsSync(running.sentinelFile)) + ); +} + +async function waitForWatcherCleanup( + running: RunningSubagent, + signal: AbortSignal | undefined, + timeoutMs = 2_000, +): Promise { + const deadline = Date.now() + timeoutMs; + while (runningSubagents.has(running.id) && Date.now() < deadline) { + if (signal?.aborted) return false; + await new Promise((resolve) => setTimeout(resolve, 25)); + } + return !runningSubagents.has(running.id); +} + function handleSubagentSteer( params: { name?: string; message?: string }, send: (surface: string, command: string) => void = sendCommand, @@ -1466,12 +1603,26 @@ function startStatusRefresh() { // Resuming a finished session is always autonomous: the relaunched agent runs // its follow-up task to completion and the harness delivers the result as a -// steer message (fire-and-forget). An interactive resume would park the pane +// queued follow-up notification (fire-and-forget). An interactive resume would park the pane // waiting for the user, contradicting that result-delivery model. function resolveResumeLaunchBehavior(): { autoExit: boolean; interactive: boolean } { return { autoExit: true, interactive: false }; } +function shouldDeliverRunningResult(running: RunningSubagent): boolean { + return !running.suppressDelivery && running.parentGeneration === sharedRuntime.parentGeneration; +} + +function assertLaunchActive(signal: AbortSignal | undefined, runtimeToken: symbol): void { + if ( + signal?.aborted + || sharedRuntime.activeRuntimeToken !== runtimeToken + || sharedRuntime.registeredRuntimeToken !== runtimeToken + ) { + throw new Error("Parent session ended or was replaced before the subagent launch completed."); + } +} + export const __test__ = { borderLine, defaultShellReadyDelayMs, @@ -1481,8 +1632,12 @@ export const __test__ = { updateWidget, loadAgentDefaults, discoverAgentDefinitions, + resolveSubagentPaths, resolveEffectiveSessionMode, resolveLaunchBehavior, + resolveEffectiveAutoExit, + autoExitEnvAssignment, + resolveClaudeSentinelFile, resolveEffectiveInteractive, buildSubagentToolAllowlist, applySandboxToParts, @@ -1493,10 +1648,18 @@ export const __test__ = { resolveRunningByName, uniqueRunningName, reservedNames, + resumingNames, + reserveResumeName, + clearStaleSessionSignals, + hasTerminalCompletionSignal, + waitForWatcherCleanup, + deliverPendingQuestion, steerSubagent, handleSubagentSteer, resolveResultPresentation, resolveResumeLaunchBehavior, + shouldDeliverRunningResult, + assertLaunchActive, wrapWithCompletionFile, runningSubagents, formatElapsed, @@ -1506,6 +1669,8 @@ export const __test__ = { formatUsageSegments, widgetIcon, sendWhenRuntimeReady, + sendResultWhenRuntimeReady, + AUTOMATIC_DELIVERY_OPTIONS, flushPendingDeliveries, sharedRuntime, }; @@ -1534,8 +1699,9 @@ function startWidgetRefresh() { async function launchSubagent( params: Static, ctx: { sessionManager: { getSessionFile(): string | null | undefined; getSessionId(): string; getSessionDir(): string }; cwd: string }, - options?: { surface?: string }, + options?: { surface?: string; ensureActive?: () => void }, ): Promise { + options?.ensureActive?.(); const startTime = Date.now(); const id = Math.random().toString(16).slice(2, 10); const name = params.name?.trim() || params.agent; @@ -1545,6 +1711,7 @@ async function launchSubagent( const effectiveTools = agentDefs?.tools; const effectiveSkills = agentDefs?.skills; const effectiveThinking = agentDefs?.thinking; + const effectiveAutoExit = resolveEffectiveAutoExit(agentDefs); const effectiveInteractive = resolveEffectiveInteractive(params, agentDefs); const sessionFile = ctx.sessionManager.getSessionFile(); @@ -1554,7 +1721,11 @@ async function launchSubagent( const completionFile = join(artifactDir, "subagent-exits", `${id}.exit-code`); mkdirSync(dirname(completionFile), { recursive: true }); - const { effectiveCwd, localAgentDir, effectiveAgentDir } = resolveSubagentPaths(params, agentDefs); + const { effectiveCwd, localAgentDir, effectiveAgentDir } = resolveSubagentPaths( + params, + agentDefs, + ctx.cwd, + ); const targetCwdForSession = effectiveCwd ?? ctx.cwd; const sessionDir = getDefaultSessionDirFor(targetCwdForSession, effectiveAgentDir); @@ -1579,6 +1750,7 @@ async function launchSubagent( } try { + options?.ensureActive?.(); const launchBehavior = resolveLaunchBehavior(params, agentDefs); if (launchBehavior.seededSessionMode) { @@ -1597,10 +1769,10 @@ async function launchSubagent( // Build the task message // Only full-context fork mode inherits prior conversation state. // Blank-session modes need the wrapper instructions and artifact-backed handoff. - const modeHint = agentDefs?.autoExit + const modeHint = effectiveAutoExit ? "Complete your task autonomously. When you are finished, simply stop — your session ends automatically." : "Complete your task. The user can interact with you at any time, and the session ends when the user exits the pane."; - const summaryInstruction = agentDefs?.autoExit + const summaryInstruction = effectiveAutoExit ? "Your FINAL assistant message should summarize what you accomplished." : "Your FINAL assistant message (before the user exits) should summarize what you accomplished."; // An agent with a non-empty subagent_agents list is granted the spawning @@ -1615,11 +1787,13 @@ async function launchSubagent( : `${roleBlock}\n\n${modeHint}\n\n${params.task}\n\n${summaryInstruction}`; // ── Claude Code CLI path ── if (agentDefs?.cli === "claude") { - const sentinelFile = `/tmp/pi-claude-${id}-done`; + const sentinelFile = resolveClaudeSentinelFile(id, effectiveAutoExit); const pluginDir = join(SUBAGENTS_DIR, "plugin"); const cmdParts: string[] = []; - cmdParts.push(`PI_CLAUDE_SENTINEL=${shellEscape(sentinelFile)}`); + if (sentinelFile) { + cmdParts.push(`PI_CLAUDE_SENTINEL=${shellEscape(sentinelFile)}`); + } cmdParts.push("claude"); cmdParts.push("--dangerously-skip-permissions"); @@ -1651,14 +1825,17 @@ async function launchSubagent( .replace(/^-|-$/g, "") || "subagent"}-${id}.sh`; const launchScriptFile = join(artifactDir, "subagent-scripts", launchScriptName); + options?.ensureActive?.(); sendLongCommand(surface, command, { scriptPath: launchScriptFile, + replaceShell: true, scriptPreamble: [ `# Claude Code subagent launch script for ${name}`, `# Generated: ${new Date().toISOString()}`, `# Surface: ${surface}`, ].join("\n"), }); + options?.ensureActive?.(); const running: RunningSubagent = { id, @@ -1677,6 +1854,7 @@ async function launchSubagent( source: "claude", startTimeMs: startTime, }), + parentGeneration: sharedRuntime.parentGeneration, }; runningSubagents.set(id, running); @@ -1717,7 +1895,7 @@ async function launchSubagent( systemPromptMode: systemPromptMode ?? null, identity: identityInSystemPrompt ? identity : null, spawnable: agentDefs?.subagentAgents ?? null, - autoExit: agentDefs?.autoExit ?? false, + autoExit: effectiveAutoExit, cwd: effectiveCwd ?? null, agentDir: resolvedAgentDir, }; @@ -1741,9 +1919,7 @@ async function launchSubagent( if (params.agent) { envParts.push(`PI_SUBAGENT_AGENT=${shellEscape(params.agent)}`); } - if (agentDefs?.autoExit) { - envParts.push(`PI_SUBAGENT_AUTO_EXIT=1`); - } + envParts.push(autoExitEnvAssignment(effectiveAutoExit)); envParts.push(`PI_SUBAGENT_SESSION=${shellEscape(subagentSessionFile)}`); envParts.push(`PI_SUBAGENT_ID=${shellEscape(id)}`); envParts.push(`PI_SUBAGENT_ACTIVITY_FILE=${shellEscape(activityFile)}`); @@ -1793,8 +1969,10 @@ async function launchSubagent( .replace(/-+/g, "-") .replace(/^-|-$/g, "") || "subagent"}-${id}.sh`; const launchScriptFile = join(artifactDir, "subagent-scripts", launchScriptName); + options?.ensureActive?.(); sendLongCommand(surface, command, { scriptPath: launchScriptFile, + replaceShell: true, scriptPreamble: [ `# Subagent launch script for ${name}`, `# Generated: ${new Date().toISOString()}`, @@ -1802,6 +1980,7 @@ async function launchSubagent( `# Surface: ${surface}`, ].join("\n"), }); + options?.ensureActive?.(); const running: RunningSubagent = { id, @@ -1819,6 +1998,7 @@ async function launchSubagent( source: "pi", startTimeMs: startTime, }), + parentGeneration: sharedRuntime.parentGeneration, }; runningSubagents.set(id, running); @@ -1870,13 +2050,15 @@ function copyClaudeSession(sentinelFile: string): string | null { * delivery so it fires once per question (a subagent may ask again later). */ function deliverPendingQuestion(running: RunningSubagent): void { + if (!shouldDeliverRunningResult(running)) return; const askFile = `${running.sessionFile}.ask`; let payload: any = null; try { if (!existsSync(askFile)) return; payload = JSON.parse(readFileSync(askFile, "utf-8")); } catch { - // Malformed/partway-written file — drop it and move on. + // A writer may still be replacing the signal. Leave it for the next tick. + return; } try { unlinkSync(askFile); @@ -1889,6 +2071,7 @@ function deliverPendingQuestion(running: RunningSubagent): void { const replyHint = `\n\nReply with subagent_message({ name: "${name}", message: "…" }) — the same name works whether it is still running or has since exited. It stays open until you reply.`; sendWhenRuntimeReady((pi) => { + if (!shouldDeliverRunningResult(running)) return; pi.sendMessage( { customType: "subagent_question", @@ -1901,7 +2084,7 @@ function deliverPendingQuestion(running: RunningSubagent): void { ...(sessionId ? { sessionId } : {}), }, }, - { triggerTurn: true, deliverAs: "steer" }, + AUTOMATIC_DELIVERY_OPTIONS, ); }); } @@ -1922,6 +2105,9 @@ async function watchSubagent( observeRunningSubagent(running); deliverPendingQuestion(running); }, + onExitDetected() { + running.completionObserved = true; + }, }); const elapsed = Math.floor((Date.now() - startTime) / 1000); @@ -2026,6 +2212,19 @@ async function watchSubagent( } } +function invalidateParentLifecycle(): void { + sharedRuntime.pendingDeliveries.length = 0; + sharedRuntime.resultDeliveryStates.clear(); + sharedRuntime.parentGeneration += 1; + resumingNames.clear(); + for (const agent of runningSubagents.values()) { + agent.suppressDelivery = true; + agent.abortController?.abort(); + } + runningSubagents.clear(); + clearRefreshIntervals(); +} + export default function subagentsExtension(pi: ExtensionAPI) { const runtimeToken = Symbol("pi-subagents/runtime"); // The newest registered extension instance is the only one allowed to bind @@ -2036,9 +2235,18 @@ export default function subagentsExtension(pi: ExtensionAPI) { // Bind only after Pi has installed the fresh runtime. During /reload the old // session_shutdown handler clears these references first, so watcher results // cannot enqueue a turn through stale extension APIs. - pi.on("session_start", (_event, ctx) => { + pi.on("session_start", (event, ctx) => { if (sharedRuntime.registeredRuntimeToken !== runtimeToken) return; + // Session replacement can start before the old runtime's shutdown event. + // Invalidate the old parent's children and queued callbacks at the start + // boundary too; reload is the only transition that preserves them. + if (event.reason === "new" || event.reason === "resume" || event.reason === "fork") { + clearStandaloneWidget(); + sharedRuntime.statusWidgetClaim = null; + invalidateParentLifecycle(); + } + // A fresh runtime may start before the old runtime emits shutdown. Drop // old UI ownership and intervals before binding the fresh context. if (sharedRuntime.activeRuntimeToken !== runtimeToken) { @@ -2085,12 +2293,7 @@ export default function subagentsExtension(pi: ExtensionAPI) { // Real session replacement/quit owns a different lifecycle. Cancel those // children and suppress the synthetic cancellation result; there is no // valid destination session to receive it. - sharedRuntime.pendingDeliveries.length = 0; - for (const agent of runningSubagents.values()) { - agent.suppressDelivery = true; - agent.abortController?.abort(); - } - runningSubagents.clear(); + invalidateParentLifecycle(); }); // The spawning tools are always registered here. Whether a child process can @@ -2105,14 +2308,14 @@ export default function subagentsExtension(pi: ExtensionAPI) { description: "Spawn a sub-agent in a dedicated terminal multiplexer pane. " + "This is a fire-and-forget async tool: the call returns immediately with only an acknowledgement. " + - "When the sub-agent finishes, the harness AUTOMATICALLY delivers its result as a steer message that wakes you up and starts a new turn — you do not need to do anything to receive it. " + + "When the sub-agent finishes, the harness AUTOMATICALLY delivers its result as a follow-up notification that wakes you up and starts a new turn — you do not need to do anything to receive it. " + "DO NOT write polling loops, sleep/wait commands, tail/watch scripts, or repeatedly read session/log files to detect completion. DO NOT call subagents_list or any other tool to 'check' status. All of that is wasted work — the harness handles delivery for you. " + "DO NOT fabricate, assume, or summarize results after calling this tool. " + "After spawning, either end your turn immediately, or work on other independent tasks (including spawning more subagents in parallel). The harness will wake you with the result when it is ready.", promptSnippet: "Spawn a sub-agent in a dedicated terminal multiplexer pane. " + "This is a fire-and-forget async tool: the call returns immediately with only an acknowledgement. " + - "When the sub-agent finishes, the harness AUTOMATICALLY delivers its result as a steer message that wakes you up and starts a new turn — you do not need to do anything to receive it. " + + "When the sub-agent finishes, the harness AUTOMATICALLY delivers its result as a follow-up notification that wakes you up and starts a new turn — you do not need to do anything to receive it. " + "DO NOT write polling loops, sleep/wait commands, tail/watch scripts, or repeatedly read session/log files to detect completion. DO NOT call subagents_list or any other tool to 'check' status. All of that is wasted work — the harness handles delivery for you. " + "DO NOT fabricate, assume, or summarize results after calling this tool. " + "After spawning, either end your turn immediately, or work on other independent tasks (including spawning more subagents in parallel). The harness will wake you with the result when it is ready.", @@ -2201,27 +2404,33 @@ export default function subagentsExtension(pi: ExtensionAPI) { ctx.sessionManager.getSessionId(), ); - // Default the cosmetic pane label to the agent name when omitted, - // disambiguating against running subagents, in-flight reservations, and - // every name already in the registry — so names stay unique across the - // whole session, running or finished. Reserve the chosen name - // synchronously (before any await) so parallel spawns don't collide. - let reservedName: string | null = null; - if (!params.name?.trim()) { - const registryNames = new Set(Object.keys(readNameRegistry(parentArtifactDir))); - params.name = uniqueRunningName(params.agent, registryNames); - reservedName = params.name; - reservedNames.add(reservedName); - } + const ensureActive = () => assertLaunchActive(_signal, runtimeToken); + ensureActive(); + + // Disambiguate both defaulted and explicit cosmetic names against + // running children, parallel launch reservations, and the persistent + // registry. Reserve synchronously before the first await. + const registryNames = new Set(Object.keys(readNameRegistry(parentArtifactDir))); + params.name = uniqueRunningName(params.name?.trim() || params.agent, registryNames); + const reservedName = params.name; + reservedNames.add(reservedName); // Launch the subagent (creates pane, sends command). Release the name // reservation once it registers in runningSubagents (or launch fails) — // from then on uniqueRunningName tracks it via the running map. - let running; + let running: RunningSubagent | undefined; try { - running = await launchSubagent(params, ctx); + running = await launchSubagent(params, ctx, { ensureActive }); + ensureActive(); + } catch (error) { + if (running) { + running.suppressDelivery = true; + runningSubagents.delete(running.id); + try { closeSurface(running.surface); } catch {} + } + throw error; } finally { - if (reservedName) reservedNames.delete(reservedName); + reservedNames.delete(reservedName); } // Persist name → session so subagent_message({ name }) can resume this @@ -2247,10 +2456,10 @@ export default function subagentsExtension(pi: ExtensionAPI) { watchSubagent(running, watcherAbort.signal) .then((result) => { updateWidget(); // reflect removal from Map immediately - if (running.suppressDelivery) return; + if (!shouldDeliverRunningResult(running)) return; const presentation = resolveResultPresentation(result, running.name); - sendWhenRuntimeReady((currentPi) => { + sendResultWhenRuntimeReady(running.id, (currentPi) => { currentPi.sendMessage( { customType: "subagent_result", @@ -2269,14 +2478,14 @@ export default function subagentsExtension(pi: ExtensionAPI) { ...(result.stats ? { stats: result.stats } : {}), }, }, - { triggerTurn: true, deliverAs: "steer" }, + AUTOMATIC_DELIVERY_OPTIONS, ); }); }) .catch((err) => { updateWidget(); - if (running.suppressDelivery) return; - sendWhenRuntimeReady((currentPi) => { + if (!shouldDeliverRunningResult(running)) return; + sendResultWhenRuntimeReady(running.id, (currentPi) => { currentPi.sendMessage( { customType: "subagent_result", @@ -2284,7 +2493,7 @@ export default function subagentsExtension(pi: ExtensionAPI) { display: true, details: { name: running.name, task: running.task, error: err?.message }, }, - { triggerTurn: true, deliverAs: "steer" }, + AUTOMATIC_DELIVERY_OPTIONS, ); }); }); @@ -2297,7 +2506,7 @@ export default function subagentsExtension(pi: ExtensionAPI) { text: `Sub-agent "${params.name}" launched and is now running in the background. ` + `Do NOT generate or assume any results — you have no idea what the sub-agent will do or produce. ` + - `The results will be delivered to you automatically as a steer message when the sub-agent finishes. ` + + `The results will be delivered to you automatically as a follow-up notification when the sub-agent finishes. ` + `Until then, move on to other work or tell the user you're waiting.`, }, ], @@ -2440,12 +2649,12 @@ export default function subagentsExtension(pi: ExtensionAPI) { "if it has finished, your message resumes that session and continues it. " + "`name` and `message` are both required. " + "Steering a running subagent returns immediately with a local acknowledgement and does NOT, by itself, emit a new result. " + - "Resuming is a fire-and-forget async call: when the resumed sub-agent finishes, the harness AUTOMATICALLY delivers its result as a steer message that wakes you up. " + + "Resuming is a fire-and-forget async call: when the resumed sub-agent finishes, the harness AUTOMATICALLY delivers its result as a follow-up notification that wakes you up. " + "DO NOT poll, sleep, tail logs, or read session files to detect completion — the harness handles delivery. " + "DO NOT fabricate or assume results. After calling, either end your turn or work on other independent tasks.", promptSnippet: "Message a subagent by name: steers it if running, resumes it if finished (same name either way). " + - "`name` and `message` are required. Steering returns immediately; resuming delivers its result later as a steer message. " + + "`name` and `message` are required. Steering returns immediately; resuming delivers its result later as a follow-up notification. " + "Do not poll or fabricate results.", parameters: Type.Object({ name: Type.String({ @@ -2509,11 +2718,37 @@ export default function subagentsExtension(pi: ExtensionAPI) { return muxUnavailableResult(); } + const ensureActive = () => assertLaunchActive(_signal, runtimeToken); + ensureActive(); + // ── Steer a running subagent ── - // A name that matches a currently-running subagent always steers it. - const runningMatch = Array.from(runningSubagents.values()).find((r) => r.name === requestedName); + // A completion signal can arrive just before the watcher removes the + // running entry. Let that cleanup settle so a late message resumes the + // finished session instead of being typed into a dead pane's shell. + let runningMatch = Array.from(runningSubagents.values()).find((r) => r.name === requestedName); + if (runningMatch && hasTerminalCompletionSignal(runningMatch)) { + await waitForWatcherCleanup(runningMatch, _signal); + ensureActive(); + runningMatch = Array.from(runningSubagents.values()).find((r) => r.name === requestedName); + } if (runningMatch) { - return handleSubagentSteer({ name: requestedName, message: params.message }); + if (hasTerminalCompletionSignal(runningMatch)) { + const err = `Subagent "${requestedName}" is finishing. Retry the message after its completion result arrives.`; + return { content: [{ type: "text" as const, text: err }], details: { error: err } }; + } + + const steerResult = handleSubagentSteer({ name: requestedName, message: params.message }); + if (!steerResult.details.error || await isSurfaceAlive(runningMatch.surface, _signal)) { + return steerResult; + } + + // The child exited between the completion check and terminal send. + // Wait for its watcher to release the session, then fall through to + // the normal finished-session resume path. + await waitForWatcherCleanup(runningMatch, _signal); + ensureActive(); + runningMatch = Array.from(runningSubagents.values()).find((r) => r.name === requestedName); + if (runningMatch) return steerResult; } // ── Resume a finished session by name ── @@ -2569,6 +2804,12 @@ export default function subagentsExtension(pi: ExtensionAPI) { return { content: [{ type: "text" as const, text: err }], details: { error: err } }; } + if (!reserveResumeName(name)) { + const err = `Subagent "${name}" is already being resumed. Wait for its launch acknowledgement before messaging it again.`; + return { content: [{ type: "text" as const, text: err }], details: { error: err } }; + } + + clearStaleSessionSignals(sessionPath); const resumedSessionId = entry.sessionId ?? getSessionId(sessionPath) ?? requestedName; // Record entry count before resuming so we can extract new messages. @@ -2576,10 +2817,11 @@ export default function subagentsExtension(pi: ExtensionAPI) { // transcript doesn't block the UI. const entryCountBefore = countSessionEntryLines(sessionPath); - const surface = createSurface(name); - await new Promise((resolve) => setTimeout(resolve, getShellReadyDelayMs())); - + let surface: string | undefined; try { + surface = createSurface(name); + await new Promise((resolve) => setTimeout(resolve, getShellReadyDelayMs())); + ensureActive(); // Build pi resume command const parts = ["pi", "--session", shellEscape(sessionPath)]; @@ -2656,8 +2898,10 @@ export default function subagentsExtension(pi: ExtensionAPI) { .replace(/-+/g, "-") .replace(/^-|-$/g, "") || "resume"}-resume-${Date.now()}.sh`, ); + ensureActive(); sendLongCommand(surface, command, { scriptPath: launchScriptFile, + replaceShell: true, scriptPreamble: [ `# Subagent resume script for ${name}`, `# Generated: ${new Date().toISOString()}`, @@ -2666,6 +2910,7 @@ export default function subagentsExtension(pi: ExtensionAPI) { ...(resumeMsgFile ? [`# Resume message file: ${resumeMsgFile}`] : []), ].join("\n"), }); + ensureActive(); // Register as a running subagent for widget tracking const running: RunningSubagent = { @@ -2683,6 +2928,7 @@ export default function subagentsExtension(pi: ExtensionAPI) { source: "pi", startTimeMs: startTime, }), + parentGeneration: sharedRuntime.parentGeneration, }; runningSubagents.set(id, running); startWidgetRefresh(); @@ -2695,7 +2941,7 @@ export default function subagentsExtension(pi: ExtensionAPI) { watchSubagent(running, watcherAbort.signal) .then((result) => { updateWidget(); - if (running.suppressDelivery) return; + if (!shouldDeliverRunningResult(running)) return; const allEntries = getNewEntries(sessionPath, entryCountBefore); const summary = findLastAssistantMessage(allEntries) ?? @@ -2709,7 +2955,7 @@ export default function subagentsExtension(pi: ExtensionAPI) { name, ); - sendWhenRuntimeReady((currentPi) => { + sendResultWhenRuntimeReady(running.id, (currentPi) => { currentPi.sendMessage( { customType: "subagent_result", @@ -2725,14 +2971,14 @@ export default function subagentsExtension(pi: ExtensionAPI) { ...(result.errorMessage ? { errorMessage: result.errorMessage } : {}), }, }, - { triggerTurn: true, deliverAs: "steer" }, + AUTOMATIC_DELIVERY_OPTIONS, ); }); }) .catch((err) => { updateWidget(); - if (running.suppressDelivery) return; - sendWhenRuntimeReady((currentPi) => { + if (!shouldDeliverRunningResult(running)) return; + sendResultWhenRuntimeReady(running.id, (currentPi) => { currentPi.sendMessage( { customType: "subagent_result", @@ -2740,7 +2986,7 @@ export default function subagentsExtension(pi: ExtensionAPI) { display: true, details: { name, error: err?.message }, }, - { triggerTurn: true, deliverAs: "steer" }, + AUTOMATIC_DELIVERY_OPTIONS, ); }); }); @@ -2758,12 +3004,16 @@ export default function subagentsExtension(pi: ExtensionAPI) { }; } catch (error) { runningSubagents.delete(id); - try { - closeSurface(surface); - } catch { - // Preserve the resume error; cleanup is best-effort. + if (surface) { + try { + closeSurface(surface); + } catch { + // Preserve the resume error; cleanup is best-effort. + } } throw error; + } finally { + resumingNames.delete(name); } }, }); diff --git a/extensions/interactive-subagents/pi-extension/subagents/orca.ts b/extensions/interactive-subagents/pi-extension/subagents/orca.ts index 338631f..e1aa31e 100644 --- a/extensions/interactive-subagents/pi-extension/subagents/orca.ts +++ b/extensions/interactive-subagents/pi-extension/subagents/orca.ts @@ -281,11 +281,13 @@ async function readScreenWithCli( surface: string, lines: number, timeout = ASYNC_READ_SCREEN_TIMEOUT_MS, + signal?: AbortSignal, ): Promise { const { stdout } = await execFileAsync(cli, readScreenArgs(surface, lines), { encoding: "utf8", timeout, killSignal: "SIGKILL", + signal, }); return parseReadTail(stdout); } @@ -294,6 +296,15 @@ export async function readScreenAsync(surface: string, lines = 50): Promise { + try { + await readScreenWithCli(requireOrca(), surface, 1, ASYNC_READ_SCREEN_TIMEOUT_MS, signal); + return true; + } catch { + return false; + } +} + export const __orcaTest__ = { readScreenWithCli }; export function closeSurface(surface: string): void { @@ -301,7 +312,11 @@ export function closeSurface(surface: string): void { const args = ["terminal", "close", "--terminal", surface]; if (tabByHandle.has(surface)) args.push("--tab"); args.push("--json"); - const output = execFileSync(cli, args, { encoding: "utf8" }); - parseResponse("terminal close", output); + try { + const output = execFileSync(cli, args, { encoding: "utf8" }); + parseResponse("terminal close", output); + } catch { + // Replaced-shell launches may close their own terminal before cleanup. + } tabByHandle.delete(surface); } diff --git a/extensions/interactive-subagents/pi-extension/subagents/subagent-done.ts b/extensions/interactive-subagents/pi-extension/subagents/subagent-done.ts index 0d4605a..2652c88 100644 --- a/extensions/interactive-subagents/pi-extension/subagents/subagent-done.ts +++ b/extensions/interactive-subagents/pi-extension/subagents/subagent-done.ts @@ -15,7 +15,7 @@ import type { ExtensionAPI } from "@earendil-works/pi-coding-agent"; import { Box, Text } from "@earendil-works/pi-tui"; import { Type } from "typebox"; -import { writeFileSync } from "node:fs"; +import { renameSync, writeFileSync } from "node:fs"; import { createSubagentActivityRecorder } from "./activity.ts"; export function shouldMarkUserTookOver(agentStarted: boolean): boolean { @@ -373,7 +373,10 @@ export default function (pi: ExtensionAPI) { agent: process.env.PI_SUBAGENT_AGENT ?? "", question: params.question, }; - writeFileSync(`${sessionFile}.ask`, JSON.stringify(askData)); + const askFile = `${sessionFile}.ask`; + const tempAskFile = `${askFile}.tmp-${process.pid}-${Math.random().toString(16).slice(2, 8)}`; + writeFileSync(tempAskFile, JSON.stringify(askData)); + renameSync(tempAskFile, askFile); return { content: [ diff --git a/extensions/interactive-subagents/pi-extension/subagents/surface.ts b/extensions/interactive-subagents/pi-extension/subagents/surface.ts index c8c2d85..0f08121 100644 --- a/extensions/interactive-subagents/pi-extension/subagents/surface.ts +++ b/extensions/interactive-subagents/pi-extension/subagents/surface.ts @@ -22,6 +22,7 @@ type SurfaceBackend = { sendCommand(surface: string, command: string): void; readScreen(surface: string, lines?: number): string; readScreenAsync(surface: string, lines?: number): Promise; + isSurfaceAlive(surface: string, signal?: AbortSignal): Promise; closeSurface(surface: string): void; }; @@ -87,6 +88,10 @@ export function readScreenAsync(surface: string, lines = 50): Promise { return getBackend().readScreenAsync(surface, lines); } +export function isSurfaceAlive(surface: string, signal?: AbortSignal): Promise { + return getBackend().isSurfaceAlive(surface, signal); +} + export function closeSurface(surface: string): void { getBackend().closeSurface(surface); } @@ -106,7 +111,7 @@ export function wrapWithCompletionFile(command: string, completionFile: string): export function sendLongCommand( surface: string, command: string, - options?: { scriptPath?: string; scriptPreamble?: string }, + options?: { scriptPath?: string; scriptPreamble?: string; replaceShell?: boolean }, ): string { const scriptPath = options?.scriptPath ?? @@ -122,7 +127,8 @@ export function sendLongCommand( scriptParts.push(command); writeFileSync(scriptPath, scriptParts.join("\n") + "\n", { mode: 0o755 }); - sendCommand(surface, `bash ${shellEscape(scriptPath)}`); + const launch = `${options?.replaceShell ? "exec " : ""}bash ${shellEscape(scriptPath)}`; + sendCommand(surface, launch); return scriptPath; } @@ -159,6 +165,10 @@ type PollOptions = { sentinelFile?: string; completionFile?: string; onTick?: (elapsed: number) => void; + onExitDetected?: () => void; + livenessCheck?: (surface: string, signal?: AbortSignal) => Promise; + livenessIntervalMs?: number; + livenessFailureThreshold?: number; }; /** Avoid launching a terminal CLI reader when a local completion signal is authoritative. */ @@ -177,6 +187,11 @@ export async function pollForExit( readSurface: (surface: string, lines?: number) => Promise = readScreenAsync, ): Promise { const start = Date.now(); + const livenessCheck = options.livenessCheck ?? isSurfaceAlive; + const livenessIntervalMs = options.livenessIntervalMs ?? 5_000; + const livenessFailureThreshold = options.livenessFailureThreshold ?? 3; + let nextLivenessCheck = start + livenessIntervalMs; + let livenessFailures = 0; for (;;) { if (signal.aborted) throw new Error("Aborted while waiting for subagent to finish"); @@ -186,6 +201,7 @@ export async function pollForExit( const exitFile = `${options.sessionFile}.exit`; if (existsSync(exitFile)) { const result = interpretExitSidecar(JSON.parse(readFileSync(exitFile, "utf8"))); + options.onExitDetected?.(); rmSync(exitFile, { force: true }); return result; } @@ -195,6 +211,7 @@ export async function pollForExit( if (options.sentinelFile) { try { if (existsSync(options.sentinelFile)) { + options.onExitDetected?.(); return { reason: "sentinel", exitCode: 0 }; } } catch {} @@ -203,7 +220,10 @@ export async function pollForExit( if (options.completionFile) { try { const result = readCompletionFile(options.completionFile); - if (result) return result; + if (result) { + options.onExitDetected?.(); + return result; + } } catch {} } @@ -211,10 +231,24 @@ export async function pollForExit( try { const screen = await readSurface(surface, 5); const match = screen.match(/__SUBAGENT_DONE_(\d+)__/); - if (match) return { reason: "sentinel", exitCode: Number.parseInt(match[1], 10) }; + if (match) { + options.onExitDetected?.(); + return { reason: "sentinel", exitCode: Number.parseInt(match[1], 10) }; + } } catch { // Surface reads are a best-effort fallback when no completion file is available. } + } else if (Date.now() >= nextLivenessCheck) { + nextLivenessCheck = Date.now() + livenessIntervalMs; + try { + livenessFailures = await livenessCheck(surface, signal) ? 0 : livenessFailures + 1; + } catch { + livenessFailures += 1; + } + if (signal.aborted) throw new Error("Aborted while checking subagent surface liveness"); + if (livenessFailures >= livenessFailureThreshold) { + throw new Error("Subagent terminal surface closed before completion."); + } } const elapsed = Math.floor((Date.now() - start) / 1000); diff --git a/extensions/interactive-subagents/pi-extension/subagents/tmux.ts b/extensions/interactive-subagents/pi-extension/subagents/tmux.ts index e0e0f51..5af95eb 100644 --- a/extensions/interactive-subagents/pi-extension/subagents/tmux.ts +++ b/extensions/interactive-subagents/pi-extension/subagents/tmux.ts @@ -112,8 +112,27 @@ export async function readScreenAsync(surface: string, lines = 50): Promise { + requireTmux(); + try { + await execFileAsync("tmux", ["display-message", "-p", "-t", surface, "#{pane_id}"], { + encoding: "utf8", + timeout: 5_000, + killSignal: "SIGKILL", + signal, + }); + return true; + } catch { + return false; + } +} + export function closeSurface(surface: string): void { requireTmux(); - execFileSync("tmux", ["kill-pane", "-t", surface], { encoding: "utf8" }); + try { + execFileSync("tmux", ["kill-pane", "-t", surface], { encoding: "utf8" }); + } catch { + // Replaced-shell launches may close their own pane before cleanup. + } rebalanceSurfaces(); } diff --git a/extensions/interactive-subagents/test/integration/agents/test-default-autonomous.md b/extensions/interactive-subagents/test/integration/agents/test-default-autonomous.md new file mode 100644 index 0000000..79b757a --- /dev/null +++ b/extensions/interactive-subagents/test/integration/agents/test-default-autonomous.md @@ -0,0 +1,9 @@ +--- +name: test-default-autonomous +description: Integration test agent with omitted auto-exit +model: anthropic/claude-haiku-4-5 +tools: bash +disable-model-invocation: true +--- + +You are a test agent. Run the requested bash command immediately, then report completion concisely. Do not ask questions or do anything else. diff --git a/extensions/interactive-subagents/test/integration/subagent-lifecycle.test.ts b/extensions/interactive-subagents/test/integration/subagent-lifecycle.test.ts index 7b6dcea..5a22c4c 100644 --- a/extensions/interactive-subagents/test/integration/subagent-lifecycle.test.ts +++ b/extensions/interactive-subagents/test/integration/subagent-lifecycle.test.ts @@ -18,6 +18,7 @@ import { describe, it, before, after } from "node:test"; import assert from "node:assert/strict"; import { existsSync, readFileSync, unlinkSync } from "node:fs"; +import { join } from "node:path"; import { getAvailableBackends, createTestEnv, @@ -31,6 +32,7 @@ import { trackTempFile, readScreen, sendCommand, + shellEscape, PI_TIMEOUT, type TestEnv, } from "./harness.ts"; @@ -202,42 +204,115 @@ for (const backend of backends) { // ── Parallel subagent spawn ── - it("spawns two subagents in parallel and both complete", async () => { + it("delivers two staggered sibling launches exactly once after the parent becomes idle", async () => { const id = uniqueId(); + const nameA = `ParaA-${id}`; + const nameB = `ParaB-${id}`; const fileA = `/tmp/pi-integ-para-${id}-a.txt`; const fileB = `/tmp/pi-integ-para-${id}-b.txt`; + const parentWorkFile = `/tmp/pi-integ-para-${id}-parent.txt`; + const parentSession = join(env.dir, `parallel-parent-${id}.jsonl`); trackTempFile(env, fileA); trackTempFile(env, fileB); + trackTempFile(env, parentWorkFile); const surface = createTrackedSurface(env, `parallel-${id}`); await sleep(1000); const task = [ - `You must call the subagent tool TWICE. Make both calls before waiting for results.`, + `Call the subagent tool twice in ONE assistant response so Pi executes them as parallel sibling tool calls.`, + `Use these exact parameters:`, ``, `First call:`, - ` name: "ParaA-${id}"`, - ` agent: "test-echo"`, - ` task: "Run: echo 'DONE_A_${id}' > '${fileA}'"`, + ` name: "${nameA}"`, + ` agent: "test-default-autonomous"`, + ` cwd: "${env.dir}"`, + ` task: "Run: sleep 6; echo 'DONE_A_${id}' > '${fileA}'"`, ``, `Second call:`, - ` name: "ParaB-${id}"`, - ` agent: "test-echo"`, - ` task: "Run: echo 'DONE_B_${id}' > '${fileB}'"`, + ` name: "${nameB}"`, + ` agent: "test-default-autonomous"`, + ` cwd: "${env.dir}"`, + ` task: "Run: sleep 12; echo 'DONE_B_${id}' > '${fileB}'"`, ``, - `Call both subagent tools NOW, do not wait between them.`, + `After both launch acknowledgements, use bash yourself to write PARENT_WORK_${id} to '${parentWorkFile}', then say you are waiting and stop.`, + `Never call subagent_message. When the first result arrives, acknowledge it and stop while the other child runs.`, + `When both results have arrived, say PARALLEL_IDLE_COMPLETE_${id} exactly once.`, ].join("\n"); - startPi(surface, env.dir, task); + startPi(surface, env.dir, task, { + extraArgs: `--session ${shellEscape(parentSession)}`, + }); - // Both marker files should appear - const [contentA, contentB] = await Promise.all([ - waitForFile(fileA, PI_TIMEOUT, /DONE_A/), - waitForFile(fileB, PI_TIMEOUT, /DONE_B/), - ]); + await waitForFile(parentWorkFile, PI_TIMEOUT, /PARENT_WORK_/); + const screen = await waitForScreen( + surface, + new RegExp(`^ PARALLEL_IDLE_COMPLETE_${id}\\s*$`, "m"), + PI_TIMEOUT, + 400, + ); + assert.equal( + screen.match(new RegExp(`^ PARALLEL_IDLE_COMPLETE_${id}\\s*$`, "gm"))?.length, + 1, + ); + assert.match(await waitForFile(fileA, PI_TIMEOUT, /DONE_A_/), new RegExp(`DONE_A_${id}`)); + assert.match(await waitForFile(fileB, PI_TIMEOUT, /DONE_B_/), new RegExp(`DONE_B_${id}`)); - assert.ok(contentA.includes(`DONE_A_${id}`), `File A should contain marker`); - assert.ok(contentB.includes(`DONE_B_${id}`), `File B should contain marker`); + const entries = readFileSync(parentSession, "utf8") + .trim() + .split("\n") + .map((line) => JSON.parse(line)); + const siblingCall = entries.find((entry) => { + const calls = entry.message?.role === "assistant" + ? entry.message.content?.filter((part: any) => part.type === "toolCall" && part.name === "subagent") + : []; + return calls?.length === 2; + }); + assert.ok(siblingCall, "both wrapped launches should reach Pi as sibling subagent calls"); + + const calls = siblingCall.message.content.filter( + (part: any) => part.type === "toolCall" && part.name === "subagent", + ); + assert.deepEqual(calls.map((call: any) => call.arguments.name), [nameA, nameB]); + assert.deepEqual(calls.map((call: any) => call.arguments.cwd), [env.dir, env.dir]); + + const acknowledgements = entries.filter( + (entry) => entry.message?.role === "toolResult" && entry.message.toolName === "subagent", + ); + assert.deepEqual( + acknowledgements.map((entry) => entry.message.details?.name).sort(), + [nameA, nameB].sort(), + ); + for (const acknowledgement of acknowledgements) { + assert.equal(acknowledgement.message.details?.status, "started"); + const childHeader = JSON.parse( + readFileSync(acknowledgement.message.details.sessionFile, "utf8").split("\n")[0], + ); + assert.equal(childHeader.cwd, env.dir); + } + + const completions = entries.filter( + (entry) => entry.type === "custom_message" && entry.customType === "subagent_result", + ); + assert.deepEqual(completions.map((entry) => entry.details?.name).sort(), [nameA, nameB].sort()); + assert.equal(completions.length, 2, "each child result should be delivered exactly once"); + assert.ok(completions.every((entry) => entry.details?.exitCode === 0)); + + const idleBeforeCompletion = entries.findIndex((entry, index) => { + if (index <= entries.indexOf(acknowledgements.at(-1))) return false; + if (entry.message?.role !== "assistant" || entry.message.stopReason !== "stop") return false; + return completions.every((completion) => Date.parse(completion.timestamp) > Date.parse(entry.timestamp)); + }); + assert.notEqual(idleBeforeCompletion, -1, "the parent should settle before either child completes"); + for (const completion of completions) { + assert.ok( + entries.some( + (entry) => entry.message?.role === "assistant" + && Date.parse(entry.timestamp) > Date.parse(completion.timestamp), + ), + `${completion.details?.name} should trigger a parent turn while idle`, + ); + } }); // ── Fork mode ── diff --git a/extensions/interactive-subagents/test/test.ts b/extensions/interactive-subagents/test/test.ts index b6f8907..1bbf55b 100644 --- a/extensions/interactive-subagents/test/test.ts +++ b/extensions/interactive-subagents/test/test.ts @@ -158,13 +158,16 @@ async function emitMockEvent( type SharedRuntimeTestState = { runningSubagents: Map; + resumingNames: Set; latestCtx: unknown; latestPi: unknown; widgetInterval: ReturnType | null; statusInterval: ReturnType | null; standaloneWidgetSyncScheduled: boolean; pendingDeliveries: { length: number }; + resultDeliveryStates: Map; deliveryFlushScheduled: boolean; + parentGeneration: number; activeRuntimeToken: symbol | null; registeredRuntimeToken: symbol | null; statusWidgetClaim: unknown; @@ -178,11 +181,14 @@ function resetSharedRuntimeForTest(shared: SharedRuntimeTestState): void { shared.statusInterval = null; shared.standaloneWidgetSyncScheduled = false; shared.runningSubagents.clear(); + shared.resumingNames.clear(); shared.latestCtx = null; shared.latestPi = null; shared.statusWidgetClaim = null; shared.pendingDeliveries.length = 0; + shared.resultDeliveryStates.clear(); shared.deliveryFlushScheduled = false; + shared.parentGeneration = 0; shared.activeRuntimeToken = null; shared.registeredRuntimeToken = null; } @@ -1201,28 +1207,53 @@ describe("subagent discovery", () => { }); }); - it("resolveEffectiveInteractive defaults to the inverse of auto-exit", () => { - // Autonomous agents (auto-exit: true) are NOT interactive — parent gets stall pings. + it("defaults unspecified launch behavior to autonomous while preserving explicit interactive profiles", () => { + assert.equal(testApi.resolveEffectiveAutoExit({ autoExit: true }), true); + assert.equal(testApi.resolveEffectiveAutoExit({ autoExit: false }), false); + assert.equal(testApi.resolveEffectiveAutoExit({}), true); + assert.equal(testApi.resolveEffectiveAutoExit(null), true); + assert.equal(testApi.autoExitEnvAssignment(true), "PI_SUBAGENT_AUTO_EXIT=1"); + assert.equal(testApi.autoExitEnvAssignment(false), "PI_SUBAGENT_AUTO_EXIT=0"); + assert.match(testApi.resolveClaudeSentinelFile("child", true), /pi-claude-child-done$/); + assert.equal(testApi.resolveClaudeSentinelFile("child", false), undefined); + + // Autonomous agents are not user-driven, so parent status transitions stay enabled. assert.equal( testApi.resolveEffectiveInteractive({ name: "A", task: "T" }, { autoExit: true }), false, ); - // Agents without auto-exit ARE interactive — parent does not receive status transition pings. + // Explicitly non-auto-exit profiles remain interactive. assert.equal( testApi.resolveEffectiveInteractive({ name: "A", task: "T" }, { autoExit: false }), true, ); + // Missing auto-exit follows the fire-and-forget tool contract. assert.equal( testApi.resolveEffectiveInteractive({ name: "A", task: "T" }, {}), - true, + false, ); - // Bare spawn with no agent defs (e.g. /iterate fork) is interactive by default. assert.equal( testApi.resolveEffectiveInteractive({ name: "A", task: "T" }, null), - true, + false, ); }); + it("resolves relative launch cwd from the parent session rather than the process", () => { + const resolved = testApi.resolveSubagentPaths( + { agent: "worker", task: "T", cwd: "roles/reviewer" }, + {}, + "/workspace/current-session", + ); + assert.equal(resolved.effectiveCwd, "/workspace/current-session/roles/reviewer"); + + const inherited = testApi.resolveSubagentPaths( + { agent: "worker", task: "T" }, + {}, + "/workspace/current-session", + ); + assert.equal(inherited.effectiveCwd, null); + }); + it("resolveEffectiveInteractive honors explicit frontmatter over the auto-exit default", () => { // Autonomous agent that still wants to be treated as interactive. assert.equal( @@ -1851,6 +1882,42 @@ describe("surface exit signals", () => { } }); + it("terminates after repeated liveness failures when a completion-file surface is closed", async () => { + await assert.rejects( + pollForExit( + "closed-surface", + new AbortController().signal, + { + interval: 1, + completionFile: "/tmp/nonexistent-subagent-completion", + livenessCheck: async () => false, + livenessIntervalMs: 0, + livenessFailureThreshold: 2, + }, + ), + /surface closed before completion/i, + ); + }); + + it("aborts an in-flight completion-file liveness probe", async () => { + const controller = new AbortController(); + const polling = pollForExit( + "blocked-surface", + controller.signal, + { + interval: 1, + completionFile: "/tmp/nonexistent-subagent-completion", + livenessIntervalMs: 0, + livenessCheck: async (_surface: string, signal?: AbortSignal) => + await new Promise((_resolve, reject) => { + signal?.addEventListener("abort", () => reject(new Error("probe aborted")), { once: true }); + }), + }, + ); + setTimeout(() => controller.abort(), 10); + await assert.rejects(polling, /aborted/i); + }); + it("bounds an asynchronous Orca screen read and kills its CLI process", async () => { const dir = createTestDir(); const pidFile = join(dir, "reader.pid"); @@ -1966,6 +2033,182 @@ describe("reload lifecycle", () => { beforeEach(() => resetSharedRuntimeForTest(shared)); after(() => resetSharedRuntimeForTest(shared)); + it("delivers parallel terminal results exactly once across active and temporarily unbound runtimes", async () => { + const activeRuntime = createMockExtensionApi(); + (subagentsModule as any).default(activeRuntime.api); + await emitMockEvent(activeRuntime, "session_start", { reason: "startup" }, ctx); + + const deliver = (name: string) => (pi: any) => { + pi.sendMessage({ customType: "subagent_result", content: name, display: true }); + }; + + assert.equal(testApi.sendResultWhenRuntimeReady("child-a", deliver("child-a")), true); + assert.equal(testApi.sendResultWhenRuntimeReady("child-a", deliver("child-a duplicate")), false); + + let transientAttempts = 0; + assert.equal(testApi.sendResultWhenRuntimeReady("child-retry", (pi: any) => { + transientAttempts += 1; + if (transientAttempts === 1) throw new Error("stale runtime"); + pi.sendMessage({ customType: "subagent_result", content: "child-retry", display: true }); + }), true); + assert.equal(testApi.sendResultWhenRuntimeReady("child-independent", deliver("child-independent")), true); + assert.deepEqual( + activeRuntime.sentMessages.map(({ message }) => message.content), + ["child-a", "child-independent"], + "one failing delivery must not block its sibling", + ); + + // Registration happens before session_start during reload. Results in this + // gap must queue rather than use the still-bound old ExtensionAPI. + const freshRuntime = createMockExtensionApi(); + (subagentsModule as any).default(freshRuntime.api); + assert.equal(testApi.sendResultWhenRuntimeReady("child-b", deliver("child-b")), true); + assert.equal(activeRuntime.sentMessages.length, 2, "the stale runtime must not receive child-b"); + + await emitMockEvent(activeRuntime, "session_shutdown", { reason: "reload" }, ctx); + assert.equal(testApi.sendResultWhenRuntimeReady("child-c", deliver("child-c failed")), true); + assert.equal(testApi.sendResultWhenRuntimeReady("child-b", deliver("child-b duplicate")), false); + assert.equal(shared.pendingDeliveries.length, 3); + + await emitMockEvent(freshRuntime, "session_start", { reason: "reload" }, ctx); + await new Promise((resolve) => setImmediate(resolve)); + + assert.deepEqual( + freshRuntime.sentMessages.map(({ message }) => message.content), + ["child-retry", "child-b", "child-c failed"], + ); + testApi.flushPendingDeliveries(); + assert.equal(freshRuntime.sentMessages.length, 3); + assert.equal(testApi.sendResultWhenRuntimeReady("child-c", deliver("post-flush duplicate")), false); + }); + + it("bounds delivered result tombstones without evicting queued results", () => { + shared.latestPi = { sendMessage() {} }; + shared.activeRuntimeToken = Symbol.for("test-runtime"); + shared.registeredRuntimeToken = shared.activeRuntimeToken; + for (let index = 0; index < 1_100; index += 1) { + assert.equal(testApi.sendResultWhenRuntimeReady(`finished-${index}`, () => {}), true); + } + assert.ok(shared.resultDeliveryStates.size <= 1_024); + }); + + it("rejects launches after cancellation or runtime replacement", () => { + const controller = new AbortController(); + const runtimeToken = Symbol("runtime"); + shared.activeRuntimeToken = runtimeToken; + shared.registeredRuntimeToken = runtimeToken; + assert.doesNotThrow(() => testApi.assertLaunchActive(controller.signal, runtimeToken)); + + controller.abort(); + assert.throws( + () => testApi.assertLaunchActive(controller.signal, runtimeToken), + /parent session ended/i, + ); + + const currentController = new AbortController(); + shared.registeredRuntimeToken = Symbol("replacement"); + assert.throws( + () => testApi.assertLaunchActive(currentController.signal, runtimeToken), + /parent session ended/i, + ); + }); + + it("rejects a completion from a replaced parent even after its watcher left the running map", () => { + const completed = { + id: "old-child", + name: "old-child", + task: "done", + surface: "old-pane", + startTime: 0, + sessionFile: "/tmp/old-child.jsonl", + interactive: false, + parentGeneration: shared.parentGeneration, + statusState: createStatusState({ source: "pi", startTimeMs: 0 }), + }; + assert.equal(testApi.shouldDeliverRunningResult(completed), true); + shared.parentGeneration += 1; + assert.equal(testApi.shouldDeliverRunningResult(completed), false); + }); + + it("drops a queued subagent question after parent replacement", () => { + const dir = createTestDir(); + const sessionFile = join(dir, "question-child.jsonl"); + const running = { + id: "question-child", + name: "question-child", + task: "ask", + agent: "worker", + surface: "question-pane", + startTime: 0, + sessionFile, + interactive: false, + parentGeneration: shared.parentGeneration, + statusState: createStatusState({ source: "pi", startTimeMs: 0 }), + }; + writeFileSync(`${sessionFile}.ask`, "{"); + + try { + testApi.deliverPendingQuestion(running); + assert.equal(existsSync(`${sessionFile}.ask`), true, "partial signals must be retried"); + assert.equal(shared.pendingDeliveries.length, 0); + + writeFileSync(`${sessionFile}.ask`, JSON.stringify({ question: "Still there?" })); + testApi.deliverPendingQuestion(running); + assert.equal(existsSync(`${sessionFile}.ask`), false); + assert.equal(shared.pendingDeliveries.length, 1); + shared.parentGeneration += 1; + const replacement = createMockExtensionApi(); + shared.latestPi = replacement.api; + shared.activeRuntimeToken = Symbol.for("replacement-runtime"); + shared.registeredRuntimeToken = shared.activeRuntimeToken; + testApi.flushPendingDeliveries(); + assert.equal(replacement.sentMessages.length, 0); + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }); + + it("invalidates old children when replacement session_start wins the shutdown race", async () => { + const oldRuntime = createMockExtensionApi(); + (subagentsModule as any).default(oldRuntime.api); + await emitMockEvent(oldRuntime, "session_start", { reason: "startup" }, ctx); + + const abortController = new AbortController(); + shared.runningSubagents.set("replaced-child", { + id: "replaced-child", + name: "replaced-child", + task: "old work", + surface: "old-pane", + startTime: 0, + sessionFile: "/tmp/replaced-child.jsonl", + interactive: false, + parentGeneration: shared.parentGeneration, + statusState: createStatusState({ source: "pi", startTimeMs: 0 }), + abortController, + }); + testApi.sendWhenRuntimeReady(() => { + throw new Error("queue this old callback"); + }); + assert.equal(shared.pendingDeliveries.length, 1); + + const freshRuntime = createMockExtensionApi(); + (subagentsModule as any).default(freshRuntime.api); + await emitMockEvent( + freshRuntime, + "session_start", + { reason: "new", previousSessionFile: "/tmp/old-parent.jsonl" }, + ctx, + ); + + assert.equal(abortController.signal.aborted, true); + assert.equal(shared.runningSubagents.size, 0); + assert.equal(shared.pendingDeliveries.length, 0); + assert.equal(shared.parentGeneration, 1); + + await emitMockEvent(oldRuntime, "session_shutdown", { reason: "new" }, ctx); + assert.equal(shared.parentGeneration, 1, "delayed old shutdown must not invalidate twice"); + }); + it("preserves a live subagent and delivers its queued result through the fresh runtime", async () => { const oldRuntime = createMockExtensionApi(); (subagentsModule as any).default(oldRuntime.api); @@ -1981,6 +2224,7 @@ describe("reload lifecycle", () => { sessionFile: string; statusState: ReturnType; interactive: boolean; + parentGeneration: number; abortController: AbortController; suppressDelivery?: boolean; } = { @@ -1992,6 +2236,7 @@ describe("reload lifecycle", () => { sessionFile: "/tmp/reload-child.jsonl", statusState: createStatusState({ source: "pi", startTimeMs: Date.now() }), interactive: false, + parentGeneration: shared.parentGeneration, abortController, }; shared.runningSubagents.set(running.id, running); @@ -2032,9 +2277,13 @@ describe("reload lifecycle", () => { }); describe("tool registration", () => { - it("always resumes subagents as autonomous (auto-exit, non-interactive tracking)", () => { + it("always resumes subagents as autonomous with reliable follow-up delivery", () => { const testApi = (subagentsModule as any).__test__; + assert.deepEqual(testApi.AUTOMATIC_DELIVERY_OPTIONS, { + triggerTurn: true, + deliverAs: "followUp", + }); assert.deepEqual(testApi.resolveResumeLaunchBehavior(), { autoExit: true, interactive: false, @@ -2367,7 +2616,7 @@ describe("subagent interruption", () => { } }); - it("uniqueRunningName suffixes defaulted names that collide with running subagents", () => { + it("uniqueRunningName suffixes explicit or defaulted names that collide with running subagents", () => { const testApi = (subagentsModule as any).__test__; const runningMap = testApi.runningSubagents as Map; runningMap.clear(); @@ -2432,6 +2681,61 @@ describe("subagent interruption", () => { } }); + it("serializes concurrent finished-session resumes by name", () => { + const testApi = (subagentsModule as any).__test__; + const resumingNames = testApi.resumingNames as Set; + resumingNames.clear(); + try { + assert.equal(testApi.reserveResumeName("Worker"), true); + assert.equal(testApi.reserveResumeName("Worker"), false); + assert.equal(testApi.reserveResumeName("Scout"), true); + } finally { + resumingNames.clear(); + } + }); + + it("clears stale completion signals before resuming a finished session", () => { + const testApi = (subagentsModule as any).__test__; + const dir = createTestDir(); + const sessionFile = join(dir, "finished.jsonl"); + writeFileSync(`${sessionFile}.exit`, JSON.stringify({ type: "error" })); + writeFileSync(`${sessionFile}.ask`, JSON.stringify({ question: "stale" })); + try { + testApi.clearStaleSessionSignals(sessionFile); + assert.equal(existsSync(`${sessionFile}.exit`), false); + assert.equal(existsSync(`${sessionFile}.ask`), false); + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }); + + it("recognizes terminal completion and waits for the watcher before late steering", async () => { + const testApi = (subagentsModule as any).__test__; + const runningMap = testApi.runningSubagents as Map; + const dir = createTestDir(); + const completionFile = join(dir, "child.exit-code"); + const running: any = makeRunning({ sessionFile: join(dir, "child.jsonl"), completionFile }); + runningMap.set(running.id, running); + + try { + assert.equal(testApi.hasTerminalCompletionSignal(running), false); + writeFileSync(completionFile, "0\n"); + assert.equal(testApi.hasTerminalCompletionSignal(running), true); + rmSync(completionFile); + running.completionObserved = true; + assert.equal( + testApi.hasTerminalCompletionSignal(running), + true, + "watcher state must preserve completion after consuming the file", + ); + setTimeout(() => runningMap.delete(running.id), 10); + assert.equal(await testApi.waitForWatcherCleanup(running, new AbortController().signal, 200), true); + } finally { + runningMap.clear(); + rmSync(dir, { recursive: true, force: true }); + } + }); + it("steers a running subagent by typing into its pane (newlines flattened)", () => { const testApi = (subagentsModule as any).__test__; let sentSurface = "";