diff --git a/src/main/acp/acp-background-task-timeline.ts b/src/main/acp/acp-background-task-timeline.ts index c3aa3f51f00..5b2a3c2f9f9 100644 --- a/src/main/acp/acp-background-task-timeline.ts +++ b/src/main/acp/acp-background-task-timeline.ts @@ -1,5 +1,8 @@ import { BoundedMap } from '../../shared/bounded-map' -import { backgroundTaskJournalBody } from '../../shared/native-chat-background-task-row' +import { + backgroundTaskJournalBody, + isSettledBackgroundTaskState +} from '../../shared/native-chat-background-task-row' import type { NativeChatBackgroundTaskBlock } from '../../shared/native-chat-types' import { boundInlineText, @@ -11,11 +14,12 @@ import type { } from '../native-chat/agent-session-timeline/provider-timeline-event' import type { AcpBackgroundTaskUpdate } from './acp-dialects/acp-dialect' -/** Partial provider snapshots only; the assembler owns settlement and durable placement. A task - * the bound evicted (128 others updated since) loses its earlier label, parent tool and output - * file on its next update; its row keeps the turn it was placed in. */ +/** Partial provider snapshots; the assembler owns settlement and durable placement. A running + * task's snapshot lives until it settles. Settled ones are history, bounded, so a later ending + * (after a Stop found it gone) keeps its label; one evicted keeps only its row's turn. */ export class AcpBackgroundTaskTimeline { - private readonly snapshots = new BoundedMap({ + private readonly running = new Map() + private readonly settled = new BoundedMap({ maxEntries: 128, maxBytes: 1024 * 1024, sizeOf: (block, key) => Buffer.byteLength(key) + Buffer.byteLength(JSON.stringify(block)) @@ -29,7 +33,7 @@ export class AcpBackgroundTaskTimeline { ): ProviderTimelineEvent[] { return updates.map((update) => { const { fallbackLabel, fallbackKind, ...fields } = update - const previous = this.snapshots.get(update.taskId) + const previous = this.running.get(update.taskId) ?? this.settled.get(update.taskId) const block: NativeChatBackgroundTaskBlock = { type: 'background-task', ...previous, @@ -44,7 +48,13 @@ export class AcpBackgroundTaskTimeline { block[key] = boundInlineText(text, DEFAULT_JOURNAL_PAYLOAD_LIMITS).text } } - this.snapshots.set(block.taskId, block) + if (isSettledBackgroundTaskState(block.state)) { + this.running.delete(block.taskId) + this.settled.set(block.taskId, block) + } else { + this.settled.delete(block.taskId) + this.running.set(block.taskId, block) + } const turn = block.parentToolUseId ? this.toolTurn(block.parentToolUseId) : undefined return { type: 'item.update', diff --git a/src/main/acp/acp-subagents.ts b/src/main/acp/acp-subagents.ts index a8ce14c9a55..e4b53b2f439 100644 --- a/src/main/acp/acp-subagents.ts +++ b/src/main/acp/acp-subagents.ts @@ -6,7 +6,6 @@ // not known here. import { createHash } from 'node:crypto' -import { BoundedMap } from '../../shared/bounded-map' import type { AgentChildWorkOutcome } from '../../shared/agent-status-child-work' import type { AgentChildWorkEvidence, @@ -24,9 +23,6 @@ import type { ProviderTimelineJoin } from '../native-chat/agent-session-timeline import type { AcpSubagentUpdate } from './acp-dialects/acp-dialect' import type { AcpTimelineEvent } from './acp-timeline-event' -/** Subagents remembered, so a later report about one is still known as a subagent's. */ -const MAX_KNOWN_SUBAGENTS = 2048 - type KnownSubagent = { /** Where its reply row goes: the turn that spawned it. */ placement: ProviderTimelineJoin @@ -54,9 +50,8 @@ export function acpSubagentOutcome( } export class AcpSubagents { - private readonly known = new BoundedMap({ - maxEntries: MAX_KNOWN_SUBAGENTS - }) + /** Every subagent this provider child reported, so a later report is still known as one's. */ + private readonly known = new Map() private readonly pending: AgentChildWorkEvidence[] = [] /** Known children include settled ones, so they cannot become background tasks. */ diff --git a/src/main/acp/acp-timeline-background-tasks.test.ts b/src/main/acp/acp-timeline-background-tasks.test.ts index 798e96cdaef..27808c487b6 100644 --- a/src/main/acp/acp-timeline-background-tasks.test.ts +++ b/src/main/acp/acp-timeline-background-tasks.test.ts @@ -141,7 +141,7 @@ describe('Grok background tasks through the shared timeline', () => { } ) - it('keeps the row and placement of a task whose snapshot the bound evicted', async () => { + it('keeps the label and placement of a running task beside 130 other running tasks', async () => { const fixture = await openAcpFixtureRig() await fixture.feed(await runningFrames()) const [before] = await taskRows(fixture) @@ -156,7 +156,11 @@ describe('Grok background tasks through the shared timeline', () => { const original = (await taskRows(fixture)).find((task) => task.block.taskId === taskId) expect(original?.row.itemId).toBe(before?.row.itemId) expect(original?.row.turnScope).toEqual(before?.row.turnScope) - expect(original?.block.state).toBe('done') + expect(original?.block).toMatchObject({ + state: 'done', + label: before?.block.label, + parentToolUseId: 'call-1' + }) }) it('places a first task notice beside its original tool after its turn ended and another began', async () => { diff --git a/src/main/agent-hooks/server-claude-child-permission-lifecycle.test.ts b/src/main/agent-hooks/server-claude-child-permission-lifecycle.test.ts index e540598f1a1..18a38aef82e 100644 --- a/src/main/agent-hooks/server-claude-child-permission-lifecycle.test.ts +++ b/src/main/agent-hooks/server-claude-child-permission-lifecycle.test.ts @@ -1,5 +1,4 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -import { AGENT_STATUS_MAX_SUBAGENTS } from '../../shared/agent-status-types' import { AgentHookServer, _internals } from './server' import { buildBody, PANE } from './server.test-fixtures' @@ -210,33 +209,4 @@ describe('Claude child permission lifecycle', () => { server.stop() } }) - - it('clears an omitted teammate permission when a full roster excludes its row', async () => { - const { server, postClaudeHook } = await createServer() - try { - await postClaudeHook({ hook_event_name: 'UserPromptSubmit', prompt: 'guarded task' }) - for (let index = 0; index < AGENT_STATUS_MAX_SUBAGENTS; index += 1) { - await postClaudeHook({ hook_event_name: 'SubagentStart', agent_id: `acapped${index}` }) - } - await postClaudeHook({ - hook_event_name: 'PermissionRequest', - agent_id: 'areviewer-6d3cb5b5', - agent_type: 'reviewer', - tool_name: 'Bash', - tool_input: { command: 'false' } - }) - expect(server.getStatusSnapshot()[0]).toMatchObject({ state: 'waiting', toolName: 'Bash' }) - expect(server.getStatusSnapshot()[0]?.subagents).toHaveLength(AGENT_STATUS_MAX_SUBAGENTS) - - await postClaudeHook({ hook_event_name: 'TeammateIdle', teammate_name: 'reviewer' }) - - const status = server.getStatusSnapshot()[0] - expect(status).toMatchObject({ paneKey: PANE, state: 'working', agentType: 'claude' }) - expect(status?.toolName).toBeUndefined() - expect(status?.toolInput).toBeUndefined() - expect(status?.subagents).toHaveLength(AGENT_STATUS_MAX_SUBAGENTS) - } finally { - server.stop() - } - }) }) diff --git a/src/main/agent-hooks/server-nested-agent-roster.test.ts b/src/main/agent-hooks/server-nested-agent-roster.test.ts index 238108f17cd..c015197b7d6 100644 --- a/src/main/agent-hooks/server-nested-agent-roster.test.ts +++ b/src/main/agent-hooks/server-nested-agent-roster.test.ts @@ -5,10 +5,7 @@ import { join } from 'node:path' import { AgentHookServer } from './server' import { buildBody, PANE, postHookEvent } from './server.test-fixtures' import { RelayAgentHookServer } from '../../wsl-guest/agent-hook-server' -import { - AGENT_STATUS_MAX_SUBAGENTS, - AGENT_STATUS_STALE_AFTER_MS -} from '../../shared/agent-status-types' +import { AGENT_STATUS_STALE_AFTER_MS } from '../../shared/agent-status-types' import type { AgentHookSource } from '../../shared/agent-hook-relay' vi.mock('../telemetry/client', () => ({ track: vi.fn() })) @@ -16,6 +13,8 @@ vi.mock('../telemetry/cohort-classifier', () => ({ getCohortAtEmit: vi.fn(() => ({ nth_repo_added: 2 })) })) +const CHILD_COUNT = 32 + afterEach(() => vi.restoreAllMocks()) describe('nested agent hooks preserve the owning pane', () => { @@ -58,7 +57,7 @@ describe('nested agent hooks preserve the owning pane', () => { await post('claude', { hook_event_name: 'Stop', session_id: 'claude-parent', - background_tasks: Array.from({ length: AGENT_STATUS_MAX_SUBAGENTS }, (_, i) => ({ + background_tasks: Array.from({ length: CHILD_COUNT }, (_, i) => ({ id: `child${i}`, type: 'subagent', status: 'running', @@ -72,7 +71,7 @@ describe('nested agent hooks preserve the owning pane', () => { mainAgent: { state: 'done' }, providerSession: { id: 'claude-parent' } }) - expect(before[0]?.subagents).toHaveLength(AGENT_STATUS_MAX_SUBAGENTS) + expect(before[0]?.subagents).toHaveLength(CHILD_COUNT) pushed.mockClear() for (const hook_event_name of [ 'SessionStart', @@ -111,9 +110,7 @@ describe('nested agent hooks preserve the owning pane', () => { session_id: 'claude-parent', agent_id: 'child0' }) - expect(server.getStatusSnapshot()[0]?.subagents).toHaveLength( - AGENT_STATUS_MAX_SUBAGENTS - 1 - ) + expect(server.getStatusSnapshot()[0]?.subagents).toHaveLength(CHILD_COUNT - 1) await post('claude', { hook_event_name: 'Stop', session_id: 'claude-parent', diff --git a/src/main/claude/claude-background-task-roster-merge.ts b/src/main/claude/claude-background-task-roster-merge.ts index 2b825a63835..729ea081fea 100644 --- a/src/main/claude/claude-background-task-roster-merge.ts +++ b/src/main/claude/claude-background-task-roster-merge.ts @@ -3,22 +3,15 @@ import type { TrackedClaudeBackgroundTask } from './claude-settled-background-ta /** Keep live foreground work beside the provider's authoritative background roster. */ export function combineClaudeBackgroundTaskRoster( prior: ReadonlyMap, - roster: ReadonlyMap, - limit: number + roster: ReadonlyMap ): Map { const combined = new Map() - const retainable = [...prior].filter( - ([id, task]) => !task.backgrounded && task.liveInTurn && !roster.has(id) - ) - let evict = Math.max(0, roster.size + retainable.length - limit) - // Drop oldest foreground rows first; preserve the provider's roster order. - for (const [id, task] of retainable) { - if (evict > 0) { - evict -= 1 - continue + for (const [id, task] of prior) { + if (!task.backgrounded && task.liveInTurn && !roster.has(id)) { + combined.set(id, task) } - combined.set(id, task) } + // Preserve the provider's roster order. for (const [id, task] of roster) { combined.set(id, task) } diff --git a/src/main/claude/claude-background-task-tracker.test.ts b/src/main/claude/claude-background-task-tracker.test.ts index b23602b0fff..031af23c3b0 100644 --- a/src/main/claude/claude-background-task-tracker.test.ts +++ b/src/main/claude/claude-background-task-tracker.test.ts @@ -513,7 +513,7 @@ describe('ClaudeBackgroundTaskTracker', () => { } }) - it('admits unknown background updates conservatively and bounds edge-only fallback ids', () => { + it('admits unknown background updates conservatively and tracks every edge-only start', () => { const tracker = trackerAt([100]) tracker.observe( system('task_updated', { task_id: 'unknown', patch: { is_backgrounded: true } }) @@ -529,20 +529,19 @@ describe('ClaudeBackgroundTaskTracker', () => { }) ) } - expect(tracker.stoppableTaskIds.length).toBeLessThanOrEqual(256) - expect(tracker.goalBackgroundWork).toBe('live') + expect(tracker.stoppableTaskIds).toHaveLength(401) tracker.observe(system('task_notification', { task_id: 'unknown', status: 'completed' })) - for (let index = 0; index < 256; index += 1) { + for (let index = 0; index < 399; index += 1) { tracker.observe( system('task_notification', { task_id: `task-${index}`, status: 'completed' }) ) } - expect(tracker.goalBackgroundWork).toBe('unverifiable') - tracker.observe(aggregate([])) + expect(tracker.goalBackgroundWork).toBe('live') + tracker.observe(system('task_notification', { task_id: 'task-399', status: 'completed' })) expect(tracker.goalBackgroundWork).toBe('exited') }) - it('bounds aggregate rosters and resets to the edge-only fallback on clear', () => { + it('lists every aggregate roster entry and resets to the edge-only fallback on clear', () => { const tracker = trackerAt([100]) tracker.observe( aggregate( @@ -553,7 +552,7 @@ describe('ClaudeBackgroundTaskTracker', () => { })) ) ) - expect(tracker.stoppableTaskIds).toHaveLength(256) + expect(tracker.stoppableTaskIds).toHaveLength(400) tracker.clear() tracker.observe( @@ -649,7 +648,7 @@ describe('ClaudeBackgroundTaskTracker', () => { expect(rows(tracker)).toEqual([{ id: 'back-1' }]) }) - it('drops a backgrounded start the roster no longer lists but bounds what it retains', () => { + it('drops a stale backgrounded start and retains every live foreground row', () => { const tracker = new ClaudeBackgroundTaskTracker() tracker.observe({ type: 'user' }, true) for (let index = 0; index < 300; index += 1) { @@ -660,12 +659,8 @@ describe('ClaudeBackgroundTaskTracker', () => { ) const ids = rows(tracker).map((row) => row.id) - // 255 retained foreground rows plus the roster's own entry: retention is - // real and still counts against the cap. - expect(ids).toHaveLength(256) - // When the cap bites, the STALEST retained row goes, not the newest. - expect(ids).toContain('fore-299') - expect(ids).not.toContain('fore-44') + expect(ids).toHaveLength(301) + expect(ids).toContain('fore-0') expect(ids).toContain('back-1') // Aggregate authority over its OWN class is unchanged. diff --git a/src/main/claude/claude-background-task-tracker.ts b/src/main/claude/claude-background-task-tracker.ts index 77d3b298aea..c59b41d0345 100644 --- a/src/main/claude/claude-background-task-tracker.ts +++ b/src/main/claude/claude-background-task-tracker.ts @@ -31,7 +31,8 @@ export { } from './claude-background-task-frames' export type { ClaudeBackgroundTaskKind } from './claude-background-task-frames' -const MAX_TRACKED_TASKS = 256 +/** Ended ids remembered to refuse a late start; live tasks take no bound. */ +const MAX_TERMINAL_TASK_IDS = 256 export type ClaudeGoalBackgroundWork = 'live' | 'unverifiable' | 'exited' export class ClaudeBackgroundTaskTracker { @@ -245,10 +246,6 @@ export class ClaudeBackgroundTaskTracker { coverageIncomplete = true continue } - if (!roster.has(id) && roster.size >= MAX_TRACKED_TASKS) { - coverageIncomplete = true - continue - } // An authoritative live roster supersedes an earlier terminal edge — for // the ids it actually lists. Wiping the whole set left a finished // FOREGROUND id undefended, since the start guard now convicts only @@ -268,7 +265,7 @@ export class ClaudeBackgroundTaskTracker { }) } this.coverageIncomplete = coverageIncomplete - for (const [id, task] of combineClaudeBackgroundTaskRoster(prior, roster, MAX_TRACKED_TASKS)) { + for (const [id, task] of combineClaudeBackgroundTaskRoster(prior, roster)) { this.tasks.set(id, task) } for (const [id, task] of prior) { @@ -279,22 +276,6 @@ export class ClaudeBackgroundTaskTracker { } private upsert(id: string, task: Omit): void { - if (!this.tasks.has(id) && this.tasks.size >= MAX_TRACKED_TASKS) { - let foregroundId: string | undefined - for (const [candidateId, candidate] of this.tasks) { - if (!candidate.backgrounded) { - foregroundId = candidateId - break - } - } - if (!foregroundId) { - if (task.backgrounded) { - this.coverageIncomplete = true - } - return - } - this.tasks.delete(foregroundId) - } const existing = this.tasks.get(id) ?? this.retention.resume(id) this.terminalTaskIds.delete(id) if (existing) { @@ -318,7 +299,7 @@ export class ClaudeBackgroundTaskTracker { this.tasks.delete(id) this.terminalTaskIds.delete(id) this.terminalTaskIds.add(id) - if (this.terminalTaskIds.size > MAX_TRACKED_TASKS) { + if (this.terminalTaskIds.size > MAX_TERMINAL_TASK_IDS) { const oldest = this.terminalTaskIds.values().next() if (!oldest.done) { this.terminalTaskIds.delete(oldest.value) diff --git a/src/main/claude/claude-child-work-decoder.test.ts b/src/main/claude/claude-child-work-decoder.test.ts index 084fd6eac0e..49351c061e9 100644 --- a/src/main/claude/claude-child-work-decoder.test.ts +++ b/src/main/claude/claude-child-work-decoder.test.ts @@ -154,6 +154,13 @@ describe('Claude child-work decoder', () => { status: 'completed' }) ) + // However many other children ended meanwhile, the wake still reads as this run's restart. + for (let index = 0; index < 300; index += 1) { + const id = `other-${index}` + decoder.observe({ ...backgroundAgent, task_id: id, tool_use_id: `toolu_${index}` }) + decoder.observe(system('task_notification', { task_id: id, status: 'completed' })) + } + decoder.drain(400) decoder.observe(backgroundAgent) decoder.observe(system('task_progress', { task_id: 'agent-bg', last_tool_name: 'Read' })) expect(decoder.drain(500)).toEqual([ @@ -208,7 +215,7 @@ describe('Claude child-work decoder', () => { expect(decoder.drain(500)).toEqual([]) }) - it('admits every live child: only the host store bounds them', () => { + it('admits every live child', () => { const decoder = new ClaudeChildWorkDecoder() for (let index = 0; index < 300; index += 1) { decoder.observe({ ...backgroundAgent, task_id: `agent-${index}` }) diff --git a/src/main/claude/claude-child-work-decoder.ts b/src/main/claude/claude-child-work-decoder.ts index 6cdab01d8f3..341cca0c3e8 100644 --- a/src/main/claude/claude-child-work-decoder.ts +++ b/src/main/claude/claude-child-work-decoder.ts @@ -36,10 +36,6 @@ import { type ClaudeTaskFacts } from './claude-child-work-evidence' -/** Ended ids remembered to recognise a restart. Live children take no bound here: refusing one - * would hide a running child; the host's global limits bound them. */ -const MAX_ENDED_TASKS = 256 - type DecodedClaudeTask = { kind: AgentChildWorkKind backgrounded: boolean @@ -85,7 +81,8 @@ function observation( export class ClaudeChildWorkDecoder { private readonly live = new Map() - /** Ended task ids, with the spawn call each ended under and how. */ + /** Ended task ids, with the spawn call each ended under and how. Kept for the session: a + * forgotten id would read a wake under the same spawn call as a settled child. */ private readonly ended = new Map< string, { run: string | undefined; outcome: AgentChildWorkOutcome } @@ -266,14 +263,7 @@ export class ClaudeChildWorkDecoder { const task = this.live.get(id) this.live.delete(id) const outcome = claudeChildWorkOutcome(status) - this.ended.delete(id) this.ended.set(id, { run: reported.runId ?? task?.toolUseId, outcome }) - if (this.ended.size > MAX_ENDED_TASKS) { - const [oldest] = this.ended.keys() - if (oldest !== undefined) { - this.ended.delete(oldest) - } - } this.pending.push((observedAt) => ({ type: 'ended', observedAt, diff --git a/src/main/codex/codex-background-command-tracker.test.ts b/src/main/codex/codex-background-command-tracker.test.ts index f720a92a44c..ade16f23705 100644 --- a/src/main/codex/codex-background-command-tracker.test.ts +++ b/src/main/codex/codex-background-command-tracker.test.ts @@ -52,25 +52,6 @@ describe('persistent command ownership', () => { expect(tracker.tasks()).toEqual([]) }) - it('preflights finite metadata capacity and admits work again after process completion', () => { - const tracker = new CodexBackgroundCommandTracker('root', 700) - const first = command('item/started', 'first') - const second = command('item/started', 'second') - expect(tracker.canObserve(first)).toBe(true) - tracker.observe(first) - expect(tracker.canObserve(second)).toBe(false) - expect(() => tracker.observe(second)).toThrow('not admitted') - expect(tracker.threadCommands('root')).toHaveLength(1) - expect(tracker.retainedMetadataBytes).toBeLessThanOrEqual(700) - tracker.observe(command('item/completed', 'first')) - expect(tracker.canObserve(second)).toBe(true) - tracker.observe(second) - expect(tracker.threadCommands('root')).toHaveLength(1) - expect(tracker.retainedMetadataBytes).toBeLessThanOrEqual(700) - tracker.clear() - expect(tracker.retainedMetadataBytes).toBe(0) - }) - it('keeps the journal running across turn completion and accepts late output and exit', () => { const rows: { key: string; body: AgentJournalItemBody }[] = [] const translator = createCodexJournalTranslator({ diff --git a/src/main/codex/codex-background-command-tracker.ts b/src/main/codex/codex-background-command-tracker.ts index 4327ce1f262..65d352424ec 100644 --- a/src/main/codex/codex-background-command-tracker.ts +++ b/src/main/codex/codex-background-command-tracker.ts @@ -6,7 +6,6 @@ import { import type { CodexBackgroundTaskEvent } from './codex-background-task-frames' import { readRecord, readString } from './codex-item-field-readers' import { readCodexThreadItem } from './codex-structured-item-translation' -import { MAX_CODEX_ITEM_STREAM_METADATA_BYTES } from './codex-item-stream-retention' import type { CodexAbandonedCommand } from './codex-prompt-registry' import { readCodexTurnId } from './codex-structured-thread-facts' import { codexPrimaryCommandTaskId } from '../../shared/structured-session-foreground-commands' @@ -20,7 +19,6 @@ type Command = { task: AgentSessionBackgroundTask /** Codex's id for the command's process, which a background-terminal stop names. */ processId?: string - bytes: number } /** A command process starting, or ending: it exited, its thread closed, or the session ended. */ @@ -37,29 +35,9 @@ export type CodexBackgroundCommandChange = export class CodexBackgroundCommandTracker { private readonly commands = new Map() - private readonly settled = new Map() - private liveBytes = 0 - private settledBytes = 0 + private readonly settled = new Set() - constructor( - private readonly primaryThreadId: string, - private readonly maxMetadataBytes = MAX_CODEX_ITEM_STREAM_METADATA_BYTES - ) {} - - get retainedMetadataBytes(): number { - return this.liveBytes + this.settledBytes - } - - canObserve(event: CodexBackgroundTaskEvent): boolean { - const parsed = this.parse(event) - return ( - !parsed || - parsed.completed || - this.commands.has(parsed.key) || - this.settled.has(parsed.key) || - this.liveBytes + parsed.command.bytes <= this.maxMetadataBytes - ) - } + constructor(private readonly primaryThreadId: string) {} observe(event: CodexBackgroundTaskEvent): CodexBackgroundCommandChange | null { const parsed = this.parse(event) @@ -73,12 +51,7 @@ export class CodexBackgroundCommandTracker { if (this.commands.has(key)) { return null } - if (this.liveBytes + command.bytes > this.maxMetadataBytes) { - throw new Error('Codex command metadata was not admitted before observation') - } this.commands.set(key, command) - this.liveBytes += command.bytes - this.trimSettled() return this.liveChange(command) } @@ -176,8 +149,6 @@ export class CodexBackgroundCommandTracker { ) this.commands.clear() this.settled.clear() - this.liveBytes = 0 - this.settledBytes = 0 return ended } @@ -185,7 +156,6 @@ export class CodexBackgroundCommandTracker { private end(key: string): CodexBackgroundCommandChange | null { const existing = this.commands.get(key) if (existing) { - this.liveBytes -= existing.bytes this.commands.delete(key) } this.rememberSettled(key) @@ -199,28 +169,12 @@ export class CodexBackgroundCommandTracker { } private rememberSettled(key: string): void { - if (this.settled.has(key)) { - return - } - const bytes = Buffer.byteLength(key, 'utf8') + 256 - if (this.liveBytes + bytes <= this.maxMetadataBytes) { - this.settled.set(key, bytes) - this.settledBytes += bytes - } - this.trimSettled() - } - - private trimSettled(): void { - while ( - this.settled.size > MAX_SETTLED_COMMANDS || - this.retainedMetadataBytes > this.maxMetadataBytes - ) { - const oldest = this.settled.entries().next().value - if (!oldest) { - break + this.settled.add(key) + if (this.settled.size > MAX_SETTLED_COMMANDS) { + const oldest = this.settled.values().next().value + if (oldest !== undefined) { + this.settled.delete(oldest) } - this.settled.delete(oldest[0]) - this.settledBytes -= oldest[1] } } @@ -242,7 +196,7 @@ export class CodexBackgroundCommandTracker { .trim() const turnId = readCodexTurnId(event.params) const processId = readString(item, 'processId') ?? undefined - const value = { + const command = { threadId: event.threadId, turnId, backgrounded: turnId !== null && this.settled.has(this.turnKey(event.threadId, turnId)), @@ -256,14 +210,6 @@ export class CodexBackgroundCommandTracker { }, ...(processId !== undefined ? { processId } : {}) } - return { - key, - completed, - command: { - ...value, - bytes: - Buffer.byteLength(key, 'utf8') + Buffer.byteLength(JSON.stringify(value), 'utf8') + 256 - } - } + return { key, completed, command } } } diff --git a/src/main/codex/codex-background-task-tracker.ts b/src/main/codex/codex-background-task-tracker.ts index 69e0f11e4d5..cf531f0f59e 100644 --- a/src/main/codex/codex-background-task-tracker.ts +++ b/src/main/codex/codex-background-task-tracker.ts @@ -133,10 +133,6 @@ export class CodexBackgroundTaskTracker { return this.publishedState } - canObserve(event: CodexBackgroundTaskEvent): boolean { - return this.commands.canObserve(event) - } - /** `unapproved`: commands whose approval the journal dropped with this frame's turn ending. */ observe( event: CodexBackgroundTaskEvent, diff --git a/src/main/codex/codex-child-work-evidence.test.ts b/src/main/codex/codex-child-work-evidence.test.ts index 00023b11bac..b53cdfb7ed4 100644 --- a/src/main/codex/codex-child-work-evidence.test.ts +++ b/src/main/codex/codex-child-work-evidence.test.ts @@ -548,7 +548,7 @@ describe('Codex child-work evidence', () => { expect(byKind('command')).toEqual([expect.objectContaining({ membership: 'live' })]) }) - it('leaves no record behind a finished shell, so none displaces a finished child', () => { + it('leaves no record behind a finished shell', () => { const { send, byKind, records, log } = runningChild() send(turn('turn/completed', CHILD, 'c1')) const before = log.length diff --git a/src/main/codex/codex-child-work-evidence.ts b/src/main/codex/codex-child-work-evidence.ts index 9569acac65c..c8f3f27dfa6 100644 --- a/src/main/codex/codex-child-work-evidence.ts +++ b/src/main/codex/codex-child-work-evidence.ts @@ -34,9 +34,6 @@ import { codexThreadWaitsOnUser, readCodexTurnId } from './codex-structured-thre import { CODEX_TOKEN_USAGE_METHOD, readCodexThreadTokenTotal } from './codex-subagent-activity' import type { CodexExecutionChild, CodexSubagentExecutions } from './codex-subagent-executions' -/** Facts kept for settled children; a working child always has its own. */ -const MAX_CHILD_FACTS = 128 -const MAX_OPEN_CALLS_PER_CHILD = 16 const CHILD_FRAME_METHODS: ReadonlySet = new Set([ 'item/started', 'item/completed', @@ -196,9 +193,6 @@ export class CodexChildWorkEvidence { const call = codexChildToolCall(item) if (call && !facts.openCalls.has(item.id)) { facts.openCalls.set(item.id, { ...call, turnId }) - for (const stale of [...facts.openCalls.keys()].slice(0, -MAX_OPEN_CALLS_PER_CHILD)) { - facts.openCalls.delete(stale) - } } } @@ -304,15 +298,6 @@ export class CodexChildWorkEvidence { if (existing) { return existing } - if (this.facts.size >= MAX_CHILD_FACTS) { - // Refusing a working child its facts would drop its every edge; the host's limits bound them. - const idle = [...this.facts.keys()].find( - (id) => this.executions.find(id)?.execution?.state !== 'working' - ) - if (idle !== undefined) { - this.facts.delete(idle) - } - } const facts: ChildFacts = { openCalls: new Map(), waiting: false } this.facts.set(threadId, facts) return facts diff --git a/src/main/codex/codex-item-stream-retention.ts b/src/main/codex/codex-item-stream-retention.ts index ace12f34667..f21ba0ba0a9 100644 --- a/src/main/codex/codex-item-stream-retention.ts +++ b/src/main/codex/codex-item-stream-retention.ts @@ -5,35 +5,23 @@ import { } from './codex-structured-item-stream-bounds' import type { CodexItemStreamState } from './codex-structured-item-stream-contracts' -// Preserve the previous metadata ceiling while letting small live commands share it. -export const MAX_CODEX_ITEM_STREAM_METADATA_BYTES = - MAX_CODEX_ITEM_STREAM_STATES * MAX_CODEX_ITEM_STREAM_ITEM_BYTES +// Bounds only evictable streams; a command that outlives its turn is kept until it ends. +const MAX_EVICTABLE_STATE_BYTES = MAX_CODEX_ITEM_STREAM_STATES * MAX_CODEX_ITEM_STREAM_ITEM_BYTES type RetainedState = { state: CodexItemStreamState; bytes: number; persistent: boolean } export class CodexItemStreamRetention { private readonly states = new Map() - private bytes = 0 - private persistentBytes = 0 + private evictableBytes = 0 private persistentCount = 0 - constructor(private readonly maxBytes = MAX_CODEX_ITEM_STREAM_METADATA_BYTES) {} - - get retainedBytes(): number { - return this.bytes - } - - get size(): number { - return this.states.size - } - get persistentSize(): number { return this.persistentCount } get overCapacity(): boolean { return ( - this.bytes > this.maxBytes || + this.evictableBytes > MAX_EVICTABLE_STATE_BYTES || this.states.size - this.persistentCount > MAX_CODEX_ITEM_STREAM_STATES ) } @@ -46,30 +34,16 @@ export class CodexItemStreamRetention { return this.states.get(key)?.persistent === true } - canRetain(key: string, state: CodexItemStreamState): boolean { - const previous = this.states.get(key) - return ( - this.persistentBytes - - (previous?.persistent ? previous.bytes : 0) + - this.stateBytes(key, state) <= - this.maxBytes - ) - } - - retain(key: string, state: CodexItemStreamState): boolean { - if (!this.canRetain(key, state)) { - return false - } + retain(key: string, state: CodexItemStreamState): void { this.forget(key) const bytes = this.stateBytes(key, state) const persistent = codexCommandOutlivesTurn(state.item) this.states.set(key, { state, bytes, persistent }) - this.bytes += bytes if (persistent) { - this.persistentBytes += bytes this.persistentCount += 1 + } else { + this.evictableBytes += bytes } - return true } oldestEvictable(): string | undefined { @@ -86,18 +60,17 @@ export class CodexItemStreamRetention { if (!entry) { return } - this.bytes -= entry.bytes if (entry.persistent) { - this.persistentBytes -= entry.bytes this.persistentCount -= 1 + } else { + this.evictableBytes -= entry.bytes } this.states.delete(key) } clear(): void { this.states.clear() - this.bytes = 0 - this.persistentBytes = 0 + this.evictableBytes = 0 this.persistentCount = 0 } diff --git a/src/main/codex/codex-persistent-command-retention.test.ts b/src/main/codex/codex-persistent-command-retention.test.ts index 5c70b2bbcf6..22396998943 100644 --- a/src/main/codex/codex-persistent-command-retention.test.ts +++ b/src/main/codex/codex-persistent-command-retention.test.ts @@ -3,10 +3,6 @@ import { describe, expect, it, vi } from 'vitest' import type { AgentJournalItemBody } from '../../shared/agent-session-journal-types' import { agentJournalItemKey } from '../../shared/agent-session-journal-item-key' import { CodexBackgroundCommandTracker } from './codex-background-command-tracker' -import { - CodexItemStreamRetention, - MAX_CODEX_ITEM_STREAM_METADATA_BYTES -} from './codex-item-stream-retention' import { CodexJournalItems } from './codex-structured-journal-items' import { settleCodexJournalTurn } from './codex-structured-journal-settlement' @@ -29,7 +25,7 @@ function command(threadId: string, id: string, method = 'item/started') { } } -function fixture(maxMetadataBytes?: number) { +function fixture() { const rows = new Map() const scheduled = new Set<() => void>() const sink = { @@ -45,7 +41,6 @@ function fixture(maxMetadataBytes?: number) { const items = new CodexJournalItems( { sink, - maxMetadataBytes, attributionFor: () => ({ turnScope: AGENT_JOURNAL_THREAD_SCOPE }), schedule: (run) => { scheduled.add(run) @@ -99,7 +94,6 @@ describe('persistent command retention', () => { Array.from({ length: 64 }, (_, index) => command(`thread-${thread}`, `exec-${index}`)) ).flat() for (const event of events) { - expect(tracker.canObserve(event)).toBe(true) expect(items.handle(event)).toMatchObject({ admission: { accepted: true } }) tracker.observe(event) expect( @@ -134,7 +128,6 @@ describe('persistent command retention', () => { expect(items.activeItems.size).toBe(448) expect(items.streams.persistentCount).toBe(448) expect(tracker.tasks()).toHaveLength(448) - expect(tracker.retainedMetadataBytes).toBeLessThan(256 * 1024) for (const event of events) { items.streams.handle(event.threadId, 'item/commandExecution/outputDelta', { turnId: 'turn', @@ -165,50 +158,10 @@ describe('persistent command retention', () => { expect(items.activeItems.size).toBe(0) expect(items.streams.persistentCount).toBe(0) expect(tracker.tasks()).toEqual([]) - expect(tracker.retainedMetadataBytes).toBeLessThan(64 * 1024) items.dispose() - tracker.clear() - expect(tracker.retainedMetadataBytes).toBe(0) expect(scheduled.size).toBe(0) }) - it('rejects command metadata exhaustion before appending or evicting live state and frees it on completion', () => { - const { items, rows } = fixture(800) - const first = command('root', 'first') - const second = command('root', 'second') - expect(items.handle(first)).toMatchObject({ admission: { accepted: true } }) - const prior = [...rows] - expect(items.handle(second)).toMatchObject({ admission: { accepted: false, reason: 'failed' } }) - expect([...rows]).toEqual(prior) - expect(items.activeItems.size).toBe(1) - expect(items.handle(command('root', 'first', 'item/completed'))).toMatchObject({ - admission: { accepted: true } - }) - expect(items.handle(second)).toMatchObject({ admission: { accepted: true } }) - items.dispose() - }) - - it('accounts metadata bytes instead of interpreting the item count as liveness', () => { - const retention = new CodexItemStreamRetention() - for (let index = 0; index < 448; index += 1) { - const item = command('root', `exec-${index}`).params.item - expect( - retention.retain(item.id, { - item, - identity: { provider: 'orca', clientMessageId: item.id } - }) - ).toBe(true) - } - expect(retention.size).toBe(448) - expect(retention.retainedBytes).toBeLessThan(256 * 1024) - expect(retention.retainedBytes).toBeLessThan(MAX_CODEX_ITEM_STREAM_METADATA_BYTES) - expect(retention.overCapacity).toBe(false) - expect(retention.oldestEvictable()).toBeUndefined() - retention.clear() - expect(retention.retainedBytes).toBe(0) - expect(retention.persistentSize).toBe(0) - }) - it('retains startup provenance when large command metadata is bounded', () => { const { items, sink } = fixture() const event = command('root', 'large') diff --git a/src/main/codex/codex-structured-item-stream-contracts.ts b/src/main/codex/codex-structured-item-stream-contracts.ts index 38b9a9d3038..af61e3357b6 100644 --- a/src/main/codex/codex-structured-item-stream-contracts.ts +++ b/src/main/codex/codex-structured-item-stream-contracts.ts @@ -17,7 +17,6 @@ export type CodexItemStreamDeps = { coalesceMs?: number maxRetainedBytes?: number maxTotalRetainedBytes?: number - maxMetadataBytes?: number schedule?: AgentSessionDeltaCoalescerDeps['schedule'] } @@ -44,14 +43,13 @@ export type CodexStructuredItemStreamHandleResult = { export type CodexStructuredItemStreams = { readonly persistentCount: number - canTrack: (threadId: string, item: CodexThreadItem, identity: AgentJournalItemIdentity) => boolean track: ( threadId: string, turnId: string | null, item: CodexThreadItem, identity: AgentJournalItemIdentity, startedAt?: number - ) => boolean + ) => void handle: ( threadId: string, method: string, diff --git a/src/main/codex/codex-structured-item-streams.ts b/src/main/codex/codex-structured-item-streams.ts index c783458ca29..261b8eea217 100644 --- a/src/main/codex/codex-structured-item-streams.ts +++ b/src/main/codex/codex-structured-item-streams.ts @@ -48,7 +48,7 @@ export { export function createCodexStructuredItemStreams( deps: CodexItemStreamDeps ): CodexStructuredItemStreams { - const states = new CodexItemStreamRetention(deps.maxMetadataBytes) + const states = new CodexItemStreamRetention() const checkpointLengths = new Map() const pendingCheckpoints = new Set() // Which thread and turn produced each stream, resolved to its attribution per append @@ -158,7 +158,7 @@ export function createCodexStructuredItemStreams( itemId: string, type: string, params: unknown - ): CodexItemStreamState | null => { + ): CodexItemStreamState => { const key = codexStructuredItemKey(threadId, itemId) const existing = states.get(key) if (existing) { @@ -167,9 +167,7 @@ export function createCodexStructuredItemStreams( const item = { type, id: itemId } const turnId = deps.turnIdFor(threadId, params) const state = { item, identity: deps.identityFor(threadId, turnId, item) } - if (!states.retain(key, state)) { - return null - } + states.retain(key, state) producers.set(key, { threadId, turnId }) trimStates() return state @@ -214,19 +212,11 @@ export function createCodexStructuredItemStreams( get persistentCount() { return states.persistentSize }, - canTrack: (threadId, item, identity) => - states.canRetain( - codexStructuredItemKey(threadId, item.id), - codexItemStreamState(item, identity) - ), track: (threadId, turnId, item, identity, startedAt) => { const key = codexStructuredItemKey(threadId, item.id) - if (!states.retain(key, codexItemStreamState(item, identity, startedAt))) { - return false - } + states.retain(key, codexItemStreamState(item, identity, startedAt)) producers.set(key, { threadId, turnId }) trimStates() - return true }, handle: (threadId, method, params) => { const paramsRecord = readCodexItemStreamRecord(params) @@ -243,9 +233,6 @@ export function createCodexStructuredItemStreams( const key = codexStructuredItemKey(threadId, itemId) const streamFlushed = coalescer.flush(key) const state = ensureState(threadId, itemId, 'fileChange', params) - if (!state) { - return { handled: true, admission: { accepted: false, reason: 'failed' } } - } state.item = { ...state.item, changes: paramsRecord.changes } const translated = codexJournalItem(state.item) if (translated.body) { @@ -285,9 +272,6 @@ export function createCodexStructuredItemStreams( return { handled: true, admission: { accepted: true } } } const state = ensureState(threadId, itemId, type ?? 'reasoning', params) - if (!state) { - return { handled: true, admission: { accepted: false, reason: 'failed' } } - } const delta = method === REASONING_PART_METHOD ? '\n' : paramsRecord.delta if (typeof delta === 'string') { const key = codexStructuredItemKey(threadId, state.item.id) diff --git a/src/main/codex/codex-structured-journal-items.ts b/src/main/codex/codex-structured-journal-items.ts index 17a002225e7..85db92a578e 100644 --- a/src/main/codex/codex-structured-journal-items.ts +++ b/src/main/codex/codex-structured-journal-items.ts @@ -54,7 +54,7 @@ export class CodexJournalItems { private readonly deps: Pick< CodexJournalTranslatorDeps, 'sink' | 'coalesceMs' | 'maxRetainedBytes' | 'schedule' | 'now' - > & { maxMetadataBytes?: number; attributionFor: CodexRowAttribution }, + > & { attributionFor: CodexRowAttribution }, private readonly activeTurn: (threadId: string) => string | null, private readonly suppress: (threadId: string, turnId: string) => void, private readonly helperName?: CodexHelperName @@ -64,7 +64,6 @@ export class CodexJournalItems { coalesceMs: deps.coalesceMs, maxRetainedBytes: deps.maxRetainedBytes, schedule: deps.schedule, - maxMetadataBytes: deps.maxMetadataBytes, turnIdFor: (threadId, params) => readCodexTurnId(params) ?? this.activeTurn(threadId), identityFor: (threadId, turnId, item) => this.identityFor(threadId, turnId, item), attributionFor: deps.attributionFor @@ -101,12 +100,6 @@ export class CodexJournalItems { if (item.type === 'contextCompaction' && event.method === 'item/started') { return { handled: true, admission: CODEX_JOURNAL_ADMITTED } } - if ( - event.method !== 'item/completed' && - !this.streams.canTrack(event.threadId, item, identity) - ) { - return { handled: true, admission: { accepted: false, reason: 'failed' } } - } const itemKey = codexStructuredItemKey(event.threadId, item.id) const active = event.method === 'item/completed' ? this.activeItems.get(itemKey) : undefined const receivedAt = event.observedAt ?? this.deps.now?.() ?? Date.now() diff --git a/src/main/codex/codex-structured-journal-translation-turn-boundaries.ts b/src/main/codex/codex-structured-journal-translation-turn-boundaries.ts index 2a1b3d92fce..19b60061804 100644 --- a/src/main/codex/codex-structured-journal-translation-turn-boundaries.ts +++ b/src/main/codex/codex-structured-journal-translation-turn-boundaries.ts @@ -76,9 +76,6 @@ export class CodexJournalTurnBoundaries { if (!turnId) { return CODEX_JOURNAL_ADMITTED } - if (!this.deps.activeTurns.canRemember(event.threadId, turnId)) { - return { accepted: false, reason: 'backpressure' } - } const startedAt = this.receiptTime(event) const command = event.threadId === this.deps.primaryThreadId() ? this.deps.commands.claim(turnId) : null diff --git a/src/main/codex/codex-structured-journal-translation-turn-state.test.ts b/src/main/codex/codex-structured-journal-translation-turn-state.test.ts index dda4eaca063..bd3699cbe22 100644 --- a/src/main/codex/codex-structured-journal-translation-turn-state.test.ts +++ b/src/main/codex/codex-structured-journal-translation-turn-state.test.ts @@ -2,53 +2,16 @@ import { describe, expect, it } from 'vitest' import { CodexJournalActiveTurns, CodexJournalRecentTurns, - MAX_CODEX_ACTIVE_TURN_BYTES, - MAX_CODEX_ACTIVE_TURNS, MAX_CODEX_RECENT_TURN_BYTES, MAX_CODEX_RECENT_TURNS } from './codex-structured-journal-translation-turn-state' describe('CodexJournalActiveTurns', () => { - it('refuses new turns at the bounded active capacity without evicting live state', () => { - const active = new CodexJournalActiveTurns() - for (let index = 0; index < MAX_CODEX_ACTIVE_TURNS; index += 1) { - expect(active.remember(`thread-${index}`, `turn-${index}`)).toBe(true) - } - - expect(active.remember('thread-overflow', 'turn-overflow')).toBe(false) - expect(active.size).toBe(MAX_CODEX_ACTIVE_TURNS) - expect(active.byThread.size).toBe(MAX_CODEX_ACTIVE_TURNS) - expect(active.current('thread-0')).toBe('turn-0') - expect(active.current(`thread-${MAX_CODEX_ACTIVE_TURNS - 1}`)).toBe( - `turn-${MAX_CODEX_ACTIVE_TURNS - 1}` - ) - }) - - it('admits a new turn after an earlier turn settles', () => { - const active = new CodexJournalActiveTurns() - for (let index = 0; index < MAX_CODEX_ACTIVE_TURNS; index += 1) { - active.remember('thread', `turn-${index}`) - } - active.forget('thread', 'turn-0') - - expect(active.remember('thread', 'turn-new')).toBe(true) - expect(active.size).toBe(MAX_CODEX_ACTIVE_TURNS) - expect(active.current('thread')).toBe('turn-new') - }) - - it('refuses provider identifiers that would exceed the aggregate byte bound', () => { - const active = new CodexJournalActiveTurns() - - expect(active.remember('thread', 'x'.repeat(MAX_CODEX_ACTIVE_TURN_BYTES))).toBe(false) - expect(active.size).toBe(0) - expect(active.bytes).toBe(0) - }) - it('remembers each turn start time until the turn is forgotten', () => { const active = new CodexJournalActiveTurns() - expect(active.remember('thread', 'turn-1', 1_000)).toBe(true) - expect(active.remember('thread', 'turn-1', 2_000)).toBe(true) + active.remember('thread', 'turn-1', 1_000) + active.remember('thread', 'turn-1', 2_000) expect(active.startedAt('thread', 'turn-1')).toBe(1_000) expect(active.startedAt('thread', 'turn-missing')).toBeUndefined() diff --git a/src/main/codex/codex-structured-journal-translation-turn-state.ts b/src/main/codex/codex-structured-journal-translation-turn-state.ts index f42abd4b26f..1146a73e0ad 100644 --- a/src/main/codex/codex-structured-journal-translation-turn-state.ts +++ b/src/main/codex/codex-structured-journal-translation-turn-state.ts @@ -1,8 +1,6 @@ import type { AgentJournalTurnLifecycle } from '../../shared/agent-session-journal-types' import type { CodexDispatchRequestOrigin } from './codex-structured-dispatch-echo' -export const MAX_CODEX_ACTIVE_TURNS = 256 -export const MAX_CODEX_ACTIVE_TURN_BYTES = 256 * 1024 export const MAX_CODEX_RECENT_TURNS = 256 export const MAX_CODEX_RECENT_TURN_BYTES = 256 * 1024 @@ -15,9 +13,8 @@ function earlierRequestOrigin( return current === undefined || candidate.sequence < current.sequence } +/** Running turns across provider threads; each is forgotten when it ends, all at session end. */ export class CodexJournalActiveTurns { - /** Bounds active turn keys retained across provider threads. */ - static readonly MAX_ENTRIES = MAX_CODEX_ACTIVE_TURNS readonly byThread = new Map>() /** Host turn-start receipt per remembered turn; the terminal row carries it forward. */ private readonly startedAtByTurn = new Map() @@ -25,34 +22,11 @@ export class CodexJournalActiveTurns { private readonly requestOriginByTurn = new Map() /** Last dispatch armed before each provider turn-start event. */ private readonly latestDispatchSequenceByTurn = new Map() - private activeCount = 0 - private retainedBytes = 0 - - get size(): number { - return this.activeCount - } - - get bytes(): number { - return this.retainedBytes - } private turnKey(threadId: string, turnId: string): string { return `${encodeURIComponent(threadId)}:${encodeURIComponent(turnId)}` } - private entryBytes(threadId: string, turnId: string): number { - return Buffer.byteLength(threadId, 'utf8') + Buffer.byteLength(turnId, 'utf8') - } - - canRemember(threadId: string, turnId: string): boolean { - const active = this.byThread.get(threadId) - return ( - active?.has(turnId) === true || - (this.activeCount < CodexJournalActiveTurns.MAX_ENTRIES && - this.retainedBytes + this.entryBytes(threadId, turnId) <= MAX_CODEX_ACTIVE_TURN_BYTES) - ) - } - current(threadId: string): string | null { return [...(this.byThread.get(threadId) ?? [])].at(-1) ?? null } @@ -113,13 +87,10 @@ export class CodexJournalActiveTurns { turnId: string, startedAt?: number, latestDispatchSequence = Number.MAX_SAFE_INTEGER - ): boolean { + ): void { const active = this.byThread.get(threadId) if (active?.has(turnId)) { - return true - } - if (!this.canRemember(threadId, turnId)) { - return false + return } if (startedAt !== undefined) { this.startedAtByTurn.set(this.turnKey(threadId, turnId), startedAt) @@ -130,9 +101,6 @@ export class CodexJournalActiveTurns { } else { this.byThread.set(threadId, new Set([turnId])) } - this.activeCount += 1 - this.retainedBytes += this.entryBytes(threadId, turnId) - return true } forget(threadId: string, turnId: string): void { @@ -140,10 +108,7 @@ export class CodexJournalActiveTurns { this.requestOriginByTurn.delete(this.turnKey(threadId, turnId)) this.latestDispatchSequenceByTurn.delete(this.turnKey(threadId, turnId)) const active = this.byThread.get(threadId) - if (active?.delete(turnId)) { - this.activeCount -= 1 - this.retainedBytes = Math.max(0, this.retainedBytes - this.entryBytes(threadId, turnId)) - } + active?.delete(turnId) if (!active?.size) { this.byThread.delete(threadId) } @@ -154,8 +119,6 @@ export class CodexJournalActiveTurns { this.startedAtByTurn.clear() this.requestOriginByTurn.clear() this.latestDispatchSequenceByTurn.clear() - this.activeCount = 0 - this.retainedBytes = 0 } } diff --git a/src/main/codex/codex-structured-journal-translation.test.ts b/src/main/codex/codex-structured-journal-translation.test.ts index 908eb86cbad..3d417bcff66 100644 --- a/src/main/codex/codex-structured-journal-translation.test.ts +++ b/src/main/codex/codex-structured-journal-translation.test.ts @@ -10,7 +10,6 @@ import { type StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' import { createCodexJournalTranslator } from './codex-structured-journal-translation' -import { MAX_CODEX_ACTIVE_TURNS } from './codex-structured-journal-translation-turn-state' import { CODEX_COMMAND_APPROVAL_METHOD, CODEX_USER_INPUT_METHOD @@ -111,31 +110,6 @@ function hardWatermarkDeferred() { } describe('codex journal translation', () => { - it('refuses an active-turn overflow before publishing an un-settleable lifecycle row', () => { - const tap = recorder() - const translator = createCodexJournalTranslator({ - sink: tap.sink, - primaryThreadId: () => THREAD_ID - }) - - for (let index = 0; index < MAX_CODEX_ACTIVE_TURNS; index += 1) { - expect( - translator.handle(notification('turn/started', { turn: { id: `turn-${index}` } })) - ).toEqual({ accepted: true }) - } - expect( - translator.handle(notification('turn/started', { turn: { id: 'turn-overflow' } })) - ).toEqual({ accepted: false, reason: 'backpressure' }) - expect(tap.rows.filter((row) => row.body.kind === 'turn')).toHaveLength(MAX_CODEX_ACTIVE_TURNS) - - expect(translator.handle(notification('turn/completed', { turn: { id: 'turn-0' } }))).toEqual({ - accepted: true - }) - expect( - translator.handle(notification('turn/started', { turn: { id: 'turn-overflow' } })) - ).toEqual({ accepted: true }) - }) - it('projects turns restored by thread/resume into durable conversation rows', () => { const { translator, tap } = translatorWith() diff --git a/src/main/codex/codex-structured-session-adapter.ts b/src/main/codex/codex-structured-session-adapter.ts index 50cb90a45f9..806d8ff4023 100644 --- a/src/main/codex/codex-structured-session-adapter.ts +++ b/src/main/codex/codex-structured-session-adapter.ts @@ -124,9 +124,6 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap session: CodexSession, event: CodexStructuredSessionEvent ): CodexJournalTranslationAdmission { - if (event.type === 'notification' && !session.backgroundTasks.canObserve(event)) { - return { accepted: false, reason: 'failed' } - } if (event.type === 'notification') { noteCodexTurnOpened(session, event.method, event.params) } diff --git a/src/main/codex/codex-structured-session-background-tasks.test.ts b/src/main/codex/codex-structured-session-background-tasks.test.ts index 466cadce8a1..8dfc3a15914 100644 --- a/src/main/codex/codex-structured-session-background-tasks.test.ts +++ b/src/main/codex/codex-structured-session-background-tasks.test.ts @@ -7,7 +7,7 @@ import type { openCodexAppServerConnection } from './codex-app-server-connection' import { CodexStructuredSessionAdapter } from './codex-structured-session-adapter' -import { CodexBackgroundTaskTracker } from './codex-background-task-tracker' +import { CodexJournalItems } from './codex-structured-journal-items' import type { CodexStructuredSessionEvent } from './codex-structured-session-state' import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' import { codexProviderHandle } from '../../shared/agent-session-provider-handle-encoding' @@ -173,7 +173,7 @@ describe("codex background tasks reach the host's child records", () => { await adapter.closeSession('session-1') }) - it('recovers the exact provider generation when command metadata cannot be admitted', async () => { + it('recovers the exact provider generation when a notification cannot be admitted', async () => { const published: Published[] = [] const observed: CodexStructuredSessionEvent[] = [] const appendItem = vi.fn() @@ -185,15 +185,15 @@ describe("codex background tasks reach the host's child records", () => { appendItem.mockClear() observed.length = 0 const admission = vi - .spyOn(CodexBackgroundTaskTracker.prototype, 'canObserve') - .mockReturnValue(false) + .spyOn(CodexJournalItems.prototype, 'handle') + .mockReturnValue({ handled: true, admission: { accepted: false, reason: 'failed' } }) try { codex.handlers().onNotification?.('item/started', { threadId: THREAD_ID, turnId: PARENT_TURN, item: { type: 'commandExecution', - id: 'over-budget', + id: 'refused', command: 'sleep 1', source: 'unifiedExecStartup', status: 'inProgress' @@ -220,6 +220,41 @@ describe("codex background tasks reach the host's child records", () => { } }) + it('publishes a child whose turn starts while 256 others are running', async () => { + const published: Published[] = [] + const { adapter, codex } = await adapterWithSession(published, { + appendItem: () => {}, + appendTombstone: () => {}, + publish: () => {} + }) + const startChild = (index: number): void => { + const childId = `child-${index}` + codex.handlers().onNotification?.('turn/started', { + threadId: childId, + turn: { id: `turn-${index}`, status: 'inProgress' } + }) + codex.handlers().onNotification?.('item/started', { + threadId: THREAD_ID, + turnId: PARENT_TURN, + item: { + type: 'subAgentActivity', + id: `call-${index}`, + kind: 'started', + agentThreadId: childId, + agentPath: `/root/worker_${index}` + } + }) + } + // The parent and its first child already hold two running turns. + for (let index = 0; index < 254; index += 1) { + startChild(index) + } + published.length = 0 + startChild(254) + expect(published).toEqual([{ sessionId: 'session-1', evidence: ['live'] }]) + await adapter.closeSession('session-1') + }) + it('publishes the fan-out while it runs and keeps it past the spawning turn', async () => { const published: Published[] = [] const { adapter, codex } = await adapterWithSession(published) diff --git a/src/main/codex/codex-subagent-executions.test.ts b/src/main/codex/codex-subagent-executions.test.ts index d298d2d351a..319ddb427ab 100644 --- a/src/main/codex/codex-subagent-executions.test.ts +++ b/src/main/codex/codex-subagent-executions.test.ts @@ -1,8 +1,9 @@ import { describe, expect, it } from 'vitest' +import { CodexBackgroundTaskTracker } from './codex-background-task-tracker' import { CodexSubagentExecutions } from './codex-subagent-executions' describe('CodexSubagentExecutions retention and identity', () => { - it('bounds settled history through repeated execution without evicting live children', () => { + it('bounds settled turn history through repeated execution without evicting live children', () => { const executions = new CodexSubagentExecutions() executions.register('long-lived', 'long-lived') executions.observeTurn('long-lived', 'long-lived-turn', 'working') @@ -13,12 +14,10 @@ describe('CodexSubagentExecutions retention and identity', () => { executions.observeTurn(id, id, 'completed') } expect(executions.workingChildren().map((child) => child.agentThreadId)).toEqual(['long-lived']) - const { children, settledTurns } = executions.retentionSizes() - expect(children).toBeLessThanOrEqual(128) - expect(settledTurns).toBeLessThanOrEqual(256) + expect(executions.retentionSizes().settledTurns).toBeLessThanOrEqual(256) }) - it('keeps every running child past the settled-history bound', () => { + it('keeps every child, running or ended, however many run', () => { const executions = new CodexSubagentExecutions() for (let index = 0; index < 200; index++) { expect(executions.observeTurn(`child-${index}`, `turn-${index}`, 'working')).not.toBeNull() @@ -26,36 +25,42 @@ describe('CodexSubagentExecutions retention and identity', () => { } expect(executions.workingChildren()).toHaveLength(200) executions.observeTurn('child-0', 'turn-0', 'completed') - executions.register('next', 'next') - expect(executions.find('child-0')).toBeUndefined() + expect(executions.find('child-0')).toBeDefined() expect(executions.workingChildren()).toHaveLength(199) }) - it('evicts a child that ended before one announced but not yet started, which keeps its spawner', () => { + it('targets live descendants through settled ancestors without losing historical lineage order', () => { const executions = new CodexSubagentExecutions() - executions.register('announced', 'announced', 'spawner') - for (let index = 0; index < 127; index++) { - executions.observeTurn(`child-${index}`, `turn-${index}`, 'working') + executions.register('root', 'root', 'primary') + executions.observeTurn('root', 'root-turn', 'working') + const finished = Array.from({ length: 10_000 }, (_, index) => `finished-${index}`) + for (const id of finished) { + executions.register(id, id, 'root') + executions.observeTurn(id, `${id}-turn`, 'completed') } - executions.observeTurn('child-5', 'turn-5', 'completed') - executions.register('next', 'next') - expect(executions.find('child-5')).toBeUndefined() - expect(executions.find('announced')).toMatchObject({ - registered: true, - spawnerThreadId: 'spawner' - }) - executions.observeTurn('announced', 'turn-a', 'working') - expect(executions.workingChildren().map((child) => child.agentThreadId)).toContain('announced') - }) + executions.register('settled-parent', 'settled parent', 'root') + executions.observeTurn('settled-parent', 'settled-turn', 'completed') + executions.register('live-descendant', 'live descendant', 'settled-parent') + executions.observeTurn('live-descendant', 'live-turn', 'working') + executions.register('sibling', 'sibling', 'root') + executions.observeTurn('sibling', 'sibling-turn', 'working') - it('past the bound, evicts the oldest never-started child once no child has ended', () => { - const executions = new CodexSubagentExecutions() - for (let index = 0; index < 200; index++) { - executions.register(`restored-${index}`, `restored-${index}`) - } - expect(executions.retentionSizes().children).toBe(128) - expect(executions.find('restored-0')).toBeUndefined() - expect(executions.find('restored-199')).toBeDefined() + expect(executions.lineage('root').map((child) => child.agentThreadId)).toEqual([ + 'root', + 'sibling', + 'settled-parent', + 'live-descendant', + ...finished.toReversed() + ]) + const tracker = new CodexBackgroundTaskTracker('primary', executions) + expect(tracker.subagentStopTargets(['root'])).toEqual({ + turns: [ + { threadId: 'root', turnId: 'root-turn' }, + { threadId: 'sibling', turnId: 'sibling-turn' }, + { threadId: 'live-descendant', turnId: 'live-turn' } + ], + terminals: [] + }) }) it('collapses a label’s whitespace, and keeps the first label a child was given', () => { diff --git a/src/main/codex/codex-subagent-executions.ts b/src/main/codex/codex-subagent-executions.ts index bd9623821bb..dc155bc99bb 100644 --- a/src/main/codex/codex-subagent-executions.ts +++ b/src/main/codex/codex-subagent-executions.ts @@ -4,9 +4,6 @@ import { canReplaceSubagentState } from '../../shared/native-chat-subagent-summary' -/** Children kept past this evict the oldest that ran and ended, else the oldest never started; a - * running child is always kept, however many run. */ -const MAX_CHILDREN = 128 const MAX_SETTLED_TURNS = 256 /** Turn ordinals remembered per child; a row from an older run reads as its first. */ const MAX_TURN_ORDINALS_PER_CHILD = 64 @@ -123,6 +120,15 @@ export class CodexSubagentExecutions { /** The child and every child it spawned, at any depth. */ lineage(agentThreadId: string): CodexExecutionChild[] { + const spawnedByThread = new Map() + for (const child of this.children.values()) { + if (child.spawnerThreadId === null) { + continue + } + const siblings = spawnedByThread.get(child.spawnerThreadId) ?? [] + siblings.push(child.agentThreadId) + spawnedByThread.set(child.spawnerThreadId, siblings) + } const lineage: CodexExecutionChild[] = [] const seen = new Set() const pending = [agentThreadId] @@ -133,10 +139,8 @@ export class CodexSubagentExecutions { } seen.add(id) lineage.push(child) - for (const spawned of this.children.values()) { - if (spawned.spawnerThreadId === id) { - pending.push(spawned.agentThreadId) - } + for (const spawned of spawnedByThread.get(id) ?? []) { + pending.push(spawned) } } return lineage @@ -163,17 +167,6 @@ export class CodexSubagentExecutions { if (existing) { return existing } - if (this.children.size >= MAX_CHILDREN) { - // Refusing a running child would hide it while it runs. One announced but not started is - // about to run, so it goes only when no child has ended; a restored history announces many. - const children = [...this.children.values()] - const evicted = - children.find((child) => child.execution !== null && child.execution.state !== 'working') ?? - children.find((child) => child.execution === null) - if (evicted) { - this.children.delete(evicted.agentThreadId) - } - } const child: CodexExecutionChild = { agentThreadId, registered: false, diff --git a/src/main/ipc/dashboard-payload-validation.ts b/src/main/ipc/dashboard-payload-validation.ts index 8463be64bba..2116b0b5a0f 100644 --- a/src/main/ipc/dashboard-payload-validation.ts +++ b/src/main/ipc/dashboard-payload-validation.ts @@ -22,7 +22,6 @@ import { isDashboardFilterOptions } from './dashboard-filter-payload-validation' export { isDashboardSpawnAgentArgs } from './dashboard-agent-launch-validation' const MAX_DASHBOARD_CARDS = 1_000 -const MAX_DASHBOARD_SUBAGENTS = 100 const MAX_DASHBOARD_REPO_ICONS = 500 // Why: sanitizing an image icon base64-decodes the whole data URI to read a // 24-byte header, and the renderer republishes the same icons every 250 ms. @@ -196,7 +195,6 @@ function isDashboardSubagents(value: unknown): boolean { } return ( Array.isArray(value) && - value.length <= MAX_DASHBOARD_SUBAGENTS && value.every((entry) => { if (!entry || typeof entry !== 'object' || Array.isArray(entry)) { return false diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-background-task-channel.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-background-task-channel.test.ts index 67bea7e7971..33443a00d8a 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-background-task-channel.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-background-task-channel.test.ts @@ -74,7 +74,7 @@ describe('the background-task channel', () => { // A task that finished while no pane listened, then an idle sweep that stopped the provider: // a pane resuming must read "none", not silence it would take as "unchanged" (#24227). - // An idle parent whose live background child the store had no room for: no row lists it, so + // An idle parent whose live background child the store refused: no row lists it, so // the strip must stay and say so rather than read as nothing running. it('keeps the strip, with its Stop, for work the host could not record', () => { const { sessions, channel, session } = channelOver( diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-subagent-row.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-subagent-row.test.ts index 5a3f70c2d1f..5f608639aa5 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-subagent-row.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-subagent-row.test.ts @@ -1,7 +1,7 @@ // A Claude background helper through the shipping adapter, host and child records: its saved row // stays working past its spawn result and its turn's end until its own end frame; and a helper the -// store had no room for keeps the strip incomplete, and the agent past idle cleanup, until the -// strip's Stop ends the agent. +// store refused keeps the strip incomplete, and the agent past idle cleanup, until the strip's Stop +// ends the agent. import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' @@ -9,8 +9,6 @@ import { join } from 'node:path' import { afterEach, beforeEach, expect, it, vi } from 'vitest' import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope' import { readAgentJournalTurn } from '../../../shared/agent-session-turn-record' -import { AGENT_STATUS_STORE_LIMITS } from '../../../shared/agent-status-store-contract' -import { makeStructuredAgentStatusSubject } from '../../../shared/agent-status-subject' import { isSubagentBlock } from '../../../shared/native-chat-types' import { activeStructuredAgentSessionTurnId } from '../../../shared/structured-agent-session-live-turn' import { acpChildWorkStatusSink } from '../../acp/acp-structured-child-work.test-support' @@ -184,38 +182,9 @@ it("keeps a background helper's row working past its spawn result and its turn's ) }) -it("keeps an agent whose helper the store had no room for past idle cleanup, until the strip's Stop", async () => { - // Another session holds every child record the store has room for. - const busy = makeStructuredAgentStatusSubject( - { executionHostId: 'local', wslDistro: null, workspaceId: 'busy', workspaceKind: 'folder' }, - 'busy-session' - ) - records.sink.publish( - { - sessionId: 'busy-session', - workspaceId: 'busy', - agent: 'claude', - status: 'working', - latestPrompt: 'Busy', - updatedAt: 1 - }, - busy - ) - records.sink.publishChildWork?.( - busy, - Array.from({ length: AGENT_STATUS_STORE_LIMITS.children }, (_unused, index) => ({ - type: 'live' as const, - observedAt: 1, - child: { - handle: { idKind: 'task_id' as const, id: `busy-${index}` }, - kind: 'agent' as const, - residency: 'background' as const, - state: 'working' as const, - stoppable: true - } - })), - 'claude' - ) +it("keeps an agent whose helper the store refused past idle cleanup, until the strip's Stop", async () => { + // The store refuses every child record this session reports. + records.sink.publishChildWork = () => undefined const connection = claude.connections[0]! await sendAndEcho(connection, 'Run it in the background.') for (const [index, captured] of RESUMED_BY_MESSAGE.filter(({ at }) => at <= 3_349).entries()) { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed-children.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed-children.test.ts index f0ec3c5dabc..b82e1653d35 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed-children.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed-children.test.ts @@ -1,5 +1,6 @@ // The status summary reads a session's child records from the sink its row landed in, and derives -// the legacy task list an older client folds from those same records. +// the legacy task list an older client folds from those same records. One chat's helpers are all +// recorded whatever their number, and its next accepted send retires every settled one. import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' @@ -13,6 +14,7 @@ import type { } from '../../../shared/agent-session-wire' import type { AgentChildWorkEvidence } from '../../../shared/agent-status-child-work-evidence' import type { AgentChildWorkView } from '../../../shared/agent-status-child-work-view' +import { acpChildWorkStatusSink } from '../../acp/acp-structured-child-work.test-support' import { createTrackedJournalOpener } from '../agent-session-journal/journal-host-database-test-support' import { StructuredAgentSessionStatusFeed, @@ -352,3 +354,78 @@ describe('structured status summary child records', () => { ) }) }) + +describe('structured status feed over the real child-record store', () => { + it('records every helper of a busy chat, and its next accepted send removes every settled one', async () => { + const journal = await journals.open({ + identity: { + sessionId: SESSION, + workspaceId: 'workspace-1', + hostId: 'local', + agent: 'codex', + providerHandle: codexProviderHandle('thread-1') + }, + stateDirectory: join(root, SESSION) + }) + const { sink, views } = acpChildWorkStatusSink() + const feed = new StructuredAgentSessionStatusFeed({ + logger: createStructuredAgentSessionLogger(), + sessions: new Map([[SESSION, indexedStatusFeedSession({ journal })]]), + getRecord: () => null, + now: () => 1_000, + statusSink: () => sink + }) + await journal.appendItem( + USER_IDENTITY, + { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'fan out' }] }, + { fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE } + ) + feed.publish(SESSION) + const live = (id: string, kind: 'agent' | 'command'): AgentChildWorkEvidence => ({ + type: 'live', + observedAt: 100, + child: { + handle: { idKind: 'task_id', id }, + kind, + residency: 'background', + state: 'working', + stoppable: true + } + }) + const finished = Array.from({ length: 2_049 }, (_, index) => `finished-${index}`) + feed.publishChildWork(SESSION, [ + ...finished.map((id) => live(id, 'agent')), + ...Array.from({ length: 40 }, (_, index) => live(`subagent-${index}`, 'agent')), + ...Array.from({ length: 300 }, (_, index) => live(`task-${index}`, 'command')) + ]) + feed.publishChildWork( + SESSION, + finished.map((id): AgentChildWorkEvidence => ({ + type: 'ended', + observedAt: 200, + handle: { idKind: 'task_id', id }, + outcome: 'succeeded' + })) + ) + expect(views()).toHaveLength(2_389) + expect(views().filter((view) => view.membership === 'settled')).toHaveLength(2_049) + expect(feed.childWorkIncomplete(SESSION)).toBe(false) + + await journal.appendSubmission({ + clientMessageId: 'send-1', + payloadFingerprint: 'send-1', + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'next' }] }, + fence: 1 + }) + await journal.resolveDispatch({ + clientMessageId: 'send-1', + state: 'accepted', + providerIdentity: { ...USER_IDENTITY, ordinal: USER_IDENTITY.ordinal + 1 }, + fence: 1 + }) + feed.publish(SESSION) + expect(views()).toHaveLength(340) + expect(views().every((view) => view.membership === 'live')).toBe(true) + expect(feed.childWorkIncomplete(SESSION)).toBe(false) + }, 60_000) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-status-ownership.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-status-ownership.test.ts index 03fba788aab..31f7ab677d4 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-status-ownership.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-status-ownership.test.ts @@ -5,7 +5,6 @@ import type { AgentChildWorkEvidence, AgentChildWorkLiveEvidence } from '../../../shared/agent-status-child-work-evidence' -import { AGENT_STATUS_STORE_LIMITS } from '../../../shared/agent-status-store-contract' import { makeStructuredAgentStatusSubject } from '../../../shared/agent-status-subject' import { AgentHookServer } from '../../agent-hooks/server' import { StructuredAgentSessionStatusOwnership } from './structured-agent-session-status-ownership' @@ -161,23 +160,21 @@ describe('child work the provider reported, recorded or not', () => { outcome: 'succeeded' }) - /** Another session holds every child record the store has room for; idle parent beside it. */ - function storeFull() { - const { owner } = owners() - owner.publish({ ...summary, sessionId: 'structured-busy' }, location) - owner.publishChildWork( - 'structured-busy', - Array.from({ length: AGENT_STATUS_STORE_LIMITS.children }, (_unused, index) => - agentLive(`busy-${index}`) - ), - 'claude' - ) + /** An idle parent whose child records the store refuses until `admit`. */ + function refusingStore() { + let refusing = true + const { owner } = owners(() => refusing) owner.publish({ ...summary, status: 'idle' }, location) - return owner + return { + owner, + admit: () => { + refusing = false + } + } } it('marks an idle parent incomplete when the store refuses its live background child', () => { - const owner = storeFull() + const { owner } = refusingStore() owner.publishChildWork(summary.sessionId, [agentLive('background-agent')], 'claude') expect(owner.readChildWork(summary.sessionId)).toEqual([]) @@ -194,18 +191,17 @@ describe('child work the provider reported, recorded or not', () => { }) it('is whole again once the refused child is admitted on a later report', () => { - const owner = storeFull() + const { owner, admit } = refusingStore() owner.publishChildWork(summary.sessionId, [agentLive('background-agent')], 'claude') expect(owner.childWorkIncomplete(summary.sessionId)).toBe(true) - // The other session goes, and with it the records that filled the store. - owner.forget('structured-busy') + admit() owner.publishChildWork(summary.sessionId, [agentLive('background-agent')], 'claude') expect(owner.readChildWork(summary.sessionId)).toHaveLength(1) expect(owner.childWorkIncomplete(summary.sessionId)).toBe(false) }) it('is whole again once every refused child reports its own end', () => { - const owner = storeFull() + const { owner } = refusingStore() owner.publishChildWork(summary.sessionId, [agentLive('first'), agentLive('second')], 'claude') owner.publishChildWork(summary.sessionId, [ended('first')], 'claude') expect(owner.childWorkIncomplete(summary.sessionId)).toBe(true) @@ -214,19 +210,6 @@ describe('child work the provider reported, recorded or not', () => { expect(owner.childWorkOpen(summary.sessionId)).toBe(false) }) - it('marks a parent the store refused incomplete, since none of its children is recorded', () => { - const { owner } = owners() - for (let index = 0; index < AGENT_STATUS_STORE_LIMITS.parents; index++) { - owner.publish({ ...summary, sessionId: `structured-other-${index}` }, location) - } - owner.publish({ ...summary, status: 'idle' }, location) - owner.publishChildWork(summary.sessionId, [agentLive('background-agent')], 'claude') - expect(owner.readChildWork(summary.sessionId)).toEqual([]) - expect(owner.childWorkIncomplete(summary.sessionId)).toBe(true) - owner.publishChildWork(summary.sessionId, [ended('background-agent')], 'claude') - expect(owner.childWorkIncomplete(summary.sessionId)).toBe(false) - }) - // Claude names the run a task ended on, and a resumed agent's earlier run may end late. it('holds recorded work open, never incomplete, through a late ending of its earlier run', () => { const { owner } = owners() diff --git a/src/main/pi/agent-status-extension-async-subagents.test.ts b/src/main/pi/agent-status-extension-async-subagents.test.ts index 5eb9196163c..da3fc0646ef 100644 --- a/src/main/pi/agent-status-extension-async-subagents.test.ts +++ b/src/main/pi/agent-status-extension-async-subagents.test.ts @@ -1,6 +1,10 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -import { AGENT_STATUS_MAX_SUBAGENTS } from '../../shared/agent-status-types' +import { + HOOK_REQUEST_MAX_BYTES, + parseAgentHookJson +} from '../../shared/agent-hook-listener/request-body' +import { normalizeAgentSubagentsField } from '../../shared/agent-status-subagent-snapshot' import { createAgentStatusExtensionHarness } from './agent-status-extension-test-harness' import { agentEndCount, @@ -309,18 +313,49 @@ describe('Pi child rows', () => { expect(childIds(posts(harness)[1])).toEqual(['run-b']) }) - // Why: the generated cap is interpolated from the host's, so a drift would show up here - // as an over-long roster the host would silently truncate on arrival. - it('caps the posted roster at the host limit', async () => { + it('posts every running child', async () => { const harness = createAgentStatusExtensionHarness({ kind: 'pi' }) await harness.callHook('agent_start') - for (let index = 0; index < AGENT_STATUS_MAX_SUBAGENTS + 4; index++) { + for (let index = 0; index < 40; index++) { startAsync(harness, `run-${index}`, 'scout') } await vi.advanceTimersByTimeAsync(0) - expect(childIds(posts(harness).at(-1))).toHaveLength(AGENT_STATUS_MAX_SUBAGENTS) - expect(childIds(posts(harness).at(-1))?.at(-1)).toBe(`run-${AGENT_STATUS_MAX_SUBAGENTS - 1}`) + expect(childIds(posts(harness).at(-1))).toHaveLength(40) + }) + + it.each(['pi', 'omp'] as const)('fits %s children using the host previews', async (kind) => { + const harness = createAgentStatusExtensionHarness({ kind }) + const children = [ + { + id: 'run-a', + agentType: `\u2003${'a'.repeat(39)}🙂${'a'.repeat(100)}`, + description: `\ufeff \t${'d'.repeat(159)}🙂${'d'.repeat(300_000)}` + }, + { + id: 'run-b', + agentType: ` \tReview\r\n\u2028 module\t\t${'a'.repeat(100)}`, + description: ` \tMap\r\n\u2029 module\t\t${'d'.repeat(300_000)}` + } + ] + await harness.callHook('agent_start') + for (const child of children) { + harness.emitPiEvent('task:subagent:lifecycle', { + ...child, + agent: child.agentType, + status: 'started' + }) + } + await vi.advanceTimersByTimeAsync(0) + + const body = String(harness.fetchMock.mock.calls.at(-1)?.[1]?.body) + expect(Buffer.byteLength(body)).toBeLessThanOrEqual(HOOK_REQUEST_MAX_BYTES) + expect(() => parseAgentHookJson(body)).not.toThrow() + expect(normalizeAgentSubagentsField(posts(harness).at(-1)?.subagents)).toEqual( + normalizeAgentSubagentsField( + children.map((child) => ({ ...child, state: 'working', startedAt: 0 })) + )?.map((child) => ({ ...child, startedAt: expect.any(Number) })) + ) }) it('posts nothing for the end of a run it is not tracking', async () => { diff --git a/src/main/pi/agent-status-extension-source.ts b/src/main/pi/agent-status-extension-source.ts index 5a3d23f47a7..78173b1a15a 100644 --- a/src/main/pi/agent-status-extension-source.ts +++ b/src/main/pi/agent-status-extension-source.ts @@ -124,8 +124,8 @@ export function getPiAgentStatusExtensionSource(kind: PiAgentKind = 'pi'): strin // Why: Pi resumes from an existing transcript; OMP resumes directly by session id (#8962). const payloadLine = kind !== 'omp' - ? ' payload: { hook_event_name: hookEventName, ...(ompRuntime ? metadata : getPersistedSessionMetadata(metadata)), ...(final ? {} : subagentPayload()), ...extra },' - : ' payload: { hook_event_name: hookEventName, ...metadata, ...(final ? {} : subagentPayload()), ...extra },' + ? ' payload: { hook_event_name: hookEventName, ...(ompRuntime ? metadata : getPersistedSessionMetadata(metadata)), ...subagents, ...extra },' + : ' payload: { hook_event_name: hookEventName, ...metadata, ...subagents, ...extra },' // Why: keep this string self-contained — it runs inside the pi process, // so it cannot import from Orca's main bundle. fs/http coords come from @@ -246,7 +246,7 @@ export function getPiAgentStatusExtensionSource(kind: PiAgentKind = 'pi'): strin ' const paneKey = process.env.ORCA_PANE_KEY', ' if (!coords.port || !coords.token || !paneKey) return', ' const url = `http://127.0.0.1:${coords.port}${resolveHookPath(ompRuntime)}`', - ' const body = JSON.stringify({', + ' const bodyWith = (subagents: Record): string => JSON.stringify({', ' paneKey,', " launchToken: process.env.ORCA_AGENT_LAUNCH_TOKEN || '',", " tabId: process.env.ORCA_TAB_ID || '',", @@ -255,6 +255,8 @@ export function getPiAgentStatusExtensionSource(kind: PiAgentKind = 'pi'): strin ' version: coords.version,', payloadLine, ' })', + ' const bareBody = bodyWith({})', + ' const body = final ? bareBody : bodyWith(subagentPayload(HOOK_BODY_BUDGET_BYTES - Buffer.byteLength(bareBody)))', " const controller = typeof AbortController !== 'undefined' ? new AbortController() : null", ' let timeout: ReturnType | undefined', ' const timeoutPromise = new Promise((_resolve, reject) => {', diff --git a/src/main/pi/agent-status-subagent-roster-source.ts b/src/main/pi/agent-status-subagent-roster-source.ts index a78cceec874..f462def4b76 100644 --- a/src/main/pi/agent-status-subagent-roster-source.ts +++ b/src/main/pi/agent-status-subagent-roster-source.ts @@ -2,7 +2,11 @@ // generated extension holds the pane's completion until every child it saw start is gone. import type { PiAgentKind } from '../../shared/pi-agent-kind' -import { AGENT_STATUS_MAX_SUBAGENTS } from '../../shared/agent-status-types' +import { HOOK_REQUEST_MAX_BYTES } from '../../shared/agent-hook-listener/request-body' +import { + AGENT_STATUS_TOOL_INPUT_MAX_LENGTH, + AGENT_TYPE_MAX_LENGTH +} from '../../shared/agent-status-field-normalization' // Module scope: post() reads the roster when a body is built, so a coalesced or // retried post always carries the children live at delivery. @@ -10,8 +14,10 @@ export function getPiSubagentSnapshotSourceLines(): string[] { return [ 'type SubagentDetail = { agentType?: string; description?: string; startedAt: number; workflow?: boolean; parent?: string; registration?: object }', 'type SubagentRoster = { active: Set; exited?: Set; details?: Map; waiting: boolean; ownsPane?: boolean; runGeneration?: number; endedRunGeneration?: number; completionPostedGeneration?: number; parked?: Map>; onEvent?: (event: unknown, forcedStatus?: string) => void; listener?: (event: unknown) => void; onRunnerExit?: (event: unknown) => void; runnerExitListener?: (event: unknown) => void; runnerExitCheck?: ReturnType | null; onRunnerExitSettled?: () => void }', - // Why: interpolated, not re-typed, so the extension cap cannot drift from the host's. - `const MAX_SUBAGENT_SNAPSHOT = ${AGENT_STATUS_MAX_SUBAGENTS}`, + // Why: the host refuses an oversized body whole. Half its limit also keeps the roster under + // the host's JSON structure-token limit; interpolated so it cannot drift from the host's. + `const HOOK_BODY_BUDGET_BYTES = ${Math.floor(HOOK_REQUEST_MAX_BYTES / 2)}`, + 'const SUBAGENTS_FIELD_BYTES = Buffer.byteLength(\',"subagents":[]\')', 'let subagentRoster: SubagentRoster | null = null', '', // Why: an exited runner is already gone, and a pi-subagents workflow run is the lead @@ -20,14 +26,31 @@ export function getPiSubagentSnapshotSourceLines(): string[] { ' return roster.active.has(id) && !roster.exited?.has(id) && roster.details?.get(id)?.workflow !== true', '}', '', - 'function subagentPayload(): Record {', + // Match the host's bounded single-line previews before reserving body bytes. + 'function subagentPreview(value: string | undefined, maxLength: number): string | undefined {', + " if (typeof value !== 'string') return undefined", + " let preview = value.slice(0, maxLength * 8 + 64).trimStart().replace(/[\\r\\n\\u2028\\u2029]+/g, ' ').slice(0, maxLength)", + ' if (preview.length < maxLength) preview = preview.trimEnd()', + ' else if (/[\\ud800-\\udbff]$/.test(preview)) preview = preview.slice(0, -1)', + ' return preview || undefined', + '}', + '', + // Only the posted rows are fitted to the body; the active set keeps every child. + 'function subagentPayload(byteBudget: number): Record {', ' if (!subagentRoster) return {}', ' const subagents: Record[] = []', + ' let bytes = SUBAGENTS_FIELD_BYTES', ' for (const id of subagentRoster.active) {', ' if (!isVisibleSubagent(subagentRoster, id)) continue', ' const detail = subagentRoster.details?.get(id)', - " subagents.push({ id, state: 'working', startedAt: detail?.startedAt ?? 0, ...(detail?.agentType ? { agentType: detail.agentType } : {}), ...(detail?.description ? { description: detail.description } : {}) })", - ' if (subagents.length >= MAX_SUBAGENT_SNAPSHOT) break', + ` const agentType = subagentPreview(detail?.agentType, ${AGENT_TYPE_MAX_LENGTH})`, + ` const description = subagentPreview(detail?.description, ${AGENT_STATUS_TOOL_INPUT_MAX_LENGTH})`, + " const subagent = { id, state: 'working', startedAt: detail?.startedAt ?? 0, ...(agentType ? { agentType } : {}), ...(description ? { description } : {}) }", + ' const rowBytes = Buffer.byteLength(JSON.stringify(subagent)) + (subagents.length > 0 ? 1 : 0)', + // A row that doesn't fit is skipped, not a stop: a later, shorter row may still fit. + ' if (bytes + rowBytes > byteBudget) continue', + ' bytes += rowBytes', + ' subagents.push(subagent)', ' }', ' return subagents.length > 0 ? { subagents } : {}', '}', diff --git a/src/shared/agent-hook-listener-claude-subagents.test.ts b/src/shared/agent-hook-listener-claude-subagents.test.ts index ec554205888..e8f19b0f55b 100644 --- a/src/shared/agent-hook-listener-claude-subagents.test.ts +++ b/src/shared/agent-hook-listener-claude-subagents.test.ts @@ -11,7 +11,6 @@ import { } from './agent-hook-listener/listener-state' import { normalizeHookPayload } from './agent-hook-listener' import { clearGrokSessionPathLookupCacheForTests } from './grok-session-paths' -import { AGENT_STATUS_MAX_SUBAGENTS } from './agent-status-types' import { makePaneKey } from './stable-pane-id' import { PANE_KEY } from './agent-hook-listener-test-harness' @@ -509,7 +508,7 @@ describe('shared agent-hook-listener', () => { ]) }) - it('keeps a live child omitted by the background task snapshot cap', () => { + it('keeps a live child omitted by an inventory with an unreadable entry', () => { claudeEvent({ hook_event_name: 'SubagentStart', agent_id: 'alive-after-cap', @@ -517,15 +516,14 @@ describe('shared agent-hook-listener', () => { }) const stop = claudeEvent({ hook_event_name: 'Stop', - background_tasks: Array.from({ length: AGENT_STATUS_MAX_SUBAGENTS + 1 }, (_, index) => ({ - id: index === AGENT_STATUS_MAX_SUBAGENTS ? 'alive-after-cap' : `a${index}`, - type: 'subagent', - status: 'running' - })) + background_tasks: [ + { id: 'a0', type: 'subagent', status: 'running' }, + { id: 'unreadable', status: 'running' } + ] }) - // Why: the inventory was capped before this id, so omission cannot - // prove the lifecycle-tracked child finished or was killed. + // Why: an unreadable entry may be this id, so omission cannot prove + // the lifecycle-tracked child finished or was killed. expect(stop?.payload.subagents).toContainEqual( expect.objectContaining({ id: 'alive-after-cap', state: 'working' }) ) diff --git a/src/shared/agent-hook-listener-startless-child-lifecycle.test.ts b/src/shared/agent-hook-listener-startless-child-lifecycle.test.ts index 4046eb3e645..70678dae1b6 100644 --- a/src/shared/agent-hook-listener-startless-child-lifecycle.test.ts +++ b/src/shared/agent-hook-listener-startless-child-lifecycle.test.ts @@ -5,7 +5,6 @@ import { } from './agent-hook-listener/listener-state' import { normalizeHookPayload } from './agent-hook-listener' import { seedClaudeSubagentRosterFromSnapshots } from './agent-hook-listener/providers/claude-roster-state' -import { AGENT_STATUS_MAX_SUBAGENTS } from './agent-status-types' import { makePaneKey } from './stable-pane-id' const LEAF_ID = '22222222-2222-4222-8222-222222222222' @@ -286,11 +285,10 @@ describe('Claude child lifecycle events with no cached lead state', () => { const stopped = claudeEvent(state, paneKey, { hook_event_name: 'Stop', - background_tasks: Array.from({ length: AGENT_STATUS_MAX_SUBAGENTS + 1 }, (_, index) => ({ - id: `finished-${index}`, - type: 'subagent', - status: 'completed' - })) + background_tasks: [ + { id: 'finished-0', type: 'subagent', status: 'completed' }, + { id: 'unreadable', status: 'completed' } + ] }) expect(stopped).toMatchObject({ diff --git a/src/shared/agent-hook-listener/providers/muse-events.ts b/src/shared/agent-hook-listener/providers/muse-events.ts index 7c4889bb14a..92b136090b4 100644 --- a/src/shared/agent-hook-listener/providers/muse-events.ts +++ b/src/shared/agent-hook-listener/providers/muse-events.ts @@ -13,8 +13,6 @@ import { import { extractToolFields, isNewTurnEvent } from '../provider-event-routing' import { readString } from '../tool-input-preview' -const MAX_TRACKED_CHILD_SESSIONS = 64 - function getMusePaneState(state: HookListenerState, paneKey: string): MusePaneState { let pane = state.musePaneStateByPaneKey.get(paneKey) if (!pane) { @@ -24,16 +22,6 @@ function getMusePaneState(state: HookListenerState, paneKey: string): MusePaneSt return pane } -function rememberChildSession(pane: MusePaneState, childSessionId: string): void { - pane.childSessionIds.add(childSessionId) - if (pane.childSessionIds.size > MAX_TRACKED_CHILD_SESSIONS) { - const oldest = pane.childSessionIds.values().next().value - if (oldest !== undefined) { - pane.childSessionIds.delete(oldest) - } - } -} - function isChildSessionEvent(pane: MusePaneState, hookPayload: Record): boolean { const sessionId = readString(hookPayload, 'session_id') if (!sessionId) { @@ -66,7 +54,7 @@ export function normalizeMuseEvent( const childSessionId = readString(hookPayload, 'child_session_id') ?? readString(hookPayload, 'session_id') if (childSessionId) { - rememberChildSession(pane, childSessionId) + pane.childSessionIds.add(childSessionId) } return null } diff --git a/src/shared/agent-session-background-task-wire.ts b/src/shared/agent-session-background-task-wire.ts index 100c674544d..c6db42ee917 100644 --- a/src/shared/agent-session-background-task-wire.ts +++ b/src/shared/agent-session-background-task-wire.ts @@ -55,10 +55,10 @@ export type AgentSessionBackgroundTaskState = { * advertise `agent-session.background-task-child-views.v1` never receives a state whose only * rows are settled: to it, as before, the roster is live work. */ children?: AgentChildWorkView[] - /** The host refused some of this session's child work for room, so what is listed is not all - * that runs; no row's absence means it ended. A refused child stops counting once a later - * report of it is admitted or it reports its own end; past the host's bound, and in every case - * once the provider session ends, the whole flag goes. */ + /** Some of this session's running child work has no row (its record was refused, has not landed + * yet, or a bounded frame dropped it), so what is listed is not all that runs; no row's absence + * means it ended. A missing child stops counting once a later report of it is admitted or it + * reports its own end, and the whole flag goes once the provider session ends. */ incomplete?: true } diff --git a/src/shared/agent-status-child-work-evidence-admission.ts b/src/shared/agent-status-child-work-evidence-admission.ts index c3a49136014..ba7ad1e3746 100644 --- a/src/shared/agent-status-child-work-evidence-admission.ts +++ b/src/shared/agent-status-child-work-evidence-admission.ts @@ -16,16 +16,12 @@ import { agentChildWorkHandleAliases, currentAgentChildWorkAliases, isPreviousAgentChildWorkRun, - ownedStructuredChildWork, resolveAgentChildWorkHandle, resolveAgentChildWorkOwner, STRUCTURED_CHILD_WORK_PROVENANCE, type AgentChildWorkEvidenceScope } from './agent-status-child-work-evidence-resolution' -/** Live children other than agents admitted per session. A subagent has no such bound: refusing - * one would cost its saved row while it runs, so only the store's global limits bound them. */ -export const STRUCTURED_CHILD_WORK_MAX_LIVE = 256 /** A child's first run, matching the provider roster's first attempt. */ const FIRST_GENERATION = 1 @@ -135,15 +131,6 @@ export function applyAgentChildWorkLive( const request = { parent: ctx.parent, provider: ctx.provider } const aliases = agentChildWorkHandleAliases(handle) if (!existing) { - if ( - child.kind !== 'agent' && - ownedStructuredChildWork(ctx).filter( - (record) => record.membership === 'live' && record.kind !== 'agent' - ).length >= STRUCTURED_CHILD_WORK_MAX_LIVE - ) { - ctx.outcome.rejected.push({ handleId: handle.id, reason: 'ingestion-limit' }) - return - } // Bindings with no record left are retired lifetimes: a forgotten session took its children // with it. The id is free again, under a generation past the retired one. const retired = resolution.highestGeneration diff --git a/src/shared/agent-status-child-work-projection.test.ts b/src/shared/agent-status-child-work-projection.test.ts index e9e3d87e8cf..6473d729d48 100644 --- a/src/shared/agent-status-child-work-projection.test.ts +++ b/src/shared/agent-status-child-work-projection.test.ts @@ -1,5 +1,4 @@ import { describe, expect, it } from 'vitest' -import { AGENT_STATUS_MAX_SUBAGENTS } from './agent-status-types' import { resolveAgentChildWorkFreshness } from './agent-status-child-work-freshness' import { agentChildWorkProjectionCandidateFromBackgroundTask, @@ -80,17 +79,6 @@ describe('agent child-work legacy projection', () => { ]) }) - it.each([31, 32, 33])('caps after accepting %i valid rows in source order', (count) => { - const interleaved = Array.from({ length: count }, (_, index) => [ - candidate('', { kind: 'agent' }), - candidate(`task-${index}`) - ]).flat() - const projected = projectAgentChildWorkLegacySubagents(interleaved) - - expect(projected).toHaveLength(Math.min(count, AGENT_STATUS_MAX_SUBAGENTS)) - expect(projected?.at(-1)?.id).toBe(`task-${Math.min(count, AGENT_STATUS_MAX_SUBAGENTS) - 1}`) - }) - it('rejects missing host first-observation time instead of inventing zero', () => { expect( projectAgentChildWorkLegacySubagents([ diff --git a/src/shared/agent-status-child-work-projection.ts b/src/shared/agent-status-child-work-projection.ts index 9d4e26230ca..3ea34ab0c86 100644 --- a/src/shared/agent-status-child-work-projection.ts +++ b/src/shared/agent-status-child-work-projection.ts @@ -1,5 +1,5 @@ import type { AgentSessionBackgroundTask } from './agent-session-background-task-wire' -import { AGENT_STATUS_MAX_SUBAGENTS, type AgentSubagentSnapshot } from './agent-status-types' +import type { AgentSubagentSnapshot } from './agent-status-types' import { isAgentChildWorkKind } from './agent-status-child-work-liveness' import type { AgentChildWorkOutcome, AgentChildWorkState } from './agent-status-child-work' import type { AgentChildWorkView } from './agent-status-child-work-view' @@ -115,9 +115,6 @@ export function projectAgentChildWorkLegacySubagents( ...(candidate.model !== undefined ? { model: candidate.model } : {}), ...(candidate.description !== undefined ? { description: candidate.description } : {}) }) - if (projected.length === AGENT_STATUS_MAX_SUBAGENTS) { - break - } } return projected.length > 0 ? projected : undefined } diff --git a/src/shared/agent-status-child-work-reconciliation.test.ts b/src/shared/agent-status-child-work-reconciliation.test.ts index e1b2b9772cc..479c4fd4487 100644 --- a/src/shared/agent-status-child-work-reconciliation.test.ts +++ b/src/shared/agent-status-child-work-reconciliation.test.ts @@ -6,7 +6,6 @@ import type { AgentChildWorkLiveObservation } from './agent-status-child-work-evidence' import { reconcileAgentChildWorkEvidence } from './agent-status-child-work-reconciliation' -import { STRUCTURED_CHILD_WORK_MAX_LIVE } from './agent-status-child-work-evidence-admission' import { createAgentStatusStore, type AgentStatusStore } from './agent-status-store' import { projectAgentChildWorkViews } from './agent-status-child-work-view' import { structuredRunningChildWork } from './agent-child-work-listing' @@ -385,21 +384,6 @@ describe('structured child-work reconciliation', () => { ]) }) - it('bounds the live tasks one session may admit, and never refuses a subagent for it', () => { - const { store, apply } = harness() - const command = (id: string) => child(id, { kind: 'command' }) - apply( - ...Array.from({ length: STRUCTURED_CHILD_WORK_MAX_LIVE }, (_, index) => - live(command(`task-${index}`)) - ) - ) - expect(apply(live(command('one-too-many'))).rejected).toEqual([ - { handleId: 'one-too-many', reason: 'ingestion-limit' } - ]) - expect(apply(live(child('agent-past-the-limit'))).rejected).toEqual([]) - expect(records(store)).toHaveLength(STRUCTURED_CHILD_WORK_MAX_LIVE + 1) - }) - it('keeps every settled child until the parent row goes', () => { const { store, apply } = harness() for (let index = 0; index < 100; index += 1) { diff --git a/src/shared/agent-status-run-alias-index.test.ts b/src/shared/agent-status-run-alias-index.test.ts index 1502d3e3546..32a2282d125 100644 --- a/src/shared/agent-status-run-alias-index.test.ts +++ b/src/shared/agent-status-run-alias-index.test.ts @@ -9,7 +9,6 @@ import { type AgentStatusRunAliasIndex } from './agent-status-run-alias-index' import type { AgentStatusExecutionScope } from './agent-status-subject' -import { AGENT_STATUS_STORE_LIMITS } from './agent-status-store-contract' function scope(overrides: Partial = {}): AgentStatusExecutionScope { return { @@ -74,11 +73,9 @@ describe('agent status provider alias index', () => { expect(decoded?.get(aliasKey)).toEqual(new Set(['run-a', 'run-b'])) }) - it('round-trips every run allowed by the canonical parent-store limit for one alias', () => { + it('round-trips more than 2,048 run ids for one alias', () => { const aliasKey = serializeAgentStatusProviderAliasKey(alias()) - const runIds = new Set( - Array.from({ length: AGENT_STATUS_STORE_LIMITS.parents }, (_, index) => `run-${index}`) - ) + const runIds = new Set(Array.from({ length: 2_049 }, (_, index) => `run-${index}`)) const decoded = deserializeAgentStatusRunAliasIndex( serializeAgentStatusRunAliasIndex(new Map([[aliasKey, runIds]])) diff --git a/src/shared/agent-status-run-alias-index.ts b/src/shared/agent-status-run-alias-index.ts index 80b917fd08f..2e012072d5e 100644 --- a/src/shared/agent-status-run-alias-index.ts +++ b/src/shared/agent-status-run-alias-index.ts @@ -8,13 +8,11 @@ import { parseAgentStatusExecutionScope, type AgentStatusExecutionScope } from './agent-status-subject' -import { AGENT_STATUS_STORE_LIMITS } from './agent-status-store-contract' import { assertJsonTextStructureWithinLimits } from './json-text-structure-limit' import { measureUtf8ByteLength } from './utf8-byte-limits' const PROVIDER_ALIAS_KEY_PREFIX = 'agent-status-provider-alias-v1:' const MAX_ALIAS_INDEX_ENTRIES = 4096 -const MAX_RUN_IDS_PER_ALIAS = AGENT_STATUS_STORE_LIMITS.parents const MAX_ALIAS_INDEX_RUN_REFERENCES = 16_384 const MAX_ALIAS_INDEX_SERIALIZED_BYTES = 4 * 1024 * 1024 const ALIAS_INDEX_JSON_STRUCTURE_LIMITS = { @@ -150,11 +148,7 @@ export function serializeAgentStatusRunAliasIndex( const entries: SerializedAliasIndexEntry[] = [] let runReferenceCount = 0 for (const [aliasKey, runIds] of index) { - if ( - !deserializeAgentStatusProviderAliasKey(aliasKey) || - runIds.size === 0 || - runIds.size > MAX_RUN_IDS_PER_ALIAS - ) { + if (!deserializeAgentStatusProviderAliasKey(aliasKey) || runIds.size === 0) { throw new Error('Invalid agent status alias index entry') } const serializedRunIds = [...runIds] @@ -207,7 +201,6 @@ export function deserializeAgentStatusRunAliasIndex( !deserializeAgentStatusProviderAliasKey(entry.alias) || !Array.isArray(entry.runIds) || entry.runIds.length === 0 || - entry.runIds.length > MAX_RUN_IDS_PER_ALIAS || !entry.runIds.every(isAgentStatusRunId) || new Set(entry.runIds).size !== entry.runIds.length || index.has(entry.alias) diff --git a/src/shared/agent-status-store-bounds.test.ts b/src/shared/agent-status-store-bounds.test.ts index 454d5ff843d..1be59503d90 100644 --- a/src/shared/agent-status-store-bounds.test.ts +++ b/src/shared/agent-status-store-bounds.test.ts @@ -1,9 +1,8 @@ -import { describe, expect, it, vi } from 'vitest' +import { describe, expect, it } from 'vitest' import type { AgentChildWorkAliasInput } from './agent-status-child-work-alias' import type { AgentChildWorkInput } from './agent-status-child-work' import { serializeAgentStatusRunAliasIndex } from './agent-status-run-alias-index' import { createAgentStatusStore } from './agent-status-store' -import { AGENT_STATUS_STORE_LIMITS } from './agent-status-store-contract' import { commitAgentStatusStoreMutation } from './agent-status-store-commit' import { indexAgentStatusStoreState } from './agent-status-store-indexes' import { agentStatusStoreStateFromSnapshot } from './agent-status-store-state' @@ -24,7 +23,7 @@ const scope: AgentStatusExecutionScope = { function child( parent: ReturnType, childIndex: number, - description = 'x'.repeat(8_000) + description: string ) { return { childWorkId: `child-${childIndex}`, @@ -42,23 +41,7 @@ function child( } satisfies AgentChildWorkInput } -function children(parent: ReturnType, offset: number) { - return Array.from({ length: AGENT_STATUS_STORE_LIMITS.mutationEntries / 2 }, (_, index) => - child(parent, offset + index) - ) -} - describe('AgentStatusStore bounds', () => { - it('rejects a mutation that would make the aggregate snapshot exceed its byte budget', () => { - const parent = makeStructuredAgentStatusSubject(scope, 'session-1') - const store = createAgentStatusStore({ epoch: 'epoch-a', mode: 'authority' }) - expect(store.applyMutation({ parent: { subject: parent } })).not.toBeNull() - expect(store.applyMutation({ children: children(parent, 0) })).not.toBeNull() - - expect(store.applyMutation({ children: children(parent, 1_024) })).toBeNull() - expect(store.getSnapshot().children).toHaveLength(1_024) - }) - it('serializes a derived provider-alias Set above the former 256-run ceiling', () => { const store = createAgentStatusStore({ epoch: 'epoch-a', mode: 'authority' }) for (let index = 0; index < 257; index += 1) { @@ -90,29 +73,6 @@ describe('AgentStatusStore bounds', () => { expect(() => serializeAgentStatusRunAliasIndex(store.getRunAliasIndex())).not.toThrow() }) - it('reuses immutable record byte measurements when one child changes', () => { - const parent = makeStructuredAgentStatusSubject(scope, 'session-1') - const store = createAgentStatusStore({ epoch: 'epoch-a', mode: 'authority' }) - const first = child(parent, 0, 'first-description') - const unchanged = child(parent, 1, 'unchanged-description') - expect( - store.applyMutation({ parent: { subject: parent }, children: [first, unchanged] }) - ).not.toBeNull() - const stringify = vi.spyOn(JSON, 'stringify') - - try { - expect( - store.applyMutation({ children: [{ ...first, observedAt: first.observedAt + 1 }] }) - ).not.toBeNull() - const serializedValues = stringify.mock.results - .map((result) => result.value) - .filter((value): value is string => typeof value === 'string') - expect(serializedValues.some((value) => value.includes('unchanged-description'))).toBe(false) - } finally { - stringify.mockRestore() - } - }) - it("removes a child batch's aliases without reading any other child's aliases", () => { const parent = makeStructuredAgentStatusSubject(scope, 'session-1') const bystander = makeStructuredAgentStatusSubject(scope, 'session-2') diff --git a/src/shared/agent-status-store-byte-budget.test.ts b/src/shared/agent-status-store-byte-budget.test.ts deleted file mode 100644 index ef49288d761..00000000000 --- a/src/shared/agent-status-store-byte-budget.test.ts +++ /dev/null @@ -1,75 +0,0 @@ -import { expect, it } from 'vitest' -import { createAgentStatusStore } from './agent-status-store' -import { AGENT_STATUS_STORE_LIMITS } from './agent-status-store-contract' -import { - deserializeAgentStatusStoreSnapshot, - serializeAgentStatusStoreSnapshot -} from './agent-status-store-persistence' -import { makeStructuredAgentStatusSubject } from './agent-status-subject' - -it('rejects cumulative snapshot overflow atomically even when each mutation fits', () => { - const subject = makeStructuredAgentStatusSubject( - { executionHostId: 'local', wslDistro: null, workspaceId: 'folder-a', workspaceKind: 'folder' }, - 'session_11111111-1111-4111-8111-111111111111' - ) - const store = createAgentStatusStore({ epoch: 'epoch-a', mode: 'authority' }) - expect(store.applyMutation({ parent: { subject } })).not.toBeNull() - const batch = (offset: number) => ({ - facts: Array.from({ length: 1024 }, (_, index) => ({ - subject, - key: `fact-${offset + index}`, - value: 'x'.repeat(4096) - })) - }) - for (const offset of [0, 1024, 2048]) { - expect(store.applyMutation(batch(offset))).not.toBeNull() - } - const before = store.getSnapshot() - expect(store.applyMutation(batch(3072))).toBeNull() - expect(store.getSnapshot()).toEqual(before) - expect(serializeAgentStatusStoreSnapshot(before).length).toBeLessThan( - AGENT_STATUS_STORE_LIMITS.serializedBytes - ) -}) - -it('can deserialize a dense valid snapshot within the declared record and byte budgets', () => { - const subject = makeStructuredAgentStatusSubject( - { executionHostId: 'local', wslDistro: null, workspaceId: 'folder-a', workspaceKind: 'folder' }, - 'session_11111111-1111-4111-8111-111111111111' - ) - const store = createAgentStatusStore({ epoch: 'epoch-a', mode: 'authority' }) - expect( - store.applySnapshot({ - version: 1, - epoch: 'persisted', - revision: 1, - parents: [{ subject, revision: 1 }], - children: Array.from({ length: 8192 }, (_, index) => ({ - childWorkId: `child-${index}`, - parent: subject, - provider: 'claude', - kind: 'agent', - state: 'working', - membership: 'live', - firstObservedAt: 10, - observedAt: 10, - stoppable: false, - invocation: { invocationId: 'invocation-1', generation: 1 }, - provenance: { source: 'structured-session', producerId: 'journal' }, - revision: 1 - })), - aliases: [], - facts: Array.from({ length: 8192 }, (_, index) => ({ - subject, - key: `fact-${index}`, - value: true, - revision: 1 - })), - tombstones: [] - }) - ).toBe(true) - const snapshot = store.getSnapshot() - expect(deserializeAgentStatusStoreSnapshot(serializeAgentStatusStoreSnapshot(snapshot))).toEqual( - snapshot - ) -}) diff --git a/src/shared/agent-status-store-byte-budget.ts b/src/shared/agent-status-store-byte-budget.ts deleted file mode 100644 index 1a64866c3e4..00000000000 --- a/src/shared/agent-status-store-byte-budget.ts +++ /dev/null @@ -1,77 +0,0 @@ -import type { AgentChildWorkAliasRecord } from './agent-status-child-work-alias' -import type { AgentChildWorkRecord } from './agent-status-child-work' -import { - AGENT_STATUS_STORE_LIMITS, - AGENT_STATUS_STORE_SNAPSHOT_VERSION, - type AgentStatusFactRecord, - type AgentStatusStoreSnapshot, - type AgentStatusTombstoneRecord -} from './agent-status-store-contract' -import type { AgentStatusParentRecord } from './agent-status-store-parent' -import type { AgentStatusStoreState } from './agent-status-store-state' -import { measureUtf8ByteLength } from './utf8-byte-limits' - -/** Either the snapshot header or a single owner/record measured while accumulating the budget. */ -type AgentStatusStoreByteBudgetRecord = - | AgentStatusStoreSnapshot - | AgentStatusParentRecord - | AgentChildWorkRecord - | AgentChildWorkAliasRecord - | AgentStatusFactRecord - | AgentStatusTombstoneRecord - -const recordBytes = new WeakMap() - -function serializedBytes(record: AgentStatusStoreByteBudgetRecord): number { - const cached = recordBytes.get(record) - if (cached !== undefined) { - return cached - } - const bytes = measureUtf8ByteLength(JSON.stringify(record)).byteLength - if (Object.isFrozen(record)) { - recordBytes.set(record, bytes) - } - return bytes -} - -/** One stored record's share of the serialized snapshot. */ -export function agentStatusStoreRecordBytes( - record: Exclude -): number { - return serializedBytes(record) -} - -/** The snapshot's own envelope, without any records or the commas between them. */ -export function agentStatusStoreHeaderBytes(epoch: string, revision: number): number { - return serializedBytes({ - version: AGENT_STATUS_STORE_SNAPSHOT_VERSION, - epoch, - revision, - parents: [], - children: [], - aliases: [], - facts: [], - tombstones: [] - }) -} - -/** Enforce the complete snapshot budget without allocating a full snapshot. */ -export function agentStatusStoreFitsByteBudget(state: AgentStatusStoreState): boolean { - let bytes = agentStatusStoreHeaderBytes(state.epoch, state.revision) - for (const records of [ - state.parents, - state.children, - state.aliases, - state.facts, - state.tombstones - ]) { - bytes += Math.max(0, records.size - 1) - for (const record of records.values()) { - bytes += serializedBytes(record) - if (bytes > AGENT_STATUS_STORE_LIMITS.serializedBytes) { - return false - } - } - } - return bytes <= AGENT_STATUS_STORE_LIMITS.serializedBytes -} diff --git a/src/shared/agent-status-store-codec.ts b/src/shared/agent-status-store-codec.ts index 326fec5aa72..ba060180f5c 100644 --- a/src/shared/agent-status-store-codec.ts +++ b/src/shared/agent-status-store-codec.ts @@ -7,8 +7,8 @@ import { parseAgentChildWorkRecord } from './agent-status-child-work-codec' import { - AGENT_STATUS_STORE_LIMITS, AGENT_STATUS_STORE_SNAPSHOT_VERSION, + AGENT_STATUS_STORE_TOMBSTONE_LIMIT, type AgentStatusStoreMutation, type AgentStatusStoreSnapshot, type AgentStatusTombstoneEntity, @@ -25,7 +25,6 @@ import { parseAgentStatusParentRecord } from './agent-status-store-parent' import { parseAgentStatusSubject } from './agent-status-subject' -import { measureUtf8ByteLength } from './utf8-byte-limits' const MAX_EPOCH_LENGTH = 256 const MAX_TOMBSTONE_KEY_LENGTH = 32_768 @@ -50,16 +49,6 @@ function isRevision(value: unknown): value is number { return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0 } -function isWithinSerializedLimit(value: unknown): boolean { - try { - return !measureUtf8ByteLength(JSON.stringify(value), { - stopAfterBytes: AGENT_STATUS_STORE_LIMITS.serializedBytes - }).exceededLimit - } catch { - return false - } -} - function isBoundedKey(value: unknown, maxLength: number): value is string { if ( typeof value !== 'string' || @@ -111,7 +100,7 @@ export function parseAgentStatusTombstoneRecord(value: unknown): AgentStatusTomb function parseArray( value: unknown, parser: (candidate: unknown) => T | null, - maxLength: number + maxLength = Number.POSITIVE_INFINITY ): T[] | null { if (!Array.isArray(value) || value.length > maxLength) { return null @@ -127,15 +116,8 @@ function parseArray( return parsed } -function parseMutationArray( - value: unknown, - parser: (candidate: unknown) => T | null -): T[] | null { - return parseArray(value, parser, AGENT_STATUS_STORE_LIMITS.mutationEntries) -} - function parseStringArray(value: unknown): string[] | null { - return parseMutationArray(value, (candidate) => + return parseArray(value, (candidate) => isBoundedKey(candidate, MAX_TOMBSTONE_KEY_LENGTH) ? candidate : null ) } @@ -153,7 +135,6 @@ export function parseAgentStatusStoreMutation(value: unknown): AgentStatusStoreM 'tombstones' ] if ( - !isWithinSerializedLimit(value) || !isRecord(value) || !hasOnlyKeys(value, [], optionalKeys) || Object.keys(value).length === 0 @@ -164,29 +145,25 @@ export function parseAgentStatusStoreMutation(value: unknown): AgentStatusStoreM const removeParent = value.removeParent === undefined ? undefined : parseAgentStatusSubject(value.removeParent) const children = - value.children === undefined - ? undefined - : parseMutationArray(value.children, parseAgentChildWorkInput) + value.children === undefined ? undefined : parseArray(value.children, parseAgentChildWorkInput) const removeChildren = value.removeChildren === undefined ? undefined : parseStringArray(value.removeChildren) const aliases = value.aliases === undefined ? undefined - : parseMutationArray(value.aliases, parseAgentChildWorkAliasInput) + : parseArray(value.aliases, parseAgentChildWorkAliasInput) const removeAliases = value.removeAliases === undefined ? undefined : parseStringArray(value.removeAliases) const facts = - value.facts === undefined - ? undefined - : parseMutationArray(value.facts, parseAgentStatusFactInput) + value.facts === undefined ? undefined : parseArray(value.facts, parseAgentStatusFactInput) const removeFacts = value.removeFacts === undefined ? undefined - : parseMutationArray(value.removeFacts, parseAgentStatusFactIdentity) + : parseArray(value.removeFacts, parseAgentStatusFactIdentity) const tombstones = value.tombstones === undefined ? undefined - : parseMutationArray(value.tombstones, parseAgentStatusTombstoneInput) + : parseArray(value.tombstones, parseAgentStatusTombstoneInput) const parsedValues = [ parent, removeParent, @@ -202,18 +179,6 @@ export function parseAgentStatusStoreMutation(value: unknown): AgentStatusStoreM if (sourceValues.some((item, index) => item !== undefined && !parsedValues[index])) { return null } - const mutationEntryCount = [ - children, - removeChildren, - aliases, - removeAliases, - facts, - removeFacts, - tombstones - ].reduce((sum, items) => sum + (items?.length ?? 0), 0) - if (mutationEntryCount > AGENT_STATUS_STORE_LIMITS.mutationEntries) { - return null - } return { ...(parent ? { parent } : {}), ...(removeParent ? { removeParent } : {}), @@ -239,7 +204,6 @@ export function parseAgentStatusStoreSnapshot(value: unknown): AgentStatusStoreS 'tombstones' ] if ( - !isWithinSerializedLimit(value) || !isRecord(value) || !hasOnlyKeys(value, keys) || value.version !== AGENT_STATUS_STORE_SNAPSHOT_VERSION || @@ -248,26 +212,14 @@ export function parseAgentStatusStoreSnapshot(value: unknown): AgentStatusStoreS ) { return null } - const parents = parseArray( - value.parents, - parseAgentStatusParentRecord, - AGENT_STATUS_STORE_LIMITS.parents - ) - const children = parseArray( - value.children, - parseAgentChildWorkRecord, - AGENT_STATUS_STORE_LIMITS.children - ) - const aliases = parseArray( - value.aliases, - parseAgentChildWorkAliasRecord, - AGENT_STATUS_STORE_LIMITS.aliases - ) - const facts = parseArray(value.facts, parseAgentStatusFactRecord, AGENT_STATUS_STORE_LIMITS.facts) + const parents = parseArray(value.parents, parseAgentStatusParentRecord) + const children = parseArray(value.children, parseAgentChildWorkRecord) + const aliases = parseArray(value.aliases, parseAgentChildWorkAliasRecord) + const facts = parseArray(value.facts, parseAgentStatusFactRecord) const tombstones = parseArray( value.tombstones, parseAgentStatusTombstoneRecord, - AGENT_STATUS_STORE_LIMITS.tombstones + AGENT_STATUS_STORE_TOMBSTONE_LIMIT ) return parents && children && aliases && facts && tombstones ? { diff --git a/src/shared/agent-status-store-commit.ts b/src/shared/agent-status-store-commit.ts index 2ea07e07f5d..2174c34aed2 100644 --- a/src/shared/agent-status-store-commit.ts +++ b/src/shared/agent-status-store-commit.ts @@ -3,11 +3,8 @@ // valid, so a record the mutation did not touch, and whose dependencies it did not touch, still is. import type { AgentStatusStoreMutation } from './agent-status-store-contract' -import { agentStatusStoreHeaderBytes } from './agent-status-store-byte-budget' -import { AGENT_STATUS_STORE_LIMITS } from './agent-status-store-contract' import { commitAgentStatusStoreIndexes, - draftedRecordBytes, inMapOrder, type AgentStatusStoreDrafts, type AgentStatusStoreIndexes @@ -17,7 +14,6 @@ import { type AgentStatusStoreMutationTables } from './agent-status-store-mutation' import { - agentStatusStoreSizesFit, storedAliasIsValid, storedChildIsValid, storedFactIsValid, @@ -156,19 +152,6 @@ function touchedAreValid( return true } -function fitsByteBudget( - epoch: string, - revision: number, - drafts: AgentStatusStoreDrafts, - recordBytes: Record -): boolean { - let bytes = agentStatusStoreHeaderBytes(epoch, revision) - for (const name of ['parents', 'children', 'aliases', 'facts', 'tombstones'] as const) { - bytes += Math.max(0, drafts[name].size - 1) + recordBytes[name] - } - return bytes <= AGENT_STATUS_STORE_LIMITS.serializedBytes -} - /** Apply one mutation to the committed store in place; false (and nothing changed) on refusal. */ export function commitAgentStatusStoreMutation( state: AgentStatusStoreState, @@ -179,16 +162,11 @@ export function commitAgentStatusStoreMutation( const { drafts, tables } = draftTables(state, indexes, revision) if ( !applyAgentStatusStoreMutationSteps(tables, mutation, revision) || - !agentStatusStoreSizesFit(drafts) || !touchedAreValid(drafts, tables) ) { return false } - const recordBytes = draftedRecordBytes(indexes, drafts) - if (!fitsByteBudget(state.epoch, revision, drafts, recordBytes)) { - return false - } - commitAgentStatusStoreIndexes(indexes, drafts, recordBytes) + commitAgentStatusStoreIndexes(indexes, drafts) drafts.parents.commitInto(state.parents) drafts.children.commitInto(state.children) drafts.aliases.commitInto(state.aliases) diff --git a/src/shared/agent-status-store-contract.ts b/src/shared/agent-status-store-contract.ts index 8c9fc96b1c3..5464d73d79b 100644 --- a/src/shared/agent-status-store-contract.ts +++ b/src/shared/agent-status-store-contract.ts @@ -7,16 +7,7 @@ import type { AgentStatusParentInput, AgentStatusParentRecord } from './agent-st import type { AgentStatusSubject } from './agent-status-subject' export const AGENT_STATUS_STORE_SNAPSHOT_VERSION = 1 as const -export const AGENT_STATUS_STORE_LIMITS = { - parents: 2_048, - children: 8_192, - aliases: 16_384, - facts: 16_384, - tombstones: 1_024, - mutationEntries: 2_048, - serializedBytes: 16 * 1024 * 1024 -} as const - +export const AGENT_STATUS_STORE_TOMBSTONE_LIMIT = 1_024 export const AGENT_STATUS_STORE_TOMBSTONE_RETENTION_REVISIONS = 4_096 export type AgentStatusFactValue = string | number | boolean | null diff --git a/src/shared/agent-status-store-in-place.test.ts b/src/shared/agent-status-store-in-place.test.ts index 0d7720cb0da..85d77eaf5a0 100644 --- a/src/shared/agent-status-store-in-place.test.ts +++ b/src/shared/agent-status-store-in-place.test.ts @@ -6,7 +6,7 @@ import { createAgentStatusStore } from './agent-status-store' import { parseAgentStatusStoreMutation } from './agent-status-store-codec' import { commitAgentStatusStoreMutation } from './agent-status-store-commit' import { - AGENT_STATUS_STORE_LIMITS, + AGENT_STATUS_STORE_TOMBSTONE_LIMIT, AGENT_STATUS_STORE_TOMBSTONE_RETENTION_REVISIONS, type AgentStatusStoreMutation } from './agent-status-store-contract' @@ -122,7 +122,6 @@ function indexContents(indexes: AgentStatusStoreIndexes) { const ranked = (order: Map) => [...order].sort((left, right) => left[1] - right[1]).map(([key]) => key) return { - recordBytes: indexes.recordBytes, childrenByParent: sets(indexes.childrenByParent), factsByParent: sets(indexes.factsByParent), aliasesByChild: sets(indexes.aliasesByChild), @@ -149,7 +148,6 @@ describe('AgentStatusStore applied in place', () => { if (mutation) { commitAgentStatusStoreMutation(state, indexes, mutation, state.revision + 1) } - // The running byte total is only observable at the budget, so it is held to a rebuild. expect({ step, ...indexContents(indexes) }).toEqual({ step, ...indexContents(indexAgentStatusStoreState(state)) @@ -190,7 +188,7 @@ describe('AgentStatusStore applied in place', () => { it('keeps the newest tombstones in order when their count overflows', () => { const store = createAgentStatusStore({ epoch: 'epoch-a', mode: 'authority' }) const childWorkIds = Array.from( - { length: AGENT_STATUS_STORE_LIMITS.tombstones }, + { length: AGENT_STATUS_STORE_TOMBSTONE_LIMIT }, (_, index) => `removed-${index}` ) for (const mutation of [{ removeChildren: ['oldest'] }, { removeChildren: childWorkIds }]) { diff --git a/src/shared/agent-status-store-indexes.ts b/src/shared/agent-status-store-indexes.ts index c423ab989b4..2a9858bda77 100644 --- a/src/shared/agent-status-store-indexes.ts +++ b/src/shared/agent-status-store-indexes.ts @@ -6,7 +6,6 @@ import { type AgentChildWorkAliasRecord } from './agent-status-child-work-alias' import type { AgentChildWorkRecord } from './agent-status-child-work' -import { agentStatusStoreRecordBytes } from './agent-status-store-byte-budget' import type { AgentStatusFactRecord, AgentStatusTombstoneRecord @@ -22,8 +21,6 @@ type TableName = (typeof TABLE_NAMES)[number] type OrderedTableName = Exclude export type AgentStatusStoreIndexes = { - /** Serialized bytes of every record per table, for the snapshot budget. */ - recordBytes: Record childrenByParent: Map> factsByParent: Map> aliasesByChild: Map> @@ -109,7 +106,6 @@ function indexTombstone( export function indexAgentStatusStoreState(state: AgentStatusStoreState): AgentStatusStoreIndexes { const indexes: AgentStatusStoreIndexes = { - recordBytes: { parents: 0, children: 0, aliases: 0, facts: 0, tombstones: 0 }, childrenByParent: new Map(), factsByParent: new Map(), aliasesByChild: new Map(), @@ -118,48 +114,25 @@ export function indexAgentStatusStoreState(state: AgentStatusStoreState): AgentS order: { children: new Map(), aliases: new Map(), facts: new Map(), tombstones: new Map() }, nextOrder: 0 } - for (const parent of state.parents.values()) { - indexes.recordBytes.parents += agentStatusStoreRecordBytes(parent) - } for (const [id, child] of state.children) { - indexes.recordBytes.children += agentStatusStoreRecordBytes(child) indexes.order.children.set(id, indexes.nextOrder++) indexChild(indexes, id, child, true) } for (const [key, alias] of state.aliases) { - indexes.recordBytes.aliases += agentStatusStoreRecordBytes(alias) indexes.order.aliases.set(key, indexes.nextOrder++) indexAlias(indexes, key, alias, true) } for (const [key, fact] of state.facts) { - indexes.recordBytes.facts += agentStatusStoreRecordBytes(fact) indexes.order.facts.set(key, indexes.nextOrder++) indexFact(indexes, key, fact, true) } for (const [key, tombstone] of state.tombstones) { - indexes.recordBytes.tombstones += agentStatusStoreRecordBytes(tombstone) indexes.order.tombstones.set(key, indexes.nextOrder++) indexTombstone(indexes, key, tombstone, true) } return indexes } -/** Record bytes per table once the drafts land. */ -export function draftedRecordBytes( - indexes: AgentStatusStoreIndexes, - drafts: AgentStatusStoreDrafts -): Record { - const bytes = { ...indexes.recordBytes } - for (const name of TABLE_NAMES) { - for (const { previous, next } of drafts[name].touched()) { - bytes[name] += - (next ? agentStatusStoreRecordBytes(next) : 0) - - (previous ? agentStatusStoreRecordBytes(previous) : 0) - } - } - return bytes -} - function moveIndex( draft: AgentStatusStoreDraftTable, index: (key: string, record: V, add: boolean) => void @@ -191,10 +164,8 @@ function moveOrder( /** Move every index by what the drafts changed; call before the drafts land in the maps. */ export function commitAgentStatusStoreIndexes( indexes: AgentStatusStoreIndexes, - drafts: AgentStatusStoreDrafts, - recordBytes: Record + drafts: AgentStatusStoreDrafts ): void { - indexes.recordBytes = recordBytes moveIndex(drafts.children, (key, child, add) => indexChild(indexes, key, child, add)) moveIndex(drafts.aliases, (key, alias, add) => indexAlias(indexes, key, alias, add)) moveIndex(drafts.facts, (key, fact, add) => indexFact(indexes, key, fact, add)) diff --git a/src/shared/agent-status-store-mutation.ts b/src/shared/agent-status-store-mutation.ts index d22f4e17c9f..9849c3c4652 100644 --- a/src/shared/agent-status-store-mutation.ts +++ b/src/shared/agent-status-store-mutation.ts @@ -13,7 +13,7 @@ import type { AgentStatusTombstoneRecord } from './agent-status-store-contract' import { - AGENT_STATUS_STORE_LIMITS, + AGENT_STATUS_STORE_TOMBSTONE_LIMIT, AGENT_STATUS_STORE_TOMBSTONE_RETENTION_REVISIONS } from './agent-status-store-contract' import { parseAgentStatusFactRecord } from './agent-status-store-fact-codec' @@ -63,7 +63,7 @@ function addTombstone( function compactTombstones(state: Tables): void { for (const [key, tombstone] of state.tombstones.entries()) { if ( - state.tombstones.size <= AGENT_STATUS_STORE_LIMITS.tombstones && + state.tombstones.size <= AGENT_STATUS_STORE_TOMBSTONE_LIMIT && state.revision - tombstone.revision < AGENT_STATUS_STORE_TOMBSTONE_RETENTION_REVISIONS ) { break diff --git a/src/shared/agent-status-store-persistence.ts b/src/shared/agent-status-store-persistence.ts index ffab63d4557..b98ad962eb2 100644 --- a/src/shared/agent-status-store-persistence.ts +++ b/src/shared/agent-status-store-persistence.ts @@ -1,43 +1,27 @@ -import { - AGENT_STATUS_STORE_LIMITS, - type AgentStatusStoreSnapshot -} from './agent-status-store-contract' +import type { AgentStatusStoreSnapshot } from './agent-status-store-contract' import { parseAgentStatusStoreSnapshot } from './agent-status-store-codec' import { assertJsonTextStructureWithinLimits } from './json-text-structure-limit' -import { measureUtf8ByteLength } from './utf8-byte-limits' -const SNAPSHOT_STRUCTURE_LIMITS = { - structuralTokens: AGENT_STATUS_STORE_LIMITS.serializedBytes, - nestingDepth: 32 -} as const +const SNAPSHOT_NESTING_DEPTH = 32 -function isWithinByteLimit(value: string): boolean { - return !measureUtf8ByteLength(value, { - stopAfterBytes: AGENT_STATUS_STORE_LIMITS.serializedBytes - }).exceededLimit -} - -/** Bounded persistence form; callers write the returned string with their existing owner. */ +/** Persistence form; callers write the returned string with their existing owner. */ export function serializeAgentStatusStoreSnapshot(snapshot: AgentStatusStoreSnapshot): string { const parsed = parseAgentStatusStoreSnapshot(snapshot) if (!parsed) { throw new Error('Invalid agent status store snapshot') } - const serialized = JSON.stringify(parsed) - if (!isWithinByteLimit(serialized)) { - throw new Error('Agent status store snapshot exceeds its serialized-byte limit') - } - return serialized + return JSON.stringify(parsed) } export function deserializeAgentStatusStoreSnapshot( serialized: string ): AgentStatusStoreSnapshot | null { - if (!isWithinByteLimit(serialized)) { - return null - } try { - assertJsonTextStructureWithinLimits(serialized, SNAPSHOT_STRUCTURE_LIMITS) + // Size follows the store, which its owners' lifetimes bound; only nesting is limited here. + assertJsonTextStructureWithinLimits(serialized, { + structuralTokens: serialized.length, + nestingDepth: SNAPSHOT_NESTING_DEPTH + }) return parseAgentStatusStoreSnapshot(JSON.parse(serialized)) } catch { return null diff --git a/src/shared/agent-status-store-state.ts b/src/shared/agent-status-store-state.ts index 20c6d9ea875..207b2f49bc2 100644 --- a/src/shared/agent-status-store-state.ts +++ b/src/shared/agent-status-store-state.ts @@ -9,10 +9,9 @@ import { type AgentChildWorkRecord } from './agent-status-child-work' import { parseAgentChildWorkRecord } from './agent-status-child-work-codec' -import { agentStatusStoreFitsByteBudget } from './agent-status-store-byte-budget' import { - AGENT_STATUS_STORE_LIMITS, AGENT_STATUS_STORE_SNAPSHOT_VERSION, + AGENT_STATUS_STORE_TOMBSTONE_LIMIT, AGENT_STATUS_STORE_TOMBSTONE_RETENTION_REVISIONS, type AgentStatusFactIdentity, type AgentStatusFactRecord, @@ -112,22 +111,6 @@ export type AgentStatusStoreRecordLookups = { tombstones: Lookup } -export function agentStatusStoreSizesFit(sizes: { - parents: { size: number } - children: { size: number } - aliases: { size: number } - facts: { size: number } - tombstones: { size: number } -}): boolean { - return ( - sizes.parents.size <= AGENT_STATUS_STORE_LIMITS.parents && - sizes.children.size <= AGENT_STATUS_STORE_LIMITS.children && - sizes.aliases.size <= AGENT_STATUS_STORE_LIMITS.aliases && - sizes.facts.size <= AGENT_STATUS_STORE_LIMITS.facts && - sizes.tombstones.size <= AGENT_STATUS_STORE_LIMITS.tombstones - ) -} - export function storedParentIsValid( lookups: AgentStatusStoreRecordLookups, key: string, @@ -196,7 +179,7 @@ export function storedTombstoneIsValid( /** Every invariant over the whole store: for snapshot restore, and the oracle mutations are held to. */ export function validateAgentStatusStoreState(state: AgentStatusStoreState): boolean { - if (!agentStatusStoreSizesFit(state)) { + if (state.tombstones.size > AGENT_STATUS_STORE_TOMBSTONE_LIMIT) { return false } for (const [key, parent] of state.parents) { @@ -224,7 +207,7 @@ export function validateAgentStatusStoreState(state: AgentStatusStoreState): boo return false } } - return agentStatusStoreFitsByteBudget(state) + return true } export function snapshotFromAgentStatusStoreState( diff --git a/src/shared/agent-status-store-status-codec.ts b/src/shared/agent-status-store-status-codec.ts index 21f5bcb135e..8fd4786b959 100644 --- a/src/shared/agent-status-store-status-codec.ts +++ b/src/shared/agent-status-store-status-codec.ts @@ -48,16 +48,22 @@ function isBoundedString(value: unknown): value is string { function copyJsonRecord(value: unknown): Record | null { let serialized: string + let bounded: string try { serialized = JSON.stringify(value) + // Why: a pane's subagent roster has no count limit; each row's fields are bounded on their own. + bounded = + isRecord(value) && Object.hasOwn(value, 'subagents') + ? JSON.stringify({ ...value, subagents: undefined }) + : serialized } catch { return null } - if (measureUtf8ByteLength(serialized, { stopAfterBytes: MAX_STATUS_BYTES }).exceededLimit) { + if (measureUtf8ByteLength(bounded, { stopAfterBytes: MAX_STATUS_BYTES }).exceededLimit) { return null } try { - assertJsonTextStructureWithinLimits(serialized, STATUS_STRUCTURE_LIMITS) + assertJsonTextStructureWithinLimits(bounded, STATUS_STRUCTURE_LIMITS) const parsed: unknown = JSON.parse(serialized) return isRecord(parsed) ? parsed : null } catch { diff --git a/src/shared/agent-status-store.test.ts b/src/shared/agent-status-store.test.ts index 86ddd90e953..dfbc149d892 100644 --- a/src/shared/agent-status-store.test.ts +++ b/src/shared/agent-status-store.test.ts @@ -1,11 +1,11 @@ -import { describe, expect, it, vi } from 'vitest' +import { describe, expect, it } from 'vitest' import type { AgentStatusIpcPayload } from './agent-status-ipc-payload' import type { AgentChildWorkAliasInput } from './agent-status-child-work-alias' import type { AgentChildWorkInput } from './agent-status-child-work' import { serializeAgentStatusProviderAliasKey } from './agent-status-run-alias-index' import { createAgentStatusStore } from './agent-status-store' import { - AGENT_STATUS_STORE_LIMITS, + AGENT_STATUS_STORE_TOMBSTONE_LIMIT, AGENT_STATUS_STORE_TOMBSTONE_RETENTION_REVISIONS, type AgentStatusStoreSnapshot } from './agent-status-store-contract' @@ -260,12 +260,12 @@ describe('AgentStatusStore', () => { expect( store.applyMutation({ removeChildren: Array.from( - { length: AGENT_STATUS_STORE_LIMITS.tombstones + 1 }, + { length: AGENT_STATUS_STORE_TOMBSTONE_LIMIT + 1 }, (_, index) => `unused-child-${index}` ) }) ).not.toBeNull() - expect(store.getSnapshot().tombstones).toHaveLength(AGENT_STATUS_STORE_LIMITS.tombstones) + expect(store.getSnapshot().tombstones).toHaveLength(AGENT_STATUS_STORE_TOMBSTONE_LIMIT) expect(store.getSnapshot().tombstones.some((item) => item.entity === 'parent')).toBe(false) expect(replica.applySnapshot(store.getSnapshot())).toBe(true) expect(replica.applyTransportEnvelope(first)).toBe(false) @@ -371,24 +371,8 @@ describe('AgentStatusStore', () => { expect(restarted.getChildren(parent)[0]?.firstObservedAt).toBe(12) }) - it('fails closed before parsing oversized persistence payloads or allocating huge mutations', () => { - const parse = vi.spyOn(JSON, 'parse') - const oversized = ' '.repeat(AGENT_STATUS_STORE_LIMITS.serializedBytes + 1) - - expect(deserializeAgentStatusStoreSnapshot(oversized)).toBeNull() - expect(parse).not.toHaveBeenCalled() - parse.mockRestore() - + it('refuses a parent status beyond its field limit', () => { const store = createAgentStatusStore({ epoch: 'epoch-a', mode: 'authority' }) - expect( - store.applyMutation({ - removeChildren: Array.from( - { length: AGENT_STATUS_STORE_LIMITS.mutationEntries + 1 }, - (_, index) => `child-${index}` - ) - }) - ).toBeNull() - expect(store.getSnapshot().revision).toBe(0) const scopedSubject = subject() expect( store.applyMutation({ @@ -407,6 +391,22 @@ describe('AgentStatusStore', () => { ).toBeNull() }) + it('accepts a parent status whose subagent roster outgrows the status field limit', () => { + const store = createAgentStatusStore({ epoch: 'epoch-a', mode: 'authority' }) + const scopedSubject = subject() + const subagents = Array.from({ length: 2_000 }, (_, index) => ({ + id: `a${index}`, + state: 'working' as const, + startedAt: 1, + description: 'x'.repeat(200) + })) + expect( + store.applyMutation({ + parent: { subject: scopedSubject, status: status(scopedSubject, { subagents }) } + }) + ).not.toBeNull() + }) + it('rejects malformed and scope-mismatched snapshots without changing state', () => { const { store } = populatedStore() const before = store.getSnapshot() diff --git a/src/shared/agent-status-subagent-snapshot.ts b/src/shared/agent-status-subagent-snapshot.ts index 4f10b1e89bb..9ce8952729b 100644 --- a/src/shared/agent-status-subagent-snapshot.ts +++ b/src/shared/agent-status-subagent-snapshot.ts @@ -24,9 +24,6 @@ export type AgentSubagentSnapshot = { startedAt: number } -/** Maximum subagent child rows carried per status entry. Bounds per-pane cache - * and IPC fanout against a runaway spawner. */ -export const AGENT_STATUS_MAX_SUBAGENTS = 32 export const AGENT_SUBAGENT_ID_MAX_LENGTH = 64 function normalizeSubagentSnapshot( @@ -81,9 +78,6 @@ function readAgentSubagents( const snapshot = normalizeSubagentSnapshot(item, allowFutureState) if (snapshot) { normalized.push(snapshot) - if (normalized.length >= AGENT_STATUS_MAX_SUBAGENTS) { - break - } } } return normalized.length > 0 ? normalized : undefined diff --git a/src/shared/agent-status-transport-envelope.test.ts b/src/shared/agent-status-transport-envelope.test.ts index 81007bc7e39..6e5307bc4f4 100644 --- a/src/shared/agent-status-transport-envelope.test.ts +++ b/src/shared/agent-status-transport-envelope.test.ts @@ -1,6 +1,5 @@ -import { describe, expect, it, vi } from 'vitest' +import { describe, expect, it } from 'vitest' import { createAgentStatusStore } from './agent-status-store' -import { AGENT_STATUS_STORE_LIMITS } from './agent-status-store-contract' import { makeStructuredAgentStatusSubject } from './agent-status-subject' import { deserializeAgentStatusTransportEnvelope, @@ -68,7 +67,7 @@ describe('agent status transport envelope', () => { expect(replica.getSnapshot().facts.map((fact) => fact.key)).toEqual(['retained']) }) - it('round-trips a mutation envelope and fails closed on malformed or oversized input', () => { + it('round-trips a mutation envelope and fails closed on malformed input', () => { const authority = createAgentStatusStore({ epoch: 'epoch-a', mode: 'authority' }) const mutation = authority.applyMutation({ parent: { subject: parent } }) expect(mutation).not.toBeNull() @@ -84,15 +83,6 @@ describe('agent status transport envelope', () => { JSON.stringify({ ...mutation, revision: mutation.revision + 2 }) ) ).toBeNull() - - const parse = vi.spyOn(JSON, 'parse') - expect( - deserializeAgentStatusTransportEnvelope( - ' '.repeat(AGENT_STATUS_STORE_LIMITS.serializedBytes + 1) - ) - ).toBeNull() - expect(parse).not.toHaveBeenCalled() - parse.mockRestore() }) it('does not resurrect an exact removal through stale replay', () => { diff --git a/src/shared/agent-status-transport-envelope.ts b/src/shared/agent-status-transport-envelope.ts index 51fd0800db8..9b1134a5876 100644 --- a/src/shared/agent-status-transport-envelope.ts +++ b/src/shared/agent-status-transport-envelope.ts @@ -1,7 +1,6 @@ -import { - AGENT_STATUS_STORE_LIMITS, - type AgentStatusStoreMutation, - type AgentStatusStoreSnapshot +import type { + AgentStatusStoreMutation, + AgentStatusStoreSnapshot } from './agent-status-store-contract' import { isAgentStatusStoreEpoch, @@ -9,10 +8,10 @@ import { parseAgentStatusStoreSnapshot } from './agent-status-store-codec' import { assertJsonTextStructureWithinLimits } from './json-text-structure-limit' -import { measureUtf8ByteLength } from './utf8-byte-limits' +// Why: the store has no size limit, so only nesting is bounded; any envelope it can serialize must parse. const ENVELOPE_STRUCTURE_LIMITS = { - structuralTokens: 512 * 1024, + structuralTokens: Number.MAX_SAFE_INTEGER, nestingDepth: 40 } as const @@ -83,27 +82,12 @@ export function serializeAgentStatusTransportEnvelope( if (!parsed) { throw new Error('Invalid agent status transport envelope') } - const serialized = JSON.stringify(parsed) - if ( - measureUtf8ByteLength(serialized, { - stopAfterBytes: AGENT_STATUS_STORE_LIMITS.serializedBytes - }).exceededLimit - ) { - throw new Error('Agent status transport envelope exceeds its serialized-byte limit') - } - return serialized + return JSON.stringify(parsed) } export function deserializeAgentStatusTransportEnvelope( serialized: string ): AgentStatusTransportEnvelope | null { - if ( - measureUtf8ByteLength(serialized, { - stopAfterBytes: AGENT_STATUS_STORE_LIMITS.serializedBytes - }).exceededLimit - ) { - return null - } try { assertJsonTextStructureWithinLimits(serialized, ENVELOPE_STRUCTURE_LIMITS) return parseAgentStatusTransportEnvelope(JSON.parse(serialized)) diff --git a/src/shared/agent-status-types.test.ts b/src/shared/agent-status-types.test.ts index b591fae683c..59bbfff318a 100644 --- a/src/shared/agent-status-types.test.ts +++ b/src/shared/agent-status-types.test.ts @@ -8,7 +8,6 @@ import { pickParsedAgentStatusPayload, AGENT_STATUS_JSON_STRUCTURE_LIMITS, AGENT_STATUS_MAX_FIELD_LENGTH, - AGENT_STATUS_MAX_SUBAGENTS, AGENT_STATUS_TOOL_NAME_MAX_LENGTH, AGENT_STATUS_TOOL_INPUT_MAX_LENGTH, AGENT_STATUS_ASSISTANT_MESSAGE_MAX_LENGTH, @@ -572,7 +571,7 @@ Fix dispatch fallback preview for normalized status prompts` } }) - it('normalizes the subagents field, dropping invalid entries and bounding count', () => { + it('normalizes the subagents field, dropping invalid entries and keeping every valid one', () => { const result = parseAgentStatusPayload( JSON.stringify({ state: 'working', @@ -583,7 +582,7 @@ Fix dispatch fallback preview for normalized status prompts` { id: '', state: 'working', startedAt: 1 }, { id: 'bad-state', state: 'running', startedAt: 1 }, 'garbage', - ...Array.from({ length: AGENT_STATUS_MAX_SUBAGENTS + 5 }, (_, i) => ({ + ...Array.from({ length: 37 }, (_, i) => ({ id: `extra-${i}`, state: 'idle', startedAt: i @@ -591,7 +590,7 @@ Fix dispatch fallback preview for normalized status prompts` ] }) ) - expect(result?.subagents?.length).toBe(AGENT_STATUS_MAX_SUBAGENTS) + expect(result?.subagents?.length).toBe(40) expect(result?.subagents?.[0]).toEqual({ id: 'a1', state: 'working', diff --git a/src/shared/agent-status-types.ts b/src/shared/agent-status-types.ts index a61ec507be8..9e9827884fb 100644 --- a/src/shared/agent-status-types.ts +++ b/src/shared/agent-status-types.ts @@ -35,7 +35,6 @@ export { AGENT_TYPE_MAX_LENGTH } from './agent-status-field-normalization' export { - AGENT_STATUS_MAX_SUBAGENTS, agentSubagentsEqual, type AgentSubagentSnapshot, type AgentSubagentState diff --git a/src/shared/claude-background-task-inventory.ts b/src/shared/claude-background-task-inventory.ts index ed94e821ce5..bbffb79a1ed 100644 --- a/src/shared/claude-background-task-inventory.ts +++ b/src/shared/claude-background-task-inventory.ts @@ -1,4 +1,3 @@ -import { AGENT_STATUS_MAX_SUBAGENTS } from './agent-status-types' import { classifyClaudeBackgroundTaskKind } from './claude-background-task-kind' import { isAgentChildWorkKind } from './agent-status-child-work-liveness' @@ -94,12 +93,6 @@ export function readClaudeBackgroundAgentTasks(hookPayload: Record= AGENT_STATUS_MAX_SUBAGENTS) { - // Why: a capped inventory cannot prove a tracked id is absent; callers - // must retain unlisted rows rather than deleting live overflow tasks. - truncated = true - continue - } tasks.push({ id: obj.id.trim(), agentType: typeof obj.agent_type === 'string' ? obj.agent_type : undefined, diff --git a/src/shared/claude-background-task-status.test.ts b/src/shared/claude-background-task-status.test.ts index b823dd53238..03c2b490cbb 100644 --- a/src/shared/claude-background-task-status.test.ts +++ b/src/shared/claude-background-task-status.test.ts @@ -11,7 +11,6 @@ import { seedClaudeSubagentRosterFromSnapshots } from './agent-hook-listener/providers/claude-roster-state' import { normalizeHookPayload } from './agent-hook-listener' -import { AGENT_STATUS_MAX_SUBAGENTS } from './agent-status-types' import { readClaudeBackgroundAgentTasks } from './claude-background-task-inventory' import { makePaneKey } from './stable-pane-id' @@ -30,8 +29,8 @@ function claudeEvent(state: HookListenerState, paneKey: string, payload: Record< } describe('Claude background task status', () => { - it('finds pending monitor work after the visible child-row cap', () => { - const agentTasks = Array.from({ length: AGENT_STATUS_MAX_SUBAGENTS + 1 }, (_, index) => ({ + it('lists every agent row and still finds pending monitor work after them', () => { + const agentTasks = Array.from({ length: 40 }, (_, index) => ({ id: `agent-${index}`, type: 'subagent', status: 'running' @@ -40,8 +39,8 @@ describe('Claude background task status', () => { background_tasks: [...agentTasks, { id: 'monitor-1', type: 'monitor', status: 'pending' }] }) - expect(result).toMatchObject({ truncated: true, hasRunningNonAgentTask: true }) - expect(result.tasks).toHaveLength(AGENT_STATUS_MAX_SUBAGENTS) + expect(result).toMatchObject({ truncated: false, hasRunningNonAgentTask: true }) + expect(result.tasks).toHaveLength(40) const state = createHookListenerState() expect( diff --git a/src/shared/claude-subagent-roster.test.ts b/src/shared/claude-subagent-roster.test.ts index 3ccb8b67a82..7bd53290ffb 100644 --- a/src/shared/claude-subagent-roster.test.ts +++ b/src/shared/claude-subagent-roster.test.ts @@ -1,5 +1,4 @@ import { describe, expect, it } from 'vitest' -import { AGENT_STATUS_MAX_SUBAGENTS } from './agent-status-types' import { readClaudeBackgroundAgentTasks } from './claude-background-task-inventory' import { claudeRosterHasRuntimeWorkingSubagent, @@ -114,58 +113,12 @@ describe('claude-subagent-roster', () => { expect(roster.size).toBe(0) }) - it('drops new spawns at the cap rather than evicting working children', () => { + it('keeps 40 working subagents', () => { const roster: ClaudeSubagentRoster = new Map() - for (let i = 0; i < AGENT_STATUS_MAX_SUBAGENTS; i++) { + for (let i = 0; i < 40; i++) { upsertWorkingClaudeSubagent(roster, `a${i}`, {}, i) } - // Why: every tracked entry is working — nothing is safe to evict, so the - // overflow spawn is dropped (it would be invisible past the wire cap). - upsertWorkingClaudeSubagent(roster, 'overflow', {}, 999) - expect(roster.has('overflow')).toBe(false) - expect(roster.size).toBe(AGENT_STATUS_MAX_SUBAGENTS) - - // Once a child finishes, a new spawn takes the freed slot. - stopClaudeSubagent(roster, 'a0') - upsertWorkingClaudeSubagent(roster, 'replacement', {}, 1000) - expect(roster.has('replacement')).toBe(true) - expect(roster.size).toBe(AGENT_STATUS_MAX_SUBAGENTS) - }) - - it('evicts the oldest idle teammate to admit a new spawn at the cap', () => { - const roster: ClaudeSubagentRoster = new Map() - upsertWorkingClaudeSubagent(roster, 'aold-teammate-6d3cb5b5', {}, 1) - upsertWorkingClaudeSubagent(roster, 'anew-teammate-6d3cb5b5', {}, 2) - stopClaudeSubagent(roster, 'aold-teammate-6d3cb5b5') - stopClaudeSubagent(roster, 'anew-teammate-6d3cb5b5') - for (let i = 2; i < AGENT_STATUS_MAX_SUBAGENTS; i++) { - upsertWorkingClaudeSubagent(roster, `a${i}`, {}, 10 + i) - } - // Why: a parked idle row is the only thing safe to displace — a working - // spawn must never be dropped just because idle teammates fill the cap. - upsertWorkingClaudeSubagent(roster, 'overflow', {}, 999) - expect(roster.has('overflow')).toBe(true) - expect(roster.has('aold-teammate-6d3cb5b5')).toBe(false) - expect(roster.has('anew-teammate-6d3cb5b5')).toBe(true) - expect(roster.size).toBe(AGENT_STATUS_MAX_SUBAGENTS) - }) - - it('reconciles stale entries before adding replacement tasks at the cap', () => { - const roster: ClaudeSubagentRoster = new Map() - for (let i = 0; i < AGENT_STATUS_MAX_SUBAGENTS; i++) { - upsertWorkingClaudeSubagent(roster, `a${i}`, {}, i) - } - const tasks = Array.from({ length: AGENT_STATUS_MAX_SUBAGENTS }, (_, index) => - task({ id: index === 0 ? 'replacement' : `a${index}` }) - ) - - // Why: a complete inventory can replace a stale child while the roster is - // full. Reap the stale slot first so the new live child keeps the done-gate. - foldClaudeBackgroundTasksIntoRoster(roster, tasks, 999) - - expect(roster.has('a0')).toBe(false) - expect(roster.has('replacement')).toBe(true) - expect(roster.size).toBe(AGENT_STATUS_MAX_SUBAGENTS) + expect(claudeRosterToSnapshots(roster)).toHaveLength(40) }) it('reads only agent-typed background_tasks entries', () => { @@ -208,17 +161,6 @@ describe('claude-subagent-roster', () => { expect(readClaudeBackgroundAgentTasks({ background_tasks: 'nope' }).present).toBe(false) }) - it('marks a background task inventory truncated after the snapshot cap', () => { - const tasks = Array.from({ length: AGENT_STATUS_MAX_SUBAGENTS + 1 }, (_, index) => ({ - id: `a${index}`, - type: 'subagent', - status: 'running' - })) - const result = readClaudeBackgroundAgentTasks({ background_tasks: tasks }) - expect(result.tasks).toHaveLength(AGENT_STATUS_MAX_SUBAGENTS) - expect(result.truncated).toBe(true) - }) - it('trusts id-exact subagent matches and ignores teammate-typed entries', () => { const roster: ClaudeSubagentRoster = new Map() upsertWorkingClaudeSubagent(roster, 'a1', {}, 100) @@ -319,19 +261,19 @@ describe('claude-subagent-roster', () => { it('retains an unlisted live child when the background task inventory was truncated', () => { const roster: ClaudeSubagentRoster = new Map() - upsertWorkingClaudeSubagent(roster, 'alive-after-cap', {}, 100) + upsertWorkingClaudeSubagent(roster, 'alive-unlisted', {}, 100) const parsed = readClaudeBackgroundAgentTasks({ - background_tasks: Array.from({ length: AGENT_STATUS_MAX_SUBAGENTS + 1 }, (_, index) => ({ - id: index === AGENT_STATUS_MAX_SUBAGENTS ? 'alive-after-cap' : `a${index}`, - type: 'subagent', - status: 'running' - })) + background_tasks: [ + { id: 'a0', type: 'subagent', status: 'running' }, + { id: 'unreadable', status: 'running' } + ] }) + expect(parsed.truncated).toBe(true) foldClaudeBackgroundTasksIntoRoster(roster, parsed.tasks, 200, { inventoryComplete: !parsed.truncated }) - expect(roster.has('alive-after-cap')).toBe(true) + expect(roster.has('alive-unlisted')).toBe(true) }) it('recreates unmatched running one-shot subagents after a listener restart', () => { @@ -428,7 +370,7 @@ describe('claude-subagent-roster', () => { it('does not idle an unrelated one-shot whose agent_type matches the teammate name', () => { const roster: ClaudeSubagentRoster = new Map() - // Why: a teammate's start hook may be missing (restart, cap, or lost + // Why: a teammate's start hook may be missing (restart or lost // delivery). Agent type is not identity, so its idle hook must not park // another live child that happens to use the same type name. upsertWorkingClaudeSubagent(roster, 'aoneshot00000001', { agentType: 'reviewer' }, 100) diff --git a/src/shared/claude-subagent-roster.ts b/src/shared/claude-subagent-roster.ts index 533d6233f5d..2ea71d279aa 100644 --- a/src/shared/claude-subagent-roster.ts +++ b/src/shared/claude-subagent-roster.ts @@ -1,4 +1,4 @@ -import { AGENT_STATUS_MAX_SUBAGENTS, type AgentSubagentSnapshot } from './agent-status-types' +import type { AgentSubagentSnapshot } from './agent-status-types' import type { ClaudeBackgroundAgentTask } from './claude-background-task-inventory' /** Mirrors the wire-normalization id cap in agent-status-types. Enforced at @@ -93,11 +93,6 @@ export function upsertWorkingClaudeSubagent( existing.restoredFromSnapshot = undefined return } - // Why: beyond the wire cap extra rows would be invisible anyway; idle - // teammates are the only safe eviction — never displace a working child. - if (roster.size >= AGENT_STATUS_MAX_SUBAGENTS && !evictOldestIdleClaudeSubagent(roster)) { - return - } roster.set(id, { state: 'working', startedAt: now, @@ -106,22 +101,6 @@ export function upsertWorkingClaudeSubagent( }) } -function evictOldestIdleClaudeSubagent(roster: ClaudeSubagentRoster): boolean { - let oldestId: string | null = null - let oldestStartedAt = Infinity - for (const [id, tracked] of roster) { - if (tracked.state === 'idle' && tracked.startedAt < oldestStartedAt) { - oldestId = id - oldestStartedAt = tracked.startedAt - } - } - if (oldestId === null) { - return false - } - roster.delete(oldestId) - return true -} - /** SubagentStop. A one-shot child is finished — the row leaves immediately. * A teammate-shaped id is only ending a TURN on claude 2.1.21x (the teammate * stays alive awaiting mail), so its row flips to idle instead — unless a @@ -174,7 +153,6 @@ export function foldClaudeBackgroundTasksIntoRoster( return } const listedIds = new Set() - const pendingRunningTasks = new Map() const hasTeammateTypedTask = tasks.some((task) => task.teammate) for (const task of tasks) { if (task.teammate) { @@ -185,7 +163,6 @@ export function foldClaudeBackgroundTasksIntoRoster( if (existing) { if (!task.running) { roster.delete(task.id) - pendingRunningTasks.delete(task.id) continue } // Why: a Stop can park the row before the lead inventory confirms the @@ -200,7 +177,6 @@ export function foldClaudeBackgroundTasksIntoRoster( continue } if (!task.running) { - pendingRunningTasks.delete(task.id) continue } upsertWorkingClaudeSubagent( @@ -213,10 +189,6 @@ export function foldClaudeBackgroundTasksIntoRoster( if (created) { created.backgroundTasksAuthoritative = true created.listedAsSubagentTask = true - } else { - // Why: a full roster may still contain stale entries that this same - // inventory will reap. Retry after cleanup so a replacement stays live. - pendingRunningTasks.set(task.id, task) } } if (options?.inventoryComplete !== false) { @@ -239,22 +211,6 @@ export function foldClaudeBackgroundTasksIntoRoster( roster.delete(id) } } - for (const task of pendingRunningTasks.values()) { - if (roster.size >= AGENT_STATUS_MAX_SUBAGENTS) { - break - } - upsertWorkingClaudeSubagent( - roster, - task.id, - { agentType: task.agentType, description: task.description }, - now - ) - const created = roster.get(task.id) - if (created) { - created.backgroundTasksAuthoritative = true - created.listedAsSubagentTask = true - } - } } /** Drop restored rows that no current-runtime child activity has confirmed. */ diff --git a/src/shared/codex-subagent-roster.test.ts b/src/shared/codex-subagent-roster.test.ts index 9d759850476..8b35d608689 100644 --- a/src/shared/codex-subagent-roster.test.ts +++ b/src/shared/codex-subagent-roster.test.ts @@ -1,9 +1,5 @@ import { describe, expect, it } from 'vitest' -import { - AGENT_MODEL_MAX_LENGTH, - AGENT_STATUS_MAX_SUBAGENTS, - AGENT_TYPE_MAX_LENGTH -} from './agent-status-types' +import { AGENT_MODEL_MAX_LENGTH, AGENT_TYPE_MAX_LENGTH } from './agent-status-types' import { codexRosterChildWorkLiveness, codexRosterToSnapshots, @@ -48,22 +44,6 @@ describe('Codex subagent roster', () => { expect(roster.size).toBe(0) }) - it('bounds live storage while admitting a replacement after one child stops', () => { - const roster: CodexSubagentRoster = new Map() - for (let index = 0; index <= AGENT_STATUS_MAX_SUBAGENTS; index += 1) { - upsertCodexSubagent(roster, `child-${index}`, { state: 'working' }, index) - } - - expect(roster.size).toBe(AGENT_STATUS_MAX_SUBAGENTS) - expect(roster.has(`child-${AGENT_STATUS_MAX_SUBAGENTS}`)).toBe(false) - - finishCodexSubagent(roster, 'child-0') - upsertCodexSubagent(roster, 'replacement', { state: 'working' }, 100) - - expect(roster.size).toBe(AGENT_STATUS_MAX_SUBAGENTS) - expect(roster.has('replacement')).toBe(true) - }) - describe('codexRosterChildWorkLiveness', () => { it('reads every child as agent work, a waiting one above the rest, and nothing as a watch loop', () => { const roster: CodexSubagentRoster = new Map() diff --git a/src/shared/codex-subagent-roster.ts b/src/shared/codex-subagent-roster.ts index 805d459e059..37f8d6f909e 100644 --- a/src/shared/codex-subagent-roster.ts +++ b/src/shared/codex-subagent-roster.ts @@ -1,6 +1,5 @@ import { AGENT_MODEL_MAX_LENGTH, - AGENT_STATUS_MAX_SUBAGENTS, AGENT_STATUS_TOOL_INPUT_MAX_LENGTH, AGENT_TYPE_MAX_LENGTH, type AgentSubagentSnapshot @@ -49,9 +48,6 @@ export function upsertCodexSubagent( existing.state = fields.state return } - if (roster.size >= AGENT_STATUS_MAX_SUBAGENTS) { - return - } roster.set(normalizedId, { agentType, description, diff --git a/src/shared/program-status/agent-status-projector.ts b/src/shared/program-status/agent-status-projector.ts index b01fab8853c..684791fc89c 100644 --- a/src/shared/program-status/agent-status-projector.ts +++ b/src/shared/program-status/agent-status-projector.ts @@ -1,10 +1,7 @@ // Projects OSC 7501 records onto agent-status payloads so they ride the OSC 9999 terminal-status ingress. import type { AgentStatusState, ParsedAgentStatusPayload } from '../agent-status-types' import type { AgentSubagentSnapshot, AgentSubagentState } from '../agent-status-subagent-snapshot' -import { - AGENT_STATUS_MAX_SUBAGENTS, - AGENT_SUBAGENT_ID_MAX_LENGTH -} from '../agent-status-subagent-snapshot' +import { AGENT_SUBAGENT_ID_MAX_LENGTH } from '../agent-status-subagent-snapshot' import type { AgentTurnOutcome } from '../agent-turn-outcome' import { createProgramStatusOscScanner } from './osc-scanner' import { @@ -160,7 +157,7 @@ export function createProgramStatusProjector(now: () => number = Date.now): Prog childStartedAt.delete(id) } } - return children.slice(-AGENT_STATUS_MAX_SUBAGENTS) + return children } const project = (): ParsedAgentStatusPayload | null => {