Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
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: 1 addition & 0 deletions src/main/acp/acp-structured-acquire.ts
Original file line number Diff line number Diff line change
Expand Up @@ -226,6 +226,7 @@ export async function acquireAcpStructuredSession(input: {
await identity.onSpawned(pid)
const process = await identity.read(pid)
const child: AcpStructuredChild = {
events,
sessionId,
fence: acquire.fence,
acquisitionGeneration: generation,
Expand Down
1 change: 1 addition & 0 deletions src/main/acp/acp-structured-child.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ import type { AcpStructuredSession } from './acp-structured-session'

export type AcpStructuredChild = Pick<
AcpStructuredSession,
| 'events'
| 'sessionId'
| 'fence'
| 'acquisitionGeneration'
Expand Down
23 changes: 23 additions & 0 deletions src/main/acp/acp-structured-host-lifecycle.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -289,3 +289,26 @@ describe('closing or stopping a Grok chat while it starts', () => {
expect(await rig.adapter.closeSession(SESSION)).toBe(true)
})
})

describe('a Grok send whose acceptance could not be written', () => {
it('is Working while Grok holds it, and stops once Grok lets it go', async () => {
const { rig, host } = await openAttachedHostRig()
const journal = host.collaboratorsForTests().sessions.get(SESSION)!.journal
vi.spyOn(journal, 'acceptLate').mockRejectedValue(new Error('disk full'))
const sent = await send(host, 'hello')
const prompt = await rig.frame('session/prompt')
// Grok names the prompt only with its first frame: until then it holds the send.
await waitFor(() => expect(host.currentWork(SESSION)?.working()).toBe(true))

rig.child().agent.notify('session/update', replyChunk(promptIdOf(prompt), 'hi'))
rig.child().agent.reply(prompt, { stopReason: 'end_turn' })
await rig.settle()

await waitFor(() => expect(host.currentWork(SESSION)?.working()).toBe(false))
// The acceptance never landed: the row is still the handover's alone.
expect(
(await host.journalSnapshot(SESSION)).submissions.find((row) => row.clientMessageId === sent)
).toMatchObject({ dispatchState: 'pending' })
await host.close(SESSION, 'user-close')
})
})
7 changes: 6 additions & 1 deletion src/main/acp/acp-structured-host.test-support.ts
Original file line number Diff line number Diff line change
Expand Up @@ -104,7 +104,12 @@ export async function openHostRig(
onEvent: (event) => void hosted.host?.handleAdapterEvent(event),
onChildWorkEvidence: (sessionId, evidence) =>
hosted.host?.publishChildWorkEvidence(sessionId, evidence),
onDispatchSettledLate: (settlement) => void hosted.host?.settleLateDispatch(settlement),
onDispatchSettledLate: (settlement) =>
void hosted.host
?.settleLateDispatch(settlement)
.catch((error: unknown) =>
state.log.logger.warn('late settlement failed', { scope: 'late-settlement', error })
),
now: () => HOST_TEST_NOW,
readProcessStartTime: async () => 1_700_000_000_000 + ++generation,
mintGeneration: () => `generation-${generation}`,
Expand Down
4 changes: 4 additions & 0 deletions src/main/acp/acp-structured-session-adapter.ts
Original file line number Diff line number Diff line change
Expand Up @@ -295,6 +295,10 @@ export class AcpStructuredSessionAdapter implements StructuredAgentSessionAdapte
}
}

// Undefined until the session opens: the host holds what it handed a starting child.
heldSends = (sessionId: string): ReadonlySet<string> | undefined =>
this.children.sessions.get(sessionId)?.turns?.heldSends

stopBackgroundTasks: NonNullable<StructuredAgentSessionAdapter['stopBackgroundTasks']> = (
input
) => stopAcpChildren(this.live(input.sessionId), input.fence, input.taskIds, input.acknowledge)
Expand Down
4 changes: 4 additions & 0 deletions src/main/acp/acp-structured-session.ts
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,11 @@ import type { AcpStructuredLane } from './acp-structured-lane'
import type { AcpStructuredOptions } from './acp-structured-options'
import type { AcpStructuredPrompts } from './acp-structured-prompts'
import type { AcpStructuredTurns } from './acp-structured-turns'
import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink'
import type { StructuredAgentSessionProviderOpenResult } from '../native-chat/agent-session-wire/structured-agent-session-provider-open-contract'

export type AcpStructuredSession = {
events?: StructuredAgentSessionEventSink
sessionId: string
fence: number
acquisitionGeneration: string
Expand Down Expand Up @@ -101,6 +103,7 @@ export function closeAcpSessionJournal(
if (session.journalClosed !== null) {
return
}
session.events?.observeExit?.(session.turns.heldSends)
session.journalClosed = reason
session.prompts.clear()
session.turns.end(reason, cause)
Expand All @@ -121,6 +124,7 @@ export function endAcpStructuredSession(
return
}
session.ended = true
session.events?.observeExit?.(undefined, observedAt)
session.exitObservedAt = observedAt
const stderr = session.connection.stderrTail()
// Read at the proven exit: a crash usually ends the agent's stdout before its exit is observed,
Expand Down
5 changes: 5 additions & 0 deletions src/main/acp/acp-structured-turns.ts
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,11 @@ export class AcpStructuredTurns {
)
}

/** The sends this agent holds: each from its dispatch until its settlement. */
get heldSends(): ReadonlySet<string> {
return this.unsettled
}

/** Steers wait behind the running prompt's cancel. */
get holdsSteers(): boolean {
return this.steers.length > 0
Expand Down
33 changes: 2 additions & 31 deletions src/main/claude/claude-command-lifecycle.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,6 @@ async function replayCapture(
) {
const capture = loadCapture(name)
const settlements: Settlement[] = []
const idles: string[] = []
const turnStates = new Map<string, string>()
// The capture supplies every frame, startup proof included.
const claude = fakeClaude({ initProof: 'none', replayUuid: null })
Expand All @@ -130,8 +129,7 @@ async function replayCapture(
readProcessStartTime: async () => 1,
now: () => 1_700_000_200_000,
persistHandle: async () => {},
onDispatchSettledLate: (settlement) => settlements.push(settlement),
onSessionIdle: ({ sessionId }) => idles.push(sessionId)
onDispatchSettledLate: (settlement) => settlements.push(settlement)
})
await adapter.acquire({
identity: identityFor(),
Expand Down Expand Up @@ -248,7 +246,7 @@ async function replayCapture(
settlements
.filter((settlement) => settlement.clientMessageId === clientMessageId)
.map(({ sessionId: _sessionId, clientMessageId: _id, ...outcome }) => outcome)
return { settlementsFor, idles, turnStates, liveUuid, connection, adapter }
return { settlementsFor, turnStates, liveUuid, connection, adapter }
}

const WITHDRAWN = {
Expand Down Expand Up @@ -574,7 +572,6 @@ describe('a send the CLI holds', () => {
isLifecycleFrame(frame, 'cancelled')
})

expect(replay.idles).toEqual(['session-1'])
expect(replay.adapter.holdsDispatch('session-1')).toBe(true)
expect(replay.settlementsFor('client-A')).toEqual([])
})
Expand Down Expand Up @@ -621,29 +618,3 @@ describe('a send the CLI holds', () => {
}
)
})

describe('the CLI reporting its session idle', () => {
it.each(['interrupt-lost', 'cancel-async', 'batch-lead', 'auth-failed'])(
'is reported once per idle frame (%s), never for running',
async (name) => {
const replay = await replayCapture(name)

expect(replay.idles).toEqual(['session-1'])
}
)

it('is not reported from a child the session no longer holds', async () => {
const replay = await replayCapture('auth-failed')
await replay.adapter.closeSession('session-1')

replay.connection.handlers.onMessage?.({
type: 'system',
subtype: 'session_state_changed',
state: 'idle',
uuid: 'late-idle',
session_id: PROVIDER_SESSION_ID
})

expect(replay.idles).toEqual(['session-1'])
})
})
21 changes: 19 additions & 2 deletions src/main/claude/claude-command-lifecycle.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
// queued. None ran. After `started`, 2.1.280 sends only `cancelled`, when an interrupt or a hard
// failure ends the turn (measured), so the send may be in the conversation: any end is doubt.
// An unstarted send's `cancelled` lands before the Stop's answer, so it settles if that is lost.
// `completed` with no `started` names a send its transcript already holds: doubt, never a verdict.

import {
agentSessionFailureFact,
Expand Down Expand Up @@ -63,6 +64,13 @@ export function observeClaudeCommandLifecycle(
agentSessionFailureFact(UNSTARTED_VERDICT[state]),
onSettledLate
)
} else if (state === 'completed' && waiter.commandLifecycle !== 'started' && !waiter.retired) {
releaseClaudeDispatchWaitersInDoubt(
session,
[waiter],
DISPATCH_DOUBT_PROVIDER_ENDED_UNANSWERED,
onSettledLate
)
}
}

Expand All @@ -80,9 +88,18 @@ export function claudeStartedUnechoedSend(session: ClaudeSession): ClaudeStarted
: null
}

/** The sends the CLI holds: every waiter its echo, a lifecycle frame, idle or exit has not ended,
* and each one a frame ended until the host is told, so no reader sees it held by neither. */
export function claudeHeldSends(session: ClaudeSession): ReadonlySet<string> {
return new Set([
...session.dispatchWaiters.flatMap((waiter) => waiter.clientMessageId ?? []),
...(session.unreportedSettlements ?? [])
])
}

/** A send the CLI took, until its echo, a terminal state or exit: the sweep must not rest its child. */
export function claudeHoldsDispatch(session: ClaudeSession): boolean {
return session.dispatchWaiters.some((waiter) => waiter.commandLifecycle !== undefined)
export function claudeHoldsDispatch(session: ClaudeSession | undefined): boolean {
return session?.dispatchWaiters.some((waiter) => waiter.commandLifecycle !== undefined) ?? false
}

/**
Expand Down
19 changes: 11 additions & 8 deletions src/main/claude/claude-structured-dispatch.ts
Original file line number Diff line number Diff line change
Expand Up @@ -59,14 +59,8 @@ export function rejectClaudeDispatchWaiters(
const retiredWaiters = session.retiredDispatchWaiters.filter((waiter) =>
named.has(waiter.sentUuid)
)
for (const waiter of activeWaiters) {
forgetWaiter(session, waiter)
waiter.resolve(null)
}
for (const waiter of retiredWaiters) {
forgetRetiredWaiter(session, waiter)
}
for (const waiter of [...activeWaiters, ...retiredWaiters]) {
// Each released as it is reported: one dropped before its report reads as held by nobody.
const report = (waiter: ClaudeDispatchWaiter): void => {
if (waiter.clientMessageId) {
onSettledLate?.({
clientMessageId: waiter.clientMessageId,
Expand All @@ -75,6 +69,15 @@ export function rejectClaudeDispatchWaiters(
})
}
}
for (const waiter of activeWaiters) {
forgetWaiter(session, waiter)
waiter.resolve(null)
report(waiter)
}
for (const waiter of retiredWaiters) {
forgetRetiredWaiter(session, waiter)
report(waiter)
}
}

/** Releases sends the CLI took and let go without an echo or a verdict: each may have run, so each
Expand Down
69 changes: 43 additions & 26 deletions src/main/claude/claude-structured-session-acquisition.ts
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,8 @@ import {
type ClaudeSession,
type ClaudeSessionExit,
type ClaudeStructuredSessionAdapterDeps,
type ClaudeAcquireCallbacks
type ClaudeAcquireCallbacks,
type ClaudeLateDispatchOutcome
} from './claude-structured-session-state'
import { resolveClaudeAcquisitionError } from './claude-structured-session-close'
import { withObservedProviderExit } from '../native-chat/agent-session-wire/structured-agent-session-failure-text'
Expand Down Expand Up @@ -107,49 +108,65 @@ export async function acquireClaudeSession({
if (message.type === 'result' && sessions.get(sessionId) === liveSession) {
persistClaudeTurnResumePoint(sessionId, liveSession, deps)
}
// The CLI's idle releases its doubted sends; a late echo still accepts one it goes on to run.
// The CLI's idle wakes whatever waits for its turn to end.
if (claudeSessionStateEndsTurn(message) && sessions.get(sessionId) === liveSession) {
settleClaudeTurnEndWaiters(liveSession)
deps.onSessionIdle?.({ sessionId })
}
}
// Settled after the turn this echo opens is emitted: a send read as answered before its turn
// lands reads as nothing running, and Stop and Working blink off in between.
const settlements: (() => void)[] = []
const settlements: ClaudeLateDispatchOutcome[] = []
const turnOrigin = liveSession
? resolveClaudeReplayTurn(liveSession, message, (settlement) =>
settlements.push(() => deps.onDispatchSettledLate?.({ sessionId, ...settlement }))
)
? resolveClaudeReplayTurn(liveSession, message, (settlement) => settlements.push(settlement))
: null
const startsTurn = turnOrigin !== null
// Turn endpoints are stamped on the host clock, never the frame's own timestamp.
const observedAt =
startsTurn || message.type === 'result' ? { observedAt: deps.now?.() ?? Date.now() } : {}
const requestedAt = turnOrigin?.requestedAt
const startedSend = !startsTurn && liveSession ? claudeStartedUnechoedSend(liveSession) : null
callbacks.deliver(attempt, sessionId, () => {
if (modelChanged && liveSession && init?.model) {
// Still held until reported: the emit can publish at once, and the host holds a settled send
// only from its report, so dropping it any earlier reads it as held by nobody (doubt).
const session = liveSession
const reported = session?.unreportedSettlements
if (session && settlements.length > 0) {
session.unreportedSettlements = new Set([
...(reported ?? []),
...settlements.map((settlement) => settlement.clientMessageId)
])
}
try {
callbacks.deliver(attempt, sessionId, () => {
if (modelChanged && liveSession && init?.model) {
callbacks.emit(liveSession, input.events, {
type: 'options-reported',
sessionId,
fence: liveSession.fence,
acquisitionGeneration: liveSession.acquisitionGeneration,
reportedOptions: { model: init.model }
})
}
callbacks.emit(liveSession, input.events, {
type: 'options-reported',
type: 'message',
sessionId,
fence: liveSession.fence,
acquisitionGeneration: liveSession.acquisitionGeneration,
reportedOptions: { model: init.model }
message,
...(startsTurn ? { startsTurn: true } : {}),
...(requestedAt === null || requestedAt === undefined ? {} : { requestedAt }),
...(turnOrigin?.clientMessageId ? { clientMessageId: turnOrigin.clientMessageId } : {}),
...(startedSend ? { startedSend } : {}),
...observedAt
})
}
callbacks.emit(liveSession, input.events, {
type: 'message',
sessionId,
message,
...(startsTurn ? { startsTurn: true } : {}),
...(requestedAt === null || requestedAt === undefined ? {} : { requestedAt }),
...(turnOrigin?.clientMessageId ? { clientMessageId: turnOrigin.clientMessageId } : {}),
...(startedSend ? { startedSend } : {}),
...observedAt
})
})
for (const settle of settlements) {
settle()
} finally {
try {
for (const settlement of settlements) {
deps.onDispatchSettledLate?.({ sessionId, ...settlement })
}
} finally {
if (session) {
session.unreportedSettlements = reported
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.
}
}
}
const emit = (event: Parameters<typeof callbacks.emit>[2]): void =>
Expand Down
10 changes: 6 additions & 4 deletions src/main/claude/claude-structured-session-adapter.ts
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@ import type {
StructuredAgentSessionAdapter
} from '../native-chat/agent-session-wire/structured-agent-session-adapter'
import { stopCurrentClaudeBackgroundTasks } from './claude-structured-control-actions'
import { claudeHoldsDispatch } from './claude-command-lifecycle'
import { claudeHeldSends, claudeHoldsDispatch } from './claude-command-lifecycle'
import { releaseClaudeAcquisition } from './claude-structured-acquisition-release'
import { acquireClaudeSession } from './claude-structured-session-acquisition'
import { supportsClaudeStructuredLocation } from './claude-structured-location-support'
Expand Down Expand Up @@ -217,9 +217,11 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda
readCommands: NonNullable<StructuredAgentSessionAdapter['readCommands']> = (sessionId) =>
this.sessions.get(sessionId)?.commands.commands
readonly atRestCommands: ClaudeStructuredSessionAdapterDeps['atRestCommands']
holdsDispatch = (sessionId: string): boolean => {
const session = this.sessions.get(sessionId)
return session ? claudeHoldsDispatch(session) : false
holdsDispatch = (sessionId: string): boolean => claudeHoldsDispatch(this.sessions.get(sessionId))
heldSends = (sessionId: string): ReadonlySet<string> | undefined => {
const session = this.sessions.get(sessionId) ?? this.exits.get(sessionId)?.session
const held = session && claudeHeldSends(session)
return held ?? (this.settledExitErrors.has(sessionId) ? new Set() : undefined)
}
holdsLiveProviderProcess = (sessionId: string, acquisitionGeneration: string): boolean => {
const session = this.sessions.get(sessionId)
Expand Down
Loading
Loading