Skip to content
Draft
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
16 changes: 16 additions & 0 deletions src/main/claude/claude-request-end-wait.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import {
awaitClaudeRequestEnd,
CLAUDE_STOP_GRACE_MS,
claudeLiveWork,
claudeStoppedRequestEndWait,
settleClaudeTurnEndWaiters
} from './claude-request-end-wait'
Expand Down Expand Up @@ -90,3 +91,18 @@ describe("a Stop's wait for Claude to wind down what it has in flight", () => {
expect(await settledWithin(wait('session-2', Date.now()), 0)).toBe(true)
})
})

// What a host stop reads, in its own step and with no wait, to judge whether it ends work.
describe("Claude's live view of what its child has in flight", () => {
it('is its open turn', () => {
expect(claudeLiveWork(session('turn-1'))).toEqual({ turnId: 'turn-1' })
})

it('is a send with no turn yet before its echo', () => {
expect(claudeLiveWork(claudeWith(null, 1).claude)).toEqual({ turnId: null })
})

it('is nothing at rest', () => {
expect(claudeLiveWork(session(null))).toBeUndefined()
})
})
10 changes: 6 additions & 4 deletions src/main/claude/claude-request-end-wait.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3,16 +3,18 @@
// ends with its own result, and Claude writes both to its transcript. Keyed on Claude's own request,
// not on a journal turn: a Stop pressed before the echo has no turn row yet.

import type { StructuredAgentSessionLiveWork } from '../native-chat/agent-session-wire/structured-agent-session-adapter'
import type { ClaudeSession } from './claude-structured-session-state'

/** One budget from the Stop's interrupt: for Claude to answer it and wind its request down. */
export const CLAUDE_STOP_GRACE_MS = 3_000

const waiters = new WeakMap<ClaudeSession, Set<() => void>>()

/** Claude has a turn open, or a send it has not answered yet. */
function claudeRequestInFlight(session: ClaudeSession): boolean {
return (session.translator?.currentTurnId ?? null) !== null || session.dispatchWaiters.length > 0
/** What Claude has in flight (`liveWork`): its open turn, or a send it has not answered yet. */
export function claudeLiveWork(session: ClaudeSession): StructuredAgentSessionLiveWork | undefined {
const turnId = session.translator?.currentTurnId ?? null
return turnId !== null || session.dispatchWaiters.length > 0 ? { turnId } : undefined
}

function nextSettle(session: ClaudeSession): { settled: Promise<void>; forget: () => void } {
Expand Down Expand Up @@ -43,7 +45,7 @@ export async function awaitClaudeRequestEnd(session: ClaudeSession, ms: number):
})
try {
// A settle that leaves something in flight, such as a subagent's result, waits on.
while (!elapsed && claudeRequestInFlight(session)) {
while (!elapsed && claudeLiveWork(session) !== undefined) {
const next = nextSettle(session)
try {
await Promise.race([next.settled, budget])
Expand Down
6 changes: 5 additions & 1 deletion src/main/claude/claude-structured-session-adapter.ts
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ import {
type ClaudeStructuredSessionEvent
} from './claude-structured-session-state'
import { closeAllClaudeSessions, closeClaudeSession } from './claude-structured-session-close'
import { claudeStoppedRequestEndWait } from './claude-request-end-wait'
import { claudeLiveWork, claudeStoppedRequestEndWait } from './claude-request-end-wait'
import {
drainClaudeObservedExits,
observeClaudeSessionExit,
Expand Down Expand Up @@ -239,6 +239,10 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda
const session = this.sessions.get(sessionId)
return session ? claudeHoldsDispatch(session) : false
}
liveWork: NonNullable<StructuredAgentSessionAdapter['liveWork']> = (sessionId) => {
const session = this.sessions.get(sessionId)
return session ? claudeLiveWork(session) : undefined
}
answerPrompt: StructuredAgentSessionAdapter['answerPrompt'] = (request) =>
answerClaudeStructuredPrompt({ request, sessions: this.sessions })
setOption: StructuredAgentSessionAdapter['setOption'] = (input) =>
Expand Down
9 changes: 9 additions & 0 deletions src/main/codex/codex-command-turn-claim.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -349,6 +349,15 @@ describe('a Codex turn a conversation command claims', () => {
expect(translator.commandProviderTurnId('ordinary')).toBe('ordinary')
})

it("maps the provider turn carrying the command to the command's journal turn", () => {
const { translator, emit } = harness()
translator.beginCommand(COMMAND)
emit(notification('turn/started', { turn: { id: PROVIDER_TURN } }))

expect(translator.commandJournalTurnId(PROVIDER_TURN)).toBe(COMMAND.turnId)
expect(translator.commandJournalTurnId('ordinary')).toBe('ordinary')
})

it('claims the same provider turn when the refused start is retried', () => {
const { writes, translator, emit } = harness(1)
translator.beginCommand(COMMAND)
Expand Down
6 changes: 6 additions & 0 deletions src/main/codex/codex-journal-command-turn.ts
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,12 @@ export class CodexJournalCommandTurn {
return turnId
}

/** The journal turn provider turn `providerTurnId` writes to: the command's it carries, or its
* own. */
journalTurnId(providerTurnId: string): string {
return this.carried.get(providerTurnId)?.command.turnId ?? providerTurnId
}

/** Codex reported the compaction, in the turn carrying the command. */
compacted(providerTurnId: string): void {
const carried = this.carried.get(providerTurnId)
Expand Down
3 changes: 3 additions & 0 deletions src/main/codex/codex-structured-journal-contracts.ts
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,9 @@ export type CodexJournalTranslator = {
forgetCommand: (turnId: string) => void
/** The provider turn a Stop naming `turnId` interrupts; undefined while a command has none. */
commandProviderTurnId: (turnId: string) => string | undefined
/** The journal turn provider turn `providerTurnId` writes to: the command's it carries, or its
* own. */
commandJournalTurnId: (providerTurnId: string) => string
cancelPrompt: (journalItemId: string) => CodexJournalTranslationAdmission
restoreThread: (
threadId: string,
Expand Down
1 change: 1 addition & 0 deletions src/main/codex/codex-structured-journal-translation.ts
Original file line number Diff line number Diff line change
Expand Up @@ -287,6 +287,7 @@ export function createCodexJournalTranslator(
beginCommand: (command) => commands.begin(command),
forgetCommand: (turnId) => commands.forget(turnId),
commandProviderTurnId: (turnId) => commands.providerTurnId(turnId),
commandJournalTurnId: (providerTurnId) => commands.journalTurnId(providerTurnId),
cancelPrompt: (journalItemId) => prompts.cancel(journalItemId),
resolvePrompt: (journalItemId) => prompts.resolve(journalItemId),
flush: () => {
Expand Down
6 changes: 6 additions & 0 deletions src/main/codex/codex-structured-session-adapter.ts
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ import type {
} from '../native-chat/agent-session-wire/structured-agent-session-adapter'
import type { CodexJournalTranslationAdmission } from './codex-structured-journal-translation'
import { dispatchCodexTurn, isCodexTurnOptionKey } from './codex-structured-turn-start'
import { codexLiveWork } from './codex-structured-turn-open-wait'
import { agentSessionFailureFact, providerDiagnosticOf } from '../../shared/agent-session-failure'
import { supportsCodexStructuredLocation } from './codex-structured-location-support'
import { CodexStructuredSessionTeardown } from './codex-structured-session-teardown'
Expand Down Expand Up @@ -175,6 +176,11 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap
) =>
this.sessions.has(sessionId) ? { supportsTaskStop: false, supportsStopAll: false } : undefined

liveWork: NonNullable<StructuredAgentSessionAdapter['liveWork']> = (sessionId) => {
const session = this.sessions.get(sessionId)
return session ? codexLiveWork(session) : undefined
}

bindPromptItemId = (
sessionId: string,
journalItemId: string,
Expand Down
55 changes: 54 additions & 1 deletion src/main/codex/codex-structured-turn-open-wait.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ import {
codexTurnLifecycleRig,
settledWithin
} from './codex-structured-dispatch-test-support'
import { CODEX_TURN_OPEN_WAIT_MS } from './codex-structured-turn-open-wait'
import { CODEX_TURN_OPEN_WAIT_MS, codexLiveWork } from './codex-structured-turn-open-wait'

type Rig = Awaited<ReturnType<typeof codexTurnLifecycleRig>>

Expand Down Expand Up @@ -145,3 +145,56 @@ describe("a no-turn Stop in the window between Codex's answer and its turn openi
expect(rig.interrupts()).toEqual([])
})
})

// What a host stop reads, in its own step and with no wait, to judge whether it ends work.
describe("Codex's live view of what its child has in flight", () => {
const liveWork = (rig: Rig) => rig.adapter.liveWork('session-1')

it('is nothing at rest', async () => {
const rig = await codexTurnLifecycleRig()

expect(liveWork(rig)).toBeUndefined()
})

it('is a send with no turn yet from the answer until its turn opens, then that turn', async () => {
const rig = await codexTurnLifecycleRig()
await answeredColdSend(rig)

expect(liveWork(rig)).toEqual({ turnId: null })
rig.turns.start()
expect(liveWork(rig)).toEqual({ turnId: 'turn-1' })
})

// Codex carries a command out as a turn of its own, whose rows join the command's turn.
it("names a turn carrying a command by the command's journal turn", () => {
const work = codexLiveWork({
threadId: CODEX_TEST_THREAD_ID,
activeTurnIds: new Set(['provider-turn']),
dispatchEchoes: { answeredUnopenedTurn: () => null },
translator: {
commandJournalTurnId: (turnId) => (turnId === 'provider-turn' ? 'command-turn' : turnId)
}
})

expect(work).toEqual({ turnId: 'command-turn' })
})

it('is nothing once the turn ends', async () => {
const rig = await codexTurnLifecycleRig()
await answeredColdSend(rig)
rig.turns.start()

rig.turns.end('completed')

expect(liveWork(rig)).toBeUndefined()
})

it('is nothing once the answered turn ends without opening', async () => {
const rig = await codexTurnLifecycleRig()
await answeredColdSend(rig)

rig.turns.end('interrupted')

expect(liveWork(rig)).toBeUndefined()
})
})
23 changes: 23 additions & 0 deletions src/main/codex/codex-structured-turn-open-wait.ts
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
// A Stop's or a send's wait for the turn Codex answered a send into to open, or provably not
// to: it ended, the thread stopped running, or the child is gone. Held in memory only.

import type { StructuredAgentSessionLiveWork } from '../native-chat/agent-session-wire/structured-agent-session-adapter'
import type { CodexDispatchEchoes } from './codex-structured-dispatch-echo'
import type { CodexJournalTranslator } from './codex-structured-journal-contracts'
import {
codexThreadStoppedRunning,
readCodexThreadId,
Expand Down Expand Up @@ -61,6 +63,27 @@ export function createCodexTurnOpenWaits(): CodexTurnOpenWaits {
}
}

/** What Codex has in flight (`liveWork`), with no wait: the turn it reports running, by the journal
* id of the command it carries when it carries one; or none yet for a send it has not answered, or
* answered into a turn that has not opened (`codexRunningOrOpeningTurn` waits that one out). */
export function codexLiveWork(session: {
threadId: string
activeTurnIds?: ReadonlySet<string>
dispatchPending?: boolean
dispatchEchoes: Pick<CodexDispatchEchoes, 'answeredUnopenedTurn'>
translator: Pick<CodexJournalTranslator, 'commandJournalTurnId'> | null
}): StructuredAgentSessionLiveWork | undefined {
const open = session.activeTurnIds ?? new Set<string>()
const running = [...open].at(-1)
if (running) {
return { turnId: session.translator?.commandJournalTurnId(running) ?? running }
}
return session.dispatchPending === true ||
session.dispatchEchoes.answeredUnopenedTurn(session.threadId, open) !== null
? { turnId: null }
: undefined
}

/**
* The turn a Stop or a send names: the latest Codex reported started and not ended, or else the
* one Codex answered a send into, once it opens. Null when none is running and that one ends, the
Expand Down
11 changes: 1 addition & 10 deletions src/main/native-chat/agent-session-journal/journal-stop-marks.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,7 @@
// the turn-end rule reads (`journal-stop-turn-end.ts`), and whether a person's still decides.

import type { JournalReducerState } from './journal-reducer'
import {
latestAcceptedSendUnopened,
personStopDecidesTurn,
type JournalLatestStop
} from './journal-stop-turn-end'
import { personStopDecidesTurn, type JournalLatestStop } from './journal-stop-turn-end'

export class JournalStopMarks {
constructor(private readonly deps: { state: () => JournalReducerState }) {}
Expand All @@ -15,11 +11,6 @@ export class JournalStopMarks {
return this.deps.state().queuePauseMarks.latestStop
}

/** `latestAcceptedSendUnopened`: the latest accepted send's turn row may still be on its way. */
latestAcceptedSendUnopened(): boolean {
return latestAcceptedSendUnopened(this.deps.state())
}

/** `personStopDecidesTurn`: a person's Stop decides how turn `turnId` ends. */
personStopDecides(turnId: string | null, endedAt?: number, openedBy?: string): boolean {
return personStopDecidesTurn(this.deps.state(), turnId, endedAt, openedBy)
Expand Down
28 changes: 16 additions & 12 deletions src/main/native-chat/agent-session-journal/journal-stop-turn-end.ts
Original file line number Diff line number Diff line change
Expand Up @@ -54,10 +54,10 @@ function openingSubmission(
: undefined
}

/** Whether the latest send the agent accepted to open a turn has opened none the journal holds: its
* turn's row may still be on its way. A send delivered into a running turn (a steer, a fold)
/** The latest send the agent accepted to open a turn, while the journal holds no turn it opened:
* that turn's row may still be on its way. A send delivered into a running turn (a steer, a fold)
* opens none, and its item carries that turn's scope (`placeHandedOverMessage`). */
export function latestAcceptedSendUnopened(state: TurnEndState): boolean {
function latestAcceptedSendUnopened(state: TurnEndState): AgentJournalSubmission | undefined {
let latest: { submission: AgentJournalSubmission; sequence: number } | undefined
for (const submission of state.submissions.values()) {
const item = state.items.get(agentJournalSubmissionKey(submission.clientMessageId))
Expand All @@ -71,16 +71,16 @@ export function latestAcceptedSendUnopened(state: TurnEndState): boolean {
}
}
if (!latest) {
return false
return undefined
}
for (const item of state.items.values()) {
if (
openingSubmission(state, readAgentJournalTurn(item.body)?.userItemId) === latest.submission
) {
return false
return undefined
}
}
return true
return latest.submission
}

/** A send a Stop that named no turn stopped: one already handed to the agent at the Stop's
Expand Down Expand Up @@ -136,14 +136,18 @@ function stopEndsTurnAsCancellation(
)
}

/** With no turn running, the work in flight is the person's Stop's: every send still unanswered is
* one it stopped. */
function unansweredSendsAreStopTargets(state: TurnEndState, stop: JournalLatestStop): boolean {
const unanswered = [...state.submissions.values()].filter((submission) =>
/** With no turn row running, the work in flight is the person's Stop's: every send still
* unanswered, and one accepted whose turn's row has yet to land, is one it stopped. */
function sendsInFlightAreStopTargets(state: TurnEndState, stop: JournalLatestStop): boolean {
const inFlight = [...state.submissions.values()].filter((submission) =>
isUnansweredStructuredAgentSessionDispatch(submission)
)
const unopened = latestAcceptedSendUnopened(state)
if (unopened) {
inFlight.push(unopened)
}
return (
unanswered.length > 0 && unanswered.every((submission) => isStopTarget(state, stop, submission))
inFlight.length > 0 && inFlight.every((submission) => isStopTarget(state, stop, submission))
)
}

Expand All @@ -165,7 +169,7 @@ export function personStopDecidesTurn(
return false
}
if (turnId === null) {
return stop.event.turnId === undefined && unansweredSendsAreStopTargets(state, stop)
return stop.event.turnId === undefined && sendsInFlightAreStopTargets(state, stop)
}
const turn = [...state.items.values()]
.map((item) => readAgentJournalTurn(item.body))
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -189,6 +189,17 @@ describe('StructuredAgentSessionAdapterRouter.stopEndsSession', () => {
expect(router.stopEndsSession('session-1')).toBe(true)
})

it("reports the live owner's work in flight, and none with no live owner", async () => {
const claude = adapterOf(vi.fn(async () => true))
claude.liveWork = () => ({ turnId: 'turn-1' })
const codex = adapterOf(vi.fn(async () => false))
const router = new StructuredAgentSessionAdapterRouter({ claude, codex }, async () => {})

expect(router.liveWork('session-1')).toBeUndefined()
await router.acquire({ identity: claudeIdentity('session-1'), fence: 1, spawnToken: 'spawn-1' })
expect(router.liveWork('session-1')).toEqual({ turnId: 'turn-1' })
})

it("waits on the session's live owner to wind the Stop down, and on nothing with none", async () => {
const claude = adapterOf(vi.fn(async () => true))
claude.awaitStoppedRequestEnd = vi.fn(async () => undefined)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -116,6 +116,9 @@ export class StructuredAgentSessionAdapterRouter implements StructuredAgentSessi
holdsDispatch = (sessionId: string): boolean =>
this.liveOwnerOrNull(sessionId)?.holdsDispatch?.(sessionId) ?? false

liveWork: NonNullable<StructuredAgentSessionAdapter['liveWork']> = (sessionId) =>
this.liveOwnerOrNull(sessionId)?.liveWork?.(sessionId)

stopEndsSession = (sessionId: string): boolean =>
this.liveOwnerOrNull(sessionId)?.stopEndsSession?.(sessionId) ?? false

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -240,6 +240,8 @@ export type StructuredAgentSessionSetOptionInput = {
fence: number
}

export type StructuredAgentSessionLiveWork = { turnId: string | null }

export type StructuredAgentSessionAdapter = StructuredAgentSessionAdapterStop & {
/** Provider-aware capability check for hosts that route more than one adapter. */
supportsCreate?(location: AgentSessionExecutionLocation, agent: string): boolean
Expand Down Expand Up @@ -335,6 +337,10 @@ export type StructuredAgentSessionAdapter = StructuredAgentSessionAdapterStop &
/** The provider reported taking a send it has neither answered nor ended, as a queued follow-up
* or a silent retry does. Derived from the live child; false with none. */
holdsDispatch?(sessionId: string): boolean
/** What the live child has in flight, read off the provider's own frames as they arrive, ahead
* of the journal rows they become: the turn it has open, by its journal id, or null for a send
* it has not answered with no turn open yet. Undefined with nothing in flight. */
liveWork?(sessionId: string): StructuredAgentSessionLiveWork | undefined
/** The `/` surface the running provider reports for itself. Undefined when the
* provider never reports one, which is what keeps the client on its catalog. */
readCommands?(sessionId: string): AgentSessionSlashCommand[] | undefined
Expand Down
Loading
Loading