diff --git a/config/scripts/vitest-sqlite-runtime-files.mjs b/config/scripts/vitest-sqlite-runtime-files.mjs index 3d224533442..17d1522c97b 100644 --- a/config/scripts/vitest-sqlite-runtime-files.mjs +++ b/config/scripts/vitest-sqlite-runtime-files.mjs @@ -45,6 +45,7 @@ export const SQLITE_RUNTIME_INCLUDE = [ 'src/main/claude/claude-subagent-permission-request.test.ts', 'src/main/claude/claude-subagent-resume-after-restart.test.ts', 'src/main/codex/codex-provider-retry-idle-sweep.test.ts', + 'src/main/codex/codex-subagent-roster-retention.test.ts', 'src/main/codex/codex-structured-journal-exit-convergence.test.ts', 'src/main/codex/codex-structured-journal-goal-revision.test.ts', 'src/main/codex/codex-structured-journal-repeated-questions.test.ts', diff --git a/src/main/acp/acp-subagent-timeline.test.ts b/src/main/acp/acp-subagent-timeline.test.ts index 9a8a831cbe9..a3848842816 100644 --- a/src/main/acp/acp-subagent-timeline.test.ts +++ b/src/main/acp/acp-subagent-timeline.test.ts @@ -16,7 +16,7 @@ const retainedSchema = z.object({ }) ), groupOf: z.map(z.string(), z.string()), - settledIds: z.object({ size: z.number() }) + retention: z.object({ settledIdentities: z.object({ size: z.number() }) }) }) function spawnGroups( @@ -99,7 +99,7 @@ describe('ACP subagent retention', () => { expect( [...retained.groups.values()].reduce((total, group) => total + group.results.size, 0) ).toBe(31 * 64) - expect(retained.settledIds.size).toBe(32 * 64) + expect(retained.retention.settledIdentities.size).toBe(32 * 64) expect(timeline.has('live')).toBe(true) expect(timeline.has('0:0')).toBe(false) expect(timeline.has('33:0')).toBe(true) @@ -149,7 +149,7 @@ describe('ACP subagent retention', () => { const retained = retainedSchema.parse(timeline) expect(retained.groups.size).toBe(0) expect(retained.groupOf.size).toBe(0) - expect(retained.settledIds.size).toBe(0) + expect(retained.retention.settledIdentities.size).toBe(0) expect(timeline.translate([{ id: 'child-1', tokens: 9 }, { id: 'new' }], {}, 101)).toEqual([]) expect(retainedSchema.parse(timeline).groups.size).toBe(0) }) diff --git a/src/main/acp/acp-subagent-timeline.ts b/src/main/acp/acp-subagent-timeline.ts index 60eed96d175..532f782210c 100644 --- a/src/main/acp/acp-subagent-timeline.ts +++ b/src/main/acp/acp-subagent-timeline.ts @@ -1,4 +1,3 @@ -import { BoundedMap } from '../../shared/bounded-map' import { canReplaceSubagentState, isTerminalSubagentState @@ -15,6 +14,7 @@ import { import type { ProviderTimelineJoin } from '../native-chat/agent-session-timeline/provider-timeline-event' import { acpSubagentIdentity, type AcpTimelineEvent } from './acp-timeline-event' import type { AcpSubagentUpdate } from './acp-dialects/acp-dialect' +import { SubagentRosterRetention } from '../native-chat/subagent-roster-retention' /** Settled history budget; live ownership outlives it until an outcome or session close. */ const MAX_GROUPS = 32 @@ -36,16 +36,22 @@ type RosterGroup = { export class AcpSubagentTimeline { private readonly groups = new Map() private readonly groupOf = new Map() - /** An eviction index, re-derived from the changed groups' entries after each batch. */ - private readonly settledGroups = new Set() - private readonly settledIds = new BoundedMap({ - maxEntries: MAX_GROUPS * MAX_SUBAGENTS_PER_GROUP + private readonly retention = new SubagentRosterRetention(this.groups, { + maxGroups: MAX_GROUPS, + maxSettledIdentities: MAX_GROUPS * MAX_SUBAGENTS_PER_GROUP, + entries: (group) => group.entries.values(), + identities: (group) => group.entries.keys(), + onEvict: (group) => { + for (const id of group.entries.keys()) { + this.groupOf.delete(id) + } + } }) private disposed = false /** Known children include recently evicted outcomes, so they cannot become background tasks. */ has(id: string): boolean { - return this.groupOf.has(id) || this.settledIds.has(id) + return this.groupOf.has(id) || this.retention.hasSettled(id) } /** The journal owns session-end settlement; no later frame may reacquire working ownership. */ @@ -53,8 +59,7 @@ export class AcpSubagentTimeline { this.disposed = true this.groups.clear() this.groupOf.clear() - this.settledGroups.clear() - this.settledIds.clear() + this.retention.clear() } translate( @@ -93,7 +98,8 @@ export class AcpSubagentTimeline { join: groupJoin(group, join) }) } - this.trimSettledGroups(changed) + // Emit the final roster and reply before releasing their cached ownership together. + this.retention.trim(changed) return events } @@ -102,7 +108,7 @@ export class AcpSubagentTimeline { join: ProviderTimelineJoin, at: number ): RosterGroup | undefined { - if (this.settledIds.has(update.id)) { + if (this.retention.hasSettled(update.id)) { return undefined } const known = this.groups.get(this.groupOf.get(update.id) ?? '') @@ -169,31 +175,6 @@ export class AcpSubagentTimeline { return boundSubagentField(seen === 0 ? bounded : `${bounded} ${seen + 1}`, index) } - private trimSettledGroups(changed: Set): void { - for (const group of changed) { - if ([...group.entries.values()].every((entry) => isTerminalSubagentState(entry.state))) { - this.settledGroups.add(group.groupId) - } else { - this.settledGroups.delete(group.groupId) - } - } - // Emit the final roster and reply before releasing their cached ownership together. - for (const groupId of this.settledGroups) { - if (this.groups.size <= MAX_GROUPS) { - break - } - const group = this.groups.get(groupId) - if (group) { - for (const id of group.entries.keys()) { - this.groupOf.delete(id) - this.settledIds.set(id, true) - } - } - this.groups.delete(groupId) - this.settledGroups.delete(groupId) - } - } - private groupEvent(group: RosterGroup, join: ProviderTimelineJoin): AcpTimelineEvent[] { const entries = [...group.entries.values()] const body = subagentGroupJournalBody(group.groupId, entries) diff --git a/src/main/claude/claude-subagent-group-row.ts b/src/main/claude/claude-subagent-group-row.ts index 051970a5b36..be51effa2cc 100644 --- a/src/main/claude/claude-subagent-group-row.ts +++ b/src/main/claude/claude-subagent-group-row.ts @@ -93,13 +93,28 @@ export function writeClaudeSubagentGroupRow( ): void { const agents = [...group.entries.values()].map((tracked) => tracked.entry) if (agents.length === 0) { - // The row's last child turned out not to be a subagent. An empty roster is - // not a roster of nothing, so the row goes rather than reading "Ran 0". - if (group.lastSerialized !== null) { + const serialized = JSON.stringify(claudeSubagentGroupBody(group.groupId, [])) + if (serialized === group.lastSerialized) { + return + } + // Reclassification removes the row; refused tombstones remain retryable. + const admission = sink.tryAppendTombstone?.(group.identity) + if (admission && !admission.accepted) { group.lastSerialized = null + return + } + if (!admission) { sink.appendTombstone(group.identity) + } + const publication = sink.tryPublish?.() + if (publication && !publication.accepted) { + group.lastSerialized = null + return + } + if (!publication) { sink.publish() } + group.lastSerialized = serialized return } const body = claudeSubagentGroupBody(group.groupId, agents) @@ -108,7 +123,22 @@ export function writeClaudeSubagentGroupRow( // Nothing changed — a duplicate delivery must not burn a revision. return } + const options = { turnScope: group.turnScope } + const admission = sink.tryAppendItem?.(group.identity, body, options) + if (admission && !admission.accepted) { + group.lastSerialized = null + return + } + if (!admission) { + sink.appendItem(group.identity, body, options) + } + const publication = sink.tryPublish?.() + if (publication && !publication.accepted) { + group.lastSerialized = null + return + } + if (!publication) { + sink.publish() + } group.lastSerialized = serialized - sink.appendItem(group.identity, body, { turnScope: group.turnScope }) - sink.publish() } diff --git a/src/main/claude/claude-subagent-roster-groups.test.ts b/src/main/claude/claude-subagent-roster-groups.test.ts index 0a790f44111..eee98ae3eaf 100644 --- a/src/main/claude/claude-subagent-roster-groups.test.ts +++ b/src/main/claude/claude-subagent-roster-groups.test.ts @@ -7,6 +7,7 @@ import type { NativeChatSubagentEntry } from '../../shared/native-chat-types' import type { JournaledClaudeSubagentGroup } from './claude-subagent-group-row' import type { ClaudeJournaledRosterSource } from './claude-subagent-journaled-roster' import { ClaudeSubagentRosterGroups } from './claude-subagent-roster-groups' +import { inheritedClaudeSubagentGroup } from './claude-subagent-group-row' const entry = (id: string): NativeChatSubagentEntry => ({ id, @@ -51,8 +52,7 @@ describe('ClaudeSubagentRosterGroups', () => { ]) { const groups = new ClaudeSubagentRosterGroups({ journaled: journaledTwiceListed(), - currentTurnScope: () => AGENT_JOURNAL_THREAD_SCOPE, - onEvicted: () => {} + currentTurnScope: () => AGENT_JOURNAL_THREAD_SCOPE }) for (const id of order) { groups.locateOrInherit(id) @@ -65,14 +65,16 @@ describe('ClaudeSubagentRosterGroups', () => { it('keeps a child placed in a held row when another row listing it is evicted', () => { const groups = new ClaudeSubagentRosterGroups({ journaled: journaledTwiceListed(), - currentTurnScope: () => AGENT_JOURNAL_THREAD_SCOPE, - onEvicted: () => {} + currentTurnScope: () => AGENT_JOURNAL_THREAD_SCOPE }) - groups.locateOrInherit('agent-b') + const inherited = groups.locateOrInherit('agent-b') groups.locateOrInherit('agent-a') + if (inherited) { + groups.trim(inherited.group) + } // Enough of this run's own rows to push the older inherited row out. for (let turn = 0; turn < 31; turn += 1) { - groups.groupFor(`turn-${turn}-new`) + groups.trim(groups.groupFor(`turn-${turn}-new`)) } expect(groups.get('turn-a')).toBeUndefined() expect(groups.locate('agent-b')).toBeNull() @@ -83,10 +85,68 @@ describe('ClaudeSubagentRosterGroups', () => { const live = turnScope('turn-live') const groups = new ClaudeSubagentRosterGroups({ journaled: journaledTwiceListed(), - currentTurnScope: () => live, - onEvicted: () => {} + currentTurnScope: () => live }) expect(groups.locateOrInherit('agent-b')?.group.turnScope).toEqual(turnScope('turn-a')) expect(groups.groupFor('turn-live').turnScope).toEqual(live) }) }) + +it('bounds settled groups and invocation identities without evicting live groups', () => { + const groups = new ClaudeSubagentRosterGroups({ + currentTurnScope: () => AGENT_JOURNAL_THREAD_SCOPE + }) + for (let index = 0; index < 35; index++) { + const group = groups.groupFor(`live-${index}`) + const tracked = inheritedClaudeSubagentGroup( + 'seed', + { + entries: [{ ...entry(`live-child-${index}`), state: 'working' }], + turnScope: AGENT_JOURNAL_THREAD_SCOPE + }, + () => 1 + ).entries.get(`live-child-${index}`) + if (!tracked) { + throw new Error('missing fixture child') + } + group.entries.set(tracked.entry.id, tracked) + group.lastSerialized = 'written' + groups.place(tracked.entry.id, group.groupId) + groups.trim(group) + } + for (let index = 0; index < 1100; index++) { + const group = groups.groupFor(`settled-${index}`) + const id = `settled-child-${index}` + const tracked = inheritedClaudeSubagentGroup( + 'seed', + { + entries: [entry(id)], + turnScope: AGENT_JOURNAL_THREAD_SCOPE + }, + () => 1 + ).entries.get(id) + if (!tracked) { + throw new Error('missing fixture child') + } + tracked.invocationIds = new Set([`tool-${index}`]) + group.entries.set(id, tracked) + group.lastSerialized = 'written' + groups.place(id, group.groupId) + groups.trim(group) + } + expect([...groups.values()]).toHaveLength(35) + expect(groups.locate('live-child-0')?.tracked.entry.state).toBe('working') + expect(groups.locate('settled-child-1099')).toBeNull() + expect(groups.hasSettled('settled-child-1099', 'tool-1099')).toBe(true) + expect(groups.hasSettled('settled-child-0', 'tool-0')).toBe(false) + for (const group of groups.values()) { + for (const tracked of group.entries.values()) { + tracked.entry.state = 'completed' + } + groups.trim(group) + } + expect([...groups.values()]).toHaveLength(32) + groups.clear() + expect([...groups.values()]).toHaveLength(0) + expect(groups.hasSettled('settled-child-1099', 'tool-1099')).toBe(false) +}) diff --git a/src/main/claude/claude-subagent-roster-groups.ts b/src/main/claude/claude-subagent-roster-groups.ts index ea4ce76cdfb..757ceb44c61 100644 --- a/src/main/claude/claude-subagent-roster-groups.ts +++ b/src/main/claude/claude-subagent-roster-groups.ts @@ -2,6 +2,7 @@ // wrote, and the ones earlier runs of the session journaled, inherited one at a // time as this run's own events reach them. +import { SubagentRosterRetention } from '../native-chat/subagent-roster-retention' import { claudeSubagentGroupIdentity, inheritedClaudeSubagentGroup @@ -10,8 +11,7 @@ import type { AgentJournalTurnScope } from '../../shared/agent-session-journal-t import type { ClaudeJournaledRosterSource } from './claude-subagent-journaled-roster' import type { RosterGroup, TrackedEntry } from './claude-subagent-roster-state' -/** Spawn-group rows kept live per session. Bounds an event-accumulated map that - * no provider snapshot ever prunes. */ +/** Settled history target; groups with unfinished children remain owned. */ const MAX_SUBAGENT_GROUPS = 32 export type LocatedClaudeSubagent = { group: RosterGroup; tracked: TrackedEntry } @@ -22,14 +22,33 @@ export class ClaudeSubagentRosterGroups { * from an earlier turn revises that turn's row instead of the live one. */ private readonly groupIdByEntry = new Map() + private readonly retention = new SubagentRosterRetention(this.groups, { + maxGroups: MAX_SUBAGENT_GROUPS, + maxSettledIdentities: 2048, + entries: (group) => [...group.entries.values()].map((tracked) => tracked.entry), + identities: (group) => + [...group.entries] + .filter(([id]) => this.groupIdByEntry.get(id) === group.groupId) + .flatMap(([id, tracked]) => [ + claudeSubagentRetentionIdentity(id, null), + ...[...new Set([...(tracked.invocationIds ?? []), tracked.toolUseId])].map((toolUseId) => + claudeSubagentRetentionIdentity(id, toolUseId) + ) + ]), + onEvict: (group) => { + for (const id of group.entries.keys()) { + if (this.groupIdByEntry.get(id) === group.groupId) { + this.groupIdByEntry.delete(id) + } + } + } + }) + constructor( private readonly deps: { journaled?: ClaudeJournaledRosterSource /** The open turn's scope, which a group this run creates belongs to. */ currentTurnScope: () => AgentJournalTurnScope - /** Once a group leaves the map nothing can reach its children again — not - * even a session sweep — so contact is lost here. */ - onEvicted: (group: RosterGroup) => void } ) {} @@ -88,7 +107,16 @@ export class ClaudeSubagentRosterGroups { this.groupIdByEntry.delete(id) } + hasSettled(id: string, toolUseId: string | null): boolean { + return this.retention.hasSettled(claudeSubagentRetentionIdentity(id, toolUseId)) + } + + trim(group: RosterGroup, retainAccessed = false): void { + this.retention.trim([group], retainAccessed ? group.groupId : undefined) + } + clear(): void { + this.retention.clear() this.groups.clear() this.groupIdByEntry.clear() } @@ -114,21 +142,9 @@ export class ClaudeSubagentRosterGroups { private admit(group: RosterGroup): void { this.groups.set(group.groupId, group) - while (this.groups.size > MAX_SUBAGENT_GROUPS) { - const oldest = this.groups.keys().next() - if (oldest.done || oldest.value === group.groupId) { - break - } - const evicted = this.groups.get(oldest.value) - if (evicted) { - this.deps.onEvicted(evicted) - } - for (const id of evicted?.entries.keys() ?? []) { - if (this.groupIdByEntry.get(id) === oldest.value) { - this.groupIdByEntry.delete(id) - } - } - this.groups.delete(oldest.value) - } } } + +function claudeSubagentRetentionIdentity(id: string, toolUseId: string | null): string { + return JSON.stringify([id, toolUseId]) +} diff --git a/src/main/claude/claude-subagent-roster-retention.test.ts b/src/main/claude/claude-subagent-roster-retention.test.ts new file mode 100644 index 00000000000..17776d48b27 --- /dev/null +++ b/src/main/claude/claude-subagent-roster-retention.test.ts @@ -0,0 +1,196 @@ +import { describe, expect, it } from 'vitest' +import { + AGENT_JOURNAL_THREAD_SCOPE, + type AgentJournalItemBody, + type AgentJournalItemIdentity +} from '../../shared/agent-session-journal-types' +import { isSubagentGroupBlock } from '../../shared/native-chat-types' +import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' +import { ClaudeSubagentRoster } from './claude-subagent-roster' +import type { ClaudeJournaledRosterSource } from './claude-subagent-journaled-roster' + +function harness(journaled?: ClaudeJournaledRosterSource) { + const rows = new Map() + let group = 'oldest' + let writes = 0 + let refuseAppend = false + let refusePublish = false + const append = (identity: AgentJournalItemIdentity, body: AgentJournalItemBody): void => { + if (identity.provider !== 'orca') { + throw new Error('unexpected roster identity') + } + rows.set(identity.clientMessageId, body) + writes++ + } + const sink: StructuredAgentSessionEventSink = { + appendItem: append, + appendTombstone: () => {}, + publish: () => {}, + tryAppendItem: (identity, body) => { + if (refuseAppend) { + return { accepted: false, reason: 'backpressure' } + } + append(identity, body) + return { accepted: true } + }, + tryPublish: () => + refusePublish ? { accepted: false, reason: 'backpressure' } : { accepted: true } + } + const roster = new ClaudeSubagentRoster({ + sink, + journaled, + currentGroupKey: () => group, + currentTurnScope: () => AGENT_JOURNAL_THREAD_SCOPE + }) + const start = (id: string, tool = id, backgrounded = false): void => { + roster.observeSystemFrame({ + type: 'system', + subtype: 'task_started', + task_type: 'local_agent', + task_id: id, + tool_use_id: tool, + description: id, + is_backgrounded: backgrounded + }) + } + const finish = (id: string, tool = id): void => { + roster.observeSystemFrame({ + type: 'system', + subtype: 'task_notification', + task_id: id, + tool_use_id: tool, + status: 'completed' + }) + } + const agents = (key: string) => { + const body = rows.get(`claude-subagents:${key}`) + return body?.kind === 'message' ? (body.blocks.find(isSubagentGroupBlock)?.agents ?? []) : [] + } + const churn = (count: number, settled = true): void => { + for (let index = 0; index < count; index++) { + group = `later-${index}` + start(`later-child-${index}`) + if (settled) { + finish(`later-child-${index}`) + } + } + } + return { + roster, + start, + finish, + agents, + churn, + writes: () => writes, + setGroup: (key: string) => { + group = key + }, + refuse: (append: boolean, publish: boolean) => { + refuseAppend = append + refusePublish = publish + } + } +} + +describe('Claude shared settled-only roster retention', () => { + it('keeps every child of the oldest live group past 33 later spawn groups', () => { + const h = harness() + h.start('a', 'tool-a', true) + h.start('b', 'tool-b', true) + h.roster.settleTurn('oldest') + h.churn(34) + expect(h.agents('oldest').map((entry) => entry.state)).toEqual(['working', 'working']) + h.finish('a', 'tool-a') + expect(h.agents('oldest').map((entry) => entry.state)).toEqual(['completed', 'working']) + h.finish('b', 'tool-b') + expect(h.agents('oldest').map((entry) => entry.id)).toEqual(['a', 'b']) + expect(h.agents('oldest').map((entry) => entry.state)).toEqual(['completed', 'completed']) + }) + + it('reaches every still-running group in the session-end sweep', () => { + const h = harness() + h.start('a', 'tool-a', true) + h.start('b') + h.churn(34, false) + h.roster.settleSession() + expect(h.agents('oldest').map((entry) => entry.state)).toEqual(['unverifiable', 'unverifiable']) + expect(h.agents('later-33')[0].state).toBe('unverifiable') + }) + + it('ignores old announcements after settled eviction and alias-cache churn, but admits a new invocation', () => { + const h = harness() + h.start('a', 'first') + h.finish('a', 'first') + h.start('a', 'second') + h.finish('a', 'second') + h.churn(600) + const before = h.writes() + h.start('a', 'first') + h.start('a', 'second') + h.roster.observeSystemFrame({ + type: 'system', + subtype: 'task_started', + task_type: 'local_agent', + task_id: 'a' + }) + h.roster.observeChildActivity('first') + expect(h.writes()).toBe(before) + expect(h.roster.linkage.settledLinkageFor('first').linkage.agentId).toBe('a') + h.setGroup('resumed') + h.start('a', 'third') + expect(h.agents('resumed')).toEqual([expect.objectContaining({ id: 'a', state: 'working' })]) + }) + + it('keeps an inherited row reachable when child traffic precedes its resume announcement', () => { + let claimed = false + const h = harness({ + canonical: (tool) => (tool === 'original-spawn' ? 'a' : null), + groupOf: (id) => (['a', 'b'].includes(id) ? 'inherited' : null), + claimGroup: (group) => { + if (group !== 'inherited' || claimed) { + return null + } + claimed = true + return { + entries: ['a', 'b'].map((id) => ({ + id, + label: id, + state: 'completed', + startedAt: 1, + settledAt: 2 + })), + turnScope: AGENT_JOURNAL_THREAD_SCOPE + } + }, + attempt: () => 4 + }) + h.churn(32, false) + h.roster.observeChildActivity('original-spawn') + h.setGroup('resumed') + h.start('a', 'resume-tool', true) + expect(h.agents('inherited').map((entry) => entry.state)).toEqual(['working', 'completed']) + expect(h.agents('resumed')).toHaveLength(0) + expect(h.roster.linkage.settledLinkageFor('original-spawn').linkage).toMatchObject({ + agentId: 'a', + attempt: 5 + }) + }) + + for (const failure of ['append', 'publish']) { + it(`retains a terminal group whose final ${failure} was refused and retries it`, () => { + const h = harness() + h.start('a') + h.start('b') + h.refuse(failure === 'append', failure === 'publish') + h.finish('a') + h.finish('b') + h.refuse(false, false) + h.churn(34) + h.finish('b') + expect(h.agents('oldest').map((entry) => entry.state)).toEqual(['completed', 'completed']) + const before = h.writes() + h.start('a') + expect(h.writes()).toBe(before) + }) + } +}) diff --git a/src/main/claude/claude-subagent-roster.test.ts b/src/main/claude/claude-subagent-roster.test.ts index db942502ee0..1547f4bd330 100644 --- a/src/main/claude/claude-subagent-roster.test.ts +++ b/src/main/claude/claude-subagent-roster.test.ts @@ -367,13 +367,13 @@ describe('ClaudeSubagentRoster', () => { }) describe('groups that no later event can reach', () => { - it('loses contact with a group evicted past the bound', () => { + it('retains live groups past the settled-history bound', () => { const { roster, rolesIn, setGroupKey } = harness('turn-0') for (let index = 0; index < 33; index += 1) { setGroupKey(`turn-${index}`) roster.observeSystemFrame(started({ task_id: `task-${index}`, description: 'Audit' })) } - expect(rolesIn('turn-0')).toEqual([expect.objectContaining({ state: 'unverifiable' })]) + expect(rolesIn('turn-0')).toEqual([expect.objectContaining({ state: 'working' })]) expect(rolesIn('turn-32')).toEqual([expect.objectContaining({ state: 'working' })]) }) diff --git a/src/main/claude/claude-subagent-roster.ts b/src/main/claude/claude-subagent-roster.ts index baeb72e3ffb..548613136d0 100644 --- a/src/main/claude/claude-subagent-roster.ts +++ b/src/main/claude/claude-subagent-roster.ts @@ -87,8 +87,7 @@ export class ClaudeSubagentRoster { this.now = deps.now ?? (() => Date.now()) this.groups = new ClaudeSubagentRosterGroups({ journaled: deps.journaled, - currentTurnScope: deps.currentTurnScope, - onEvicted: (group) => this.sweep(group, true) + currentTurnScope: deps.currentTurnScope }) this.ids = new ClaudeSubagentIds(deps.journaled?.canonical) this.linkage = new ClaudeSubagentLinkage({ @@ -124,6 +123,12 @@ export class ClaudeSubagentRoster { if (frame.toolUseId) { this.ids.alias(frame.toolUseId, frame.taskId) } + if ( + this.groups.hasSettled(frame.taskId, frame.toolUseId) && + (frame.toolUseId !== null || !this.groups.locate(frame.taskId)) + ) { + return true + } const located = this.groups.locateOrInherit(frame.taskId) ?? (frame.toolUseId ? this.adopt(frame.toolUseId, frame.taskId) : null) @@ -158,10 +163,20 @@ export class ClaudeSubagentRoster { */ observeChildActivity(parentToolUseId: string): void { const canonical = this.ids.canonical(parentToolUseId) - if (this.ids.isExcluded(parentToolUseId, canonical)) { + if ( + this.ids.isExcluded(parentToolUseId, canonical) || + this.groups.hasSettled(canonical, parentToolUseId) || + (!this.groups.locate(canonical) && this.groups.hasSettled(canonical, null)) + ) { return } - if (this.groups.locateOrInherit(canonical)) { + if (this.groups.locate(canonical)) { + return + } + const inherited = this.groups.locateOrInherit(canonical) + if (inherited) { + // Child traffic can precede the announcement that reopens an inherited row. + this.groups.trim(inherited.group, true) return } if (this.announcesTasks) { @@ -242,7 +257,6 @@ export class ClaudeSubagentRoster { if (!group) { return } - let changed = false for (const [id, tracked] of group.entries) { if (isTerminalSubagentState(tracked.entry.state)) { continue @@ -254,11 +268,13 @@ export class ClaudeSubagentRoster { ...tracked, entry: { ...tracked.entry, state: 'unverifiable', settledAt: this.now() } }) - changed = true - } - if (changed) { - writeClaudeSubagentGroupRow(this.deps.sink, group) } + this.write(group) + } + + private write(group: RosterGroup): void { + writeClaudeSubagentGroupRow(this.deps.sink, group) + this.groups.trim(group) } private create( @@ -270,6 +286,7 @@ export class ClaudeSubagentRoster { ): void { const group = this.groups.groupFor(this.deps.currentGroupKey() ?? OUTSIDE_TURN) if (group.admittedEntries >= MAX_SUBAGENTS_PER_GROUP) { + this.write(group) return } group.admittedEntries += 1 @@ -291,7 +308,7 @@ export class ClaudeSubagentRoster { } }) this.groups.place(id, group.groupId) - writeClaudeSubagentGroupRow(this.deps.sink, group) + this.write(group) } private revise( @@ -332,7 +349,7 @@ export class ClaudeSubagentRoster { } } group.entries.set(id, next) - writeClaudeSubagentGroupRow(this.deps.sink, group) + this.write(group) } /** Re-key a provisional entry from its tool id onto the canonical task id the @@ -362,6 +379,6 @@ export class ClaudeSubagentRoster { } located.group.entries.delete(id) this.groups.forget(id) - writeClaudeSubagentGroupRow(this.deps.sink, located.group) + this.write(located.group) } } diff --git a/src/main/codex/codex-structured-journal-limits.ts b/src/main/codex/codex-structured-journal-limits.ts index d4c9bcfa0de..82ad7a38235 100644 --- a/src/main/codex/codex-structured-journal-limits.ts +++ b/src/main/codex/codex-structured-journal-limits.ts @@ -11,9 +11,9 @@ export const MAX_CODEX_PENDING_PROMPTS = 128 export const MAX_CODEX_IDENTITY_ENTRIES = 512 export const MAX_CODEX_DETAIL_ENTRIES = 512 export const MAX_CODEX_DETAIL_BYTES = 64 * 1024 -/** Spawn-group rows kept live per session, and children per row. Both bound an - * event-accumulated map that no provider snapshot ever prunes. */ +/** Settled spawn-group history budget; live groups remain until their final write succeeds. */ export const MAX_CODEX_SUBAGENT_GROUPS = 32 +/** Children admitted per spawn-group row. */ export const MAX_CODEX_SUBAGENTS_PER_GROUP = 64 /** Threads whose latest token total is retained. Usage frames arrive for * threads that are not yet (or never become) roster children. */ diff --git a/src/main/codex/codex-subagent-roster-retention.test.ts b/src/main/codex/codex-subagent-roster-retention.test.ts new file mode 100644 index 00000000000..92de0fc2aea --- /dev/null +++ b/src/main/codex/codex-subagent-roster-retention.test.ts @@ -0,0 +1,322 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, describe, expect, it } from 'vitest' +import { codexProviderHandle } from '../../shared/agent-session-provider-handle-encoding' +import { isSubagentGroupBlock, type NativeChatSubagentState } from '../../shared/native-chat-types' +import { createTrackedJournalOpener } from '../native-chat/agent-session-journal/journal-host-database-test-support' +import { + createDeferredStructuredAgentSessionEventSink, + type StructuredAgentSessionEventSink +} from '../native-chat/agent-session-wire/structured-agent-session-event-sink' +import { testEventSinkLogging } from '../native-chat/agent-session-wire/structured-agent-session-logger-test-support' +import { CodexSubagentRoster } from './codex-subagent-roster' +import { MAX_CODEX_SUBAGENT_GROUPS } from './codex-structured-journal-limits' + +const THREAD = 'parent' +const ORIGINAL = 'original' +const journals = createTrackedJournalOpener() +const cleanups: (() => Promise)[] = [] +afterEach(async () => { + for (const cleanup of cleanups.splice(0)) { + await cleanup() + } + await journals.closeAll() +}) + +async function session(refuse?: 'append' | 'publish') { + const root = await mkdtemp(join(tmpdir(), 'orca-codex-roster-retention-')) + const journal = await journals.open({ + identity: { + sessionId: 'session-codex-roster-retention', + workspaceId: 'workspace-1', + hostId: 'local', + agent: 'codex', + providerHandle: codexProviderHandle(THREAD) + }, + stateDirectory: root, + now: () => 1_000 + }) + const deferred = createDeferredStructuredAgentSessionEventSink(testEventSinkLogging()) + deferred.bind({ journal, fence: 1, publish: () => {} }) + const append = deferred.sink.tryAppendItem + const publish = deferred.sink.tryPublish + if (!append || !publish) { + throw new Error('the deferred sink must expose append and publish admission') + } + let refusing = false + let writingOriginal = false + const refusal = { accepted: false, reason: 'backpressure' } as const + const sink: StructuredAgentSessionEventSink = { + ...deferred.sink, + tryAppendItem: (identity, body, options) => { + writingOriginal = + identity.provider === 'orca' && + identity.clientMessageId === `codex-subagents:${THREAD}:${ORIGINAL}` + return refusing && writingOriginal && refuse === 'append' + ? refusal + : append(identity, body, options) + }, + tryPublish: () => (refusing && writingOriginal && refuse === 'publish' ? refusal : publish()) + } + let parentTurn = ORIGINAL + let clock = 1_000 + const roster = new CodexSubagentRoster({ + sink, + primaryThreadId: () => THREAD, + activeTurn: () => parentTurn, + turnScopeFor: (_thread, turnId) => ({ kind: 'turn', turnItemId: turnId ?? 'outside-turn' }), + now: () => ++clock + }) + cleanups.push(async () => { + roster.dispose() + deferred.close() + await journal.close() + await rm(root, { recursive: true, force: true }) + }) + const turn = (id: string, state: NativeChatSubagentState, turnId = `execution:${id}`) => + roster.handleTurn({ threadId: id, turnId, state }) + const announce = (id: string, kind = 'started') => + roster.handleItem({ + threadId: THREAD, + turnId: parentTurn, + item: { + type: 'subAgentActivity', + id: `activity:${id}:${kind}`, + agentThreadId: id, + agentPath: '/root/task', + kind + } + }) + function spawn(id: string, parent = ORIGINAL) { + parentTurn = parent + expect(turn(id, 'working')).toEqual({ accepted: true }) + expect(announce(id)).toEqual({ accepted: true }) + } + async function rows() { + await deferred.drained() + return journal + .snapshot() + .items.flatMap((row) => + row.body.kind === 'message' + ? row.body.blocks.filter(isSubagentGroupBlock).map((group) => ({ row, group })) + : [] + ) + } + return { + roster, + turn, + announce, + spawn, + rows, + original: async () => + (await rows()).find(({ group }) => group.groupId === `${THREAD}:${ORIGINAL}`), + parent: (turnId: string) => { + parentTurn = turnId + }, + refuse: (value: boolean) => { + refusing = value + } + } +} + +type Session = Awaited> +async function laterGroups(s: Session, settled: boolean, count = MAX_CODEX_SUBAGENT_GROUPS + 2) { + for (let index = 0; index < count; index++) { + const id = `later-child-${index}` + s.spawn(id, `later-parent-${index}`) + if (settled) { + expect(s.turn(id, 'completed')).toEqual({ accepted: true }) + } + if (index % 32 === 31) { + await s.rows() + } + } + await s.rows() +} + +function states(s: Awaited>) { + return s?.group.agents.map(({ id, state }) => ({ id, state })) +} + +describe('Codex roster retention through the real sink and journal', () => { + it.each([ + [false, 'completed'], + [true, 'completed'], + [false, 'sweep'], + [true, 'sweep'] + ] as const)( + 'preserves all oldest live children when later groups settled=%s and final event=%s', + async (settled, final) => { + const s = await session() + for (const id of ['old-a', 'old-b', 'old-c']) { + s.spawn(id) + } + const before = await s.original() + expect(before?.group.agents).toHaveLength(3) + await laterGroups(s, settled) + s.parent('current') + s.announce('old-a', 'interacted') + expect(states(await s.original())).toEqual( + ['old-a', 'old-b', 'old-c'].map((id) => ({ id, state: 'working' })) + ) + if (final === 'sweep') { + expect(s.roster.settleSession()).toEqual({ accepted: true }) + } else { + for (const id of ['old-a', 'old-b', 'old-c']) { + s.turn(id, 'completed') + } + } + const after = await s.original() + expect(after?.row.itemId).toBe(before?.row.itemId) + expect(after?.row.turnScope).toEqual(before?.row.turnScope) + expect(states(after)).toEqual( + ['old-a', 'old-b', 'old-c'].map((id) => ({ + id, + state: final === 'sweep' ? 'unverifiable' : 'completed' + })) + ) + expect(after?.group.agents.every((entry) => entry.settledAt !== undefined)).toBe(true) + expect( + (await s.rows()).flatMap(({ group }) => + group.agents.filter((entry) => entry.id === 'old-a') + ) + ).toHaveLength(1) + } + ) + + it('ignores late reports for evicted settled executions even after the execution cache has pruned them', async () => { + const s = await session() + s.spawn('old-a') + s.spawn('old-b') + s.turn('old-a', 'stopped') + s.turn('old-b', 'completed') + const before = await s.original() + await laterGroups(s, true, 270) + expect(s.roster.executions.find('old-a')).toBeUndefined() + expect(s.roster.retentionSizes().groups).toBe(MAX_CODEX_SUBAGENT_GROUPS) + s.parent('current') + s.turn('old-a', 'working') + s.announce('old-a') + s.announce('old-a', 'interacted') + s.turn('old-a', 'completed') + s.announce('old-a', 'completed') + s.roster.handleTokenUsage({ threadId: 'old-a', tokenUsage: { total: { totalTokens: 123 } } }) + expect(await s.original()).toEqual(before) + expect(s.roster.executions.find('old-a')?.execution).toBeNull() + expect( + (await s.rows()).flatMap(({ group }) => group.agents.filter((entry) => entry.id === 'old-a')) + ).toHaveLength(1) + }) + + it.each([ + ['before', false], + ['after', false], + ['before', true], + ['after', true] + ] as const)( + 'admits a genuine new child turn when the announcement is %s its start and execution cache pruned=%s', + async (order, pruned) => { + const s = await session() + s.spawn('old-a') + s.turn('old-a', 'completed') + const before = await s.original() + await laterGroups(s, true, pruned ? 270 : MAX_CODEX_SUBAGENT_GROUPS + 2) + s.parent('follow-up') + if (order === 'before') { + s.announce('old-a', 'interacted') + } + s.turn('old-a', 'working', 'new-execution') + if (order === 'after') { + s.announce('old-a', 'interacted') + } + const followup = () => + s.rows().then((rows) => rows.find(({ group }) => group.groupId === `${THREAD}:follow-up`)) + expect(states(await followup())).toEqual([{ id: 'old-a', state: 'working' }]) + s.turn('old-a', 'working') + s.turn('old-a', 'completed') + expect(states(await followup())).toEqual([{ id: 'old-a', state: 'working' }]) + s.turn('old-a', 'completed', 'new-execution') + expect(states(await followup())).toEqual([{ id: 'old-a', state: 'completed' }]) + expect(await s.original()).toEqual(before) + } + ) + + it('suppresses both settled executions after same-group reuse, eviction and execution-cache churn', async () => { + const s = await session() + s.spawn('old-a') + s.spawn('old-b') + s.turn('old-a', 'completed') + s.announce('old-a', 'interacted') + s.turn('old-a', 'working', 'second-execution') + s.turn('old-a', 'completed', 'second-execution') + s.turn('old-b', 'completed') + const original = await s.original() + await laterGroups(s, true, 270) + expect(s.roster.executions.find('old-a')).toBeUndefined() + expect(s.roster.retentionSizes().settledIdentities).toBeLessThan(2048) + const before = await s.rows() + s.parent('current') + for (const turnId of ['execution:old-a', 'second-execution']) { + s.turn('old-a', 'working', turnId) + s.announce('old-a') + s.turn('old-a', 'completed', turnId) + s.announce('old-a', 'completed') + } + expect(await s.rows()).toEqual(before) + s.announce('old-a', 'interacted') + s.turn('old-a', 'working', 'third-execution') + const current = (await s.rows()).find(({ group }) => group.groupId === `${THREAD}:current`) + expect(states(current)).toEqual([{ id: 'old-a', state: 'working' }]) + s.turn('old-a', 'completed', 'third-execution') + expect(await s.original()).toEqual(original) + }) + + it.each(['append', 'publish'] as const)( + 'retains final retry ownership after %s refuses settlement under group pressure', + async (refuse) => { + const s = await session(refuse) + s.spawn('old-a') + s.spawn('old-b') + const before = await s.original() + await laterGroups(s, false) + s.turn('old-b', 'completed') + s.refuse(true) + expect(s.turn('old-a', 'completed')).toEqual({ accepted: false, reason: 'backpressure' }) + s.spawn('more-live', 'current') + expect(s.roster.retentionSizes().groups).toBeGreaterThan(MAX_CODEX_SUBAGENT_GROUPS) + s.refuse(false) + expect(s.roster.settleSession()).toEqual({ accepted: true }) + const after = await s.original() + expect(after?.row.itemId).toBe(before?.row.itemId) + expect(states(after)).toEqual([ + { id: 'old-a', state: 'completed' }, + { id: 'old-b', state: 'completed' } + ]) + expect(s.roster.retentionSizes().groups).toBe(MAX_CODEX_SUBAGENT_GROUPS) + } + ) + + it.each(['append', 'publish'] as const)( + 'retries a refused final sweep %s after more live groups arrive', + async (refuse) => { + const s = await session(refuse) + s.spawn('old-a') + s.spawn('old-b') + const before = await s.original() + await laterGroups(s, false) + s.refuse(true) + expect(s.roster.settleSession()).toEqual({ accepted: false, reason: 'backpressure' }) + s.spawn('more-live', 'current') + s.refuse(false) + expect(s.roster.settleSession()).toEqual({ accepted: true }) + const after = await s.original() + expect(after?.row.itemId).toBe(before?.row.itemId) + expect(states(after)).toEqual([ + { id: 'old-a', state: 'unverifiable' }, + { id: 'old-b', state: 'unverifiable' } + ]) + expect(s.roster.retentionSizes().groups).toBe(MAX_CODEX_SUBAGENT_GROUPS) + } + ) +}) diff --git a/src/main/codex/codex-subagent-roster-state.ts b/src/main/codex/codex-subagent-roster-state.ts index 02c79411b79..ccbef83ade6 100644 --- a/src/main/codex/codex-subagent-roster-state.ts +++ b/src/main/codex/codex-subagent-roster-state.ts @@ -5,6 +5,12 @@ import type { AgentJournalTurnScope } from '../../shared/agent-session-journal-types' import type { NativeChatSubagentEntry } from '../../shared/native-chat-types' +import { isTerminalSubagentState } from '../../shared/native-chat-subagent-summary' +import { SubagentRosterRetention } from '../native-chat/subagent-roster-retention' +import { + MAX_CODEX_SUBAGENT_GROUPS, + MAX_CODEX_SUBAGENTS_PER_GROUP +} from './codex-structured-journal-limits' /** The turn a group belongs to when Codex reports activity outside any turn. * Mirrors the generic-frame bucket name so the two read alike in the journal. */ @@ -22,6 +28,41 @@ export type RosterGroup = { lastSerialized: string | null } +/** A reused child thread starts a new execution only when its turn id changes. */ +export function codexSubagentExecutionIdentity(threadId: string, turnId: string): string { + return JSON.stringify([threadId, turnId]) +} + +export function retainSupersededCodexExecution( + group: RosterGroup, + threadId: string, + nextTurnId: string | null, + retention: SubagentRosterRetention +): void { + const previousTurnId = group.executionTurns.get(threadId) + const previous = group.entries.get(threadId) + if ( + typeof previousTurnId === 'string' && + previousTurnId !== nextTurnId && + previous && + isTerminalSubagentState(previous.state) + ) { + retention.rememberSettled(codexSubagentExecutionIdentity(threadId, previousTurnId)) + } +} + +export function createCodexSubagentRosterRetention(groups: Map) { + return new SubagentRosterRetention(groups, { + maxGroups: MAX_CODEX_SUBAGENT_GROUPS, + maxSettledIdentities: MAX_CODEX_SUBAGENT_GROUPS * MAX_CODEX_SUBAGENTS_PER_GROUP, + entries: (group) => group.entries.values(), + identities: (group) => + [...group.executionTurns].flatMap(([threadId, turnId]) => + turnId === null ? [] : [codexSubagentExecutionIdentity(threadId, turnId)] + ) + }) +} + /** Group identity: the parent turn that spawned the children. `agentPath` is a * tree rooted at the parent thread, so every child of one turn shares a row * no matter which thread's stream carried its activity item. */ diff --git a/src/main/codex/codex-subagent-roster.test.ts b/src/main/codex/codex-subagent-roster.test.ts index 3f411fff1a3..ac56c042ba2 100644 --- a/src/main/codex/codex-subagent-roster.test.ts +++ b/src/main/codex/codex-subagent-roster.test.ts @@ -569,10 +569,7 @@ describe('CodexSubagentRoster', () => { expect(appended).toHaveLength(atCap) }) - // The eviction is the KNOWN LIMITATION the module documents: `groups` is never - // seeded from the journal, so the evicted group's next child rebuilds its - // durable row from that one child. Pinned so the boundary cannot move silently. - it('caps live spawn groups, and an evicted group rebuilds its row from one child', () => { + it('retains live spawn groups beyond the settled history budget', () => { const { roster, appended, agents } = createHarness() for (let index = 0; index <= MAX_CODEX_SUBAGENT_GROUPS; index++) { deliver( @@ -581,10 +578,7 @@ describe('CodexSubagentRoster', () => { `turn-${index}` ) } - const evicted = codexSubagentGroupIdentity(codexSubagentGroupId(THREAD, 'turn-0')) - const rowsFor = (identity: AgentJournalItemIdentity): Appended[] => - appended.filter((entry) => JSON.stringify(entry.identity) === JSON.stringify(identity)) - expect(rowsFor(evicted)).toHaveLength(1) + const original = codexSubagentGroupIdentity(codexSubagentGroupId(THREAD, 'turn-0')) deliver( roster, @@ -592,8 +586,9 @@ describe('CodexSubagentRoster', () => { 'turn-0' ) - expect(latestIdentity(appended)).toEqual(evicted) - expect(agents().map((agent) => agent.id)).toEqual(['child-late']) + expect(latestIdentity(appended)).toEqual(original) + expect(agents().map((agent) => agent.id)).toEqual(['child-0', 'child-late']) + expect(roster.retentionSizes().groups).toBe(MAX_CODEX_SUBAGENT_GROUPS + 1) }) it('keeps a token count a later thread-map eviction would otherwise retract', () => { diff --git a/src/main/codex/codex-subagent-roster.ts b/src/main/codex/codex-subagent-roster.ts index 43d31756954..0b55d231fbe 100644 --- a/src/main/codex/codex-subagent-roster.ts +++ b/src/main/codex/codex-subagent-roster.ts @@ -5,15 +5,11 @@ // child turn events supply execution state. // // KNOWN LIMITATION: `groups` is process-local and is never seeded from the -// journal, while the row's identity is keyed on the group id alone. So once a -// group leaves the map its row stays, and the next activity item rebuilds that -// row from one child — rewriting N down to one. Two ways in: eviction past -// MAX_CODEX_SUBAGENT_GROUPS, which drops the oldest-inserted group in-process -// even while it is live, and skips the sweep so its children never latch -// `unverifiable`; and a restart on `threadId:outside-turn`, the one group id -// that outlives the process — `thread/resume` is verified to return the same -// thread, and a real turn id is assumed freshly minted per turn. Seeding from -// the journal is the fix. +// journal, while the row's identity is keyed on the group id alone. After a +// restart loses membership for `threadId:outside-turn`, the next activity +// rebuilds that row from one child — rewriting N down to one. This group id +// outlives the process: `thread/resume` returns the same thread, while real turn +// ids are assumed freshly minted per turn. Seeding from the journal is the fix. import type { AgentJournalTurnScope } from '../../shared/agent-session-journal-types' import { isTerminalSubagentState } from '../../shared/native-chat-subagent-summary' @@ -43,13 +39,13 @@ export { codexSubagentGroupId, codexSubagentGroupIdentity } from './codex-subage import { codexSubagentGroupId, codexSubagentGroupIdentity, + codexSubagentExecutionIdentity as executionIdentity, + createCodexSubagentRosterRetention, + retainSupersededCodexExecution, type RosterGroup } from './codex-subagent-roster-state' import type { CodexThreadItem } from './codex-structured-item-translation' -import { - MAX_CODEX_SUBAGENT_GROUPS, - MAX_CODEX_SUBAGENTS_PER_GROUP -} from './codex-structured-journal-limits' +import { MAX_CODEX_SUBAGENTS_PER_GROUP } from './codex-structured-journal-limits' import { CodexThreadTokenTotals } from './codex-thread-token-totals' const ADMITTED: StructuredAgentSessionSinkAdmission = { accepted: true } @@ -66,6 +62,7 @@ export type CodexSubagentRosterDeps = { export class CodexSubagentRoster { private readonly groups = new Map() + private readonly retention = createCodexSubagentRosterRetention(this.groups) /** Every thread's total, members or not; children are selected at write time. */ private readonly tokensByThread = new CodexThreadTokenTotals() private readonly now: () => number @@ -115,7 +112,10 @@ export class CodexSubagentRoster { // Only a spawn names the spawner: other announcements ride whichever agent acted. announcement.spawned ? input.threadId : undefined ) - if (!child?.execution) { + if ( + !child?.execution || + this.retention.hasSettled(executionIdentity(child.agentThreadId, child.execution.turnId)) + ) { return ADMITTED } const group = @@ -150,7 +150,10 @@ export class CodexSubagentRoster { turnId: string state: NativeChatSubagentState }): StructuredAgentSessionSinkAdmission { - if (input.threadId === this.deps.primaryThreadId()) { + if ( + input.threadId === this.deps.primaryThreadId() || + this.retention.hasSettled(executionIdentity(input.threadId, input.turnId)) + ) { return ADMITTED } const observed = this.executions.observeTurn(input.threadId, input.turnId, input.state) @@ -162,7 +165,10 @@ export class CodexSubagentRoster { child: Readonly, execution: CodexChildExecution ): StructuredAgentSessionSinkAdmission { - if (!child.registered) { + if ( + !child.registered || + this.retention.hasSettled(executionIdentity(child.agentThreadId, execution.turnId)) + ) { return ADMITTED } if (execution.state === 'working') { @@ -227,13 +233,15 @@ export class CodexSubagentRoster { dispose(): void { this.unfollow() this.groups.clear() + this.retention.clear() this.tokensByThread.clear() } - private sweep(group: RosterGroup | undefined): StructuredAgentSessionSinkAdmission { - if (!group) { - return ADMITTED - } + retentionSizes(): { groups: number; settledIdentities: number } { + return this.retention.sizes() + } + + private sweep(group: RosterGroup): StructuredAgentSessionSinkAdmission { let changed = false for (const [id, entry] of group.entries) { if (isTerminalSubagentState(entry.state)) { @@ -268,13 +276,6 @@ export class CodexSubagentRoster { lastSerialized: null } this.groups.set(groupId, group) - while (this.groups.size > MAX_CODEX_SUBAGENT_GROUPS) { - const oldest = this.groups.keys().next().value - if (typeof oldest !== 'string' || oldest === groupId) { - break - } - this.groups.delete(oldest) - } return group } @@ -307,6 +308,7 @@ export class CodexSubagentRoster { return } const now = this.now() + retainSupersededCodexExecution(group, child.agentThreadId, turnId, this.retention) group.executionTurns.set(child.agentThreadId, turnId) group.entries.set(child.agentThreadId, { id: child.agentThreadId, @@ -345,6 +347,7 @@ export class CodexSubagentRoster { : (this.deps.sink.appendItem(group.identity, body, options), ADMITTED) if (!admission.accepted) { group.lastSerialized = null + this.retention.trim([group]) return admission } const published = this.deps.sink.tryPublish @@ -357,6 +360,7 @@ export class CodexSubagentRoster { // roster stays queued but never reaches the renderer. group.lastSerialized = null } + this.retention.trim([group]) return published } } diff --git a/src/main/native-chat/subagent-roster-retention.test.ts b/src/main/native-chat/subagent-roster-retention.test.ts new file mode 100644 index 00000000000..5eaacca3568 --- /dev/null +++ b/src/main/native-chat/subagent-roster-retention.test.ts @@ -0,0 +1,150 @@ +import { describe, expect, it } from 'vitest' +import type { + NativeChatSubagentEntry, + NativeChatSubagentState +} from '../../shared/native-chat-types' +import { SubagentRosterRetention } from './subagent-roster-retention' + +type Group = { + groupId: string + entries: Map + lastSerialized: string | null +} + +function group(groupId: string, state: NativeChatSubagentState, children = 64): Group { + return { + groupId, + lastSerialized: 'admitted', + entries: new Map( + Array.from({ length: children }, (_, index) => { + const id = `${groupId}:${index}` + return [id, { id, label: id, state }] + }) + ) + } +} + +describe('host subagent roster settled retention', () => { + it('shares one bound between superseded executions and evicted group identities', () => { + const groups = new Map() + const retention = new SubagentRosterRetention(groups, { + maxGroups: 1, + maxSettledIdentities: 2048, + entries: (group) => group.entries.values(), + identities: (roster) => roster.entries.keys() + }) + const live = group('live', 'working', 1) + groups.set(live.groupId, live) + for (let index = 0; index < 2100; index++) { + retention.rememberSettled(`superseded-${index}`) + } + const settled = group('settled', 'completed', 1) + groups.set(settled.groupId, settled) + retention.trim([settled]) + expect(retention.sizes()).toEqual({ groups: 1, settledIdentities: 2048 }) + expect(retention.hasSettled('superseded-52')).toBe(false) + expect(retention.hasSettled('superseded-2099')).toBe(true) + expect(retention.hasSettled('settled:0')).toBe(true) + retention.clear() + expect(retention.sizes().settledIdentities).toBe(0) + }) + + it('bounds settled groups and recent identities while retaining every live group', () => { + const groups = new Map() + const retention = new SubagentRosterRetention(groups, { + maxGroups: 32, + maxSettledIdentities: 32 * 64, + entries: (group) => group.entries.values(), + identities: (roster) => roster.entries.keys() + }) + for (let index = 0; index < 35; index++) { + const live = group(`live-${index}`, 'working') + groups.set(live.groupId, live) + retention.trim([live]) + } + for (let index = 0; index < 70; index++) { + const settled = group(`settled-${index}`, 'completed') + groups.set(settled.groupId, settled) + retention.trim([settled]) + } + expect(retention.sizes()).toEqual({ groups: 35, settledIdentities: 32 * 64 }) + expect(retention.hasSettled('settled-0:0')).toBe(false) + expect(retention.hasSettled('settled-69:0')).toBe(true) + expect([...groups.keys()].every((id) => id.startsWith('live-'))).toBe(true) + for (let index = 0; index < 70; index++) { + const settled = group(`more-settled-${index}`, 'completed') + const live = groups.get(`live-${index}`) + if (live) { + for (const entry of live.entries.values()) { + entry.state = 'completed' + } + retention.trim([live]) + } + groups.set(settled.groupId, settled) + retention.trim([settled]) + } + expect(retention.sizes()).toEqual({ groups: 32, settledIdentities: 32 * 64 }) + }) + + it('re-derives settled eligibility when a retained child runs again or its write is refused', () => { + const groups = new Map() + const retention = new SubagentRosterRetention(groups, { + maxGroups: 1, + maxSettledIdentities: 64, + entries: (group) => group.entries.values(), + identities: (roster) => roster.entries.keys() + }) + const old = group('old', 'completed', 1) + groups.set(old.groupId, old) + retention.trim([old]) + old.entries.set('old:0', { id: 'old:0', label: 'old', state: 'working' }) + retention.trim([old]) + const newGroup = group('new', 'working', 1) + groups.set(newGroup.groupId, newGroup) + retention.trim([newGroup]) + expect(groups.has('old')).toBe(true) + old.entries.set('old:0', { id: 'old:0', label: 'old', state: 'completed' }) + old.lastSerialized = null + retention.trim([old]) + expect(groups.has('old')).toBe(true) + old.lastSerialized = 'admitted' + retention.trim([old]) + expect(groups.has('old')).toBe(false) + expect(retention.hasSettled('old:0')).toBe(true) + groups.clear() + retention.clear() + expect(retention.sizes()).toEqual({ groups: 0, settledIdentities: 0 }) + }) +}) + +it('keeps only the accessed inherited row during a trim and re-derives it on the next event', () => { + const groups = new Map() + const retention = new SubagentRosterRetention(groups, { + maxGroups: 32, + maxSettledIdentities: 2048, + entries: (roster) => roster.entries.values(), + identities: (roster) => roster.entries.keys() + }) + for (let index = 0; index < 32; index++) { + const live = group(`live-${index}`, 'working', 1) + groups.set(live.groupId, live) + retention.trim([live]) + } + for (let index = 0; index < 70; index++) { + const inherited = group(`inherited-${index}`, 'completed', 1) + groups.set(inherited.groupId, inherited) + retention.trim([inherited], inherited.groupId) + expect(groups.size).toBe(33) + expect(groups.has(inherited.groupId)).toBe(true) + } + retention.trim([]) + expect(groups.size).toBe(32) + expect(retention.hasSettled('inherited-69:0')).toBe(true) + const resumed = group('resumed', 'completed', 1) + groups.set(resumed.groupId, resumed) + retention.trim([resumed], resumed.groupId) + resumed.entries.set('resumed:0', { id: 'resumed:0', label: 'resumed', state: 'working' }) + retention.trim([resumed]) + expect(groups.has(resumed.groupId)).toBe(true) + expect(groups.size).toBe(33) +}) diff --git a/src/main/native-chat/subagent-roster-retention.ts b/src/main/native-chat/subagent-roster-retention.ts new file mode 100644 index 00000000000..6d4f5d865f5 --- /dev/null +++ b/src/main/native-chat/subagent-roster-retention.ts @@ -0,0 +1,75 @@ +import { BoundedMap } from '../../shared/bounded-map' +import { isTerminalSubagentState } from '../../shared/native-chat-subagent-summary' +import type { NativeChatSubagentEntry } from '../../shared/native-chat-types' + +type RetainedRosterGroup = { + groupId: string + lastSerialized: string | null +} + +/** Bounds settled history; live children and refused final writes keep their ownership. */ +export class SubagentRosterRetention { + private readonly settledGroups = new Set() + private readonly settledIdentities: BoundedMap + + constructor( + private readonly groups: Map, + private readonly options: { + maxGroups: number + maxSettledIdentities: number + entries: (group: Group) => Iterable + identities: (group: Group) => Iterable + onEvict?: (group: Group) => void + } + ) { + this.settledIdentities = new BoundedMap({ maxEntries: options.maxSettledIdentities }) + } + + hasSettled(identity: string): boolean { + return this.settledIdentities.has(identity) + } + + /** Superseded executions share the evicted history's single bounded budget. */ + rememberSettled(identity: string): void { + this.settledIdentities.set(identity, true) + } + + trim(changed: Iterable, retainedGroupId?: string): void { + for (const group of changed) { + if ( + group.lastSerialized !== null && + [...this.options.entries(group)].every((entry) => isTerminalSubagentState(entry.state)) + ) { + this.settledGroups.add(group.groupId) + } else { + this.settledGroups.delete(group.groupId) + } + } + for (const groupId of this.settledGroups) { + if (this.groups.size <= this.options.maxGroups) { + break + } + if (groupId === retainedGroupId) { + continue + } + const group = this.groups.get(groupId) + if (group) { + for (const identity of this.options.identities(group)) { + this.rememberSettled(identity) + } + this.options.onEvict?.(group) + } + this.groups.delete(groupId) + this.settledGroups.delete(groupId) + } + } + + sizes(): { groups: number; settledIdentities: number } { + return { groups: this.groups.size, settledIdentities: this.settledIdentities.size } + } + + clear(): void { + this.settledGroups.clear() + this.settledIdentities.clear() + } +}