mirror of
https://github.com/stablyai/orca.git
synced 2026-10-30 16:47:34 +00:00
fix(native-chat): preserve running subagent rows in long chats (#26798)
* fix(codex): keep running subagent groups in long chats * fix(codex): remember superseded settled child turns * fix(native-chat): retain running Claude subagent groups
This commit is contained in:
@@ -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',
|
||||
|
||||
@@ -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)
|
||||
})
|
||||
|
||||
@@ -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<string, RosterGroup>()
|
||||
private readonly groupOf = new Map<string, string>()
|
||||
/** An eviction index, re-derived from the changed groups' entries after each batch. */
|
||||
private readonly settledGroups = new Set<string>()
|
||||
private readonly settledIds = new BoundedMap<string, true>({
|
||||
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<RosterGroup>): 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)
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
})
|
||||
|
||||
@@ -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<string, string>()
|
||||
|
||||
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])
|
||||
}
|
||||
|
||||
@@ -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<string, AgentJournalItemBody>()
|
||||
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)
|
||||
})
|
||||
}
|
||||
})
|
||||
@@ -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' })])
|
||||
})
|
||||
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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. */
|
||||
|
||||
@@ -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<void>)[] = []
|
||||
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<ReturnType<typeof session>>
|
||||
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<ReturnType<Session['original']>>) {
|
||||
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)
|
||||
}
|
||||
)
|
||||
})
|
||||
@@ -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<RosterGroup>
|
||||
): 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<string, RosterGroup>) {
|
||||
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. */
|
||||
|
||||
@@ -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', () => {
|
||||
|
||||
@@ -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<string, RosterGroup>()
|
||||
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<CodexExecutionChild>,
|
||||
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
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<string, NativeChatSubagentEntry>
|
||||
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<string, Group>()
|
||||
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<string, Group>()
|
||||
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<string, Group>()
|
||||
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<string, Group>()
|
||||
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)
|
||||
})
|
||||
@@ -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<Group extends RetainedRosterGroup> {
|
||||
private readonly settledGroups = new Set<string>()
|
||||
private readonly settledIdentities: BoundedMap<string, true>
|
||||
|
||||
constructor(
|
||||
private readonly groups: Map<string, Group>,
|
||||
private readonly options: {
|
||||
maxGroups: number
|
||||
maxSettledIdentities: number
|
||||
entries: (group: Group) => Iterable<NativeChatSubagentEntry>
|
||||
identities: (group: Group) => Iterable<string>
|
||||
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<Group>, 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()
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user