From 4f4fe60f09b695469f8eaf42890d8d13d54d5b8b Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Tue, 29 Sep 2026 16:24:07 -0700 Subject: [PATCH] fix(subagent): digest mailbox reports instead of pasting them whole --- docs/ARCHITECTURE.md | 4 +- docs/IMPLEMENTATION.md | 2 +- docs/TUI.md | 4 +- src/subagent/fleet-dry-drive.test.ts | 102 ++++++++++ src/subagent/fleet-dry-drive.ts | 235 +++++++++++++++++++++++- src/subagent/mailbox-mail-drive.test.ts | 122 +++++++++++- src/subagent/mailbox-mail-drive.ts | 60 +++--- 7 files changed, 476 insertions(+), 53 deletions(-) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 8ba7752a3..6c0988f38 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -148,12 +148,12 @@ A worker deny-on-ask registers a harness-owned denied-call envelope (`src/permis Two directors, selected by role: -- **ChatDirector** (interactive, `src/agent/director.ts`) — Extends `DefaultDirector` with task list tracking, workflow nudges, LSP auto-activation, and multi-turn chat semantics. It never terminates the session: operator declines are surfaced as replies and the reactor stays alive for the next message. Yielding while a live fleet is running is allowed (idle-with-fleet); the open-task nudge does not rewrite that wait/reply, and the workflow idle rail does not declare the parent stuck while helpers run. The TUI idle-with-fleet seed alone is not occupancy: an empty fleet still auto-continues slash workflows. Occupancy, mailbox, and operator inbounds reset the workflow idle counter. When a worker finishes or fails while the parent is idle (including idle-with-fleet), occupancy delivers mailbox mail as system inbound so Skywalker starts a new turn without polling `wait_agents`. When the fleet goes dry with tasks still todo/doing, the TUI runtime re-enters the parent with collected worker reports rather than settling idle. +- **ChatDirector** (interactive, `src/agent/director.ts`) — Extends `DefaultDirector` with task list tracking, workflow nudges, LSP auto-activation, and multi-turn chat semantics. It never terminates the session: operator declines are surfaced as replies and the reactor stays alive for the next message. Yielding while a live fleet is running is allowed (idle-with-fleet); the open-task nudge does not rewrite that wait/reply, and the workflow idle rail does not declare the parent stuck while helpers run. The TUI idle-with-fleet seed alone is not occupancy: an empty fleet still auto-continues slash workflows. Occupancy, mailbox, and operator inbounds reset the workflow idle counter. When a worker finishes or fails while the parent is idle (including idle-with-fleet), occupancy delivers mailbox mail as system inbound so Skywalker starts a new turn without polling `wait_agents`. Mailbox mail is a Summary + Blockers digest with a blob pointer to the full report, not the full report JSON. When the fleet goes dry with tasks still todo/doing, the TUI runtime re-enters the parent with collected worker reports rather than settling idle. Already-collected IDs in that continuation are id/status/description only. Auto mode is toggled by CLI flags (`--auto` / `--no-auto`); there is currently no in-session key to toggle it (default on; constrained envelope — workspace writes and unconstrained shell auto-allow; installs, recursive rm, force/uncontained worktree changes, sensitive-path and opaque-wrapper shell still ask; contained non-force `git worktree add`/`remove`/`prune` and `list` auto-allow; shell file-mutation denied). It is not a separate edit/plan mode. - **SubAgentDirector** (delegated work, `src/subagent/index.ts`) — Drives a dispatched worker until a turn arrives with no tool calls, then replies with the final assistant text and ends the run. A tool-less turn **after tools** completes only with the four-heading envelope (Summary, Findings, Blockers, Paths). Assistant text that prints explicit `` markup is treated as attempted tool use, not narration: one **verbatim-tool-call** nudge asks the worker to re-issue a real `tool_call` and does not count toward the tool-less spiral. A missing envelope otherwise nudges once (**incomplete-report**) and a second tool-less turn still without the envelope salvages as **incomplete-report-stop** (once; later tool-less turns wait). Explore/read-only workers that used tools then replied with findings remain normal completes; `requireEvidence` (off by default, set per director) additionally requires at least one read before a tool-less spawn-only reply can complete. `requirePlanSubstance` (counsel or `intent=plan`, not `modelRole === "plan"`) additionally requires Findings to contain files/paths, acceptance criteria, non-goals, risks, and ordered steps with a non-placeholder line each — four headings with stub Findings are incomplete-report, not an attachable plan. After real tool work, wrap-up Findings that are not placeholder or outline-only complete instead of being salvaged as a stub plan. Reads done through `run_shell` count as evidence too — `src/subagent/shell-evidence.ts` classifies shell reads (`cat`, `grep`, `sed` without `-i`, …) over the same subject expansion the auto-shell policy uses — but there is no corresponding shell-write evidence or file-write requirement: a run that never touches a file still completes normally once it replies with the envelope. There is no turn budget. Operator/parent cancel after any progress returns a **cancelled** salvage report (partial findings + tool activity) instead of a bare cancel string; cancel before progress still surfaces as cancelled-by-operator. There is no repetition/no-progress/never-acted/never-edited hard stop and no fingerprint-based re-dispatch block — a genuinely stuck worker runs until it completes, stalls, hits an opt-in wall-clock deadline, or is cancelled. - `spawn_agent` starts each worker and records it in the caller's fleet mailbox. On the TUI primary, mailbox mail is the collect path: occupancy takes uncollected terminals and re-enters the parent as system inbound. Nested orchestrators collect through mailbox mail the same way. A mounted `wait_agents` (exec primary) collects worker reports directly instead. Already-collected waits return status without a second report or error body. Wait JSON includes `stop_reason` from the session when present so a salvage that is wait-`done` is not mistaken for a clean complete, and so parent-initiated interrupt (`interrupted`) is not mistaken for operator-cancel (`cancelled`). Deadline salvage prepends an advisory parent hint suggesting continuation plus a longer deadline if more wall-clock time is warranted. Failed and incomplete-report salvage tell the parent to diagnose from the report or error and MAY spawn one successor with a changed brief. A parent-initiated interrupt is a resumable pause: wait unblocks with `stop_reason: interrupted` (often while the session is still running and has no report); the parent should `resume_agent` or re-wait, and must not spawn a successor against a still-live worker. Successor only if that session is no longer resumable. Operator-cancelled salvage asks the parent to synthesize Findings and Paths and wait for the operator instead of auto-starting another specialist. Identical re-dispatch of the same brief stays refused at the prompt / spawn-handoff layer, except a recoverable/continuable child failure (`continuable: true`) MAY spawn one successor with the same brief; there is no fingerprint-based re-dispatch hard-block. Deadline hints are advisory only — an identical re-dispatch is still admitted at runtime. Parent hints are prepended on salvage reports returned to the parent. The runtime does not auto-spawn successors. + `spawn_agent` starts each worker and records it in the caller's fleet mailbox. On the TUI primary, mailbox mail is the collect path: occupancy takes uncollected terminals and re-enters the parent as system inbound with a Summary + Blockers digest and a blob pointer to the full report. Nested orchestrators collect through mailbox mail the same way. A mounted `wait_agents` (exec primary) collects worker reports directly instead. Already-collected waits return status without a second report or error body. Wait JSON includes `stop_reason` from the session when present so a salvage that is wait-`done` is not mistaken for a clean complete, and so parent-initiated interrupt (`interrupted`) is not mistaken for operator-cancel (`cancelled`). Deadline salvage prepends an advisory parent hint suggesting continuation plus a longer deadline if more wall-clock time is warranted. Failed and incomplete-report salvage tell the parent to diagnose from the report or error and MAY spawn one successor with a changed brief. A parent-initiated interrupt is a resumable pause: wait unblocks with `stop_reason: interrupted` (often while the session is still running and has no report); the parent should `resume_agent` or re-wait, and must not spawn a successor against a still-live worker. Successor only if that session is no longer resumable. Operator-cancelled salvage asks the parent to synthesize Findings and Paths and wait for the operator instead of auto-starting another specialist. Identical re-dispatch of the same brief stays refused at the prompt / spawn-handoff layer, except a recoverable/continuable child failure (`continuable: true`) MAY spawn one successor with the same brief; there is no fingerprint-based re-dispatch hard-block. Deadline hints are advisory only — an identical re-dispatch is still admitted at runtime. Parent hints are prepended on salvage reports returned to the parent. The runtime does not auto-spawn successors. #### Model-family policy (`src/agent/model-family-policy.ts`) diff --git a/docs/IMPLEMENTATION.md b/docs/IMPLEMENTATION.md index 62a1bd75b..429ee35cf 100644 --- a/docs/IMPLEMENTATION.md +++ b/docs/IMPLEMENTATION.md @@ -212,7 +212,7 @@ Listings are list-free, dumps are dump-locked: a bounded `ls`/`tree` prints name - **Alt+Enter** queues a follow-up (kind `"queue"`) delivered only on **session-idle** — parent-idle **and** no live fleet lanes (`run` goes idle). Session-idle Alt+Enter is a no-op. **Ctrl+C** stops the run. -Idle-with-fleet is shipped: after a non-blocking `spawn_agent` dispatch the parent turn can settle while workers keep running. The runner emits a `fleet` event carrying the live-lane count (same liveness rule as the progress strip: interrupted leftovers are not live); the bridge holds the run busy on that count, so mid-hold Enter upgrades to a new primary turn (sent immediately) instead of queueing a steer, follow-ups keep waiting for true session-idle, and any steer left pending at the hold's engagement delivers immediately — the parent it was steering has already stopped. The workflow idle rail honors live helper occupancy, not the TUI idle-with-fleet seed: a live fleet is not a stuck parent, an empty fleet still auto-continues slash workflows, and occupancy / mailbox / operator inbounds reset the workflow idle counter. While the hold is up and the parent is not processing, occupancy flushes mailbox mail (`driveMailboxMail` + `buildMailboxMailMessage`) on store subscribe and idle-with-fleet settle — one child done while siblings run is enough. Skip that shot when a fleet-dry open-task continuation is latched. Both wakes wait for send to settle as accepted before taking reports; a pending Promise is not a continuation, and a failed or uncertain send idles the parent. Exec-primary `wait_agents` mounts with no yield predicate (`createWaitAgentsTool({ sessions, fleetRecords })`) and blocks to ready/timeout/abort; parent interrupt/abort cancels that parked waiter via the tool AbortSignal (`timed_out`, workers stay running) rather than a second waiter table. `interrupt_agent` on a child unblocks wait as interrupted through the mailbox overlay. `shouldYieldWait` remains a supported-but-unwired tool option (no production mount passes it). +Idle-with-fleet is shipped: after a non-blocking `spawn_agent` dispatch the parent turn can settle while workers keep running. The runner emits a `fleet` event carrying the live-lane count (same liveness rule as the progress strip: interrupted leftovers are not live); the bridge holds the run busy on that count, so mid-hold Enter upgrades to a new primary turn (sent immediately) instead of queueing a steer, follow-ups keep waiting for true session-idle, and any steer left pending at the hold's engagement delivers immediately — the parent it was steering has already stopped. The workflow idle rail honors live helper occupancy, not the TUI idle-with-fleet seed: a live fleet is not a stuck parent, an empty fleet still auto-continues slash workflows, and occupancy / mailbox / operator inbounds reset the workflow idle counter. While the hold is up and the parent is not processing, occupancy flushes mailbox mail (`driveMailboxMail` + `buildMailboxMailMessage`) on store subscribe and idle-with-fleet settle — one child done while siblings run is enough. The mail payload is a Summary + Blockers digest with a blob pointer, not the full report JSON; fleet-dry continuation lists already-collected IDs as id/status/description only and does not re-deliver an ID mailbox mail is already sending. Skip that shot when a fleet-dry open-task continuation is latched. Both wakes wait for send to settle as accepted before taking reports; a pending Promise is not a continuation, and a failed or uncertain send idles the parent. Exec-primary `wait_agents` mounts with no yield predicate (`createWaitAgentsTool({ sessions, fleetRecords })`) and blocks to ready/timeout/abort; parent interrupt/abort cancels that parked waiter via the tool AbortSignal (`timed_out`, workers stay running) rather than a second waiter table. `interrupt_agent` on a child unblocks wait as interrupted through the mailbox overlay. `shouldYieldWait` remains a supported-but-unwired tool option (no production mount passes it). `src/tui/stream-event-map.ts` maps reactor events onto the bridge's inbound events, and `src/tui/turn-state.ts` tracks the turn's status. `src/tui/turns-to-blocks.ts` hydrates a resumed session's stored turns into the same content blocks. diff --git a/docs/TUI.md b/docs/TUI.md index f4dab17a3..73c4f8060 100644 --- a/docs/TUI.md +++ b/docs/TUI.md @@ -723,7 +723,9 @@ to idle — unless todo/doing tasks remain, in which case a system continuation starts before the fleet-0 event so the run stays busy and follow-ups wait one more turn. Occupancy wakes consume mailbox reports only after send settles as accepted; a failed or uncertain send idles the parent -so a later flush can retry. The mail and that fleet-dry continuation +so a later flush can retry. Mailbox mail is a Summary + Blockers digest with +a blob pointer to the full report. Fleet-dry continuation lists +already-collected IDs as id/status/description only. The mail and that fleet-dry continuation are runtime-to-agent traffic — the fleet board owns worker status — so neither paints a transcript row, and neither rehydrates as one. diff --git a/src/subagent/fleet-dry-drive.test.ts b/src/subagent/fleet-dry-drive.test.ts index 8d601f86b..27103e9e0 100644 --- a/src/subagent/fleet-dry-drive.test.ts +++ b/src/subagent/fleet-dry-drive.test.ts @@ -12,6 +12,7 @@ import { shouldDriveOpenTasks, type FleetDryMailboxRecord, } from "./fleet-dry-drive.js"; +import { driveMailboxMail } from "./mailbox-mail-drive.js"; import { createSubAgentSessionStore } from "./session-store.js"; import { ACCEPTED_DELIVERY, @@ -497,6 +498,107 @@ describe("driveOpenTasksAfterFleetDry", () => { expect(records.get("w1")?.collected).toBe(true); }); + test("already-collected IDs are id, status, and description only", async () => { + const records = recordsOf({ + prior: { + status: "done", + report: "SECRET_PRIOR_REPORT", + description: "already mailed", + collected: true, + }, + fresh: { + status: "done", + report: "fresh report", + description: "new lane", + }, + }); + const sent: string[] = []; + const driven = await driveOpenTasksAfterFleetDry({ + previousRunning: 1, + running: 0, + openTasks: [openTask], + ...driveFixture(records, { + send: (prompt) => { + sent.push(prompt); + return ACCEPTED_DELIVERY; + }, + }), + }); + expect(driven).toBe(true); + const parsed = reportsJSONFromPrompt(sent[0] ?? "") as { + agent_id: string; + status: string; + description?: string; + report?: string; + }[]; + expect(parsed).toEqual([ + { + agent_id: "fresh", + status: "done", + description: "new lane", + report: "fresh report", + }, + { + agent_id: "prior", + status: "done", + description: "already mailed", + }, + ]); + expect(sent[0]).not.toContain("SECRET_PRIOR_REPORT"); + expect(records.get("fresh")?.collected).toBe(true); + expect(records.get("prior")?.collected).toBe(true); + }); + + test("omits IDs mailbox mail is already delivering in the same window", async () => { + const records = recordsOf({ + w1: { + status: "done", + report: "SECRET_MAILBOX_BODY", + description: "lane", + }, + }); + const mailbox = collectingMailbox(records); + let resolveSend: ((result: typeof ACCEPTED_DELIVERY) => void) | undefined; + let sendStarted: (() => void) | undefined; + const sendSeen = new Promise((resolve) => { + sendStarted = resolve; + }); + const mailboxDrive = driveMailboxMail({ + parentProcessing: false, + mailbox, + lanes: [], + beginSystemContinuation: () => undefined, + send: () => { + sendStarted?.(); + return new Promise((resolve) => { + resolveSend = resolve; + }); + }, + }); + await sendSeen; + const drySent: string[] = []; + const dryDriven = await driveOpenTasksAfterFleetDry({ + previousRunning: 1, + running: 0, + openTasks: [openTask], + parentProcessing: false, + mailbox, + lanes: [], + beginSystemContinuation: () => undefined, + send: (prompt) => { + drySent.push(prompt); + return ACCEPTED_DELIVERY; + }, + }); + expect(dryDriven).toBe(true); + expect(reportsJSONFromPrompt(drySent[0] ?? "")).toEqual([]); + expect(drySent[0]).not.toContain("SECRET_MAILBOX_BODY"); + expect(records.get("w1")?.collected).not.toBe(true); + resolveSend?.(ACCEPTED_DELIVERY); + expect(await mailboxDrive).toBe(true); + expect(records.get("w1")?.collected).toBe(true); + }); + test("dry+open continuation JSON includes a spill URI for oversized reports", async () => { const original = `head-${"x".repeat(FLEET_DRY_REPORT_CHARS)}TAIL-MARKER`; const records = recordsOf({ diff --git a/src/subagent/fleet-dry-drive.ts b/src/subagent/fleet-dry-drive.ts index 06c3ef860..9ea3c8983 100644 --- a/src/subagent/fleet-dry-drive.ts +++ b/src/subagent/fleet-dry-drive.ts @@ -10,10 +10,14 @@ import { truncateWithReservedNotice, } from "../plugins/result-truncation-plugin.js"; import { isLiveWaitStatus, type WaitJSONStatus } from "./lifecycle.js"; +import { parseSubAgentReport } from "./report.js"; /** Enough of a lane report for a parent continuation; traces stay on disk. */ export const FLEET_DRY_REPORT_CHARS = 8_192; +/** Summary/Blockers inline in mailbox digest; the blob holds the rest. */ +export const MAILBOX_DIGEST_SECTION_CHARS = 2_048; + export const FLEET_DRY_CONTINUATION_PREFIX = "The fleet has gone dry. Remaining open tasks:"; @@ -45,6 +49,49 @@ export interface FleetDryMailbox { take(id: string): FleetDryMailboxRecord | undefined; } +/** + * Ids occupancy has snapshotted and handed to send, but not yet taken. + * Shared by mailbox mail and fleet-dry so one parent window cannot paste the + * same agent twice. Failed send clears the set so a later flush can retry. + * Weak-keyed so a mailbox object can go away without a leak. + */ +const occupancyDeliveringByMailbox = new WeakMap< + FleetDryMailbox, + Set +>(); + +export function occupancyDeliveringSet(mailbox: FleetDryMailbox): Set { + let ids = occupancyDeliveringByMailbox.get(mailbox); + if (ids === undefined) { + ids = new Set(); + occupancyDeliveringByMailbox.set(mailbox, ids); + } + return ids; +} + +export function releaseOccupancyDelivering( + mailbox: FleetDryMailbox | undefined, + ids: readonly string[], +): void { + if (mailbox === undefined) return; + const delivering = occupancyDeliveringByMailbox.get(mailbox); + if (delivering === undefined) return; + for (const id of ids) delivering.delete(id); +} + +export function dedupeByAgentId( + reports: readonly T[], +): T[] { + const seen = new Set(); + const out: T[] = []; + for (const report of reports) { + if (seen.has(report.agent_id)) continue; + seen.add(report.agent_id); + out.push(report); + } + return out; +} + export interface FleetDryLane { readonly id: string; readonly description?: string; @@ -70,6 +117,23 @@ export interface CollectedWorkerReport { stop_reason?: string; } +/** Mailbox parent payload: envelope digest plus a blob pointer, not the full report. */ +export interface MailboxWorkerDigest { + agent_id: string; + status: string; + description?: string; + summary?: string; + blockers?: string; + report_uri?: string; + error?: string; + error_uri?: string; + hint?: string; + provider_failure?: true; + continuable?: true; + continue_with?: string; + stop_reason?: string; +} + /** * Single-successor guidance for a failed+continuable entry. The marker is * advisory only — no runtime auto-retry backs it. @@ -140,6 +204,137 @@ export function fleetDrySpillKey( return `fleet-dry:${agentId}:${field}`; } +function clipDigestSection(text: string): string { + if (text.length <= MAILBOX_DIGEST_SECTION_CHARS) return text; + return `${text.slice(0, MAILBOX_DIGEST_SECTION_CHARS - 1).trimEnd()}…`; +} + +export async function spillWorkerField( + text: string | undefined, + agentId: string, + field: "report" | "error", + writeBlob?: FleetDryBlobWriter, +): Promise { + if (text === undefined || writeBlob === undefined) return undefined; + const key = fleetDrySpillKey(agentId, field); + try { + await writeBlob(key, new TextEncoder().encode(text), "text/plain"); + return `tool-output:///${key}`; + } catch { + return undefined; + } +} + +export function slimCollectedReport( + report: CollectedWorkerReport, +): Pick { + return { + agent_id: report.agent_id, + status: report.status, + ...(report.description !== undefined && report.description.length > 0 + ? { description: report.description } + : {}), + }; +} + +export async function digestCollectedReport( + report: CollectedWorkerReport, + writeBlob?: FleetDryBlobWriter, +): Promise { + const parsed = + report.report !== undefined + ? parseSubAgentReport(report.report) + : undefined; + const reportUri = await spillWorkerField( + report.report, + report.agent_id, + "report", + writeBlob, + ); + const errorUri = await spillWorkerField( + report.error, + report.agent_id, + "error", + writeBlob, + ); + const summary = + parsed !== undefined && parsed.summary.length > 0 + ? clipDigestSection(parsed.summary) + : undefined; + const blockers = + parsed !== undefined + ? clipDigestSection( + parsed.blockers.length > 0 ? parsed.blockers : "None.", + ) + : undefined; + return { + agent_id: report.agent_id, + status: report.status, + ...(report.description !== undefined && report.description.length > 0 + ? { description: report.description } + : {}), + ...(summary !== undefined ? { summary } : {}), + ...(blockers !== undefined ? { blockers } : {}), + ...(reportUri !== undefined ? { report_uri: reportUri } : {}), + ...(report.error !== undefined + ? { error: clipDigestSection(report.error) } + : {}), + ...(errorUri !== undefined ? { error_uri: errorUri } : {}), + ...(report.hint !== undefined ? { hint: report.hint } : {}), + ...(report.provider_failure === true ? { provider_failure: true } : {}), + ...(report.continuable === true + ? { + continuable: true as const, + ...(report.continue_with !== undefined + ? { continue_with: report.continue_with } + : {}), + } + : {}), + ...(report.stop_reason !== undefined + ? { stop_reason: report.stop_reason } + : {}), + }; +} + +export async function digestCollectedReports( + reports: readonly CollectedWorkerReport[], + writeBlob?: FleetDryBlobWriter, +): Promise { + const out: MailboxWorkerDigest[] = []; + for (const report of dedupeByAgentId(reports)) { + out.push(await digestCollectedReport(report, writeBlob)); + } + return out; +} + +export function collectAlreadyCollectedStubs( + mailbox: FleetDryMailbox | undefined, + lanes: readonly FleetDryLane[], +): CollectedWorkerReport[] { + if (mailbox === undefined) return []; + const byId = new Map(lanes.map((lane) => [lane.id, lane])); + const stubs: CollectedWorkerReport[] = []; + const seen = new Set(); + for (const id of mailbox.ids()) { + if (seen.has(id)) continue; + seen.add(id); + const peeked = mailbox.peek(id); + if (peeked === undefined || peeked.collected !== true) continue; + if (isLiveWaitStatus(peeked.status)) continue; + const description = peeked.description ?? byId.get(id)?.description; + stubs.push( + slimCollectedReport({ + agent_id: id, + status: peeked.status, + ...(description !== undefined && description.length > 0 + ? { description } + : {}), + }), + ); + } + return stubs; +} + async function clipField( text: string | undefined, agentId: string, @@ -244,11 +439,15 @@ export async function collectUncollectedTerminals( lanes: readonly FleetDryLane[], consume: boolean, writeBlob?: FleetDryBlobWriter, + clip = true, ): Promise { if (mailbox === undefined) return []; const byId = new Map(lanes.map((lane) => [lane.id, lane])); const reports: CollectedWorkerReport[] = []; + const seen = new Set(); for (const id of mailbox.ids()) { + if (seen.has(id)) continue; + seen.add(id); const peeked = mailbox.peek(id); if (peeked === undefined) continue; if (peeked.collected === true) continue; @@ -257,7 +456,9 @@ export async function collectUncollectedTerminals( ? takeAndProjectMailboxRecord(mailbox, id, byId.get(id)) : projectMailboxRecord(id, peeked, byId.get(id)); if (projected === undefined) continue; - reports.push(await clipCollectedReport(projected, writeBlob)); + reports.push( + clip ? await clipCollectedReport(projected, writeBlob) : projected, + ); } return reports; } @@ -277,7 +478,7 @@ export function buildFleetDryContinuationPrompt( taskLines, "", "Collected worker reports (already collected — do not call wait_agents for these agent_ids):", - JSON.stringify(reports), + JSON.stringify(dedupeByAgentId(reports)), "", "Continue the remaining work. Mark each task done or cancelled with manage_tasks", "when finished, or spawn_agent the next specialist. Do not end this turn while", @@ -317,19 +518,35 @@ async function driveOpenTasksAfterFleetDrySpill( args: Parameters[0], tasks: Task[], ): Promise { - const reports = await collectUncollectedTerminals( - args.mailbox, - args.lanes, - false, - args.writeBlob, + const delivering = + args.mailbox !== undefined + ? occupancyDeliveringSet(args.mailbox) + : new Set(); + const uncollected = ( + await collectUncollectedTerminals( + args.mailbox, + args.lanes, + false, + args.writeBlob, + ) + ).filter((report) => !delivering.has(report.agent_id)); + const uncollectedIds = new Set(uncollected.map((report) => report.agent_id)); + const stubs = collectAlreadyCollectedStubs(args.mailbox, args.lanes).filter( + (stub) => + !uncollectedIds.has(stub.agent_id) && !delivering.has(stub.agent_id), ); + const reports = dedupeByAgentId([...uncollected, ...stubs]); + const takeIds = uncollected.map((report) => report.agent_id); + for (const id of takeIds) delivering.add(id); const prompt = buildFleetDryContinuationPrompt(tasks, reports); const takeReports = (): void => { - for (const report of reports) { - args.mailbox?.take(report.agent_id); + for (const id of takeIds) { + args.mailbox?.take(id); } + releaseOccupancyDelivering(args.mailbox, takeIds); }; const fail = (): boolean => { + releaseOccupancyDelivering(args.mailbox, takeIds); args.onSendFailure?.(); return false; }; diff --git a/src/subagent/mailbox-mail-drive.test.ts b/src/subagent/mailbox-mail-drive.test.ts index c4586712b..e4517af7a 100644 --- a/src/subagent/mailbox-mail-drive.test.ts +++ b/src/subagent/mailbox-mail-drive.test.ts @@ -9,7 +9,11 @@ import { occupancyShouldYieldWait, } from "./mailbox-mail-drive.js"; import { createSubAgentSessionStore } from "./session-store.js"; -import type { FleetDryMailboxRecord } from "./fleet-dry-drive.js"; +import { + digestCollectedReports, + fleetDrySpillKey, + type FleetDryMailboxRecord, +} from "./fleet-dry-drive.js"; import { ACCEPTED_DELIVERY, collectingMailbox, @@ -25,6 +29,41 @@ function recordsOf( return new Map(Object.entries(entries)); } +function fakeBlobStore() { + const blobs = new Map(); + return { + blobs, + writeBlob: (key: string, bytes: Uint8Array, contentType: string) => { + blobs.set(key, { bytes, contentType }); + }, + }; +} + +function mailboxReportsFromPrompt(prompt: string): Record[] { + const line = mailboxMailWakeLine(); + const start = prompt.indexOf(line); + expect(start).toBeGreaterThanOrEqual(0); + const jsonStart = start + line.length + 1; + const jsonEnd = prompt.indexOf("\n", jsonStart); + return JSON.parse( + prompt.slice(jsonStart, jsonEnd === -1 ? undefined : jsonEnd), + ) as Record[]; +} + +const ENVELOPE_REPORT = [ + "## Summary", + "Shipped the digest.", + "", + "## Findings", + "SECRET_FINDINGS_BODY", + "", + "## Blockers", + "Need a follow-up.", + "", + "## Paths", + "src/subagent/mailbox-mail-drive.ts", +].join("\n"); + const NOOP_DRIVE = { beginSystemContinuation: () => { throw new Error("must not begin"); @@ -64,6 +103,50 @@ describe("buildMailboxMailPrompt", () => { expect(prompt).toContain(mailboxMailWakeLine()); expect(prompt).not.toContain("already collected"); }); + + test("does not paste a duplicate agent_id", () => { + const prompt = buildMailboxMailPrompt([ + { agent_id: "w1", status: "done" }, + { agent_id: "w1", status: "failed" }, + ]); + expect(prompt.match(/"agent_id":"w1"/g)?.length).toBe(1); + }); +}); + +describe("digestCollectedReports", () => { + test("keeps Summary and Blockers and spills the full report", async () => { + const store = fakeBlobStore(); + const [digest] = await digestCollectedReports( + [ + { + agent_id: "worker-1", + status: "done", + description: "lane", + report: ENVELOPE_REPORT, + }, + { + agent_id: "worker-1", + status: "done", + report: "duplicate must drop", + }, + ], + store.writeBlob, + ); + expect(digest).toEqual({ + agent_id: "worker-1", + status: "done", + description: "lane", + summary: "Shipped the digest.", + blockers: "Need a follow-up.", + report_uri: `tool-output:///${fleetDrySpillKey("worker-1", "report")}`, + }); + expect( + new TextDecoder().decode( + store.blobs.get(fleetDrySpillKey("worker-1", "report"))?.bytes ?? + new Uint8Array(), + ), + ).toBe(ENVELOPE_REPORT); + }); }); describe("driveMailboxMail", () => { @@ -105,6 +188,43 @@ describe("driveMailboxMail", () => { expect(records.get("fail")?.collected).toBe(true); }); + test("delivers a digest and blob pointer instead of the full report JSON", async () => { + const records = recordsOf({ + done: { status: "done", report: ENVELOPE_REPORT, description: "lane" }, + }); + const store = fakeBlobStore(); + const sent: string[] = []; + const driven = await driveMailboxMail({ + ...driveFixture(records, { + send: (prompt) => { + sent.push(prompt); + return ACCEPTED_DELIVERY; + }, + }), + writeBlob: store.writeBlob, + }); + expect(driven).toBe(true); + const parsed = mailboxReportsFromPrompt(sent[0] ?? ""); + expect(parsed).toEqual([ + { + agent_id: "done", + status: "done", + description: "lane", + summary: "Shipped the digest.", + blockers: "Need a follow-up.", + report_uri: `tool-output:///${fleetDrySpillKey("done", "report")}`, + }, + ]); + expect(sent[0]).not.toContain("SECRET_FINDINGS_BODY"); + expect(sent[0]).not.toContain('"report":'); + expect( + new TextDecoder().decode( + store.blobs.get(fleetDrySpillKey("done", "report"))?.bytes ?? + new Uint8Array(), + ), + ).toBe(ENVELOPE_REPORT); + }); + test("parentProcessing or empty mailbox is a no-op", async () => { const records = recordsOf({ done: { status: "done", report: "ok" }, diff --git a/src/subagent/mailbox-mail-drive.ts b/src/subagent/mailbox-mail-drive.ts index 7fe018b55..a73205b1b 100644 --- a/src/subagent/mailbox-mail-drive.ts +++ b/src/subagent/mailbox-mail-drive.ts @@ -8,8 +8,11 @@ import { isLiveWaitStatus } from "./lifecycle.js"; import { collectUncollectedTerminals, + dedupeByAgentId, + digestCollectedReports, + occupancyDeliveringSet, + releaseOccupancyDelivering, settleOccupancySend, - type CollectedWorkerReport, type FleetDryBlobWriter, type FleetDryLane, type FleetDryMailbox, @@ -30,7 +33,7 @@ export function mailboxMailWakeLine(): string { /** * Whether inbound text is occupancy's mailbox mail. Internal runtime→agent * traffic — the fleet board already owns worker status and the payload is - * model-facing report JSON, so the transcript never paints it. The live event + * a model-facing digest, so the transcript never paints it. The live event * map recognises it by content; history hydration keys on the persisted * origin marker instead (see isPersistedOccupancyWakeText). */ @@ -62,33 +65,6 @@ export function isPersistedOccupancyWakeText(text: string): boolean { return isMailboxMailText(bare) || isFleetDryContinuationText(bare); } -/** - * Ids occupancy has snapshotted and handed to send, but not yet taken. - * A second flush must not start another parent turn for the same reports. - * Failed send clears the set so a later flush can retry. Weak-keyed so a - * mailbox object can go away without a leak. - */ -const deliveringByMailbox = new WeakMap>(); - -function deliveringSet(mailbox: FleetDryMailbox): Set { - let ids = deliveringByMailbox.get(mailbox); - if (ids === undefined) { - ids = new Set(); - deliveringByMailbox.set(mailbox, ids); - } - return ids; -} - -function releaseDelivering( - mailbox: FleetDryMailbox | undefined, - ids: readonly string[], -): void { - if (mailbox === undefined) return; - const delivering = deliveringByMailbox.get(mailbox); - if (delivering === undefined) return; - for (const id of ids) delivering.delete(id); -} - export function occupancyShouldYieldWait( mailbox: FleetDryMailbox | undefined, ): boolean { @@ -103,19 +79,21 @@ export function occupancyShouldYieldWait( return false; } -export function buildMailboxMailPrompt( - reports: readonly CollectedWorkerReport[], +export function buildMailboxMailPrompt( + reports: readonly T[], ): string { - return [mailboxMailWakeLine(), JSON.stringify(reports)].join("\n"); + return [mailboxMailWakeLine(), JSON.stringify(dedupeByAgentId(reports))].join( + "\n", + ); } function hasDeliverableMailboxMail( mailbox: FleetDryMailbox | undefined, ): boolean { if (mailbox === undefined) return false; - const delivering = deliveringByMailbox.get(mailbox); + const delivering = occupancyDeliveringSet(mailbox); for (const id of mailbox.ids()) { - if (delivering?.has(id)) continue; + if (delivering.has(id)) continue; const record = mailbox.peek(id); if (record === undefined || record.collected === true) continue; if (isLiveWaitStatus(record.status)) continue; @@ -152,33 +130,37 @@ async function driveMailboxMailAfterCollect( ): Promise { const delivering = args.mailbox !== undefined - ? deliveringSet(args.mailbox) + ? occupancyDeliveringSet(args.mailbox) : new Set(); const reports = ( await collectUncollectedTerminals( args.mailbox, args.lanes, false, - args.writeBlob, + undefined, + false, ) ).filter((report) => !delivering.has(report.agent_id)); if (reports.length === 0) return false; if (parentIsProcessing(args)) return false; const ids = reports.map((report) => report.agent_id); for (const id of ids) delivering.add(id); - const prompt = buildMailboxMailPrompt(reports); const takeReports = (): void => { for (const id of ids) { args.mailbox?.take(id); } - releaseDelivering(args.mailbox, ids); + releaseOccupancyDelivering(args.mailbox, ids); }; const fail = (): boolean => { - releaseDelivering(args.mailbox, ids); + releaseOccupancyDelivering(args.mailbox, ids); args.onSendFailure?.(); return false; }; + let prompt: string; try { + prompt = buildMailboxMailPrompt( + await digestCollectedReports(reports, args.writeBlob), + ); args.beginSystemContinuation(prompt); } catch { return fail();