mirror of
https://github.com/stablyai/orca.git
synced 2026-10-26 16:02:42 +00:00
Run a Codex session under the agent-session lease
Implements the structured agent-session adapter for Codex: one long-lived app-server child per session, started or resumed under the lease the host already reserved, reporting the process identity the lease re-proves and the provider handle link the journal binds to. The thread the adapter reports is the one Codex actually opened, never the one a client declared, and a resume that lands on a different thread is refused rather than recorded as a resume. A turn is named from the start response when Codex sends an id there and from the turn-started notification when it acks first; when neither names it in time the submission settles unknown, which the journal renders as delivery unconfirmed rather than failure. Only a refusal Codex actually answered becomes a rejection. Model, effort, and policy are per-turn overrides — there is no thread-settings write — so an option set between turns is applied to the next turn/start, and unlisted keys are refused instead of becoming arbitrary client-controlled parameters.
This commit is contained in:
Brennan Benson
@@ -0,0 +1,48 @@
|
||||
import type { AgentSessionJournalIdentity } from '../../shared/agent-session-journal-types'
|
||||
import type { AgentSessionProviderHandleLink } from '../../shared/agent-session-provider-handle'
|
||||
import type { AgentSessionProcessIdentity } from '../../shared/agent-session-record'
|
||||
import { readProcessStartTimeMs } from '../runtime/agent-session-process-identity-probe'
|
||||
|
||||
// What the lease records about the child Codex just handed back: the process it
|
||||
// will later re-prove, and the provider handle link the journal binds to. Both
|
||||
// must describe the thread Codex actually opened, never the one a client asked
|
||||
// for.
|
||||
|
||||
/** The child echoes its spawn token here so the owner probe can tell a live
|
||||
* child of THIS reservation from a same-pid stranger. */
|
||||
export const CODEX_SPAWN_TOKEN_ENV = 'ORCA_AGENT_SESSION_SPAWN_TOKEN'
|
||||
|
||||
export async function codexProcessIdentity(
|
||||
input: {
|
||||
identity: AgentSessionJournalIdentity
|
||||
spawnToken: string
|
||||
pid: number | undefined
|
||||
},
|
||||
readStartTime: (pid: number) => Promise<number | null> = readProcessStartTimeMs
|
||||
): Promise<AgentSessionProcessIdentity> {
|
||||
if (input.pid === undefined) {
|
||||
throw new Error('codex app-server started without a pid')
|
||||
}
|
||||
return {
|
||||
hostId: input.identity.hostId,
|
||||
pid: input.pid,
|
||||
processStartTimeMs: await readStartTime(input.pid),
|
||||
spawnToken: input.spawnToken
|
||||
}
|
||||
}
|
||||
|
||||
export function codexProviderHandleLink(input: {
|
||||
threadId: string
|
||||
resumed: boolean
|
||||
fence: number
|
||||
linkId?: string
|
||||
observedAt: number
|
||||
}): AgentSessionProviderHandleLink {
|
||||
return {
|
||||
linkId: input.linkId ?? `codex-${input.fence}-${input.threadId}`.slice(0, 128),
|
||||
handle: { provider: 'codex', threadId: input.threadId },
|
||||
origin: input.resumed ? 'resumed' : 'created',
|
||||
mintedAtFence: input.fence,
|
||||
observedAt: input.observedAt
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,600 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import type {
|
||||
AgentJournalMessageItem,
|
||||
AgentSessionJournalIdentity
|
||||
} from '../../shared/agent-session-journal-types'
|
||||
import { CodexAppServerRequestError } from './codex-app-server-connection'
|
||||
import type {
|
||||
CodexAppServerConnection,
|
||||
CodexAppServerConnectionHandlers,
|
||||
CodexAppServerLaunch,
|
||||
openCodexAppServerConnection
|
||||
} from './codex-app-server-connection'
|
||||
import { CodexAppServerUnsupportedError } from './codex-app-server-session'
|
||||
import { CODEX_SPAWN_TOKEN_ENV } from './codex-structured-owner-identity'
|
||||
import { encodeCodexQuestionOptionId } from './codex-structured-prompt-replies'
|
||||
import {
|
||||
CodexStructuredSessionAdapter,
|
||||
type CodexStructuredLaunch,
|
||||
type CodexStructuredSessionEvent
|
||||
} from './codex-structured-session-adapter'
|
||||
|
||||
const THREAD_ID = 'thread-abc'
|
||||
|
||||
function identityFor(sessionId: string): AgentSessionJournalIdentity {
|
||||
return {
|
||||
sessionId,
|
||||
workspaceId: 'ws-1',
|
||||
hostId: 'host-1',
|
||||
agent: 'codex',
|
||||
providerHandle: { kind: 'codex', threadId: THREAD_ID }
|
||||
}
|
||||
}
|
||||
|
||||
const USER_MESSAGE: AgentJournalMessageItem = {
|
||||
kind: 'message',
|
||||
role: 'user',
|
||||
blocks: [{ type: 'text', text: 'ship it' }]
|
||||
}
|
||||
|
||||
type Route = (params: Record<string, unknown> | undefined) => unknown
|
||||
|
||||
type FakeConnection = CodexAppServerConnection & {
|
||||
launch: CodexAppServerLaunch
|
||||
handlers: CodexAppServerConnectionHandlers
|
||||
calls: { method: string; params?: Record<string, unknown> }[]
|
||||
replies: { id: number | string; result?: unknown; code?: number; message?: string }[]
|
||||
closeCount: number
|
||||
}
|
||||
|
||||
/** Stands in for a live `codex app-server`: every RPC is answered from `routes`,
|
||||
* and the test drives Codex's own traffic through `handlers`. */
|
||||
function fakeCodex(routes: Record<string, Route> = {}): {
|
||||
connections: FakeConnection[]
|
||||
openConnection: typeof openCodexAppServerConnection
|
||||
routes: Record<string, Route>
|
||||
} {
|
||||
const connections: FakeConnection[] = []
|
||||
const openConnection = (async (launch, handlers = {}) => {
|
||||
const connection: FakeConnection = {
|
||||
launch,
|
||||
handlers,
|
||||
calls: [],
|
||||
replies: [],
|
||||
closeCount: 0,
|
||||
pid: 4321,
|
||||
closed: false,
|
||||
request: async (method, params) => {
|
||||
connection.calls.push({ method, params })
|
||||
const route = routes[method]
|
||||
return route ? route(params) : {}
|
||||
},
|
||||
notify: () => {},
|
||||
respond: (id, result) => connection.replies.push({ id, result }),
|
||||
respondWithError: (id, code, message) => connection.replies.push({ id, code, message }),
|
||||
close: async () => {
|
||||
connection.closeCount += 1
|
||||
}
|
||||
}
|
||||
connections.push(connection)
|
||||
return connection
|
||||
}) as typeof openCodexAppServerConnection
|
||||
routes['thread/start'] ??= () => ({ thread: { id: THREAD_ID, path: '/rollouts/abc.jsonl' } })
|
||||
routes['thread/resume'] ??= (params) => ({
|
||||
thread: { id: (params as { threadId: string }).threadId }
|
||||
})
|
||||
return { connections, openConnection, routes }
|
||||
}
|
||||
|
||||
function adapterFor(
|
||||
codex: ReturnType<typeof fakeCodex>,
|
||||
launch: Partial<CodexStructuredLaunch> = {},
|
||||
events: CodexStructuredSessionEvent[] = []
|
||||
): CodexStructuredSessionAdapter {
|
||||
return new CodexStructuredSessionAdapter({
|
||||
resolveLaunch: async () => ({
|
||||
command: 'codex',
|
||||
args: ['app-server'],
|
||||
cwd: '/work/repo',
|
||||
codexHome: null,
|
||||
resumeThreadId: null,
|
||||
...launch
|
||||
}),
|
||||
onEvent: (event) => events.push(event),
|
||||
openConnection: codex.openConnection,
|
||||
readProcessStartTime: async () => 1_700_000_000_000,
|
||||
now: () => 1_700_000_000_500
|
||||
})
|
||||
}
|
||||
|
||||
async function acquired(
|
||||
codex: ReturnType<typeof fakeCodex>,
|
||||
launch: Partial<CodexStructuredLaunch> = {},
|
||||
events: CodexStructuredSessionEvent[] = []
|
||||
): Promise<CodexStructuredSessionAdapter> {
|
||||
const adapter = adapterFor(codex, launch, events)
|
||||
await adapter.acquire({ identity: identityFor('session-1'), fence: 7, spawnToken: 'spawn-9' })
|
||||
return adapter
|
||||
}
|
||||
|
||||
describe('CodexStructuredSessionAdapter.acquire', () => {
|
||||
it('starts a new thread and reports the process and link the lease will prove', async () => {
|
||||
const codex = fakeCodex()
|
||||
const adapter = adapterFor(codex, { codexHome: '/codex/home' })
|
||||
|
||||
const acquisition = await adapter.acquire({
|
||||
identity: identityFor('session-1'),
|
||||
fence: 7,
|
||||
spawnToken: 'spawn-9'
|
||||
})
|
||||
|
||||
expect(codex.connections[0].launch.env).toEqual({
|
||||
[CODEX_SPAWN_TOKEN_ENV]: 'spawn-9',
|
||||
CODEX_HOME: '/codex/home'
|
||||
})
|
||||
expect(codex.connections[0].calls[0]).toEqual({
|
||||
method: 'thread/start',
|
||||
params: { cwd: '/work/repo' }
|
||||
})
|
||||
expect(acquisition.process).toEqual({
|
||||
hostId: 'host-1',
|
||||
pid: 4321,
|
||||
processStartTimeMs: 1_700_000_000_000,
|
||||
spawnToken: 'spawn-9'
|
||||
})
|
||||
expect(acquisition.link).toEqual({
|
||||
linkId: `codex-7-${THREAD_ID}`,
|
||||
handle: { provider: 'codex', threadId: THREAD_ID },
|
||||
origin: 'created',
|
||||
mintedAtFence: 7,
|
||||
observedAt: 1_700_000_000_500
|
||||
})
|
||||
})
|
||||
|
||||
it('resumes the thread the durable handle chain names, not the client one', async () => {
|
||||
const codex = fakeCodex()
|
||||
const adapter = adapterFor(codex, { resumeThreadId: 'thread-proven' })
|
||||
|
||||
const acquisition = await adapter.acquire({
|
||||
identity: identityFor('session-1'),
|
||||
fence: 9,
|
||||
spawnToken: 'spawn-9'
|
||||
})
|
||||
|
||||
expect(codex.connections[0].calls[0]).toEqual({
|
||||
method: 'thread/resume',
|
||||
params: { threadId: 'thread-proven', cwd: '/work/repo' }
|
||||
})
|
||||
expect(acquisition.link.origin).toBe('resumed')
|
||||
expect(acquisition.link.handle).toEqual({ provider: 'codex', threadId: 'thread-proven' })
|
||||
})
|
||||
|
||||
it('refuses a resume that lands on a different thread and reaps the child', async () => {
|
||||
const codex = fakeCodex({ 'thread/resume': () => ({ thread: { id: 'thread-other' } }) })
|
||||
const adapter = adapterFor(codex, { resumeThreadId: 'thread-proven' })
|
||||
|
||||
await expect(
|
||||
adapter.acquire({ identity: identityFor('session-1'), fence: 9, spawnToken: 'spawn-9' })
|
||||
).rejects.toThrow('resumed thread-other instead of thread-proven')
|
||||
expect(codex.connections[0].closeCount).toBe(1)
|
||||
})
|
||||
|
||||
it('refuses a thread Codex never named', async () => {
|
||||
const codex = fakeCodex({ 'thread/start': () => ({}) })
|
||||
const adapter = adapterFor(codex)
|
||||
|
||||
await expect(
|
||||
adapter.acquire({ identity: identityFor('session-1'), fence: 1, spawnToken: 'spawn-9' })
|
||||
).rejects.toThrow('did not name the thread')
|
||||
expect(codex.connections[0].closeCount).toBe(1)
|
||||
})
|
||||
|
||||
it('closes the previous child before re-acquiring at a new fence', async () => {
|
||||
const codex = fakeCodex()
|
||||
const adapter = await acquired(codex)
|
||||
|
||||
await adapter.acquire({ identity: identityFor('session-1'), fence: 8, spawnToken: 'spawn-10' })
|
||||
|
||||
expect(codex.connections).toHaveLength(2)
|
||||
expect(codex.connections[0].closeCount).toBe(1)
|
||||
expect(codex.connections[1].closeCount).toBe(0)
|
||||
})
|
||||
|
||||
it('reports the rollout path Codex named, and null when it named none', async () => {
|
||||
const withPath = fakeCodex()
|
||||
const adapter = await acquired(withPath)
|
||||
expect(await adapter.historyFilePath({ identity: identityFor('session-1') })).toBe(
|
||||
'/rollouts/abc.jsonl'
|
||||
)
|
||||
|
||||
const withoutPath = fakeCodex({ 'thread/start': () => ({ thread: { id: THREAD_ID } }) })
|
||||
const bare = await acquired(withoutPath)
|
||||
expect(await bare.historyFilePath({ identity: identityFor('session-1') })).toBeNull()
|
||||
})
|
||||
})
|
||||
|
||||
describe('CodexStructuredSessionAdapter.dispatch', () => {
|
||||
it('accepts a turn Codex names in its response', async () => {
|
||||
const codex = fakeCodex({ 'turn/start': () => ({ turn: { id: 'turn-1' } }) })
|
||||
const adapter = await acquired(codex)
|
||||
|
||||
const outcome = await adapter.dispatch({
|
||||
sessionId: 'session-1',
|
||||
clientMessageId: 'client-1',
|
||||
body: {
|
||||
kind: 'message',
|
||||
role: 'user',
|
||||
blocks: [
|
||||
{ type: 'text', text: 'ship it' },
|
||||
{ type: 'image-ref', path: '/tmp/shot.png' },
|
||||
{ type: 'image-ref', url: 'https://example.test/a.png' }
|
||||
]
|
||||
},
|
||||
fence: 7
|
||||
})
|
||||
|
||||
expect(outcome).toEqual({
|
||||
state: 'accepted',
|
||||
providerIdentity: { provider: 'codex', threadId: THREAD_ID, turnId: 'turn-1', ordinal: 0 }
|
||||
})
|
||||
expect(codex.connections[0].calls[1].params).toEqual({
|
||||
threadId: THREAD_ID,
|
||||
clientUserMessageId: 'client-1',
|
||||
input: [
|
||||
{ type: 'text', text: 'ship it' },
|
||||
{ type: 'localImage', path: '/tmp/shot.png' },
|
||||
{ type: 'image', url: 'https://example.test/a.png' }
|
||||
]
|
||||
})
|
||||
})
|
||||
|
||||
it('accepts a turn named only by the notification that raced the ack', async () => {
|
||||
const codex = fakeCodex()
|
||||
const events: CodexStructuredSessionEvent[] = []
|
||||
const adapter = await acquired(codex, {}, events)
|
||||
codex.routes['turn/start'] = () => {
|
||||
codex.connections[0].handlers.onNotification?.('turn/started', {
|
||||
threadId: THREAD_ID,
|
||||
turn: { id: 'turn-late' }
|
||||
})
|
||||
return {}
|
||||
}
|
||||
|
||||
const outcome = await adapter.dispatch({
|
||||
sessionId: 'session-1',
|
||||
clientMessageId: 'client-1',
|
||||
body: USER_MESSAGE,
|
||||
fence: 7
|
||||
})
|
||||
|
||||
expect(outcome).toMatchObject({ state: 'accepted' })
|
||||
expect(outcome).toMatchObject({ providerIdentity: { turnId: 'turn-late' } })
|
||||
expect(events.at(-1)).toMatchObject({ type: 'notification', method: 'turn/started' })
|
||||
})
|
||||
|
||||
it('settles unknown rather than failed when Codex never names the turn', async () => {
|
||||
vi.useFakeTimers()
|
||||
try {
|
||||
const codex = fakeCodex()
|
||||
const adapter = await acquired(codex)
|
||||
|
||||
const dispatching = adapter.dispatch({
|
||||
sessionId: 'session-1',
|
||||
clientMessageId: 'client-1',
|
||||
body: USER_MESSAGE,
|
||||
fence: 7
|
||||
})
|
||||
await vi.advanceTimersByTimeAsync(10_000)
|
||||
|
||||
expect(await dispatching).toEqual({
|
||||
state: 'unknown',
|
||||
reason: 'codex app-server started a turn it did not name in time'
|
||||
})
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('rejects only when Codex answered and declined', async () => {
|
||||
const codex = fakeCodex({
|
||||
'turn/start': () => {
|
||||
throw new CodexAppServerRequestError('turn/start', -32602, 'turn already running')
|
||||
}
|
||||
})
|
||||
const adapter = await acquired(codex)
|
||||
|
||||
expect(
|
||||
await adapter.dispatch({
|
||||
sessionId: 'session-1',
|
||||
clientMessageId: 'client-1',
|
||||
body: USER_MESSAGE,
|
||||
fence: 7
|
||||
})
|
||||
).toEqual({ state: 'rejected', reason: 'turn already running' })
|
||||
})
|
||||
|
||||
it('rethrows a dead child so the wire settles the submission unknown', async () => {
|
||||
const codex = fakeCodex({
|
||||
'turn/start': () => {
|
||||
throw new Error('codex app-server connection ended')
|
||||
}
|
||||
})
|
||||
const adapter = await acquired(codex)
|
||||
|
||||
await expect(
|
||||
adapter.dispatch({
|
||||
sessionId: 'session-1',
|
||||
clientMessageId: 'client-1',
|
||||
body: USER_MESSAGE,
|
||||
fence: 7
|
||||
})
|
||||
).rejects.toThrow('connection ended')
|
||||
})
|
||||
|
||||
it('applies an option change to the next turn only', async () => {
|
||||
const codex = fakeCodex({ 'turn/start': () => ({ turn: { id: 'turn-1' } }) })
|
||||
const adapter = await acquired(codex)
|
||||
|
||||
await adapter.setOption({ sessionId: 'session-1', key: 'model', value: 'gpt-5', fence: 7 })
|
||||
await adapter.setOption({ sessionId: 'session-1', key: 'effort', value: 'high', fence: 7 })
|
||||
await expect(
|
||||
adapter.setOption({ sessionId: 'session-1', key: 'sandboxEscape', value: 'yes', fence: 7 })
|
||||
).rejects.toThrow('no thread option named sandboxEscape')
|
||||
await adapter.dispatch({
|
||||
sessionId: 'session-1',
|
||||
clientMessageId: 'client-1',
|
||||
body: USER_MESSAGE,
|
||||
fence: 7
|
||||
})
|
||||
|
||||
expect(codex.connections[0].calls[1].params).toMatchObject({ model: 'gpt-5', effort: 'high' })
|
||||
expect(codex.connections[0].calls[1].params).not.toHaveProperty('sandboxEscape')
|
||||
})
|
||||
})
|
||||
|
||||
describe('CodexStructuredSessionAdapter.cancelTurn', () => {
|
||||
it('confirms an interrupt Codex acknowledged', async () => {
|
||||
const codex = fakeCodex()
|
||||
const adapter = await acquired(codex)
|
||||
|
||||
expect(
|
||||
await adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 })
|
||||
).toEqual({ cancelled: true })
|
||||
expect(codex.connections[0].calls[1]).toEqual({
|
||||
method: 'turn/interrupt',
|
||||
params: { threadId: THREAD_ID, turnId: 'turn-1' }
|
||||
})
|
||||
})
|
||||
|
||||
it('reports not-cancelled when Codex declines or lacks the method', async () => {
|
||||
const declined = fakeCodex({
|
||||
'turn/interrupt': () => {
|
||||
throw new CodexAppServerRequestError('turn/interrupt', -32602, 'no such turn')
|
||||
}
|
||||
})
|
||||
const absent = fakeCodex({
|
||||
'turn/interrupt': () => {
|
||||
throw new CodexAppServerUnsupportedError('no turn/interrupt')
|
||||
}
|
||||
})
|
||||
|
||||
expect(
|
||||
await (
|
||||
await acquired(declined)
|
||||
).cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 })
|
||||
).toEqual({ cancelled: false })
|
||||
expect(
|
||||
await (
|
||||
await acquired(absent)
|
||||
).cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 })
|
||||
).toEqual({ cancelled: false })
|
||||
})
|
||||
|
||||
it('rethrows an unsettled interrupt so the turn is not shown as cancelled', async () => {
|
||||
const codex = fakeCodex({
|
||||
'turn/interrupt': () => {
|
||||
throw new Error('codex app-server turn/interrupt exceeded 30000ms')
|
||||
}
|
||||
})
|
||||
const adapter = await acquired(codex)
|
||||
|
||||
await expect(
|
||||
adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 })
|
||||
).rejects.toThrow('exceeded 30000ms')
|
||||
})
|
||||
})
|
||||
|
||||
describe('CodexStructuredSessionAdapter prompts', () => {
|
||||
function askApproval(codex: ReturnType<typeof fakeCodex>): void {
|
||||
codex.connections[0].handlers.onServerRequest?.({
|
||||
id: 11,
|
||||
method: 'item/commandExecution/requestApproval',
|
||||
params: { itemId: 'codex-item-1', threadId: THREAD_ID, turnId: 'turn-1' }
|
||||
})
|
||||
}
|
||||
|
||||
it('surfaces an approval request and answers it exactly once', async () => {
|
||||
const codex = fakeCodex()
|
||||
const events: CodexStructuredSessionEvent[] = []
|
||||
const adapter = await acquired(codex, {}, events)
|
||||
|
||||
askApproval(codex)
|
||||
adapter.bindPromptItemId('session-1', 'codex:thread-abc:turn-1:3', 'codex-item-1')
|
||||
await adapter.answerPrompt({
|
||||
sessionId: 'session-1',
|
||||
itemId: 'codex:thread-abc:turn-1:3',
|
||||
kind: 'approval',
|
||||
optionId: 'accept',
|
||||
fence: 7
|
||||
})
|
||||
|
||||
expect(events.at(-1)).toMatchObject({ type: 'prompt', codexItemId: 'codex-item-1' })
|
||||
expect(codex.connections[0].replies).toEqual([{ id: 11, result: { decision: 'accept' } }])
|
||||
|
||||
await expect(
|
||||
adapter.answerPrompt({
|
||||
sessionId: 'session-1',
|
||||
itemId: 'codex:thread-abc:turn-1:3',
|
||||
kind: 'approval',
|
||||
optionId: 'decline',
|
||||
fence: 7
|
||||
})
|
||||
).rejects.toThrow('no longer waiting on')
|
||||
expect(codex.connections[0].replies).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('rejects an option id that is not a Codex decision', async () => {
|
||||
const codex = fakeCodex()
|
||||
const adapter = await acquired(codex)
|
||||
|
||||
askApproval(codex)
|
||||
|
||||
await expect(
|
||||
adapter.answerPrompt({
|
||||
sessionId: 'session-1',
|
||||
itemId: 'codex-item-1',
|
||||
kind: 'approval',
|
||||
optionId: 'yolo',
|
||||
fence: 7
|
||||
})
|
||||
).rejects.toThrow('is not a Codex approval decision')
|
||||
expect(codex.connections[0].replies).toEqual([])
|
||||
})
|
||||
|
||||
it('holds a multi-question request until every question is answered', async () => {
|
||||
const codex = fakeCodex()
|
||||
const adapter = await acquired(codex)
|
||||
codex.connections[0].handlers.onServerRequest?.({
|
||||
id: 12,
|
||||
method: 'item/tool/requestUserInput',
|
||||
params: {
|
||||
itemId: 'codex-item-2',
|
||||
threadId: THREAD_ID,
|
||||
turnId: 'turn-1',
|
||||
questions: [{ id: 'q1' }, { id: 'q2' }]
|
||||
}
|
||||
})
|
||||
|
||||
await adapter.answerPrompt({
|
||||
sessionId: 'session-1',
|
||||
itemId: 'codex-item-2',
|
||||
kind: 'question',
|
||||
optionId: encodeCodexQuestionOptionId('q1', 'yes'),
|
||||
fence: 7
|
||||
})
|
||||
expect(codex.connections[0].replies).toEqual([])
|
||||
|
||||
await adapter.answerPrompt({
|
||||
sessionId: 'session-1',
|
||||
itemId: 'codex-item-2',
|
||||
kind: 'question',
|
||||
optionId: encodeCodexQuestionOptionId('q2', 'no'),
|
||||
fence: 7
|
||||
})
|
||||
|
||||
expect(codex.connections[0].replies).toEqual([
|
||||
{ id: 12, result: { answers: { q1: { answers: ['yes'] }, q2: { answers: ['no'] } } } }
|
||||
])
|
||||
})
|
||||
|
||||
it('refuses a request it does not model instead of leaving the turn blocked', async () => {
|
||||
const codex = fakeCodex()
|
||||
const events: CodexStructuredSessionEvent[] = []
|
||||
await acquired(codex, {}, events)
|
||||
|
||||
codex.connections[0].handlers.onServerRequest?.({
|
||||
id: 13,
|
||||
method: 'mcpServer/elicitation/request',
|
||||
params: { itemId: 'codex-item-3', threadId: THREAD_ID }
|
||||
})
|
||||
|
||||
expect(codex.connections[0].replies).toEqual([
|
||||
{ id: 13, code: -32601, message: 'Orca does not handle mcpServer/elicitation/request' }
|
||||
])
|
||||
expect(events.some((event) => event.type === 'prompt')).toBe(false)
|
||||
})
|
||||
|
||||
it('surfaces an answer to a prompt Codex already forgot', async () => {
|
||||
const codex = fakeCodex()
|
||||
const adapter = await acquired(codex)
|
||||
|
||||
await expect(
|
||||
adapter.answerPrompt({
|
||||
sessionId: 'session-1',
|
||||
itemId: 'codex-item-gone',
|
||||
kind: 'approval',
|
||||
optionId: 'accept',
|
||||
fence: 7
|
||||
})
|
||||
).rejects.toThrow('no longer waiting on codex-item-gone')
|
||||
})
|
||||
})
|
||||
|
||||
describe('CodexStructuredSessionAdapter lifecycle', () => {
|
||||
it('keeps sessions isolated and closes each child once', async () => {
|
||||
const codex = fakeCodex()
|
||||
const adapter = adapterFor(codex)
|
||||
await adapter.acquire({ identity: identityFor('session-1'), fence: 1, spawnToken: 'spawn-a' })
|
||||
await adapter.acquire({ identity: identityFor('session-2'), fence: 1, spawnToken: 'spawn-b' })
|
||||
|
||||
codex.connections[0].handlers.onServerRequest?.({
|
||||
id: 21,
|
||||
method: 'item/fileChange/requestApproval',
|
||||
params: { itemId: 'codex-item-1', threadId: THREAD_ID, turnId: 'turn-1' }
|
||||
})
|
||||
await expect(
|
||||
adapter.answerPrompt({
|
||||
sessionId: 'session-2',
|
||||
itemId: 'codex-item-1',
|
||||
kind: 'approval',
|
||||
optionId: 'accept',
|
||||
fence: 1
|
||||
})
|
||||
).rejects.toThrow('no longer waiting on')
|
||||
|
||||
await adapter.closeAll()
|
||||
expect(codex.connections.map((connection) => connection.closeCount)).toEqual([1, 1])
|
||||
await expect(
|
||||
adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 1 })
|
||||
).rejects.toThrow('no live codex app-server for session session-1')
|
||||
})
|
||||
|
||||
it('drops a session whose child died and reports it once', async () => {
|
||||
const codex = fakeCodex()
|
||||
const events: CodexStructuredSessionEvent[] = []
|
||||
const adapter = await acquired(codex, {}, events)
|
||||
|
||||
codex.connections[0].handlers.onExit?.(new Error('codex app-server connection ended'))
|
||||
|
||||
expect(events.at(-1)).toEqual({
|
||||
type: 'ended',
|
||||
sessionId: 'session-1',
|
||||
reason: 'codex app-server connection ended'
|
||||
})
|
||||
await expect(
|
||||
adapter.dispatch({
|
||||
sessionId: 'session-1',
|
||||
clientMessageId: 'client-1',
|
||||
body: USER_MESSAGE,
|
||||
fence: 7
|
||||
})
|
||||
).rejects.toThrow('no live codex app-server')
|
||||
expect(await adapter.historyFilePath({ identity: identityFor('session-1') })).toBeNull()
|
||||
})
|
||||
|
||||
it('ignores Codex traffic that arrives after the session is gone', async () => {
|
||||
const codex = fakeCodex()
|
||||
const adapter = await acquired(codex)
|
||||
const connection = codex.connections[0]
|
||||
|
||||
await adapter.closeSession('session-1')
|
||||
connection.handlers.onNotification?.('item/agentMessage/delta', { delta: 'x' })
|
||||
connection.handlers.onServerRequest?.({
|
||||
id: 31,
|
||||
method: 'item/fileChange/requestApproval',
|
||||
params: { itemId: 'codex-item-9', threadId: THREAD_ID }
|
||||
})
|
||||
|
||||
expect(connection.replies).toEqual([])
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,333 @@
|
||||
import type {
|
||||
AgentJournalMessageItem,
|
||||
AgentSessionJournalIdentity
|
||||
} from '../../shared/agent-session-journal-types'
|
||||
import type {
|
||||
AgentSessionAcquisition,
|
||||
AgentSessionDispatchOutcome,
|
||||
StructuredAgentSessionAdapter
|
||||
} from '../native-chat/agent-session-wire/structured-agent-session-adapter'
|
||||
import {
|
||||
isCodexAppServerRequestError,
|
||||
openCodexAppServerConnection,
|
||||
type CodexAppServerConnection,
|
||||
type CodexAppServerServerRequest
|
||||
} from './codex-app-server-connection'
|
||||
import { isCodexAppServerUnsupportedError } from './codex-app-server-session'
|
||||
import {
|
||||
CODEX_SPAWN_TOKEN_ENV,
|
||||
codexProcessIdentity,
|
||||
codexProviderHandleLink
|
||||
} from './codex-structured-owner-identity'
|
||||
import {
|
||||
answerCodexPrompt,
|
||||
CodexPromptRegistry,
|
||||
receiveCodexPromptRequest
|
||||
} from './codex-structured-prompt-replies'
|
||||
import {
|
||||
readCodexThreadId,
|
||||
readCodexThreadPath,
|
||||
readCodexTurnId
|
||||
} from './codex-structured-thread-facts'
|
||||
import {
|
||||
CODEX_USER_MESSAGE_ORDINAL,
|
||||
isCodexTurnOptionKey,
|
||||
startCodexTurn
|
||||
} from './codex-structured-turn-start'
|
||||
|
||||
// The Codex half of the structured agent-session wire: one long-lived
|
||||
// `codex app-server` child per session, started or resumed under the lease the
|
||||
// host already reserved. Everything durable — journal, fence, idempotency —
|
||||
// belongs to the wire; this adapter only starts the process, names the turn the
|
||||
// provider accepted, and answers Codex's blocking prompt requests.
|
||||
|
||||
export type CodexStructuredLaunch = {
|
||||
command: string
|
||||
args: string[]
|
||||
/** Thread working directory, passed to Codex rather than to the child. */
|
||||
cwd: string
|
||||
/** Pinned account home. Null inherits whatever CODEX_HOME the host has. */
|
||||
codexHome: string | null
|
||||
/** Thread this session already proved. Null starts a new one. Taken from the
|
||||
* durable handle chain, never from the client-declared identity. */
|
||||
resumeThreadId: string | null
|
||||
}
|
||||
|
||||
export type CodexStructuredSessionEvent =
|
||||
| { type: 'notification'; sessionId: string; threadId: string; method: string; params: unknown }
|
||||
| {
|
||||
type: 'prompt'
|
||||
sessionId: string
|
||||
threadId: string
|
||||
method: string
|
||||
params: unknown
|
||||
codexItemId: string
|
||||
}
|
||||
/** The child died without the host asking it to. The lease is now stale. */
|
||||
| { type: 'ended'; sessionId: string; reason: string }
|
||||
|
||||
export type CodexStructuredSessionAdapterDeps = {
|
||||
/** Command, args, cwd, and pinned home for one session. The wire's adapter
|
||||
* contract carries none of these, so the runtime resolves them. */
|
||||
resolveLaunch: (input: {
|
||||
identity: AgentSessionJournalIdentity
|
||||
}) => Promise<CodexStructuredLaunch>
|
||||
/** Every Codex notification and prompt, in arrival order. The translation
|
||||
* module subscribes here; without it the session still runs, unjournaled. */
|
||||
onEvent?: (event: CodexStructuredSessionEvent) => void
|
||||
openConnection?: typeof openCodexAppServerConnection
|
||||
readProcessStartTime?: (pid: number) => Promise<number | null>
|
||||
mintLinkId?: () => string
|
||||
now?: () => number
|
||||
requestTimeoutMs?: number
|
||||
}
|
||||
|
||||
type CodexSession = {
|
||||
connection: CodexAppServerConnection
|
||||
threadId: string
|
||||
historyPath: string | null
|
||||
prompts: CodexPromptRegistry
|
||||
/** Applied to the next `turn/start`; Codex has no thread-settings write. */
|
||||
options: Map<string, string>
|
||||
turnIdWaiters: ((turnId: string) => void)[]
|
||||
}
|
||||
|
||||
export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdapter {
|
||||
private readonly sessions = new Map<string, CodexSession>()
|
||||
|
||||
constructor(private readonly deps: CodexStructuredSessionAdapterDeps) {}
|
||||
|
||||
private get requestTimeoutMs(): number | undefined {
|
||||
return this.deps.requestTimeoutMs
|
||||
}
|
||||
|
||||
async acquire(input: {
|
||||
identity: AgentSessionJournalIdentity
|
||||
fence: number
|
||||
spawnToken: string
|
||||
}): Promise<AgentSessionAcquisition> {
|
||||
// A re-acquire at a new fence must not leave the old child writing.
|
||||
await this.closeSession(input.identity.sessionId)
|
||||
const sessionId = input.identity.sessionId
|
||||
const launch = await this.deps.resolveLaunch({ identity: input.identity })
|
||||
const prompts = new CodexPromptRegistry()
|
||||
const open = this.deps.openConnection ?? openCodexAppServerConnection
|
||||
const connection = await open(
|
||||
{
|
||||
command: launch.command,
|
||||
args: launch.args,
|
||||
env: {
|
||||
[CODEX_SPAWN_TOKEN_ENV]: input.spawnToken,
|
||||
...(launch.codexHome ? { CODEX_HOME: launch.codexHome } : {})
|
||||
}
|
||||
},
|
||||
{
|
||||
onNotification: (method, params) => this.handleNotification(sessionId, method, params),
|
||||
onServerRequest: (request) => this.handleServerRequest(sessionId, request),
|
||||
onExit: (error) => {
|
||||
this.sessions.delete(sessionId)
|
||||
prompts.clear()
|
||||
this.deps.onEvent?.({ type: 'ended', sessionId, reason: error.message })
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
try {
|
||||
const started = await connection.request(
|
||||
launch.resumeThreadId ? 'thread/resume' : 'thread/start',
|
||||
launch.resumeThreadId
|
||||
? { threadId: launch.resumeThreadId, cwd: launch.cwd }
|
||||
: { cwd: launch.cwd },
|
||||
{ timeoutMs: this.requestTimeoutMs }
|
||||
)
|
||||
const threadId = readCodexThreadId(started)
|
||||
if (!threadId) {
|
||||
throw new Error('codex app-server did not name the thread it opened')
|
||||
}
|
||||
// A resume that lands on a different thread is a fork wearing a resume's
|
||||
// name; the handle chain would record a lie.
|
||||
if (launch.resumeThreadId && threadId !== launch.resumeThreadId) {
|
||||
throw new Error(`codex app-server resumed ${threadId} instead of ${launch.resumeThreadId}`)
|
||||
}
|
||||
this.sessions.set(sessionId, {
|
||||
connection,
|
||||
threadId,
|
||||
historyPath: readCodexThreadPath(started),
|
||||
prompts,
|
||||
options: new Map(),
|
||||
turnIdWaiters: []
|
||||
})
|
||||
return {
|
||||
process: await codexProcessIdentity(
|
||||
{ ...input, pid: connection.pid },
|
||||
this.deps.readProcessStartTime
|
||||
),
|
||||
link: codexProviderHandleLink({
|
||||
threadId,
|
||||
resumed: launch.resumeThreadId !== null,
|
||||
fence: input.fence,
|
||||
linkId: this.deps.mintLinkId?.(),
|
||||
observedAt: this.deps.now?.() ?? Date.now()
|
||||
})
|
||||
}
|
||||
} catch (error) {
|
||||
await connection.close()
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
private handleNotification(sessionId: string, method: string, params: unknown): void {
|
||||
const session = this.sessions.get(sessionId)
|
||||
if (!session) {
|
||||
return
|
||||
}
|
||||
if (method === 'turn/started') {
|
||||
const turnId = readCodexTurnId(params)
|
||||
const waiter = turnId ? session.turnIdWaiters.shift() : undefined
|
||||
waiter?.(turnId as string)
|
||||
}
|
||||
this.deps.onEvent?.({
|
||||
type: 'notification',
|
||||
sessionId,
|
||||
threadId: session.threadId,
|
||||
method,
|
||||
params
|
||||
})
|
||||
}
|
||||
|
||||
private handleServerRequest(sessionId: string, request: CodexAppServerServerRequest): void {
|
||||
const session = this.sessions.get(sessionId)
|
||||
if (!session) {
|
||||
return
|
||||
}
|
||||
const prompt = receiveCodexPromptRequest(session.prompts, session.connection, request)
|
||||
if (!prompt) {
|
||||
return
|
||||
}
|
||||
this.deps.onEvent?.({
|
||||
type: 'prompt',
|
||||
sessionId,
|
||||
threadId: session.threadId,
|
||||
method: request.method,
|
||||
params: request.params,
|
||||
codexItemId: prompt.codexItemId
|
||||
})
|
||||
}
|
||||
|
||||
/** Lets the translation module address a live prompt by the journal item it
|
||||
* became; until then the Codex item id addresses it directly. */
|
||||
bindPromptItemId(sessionId: string, journalItemId: string, codexItemId: string): void {
|
||||
this.sessions.get(sessionId)?.prompts.bindJournalItemId(journalItemId, codexItemId)
|
||||
}
|
||||
|
||||
async dispatch(input: {
|
||||
sessionId: string
|
||||
clientMessageId: string
|
||||
body: AgentJournalMessageItem
|
||||
fence: number
|
||||
}): Promise<AgentSessionDispatchOutcome> {
|
||||
const session = this.session(input.sessionId)
|
||||
let turnId: string | null
|
||||
try {
|
||||
turnId = await startCodexTurn(session, { ...input, timeoutMs: this.requestTimeoutMs })
|
||||
} catch (error) {
|
||||
// Codex answering and declining is a rejection; a timeout or a dead child
|
||||
// is not, and the wire must not tell the user their message failed.
|
||||
if (isCodexAppServerRequestError(error) || isCodexAppServerUnsupportedError(error)) {
|
||||
return { state: 'rejected', reason: (error as Error).message }
|
||||
}
|
||||
throw error
|
||||
}
|
||||
return turnId === null
|
||||
? {
|
||||
state: 'unknown',
|
||||
reason: 'codex app-server started a turn it did not name in time'
|
||||
}
|
||||
: {
|
||||
state: 'accepted',
|
||||
providerIdentity: {
|
||||
provider: 'codex',
|
||||
threadId: session.threadId,
|
||||
turnId,
|
||||
ordinal: CODEX_USER_MESSAGE_ORDINAL
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async cancelTurn(input: {
|
||||
sessionId: string
|
||||
turnId: string
|
||||
fence: number
|
||||
}): Promise<{ cancelled: boolean }> {
|
||||
const session = this.session(input.sessionId)
|
||||
try {
|
||||
await session.connection.request(
|
||||
'turn/interrupt',
|
||||
{ threadId: session.threadId, turnId: input.turnId },
|
||||
{ timeoutMs: this.requestTimeoutMs }
|
||||
)
|
||||
return { cancelled: true }
|
||||
} catch (error) {
|
||||
// Codex declining names a turn it no longer owns; anything else leaves the
|
||||
// cancel unconfirmed and must surface as such.
|
||||
if (isCodexAppServerRequestError(error) || isCodexAppServerUnsupportedError(error)) {
|
||||
return { cancelled: false }
|
||||
}
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
async answerPrompt(input: {
|
||||
sessionId: string
|
||||
itemId: string
|
||||
kind: 'approval' | 'question'
|
||||
optionId: string
|
||||
fence: number
|
||||
}): Promise<void> {
|
||||
const session = this.session(input.sessionId)
|
||||
answerCodexPrompt(session.prompts, session.connection, input.itemId, input.optionId)
|
||||
}
|
||||
|
||||
async setOption(input: {
|
||||
sessionId: string
|
||||
key: string
|
||||
value: string
|
||||
fence: number
|
||||
}): Promise<void> {
|
||||
const session = this.session(input.sessionId)
|
||||
if (!isCodexTurnOptionKey(input.key)) {
|
||||
throw new Error(`codex app-server has no thread option named ${input.key}`)
|
||||
}
|
||||
// Codex applies model, effort, and policy on `turn/start`, so an option set
|
||||
// between turns takes effect on the next one rather than immediately.
|
||||
session.options.set(input.key, input.value)
|
||||
}
|
||||
|
||||
async historyFilePath(input: { identity: AgentSessionJournalIdentity }): Promise<string | null> {
|
||||
return this.sessions.get(input.identity.sessionId)?.historyPath ?? null
|
||||
}
|
||||
|
||||
/** Reaps one session's child. The proven handle chain is already durable, so
|
||||
* a graceful close loses nothing. */
|
||||
async closeSession(sessionId: string): Promise<void> {
|
||||
const session = this.sessions.get(sessionId)
|
||||
if (!session) {
|
||||
return
|
||||
}
|
||||
this.sessions.delete(sessionId)
|
||||
session.prompts.clear()
|
||||
await session.connection.close()
|
||||
}
|
||||
|
||||
async closeAll(): Promise<void> {
|
||||
await Promise.all([...this.sessions.keys()].map((sessionId) => this.closeSession(sessionId)))
|
||||
}
|
||||
|
||||
private session(sessionId: string): CodexSession {
|
||||
const session = this.sessions.get(sessionId)
|
||||
if (!session) {
|
||||
throw new Error(`no live codex app-server for session ${sessionId}`)
|
||||
}
|
||||
return session
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,39 @@
|
||||
// The handful of facts Orca reads out of Codex app-server payloads. Codex has
|
||||
// moved these fields between the envelope and a nested `thread` / `turn` object
|
||||
// across releases, so each reader accepts both shapes rather than pinning one.
|
||||
|
||||
function record(value: unknown): Record<string, unknown> | null {
|
||||
return typeof value === 'object' && value !== null && !Array.isArray(value)
|
||||
? (value as Record<string, unknown>)
|
||||
: null
|
||||
}
|
||||
|
||||
function nonEmptyString(value: unknown): string | null {
|
||||
return typeof value === 'string' && value.length > 0 ? value : null
|
||||
}
|
||||
|
||||
/** `thread/start`, `thread/resume`, and `thread/started` all name the thread. */
|
||||
export function readCodexThreadId(payload: unknown): string | null {
|
||||
const root = record(payload)
|
||||
if (!root) {
|
||||
return null
|
||||
}
|
||||
return nonEmptyString(record(root.thread)?.id) ?? nonEmptyString(root.threadId)
|
||||
}
|
||||
|
||||
/** Rollout file for the thread, when Codex reports one. Journal recovery reads
|
||||
* it; a null just falls back to the existing session-file resolver. */
|
||||
export function readCodexThreadPath(payload: unknown): string | null {
|
||||
const root = record(payload)
|
||||
return root ? nonEmptyString(record(root.thread)?.path) : null
|
||||
}
|
||||
|
||||
/** `turn/start` responses carry `turn.id`; `turn/started` notifications carry
|
||||
* the same under `turn`, and older builds put `turnId` on the envelope. */
|
||||
export function readCodexTurnId(payload: unknown): string | null {
|
||||
const root = record(payload)
|
||||
if (!root) {
|
||||
return null
|
||||
}
|
||||
return nonEmptyString(record(root.turn)?.id) ?? nonEmptyString(root.turnId)
|
||||
}
|
||||
@@ -0,0 +1,91 @@
|
||||
import type { AgentJournalMessageItem } from '../../shared/agent-session-journal-types'
|
||||
import type { NativeChatBlock } from '../../shared/native-chat-types'
|
||||
import type { CodexAppServerConnection } from './codex-app-server-connection'
|
||||
import { readCodexTurnId } from './codex-structured-thread-facts'
|
||||
|
||||
// Starting a Codex turn and learning its id, which are not the same event:
|
||||
// `turn/start` returns the id on newer builds and acks before it exists on
|
||||
// older ones, where it arrives as a `turn/started` notification instead.
|
||||
|
||||
/** Codex records the user message first in a turn, so the submission Orca just
|
||||
* accepted is ordinal 0 of `(threadId, turnId)`. */
|
||||
export const CODEX_USER_MESSAGE_ORDINAL = 0
|
||||
|
||||
/** Past this the turn is real but unnameable, which the journal renders as
|
||||
* delivery unconfirmed rather than failure. */
|
||||
const TURN_ID_WAIT_MS = 10_000
|
||||
|
||||
/** Keys Codex accepts as per-turn overrides. An unlisted key would otherwise
|
||||
* become an arbitrary client-controlled `turn/start` parameter. */
|
||||
const CODEX_TURN_OPTION_KEYS = new Set([
|
||||
'model',
|
||||
'effort',
|
||||
'approvalPolicy',
|
||||
'approvalsReviewer',
|
||||
'personality',
|
||||
'serviceTier'
|
||||
])
|
||||
|
||||
export function isCodexTurnOptionKey(key: string): boolean {
|
||||
return CODEX_TURN_OPTION_KEYS.has(key)
|
||||
}
|
||||
|
||||
/** The session state one turn needs. `turnIdWaiters` is shared with the
|
||||
* notification handler, which resolves the head of the queue — correct because
|
||||
* Codex runs one turn per thread, so starts and `turn/started` share an order. */
|
||||
export type CodexTurnHost = {
|
||||
connection: Pick<CodexAppServerConnection, 'request'>
|
||||
threadId: string
|
||||
options: Map<string, string>
|
||||
turnIdWaiters: ((turnId: string) => void)[]
|
||||
}
|
||||
|
||||
function turnInputFor(body: AgentJournalMessageItem): Record<string, unknown>[] {
|
||||
const input: Record<string, unknown>[] = []
|
||||
for (const block of body.blocks as NativeChatBlock[]) {
|
||||
if (block.type === 'text' && block.text.length > 0) {
|
||||
input.push({ type: 'text', text: block.text })
|
||||
} else if (block.type === 'image-ref' && block.path) {
|
||||
input.push({ type: 'localImage', path: block.path })
|
||||
} else if (block.type === 'image-ref' && block.url) {
|
||||
input.push({ type: 'image', url: block.url })
|
||||
}
|
||||
}
|
||||
return input
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolves the turn id, or null when Codex owns a turn it never named. Throws
|
||||
* only for outcomes the wire must not read as acceptance.
|
||||
*/
|
||||
export async function startCodexTurn(
|
||||
host: CodexTurnHost,
|
||||
input: { clientMessageId: string; body: AgentJournalMessageItem; timeoutMs?: number }
|
||||
): Promise<string | null> {
|
||||
// Registered BEFORE the call: on builds that ack first, `turn/started` can
|
||||
// land while the response is still in flight.
|
||||
let notified: ((turnId: string) => void) | null = null
|
||||
const fromNotification = new Promise<string | null>((resolve) => {
|
||||
notified = resolve
|
||||
host.turnIdWaiters.push(resolve)
|
||||
setTimeout(() => resolve(null), TURN_ID_WAIT_MS).unref?.()
|
||||
})
|
||||
try {
|
||||
const started = await host.connection.request(
|
||||
'turn/start',
|
||||
{
|
||||
threadId: host.threadId,
|
||||
clientUserMessageId: input.clientMessageId,
|
||||
input: turnInputFor(input.body),
|
||||
...Object.fromEntries(host.options)
|
||||
},
|
||||
{ timeoutMs: input.timeoutMs }
|
||||
)
|
||||
return readCodexTurnId(started) ?? (await fromNotification)
|
||||
} finally {
|
||||
const index = notified ? host.turnIdWaiters.indexOf(notified) : -1
|
||||
if (index !== -1) {
|
||||
host.turnIdWaiters.splice(index, 1)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user