Skip to content
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,8 @@ export type StructuredAgentSessionHostSession = {
* observed exit. */
child: StructuredAgentSessionProviderChild | null
lastEndedChild?: StructuredAgentSessionEndedChild
/** Owned by one live restart action, so another caller cannot replace its progress. */
restartResume?: NonNullable<AgentSessionStatusSummary['restartResume']> & { operationId: string }
}

export type StructuredAgentSessionHostDeps = {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -152,7 +152,8 @@ export class StructuredAgentSessionHost {
})
this.restartResume = createStructuredAgentSessionRestartResume(deps, this.sessions, {
...structuredAgentSessionRestartResumeSurfaces(this, this.now),
readChildWork: this.clientDelivery.readChildWork
readChildWork: this.clientDelivery.readChildWork,
publishStatus: this.clientDelivery.publishStatusAndSettlement
})
this.lifetime = createStructuredAgentSessionConversationLifetime({
context: () => this.lifetimeContext(),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@ it('neither continues nor reserves an offer the caller cannot show, named or not
expect(await host.restartResume.continueAfterRestart(named, 'modal', cannotShowCodex)).toEqual({
resumed: [],
continued: [],
...(named ? { skipped: [SESSION] } : {}),
sessions: [],
failed: []
})
Expand Down
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// "Resume all" starts agents a few at a time. A continuation is a send, so a chat counts as started
// Resumes start agents a few at a time host-wide. A continuation is a send, so a chat counts as started
// once its agent took the message or its start failed — not when the message was accepted, which
// is instant and would let every agent start at once.

Expand All @@ -9,14 +9,52 @@ import {
} from './structured-agent-session-restart-continuation'
import {
resumeStructuredAgentSessionsFromRestart,
STRUCTURED_AGENT_SESSION_RESUME_CONCURRENCY,
StructuredAgentSessionResumeAdmission
} from './structured-agent-session-restart-resume-runner'
import { marker } from './structured-agent-session-restart-resume-test-harness'
import { createStructuredAgentSessionLogger } from './structured-agent-session-logger'

type Handover = { dispatchState: string; reason?: string | null } | undefined

const LIMIT = 3

it('publishes queued chats separately from the chat holding the start slot', async () => {
const first = Promise.withResolvers<void>()
const phases = new Map<string, string>()
const action = resumeStructuredAgentSessionsFromRestart(
{
admission: new StructuredAgentSessionResumeAdmission(1),
onPhase: (sessionId, phase) => phases.set(sessionId, phase),
consumeMarker: async () => true,
resume: async (sessionId) => {
if (sessionId === 'a') {
await first.promise
}
}
},
['a', 'b'].map((sessionId) => ({
sessionId,
workspaceId: 'folder',
executionHostId: 'local',
workspaceKind: 'folder',
agent: 'codex',
work: { kind: 'turn', id: 'turn' },
trigger: 'quit',
recordedAt: 1,
latestPrompt: ''
})),
'modal'
)
try {
await vi.waitFor(() => expect(phases.get('a')).toBe('starting'))
expect(phases.get('b')).toBe('queued')
} finally {
first.resolve()
await action
}
expect(phases.get('b')).toBe('starting')
})

it('holds each slot from accept until handover or rejection, and frees it when the wait ends (P2-27)', async () => {
const handovers = new Map<string, PromiseWithResolvers<Handover>>()
let waiting = 0
Expand All @@ -42,7 +80,7 @@ it('holds each slot from accept until handover or rejection, and frees it when t
const sessions = Array.from({ length: 8 }, (_, index) => `session-${index}`)
const outcomes = resumeStructuredAgentSessionsFromRestart(
{
admission: new StructuredAgentSessionResumeAdmission(),
admission: new StructuredAgentSessionResumeAdmission(LIMIT),
consumeMarker: async () => true,
resume: async (sessionId) => {
const started = await startStructuredAgentSessionContinuation(
Expand All @@ -65,10 +103,10 @@ it('holds each slot from accept until handover or rejection, and frees it when t
handovers.get(sessionId)!.resolve(handover)
}

await vi.waitFor(() => expect(waiting).toBe(STRUCTURED_AGENT_SESSION_RESUME_CONCURRENCY))
await vi.waitFor(() => expect(waiting).toBe(LIMIT))
// Accepted but not yet handed over: a slow start keeps its slot.
await new Promise((resolve) => setTimeout(resolve, 20))
expect(handovers.size).toBe(STRUCTURED_AGENT_SESSION_RESUME_CONCURRENCY)
expect(handovers.size).toBe(LIMIT)
await settle('session-0', { dispatchState: 'pending' })
await settle('session-1', { dispatchState: 'rejected', reason: 'Codex could not start.' })
// The session closed, or too many waited: the wait proves nothing, and the slot is freed.
Expand All @@ -78,7 +116,7 @@ it('holds each slot from accept until handover or rejection, and frees it when t
}

const results = await outcomes
expect(mostWaiting).toBe(STRUCTURED_AGENT_SESSION_RESUME_CONCURRENCY)
expect(mostWaiting).toBe(LIMIT)
expect(results.find((result) => result.sessionId === 'session-1')).toMatchObject({
outcome: 'refused',
reason: 'Codex could not start.'
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ export type StructuredAgentSessionRestartOfferWithdrawal = ReturnType<

export type StructuredAgentSessionRestartOfferSession = Pick<
StructuredAgentSessionHostSession,
'journal' | 'child' | 'lastEndedChild'
'journal' | 'child' | 'lastEndedChild' | 'restartResume'
>

export function createStructuredAgentSessionRestartOfferWithdrawal(deps: {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -326,6 +326,7 @@ it('fails closed on corrupt recovery storage while an ordinary send still works'
expect(await host.restartResume.continueAfterRestart([SESSION], 'modal')).toEqual({
resumed: [],
continued: [],
skipped: [SESSION],
sessions: [],
failed: []
})
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,8 @@ export type StructuredAgentSessionRestartResume = {
) => Promise<{
resumed: StructuredAgentSessionResumeOutcome[]
continued: StructuredAgentSessionContinuationOutcome[]
/** Requested chats excluded from this action. */
skipped?: string[]
sessions?: StructuredAgentSessionResumeCandidate[]
failed?: StructuredAgentSessionResumeFailure[]
}>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
// action removes them — or files what went wrong. Starting the chat's agent again withdraws them.

import { randomUUID } from 'node:crypto'
import { createRestartResumeProgress } from './structured-agent-session-restart-resume-progress'
import {
AgentSessionRefusalError,
agentSessionRefusalFromReference
Expand Down Expand Up @@ -139,13 +140,14 @@ export function createStructuredAgentSessionRestartResume(
stillResumable: (marker) => deriveAtSend([marker], 'may-be-held').candidates.length === 1
}

/** One explicit action: reserve the offers, then continue each through `continueOne`, a few at
* a time. The runner counts a chat as done once `continueOne` returns. */
/** One explicit action: reserve the offers, then continue each through `continueOne`, as many at
* once as the admission's host-wide start limit allows. A chat is done once `continueOne` returns. */
const run = async (
sessionIds: readonly string[] | undefined,
owner: string,
audience: StructuredAgentSessionRestartAudience | undefined,
continueOne: (marker: AgentSessionResumeMarker, continuationId: string) => Promise<void>
continueOne: (marker: AgentSessionResumeMarker, continuationId: string) => Promise<void>,
progress: ReturnType<typeof createRestartResumeProgress>
) => {
// An explicit action supersedes teardown witnesses captured by this host. The durable mutation
// lane below also drains a publication already in flight before completion.
Expand All @@ -159,6 +161,7 @@ export function createStructuredAgentSessionRestartResume(
const eligible = restartRowsFor(derived.candidates, audience).filter((candidate) =>
requested.has(candidate.sessionId)
)
progress.skipExcluded(sessionIds ?? [], eligible)
if (eligible.length === 0) {
return null
}
Expand All @@ -178,10 +181,16 @@ export function createStructuredAgentSessionRestartResume(
)) ?? []
const markersBySession = new Map(reserved.map((marker) => [marker.sessionId, marker]))
const candidates = derive(reserved, 'may-be-held').candidates
progress.skipExcluded(
eligible.map((candidate) => candidate.sessionId),
candidates
)
try {
const outcomes = await resumeStructuredAgentSessionsFromRestart(
{
admission,
onPhase: progress.set,
onRefused: (sessionId) => progress.set(sessionId, 'refused'),
consumeMarker: async (sessionId) => {
const marker = markersBySession.get(sessionId)
return (
Expand Down Expand Up @@ -219,55 +228,75 @@ export function createStructuredAgentSessionRestartResume(
owner,
audience
) => {
const continued: StructuredAgentSessionContinuationOutcome[] = []
const verdicts: Promise<void>[] = []
// A chat holds its slot until its agent took the continuation or its start failed, so a batch
// never starts more agents at once than the runner allows; the provider's answer comes after.
const action = await run(sessionIds, owner, audience, async (marker, continuationId) => {
const started = await startStructuredAgentSessionContinuation(
continuationDeps(continuationHost, () => continuationHost.stillResumable(marker)),
marker.sessionId,
marker,
continuationId
const progress = createRestartResumeProgress(sessions, surfaces.publishStatus)
try {
const continued: StructuredAgentSessionContinuationOutcome[] = []
const verdicts: Promise<void>[] = []
// A chat holds its slot until its agent took the continuation or its start failed, so a batch
// never starts more agents at once than the runner allows; the provider's answer comes after.
const action = await run(
sessionIds,
owner,
audience,
async (marker, continuationId) => {
const started = await startStructuredAgentSessionContinuation(
continuationDeps(continuationHost, () => continuationHost.stillResumable(marker)),
marker.sessionId,
marker,
continuationId
)
if ('done' in started) {
continued.push(started.done)
progress.verdict(started.done)
const { outcome, reason, refusal } = started.done
if (outcome === 'refused') {
// Thrown as a refusal so the filed failure keeps its details beside the code.
throw refusal
? new AgentSessionRefusalError(
agentSessionRefusalFromReference(refusal, refusal.code)
)
: new Error(reason ?? 'agent_session_continuation_refused')
}
return
}
verdicts.push(
started.verdict().then((outcome) => {
continued.push(outcome)
progress.verdict(outcome)
})
)
},
progress
)
if ('done' in started) {
continued.push(started.done)
const { outcome, reason, refusal } = started.done
if (outcome === 'refused') {
// Thrown as a refusal so the filed failure keeps its details beside the code.
throw refusal
? new AgentSessionRefusalError(agentSessionRefusalFromReference(refusal, refusal.code))
: new Error(reason ?? 'agent_session_continuation_refused')
}
return
await Promise.all(verdicts)
const resumed = action?.outcomes ?? []
if (action) {
await failures.settle(action.operationId, resumed, {
candidates: action.candidates,
markers: action.markers,
failureAfterResume: (sessionId) => {
const outcome = continued.find((entry) => entry.sessionId === sessionId)
return outcome ? continuationFailureOutcome(outcome.outcome) : null
},
failureReason: (sessionId) => {
const outcome = continued.find((entry) => entry.sessionId === sessionId)
return outcome?.reason ?? outcome?.outcome ?? 'agent_session_continuation_unknown'
}
})
}
verdicts.push(started.verdict().then((outcome) => void continued.push(outcome)))
})
await Promise.all(verdicts)
const resumed = action?.outcomes ?? []
if (action) {
await failures.settle(action.operationId, resumed, {
candidates: action.candidates,
markers: action.markers,
failureAfterResume: (sessionId) => {
const outcome = continued.find((entry) => entry.sessionId === sessionId)
return outcome ? continuationFailureOutcome(outcome.outcome) : null
},
failureReason: (sessionId) => {
const outcome = continued.find((entry) => entry.sessionId === sessionId)
return outcome?.reason ?? outcome?.outcome ?? 'agent_session_continuation_unknown'
}
})
}
continued.push(...unstartedRestartRefusals(resumed, continued))
return {
resumed,
continued,
...(await remainingRestartRows(
() => list(audience),
() => listFailures(audience),
deps.logger
))
continued.push(...unstartedRestartRefusals(resumed, continued))
return {
resumed,
continued,
...(progress.skipped.length > 0 ? { skipped: progress.skipped } : {}),
...(await remainingRestartRows(
() => list(audience),
() => listFailures(audience),
deps.logger
))
}
} finally {
progress.clear()
}
}

Expand Down
Loading
Loading