diff --git a/src/main/claude/claude-structured-prompt-ownership.ts b/src/main/claude/claude-structured-prompt-ownership.ts index 8d7db516959..399dd98506a 100644 --- a/src/main/claude/claude-structured-prompt-ownership.ts +++ b/src/main/claude/claude-structured-prompt-ownership.ts @@ -111,13 +111,12 @@ export async function cancelClaudeStructuredTurn(input: { return { cancelled: false } } // Judge against the published journal, because that is the only turn a client could have been - // shown; direct adapter callers with no journal fall back to the in-memory turn, which can - // already name a row the sink has not drained. No live turn either way means nothing has - // published an identity this request can contradict. + // shown — but only while it HAS an answer. The journal drains through a serialized async queue, + // so a null read means the row has not landed yet, not that nothing is running; falling back to + // the in-memory turn there keeps Stop from being gated on bookkeeping. No live turn either way + // means nothing has published an identity this request can contradict. const ownsRequestedTurn = (): boolean => { - const liveTurnId = request.resolveLiveTurnId - ? request.resolveLiveTurnId() - : (session.translator?.currentTurnId ?? null) + const liveTurnId = request.resolveLiveTurnId?.() ?? session.translator?.currentTurnId ?? null return liveTurnId === null ? session.dispatchSequence === 0 : liveTurnId === request.turnId } // The host supplies the durable latest submission; direct adapter callers fall back to diff --git a/src/main/claude/claude-turn-ownership.test.ts b/src/main/claude/claude-turn-ownership.test.ts index 8daea7b438a..05c8aa97a3c 100644 --- a/src/main/claude/claude-turn-ownership.test.ts +++ b/src/main/claude/claude-turn-ownership.test.ts @@ -394,6 +394,25 @@ describe('Claude turn ownership', () => { expect(interrupt).toHaveBeenCalledOnce() }) + // The journal drains through a serialized async queue, so a live turn routinely has no published + // row yet. Refusing there would gate a user's Stop on bookkeeping, so the in-memory turn covers + // the lag — the journal is authoritative only while it has an answer. + it('admits a Stop for the live turn while the journal has not drained its row', async () => { + const session = sessionHoldingTurn('turn-live') + const interrupt = vi.fn().mockResolvedValue(undefined) + session.connection.interrupt = interrupt + + await expect( + cancellationOf(session, { + sessionId: 'session-1', + turnId: 'turn-live', + fence: 1, + resolveLiveTurnId: () => null + }) + ).resolves.toEqual({ cancelled: true }) + expect(interrupt).toHaveBeenCalledOnce() + }) + it('refuses a Stop the adapter still holds once the journal published a newer turn', async () => { const session = sessionHoldingTurn('turn-stale') const interrupt = vi.fn().mockResolvedValue(undefined) diff --git a/src/main/native-chat/agent-session-journal/journal-store.test.ts b/src/main/native-chat/agent-session-journal/journal-store.test.ts index 3592947faaa..ffaea5c5d6f 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.test.ts @@ -18,6 +18,7 @@ import { boundPayload, DEFAULT_JOURNAL_PAYLOAD_LIMITS } from './journal-payload-bounds' +import { activeStructuredAgentSessionTurnId } from '../../../shared/structured-agent-session-live-turn' import { journalDatabaseFile, journalDirectoryFor, journalPathSegment } from './journal-paths' import { AgentSessionJournalError, type AgentSessionJournal } from './journal-store' import type { openAgentSessionJournal } from './journal-store-factory' @@ -112,6 +113,45 @@ describe('sequences', () => { ]) }) + it('reads the live turn off reduced items, agreeing with the rendered snapshot', async () => { + const journal = await open() + const turnItem = (turnId: string): AgentJournalItemIdentity => ({ + provider: 'legacy', + agent: 'codex', + sessionId: 'session-1', + recordId: `turn-lifecycle:${turnId}` + }) + const rendered = (): string | null => + activeStructuredAgentSessionTurnId(journal.snapshot().items) + const bothAgreeOn = async (turnId: string | null): Promise => { + expect(journal.activeTurnId()).toBe(turnId) + expect(rendered()).toBe(turnId) + } + + await journal.appendItem( + turnItem('turn-1'), + { kind: 'turn', turnId: 'turn-1', state: 'running' }, + { fence: 1 } + ) + await journal.appendItem(item(0), body('work'), { fence: 1 }) + await bothAgreeOn('turn-1') + + // The completion is a revision, so it keeps the row's creation sequence rather than moving it. + await journal.appendItem( + turnItem('turn-1'), + { kind: 'turn', turnId: 'turn-1', state: 'completed' }, + { fence: 1 } + ) + await bothAgreeOn(null) + + await journal.appendItem( + turnItem('turn-2'), + { kind: 'turn', turnId: 'turn-2', state: 'running' }, + { fence: 1 } + ) + await bothAgreeOn('turn-2') + }) + it('preserves an oversized identity and its raw digest-form mimic across reopen', async () => { const oversizedTurnId = 'a'.repeat(MAX_JOURNAL_KEY_COMPONENT_CHARS + 1) const digestFormMimic = boundJournalKeyComponent(oversizedTurnId) diff --git a/src/main/native-chat/agent-session-journal/journal-store.ts b/src/main/native-chat/agent-session-journal/journal-store.ts index 2e372f9abae..b3aab6b1e3b 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.ts @@ -11,6 +11,7 @@ import type { AgentSessionJournalIdentity } from '../../../shared/agent-session-journal-types' import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key' +import { activeStructuredAgentSessionTurnIdBySequence } from '../../../shared/structured-agent-session-live-turn' import { agentSessionJournalCloseRetries } from './journal-close-retry' import { openJournalDatabase, type OpenJournalDatabase } from './journal-database' import type { JournalReplacementItem } from './journal-epoch-replacement' @@ -169,6 +170,11 @@ export class AgentSessionJournal { } } + /** The turn this journal has published as running — the same read a client's snapshot gives, + * without materialising one. */ + activeTurnId = (): string | null => + activeStructuredAgentSessionTurnIdBySequence(this.state.items.values()) + /** Includes revisions and completion tombstones, whose timestamps disappear from render items. */ lastActivityAt = (): number => this.state.lastActivityAt diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-handoff.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-handoff.ts index a778b252546..2720a97784c 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-handoff.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-handoff.ts @@ -20,7 +20,6 @@ import { StructuredTuiTranscriptCatchup } from './structured-tui-transcript-catc import { adapterSupportsCreateIfDeclared } from './structured-agent-session-provider-support' import { retryLoadedStructuredAgentSessionSettlement } from './structured-agent-session-settlement-retry' import { latestJournalDispatchObservation } from '../agent-session-journal/journal-dispatch-observation' -import { activeStructuredAgentSessionTurnId } from '../../../shared/structured-agent-session-projection' type HostHandoffAccess = { session: (sessionId: string) => StructuredAgentSessionHostSession @@ -174,7 +173,7 @@ export async function stopNativeHandoffTurn( await adapter.cancelTurn({ ...input, // The journal is what the client read to name a turn, so it is what judges the request. - resolveLiveTurnId: () => activeStructuredAgentSessionTurnId(session.journal.snapshot().items), + resolveLiveTurnId: () => session.journal.activeTurnId(), ...(dispatchStatus ? { dispatchStatus } : {}) }) ).cancelled diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-turns.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-turns.ts index 9fe1f5b8a17..2b10b6dba41 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-turns.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-turns.ts @@ -18,7 +18,6 @@ import type { import { DISPATCH_DOUBT_PERSISTENCE_FAILED } from '../agent-session-journal/journal-dispatch-doubt-reasons' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import { latestJournalDispatchObservation } from '../agent-session-journal/journal-dispatch-observation' -import { activeStructuredAgentSessionTurnId } from '../../../shared/structured-agent-session-projection' import type { AgentSessionDispatchOutcome, StructuredAgentSessionAdapter @@ -218,8 +217,7 @@ export async function performCancel( turnId: input.turnId, fence: ctx.fence, // The journal is what the client read to name a turn, so it is what judges the request. - resolveLiveTurnId: () => - activeStructuredAgentSessionTurnId(ctx.journal.snapshot().items), + resolveLiveTurnId: () => ctx.journal.activeTurnId(), ...(dispatchStatus ? { dispatchStatus } : {}), ...(input.prompt ? { prompt: { itemId: input.prompt.itemId } } : {}) }) diff --git a/src/shared/structured-agent-session-live-turn.ts b/src/shared/structured-agent-session-live-turn.ts index 319a6d7d579..52f3ee6a549 100644 --- a/src/shared/structured-agent-session-live-turn.ts +++ b/src/shared/structured-agent-session-live-turn.ts @@ -5,7 +5,8 @@ import type { AgentJournalRenderItem, - AgentJournalToolCallItem + AgentJournalToolCallItem, + AgentJournalTurnLifecycle } from './agent-session-journal-types' import { readAgentJournalTurn } from './agent-session-turn-record' @@ -21,6 +22,27 @@ export function activeStructuredAgentSessionTurnId( return null } +/** The same verdict for reduced items a caller holds unordered, so a reader that already has them + * need not render and sort a whole snapshot to ask. Sequence is the ordering key the render pass + * sorts on, and ties resolve to the later-reduced item exactly as that stable sort would. */ +export function activeStructuredAgentSessionTurnIdBySequence( + items: Iterable +): string | null { + let newestSequence = 0 + let newest: AgentJournalTurnLifecycle | null = null + for (const item of items) { + if (item.sequence < newestSequence) { + continue + } + const turn = readAgentJournalTurn(item.body) + if (turn) { + newestSequence = item.sequence + newest = turn + } + } + return newest?.state === 'running' ? newest.turnId : null +} + /** * Whether the newest thing the active turn produced is the model's own reasoning. *