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
1 change: 0 additions & 1 deletion config/scripts/ci-shard-timings.json
Original file line number Diff line number Diff line change
Expand Up @@ -1300,7 +1300,6 @@
"src/main/codex/codex-structured-session-shutdown.test.ts": 471,
"src/main/codex/codex-structured-thread-goal.test.ts": 73,
"src/main/codex/codex-structured-thread-open.test.ts": 300,
"src/main/codex/codex-structured-turn-processes.integration.test.ts": 194,
"src/main/codex/codex-subagent-execution-projection.test.ts": 202,
"src/main/codex/codex-subagent-executions.test.ts": 71,
"src/main/codex/codex-subagent-roster.test.ts": 118,
Expand Down
2 changes: 1 addition & 1 deletion src/main/codex/codex-prompt-registry-retention.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -83,7 +83,7 @@ describe('Codex prompt claim lifetime', () => {
await collect()
expect(old.deref()).toBeUndefined()
expect(registry.find(replacement.itemId)).toBe(replacement.prompt)
expect(registry.ownsBoundClaim(claim, replacement.itemId, 'thread', 'new-turn')).toBe(true)
expect(registry.ownsClaim(claim)).toBe(true)
registry.clearTurn('thread', 'new-turn')
expect(registry.ownsClaim(claim)).toBe(false)
})
Expand Down
17 changes: 0 additions & 17 deletions src/main/codex/codex-prompt-registry.ts
Original file line number Diff line number Diff line change
Expand Up @@ -191,23 +191,6 @@ export class CodexPromptRegistry {
return this.claims.get(claim.prompt) === claim && this.find(claim.itemId) === claim.prompt
}

ownsBoundClaim(
claim: CodexPromptClaim,
journalItemId: string,
threadId: string,
turnId: string
): boolean {
return (
claim.itemId === journalItemId &&
this.claims.get(claim.prompt) === claim &&
this.journalItemIds.get(journalItemId) ===
this.address(claim.prompt.threadId, claim.prompt.promptKey) &&
this.boundPrompts.get(journalItemId) === claim.prompt &&
claim.prompt.threadId === threadId &&
codexPromptMatchesTurn(claim.prompt, turnId)
)
}

releaseClaim(claim: CodexPromptClaim): void {
if (this.claims.get(claim.prompt) === claim) {
this.claims.delete(claim.prompt)
Expand Down
40 changes: 12 additions & 28 deletions src/main/codex/codex-structured-dispatch-boundary.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6,41 +6,25 @@ import {
fakeCodexAppServer
} from './codex-structured-dispatch-test-support'

it('checks continuation authority after process capture and before writing turn/start', async () => {
const capturing = Promise.withResolvers<void>()
const captured = Promise.withResolvers<null>()
it('checks continuation authority before writing turn/start', async () => {
const codex = fakeCodexAppServer()
const adapter = await acquiredCodexAdapter({
codex,
settlements: [],
captureTurnProcesses: () => {
capturing.resolve()
return captured.promise
}
})
let superseded = false
const adapter = await acquiredCodexAdapter({ codex, settlements: [] })
const beforeDispatch = vi.fn(async () => {
if (superseded) {
throw new AgentSessionPreDispatchError('agent_session_restart_work_superseded')
}
throw new AgentSessionPreDispatchError('agent_session_restart_work_superseded')
})
try {
const result = adapter.dispatch({
sessionId: 'session-1',
clientMessageId: 'continuation',
body: CODEX_TEST_USER_MESSAGE,
fence: 7,
beforeDispatch
})
const verdict = result.catch((error: unknown) => error)
await capturing.promise
superseded = true
captured.resolve(null)
expect(await verdict).toBeInstanceOf(AgentSessionPreDispatchError)
await expect(
adapter.dispatch({
sessionId: 'session-1',
clientMessageId: 'continuation',
body: CODEX_TEST_USER_MESSAGE,
fence: 7,
beforeDispatch
})
).rejects.toBeInstanceOf(AgentSessionPreDispatchError)
expect(beforeDispatch).toHaveBeenCalledOnce()
expect(codex.connections[0]?.calls.some((call) => call.method === 'turn/start')).toBe(false)
} finally {
captured.resolve(null)
await adapter.closeAll()
}
})
3 changes: 2 additions & 1 deletion src/main/codex/codex-structured-dispatch-echo.ts
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,8 @@ export type CodexDispatchEchoes = {
answeredUnopenedTurn: (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.
* completed, which echoes its pending input first, so one it never echoed waits for recovery. An
* interrupt withdraws an un-echoed send, steered or the turn's own input: neither reached history.
*/
endTurn: (threadId: string, turnId: string, end: CodexTurnEnd) => string[]
/** Submission origin for this exact send, retained until its echo settles it. */
Expand Down
4 changes: 0 additions & 4 deletions src/main/codex/codex-structured-dispatch-test-support.ts
Original file line number Diff line number Diff line change
Expand Up @@ -84,7 +84,6 @@ export async function acquiredCodexAdapter(input: {
codex: ReturnType<typeof fakeCodexAppServer>
settlements: LateSettlement[]
sink?: StructuredAgentSessionEventSink
captureTurnProcesses?: CodexStructuredSessionAdapterDeps['captureTurnProcesses']
requestTimeoutMs?: number
}): Promise<CodexStructuredSessionAdapter> {
const adapter = new CodexStructuredSessionAdapter({
Expand All @@ -97,9 +96,6 @@ export async function acquiredCodexAdapter(input: {
}),
openConnection: input.codex.openConnection,
readProcessStartTime: async () => 1_700_000_000_000,
captureTurnProcesses: input.captureTurnProcesses ?? (async () => null),
// A Stop must never reach for real processes on this machine under the fake pid.
terminateTurnProcesses: async () => true,
now: () => 1_700_000_000_500,
onDispatchSettledLate: (settlement) => input.settlements.push(settlement),
...(input.requestTimeoutMs === undefined ? {} : { requestTimeoutMs: input.requestTimeoutMs })
Expand Down
29 changes: 11 additions & 18 deletions src/main/codex/codex-structured-prompt-ownership.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -277,8 +277,7 @@ describe('Codex live prompt ownership', () => {
const codex = fakeCodex({
'turn/interrupt': () => completeTurn(codex, 'thread-child', 'child-turn')
})
const terminateTurnProcesses = vi.fn(async () => true)
const adapter = adapterFor(codex, {}, [], { terminateTurnProcesses })
const adapter = adapterFor(codex)
await adapter.acquire({
identity: identityFor('session-1'),
fence: 7,
Expand All @@ -298,7 +297,6 @@ describe('Codex live prompt ownership', () => {
method: 'turn/interrupt',
params: { threadId: 'thread-child', turnId: 'child-turn' }
})
expect(terminateTurnProcesses).not.toHaveBeenCalled()
expect(codex.connections[0]?.closed).toBe(false)
})

Expand Down Expand Up @@ -553,7 +551,7 @@ describe('Codex live prompt ownership', () => {
expect(commit).not.toHaveBeenCalled()
})

it('does not report success when a deferred provider completion is backpressured', async () => {
it("confirms on Codex's answer and keeps the prompt's claim when the turn's end is backpressured", async () => {
const recorded = lifecycleRecorder(true, false)
const codex = fakeCodex({
'turn/interrupt': () => {
Expand All @@ -580,7 +578,7 @@ describe('Codex live prompt ownership', () => {
fence: 7,
prompt: { itemId: promptItemId }
})
).rejects.toThrow(/deferred turn completion lifecycle was not admitted/)
).resolves.toEqual({ cancelled: true })
const commit = vi.fn(async () => undefined)
await expect(
adapter.answerPrompt({
Expand All @@ -597,7 +595,7 @@ describe('Codex live prompt ownership', () => {
await adapter.closeSession('session-1')
})

it('defers only the matching thread and emits its terminal event before cancel resolves', async () => {
it("publishes every thread's turn end as Codex sends it, ahead of the Stop's answer", async () => {
const interruptGate = deferred()
const events: CodexStructuredSessionEvent[] = []
const codex = fakeCodex({
Expand All @@ -610,18 +608,13 @@ describe('Codex live prompt ownership', () => {
const adapter = await acquired(codex, {}, events)
registerPrompt(adapter, codex, 'child-prompt', 'thread-child')

const cancellation = adapter
.cancelTurn({
sessionId: 'session-1',
turnId: 'turn-1',
fence: 7,
prompt: { itemId: 'child-prompt' }
})
.then((result) => {
expect(completionThreads(events)).toEqual([THREAD_ID, 'thread-child'])
return result
})
await vi.waitFor(() => expect(completionThreads(events)).toEqual([THREAD_ID]))
const cancellation = adapter.cancelTurn({
sessionId: 'session-1',
turnId: 'turn-1',
fence: 7,
prompt: { itemId: 'child-prompt' }
})
await vi.waitFor(() => expect(completionThreads(events)).toEqual([THREAD_ID, 'thread-child']))

interruptGate.resolve()
await expect(cancellation).resolves.toEqual({ cancelled: true })
Expand Down
50 changes: 19 additions & 31 deletions src/main/codex/codex-structured-prompt-ownership.ts
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@ import {
type CodexPreparedAnswer
} from './codex-structured-prompt-replies'
import { requireLiveCodexSession, type CodexSession } from './codex-structured-session-state'
import type { CodexStructuredTurnCancellation } from './codex-structured-turn-cancellation'
import { interruptCodexTurn } from './codex-structured-turn-cancellation'

type CancelInput = Parameters<StructuredAgentSessionAdapter['cancelTurn']>[0]
type AnswerInput = Parameters<StructuredAgentSessionAdapter['answerPrompt']>[0]
Expand All @@ -35,27 +35,23 @@ async function cancelCodexConversation(
input: Parameters<typeof cancelCodexStructuredTurn>[0],
session: CodexSession
): Promise<AgentSessionCancelOutcome> {
const { request, sessions, cancellation } = input
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
? ([...(session.activeTurnIds ?? [])].at(-1) ?? (await openedAnsweredTurn(session)))
: providerTurnId(session, liveTurnId)
if (!turnId) {
// 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
) {
return { cancelled: false }
}
const acquisitionGeneration = session.acquisitionGeneration
return cancellation.cancel(
session,
session.threadId,
turnId,
() =>
sessions.get(request.sessionId) === session &&
!session.ended &&
session.fence === request.fence &&
session.acquisitionGeneration === acquisitionGeneration
)
return interruptCodexTurn({ session, threadId: session.threadId, turnId, requestTimeoutMs })
}

/** The turn Codex answered a send into and has not opened yet, once it opens; null when it ends,
Expand All @@ -74,9 +70,9 @@ async function openedAnsweredTurn(session: CodexSession): Promise<string | null>
export async function cancelCodexStructuredTurn(input: {
request: CancelInput
sessions: Map<string, CodexSession>
cancellation: CodexStructuredTurnCancellation
requestTimeoutMs?: number
}): Promise<AgentSessionCancelOutcome> {
const { request, sessions, cancellation } = input
const { request, sessions, requestTimeoutMs } = input
const session = requireLiveCodexSession(sessions, request.sessionId)
const prompt = request.prompt
const requestedTurnId = request.turnId
Expand All @@ -88,12 +84,11 @@ export async function cancelCodexStructuredTurn(input: {
return { cancelled: false }
}
if (!prompt) {
return cancellation.cancel(session, session.threadId, turnId)
return interruptCodexTurn({ session, threadId: session.threadId, turnId, requestTimeoutMs })
}
if (session.fence !== request.fence) {
return { cancelled: false }
}
const acquisitionGeneration = session.acquisitionGeneration
const claim = session.prompts.claimBound(prompt.itemId)
const promptTurnId = claim?.prompt.turnId
if (!claim || !promptTurnId) {
Expand All @@ -102,25 +97,18 @@ export async function cancelCodexStructuredTurn(input: {
}
return { cancelled: false }
}
const isCurrent = (): boolean =>
sessions.get(request.sessionId) === session &&
!session.ended &&
session.fence === request.fence &&
session.acquisitionGeneration === acquisitionGeneration &&
providerTurnId(session, requestedTurnId) === turnId &&
session.prompts.ownsBoundClaim(claim, prompt.itemId, claim.prompt.threadId, promptTurnId)
let interruptConfirmed = false
try {
const result = await cancellation.cancel(
const result = await interruptCodexTurn({
session,
claim.prompt.threadId,
promptTurnId,
isCurrent,
() => {
threadId: claim.prompt.threadId,
turnId: promptTurnId,
requestTimeoutMs,
onConfirmed: () => {
interruptConfirmed = true
return session.translator?.cancelPrompt(prompt.itemId) ?? { accepted: true }
}
)
})
Comment on lines 100 to +111

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win

🔎 Supported by static analysis

🏁 Script executed:

sed -n '1,170p' src/main/codex/codex-structured-prompt-ownership.ts
sed -n '130,260p' src/main/codex/codex-prompt-registry.ts
rg -n 'cancelPrompt|claimBoundPrompt|ownsClaim|interruptConfirmed' src/main/codex
git diff 280733273545f0b3eeedc1be54b14d406239030e 179837a1aaecfd1f34fa9dcf30bbbbbdb6030741 -- src/main/codex/codex-structured-prompt-ownership.ts src/main/codex/codex-structured-turn-cancellation.ts

Repository: stablyai/orca

Length of output: 24879


🏁 Script executed:

set -e
printf '%s\n' '--- translator contract and implementation ---'
sed -n '1,120p' src/main/codex/codex-structured-journal-contracts.ts
sed -n '220,320p' src/main/codex/codex-structured-journal-translation.ts
rg -n -C 6 'cancelPrompt|function cancel|cancel\(|bindJournalItemId|clearTurn|forget\(|replace|replacement|turn/(completed|started)|activeTurnIds' src/main/codex --glob '*.ts'
printf '%s\n' '--- current cancellation tests ---'
rg -n -C 10 'cancelCodexStructuredTurn|interruptCodexTurn|cancelPrompt|prompt.*cancel|replacement|acquisitionGeneration|ownsBoundClaim' src/main/codex/codex-structured-prompt-ownership.test.ts src/main/codex --glob '*test.ts'
printf '%s\n' '--- exact base ownership implementation ---'
git show 280733273545f0b3eeedc1be54b14d406239030e:src/main/codex/codex-structured-prompt-ownership.ts | sed -n '1,145p'
printf '%s\n' '--- relevant current ownership and registry ranges ---'
cat -n src/main/codex/codex-structured-prompt-ownership.ts | sed -n '65,125p'
cat -n src/main/codex/codex-prompt-registry.ts | sed -n '145,225p'

Repository: stablyai/orca

Length of output: 45659


🏁 Script executed:

printf '%s\n' '--- cancelPrompt definitions ---'
rg -n -C 12 'cancelPrompt' src/main/codex
printf '%s\n' '--- prompt registry lifecycle calls ---'
rg -n -C 8 'bindJournalItemId|clearTurn|prompts\.forget|prompts\.clear|prompts\.cancel|cancel\(' src/main/codex --glob '*.ts'
printf '%s\n' '--- base cancellation wrapper ---'
git show 280733273545f0b3eeedc1be54b14d406239030e:src/main/codex/codex-structured-turn-cancellation.ts | sed -n '55,155p'

Repository: stablyai/orca

Length of output: 41780


🏁 Script executed:

set -e
printf '%s\n' '--- prompt cancellation implementation ---'
cat -n src/main/codex/codex-structured-journal-prompts.ts | sed -n '80,145p'
printf '%s\n' '--- cancellation and turn lifecycle tests ---'
rg -n -C 14 'cancel\(|cancelled|cancelPrompt|same-item|replacement|turn/completed|clearPromptTurn' src/main/codex/codex-structured-journal-translation-turn-lifecycle.test.ts src/main/codex/codex-structured-prompt-ownership.test.ts src/main/codex/codex-prompt-registry-retention.test.ts

Repository: stablyai/orca

Length of output: 42362


Preserve prompt ownership through interrupt confirmation.

turn/completed can clear the claimed prompt while turn/interrupt is pending. A replacement can reuse the same journal item ID. The onConfirmed callback then passes that ID to cancelPrompt, which can cancel the replacement and other prompts in its turn. Recheck ownership before cancelling.

Suggested fix
   if (!claim || !promptTurnId) {
     if (claim) {
       session.prompts.releaseClaim(claim)
     }
     return { cancelled: false }
   }
+  const acquisitionGeneration = session.acquisitionGeneration
   let interruptConfirmed = false
   try {
     const result = await interruptCodexTurn({
       session,
       threadId: claim.prompt.threadId,
       turnId: promptTurnId,
       requestTimeoutMs,
       onConfirmed: () => {
+        if (
+          sessions.get(request.sessionId) !== session ||
+          session.ended ||
+          session.fence !== request.fence ||
+          session.acquisitionGeneration !== acquisitionGeneration ||
+          providerTurnId(session, requestedTurnId) !== turnId ||
+          !session.prompts.ownsClaim(claim)
+        ) {
+          session.prompts.releaseClaim(claim)
+          return { accepted: true }
+        }
         interruptConfirmed = true
         return session.translator?.cancelPrompt(prompt.itemId) ?? { accepted: true }
       }
     })

if (!result.cancelled) {
session.prompts.releaseClaim(claim)
}
Expand Down
7 changes: 1 addition & 6 deletions src/main/codex/codex-structured-provider-events.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,30 +4,25 @@ import type { CodexJournalTranslationAdmission } from './codex-structured-journa
import * as codexRewind from './codex-structured-rewind'
import type { CodexSession, CodexStructuredSessionEvent } from './codex-structured-session-state'
import { readCodexThreadId } from './codex-structured-thread-facts'
import type { CodexStructuredTurnCancellation } from './codex-structured-turn-cancellation'

type EmitCodexEvent = (
session: CodexSession,
event: CodexStructuredSessionEvent
) => CodexJournalTranslationAdmission

/** One live notification's journal entry: rewind bookkeeping, cancellation deferral, delivery. */
/** One live notification's journal entry: rewind bookkeeping, delivery. */
export function translateCodexNotification(input: {
sessionId: string
session: CodexSession
method: string
params: unknown
observedAt?: number
dispatchSequenceAtReceipt?: number
turnCancellation: Pick<CodexStructuredTurnCancellation, 'handleNotification'>
emit: EmitCodexEvent
}): CodexJournalTranslationAdmission {
const { sessionId, session, method, params, observedAt, dispatchSequenceAtReceipt } = input
codexRewind.observeCodexRewindActivity(session, method, params)
session.turnOpenWaits.observe(session.threadId, method, params)
if (input.turnCancellation.handleNotification(sessionId, session, method, params, observedAt)) {
return { accepted: true }
}
return deliverCodexNotification(
sessionId,
session,
Expand Down
12 changes: 1 addition & 11 deletions src/main/codex/codex-structured-session-acquire.ts
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,6 @@ import {
type CodexSession,
type CodexStructuredSessionAdapterDeps
} from './codex-structured-session-state'
import type { CodexStructuredTurnCancellation } from './codex-structured-turn-cancellation'
import type { CodexStructuredSessionTeardown } from './codex-structured-session-teardown'
import type { CodexStructuredNotificationRetry } from './codex-structured-notification-retry'
import type { deliverCodexServerRequest } from './codex-structured-provider-events'
Expand All @@ -53,7 +52,6 @@ export async function acquireCodexStructuredSession(input: {
deps: CodexStructuredSessionAdapterDeps
sessions: Map<string, CodexSession>
acquisitions: CodexAcquisitionRegistry
turnCancellation: CodexStructuredTurnCancellation
notificationRetries: CodexStructuredNotificationRetry
deliver: (
acquisition: CodexAcquisitionAttempt['window'],
Expand All @@ -68,14 +66,7 @@ export async function acquireCodexStructuredSession(input: {
handleUnhandledFrame: (sessionId: string, kind: string, payload: unknown) => void
forceCloseUnexpected: CodexStructuredSessionTeardown['forceCloseUnexpected']
}): Promise<AgentSessionAcquisition> {
const {
input: acquireInput,
deps,
sessions,
acquisitions,
turnCancellation,
notificationRetries
} = input
const { input: acquireInput, deps, sessions, acquisitions, notificationRetries } = input
const sessionId = acquireInput.identity.sessionId
const { previousAttempt, attempt } = acquisitions.start(sessionId)
const acquisition = attempt.window
Expand Down Expand Up @@ -271,7 +262,6 @@ export async function acquireCodexStructuredSession(input: {
?.supportsFastMode
})
}
turnCancellation.register(session)
sessions.set(sessionId, session)
for (const event of acquisition.drain()) {
event()
Expand Down
Loading
Loading