Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
24 commits
Select commit Hold shift + click to select a range
486a4aa
fix(native-chat): a Stop binds only the turn it actually stopped
brennanb2025 Oct 2, 2026
4d5e13e
fix(native-chat): keep the person's queue pause and close binding aft…
brennanb2025 Oct 2, 2026
8447d44
Merge origin/main into stop-binds-at-settle
brennanb2025 Oct 2, 2026
9f61404
fix(native-chat): a Codex Stop agrees on both presses when a turn is …
brennanb2025 Oct 2, 2026
34acafa
Merge remote-tracking branch 'origin/main' into brennanb2025/stop-bin…
brennanb2025 Oct 3, 2026
67a06b8
Merge remote-tracking branch 'origin/main' into brennanb2025/stop-bin…
brennanb2025 Oct 3, 2026
057e364
Merge remote-tracking branch 'origin/main' into brennanb2025/stop-bin…
brennanb2025 Oct 3, 2026
06261d0
test(native-chat): a Stop whose event row failed binds no turn to an …
brennanb2025 Oct 3, 2026
60572cc
test(native-chat): name the settle, not a stream drain, in the Stop's…
brennanb2025 Oct 3, 2026
5fdc221
Merge remote-tracking branch 'origin/main' into brennanb2025/stop-bin…
brennanb2025 Oct 3, 2026
0855e21
test(native-chat): a Codex Stop answered before Codex ends the turn r…
brennanb2025 Oct 3, 2026
bd95932
fix(native-chat): a Codex Stop interrupts a turn Codex answered but h…
brennanb2025 Oct 3, 2026
ffd8d6a
Merge remote-tracking branch 'origin/main' into brennanb2025/stop-bin…
brennanb2025 Oct 3, 2026
980f0e7
fix(native-chat): every Codex Stop waits for an answered turn to star…
brennanb2025 Oct 3, 2026
a2ba0c5
fix(native-chat): a Codex Stop that ends the child before any turn op…
brennanb2025 Oct 3, 2026
a6b8185
fix(native-chat): derive the withdrawal of a Codex send whose turn ne…
brennanb2025 Oct 3, 2026
70601f5
fix(native-chat): keep a send in doubt when a turn was open for it
brennanb2025 Oct 3, 2026
8e3240e
fix(native-chat): withdraw a Codex send by whether it started its own…
brennanb2025 Oct 3, 2026
5b60057
fix(native-chat): leave no Stop row when the Stop took back a send th…
brennanb2025 Oct 3, 2026
b47e696
fix(native-chat): count a send whose answer was lost when a Stop take…
brennanb2025 Oct 4, 2026
22d2271
test(native-chat): pin which sends a Stop's child end can take back
brennanb2025 Oct 4, 2026
2daeeea
Merge remote-tracking branch 'origin/main' into brennanb2025/stop-bin…
brennanb2025 Oct 4, 2026
f0c185e
Merge remote-tracking branch 'origin/main' into brennanb2025/stop-bin…
brennanb2025 Oct 5, 2026
d8422dd
test(native-chat): read the outbox reconcile from where main moved it
brennanb2025 Oct 5, 2026
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
5 changes: 1 addition & 4 deletions src/main/claude/claude-message-journaling.ts
Original file line number Diff line number Diff line change
Expand Up @@ -63,9 +63,7 @@ export function journalClaudeMessage(
startsTurn: boolean,
observedAt: number,
/** Host clock on the submission that produced this send, when known. */
requestedAt?: number,
/** The submission this send echo acknowledged. */
openedBy?: string
requestedAt?: number
): boolean {
const envelope = readClaudeMessageEnvelope(message)
if (!envelope) {
Expand Down Expand Up @@ -178,7 +176,6 @@ export function journalClaudeMessage(
startsTurn,
observedAt,
...(requestedAt === undefined ? {} : { requestedAt }),
...(openedBy === undefined ? {} : { openedBy }),
userItemId: agentJournalItemKey(identity)
})
if (sendEchoTurn) {
Expand Down
5 changes: 0 additions & 5 deletions src/main/claude/claude-open-turn.ts
Original file line number Diff line number Diff line change
Expand Up @@ -60,11 +60,6 @@ export class ClaudeOpenTurn {
}

/** The open turn's row, where a fact about the running turn lands. */
/** The submission that opened the open turn, when known (`ClaudeCurrentTurn.openedBy`). */
get openedBy(): string | null {
return this.current?.openedBy ?? null
}

get identity(): AgentJournalItemIdentity | null {
return this.current ? claudeCurrentTurnIdentity(this.current) : null
}
Expand Down
4 changes: 1 addition & 3 deletions src/main/claude/claude-result-journaling.ts
Original file line number Diff line number Diff line change
Expand Up @@ -59,9 +59,7 @@ export function journalClaudeResult(
// Read before the settle below closes the turn: an error end the journal's Stop rule makes a
// person's cancellation is theirs to decide as the end is written (`turnEndAfterStop`).
const turnId = settlesTurn ? turn.id : null
const leftToStop =
turnId !== null &&
sink.journalStopDecidesTurn?.(turnId, observedAt, turn.openedBy ?? undefined) === true
const leftToStop = turnId !== null && sink.journalStopDecidesTurn?.(turnId, observedAt) === true
if (settlesTurn) {
prompts.retryPendingCancellations()
turn.suppressReopenOnFailure(message.is_error === true)
Expand Down
9 changes: 3 additions & 6 deletions src/main/claude/claude-structured-journal-translation.ts
Original file line number Diff line number Diff line change
Expand Up @@ -179,10 +179,8 @@ export function createClaudeJournalTranslator(
message: Record<string, unknown>,
startsTurn: boolean,
observedAt: number,
requestedAt?: number,
openedBy?: string
): boolean =>
journalClaudeMessage(messageContext, message, startsTurn, observedAt, requestedAt, openedBy)
requestedAt?: number
): boolean => journalClaudeMessage(messageContext, message, startsTurn, observedAt, requestedAt)

return {
handle: (event) => {
Expand Down Expand Up @@ -250,8 +248,7 @@ export function createClaudeJournalTranslator(
event.message,
event.startsTurn === true,
event.observedAt ?? Date.now(),
event.requestedAt,
event.clientMessageId
event.requestedAt
)
) {
providerFallback.append(
Expand Down
2 changes: 0 additions & 2 deletions src/main/claude/claude-turn-lifecycle-item.ts
Original file line number Diff line number Diff line change
Expand Up @@ -19,8 +19,6 @@ export type ClaudeCurrentTurn = {
/** Provider key of the user echo, or the lifecycle row itself when provider
* output opened a turn with no user row to receive its timing. */
userItemId: string
/** The submission whose send opened the turn, while its echo has yet to land in the journal. */
openedBy?: string
/** Present when the turn is the host's record of a conversation command. */
command?: ClaudeCommandTurn
}
Expand Down
5 changes: 1 addition & 4 deletions src/main/claude/claude-turn-opening.ts
Original file line number Diff line number Diff line change
Expand Up @@ -30,8 +30,6 @@ export type ClaudeSendEchoTurnInput = {
requestedAt?: number
/** Provider key of the user row this turn is anchored to. */
userItemId: string
/** The submission this echo acknowledged. */
openedBy?: string
}

/** The turn a replayed send echo opens, or null when this frame is not one. */
Expand All @@ -48,8 +46,7 @@ export function claudeTurnOpenedBySendEcho(
turnId: envelope.uuid,
startedAt: input.observedAt,
...(input.requestedAt === undefined ? {} : { requestedAt: input.requestedAt }),
userItemId: input.userItemId,
...(input.openedBy === undefined ? {} : { openedBy: input.openedBy })
userItemId: input.userItemId
}
: null
}
Expand Down
5 changes: 3 additions & 2 deletions src/main/codex/codex-structured-conversation-stop.test.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
// A Codex Stop that names no turn. The fake keeps Codex 0.157's turn bookkeeping: it answers
// `turn/start` before it opens the turn, and refuses an interrupt until then.
// `turn/start` before it opens the turn, and refuses an interrupt with no turn active until the
// thread runs.

import { describe, expect, it, vi } from 'vitest'
import { structuredAgentSessionCommandTurn } from '../native-chat/agent-session-wire/structured-agent-session-command-turn'
Expand Down Expand Up @@ -29,7 +30,7 @@ describe('a Codex Stop that names no turn', () => {
const rig = await codexTurnLifecycleRig()
await openedTurn(rig)

await expect(stop(rig, () => null)).resolves.toEqual({ cancelled: true })
await expect(stop(rig, () => null)).resolves.toEqual({ cancelled: true, turnId: 'turn-1' })
expect(rig.interrupts().map((call) => call.params?.turnId)).toEqual(['turn-1'])
})

Expand Down
31 changes: 20 additions & 11 deletions src/main/codex/codex-structured-dispatch-echo.ts
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,9 @@ export type CodexDispatchEchoes = {
answeredUnopenedTurn: (threadId: string, openTurnIds: ReadonlySet<string>) => string | null
/** Codex did not open this answered turn within a wait, so no later wait is spent on it. */
leftUnopened: (threadId: string, turnId: string) => void
/** The latest armed send's answered turn a wait left unopened, neither open nor ended: Codex may
* still open it, though no wait is spent on it again. */
answeredTurnLeftUnopened: (threadId: string, openTurnIds: ReadonlySet<string>) => string | null
/**
* Records a turn's end and returns the sends bound to it that it settles: all of them unless it
* completed, which echoes its pending input first, so one it never echoed waits for recovery. An
Expand All @@ -68,6 +71,19 @@ export function createCodexDispatchEchoes(): CodexDispatchEchoes {
let nextSequence = 0
const turnKey = (threadId: string, turnId: string): string => JSON.stringify([threadId, turnId])
const settles = (end: CodexTurnEnd): boolean => end.status !== 'completed'
const answeredTurn = (
threadId: string,
openTurnIds: ReadonlySet<string>,
admits: (key: string) => boolean
): string | null => {
const answered = [...armed.values()].flatMap(({ turn }) => {
const key = turn?.threadId === threadId ? turnKey(threadId, turn.turnId) : null
return turn && key && !openTurnIds.has(turn.turnId) && !endedTurns.has(key) && admits(key)
? [turn.turnId]
: []
})
return answered.at(-1) ?? null
}
return {
arm(clientMessageId, requestedAt) {
const existing = armed.get(clientMessageId)
Expand Down Expand Up @@ -97,17 +113,10 @@ export function createCodexDispatchEchoes(): CodexDispatchEchoes {
}
return end
},
answeredUnopenedTurn: (threadId, openTurnIds) => {
const answered = [...armed.values()].flatMap(({ turn }) =>
turn?.threadId === threadId &&
!openTurnIds.has(turn.turnId) &&
!endedTurns.has(turnKey(threadId, turn.turnId)) &&
!unopenedTurns.has(turnKey(threadId, turn.turnId))
? [turn.turnId]
: []
)
return answered.at(-1) ?? null
},
answeredUnopenedTurn: (threadId, openTurnIds) =>
answeredTurn(threadId, openTurnIds, (key) => !unopenedTurns.has(key)),
answeredTurnLeftUnopened: (threadId, openTurnIds) =>
answeredTurn(threadId, openTurnIds, (key) => unopenedTurns.has(key)),
leftUnopened: (threadId, turnId) => {
unopenedTurns.add(turnKey(threadId, turnId))
for (const oldest of unopenedTurns) {
Expand Down
2 changes: 1 addition & 1 deletion src/main/codex/codex-structured-prompt-ownership.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -509,7 +509,7 @@ describe('Codex live prompt ownership', () => {

await expect(
adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 })
).resolves.toEqual({ cancelled: true })
).resolves.toEqual({ cancelled: true, turnId: 'turn-1' })
expect(completionThreads(events)).toEqual([])
})

Expand Down
60 changes: 45 additions & 15 deletions src/main/codex/codex-structured-prompt-ownership.ts
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@ import {
} from './codex-structured-prompt-replies'
import { requireLiveCodexSession, type CodexSession } from './codex-structured-session-state'
import { interruptCodexTurn } from './codex-structured-turn-cancellation'
import { codexRunningOrOpeningTurn } from './codex-structured-turn-open-wait'
import { codexStopTarget, interruptOpeningCodexTurn } from './codex-structured-turn-open-wait'

type CancelInput = Parameters<StructuredAgentSessionAdapter['cancelTurn']>[0]
type AnswerInput = Parameters<StructuredAgentSessionAdapter['answerPrompt']>[0]
Expand All @@ -22,10 +22,25 @@ function providerTurnId(session: CodexSession, turnId: string): string | undefin
return session.translator ? session.translator.commandProviderTurnId(turnId) : turnId
}

/** A send whose answer this child lost: Codex may still open a turn for it that no Stop can name.
* A pending one Codex answered is not: its turn is the one the Stop interrupts. */
function sendInDoubt(dispatchStatus: CancelInput['dispatchStatus']): boolean {
return dispatchStatus?.state === 'unknown' && !dispatchStatus.recovered
}

/** The Stop's own answer, naming the journal turn it stopped when it took. */
function stoppedTurn(
outcome: AgentSessionCancelOutcome,
turnId: string
): AgentSessionCancelOutcome {
return outcome.cancelled ? { ...outcome, turnId } : outcome
}

/**
* A Stop that names no turn: interrupt the turn the journal shows, or else the one Codex reported
* started and not yet ended, which the journal can trail by a publish, or else the one Codex
* answered a send into, once it opens.
* answered a send into, at once (`interruptOpeningCodexTurn`). A turn that may still open but
* cannot be reached, or a send whose answer was lost, answers refused, and the host ends the child.
*/
async function cancelCodexConversation(
input: Parameters<typeof cancelCodexStructuredTurn>[0],
Expand All @@ -34,20 +49,32 @@ async function cancelCodexConversation(
const { request, sessions, requestTimeoutMs } = input
const liveTurnId = request.resolveLiveTurnId?.() ?? null
// A turn the journal shows that Codex has not started yet (a compaction's) has nothing to stop.
const turnId =
liveTurnId === null
? await codexRunningOrOpeningTurn(session)
: providerTurnId(session, liveTurnId)
// The wait for the turn to open can outlive the session it began on.
if (
!turnId ||
sessions.get(request.sessionId) !== session ||
session.ended ||
session.fence !== request.fence
) {
const target = liveTurnId === null ? codexStopTarget(session) : { turnId: liveTurnId }
// Read again after any wait for the turn to open, which can outlive the session it began on.
const current = (): boolean =>
sessions.get(request.sessionId) === session && !session.ended && session.fence === request.fence
if (!current()) {
return { cancelled: false }
}
if (target === null) {
return sendInDoubt(request.dispatchStatus)
? { cancelled: false, refusal: { turnMayOpen: true } }
: { cancelled: false }
}
const journalTurnId = 'opening' in target ? target.opening : target.turnId
const turnId = liveTurnId === null ? journalTurnId : providerTurnId(session, liveTurnId)
if (!turnId) {
return { cancelled: false }
}
return interruptCodexTurn({ session, threadId: session.threadId, turnId, requestTimeoutMs })
const interrupt = () =>
interruptCodexTurn({ session, threadId: session.threadId, turnId, requestTimeoutMs })
if (!('opening' in target)) {
return stoppedTurn(await interrupt(), journalTurnId)
}
return stoppedTurn(
await interruptOpeningCodexTurn(session, turnId, interrupt, current),
journalTurnId
)
}

export async function cancelCodexStructuredTurn(input: {
Expand All @@ -67,7 +94,10 @@ export async function cancelCodexStructuredTurn(input: {
return { cancelled: false }
}
if (!prompt) {
return interruptCodexTurn({ session, threadId: session.threadId, turnId, requestTimeoutMs })
return stoppedTurn(
await interruptCodexTurn({ session, threadId: session.threadId, turnId, requestTimeoutMs }),
requestedTurnId
)
}
if (session.fence !== request.fence) {
return { cancelled: false }
Expand Down
6 changes: 3 additions & 3 deletions src/main/codex/codex-structured-session-cancel.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -143,7 +143,7 @@ describe('CodexStructuredSessionAdapter.cancelTurn', () => {

await expect(
adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 })
).resolves.toEqual({ cancelled: true })
).resolves.toEqual({ cancelled: true, turnId: 'turn-1' })
expect(codex.connections[0].calls.at(-1)).toEqual({
method: 'turn/interrupt',
params: { threadId: THREAD_ID, turnId: 'turn-1' }
Expand Down Expand Up @@ -288,7 +288,7 @@ describe('a Codex Stop is the interrupt alone', () => {
params: expect.objectContaining({ turn: { id: 'turn-1', status: 'interrupted' } })
})
)
await expect(stopped).resolves.toEqual({ cancelled: true })
await expect(stopped).resolves.toEqual({ cancelled: true, turnId: 'turn-1' })
})

it('reads and kills no processes, from the send through the Stop', async () => {
Expand All @@ -299,7 +299,7 @@ describe('a Codex Stop is the interrupt alone', () => {
for (const work of Object.values(processWork)) {
expect(work).not.toHaveBeenCalled()
}
await expect(stopped).resolves.toEqual({ cancelled: true })
await expect(stopped).resolves.toEqual({ cancelled: true, turnId: 'turn-1' })
for (const work of Object.values(processWork)) {
expect(work).not.toHaveBeenCalled()
}
Expand Down
Loading
Loading