fix(native-chat): a Stop Codex took settles the message whose turn never opened (follow-up to #25217)

src/main/codex/codex-structured-journal-translation-turn-boundaries.ts

@@ -174,14 +174,25 @@ export class CodexJournalTurnBoundaries {
       error: readString(readRecord(readRecord(readRecord(event.params).turn).error), 'message'),
       completedAt
     })
-    const turnLifecycle = this.ownsRecord(event.threadId, turnId)
-      ? this.settled(event.threadId, turnId, {
-          state: codexTurnLifecycleState(status),
-          outcome: codexTurnOutcome(status),
-          completedAt,
-          durationMs: readCodexTurnDurationMs(event.params)
-        })
-      : null
+    // Codex records a turn's prompt only once the turn starts, so an interrupted turn this child
+    // never started, with no item or prompt read, ran nothing and gets no record: the Stop that
+    // took it settles its send.
+    const ranNothing =
+      status === 'interrupted' &&
+      !this.deps.activeTurns.has(event.threadId, turnId) &&
+      !this.deps.items.ordinals.hasItems(event.threadId, turnId) &&
+      ![...this.deps.pendingPrompts.values()].some(
+        (prompt) => prompt.threadId === event.threadId && prompt.turnId === turnId
+      )
+    const turnLifecycle =
+      this.ownsRecord(event.threadId, turnId) && !ranNothing
+        ? this.settled(event.threadId, turnId, {
+            state: codexTurnLifecycleState(status),
+            outcome: codexTurnOutcome(status),
+            completedAt,
+            durationMs: readCodexTurnDurationMs(event.params)
+          })
+        : null
     const requestOrigin = this.deps.activeTurns.requestOrigin(event.threadId, turnId)
     const latestDispatchSequence = this.deps.activeTurns.latestDispatchSequence(
       event.threadId,

src/main/codex/codex-structured-journal-translation-turn-state.ts

@@ -57,6 +57,11 @@ export class CodexJournalActiveTurns {
     return [...(this.byThread.get(threadId) ?? [])].at(-1) ?? null
   }
 
+  /** Whether this child started the turn: its running record is this translator's. */
+  has(threadId: string, turnId: string): boolean {
+    return this.byThread.get(threadId)?.has(turnId) === true
+  }
+
   startedAt(threadId: string, turnId: string): number | undefined {
     return this.startedAtByTurn.get(this.turnKey(threadId, turnId))
   }

src/main/codex/codex-turn-ordinals.ts

@@ -100,6 +100,11 @@ export class CodexTurnOrdinals {
     }
   }
 
+  /** Whether any of the turn's items was read. */
+  hasItems(threadId: string, turnId: string): boolean {
+    return this.turns.has(this.turnKey(threadId, turnId))
+  }
+
   ordinalFor(threadId: string, turnId: string, codexItemId: string): number {
     const turnKey = this.turnKey(threadId, turnId)
     let turn = this.turns.get(turnKey)

src/main/native-chat/agent-session-wire/structured-agent-session-claude-opening-send-hold.test.ts

@@ -35,6 +35,7 @@ import {
 } from './structured-agent-session-host-test-data'
 import { recordingStructuredAgentSessionLogger } from './structured-agent-session-logger-test-support'
 import { claudeAndCodexDeclared } from './structured-agent-session-adapter-router-test-support'
+import { isStructuredAgentSessionMainAgentWorking } from '../../../shared/structured-agent-session-main-agent-working'
 import { DISPATCH_DOUBT_PROVIDER_IDLE } from '../agent-session-journal/journal-dispatch-doubt-reasons'
 import { claudeUnwrittenUserMessageError } from '../../claude/claude-agent-sdk-user-message-queue'
 
@@ -290,6 +291,52 @@ it('releases the follow-up when the CLI goes idle on a started send it never ech
   expect(submissions.find((entry) => entry.clientMessageId === steer)?.handedOverAt).toBeDefined()
 })
 
+/** A person's Stop naming no turn. */
+function stop() {
+  return host.cancel(CALLER, {
+    envelope: {
+      sessionId: SESSION,
+      clientOperationId: hostTestOperationId(),
+      expectedRuntimeFence: store.getRecord(SESSION)!.lease.runtimeFence,
+      payloadFingerprint: computeAgentSessionPayloadFingerprint({
+        method: 'agentSession.cancel',
+        sessionId: SESSION,
+        fields: {}
+      })
+    }
+  })
+}
+
+// Claude's interrupt names no turn, but its Stop ends the CLI, and that end settles the send.
+it('settles a send a Stop took before its echo, and the next message goes out', async () => {
+  const connection = claude.connections[0]!
+  const first = await send('FIRST prompt')
+  await eventually(() => expect(written(connection, 'FIRST prompt')).toBeDefined())
+
+  expect(await stop()).toMatchObject({ ok: true, value: { cancelled: true } })
+  await eventually(() => expect(connection.closeCount).toBe(1))
+  await eventually(async () =>
+    expect(
+      (await snapshot()).submissions.find((entry) => entry.clientMessageId === first)?.dispatchState
+    ).not.toBe('pending')
+  )
+  const after = await snapshot()
+  expect(
+    isStructuredAgentSessionMainAgentWorking(
+      null,
+      after.submissions,
+      store.getRecord(SESSION)!.lease.runtimeFence
+    )
+  ).toBe(false)
+
+  await send('NEXT prompt')
+  await eventually(() => {
+    const resumed = claude.connections.at(-1)!
+    expect(resumed).not.toBe(connection)
+    expect(written(resumed, 'NEXT prompt')).toBeDefined()
+  })
+})
+
 /** As the real connection: once a close begins it refuses every write; the first `failures`
  *  closes come back unproven. */
 function closeUnprovenFor(connection: FakeConnection, failures: number): void {
@@ -339,19 +386,7 @@ async function stoppedWithUnprovenClose(): Promise<FakeConnection> {
     ).toBe('accepted')
   )
   closeUnprovenFor(connection, 1)
-  const stopped = await host.cancel(CALLER, {
-    envelope: {
-      sessionId: SESSION,
-      clientOperationId: hostTestOperationId(),
-      expectedRuntimeFence: store.getRecord(SESSION)!.lease.runtimeFence,
-      payloadFingerprint: computeAgentSessionPayloadFingerprint({
-        method: 'agentSession.cancel',
-        sessionId: SESSION,
-        fields: {}
-      })
-    }
-  })
-  expect(stopped).toMatchObject({ ok: true })
+  expect(await stop()).toMatchObject({ ok: true })
   frame(connection, {
     type: 'result',
     subtype: 'error_during_execution',

src/main/native-chat/agent-session-wire/structured-agent-session-command-turn.ts

@@ -18,7 +18,8 @@ import {
 } from '../../../shared/agent-session-failure-words'
 import {
   agentJournalItemKey,
-  agentJournalSubmissionKey
+  agentJournalSubmissionKey,
+  parseAgentJournalItemKey
 } from '../../../shared/agent-session-journal-item-key'
 import {
   AGENT_JOURNAL_THREAD_SCOPE,
@@ -35,10 +36,6 @@ import {
   readAgentJournalTurn
 } from '../../../shared/agent-session-turn-record'
 import type { JournalLifecycleMutationInput } from '../agent-session-journal/journal-row-builders'
-import {
-  isStructuredAgentSessionStopNote,
-  structuredAgentSessionStopNoteIdentity
-} from '../../../shared/structured-agent-session-stop-note-key'
 import type { AgentSessionJournal } from '../agent-session-journal/journal-store'
 import type {
   AgentSessionCommandAdmission,
@@ -122,7 +119,18 @@ export function isStructuredAgentSessionCommandTurnId(turnId: string): boolean {
   return turnId.startsWith('compact:')
 }
 
-export { isStructuredAgentSessionStopNote, structuredAgentSessionStopNoteIdentity }
+const STOP_NOTE_PREFIX = 'stop:'
+
+/** A Stop's note, keyed by the turn it stopped (or, with no turn, by its operation). */
+export function structuredAgentSessionStopNoteIdentity(stopKey: string): AgentJournalItemIdentity {
+  return { provider: 'orca', clientMessageId: `${STOP_NOTE_PREFIX}${stopKey}` }
+}
+
+/** Whether a journal row is a Stop's note. */
+export function isStructuredAgentSessionStopNote(itemId: string): boolean {
+  const identity = parseAgentJournalItemKey(itemId)
+  return identity?.provider === 'orca' && identity.clientMessageId.startsWith(STOP_NOTE_PREFIX)
+}
 
 /** Whether an earlier Stop already asked the running command `turnId` names to end. Read from the
  *  journal, so nothing is held that could outlive the command. */

src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts

@@ -252,6 +252,123 @@ describe('a Stop that names no turn', () => {
     pending.resolve({ data: [], nextCursor: null })
   })
 
+  // Codex can abort a turn before it starts: it answers the interrupt, then ends the turn it never
+  // started. That turn ran nothing, so it draws no turn of its own, whichever of the two is read first.
+  it.each([
+    ['its end follows the answer', 'after'],
+    ['its end is read before the answer', 'before'],
+    ['no end comes', 'none']
+  ])('withdraws a send whose turn Codex never started when it takes a Stop: %s', async (_, end) => {
+    const codex = fakeCodex()
+    codex.routes['turn/start'] = () => ({ turn: { id: 'turn-1' } })
+    const ended = () =>
+      codex.connections[0].handlers.onNotification?.('turn/completed', {
+        threadId: THREAD_ID,
+        turn: { id: 'turn-1', status: 'interrupted' }
+      })
+    codex.routes['turn/interrupt'] = () => {
+      if (end === 'before') {
+        ended()
+      } else if (end === 'after') {
+        setTimeout(ended)
+      }
+      return {}
+    }
+    const adapter = adapterFor(codex, { codexHome: '/codex/home' })
+    await adapter.acquire({
+      identity: identityFor(SESSION),
+      fence: 1,
+      spawnToken: 'spawn-unstarted',
+      events: hostEvents()
+    })
+    dispatch.mockImplementation((input) => adapter.dispatch(input))
+    cancelTurn.mockImplementation((input) => adapter.cancelTurn(input))
+    const first = send('first send')
+    await first.result
+    await eventually(async () => expect((await submission(first.id))?.handedOverAt).toBeDefined())
+
+    expect(await stop()).toMatchObject({ ok: true, value: { cancelled: true } })
+    await new Promise((resolve) => setTimeout(resolve, 20))
+    await host.flushStreamedEvents(SESSION)
+
+    expect(await submission(first.id)).toMatchObject({
+      dispatchState: 'rejected',
+      reason: DISPATCH_REJECTED_CANCELLED
+    })
+    const { items } = await host.journalSnapshot(SESSION)
+    expect(items.filter((item) => item.body.kind === 'turn')).toEqual([])
+    expect(await statusRows()).toEqual([])
+  })
+
+  it('keeps the record of a turn Codex never started when something of it was read', async () => {
+    const codex = fakeCodex()
+    codex.routes['turn/start'] = () => ({ turn: { id: 'turn-1' } })
+    codex.routes['turn/interrupt'] = () => {
+      const notify = codex.connections[0].handlers.onNotification
+      notify?.('item/completed', {
+        threadId: THREAD_ID,
+        turnId: 'turn-1',
+        item: { type: 'agentMessage', id: 'message-1', text: 'partial' }
+      })
+      notify?.('turn/completed', {
+        threadId: THREAD_ID,
+        turn: { id: 'turn-1', status: 'interrupted' }
+      })
+      return {}
+    }
+    const adapter = adapterFor(codex, { codexHome: '/codex/home' })
+    await adapter.acquire({
+      identity: identityFor(SESSION),
+      fence: 1,
+      spawnToken: 'spawn-unstarted-item',
+      events: hostEvents()
+    })
+    dispatch.mockImplementation((input) => adapter.dispatch(input))
+    cancelTurn.mockImplementation((input) => adapter.cancelTurn(input))
+    const first = send('first send')
+    await first.result
+    await eventually(async () => expect((await submission(first.id))?.handedOverAt).toBeDefined())
+
+    expect(await stop()).toMatchObject({ ok: true, value: { cancelled: true } })
+    await host.flushStreamedEvents(SESSION)
+
+    const { items } = await host.journalSnapshot(SESSION)
+    expect(items.filter((item) => item.body.kind === 'turn').map((item) => item.body)).toEqual([
+      expect.objectContaining({ turnId: 'turn-1', state: 'interrupted' })
+    ])
+    // Something of it ran, so it is never taken back as unrun. Codex sends no item before a turn's
+    // start, so this order only guards the record; the send's own end is not this test's.
+    expect((await submission(first.id))?.dispatchState).not.toBe('rejected')
+  })
+
+  // Only an interrupted end ran nothing for certain: a failure keeps its record, which carries why.
+  it('keeps the record of a turn Codex failed without starting it', async () => {
+    const codex = fakeCodex()
+    codex.routes['turn/start'] = () => ({ turn: { id: 'turn-1' } })
+    const adapter = adapterFor(codex, { codexHome: '/codex/home' })
+    await adapter.acquire({
+      identity: identityFor(SESSION),
+      fence: 1,
+      spawnToken: 'spawn-unstarted-failed',
+      events: hostEvents()
+    })
+    dispatch.mockImplementation((input) => adapter.dispatch(input))
+    const first = send('first send')
+    await first.result
+    await eventually(async () => expect((await submission(first.id))?.handedOverAt).toBeDefined())
+
+    codex.connections[0].handlers.onNotification?.('turn/completed', {
+      threadId: THREAD_ID,
+      turn: { id: 'turn-1', status: 'failed', error: { message: 'model overloaded' } }
+    })
+    await host.flushStreamedEvents(SESSION)
+
+    const { items } = await host.journalSnapshot(SESSION)
+    expect(items.filter((item) => item.body.kind === 'turn').map((item) => item.body)).toEqual([
+      expect.objectContaining({ turnId: 'turn-1' })
+    ])
+  })
+
   it('fails a turn Codex refuses for a model picked before the list, then sends again', async () => {
     const pending = Promise.withResolvers<unknown>()
     const codex = fakeCodex({ 'model/list': () => pending.promise })
@@ -301,12 +418,20 @@ describe('a Stop that names no turn', () => {
     pending.resolve({ data: [], nextCursor: null })
   })
 
+  // The provider took the Stop on the turn it named and never opened it: no turn-ended event will
+  // come, so the Stop's answer settles the send, and its own row says it never started.
   it('delivers a newly accepted message after Stop withdraws the prior send', async () => {
     const first = send('first send')
     await first.result
     await eventually(() => expect(dispatch).toHaveBeenCalledOnce())
+    cancelTurn.mockResolvedValueOnce({ cancelled: true, turnId: 'turn-never-opened' })
 
     expect(await stop()).toMatchObject({ ok: true, value: { cancelled: true } })
+    expect(await submission(first.id)).toMatchObject({
+      dispatchState: 'rejected',
+      reason: DISPATCH_REJECTED_CANCELLED
+    })
+    expect(await statusRows()).toEqual([])
     const replacement = send('replacement send')
     await replacement.result
 
@@ -338,6 +463,69 @@ describe('a Stop that names no turn', () => {
     expect(note?.turnScope).toEqual({ kind: 'turn', turnItemId: record?.itemId })
   })
 
+  // Codex's own shape: the turn's start is read before the interrupt's answer, so the turn has a
+  // record, which the Stop ends; the send opened it and stays as it is.
+  it('ends the turn the provider opened before answering, and withdraws nothing', async () => {
+    const first = send('first send')
+    await first.result
+    await eventually(() => expect(dispatch).toHaveBeenCalledOnce())
+    cancelTurn.mockImplementationOnce(async () => {
+      hostEvents().appendItem(
+        { provider: 'legacy', agent: 'codex', sessionId: SESSION, recordId: 'turn:turn-1' },
+        { kind: 'turn', turnId: 'turn-1', state: 'running' },
+        { turnScope: AGENT_JOURNAL_THREAD_SCOPE }
+      )
+      return { cancelled: true, turnId: 'turn-1' }
+    })
+
+    expect(await stop()).toMatchObject({ ok: true, value: { cancelled: true } })
+
+    expect((await submission(first.id))?.dispatchState).toBe('pending')
+    const turn = (await host.journalSnapshot(SESSION)).items.find(
+      (item) => item.body.kind === 'turn'
+    )
+    expect(turn?.body).toMatchObject({ state: 'interrupted' })
+    expect(await statusRows()).toEqual(['Cancellation requested.'])
+  })
+
+  it('withdraws nothing on a refusal, even one that names the turn', async () => {
+    const first = send('first send')
+    await first.result
+    await eventually(async () => expect((await submission(first.id))?.handedOverAt).toBeDefined())
+    cancelTurn.mockResolvedValueOnce({
+      cancelled: false,
+      turnId: 'turn-never-opened',
+      refusal: {
+        detail: { text: 'no active turn to interrupt', audience: 'person' },
+        turnNotRunning: true
+      }
+    })
+
+    expect(await stop()).toMatchObject({ ok: true, value: { cancelled: false } })
+
+    expect((await submission(first.id))?.dispatchState).toBe('pending')
+  })
+
+  it('answers the Stop when settling the send it took fails, and reports the failure', async () => {
+    const first = send('first send')
+    await first.result
+    await eventually(() => expect(dispatch).toHaveBeenCalledOnce())
+    const journal = host.collaboratorsForTests().sessions.get(SESSION)?.journal
+    if (!journal) {
+      throw new Error('expected the conversation open')
+    }
+    vi.spyOn(journal, 'resolveDispatch').mockRejectedValueOnce(new Error('the journal threw'))
+    cancelTurn.mockResolvedValueOnce({ cancelled: true, turnId: 'turn-never-opened' })
+
+    expect(await stop()).toMatchObject({ ok: true, value: { cancelled: true } })
+
+    expect(log.entries).toContainEqual(
+      expect.objectContaining({
+        fields: expect.objectContaining({ scope: 'stop-settle', sessionId: SESSION })
+      })
+    )
+  })
+
   it('is a valid cancel, and only a plain Stop may omit the turn', () => {
     const envelope = {
       sessionId: SESSION,