diff --git a/src/main/native-chat/agent-session-journal/journal-store.ts b/src/main/native-chat/agent-session-journal/journal-store.ts index ec445cb7004..568fbe4d39d 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.ts @@ -76,6 +76,7 @@ export class AgentSessionJournal { private readOnly = false private malformedRows = 0 private database: OpenJournalDatabase | null = null + private onCommitted: (() => void) | null = null private readonly queue: JournalWriteQueue private readonly closer: JournalConnectionCloser private readonly rowWriter: JournalRowWriter @@ -111,8 +112,14 @@ export class AgentSessionJournal { this.readOnly = readOnly }, cursor: this.cursor, - adopt: (loaded) => this.adoptLoadedJournal(loaded), - commit: (row) => applyJournalRow(this.state, row), + adopt: (loaded) => { + this.adoptLoadedJournal(loaded) + this.onCommitted?.() + }, + commit: (row) => { + applyJournalRow(this.state, row) + this.onCommitted?.() + }, loaded: () => this.loaded, malformedRows: () => this.malformedRows, setMalformedRows: (count) => { @@ -167,6 +174,12 @@ export class AgentSessionJournal { return this.closer.close() } + /** Told of every durable change, epoch replacements included, so a reader learns of a write + * without its writer saying so. One listener: a later call replaces it. It must not throw. */ + observeCommits(listener: () => void): void { + this.onCommitted = listener + } + cursor = (): AgentJournalCursor => ({ epoch: this.state.epoch, sequence: this.state.lastSequence diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-append-delivery.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-append-delivery.test.ts new file mode 100644 index 00000000000..aa955eeee9c --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-append-delivery.test.ts @@ -0,0 +1,188 @@ +// What an open chat receives, asserted at its subscriber rather than in the journal: a fresh +// subscribe re-reads the journal and hides a write that never reached the readers already open. + +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi, type Mock } from 'vitest' +import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope' +import type { + AgentJournalRenderItem, + AgentJournalSubmission +} from '../../../shared/agent-session-journal-types' +import type { AgentSessionSubscribeEvent } from '../../../shared/agent-session-wire' +import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' +import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter' +import { StructuredAgentSessionHost } from './structured-agent-session-host' +import { + HOST_TEST_NOW as NOW, + HOST_TEST_SESSION as SESSION, + HOST_TEST_THREAD as THREAD, + hostTestAttachParams, + hostTestMessage, + hostTestOperationId, + resetHostTestOperationIds +} from './structured-agent-session-host-test-data' + +const CALLER = { callerKey: 'client-1' } +const EXIT_REASON = 'Claude Code is not signed in. Sign in with the Claude CLI' + +let root: string +let store: AgentSessionRecordStore +let host: StructuredAgentSessionHost +let acquire: Mock +let generation = 0 + +/** Everything a live subscriber was sent after it opened. */ +function liveReader() { + const events: AgentSessionSubscribeEvent[] = [] + host.subscribe({ id: 'pane', sessionId: SESSION, emit: (event) => events.push(event) }) + const opened = events.length + const received = () => { + const items: AgentJournalRenderItem[] = [] + const submissions: AgentJournalSubmission[] = [] + for (const event of events.slice(opened)) { + if (event.type === 'batch') { + items.push(...event.batch.items) + submissions.push(...event.batch.submissions) + } else if (event.type === 'snapshot' || event.type === 'reset') { + items.push(...event.page.items) + submissions.push(...event.page.submissions) + } + } + return { + statuses: items.flatMap((item) => (item.body.kind === 'status' ? [item.body.text] : [])), + submissions + } + } + return { received } +} + +async function send(text: string): Promise { + const body = hostTestMessage(text) + const sent = await host.send(CALLER, { + envelope: { + sessionId: SESSION, + clientOperationId: hostTestOperationId(), + expectedRuntimeFence: store.getRecord(SESSION)?.lease.runtimeFence ?? 0, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.send', + sessionId: SESSION, + fields: { body } + }) + }, + body + }) + expect(sent, JSON.stringify(sent)).toMatchObject({ ok: true }) + return sent.ok ? sent.value.clientMessageId : '' +} + +function exitBeforeProof(): Promise { + return host.handleAdapterEvent({ + type: 'ended', + sessionId: SESSION, + fence: store.getRecord(SESSION)?.lease.runtimeFence ?? 0, + acquisitionGeneration: `generation-${generation}`, + reason: EXIT_REASON, + cause: 'unexpected-exit', + startupUnproven: true + }) +} + +function providerSink() { + const events = acquire.mock.calls.at(-1)?.[0].events + if (!events) { + throw new Error('no acquired provider sink') + } + return events +} + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-append-delivery-')) + resetHostTestOperationIds() + generation = 0 + acquire = vi.fn(async ({ fence, spawnToken }) => ({ + process: { hostId: 'local', pid: 4242, processStartTimeMs: 1_700_000_000_000, spawnToken }, + link: { + linkId: `link-${fence}`, + handle: { provider: 'codex', threadId: THREAD }, + origin: generation === 0 ? ('created' as const) : ('resumed' as const), + mintedAtFence: fence, + observedAt: NOW + }, + acquisitionGeneration: `generation-${++generation}`, + providerChildPhase: 'starting' as const + })) + store = await AgentSessionRecordStore.open({ directory: join(root, 'store'), hostId: 'local' }) + host = new StructuredAgentSessionHost({ + store, + adapter: { + acquire, + releaseAcquisition: vi.fn(async () => true), + closeSession: vi.fn(async () => true), + dispatch: vi.fn(async () => ({ state: 'admitted' as const })), + cancelTurn: vi.fn(async () => ({ cancelled: true })), + answerPrompt: vi.fn(async () => undefined), + setOption: vi.fn(async () => undefined) + }, + journalRoot: root, + claimKeyId: 'key-1', + mintSpawnToken: () => `spawn-${generation + 1}`, + now: () => NOW + }) + await expect(host.attach(CALLER, hostTestAttachParams(null))).resolves.toMatchObject({ + ok: true + }) +}) + +afterEach(async () => { + await host.flushAllStreamedEvents() + await rm(root, { recursive: true, force: true }) +}) + +describe('an open chat receives every row its journal commits', () => { + it('shows a failed start whose lease could not be handed back', async () => { + const held = await send('hello') + const pane = liveReader() + // The exit settles the journal, then fails to release the lease: nothing moves the fence. + vi.spyOn(store, 'transitionHandoff').mockRejectedValueOnce(new Error('record store busy')) + + await exitBeforeProof() + + expect(pane.received().statuses).toEqual([ + expect.stringMatching(/stopped before it finished starting: .*not signed in/) + ]) + expect(pane.received().submissions).toContainEqual( + expect.objectContaining({ clientMessageId: held, dispatchState: 'rejected' }) + ) + }) + + it('shows a revision the provider queued with no publish behind it', async () => { + const pane = liveReader() + const identity = { provider: 'orca' as const, clientMessageId: 'context-usage' } + const body = { kind: 'status' as const, text: 'context usage answered after the turn' } + + expect(providerSink().tryReviseResolvedItem?.(4_096, () => ({ identity, body }))).toEqual({ + accepted: true + }) + await host.flushStreamedEvents(SESSION) + + expect(pane.received().statuses).toEqual(['context usage answered after the turn']) + }) + + it('shows a row appended straight to the journal', async () => { + const pane = liveReader() + const journal = host['sessions'].get(SESSION)?.journal + if (!journal) { + throw new Error('the attached chat has no journal') + } + + await journal.appendItem( + { provider: 'orca', clientMessageId: 'host-note' }, + { kind: 'status', text: 'written by a writer that publishes nothing' }, + { fence: store.getRecord(SESSION)?.lease.runtimeFence ?? 0 } + ) + + expect(pane.received().statuses).toEqual(['written by a writer that publishes nothing']) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-conversations.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-conversations.test.ts new file mode 100644 index 00000000000..f76c84d33fc --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-conversations.test.ts @@ -0,0 +1,126 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, describe, expect, it, vi } from 'vitest' +import type { AgentSessionJournalIdentity } 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 { StructuredAgentSessionConversations } from './structured-agent-session-conversations' +import { hostTestAttachParams } from './structured-agent-session-host-test-data' + +const IDENTITY: AgentSessionJournalIdentity = { + sessionId: 'session-1', + workspaceId: 'workspace-1', + hostId: 'host-1', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: 'thread-1' } +} + +const journals = createTrackedJournalOpener() +let root: string | null = null + +afterEach(async () => { + await journals.closeAll() + if (root) { + await rm(root, { recursive: true, force: true }) + root = null + } +}) + +async function openJournal(name: string): Promise { + root ??= await mkdtemp(join(tmpdir(), 'orca-conversations-')) + return journals.open({ identity: IDENTITY, journalDir: join(root, name) }) +} + +function session(journal: AgentSessionJournal) { + return { + journal, + params: hostTestAttachParams(null), + fence: 0, + hasProviderChild: false, + providerChildPhase: 'ready' as const, + acquisitionGeneration: null + } +} + +function appendStatus(journal: AgentSessionJournal, text: string) { + return journal.appendItem( + { provider: 'orca', clientMessageId: text }, + { kind: 'status', text }, + { fence: 0 } + ) +} + +describe('a conversation delivers what its journal commits', () => { + it('delivers an append that no writer published', async () => { + const deliver = vi.fn() + const conversations = new StructuredAgentSessionConversations({ + deliver, + onDeliveryError: vi.fn() + }) + const journal = await openJournal('a') + conversations.set('session-1', session(journal)) + + await appendStatus(journal, 'written') + + // Already delivered when the writer's await returns, as an explicit publish would have been. + expect(deliver).toHaveBeenCalledOnce() + expect(deliver).toHaveBeenCalledWith('session-1', journal) + }) + + it('delivers nothing for a handle the conversation has replaced', async () => { + const deliver = vi.fn() + const conversations = new StructuredAgentSessionConversations({ + deliver, + onDeliveryError: vi.fn() + }) + const replaced = await openJournal('a') + const current = await openJournal('b') + conversations.set('session-1', session(replaced)) + conversations.set('session-1', session(current)) + + await appendStatus(replaced, 'stale') + expect(deliver).not.toHaveBeenCalled() + + await appendStatus(current, 'live') + expect(deliver).toHaveBeenCalledExactlyOnceWith('session-1', current) + }) + + it('delivers nothing once the conversation is dropped', async () => { + const deliver = vi.fn() + const conversations = new StructuredAgentSessionConversations({ + deliver, + onDeliveryError: vi.fn() + }) + const journal = await openJournal('a') + conversations.set('session-1', session(journal)) + conversations.delete('session-1') + + await appendStatus(journal, 'after close') + + expect(deliver).not.toHaveBeenCalled() + }) + + it('reports a reader failure without failing the durable write', async () => { + const failure = new Error('reader failed') + const onDeliveryError = vi.fn() + const conversations = new StructuredAgentSessionConversations({ + deliver: () => { + throw failure + }, + onDeliveryError + }) + const journal = await openJournal('a') + conversations.set('session-1', session(journal)) + + await expect(appendStatus(journal, 'durable')).resolves.toMatchObject({ + itemId: expect.any(String) + }) + + expect(onDeliveryError).toHaveBeenCalledExactlyOnceWith('session-1', failure) + expect(journal.snapshot().items.map((item) => item.body)).toContainEqual({ + kind: 'status', + text: 'durable' + }) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-conversations.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-conversations.ts new file mode 100644 index 00000000000..2f38ddf4da2 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-conversations.ts @@ -0,0 +1,47 @@ +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import type { StructuredAgentSessionHostSession } from './structured-agent-session-host-types' + +/** + * The host's open conversations. A journal handle becomes a conversation's when it is set here, + * and from then on every change it commits reaches that conversation's readers: no writer + * publishes what it appended, so none can forget to. + * + * Delivery runs a microtask after the commit, so a reader that throws cannot fail a write that is + * already durable. A handle this map has since replaced or dropped delivers nothing. + */ +export class StructuredAgentSessionConversations extends Map< + string, + StructuredAgentSessionHostSession +> { + constructor( + private readonly delivery: { + deliver: (sessionId: string, journal: AgentSessionJournal) => void + onDeliveryError: (sessionId: string, error: unknown) => void + } + ) { + super() + } + + override set(sessionId: string, session: StructuredAgentSessionHostSession): this { + const { journal } = session + let queued = false + journal.observeCommits(() => { + if (queued) { + return + } + queued = true + queueMicrotask(() => { + queued = false + if (this.get(sessionId)?.journal !== journal) { + return + } + try { + this.delivery.deliver(sessionId, journal) + } catch (error) { + this.delivery.onDeliveryError(sessionId, error) + } + }) + }) + return super.set(sessionId, session) + } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts index 364f1e7c0dd..906025413a9 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts @@ -224,7 +224,6 @@ export async function settleStructuredAgentSessionLateDispatch( fence: session.fence } ) - context.publish(input.sessionId, session.journal) } /** @@ -240,7 +239,7 @@ export async function settleStructuredAgentSessionLateDispatch( * it never makes a send re-deliverable, because the provider may well have run it. */ export async function releaseStructuredAgentSessionUnansweredDispatches( - context: Pick, + context: Pick, input: { sessionId: string; reason: string } ): Promise { const session = context.sessions.get(input.sessionId) @@ -263,7 +262,6 @@ export async function releaseStructuredAgentSessionUnansweredDispatches( recovered: true }) } - context.publish(input.sessionId, session.journal) } /** The host's thin mutation surface. Each call re-reads the context, so a session diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts index ba508b67bcf..c6aaa8db662 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts @@ -46,6 +46,7 @@ import type { StructuredAgentSessionTurnCompletionSubscriber } from './structure import { StructuredAgentSessionEventRecovery } from './structured-agent-session-event-recovery' import { StructuredAgentSessionBackgroundTaskChannel } from './structured-agent-session-background-task-channel' import { StructuredAgentSessionClientDelivery } from './structured-agent-session-client-delivery' +import { StructuredAgentSessionConversations } from './structured-agent-session-conversations' import { createStructuredAgentSessionRestartResume, type StructuredAgentSessionRestartResume @@ -58,7 +59,10 @@ export class StructuredAgentSessionHost { () => this.mutationContext(), this ) - private readonly sessions = new Map() + private readonly sessions = new StructuredAgentSessionConversations({ + deliver: (sessionId, journal) => this.subscribers.publish(sessionId, journal), + onDeliveryError: (sessionId, error) => this.deps.onEventSinkError?.({ sessionId, error }) + }) private readonly clientDelivery = new StructuredAgentSessionClientDelivery( this.sessions, () => this.now(), @@ -127,10 +131,11 @@ export class StructuredAgentSessionHost { ensureProviderChild: (id, options) => this.holds.ensureProviderChild(id, options), onBarrierError: (sessionId, error) => deps.onEventSinkError?.({ sessionId, error }) }) - this.restartResume = createStructuredAgentSessionRestartResume(deps, this.sessions, { - ...structuredAgentSessionRestartResumeSurfaces(this, this.now), - publish: this.subscribers.publish.bind(this.subscribers) - }) + this.restartResume = createStructuredAgentSessionRestartResume( + deps, + this.sessions, + structuredAgentSessionRestartResumeSurfaces(this, this.now) + ) this.runtimeState.startLeaseRenewal() } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-continuation.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-continuation.ts index 58d104a651c..0dd2337f4ab 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-continuation.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-continuation.ts @@ -55,7 +55,6 @@ export type StructuredAgentSessionContinuationHost = { clientMessageId: string ) => Promise<{ value: AgentSessionSendResult } | undefined> onNoteFailed: (sessionId: string, error: unknown) => void - publish: (sessionId: string, journal: AgentSessionJournal) => void now: () => number /** Whether the marker is still an offer, with the continuation's own submission set aside. * Re-asked right before dispatch, so a newer user message refuses the send; a provider turn @@ -106,9 +105,9 @@ export function noteRestartReattachFailed( ) } -/** Writes a host-authored status note into the chat and publishes it to open panes. */ +/** Writes a host-authored status note into the chat. */ function restartNoteWriter( - host: Pick + host: Pick ): StructuredAgentSessionContinuationDeps['note'] { return async (sessionId, text, tone) => { const session = host.sessions.get(sessionId) @@ -120,7 +119,6 @@ function restartNoteWriter( { kind: 'status', text, ...(tone ? { tone } : {}) }, { fence: session.fence } ) - host.publish(sessionId, session.journal) } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-host.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-host.ts index 0770190c2f4..29453e74652 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-host.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-host.ts @@ -45,7 +45,6 @@ import { createStructuredAgentSessionRestartWitnesses } from './structured-agent type LiveSession = { journal: AgentSessionJournal; hasProviderChild: boolean; fence: number } export type StructuredAgentSessionRestartResumeSurfaces = { - publish: (sessionId: string, journal: AgentSessionJournal) => void revealSession: (sessionId: string) => Promise<{ readable: boolean }> hold: (sessionId: string, holderId: string) => Promise release: (sessionId: string, holderId: string) => void diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-wiring.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-wiring.ts index 4494392fb48..07ec6b81599 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-wiring.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-wiring.ts @@ -41,7 +41,7 @@ type RestartResumeHostBindings = { export function structuredAgentSessionRestartResumeSurfaces( host: RestartResumeHostBindings, now: () => number -): Omit { +): StructuredAgentSessionRestartResumeSurfaces { return { revealSession: host.revealSession, hold: host.hold, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-send-preparation.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-send-preparation.ts index e4a332f7acf..e277889345f 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-send-preparation.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-send-preparation.ts @@ -114,7 +114,7 @@ export function structuredAgentSessionSendNeedsOwner( type SendPreparationContext = Pick< StructuredAgentSessionMutationContext, - 'deps' | 'sessions' | 'holds' | 'restoreReadable' | 'publish' + 'deps' | 'sessions' | 'holds' | 'restoreReadable' > export async function prepareStructuredAgentSessionSend( @@ -224,7 +224,6 @@ async function recordFailedRestart( } ] }) - context.publish(sessionId, session.journal) } catch (error) { context.deps.onEventSinkError?.({ sessionId, error }) } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-surface-lifetime.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-surface-lifetime.test.ts index d1b17ef49b5..cb958d8364f 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-surface-lifetime.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-surface-lifetime.test.ts @@ -246,7 +246,7 @@ describe('a chat that closes', () => { expect(host.hasSession(SESSION)).toBe(true) }) - it('releases a compatibility wait when the session is evicted', async () => { + it('answers a compatibility wait with what eviction recorded', async () => { await attach() dispatch.mockResolvedValueOnce({ state: 'admitted' }) const body = hostTestMessage('pending until close') @@ -265,7 +265,12 @@ describe('a chat that closes', () => { await host.close(SESSION) - await expect(settlement).resolves.toBeUndefined() + // Eviction's settlement is a journal write, so the wait sees it rather than timing out. + await expect(settlement).resolves.toMatchObject({ + value: { + submission: { dispatchState: 'unknown', reason: 'provider_closed_before_acknowledgement' } + } + }) }) it('retries teardown after journal close loses its result', async () => { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-thread-goal.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-thread-goal.ts index 4e49e00f587..3f63679d55f 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-thread-goal.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-thread-goal.ts @@ -68,13 +68,11 @@ export async function performThreadGoalChange( }, { fence: ctx.fence } ) - ctx.publish() } const withdrawObjective = async (): Promise => { if (change.kind === 'set') { // Nothing was sent as a goal. await ctx.journal.appendTombstone(identity, { fence: ctx.fence }) - ctx.publish() } } let result: Awaited> diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-turns-prompt.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-turns-prompt.ts index 6ac29cab0ab..2f6a1a1f1f2 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-turns-prompt.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-turns-prompt.ts @@ -73,7 +73,6 @@ export async function performPrompt( fence: ctx.fence } ) - ctx.publish() } }) } catch (error) { @@ -93,7 +92,6 @@ export async function performPrompt( }, { fence: ctx.fence } ) - ctx.publish() } const appended = committed.item if (!appended) { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-turns.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-turns.ts index bea393eddbb..e8103dac62f 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-turns.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-turns.ts @@ -42,6 +42,8 @@ export type AgentSessionTurnContext = { persistOptions: (options: Readonly>) => Promise /** Opaque client identity recorded as the resolver of a prompt. */ resolvedBy: string + /** Republishes state kept outside the journal, such as the record's options or rewind phase. + * Journal appends reach readers on their own. */ publish: () => void /** Drains provider lifecycle already accepted by the execution host. */ flushStreamedEvents: () => Promise @@ -100,7 +102,6 @@ async function appendStatus( { kind: 'status', text }, { fence: ctx.fence } ) - ctx.publish() } /** @@ -136,7 +137,6 @@ export async function performSend( } catch { return invalid('The message could not be recorded and was not sent.') } - ctx.publish() // The row just written is the send's instant on the host clock; the turn this // dispatch opens records it so the live counter never re-anchors at turn-open. @@ -161,14 +161,12 @@ export async function performSend( if (!recorded) { console.warn('[structured-agent-session] pre-dispatch refusal persistence failed') } - ctx.publish() } throw error } ) // An admission needs no dispatch row: the submission is already pending. if (outcome.state === 'admitted') { - ctx.publish() return { ok: true, value: { @@ -206,10 +204,8 @@ export async function performSend( } catch { // Nothing further to record; the pending row is settled on the next attach. } - ctx.publish() throw error } - ctx.publish() return { ok: true, value: { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-unanswered-dispatch-release.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-unanswered-dispatch-release.test.ts index fc8440500a4..79f21a349b4 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-unanswered-dispatch-release.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-unanswered-dispatch-release.test.ts @@ -35,9 +35,8 @@ function contextWith(submissions: AgentJournalSubmission[]) { // The mutation reads only `journal.submissions`, `journal.resolveDispatch` and `fence`. // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: all three are supplied here; the rest of the host session is unreachable from this mutation. const session = { journal, fence: FENCE } as unknown as ReleaseSession - const publish = vi.fn() - const context: ReleaseContext = { sessions: new Map([['s-1', session]]), publish } - return { context, resolved, publish, journal } + const context: ReleaseContext = { sessions: new Map([['s-1', session]]) } + return { context, resolved, journal } } describe('releasing dispatches the provider can no longer answer', () => { @@ -45,7 +44,7 @@ describe('releasing dispatches the provider can no longer answer', () => { const live = [submission({})] // POSITIVE CONTROL: this is the latch being dissolved. expect(projectStructuredAgentSessionStatus([], live, FENCE)).toBe('working') - const { context, resolved, publish } = contextWith(live) + const { context, resolved } = contextWith(live) await releaseStructuredAgentSessionUnansweredDispatches(context, { sessionId: 's-1', @@ -62,14 +61,13 @@ describe('releasing dispatches the provider can no longer answer', () => { recovered: true } ]) - expect(publish).toHaveBeenCalledOnce() expect(projectStructuredAgentSessionStatus([], [submission({ recovered: true })], FENCE)).toBe( 'idle' ) }) it('never touches a pending send, whose dispatch may still be in flight', async () => { - const { context, resolved, publish } = contextWith([ + const { context, resolved } = contextWith([ submission({ clientMessageId: 'm-2', dispatchState: 'pending', reason: null }) ]) @@ -79,7 +77,6 @@ describe('releasing dispatches the provider can no longer answer', () => { }) expect(resolved).toEqual([]) - expect(publish).not.toHaveBeenCalled() }) it('leaves an already recovered unknown alone', async () => { diff --git a/src/main/native-chat/agent-session-wire/structured-conversation-command.ts b/src/main/native-chat/agent-session-wire/structured-conversation-command.ts index 022343d9407..5894defc483 100644 --- a/src/main/native-chat/agent-session-wire/structured-conversation-command.ts +++ b/src/main/native-chat/agent-session-wire/structured-conversation-command.ts @@ -176,7 +176,6 @@ export function runStructuredConversationCommand( }, { fence: ctx.fence } ) - ctx.publish() try { error = ( await ctx.adapter.compact({ @@ -212,7 +211,6 @@ export function runStructuredConversationCommand( conversationCommand: matching()! } }) - ctx.publish() }) }) ).error @@ -223,7 +221,6 @@ export function runStructuredConversationCommand( { kind: 'status', text: 'Compaction completion is unconfirmed.' }, { fence: ctx.fence } ) - ctx.publish() throw cause } await ctx.journal.appendItem( @@ -231,7 +228,6 @@ export function runStructuredConversationCommand( { kind: 'status', text: error ?? 'Conversation compacted.' }, { fence: ctx.fence } ) - ctx.publish() } const completed = { ...prepared,