From a2d7f63a8317485486e8acda319310052b5ad097 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 17 Sep 2026 16:36:12 -0700 Subject: [PATCH] Resolve mainline conflicts in settlement plumbing --- .../codex-structured-dispatch-test-support.ts | 4 +- .../structured-agent-session-event-sink.ts | 109 +++++++----------- 2 files changed, 46 insertions(+), 67 deletions(-) diff --git a/src/main/codex/codex-structured-dispatch-test-support.ts b/src/main/codex/codex-structured-dispatch-test-support.ts index b5722198f54..768a442e61c 100644 --- a/src/main/codex/codex-structured-dispatch-test-support.ts +++ b/src/main/codex/codex-structured-dispatch-test-support.ts @@ -10,6 +10,7 @@ import type { } from './codex-app-server-connection' import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' import { CodexStructuredSessionAdapter } from './codex-structured-session-adapter' +import type { CodexStructuredSessionAdapterDeps } from './codex-structured-session-state' import type { CodexStructuredSessionEvent } from './codex-structured-session-state' import { readCodexTurnId } from './codex-structured-thread-facts' import type { @@ -91,6 +92,7 @@ export async function acquiredCodexAdapter(input: { settleLateDispatch?: ( settlement: LateSettlement ) => Promise + captureTurnProcesses?: CodexStructuredSessionAdapterDeps['captureTurnProcesses'] }): Promise { const events = input.events const terminalTurnIds = new Set() @@ -104,7 +106,7 @@ export async function acquiredCodexAdapter(input: { }), openConnection: input.codex.openConnection, readProcessStartTime: async () => 1_700_000_000_000, - captureTurnProcesses: async () => null, + captureTurnProcesses: input.captureTurnProcesses ?? (async () => null), now: () => 1_700_000_000_500, onDispatchSettledLate: async (settlement) => { if (input.settleLateDispatch) { 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 3853559de50..1146bde634b 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 @@ -8,6 +8,7 @@ import type { AgentSessionJournal } from '../agent-session-journal/journal-store import type { JournalLifecycleMutationInput } from '../agent-session-journal/journal-row-builders' import { estimateStructuredAgentSessionItemBytes } from './structured-agent-session-event-sink-estimate' import { StructuredAgentSessionSinkQueue } from './structured-agent-session-event-sink-queue' +import { createStructuredAgentSessionResolvedAppend } from './structured-agent-session-resolved-append' export type StructuredAgentSessionSinkAdmission = | { accepted: true } @@ -23,18 +24,15 @@ export type StructuredAgentSessionSinkState = { export type StructuredAgentSessionSinkBarrier = { ok: true } | { ok: false; error: unknown } export type StructuredAgentSessionAppendOptions = { + ownerEndedClientMessageIds?: readonly string[] + onCommitted?: () => void + onAbandoned?: () => void /** Pending checkpoints with this key replace one another before they run. */ coalescingKey?: string /** Marks a critical lifecycle operation for lifecycle barriers and diagnostics. */ lifecycle?: boolean /** Host clock to stamp on the row instead of its append time. */ observedAt?: number - /** Exact sends whose provider ownership ends with this terminal lifecycle row. */ - ownerEndedClientMessageIds?: readonly string[] - /** Runs only after the queued lifecycle append is durable. */ - onCommitted?: () => void - /** Releases in-memory bookkeeping if the queued lifecycle append is discarded. */ - onAbandoned?: () => void } export type StructuredAgentSessionLifecycleJournal = Pick< @@ -42,12 +40,14 @@ export type StructuredAgentSessionLifecycleJournal = Pick< 'epoch' | 'visitItems' > -export type StructuredAgentSessionLifecycleIdentityResolver = ( +export type StructuredAgentSessionIdentityResolver = ( journal: StructuredAgentSessionLifecycleJournal ) => AgentJournalItemIdentity | null +/** Compatibility alias for lifecycle callers that already use this resolver. */ +export type StructuredAgentSessionLifecycleIdentityResolver = StructuredAgentSessionIdentityResolver + export type StructuredAgentSessionEventSink = { - /** This sink invokes lifecycle callbacks at durable completion or abandonment. */ durableLifecycleCallbacks?: true appendItem( identity: AgentJournalItemIdentity, @@ -69,11 +69,25 @@ export type StructuredAgentSessionEventSink = { body: AgentJournalItemBody, options?: StructuredAgentSessionAppendOptions ): StructuredAgentSessionSinkAdmission + /** Queues an ordinary append whose identity is resolved after journal bind. */ + tryAppendResolvedItem?( + identitySizeBound: AgentJournalItemIdentity, + body: AgentJournalItemBody, + resolveIdentity: StructuredAgentSessionIdentityResolver, + options?: StructuredAgentSessionAppendOptions + ): StructuredAgentSessionSinkAdmission + /** Queues one resolved append and its publication as a single admitted operation. */ + tryAppendResolvedItemAndPublish?( + identitySizeBound: AgentJournalItemIdentity, + body: AgentJournalItemBody, + resolveIdentity: StructuredAgentSessionIdentityResolver, + options?: StructuredAgentSessionAppendOptions + ): StructuredAgentSessionSinkAdmission /** Queues one journal-derived lifecycle append; a null resolution is a no-op. */ tryAppendLifecycleTransition?( identitySizeBound: AgentJournalItemIdentity, body: AgentJournalItemBody, - resolveIdentity: StructuredAgentSessionLifecycleIdentityResolver + resolveIdentity: StructuredAgentSessionIdentityResolver ): StructuredAgentSessionSinkAdmission /** Current durable epoch, when this deferred sink is bound to its journal. */ journalEpoch?(): string | null @@ -150,46 +164,26 @@ export function createDeferredStructuredAgentSessionEventSink( ...(deps.readingControl ? { readingControl: deps.readingControl } : {}), ...(deps.onBackpressureChange ? { onBackpressureChange: deps.onBackpressureChange } : {}) }) + const resolvedAppend = createStructuredAgentSessionResolvedAppend(queue) const appendLifecycleBatch = ( settlementId: string, mutations: readonly JournalLifecycleMutationInput[], options: StructuredAgentSessionAppendOptions = {} - ): StructuredAgentSessionSinkAdmission => { - const ownerEndedClientMessageIds = options.ownerEndedClientMessageIds?.slice() - return queue.submit( + ): StructuredAgentSessionSinkAdmission => + queue.submit( { - bytes: - Buffer.byteLength( - JSON.stringify({ - settlementId, - mutations, - ownerEndedClientMessageIds - }), - 'utf8' - ) + 512, + bytes: Buffer.byteLength(JSON.stringify({ settlementId, mutations }), 'utf8') + 512, coalescingKey: `lifecycle:${settlementId}`, run: (bound) => - bound.journal - .appendLifecycleBatch({ - settlementId, - mutations, - fence: bound.fence, - ...(ownerEndedClientMessageIds ? { ownerEndedClientMessageIds } : {}) - }) - .then(options.onCommitted, (error: unknown) => { - options.onAbandoned?.() - throw error - }), - ...(options.onAbandoned ? { onDiscarded: options.onAbandoned } : {}) + bound.journal.appendLifecycleBatch({ + settlementId, + mutations, + fence: bound.fence + }) }, - { - ...options, - lifecycle: true, - ...(ownerEndedClientMessageIds ? { ownerEndedClientMessageIds } : {}) - } + { ...options, lifecycle: true } ) - } const publish = ( options: StructuredAgentSessionAppendOptions = {} @@ -206,49 +200,33 @@ export function createDeferredStructuredAgentSessionEventSink( return { sink: { appendItem: (identity, body, options = {}) => { - const ownerEndedClientMessageIds = options.ownerEndedClientMessageIds - ? [...options.ownerEndedClientMessageIds] - : undefined queue.submit( { - bytes: - estimateStructuredAgentSessionItemBytes(identity, body) + - (ownerEndedClientMessageIds - ? Buffer.byteLength(JSON.stringify(ownerEndedClientMessageIds), 'utf8') - : 0), + bytes: estimateStructuredAgentSessionItemBytes(identity, body), coalescingKey: options.coalescingKey, run: (bound) => bound.journal.appendItem(identity, body, { fence: bound.fence, - ...(options.observedAt === undefined ? {} : { observedAt: options.observedAt }), - ...(ownerEndedClientMessageIds ? { ownerEndedClientMessageIds } : {}) + ...(options.observedAt === undefined ? {} : { observedAt: options.observedAt }) }) }, - { ...options, ...(ownerEndedClientMessageIds ? { ownerEndedClientMessageIds } : {}) } + options ) }, - tryAppendItem: (identity, body, options = {}) => { - const ownerEndedClientMessageIds = options.ownerEndedClientMessageIds - ? [...options.ownerEndedClientMessageIds] - : undefined - return queue.submit( + tryAppendItem: (identity, body, options = {}) => + queue.submit( { - bytes: - estimateStructuredAgentSessionItemBytes(identity, body) + - (ownerEndedClientMessageIds - ? Buffer.byteLength(JSON.stringify(ownerEndedClientMessageIds), 'utf8') - : 0), + bytes: estimateStructuredAgentSessionItemBytes(identity, body), coalescingKey: options.coalescingKey, run: (bound) => bound.journal.appendItem(identity, body, { fence: bound.fence, - ...(options.observedAt === undefined ? {} : { observedAt: options.observedAt }), - ...(ownerEndedClientMessageIds ? { ownerEndedClientMessageIds } : {}) + ...(options.observedAt === undefined ? {} : { observedAt: options.observedAt }) }) }, - { ...options, ...(ownerEndedClientMessageIds ? { ownerEndedClientMessageIds } : {}) } - ) - }, + options + ), + ...resolvedAppend, tryAppendLifecycleTransition: (identitySizeBound, body, resolveIdentity) => { const bytes = estimateStructuredAgentSessionItemBytes(identitySizeBound, body) return queue.submit( @@ -282,7 +260,6 @@ export function createDeferredStructuredAgentSessionEventSink( } return admission }, - durableLifecycleCallbacks: true, tryAppendLifecycleBatch: appendLifecycleBatch, bindReadingControl: (control) => { return queue.bindReadingControl(control)