diff --git a/src/main/native-chat/agent-session-wire/agent-session-history-page.test.ts b/src/main/native-chat/agent-session-wire/agent-session-history-page.test.ts index 5b84b8417e5..b8950517c20 100644 --- a/src/main/native-chat/agent-session-wire/agent-session-history-page.test.ts +++ b/src/main/native-chat/agent-session-wire/agent-session-history-page.test.ts @@ -123,6 +123,23 @@ describe('readAgentSessionHistory', () => { expect(page.page.window.nextCursor.sequence).toBeGreaterThan(cursor.sequence) }) + it('carries tombstones in a forward catch-up page', async () => { + await appendItems(1) + const cursor = journal.cursor() + await journal.appendTombstone(item(1), { fence: 1 }) + + const page = readAgentSessionHistory(journal, { + sessionId: 'session-1', + direction: 'after', + cursor + }) + if (!page.ok) { + throw new Error(`expected a page, got reset ${page.reset}`) + } + expect(page.page.items).toHaveLength(0) + expect(page.page.removedItemIds).toEqual(['codex:thread-1:turn-1:1']) + }) + it('reports a forward read with no cursor as cursor_ahead rather than serving the tail', async () => { await appendItems(1) expect( diff --git a/src/main/native-chat/agent-session-wire/agent-session-history-page.ts b/src/main/native-chat/agent-session-wire/agent-session-history-page.ts index c3c621bb1f0..2f7a398e5c5 100644 --- a/src/main/native-chat/agent-session-wire/agent-session-history-page.ts +++ b/src/main/native-chat/agent-session-wire/agent-session-history-page.ts @@ -102,6 +102,7 @@ function readForward( snapshot, direction: 'after', items: projected.batch.items, + removedItemIds: projected.batch.removedItemIds, // Reading after a position means there is something before it. hasOlder: cursor.sequence > 0, hasNewer: since.rows.length > rows.length, @@ -115,6 +116,7 @@ function buildPage(input: { snapshot: AgentJournalSnapshot direction: AgentSessionHistoryDirection items: AgentJournalRenderItem[] + removedItemIds?: string[] hasOlder: boolean hasNewer: boolean fallbackCursor: AgentJournalCursor @@ -129,6 +131,7 @@ function buildPage(input: { epoch, direction: input.direction, items: input.items, + removedItemIds: input.removedItemIds ?? [], submissions: input.snapshot.submissions.filter((submission) => pageItemIds.has(agentJournalSubmissionKey(submission.clientMessageId)) ), diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-flow.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-flow.ts index 1d0292fefaf..4ce563ec9ce 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-flow.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-flow.ts @@ -5,10 +5,12 @@ // the decisions that must not be client-supplied — the spawn token, the claim // key, the owner probe — and passes them in. +import { isDeepStrictEqual } from 'node:util' import type { AgentSessionAttachResult, AgentSessionMutationResult } from '../../../shared/agent-session-wire' +import { agentSessionLeaseAdmitsWriter } from '../../../shared/agent-session-lease-adjudication' import type { AgentSessionRecord } from '../../../shared/agent-session-record' import { admitAttachOrRefuse, @@ -62,7 +64,7 @@ export async function performAttach( ) record = reserved.record replayed = reserved.disposition === 'replayed' - if (record.lease.ownerProcess === null) { + if (!agentSessionLeaseAdmitsWriter(record.lease)) { record = await acquireOwner(input, record) } } catch (error) { @@ -107,17 +109,26 @@ async function acquireOwner( record: AgentSessionRecord ): Promise { const fence = record.lease.runtimeFence + const spawnToken = record.lease.reservedSpawnToken + if (!spawnToken) { + throw new Error('agent_session_ownership_unknown') + } const acquired = await input.adapter.acquire({ identity: journalIdentityFor(record.sessionId, input.params), fence, - spawnToken: input.authority.spawnToken - }) - await input.store.commitProcessIdentity({ - sessionId: record.sessionId, - fence, - process: acquired.process, - now: input.now() + // Retries must recover the original reservation, not mint a second child. + spawnToken }) + if (record.lease.ownerProcess === null) { + await input.store.commitProcessIdentity({ + sessionId: record.sessionId, + fence, + process: acquired.process, + now: input.now() + }) + } else if (!isDeepStrictEqual(record.lease.ownerProcess, acquired.process)) { + throw new Error('agent_session_ownership_unknown') + } return input.store.proveOwner({ sessionId: record.sessionId, fence, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-attach.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-attach.ts index 27fcfd962c7..facca4d142c 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-attach.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-attach.ts @@ -50,7 +50,7 @@ export type AgentSessionAttachParams = { agent: AgentType accountHome: AgentSessionAccountHome runtimeKind: AgentSessionOwnerRuntimeKind - providerHandle: AgentSessionProviderHandle + providerHandle: Exclude } /** Host-supplied half of the reservation. */ @@ -81,6 +81,15 @@ export function attachFingerprintFields(params: AgentSessionAttachParams): Recor export function admitAttachOrRefuse( params: AgentSessionAttachParams ): { ok: true; fingerprint: string } | { ok: false; refusal: AgentSessionWireRefusal } { + if (params.providerHandle.kind !== params.provider) { + return { + ok: false, + refusal: { + code: 'agent_session_operation_invalid', + message: `A ${params.provider} session requires a ${params.provider} provider handle.` + } + } + } const fingerprint = computeAgentSessionPayloadFingerprint({ method: 'agentSession.attach', sessionId: params.envelope.sessionId, 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 0fd4a23ee38..68ec4bcbce0 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 @@ -238,6 +238,18 @@ describe('attach', () => { }) }) + it('refuses a provider handle that belongs to a different provider', async () => { + const params = attachParams({ + providerHandle: { kind: 'claude', sessionId: 'claude-session', leafUuid: null } + }) + + expect(await host.attach(CALLER, params)).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_operation_invalid' } + }) + expect(store.getRecord(SESSION)).toBeNull() + }) + it('refuses a second create against a live session', async () => { await attach() expect(await host.attach(CALLER, attachParams())).toMatchObject({ ok: false }) @@ -249,6 +261,99 @@ describe('attach', () => { const retry = await host.attach(CALLER, params) expect(retry).toMatchObject({ ok: true, replayed: true }) }) + + it('reuses the persisted spawn token when acquisition is retried', async () => { + const spawnTokens: string[] = [] + const acquire = vi + .fn() + .mockImplementationOnce(async ({ spawnToken }) => { + spawnTokens.push(spawnToken) + throw new Error('reply lost') + }) + .mockImplementation(async ({ fence, spawnToken }) => { + spawnTokens.push(spawnToken) + return { + process: { + hostId: 'local', + pid: 4242, + processStartTimeMs: 1_700_000_000_000, + spawnToken + }, + link: { + linkId: `link-${fence}`, + handle: { provider: 'codex', threadId: THREAD }, + origin: 'created', + mintedAtFence: fence, + observedAt: NOW + } + } + }) + let token = 0 + host = new StructuredAgentSessionHost({ + store, + adapter: { ...adapter(), acquire }, + journalRoot: root, + claimKeyId: 'key-1', + mintSpawnToken: () => `spawn-${++token}`, + now: () => NOW + }) + const params = attachParams() + + await expect(host.attach(CALLER, params)).rejects.toThrow('reply lost') + expect(await host.attach(CALLER, params)).toMatchObject({ ok: true, replayed: true }) + expect(spawnTokens).toEqual(['spawn-1', 'spawn-1']) + }) + + it('finishes proof when a retry finds the process identity already committed', async () => { + const acquire = vi + .fn() + .mockImplementationOnce(async ({ fence, spawnToken }) => ({ + process: { + hostId: 'local', + pid: 4242, + processStartTimeMs: 1_700_000_000_000, + spawnToken + }, + link: { + linkId: 'stale-link', + handle: { provider: 'codex', threadId: THREAD }, + origin: 'created', + mintedAtFence: fence + 1, + observedAt: NOW + } + })) + .mockImplementation(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: 'created', + mintedAtFence: fence, + observedAt: NOW + } + })) + host = new StructuredAgentSessionHost({ + store, + adapter: { ...adapter(), acquire }, + journalRoot: root, + claimKeyId: 'key-1', + mintSpawnToken: () => 'spawn-a', + now: () => NOW + }) + const params = attachParams() + + await expect(host.attach(CALLER, params)).rejects.toThrow( + 'agent_session_provider_handle_stale_fence' + ) + expect(await host.attach(CALLER, params)).toMatchObject({ ok: true, replayed: true }) + expect(acquire).toHaveBeenCalledTimes(2) + expect(store.getRecord(SESSION)?.lease.claimStatus).toBe('live') + }) }) describe('send', () => { @@ -429,6 +534,37 @@ describe('respondToPrompt', () => { expect(answerPrompt).not.toHaveBeenCalled() }) + it("does not turn a recorded refusal into another client's successful answer", async () => { + const prompt = await seedApproval() + await attach() + const rejectedFields = { + itemId: prompt.itemId, + expectedRevision: prompt.revision, + optionId: 'deny' + } + const rejected = { + envelope: envelope('agentSession.respondTo:approval', rejectedFields), + kind: 'approval' as const, + ...rejectedFields + } + await host.respondToPrompt(CALLER, rejected) + + const acceptedFields = { ...rejectedFields, optionId: 'allow' } + await host.respondToPrompt( + { callerKey: 'client-2' }, + { + envelope: envelope('agentSession.respondTo:approval', acceptedFields), + kind: 'approval', + ...acceptedFields + } + ) + + expect(await host.respondToPrompt(CALLER, rejected)).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_operation_invalid' } + }) + }) + it('keeps the answer and reports it undelivered when the provider callback throws', async () => { const prompt = await seedApproval() await attach() @@ -462,6 +598,23 @@ describe('setOption', () => { const page = host.history({ sessionId: SESSION, direction: 'tail' }) expect(page.ok && page.page.items).toHaveLength(0) }) + + it('does not turn an unknown provider outcome into a successful replay', async () => { + await attach() + setOption.mockRejectedValueOnce(new Error('reply lost')) + const fields = { key: 'model', value: 'gpt-5' } + const params = { + envelope: envelope('agentSession.setOption', fields), + ...fields + } + + await expect(host.setOption(CALLER, params)).rejects.toThrow('reply lost') + expect(await host.setOption(CALLER, params)).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_operation_unknown' } + }) + expect(setOption).toHaveBeenCalledTimes(1) + }) }) describe('restart', () => { @@ -589,6 +742,33 @@ describe('subscribe', () => { expect(events.some((event) => event.type === 'snapshot')).toBe(false) }) + it('drops a failed transport without aborting the mutation or other subscribers', async () => { + await attach() + const events: AgentSessionSubscribeEvent[] = [] + host.subscribe({ + id: 'dead-sub', + sessionId: SESSION, + emit: () => { + throw new Error('socket closed') + } + }) + host.subscribe({ + id: 'live-sub', + sessionId: SESSION, + emit: (event) => events.push(event) + }) + const body = message('survive subscriber failure') + + const result = await host.send(CALLER, { + envelope: envelope('agentSession.send', { body }), + body + }) + + expect(result).toMatchObject({ ok: true, value: { submission: { dispatchState: 'accepted' } } }) + expect(dispatch).toHaveBeenCalledTimes(1) + expect(events.some((event) => event.type === 'batch')).toBe(true) + }) + it('resets a subscriber whose epoch is gone', async () => { await attach() const events: AgentSessionSubscribeEvent[] = [] @@ -598,6 +778,28 @@ describe('subscribe', () => { emit: (event) => events.push(event), cursor: { epoch: 'epoch-from-a-previous-life', sequence: 3 } }) - expect(events[0]).toMatchObject({ type: 'reset', reset: 'epoch_changed' }) + expect(events[0]).toMatchObject({ type: 'reset', reset: 'epoch_changed', fence: 1 }) + }) + + it('publishes the replacement fence when the owner generation changes', async () => { + const record = await attach() + const events: AgentSessionSubscribeEvent[] = [] + host.subscribe({ + id: 'sub-4', + sessionId: SESSION, + emit: (event) => events.push(event) + }) + const released = await store.evictProvenDeadOwner({ + sessionId: SESSION, + expectedFence: record?.lease.runtimeFence ?? 1, + probe: { outcome: 'pid-absent' }, + now: NOW + }) + + const replacement = await host.attach(CALLER, ensureParams(released.lease.runtimeFence)) + if (!replacement.ok) { + throw new Error(`expected replacement owner, got ${replacement.refusal.code}`) + } + expect(events.at(-1)).toMatchObject({ type: 'snapshot', fence: replacement.fence }) }) }) 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 4f77a503d7e..4bf5940ecea 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 @@ -45,11 +45,13 @@ import { setOptionPlan, type MutationPlan } from './structured-agent-session-mutation-plans' +import { runSettledAgentSessionMutation } from './structured-agent-session-operation-settlement' +import { resolveAgentSessionReplayOutcome } from './structured-agent-session-replay-outcome' import { AgentSessionSubscribers, type AgentSessionSubscriberEmit } from './structured-agent-session-subscribers' -import type { AgentSessionTurnContext, TurnOutcome } from './structured-agent-session-turns' +import type { AgentSessionTurnContext } from './structured-agent-session-turns' export type StructuredAgentSessionCaller = { /** Stable per-client identity; scopes the operation ledger and records who @@ -68,7 +70,11 @@ export type StructuredAgentSessionHostDeps = { now?: () => number } -type SessionState = { journal: AgentSessionJournal; params: AgentSessionAttachParams } +type SessionState = { + journal: AgentSessionJournal + params: AgentSessionAttachParams + fence: number +} function refuse(refusal: AgentSessionWireRefusal): { ok: false; refusal: AgentSessionWireRefusal } { return { ok: false, refusal } @@ -93,9 +99,7 @@ export class StructuredAgentSessionHost { }) } - private now(): number { - return this.deps.now?.() ?? Date.now() - } + private now = (): number => this.deps.now?.() ?? Date.now() hasSession(sessionId: string): boolean { return this.sessions.has(sessionId) @@ -140,9 +144,13 @@ export class StructuredAgentSessionHost { params, now: () => this.now(), onAttached: (attached) => { - this.sessions.set(sessionId, { journal: attached.journal, params }) + const fence = this.deps.store.getRecord(sessionId)?.lease.runtimeFence ?? 0 + const previousFence = this.sessions.get(sessionId)?.fence + this.sessions.set(sessionId, { journal: attached.journal, params, fence }) if (attached.recovery) { - this.subscribers.reset(sessionId, attached.journal, attached.recovery.reset) + this.subscribers.reset(sessionId, attached.journal, attached.recovery.reset, fence) + } else if (previousFence !== undefined && previousFence !== fence) { + this.subscribers.snapshot(sessionId, attached.journal, fence) } else { this.subscribers.publish(sessionId, attached.journal) } @@ -254,9 +262,22 @@ export class StructuredAgentSessionHost { const fence = record.lease.runtimeFence const ctx = this.contextFor(caller, session, envelope.sessionId, fence) if (admission.decision === 'replay') { - const recorded = plan.replay(ctx) - if (recorded) { - return { ok: true, replayed: true, fence, cursor: ctx.journal.cursor(), value: recorded } + const replay = resolveAgentSessionReplayOutcome({ + operationId: envelope.clientOperationId, + outcome: admission.row.outcome, + reconstruct: () => plan.replay(ctx, admission.row.outcome) + }) + if (replay.decision === 'refuse') { + return refuse(replay.refusal) + } + if (replay.decision === 'replay') { + return { + ok: true, + replayed: true, + fence, + cursor: ctx.journal.cursor(), + value: replay.value + } } // Nothing durable landed, so this id is about to run for the first time. // A refused call leaves its ledger row behind, and replaying past the @@ -273,7 +294,13 @@ export class StructuredAgentSessionHost { } } - const outcome = await this.finish(caller, envelope, plan, ctx) + const outcome = await runSettledAgentSessionMutation({ + store: this.deps.store, + callerKey: caller.callerKey, + envelope, + plan, + context: ctx + }) return outcome.ok ? { ok: true, replayed: false, fence, cursor: ctx.journal.cursor(), value: outcome.value } : refuse(outcome.refusal) @@ -296,36 +323,6 @@ export class StructuredAgentSessionHost { } } - /** Runs the effect and settles its ledger row, including on a throw — a row - * left `pending` would replay as unresolvable forever. */ - private async finish( - caller: StructuredAgentSessionCaller, - envelope: AgentSessionMutationEnvelope, - plan: MutationPlan, - ctx: AgentSessionTurnContext - ): Promise> { - const settle = ( - outcome: Parameters[0]['outcome'] - ) => - this.deps.store.recordOperationOutcome({ - callerKey: caller.callerKey, - operationId: envelope.clientOperationId, - outcome - }) - try { - const outcome = await plan.run(ctx) - await settle( - outcome.ok - ? { status: 'succeeded', sessionId: envelope.sessionId } - : { status: 'failed', code: outcome.refusal.code } - ) - return outcome - } catch (error) { - await settle({ status: 'unknown' }) - throw error - } - } - /** * Paged read. Throws `agent_session_ownership_unknown` for a session this host * has not attached — a reset would tell the client to reload something that diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans.ts index 0f101a6ad70..0d8bad1d19e 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans.ts @@ -7,6 +7,7 @@ // which is exactly right when the crash landed before the journal write. import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types' +import type { AgentSessionOperationOutcome } from '../../../shared/agent-session-operation-ledger' import type { AgentSessionCancelResult, AgentSessionMutationEnvelope, @@ -27,7 +28,7 @@ export type MutationPlan = { method: string fields: Record run: (ctx: AgentSessionTurnContext) => Promise> - replay: (ctx: AgentSessionTurnContext) => TValue | null + replay: (ctx: AgentSessionTurnContext, outcome: AgentSessionOperationOutcome) => TValue | null } export function sendPlan(params: { @@ -108,6 +109,9 @@ export function setOptionPlan(params: { method: 'agentSession.setOption', fields: { key: params.key, value: params.value }, run: (ctx) => performSetOption(ctx, params), - replay: () => ({ key: params.key, value: params.value }) + // A pending row may have crashed before the adapter call. Reapplying the + // same assignment is safe; only a settled success can be answered directly. + replay: (_ctx, outcome) => + outcome.status === 'succeeded' ? { key: params.key, value: params.value } : null } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-operation-settlement.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-operation-settlement.ts new file mode 100644 index 00000000000..4cb24518c69 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-operation-settlement.ts @@ -0,0 +1,33 @@ +import type { AgentSessionMutationEnvelope } from '../../../shared/agent-session-wire' +import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' +import type { MutationPlan } from './structured-agent-session-mutation-plans' +import type { AgentSessionTurnContext, TurnOutcome } from './structured-agent-session-turns' + +export async function runSettledAgentSessionMutation(input: { + store: AgentSessionRecordStore + callerKey: string + envelope: AgentSessionMutationEnvelope + plan: MutationPlan + context: AgentSessionTurnContext +}): Promise> { + const settle = ( + outcome: Parameters[0]['outcome'] + ) => + input.store.recordOperationOutcome({ + callerKey: input.callerKey, + operationId: input.envelope.clientOperationId, + outcome + }) + try { + const outcome = await input.plan.run(input.context) + await settle( + outcome.ok + ? { status: 'succeeded', sessionId: input.envelope.sessionId } + : { status: 'failed', code: outcome.refusal.code } + ) + return outcome + } catch (error) { + await settle({ status: 'unknown' }) + throw error + } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-replay-outcome.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-replay-outcome.ts new file mode 100644 index 00000000000..1121dbc8124 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-replay-outcome.ts @@ -0,0 +1,53 @@ +import type { AgentSessionOperationOutcome } from '../../../shared/agent-session-operation-ledger' +import { + AGENT_SESSION_WIRE_REFUSAL_CODES, + type AgentSessionWireRefusal, + type AgentSessionWireRefusalCode +} from '../../../shared/agent-session-wire' + +export type AgentSessionReplayOutcomeDecision = + | { decision: 'replay'; value: TValue } + | { decision: 'rerun' } + | { decision: 'refuse'; refusal: AgentSessionWireRefusal } + +export function resolveAgentSessionReplayOutcome(input: { + operationId: string + outcome: AgentSessionOperationOutcome + reconstruct: () => TValue | null +}): AgentSessionReplayOutcomeDecision { + const { operationId, outcome } = input + if (outcome.status === 'failed') { + const code = (AGENT_SESSION_WIRE_REFUSAL_CODES as readonly string[]).includes(outcome.code) + ? (outcome.code as AgentSessionWireRefusalCode) + : 'agent_session_operation_invalid' + return { + decision: 'refuse', + refusal: { + code, + message: `Operation ${operationId} was already refused: ${outcome.code}.` + } + } + } + if (outcome.status === 'unknown') { + return { + decision: 'refuse', + refusal: { + code: 'agent_session_operation_unknown', + message: `The outcome of operation ${operationId} is unknown; it was not run again.` + } + } + } + const recorded = input.reconstruct() + if (recorded) { + return { decision: 'replay', value: recorded } + } + return outcome.status === 'succeeded' + ? { + decision: 'refuse', + refusal: { + code: 'agent_session_operation_unknown', + message: `Operation ${operationId} succeeded, but its result is no longer reconstructable.` + } + } + : { decision: 'rerun' } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-reconcile.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-reconcile.test.ts new file mode 100644 index 00000000000..ac0ae832c0b --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-reconcile.test.ts @@ -0,0 +1,29 @@ +import { describe, expect, it, vi } from 'vitest' +import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' +import { createRestartReconciler } from './structured-agent-session-restart-reconcile' + +describe('createRestartReconciler', () => { + it('reruns after an external store refresh introduces unreconciled leases', async () => { + let record = { sessionId: 'session-1', lease: { unreconciled: true } } as AgentSessionRecord + const reconcileOnRestart = vi.fn(async () => { + record = { ...record, lease: { ...record.lease, unreconciled: false } } + return new Map() + }) + const store = { + listRecords: () => [record], + getRecord: () => record, + reconcileOnRestart + } as unknown as AgentSessionRecordStore + const reconcile = createRestartReconciler({ + store, + probe: async () => ({ outcome: 'pid-absent' }), + now: () => 1 + }) + + expect(await reconcile('session-1')).toBeNull() + record = { ...record, lease: { ...record.lease, unreconciled: true } } + expect(await reconcile('session-1')).toBeNull() + expect(reconcileOnRestart).toHaveBeenCalledTimes(2) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-reconcile.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-reconcile.ts index 472fe0140f9..895381f4dbf 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-reconcile.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-reconcile.ts @@ -9,9 +9,8 @@ import type { AgentSessionWireRefusal } from '../../../shared/agent-session-wire import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' import { classifyStoreFailure } from './structured-agent-session-attach' -/** Adjudicates the loaded leases once per host, on the first attach. Answers - * with the refusal that attach owes its caller, or null once the leases are - * settled. */ +/** Adjudicates leases loaded by this process or refreshed from another writer. + * Answers with the refusal attach owes its caller, or null once settled. */ export function createRestartReconciler(deps: { store: AgentSessionRecordStore probe: (record: AgentSessionRecord) => Promise @@ -19,15 +18,15 @@ export function createRestartReconciler(deps: { }): (sessionId: string) => Promise { let pending: Promise | null = null return async (sessionId) => { + if (!deps.store.listRecords().some((record) => record.lease.unreconciled)) { + return null + } if (!pending) { const run = deps.store .reconcileOnRestart({ probe: deps.probe, now: deps.now() }) .then(() => undefined) - // A failed adjudication must not be remembered as done, or one unlucky - // startup would strand every persisted session for this host's lifetime. - pending = run.catch((error: unknown) => { + pending = run.finally(() => { pending = null - throw error }) } try { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts index 8c56efb4c18..a3b4e3484ee 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts @@ -19,6 +19,7 @@ type Subscriber = { sessionId: string emit: AgentSessionSubscriberEmit cursor: AgentJournalCursor + fence: number } export class AgentSessionSubscribers { @@ -39,7 +40,8 @@ export class AgentSessionSubscribers { id: input.id, sessionId: input.sessionId, emit: input.emit, - cursor: input.cursor ?? { epoch: snapshot.cursor.epoch, sequence: 0 } + cursor: input.cursor ?? { epoch: snapshot.cursor.epoch, sequence: 0 }, + fence: input.fence } const session = this.bySession.get(input.sessionId) ?? new Map() session.set(input.id, subscriber) @@ -48,7 +50,7 @@ export class AgentSessionSubscribers { if (input.cursor) { this.deliver(subscriber, input.journal) } else { - subscriber.emit({ + this.emit(subscriber, { type: 'snapshot', sessionId: input.sessionId, snapshot, @@ -65,11 +67,12 @@ export class AgentSessionSubscribers { if (!session || !subscriber) { return } - session.delete(id) - if (session.size === 0) { - this.bySession.delete(sessionId) + this.drop(subscriber) + try { + subscriber.emit({ type: 'end' }) + } catch { + // The transport is already gone; teardown must remain idempotent. } - subscriber.emit({ type: 'end' }) } /** Fan out whatever each subscriber has not yet seen. */ @@ -81,11 +84,26 @@ export class AgentSessionSubscribers { /** Force every subscriber back to a clean snapshot — recovery, epoch * rollover, an unreadable schema. */ - reset(sessionId: string, journal: AgentSessionJournal, reason: AgentJournalResetReason): void { + reset( + sessionId: string, + journal: AgentSessionJournal, + reason: AgentJournalResetReason, + fence: number + ): void { const snapshot = journal.snapshot() for (const subscriber of this.subscribers(sessionId)) { - subscriber.emit({ type: 'reset', sessionId, reset: reason, snapshot }) + this.emit(subscriber, { type: 'reset', sessionId, reset: reason, snapshot, fence }) subscriber.cursor = snapshot.cursor + subscriber.fence = fence + } + } + + snapshot(sessionId: string, journal: AgentSessionJournal, fence: number): void { + const snapshot = journal.snapshot() + for (const subscriber of this.subscribers(sessionId)) { + this.emit(subscriber, { type: 'snapshot', sessionId, snapshot, fence }) + subscriber.cursor = snapshot.cursor + subscriber.fence = fence } } @@ -97,11 +115,12 @@ export class AgentSessionSubscribers { const since = journal.readSince(subscriber.cursor) const snapshot = journal.snapshot() if (!since.ok) { - subscriber.emit({ + this.emit(subscriber, { type: 'reset', sessionId: subscriber.sessionId, reset: since.reset, - snapshot + snapshot, + fence: subscriber.fence }) subscriber.cursor = snapshot.cursor return @@ -115,16 +134,39 @@ export class AgentSessionSubscribers { afterSequence: subscriber.cursor.sequence }) if (!projected.ok) { - subscriber.emit({ + this.emit(subscriber, { type: 'reset', sessionId: subscriber.sessionId, reset: projected.reset, - snapshot + snapshot, + fence: subscriber.fence }) subscriber.cursor = snapshot.cursor return } - subscriber.emit({ type: 'batch', sessionId: subscriber.sessionId, batch: projected.batch }) + this.emit(subscriber, { + type: 'batch', + sessionId: subscriber.sessionId, + batch: projected.batch + }) subscriber.cursor = projected.batch.cursor } + + /** A dead transport cannot be allowed to turn a durable mutation into an + * unknown outcome or poison every later publication. */ + private emit(subscriber: Subscriber, event: AgentSessionSubscribeEvent): void { + try { + subscriber.emit(event) + } catch { + this.drop(subscriber) + } + } + + private drop(subscriber: Subscriber): void { + const session = this.bySession.get(subscriber.sessionId) + session?.delete(subscriber.id) + if (session?.size === 0) { + this.bySession.delete(subscriber.sessionId) + } + } } diff --git a/src/main/runtime/rpc/methods/structured-agent-session-schemas.ts b/src/main/runtime/rpc/methods/structured-agent-session-schemas.ts index ea2681df042..d766399202a 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session-schemas.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session-schemas.ts @@ -55,13 +55,6 @@ const ProviderHandle = z.discriminatedUnion('kind', [ sessionId: Identifier('Invalid provider session id'), leafUuid: Identifier('Invalid leaf uuid').nullable() }) - .strict(), - z - .object({ - kind: z.literal('opaque'), - agent: Identifier('Invalid agent'), - value: Identifier('Invalid provider handle') - }) .strict() ]) diff --git a/src/main/runtime/rpc/methods/structured-agent-session.test.ts b/src/main/runtime/rpc/methods/structured-agent-session.test.ts index 110b0d0e390..85460a63448 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session.test.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session.test.ts @@ -232,6 +232,13 @@ describe('parameter validation', () => { ) }) + it('rejects a journal-only opaque provider handle', async () => { + await rejects( + 'agentSession.create', + attachParams({ providerHandle: { kind: 'opaque', agent: 'codex', value: 'thread-1' } }) + ) + }) + it('requires a sha256 fingerprint and a positive fence', async () => { await rejects( 'agentSession.send', diff --git a/src/shared/agent-session-wire.ts b/src/shared/agent-session-wire.ts index fc6982036fe..4dcfe039525 100644 --- a/src/shared/agent-session-wire.ts +++ b/src/shared/agent-session-wire.ts @@ -37,6 +37,8 @@ export type AgentSessionHistoryPage = { epoch: string direction: AgentSessionHistoryDirection items: AgentJournalRenderItem[] + /** Populated by `after` reads so a disconnected client can apply tombstones. */ + removedItemIds: string[] /** Submissions overlapping this page, so an unconfirmed bubble renders with * its dispatch state instead of as a plain message. */ submissions: AgentJournalSubmission[] @@ -75,6 +77,7 @@ export type AgentSessionSubscribeEvent = sessionId: string reset: AgentJournalResetReason snapshot: AgentJournalSnapshot + fence: number } | { type: 'end' } @@ -103,6 +106,7 @@ export const AGENT_SESSION_WIRE_REFUSAL_CODES = [ 'agent_session_operation_expired', 'agent_session_operation_capacity', 'agent_session_operation_invalid', + 'agent_session_operation_unknown', 'agent_session_item_revision_stale', 'agent_session_already_resolved', 'agent_session_identity_required',