Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions src/agent/tools.ts
Original file line number Diff line number Diff line change
Expand Up @@ -282,6 +282,7 @@ export async function createAgentToolset(args: AgentToolsetArgs): Promise<AgentT
const orchestratorTools: AgentTool[] = [];
if (subAgentsEnabled && args.subAgent !== undefined) {
const sa = args.subAgent;
const fleetRecords = sa.sessions !== undefined ? createFleetRecords() : undefined;
orchestratorTools.push(
createTaskTool({
cwd,
Expand All @@ -302,6 +303,7 @@ export async function createAgentToolset(args: AgentToolsetArgs): Promise<AgentT
...(args.getBlobReader !== undefined ? { getBlobReader: args.getBlobReader } : {}),
...(sa.useWorktree !== undefined ? { useWorktree: sa.useWorktree } : {}),
...(args.telemetry !== undefined ? { telemetry: args.telemetry } : {}),
...(fleetRecords !== undefined ? { fleetRecords } : {}),
}),
);
if (sa.profiles !== undefined) {
Expand All @@ -321,9 +323,8 @@ export async function createAgentToolset(args: AgentToolsetArgs): Promise<AgentT
// Mirror nested runSubAgent's orchestrator fleet mount (run.ts), but
// reuse the existing TUI/exec session store — do not allocate a private
// store only for these verbs. spawnAllowlist stays unwired on primary.
if (sa.sessions !== undefined) {
if (sa.sessions !== undefined && fleetRecords !== undefined) {
const fleetSessions = sa.sessions;
const fleetRecords = createFleetRecords();
const fleetDeps = {
permissionGate,
inheritMcpTools: () => inheritedMcpTools,
Expand Down
31 changes: 24 additions & 7 deletions src/subagent/agent-fleet.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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"]);

Expand Down Expand Up @@ -363,6 +365,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;
Expand Down Expand Up @@ -660,7 +664,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 {
Expand Down Expand Up @@ -716,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", {
Expand Down
14 changes: 5 additions & 9 deletions src/subagent/run.ts
Original file line number Diff line number Diff line change
Expand Up @@ -521,6 +521,8 @@ export async function runSubAgent(params: RunSubAgentParams): Promise<RunSubAgen
);
}
const nd = params.nestedDispatch;
const fleetSessions = nd.sessions ?? createSubAgentSessionStore();
const fleetRecords = createFleetRecords();
tools = [
...tools,
createTaskTool({
Expand All @@ -542,7 +544,8 @@ export async function runSubAgent(params: RunSubAgentParams): Promise<RunSubAgen
telemetry: liveTelemetry,
...(nd.onEvent !== undefined ? { onEvent: nd.onEvent } : {}),
...(nd.onProgress !== undefined ? { onProgress: nd.onProgress } : {}),
...(nd.sessions !== undefined ? { sessions: nd.sessions } : {}),
sessions: fleetSessions,
fleetRecords,
...(nd.settings !== undefined ? { settings: nd.settings } : {}),
...(nd.catalog !== undefined ? { catalog: nd.catalog } : {}),
...(nd.profiles !== undefined ? { profiles: nd.profiles } : {}),
Expand All @@ -569,16 +572,9 @@ export async function runSubAgent(params: RunSubAgentParams): Promise<RunSubAgen
createReadAgentTraceTool(nd.getWorkdirBase, {
actorId: params.id,
tier,
getNodes: () => 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,
Expand Down
Loading
Loading