fix(orchestration): honor structured worker transport admission

src/main/runtime/claude-structured-session-integration.test.ts

@@ -27,6 +27,7 @@ import type { OrcaRuntimeService } from './orca-runtime'
 import type { RpcRequest, RpcResponse } from './rpc/core'
 import type { ClaudeStructuredAuthPolicy } from '../claude-accounts/claude-structured-auth-policy'
 import { RpcDispatcher } from './rpc/dispatcher'
+import { expectStructuredWorkerAdmission } from './structured-worker-admission.test-support'
 import { STRUCTURED_AGENT_SESSION_METHODS } from './rpc/methods/structured-agent-session'
 import {
   ensureStructuredAgentSessionHost,
@@ -532,6 +533,15 @@ describe('a structured Claude session over agentSession.*', () => {
     expect(leaseOf(SESSION)).toMatchObject({ claimStatus: 'released', handoffStage: null })
   })
 
+  it('admits a fresh worker preamble before the provider echo without resending', async () => {
+    await ok('agentSession.create', createIntentParams())
+    claude.live().send = async (message) => {
+      claude.live().sent.push(message)
+    }
+    await expectStructuredWorkerAdmission(SESSION)
+    expect(claude.live().sent.filter((message) => message.type === 'user')).toHaveLength(1)
+  })
+
   it('creates, sends, streams, approves, interrupts, and resumes from the chain head', async () => {
     vi.stubEnv('ANTHROPIC_API_KEY', 'sk-ant-SHELL-LEAK')
     const created = await ok<{ fence: number }>('agentSession.create', createIntentParams())

src/main/runtime/orchestration/structured-mailbox-pointer-delivery.test.ts

@@ -74,7 +74,7 @@ function attentionJournal(): AgentJournalRenderItem[] {
 
 function harness(options: {
   journal: AgentJournalRenderItem[] | null
-  dispatchState?: 'accepted' | 'rejected' | 'unknown'
+  dispatchState?: 'pending' | 'accepted' | 'rejected' | 'unknown'
   refusal?: AgentSessionPtyWriteRefusal
   /** The coordinator of this worker's Run is mid-batch: it checked and has not acked yet. */
   outstandingRunDelivery?: boolean
@@ -269,6 +269,27 @@ describe('structured mailbox pointer delivery', () => {
     expect(send).not.toHaveBeenCalled()
   })
 
+  it('retains admitted mail and replays its operation until provider acceptance', async () => {
+    const { delivery, send, markAsDelivered, stored } = harness({
+      journal: idleJournal(),
+      dispatchState: 'pending'
+    })
+    delivery.deliverForHandle('dispatch:d1')
+    await flush()
+    expect(markAsDelivered).not.toHaveBeenCalled()
+    const first = send.mock.calls[0]![0].operationId
+    delivery.onJournalActivity('session-1')
+    await flush()
+    expect(send.mock.calls[1]![0].operationId).toBe(first)
+    expect(markAsDelivered).not.toHaveBeenCalled()
+    send.mockResolvedValueOnce({ kind: 'sent', state: 'accepted' })
+    delivery.onJournalActivity('session-1')
+    await flush()
+    expect(send.mock.calls[2]![0].operationId).toBe(first)
+    expect(markAsDelivered).toHaveBeenCalledExactlyOnceWith(['m1'])
+    expect(stored.size).toBe(0)
+  })
+
   it('retries a rejected nudge on the next journal edge', async () => {
     // A rejection consumes no mail and nothing else redrives this mailbox, so leaving it unparked
     // stranded the worker until unrelated mail happened to arrive.

src/main/runtime/orchestration/structured-mailbox-pointer-host.test.ts

@@ -80,9 +80,7 @@ describe('structured mailbox pointer host', () => {
   it.each([
     ['accepted', 'accepted'],
     ['rejected', 'rejected'],
-    // Neither is an acknowledgement, and only `accepted` may consume mail: both have to reach the
-    // caller as `unknown` so the pointer is retained for the next journal edge.
-    ['pending', 'unknown'],
+    ['pending', 'pending'],
     ['unknown', 'unknown']
   ])('maps a %s submission to %s', async (dispatchState, expected) => {
     const send = vi.fn(

src/main/runtime/orchestration/structured-mailbox-pointer-host.ts

@@ -94,11 +94,12 @@ export function createStructuredMailboxPointerHost(): StructuredMailboxPointerHo
           ? { kind: 'unattached' }
           : { kind: 'sent', state: 'rejected' }
       }
-      // `pending` is not yet an acknowledgement; only `accepted` may consume mail.
+      // Mail consumption still requires provider acceptance, beyond transport admission.
       const state = result.value.submission.dispatchState
       return {
         kind: 'sent',
-        state: state === 'accepted' ? 'accepted' : state === 'rejected' ? 'rejected' : 'unknown'
+        state:
+          state === 'pending' || state === 'accepted' || state === 'rejected' ? state : 'unknown'
       }
     }
   }

src/main/runtime/orchestration/structured-session-pointer-delivery.test.ts

@@ -161,13 +161,15 @@ describe('dispatch outcome classification', () => {
   it('marks mail delivered only on an accepted dispatch', () => {
     expect(structuredDispatchDelivered('accepted')).toBe(true)
     expect(structuredDispatchDelivered('rejected')).toBe(false)
+    expect(structuredDispatchDelivered('pending')).toBe(false)
   })
 
   it('does not treat unknown as delivered, because a dead child settles unknown', () => {
     expect(structuredDispatchDelivered('unknown')).toBe(false)
   })
 
   it('names the retain reason for each non-accepted dispatch', () => {
+    expect(retainReasonForDispatch('pending')).toBe('dispatch-pending')
     expect(retainReasonForDispatch('rejected')).toBe('dispatch-rejected')
     expect(retainReasonForDispatch('unknown')).toBe('dispatch-unknown')
   })

src/main/runtime/orchestration/structured-session-pointer-delivery.ts

@@ -23,6 +23,7 @@ export type StructuredPointerRetainReason =
   | 'session-not-attached'
   | 'turn-unsettled'
   | 'awaiting-human'
+  | 'dispatch-pending'
   | 'dispatch-rejected'
   | 'dispatch-unknown'
 
@@ -31,7 +32,7 @@ export type StructuredPointerDecision =
   | { deliver: false; retain: StructuredPointerRetainReason }
 
 /** The dispatch states both provider adapters converge on. */
-export type StructuredDispatchState = 'accepted' | 'rejected' | 'unknown'
+export type StructuredDispatchState = 'pending' | 'accepted' | 'rejected' | 'unknown'
 
 /**
  * A refusal names an owner this pointer may be redirected to only when that
@@ -128,9 +129,8 @@ export function decideStructuredSessionPointerDelivery(input: {
 /**
  * Only an accepted dispatch may mark mail delivered.
  *
- * `unknown` covers a dead provider child and a slow acknowledgement alike — the
- * adapters cannot tell them apart — so it must retain. Treating it as delivered
- * would drop mail whenever a child died mid-send.
+ * Pending admission awaits provider acceptance; unknown delivery may never settle.
+ * Both retain the mail and its operation ID for journal-driven reconciliation.
  */
 export function structuredDispatchDelivered(state: StructuredDispatchState): boolean {
   return state === 'accepted'
@@ -139,7 +139,11 @@ export function structuredDispatchDelivered(state: StructuredDispatchState): boo
 export function retainReasonForDispatch(
   state: Exclude<StructuredDispatchState, 'accepted'>
 ): StructuredPointerRetainReason {
-  return state === 'rejected' ? 'dispatch-rejected' : 'dispatch-unknown'
+  return state === 'pending'
+    ? 'dispatch-pending'
+    : state === 'rejected'
+      ? 'dispatch-rejected'
+      : 'dispatch-unknown'
 }
 
 /**

src/main/runtime/rpc/methods/orchestration-structured-worker-session.test.ts

@@ -234,18 +234,16 @@ describe('structured worker dispatch preamble', () => {
   const send = (host: never) =>
     sendStructuredWorkerPreamble({ host, sessionId: 's1', dispatchId: 'd1', preamble: 'spec' })
 
-  it('reports the preamble delivered only on an accepted submission', async () => {
-    await expect(
-      send(hostWithSubmission({ dispatchState: 'accepted', reason: null }))
-    ).resolves.toBeUndefined()
+  it.each(['pending', 'accepted'])('admits a %s preamble', async (dispatchState) => {
+    await expect(send(hostWithSubmission({ dispatchState, reason: null }))).resolves.toBeUndefined()
   })
 
   it('never claims delivery for a submission the provider never acknowledged', async () => {
     // `dispatchSafely` turns ANY thrown adapter call — provider child dead, transport dropped —
     // into `unknown`, and `performSend` still returns ok. Reporting that as `dispatch_input:
     // accepted` marks the worker ready with no task, and the coordinator blocks in
     // `check --wait --types worker_done` until it times out.
-    for (const dispatchState of ['unknown', 'pending'] as const) {
+    for (const dispatchState of ['unknown', 'future-state'] as const) {
       const error = await send(
         hostWithSubmission({ dispatchState, reason: 'provider child exited' })
       ).catch((thrown: unknown) => thrown)

src/main/runtime/rpc/methods/orchestration-structured-worker-session.ts

@@ -245,7 +245,8 @@ export async function sendStructuredWorkerPreamble(args: {
     throw new Error(`The dispatch preamble was refused: ${result.refusal.message}`)
   }
   const submission = result.value.submission
-  if (submission.dispatchState === 'accepted') {
+  // The host returns pending after transport admission; provider identity arrives through the journal.
+  if (submission.dispatchState === 'pending' || submission.dispatchState === 'accepted') {
     return
   }
   if (submission.dispatchState === 'rejected') {
@@ -259,10 +260,7 @@ export async function sendStructuredWorkerPreamble(args: {
       `The dispatch preamble was not delivered: ${submission.reason ?? 'no reason given'}.`
     )
   }
-  // Only `accepted` is an acknowledgement — the same rule the mail lane already applies. A thrown
-  // adapter call settles as `unknown`, which is indistinguishable from a lost reply, so the start
-  // may claim neither delivery nor failure: `operation_unknown` is what turns this into the
-  // `outcome_unknown` receipt whose nextCommands send the coordinator to look.
+  // A lost reply cannot prove admission or non-delivery; never retry the preamble here.
   throw new OrchestrationError(
     'operation_unknown',
     `The dispatch preamble was submitted but not acknowledged (${submission.dispatchState}): ${submission.reason ?? 'no reason given'}.`

src/main/runtime/rpc/methods/orchestration-structured-worker-start-failure.test.ts

@@ -10,6 +10,8 @@
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 import { setStructuredAgentSessionHost } from '../../../native-chat/agent-session-wire/structured-agent-session-registry'
 import type { OrcaRuntimeService } from '../../orca-runtime'
+import { releaseStructuredWorkerSession } from './orchestration-structured-worker-session'
+import { isUnknownWorkerStartOutcome } from './orchestration/worker/worker-topology'
 import type { OrchestrationDb } from '../../orchestration/db'
 import { structuredWorkerIdentities } from '../../structured-worker-identity'
 
@@ -19,16 +21,9 @@ vi.mock('./structured-agent-session-create', () => ({
     value: { sessionId: args.envelope.sessionId }
   })
 }))
-vi.mock('./orchestration-structured-worker-session', async (importOriginal) => ({
-  ...(await importOriginal<Record<string, unknown>>()),
-  // The realistic post-create failure: the session is live, the preamble turn is not acknowledged.
-  sendStructuredWorkerPreamble: async () => {
-    throw new Error('The dispatch preamble was rejected: no capacity')
-  }
-}))
 vi.mock('./orchestration/worker/worker-start-validation', () => ({
-  prepareLocalWorkerStart: () => ({
-    agent: 'claude',
+  prepareLocalWorkerStart: (args: { params: { agent: string } }) => ({
+    agent: args.params.agent,
     launch: { receipt: { requested: null, effective: null }, preferences: undefined }
   })
 }))
@@ -37,9 +32,10 @@ vi.mock('./orchestration/worker/worker-setup-gate', () => ({
   persistWorkerReadinessStage: () => {},
   persistWorkerSetupWaitOutcome: () => {}
 }))
-vi.mock('./orchestration/worker/worker-start-receipt', () => ({
-  failWorkerStartWithReceipt: (args: { failedStage: string }) => ({
-    state: 'failed',
+vi.mock('./orchestration/worker/worker-start-receipt', async (importOriginal) => ({
+  ...(await importOriginal<Record<string, unknown>>()),
+  failWorkerStartWithReceipt: (args: { failedStage: string; error: unknown }) => ({
+    state: isUnknownWorkerStartOutcome(args.error, args.failedStage) ? 'outcome_unknown' : 'failed',
     stage: args.failedStage
   })
 }))
@@ -52,10 +48,17 @@ const { startLocalWorker } = await import('./orchestration/worker/local-worker-s
 
 const WORKTREE = 'wt_1'
 
-function installHost() {
+function installHost(dispatchState = 'rejected') {
   const closed: string[] = []
   const visibility: [string, boolean][] = []
-  setStructuredAgentSessionHost({
+  const send = vi.fn(async () => ({
+    ok: true,
+    value: { submission: { dispatchState, reason: 'provider outcome' } }
+  }))
+  const release = vi.fn()
+  const unsubscribe = vi.fn()
+  const host = {
+    send,
     setSessionTabVisibility: async (sessionId: string, visible: boolean) => {
       visibility.push([sessionId, visible])
     },
@@ -64,8 +67,8 @@ function installHost() {
     },
     hasSession: () => true,
     hold: async () => {},
-    release: () => {},
-    subscribe: () => () => {},
+    release,
+    subscribe: () => unsubscribe,
     deps: {
       store: {
         getRecord: () => ({
@@ -79,8 +82,10 @@ function installHost() {
         })
       }
     }
-  } as never)
-  return { closed, visibility }
+  }
+  // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: This fixture implements every host member used by structured worker creation, dispatch and cleanup.
+  setStructuredAgentSessionHost(host as never)
+  return { closed, visibility, send, release, unsubscribe }
 }
 
 function fakes() {
@@ -104,14 +109,18 @@ function fakes() {
     getTerminalPaneKey: vi.fn(() => 'pane_1'),
     retireStructuredAgentSessionTabFromSnapshot
   } as unknown as OrcaRuntimeService
-  const db = {
+  const dbFixture = {
     createStartingWorkerDispatch: () => ({
       dispatch: { id: 'd_fail', depth: 0 },
       task: { id: 't1', spec: 'do the thing' }
     }),
     recordWorkerStage: () => {},
-    prepareStartingWorkerAuthority: () => 'capability'
-  } as unknown as OrchestrationDb
+    prepareStartingWorkerAuthority: () => 'capability',
+    getWorkerDispatch: () => ({ state: 'starting' }),
+    markWorkerDispatchReady: () => ({ state: 'ready', stage: 'ready' })
+  }
+  // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: This fixture implements the database calls reached by local worker start; receipt persistence is mocked.
+  const db = dbFixture as unknown as OrchestrationDb
   return { runtime, db, retireStructuredAgentSessionTabFromSnapshot }
 }
 
@@ -149,3 +158,53 @@ describe('a structured worker-start that fails after the session exists', () =>
     expect(retireStructuredAgentSessionTabFromSnapshot).toHaveBeenCalledWith(sessionId)
   })
 })
+
+describe.each(['codex', 'claude'] as const)('%s structured worker admission', (agent) => {
+  it.each(['pending', 'accepted', 'rejected', 'unknown'])(
+    'preserves readiness and cleanup for %s',
+    async (dispatchState) => {
+      const host = installHost(dispatchState)
+      const { runtime, db, retireStructuredAgentSessionTabFromSnapshot } = fakes()
+      try {
+        const receipt = await startLocalWorker({
+          params: { from: 'term_c', timeoutMs: 1_000, agent },
+          mode: {
+            mode: 'structured',
+            preferred: 'structured',
+            reason: 'user_default',
+            detail: 'structured by default'
+          },
+          runtime,
+          db,
+          // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Only id is read by this start fixture.
+          run: { id: 'run_1' } as never,
+          // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Only id and spec are read by this start fixture.
+          existingTask: { id: 't1', spec: 'do the thing' } as never,
+          coordinatorPane: null,
+          orchestrationMutation: undefined
+        })
+        expect(host.send).toHaveBeenCalledTimes(1)
+        if (dispatchState === 'pending' || dispatchState === 'accepted') {
+          expect(receipt).toMatchObject({ state: 'ready', turnStart: 'unsupported' })
+          expect(host.closed).toHaveLength(0)
+          expect(host.release).not.toHaveBeenCalled()
+          expect(host.unsubscribe).not.toHaveBeenCalled()
+          expect(retireStructuredAgentSessionTabFromSnapshot).not.toHaveBeenCalled()
+        } else {
+          expect(receipt).toMatchObject({
+            state: dispatchState === 'unknown' ? 'outcome_unknown' : 'failed',
+            stage: 'dispatch_input'
+          })
+          expect(host.closed).toHaveLength(1)
+          expect(host.release).toHaveBeenCalledTimes(1)
+          expect(host.unsubscribe).toHaveBeenCalledTimes(1)
+          expect(retireStructuredAgentSessionTabFromSnapshot).toHaveBeenCalledTimes(1)
+        }
+      } finally {
+        releaseStructuredWorkerSession('d_fail', runtime)
+      }
+      expect(host.release).toHaveBeenCalledTimes(1)
+      expect(host.unsubscribe).toHaveBeenCalledTimes(1)
+    }
+  )
+})

src/main/runtime/rpc/methods/orchestration/worker/deliver-worker-dispatch-preamble.ts

@@ -11,7 +11,7 @@ type StructuredSession = Awaited<ReturnType<typeof createStructuredWorkerSession
  *
  * The preamble itself is identical for both: a worker is taught the same verbs whichever mode it
  * runs in, and only the delivery differs — a PTY write returns a queued/accepted receipt, while a
- * structured turn either is acknowledged or throws.
+ * structured turn either is admitted or throws.
  */
 export async function deliverWorkerDispatchPreamble(args: {
   runtime: OrcaRuntimeService

src/main/runtime/rpc/methods/orchestration/worker/worker-start-readiness-settlement.ts

@@ -71,10 +71,9 @@ export async function deliverAndSettleWorkerStartReadiness(args: {
   // The write above was accepted without waiting on provider hooks; now demand the positive
   // evidence the receipt claims is observable. A worker whose turn never starts must not be
   // reported ready — a wedged agent and a working one looked identical before this gate.
-  // A structured preamble send is acknowledged by the provider or throws, so it is already
-  // positive evidence.
+  // Structured sends prove transport admission; this path has no correlated turn-start observer.
   const turnStart: WorkerTurnStartObservation = structuredSession
-    ? { verdict: 'observed' }
+    ? { verdict: 'unsupported' }
     : await observeWorkerTurnStart({ runtime, terminalHandle, prompt: promptDelivery })
   const deliveredPrompt = turnStart.prompt ?? promptDelivery
   monitorWorkerSetup({

src/main/runtime/structured-agent-session-integration.test.ts

@@ -37,6 +37,7 @@ import { createTrackedJournalOpener } from '../native-chat/agent-session-journal
 import type { OrcaRuntimeService } from './orca-runtime'
 import type { RpcRequest, RpcResponse } from './rpc/core'
 import { RpcDispatcher } from './rpc/dispatcher'
+import { expectStructuredWorkerAdmission } from './structured-worker-admission.test-support'
 import { STRUCTURED_AGENT_SESSION_METHODS } from './rpc/methods/structured-agent-session'
 import {
   ensureStructuredAgentSessionHost,
@@ -433,6 +434,12 @@ describe('a structured codex session over agentSession.*', () => {
     })
   })
 
+  it('admits a fresh worker preamble before the provider echo without resending', async () => {
+    await ok('agentSession.create', createIntentParams())
+    await expectStructuredWorkerAdmission(SESSION)
+    expect(codex.live().calls.filter((call) => call.method === 'turn/start')).toHaveLength(1)
+  })
+
   it('dispatches and streams a plain first send from a fresh session', async () => {
     const created = await ok<{ fence: number; page: { items: unknown[] } }>(
       'agentSession.create',