refactor: give agent startup one readiness owner

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',

src/main/native-chat/agent-session-wire/structured-agent-session-claude-hung-start-stop.test.ts

@@ -131,12 +131,10 @@ async function readsWorking(): Promise<boolean> {
   )
 }
 
-/** Whether the CLI was written `hello`, once the delivery loop has settled what it does with it. */
-async function helloWritten(): Promise<boolean> {
-  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<void> {
+  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<string> {
   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
 }
 

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<StructuredAgentSessionResumeOutcome>
+  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<StructuredAgentSessionResumeOutcome, { ok: false }>
 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.

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<StructuredAgentSessionResumeOutcome>
+  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<void>
   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) => {

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<typeof sendPreparation>[2]
 ): Promise<AgentSessionMutationResult<AgentSessionSendResult>> {
   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<AgentSessionMutationResult<AgentSessionThreadGoalResult>> {
-  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)
-      )
-    )
+    }
   )
 }
 

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),

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()

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<StructuredAgentSessionHostSession | null>
   /** Gives the session a provider child; inside the caller's serialize. */
   ensureAgent: (sessionId: string) => Promise<AgentSessionMutationSessionPreparation>
+  startupAttempts: Pick<StructuredAgentSessionStartupAttempts, 'readiness'>
   /** 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<AgentSessionMutationSessionPreparation>
@@ -63,19 +65,30 @@ export function mutateStructuredAgentSession<TValue>(
   prepareSession?: AgentSessionMutationRequest<TValue>['prepareSession']
 ): Promise<AgentSessionMutationResult<TValue>> {
   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<TValue>(
+  context: StructuredAgentSessionMutationContext,
+  caller: StructuredAgentSessionCaller,
+  envelope: AgentSessionMutationEnvelope,
+  plan: MutationPlan<TValue>,
+  prepareSession?: AgentSessionMutationRequest<TValue>['prepareSession']
+): Promise<AgentSessionMutationResult<TValue>> {
+  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()
+  })
+}