fix(agent-status): never drop a running helper for lack of room (#30416)

* fix(agent-status): never drop a running helper for lack of room

Helper state (subagents, background tasks and commands) was bounded by counts and byte budgets that refused or evicted live records once hit, so the agent kept working while Orca listed it wrong. Every one of these records already dies with its session, pane or chat, so the count and size caps go; identity-history, transcript and transport bounds stay.

* fix(agent-status): keep the status field limit, measured without the subagent roster

The previous commit dropped the 256 KiB status check outright; it still guards every other field. Only the roster, whose rows are bounded individually, is left out of the measurement.

* fix(agent-status): fit helper previews and walk lineage once

* fix(agent-status): skip an oversized Pi row instead of stopping, and parse every envelope the store can write

Review follow-up: one long row no longer hides later rows that still fit the request, and the transport envelope's reader no longer refuses a size its writer now produces.
This commit is contained in:
Brennan Benson
2026-10-19 00:31:42 -07:00
committed by GitHub
parent add1eccae3
commit 978b207cc1
71 changed files with 465 additions and 1270 deletions
+17 -7
View File
@@ -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<string, NativeChatBackgroundTaskBlock>({
private readonly running = new Map<string, NativeChatBackgroundTaskBlock>()
private readonly settled = new BoundedMap<string, NativeChatBackgroundTaskBlock>({
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',
+2 -7
View File
@@ -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<string, KnownSubagent>({
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<string, KnownSubagent>()
private readonly pending: AgentChildWorkEvidence[] = []
/** Known children include settled ones, so they cannot become background tasks. */
@@ -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 () => {
@@ -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()
}
})
})
@@ -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',
@@ -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<string, TrackedClaudeBackgroundTask>,
roster: ReadonlyMap<string, TrackedClaudeBackgroundTask>,
limit: number
roster: ReadonlyMap<string, TrackedClaudeBackgroundTask>
): Map<string, TrackedClaudeBackgroundTask> {
const combined = new Map<string, TrackedClaudeBackgroundTask>()
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)
}
@@ -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.
@@ -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<TrackedClaudeBackgroundTask, 'liveInTurn'>): 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)
@@ -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}` })
+2 -12
View File
@@ -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<string, DecodedClaudeTask>()
/** 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,
@@ -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({
@@ -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<string, Command>()
private readonly settled = new Map<string, number>()
private liveBytes = 0
private settledBytes = 0
private readonly settled = new Set<string>()
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 }
}
}
@@ -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,
@@ -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
@@ -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<string> = 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
+10 -37
View File
@@ -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<string, RetainedState>()
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
}
@@ -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<string, AgentJournalItemBody>()
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')
@@ -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,
@@ -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<string, number>()
const pendingCheckpoints = new Set<string>()
// 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)
@@ -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()
@@ -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
@@ -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()
@@ -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<string, Set<string>>()
/** Host turn-start receipt per remembered turn; the terminal row carries it forward. */
private readonly startedAtByTurn = new Map<string, number>()
@@ -25,34 +22,11 @@ export class CodexJournalActiveTurns {
private readonly requestOriginByTurn = new Map<string, CodexJournalRequestOrigin>()
/** Last dispatch armed before each provider turn-start event. */
private readonly latestDispatchSequenceByTurn = new Map<string, number>()
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
}
}
@@ -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()
@@ -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)
}
@@ -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)
@@ -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', () => {
+11 -18
View File
@@ -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<string, string[]>()
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<string>()
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,
@@ -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
@@ -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(
@@ -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()) {
@@ -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)
})
@@ -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()
@@ -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 () => {
+5 -3
View File
@@ -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, unknown>): 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<typeof setTimeout> | undefined',
' const timeoutPromise = new Promise<never>((_resolve, reject) => {',
@@ -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<string>; exited?: Set<string>; details?: Map<string, SubagentDetail>; waiting: boolean; ownsPane?: boolean; runGeneration?: number; endedRunGeneration?: number; completionPostedGeneration?: number; parked?: Map<string, Map<string, SubagentDetail>>; onEvent?: (event: unknown, forcedStatus?: string) => void; listener?: (event: unknown) => void; onRunnerExit?: (event: unknown) => void; runnerExitListener?: (event: unknown) => void; runnerExitCheck?: ReturnType<typeof setTimeout> | 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<string, unknown> {',
// 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<string, unknown> {',
' if (!subagentRoster) return {}',
' const subagents: Record<string, unknown>[] = []',
' 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 } : {}',
'}',
@@ -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' })
)
@@ -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({
@@ -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<string, unknown>): 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
}
@@ -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
}
@@ -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
@@ -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([
@@ -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
}
@@ -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) {
@@ -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> = {}): 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]]))
+1 -8
View File
@@ -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)
+2 -42
View File
@@ -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<typeof makeStructuredAgentStatusSubject>,
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<typeof makeStructuredAgentStatusSubject>, 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')
@@ -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
)
})
@@ -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<AgentStatusStoreByteBudgetRecord, number>()
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<AgentStatusStoreByteBudgetRecord, AgentStatusStoreSnapshot>
): 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
}
+13 -61
View File
@@ -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<T>(
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<T>(
return parsed
}
function parseMutationArray<T>(
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
? {
+1 -23
View File
@@ -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<keyof AgentStatusStoreDrafts, number>
): 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)
+1 -10
View File
@@ -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
@@ -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<string, number>) =>
[...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 }]) {
+1 -30
View File
@@ -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<TableName, 'parents'>
export type AgentStatusStoreIndexes = {
/** Serialized bytes of every record per table, for the snapshot budget. */
recordBytes: Record<TableName, number>
childrenByParent: Map<string, Set<string>>
factsByParent: Map<string, Set<string>>
aliasesByChild: Map<string, Set<string>>
@@ -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<TableName, number> {
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<V>(
draft: AgentStatusStoreDraftTable<V>,
index: (key: string, record: V, add: boolean) => void
@@ -191,10 +164,8 @@ function moveOrder<V>(
/** 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<TableName, number>
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))
+2 -2
View File
@@ -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
+9 -25
View File
@@ -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
+3 -20
View File
@@ -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<AgentStatusTombstoneRecord>
}
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(
@@ -48,16 +48,22 @@ function isBoundedString(value: unknown): value is string {
function copyJsonRecord(value: unknown): Record<string, unknown> | 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 {
+21 -21
View File
@@ -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()
@@ -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
@@ -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', () => {
+6 -22
View File
@@ -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))
+3 -4
View File
@@ -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',
-1
View File
@@ -35,7 +35,6 @@ export {
AGENT_TYPE_MAX_LENGTH
} from './agent-status-field-normalization'
export {
AGENT_STATUS_MAX_SUBAGENTS,
agentSubagentsEqual,
type AgentSubagentSnapshot,
type AgentSubagentState
@@ -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<string, unkno
truncated = true
continue
}
if (tasks.length >= 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,
@@ -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(
+11 -69
View File
@@ -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)
+1 -45
View File
@@ -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<string>()
const pendingRunningTasks = new Map<string, ClaudeBackgroundAgentTask>()
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. */
+1 -21
View File
@@ -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()
-4
View File
@@ -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,
@@ -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 => {