From c193943147d171c432dfaef55839b9492b4ab257 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 23 Aug 2026 22:20:27 -0700 Subject: [PATCH 1/2] Add read_agent_trace fleet verb for orchestrator tiers Lets an orchestrator or nested orchestrator read a worker's on-disk turns.jsonl directly, so a cancelled or interrupted worker's completed work is no longer invisible once its in-memory session record is gone. Every response is bounded (turn window, entry count, per-entry chars), tolerates partially written/malformed lines, and reports a clean error for an unknown target. Fixes a latest-symlink double-count in directory enumeration by resolving symlinks and de-duping by real path. progress_note for leaf workers is a separate follow-up, not included here. --- CHANGELOG.md | 16 ++ src/agent/tools.ts | 4 + src/subagent/authority.ts | 7 +- src/subagent/run.ts | 8 +- src/subagent/trace-reader.test.ts | 228 ++++++++++++++++++ src/subagent/trace-reader.ts | 374 ++++++++++++++++++++++++++++++ src/subagent/trace-tool.test.ts | 48 ++++ src/subagent/trace-tool.ts | 128 ++++++++++ 8 files changed, 809 insertions(+), 4 deletions(-) create mode 100644 src/subagent/trace-reader.test.ts create mode 100644 src/subagent/trace-reader.ts create mode 100644 src/subagent/trace-tool.test.ts create mode 100644 src/subagent/trace-tool.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 6d4314291..ae7fc9e9c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -39,6 +39,22 @@ parallel copies under `docs/` or `scripts/notes/`. At cut time: rename own), so the two layers no longer multiply. Attempt counts are now logged on each recovery so retry storms are visible in traces. +- **`read_agent_trace` lets an orchestrator inspect a worker's on-disk trace + directly**, so a cancelled or interrupted worker's completed work is no + longer invisible just because its in-memory session record is gone. Reads + turns, tool calls, and tool errors straight from the worker's + `turns.jsonl`, tolerating a partially written or malformed line without + failing. Every response is bounded on four independent axes — turn window, + entry count, per-entry characters, and total output characters (the first + three multiply, so a total-output ceiling caps them together) — each with + a hard maximum the caller cannot exceed, and a truncated response says + exactly what was left out and how to page for the rest. A Tier 2 nested + orchestrator can only read its own descendants' traces, enforced by + reusing `SubAgentSessionStore`'s existing parentSessionId chain + (`assertCanTargetAgent`'s first live call site); leaf directors never see + the tool at all. `progress_note` for leaf workers is a separate, + not-yet-implemented follow-up. + ### Fixed - **Interrupting a turn no longer risks a startup crash.** If an interrupt hit diff --git a/src/agent/tools.ts b/src/agent/tools.ts index c61881bee..a16b3035a 100644 --- a/src/agent/tools.ts +++ b/src/agent/tools.ts @@ -46,6 +46,7 @@ import { createWebSearchTool, disposeWebSearchClients } from "../tools/web-searc import { createUseSkillTool } from "./use-skill.js"; import { createToolIndex, createToolSearchTool } from "./tool-search.js"; import { createSearchAgentsTool } from "./agent-search.js"; +import { createReadAgentTraceTool } from "../subagent/trace-tool.js"; import { createCodexToolProxies, type CodexRunManageTasks, @@ -309,6 +310,9 @@ export async function createAgentToolset(args: AgentToolsetArgs): Promise { // AgentProfile with orchestrator: true from mounting task/search_agents // just because it is outside the closed director set. const tier = params.orchestratorTier ?? "leaf"; - for (const verb of ["task", "search_agents"]) { + for (const verb of ["task", "search_agents", "read_agent_trace"]) { assertTierMayMountFleetVerb(tier, verb); } if (params.nestedDispatch === undefined) { @@ -481,6 +482,11 @@ export async function runSubAgent(params: RunSubAgentParams): Promise { }), ] : []), + // Every worker at every nesting depth is created under the same root + // workdirBase (nestedDispatch.getWorkdirBase is threaded through + // unchanged, never rebound to this worker's own dir), so the trace + // reader's search root is that same function. + createReadAgentTraceTool(nd.getWorkdirBase), ]; } diff --git a/src/subagent/trace-reader.test.ts b/src/subagent/trace-reader.test.ts new file mode 100644 index 000000000..ce2981a16 --- /dev/null +++ b/src/subagent/trace-reader.test.ts @@ -0,0 +1,228 @@ +import { describe, test, expect } from "bun:test"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +import { + AgentTraceNotFoundError, + findAgentTraceDir, + listUniqueSubdirs, + readAgentTrace, + MAX_TRACE_ENTRY_LIMIT, + MAX_TRACE_TURN_WINDOW, +} from "./trace-reader.js"; + +function tempDir(): string { + return fs.mkdtempSync(path.join(os.tmpdir(), "trace-reader-")); +} + +function writeTurns(dir: string, turns: unknown[]): void { + fs.mkdirSync(dir, { recursive: true }); + const text = turns.map((t) => JSON.stringify(t)).join("\n") + (turns.length > 0 ? "\n" : ""); + fs.writeFileSync(path.join(dir, "turns.jsonl"), text); +} + +describe("listUniqueSubdirs", () => { + test("a directory containing latest plus its target enumerates the session exactly once", async () => { + const root = tempDir(); + const real = path.join(root, "01234567-89ab-7def-8123-456789abcdef"); + fs.mkdirSync(real); + fs.symlinkSync(path.basename(real), path.join(root, "latest")); + + const entries = await listUniqueSubdirs(root); + expect(entries).toHaveLength(1); + expect(entries[0]!.path).toBe(fs.realpathSync(real)); + }); + + test("two distinct real directories are both listed", async () => { + const root = tempDir(); + fs.mkdirSync(path.join(root, "a")); + fs.mkdirSync(path.join(root, "b")); + const entries = await listUniqueSubdirs(root); + expect(entries).toHaveLength(2); + }); + + test("a broken symlink is skipped, not thrown", async () => { + const root = tempDir(); + fs.symlinkSync(path.join(root, "does-not-exist"), path.join(root, "dangling")); + const entries = await listUniqueSubdirs(root); + expect(entries).toHaveLength(0); + }); + + test("missing directory returns empty rather than throwing", async () => { + const entries = await listUniqueSubdirs(path.join(tempDir(), "nope")); + expect(entries).toHaveLength(0); + }); +}); + +describe("findAgentTraceDir", () => { + test("finds a direct child under root/subagents", async () => { + const root = tempDir(); + const childDir = path.join(root, "subagents", "child-1"); + writeTurns(childDir, []); + const found = await findAgentTraceDir(root, "child-1"); + expect(found).toBe(fs.realpathSync(childDir)); + }); + + test("finds a nested descendant several levels deep", async () => { + const root = tempDir(); + const grandchildDir = path.join(root, "subagents", "child-1", "subagents", "grandchild-1"); + writeTurns(grandchildDir, []); + const found = await findAgentTraceDir(root, "grandchild-1"); + expect(found).toBe(fs.realpathSync(grandchildDir)); + }); + + test("returns null for an unknown id", async () => { + const root = tempDir(); + writeTurns(path.join(root, "subagents", "child-1"), []); + const found = await findAgentTraceDir(root, "does-not-exist"); + expect(found).toBeNull(); + }); + + test("is not confused by a latest symlink alongside the real worker dir", async () => { + const root = tempDir(); + const childDir = path.join(root, "subagents", "child-1"); + writeTurns(childDir, []); + fs.symlinkSync("child-1", path.join(root, "subagents", "latest")); + const found = await findAgentTraceDir(root, "child-1"); + expect(found).toBe(fs.realpathSync(childDir)); + }); +}); + +describe("readAgentTrace", () => { + test("throws a clean error for a missing target", async () => { + const root = tempDir(); + await expect(readAgentTrace(root, "ghost")).rejects.toBeInstanceOf(AgentTraceNotFoundError); + }); + + test("reads turns, tool calls, and tool errors", async () => { + const root = tempDir(); + const childDir = path.join(root, "subagents", "worker-1"); + writeTurns(childDir, [ + { role: "user", content: [{ type: "text", text: "do the thing" }] }, + { + role: "assistant", + content: [{ type: "tool_call", id: "call-1", name: "run_shell", arguments: { cmd: "ls" } }], + }, + { + role: "user", + content: [ + { + type: "tool_result", + callId: "call-1", + content: [{ type: "text", text: "boom" }], + isError: true, + }, + ], + }, + ]); + + const result = await readAgentTrace(root, "worker-1"); + expect(result.totalTurns).toBe(3); + expect(result.entries.map((e) => e.kind)).toEqual(["text", "tool_call", "error"]); + expect(result.entries[2]!.isError).toBe(true); + expect(result.omitted).toBeNull(); + }); + + test("skips a malformed trailing line instead of throwing", async () => { + const root = tempDir(); + const childDir = path.join(root, "subagents", "worker-1"); + fs.mkdirSync(childDir, { recursive: true }); + const good = JSON.stringify({ role: "user", content: [{ type: "text", text: "hi" }] }); + fs.writeFileSync(path.join(childDir, "turns.jsonl"), `${good}\n{"role":"assistant","cont`); + + const result = await readAgentTrace(root, "worker-1"); + expect(result.totalTurns).toBe(1); + expect(result.parseWarnings).toBe(1); + expect(result.entries).toHaveLength(1); + }); + + test("bounds the entry count to the requested limit and reports omission", async () => { + const root = tempDir(); + const childDir = path.join(root, "subagents", "worker-1"); + const turns = Array.from({ length: 5 }, (_, i) => ({ + role: "assistant", + content: [{ type: "text", text: `turn ${i}` }], + })); + writeTurns(childDir, turns); + + const result = await readAgentTrace(root, "worker-1", { limit: 2 }); + expect(result.entries).toHaveLength(2); + expect(result.entriesTruncated).toBe(true); + expect(result.omitted).not.toBeNull(); + expect(result.omitted!.hint.length).toBeGreaterThan(0); + }); + + test("never exceeds the hard entry-limit cap regardless of requested limit", async () => { + const root = tempDir(); + const childDir = path.join(root, "subagents", "worker-1"); + const turns = Array.from({ length: 10 }, (_, i) => ({ + role: "assistant", + content: [{ type: "text", text: `turn ${i}` }], + })); + writeTurns(childDir, turns); + + const result = await readAgentTrace(root, "worker-1", { limit: 1_000_000 }); + expect(result.entries.length).toBeLessThanOrEqual(MAX_TRACE_ENTRY_LIMIT); + }); + + test("never exceeds the hard turn-window cap regardless of requested range", async () => { + const root = tempDir(); + const childDir = path.join(root, "subagents", "worker-1"); + const turns = Array.from({ length: 500 }, (_, i) => ({ + role: "assistant", + content: [{ type: "text", text: `turn ${i}` }], + })); + writeTurns(childDir, turns); + + const result = await readAgentTrace(root, "worker-1", { + fromTurn: 0, + toTurn: 500, + limit: MAX_TRACE_ENTRY_LIMIT, + }); + expect(result.toTurn - result.fromTurn).toBeLessThanOrEqual(MAX_TRACE_TURN_WINDOW); + }); + + test("filters entries by kind", async () => { + const root = tempDir(); + const childDir = path.join(root, "subagents", "worker-1"); + writeTurns(childDir, [ + { + role: "assistant", + content: [ + { type: "thinking", thinking: "hmm" }, + { type: "text", text: "hello" }, + ], + }, + ]); + + const result = await readAgentTrace(root, "worker-1", { kinds: ["text"] }); + expect(result.entries.map((e) => e.kind)).toEqual(["text"]); + }); + + test("truncates an oversized entry body and marks it truncated", async () => { + const root = tempDir(); + const childDir = path.join(root, "subagents", "worker-1"); + writeTurns(childDir, [ + { role: "assistant", content: [{ type: "text", text: "x".repeat(10_000) }] }, + ]); + + const result = await readAgentTrace(root, "worker-1"); + expect(result.entries[0]!.truncated).toBe(true); + expect(result.entries[0]!.content.length).toBeLessThan(10_000); + }); + + test("a partially written trace (worker still running) reads what exists so far", async () => { + const root = tempDir(); + const childDir = path.join(root, "subagents", "worker-1"); + fs.mkdirSync(childDir, { recursive: true }); + fs.writeFileSync( + path.join(childDir, "turns.jsonl"), + `${JSON.stringify({ role: "user", content: [{ type: "text", text: "go" }] })}\n`, + ); + + const result = await readAgentTrace(root, "worker-1"); + expect(result.totalTurns).toBe(1); + expect(result.entries).toHaveLength(1); + }); +}); diff --git a/src/subagent/trace-reader.ts b/src/subagent/trace-reader.ts new file mode 100644 index 000000000..d199c64ba --- /dev/null +++ b/src/subagent/trace-reader.ts @@ -0,0 +1,374 @@ +/** + * On-disk trace reader backing the `read_agent_trace` fleet verb (CL-6951). + * + * Every sub-agent worker writes its full turn history to `turns.jsonl` + * (segmented — see incremental-jsonl.ts) under its own workdir, but nothing + * in the runtime reads it back. That means a cancelled or interrupted + * worker's completed work — everything it did before it stopped — is + * invisible to the orchestrator even though it is sitting on disk. This + * module reads it directly, independent of the in-memory + * SubAgentSessionStore (which a process restart or a killed worker can + * leave with nothing). + * + * Every read here is bounded: a fixed turn window, a fixed entry count, and + * a per-entry character cap, each capped again by a hard maximum regardless + * of what the caller asks for. No argument combination can pull an + * unbounded blob into the parent's context. + */ + +import fs from "node:fs"; +import { readdir, realpath } from "node:fs/promises"; +import path, { join } from "node:path"; + +import { listSegmentFiles } from "../session/incremental-jsonl.js"; + +const TURNS_FILE = "turns.jsonl"; + +export const DEFAULT_TRACE_TURN_WINDOW = 40; +export const MAX_TRACE_TURN_WINDOW = 200; +export const DEFAULT_TRACE_ENTRY_LIMIT = 200; +export const MAX_TRACE_ENTRY_LIMIT = 500; +export const MAX_TRACE_ENTRY_CHARS = 4_000; + +// A pathological or runaway fleet tree should fail the search cheaply rather +// than walk forever; a worker this deep or a fleet this large is itself a +// signal something upstream is wrong. +const MAX_SEARCH_DIRS = 4_000; +const MAX_SEARCH_DEPTH = 16; + +export type TraceEntryKind = "text" | "thinking" | "tool_call" | "tool_result" | "error"; + +export interface TraceEntry { + turn: number; + role: string; + kind: TraceEntryKind; + name?: string; + callId?: string; + isError?: boolean; + content: string; + truncated?: boolean; +} + +export interface TraceOmission { + reason: string; + turnsBefore: number; + turnsAfter: number; + hint: string; +} + +export interface TraceReadResult { + agentId: string; + totalTurns: number; + fromTurn: number; + toTurn: number; + entries: TraceEntry[]; + entriesTruncated: boolean; + parseWarnings: number; + omitted: TraceOmission | null; +} + +export interface TraceReadOptions { + kinds?: readonly TraceEntryKind[]; + fromTurn?: number; + toTurn?: number; + limit?: number; +} + +export class AgentTraceNotFoundError extends Error { + constructor(target: string) { + super( + `No on-disk trace found for agent "${target}". It may not exist, may not have started ` + + "writing turns yet, or may belong to a different fleet than the one you can see.", + ); + this.name = "AgentTraceNotFoundError"; + } +} + +interface DirEntry { + name: string; + path: string; +} + +/** + * Subdirectories of `dir`, symlinks resolved and de-duplicated by real path. + * A `latest`-style symlink pointing at a sibling entry would otherwise be + * visited as a second, distinct directory by a naive `readdir` — every + * enumeration in this module goes through here instead so that case can + * never double-count. `name` is derived from the resolved real path's own + * basename (not the raw dirent name), so a symlink alias and its target + * always report the same canonical name no matter which one `readdir` + * happens to return first. + */ +export async function listUniqueSubdirs(dir: string): Promise { + let entries: fs.Dirent[]; + try { + entries = await readdir(dir, { withFileTypes: true }); + } catch { + return []; + } + const seen = new Set(); + const result: DirEntry[] = []; + for (const entry of entries) { + if (!entry.isDirectory() && !entry.isSymbolicLink()) continue; + const full = join(dir, entry.name); + let real: string; + try { + real = await realpath(full); + } catch { + continue; // broken symlink + } + if (seen.has(real)) continue; + seen.add(real); + result.push({ name: path.basename(real), path: real }); + } + return result; +} + +/** + * Locate the trace directory for `targetId` under `rootWorkdirBase`'s + * `subagents/` tree, at any depth. A shallower match wins over a deeper one + * with the same name (there should never be two, since ids are generated + * uuids, but shallowest-first keeps the search deterministic either way). + */ +export async function findAgentTraceDir( + rootWorkdirBase: string, + targetId: string, +): Promise { + let scanned = 0; + + async function walk(dir: string, depth: number): Promise { + if (depth > MAX_SEARCH_DEPTH) return null; + const children = await listUniqueSubdirs(join(dir, "subagents")); + + for (const child of children) { + scanned += 1; + if (scanned > MAX_SEARCH_DIRS) return null; + if (child.name === targetId) return child.path; + } + for (const child of children) { + const nested = await walk(child.path, depth + 1); + if (nested !== null) return nested; + } + return null; + } + + return walk(rootWorkdirBase, 0); +} + +interface RawTurn { + role: string; + content: unknown[]; +} + +function isRawTurn(value: unknown): value is RawTurn { + return ( + typeof value === "object" && + value !== null && + typeof (value as { role?: unknown }).role === "string" && + Array.isArray((value as { content?: unknown }).content) + ); +} + +/** + * Tolerant line-oriented parse: a torn or malformed line (the file is being + * appended to live while we read it) is skipped, not thrown. Null bytes from + * a stale truncate-past-EOF are stripped first for the same reason + * optimized-context-store.ts strips them on resume. + */ +function parseTurnsTolerant(text: string): { turns: RawTurn[]; warnings: number } { + const cleaned = text.includes("\0") ? text.replaceAll("\0", "") : text; + if (cleaned.length === 0) return { turns: [], warnings: 0 }; + const lines = cleaned.split("\n"); + if (lines[lines.length - 1] === "") lines.pop(); + + const turns: RawTurn[] = []; + let warnings = 0; + for (const line of lines) { + if (line.length === 0) continue; + try { + const raw: unknown = JSON.parse(line); + if (isRawTurn(raw)) turns.push(raw); + else warnings += 1; + } catch { + warnings += 1; + } + } + return { turns, warnings }; +} + +async function readAllTurns(dir: string): Promise<{ turns: RawTurn[]; warnings: number }> { + const segments = await listSegmentFiles(dir, TURNS_FILE); + const turns: RawTurn[] = []; + let warnings = 0; + for (const name of segments) { + let text: string; + try { + text = await fs.promises.readFile(join(dir, name), "utf-8"); + } catch { + continue; + } + const parsed = parseTurnsTolerant(text); + turns.push(...parsed.turns); + warnings += parsed.warnings; + } + return { turns, warnings }; +} + +function truncateContent(text: string): { content: string; truncated: boolean } { + if (text.length <= MAX_TRACE_ENTRY_CHARS) return { content: text, truncated: false }; + return { content: text.slice(0, MAX_TRACE_ENTRY_CHARS), truncated: true }; +} + +function safeStringify(value: unknown): string { + try { + return JSON.stringify(value) ?? ""; + } catch { + return String(value); + } +} + +function toolResultText(content: unknown): string { + if (!Array.isArray(content)) return ""; + return content + .map((block) => { + const b = block as { type?: unknown; text?: unknown }; + if (b?.type === "text" && typeof b.text === "string") return b.text; + return `[${typeof b?.type === "string" ? b.type : "unknown"} block]`; + }) + .join("\n"); +} + +function blockToEntry(turnIndex: number, role: string, block: unknown): TraceEntry | null { + const b = block as { type?: unknown } & Record; + switch (b?.type) { + case "text": { + const { content, truncated } = truncateContent(typeof b.text === "string" ? b.text : ""); + return { turn: turnIndex, role, kind: "text", content, ...(truncated && { truncated }) }; + } + case "thinking": { + const { content, truncated } = truncateContent( + typeof b.thinking === "string" ? b.thinking : "", + ); + return { + turn: turnIndex, + role, + kind: "thinking", + content, + ...(truncated && { truncated }), + }; + } + case "tool_call": { + const { content, truncated } = truncateContent(safeStringify(b.arguments)); + return { + turn: turnIndex, + role, + kind: "tool_call", + ...(typeof b.name === "string" && { name: b.name }), + ...(typeof b.id === "string" && { callId: b.id }), + content, + ...(truncated && { truncated }), + }; + } + case "tool_result": { + const isError = b.isError === true; + const { content, truncated } = truncateContent(toolResultText(b.content)); + return { + turn: turnIndex, + role, + kind: isError ? "error" : "tool_result", + ...(typeof b.callId === "string" && { callId: b.callId }), + isError, + content, + ...(truncated && { truncated }), + }; + } + case "refusal": { + const { content, truncated } = truncateContent(typeof b.reason === "string" ? b.reason : ""); + return { turn: turnIndex, role, kind: "error", content, ...(truncated && { truncated }) }; + } + default: + return null; + } +} + +/** + * Read a bounded slice of one worker's on-disk trace. Defaults to the most + * recent `DEFAULT_TRACE_TURN_WINDOW` turns; every window and entry cap has a + * hard maximum the caller cannot exceed. `omitted` is populated whenever any + * turns or entries were left out, with enough information (turn counts plus + * a concrete hint) to fetch the rest across follow-up calls. + */ +export async function readAgentTrace( + rootWorkdirBase: string, + target: string, + options: TraceReadOptions = {}, +): Promise { + const dir = await findAgentTraceDir(rootWorkdirBase, target); + if (dir === null) throw new AgentTraceNotFoundError(target); + + const { turns, warnings } = await readAllTurns(dir); + const totalTurns = turns.length; + + const toTurn = Math.min(Math.max(options.toTurn ?? totalTurns, 0), totalTurns); + let fromTurn = Math.min( + Math.max(options.fromTurn ?? Math.max(0, toTurn - DEFAULT_TRACE_TURN_WINDOW), 0), + toTurn, + ); + const maxWindow = MAX_TRACE_TURN_WINDOW; + if (toTurn - fromTurn > maxWindow) fromTurn = toTurn - maxWindow; + + const kindsFilter = options.kinds !== undefined ? new Set(options.kinds) : null; + const limit = Math.min( + Math.max(options.limit ?? DEFAULT_TRACE_ENTRY_LIMIT, 1), + MAX_TRACE_ENTRY_LIMIT, + ); + + const entries: TraceEntry[] = []; + let entriesTruncated = false; + let lastReadTurn = fromTurn; + outer: for (let i = fromTurn; i < toTurn; i++) { + lastReadTurn = i; + const turn = turns[i]!; + for (const block of turn.content) { + const entry = blockToEntry(i, turn.role, block); + if (entry === null) continue; + if (kindsFilter !== null && !kindsFilter.has(entry.kind)) continue; + if (entries.length >= limit) { + entriesTruncated = true; + break outer; + } + entries.push(entry); + } + } + // If we stopped mid-window, only turns strictly before lastReadTurn were + // fully read; report the boundary honestly for the resume hint. + const readThrough = entriesTruncated ? lastReadTurn : toTurn; + + const turnsBefore = fromTurn; + const turnsAfter = totalTurns - readThrough; + const omitted: TraceOmission | null = + turnsBefore > 0 || turnsAfter > 0 + ? { + reason: entriesTruncated + ? "entry limit reached before the requested turn range finished reading" + : "turn window bounded to the default/requested range", + turnsBefore, + turnsAfter, + hint: + turnsBefore > 0 + ? `call again with toTurn=${fromTurn} to page backward (totalTurns=${totalTurns})` + : `call again with fromTurn=${readThrough} to page forward (totalTurns=${totalTurns})`, + } + : null; + + return { + agentId: target, + totalTurns, + fromTurn, + toTurn, + entries, + entriesTruncated, + parseWarnings: warnings, + omitted, + }; +} diff --git a/src/subagent/trace-tool.test.ts b/src/subagent/trace-tool.test.ts new file mode 100644 index 000000000..3c373db3d --- /dev/null +++ b/src/subagent/trace-tool.test.ts @@ -0,0 +1,48 @@ +import { describe, test, expect } from "bun:test"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +import { createReadAgentTraceTool } from "./trace-tool.js"; + +function tempDir(): string { + return fs.mkdtempSync(path.join(os.tmpdir(), "trace-tool-")); +} + +function writeTurns(dir: string, turns: unknown[]): void { + fs.mkdirSync(dir, { recursive: true }); + fs.writeFileSync( + path.join(dir, "turns.jsonl"), + turns.map((t) => JSON.stringify(t)).join("\n") + (turns.length > 0 ? "\n" : ""), + ); +} + +describe("createReadAgentTraceTool", () => { + test("rejects a missing target with a clean, non-throwing error result", async () => { + const root = tempDir(); + const tool = createReadAgentTraceTool(() => root); + if (tool.kind !== "string") throw new Error("expected string tool"); + const text = await tool.handler({ target: "ghost" }, new AbortController().signal); + expect(text).toContain("No on-disk trace found"); + }); + + test("rejects a missing required arg without throwing", async () => { + const root = tempDir(); + const tool = createReadAgentTraceTool(() => root); + if (tool.kind !== "string") throw new Error("expected string tool"); + const text = await tool.handler({}, new AbortController().signal); + expect(text).toContain("Error"); + }); + + test("formats turns, tool calls, and truncation info for a real worker", async () => { + const root = tempDir(); + writeTurns(path.join(root, "subagents", "worker-1"), [ + { role: "user", content: [{ type: "text", text: "hello" }] }, + ]); + const tool = createReadAgentTraceTool(() => root); + if (tool.kind !== "string") throw new Error("expected string tool"); + const text = await tool.handler({ target: "worker-1" }, new AbortController().signal); + expect(text).toContain("worker-1"); + expect(text).toContain("hello"); + }); +}); diff --git a/src/subagent/trace-tool.ts b/src/subagent/trace-tool.ts new file mode 100644 index 000000000..474b92447 --- /dev/null +++ b/src/subagent/trace-tool.ts @@ -0,0 +1,128 @@ +/** + * `read_agent_trace` tool (CL-6951): lets an orchestrator or nested + * orchestrator inspect what a worker has actually done on disk — its turns, + * tool calls, and errors — even if the worker is still running or was + * cancelled/interrupted and its in-memory session record is gone. + * + * Only orchestrator tiers may hold this tool; see authority.ts / + * assertTierMayMountFleetVerb for the mount-point gate. Leaves never see it. + */ + +import { stringTool } from "@intx/agent"; +import type { AgentTool } from "@intx/agent"; +import type { ToolDefinition } from "@intx/types/runtime"; +import { type } from "arktype"; + +import { + AgentTraceNotFoundError, + readAgentTrace, + type TraceEntryKind, + type TraceReadResult, +} from "./trace-reader.js"; + +const TRACE_ENTRY_KINDS: readonly TraceEntryKind[] = [ + "text", + "thinking", + "tool_call", + "tool_result", + "error", +]; + +export const readAgentTraceDefinition: ToolDefinition = { + name: "read_agent_trace", + description: + "Read a worker's on-disk trace — its turns, tool calls, and errors — directly from " + + "disk, so you can see what it has actually done even if it is still running, was " + + "cancelled, or was interrupted. Returns a bounded window; a truncated response tells " + + "you what was omitted and how to page for the rest.", + inputSchema: { + type: "object", + properties: { + target: { + type: "string", + description: "Worker id (the id you spawned/see for it) whose trace to read.", + }, + kinds: { + type: "array", + items: { type: "string", enum: [...TRACE_ENTRY_KINDS] }, + description: + "Only return entries of these kinds (text, thinking, tool_call, tool_result, error). Omit for all kinds.", + }, + fromTurn: { + type: "number", + description: "0-based inclusive start turn index. Omit to default to the recent tail.", + }, + toTurn: { + type: "number", + description: "0-based exclusive end turn index. Omit to default to the end of the trace.", + }, + limit: { + type: "number", + description: "Max entries to return (default 200, hard cap 500).", + }, + }, + required: ["target"], + }, +}; + +const ReadAgentTraceArgs = type({ + target: "string", + "kinds?": "('text'|'thinking'|'tool_call'|'tool_result'|'error')[]", + "fromTurn?": "number", + "toTurn?": "number", + "limit?": "number", +}); + +function formatTraceResult(result: TraceReadResult): string { + const lines: string[] = [ + `agent: ${result.agentId}`, + `turns: ${result.fromTurn}-${result.toTurn} of ${result.totalTurns} total`, + ]; + if (result.parseWarnings > 0) { + lines.push(`(skipped ${result.parseWarnings} malformed/partial line(s) while reading)`); + } + if (result.entries.length === 0) { + lines.push("", "No matching entries in this range."); + } else { + lines.push(""); + for (const entry of result.entries) { + const tag = entry.name !== undefined ? `${entry.kind}:${entry.name}` : entry.kind; + const callId = entry.callId !== undefined ? ` [${entry.callId}]` : ""; + const truncatedMark = entry.truncated === true ? " …[truncated]" : ""; + lines.push(`--- turn ${entry.turn} ${entry.role} ${tag}${callId} ---`); + lines.push(`${entry.content}${truncatedMark}`); + } + } + if (result.omitted !== null) { + lines.push( + "", + `[omitted: ${result.omitted.reason}; ${result.omitted.turnsBefore} turn(s) before, ` + + `${result.omitted.turnsAfter} turn(s) after this window — ${result.omitted.hint}]`, + ); + } + return lines.join("\n"); +} + +export function createReadAgentTraceTool(getRootWorkdirBase: () => string): AgentTool { + return stringTool({ + definition: readAgentTraceDefinition, + handler: async (rawArgs: Record): Promise => { + const parsed = ReadAgentTraceArgs(rawArgs); + if (parsed instanceof type.errors) { + return `Error: read_agent_trace requires target (string); ${parsed.summary}`; + } + try { + const result = await readAgentTrace(getRootWorkdirBase(), parsed.target, { + ...(parsed.kinds !== undefined ? { kinds: parsed.kinds } : {}), + ...(parsed.fromTurn !== undefined ? { fromTurn: parsed.fromTurn } : {}), + ...(parsed.toTurn !== undefined ? { toTurn: parsed.toTurn } : {}), + ...(parsed.limit !== undefined ? { limit: parsed.limit } : {}), + }); + return formatTraceResult(result); + } catch (cause) { + if (cause instanceof AgentTraceNotFoundError) return `Error: ${cause.message}`; + throw cause; + } + }, + }); +} From a4380262fe7431d04b337d9a14ae502173e27558 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 23 Aug 2026 22:32:17 -0700 Subject: [PATCH 2/2] Fix cross-subtree trace leak and add an aggregate output cap MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review on #599 found two real issues: - findAgentTraceDir walked the whole (flat, shared) subagents/ tree from the root workdirBase, so a Tier-2 nested orchestrator could read any worker's trace, not just its own descendants. Wired assertCanTargetAgent (authority.ts's first live call site) using the worker's own SubAgentSessionStore id and the store's existing parentSessionId chain, rather than reshaping the on-disk layout — the disk tree is deliberately flat across the whole fleet (shared by worktrees and intervention logs too), so a structural per-subtree root would be a much larger change. To make that id available, run.ts now names a worker's trace directory after its session-store id when one is supplied (task-tool.ts passes it), instead of always minting a fresh disk-only id. - The per-entry, entry-count, and turn-window caps multiply (500 * 4,000 = 2,000,000 chars). Added a total-output character cap that stops filling entries once reached and reports the remainder via the existing `omitted` block. Noted but not changed: readAllTurns loads each full segment before bounds apply. Segments are already bounded to ~256KB by the writer, so this isn't unbounded, but avoiding the read entirely needs a line-count index or a streaming reader — left as a follow-up rather than expanding this fix. --- src/agent/tools.ts | 6 ++- src/subagent/run.ts | 21 +++++++-- src/subagent/task-tool.ts | 4 ++ src/subagent/trace-reader.test.ts | 22 ++++++++++ src/subagent/trace-reader.ts | 43 ++++++++++++++++--- src/subagent/trace-tool.test.ts | 71 +++++++++++++++++++++++++++++++ src/subagent/trace-tool.ts | 47 +++++++++++++++++++- src/subagent/types.ts | 9 ++++ 8 files changed, 210 insertions(+), 13 deletions(-) diff --git a/src/agent/tools.ts b/src/agent/tools.ts index a16b3035a..cf6fcf5ab 100644 --- a/src/agent/tools.ts +++ b/src/agent/tools.ts @@ -310,8 +310,10 @@ export async function createAgentToolset(args: AgentToolsetArgs): Promise { // Every worker at every nesting depth is created under the same root // workdirBase (nestedDispatch.getWorkdirBase is threaded through // unchanged, never rebound to this worker's own dir), so the trace - // reader's search root is that same function. - createReadAgentTraceTool(nd.getWorkdirBase), + // reader's search root is that same function. Descendant-only + // scoping is enforced inside the tool via assertCanTargetAgent, + // reusing the fleet nodes SubAgentSessionStore already tracks and + // this worker's own store id (params.id) — not the disk layout, + // which is intentionally flat across the whole fleet. + createReadAgentTraceTool(nd.getWorkdirBase, { + actorId: params.id, + tier, + getNodes: () => nd.sessions?.list() ?? [], + }), ]; } @@ -584,7 +592,14 @@ export async function runSubAgent(params: RunSubAgentParams): Promise { }, }); - const workdir = join(params.workdirBase, "subagents", generateSessionId()); + // Reuse the caller's session-store id as the on-disk directory name + // when it is safe as a path segment, so read_agent_trace's descendant + // check can walk the same parentSessionId chain SubAgentSessionStore + // already tracks instead of needing a second, disk-only identity + // scheme. + const safeRequestedId = + params.id !== undefined && /^[A-Za-z0-9_-]+$/.test(params.id) ? params.id : undefined; + const workdir = join(params.workdirBase, "subagents", safeRequestedId ?? generateSessionId()); await mkdir(workdir, { recursive: true }); // One record per stop/nudge, with its measured value beside its threshold, // written into this leaf's own trace dir (CL-6938). diff --git a/src/subagent/task-tool.ts b/src/subagent/task-tool.ts index 12f038d98..d1d8707c0 100644 --- a/src/subagent/task-tool.ts +++ b/src/subagent/task-tool.ts @@ -804,6 +804,10 @@ export function createTaskTool(deps: TaskToolDeps): AgentTool { ...sandbox, cwd: worktreeCwd ?? deps.cwd, workdirBase: deps.getWorkdirBase(), + // Same id as the SubAgentSessionStore record so read_agent_trace's + // descendant check (authority.ts assertCanTargetAgent) can reuse the + // store's parentSessionId chain instead of a second identity scheme. + ...(session !== undefined ? { id: session.id } : {}), provider, ...(settings !== undefined ? { settings } : {}), ...(catalog !== undefined ? { catalog } : {}), diff --git a/src/subagent/trace-reader.test.ts b/src/subagent/trace-reader.test.ts index ce2981a16..3519dcc92 100644 --- a/src/subagent/trace-reader.test.ts +++ b/src/subagent/trace-reader.test.ts @@ -9,6 +9,7 @@ import { listUniqueSubdirs, readAgentTrace, MAX_TRACE_ENTRY_LIMIT, + MAX_TRACE_TOTAL_CHARS, MAX_TRACE_TURN_WINDOW, } from "./trace-reader.js"; @@ -153,6 +154,27 @@ describe("readAgentTrace", () => { expect(result.omitted!.hint.length).toBeGreaterThan(0); }); + test("never exceeds the total-output character cap regardless of entry/window caps", async () => { + const root = tempDir(); + const childDir = path.join(root, "subagents", "worker-1"); + const turns = Array.from({ length: 600 }, (_, i) => ({ + role: "assistant", + content: [{ type: "text", text: `turn ${i} `.repeat(1000) }], // ~5,000 chars each + })); + writeTurns(childDir, turns); + + const result = await readAgentTrace(root, "worker-1", { + fromTurn: 0, + toTurn: 600, + limit: MAX_TRACE_ENTRY_LIMIT, + }); + const totalChars = result.entries.reduce((sum, e) => sum + e.content.length, 0); + expect(totalChars).toBeLessThanOrEqual(MAX_TRACE_TOTAL_CHARS); + expect(result.entriesTruncated).toBe(true); + expect(result.omitted).not.toBeNull(); + expect(result.omitted!.reason).toContain("total output cap"); + }); + test("never exceeds the hard entry-limit cap regardless of requested limit", async () => { const root = tempDir(); const childDir = path.join(root, "subagents", "worker-1"); diff --git a/src/subagent/trace-reader.ts b/src/subagent/trace-reader.ts index d199c64ba..431ae36de 100644 --- a/src/subagent/trace-reader.ts +++ b/src/subagent/trace-reader.ts @@ -10,10 +10,11 @@ * SubAgentSessionStore (which a process restart or a killed worker can * leave with nothing). * - * Every read here is bounded: a fixed turn window, a fixed entry count, and - * a per-entry character cap, each capped again by a hard maximum regardless - * of what the caller asks for. No argument combination can pull an - * unbounded blob into the parent's context. + * Every read here is bounded on four independent axes — turn window, entry + * count, per-entry characters, and total output characters (the first three + * multiply, so they are each capped again by a total-output ceiling) — each + * with a hard maximum regardless of what the caller asks for. No argument + * combination can pull an unbounded blob into the parent's context. */ import fs from "node:fs"; @@ -29,6 +30,10 @@ export const MAX_TRACE_TURN_WINDOW = 200; export const DEFAULT_TRACE_ENTRY_LIMIT = 200; export const MAX_TRACE_ENTRY_LIMIT = 500; export const MAX_TRACE_ENTRY_CHARS = 4_000; +// Per-entry/entry-count/turn-window caps each bound one axis, but multiply +// together (500 entries * 4,000 chars = 2,000,000 chars in one call). This +// caps the total regardless of how the other axes are combined. +export const MAX_TRACE_TOTAL_CHARS = 20_000; // A pathological or runaway fleet tree should fail the search cheaply rather // than walk forever; a worker this deep or a fleet this large is itself a @@ -196,6 +201,18 @@ function parseTurnsTolerant(text: string): { turns: RawTurn[]; warnings: number return { turns, warnings }; } +/** + * Reads and parses every segment before the caller's window/limit bounds + * apply, so a not-yet-rotated active segment is loaded whole regardless of + * how small a slice the caller actually wants. In practice each segment is + * itself bounded to ~256KB by the writer (createSegmentedJSONLWriter's + * DEFAULT_MAX_SEGMENT_BYTES), so this cannot grow unboundedly with a + * worker's total history the way reading turns.jsonl as one file could — + * but avoiding this read entirely (only touching the segments the requested + * turn range actually falls in) needs either a cheap line-count index or a + * streaming reader, which is a larger change than this fix; tracked as a + * follow-up rather than expanding this one. + */ async function readAllTurns(dir: string): Promise<{ turns: RawTurn[]; warnings: number }> { const segments = await listSegmentFiles(dir, TURNS_FILE); const turns: RawTurn[] = []; @@ -325,6 +342,8 @@ export async function readAgentTrace( const entries: TraceEntry[] = []; let entriesTruncated = false; + let totalChars = 0; + let stopReason: "entry-limit" | "total-chars" | null = null; let lastReadTurn = fromTurn; outer: for (let i = fromTurn; i < toTurn; i++) { lastReadTurn = i; @@ -335,8 +354,15 @@ export async function readAgentTrace( if (kindsFilter !== null && !kindsFilter.has(entry.kind)) continue; if (entries.length >= limit) { entriesTruncated = true; + stopReason = "entry-limit"; + break outer; + } + if (totalChars + entry.content.length > MAX_TRACE_TOTAL_CHARS) { + entriesTruncated = true; + stopReason = "total-chars"; break outer; } + totalChars += entry.content.length; entries.push(entry); } } @@ -349,9 +375,12 @@ export async function readAgentTrace( const omitted: TraceOmission | null = turnsBefore > 0 || turnsAfter > 0 ? { - reason: entriesTruncated - ? "entry limit reached before the requested turn range finished reading" - : "turn window bounded to the default/requested range", + reason: + stopReason === "entry-limit" + ? "entry limit reached before the requested turn range finished reading" + : stopReason === "total-chars" + ? `total output cap (${MAX_TRACE_TOTAL_CHARS} chars) reached before the requested turn range finished reading` + : "turn window bounded to the default/requested range", turnsBefore, turnsAfter, hint: diff --git a/src/subagent/trace-tool.test.ts b/src/subagent/trace-tool.test.ts index 3c373db3d..184673acc 100644 --- a/src/subagent/trace-tool.test.ts +++ b/src/subagent/trace-tool.test.ts @@ -4,6 +4,7 @@ import os from "node:os"; import path from "node:path"; import { createReadAgentTraceTool } from "./trace-tool.js"; +import type { FleetNode } from "./authority.js"; function tempDir(): string { return fs.mkdtempSync(path.join(os.tmpdir(), "trace-tool-")); @@ -45,4 +46,74 @@ describe("createReadAgentTraceTool", () => { expect(text).toContain("worker-1"); expect(text).toContain("hello"); }); + + describe("descendant-only scoping (two sibling subtrees under one flat root)", () => { + // Every worker at every nesting depth lands under the same root + // subagents/ dir (see run.ts), so on disk workerA1 and workerY are + // indistinguishable siblings. Authority comes entirely from the fleet + // node list (parentSessionId chain), not from directory structure. + const nodes: FleetNode[] = [ + { id: "orchA" }, + { id: "workerA1", parentSessionId: "orchA" }, + { id: "orchB" }, + { id: "workerY", parentSessionId: "orchB" }, + ]; + + function setUpRoot(): string { + const root = tempDir(); + writeTurns(path.join(root, "subagents", "workerA1"), [ + { role: "assistant", content: [{ type: "text", text: "from A1" }] }, + ]); + writeTurns(path.join(root, "subagents", "workerY"), [ + { role: "assistant", content: [{ type: "text", text: "from Y" }] }, + ]); + return root; + } + + test("orchestratorA can read its own descendant workerA1", async () => { + const root = setUpRoot(); + const tool = createReadAgentTraceTool(() => root, { + actorId: "orchA", + tier: "nested-orchestrator", + getNodes: () => nodes, + }); + if (tool.kind !== "string") throw new Error("expected string tool"); + const text = await tool.handler({ target: "workerA1" }, new AbortController().signal); + expect(text).toContain("from A1"); + }); + + test("orchestratorA cannot read workerY, a sibling subtree's worker", async () => { + const root = setUpRoot(); + const tool = createReadAgentTraceTool(() => root, { + actorId: "orchA", + tier: "nested-orchestrator", + getNodes: () => nodes, + }); + if (tool.kind !== "string") throw new Error("expected string tool"); + const text = await tool.handler({ target: "workerY" }, new AbortController().signal); + expect(text).toContain("Error:"); + expect(text).not.toContain("from Y"); + }); + + test("an actor with no resolvable session id is denied entirely", async () => { + const root = setUpRoot(); + const tool = createReadAgentTraceTool(() => root, { + actorId: undefined, + tier: "nested-orchestrator", + getNodes: () => nodes, + }); + if (tool.kind !== "string") throw new Error("expected string tool"); + const text = await tool.handler({ target: "workerA1" }, new AbortController().signal); + expect(text).toContain("Error:"); + expect(text).not.toContain("from A1"); + }); + + test("Tier 1 (no authority context) can read any worker", async () => { + const root = setUpRoot(); + const tool = createReadAgentTraceTool(() => root); + if (tool.kind !== "string") throw new Error("expected string tool"); + const text = await tool.handler({ target: "workerY" }, new AbortController().signal); + expect(text).toContain("from Y"); + }); + }); }); diff --git a/src/subagent/trace-tool.ts b/src/subagent/trace-tool.ts index 474b92447..27b516c5b 100644 --- a/src/subagent/trace-tool.ts +++ b/src/subagent/trace-tool.ts @@ -19,6 +19,27 @@ import { type TraceEntryKind, type TraceReadResult, } from "./trace-reader.js"; +import { + assertCanTargetAgent, + FleetAuthorityError, + type FleetNode, + type SubagentTier, +} from "./authority.js"; + +/** + * Descendant-scoping context for a Tier 2 nested orchestrator's copy of this + * tool. `actorId` is this worker's own SubAgentSessionStore id (the same id + * used as its on-disk directory name — see run.ts) and `getNodes` returns + * the live fleet so `assertCanTargetAgent` can walk the existing + * parentSessionId chain rather than trusting a per-caller check that could + * be forgotten at a future mount site. Omit entirely for Tier 1 (the + * primary orchestrator), which may target anyone. + */ +export interface ReadAgentTraceAuthority { + actorId: string | undefined; + tier: SubagentTier; + getNodes: () => readonly FleetNode[]; +} const TRACE_ENTRY_KINDS: readonly TraceEntryKind[] = [ "text", @@ -103,7 +124,10 @@ function formatTraceResult(result: TraceReadResult): string { return lines.join("\n"); } -export function createReadAgentTraceTool(getRootWorkdirBase: () => string): AgentTool { +export function createReadAgentTraceTool( + getRootWorkdirBase: () => string, + authority?: ReadAgentTraceAuthority, +): AgentTool { return stringTool({ definition: readAgentTraceDefinition, handler: async (rawArgs: Record): Promise => { @@ -111,6 +135,27 @@ export function createReadAgentTraceTool(getRootWorkdirBase: () => string): Agen if (parsed instanceof type.errors) { return `Error: read_agent_trace requires target (string); ${parsed.summary}`; } + if (authority !== undefined) { + // Fails closed: an actor whose own store id could not be resolved + // (no session record for this dispatch) must never be trusted with + // fleet-wide read access, mirroring CL-6941's unresolved-tier rule. + if (authority.actorId === undefined) { + return ( + "Error: read_agent_trace is unavailable for this worker (no resolvable session " + + "id to scope descendant access)." + ); + } + try { + assertCanTargetAgent( + { id: authority.actorId, tier: authority.tier }, + parsed.target, + authority.getNodes(), + ); + } catch (cause) { + if (cause instanceof FleetAuthorityError) return `Error: ${cause.message}`; + throw cause; + } + } try { const result = await readAgentTrace(getRootWorkdirBase(), parsed.target, { ...(parsed.kinds !== undefined ? { kinds: parsed.kinds } : {}), diff --git a/src/subagent/types.ts b/src/subagent/types.ts index b247abbe1..4ce43b3e1 100644 --- a/src/subagent/types.ts +++ b/src/subagent/types.ts @@ -76,6 +76,15 @@ export type NestedDispatchDeps = SubAgentSandboxDeps & { export type RunSubAgentParams = { cwd: string; workdirBase: string; + /** + * Stable id for this worker's on-disk trace directory (subagents/). + * Callers that track a session store (task-tool.ts) pass the same id as + * the SubAgentSessionStore record so read_agent_trace's descendant check + * can reuse the store's existing parentSessionId chain instead of a + * second identity scheme. Falls back to a fresh generated id when unset + * or unsafe for a path segment. + */ + id?: string; provider: SubAgentProvider; settings?: Settings; catalog?: readonly ProviderCatalogEntry[];