Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions src/main/claude/claude-structured-conversation-stop.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -33,14 +33,14 @@ describe('Claude Stop that names no turn', () => {
expect(interrupts(claude)).toBe(1)
})

it('still refuses the placeholder a client used to name for that gap', async () => {
it('interrupts that gap the same way when the Stop names a turn that is not live', async () => {
const claude = fakeClaude({ replayUuid: null })
const adapter = await written(claude)

await expect(
adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-none', fence: 7 })
).resolves.toEqual({ cancelled: false })
expect(interrupts(claude)).toBe(0)
adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-ended', fence: 7 })
).resolves.toEqual({ cancelled: true })
expect(interrupts(claude)).toBe(1)
})

it('interrupts a turn Claude opened on its own echo', async () => {
Expand Down
94 changes: 30 additions & 64 deletions src/main/claude/claude-structured-prompt-ownership.ts
Original file line number Diff line number Diff line change
Expand Up @@ -15,11 +15,6 @@ import type { ClaudeSession } from './claude-structured-session-state'
import type { ClaudePendingPrompt } from './claude-prompt-registry'
import type { PermissionResult } from '@anthropic-ai/claude-agent-sdk'

/** Conservative user-facing window: below the 30s control deadline, trading
* residual slow-pump risk for ensuring delivery bookkeeping cannot block Stop indefinitely. */
export const CLAUDE_DISPATCH_ADMISSION_TIMEOUT_MS = 3_000
const CLAUDE_DISPATCH_ADMISSION_POLL_MS = 50

type CancelInput = Parameters<StructuredAgentSessionAdapter['cancelTurn']>[0]
type AnswerInput = Parameters<StructuredAgentSessionAdapter['answerPrompt']>[0]

Expand Down Expand Up @@ -55,42 +50,14 @@ function requireSession(sessions: Map<string, ClaudeSession>, sessionId: string)
return session
}

function waitForClaudeDispatchAdmission(
admitted: () => boolean,
timeoutMs = CLAUDE_DISPATCH_ADMISSION_TIMEOUT_MS
): Promise<boolean> {
return new Promise((resolve) => {
let settled = false
let deadline: ReturnType<typeof setTimeout> | null = null
let poll: ReturnType<typeof setInterval> | null = null
const finish = (value: boolean): void => {
if (settled) {
return
}
settled = true
if (deadline) {
clearTimeout(deadline)
}
if (poll) {
clearInterval(poll)
}
resolve(value)
}
const check = (): void => {
if (admitted()) {
finish(true)
}
}
deadline = setTimeout(() => finish(false), timeoutMs)
poll = setInterval(check, CLAUDE_DISPATCH_ADMISSION_POLL_MS)
check()
deadline.unref?.()
poll.unref?.()
})
/** The published journal's turn while it has one, else the in-memory turn (see the owner check). */
function claudeLiveTurnId(session: ClaudeSession, request: CancelInput): string | null {
return request.resolveLiveTurnId?.() ?? session.translator?.currentTurnId ?? null
}

/**
* A Stop that names no turn: the conversation asked to stop whatever this child has in flight.
* A Stop that names no turn, names the live one, or names one that ended before a written
* follow-up opened its own: the conversation asked to stop whatever this child has in flight.
* Claude's interrupt is session-scoped, so there is no turn identity to check — only that this is
* still the child the host judged, and that it has a turn open or a written message whose turn has
* not opened yet (the gap before its echo, which no client can name).
Expand All @@ -107,8 +74,7 @@ function cancelClaudeConversation(
sessions.get(request.sessionId) === session &&
session.fence === request.fence &&
session.acquisitionGeneration === acquisitionGeneration &&
((request.resolveLiveTurnId?.() ?? session.translator?.currentTurnId ?? null) !== null ||
session.dispatchWaiters.length > 0)
(claudeLiveTurnId(session, request) !== null || session.dispatchWaiters.length > 0)
return cancelClaudeTurn(session, timeoutMs, isCurrent, onDispatchSettledLate)
}

Expand All @@ -128,10 +94,27 @@ export async function cancelClaudeStructuredTurn(input: {
return { cancelled: false }
}
const requestedTurnId = request.turnId
const liveTurnId = claudeLiveTurnId(session, request)
// Naming the live turn asks for exactly what the conversation Stop interrupts, so a follow-up's
// pending handover is no reason to hold it; the queue sweep settles that follow-up. With nothing
// live, a written follow-up whose turn has not opened is one the naming client has not seen start.
if (
!prompt &&
(requestedTurnId === undefined ||
(session.translator?.commandTurnId !== requestedTurnId &&
(liveTurnId === requestedTurnId ||
(liveTurnId === null && session.dispatchWaiters.length > 0))))
) {
return cancelClaudeConversation(
session,
sessions,
request,
timeoutMs,
input.onDispatchSettledLate
)
}
if (requestedTurnId === undefined) {
return prompt
? { cancelled: false }
: cancelClaudeConversation(session, sessions, request, timeoutMs, input.onDispatchSettledLate)
return { cancelled: false }
}
if (prompt && session.fence !== request.fence) {
return { cancelled: false }
Expand All @@ -151,7 +134,7 @@ export async function cancelClaudeStructuredTurn(input: {
// 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?.() ?? session.translator?.currentTurnId ?? null
const liveTurnId = claudeLiveTurnId(session, request)
return liveTurnId === null ? session.dispatchSequence === 0 : liveTurnId === requestedTurnId
}
// The host supplies the durable latest submission; direct adapter callers fall back to
Expand All @@ -166,38 +149,21 @@ export async function cancelClaudeStructuredTurn(input: {
(waiter) => waiter.dispatchSequence === session.dispatchSequence
)
// Prompt cancellation has a separate callback-settlement contract, so only a provider with
// cancelQueued can release its uncertain queued send. Ordinary Stop gets a bounded escape below.
// cancelQueued can release its uncertain queued send.
const dispatchAdmissionAllowsCancellation = (): boolean =>
dispatchAdmissionIsCurrent() ||
(Boolean(prompt) && supportsClaudeQueuedInterruptCancellation(session))
const compactionOwnsTurn = (): boolean =>
session.translator !== null && session.translator.commandTurnId === requestedTurnId
const currentDispatchHasRetiredWaiter = (): boolean =>
session.retiredDispatchWaiters.some(
(waiter) => waiter.dispatchSequence === session.dispatchSequence
)
let dispatchAdmissionExpired = false
if (
!prompt &&
!compactionOwnsTurn() &&
!dispatchAdmissionAllowsCancellation() &&
(request.dispatchStatus !== undefined || currentDispatchHasRetiredWaiter())
) {
dispatchAdmissionExpired = !(await waitForClaudeDispatchAdmission(
dispatchAdmissionAllowsCancellation
))
}
const isCurrent = (): boolean =>
sessions.get(request.sessionId) === session &&
session.fence === request.fence &&
session.acquisitionGeneration === acquisitionGeneration &&
(claim && prompt
? ownsRequestedTurn() &&
session.prompts.ownsBoundClaim(claim, prompt.itemId, requestedTurnId) &&
(dispatchAdmissionAllowsCancellation() || dispatchAdmissionExpired)
: compactionOwnsTurn() ||
(ownsRequestedTurn() &&
(dispatchAdmissionAllowsCancellation() || dispatchAdmissionExpired)))
dispatchAdmissionAllowsCancellation()
: compactionOwnsTurn() || (ownsRequestedTurn() && dispatchAdmissionAllowsCancellation()))
let interruptConfirmed = false
try {
const result = await cancelClaudeTurn(
Expand Down
151 changes: 122 additions & 29 deletions src/main/claude/claude-structured-queued-stop.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,6 @@ import { agentJournalItemKey } from '../../shared/agent-session-journal-item-key
import { readAgentJournalTurn } from '../../shared/agent-session-turn-record'
import { DISPATCH_REJECTED_CANCELLED } from '../../shared/structured-agent-session-dispatch-rejection'
import type { ClaudeStructuredSessionAdapterDeps } from './claude-structured-session-state'
import { CLAUDE_DISPATCH_ADMISSION_TIMEOUT_MS } from './claude-structured-prompt-ownership'
import {
PROVIDER_SESSION_ID,
USER_MESSAGE,
Expand Down Expand Up @@ -82,50 +81,90 @@ async function acquired(
return { adapter, bodies, connection: claude.connections[0]! }
}

function runningTurnId(bodies: Map<string, AgentJournalItemBody>): string {
function openTurnId(bodies: Map<string, AgentJournalItemBody>): string | null {
for (const body of bodies.values()) {
const turn = readAgentJournalTurn(body)
if (turn?.state === 'running') {
return turn.turnId
}
}
throw new Error('expected a running turn')
return null
}

/** Sends A and opens its turn, queues B behind it, then Stops A as the host does. */
async function stopWithQueuedFollowUp(
cli: ReturnType<typeof claudeCli>,
openTurn: (connection: FakeConnection) => void
): Promise<{ interrupt: unknown; settlements: Settlement[] }> {
const settlements: Settlement[] = []
const { adapter, bodies, connection } = await acquired(cli.claude, settlements)
const send = (clientMessageId: string) =>
adapter.dispatch({ sessionId: 'session-1', clientMessageId, body: USER_MESSAGE, fence: 7 })
await send('client-a')
function runningTurnId(bodies: Map<string, AgentJournalItemBody>): string {
const turnId = openTurnId(bodies)
if (turnId === null) {
throw new Error('expected a running turn')
}
return turnId
}

function completeTurn(connection: FakeConnection, uuid: string): void {
connection.handlers.onMessage?.({
type: 'result',
subtype: 'success',
uuid,
session_id: PROVIDER_SESSION_ID,
is_error: false,
terminal_reason: 'completed',
duration_ms: 12
})
}

/** Writes a send and opens its turn with the echo, which adopts the client uuid. */
async function sendAndOpen(
adapter: ReturnType<typeof adapterFor>,
connection: FakeConnection,
clientMessageId: string,
openTurn: (connection: FakeConnection) => void = () => {}
): Promise<void> {
await send(adapter, clientMessageId)
openTurn(connection)
// Claude adopts the client uuid for the echo that opens the turn.
connection.handlers.onMessage?.({
...connection.sent.at(-1)!,
uuid: connection.sent.at(-1)!.uuid
})
await send('client-b')
cli.queue(String(connection.sent.at(-1)!.uuid))
}

function send(adapter: ReturnType<typeof adapterFor>, clientMessageId: string) {
return adapter.dispatch({ sessionId: 'session-1', clientMessageId, body: USER_MESSAGE, fence: 7 })
}

/** Stops `turnId` as the phone does, naming it, while the journal still reads the queued send as
* pending, and reads the outcome without moving the clock: that handover must hold no Stop. */
async function namedStop(adapter: ReturnType<typeof adapterFor>, turnId: string): Promise<unknown> {
vi.useFakeTimers()
try {
const stopped = adapter.cancelTurn({
sessionId: 'session-1',
turnId: runningTurnId(bodies),
fence: 7,
// The host reads B's handover from the journal, and it is still pending.
dispatchStatus: { state: 'pending', recovered: false }
})
await vi.advanceTimersByTimeAsync(CLAUDE_DISPATCH_ADMISSION_TIMEOUT_MS)
await expect(stopped).resolves.toEqual({ cancelled: true })
let outcome: unknown = 'still waiting'
void adapter
.cancelTurn({
sessionId: 'session-1',
turnId,
fence: 7,
dispatchStatus: { state: 'pending', recovered: false }
})
.then((result) => (outcome = result))
await vi.advanceTimersByTimeAsync(0)
return outcome
} finally {
vi.useRealTimers()
}
}

/** Sends A and opens its turn, queues B behind it, then Stops A by name as the phone does. */
async function stopWithQueuedFollowUp(
cli: ReturnType<typeof claudeCli>,
openTurn: (connection: FakeConnection) => void
): Promise<{ stopped: unknown; interrupt: unknown; settlements: Settlement[] }> {
const settlements: Settlement[] = []
const { adapter, bodies, connection } = await acquired(cli.claude, settlements)
await sendAndOpen(adapter, connection, 'client-a', openTurn)
await send(adapter, 'client-b')
cli.queue(String(connection.sent.at(-1)!.uuid))

const stopped = await namedStop(adapter, runningTurnId(bodies))
return {
stopped,
interrupt: connection.calls.find((call) => call.subtype === 'interrupt')?.params,
settlements: settlements.filter((settlement) => settlement.clientMessageId === 'client-b')
}
Expand All @@ -152,7 +191,11 @@ describe('Stop with a follow-up Claude queued behind the running turn', () => {
(connection) => connection.handlers.onMessage?.(turnInit(CAPABILITIES))
)

expect(byInit).toEqual({ interrupt: { cancelQueued: true }, settlements: WITHDRAWN })
expect(byInit).toEqual({
stopped: { cancelled: true },
interrupt: { cancelQueued: true },
settlements: WITHDRAWN
})
expect(bySessionStart).toEqual(byInit)
})

Expand All @@ -171,7 +214,11 @@ describe('Stop with a follow-up Claude queued behind the running turn', () => {
}
)

expect(result).toEqual({ interrupt: { cancelQueued: true }, settlements: WITHDRAWN })
expect(result).toEqual({
stopped: { cancelled: true },
interrupt: { cancelQueued: true },
settlements: WITHDRAWN
})
})

it('keeps what a turn advertised when the start is read after it', async () => {
Expand All @@ -191,7 +238,11 @@ describe('Stop with a follow-up Claude queued behind the running turn', () => {

const result = await stopWithQueuedFollowUp(cli, () => {})

expect(result).toEqual({ interrupt: { cancelQueued: true }, settlements: WITHDRAWN })
expect(result).toEqual({
stopped: { cancelled: true },
interrupt: { cancelQueued: true },
settlements: WITHDRAWN
})
})

it('settles a follow-up withdrawn one at a time on a CLI without cancel_queued', async () => {
Expand All @@ -200,6 +251,48 @@ describe('Stop with a follow-up Claude queued behind the running turn', () => {
() => {}
)

expect(result).toEqual({ interrupt: {}, settlements: WITHDRAWN })
expect(result).toEqual({ stopped: { cancelled: true }, interrupt: {}, settlements: WITHDRAWN })
})

it('still refuses a Stop that names an older turn than the one running', async () => {
const settlements: Settlement[] = []
const cli = claudeCli({ initProof: 'init', capabilities: CAPABILITIES })
const { adapter, bodies, connection } = await acquired(cli.claude, settlements)
await sendAndOpen(adapter, connection, 'client-a')
const olderTurnId = runningTurnId(bodies)
completeTurn(connection, 'result-a')
await sendAndOpen(adapter, connection, 'client-b')
const liveTurnId = runningTurnId(bodies)
await send(adapter, 'client-c')
cli.queue(String(connection.sent.at(-1)!.uuid))

expect(liveTurnId).not.toBe(olderTurnId)
await expect(namedStop(adapter, olderTurnId)).resolves.toEqual({ cancelled: false })
expect(connection.calls.some((call) => call.subtype === 'interrupt')).toBe(false)
expect(runningTurnId(bodies)).toBe(liveTurnId)
expect(settlements.filter((settlement) => settlement.clientMessageId === 'client-c')).toEqual(
[]
)
})

it('withdraws a follow-up whose turn has not opened when the turn the Stop names already ended', async () => {
const settlements: Settlement[] = []
const cli = claudeCli({ initProof: 'init', capabilities: CAPABILITIES })
const { adapter, bodies, connection } = await acquired(cli.claude, settlements)
await sendAndOpen(adapter, connection, 'client-a')
const shownTurnId = runningTurnId(bodies)
await send(adapter, 'client-b')
cli.queue(String(connection.sent.at(-1)!.uuid))
// The phone's copy still shows the first turn, which ends before its Stop lands.
completeTurn(connection, 'result-a')
expect(openTurnId(bodies)).toBeNull()

await expect(namedStop(adapter, shownTurnId)).resolves.toEqual({ cancelled: true })
expect(connection.calls.find((call) => call.subtype === 'interrupt')?.params).toEqual({
cancelQueued: true
})
expect(settlements.filter((settlement) => settlement.clientMessageId === 'client-b')).toEqual(
WITHDRAWN
)
})
})
Loading
Loading