diff --git a/config/scripts/vitest-sqlite-runtime-files.mjs b/config/scripts/vitest-sqlite-runtime-files.mjs index f96e2ecada22..834a6811ab77 100644 --- a/config/scripts/vitest-sqlite-runtime-files.mjs +++ b/config/scripts/vitest-sqlite-runtime-files.mjs @@ -2,6 +2,7 @@ import { SQLITE_DATABASE_RUNTIME_INCLUDE } from './vitest-sqlite-database-runtim // Real SQLite fixtures keep Node publication and close semantics. export const SQLITE_RUNTIME_INCLUDE = [ + 'src/main/native-chat/agent-session-wire/structured-agent-session-ready-operation-order.test.ts', ...SQLITE_DATABASE_RUNTIME_INCLUDE, 'src/main/acp/acp-structured-host-live-listing.test.ts', 'src/main/acp/acp-structured-host-configured-default.test.ts', diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-hung-start-stop.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-hung-start-stop.test.ts index 347de6cf6643..db87e619b933 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-hung-start-stop.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-hung-start-stop.test.ts @@ -131,12 +131,10 @@ async function readsWorking(): Promise { ) } -/** Whether the CLI was written `hello`, once the delivery loop has settled what it does with it. */ -async function helloWritten(): Promise { - await vi.waitFor(() => - expect(host.collaboratorsForTests().conversationDelivery.loop.isRunning(SESSION)).toBe(false) - ) - return claude.connections[0]!.sent.some((message) => JSON.stringify(message).includes('hello')) +async function expectHeld(id: string): Promise { + expect(await submission(id)).toMatchObject({ dispatchState: 'pending' }) + expect((await submission(id))?.handedOverAt).toBeUndefined() + expect(claude.connections[0]!.sent).toEqual([]) } it('holds the message from a start that never answers, and a Stop ends it: stopped, nothing working', async () => { @@ -150,7 +148,7 @@ it('holds the message from a start that never answers, and a Stop ends it: stopp const [connection] = claude.connections // Accepted, and held while initialize is still unanswered. - expect(await helloWritten()).toBe(false) + await expectHeld(id) expect(connection!.calls).toEqual([{ subtype: 'initialize' }]) expect(host['sessions'].get(SESSION)?.child?.phase).toBe('starting') expect(await readsWorking()).toBe(true) @@ -182,7 +180,9 @@ async function sendHello(written: boolean): Promise { if (!sent.ok) { throw new Error('send refused') } - await vi.waitFor(async () => expect(await helloWritten()).toBe(written)) + await (written + ? vi.waitFor(() => expect(claude.connections[0]!.sent).toHaveLength(1)) + : expectHeld(sent.value.clientMessageId)) return sent.value.clientMessageId } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-delivery-loop.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-delivery-loop.ts index 1858eaeddf8d..dcfc8b4de1d5 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-delivery-loop.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-delivery-loop.ts @@ -37,6 +37,7 @@ import { handOverSubmission } from './structured-agent-session-turns' import { structuredAgentSessionNextHandover } from './structured-agent-session-opening-send' import { structuredAgentSessionCommandRunning } from './structured-agent-session-command-turn' import type { StructuredAgentSessionLogger } from './structured-agent-session-logger' +import type { StructuredAgentSessionStartReadiness } from './structured-agent-session-startup-attempt' import { holdRestartedStructuredAgentSessionSends } from './structured-agent-session-host-lifetime' import { structuredAgentSessionEndedChildFailure } from './structured-agent-session-ended-child-failure' @@ -53,6 +54,10 @@ export type StructuredAgentSessionDeliveryLoopDeps = { sessionId: string, startedFor: string ) => Promise + readiness: ( + sessionId: string, + child: StructuredAgentSessionProviderChildIdentity + ) => StructuredAgentSessionStartReadiness | null /** The fence the conversation's own writes carry; see `structuredAgentSessionConversationFence`. */ conversationFence: (sessionId: string) => number /** Settles queued messages as a completed close of the chat does; false when that failed. */ @@ -77,7 +82,11 @@ type RefusedStart = Extract type Prepared = | 'stop' | RefusedStart - | { ok: true; awaited: StructuredAgentSessionProviderChildIdentity | null } + | { + ok: true + awaited: StructuredAgentSessionProviderChildIdentity | null + readiness: StructuredAgentSessionStartReadiness | null + } /** A failed start before it is worded; `fail` words it once, through the one wording point. */ type StartFailure = { startKey: string | null; cause: StructuredAgentSessionStartFailureCause } @@ -138,6 +147,9 @@ export class StructuredAgentSessionDeliveryLoop { ) return } + if (prepared.readiness) { + await prepared.readiness.wait() + } const handed = await this.deps.serialize(sessionId, () => this.handOver(sessionId, prepared.awaited) ) @@ -209,7 +221,8 @@ export class StructuredAgentSessionDeliveryLoop { // The child this run waits on; handover checks it is still the one there. return { ok: true, - awaited: child ? { generation: child.generation, fence: child.fence } : null + awaited: child ? { generation: child.generation, fence: child.fence } : null, + readiness: child?.phase === 'starting' ? this.deps.readiness(sessionId, child) : null } } @@ -247,9 +260,7 @@ export class StructuredAgentSessionDeliveryLoop { // Never steer into a turn a person's Stop is ending: the message runs after it, as its own // turn, and the turn's end is a commit that wakes the loop again. Read here, at the handover, // because a Stop can take the lane between the step that judged the send and this one. - // Nor hand anything to a child still proving its start: nothing is written until it is - // `ready`, whose settlement wakes the loop, so a start that fails leaves the message unsent. - if (this.deps.stopping(sessionId) || awaitedChild.phase === 'starting') { + if (this.deps.stopping(sessionId) || awaitedChild.phase !== 'ready') { return this.stop(sessionId) } // None while a turn is still opening: joining it would record the message ahead of it. diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-delivery.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-delivery.ts index bd9102421ed3..344d583d02a0 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-delivery.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-delivery.ts @@ -15,9 +15,11 @@ import type { StructuredAgentSessionClientDelivery } from './structured-agent-se import { StructuredAgentSessionDeliveryLoop } from './structured-agent-session-delivery-loop' import { structuredAgentSessionCommandRunning } from './structured-agent-session-command-turn' import type { StructuredAgentSessionResumeOutcome } from './structured-agent-session-agent-start' +import type { StructuredAgentSessionStartReadiness } from './structured-agent-session-startup-attempt' import type { StructuredAgentSessionHostDeps, - StructuredAgentSessionHostSession + StructuredAgentSessionHostSession, + StructuredAgentSessionProviderChildIdentity } from './structured-agent-session-host-types' import { structuredAgentSessionConversationFence } from './structured-agent-session-provider-child' import { structuredAgentSessionFailureWordsContext } from './structured-agent-session-send-preparation' @@ -57,6 +59,10 @@ export function createStructuredAgentSessionConversationDelivery(input: { sessionId: string, startedFor: string ) => Promise + readiness: ( + sessionId: string, + child: StructuredAgentSessionProviderChildIdentity + ) => StructuredAgentSessionStartReadiness | null /** Stops a child that reported it is not signed in, so the start after it reads a new login. */ stopSignedOutAgent: (sessionId: string) => Promise clientDelivery: Pick< @@ -76,7 +82,7 @@ export function createStructuredAgentSessionConversationDelivery(input: { const session = input.clientDelivery.readStopping(sessionId) ? undefined : sessions.get(sessionId) - await retireSignedOutStructuredAgentSessionChild(sessionId, session, startedFor, { + await retireSignedOutStructuredAgentSessionChild(sessionId, session, { work: { childWork: () => input.clientDelivery.readChildWork(sessionId), hasOpenDispatch: () => { @@ -91,6 +97,7 @@ export function createStructuredAgentSessionConversationDelivery(input: { }) return input.ensureProviderChild(sessionId, startedFor) }, + readiness: input.readiness, conversationFence: (sessionId) => structuredAgentSessionConversationFence(deps.store, sessionId), holdClosed: async (sessionId, which) => { 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 9d61a02a1dce..429be48b4334 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 @@ -22,11 +22,11 @@ import type { AgentSessionPromptRequest } from './structured-agent-session-turns import { refuseThreadGoalUnsupported, threadGoalPlan } from './structured-agent-session-thread-goal' import { mutateStructuredAgentSession, + runStructuredAgentSessionMutation, type StructuredAgentSessionMutationContext } from './structured-agent-session-mutation-context' import { openForProviderWrite, - openWithAgent, sendPreparation, structuredAgentSessionSendBlock } from './structured-agent-session-send-preparation' @@ -41,7 +41,11 @@ import { agentSessionMutationAdmitsNow } from './structured-agent-session-mutati import { runQueueableStructuredAgentSessionSend } from './structured-agent-session-queued-send' import { cancelStructuredAgentSessionPrompt } from './structured-agent-session-prompt-cancel' import { mutateWithChatStop } from './structured-agent-session-chat-stop' -import { runAfterProviderStart } from './structured-agent-session-provider-start-hold' +import { + runWithReadyStructuredAgentSessionOperation, + openWithReadyStructuredAgentSessionChild, + runAfterRewindStart +} from './structured-agent-session-ready-operation' import { performSetOption } from './structured-agent-session-turns-options' export type { StructuredAgentSessionMutationContext } from './structured-agent-session-mutation-context' import type { StructuredAgentSessionCaller } from './structured-agent-session-host-types' @@ -74,8 +78,8 @@ export function sendStructuredAgentSessionTurn( arrival?: Parameters[2] ): Promise> { const plan = sendPlan(params) - return runAfterProviderStart(context, params.envelope.sessionId, () => - mutateStructuredAgentSession( + return runAfterRewindStart(context, caller.callerKey, params.envelope, plan, (selection) => + runStructuredAgentSessionMutation( context, caller, params.envelope, @@ -91,7 +95,7 @@ export function sendStructuredAgentSessionTurn( (await plan.run(ctx)) ) }, - sendPreparation(context, params.envelope, arrival) + sendPreparation(context, params.envelope, arrival, selection) ) ) } @@ -217,22 +221,29 @@ export async function setStructuredAgentSessionOption( ) } -export function changeStructuredAgentSessionThreadGoal( +export async function changeStructuredAgentSessionThreadGoal( context: StructuredAgentSessionMutationContext, caller: StructuredAgentSessionCaller, params: { envelope: AgentSessionMutationEnvelope; change: AgentSessionThreadGoalChange } ): Promise> { - return runAfterProviderStart(context, params.envelope.sessionId, () => - mutateStructuredAgentSession( - context, - caller, - params.envelope, - threadGoalPlan(params), - // An agent without goals answers so at once, with no start to wait out. - openWithAgent(context, params.envelope, (record) => + const plan = threadGoalPlan(params) + return runWithReadyStructuredAgentSessionOperation( + context, + caller.callerKey, + params.envelope, + plan, + (selection) => + runStructuredAgentSessionMutation( + context, + caller, + params.envelope, + plan, + openWithReadyStructuredAgentSessionChild(context, params.envelope, selection) + ), + { + beforeAgent: (record) => refuseThreadGoalUnsupported(context.deps.adapter, context.deps.agents, record?.provider) - ) - ) + } ) } 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 8680110d9148..44deeb83a694 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 @@ -123,6 +123,8 @@ export class StructuredAgentSessionHost { trackStart: (start) => this.tasks.trackAttach(start), ensureProviderChild: (sessionId, startedFor) => agentStart.ensureStructuredAgentSessionAgent(this.attachContext(), sessionId, startedFor), + readiness: (sessionId, child) => + this.runtimeState.startupAttempts.readiness(sessionId, child), // Put to rest like an idle agent: the stop writes nothing, and the next start is new. stopSignedOutAgent: (id) => this.lifetime.stopAgent(id, { cause: 'evict', resting: true }), clientDelivery: this.clientDelivery @@ -280,6 +282,7 @@ export class StructuredAgentSessionHost { openConversation: this.conversationDelivery.open, ensureAgent: (sessionId) => agentStart.ensureStructuredAgentSessionAgentForOperation(this.attachContext(), sessionId), + startupAttempts: this.runtimeState.startupAttempts, joinChildClose: (sessionId) => agentStart.joinClosingStructuredAgentSessionChild(this.attachContext(), sessionId), wakeDelivery: (sessionId) => this.conversationDelivery.loop.wake(sessionId), diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-context.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-context.test.ts index 99120dcc23b5..6422b4b8437e 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-context.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-context.test.ts @@ -195,29 +195,48 @@ it('logs a wake error and still answers accepted', async () => { ) }) -it('logs a receipt read error and refuses unknown without accepting or waking', async () => { - const ctx = context() - const request = params() - const error = new Error('receipt read failed') - const warn = vi.spyOn(ctx.deps.logger, 'warn') - vi.spyOn(rig.store, 'readCommandReceipt').mockImplementationOnce(() => { - throw error - }) - expect(await sendStructuredAgentSessionTurn(ctx, CALLER, request)).toMatchObject({ - ok: false, - refusal: { code: 'agent_session_operation_unknown', details: { reason: 'outcomeUnknown' } } - }) - expect(warn).toHaveBeenCalledWith( - expect.any(String), - expect.objectContaining({ - scope: 'command-receipt-read', - operationId: request.envelope.clientOperationId, - error +it.each([false, true])( + 'logs a receipt read error and refuses unknown without accepting or waking (rewind: %s)', + async (rewinding) => { + const ctx = context() + const request = params() + if (rewinding) { + await rig.store.transitionHandoff(SESSION, (record) => ({ + ...record, + rewind: { + operationId: hostTestOperationId(), + callerKey: CALLER.callerKey, + itemId: 'prior-message', + providerItemId: 'prior-provider-item', + expectedEpoch: rig.host['sessions'].get(SESSION)!.journal.cursor().epoch, + phase: 'prepared', + retained: [] + } + })) + } + const ensureAgent = vi.spyOn(ctx, 'ensureAgent') + const error = new Error('receipt read failed') + const warn = vi.spyOn(ctx.deps.logger, 'warn') + vi.spyOn(rig.store, 'readCommandReceipt').mockImplementationOnce(() => { + throw error }) - ) - expect(rig.host['sessions'].get(SESSION)?.journal.submissions()).toHaveLength(0) - expect(ctx.wakeDelivery).not.toHaveBeenCalled() -}) + expect(await sendStructuredAgentSessionTurn(ctx, CALLER, request)).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_operation_unknown', details: { reason: 'outcomeUnknown' } } + }) + expect(warn).toHaveBeenCalledWith( + expect.any(String), + expect.objectContaining({ + scope: 'command-receipt-read', + operationId: request.envelope.clientOperationId, + error + }) + ) + expect(rig.host['sessions'].get(SESSION)?.journal.submissions()).toHaveLength(0) + expect(ctx.wakeDelivery).not.toHaveBeenCalled() + expect(ensureAgent).not.toHaveBeenCalled() + } +) it('wakes once for /compact and never for its replay', async () => { const ctx = context() diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-context.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-context.ts index f9b83c3b4f3a..8a077cbfe2dc 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-context.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-context.ts @@ -15,6 +15,7 @@ import type { MutationPlan } from './structured-agent-session-mutation-plans' import type { StructuredAgentSessionStopEnding } from './structured-agent-session-host-lifetime' import type { StructuredAgentSessionAcquireAborts } from './structured-agent-session-acquire-aborts' import type { StructuredAgentSessionOptionRevisions } from './structured-agent-session-option-revisions' +import type { StructuredAgentSessionStartupAttempts } from './structured-agent-session-startup-attempt' import type { StructuredAgentSessionCaller, StructuredAgentSessionHostDeps, @@ -34,6 +35,7 @@ export type StructuredAgentSessionMutationContext = { openConversation: (sessionId: string) => Promise /** Gives the session a provider child; inside the caller's serialize. */ ensureAgent: (sessionId: string) => Promise + startupAttempts: Pick /** Joins a close a stop began on the session's child, for an operation that starts no child; * inside the caller's serialize. */ joinChildClose: (sessionId: string) => Promise @@ -63,19 +65,30 @@ export function mutateStructuredAgentSession( prepareSession?: AgentSessionMutationRequest['prepareSession'] ): Promise> { return context.serialize(envelope.sessionId, () => - admitAndRunAgentSessionMutation({ - store: context.deps.store, - adapter: context.deps.adapter, - agents: context.deps.agents, - logger: context.deps.logger, - callerKey: caller.callerKey, - envelope, - plan, - journal: () => context.sessions.get(envelope.sessionId)?.journal, - prepareSession, - publish: (journal) => context.publish(envelope.sessionId, journal), - wakeDelivery: (sessionId) => context.wakeDelivery(sessionId), - now: () => context.now() - }) + runStructuredAgentSessionMutation(context, caller, envelope, plan, prepareSession) ) } + +/** Admission for a caller already inside the session's queue. */ +export function runStructuredAgentSessionMutation( + context: StructuredAgentSessionMutationContext, + caller: StructuredAgentSessionCaller, + envelope: AgentSessionMutationEnvelope, + plan: MutationPlan, + prepareSession?: AgentSessionMutationRequest['prepareSession'] +): Promise> { + return admitAndRunAgentSessionMutation({ + store: context.deps.store, + adapter: context.deps.adapter, + agents: context.deps.agents, + logger: context.deps.logger, + callerKey: caller.callerKey, + envelope, + plan, + journal: () => context.sessions.get(envelope.sessionId)?.journal, + prepareSession, + publish: (journal) => context.publish(envelope.sessionId, journal), + wakeDelivery: (sessionId) => context.wakeDelivery(sessionId), + now: () => context.now() + }) +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-owed-work-release.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-owed-work-release.test.ts index 6d188fb3b385..1ad43f3144de 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-owed-work-release.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-owed-work-release.test.ts @@ -175,24 +175,19 @@ function waitOutSeveralSweeps(): Promise { return new Promise((resolve) => setTimeout(resolve, SWEEP_MS * 20)) } -function deliverySettled(): Promise { - return vi.waitFor(() => - expect(host.collaboratorsForTests().conversationDelivery.loop.isRunning(SESSION)).toBe(false) - ) -} - describe('a Claude chat whose CLI is still starting', () => { it('holds a message until startup lands, and keeps the agent inside the idle window meanwhile', async () => { await attachStarting() const sent = await send('sent while starting') // Held, never written, while initialize is unanswered. - await deliverySettled() expect(claude.connections[0].sent).toEqual([]) expect(await dispatchState(sent)).toBe('pending') clock += IDLE_MS - 1 await waitOutSeveralSweeps() + expect(claude.connections[0].sent).toEqual([]) + expect(await dispatchState(sent)).toBe('pending') expect(host.hasSession(SESSION)).toBe(true) expect(claude.connections[0].closeCount).toBe(0) @@ -212,10 +207,12 @@ describe('a Claude chat whose CLI is still starting', () => { connection.sent.push(message) } const held = await send('sent while starting') - await deliverySettled() expect(connection.sent).toEqual([]) + expect(await dispatchState(held)).toBe('pending') clock += IDLE_MS - 1 await waitOutSeveralSweeps() + expect(connection.sent).toEqual([]) + expect(await dispatchState(held)).toBe('pending') expect(host.hasSession(SESSION)).toBe(true) // Startup lands just before the window closes and the message is handed over; the agent stays diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-provider-child-record.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-provider-child-record.test.ts index c0037c442d9c..9406290cd51a 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-provider-child-record.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-provider-child-record.test.ts @@ -594,8 +594,11 @@ describe('another child indexed after a Stop ended the starting one', () => { const replacement = currentChild() expect(replacement.acquisitionGeneration).not.toBe(stopped.acquisitionGeneration) const secondId = await accept('second') - await settleLoop() expect((await submission(secondId))?.handedOverAt).toBeUndefined() + expect(dispatch).not.toHaveBeenCalled() + await host.flushStreamedEvents(SESSION) + expect((await submission(secondId))?.handedOverAt).toBeUndefined() + expect(dispatch).not.toHaveBeenCalled() await prove() diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-provider-child-start-wait.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-provider-child-start-wait.test.ts deleted file mode 100644 index adc87fa88d25..000000000000 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-provider-child-start-wait.test.ts +++ /dev/null @@ -1,51 +0,0 @@ -// A wait on a child's start follows that child, not the object the host keeps for it: a re-attach -// rebuilds the same child, and only a different child ends the wait. - -import { describe, expect, it } from 'vitest' -import { - indexProviderChild, - markProviderChildStarted, - providerChildStartSettled -} from './structured-agent-session-provider-child' -import type { StructuredAgentSessionProviderChild } from './structured-agent-session-host-types' - -function bearer(child: StructuredAgentSessionProviderChild) { - return { child, journal: { cursor: () => ({ epoch: 'e1', sequence: 0 }) } } -} - -const starting = (generation: string): StructuredAgentSessionProviderChild => ({ - generation, - fence: 1, - phase: 'starting' -}) - -describe('a start wait across a re-index of the child', () => { - it('stays with the same child rebuilt by a re-attach, and settles ready once it starts', async () => { - const session = bearer(starting('gen-1')) - const wait = providerChildStartSettled(session) - const rebuilt = starting('gen-1') - - indexProviderChild(session, rebuilt) - markProviderChildStarted(session, rebuilt) - - expect(await wait).toBe('ready') - }) - - it('settles ready when the rebuilt child has already proven its start', async () => { - const session = bearer(starting('gen-1')) - const wait = providerChildStartSettled(session) - - indexProviderChild(session, { ...starting('gen-1'), phase: 'ready' }) - - expect(await wait).toBe('ready') - }) - - it('settles ended when a different child replaces the one waited on', async () => { - const session = bearer(starting('gen-1')) - const wait = providerChildStartSettled(session) - - indexProviderChild(session, starting('gen-2')) - - expect(await wait).toBe('ended') - }) -}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-provider-child.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-provider-child.ts index e123109b4d23..978d65d055bf 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-provider-child.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-provider-child.ts @@ -29,81 +29,14 @@ export function structuredAgentSessionConversationFence( return store.getRecord(sessionId)?.lease.runtimeFence ?? 0 } -/** How a start someone waited on ended: proven, or the child is gone. */ -export type StructuredAgentSessionChildStartOutcome = 'ready' | 'ended' - -const startWaits = new WeakMap< - StructuredAgentSessionProviderChild, - PromiseWithResolvers ->() - -function settleStartWait( - child: StructuredAgentSessionProviderChild, - outcome: StructuredAgentSessionChildStartOutcome -): void { - startWaits.get(child)?.resolve(outcome) - startWaits.delete(child) -} - -/** Settles when the session's child leaves `starting`; at once when it has none starting. Every - * move out of `starting` is made in this file, so none can leave a waiter behind. */ -export function providerChildStartSettled( - session: Pick | undefined -): Promise { - const child = session?.child - if (!child) { - return Promise.resolve('ended') - } - if (child.phase !== 'starting') { - return Promise.resolve('ready') - } - let wait = startWaits.get(child) - if (!wait) { - wait = Promise.withResolvers() - startWaits.set(child, wait) - } - return wait.promise -} - /** For the end of a successful attach only: a failed one never wrote a child to take back. */ export function indexProviderChild( session: ChildBearer, child: StructuredAgentSessionProviderChild ): void { - const previous = session.child - if (previous && previous !== child) { - if (sameProviderChild(previous, child)) { - carryStartWait(previous, child) - } else { - settleStartWait(previous, 'ended') - } - } session.child = child } -/** A re-attach rebuilds the same child (a second window, a phone, a reconnect): its start is still - * the one waited on. */ -function carryStartWait( - from: StructuredAgentSessionProviderChild, - to: StructuredAgentSessionProviderChild -): void { - const wait = startWaits.get(from) - startWaits.delete(from) - if (!wait) { - return - } - if (to.phase !== 'starting') { - wait.resolve('ready') - return - } - const existing = startWaits.get(to) - if (existing) { - void existing.promise.then(wait.resolve) - } else { - startWaits.set(to, wait) - } -} - export function markProviderChildStarted( session: ChildBearer, identity: StructuredAgentSessionProviderChildIdentity @@ -111,7 +44,6 @@ export function markProviderChildStarted( const child = matchingChild(session, identity) if (child) { child.phase = 'ready' - settleStartWait(child, 'ready') } return child !== null } @@ -159,7 +91,6 @@ export function endProviderChild( return false } session.child = null - settleStartWait(child, 'ended') session.lastEndedChild = { ...ended, ...(child.startedFor === undefined ? {} : { startedFor: child.startedFor }), diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-provider-start-hold.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-provider-start-hold.ts deleted file mode 100644 index 425d4e64766d..000000000000 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-provider-start-hold.ts +++ /dev/null @@ -1,90 +0,0 @@ -// One rule for every agent: an operation only the agent's protocol session can perform (a rewind, a -// goal, the rewind recovery a send needs) never reaches a child still proving its start, which has -// no protocol session yet. Its preparation answers `providerStarting` while the child starts; the -// call then waits here, outside the session's queue, because the start's own step runs in that -// queue, and is decided afresh once the start is proven. A start that ends unproven fails the call. - -import { - refuse, - refuseUnclassified, - type AgentSessionMutationResult, - type AgentSessionWireRefusal -} from '../../../shared/agent-session-wire' -import { agentSessionWriteNoticeEnglish } from '../../../shared/agent-session-refusal-notice' -import { - providerChildStartSettled, - type StructuredAgentSessionChildStartOutcome -} from './structured-agent-session-provider-child' -import type { - StructuredAgentSessionHostDeps, - StructuredAgentSessionHostSession -} from './structured-agent-session-host-types' -import { STRUCTURED_AGENT_SESSION_STARTUP_CEILING_MS } from './structured-agent-session-startup-attempt-contract' - -const PROVIDER_STARTING: AgentSessionWireRefusal = refuse( - 'agent_session_operation_invalid', - { reason: 'providerStarting' }, - 'The agent is still starting.' -) - -/** The refusal a preparation answers with while the session's child has not proven its start. */ -export function refuseWhileProviderStarting( - session: Pick | undefined -): { ok: false; refusal: AgentSessionWireRefusal } | null { - return session?.child?.phase === 'starting' ? { ok: false, refusal: PROVIDER_STARTING } : null -} - -/** The startup limit always ends a start; the ceiling only bounds a wait nothing else would. */ -async function awaitProviderChildStart( - session: Pick | undefined, - ceilingMs: number = STRUCTURED_AGENT_SESSION_STARTUP_CEILING_MS -): Promise { - let timer: ReturnType | undefined - const timedOut = new Promise<'timeout'>((resolve) => { - timer = setTimeout(() => resolve('timeout'), ceilingMs) - timer.unref?.() - }) - try { - return await Promise.race([providerChildStartSettled(session), timedOut]) - } finally { - clearTimeout(timer) - } -} - -function heldForStart(result: AgentSessionMutationResult): boolean { - return !result.ok && result.refusal.details?.reason === 'providerStarting' -} - -/** Runs `run`, and once more after the start it found under way is proven. Never called inside the - * session's queue. A second start found starting is refused, never waited on again. */ -export async function runAfterProviderStart( - context: { - sessions: ReadonlyMap> - deps: Pick - }, - sessionId: string, - run: () => Promise> -): Promise> { - const first = await run() - if (!heldForStart(first)) { - return first - } - const started = await awaitProviderChildStart( - context.sessions.get(sessionId), - context.deps.startupLimits?.ceilingMs - ) - if (started === 'timeout') { - return first - } - if (started === 'ended') { - // The start's own failure is the chat's to show; the call says only that it did not start. - return { - ok: false, - refusal: refuseUnclassified( - 'agent_session_owner_restart_failed', - agentSessionWriteNoticeEnglish(['restartFailed']) - ) - } - } - return run() -} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-ready-operation-order.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-ready-operation-order.test.ts new file mode 100644 index 000000000000..cbf4f840191f --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-ready-operation-order.test.ts @@ -0,0 +1,113 @@ +import { expect, it, vi } from 'vitest' +import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key' +import { + CALLER, + attach, + envelope, + hostTestState +} from './structured-agent-session-host-test-harness' +import { + HOST_TEST_SESSION as SESSION, + hostTestMessage +} from './structured-agent-session-host-test-data' + +it.each(['send', 'compact'] as const)( + 'answers an accepted %s receipt during an unconfirmed rewind without starting an agent', + async (command) => { + const { host, store, acquire, dispatch } = hostTestState() + const compact = vi.fn(async () => ({ state: 'admitted' as const })) + host.deps.adapter.compact = compact + const providerWrites = command === 'send' ? dispatch : compact + await attach() + const body = hostTestMessage('already accepted') + const params = { + envelope: envelope( + command === 'send' ? 'agentSession.send' : 'agentSession.conversationCommand', + command === 'send' ? { body } : { command: 'compact' } + ), + body, + command: 'compact' as const + } + const run = (caller = CALLER) => + command === 'send' ? host.send(caller, params) : host.conversationCommand(caller, params) + expect(await run()).toMatchObject({ ok: true }) + await vi.waitFor(() => expect(providerWrites).toHaveBeenCalledTimes(1)) + await host.close(SESSION, 'evict') + const expectedEpoch = (await host.journalSnapshot(SESSION)).cursor.epoch + await store.transitionHandoff(SESSION, (record) => ({ + ...record, + rewind: { + operationId: envelope('agentSession.rewind', {}).clientOperationId, + callerKey: CALLER.callerKey, + itemId: agentJournalSubmissionKey(params.envelope.clientOperationId), + providerItemId: 'codex:thread-1:turn-1:1', + expectedEpoch, + phase: 'prepared', + retained: [] + } + })) + acquire.mockRejectedValueOnce(new Error('restart failed')) + + expect(await run({ callerKey: 'reconnecting-phone' })).toMatchObject({ + ok: true, + replayed: true + }) + expect(acquire).toHaveBeenCalledTimes(1) + expect(providerWrites).toHaveBeenCalledTimes(1) + } +) + +it.each(['goal', 'rewind'] as const)( + 'runs a %s on a ready child before a later send is accepted', + async (operation) => { + const { host, dispatch } = hostTestState() + await attach() + const body = hostTestMessage('first turn') + const first = { envelope: envelope('agentSession.send', { body }), body } + expect(await host.send(CALLER, first)).toMatchObject({ ok: true }) + await vi.waitFor(() => expect(dispatch).toHaveBeenCalledTimes(1)) + await host.flushStreamedEvents(SESSION) + const order: string[] = [] + host.deps.adapter.changeThreadGoal = vi.fn(async () => { + order.push('goal') + return { ok: true as const } + }) + host.deps.adapter.rewindSupport = () => ({ supported: true }) + host.deps.adapter.rewind = vi.fn(async () => { + order.push('rewind') + return { ok: false as const, reason: 'history-not-paginated' as const } + }) + const itemId = agentJournalSubmissionKey(first.envelope.clientOperationId) + const expectedEpoch = (await host.journalSnapshot(SESSION)).cursor.epoch + const change = { kind: 'set' as const, objective: 'Apply before the message' } + const blocked = Promise.withResolvers() + void host.collaboratorsForTests().serialize(SESSION, () => blocked.promise) + const requested = + operation === 'goal' + ? host.changeThreadGoal(CALLER, { + envelope: envelope('agentSession.threadGoal', { change }), + change + }) + : host.rewind(CALLER, { + envelope: envelope('agentSession.rewind', { itemId, expectedEpoch }), + itemId, + expectedEpoch + }) + const laterBody = hostTestMessage('later message') + const sent = host.send(CALLER, { + envelope: envelope('agentSession.send', { body: laterBody }), + body: laterBody, + beforeRun: () => order.push('send') + }) + blocked.resolve() + + const result = await requested + expect(result).toMatchObject( + operation === 'goal' + ? { ok: true } + : { ok: false, refusal: { rewindReason: 'history-not-paginated' } } + ) + expect(await sent).toMatchObject({ ok: true }) + expect(order).toEqual([operation, 'send']) + } +) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-ready-operation.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-ready-operation.ts new file mode 100644 index 000000000000..77e8a994a914 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-ready-operation.ts @@ -0,0 +1,237 @@ +import { + agentSessionFingerprintConflict, + computeAgentSessionPayloadFingerprint +} from '../../../shared/agent-session-mutation-envelope' +import { refuse, refuseUnclassified } from '../../../shared/agent-session-wire-refusals' +import type { + AgentSessionMutationEnvelope, + AgentSessionMutationResult, + AgentSessionWireRefusal +} from '../../../shared/agent-session-wire' +import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import type { StructuredAgentSessionMutationContext } from './structured-agent-session-mutation-context' +import type { StructuredAgentSessionProviderChildIdentity } from './structured-agent-session-host-types' +import { sameProviderChild } from './structured-agent-session-provider-child' +import { STRUCTURED_AGENT_SESSION_START_WAIT_MS } from './structured-agent-session-send-settlement' +import { + openForWrite, + openForProviderWrite, + rewindInDoubt +} from './structured-agent-session-send-preparation' +import { agentSessionOperationOutcomeUnknown } from './structured-agent-session-replay-outcome' +import type { AgentSessionMutationSessionPreparation } from './structured-agent-session-mutation-admission' +import type { MutationPlan } from './structured-agent-session-mutation-plans' +import { commandReceiptScope } from '../agent-session-journal/command-receipt-schema' +import { parseAgentSessionOperationTimestamp } from '../../../shared/agent-session-host-authority' +import type { StructuredAgentSessionStartReadiness } from './structured-agent-session-startup-attempt' + +type ReadyOperationPlan = Pick< + MutationPlan, + 'method' | 'fields' | 'operationIdScope' | 'acceptsWithCommandReceipt' +> +type QueuedReadyOperation = + | { result: AgentSessionMutationResult } + | { + child: StructuredAgentSessionProviderChildIdentity + readiness: StructuredAgentSessionStartReadiness + } + +export type ReadyOperationChild = StructuredAgentSessionProviderChildIdentity | null +export type ReadyOperationSelection = { + ok: true + child: ReadyOperationChild + replay?: true + deadline?: number +} + +/** A send or conversation command waits only when an earlier rewind needs its provider. */ +export async function runAfterRewindStart( + context: StructuredAgentSessionMutationContext, + callerKey: string, + envelope: AgentSessionMutationEnvelope, + plan: ReadyOperationPlan, + run: (selection: ReadyOperationSelection) => Promise> +): Promise> { + return runWithReadyStructuredAgentSessionOperation(context, callerKey, envelope, plan, run, { + needsAgent: rewindInDoubt + }) +} + +export function operationDidNotRun(expired = false): { + ok: false + refusal: AgentSessionWireRefusal +} { + return { + ok: false, + refusal: expired + ? refuse( + 'agent_session_operation_expired', + { reason: 'operationExpired' }, + 'The operation did not run because its wait expired. Try again.' + ) + : refuseUnclassified( + 'agent_session_owner_restart_failed', + 'The operation did not run because the agent is not ready.' + ) + } +} + +/** A ready child runs in this queue turn; only a starting child releases it to wait. */ +export async function runWithReadyStructuredAgentSessionOperation( + context: StructuredAgentSessionMutationContext, + callerKey: string, + envelope: AgentSessionMutationEnvelope, + plan: ReadyOperationPlan, + run: (selection: ReadyOperationSelection) => Promise>, + options: { + beforeAgent?: ( + record: AgentSessionRecord | null + ) => AgentSessionMutationSessionPreparation | null + needsAgent?: (record: AgentSessionRecord | null) => boolean + } = {} +): Promise> { + const hostFingerprint = computeAgentSessionPayloadFingerprint({ + method: plan.method, + sessionId: envelope.sessionId, + fields: plan.fields + }) + const conflict = agentSessionFingerprintConflict(envelope, hostFingerprint) + if (conflict) { + return { ok: false, refusal: conflict } + } + const deadline = Date.now() + STRUCTURED_AGENT_SESSION_START_WAIT_MS + const selected = await context.serialize>( + envelope.sessionId, + async () => { + const record = context.deps.store.getRecord(envelope.sessionId) + if (options.needsAgent && !options.needsAgent(record)) { + return { result: await run({ ok: true, child: null }) } + } + if (plan.acceptsWithCommandReceipt) { + if (parseAgentSessionOperationTimestamp(envelope.clientOperationId) === null) { + return { result: await run({ ok: true, child: null }) } + } + // Receipt admission owns replay and unreadable-result answers, before any start. + let recorded: boolean + try { + recorded = + context.deps.store.readCommandReceipt( + commandReceiptScope(callerKey, plan.operationIdScope), + envelope.clientOperationId + ).verdict !== 'absent' + } catch (error) { + context.deps.logger.warn('reading a command receipt failed', { + scope: 'command-receipt-read', + sessionId: envelope.sessionId, + operationId: envelope.clientOperationId, + error + }) + return { + result: { + ok: false as const, + refusal: agentSessionOperationOutcomeUnknown(envelope.clientOperationId) + } + } + } + if (recorded) { + return { result: await run({ ok: true, child: null }) } + } + } else { + const evaluated = context.deps.store.evaluateMutationOperation({ + callerKey, + envelope, + hostFingerprint, + now: context.now(), + ...(plan.operationIdScope ? { operationIdScope: plan.operationIdScope } : {}) + }) + if (!evaluated || evaluated.decision.decision !== 'admit') { + return { + result: await run({ + ok: true, + child: null, + ...(evaluated?.decision.decision === 'replay' ? { replay: true as const } : {}) + }) + } + } + } + const select = async () => { + const opened = await openForWrite(context, envelope)() + if (!opened.ok) { + return opened + } + const checked = options.beforeAgent?.(context.deps.store.getRecord(envelope.sessionId)) + if (checked && !checked.ok) { + return checked + } + const ensured = await context.ensureAgent(envelope.sessionId) + if (!ensured.ok) { + return ensured + } + const child = context.sessions.get(envelope.sessionId)?.child + if (!child || child.close) { + return operationDidNotRun() + } + const identity = { generation: child.generation, fence: child.fence } + if (child.phase === 'ready') { + return { ok: true as const, child: identity, deadline } + } + const readiness = context.startupAttempts.readiness(envelope.sessionId, identity) + return readiness ? { ok: true as const, child: identity, readiness } : operationDidNotRun() + } + const child = await select().catch((error: unknown) => { + context.deps.logger.warn('selecting an agent for an operation failed', { + scope: 'operation-agent-start', + sessionId: envelope.sessionId, + error + }) + return operationDidNotRun() + }) + if (!child.ok) { + return { result: child } + } + if ('readiness' in child && child.readiness) { + return { child: child.child, readiness: child.readiness } + } + return { result: await run(child) } + } + ) + if ('result' in selected) { + return selected.result + } + const outcome = await selected.readiness.wait(deadline - Date.now()) + if (outcome !== 'ready') { + return operationDidNotRun(outcome === 'timeout') + } + if (Date.now() >= deadline) { + return operationDidNotRun(true) + } + return context.serialize(envelope.sessionId, () => + run({ ok: true, child: selected.child, deadline }) + ) +} + +/** Replay uses the ledger answer; a new operation uses only its selected ready child. */ +export function openWithReadyStructuredAgentSessionChild( + context: StructuredAgentSessionMutationContext, + envelope: AgentSessionMutationEnvelope, + selection: ReadyOperationSelection +): (ledger: 'admit' | 'replay') => Promise { + return async (ledger) => { + const opened = await openForProviderWrite(context, envelope)() + if (!opened.ok || ledger === 'replay') { + return opened + } + if (!selection.child) { + return selection.replay + ? { ok: false, refusal: agentSessionOperationOutcomeUnknown(envelope.clientOperationId) } + : operationDidNotRun() + } + if (selection.deadline !== undefined && Date.now() >= selection.deadline) { + return operationDidNotRun(true) + } + const child = context.sessions.get(envelope.sessionId)?.child + return child?.phase === 'ready' && !child.close && sameProviderChild(child, selection.child) + ? opened + : operationDidNotRun() + } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-rewind.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-rewind.ts index 8c92ddd51246..18f54eba1b4a 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-rewind.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-rewind.ts @@ -15,7 +15,11 @@ import type { import type { AgentSessionMutationResult } from '../../../shared/agent-session-wire' import { AGENT_SESSION_HISTORY_MAX_PAGE_BYTES } from './agent-session-history-page-bounds' import type { StructuredAgentSessionMutationContext } from './structured-agent-session-host-mutations' -import { openWithAgent } from './structured-agent-session-send-preparation' +import { + runWithReadyStructuredAgentSessionOperation, + openWithReadyStructuredAgentSessionChild, + type ReadyOperationSelection +} from './structured-agent-session-ready-operation' import type { StructuredAgentSessionAttachContext } from './structured-agent-session-attach-context' import type { StructuredAgentSessionCaller } from './structured-agent-session-host-types' import { admitAndRunAgentSessionMutation } from './structured-agent-session-mutation-admission' @@ -24,21 +28,9 @@ import { rewindRefusal, rewindRefusalBehindClear } from './structured-rewind-ref import { renameRewindTurnOpener } from './structured-rewind-journal-body' import { persistRewindRecord, recoverStructuredRewind } from './structured-rewind-recovery' import { mergeRetainedHostLifecycleRows } from './structured-rewind-retained-host-rows' -import { runAfterProviderStart } from './structured-agent-session-provider-start-hold' - -export function rewindStructuredAgentSession( - context: StructuredAgentSessionMutationContext, - attachContext: StructuredAgentSessionAttachContext, - caller: StructuredAgentSessionCaller, - params: AgentSessionRewindParams -): Promise> { - // Only a proven child has the protocol session a rewind reads and changes. - return runAfterProviderStart(context, params.envelope.sessionId, () => - rewindUnderSerialize(context, attachContext, caller, params) - ) -} +import { rewindInDoubt } from './structured-agent-session-send-preparation' -function rewindUnderSerialize( +export async function rewindStructuredAgentSession( context: StructuredAgentSessionMutationContext, attachContext: StructuredAgentSessionAttachContext, caller: StructuredAgentSessionCaller, @@ -46,7 +38,9 @@ function rewindUnderSerialize( ): Promise> { const { sessionId, clientOperationId } = params.envelope const store = context.deps.store - return context.serialize(sessionId, async () => { + const fields = { itemId: params.itemId, expectedEpoch: params.expectedEpoch } + const operation = { method: 'agentSession.rewind', fields } + const run = async (selection: ReadyOperationSelection) => { let completionReceipt: JournalOperationReceipt | undefined const result = await admitAndRunAgentSessionMutation({ store, @@ -56,15 +50,12 @@ function rewindUnderSerialize( callerKey: caller.callerKey, envelope: params.envelope, // Only the provider can do this, so an agent at rest is started first. - prepareSession: openWithAgent(context, params.envelope, () => - rewindRefusalBehindClear(context.sessions.get(sessionId)?.journal, params) - ), + prepareSession: openWithReadyStructuredAgentSessionChild(context, params.envelope, selection), journal: () => context.sessions.get(sessionId)?.journal, publish: (journal) => context.publish(sessionId, journal), now: context.now, plan: { - method: 'agentSession.rewind', - fields: { itemId: params.itemId, expectedEpoch: params.expectedEpoch }, + ...operation, recoverUnknownFromDurableState: true, settledOutcome: (rewind: AgentSessionRewindResult) => ({ status: 'succeeded' as const, @@ -99,10 +90,7 @@ function rewindUnderSerialize( if (!support?.supported) { return rewindRefusal(support?.reason ?? 'unsupported') } - if ( - record.rewind?.phase === 'prepared' || - record.rewind?.phase === 'provider-succeeded' - ) { + if (rewindInDoubt(record)) { return rewindRefusal('outcome-unknown') } if (conversationCommandBlocked(ctx, record, context.readChildWork(sessionId))) { @@ -301,5 +289,15 @@ function rewindUnderSerialize( cursor: context.sessions.get(sessionId)!.journal.cursor() } : result - }) + } + return runWithReadyStructuredAgentSessionOperation( + context, + caller.callerKey, + params.envelope, + operation, + run, + { + beforeAgent: () => rewindRefusalBehindClear(context.sessions.get(sessionId)?.journal, params) + } + ) } 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 78e430dfa7fc..3658b3df146d 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 @@ -18,6 +18,10 @@ import { type StructuredAgentSessionAwaitedCommandJournal } from './structured-agent-session-command-turn' import type { StructuredAgentSessionHostSession } from './structured-agent-session-host-types' +import { + operationDidNotRun, + type ReadyOperationSelection +} from './structured-agent-session-ready-operation' import { AGENT_SESSION_NOT_ATTACHED, type AgentSessionMutationSessionPreparation @@ -25,13 +29,15 @@ import { import { agentSessionOperationOutcomeUnknown } from './structured-agent-session-replay-outcome' import { rewindRefusal } from './structured-rewind-refusal' import { recoverStructuredRewind } from './structured-rewind-recovery' -import { refuseWhileProviderStarting } from './structured-agent-session-provider-start-hold' -import { structuredAgentSessionConversationFence } from './structured-agent-session-provider-child' +import { + sameProviderChild, + structuredAgentSessionConversationFence +} from './structured-agent-session-provider-child' import { conversationCommandInFlight } from './structured-conversation-command-admission' import type { StructuredAgentSessionMutationContext } from './structured-agent-session-host-mutations' import type { StructuredAgentSessionLogger } from './structured-agent-session-logger' -function rewindInDoubt(record: AgentSessionRecord | null | undefined): boolean { +export function rewindInDoubt(record: AgentSessionRecord | null | undefined): boolean { const phase = record?.rewind?.phase return phase === 'prepared' || phase === 'provider-succeeded' } @@ -97,37 +103,6 @@ export function openForProviderWrite( } } -/** For an operation only the provider can perform: the conversation, then its agent, proven - * started; the caller waits out a start under way (`runAfterProviderStart`). `beforeAgent` - * answers what needs no agent before one is started for it. */ -export function openWithAgent( - context: Pick< - StructuredAgentSessionMutationContext, - 'openConversation' | 'ensureAgent' | 'deps' | 'sessions' - >, - envelope: AgentSessionMutationEnvelope, - beforeAgent?: (record: AgentSessionRecord | null) => AgentSessionMutationSessionPreparation | null -): (ledger: 'admit' | 'replay') => Promise { - return async (ledger) => { - const opened = await openConversationForWrite( - context.openConversation, - envelope, - context.deps.logger - ) - if (!opened.ok || ledger === 'replay') { - return opened - } - const checked = beforeAgent?.(context.deps.store.getRecord(envelope.sessionId)) - if (checked && !checked.ok) { - return checked - } - const ensured = await context.ensureAgent(envelope.sessionId) - return ensured.ok - ? (refuseWhileProviderStarting(context.sessions.get(envelope.sessionId)) ?? ensured) - : ensured - } -} - /** The recovery a proven start runs, for a child already running: no start comes for it. A * recovery that stays unknown leaves the record as it was. */ async function recoverRewindOnLiveChild( @@ -160,7 +135,7 @@ async function recoverRewindOnLiveChild( /** A rewind still in doubt once the conversation is open is one only its provider can settle — * the open settles every other — so a send settles it first: an agent at rest is started, whose - * proven start recovers it while the send waits that start out (`runAfterProviderStart`), and a + * proven start recovers it before the send is accepted, and a * running one is asked as that start would. One still in doubt refuses the send here, before the * ledger records it, so a Retry of the same id is decided afresh. A * resend of a recorded id needs only the conversation, its answer's source: it starts nothing, @@ -171,10 +146,11 @@ async function recoverRewindOnLiveChild( export function sendPreparation( context: Pick< StructuredAgentSessionMutationContext, - 'openConversation' | 'ensureAgent' | 'deps' | 'sessions' | 'publish' | 'now' + 'openConversation' | 'deps' | 'sessions' | 'publish' | 'now' >, envelope: AgentSessionMutationEnvelope, - arrival: { clearInFlight?: boolean; refusesInRun?: boolean } = {} + arrival: { clearInFlight?: boolean; refusesInRun?: boolean } = {}, + selection?: ReadyOperationSelection ): (ledger: 'admit' | 'replay') => Promise { return async (ledger) => { if (ledger === 'admit' && arrival.clearInFlight) { @@ -190,22 +166,27 @@ export function sendPreparation( ? opened : { ok: false, refusal: agentSessionOperationOutcomeUnknown(envelope.clientOperationId) } } + if (selection?.deadline !== undefined && Date.now() >= selection.deadline) { + return operationDidNotRun(true) + } const { sessionId } = envelope if (!opened.ok || !rewindInDoubt(context.deps.store.getRecord(sessionId))) { return opened } - const ensured = await context.ensureAgent(sessionId) - if (!ensured.ok) { - return ensured - } - const starting = refuseWhileProviderStarting(context.sessions.get(sessionId)) - if (starting) { - return starting + const child = context.sessions.get(sessionId)?.child + if ( + !selection?.child || + !child || + child.phase !== 'ready' || + child.close || + !sameProviderChild(child, selection.child) + ) { + return rewindRefusal('outcome-unknown') } await recoverRewindOnLiveChild(context, sessionId) return !arrival.refusesInRun && rewindInDoubt(context.deps.store.getRecord(sessionId)) ? rewindRefusal('outcome-unknown') - : ensured + : opened } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-send-restarts-failed-start.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-send-restarts-failed-start.test.ts index 4e9d1503e820..b11844207c50 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-send-restarts-failed-start.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-send-restarts-failed-start.test.ts @@ -78,9 +78,7 @@ async function send( value: { submission: { dispatchState: 'pending', handoverRecorded: true } } }) const clientMessageId = sent.ok ? sent.value.clientMessageId : '' - await eventually(() => - expect(host.collaboratorsForTests().conversationDelivery.loop.isRunning(SESSION)).toBe(false) - ) + expect(await submission(clientMessageId)).toMatchObject({ dispatchState: 'pending' }) return clientMessageId } @@ -197,7 +195,7 @@ describe('a send into a published session whose child ended before startup', () const first = await send('hello again', releasedFence) // Accepted at the lost owner's fence, held for the child delivery started. - expect(acquire).toHaveBeenCalledTimes(2) + await eventually(() => expect(acquire).toHaveBeenCalledTimes(2)) expect(store.getRecord(SESSION)?.lease.claimStatus).toBe('live') expect(dispatch).not.toHaveBeenCalled() const current = store.getRecord(SESSION)?.lease.runtimeFence ?? 0 @@ -225,6 +223,7 @@ describe('a send into a published session whose child ended before startup', () const rowsBefore = (await journalStatuses()).length const held = await send('still not signed in', releasedFence) + await eventually(() => expect(acquire).toHaveBeenCalledTimes(2)) await exitBeforeProof() // The child never proved its start, so it was handed nothing: the message is rejected, so @@ -260,7 +259,7 @@ describe('a send into a published session whose child ended before startup', () // Retry is a fresh action: it restarts once and is handed to the new child once it proves itself. const retry = await send('signed in now') - expect(acquire).toHaveBeenCalledTimes(3) + await eventually(() => expect(acquire).toHaveBeenCalledTimes(3)) await proveStarted() await handedOver(retry) expect(dispatch).toHaveBeenCalledOnce() diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-signed-out-child.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-signed-out-child.test.ts index 13fa794a4eb1..5ac87ab412ba 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-signed-out-child.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-signed-out-child.test.ts @@ -11,6 +11,7 @@ import { REST_TEST_CALLER as CALLER, REST_TEST_SESSION as SESSION, restTestSend, + sweepTicks, type RestTestRig } from './structured-agent-session-rest-test-rig' import { @@ -40,6 +41,31 @@ describe('a send after the agent said it is not signed in', () => { const fence = () => rig.store.getRecord(SESSION)?.lease.runtimeFence ?? 1 + it('hands a message to the signed-out child started for it exactly once', async () => { + await foundRestTestChat(rig) + await rig.host.close(SESSION, 'evict') + rig.adapter.closeSession.mockClear() + rig.adapter.startUnavailable.mockReturnValue({ reason: 'notSignedIn' }) + const acquire = rig.adapter.acquire.getMockImplementation()! + rig.adapter.acquire + .mockImplementationOnce(acquire) + .mockRejectedValueOnce(new Error('unexpected extra restart')) + const params = restTestSend('try the newly started agent', fence()) + + expect(await rig.host.send(CALLER, params)).toMatchObject({ ok: true }) + await vi.waitFor(() => expect(rig.adapter.dispatch).toHaveBeenCalledTimes(2)) + await rig.host.flushStreamedEvents(SESSION) + await sweepTicks() + + expect( + rig.adapter.dispatch.mock.calls.filter( + ([input]) => input.clientMessageId === params.envelope.clientOperationId + ) + ).toHaveLength(1) + expect(rig.adapter.acquire).toHaveBeenCalledTimes(2) + expect(rig.adapter.closeSession).not.toHaveBeenCalled() + }) + it('goes to a new agent, and the new agent is kept for the sends after it', async () => { await foundRestTestChat(rig) rig.adapter.dispatch.mockImplementationOnce(async () => ({ @@ -205,11 +231,10 @@ describe('retireSignedOutStructuredAgentSessionChild', () => { const retire = async ( input: Parameters[0], work: Partial = {}, - stopAgent = vi.fn(async () => undefined), - startedFor = 'next-send' + stopAgent = vi.fn(async () => undefined) ) => { const warn = vi.fn() - await retireSignedOutStructuredAgentSessionChild('session', conversation(input), startedFor, { + await retireSignedOutStructuredAgentSessionChild('session', conversation(input), { work: { ...idle, ...work }, stopAgent, logger: { warn, error: vi.fn() } @@ -221,23 +246,6 @@ describe('retireSignedOutStructuredAgentSessionChild', () => { expect((await retire(signedOut)).stopAgent).toHaveBeenCalledWith('session') }) - it('keeps a child started for the send it is about to take', async () => { - const own = await retire( - { ...signedOut, child: childAt({ startedFor: 'next-send' }) }, - {}, - undefined, - 'next-send' - ) - expect(own.stopAgent).not.toHaveBeenCalled() - const earlier = await retire( - { ...signedOut, child: childAt({ startedFor: 'earlier-send' }) }, - {}, - undefined, - 'next-send' - ) - expect(earlier.stopAgent).toHaveBeenCalledWith('session') - }) - it('keeps it while it owes work the idle sweep also protects', async () => { // A pending prompt is the shared check's too; the idle sweep's tests cover that case. for (const kept of [ diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-signed-out-child.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-signed-out-child.ts index c9c891bc3791..6eb7cd88ec16 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-signed-out-child.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-signed-out-child.ts @@ -78,15 +78,12 @@ function owesWakeUp( ) } -/** For a caller inside the session's serialize, before it hands the next send (`startedFor`) to - * the child. A child started for that same send keeps it: the send has not reached any child yet, - * and the agent's own refusal settles it once, where a new child would refuse it again forever. - * Work the child still owes keeps it, as it keeps an idle one from the sweep. A stop that fails +/** For a caller inside the session's serialize, before it hands the next send to + * the child. Work the child still owes keeps it, as it keeps an idle one from the sweep. A stop that fails * leaves the close begun, which the start joins. */ export async function retireSignedOutStructuredAgentSessionChild( sessionId: string, session: SignedOutReading | undefined, - startedFor: string, deps: { work: StructuredAgentSessionChildWorkReads startUnavailable?: () => AgentSessionUnavailable | undefined @@ -96,7 +93,6 @@ export async function retireSignedOutStructuredAgentSessionChild( ): Promise { if ( !session || - session.child?.startedFor === startedFor || !structuredAgentSessionChildReportedSignedOut(session, deps.startUnavailable?.()) || structuredAgentSessionChildHasOpenWork(session.journal, deps.work) || owesWakeUp(session.journal, deps.work.childWork()) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-startup-attempt.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-startup-attempt.test.ts index 224c92c4c1e9..1794e14c38e6 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-startup-attempt.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-startup-attempt.test.ts @@ -63,6 +63,55 @@ afterEach(() => { }) describe('the startup clock', () => { + it('settles a selected caller with readiness and does not retain a completed attempt', async () => { + const attempt = mint() + attempts.track(SESSION, attempt).spawned() + attempts.published(SESSION, attempt.attemptId, CHILD) + const readiness = attempts.readiness(SESSION, CHILD) + expect(readiness).not.toBeNull() + + const waiting = readiness!.wait(120_000) + attempts.ready(SESSION, CHILD) + + await expect(waiting).resolves.toBe('ready') + expect(attempts.readiness(SESSION, CHILD)).toBeNull() + expect(vi.getTimerCount()).toBe(0) + }) + + it('ends a selected caller on child exit, expiry, or disposal', async () => { + for (const end of ['exit', 'expiry', 'dispose'] as const) { + const attempt = mint() + attempts.track(SESSION, attempt).spawned() + attempts.published(SESSION, attempt.attemptId, CHILD) + const waiting = attempts.readiness(SESSION, CHILD)!.wait(120_000) + if (end === 'exit') { + attempts.childEnded(SESSION, CHILD) + } + if (end === 'expiry') { + vi.advanceTimersByTime(SILENCE_MS) + } + if (end === 'dispose') { + attempts.dispose() + } + await expect(waiting).resolves.toBe('ended') + } + }) + + it('ends only the caller wait at its budget, leaving the attempt able to start', async () => { + const attempt = mint() + attempts.track(SESSION, attempt).spawned() + attempts.published(SESSION, attempt.attemptId, CHILD) + const waiting = attempts.readiness(SESSION, CHILD)!.wait(500) + const reattached = attempts.readiness(SESSION, { ...CHILD })!.wait(120_000) + + vi.advanceTimersByTime(500) + await expect(waiting).resolves.toBe('timeout') + expect(attempts.isOpen(SESSION)).toBe(true) + attempts.ready(SESSION, CHILD) + await expect(reattached).resolves.toBe('ready') + expect(attempts.isOpen(SESSION)).toBe(false) + }) + it('does not run before the spawn: resolving a launch is not starting', () => { attempts.track(SESSION, mint()) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-startup-attempt.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-startup-attempt.ts index 800cdb6dde5d..d9c8db6066b8 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-startup-attempt.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-startup-attempt.ts @@ -94,6 +94,11 @@ type Tracked = { /** The limit's measure, held until an acquire it aborted settles: one that is ready anyway * reports that instead. */ expiry: StructuredAgentSessionStartupSettled | null + readiness: PromiseWithResolvers<'ready' | 'ended'> +} + +export type StructuredAgentSessionStartReadiness = { + wait: (budgetMs?: number) => Promise<'ready' | 'ended' | 'timeout'> } /** Each session's open attempt and its startup clock. A session starts one child at a time, so a @@ -131,7 +136,8 @@ export class StructuredAgentSessionStartupAttempts { lastOutputAt: 0, timer: null, expired: false, - expiry: null + expiry: null, + readiness: Promise.withResolvers() } this.open.set(sessionId, tracked) const current = (): boolean => this.open.get(sessionId) === tracked && !this.disposed @@ -155,6 +161,34 @@ export class StructuredAgentSessionStartupAttempts { } } + /** Capture the attempt while the selected child still holds the session's queue. */ + readiness( + sessionId: string, + child: StructuredAgentSessionProviderChildIdentity + ): StructuredAgentSessionStartReadiness | null { + const tracked = this.open.get(sessionId) + if (!tracked?.child || !sameProviderChild(tracked.child, child)) { + return null + } + return { + wait: async (budgetMs) => { + if (budgetMs === undefined) { + return tracked.readiness.promise + } + let timer: ReturnType | undefined + const timeout = new Promise<'timeout'>((resolve) => { + timer = setTimeout(() => resolve('timeout'), Math.max(0, budgetMs)) + timer.unref?.() + }) + try { + return await Promise.race([tracked.readiness.promise, timeout]) + } finally { + clearTimeout(timer) + } + } + } + } + /** The acquire returned: its `starting` child stays on the clock until it proves its start, and * one published after the limit passed is expired now. */ published( @@ -222,6 +256,7 @@ export class StructuredAgentSessionStartupAttempts { dispose(): void { this.disposed = true for (const tracked of this.open.values()) { + tracked.readiness.resolve('ended') if (tracked.timer) { clearTimeout(tracked.timer) } @@ -286,6 +321,7 @@ export class StructuredAgentSessionStartupAttempts { } private settle(sessionId: string, tracked: Tracked, outcome: 'ready' | 'ended'): void { + tracked.readiness.resolve(outcome) this.report( tracked.expired && outcome === 'ended' ? tracked.expiry : this.measure(tracked, outcome) ) @@ -318,6 +354,7 @@ export class StructuredAgentSessionStartupAttempts { private end(sessionId: string): void { const tracked = this.open.get(sessionId) + tracked?.readiness.resolve('ended') if (tracked?.timer) { clearTimeout(tracked.timer) } 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 92d41bc1dd41..84aef7e72764 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 @@ -10,7 +10,10 @@ import { admitAndRunAgentSessionMutation } from './structured-agent-session-muta import { agentSessionOperationKey } from '../../../shared/agent-session-operation-ledger' import type { StructuredAgentSessionMutationContext } from './structured-agent-session-host-mutations' import { sendPreparation } from './structured-agent-session-send-preparation' -import { runAfterProviderStart } from './structured-agent-session-provider-start-hold' +import { + runAfterRewindStart, + type ReadyOperationSelection +} from './structured-agent-session-ready-operation' import type { StructuredAgentSessionCaller } from './structured-agent-session-host-types' import { conversationCommandBlocked } from './structured-conversation-command-admission' import type { AgentSessionFailureFact } from '../../../shared/agent-session-failure' @@ -49,16 +52,20 @@ export function runStructuredConversationCommand( caller: StructuredAgentSessionCaller, params: ConversationCommandParams ): Promise> { - // A rewind in doubt is recovered by a proven start, which the clear waits out. - return runAfterProviderStart(context, params.envelope.sessionId, () => - clearUnderSerialize(context, caller, params) + return runAfterRewindStart( + context, + caller.callerKey, + params.envelope, + { method: 'agentSession.conversationCommand', fields: { command: params.command } }, + (selection) => clearUnderSerialize(context, caller, params, selection) ) } -function clearUnderSerialize( +async function clearUnderSerialize( context: StructuredAgentSessionMutationContext, caller: StructuredAgentSessionCaller, - params: ConversationCommandParams + params: ConversationCommandParams, + selection: ReadyOperationSelection ): Promise> { const { envelope, command } = params const { sessionId, clientOperationId } = envelope @@ -69,94 +76,92 @@ function clearUnderSerialize( ? record : null } - return context.serialize(sessionId, async () => { - let clear: AgentSessionConversationClear | null = null - const result = await admitAndRunAgentSessionMutation({ - store, - adapter: context.deps.adapter, - agents: context.deps.agents, - logger: context.deps.logger, - callerKey: caller.callerKey, - envelope, - // Starts the agent only to settle a rewind in doubt, as a send does; a /clear itself starts nothing. - prepareSession: sendPreparation(context, envelope, { refusesInRun: true }), - journal: () => context.sessions.get(sessionId)?.journal, - publish: (journal) => context.publish(sessionId, journal), - now: context.now, - plan: { - method: 'agentSession.conversationCommand', - fields: { command }, - // Written to the conversation, not the agent, so whoever owns the agent does not matter. - conversationWrite: true, - recoverUnknownFromDurableState: true, - settlesWithWrite: true, - successReceipt: () => - store.conversationReceipts.clear( - () => { - if (!clear) { - throw new Error('agent_session_clear_not_completed') - } - return clear - }, - { callerKey: caller.callerKey, operationId: clientOperationId } - ), - replay: (_ctx, outcome) => { - if (outcome.status === 'succeeded' && outcome.conversationCommand) { - return outcome.conversationCommand - } - const prior = matching() - return prior?.phase === 'committed' ? prior : null - }, - // The commit is the only write, so a clear with no committed answer changed nothing. - rerunWhenReplayMissing: () => true, - run: async (ctx) => { - const record = store.getRecord(sessionId)! - const blocked = conversationCommandBlocked( - ctx, - record, - context.readChildWork(sessionId), - context.sessions.get(sessionId)?.child ? undefined : 'at-rest' - ) - if (blocked) { - return { ok: false, refusal: blocked } - } - await context.stopAgent(sessionId, { cause: 'context-clear' }) - const stopped = store.getRecord(sessionId)! - const fence = stopped.lease.runtimeFence - const rechecked = conversationCommandBlocked( - { ...ctx, fence }, - stopped, - context.readChildWork(sessionId), - 'at-rest' - ) - if (rechecked) { - return { ok: false, refusal: rechecked } - } - const completed: AgentSessionConversationClear['command'] = { - command: 'clear', - runtimeFence: fence, - operationId: clientOperationId, - callerKey: caller.callerKey, - phase: 'committed', - state: 'completed' - } - clear = { sessionId, fence, command: completed, now: context.now() } - if (!ctx.operationReceipt) { - throw new Error('agent_session_clear_receipt_missing') - } - await ctx.journal.context.clear( - providerContextBoundaryForClear(clear), - ctx.operationReceipt, - agentSessionOperationKey(caller.callerKey, clientOperationId) - ) - return { ok: true, value: completed } + let clear: AgentSessionConversationClear | null = null + const result = await admitAndRunAgentSessionMutation({ + store, + adapter: context.deps.adapter, + agents: context.deps.agents, + logger: context.deps.logger, + callerKey: caller.callerKey, + envelope, + // Starts the agent only to settle a rewind in doubt, as a send does; a /clear itself starts nothing. + prepareSession: sendPreparation(context, envelope, { refusesInRun: true }, selection), + journal: () => context.sessions.get(sessionId)?.journal, + publish: (journal) => context.publish(sessionId, journal), + now: context.now, + plan: { + method: 'agentSession.conversationCommand', + fields: { command }, + // Written to the conversation, not the agent, so whoever owns the agent does not matter. + conversationWrite: true, + recoverUnknownFromDurableState: true, + settlesWithWrite: true, + successReceipt: () => + store.conversationReceipts.clear( + () => { + if (!clear) { + throw new Error('agent_session_clear_not_completed') + } + return clear + }, + { callerKey: caller.callerKey, operationId: clientOperationId } + ), + replay: (_ctx, outcome) => { + if (outcome.status === 'succeeded' && outcome.conversationCommand) { + return outcome.conversationCommand + } + const prior = matching() + return prior?.phase === 'committed' ? prior : null + }, + // The commit is the only write, so a clear with no committed answer changed nothing. + rerunWhenReplayMissing: () => true, + run: async (ctx) => { + const record = store.getRecord(sessionId)! + const blocked = conversationCommandBlocked( + ctx, + record, + context.readChildWork(sessionId), + context.sessions.get(sessionId)?.child ? undefined : 'at-rest' + ) + if (blocked) { + return { ok: false, refusal: blocked } + } + await context.stopAgent(sessionId, { cause: 'context-clear' }) + const stopped = store.getRecord(sessionId)! + const fence = stopped.lease.runtimeFence + const rechecked = conversationCommandBlocked( + { ...ctx, fence }, + stopped, + context.readChildWork(sessionId), + 'at-rest' + ) + if (rechecked) { + return { ok: false, refusal: rechecked } + } + const completed: AgentSessionConversationClear['command'] = { + command: 'clear', + runtimeFence: fence, + operationId: clientOperationId, + callerKey: caller.callerKey, + phase: 'committed', + state: 'completed' + } + clear = { sessionId, fence, command: completed, now: context.now() } + if (!ctx.operationReceipt) { + throw new Error('agent_session_clear_receipt_missing') } + await ctx.journal.context.clear( + providerContextBoundaryForClear(clear), + ctx.operationReceipt, + agentSessionOperationKey(caller.callerKey, clientOperationId) + ) + return { ok: true, value: completed } } - }) - return result.ok && - 'runtimeFence' in result.value && - typeof result.value.runtimeFence === 'number' - ? { ...result, fence: result.value.runtimeFence } - : result + } }) + return result.ok && + 'runtimeFence' in result.value && + typeof result.value.runtimeFence === 'number' + ? { ...result, fence: result.value.runtimeFence } + : result } diff --git a/src/main/native-chat/agent-session-wire/structured-conversation-compaction.ts b/src/main/native-chat/agent-session-wire/structured-conversation-compaction.ts index bae65871c531..9ae63858ca30 100644 --- a/src/main/native-chat/agent-session-wire/structured-conversation-compaction.ts +++ b/src/main/native-chat/agent-session-wire/structured-conversation-compaction.ts @@ -13,7 +13,7 @@ import { isQueuedAgentJournalSubmission } from '../../../shared/agent-session-qu import type { AgentSessionMutationResult } from '../../../shared/agent-session-wire' import type { StructuredAgentSessionHost } from './structured-agent-session-host' import { - mutateStructuredAgentSession, + runStructuredAgentSessionMutation, type StructuredAgentSessionMutationContext } from './structured-agent-session-mutation-context' import type { StructuredAgentSessionCaller } from './structured-agent-session-host-types' @@ -26,7 +26,10 @@ import { structuredAgentSessionFailureWordsContext, structuredAgentSessionSendBlock } from './structured-agent-session-send-preparation' -import { runAfterProviderStart } from './structured-agent-session-provider-start-hold' +import { + runAfterRewindStart, + type ReadyOperationSelection +} from './structured-agent-session-ready-operation' import { STRUCTURED_AGENT_SESSION_START_WAIT_MS } from './structured-agent-session-send-settlement' import { conversationCommandFailure, @@ -147,8 +150,8 @@ function acceptStructuredConversationCommand( ...(params.delivery ? { delivery: params.delivery } : {}), priorRecord }) - return runAfterProviderStart(context, params.envelope.sessionId, () => - acceptUnderSerialize(context, caller, params, plan, arrival) + return runAfterRewindStart(context, caller.callerKey, params.envelope, plan, (selection) => + acceptUnderSerialize(context, caller, params, plan, arrival, selection) ) } @@ -157,9 +160,10 @@ function acceptUnderSerialize( caller: StructuredAgentSessionCaller, params: ConversationCommandParams, plan: ReturnType, - arrival: { clearInFlight?: boolean } + arrival: { clearInFlight?: boolean }, + selection: ReadyOperationSelection ): Promise> { - return mutateStructuredAgentSession( + return runStructuredAgentSessionMutation( context, caller, params.envelope, @@ -206,6 +210,6 @@ function acceptUnderSerialize( return plan.run(ctx) } }, - sendPreparation(context, params.envelope, arrival) + sendPreparation(context, params.envelope, arrival, selection) ) } diff --git a/src/main/runtime/claude-structured-send-during-startup.test.ts b/src/main/runtime/claude-structured-send-during-startup.test.ts index 494d012de462..de452a828a63 100644 --- a/src/main/runtime/claude-structured-send-during-startup.test.ts +++ b/src/main/runtime/claude-structured-send-during-startup.test.ts @@ -57,12 +57,17 @@ function fence(host: StructuredAgentSessionHost): number { return host.deps.store.getRecord(SESSION)?.lease.runtimeFence ?? 0 } -/** Whether the CLI was written a message, once the delivery loop has settled what it does. */ -async function written(host: StructuredAgentSessionHost): Promise { - await vi.waitFor(() => - expect(host.collaboratorsForTests().conversationDelivery.loop.isRunning(SESSION)).toBe(false) - ) - return claude.child(SESSION).calls.includes('send') +async function expectHeldBeforeStart( + host: StructuredAgentSessionHost, + clientMessageId: string +): Promise { + const entry = await submission(host, clientMessageId) + expect(entry).toMatchObject({ dispatchState: 'pending' }) + expect(entry?.handedOverAt).toBeUndefined() + expect(claude.child(SESSION).calls).not.toContain('send') + await host.flushStreamedEvents(SESSION) + expect((await submission(host, clientMessageId))?.handedOverAt).toBeUndefined() + expect(claude.child(SESSION).calls).not.toContain('send') } async function statusRows(host: StructuredAgentSessionHost): Promise { @@ -105,7 +110,7 @@ describe('a send into a Claude chat whose CLI keeps failing at startup', () => { await vi.waitFor(() => expect(host.deps.store.getRecord(SESSION)?.lease.claimStatus).toBe('live') ) - expect(await written(host)).toBe(false) + await expectHeldBeforeStart(host, held) await failLatestStart(host, 2) // Rejected with the cause, not left in doubt; one row for this attempt names it. @@ -142,7 +147,7 @@ describe('a send while the first Claude start is still answering initialize', () await host.attach(CALLER, claude.attachParams(SESSION, null)) const sent = await send(host, 'hello') - expect(await written(host)).toBe(false) + await expectHeldBeforeStart(host, sent) expect(host.collaboratorsForTests().sessions.get(SESSION)?.child?.phase).toBe('starting') // The CLI answers: the start lands, and the held message is written to it, once. @@ -165,8 +170,8 @@ describe('a send while the first Claude start is still answering initialize', () const host = await claude.install() await host.attach(CALLER, claude.attachParams(SESSION, null)) - await send(host, 'hello') - expect(await written(host)).toBe(false) + const sent = await send(host, 'hello') + await expectHeldBeforeStart(host, sent) claude.child(SESSION).answerInit() await vi.waitFor(() => @@ -184,7 +189,7 @@ describe('a send while the first Claude start is still answering initialize', () const startedFence = fence(host) const held = await send(host, 'hello') - expect(await written(host)).toBe(false) + await expectHeldBeforeStart(host, held) await failLatestStart(host, 1) await vi.waitFor(async () => diff --git a/src/main/runtime/claude-structured-send-restart-dies-before-dispatch.test.ts b/src/main/runtime/claude-structured-send-restart-dies-before-dispatch.test.ts index a05905e9599c..75e3a1fcd672 100644 --- a/src/main/runtime/claude-structured-send-restart-dies-before-dispatch.test.ts +++ b/src/main/runtime/claude-structured-send-restart-dies-before-dispatch.test.ts @@ -141,9 +141,8 @@ describe('a send whose restarted Claude child dies before it proves its start', // Accepted: the answer comes before the restart it needs. expect(answered.get(sent)).toBe('pending') await eventually(() => expect(claude.children(SESSION)).toHaveLength(2)) - await eventually(() => - expect(host.collaboratorsForTests().conversationDelivery.loop.isRunning(SESSION)).toBe(false) - ) + expect(await submission(host, sent)).toMatchObject({ dispatchState: 'pending' }) + expect((await submission(host, sent))?.handedOverAt).toBeUndefined() expect(claude.child(SESSION).calls).not.toContain('send') await failLatestStart(host, 2) diff --git a/src/main/runtime/claude-structured-startup-fault-is-not-an-exit.test.ts b/src/main/runtime/claude-structured-startup-fault-is-not-an-exit.test.ts index 81b3b9c19535..7dd9826c2960 100644 --- a/src/main/runtime/claude-structured-startup-fault-is-not-an-exit.test.ts +++ b/src/main/runtime/claude-structured-startup-fault-is-not-an-exit.test.ts @@ -101,9 +101,8 @@ describe('a Claude start that Orca fails while the CLI is still running', () => ok: true }) const held = await send(host, 'hello') - await vi.waitFor(() => - expect(host.collaboratorsForTests().conversationDelivery.loop.isRunning(SESSION)).toBe(false) - ) + expect(await submission(host, held)).toMatchObject({ dispatchState: 'pending' }) + expect((await submission(host, held))?.handedOverAt).toBeUndefined() expect(claude.child(SESSION).calls).not.toContain('send') claude.child(SESSION).answerInit() diff --git a/src/main/runtime/structured-agent-session-provider-operations-while-starting.test.ts b/src/main/runtime/structured-agent-session-provider-operations-while-starting.test.ts index 66c7fd735e72..c0310e6ca151 100644 --- a/src/main/runtime/structured-agent-session-provider-operations-while-starting.test.ts +++ b/src/main/runtime/structured-agent-session-provider-operations-while-starting.test.ts @@ -278,6 +278,26 @@ it('sets a goal asked of a starting Codex chat on the thread once it is open', a ]) }) +it('refuses a changed goal payload immediately while the handshake is held', async () => { + const child = codexScriptedChild() + const host = await openChat(child) + const change = { kind: 'set' as const, objective: 'Do not run' } + const goal = host.changeThreadGoal(CALLER, { + envelope: { + ...envelope(host, 'agentSession.threadGoal', { change }), + payloadFingerprint: 'changed-payload' + }, + change + }) + + expect(await settlesSoon(goal)).toBe(true) + expect(await goal).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_operation_conflict' } + }) + expect(child.requests().some(({ method }) => method === 'thread/goal/set')).toBe(false) +}) + it('keeps a goal waiting on a starting Codex chat through a re-attach, then sets it once the thread is open', async () => { const child = codexScriptedChild() const host = await openChat(child) @@ -301,6 +321,132 @@ it('keeps a goal waiting on a starting Codex chat through a re-attach, then sets expect(child.requests().some(({ method }) => method === 'thread/goal/set')).toBe(true) }) +it('ends a waiting goal when its starting child is closed and never runs it afterward', async () => { + const child = codexScriptedChild() + const host = await openChat(child) + const change = { kind: 'set' as const, objective: 'Do not run later' } + const goal = host.changeThreadGoal(CALLER, { + envelope: envelope(host, 'agentSession.threadGoal', { change }), + change + }) + expect(await settlesSoon(goal)).toBe(false) + + await host.close(SESSION, 'user-close') + expect(await goal).toMatchObject({ + ok: false, + refusal: { message: expect.stringContaining('did not run') } + }) + child.releaseHandshake() + expect(child.requests().some(({ method }) => method === 'thread/goal/set')).toBe(false) +}) + +it('does not run a goal after its caller budget ends, even if the child starts later', async () => { + const child = codexScriptedChild() + const host = await openChat(child) + const attempts = host.collaboratorsForTests().runtimeState.startupAttempts + const readiness = attempts.readiness.bind(attempts) + attempts.readiness = (sessionId, identity) => { + const selected = readiness(sessionId, identity) + return selected && { wait: async () => 'timeout' as const } + } + const change = { kind: 'set' as const, objective: 'Never after timeout' } + + const goal = await host.changeThreadGoal(CALLER, { + envelope: envelope(host, 'agentSession.threadGoal', { change }), + change + }) + expect(goal).toMatchObject({ + ok: false, + refusal: { message: expect.stringContaining('did not run') } + }) + attempts.readiness = readiness + child.releaseHandshake() + await host.flushStreamedEvents(SESSION) + expect(child.requests().some(({ method }) => method === 'thread/goal/set')).toBe(false) +}) + +it('does not run a ready goal when its final queued admission passes the caller budget', async () => { + const child = codexScriptedChild() + const host = await openChat(child) + const attempts = host.collaboratorsForTests().runtimeState.startupAttempts + const readiness = attempts.readiness.bind(attempts) + const held = Promise.withResolvers() + let finalQueueHeld = false + const realNow = Date.now.bind(Date) + attempts.readiness = (sessionId, identity) => { + const selected = readiness(sessionId, identity) + return ( + selected && { + wait: async (budgetMs?: number) => { + const outcome = await selected.wait(budgetMs) + attempts.readiness = readiness + void host.collaboratorsForTests().serialize(SESSION, async () => { + finalQueueHeld = true + await held.promise + }) + return outcome + } + } + ) + } + const change = { kind: 'set' as const, objective: 'Too late for admission' } + const goal = host.changeThreadGoal(CALLER, { + envelope: envelope(host, 'agentSession.threadGoal', { change }), + change + }) + child.releaseHandshake() + await eventually(() => expect(finalQueueHeld).toBe(true)) + const clock = vi.spyOn(Date, 'now').mockImplementation(() => realNow() + 121_000) + expect(await settlesSoon(goal)).toBe(false) + held.resolve() + + expect(await goal).toMatchObject({ + ok: false, + refusal: { + code: 'agent_session_operation_expired', + message: expect.stringContaining('wait expired') + } + }) + clock.mockRestore() + expect(child.requests().some(({ method }) => method === 'thread/goal/set')).toBe(false) +}) + +it('rechecks its selected child when another child replaces it before the goal runs', async () => { + const child = codexScriptedChild() + const host = await openChat(child) + const attempts = host.collaboratorsForTests().runtimeState.startupAttempts + const readiness = attempts.readiness.bind(attempts) + attempts.readiness = (sessionId, identity) => { + const selected = readiness(sessionId, identity) + return ( + selected && { + wait: async (budgetMs?: number) => { + const outcome = await selected.wait(budgetMs) + attempts.readiness = readiness + await host.close(SESSION, 'evict') + const replacement = send(host, 'replacement child') + await eventually(() => expect(child.spawns()).toBe(2)) + child.releaseHandshake() + expect(await replacement).toMatchObject({ ok: true }) + return outcome + } + } + ) + } + const change = { kind: 'set' as const, objective: 'Old child only' } + const goal = host.changeThreadGoal(CALLER, { + envelope: envelope(host, 'agentSession.threadGoal', { change }), + change + }) + child.releaseHandshake() + + expect(await goal).toMatchObject({ + ok: false, + refusal: { message: expect.stringContaining('did not run') } + }) + expect(child.requests().some(({ method }) => method === 'thread/goal/set')).toBe(false) +}) + it('answers a goal asked of an ACP agent, which has none, without waiting out its start', async () => { const spec = ACP_LAUNCH_SPECS[0]! const child = acpScriptedChild(spec.agent)() @@ -321,3 +467,28 @@ it('answers a goal asked of an ACP agent, which has none, without waiting out it expect(await send(host, 'after the goal')).toMatchObject({ ok: true }) await eventually(() => expect(child.prompts()).toEqual(['after the goal'])) }) + +it('admits an ordinary send in its first queued step before a later Stop', async () => { + const child = codexScriptedChild() + const host = await openChat(child) + const blocked = Promise.withResolvers() + void host.collaboratorsForTests().serialize(SESSION, () => blocked.promise) + const order: string[] = [] + const sent = send(host, 'before Stop').then((result) => { + order.push('send') + return result + }) + const stopped = host + .cancel(CALLER, { + envelope: envelope(host, 'agentSession.cancel', {}) + }) + .then((result) => { + order.push('Stop') + return result + }) + blocked.resolve() + + expect(await sent).toMatchObject({ ok: true }) + expect(await stopped).toMatchObject({ ok: true }) + expect(order).toEqual(['send', 'Stop']) +})