diff --git a/src/main/codex/codex-persistent-command-retention.test.ts b/src/main/codex/codex-persistent-command-retention.test.ts index 949e3156ee2d..498d5b186ef4 100644 --- a/src/main/codex/codex-persistent-command-retention.test.ts +++ b/src/main/codex/codex-persistent-command-retention.test.ts @@ -45,6 +45,7 @@ function fixture(maxMetadataBytes?: number) { { sink, maxMetadataBytes, + linkageFor: () => ({}), schedule: (run) => { scheduled.add(run) return () => { @@ -116,7 +117,8 @@ describe('persistent command retention', () => { turnLifecycle: null, sink, streams: items.streams, - activeItems: items.activeItems + activeItems: items.activeItems, + linkageFor: () => ({}) }) ).toEqual({ accepted: true }) } @@ -211,7 +213,8 @@ describe('persistent command retention', () => { turnLifecycle: null, sink, streams: items.streams, - activeItems: items.activeItems + activeItems: items.activeItems, + linkageFor: () => ({}) }) ).toEqual({ accepted: true }) expect(items.activeItems.size).toBe(1) diff --git a/src/main/codex/codex-structured-item-stream-contracts.ts b/src/main/codex/codex-structured-item-stream-contracts.ts index 9bfb8795ff1b..ebbd05731d49 100644 --- a/src/main/codex/codex-structured-item-stream-contracts.ts +++ b/src/main/codex/codex-structured-item-stream-contracts.ts @@ -2,14 +2,18 @@ import type { AgentJournalItemIdentity } from '../../shared/agent-session-journa import type { AgentSessionDeltaCoalescerDeps } from '../native-chat/agent-session-wire/agent-session-delta-coalescer' import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' import type { codexJournalItem, CodexThreadItem } from './codex-structured-item-translation' +import type { CodexRowLinkage } from './codex-subagent-linkage' export type CodexItemStreamDeps = { sink: StructuredAgentSessionEventSink + /** The turn a delta-only item belongs to, read the way its first delta names it. */ + turnIdFor: (threadId: string, params: unknown) => string | null identityFor: ( threadId: string, - params: unknown, + turnId: string | null, item: CodexThreadItem ) => AgentJournalItemIdentity + linkageFor: CodexRowLinkage coalesceMs?: number maxRetainedBytes?: number maxTotalRetainedBytes?: number @@ -39,7 +43,12 @@ export type CodexStructuredItemStreamHandleResult = { export type CodexStructuredItemStreams = { readonly persistentCount: number canTrack: (threadId: string, item: CodexThreadItem, identity: AgentJournalItemIdentity) => boolean - track: (threadId: string, item: CodexThreadItem, identity: AgentJournalItemIdentity) => boolean + track: ( + threadId: string, + turnId: string | null, + item: CodexThreadItem, + identity: AgentJournalItemIdentity + ) => boolean handle: ( threadId: string, method: string, diff --git a/src/main/codex/codex-structured-item-streams.ts b/src/main/codex/codex-structured-item-streams.ts index e67263f133d6..03085bf6913c 100644 --- a/src/main/codex/codex-structured-item-streams.ts +++ b/src/main/codex/codex-structured-item-streams.ts @@ -1,6 +1,7 @@ import { agentJournalItemKey } from '../../shared/agent-session-journal-item-key' import { createAgentSessionDeltaCoalescer } from '../native-chat/agent-session-wire/agent-session-delta-coalescer' import { CodexItemStreamRetention } from './codex-item-stream-retention' +import { appendCodexItemAndPublish } from './codex-structured-journal-sink' import { codexJournalItem, codexStreamingJournalItem, @@ -50,6 +51,13 @@ export function createCodexStructuredItemStreams( const states = new CodexItemStreamRetention(deps.maxMetadataBytes) const checkpointLengths = new Map() const pendingCheckpoints = new Set() + // Which thread and turn produced each stream, resolved to linkage per append + // so a parent learned after the first checkpoint still reaches the row. + const producers = new Map() + const linkageOf = (key: string) => { + const producer = producers.get(key) + return producer ? deps.linkageFor(producer.threadId, producer.turnId) : {} + } // Patch updates are authoritative item snapshots. Keep the latest rejected // snapshot until the journal admits it; unlike streamed deltas, there is no // coalescer timer to retry these events for us. @@ -61,6 +69,7 @@ export function createCodexStructuredItemStreams( states.forget(key) checkpointLengths.delete(key) pendingCheckpoints.delete(key) + producers.delete(key) const pending = pendingPatches.get(key) if (pending) { retainedPatchBytes = Math.max(0, retainedPatchBytes - pendingPatchBytes(pending)) @@ -98,23 +107,15 @@ export function createCodexStructuredItemStreams( } } - const append = (state: CodexItemStreamState, text: string): boolean => { + const append = (key: string, state: CodexItemStreamState, text: string): boolean => { const translated = codexStreamingJournalItem(state.item, text) if (!translated.body) { return true } - const options = { coalescingKey: `checkpoint:${agentJournalItemKey(state.identity)}` } - const admission = deps.sink.tryAppendItem - ? deps.sink.tryAppendItem(state.identity, translated.body, options) - : (deps.sink.appendItem(state.identity, translated.body, options), - { accepted: true as const }) - if (!admission.accepted) { - return false - } - const published = deps.sink.tryPublish - ? deps.sink.tryPublish() - : (deps.sink.publish(), { accepted: true as const }) - return published.accepted + return appendCodexItemAndPublish(deps.sink, state.identity, translated.body, { + coalescingKey: `checkpoint:${agentJournalItemKey(state.identity)}`, + ...linkageOf(key) + }).accepted } const persist = (key: string, text: string, force: boolean): boolean => { @@ -124,7 +125,7 @@ export function createCodexStructuredItemStreams( return true } const state = states.get(key) - if (state && append(state, text)) { + if (state && append(key, state, text)) { checkpointLengths.set(key, text.length) pendingCheckpoints.delete(key) return true @@ -155,10 +156,12 @@ export function createCodexStructuredItemStreams( return existing } const item = { type, id: itemId } - const state = { item, identity: deps.identityFor(threadId, params, item) } + const turnId = deps.turnIdFor(threadId, params) + const state = { item, identity: deps.identityFor(threadId, turnId, item) } if (!states.retain(key, state)) { return null } + producers.set(key, { threadId, turnId }) trimStates() return state } @@ -184,18 +187,15 @@ export function createCodexStructuredItemStreams( if (!pending) { return { accepted: true } } - const admission = deps.sink.tryAppendItem - ? deps.sink.tryAppendItem(pending.identity, pending.body) - : (deps.sink.appendItem(pending.identity, pending.body), { accepted: true as const }) + const admission = appendCodexItemAndPublish( + deps.sink, + pending.identity, + pending.body, + linkageOf(key) + ) if (!admission.accepted) { return admission } - const published = deps.sink.tryPublish - ? deps.sink.tryPublish() - : (deps.sink.publish(), { accepted: true as const }) - if (!published.accepted) { - return published - } retainedPatchBytes = Math.max(0, retainedPatchBytes - pendingPatchBytes(pending)) pendingPatches.delete(key) return { accepted: true } @@ -210,11 +210,12 @@ export function createCodexStructuredItemStreams( item: boundStreamItem(item) as CodexThreadItem, identity }), - track: (threadId, item, identity) => { + track: (threadId, turnId, item, identity) => { const key = codexStructuredItemKey(threadId, item.id) if (!states.retain(key, { item: boundStreamItem(item) as CodexThreadItem, identity })) { return false } + producers.set(key, { threadId, turnId }) trimStates() return true }, @@ -299,6 +300,7 @@ export function createCodexStructuredItemStreams( states.clear() checkpointLengths.clear() pendingCheckpoints.clear() + producers.clear() pendingPatches.clear() retainedPatchBytes = 0 }, diff --git a/src/main/codex/codex-structured-journal-compactions.ts b/src/main/codex/codex-structured-journal-compactions.ts index d2d6aeca4649..9aa2e8f4cf17 100644 --- a/src/main/codex/codex-structured-journal-compactions.ts +++ b/src/main/codex/codex-structured-journal-compactions.ts @@ -8,13 +8,15 @@ import { import { MAX_CODEX_GENERIC_TURN_BUCKETS } from './codex-structured-journal-limits' import { appendCodexLifecycleItem, publishCodexLifecycle } from './codex-structured-journal-sink' import { readCodexTurnId } from './codex-structured-thread-facts' +import type { CodexRowLinkage } from './codex-subagent-linkage' export class CodexJournalCompactions { private readonly turns = new Map() constructor( private readonly sink: StructuredAgentSessionEventSink, - private readonly activeTurn: (threadId: string) => string | null + private readonly activeTurn: (threadId: string) => string | null, + private readonly linkageFor: CodexRowLinkage ) {} handle(event: { @@ -41,7 +43,8 @@ export class CodexJournalCompactions { const admission = appendCodexLifecycleItem( this.sink, { provider: 'orca', clientMessageId: `codex-compaction:${key}` }, - { kind: 'status', text: 'Context compacted', presentation: 'compaction' } + { kind: 'status', text: 'Context compacted', presentation: 'compaction' }, + this.linkageFor(event.threadId, turnId) ) if (!admission.accepted) { return admission diff --git a/src/main/codex/codex-structured-journal-generic-frames.ts b/src/main/codex/codex-structured-journal-generic-frames.ts index 6c211a8f5a4d..7d9f54462fc9 100644 --- a/src/main/codex/codex-structured-journal-generic-frames.ts +++ b/src/main/codex/codex-structured-journal-generic-frames.ts @@ -12,9 +12,16 @@ import { MAX_CODEX_GENERIC_TURN_BUCKETS } from './codex-structured-journal-limits' import { readCodexTurnId } from './codex-structured-thread-facts' +import type { CodexRowLinkage } from './codex-subagent-linkage' const OVERFLOW_BUCKET = '__codex-generic-overflow__' -type SuppressedSummary = { count: number; publishedCount: number } +/** `producer` is absent only on the overflow bucket, which pools every thread's + * evicted turns and so has no single author: it reads as the session's own. */ +type SuppressedSummary = { + count: number + publishedCount: number + producer?: { threadId: string; turnId: string } +} function boundedTurnBucket(threadId: string, turnId: string): string { const encoded = `${encodeURIComponent(threadId)}:${encodeURIComponent(turnId)}` @@ -51,7 +58,9 @@ export class CodexJournalGenericFrames { private cancelSuppressionFlush: (() => void) | null = null constructor( - private readonly deps: Pick, + private readonly deps: Pick & { + linkageFor: CodexRowLinkage + }, private readonly activeTurn: (threadId: string) => string | null ) { this.schedule = deps.schedule ?? defaultSchedule @@ -61,7 +70,7 @@ export class CodexJournalGenericFrames { appendUnhandled( kind: string, payload: unknown, - threadId = 'session' + threadId: string ): CodexJournalTranslationAdmission { const translated = unhandledProviderFrameJournalItem('codex', kind, payload) // A frame the classifier declines is deliberately not journaled, which is success. @@ -69,14 +78,15 @@ export class CodexJournalGenericFrames { if (!translated) { return CODEX_JOURNAL_ADMITTED } - const turnId = readCodexTurnId(payload) ?? this.activeTurn(threadId) ?? 'outside-turn' + const frameTurnId = readCodexTurnId(payload) ?? this.activeTurn(threadId) + const turnId = frameTurnId ?? 'outside-turn' const bucket = this.bucketFor(threadId, turnId) const rowCount = this.genericRowsByTurn.get(bucket) ?? 0 // The cap bounds noise, never evidence: an error frame is always journaled, and // capped frames stay countable through one summary row per turn. const isError = translated.classification === 'error-surface' if (!isError && rowCount >= MAX_CODEX_GENERIC_ROWS_PER_TURN) { - this.addSuppressed(bucket, 1) + this.addSuppressed(bucket, 1, { threadId, turnId }) this.recordBucket(bucket) this.scheduleSuppressedRows() return CODEX_JOURNAL_ADMITTED @@ -88,16 +98,14 @@ export class CodexJournalGenericFrames { } } this.fallbackSequence += 1 + const identity = { + provider: 'orca' as const, + clientMessageId: `provider-frame:codex:${this.fallbackSequence}` + } + const linkage = this.deps.linkageFor(threadId, frameTurnId) const admission = this.deps.sink.tryAppendItem - ? this.deps.sink.tryAppendItem( - { provider: 'orca', clientMessageId: `provider-frame:codex:${this.fallbackSequence}` }, - translated.body - ) - : (this.deps.sink.appendItem( - { provider: 'orca', clientMessageId: `provider-frame:codex:${this.fallbackSequence}` }, - translated.body - ), - CODEX_JOURNAL_ADMITTED) + ? this.deps.sink.tryAppendItem(identity, translated.body, linkage) + : (this.deps.sink.appendItem(identity, translated.body, linkage), CODEX_JOURNAL_ADMITTED) if (!admission.accepted) { this.fallbackSequence -= 1 return admission @@ -109,7 +117,7 @@ export class CodexJournalGenericFrames { suppress(threadId: string, turnId: string, count = 1): void { const bucket = this.bucketFor(threadId, turnId) - this.addSuppressed(bucket, count) + this.addSuppressed(bucket, count, { threadId, turnId }) this.recordBucket(bucket) } @@ -127,20 +135,19 @@ export class CodexJournalGenericFrames { bucket === OVERFLOW_BUCKET ? `${summary.count} more provider notification${summary.count === 1 ? '' : 's'} not shown across evicted turns` : `${summary.count} more provider notification${summary.count === 1 ? '' : 's'} not shown for this turn` + const identity = { + provider: 'orca' as const, + clientMessageId: `provider-frame-suppressed:codex:${bucket}` + } + const options = { + coalescingKey: `provider-frame-suppressed:codex:${bucket}`, + ...(summary.producer + ? this.deps.linkageFor(summary.producer.threadId, summary.producer.turnId) + : {}) + } const admission = this.deps.sink.tryAppendItem - ? this.deps.sink.tryAppendItem( - { provider: 'orca', clientMessageId: `provider-frame-suppressed:codex:${bucket}` }, - { - kind: 'status', - text - }, - { coalescingKey: `provider-frame-suppressed:codex:${bucket}` } - ) - : (this.deps.sink.appendItem( - { provider: 'orca', clientMessageId: `provider-frame-suppressed:codex:${bucket}` }, - { kind: 'status', text }, - { coalescingKey: `provider-frame-suppressed:codex:${bucket}` } - ), + ? this.deps.sink.tryAppendItem(identity, { kind: 'status', text }, options) + : (this.deps.sink.appendItem(identity, { kind: 'status', text }, options), CODEX_JOURNAL_ADMITTED) if (!admission.accepted) { blocked ??= admission @@ -188,8 +195,16 @@ export class CodexJournalGenericFrames { : requested } - private addSuppressed(bucket: string, count: number): void { - const summary = this.suppressedRowsByTurn.get(bucket) ?? { count: 0, publishedCount: 0 } + private addSuppressed( + bucket: string, + count: number, + producer?: SuppressedSummary['producer'] + ): void { + const summary = this.suppressedRowsByTurn.get(bucket) ?? { + count: 0, + publishedCount: 0, + ...(producer && bucket !== OVERFLOW_BUCKET ? { producer } : {}) + } summary.count += count this.suppressedRowsByTurn.set(bucket, summary) } diff --git a/src/main/codex/codex-structured-journal-goal-admission.test.ts b/src/main/codex/codex-structured-journal-goal-admission.test.ts index 5fe926d18c93..b0a9e50ba2cf 100644 --- a/src/main/codex/codex-structured-journal-goal-admission.test.ts +++ b/src/main/codex/codex-structured-journal-goal-admission.test.ts @@ -137,7 +137,7 @@ describe('codex goal lifecycle admission', () => { appendTombstone: () => {}, publish: () => {} } satisfies StructuredAgentSessionEventSink - const goals = new CodexJournalGoals(sink) + const goals = new CodexJournalGoals(sink, () => ({})) const update = (goal: Record = {}) => goals.handle({ threadId: THREAD, method: 'thread/goal/updated', params: goalFrame(goal) }) const clear = () => @@ -178,7 +178,7 @@ describe('codex goal lifecycle admission', () => { appendTombstone: () => {}, publish: () => {} } satisfies StructuredAgentSessionEventSink - const goals = new CodexJournalGoals(sink) + const goals = new CodexJournalGoals(sink, () => ({})) const send = (threadId: string) => goals.handle({ threadId, method: 'thread/goal/updated', params: goalFrame() }) @@ -211,7 +211,7 @@ describe('codex goal lifecycle admission', () => { appendTombstone: () => {}, publish: () => {} } satisfies StructuredAgentSessionEventSink - const goals = new CodexJournalGoals(sink) + const goals = new CodexJournalGoals(sink, () => ({})) const event = { threadId: THREAD, method: 'thread/goal/updated', params: goalFrame() } goals.handle(event) diff --git a/src/main/codex/codex-structured-journal-goal-resume.test.ts b/src/main/codex/codex-structured-journal-goal-resume.test.ts index bb6c4b53615d..4609a06fb0d6 100644 --- a/src/main/codex/codex-structured-journal-goal-resume.test.ts +++ b/src/main/codex/codex-structured-journal-goal-resume.test.ts @@ -130,7 +130,7 @@ describe('codex goal lifecycle resume', () => { it('does not append a cleared snapshot when the journal has no prior goal occurrence', async () => { const journal = goalJournal() journal.unbind() - const resumed = new CodexJournalGoals(journal.sink) + const resumed = new CodexJournalGoals(journal.sink, () => ({})) expect( resumed.handle({ @@ -151,7 +151,7 @@ describe('codex goal lifecycle resume', () => { it('does not revisit durable history for accounting-only updates', async () => { const journal = goalJournal() - const goals = new CodexJournalGoals(journal.sink) + const goals = new CodexJournalGoals(journal.sink, () => ({})) goals.handle({ threadId: THREAD, method: 'thread/goal/updated', params: goalFrame() }) await journal.drained() const visits = journal.visits() @@ -176,7 +176,7 @@ describe('codex goal lifecycle resume', () => { it('rebuilds dedupe state after the journal epoch is replaced', async () => { const journal = goalJournal() - const goals = new CodexJournalGoals(journal.sink) + const goals = new CodexJournalGoals(journal.sink, () => ({})) const event = { threadId: THREAD, method: 'thread/goal/updated', params: goalFrame() } goals.handle(event) @@ -198,7 +198,7 @@ describe('codex goal lifecycle resume', () => { it('visits a large journal once per epoch when thread churn exceeds the transient LRU', async () => { const journal = goalJournal() journal.seedProviderItems(10_000) - const goals = new CodexJournalGoals(journal.sink) + const goals = new CodexJournalGoals(journal.sink, () => ({})) const threadCount = MAX_CODEX_GOAL_THREADS + 1 const sendRound = () => { for (let index = 0; index < threadCount; index += 1) { @@ -227,7 +227,7 @@ describe('codex goal lifecycle resume', () => { const journal = goalJournal() journal.seedProviderItems(10_000) journal.unbind() - const goals = new CodexJournalGoals(journal.sink) + const goals = new CodexJournalGoals(journal.sink, () => ({})) for (let index = 0; index < MAX_CODEX_GOAL_THREADS; index += 1) { goals.handle({ @@ -249,7 +249,7 @@ describe('codex goal lifecycle resume', () => { it('retries a journal-derived transition after lifecycle backpressure', async () => { const journal = goalJournal({ watermarks: { maxLifecycleQueuedOperations: 1 } }) - const goals = new CodexJournalGoals(journal.sink) + const goals = new CodexJournalGoals(journal.sink, () => ({})) journal.unbind() expect( @@ -317,7 +317,7 @@ describe('codex goal lifecycle resume', () => { }) } - const prior = new CodexJournalGoals(journal.sink) + const prior = new CodexJournalGoals(journal.sink, () => ({})) for (const state of scenario.beforeResume) { send(prior, state) } @@ -329,7 +329,7 @@ describe('codex goal lifecycle resume', () => { prior.dispose() journal.unbind() - const resumed = new CodexJournalGoals(journal.sink) + const resumed = new CodexJournalGoals(journal.sink, () => ({})) resumed.handle({ threadId: THREAD, method: scenario.resumed.method, diff --git a/src/main/codex/codex-structured-journal-goal-revision.test.ts b/src/main/codex/codex-structured-journal-goal-revision.test.ts index 9c4f0cd557a8..e6f96ad5503d 100644 --- a/src/main/codex/codex-structured-journal-goal-revision.test.ts +++ b/src/main/codex/codex-structured-journal-goal-revision.test.ts @@ -96,7 +96,7 @@ describe('codex goal accounting revisions', () => { { fence: 1 } ) const { sink, drained, subscribe, published } = journalSink(journal) - const goals = new CodexJournalGoals(sink) + const goals = new CodexJournalGoals(sink, () => ({})) goals.handle({ threadId: THREAD, method: 'thread/goal/updated', params: goalFrame() }) await drained() @@ -160,14 +160,14 @@ describe('codex goal accounting revisions', () => { root = await mkdtemp(join(tmpdir(), 'orca-goal-resume-revision-')) const journal = await journals.open({ identity: IDENTITY, journalDir: root }) const first = journalSink(journal) - const prior = new CodexJournalGoals(first.sink) + const prior = new CodexJournalGoals(first.sink, () => ({})) prior.handle({ threadId: THREAD, method: 'thread/goal/updated', params: goalFrame() }) await first.drained() prior.dispose() const [created] = journal.snapshot().items const second = journalSink(journal) - const resumed = new CodexJournalGoals(second.sink) + const resumed = new CodexJournalGoals(second.sink, () => ({})) // Only a few seconds more: a resume snapshot still refreshes, since no live tick follows. const snapshot = goalFrame({ timeUsedSeconds: 3, updatedAt: 1789067991 }) resumed.handle({ threadId: THREAD, method: 'thread/goal/updated', params: snapshot }) diff --git a/src/main/codex/codex-structured-journal-goal-rows.test.ts b/src/main/codex/codex-structured-journal-goal-rows.test.ts index 7af1d863e174..ff234c7d5fb1 100644 --- a/src/main/codex/codex-structured-journal-goal-rows.test.ts +++ b/src/main/codex/codex-structured-journal-goal-rows.test.ts @@ -35,8 +35,8 @@ function frames(): { }, publish: vi.fn() } as unknown as StructuredAgentSessionEventSink - const goals = new CodexJournalGoals(sink) - const generic = new CodexJournalGenericFrames({ sink }, () => null) + const goals = new CodexJournalGoals(sink, () => ({})) + const generic = new CodexJournalGenericFrames({ sink, linkageFor: () => ({}) }, () => null) return { rows, frames: { diff --git a/src/main/codex/codex-structured-journal-goals.ts b/src/main/codex/codex-structured-journal-goals.ts index 6d9f19178816..681a483c12f7 100644 --- a/src/main/codex/codex-structured-journal-goals.ts +++ b/src/main/codex/codex-structured-journal-goals.ts @@ -24,6 +24,7 @@ import { } from './codex-structured-journal-contracts' import { MAX_CODEX_GOAL_THREADS } from './codex-structured-journal-limits' import { appendCodexLifecycleTransition } from './codex-structured-journal-sink' +import type { CodexRowLinkage } from './codex-subagent-linkage' type GoalAccounting = { key: string; timeUsedSeconds: number; tokenBudget: number | null } @@ -67,7 +68,10 @@ export class CodexJournalGoals { private durableEpoch: string | null = null private transientEpoch: string | null = null - constructor(private readonly sink: StructuredAgentSessionEventSink) {} + constructor( + private readonly sink: StructuredAgentSessionEventSink, + private readonly linkageFor: CodexRowLinkage + ) {} handle(event: { threadId: string @@ -126,7 +130,9 @@ export class CodexJournalGoals { signatureKey, accounting, event.method === 'thread/goal/cleared' - ) + ), + // A goal belongs to its thread, not to one run of it, so no turn is named. + this.linkageFor(event.threadId, null) ) if (!admission.accepted) { return admission diff --git a/src/main/codex/codex-structured-journal-items.ts b/src/main/codex/codex-structured-journal-items.ts index b0e9cba58bf7..6717f3edabe3 100644 --- a/src/main/codex/codex-structured-journal-items.ts +++ b/src/main/codex/codex-structured-journal-items.ts @@ -1,6 +1,7 @@ import type { AgentJournalItemBody, - AgentJournalItemIdentity + AgentJournalItemIdentity, + AgentJournalProducerLinkage } from '../../shared/agent-session-journal-types' import { requiresTerminalSettlement } from '../native-chat/agent-session-journal/journal-terminal-settlement' import { @@ -30,6 +31,7 @@ import type { CodexActiveJournalItem } from './codex-structured-journal-settleme import { readCodexJournalString } from './codex-structured-journal-translation-values' import { readCodexTurnId } from './codex-structured-thread-facts' import { readCodexDispatchEcho } from './codex-structured-dispatch-echo' +import type { CodexRowLinkage } from './codex-subagent-linkage' export class CodexJournalItems { readonly ordinals = new CodexTurnOrdinals() @@ -42,7 +44,7 @@ export class CodexJournalItems { private readonly deps: Pick< CodexJournalTranslatorDeps, 'sink' | 'coalesceMs' | 'maxRetainedBytes' | 'schedule' - > & { maxMetadataBytes?: number }, + > & { maxMetadataBytes?: number; linkageFor: CodexRowLinkage }, private readonly activeTurn: (threadId: string) => string | null, private readonly suppress: (threadId: string, turnId: string) => void ) { @@ -52,10 +54,9 @@ export class CodexJournalItems { maxRetainedBytes: deps.maxRetainedBytes, schedule: deps.schedule, maxMetadataBytes: deps.maxMetadataBytes, - identityFor: (threadId, params, item) => { - const turnId = readCodexTurnId(params) ?? this.activeTurn(threadId) - return this.identityFor(threadId, turnId, item) - } + turnIdFor: (threadId, params) => readCodexTurnId(params) ?? this.activeTurn(threadId), + identityFor: (threadId, turnId, item) => this.identityFor(threadId, turnId, item), + linkageFor: deps.linkageFor }) } @@ -117,7 +118,12 @@ export class CodexJournalItems { } return { handled: true, admission: CODEX_JOURNAL_ADMITTED } } - const admission = this.appendTranslated(event.method, identity, translated) + const admission = this.appendTranslated( + event.method, + identity, + translated, + this.deps.linkageFor(event.threadId, turnId) + ) if (!admission.accepted) { return { handled: true, admission } } @@ -144,19 +150,22 @@ export class CodexJournalItems { private appendTranslated( method: string, identity: AgentJournalItemIdentity, - translated: ReturnType + translated: ReturnType, + linkage: AgentJournalProducerLinkage ): CodexJournalTranslationAdmission { if (!translated.body) { return CODEX_JOURNAL_ADMITTED } if (method === 'item/completed') { - const admission = appendCodexLifecycleItem(this.deps.sink, identity, translated.body) + const admission = appendCodexLifecycleItem(this.deps.sink, identity, translated.body, linkage) return admission.accepted ? publishCodexLifecycle(this.deps.sink) : admission } const options = requiresTerminalSettlement(translated.body) ? { lifecycle: true } : {} + const appendOptions = { ...options, ...linkage } const admission = this.deps.sink.tryAppendItem - ? this.deps.sink.tryAppendItem(identity, translated.body, options) - : (this.deps.sink.appendItem(identity, translated.body), CODEX_JOURNAL_ADMITTED) + ? this.deps.sink.tryAppendItem(identity, translated.body, appendOptions) + : (this.deps.sink.appendItem(identity, translated.body, appendOptions), + CODEX_JOURNAL_ADMITTED) if (!admission.accepted) { return admission } @@ -174,7 +183,7 @@ export class CodexJournalItems { const retainedItem = codexCommandOutlivesTurn(item) ? (boundStreamItem(item) as CodexThreadItem) : item - this.streams.track(threadId, retainedItem, identity) + this.streams.track(threadId, turnId, retainedItem, identity) this.activeItems.set(codexStructuredItemKey(threadId, item.id), { threadId, turnId, @@ -225,7 +234,8 @@ export class CodexJournalItems { const admission = appendCodexLifecycleItem( this.deps.sink, evicted.identity, - evictedActiveBody(translated) + evictedActiveBody(translated), + this.deps.linkageFor(evicted.threadId, evicted.turnId) ) if (!admission.accepted) { return admission diff --git a/src/main/codex/codex-structured-journal-prompts.ts b/src/main/codex/codex-structured-journal-prompts.ts index f72fd6264ef6..96c58f9b8b7b 100644 --- a/src/main/codex/codex-structured-journal-prompts.ts +++ b/src/main/codex/codex-structured-journal-prompts.ts @@ -20,6 +20,8 @@ import { } from './codex-structured-journal-sink' import type { CodexPendingJournalPrompt } from './codex-structured-journal-settlement' import { readCodexTurnId } from './codex-structured-thread-facts' +import type { CodexRowLinkage } from './codex-subagent-linkage' +import { journalLifecycleItemMutation } from '../native-chat/agent-session-journal/journal-row-builders' type CodexGroupedPendingJournalPrompt = CodexPendingJournalPrompt & { promptKey: string } @@ -27,7 +29,9 @@ export class CodexJournalPrompts { readonly pending = new Map() constructor( - private readonly deps: Pick, + private readonly deps: Pick & { + linkageFor: CodexRowLinkage + }, private readonly detailFor: (threadId: string, itemId: string) => string | null, private readonly activeTurn: (threadId: string) => string | null ) {} @@ -47,7 +51,7 @@ export class CodexJournalPrompts { params: event.params }) const promptItems = questions.map(({ identity, body }) => ({ identity, body })) - const admission = this.admit(event, promptItems) + const admission = this.admit(event, turnId, promptItems) if (!admission.accepted) { return admission } @@ -77,7 +81,7 @@ export class CodexJournalPrompts { params: event.params, detail: this.detailFor(event.threadId, event.codexItemId) }) - const admission = this.admit(event, [{ identity, body }]) + const admission = this.admit(event, turnId, [{ identity, body }]) if (!admission.accepted) { return admission } @@ -114,7 +118,8 @@ export class CodexJournalPrompts { ) const mutations = group.flatMap(([, prompt]) => { const body = cancelledJournalPromptBody(prompt.body) - return body ? [{ kind: 'item' as const, identity: prompt.identity, body }] : [] + const producer = this.deps.linkageFor(prompt.threadId, prompt.turnId) + return body ? [journalLifecycleItemMutation(producer, prompt.identity, body)] : [] }) const admission = appendCodexLifecycleMutations( this.deps.sink, @@ -137,6 +142,7 @@ export class CodexJournalPrompts { private admit( event: { method: string; threadId: string; promptKey: string }, + turnId: string | null, items: readonly Pick[] ): CodexJournalTranslationAdmission { return admitCodexLifecycleItems( @@ -144,7 +150,9 @@ export class CodexJournalPrompts { `prompt:${encodeURIComponent(event.method)}:${encodeURIComponent( event.threadId )}:${encodeURIComponent(event.promptKey)}`, - items + items, + // A child's approval arrives on the child's own thread, so it names the asker. + this.deps.linkageFor(event.threadId, turnId) ) } @@ -158,7 +166,12 @@ export class CodexJournalPrompts { if (evicted) { const cancelled = cancelledJournalPromptBody(evicted.body) if (cancelled) { - const admission = appendCodexLifecycleItem(this.deps.sink, evicted.identity, cancelled) + const admission = appendCodexLifecycleItem( + this.deps.sink, + evicted.identity, + cancelled, + this.deps.linkageFor(evicted.threadId, evicted.turnId) + ) if (!admission.accepted) { return admission } diff --git a/src/main/codex/codex-structured-journal-settlement.ts b/src/main/codex/codex-structured-journal-settlement.ts index 5322d3355cd1..e2164b9498dd 100644 --- a/src/main/codex/codex-structured-journal-settlement.ts +++ b/src/main/codex/codex-structured-journal-settlement.ts @@ -3,7 +3,10 @@ import type { AgentJournalItemIdentity, AgentJournalTurnLifecycle } from '../../shared/agent-session-journal-types' -import type { JournalLifecycleMutationInput } from '../native-chat/agent-session-journal/journal-row-builders' +import { + journalLifecycleItemMutation, + type JournalLifecycleMutationInput +} from '../native-chat/agent-session-journal/journal-row-builders' import type { StructuredAgentSessionEventSink, StructuredAgentSessionSinkAdmission @@ -23,6 +26,7 @@ import { codexTurnLifecycleIdentity } from './codex-structured-journal-translation-turns' import { appendCodexLifecycleMutations } from './codex-structured-journal-sink' +import type { CodexRowLinkage } from './codex-subagent-linkage' export type CodexActiveJournalItem = { threadId: string @@ -51,7 +55,9 @@ export function settleCodexJournalSession(input: { ordinals: CodexTurnOrdinals /** Terminal lifecycle for a turn the provider left running when it ended. */ settledTurnLifecycle: (threadId: string, turnId: string) => AgentJournalTurnLifecycle + linkageFor: CodexRowLinkage }): StructuredAgentSessionSinkAdmission { + // Rows from every thread settle in this one batch, so each names its own producer. const mutations: JournalLifecycleMutationInput[] = [] const turnOrdinalsToForget: { threadId: string; turnId: string }[] = [] for (const active of input.activeItems.values()) { @@ -61,17 +67,13 @@ export function settleCodexJournalSession(input: { : codexJournalItem(active.item) const body = interruptedBody(translated.body) if (body) { - mutations.push({ kind: 'item', identity: active.identity, body }) + mutations.push(settledRow(input.linkageFor, active, body)) } } for (const prompt of input.pendingPrompts.values()) { const body = cancelledJournalPromptBody(prompt.body) if (body) { - mutations.push({ - kind: 'item', - identity: prompt.identity, - body - }) + mutations.push(settledRow(input.linkageFor, prompt, body)) } } for (const [threadId, turnIds] of input.currentTurnIds) { @@ -112,6 +114,7 @@ export function settleCodexJournalTurn(input: { activeItems: Map pendingPrompts?: Map clearPromptTurn?: (threadId: string, turnId: string) => void + linkageFor: CodexRowLinkage }): StructuredAgentSessionSinkAdmission { const mutations: JournalLifecycleMutationInput[] = [] const activeItemsToForget: { key: string; threadId: string; itemId: string }[] = [] @@ -130,7 +133,7 @@ export function settleCodexJournalTurn(input: { : codexJournalItem(active.item) const body = interruptedBody(translated.body) if (body) { - mutations.push({ kind: 'item', identity: active.identity, body }) + mutations.push(settledRow(input.linkageFor, active, body)) } activeItemsToForget.push({ key, threadId: active.threadId, itemId: active.item.id }) } @@ -140,7 +143,7 @@ export function settleCodexJournalTurn(input: { } const body = cancelledJournalPromptBody(prompt.body) if (body) { - mutations.push({ kind: 'item', identity: prompt.identity, body }) + mutations.push(settledRow(input.linkageFor, prompt, body)) } pendingPromptsToForget.push(key) } @@ -179,6 +182,7 @@ export function settleCodexOversizedNotification(input: { sink: StructuredAgentSessionEventSink streams: CodexStructuredItemStreams activeItems: Map + linkageFor: CodexRowLinkage }): StructuredAgentSessionSinkAdmission { const itemType = oversizedStreamItemType(input.method) if (!itemType) { @@ -196,7 +200,7 @@ export function settleCodexOversizedNotification(input: { : codexJournalItem(active.item) const body = interruptedBody(translated.body) if (body) { - mutations.push({ kind: 'item', identity: active.identity, body }) + mutations.push(settledRow(input.linkageFor, active, body)) } activeItemsToForget.push({ key, threadId: active.threadId, itemId: active.item.id }) } @@ -246,6 +250,15 @@ function oversizedStreamItemType(method: string): CodexThreadItem['type'] | null return null } +/** A settled item or prompt, naming its producer: the settlement can be the row's first write. */ +function settledRow( + linkageFor: CodexRowLinkage, + row: { threadId: string; turnId: string | null; identity: AgentJournalItemIdentity }, + body: AgentJournalItemBody +): JournalLifecycleMutationInput { + return journalLifecycleItemMutation(linkageFor(row.threadId, row.turnId), row.identity, body) +} + function interruptedBody(body: AgentJournalItemBody | null): AgentJournalItemBody | null { if (!body) { return null diff --git a/src/main/codex/codex-structured-journal-sink.ts b/src/main/codex/codex-structured-journal-sink.ts index 5b8f4b839202..95a27702365f 100644 --- a/src/main/codex/codex-structured-journal-sink.ts +++ b/src/main/codex/codex-structured-journal-sink.ts @@ -1,14 +1,19 @@ import type { AgentJournalItemBody, - AgentJournalItemIdentity + AgentJournalItemIdentity, + AgentJournalProducerLinkage } from '../../shared/agent-session-journal-types' import type { + StructuredAgentSessionAppendOptions, StructuredAgentSessionEventSink, StructuredAgentSessionLifecycleIdentityResolver, StructuredAgentSessionSinkAdmission } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' import { partitionJournalLifecycleMutations } from '../native-chat/agent-session-journal/journal-lifecycle-batch-partition' -import type { JournalLifecycleMutationInput } from '../native-chat/agent-session-journal/journal-row-builders' +import { + journalLifecycleItemMutation, + type JournalLifecycleMutationInput +} from '../native-chat/agent-session-journal/journal-row-builders' import type { CodexPendingJournalPrompt } from './codex-structured-journal-settlement' import type { CodexJournalTranslationAdmission } from './codex-structured-journal-contracts' import { CODEX_JOURNAL_ADMITTED } from './codex-structured-journal-contracts' @@ -30,13 +35,14 @@ export function appendCodexLifecycleMutations( } else { for (const mutation of chunk) { if (mutation.kind === 'item') { + const options = { lifecycle: true, ...mutation.linkage } if (sink.tryAppendItem) { - admission = sink.tryAppendItem(mutation.identity, mutation.body, { lifecycle: true }) + admission = sink.tryAppendItem(mutation.identity, mutation.body, options) if (!admission.accepted) { return admission } } else { - sink.appendItem(mutation.identity, mutation.body, { lifecycle: true }) + sink.appendItem(mutation.identity, mutation.body, options) } } else { if (sink.tryAppendTombstone) { @@ -63,6 +69,22 @@ export function appendCodexLifecycleMutations( return ADMITTED } +/** An ordinary (non-lifecycle) append, published once admitted. */ +export function appendCodexItemAndPublish( + sink: StructuredAgentSessionEventSink, + identity: AgentJournalItemIdentity, + body: AgentJournalItemBody, + options: StructuredAgentSessionAppendOptions +): StructuredAgentSessionSinkAdmission { + const admission = sink.tryAppendItem + ? sink.tryAppendItem(identity, body, options) + : (sink.appendItem(identity, body, options), ADMITTED) + if (!admission.accepted) { + return admission + } + return sink.tryPublish ? sink.tryPublish() : (sink.publish(), ADMITTED) +} + function criticalAdmission( admission: StructuredAgentSessionSinkAdmission ): CodexJournalTranslationAdmission { @@ -72,12 +94,14 @@ function criticalAdmission( export function appendCodexLifecycleItem( sink: StructuredAgentSessionEventSink, identity: AgentJournalItemIdentity, - body: AgentJournalItemBody + body: AgentJournalItemBody, + linkage: AgentJournalProducerLinkage ): CodexJournalTranslationAdmission { + const options = { lifecycle: true, ...linkage } if (sink.tryAppendItem) { - return criticalAdmission(sink.tryAppendItem(identity, body, { lifecycle: true })) + return criticalAdmission(sink.tryAppendItem(identity, body, options)) } - sink.appendItem(identity, body, { lifecycle: true }) + sink.appendItem(identity, body, options) return CODEX_JOURNAL_ADMITTED } @@ -85,14 +109,15 @@ export function appendCodexLifecycleTransition( sink: StructuredAgentSessionEventSink, identitySizeBound: AgentJournalItemIdentity, body: AgentJournalItemBody, - resolveIdentity: StructuredAgentSessionLifecycleIdentityResolver + resolveIdentity: StructuredAgentSessionLifecycleIdentityResolver, + linkage: AgentJournalProducerLinkage ): CodexJournalTranslationAdmission { if (sink.tryAppendLifecycleTransition) { return criticalAdmission( - sink.tryAppendLifecycleTransition(identitySizeBound, body, resolveIdentity) + sink.tryAppendLifecycleTransition(identitySizeBound, body, resolveIdentity, linkage) ) } - const admission = appendCodexLifecycleItem(sink, identitySizeBound, body) + const admission = appendCodexLifecycleItem(sink, identitySizeBound, body, linkage) return admission.accepted ? publishCodexLifecycle(sink) : admission } @@ -106,10 +131,12 @@ export function publishCodexLifecycle( return CODEX_JOURNAL_ADMITTED } +/** One producer's rows, admitted together. */ export function admitCodexLifecycleItems( sink: StructuredAgentSessionEventSink, settlementId: string, - items: readonly Pick[] + items: readonly Pick[], + linkage: AgentJournalProducerLinkage ): CodexJournalTranslationAdmission { if (items.length === 0) { return { accepted: false, reason: 'untranslated' } @@ -118,14 +145,14 @@ export function admitCodexLifecycleItems( const admission = criticalAdmission( sink.tryAppendLifecycleBatch( settlementId, - items.map((item) => ({ kind: 'item' as const, identity: item.identity, body: item.body })), + items.map((item) => journalLifecycleItemMutation(linkage, item.identity, item.body)), { lifecycle: true } ) ) return admission.accepted ? publishCodexLifecycle(sink) : admission } for (const item of items) { - const admission = appendCodexLifecycleItem(sink, item.identity, item.body) + const admission = appendCodexLifecycleItem(sink, item.identity, item.body, linkage) if (!admission.accepted) { return admission } diff --git a/src/main/codex/codex-structured-journal-subagent-readers.test.ts b/src/main/codex/codex-structured-journal-subagent-readers.test.ts new file mode 100644 index 000000000000..117f204e9637 --- /dev/null +++ b/src/main/codex/codex-structured-journal-subagent-readers.test.ts @@ -0,0 +1,182 @@ +// A Codex subagent's rows, through the real path a parent's surfaces read them: +// translator → deferred sink → on-disk journal → snapshot → the shared readers. +// Every reader here answers for the SESSION'S OWN agent, so a child's row must +// never decide what it says; the transcript still renders every agent's rows. + +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 type { AgentJournalRenderItem } from '../../shared/agent-session-journal-types' +import { selectStructuredAgentTurnActivity } from '../../shared/native-chat-turn-activity' +import { latestStructuredAgentSessionAssistantMessage } from '../../shared/structured-agent-session-projection' +import { + isStructuredAgentSessionThinking, + statusStructuredAgentSessionToolCall +} from '../../shared/structured-agent-session-live-turn' +import { openAgentSessionJournal } from '../native-chat/agent-session-journal/journal-store-factory' +import type { AgentSessionJournal } from '../native-chat/agent-session-journal/journal-store' +import { createDeferredStructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' +import { createCodexJournalTranslator } from './codex-structured-journal-translation' +import type { CodexThreadItem } from './codex-thread-item-identity' + +const SESSION = 'session-codex-children' +const PARENT = 'thread-parent' +const CHILD = 'thread-child' +const PARENT_TURN = 'turn-parent' +const CHILD_TURN = 'turn-child' + +const cleanups: (() => Promise)[] = [] +afterEach(async () => { + for (const cleanup of cleanups.splice(0)) { + await cleanup() + } +}) + +async function openJournal(root: string): Promise { + return openAgentSessionJournal({ + identity: { + sessionId: SESSION, + workspaceId: 'workspace-1', + hostId: 'local', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: PARENT } + }, + journalDir: root, + now: () => 1_000 + }) +} + +async function session() { + const root = await mkdtemp(join(tmpdir(), 'orca-codex-children-')) + let journal = await openJournal(root) + const deferred = createDeferredStructuredAgentSessionEventSink() + deferred.bind({ journal, fence: 1, publish: () => {} }) + cleanups.push(async () => { + deferred.close() + await journal.close() + await rm(root, { recursive: true, force: true }) + }) + const translator = createCodexJournalTranslator({ + sink: deferred.sink, + sessionId: SESSION, + primaryThreadId: () => PARENT, + schedule: (run) => { + run() + return () => {} + } + }) + const on = (threadId: string, method: string, params: Record = {}) => + translator.handle({ + type: 'notification', + sessionId: SESSION, + threadId, + method, + params: { threadId, ...params } + }) + const item = (threadId: string, method: string, turnId: string, body: CodexThreadItem) => + on(threadId, method, { turnId, item: body }) + return { + on, + item, + /** The parent's turn is open and has spawned a running child. */ + spawnChild: () => { + on(PARENT, 'turn/started', { turn: { id: PARENT_TURN } }) + const spawn = { + type: 'subAgentActivity', + id: 'spawn-1', + kind: 'started', + agentThreadId: CHILD, + agentPath: '/root/review' + } + item(PARENT, 'item/started', PARENT_TURN, spawn) + item(PARENT, 'item/completed', PARENT_TURN, spawn) + on(CHILD, 'turn/started', { turn: { id: CHILD_TURN } }) + }, + items: async (): Promise => { + await deferred.drained() + return journal.snapshot().items + }, + /** Closes and reopens the journal file, so reads come from what was persisted. */ + reopen: async (): Promise => { + await deferred.drained() + deferred.unbind() + await journal.close() + journal = await openJournal(root) + return journal.snapshot().items + } + } +} + +describe("a Codex subagent's rows on the parent's surfaces", () => { + it("names the parent's own tool, not the child's running command", async () => { + const { spawnChild, item, items } = await session() + spawnChild() + item(PARENT, 'item/completed', PARENT_TURN, { + type: 'commandExecution', + id: 'own-cmd', + command: 'git status', + status: 'completed', + exitCode: 0 + }) + item(CHILD, 'item/started', CHILD_TURN, { + type: 'commandExecution', + id: 'child-cmd', + command: 'pnpm test', + status: 'inProgress' + }) + + const rows = await items() + const named = statusStructuredAgentSessionToolCall(rows) + expect(JSON.stringify(named)).toContain('git status') + expect(JSON.stringify(named)).not.toContain('pnpm test') + // The transcript is unscoped: the child's command is still a row. + expect(JSON.stringify(rows)).toContain('pnpm test') + }) + + it("does not read the child's reasoning as the parent thinking", async () => { + const { spawnChild, item, items } = await session() + spawnChild() + item(PARENT, 'item/completed', PARENT_TURN, { + type: 'agentMessage', + id: 'own-msg', + text: 'I asked a reviewer.' + }) + item(CHILD, 'item/completed', CHILD_TURN, { + type: 'reasoning', + id: 'child-reasoning', + summary: ['Reading the diff'] + }) + + expect(isStructuredAgentSessionThinking(await items())).toBe(false) + }) + + it("does not show the child's compaction as the parent's activity line", async () => { + const { spawnChild, item, items } = await session() + spawnChild() + item(CHILD, 'item/completed', CHILD_TURN, { type: 'contextCompaction', id: 'child-compact' }) + + const rows = await items() + expect(selectStructuredAgentTurnActivity(rows, PARENT_TURN)).toBeNull() + expect( + rows.some((row) => row.body.kind === 'status' && row.body.text === 'Context compacted') + ).toBe(true) + }) + + it("quotes the parent's own latest line, and still does after a reopen", async () => { + const { spawnChild, item, reopen } = await session() + spawnChild() + item(PARENT, 'item/completed', PARENT_TURN, { + type: 'agentMessage', + id: 'own-msg', + text: 'I asked a reviewer.' + }) + item(CHILD, 'item/completed', CHILD_TURN, { + type: 'agentMessage', + id: 'child-msg', + text: 'Looks good to me.' + }) + + expect(latestStructuredAgentSessionAssistantMessage(await reopen())).toBe('I asked a reviewer.') + }) +}) diff --git a/src/main/codex/codex-structured-journal-translation-frames.ts b/src/main/codex/codex-structured-journal-translation-frames.ts index d4cb2c825d98..f94c10bfe667 100644 --- a/src/main/codex/codex-structured-journal-translation-frames.ts +++ b/src/main/codex/codex-structured-journal-translation-frames.ts @@ -27,6 +27,7 @@ export function settleCodexOversizedNotificationFrame(input: { sink: OversizedInput['sink'] streams: OversizedInput['streams'] activeItems: OversizedInput['activeItems'] + linkageFor: OversizedInput['linkageFor'] }): CodexJournalTranslationAdmission | null { if (input.kind !== 'frame:oversized-notification') { return null @@ -39,13 +40,14 @@ export function settleCodexOversizedNotificationFrame(input: { method, sink: input.sink, streams: input.streams, - activeItems: input.activeItems + activeItems: input.activeItems, + linkageFor: input.linkageFor }) : null } export function createCodexOversizedNotificationSettler( - deps: { sink: OversizedInput['sink'] }, + deps: { sink: OversizedInput['sink']; linkageFor: OversizedInput['linkageFor'] }, items: Pick ) { return settleOversizedNotification @@ -59,7 +61,8 @@ export function createCodexOversizedNotificationSettler( ...event, sink: deps.sink, streams: items.streams, - activeItems: items.activeItems + activeItems: items.activeItems, + linkageFor: deps.linkageFor }) } } diff --git a/src/main/codex/codex-structured-journal-translation-subagent-linkage.test.ts b/src/main/codex/codex-structured-journal-translation-subagent-linkage.test.ts new file mode 100644 index 000000000000..16474874d5de --- /dev/null +++ b/src/main/codex/codex-structured-journal-translation-subagent-linkage.test.ts @@ -0,0 +1,271 @@ +import { describe, expect, it } from 'vitest' +import { agentJournalItemKey } from '../../shared/agent-session-journal-item-key' +import type { + AgentJournalItemBody, + AgentJournalProducerLinkage +} from '../../shared/agent-session-journal-types' +import { agentJournalLinkageFields } from '../../shared/agent-session-journal-producer' +import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' +import { createCodexJournalTranslator } from './codex-structured-journal-translation' +import type { CodexThreadItem } from './codex-thread-item-identity' +import { CODEX_COMMAND_APPROVAL_METHOD } from './codex-structured-prompt-replies' + +const SESSION = 'session-1' +const PARENT = 'thread-parent' +const CHILD = 'thread-child' +const GRANDCHILD = 'thread-grandchild' +const CHILD_LINKAGE = { agentId: CHILD, producerKind: 'agent' } + +type Written = { key: string; body: AgentJournalItemBody; linkage: AgentJournalProducerLinkage } + +/** Records the linkage every write carries — plain appends, batch mutations and + * lifecycle transitions alike — so an assertion about attribution can never + * pass against a harness that dropped it. */ +function harness(primaryThreadId: string | null = PARENT) { + const writes: Written[] = [] + const record = ( + identity: Parameters[0], + body: AgentJournalItemBody, + options: AgentJournalProducerLinkage | undefined + ) => + writes.push({ + key: agentJournalItemKey(identity), + body, + linkage: agentJournalLinkageFields(options) + }) + const sink: StructuredAgentSessionEventSink = { + appendItem: (identity, body, options) => record(identity, body, options), + tryAppendItem: (identity, body, options) => { + record(identity, body, options) + return { accepted: true } + }, + tryAppendLifecycleBatch: (_settlementId, mutations) => { + for (const mutation of mutations) { + if (mutation.kind === 'item') { + record(mutation.identity, mutation.body, mutation.linkage) + } + } + return { accepted: true } + }, + tryAppendLifecycleTransition: (identity, body, _resolve, options) => { + record(identity, body, options) + return { accepted: true } + }, + appendTombstone: () => {}, + publish: () => {} + } + const translator = createCodexJournalTranslator({ + sink, + sessionId: SESSION, + primaryThreadId: () => primaryThreadId, + schedule: (run) => { + run() + return () => {} + } + }) + const on = (threadId: string, method: string, params: Record = {}) => + translator.handle({ + type: 'notification', + sessionId: SESSION, + threadId, + method, + params: { threadId, ...params } + }) + const item = (threadId: string, method: string, turnId: string, body: CodexThreadItem) => + on(threadId, method, { turnId, item: body }) + const spawn = (spawner: string, turnId: string, child: string, path: string) => { + const body = { + type: 'subAgentActivity', + id: `spawn-${child}`, + kind: 'started', + agentThreadId: child, + agentPath: path + } + item(spawner, 'item/started', turnId, body) + item(spawner, 'item/completed', turnId, body) + } + /** The linkage on the newest write whose serialized body mentions `text`. */ + const linkageOf = (text: string): AgentJournalProducerLinkage | undefined => + writes.findLast((write) => JSON.stringify(write.body).includes(text))?.linkage + return { translator, writes, on, item, spawn, linkageOf } +} + +describe('codex journal translation — subagent producer linkage', () => { + it("stamps every row a child thread produces, and nothing on the session's own", () => { + const { on, item, spawn, linkageOf, writes } = harness() + on(PARENT, 'turn/started', { turn: { id: 'pt' } }) + spawn(PARENT, 'pt', CHILD, '/root/review') + on(CHILD, 'turn/started', { turn: { id: 'ct' } }) + const command = { type: 'commandExecution', id: 'cmd-1', command: 'pnpm test' } + item(CHILD, 'item/started', 'ct', { ...command, status: 'inProgress' }) + on(CHILD, 'item/commandExecution/outputDelta', { itemId: 'cmd-1', delta: 'streamed out' }) + item(CHILD, 'item/completed', 'ct', { ...command, status: 'completed', exitCode: 0 }) + item(CHILD, 'item/completed', 'ct', { type: 'agentMessage', id: 'm-1', text: 'child prose' }) + item(PARENT, 'item/completed', 'pt', { type: 'agentMessage', id: 'm-2', text: 'own prose' }) + + const commandRows = writes.filter((write) => write.key.includes('cmd-1')) + // Start, streamed checkpoint and completion: every revision restates it. + expect(commandRows.length).toBeGreaterThanOrEqual(3) + expect(commandRows.map((write) => write.linkage)).toEqual(commandRows.map(() => CHILD_LINKAGE)) + expect(linkageOf('child prose')).toEqual(CHILD_LINKAGE) + expect(linkageOf('own prose')).toEqual({}) + // The spawn-group row is the parent's list of its children, though a child's + // frames also write it. + expect( + writes.filter((write) => write.key.includes('codex-subagents')).map((w) => w.linkage) + ).toEqual(expect.arrayContaining([{}])) + expect( + writes.filter((write) => write.key.includes('codex-subagents') && write.linkage.agentId) + ).toEqual([]) + }) + + it('stamps a child row that arrives before the spawn announcement names it', () => { + // The thread already proves it is not the session's own; only the parent + // and run are learned from the announcement. + const { on, item, linkageOf } = harness() + on(PARENT, 'turn/started', { turn: { id: 'pt' } }) + item(CHILD, 'item/completed', 'ct', { type: 'agentMessage', id: 'm-1', text: 'early words' }) + + expect(linkageOf('early words')).toEqual(CHILD_LINKAGE) + }) + + it('names the child that spawned a grandchild, from the thread that announced it', () => { + const { on, item, spawn, linkageOf } = harness() + on(PARENT, 'turn/started', { turn: { id: 'pt' } }) + spawn(PARENT, 'pt', CHILD, '/root/lead') + on(CHILD, 'turn/started', { turn: { id: 'ct' } }) + spawn(CHILD, 'ct', GRANDCHILD, '/root/lead/worker') + // An `interacted` activity rides whichever agent acted, so it names no parent. + item(CHILD, 'item/completed', 'ct', { + type: 'subAgentActivity', + id: 'poke-parent', + kind: 'interacted', + agentThreadId: 'thread-sibling', + agentPath: '/root/sibling' + }) + on(GRANDCHILD, 'turn/started', { turn: { id: 'gt' } }) + item(GRANDCHILD, 'item/completed', 'gt', { type: 'agentMessage', id: 'g', text: 'grand words' }) + item(CHILD, 'item/completed', 'ct', { type: 'agentMessage', id: 'c', text: 'child words' }) + item('thread-sibling', 'item/completed', 'st', { type: 'agentMessage', id: 's', text: 'sib' }) + + expect(linkageOf('grand words')).toEqual({ + ...CHILD_LINKAGE, + agentId: GRANDCHILD, + parentAgentId: CHILD + }) + expect(linkageOf('child words')).toEqual(CHILD_LINKAGE) + expect(linkageOf('sib')).toEqual({ agentId: 'thread-sibling', producerKind: 'agent' }) + }) + + it("names the child's run by the row's own turn, so a shell outliving its turn keeps it", () => { + const { on, item, spawn, linkageOf } = harness() + on(PARENT, 'turn/started', { turn: { id: 'pt' } }) + spawn(PARENT, 'pt', CHILD, '/root/review') + on(CHILD, 'turn/started', { turn: { id: 'ct-1' } }) + const shell = { + type: 'commandExecution', + id: 'dev-server', + command: 'pnpm dev', + source: 'unifiedExecStartup' + } + item(CHILD, 'item/started', 'ct-1', { ...shell, status: 'inProgress' }) + on(CHILD, 'turn/completed', { turn: { id: 'ct-1', status: 'completed' } }) + // A follow-up from the parent is the child's second run. + on(CHILD, 'turn/started', { turn: { id: 'ct-2' } }) + item(CHILD, 'item/completed', 'ct-2', { type: 'agentMessage', id: 'r2', text: 'second run' }) + item(CHILD, 'item/completed', 'ct-1', { ...shell, status: 'completed', exitCode: 0 }) + + expect(linkageOf('second run')).toEqual({ ...CHILD_LINKAGE, attempt: 2 }) + expect(linkageOf('pnpm dev')).toEqual(CHILD_LINKAGE) + }) + + it("settles every thread's rows in one exit batch, each under its own producer", () => { + const { translator, on, item, spawn, writes } = harness() + on(PARENT, 'turn/started', { turn: { id: 'pt' } }) + spawn(PARENT, 'pt', CHILD, '/root/review') + on(CHILD, 'turn/started', { turn: { id: 'ct' } }) + item(PARENT, 'item/started', 'pt', { + type: 'commandExecution', + id: 'own', + command: 'ls', + status: 'inProgress' + }) + item(CHILD, 'item/started', 'ct', { + type: 'commandExecution', + id: 'kid', + command: 'rg x', + status: 'inProgress' + }) + translator.handle({ + type: 'prompt', + sessionId: SESSION, + threadId: CHILD, + method: CODEX_COMMAND_APPROVAL_METHOD, + params: { turnId: 'ct' }, + codexItemId: 'kid', + promptKey: 'child-approval' + }) + const beforeExit = writes.length + translator.handle({ + type: 'ended', + sessionId: SESSION, + reason: 'provider exited', + cause: 'unexpected-exit', + fence: 1, + acquisitionGeneration: 'generation-1' + }) + + const settled = writes.slice(beforeExit) + const producerOf = (fragment: string) => + settled.find((write) => write.key.includes(fragment))?.linkage + expect(producerOf('kid')).toEqual(CHILD_LINKAGE) + expect(producerOf('child-approval')).toEqual(CHILD_LINKAGE) + expect(producerOf('own')).toEqual({}) + // A turn is the session's unit of work and never carries a producer. + const turnRows = settled.filter((write) => write.body.kind === 'turn') + expect(turnRows.length).toBe(1) + expect(turnRows[0]?.linkage).toEqual({}) + // The child's approval was admitted under the child, too. + expect(writes[writes.findIndex((w) => w.key.includes('child-approval'))]?.linkage).toEqual( + CHILD_LINKAGE + ) + }) + + it("stamps a child's provider frames and goal rows, which are journaled per thread", () => { + const { on, spawn, writes } = harness() + on(PARENT, 'turn/started', { turn: { id: 'pt' } }) + spawn(PARENT, 'pt', CHILD, '/root/review') + on(CHILD, 'turn/started', { turn: { id: 'ct' } }) + on(CHILD, 'error', { + turnId: 'ct', + error: { message: 'child hit a rate limit' }, + willRetry: true + }) + on(CHILD, 'thread/goal/updated', { + turnId: 'ct', + goal: { + threadId: CHILD, + objective: 'Review the diff', + status: 'active', + tokenBudget: null, + tokensUsed: 0, + timeUsedSeconds: 0, + createdAt: 1, + updatedAt: 1 + } + }) + + const frame = writes.find((write) => JSON.stringify(write.body).includes('rate limit')) + const goal = writes.find((write) => JSON.stringify(write.body).includes('Review the diff')) + expect(frame?.linkage).toEqual(CHILD_LINKAGE) + // A goal belongs to the thread, not one run of it: no attempt, ever. + expect(goal?.linkage).toEqual(CHILD_LINKAGE) + }) + + it('stamps nothing while the session thread is still opening', () => { + const { item, linkageOf } = harness(null) + item('thread-x', 'item/completed', 't', { type: 'agentMessage', id: 'm', text: 'too early' }) + + expect(linkageOf('too early')).toEqual({}) + }) +}) diff --git a/src/main/codex/codex-structured-journal-translation-turn-boundaries.ts b/src/main/codex/codex-structured-journal-translation-turn-boundaries.ts index 5c5d6dc8c060..d4881de1897b 100644 --- a/src/main/codex/codex-structured-journal-translation-turn-boundaries.ts +++ b/src/main/codex/codex-structured-journal-translation-turn-boundaries.ts @@ -27,6 +27,7 @@ import { readCodexTurnId, readCodexTurnStatus } from './codex-structured-thread-facts' +import type { CodexRowLinkage } from './codex-subagent-linkage' type TurnBoundaryEvent = { sessionId: string @@ -50,6 +51,7 @@ export class CodexJournalTurnBoundaries { clearPromptTurn?: (threadId: string, turnId: string) => void flushSuppression: () => CodexJournalTranslationAdmission resetActivity: (threadId: string) => void + linkageFor: CodexRowLinkage now?: () => number } ) {} @@ -167,7 +169,8 @@ export class CodexJournalTurnBoundaries { streams: this.deps.items.streams, activeItems: this.deps.items.activeItems, pendingPrompts: this.deps.pendingPrompts, - ...(this.deps.clearPromptTurn ? { clearPromptTurn: this.deps.clearPromptTurn } : {}) + ...(this.deps.clearPromptTurn ? { clearPromptTurn: this.deps.clearPromptTurn } : {}), + linkageFor: this.deps.linkageFor }) if (admission.accepted) { if (turnLifecycle) { @@ -228,7 +231,8 @@ export class CodexJournalTurnBoundaries { streams: this.deps.items.streams, activeItems: this.deps.items.activeItems, pendingPrompts: this.deps.pendingPrompts, - ...(this.deps.clearPromptTurn ? { clearPromptTurn: this.deps.clearPromptTurn } : {}) + ...(this.deps.clearPromptTurn ? { clearPromptTurn: this.deps.clearPromptTurn } : {}), + linkageFor: this.deps.linkageFor }) if (admission.accepted) { if (turnLifecycle) { diff --git a/src/main/codex/codex-structured-journal-translation-writers.ts b/src/main/codex/codex-structured-journal-translation-writers.ts new file mode 100644 index 000000000000..c4fba2c13708 --- /dev/null +++ b/src/main/codex/codex-structured-journal-translation-writers.ts @@ -0,0 +1,45 @@ +// The translator's writers, built once. Split out so the translator reads as +// routing, and so one place shows that every writer is handed the same +// producer resolver: the roster's, which knows each child thread. + +import { CodexJournalCompactions } from './codex-structured-journal-compactions' +import type { CodexJournalTranslatorDeps } from './codex-structured-journal-contracts' +import { CodexJournalGenericFrames } from './codex-structured-journal-generic-frames' +import { CodexJournalGoals } from './codex-structured-journal-goals' +import { CodexJournalItems } from './codex-structured-journal-items' +import { CodexJournalPrompts } from './codex-structured-journal-prompts' +import { createCodexOversizedNotificationSettler } from './codex-structured-journal-translation-frames' +import { CodexJournalActiveTurns } from './codex-structured-journal-translation-turn-state' +import { CodexSubagentRoster } from './codex-subagent-roster' + +export function createCodexJournalTranslatorWriters(deps: CodexJournalTranslatorDeps) { + const activeTurns = new CodexJournalActiveTurns() + const activeTurn = (threadId: string): string | null => activeTurns.current(threadId) + const subagents = new CodexSubagentRoster({ + sink: deps.sink, + primaryThreadId: () => deps.primaryThreadId?.() ?? null, + activeTurn, + ...(deps.subagentExecutions ? { executions: deps.subagentExecutions } : {}) + }) + const { linkageFor } = subagents.linkage + const producerDeps = { ...deps, linkageFor } + const genericFrames = new CodexJournalGenericFrames(producerDeps, activeTurn) + const items = new CodexJournalItems(producerDeps, activeTurn, (threadId, turnId) => + genericFrames.suppress(threadId, turnId) + ) + return { + activeTurns, + subagents, + linkageFor, + genericFrames, + items, + compactions: new CodexJournalCompactions(deps.sink, activeTurn, linkageFor), + goals: new CodexJournalGoals(deps.sink, linkageFor), + prompts: new CodexJournalPrompts( + producerDeps, + (threadId, itemId) => items.detailFor(threadId, itemId), + activeTurn + ), + settleOversizedNotification: createCodexOversizedNotificationSettler(producerDeps, items) + } +} diff --git a/src/main/codex/codex-structured-journal-translation.ts b/src/main/codex/codex-structured-journal-translation.ts index 47ce3f0c46b2..fb90cb73128a 100644 --- a/src/main/codex/codex-structured-journal-translation.ts +++ b/src/main/codex/codex-structured-journal-translation.ts @@ -1,11 +1,5 @@ import { createCodexProviderActivityReader } from '../native-chat/agent-session-wire/provider-frame-activity' import { CODEX_TOKEN_USAGE_METHOD } from './codex-subagent-activity' -import { CodexSubagentRoster } from './codex-subagent-roster' -import { CodexJournalGenericFrames } from './codex-structured-journal-generic-frames' -import { CodexJournalCompactions } from './codex-structured-journal-compactions' -import { CodexJournalGoals } from './codex-structured-journal-goals' -import { CodexJournalItems } from './codex-structured-journal-items' -import { CodexJournalPrompts } from './codex-structured-journal-prompts' import { CODEX_JOURNAL_ADMITTED, type CodexJournalTranslationAdmission, @@ -13,10 +7,9 @@ import { type CodexJournalTranslatorDeps } from './codex-structured-journal-contracts' import { settleCodexJournalSession } from './codex-structured-journal-settlement' -import { createCodexOversizedNotificationSettler } from './codex-structured-journal-translation-frames' import { restoreCodexJournalThread } from './codex-structured-journal-translation-restore' import { CodexJournalTurnBoundaries } from './codex-structured-journal-translation-turn-boundaries' -import { CodexJournalActiveTurns } from './codex-structured-journal-translation-turn-state' +import { createCodexJournalTranslatorWriters } from './codex-structured-journal-translation-writers' import { publishCodexTurnLifecycle } from './codex-structured-journal-translation-turns' import { readCodexProviderVerdict } from './codex-structured-journal-provider-verdicts' import { createCodexThreadItemRouter } from './codex-structured-journal-thread-item-routing' @@ -43,31 +36,17 @@ export { export function createCodexJournalTranslator( deps: CodexJournalTranslatorDeps ): CodexJournalTranslator { - const activeTurns = new CodexJournalActiveTurns() - const compactions = new CodexJournalCompactions(deps.sink, (threadId) => - activeTurns.current(threadId) - ) - const genericFrames = new CodexJournalGenericFrames(deps, (threadId) => - activeTurns.current(threadId) - ) - const goals = new CodexJournalGoals(deps.sink) - const items = new CodexJournalItems( - deps, - (threadId) => activeTurns.current(threadId), - (threadId, turnId) => genericFrames.suppress(threadId, turnId) - ) - const settleOversizedNotification = createCodexOversizedNotificationSettler(deps, items) - const prompts = new CodexJournalPrompts( - deps, - (threadId, itemId) => items.detailFor(threadId, itemId), - (threadId) => activeTurns.current(threadId) - ) - const subagents = new CodexSubagentRoster({ - sink: deps.sink, - primaryThreadId: () => deps.primaryThreadId?.() ?? null, - activeTurn: (threadId) => activeTurns.current(threadId), - ...(deps.subagentExecutions ? { executions: deps.subagentExecutions } : {}) - }) + const { + activeTurns, + subagents, + linkageFor, + genericFrames, + items, + compactions, + goals, + prompts, + settleOversizedNotification + } = createCodexJournalTranslatorWriters(deps) const flushStreams = (): CodexJournalTranslationAdmission => items.streams.flush() ? CODEX_JOURNAL_ADMITTED : { accepted: false, reason: 'backpressure' } let readActivity = createCodexProviderActivityReader() @@ -86,6 +65,7 @@ export function createCodexJournalTranslator( ...(deps.clearPromptTurn ? { clearPromptTurn: deps.clearPromptTurn } : {}), flushSuppression: () => genericFrames.flush(), resetActivity, + linkageFor, ...(deps.now ? { now: deps.now } : {}) }) let primaryThreadStoppedRunning = false @@ -183,7 +163,8 @@ export function createCodexJournalTranslator( turnBoundaries.settled(threadId, turnId, { state: 'interrupted', completedAt: event.observedAt ?? deps.now?.() ?? Date.now() - }) + }), + linkageFor }) if (!admission.accepted) { return admission diff --git a/src/main/codex/codex-subagent-executions.ts b/src/main/codex/codex-subagent-executions.ts index d5ac62bfa740..0f74babe7341 100644 --- a/src/main/codex/codex-subagent-executions.ts +++ b/src/main/codex/codex-subagent-executions.ts @@ -3,6 +3,8 @@ import { MAX_SUBAGENT_FIELD_CHARS } from '../../shared/native-chat-subagent-summ const MAX_CHILDREN = 128 const MAX_SETTLED_TURNS = 256 +/** Turn ordinals remembered per child; a row from an older run reads as its first. */ +const MAX_TURN_ORDINALS_PER_CHILD = 64 export type CodexChildExecution = { turnId: string @@ -14,7 +16,13 @@ export type CodexExecutionChild = { registered: boolean label: string | null parentTurnId: string | null + /** The thread whose stream carried this child's `started` activity. Codex + * emits that item on the spawning agent's own session, so it names the parent. */ + spawnerThreadId: string | null execution: CodexChildExecution | null + /** Which run each observed turn was, in the order the child's turns began. */ + turnOrdinals: Map + turnCount: number } /** Child turn events own execution; activity items only identify the child. */ @@ -25,7 +33,8 @@ export class CodexSubagentExecutions { register( agentThreadId: string, label: string | null, - parentTurnId: string | null | undefined + parentTurnId: string | null | undefined, + spawnerThreadId?: string ): CodexExecutionChild | undefined { const child = this.child(agentThreadId) if (!child) { @@ -34,6 +43,8 @@ export class CodexSubagentExecutions { if (!child.registered || parentTurnId !== undefined) { child.parentTurnId = parentTurnId ?? null } + // A child is spawned once; its announcement is delivered twice, never by another thread. + child.spawnerThreadId ??= spawnerThreadId ?? null child.registered = true // Retain one overflow unit so the journal can append its per-row truncation marker. child.label ??= @@ -58,6 +69,7 @@ export class CodexSubagentExecutions { if (!child) { return null } + this.numberTurn(child, turnId) if ( state === 'working' && child.execution?.turnId === turnId && @@ -87,6 +99,16 @@ export class CodexSubagentExecutions { return this.children.get(agentThreadId)?.label ?? null } + spawnerOf(agentThreadId: string): string | null { + return this.children.get(agentThreadId)?.spawnerThreadId ?? null + } + + /** Which run of the child a turn was: 1 for the turn it was spawned into, then + * one more per follow-up turn. Null when the turn was never observed. */ + turnOrdinal(agentThreadId: string, turnId: string): number | null { + return this.children.get(agentThreadId)?.turnOrdinals.get(turnId) ?? null + } + workingChildren(): CodexExecutionChild[] { return [...this.children.values()].filter( (child) => child.registered && child.execution?.state === 'working' @@ -128,11 +150,28 @@ export class CodexSubagentExecutions { registered: false, label: null, parentTurnId: null, - execution: null + spawnerThreadId: null, + execution: null, + turnOrdinals: new Map(), + turnCount: 0 } this.children.set(agentThreadId, child) return child } + + private numberTurn(child: CodexExecutionChild, turnId: string): void { + if (child.turnOrdinals.has(turnId)) { + return + } + child.turnCount += 1 + child.turnOrdinals.set(turnId, child.turnCount) + if (child.turnOrdinals.size > MAX_TURN_ORDINALS_PER_CHILD) { + const oldest = child.turnOrdinals.keys().next().value + if (oldest !== undefined) { + child.turnOrdinals.delete(oldest) + } + } + } } export function codexChildTurnState(status: unknown): NativeChatSubagentState { diff --git a/src/main/codex/codex-subagent-linkage.ts b/src/main/codex/codex-subagent-linkage.ts new file mode 100644 index 000000000000..59d57a8e4c03 --- /dev/null +++ b/src/main/codex/codex-subagent-linkage.ts @@ -0,0 +1,48 @@ +// Who produced a Codex journal row, decided from the thread that carried it. +// +// Orca opens exactly one thread per app-server, so every other thread on the +// connection is one Codex spawned for a subagent. That settles WHETHER a row is +// a child's without waiting on anything: there is no root arm for another +// thread, announced or not, and its thread id is final from its first frame — +// unlike a tool-call reference, it is never re-minted. Only the parent and the +// run are learned from the roster's executions, and a later revision of the +// row picks them up when they arrive. + +import type { AgentJournalProducerLinkage } from '../../shared/agent-session-journal-types' +import type { CodexSubagentExecutions } from './codex-subagent-executions' + +/** Linkage for a row one thread produced, within one of that thread's turns + * (null outside any). Every Codex write site resolves through this. */ +export type CodexRowLinkage = ( + threadId: string, + turnId: string | null +) => AgentJournalProducerLinkage + +export class CodexSubagentLinkage { + constructor( + private readonly deps: { + primaryThreadId: () => string | null + executions: Pick + } + ) {} + + linkageFor: CodexRowLinkage = (threadId, turnId) => { + const primary = this.deps.primaryThreadId() + // An unknown primary means the session's thread is still opening, and no + // turn has run that could have spawned a child. + if (primary === null || threadId === primary) { + return {} + } + const spawner = this.deps.executions.spawnerOf(threadId) + const attempt = turnId === null ? null : this.deps.executions.turnOrdinal(threadId, turnId) + return { + agentId: threadId, + // Absent means the session's own agent spawned it, so only another child is named. + ...(spawner !== null && spawner !== primary && spawner !== threadId + ? { parentAgentId: spawner } + : {}), + producerKind: 'agent', + ...(attempt !== null && attempt > 1 ? { attempt } : {}) + } + } +} diff --git a/src/main/codex/codex-subagent-roster.ts b/src/main/codex/codex-subagent-roster.ts index a2b58e1a6cec..99b37e9c7723 100644 --- a/src/main/codex/codex-subagent-roster.ts +++ b/src/main/codex/codex-subagent-roster.ts @@ -38,6 +38,7 @@ import { import { readRecord } from './codex-item-field-readers' import { readCodexTurnId } from './codex-structured-thread-facts' import { codexSubagentGroupBody } from './codex-subagent-group-body' +import { CodexSubagentLinkage } from './codex-subagent-linkage' export { codexSubagentGroupBody } from './codex-subagent-group-body' import type { CodexThreadItem } from './codex-structured-item-translation' import { @@ -93,10 +94,16 @@ export class CodexSubagentRoster { private readonly tokensByThread = new Map() private readonly now: () => number private readonly executions: CodexSubagentExecutions + /** Who produced a row, from what this roster learned about each child thread. */ + readonly linkage: CodexSubagentLinkage constructor(private readonly deps: CodexSubagentRosterDeps) { this.now = deps.now ?? (() => Date.now()) this.executions = deps.executions ?? new CodexSubagentExecutions() + this.linkage = new CodexSubagentLinkage({ + primaryThreadId: deps.primaryThreadId, + executions: this.executions + }) } /** Consume a `subAgentActivity` item. Returns null when the item is not one. */ @@ -119,7 +126,9 @@ export class CodexSubagentRoster { const child = this.executions.register( activity.agentThreadId, codexSubagentLabel(activity), - activity.kind === 'started' || activity.kind === 'interacted' ? input.turnId : undefined + activity.kind === 'started' || activity.kind === 'interacted' ? input.turnId : undefined, + // Only `started` names the spawner: other kinds ride whichever agent acted. + activity.kind === 'started' ? input.threadId : undefined ) if (!child?.execution) { return ADMITTED @@ -346,6 +355,8 @@ export class CodexSubagentRoster { return ADMITTED } group.lastSerialized = serialized + // Deliberately unstamped: a child's frame can trigger this write, but the + // row is the PARENT's roster of its children. // The append coalesces per group so a burst collapses to the latest roster. // The publish must NOT reuse that key: the queue coalesces by key alone, // with no op-kind check, so a publish carrying it would splice out the diff --git a/src/main/native-chat/agent-session-journal/journal-lifecycle-batch-partition.ts b/src/main/native-chat/agent-session-journal/journal-lifecycle-batch-partition.ts index f109451edb97..11fb69863f26 100644 --- a/src/main/native-chat/agent-session-journal/journal-lifecycle-batch-partition.ts +++ b/src/main/native-chat/agent-session-journal/journal-lifecycle-batch-partition.ts @@ -1,6 +1,9 @@ import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key' import { journalRowSchemaVersion } from '../../../shared/agent-session-journal-types' -import type { JournalLifecycleMutationInput } from './journal-row-builders' +import { + journalLifecycleMutationRow, + type JournalLifecycleMutationInput +} from './journal-row-builders' import type { JournalLifecycleBatchRow, JournalLifecycleMutation } from './journal-row-schema' import { MAX_JOURNAL_LIFECYCLE_BATCH_BYTES, @@ -69,13 +72,9 @@ function serializedLifecycleBatchFits( } function toLifecycleMutationRow(mutation: JournalLifecycleMutationInput): JournalLifecycleMutation { - const itemId = agentJournalItemKey(mutation.identity) - return mutation.kind === 'item' - ? { - kind: 'item', - itemId, - revision: Number.MAX_SAFE_INTEGER, - body: mutation.body - } - : { kind: 'tombstone', itemId, revision: Number.MAX_SAFE_INTEGER } + return journalLifecycleMutationRow( + mutation, + agentJournalItemKey(mutation.identity), + Number.MAX_SAFE_INTEGER + ) } diff --git a/src/main/native-chat/agent-session-journal/journal-producer-inheritance.test.ts b/src/main/native-chat/agent-session-journal/journal-producer-inheritance.test.ts new file mode 100644 index 000000000000..07ae011b7908 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-producer-inheritance.test.ts @@ -0,0 +1,204 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key' +import type { + AgentJournalItemIdentity, + AgentJournalMessageItem, + AgentJournalRenderItem +} from '../../../shared/agent-session-journal-types' +import { applyJournalRow, createJournalReducerState } from './journal-reducer' +import { + buildJournalItemRow, + buildJournalTombstoneRow, + journalLifecycleBatchRowBuilder +} from './journal-row-builders' +import { createTrackedJournalOpener } from './journal-store-test-open' + +// A row's producer is fixed by the write that created it. A revision naming no +// producer — a settlement, a prompt answer, a reopen sweep — keeps it; a +// revision naming one replaces the whole bundle, which is how a provisional +// stamp is corrected. + +const child: AgentJournalItemIdentity = { + provider: 'codex', + threadId: 'child', + turnId: 't', + ordinal: 1 +} +const own: AgentJournalItemIdentity = { + provider: 'codex', + threadId: 'root', + turnId: 't', + ordinal: 1 +} +const linkage = { + agentId: 'child', + parentAgentId: 'spawner', + providerParentRef: 'call-1', + producerKind: 'agent' as const, + attempt: 2 +} + +function text(value: string): AgentJournalMessageItem { + return { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: value }] } +} + +function producerOf(item: AgentJournalRenderItem | undefined) { + if (!item) { + return undefined + } + const { agentId, parentAgentId, providerParentRef, producerKind, attempt } = item + return { agentId, parentAgentId, providerParentRef, producerKind, attempt } +} + +const NO_PRODUCER = { + agentId: undefined, + parentAgentId: undefined, + providerParentRef: undefined, + producerKind: undefined, + attempt: undefined +} + +function seeded() { + const state = createJournalReducerState('session-1', 'epoch-1') + let seq = 0 + const write = ( + identity: AgentJournalItemIdentity, + body: AgentJournalMessageItem, + stamp?: Parameters[0]['linkage'] + ) => { + seq += 1 + applyJournalRow( + state, + buildJournalItemRow({ + state, + identity, + body, + seq, + fence: 1, + ts: 1_000 + seq, + ...(stamp ? { linkage: stamp } : {}) + }) + ) + } + const settle = (identities: AgentJournalItemIdentity[]) => { + seq += 1 + applyJournalRow( + state, + journalLifecycleBatchRowBuilder( + () => state, + `settle-${seq}`, + identities.map((identity) => ({ kind: 'item' as const, identity, body: text('settled') })), + { fence: 1 } + )(seq, 1_000 + seq) + ) + } + const remove = (identity: AgentJournalItemIdentity) => { + seq += 1 + const itemId = agentJournalItemKey(identity) + applyJournalRow( + state, + buildJournalTombstoneRow({ state, itemId, seq, fence: 1, ts: 1_000 + seq }) + ) + } + const item = (identity: AgentJournalItemIdentity) => + state.items.get(agentJournalItemKey(identity)) + return { state, write, settle, remove, item } +} + +describe('producer inheritance in the reducer', () => { + it('keeps the whole bundle when a plain revision names no producer', () => { + const { write, item } = seeded() + write(child, text('looking'), linkage) + write(child, text('looked')) + + expect(item(child)).toMatchObject({ revision: 2, body: text('looked'), sequence: 1 }) + expect(producerOf(item(child))).toEqual(linkage) + }) + + it("keeps each row's own producer when one batch settles a child's row and the session's", () => { + const { write, settle, item } = seeded() + write(child, text('child working'), linkage) + write(own, text('own working')) + settle([child, own]) + + expect(item(child)).toMatchObject({ revision: 2, body: text('settled') }) + expect(producerOf(item(child))).toEqual(linkage) + expect(item(own)).toMatchObject({ revision: 2, body: text('settled') }) + expect(producerOf(item(own))).toEqual(NO_PRODUCER) + }) + + it('replaces the bundle wholesale when a revision names any producer', () => { + const { write, item } = seeded() + write(child, text('looking'), linkage) + write(child, text('looking'), { agentId: 'task-1', producerKind: 'agent' }) + + expect(producerOf(item(child))).toEqual({ + ...NO_PRODUCER, + agentId: 'task-1', + producerKind: 'agent' + }) + }) + + it('gives a row re-created after its removal nothing from the removed one', () => { + const { write, remove, item } = seeded() + write(child, text('first life'), linkage) + remove(child) + write(child, text('second life')) + + expect(item(child)?.body).toEqual(text('second life')) + expect(producerOf(item(child))).toEqual(NO_PRODUCER) + }) +}) + +describe('producer inheritance across a reopen', () => { + let root: string + const journals = createTrackedJournalOpener() + const open = () => + journals.open({ + identity: { + sessionId: 'session-1', + workspaceId: 'workspace-1', + hostId: 'local', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: 'root' } + }, + journalDir: root, + now: () => 1_000 + }) + + beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-producer-inheritance-')) + }) + + afterEach(async () => { + await journals.closeAll() + await rm(root, { recursive: true, force: true }) + }) + + it('replays an inherited producer from disk exactly as it was folded live', async () => { + const journal = await open() + await journal.appendItem(child, text('working'), { fence: 1, ...linkage }) + await journal.appendItem(own, text('working'), { fence: 1 }) + await journal.appendItem(child, text('still working'), { fence: 1 }) + await journal.appendLifecycleBatch({ + settlementId: 'settle-1', + fence: 1, + mutations: [ + { kind: 'item', identity: child, body: text('settled') }, + { kind: 'item', identity: own, body: text('settled') } + ] + }) + const live = journal.snapshot().items + expect(live.map((item) => [item.revision, producerOf(item)])).toEqual([ + [3, linkage], + [2, NO_PRODUCER] + ]) + + await journal.close() + const reopened = await open() + expect(reopened.snapshot().items).toEqual(live) + }) +}) diff --git a/src/main/native-chat/agent-session-journal/journal-reducer.test.ts b/src/main/native-chat/agent-session-journal/journal-reducer.test.ts index 1a0b00e5d85b..825c5aaf3473 100644 --- a/src/main/native-chat/agent-session-journal/journal-reducer.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-reducer.test.ts @@ -17,7 +17,11 @@ import { renderJournalState, type JournalReducerState } from './journal-reducer' -import { buildJournalItemRow, buildJournalTombstoneRow } from './journal-row-builders' +import { + buildJournalItemRow, + buildJournalTombstoneRow, + journalLifecycleBatchRowBuilder +} from './journal-row-builders' import type { JournalRow } from './journal-row-schema' const EPOCH = 'epoch-1' @@ -702,6 +706,34 @@ describe('producer linkage round-trips through the reducer', () => { expect(renderJournalState(state).items[0]).toMatchObject(linkage) }) + it('reads each mutation of a mixed batch as its own producer', () => { + // A batch can CREATE rows several agents produced — a settlement landing + // before any checkpoint did. The mutation that names a producer is that + // producer's; the one naming none is the session's own, beside it. + const state = createJournalReducerState('session-1', EPOCH) + applyJournalRow( + state, + journalLifecycleBatchRowBuilder( + () => state, + 'settle-mixed', + [ + { kind: 'item', identity, body: text('child'), linkage }, + { + kind: 'item', + identity: { provider: 'claude', sessionId: 'claude-session', uuid: 'own-1' }, + body: text('own') + } + ], + { fence: 1 } + )(1, 1_001) + ) + + const [child, own] = renderJournalState(state).items + expect(child).toMatchObject({ body: text('child'), ...linkage }) + expect(own?.body).toEqual(text('own')) + expect(own && 'agentId' in own).toBe(false) + }) + it('lets a correction win over the provisional row, without moving the bubble', () => { // Write-through then correct: the row is written under the spawn call's own // id, then re-appended under the canonical one. Revision is assigned inside diff --git a/src/main/native-chat/agent-session-journal/journal-reducer.ts b/src/main/native-chat/agent-session-journal/journal-reducer.ts index 03352d8e9a53..273502c8cf2b 100644 --- a/src/main/native-chat/agent-session-journal/journal-reducer.ts +++ b/src/main/native-chat/agent-session-journal/journal-reducer.ts @@ -5,7 +5,8 @@ // Rules: highest revision wins, a tombstone removes, a late lower revision is // dropped rather than resurrecting stale content, and ordering is by the // sequence of the row that CREATED an item (a later revision updates the body, -// it does not move the bubble). +// it does not move the bubble). Producer linkage is likewise the creating +// write's: a revision naming no producer keeps it, one naming any replaces it. import type { AgentJournalAcceptanceReceipt, @@ -13,11 +14,15 @@ import type { AgentJournalSnapshot, AgentJournalSubmission } from '../../../shared/agent-session-journal-types' -import { journalRenderItem } from './journal-render-item' +import { journalBatchMutationProducer, journalRenderItem } from './journal-render-item' import { agentJournalSubmissionKey, parseAgentJournalItemKey } from '../../../shared/agent-session-journal-item-key' +import { + agentJournalLinkageFields, + namesAgentJournalProducer +} from '../../../shared/agent-session-journal-producer' import { structuredAgentSessionPayloadFingerprint } from '../../../shared/structured-agent-session-mutation' import { journalItemRevisionIsStale } from './journal-item-revision' import type { JournalRow } from './journal-row-schema' @@ -96,7 +101,13 @@ export function applyJournalRow(state: JournalReducerState, row: JournalRow): vo state, itemId, mutation.revision, - journalRenderItem(itemId, mutation.revision, mutation.body, row) + journalRenderItem( + itemId, + mutation.revision, + mutation.body, + row, + journalBatchMutationProducer(row, mutation) + ) ) } else { removeItem(state, resolveItemId(state, mutation.itemId), mutation.revision) @@ -207,6 +218,9 @@ function upsertItem( parseAgentJournalItemKey(itemId)?.provider === 'orca' state.items.set(itemId, { ...next, + // Settlements, prompt answers and reopen sweeps revise rows any agent wrote + // without naming one; each would otherwise hand a subagent's row to the session. + ...(namesAgentJournalProducer(next) ? {} : agentJournalLinkageFields(existing)), // Provider history may normalize text or omit local attachments from the original send. body: submitted ? existing.body : next.body, sequence: existing.sequence, diff --git a/src/main/native-chat/agent-session-journal/journal-render-item.ts b/src/main/native-chat/agent-session-journal/journal-render-item.ts index 3324ffedbd11..72bef3e7828d 100644 --- a/src/main/native-chat/agent-session-journal/journal-render-item.ts +++ b/src/main/native-chat/agent-session-journal/journal-render-item.ts @@ -1,19 +1,25 @@ import type { AgentJournalItemBody, + AgentJournalProducerLinkage, AgentJournalRenderItem } from '../../../shared/agent-session-journal-types' -import { agentJournalLinkageFields } from '../../../shared/agent-session-journal-producer' +import { + agentJournalLinkageFields, + namesAgentJournalProducer +} from '../../../shared/agent-session-journal-producer' import type { JournalRow } from './journal-row-schema' /** One render item, built the same way by every upsert path in the reducer. * The row-level markers are copied here rather than at each call site: they * were three separate spreads that had to stay in sync, and absence is the - * claim in each case — appended live, and produced by the session's own agent. */ + * claim in each case — appended live, and (on a row's first write) produced by + * the session's own agent. */ export function journalRenderItem( itemId: string, revision: number, body: AgentJournalItemBody, - row: JournalRow + row: JournalRow, + producer: AgentJournalProducerLinkage = row ): AgentJournalRenderItem { return { itemId, @@ -22,6 +28,16 @@ export function journalRenderItem( sequence: row.seq, observedAt: row.ts, ...(row.recovered ? { recovered: row.recovered } : {}), - ...agentJournalLinkageFields(row) + ...agentJournalLinkageFields(producer) } } + +/** Who one mutation of a batch names as its producer: the mutation itself when + * it names one, else the batch row, which only a host stamping whole batches + * wrote. Naming none leaves the reducer to keep the row's existing producer. */ +export function journalBatchMutationProducer( + row: AgentJournalProducerLinkage, + mutation: AgentJournalProducerLinkage +): AgentJournalProducerLinkage { + return namesAgentJournalProducer(mutation) ? mutation : row +} diff --git a/src/main/native-chat/agent-session-journal/journal-row-builders.ts b/src/main/native-chat/agent-session-journal/journal-row-builders.ts index ba06f76792f7..e17ccf0cdf87 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-builders.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-builders.ts @@ -7,7 +7,10 @@ import type { AgentSessionProviderHandle } from '../../../shared/agent-session-journal-types' import { journalRowSchemaVersion } from '../../../shared/agent-session-journal-types' -import { agentJournalLinkageFields } from '../../../shared/agent-session-journal-producer' +import { + agentJournalLinkageFields, + namesAgentJournalProducer +} from '../../../shared/agent-session-journal-producer' import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key' import type { JournalReducerState } from './journal-reducer' import type { @@ -100,18 +103,57 @@ function boundedDispatchReason(input: ResolveDispatchInput): string | null { } export type JournalLifecycleMutationInput = - | { kind: 'item'; identity: AgentJournalItemIdentity; body: AgentJournalItemBody } + | { + kind: 'item' + identity: AgentJournalItemIdentity + body: AgentJournalItemBody + /** Who wrote the row. Absent ⇒ the session's own agent on a first write, + * and the row's existing producer on a revision. */ + linkage?: AgentJournalProducerLinkage + } | { kind: 'tombstone'; identity: AgentJournalItemIdentity } +/** An item mutation from a writer that knows who produced the row. Needed + * because a batch can CREATE a row — a Codex child's prompt, or its item + * settled before any checkpoint landed — and one batch can mix producers. + * The session's own rows carry no key at all: absence is the claim. */ +export function journalLifecycleItemMutation( + producer: AgentJournalProducerLinkage, + identity: AgentJournalItemIdentity, + body: AgentJournalItemBody +): JournalLifecycleMutationInput { + return namesAgentJournalProducer(producer) + ? { kind: 'item', identity, body, linkage: agentJournalLinkageFields(producer) } + : { kind: 'item', identity, body } +} + +/** The persisted form of one mutation, shared with the partitioner's size probe + * so a chunk is measured with the linkage it will actually carry. */ +export function journalLifecycleMutationRow( + mutation: JournalLifecycleMutationInput, + itemId: string, + revision: number +): JournalLifecycleMutation { + return mutation.kind === 'item' + ? { + kind: 'item', + itemId, + revision, + body: mutation.body, + ...agentJournalLinkageFields(mutation.linkage) + } + : { kind: 'tombstone', itemId, revision } +} + export function journalLifecycleBatchRowBuilder( state: () => JournalReducerState, settlementId: string, mutations: readonly JournalLifecycleMutationInput[], - /** No producer linkage: one batch row covers N mutations, so a row-level - * producer would stamp whoever opened the batch onto every one of them. The - * reducer still READS linkage off a batch row, because a row may come from a - * host that writes one; a mixed-producer batch would have to stamp per - * mutation, which nothing needs yet. */ + /** No ROW-level producer: one batch row covers N mutations, so a row-level + * producer would stamp whoever opened the batch onto every one of them. + * An item mutation names its own, or none to keep the row's existing one. + * The reducer still reads row-level linkage as the fallback for a mutation + * that names none, because a row may come from a host that wrote one. */ options: { fence: number; recovered?: true } ): RowBuilder { return (seq, ts) => { @@ -130,9 +172,7 @@ export function journalLifecycleBatchRowBuilder( current.tombstones.get(resolved) ?? 0 )) + 1 revisions.set(resolved, revision) - return mutation.kind === 'item' - ? { kind: 'item', itemId, revision, body: mutation.body } - : { kind: 'tombstone', itemId, revision } + return journalLifecycleMutationRow(mutation, itemId, revision) }) const row: JournalLifecycleBatchRow = { kind: 'lifecycle-batch', diff --git a/src/main/native-chat/agent-session-journal/journal-row-schema.test.ts b/src/main/native-chat/agent-session-journal/journal-row-schema.test.ts index 88a81e070f2a..870b52593576 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-schema.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-schema.test.ts @@ -6,7 +6,11 @@ import { type JournalRow } from './journal-row-schema' import { createJournalReducerState } from './journal-reducer' -import { buildJournalItemRow } from './journal-row-builders' +import { + buildJournalItemRow, + journalLifecycleBatchRowBuilder, + type JournalLifecycleMutationInput +} from './journal-row-builders' const BASE = { v: 1, epoch: 'epoch-1', seq: 1, fence: 1, ts: 1 } @@ -342,6 +346,46 @@ describe('producer linkage on the persisted row', () => { }) }) + it('round-trips a batch mutation that names its own producer, with no version bump', () => { + const state = createJournalReducerState('session-1', 'epoch-1') + const own = { kind: 'item' as const, identity: { ...identity, uuid: 'u-own' }, body } + const build = (child: JournalLifecycleMutationInput) => + journalLifecycleBatchRowBuilder(() => state, 'settle-1', [child, own], { fence: 1 })(1, 1) + const row = build({ kind: 'item', identity, body, linkage }) + + const parsed = parseJournalRow(JSON.stringify(row)) + const mutations = parsed.ok && parsed.row.kind === 'lifecycle-batch' ? parsed.row.mutations : [] + expect(mutations[0]).toMatchObject(linkage) + expect(mutations[1] && 'agentId' in mutations[1]).toBe(false) + // The same batch without the stamp writes the same version: an older host + // ignores the unknown keys rather than latching the journal read-only. + expect(row.v).toBe(build({ kind: 'item', identity, body }).v) + }) + + it('keeps a batch mutation but drops its unusable producer id', () => { + // Same policy as the row base: sanitize, never reject — a rejected batch + // would take every mutation in it out of the timeline. + const parsed = parseJournalRow( + JSON.stringify({ + v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION, + epoch: 'epoch-1', + seq: 7, + fence: 1, + ts: 1_700_000_000_000, + kind: 'lifecycle-batch', + settlementId: 'settle-2', + mutations: [ + { kind: 'item', itemId: 'i-1', revision: 1, body, agentId: '' }, + { kind: 'item', itemId: 'i-2', revision: 1, body, agentId: 'thread-child' } + ] + }) + ) + const mutations = parsed.ok && parsed.row.kind === 'lifecycle-batch' ? parsed.row.mutations : [] + expect(mutations).toHaveLength(2) + expect(mutations[0] && 'agentId' in mutations[0]).toBe(false) + expect(mutations[1]).toMatchObject({ agentId: 'thread-child' }) + }) + it("omits every key on a row the session's own agent produced", () => { const row = roundTrip(false) expect(row && 'agentId' in row).toBe(false) diff --git a/src/main/native-chat/agent-session-journal/journal-row-schema.ts b/src/main/native-chat/agent-session-journal/journal-row-schema.ts index 59b62c3b066e..99cb47971e95 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-schema.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-schema.ts @@ -88,13 +88,17 @@ export type JournalDispatchRow = JournalRowBase & { reason: string | null } +/** An item mutation may name its own producer, because one batch can CREATE + * rows several agents produced. Naming none keeps the row's existing producer. + * Inline like the row base, and for the same reason no `v` bump: an older host + * ignores the unknown keys and reads the mutation as root, as it always did. */ export type JournalLifecycleMutation = - | { + | (AgentJournalProducerLinkage & { kind: 'item' itemId: string revision: number body: AgentJournalItemBody - } + }) | { kind: 'tombstone'; itemId: string; revision: number } /** One durable append whose nested mutations share the outer ordering facts. */ @@ -159,6 +163,13 @@ export function parseJournalRow(line: string): JournalRowParse { } const upcast = upcastRow(record, version) dropUnusableProducerLinkage(upcast) + if (upcast.kind === 'lifecycle-batch' && Array.isArray(upcast.mutations)) { + for (const mutation of upcast.mutations) { + if (isPlainObject(mutation)) { + dropUnusableProducerLinkage(mutation) + } + } + } dropUnusableContextUsage(upcast) return isJournalRow(upcast) ? { ok: true, row: upcast } : { ok: false, unreadable: false } } diff --git a/src/main/native-chat/agent-session-wire/agent-session-empty-delta-retention.test.ts b/src/main/native-chat/agent-session-wire/agent-session-empty-delta-retention.test.ts index 4e2161b19a80..3907fa7a6458 100644 --- a/src/main/native-chat/agent-session-wire/agent-session-empty-delta-retention.test.ts +++ b/src/main/native-chat/agent-session-wire/agent-session-empty-delta-retention.test.ts @@ -7,7 +7,9 @@ describe('empty streamed deltas', () => { const prefix = 'empty-delta-retention-prefix' const streams = createCodexStructuredItemStreams({ sink: { appendItem() {}, appendTombstone() {}, publish() {} }, + turnIdFor: () => 'turn', identityFor: () => ({ provider: 'codex', threadId: 'thread', turnId: 'turn', ordinal: 0 }), + linkageFor: () => ({}), schedule: () => () => {} }) const append = (delta: string) => diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.test.ts index af51a9967d6a..64b7b9d03b5e 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.test.ts @@ -248,6 +248,85 @@ describe('dead structured-session generation settlement', () => { }) ]) }) + + it("keeps a subagent's settled rows the subagent's, in one batch and after a reopen", async () => { + // One batch settles rows several agents wrote and names none of them. Each + // row keeps the producer its first write named, including after a replay. + const child = { agentId: 'thread-child', producerKind: 'agent' as const } + const childCall = { + provider: 'codex' as const, + threadId: 'thread-child', + turnId: 'c', + ordinal: 1 + } + const childAsk = { + provider: 'codex' as const, + threadId: 'thread-child', + turnId: 'c', + ordinal: 2 + } + await seedUnfinishedWork() + await journal.appendItem( + childCall, + { kind: 'tool-call', name: 'shell', input: { command: 'ls' }, state: 'running' }, + { fence: 7, ...child } + ) + await journal.appendItem( + childAsk, + { + kind: 'approval', + title: 'Run ls?', + detail: null, + options: [{ id: 'yes', label: 'Allow' }], + resolution: { state: 'pending', selectedOptionId: null, resolvedBy: null, resolvedAt: null } + }, + { fence: 7, ...child } + ) + + await settleStructuredAgentSessionDeadGeneration({ + journal, + sessionId: SESSION, + fence: 8, + settlementId: `restart-eviction:${SESSION}:8`, + pendingSubmissionReason: 'provider_exited_before_acknowledgement', + verdict: { state: 'unverifiable' }, + showUnexpectedExitOutcome: false + }) + const settledProducers = (): [string, number, string | undefined][] => + journal + .snapshot() + .items.map((item): [string, number, string | undefined] => [ + item.body.kind, + item.revision, + item.agentId + ]) + + const settled = settledProducers() + // Every seeded row was revised by the batch, so these are revision-2 producers. + expect(settled).toEqual([ + ['message', 0, undefined], + ['tool-call', 2, undefined], + ['approval', 2, undefined], + ['question', 2, undefined], + ['turn', 2, undefined], + ['tool-call', 2, 'thread-child'], + ['approval', 2, 'thread-child'] + ]) + + await journal.close() + journal = await openAgentSessionJournal({ + identity: { + sessionId: SESSION, + workspaceId: 'workspace-1', + hostId: 'local', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: THREAD } + }, + journalDir: root, + now: () => 1_000 + }) + expect(settledProducers()).toEqual(settled) + }) }) describe('whether a dead generation interrupted anything', () => { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.test.ts index d9f567a8a75c..6c3747fb824a 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.test.ts @@ -442,8 +442,14 @@ describe('producer linkage reaches the journal through every append path', () => } }) - it('forwards it on the resolved-append paths, which lost it once before', async () => { - for (const append of ['tryAppendResolvedItem', 'tryAppendResolvedItemAndPublish'] as const) { + it('forwards it on the resolved-append and lifecycle-transition paths', async () => { + // The resolved paths lost it once before; a transition is how a Codex + // child's goal row is written, so dropping it there files the goal as root. + for (const append of [ + 'tryAppendResolvedItem', + 'tryAppendResolvedItemAndPublish', + 'tryAppendLifecycleTransition' + ] as const) { const log: Recorded[] = [] const deferred = createDeferredStructuredAgentSessionEventSink() deferred.bind(target(5, log)) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts index 57cd52a9b70d..923342b7fa6d 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts @@ -116,7 +116,8 @@ export type StructuredAgentSessionEventSink = { tryAppendLifecycleTransition?( identitySizeBound: AgentJournalItemIdentity, body: AgentJournalItemBody, - resolveIdentity: StructuredAgentSessionIdentityResolver + resolveIdentity: StructuredAgentSessionIdentityResolver, + options?: StructuredAgentSessionAppendOptions ): StructuredAgentSessionSinkAdmission /** Current durable epoch, when this deferred sink is bound to its journal. */ journalEpoch?(): string | null @@ -208,7 +209,7 @@ export function createDeferredStructuredAgentSessionEventSink( bound.journal.appendLifecycleBatch({ settlementId, mutations, - // Linkage is deliberately not forwarded: see the batch row builder. + // No row-level linkage: each mutation names its own (see the batch row builder). fence: bound.fence }) }, @@ -259,27 +260,6 @@ export function createDeferredStructuredAgentSessionEventSink( options ), ...resolvedAppend, - tryAppendLifecycleTransition: (identitySizeBound, body, resolveIdentity) => { - const bytes = estimateStructuredAgentSessionItemBytes(identitySizeBound, body) - return queue.submit( - { - bytes, - lifecycle: true, - run: async (bound) => { - const identity = resolveIdentity(bound.journal) - if (identity === null) { - return - } - if (estimateStructuredAgentSessionItemBytes(identity, body) > bytes) { - throw new Error('structured agent-session item identity exceeded its reserved size') - } - await bound.journal.appendItem(identity, body, { fence: bound.fence }) - bound.publish() - } - }, - { lifecycle: true } - ) - }, journalEpoch: queue.journalEpoch, appendLifecycleBatch: (settlementId, mutations, options = {}) => { const admission = appendLifecycleBatch(settlementId, mutations, options) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts index a19735657d60..cbf7b95590f8 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts @@ -358,6 +358,46 @@ describe('respondToPrompt', () => { expect(answerPrompt).toHaveBeenCalledTimes(1) }) + it("keeps a subagent's approval the subagent's once the user answers it", async () => { + // The answer revises the row without naming a producer, so it keeps the asker's. + await attach() + const child = { agentId: 'thread-child', producerKind: 'agent' as const } + const identity = { + provider: 'codex' as const, + threadId: 'thread-child', + turnId: 'c', + ordinal: 1 + } + acquire.mock.calls.at(-1)?.[0].events?.appendItem( + identity, + { + kind: 'approval', + title: 'Run ls?', + detail: null, + options: [{ id: 'allow', label: 'Allow' }], + resolution: { state: 'pending', selectedOptionId: null, resolvedBy: null, resolvedAt: null } + }, + child + ) + await host.flushStreamedEvents(SESSION) + const itemId = agentJournalItemKey(identity) + const fields = { itemId, expectedRevision: 1, optionId: 'allow' } + + await host.respondToPrompt(CALLER, { + envelope: envelope('agentSession.respondTo:approval', fields), + kind: 'approval', + ...fields + }) + + const page = host.history({ sessionId: SESSION, direction: 'tail' }) + const answered = page.ok ? page.page.items.find((item) => item.itemId === itemId) : null + expect(answered).toMatchObject({ + revision: 2, + body: { resolution: { state: 'resolved' } }, + ...child + }) + }) + it('refuses a second answer to one prompt and says which answer won', async () => { await attach() const prompt = await seedApproval() diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-journal-append-options.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-journal-append-options.ts index fdab5ebb0a63..b35690c4e7d3 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-journal-append-options.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-journal-append-options.ts @@ -1,9 +1,8 @@ -// The journal options one admitted sink append forwards. +// The journal options one admitted single-item sink append forwards. // -// Its own module because four append paths need it and the sink's own file -// already depends on two of them. Every path calls this, so a row-level field -// added to the sink's options reaches the durable row through all four rather -// than through whichever spread the next change remembers to edit. +// Every such path calls this, so a row-level field added to the sink's options +// reaches the durable row through one edit rather than through whichever spread +// the next change remembers. Lifecycle batches carry producers per mutation. import { agentJournalLinkageFields } from '../../../shared/agent-session-journal-producer' import type { JournalItemAppendOptions } from '../agent-session-journal/journal-store-contracts' diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-resolved-append.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-resolved-append.ts index b0acba120b7e..de1190aefc1a 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-resolved-append.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-resolved-append.ts @@ -24,6 +24,7 @@ export function createStructuredAgentSessionResolvedAppend( | 'tryAppendResolvedItemAndPublish' | 'tryReviseResolvedItem' | 'tryReviseResolvedItemAndPublish' + | 'tryAppendLifecycleTransition' > > { const submit = ( @@ -73,6 +74,31 @@ export function createStructuredAgentSessionResolvedAppend( tryReviseResolvedItem: (reservedBytes, resolve, options = {}) => submit(reservedBytes, resolve, options, false), tryReviseResolvedItemAndPublish: (reservedBytes, resolve, options = {}) => - submit(reservedBytes + 1, resolve, options, true) + submit(reservedBytes + 1, resolve, options, true), + tryAppendLifecycleTransition: (identitySizeBound, body, resolveIdentity, options = {}) => { + const bytes = estimateStructuredAgentSessionItemBytes(identitySizeBound, body) + return queue.submit( + { + bytes, + lifecycle: true, + run: async (bound) => { + const identity = resolveIdentity(bound.journal) + if (identity === null) { + return + } + if (estimateStructuredAgentSessionItemBytes(identity, body) > bytes) { + throw new Error('structured agent-session item identity exceeded its reserved size') + } + await bound.journal.appendItem( + identity, + body, + structuredAgentSessionJournalAppendOptions(bound.fence, options) + ) + bound.publish() + } + }, + { lifecycle: true } + ) + } } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-verdict.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-verdict.test.ts index 85072ce2a125..3264a55fff25 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-verdict.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-verdict.test.ts @@ -1,7 +1,11 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' import { describe, expect, it, vi } from 'vitest' import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key' import type { AgentJournalRenderItem } from '../../../shared/agent-session-journal-types' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import { createTrackedJournalOpener } from '../agent-session-journal/journal-store-test-open' import { runningTurnLifecycleRevisions, settleStaleSessionStateOnAcquire, @@ -257,6 +261,52 @@ describe('stale session state on a cold acquire', () => { }) }) + it("cancels a subagent's lost prompt as the subagent's, and the session's own as its own", async () => { + // The sweep names no producer, so each cancelled row keeps the one it had. + const root = await mkdtemp(join(tmpdir(), 'orca-stale-session-')) + const journals = createTrackedJournalOpener() + try { + const journal = await journals.open({ + identity: { + sessionId: 'session-1', + workspaceId: 'workspace-1', + hostId: 'local', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: THREAD } + }, + journalDir: root, + now: () => 1_000 + }) + const child = { agentId: 'thread-child', producerKind: 'agent' as const } + const { body } = promptItem('pending', 1) + const prompt = (threadId: string) => ({ + provider: 'codex' as const, + threadId, + turnId: 'turn-1', + ordinal: 1 + }) + await journal.appendItem(prompt('thread-child'), body, { fence: 1, ...child }) + await journal.appendItem(prompt(THREAD), body, { fence: 1 }) + + await settleStaleSessionStateOnAcquire({ + journal, + sessionId: 'session-1', + fence: 2, + acquisitionGeneration: 'generation-2' + }) + + expect( + journal.snapshot().items.map((item) => [item.body.kind, item.revision, item.agentId]) + ).toEqual([ + ['approval', 2, 'thread-child'], + ['approval', 2, undefined] + ]) + } finally { + await journals.closeAll() + await rm(root, { recursive: true, force: true }) + } + }) + it('writes nothing when no turn is running and keys on the journal position without a generation', async () => { const idle = journalWith([ lifecycleItem('turn-1', 'completed', 1, { startedAt: 10, completedAt: 20 }) diff --git a/src/shared/agent-session-journal-producer.ts b/src/shared/agent-session-journal-producer.ts index fe3132946775..ff51c6eaf576 100644 --- a/src/shared/agent-session-journal-producer.ts +++ b/src/shared/agent-session-journal-producer.ts @@ -5,6 +5,8 @@ // agent doing right now" reader has to say which producer it means. Readers ask // "is this NOT mine", never "is this mine": the session's own agent stamps // nothing, so root-ness is the absence of an id rather than a value to match. +// That absence is decided by a row's FIRST write: a later revision naming no +// producer keeps the row's existing one (see the journal reducer). import type { AgentJournalProducerLinkage, @@ -26,6 +28,18 @@ export function isRootAgentJournalItem( return item?.agentId == null } +/** Whether a write names its producer at all. One that does not revises a row + * without re-attributing it, so this is presence of any member, not of `agentId`. */ +export function namesAgentJournalProducer(linkage: AgentJournalProducerLinkage): boolean { + return ( + linkage.agentId !== undefined || + linkage.parentAgentId !== undefined || + linkage.providerParentRef !== undefined || + linkage.producerKind !== undefined || + linkage.attempt !== undefined + ) +} + /** Linkage as row fields, with absent members omitted rather than set to * `undefined`. Every carrier spreads this, so a new field reaches the row * through one edit instead of one per hop. */ diff --git a/src/shared/structured-agent-session-live-turn.ts b/src/shared/structured-agent-session-live-turn.ts index e31d4af2c8fe..2f5a07476671 100644 --- a/src/shared/structured-agent-session-live-turn.ts +++ b/src/shared/structured-agent-session-live-turn.ts @@ -10,11 +10,12 @@ // Each scan reads the turn record BEFORE it checks the producer, which is only // safe because a turn row can never carry linkage: a turn is the SESSION'S unit // of work, and no producer of a turn-bearing body stamps one. Both lanes were -// checked — Claude's turn rows are built with no linkage at all, Codex has no -// linkage concept, the compact row passes only a fence, and the stale-turn -// sweep goes through the lifecycle-batch path, which cannot carry linkage by -// type. So a child-linked row can never be what terminates one of these scans. -// Re-check that before giving any of those sites a producer. +// checked — Claude's turn rows are built with no linkage at all, Codex writes +// turn rows only for its primary thread (the one thread it never stamps), the +// compact row passes only a fence, and the stale-turn and dead-generation +// sweeps name no producer, so their turn revisions keep the turn row's own +// (none). So a child-linked row can never be what terminates one of these +// scans. Re-check that before giving any of those sites a producer. import type { AgentJournalRenderItem,