Skip to content
Closed
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
12 changes: 12 additions & 0 deletions src/subagent/agent-fleet.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1118,6 +1118,7 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool {
const parentTraceId = getCurrentTurnTraceId();
telemetry.capture("subagent_start", { agent_name: agentName });
const startedAt = Date.now();
let setupEndMs: number | undefined;
let settlement: Readonly<SubAgentRunSettlement> | undefined;
let endFinalized = false;
let runInterrupted = false;
Expand All @@ -1136,6 +1137,7 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool {
agentName,
status: setupFailed ? "failed" : status,
durationMs: Date.now() - startedAt,
setupMs: (setupEndMs ?? Date.now()) - startedAt,
model: settlement?.model ?? provider.model,
stopReason: setupFailed
? "setup_error"
Expand All @@ -1149,6 +1151,13 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool {
reasoning_tokens: 0,
tool_call_count: 0,
tool_error_count: 0,
tool_read_count: 0,
tool_write_count: 0,
tool_shell_count: 0,
tool_search_count: 0,
tool_agent_count: 0,
tool_other_count: 0,
hydrate_ms: 0,
},
...(parentTraceId !== undefined ? { parentTraceId } : {}),
});
Expand Down Expand Up @@ -1503,6 +1512,9 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool {

// Fire and forget: this handler must return before the worker finishes.
// Wait JSON projects stored WorkerLifecycle plus the per-install overlay.
// Setup (worktree snapshot, admission, lane checks) ends here; the
// worker run owns the remaining duration.
setupEndMs = Date.now();
deps
.run(params)
.then((result) => {
Expand Down
28 changes: 28 additions & 0 deletions src/subagent/index.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ import type {
ReactorState,
} from "@intx/types/runtime";
import { defined } from "../../tests/helpers/defined.js";
import { classifyToolFamily } from "./run.js";

describe("sub-agent teardown", () => {
test("disposeSubAgentSession closes agent, awaits stream, and disposes posix tools once", async () => {
Expand Down Expand Up @@ -1424,3 +1425,30 @@ describe("buildDispatchBrief typed spawn contract", () => {
expect(both).toContain("1. manage_tasks seed");
});
});

describe("classifyToolFamily", () => {
test("maps canonical engine names onto the six telemetry buckets", () => {
expect(classifyToolFamily("read_file")).toBe("read");
expect(classifyToolFamily("write_file")).toBe("write");
expect(classifyToolFamily("edit_file")).toBe("write");
expect(classifyToolFamily("delete_file")).toBe("write");
expect(classifyToolFamily("run_shell")).toBe("shell");
expect(classifyToolFamily("shell_collect")).toBe("shell");
expect(classifyToolFamily("grep")).toBe("search");
expect(classifyToolFamily("search_files")).toBe("search");
expect(classifyToolFamily("search_agents")).toBe("search");
expect(classifyToolFamily("spawn_agent")).toBe("agent");
expect(classifyToolFamily("wait_agents")).toBe("agent");
expect(classifyToolFamily("ask_director")).toBe("agent");
expect(classifyToolFamily("submit_result")).toBe("agent");
});

test("falls back to other for orchestration, web, and unknown tools", () => {
expect(classifyToolFamily("manage_tasks")).toBe("other");
expect(classifyToolFamily("web_fetch")).toBe("other");
expect(classifyToolFamily("web_search")).toBe("other");
expect(classifyToolFamily("mcp__plugin_tool")).toBe("other");
expect(classifyToolFamily("something_unknown")).toBe("other");
expect(classifyToolFamily("")).toBe("other");
});
});
306 changes: 306 additions & 0 deletions src/subagent/run-settlement.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,14 @@ import { tmpdir } from "node:os";
import { join } from "node:path";

import type { ReactorEmittedEvent } from "@intx/inference";
import type {
ContextTransform,
ConversationTurn,
StrategyContext,
} from "@intx/types/runtime";

import { withMockedModuleDuring } from "../../tests/helpers/mock-module.js";
import { defined } from "../../tests/helpers/defined.js";
import { createPermissionGate } from "../permission/gate.js";
import type { SubAgentRunSettlement } from "./types.js";

Expand Down Expand Up @@ -193,3 +199,303 @@ test("pre-progress cancellation settles as cancelled without changing rejection"
});
expect(Object.isFrozen(settlement)).toBe(true);
});

test("tool calls accumulate into family buckets whose sum equals tool_call_count", async () => {
const cwd = await mkdtemp(join(tmpdir(), "corbits-run-buckets-"));
const originalError = new Error("bucket probe done");
let settlement: Readonly<SubAgentRunSettlement> | undefined;
const toolNames = [
"read_file",
"read_file",
"write_file",
"run_shell",
"grep",
"spawn_agent",
"manage_tasks",
"mystery_tool",
];

const caught = await withMockedModuleDuring(
import.meta.resolve("../agent/live-tool-dispatch.js"),
(real: typeof import("../agent/live-tool-dispatch.js")) => ({
...real,
createAgentWithLiveToolDispatch: async () => ({
send: async () => {
await new Promise((resolve) => setTimeout(resolve, 20));
throw originalError;
},
stream: () =>
(async function* (): AsyncGenerator<ReactorEmittedEvent> {
for (const [index, name] of toolNames.entries()) {
yield {
type: "tool.start",
seq: index + 1,
data: {
call: { id: `call-${index}`, name, arguments: {} },
},
} as ReactorEmittedEvent;
}
})(),
deliver: () => undefined,
close: async () => undefined,
setSource: () => undefined,
setSources: () => undefined,
history: async () => [],
checkpoints: async () => [],
readAt: async () => [],
blobReader: {},
}),
}),
async () => {
const { runSubAgent } = await import("./run.js");
try {
await runSubAgent({
cwd,
workdirBase: join(cwd, ".ctx"),
permissionGate,
provider: {
providerName: "initial",
baseURL: "http://localhost",
model: "initial-model",
},
description: "bucket probe",
prompt: "use tools across families",
onRunSettled: (summary) => {
settlement = summary;
},
});
} catch (error) {
return error;
}
throw new Error("expected runSubAgent to reject");
},
);

expect(caught).toBe(originalError);
expect(settlement).toMatchObject({
tool_call_count: 8,
tool_read_count: 2,
tool_write_count: 1,
tool_shell_count: 1,
tool_search_count: 1,
tool_agent_count: 1,
tool_other_count: 2,
hydrate_ms: 0,
});
const settled = defined(settlement);
expect(settled.tool_call_count).toBe(
settled.tool_read_count +
settled.tool_write_count +
settled.tool_shell_count +
settled.tool_search_count +
settled.tool_agent_count +
settled.tool_other_count,
);
});

test("wire names bucket into families with zero other", async () => {
const cwd = await mkdtemp(join(tmpdir(), "corbits-run-wire-buckets-"));
const originalError = new Error("wire bucket probe done");
let settlement: Readonly<SubAgentRunSettlement> | undefined;
const toolNames = [
"read",
"write",
"edit",
"delete",
"bash",
"glob",
"grep",
"shell",
"default.bash",
"bash__P1",
"apply_patch",
"close_agent",
"resume_agent",
"interrupt_agent",
"send_input",
"read_agent_trace",
];

const caught = await withMockedModuleDuring(
import.meta.resolve("../agent/live-tool-dispatch.js"),
(real: typeof import("../agent/live-tool-dispatch.js")) => ({
...real,
createAgentWithLiveToolDispatch: async () => ({
send: async () => {
await new Promise((resolve) => setTimeout(resolve, 20));
throw originalError;
},
stream: () =>
(async function* (): AsyncGenerator<ReactorEmittedEvent> {
for (const [index, name] of toolNames.entries()) {
yield {
type: "tool.start",
seq: index + 1,
data: {
call: { id: `call-${index}`, name, arguments: {} },
},
} as ReactorEmittedEvent;
}
})(),
deliver: () => undefined,
close: async () => undefined,
setSource: () => undefined,
setSources: () => undefined,
history: async () => [],
checkpoints: async () => [],
readAt: async () => [],
blobReader: {},
}),
}),
async () => {
const { runSubAgent } = await import("./run.js");
try {
await runSubAgent({
cwd,
workdirBase: join(cwd, ".ctx"),
permissionGate,
provider: {
providerName: "initial",
baseURL: "http://localhost",
model: "initial-model",
},
description: "wire bucket probe",
prompt: "use tools across families",
onRunSettled: (summary) => {
settlement = summary;
},
});
} catch (error) {
return error;
}
throw new Error("expected runSubAgent to reject");
},
);

expect(caught).toBe(originalError);
expect(settlement).toMatchObject({
tool_call_count: 16,
tool_read_count: 1,
tool_write_count: 4,
tool_shell_count: 4,
tool_search_count: 2,
tool_agent_count: 5,
tool_other_count: 0,
hydrate_ms: 0,
});
const settled = defined(settlement);
expect(settled.tool_call_count).toBe(
settled.tool_read_count +
settled.tool_write_count +
settled.tool_shell_count +
settled.tool_search_count +
settled.tool_agent_count +
settled.tool_other_count,
);
});

async function runWithRehydrateDelay(
delayMs: number,
): Promise<Readonly<SubAgentRunSettlement> | undefined> {
const cwd = await mkdtemp(join(tmpdir(), "corbits-run-hydrate-"));
const originalError = new Error("hydrate probe done");
let settlement: Readonly<SubAgentRunSettlement> | undefined;
let captured: ContextTransform[] | undefined;
const testCtx = {} as StrategyContext;

await withMockedModuleDuring(
import.meta.resolve("../session/attachment-store.js"),
(real: typeof import("../session/attachment-store.js")) => ({
...real,
createAttachmentRehydrateTransform: () => ({
name: "attachment-rehydrate",
version: "1",
apply: async (turns: ConversationTurn[]) => {
if (delayMs > 0) {
await new Promise((resolve) => setTimeout(resolve, delayMs));
}
return {
output: turns,
record: {
strategy: "attachment-rehydrate",
version: "1",
parameters: {},
reason: "noop",
decisions: {},
},
};
},
}),
}),
async () =>
withMockedModuleDuring(
import.meta.resolve("../agent/live-tool-dispatch.js"),
(real: typeof import("../agent/live-tool-dispatch.js")) => ({
...real,
createAgentWithLiveToolDispatch: async (...args: unknown[]) => {
const env = args[1] as
| { deps?: { contextTransforms?: ContextTransform[] } }
| undefined;
captured = env?.deps?.contextTransforms;
return {
send: async () => {
const rehydrate = captured?.find(
(transform) => transform.name === "attachment-rehydrate",
);
await rehydrate?.apply([], testCtx);
throw originalError;
},
stream: () =>
(async function* (): AsyncGenerator<ReactorEmittedEvent> {
yield* [];
})(),
deliver: () => undefined,
close: async () => undefined,
setSource: () => undefined,
setSources: () => undefined,
history: async () => [],
checkpoints: async () => [],
readAt: async () => [],
blobReader: {},
} as unknown as Awaited<
ReturnType<typeof real.createAgentWithLiveToolDispatch>
>;
},
}),
async () => {
const { runSubAgent } = await import("./run.js");
try {
await runSubAgent({
cwd,
workdirBase: join(cwd, ".ctx"),
permissionGate,
provider: {
providerName: "initial",
baseURL: "http://localhost",
model: "initial-model",
},
description: "hydrate probe",
prompt: "rehydrate attachments",
onRunSettled: (summary) => {
settlement = summary;
},
});
} catch {
return;
}
throw new Error("expected runSubAgent to reject");
},
),
);
return settlement;
}

test("a slow attachment rehydrate accumulates into hydrate_ms", async () => {
const settlement = await runWithRehydrateDelay(60);
expect(defined(settlement).hydrate_ms).toBeGreaterThanOrEqual(50);
});

test("a noop attachment rehydrate reports hydrate_ms zero", async () => {
const settlement = await runWithRehydrateDelay(0);
expect(defined(settlement).hydrate_ms).toBe(0);
});
Loading
Loading