From c8c2c61a33ff505be5b73d489b1a697cf30c1b4d Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Tue, 25 Aug 2026 07:47:33 -0700 Subject: [PATCH 1/2] Make director task() a spawn_agent plus wait_agents wrapper Closed-director task() was a second full spawn engine. When a session store is present it now starts the worker through spawn_agent and blocks on wait_agents, so one mailbox owns completion. Custom AgentProfile lookup still uses the legacy await-run path. --- src/agent/tools.ts | 5 +- src/subagent/agent-fleet.ts | 4 +- src/subagent/run.ts | 14 +-- src/subagent/task-tool.ts | 189 ++++++++++++++++++++++++++++ src/subagent/task-via-fleet.test.ts | 48 +++++++ 5 files changed, 248 insertions(+), 12 deletions(-) create mode 100644 src/subagent/task-via-fleet.test.ts diff --git a/src/agent/tools.ts b/src/agent/tools.ts index 7c75933b9..bac8e26a0 100644 --- a/src/agent/tools.ts +++ b/src/agent/tools.ts @@ -282,6 +282,7 @@ export async function createAgentToolset(args: AgentToolsetArgs): Promise inheritedMcpTools, diff --git a/src/subagent/agent-fleet.ts b/src/subagent/agent-fleet.ts index acd2ce641..dbcc840c1 100644 --- a/src/subagent/agent-fleet.ts +++ b/src/subagent/agent-fleet.ts @@ -363,6 +363,8 @@ export type AgentFleetDeps = SubAgentSandboxDeps & { useWorktree?: boolean; /** Optional wall-clock budget (ms) forwarded to runSubAgent. */ deadlineMs?: number; + /** When false, tear the worker down on completion (task wrapper). Default true. */ + persist?: boolean; settings?: Settings | (() => Settings | undefined); catalog?: readonly ProviderCatalogEntry[] | (() => readonly ProviderCatalogEntry[]); onEvent?: (event: ReactorEmittedEvent) => void; @@ -660,7 +662,7 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { // stays alive for followup (agentRetained / interrupt keep-alive) — // matching run.ts's persisting gate so followup_task does not hit a // removed cwd. - persist: true, + persist: deps.persist !== false, onAgentReady: ({ close, interrupt, followup, deliver }) => { deps.sessions.registerClose(session.id, async (deadlineMs) => { try { diff --git a/src/subagent/run.ts b/src/subagent/run.ts index 49826c96f..0124903af 100644 --- a/src/subagent/run.ts +++ b/src/subagent/run.ts @@ -521,6 +521,8 @@ export async function runSubAgent(params: RunSubAgentParams): Promise nd.sessions?.list() ?? [], + getNodes: () => fleetSessions.list(), }), ]; - // spawn_agent/wait_agents need a session store as their mailbox; - // reuse the orchestrator's if it has one, else give this install its - // own. fleetRecords holds terminal results the session store's - // display cap would otherwise evict before wait_agents collects them - // (see agent-fleet.ts). - const fleetSessions = nd.sessions ?? createSubAgentSessionStore(); - const fleetRecords = createFleetRecords(); const lifecycleAuthority = { actorId: params.id, tier, diff --git a/src/subagent/task-tool.ts b/src/subagent/task-tool.ts index 01bb03c41..12743484e 100644 --- a/src/subagent/task-tool.ts +++ b/src/subagent/task-tool.ts @@ -6,8 +6,11 @@ import { tool } from "@intx/agent"; import type { AgentTool } from "@intx/agent"; import { type } from "arktype"; import type { ReactorEmittedEvent } from "@intx/inference"; +import { getLogger } from "@intx/log"; import type { ToolDefinition, ToolResult } from "@intx/types/runtime"; +import { LOG_NAMESPACE_ROOT } from "../branding.js"; + import { runtimeSettingsWithCatalog, type ProviderCatalogEntry } from "../config/index.js"; import { formatSubAgentTaskAuthFailureMessage } from "./inference-auth-failure.js"; import type { CapabilityFilter, AgentProfile } from "../agent/profiles.js"; @@ -30,6 +33,14 @@ import { } from "../provider/reasoning-effort.js"; import { isCodexProviderName } from "../config/codex-providers.js"; import { DEFAULT_CANCEL_REASON, type SubAgentSessionStore } from "./session-store.js"; +import { + createFleetRecords, + createSpawnAgentTool, + createWaitAgentsTool, + MAX_WAIT_TIMEOUT_MS, + type AgentFleetDeps, + type FleetRecordsHandle, +} from "./agent-fleet.js"; import { buildDispatchBrief, type TaskIntent } from "./report.js"; import { appendSubAgentParentHints, type ForcedStopReason } from "./stop-policy.js"; import { @@ -55,6 +66,8 @@ import type { SubAgentSandboxDeps, } from "./types.js"; +const log = getLogger([LOG_NAMESPACE_ROOT, "subagent", "task-tool"]); + export const TaskToolArgs = type({ description: "string", prompt: "string", @@ -186,6 +199,8 @@ export type TaskToolDeps = SubAgentSandboxDeps & { // Records sub-agent starts and outcomes. Injected so the tool has no // process-wide dependency; omitting it makes dispatch silent. telemetry?: Telemetry; + /** Shared with spawn_agent/wait_agents when this task tool is fleet-backed. */ + fleetRecords?: FleetRecordsHandle; }; function taskToolResult( @@ -249,11 +264,159 @@ function requiredTaskFieldsError( return message; } +async function runTaskViaFleet(input: { + callId: string; + signal: AbortSignal; + description: string; + prompt: string; + context: string | undefined; + agentId: string | undefined; + goals: string[]; + intent: TaskIntent | undefined; + successCriteria: string[]; + doNot: string[]; + reportFocus: string | undefined; + deps: TaskToolDeps; + sessions: SubAgentSessionStore; + fleetRecords: FleetRecordsHandle; +}): Promise { + const fleetDeps: AgentFleetDeps = { + permissionGate: input.deps.permissionGate, + ...(input.deps.inheritMcpTools !== undefined + ? { inheritMcpTools: input.deps.inheritMcpTools } + : {}), + ...(input.deps.shellTimeout !== undefined ? { shellTimeout: input.deps.shellTimeout } : {}), + ...(input.deps.shellEnv !== undefined ? { shellEnv: input.deps.shellEnv } : {}), + ...(input.deps.extraToolPlugins !== undefined + ? { extraToolPlugins: input.deps.extraToolPlugins } + : {}), + ...(input.deps.getBlobReader !== undefined ? { getBlobReader: input.deps.getBlobReader } : {}), + cwd: input.deps.cwd, + getWorkdirBase: input.deps.getWorkdirBase, + provider: input.deps.provider, + run: input.deps.run, + sessions: input.sessions, + fleetRecords: input.fleetRecords, + persist: false, + ...(input.deps.parentSessionId !== undefined + ? { parentSessionId: input.deps.parentSessionId } + : {}), + ...(input.deps.spawnAllowlist !== undefined + ? { spawnAllowlist: input.deps.spawnAllowlist } + : {}), + ...(input.deps.allowOrchestrator !== undefined + ? { allowOrchestrator: input.deps.allowOrchestrator } + : {}), + ...(input.deps.useWorktree !== undefined ? { useWorktree: input.deps.useWorktree } : {}), + ...(input.deps.deadlineMs !== undefined ? { deadlineMs: input.deps.deadlineMs } : {}), + ...(input.deps.settings !== undefined ? { settings: input.deps.settings } : {}), + ...(input.deps.catalog !== undefined ? { catalog: input.deps.catalog } : {}), + ...(input.deps.onEvent !== undefined ? { onEvent: input.deps.onEvent } : {}), + ...(input.deps.onProgress !== undefined ? { onProgress: input.deps.onProgress } : {}), + ...(input.deps.telemetry !== undefined ? { telemetry: input.deps.telemetry } : {}), + }; + const spawn = createSpawnAgentTool(fleetDeps); + const wait = createWaitAgentsTool({ + sessions: input.sessions, + fleetRecords: input.fleetRecords, + }); + if (spawn.kind !== "full" || wait.kind !== "full") { + return taskToolResult(input.callId, "Error: fleet tools are unavailable."); + } + const started = await spawn.handler( + { + id: input.callId, + name: "spawn_agent", + arguments: { + description: input.description, + prompt: input.prompt, + ...(input.context !== undefined ? { context: input.context } : {}), + ...(input.agentId !== undefined ? { agent: input.agentId } : {}), + ...(input.goals.length > 0 ? { goals: input.goals } : {}), + ...(input.intent !== undefined ? { intent: input.intent } : {}), + ...(input.successCriteria.length > 0 ? { success_criteria: input.successCriteria } : {}), + ...(input.doNot.length > 0 ? { do_not: input.doNot } : {}), + ...(input.reportFocus !== undefined ? { report_focus: input.reportFocus } : {}), + }, + }, + input.signal, + ); + const startedText = + typeof started.content === "string" ? started.content : JSON.stringify(started.content); + if (started.isError === true || startedText.startsWith("Error:")) { + return taskToolResult(input.callId, startedText); + } + let agentId: string; + try { + const parsed = JSON.parse(startedText) as { agent_id?: unknown }; + if (typeof parsed.agent_id !== "string" || parsed.agent_id.length === 0) { + return taskToolResult(input.callId, "Error: spawn_agent returned no agent_id."); + } + agentId = parsed.agent_id; + } catch (err) { + log.error("spawn_agent payload was not JSON: {error}", { + error: err instanceof Error ? err.message : String(err), + }); + return taskToolResult( + input.callId, + `Error: spawn_agent returned invalid payload: ${startedText}`, + ); + } + + while (!input.signal.aborted) { + const waited = await wait.handler( + { + id: `${input.callId}-wait`, + name: "wait_agents", + arguments: { targets: [agentId], mode: "all", timeout_ms: MAX_WAIT_TIMEOUT_MS }, + }, + input.signal, + ); + const waitedText = + typeof waited.content === "string" ? waited.content : JSON.stringify(waited.content); + if (waited.isError === true || waitedText.startsWith("Error:")) { + return taskToolResult(input.callId, waitedText); + } + let payload: { + timed_out?: boolean; + results?: { status?: string; report?: string; error?: string }[]; + }; + try { + payload = JSON.parse(waitedText) as typeof payload; + } catch (err) { + log.error("wait_agents payload was not JSON: {error}", { + error: err instanceof Error ? err.message : String(err), + }); + return taskToolResult( + input.callId, + `Error: wait_agents returned invalid payload: ${waitedText}`, + ); + } + if (payload.timed_out === true) continue; + const result = payload.results?.[0]; + if (result === undefined) { + return taskToolResult(input.callId, `Error: wait_agents returned no result for ${agentId}.`); + } + if (result.status === "failed") { + return taskToolResult( + input.callId, + `Error: sub-agent "${input.description}" failed: ${result.error ?? "unknown error"}`, + ); + } + const report = result.report ?? ""; + return taskToolResult(input.callId, `Sub-agent "${input.description}" reported:\n\n${report}`); + } + return taskToolResult(input.callId, `Sub-agent "${input.description}" cancelled by operator.`); +} + export function createTaskTool(deps: TaskToolDeps): AgentTool { const run = deps.run; const telemetry = deps.telemetry ?? NOOP_TELEMETRY; // Session-scoped re-dispatch ledger: one per parent task tool instance. const briefLedger = createBriefDispatchLedger(); + const fleetSessions = deps.sessions; + const fleetRecords = + deps.fleetRecords ?? (fleetSessions !== undefined ? createFleetRecords() : undefined); // Every completed dispatch gets an outcome record — the log otherwise // carries shape and run state but never what the run actually produced. // Tagged with the dispatched child's provider/model/family so @@ -334,6 +497,32 @@ export function createTaskTool(deps: TaskToolDeps): AgentTool { return taskToolResult(call.id, requiredTaskFieldsError(args, empty)); } + // Closed-director task() is spawn_agent + wait_agents. Custom profiles + // still use the legacy await-run path until spawn grows profile lookup. + const agentForFleet = typeof args.agent === "string" ? args.agent : undefined; + const canUseFleet = + fleetSessions !== undefined && + fleetRecords !== undefined && + (agentForFleet === undefined || agentForFleet.length === 0 || isDirectorId(agentForFleet)); + if (canUseFleet) { + return await runTaskViaFleet({ + callId: call.id, + signal, + description, + prompt, + context, + agentId, + goals, + intent, + successCriteria, + doNot, + reportFocus, + deps, + sessions: fleetSessions, + fleetRecords, + }); + } + let provider: SubAgentProvider = typeof deps.provider === "function" ? deps.provider() : deps.provider; // Snapshot parent effort before profile-inference rebuilds so role-default diff --git a/src/subagent/task-via-fleet.test.ts b/src/subagent/task-via-fleet.test.ts new file mode 100644 index 000000000..b69039865 --- /dev/null +++ b/src/subagent/task-via-fleet.test.ts @@ -0,0 +1,48 @@ +import { describe, expect, test } from "bun:test"; + +import { createTaskTool } from "./task-tool.js"; +import { createFleetRecords } from "./agent-fleet.js"; +import { createSubAgentSessionStore } from "./session-store.js"; +import { createPermissionGate } from "../permission/gate.js"; + +const testPermissionGate = createPermissionGate({ + approvals: [], + interactive: false, + skipPermissions: true, +}); + +const provider = { + providerName: "test-provider", + baseURL: "http://localhost", + model: "test-model", +}; + +describe("task via spawn_agent + wait_agents", () => { + test("a director task with a session store returns the worker report", async () => { + const sessions = createSubAgentSessionStore(); + const fleetRecords = createFleetRecords(); + const tool = createTaskTool({ + permissionGate: testPermissionGate, + cwd: "/tmp", + getWorkdirBase: () => "/tmp/workdir", + provider, + sessions, + fleetRecords, + run: async () => ({ + report: "## Summary\nshipped\n## Findings\nok\n## Blockers\n\n## Paths\n", + }), + }); + if (tool.kind !== "full") throw new Error("expected full tool"); + const result = await tool.handler( + { + id: "t1", + name: "task", + arguments: { description: "ship", prompt: "do it", intent: "explore" }, + }, + new AbortController().signal, + ); + const content = typeof result.content === "string" ? result.content : ""; + expect(content).toContain('Sub-agent "ship" reported'); + expect(content).toContain("shipped"); + }); +}); From 45b726946adf1e998ab6439ee03dd4c0301f17ce Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Tue, 25 Aug 2026 08:27:39 -0700 Subject: [PATCH 2/2] Preserve task cancel, auth, and abort contracts through the fleet wrapper Director task() now routes through spawn_agent + wait_agents, which was misclassifying AbortError as failed:aborted, dropping Re-authenticate auth wording, and leaving the child running when the parent tool aborted. Map those outcomes back to the legacy fused-task parent contract and prettier the inherited agent-progress tip. --- src/subagent/agent-fleet.ts | 27 ++++++++++++++++++------ src/subagent/task-tool.ts | 42 +++++++++++++++++++++++++++++++++++-- 2 files changed, 61 insertions(+), 8 deletions(-) diff --git a/src/subagent/agent-fleet.ts b/src/subagent/agent-fleet.ts index dbcc840c1..38c16548e 100644 --- a/src/subagent/agent-fleet.ts +++ b/src/subagent/agent-fleet.ts @@ -63,7 +63,7 @@ import type { Settings } from "../config/settings.js"; import { resolveEffortForRole } from "../provider/reasoning-effort.js"; import { isCodexProviderName } from "../config/codex-providers.js"; import { buildDispatchBrief, type TaskIntent } from "./report.js"; -import type { SubAgentSessionStore } from "./session-store.js"; +import { DEFAULT_CANCEL_REASON, type SubAgentSessionStore } from "./session-store.js"; import type { NestedDispatchDeps, RunSubAgentParams, @@ -75,6 +75,8 @@ import { cleanupSubAgentWorktree, createSubAgentWorktree, WorktreeError } from " import { NOOP_TELEMETRY, type Telemetry } from "../telemetry/index.js"; import { classifyAgentName } from "../telemetry/classify.js"; import type { DirectorPackage } from "../agent/directors/types.js"; +import { formatSubAgentTaskAuthFailureMessage } from "./inference-auth-failure.js"; +import { isSubAgentCancelError } from "./dispose.js"; const log = getLogger([LOG_NAMESPACE_ROOT, "subagent", "agent-fleet"]); @@ -718,11 +720,24 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { }) .catch((err) => { // Always terminalize fleetRecords — including pre-progress cancel that - // rethrows with no salvage — so wait_agents does not hang. fail() - // no-ops when cancel already flipped the strip status. - const message = err instanceof Error ? err.message : String(err); - deps.fleetRecords.reject(session.id, message); - deps.sessions.fail(session.id, message); + // rethrows with no salvage — so wait_agents does not hang. Prefer + // cancel semantics over fail when the strip already cancelled or the + // throw is an AbortError (legacy task() parent contract). + const alreadyCancelled = deps.sessions.get(session.id)?.status === "cancelled"; + if (alreadyCancelled || isSubAgentCancelError(err, childCtl.signal)) { + if (!alreadyCancelled) { + deps.sessions.cancel(session.id, DEFAULT_CANCEL_REASON); + } + const message = err instanceof Error ? err.message : String(err); + deps.fleetRecords.reject(session.id, message); + return; + } + // Auth failures keep the actionable Re-authenticate wording that + // task()'s fused path surfaces via formatSubAgentTaskAuthFailureMessage. + const authMessage = formatSubAgentTaskAuthFailureMessage(description, err); + const failReason = authMessage ?? (err instanceof Error ? err.message : String(err)); + deps.fleetRecords.reject(session.id, failReason); + deps.sessions.fail(session.id, failReason); }) .finally(() => { telemetry.capture("subagent_end", { diff --git a/src/subagent/task-tool.ts b/src/subagent/task-tool.ts index 12743484e..f7b37662a 100644 --- a/src/subagent/task-tool.ts +++ b/src/subagent/task-tool.ts @@ -397,16 +397,54 @@ async function runTaskViaFleet(input: { if (result === undefined) { return taskToolResult(input.callId, `Error: wait_agents returned no result for ${agentId}.`); } + // Cancel must not be misclassified as failed:abort — strip cancel / + // AbortError leaves fleetRecords as failed with "aborted" while the + // session store holds cancelled + the operator reason. A cancel that + // still resolved a salvage body (fleet status done) keeps the report, + // matching the fused task() race contract. + const session = input.sessions.get(agentId); + if (session?.status === "cancelled") { + if ( + (result.status === "done" || result.status === "interrupted") && + typeof result.report === "string" && + result.report.length > 0 + ) { + return taskToolResult( + input.callId, + `Sub-agent "${input.description}" reported:\n\n${result.report}`, + ); + } + return taskToolResult( + input.callId, + cancelledSubAgentMessage(input.description, session.error), + ); + } if (result.status === "failed") { + const errText = result.error ?? "unknown error"; + // Auth failures already carry the actionable Re-authenticate wording + // from formatSubAgentTaskAuthFailureMessage (baked in spawn catch). + if (errText.includes("Re-authenticate")) { + return taskToolResult(input.callId, `Error: ${errText}`); + } return taskToolResult( input.callId, - `Error: sub-agent "${input.description}" failed: ${result.error ?? "unknown error"}`, + `Error: sub-agent "${input.description}" failed: ${errText}`, ); } const report = result.report ?? ""; return taskToolResult(input.callId, `Sub-agent "${input.description}" reported:\n\n${report}`); } - return taskToolResult(input.callId, `Sub-agent "${input.description}" cancelled by operator.`); + // Parent tool abort must cancel the child — wait_agents itself has no + // abort side effects (workers stay waitable), so task()'s fused contract + // owns the cancel here. + if (input.sessions.get(agentId)?.status === "running") { + input.sessions.cancel(agentId); + } + const cancelled = input.sessions.get(agentId); + return taskToolResult( + input.callId, + cancelledSubAgentMessage(input.description, cancelled?.error), + ); } export function createTaskTool(deps: TaskToolDeps): AgentTool {