diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index a8025afe9..9cdb165ee 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -11,6 +11,7 @@ - Fixed long-running thinking timers to display hours and days instead of unbounded minutes. - Fixed overlapping daemon snapshot catch-ups closing healthy workers and preventing new sessions from starting. - Changed daemon and RPC session state to report literal queued actions separately from active scheduler work. +- Changed session input scheduling to use one action lifecycle and store, fixing active work appearing queued, headless completion exiting early, `/compact` consuming itself as a successor, and incompatible clients not failing cleanly at protocol 7/schema 8. ## [0.4.0] - 2026-08-01 diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index 410222d02..6ce031d38 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -183,7 +183,7 @@ import { isSessionSlashCommandMessage, } from "./messages.js"; import type { ModelRegistry } from "./model-registry.js"; -import { throwIfPromptAdmissionCancelled, waitForPromptAdmission } from "./prompt-admission.js"; +import { throwIfPromptAdmissionCancelled } from "./prompt-admission.js"; import { expandPromptTemplate, type PromptTemplate } from "./prompt-templates.js"; import { type AutoRefineReason, @@ -240,7 +240,6 @@ import { type SessionActionSnapshot, type SessionCommandPayload, type SessionTurnPayload, - shouldYieldLowerRun, transitionSessionAction, type WakePolicy, } from "./session-action-store.js"; @@ -721,14 +720,6 @@ function primaryDeliveryRecord(action: QueuedSessionAction): DeliveryRecord { return record; } -function actionPrefixMessages(action: QueuedSessionAction): CustomMessage[] { - return action.payload.kind === "turn" - ? action.payload.records - .filter((record): record is DeliveryRecord & { message: CustomMessage } => record.role === "prefix") - .map((record) => record.message) - : []; -} - function normalizeMessageContent(content: string | (TextContent | ImageContent)[]): { text: string; images?: ImageContent[]; @@ -1052,16 +1043,20 @@ export class AgentSession { /** Session-owned actions. Items are never fed into Agent.steer/followUp. */ private readonly _actionStore = new ActionStore(); private _sessionInputPump: Promise = Promise.resolve(); + // Coalesces wakes so overlapping submissions cannot start competing pumps. private _sessionInputPumpRequested = false; + // Invalidates preparation when a branch pause starts and finishes before its next await resumes. private _sessionInputPumpEpoch = 0; private _sessionInputArrivalEpoch = 0; + // Persists abort/restart suspension after the initiating call returns. private _sessionInputPumpSuspended = false; + // Branch mutation pause leases can overlap and must all release before dispatch resumes. private readonly _queuedWorkPauses = new Set(); - private readonly _queuedWorkAdmissionWaiters = new Set<() => void>(); - private _turnAdmissionTail: Promise = Promise.resolve(); - private _turnAdmissionOwner: symbol | undefined; - private readonly _turnAdmissionContext = new AsyncLocalStorage(); - private _pumpingSessionInput = false; + private _sessionActionCommitTail: Promise = Promise.resolve(); + private _sessionActionCommitOwner: symbol | undefined; + private readonly _sessionActionCommitContext = new AsyncLocalStorage(); + private readonly _sessionActionCommitDisposeAbortController = new AbortController(); + // Checkpoint and handoff waiters share lifecycle-edge notifications to avoid polling. private readonly _sessionInputCheckpointWaiters = new Set<() => void>(); /** Messages queued to be included with the next user prompt as context ("asides"). */ private _pendingNextTurnMessages: CustomMessage[] = []; @@ -1578,6 +1573,7 @@ export class AgentSession { ); const previousAnchor = preparing.at(-1); const actions = this._actionStore.remove(predicate, candidates); + const restorableMessages: CustomMessage[] = []; const removed = new Set(actions); if (previousAnchor && removed.has(previousAnchor)) { for (const action of preparing) { @@ -1608,7 +1604,7 @@ export class AgentSession { !record.durable, ) .map((record) => cloneCustomMessage(record.message)); - this._pendingNextTurnMessages.unshift(...restorable); + restorableMessages.push(...restorable); if (dispatched) { payload.captureRunMessages = new Set(payload.records.map((record) => record.message)); this.agent.state.messages = this.agent.state.messages.filter( @@ -1620,6 +1616,7 @@ export class AgentSession { this._actionStore.releaseTerminal(action); } } + this._pendingNextTurnMessages.unshift(...restorableMessages); if (actions.length > 0) this._notifySessionInputCheckpointChange(); return actions; } @@ -1910,42 +1907,16 @@ export class AgentSession { } } - private async _runOrQueueGoalContext( - kind: "continuation" | "budget_limit" | "objective_updated", - images?: ImageContent[], - ): Promise { - if (!this._goalState.objective) { - return; - } + private _runOrQueueGoalContext(kind: "continuation" | "objective_updated", images?: ImageContent[]): void { + if (!this._goalState.objective) return; this._ensureGoalRuntimeActive(); const message = createGoalContextMessage(this._goalState, kind, images); - if (this._pumpingSessionInput) { - const normalized = normalizeMessageContent(message.content); - if (kind === "budget_limit") { - await this._queuePreparedPrompt("steer", normalized.text, normalized.images, { - message, - resumeIfIdle: true, - }); - } else { - const action = this._createPreparedTurnAction("followUp", normalized.text, normalized.images, { - message, - resumeIfIdle: true, - }); - this._admitSessionInput(action, { front: true, wake: false }); - } - return; - } - if (this.isStreaming) { - if (kind === "budget_limit") this.agent.steer(message); - else this.agent.followUp(message); - return; - } - const normalized = normalizeMessageContent(message.content); - await this._promptInjectedMessage(normalized.text, message, { + const action = this._createPreparedTurnAction("followUp", normalized.text, normalized.images, { + message, resumeIfIdle: true, - executionPolicy: this._turnExecutionPolicy("goalContext"), }); + this._admitSessionInput(action, { front: true, wake: false }); } private async _handleGoalSlashCommand(text: string, images: ImageContent[] | undefined): Promise { @@ -2026,15 +1997,16 @@ export class AgentSession { } private get _steeringStopPending(): boolean { - const activity = this._runtimeActivity(true); - if (shouldYieldLowerRun(this._actionStore, activity)) return true; - return this._actionStore - .activeActions("next_turn_boundary") - .some( - (action) => - action.payload.kind === "turn" && - (action.lifecycle.state === "selected" || action.lifecycle.state === "preparing"), - ); + return ( + this._actionStore.queuedActions("next_turn_boundary").length > 0 || + this._actionStore + .activeActions("next_turn_boundary") + .some( + (action) => + action.payload.kind === "turn" && + (action.lifecycle.state === "selected" || action.lifecycle.state === "preparing"), + ) + ); } private _shouldStopBeforeTurn(): boolean { @@ -3631,7 +3603,7 @@ export class AgentSession { return; } this._disposing = true; - this._wakeQueuedWorkAdmissionWaiters(); + this._sessionActionCommitDisposeAbortController.abort(); await this._disposeAsyncOnce(); })(); return this._disposeAsyncPromise; @@ -3799,7 +3771,7 @@ export class AgentSession { return; } this._disposed = true; - this._wakeQueuedWorkAdmissionWaiters(); + this._sessionActionCommitDisposeAbortController.abort(); try { // Invalidate scheduled timers and abort any in-flight review so a late // resolution cannot write harness state or re-subscribe handlers. @@ -3834,7 +3806,6 @@ export class AgentSession { if (outcome.completion) this._settleAgentMessage(agentMessageId, "completion", completionError); } this._cancelSessionActions(() => true, deliveryError); - this._notifySessionInputCheckpointChange(); this.agent.clearAllQueues(); this._extensionRunner.invalidate( "This extension ctx is stale after session replacement or reload. Do not use a captured pi or command ctx after ctx.newSession(), ctx.fork(), ctx.switchSession(), or ctx.reload(). For newSession, fork, and switchSession, move post-replacement work into withSession and use the ctx passed to withSession. For reload, do not use the old ctx after await ctx.reload().", @@ -4236,9 +4207,6 @@ export class AgentSession { } private _canStartSessionActionImmediately(): boolean { - const hasBlockingOwnedWork = this._actionStore - .unfinishedActions() - .some((action) => action.lifecycle.state !== "queued" || action.wake !== "external_resume"); return ( !this.isStreaming && !this.isCompacting && @@ -4246,7 +4214,6 @@ export class AgentSession { !this.isBashRunning && !this._sessionInputPumpSuspended && this._queuedWorkPauses.size === 0 && - !hasBlockingOwnedWork && !this._disposed && !this._disposing ); @@ -4329,9 +4296,10 @@ export class AgentSession { options?: InternalPromptOptions & { executionPolicy?: TurnExecutionPolicy }, ): Promise { if (!this.isStreaming && options?.resumeIfIdle) this._sessionInputPumpSuspended = false; - const admission = this.isStreaming - ? undefined - : await this._acquireSessionActionAdmission({ signal: options?.signal }); + const admissionFence = await this._acquireDirectTurnAdmissionFence(options?.signal).catch((error: unknown) => { + throwIfPromptAdmissionCancelled(options?.signal); + throw error; + }); const reportPreflight = oncePreflight(options?.preflightResult); try { throwIfPromptAdmissionCancelled(options?.signal); @@ -4364,6 +4332,7 @@ export class AgentSession { queueVisible: visibleQueued, }); const result = this._admitSessionInput(action, { immediatelyEligible: !visibleQueued }); + admissionFence.release(); if (!result.accepted || !result.ticket) { if (prefixMessages) this._pendingNextTurnMessages.unshift(...prefixMessages); reportPreflight(false, false); @@ -4377,7 +4346,6 @@ export class AgentSession { () => reportPreflight(false), ); } - admission?.release(); if (options?.returnAfterAccepted) { if (result.disposition === "starts_when_admitted") await result.ticket.delivered; return; @@ -4388,15 +4356,21 @@ export class AgentSession { reportPreflight(false); throw error; } finally { - admission?.release(); + admissionFence.release(); } } private async _prompt(text: string, options?: InternalPromptOptions): Promise { - if (!this.isStreaming) this._sessionInputPumpSuspended = false; - const admission = this.isStreaming + if (!this.isStreaming) { + this._sessionInputPumpSuspended = false; + this._assertSessionActionAdmissionAvailable(); + } + const commitFence = this.isStreaming ? undefined - : await this._acquireSessionActionAdmission({ signal: options?.signal }); + : await this._acquireDirectTurnAdmissionFence(options?.signal).catch((error: unknown) => { + throwIfPromptAdmissionCancelled(options?.signal); + throw error; + }); const reportPreflight = oncePreflight(options?.preflightResult); const run = async () => { try { @@ -4414,6 +4388,7 @@ export class AgentSession { }); const normalized = normalizationResult instanceof Promise ? await normalizationResult : normalizationResult; if (normalized.kind === "extensionCommand") { + commitFence?.release(); reportPreflight(true); void normalized.completion.then( () => this._settleAgentMessage(options?.agentMessageId, "completion"), @@ -4424,6 +4399,7 @@ export class AgentSession { return; } if (normalized.kind === "handled") { + commitFence?.release(); reportPreflight(true); this._settleAgentMessage(options?.agentMessageId, "completion"); return; @@ -4447,14 +4423,14 @@ export class AgentSession { const result = this._admitSessionInput(action, { immediatelyEligible: !wasBusy && this._canStartSessionActionImmediately(), }); + commitFence?.release(); reportPreflight(result.accepted, result.disposition === "queued"); if (!result.accepted || !result.ticket) return; - admission?.release(); if (options?.returnAfterAccepted) { if (result.disposition === "starts_when_admitted") await result.ticket.delivered; return; } - if (wasRuntimeBusy) return; + if (result.disposition === "queued") return; await this.waitForSessionInputIdle(); return; } @@ -4509,7 +4485,10 @@ export class AgentSession { if (action.suppressAutonomousContinuation) { this._markAutonomousContinuationSuppressed(primaryDeliveryRecord(action).message); } - const result = this._admitSessionInput(action, { immediatelyEligible: !visibleQueued }); + const result = this._admitSessionInput(action, { + immediatelyEligible: !visibleQueued && this._canStartSessionActionImmediately(), + }); + commitFence?.release(); if (!result.accepted || !result.ticket) { if (prefixMessages) this._pendingNextTurnMessages.unshift(...prefixMessages); reportPreflight(false, false); @@ -4523,20 +4502,22 @@ export class AgentSession { () => reportPreflight(false), ); } - const rollbackObserver = - acceptedAgentMessage && options?.queueIfBusy === true && result.disposition === "starts_when_admitted" - ? this._observeSessionActionRollback(action) + const deferralObserver = + acceptedAgentMessage && + options?.queueIfBusy === true && + !options.streamingBehavior && + result.disposition === "starts_when_admitted" + ? this._observeSessionActionDeferral(action) : undefined; - admission?.release(); - if (acceptedAgentMessage && !queueForStreaming && !queueForBusy) { + if (acceptedAgentMessage && !queueForStreaming && !queueForBusy && !options?.streamingBehavior) { try { - const outcome = rollbackObserver + const outcome = deferralObserver ? await Promise.race([ result.ticket.delivered.then(() => "delivered" as const), - rollbackObserver.rolledBack.then(() => "rolled_back" as const), + deferralObserver.deferred.then(() => "deferred" as const), ]) : await result.ticket.delivered.then(() => "delivered" as const); - if (outcome === "rolled_back" && !options?.streamingBehavior) { + if (outcome === "deferred" && !options?.streamingBehavior) { const error = new Error( "Agent became busy before prompt delivery. Specify streamingBehavior ('steer' or 'followUp') to queue the message.", ); @@ -4547,11 +4528,13 @@ export class AgentSession { } return; } finally { - rollbackObserver?.stop(); + deferralObserver?.stop(); } } if (options?.returnAfterAccepted) { - if (result.disposition === "starts_when_admitted") await result.ticket.delivered; + if (result.disposition === "starts_when_admitted" || (acceptedAgentMessage && !visibleQueued)) { + await result.ticket.delivered; + } return; } if (visibleQueued) return; @@ -4561,10 +4544,10 @@ export class AgentSession { reportPreflight(false); throw error; } finally { - admission?.release(); + commitFence?.release(); } }; - return admission ? this._turnAdmissionContext.run(admission.owner, run) : run(); + return commitFence ? this._sessionActionCommitContext.run(commitFence.owner, run) : run(); } private _executeExtensionCommand(text: string): Promise | undefined { @@ -4871,7 +4854,7 @@ export class AgentSession { } private _turnExecutionPolicy( - kind: "queued" | "directPrompt" | "injected" | "customTrigger" | "goalContext", + kind: "queued" | "directPrompt" | "injected" | "customTrigger", options: { returnAfterAccepted?: boolean; skipPrePromptWork?: boolean } = {}, ): TurnExecutionPolicy { if (kind === "queued") { @@ -4924,17 +4907,17 @@ export class AgentSession { } return { preparation: { - initialRefineBarrier: kind === "goalContext" ? "skip" : "always", + initialRefineBarrier: "always", flushPendingBashBeforeValidation: false, - validateModelAndAuth: kind === "goalContext", + validateModelAndAuth: false, awaitPendingModelSelection: false, preTurnCompaction: "skip", - finalRefineBarrier: kind === "goalContext" ? "always" : "skip", + finalRefineBarrier: "skip", }, runBeforeAgentStart: false, nextTurnContextTiming: "skip", preserveEmptyExtensionPrompt: false, - completionIncludesRetryChain: kind === "goalContext", + completionIncludesRetryChain: false, }; } @@ -5024,7 +5007,7 @@ export class AgentSession { private _coalescedFollowUpOwner(action: QueuedSessionAction): QueuedSessionAction | undefined { if (action.delivery !== "when_run_idle" || action.payload.kind !== "turn" || !action.queueKey) return undefined; return this._actionStore - .unfinishedActions("when_run_idle") + .unfinishedActions() .find( (candidate) => candidate.queueKey === action.queueKey && @@ -5034,10 +5017,22 @@ export class AgentSession { ); } + private _assertSessionActionAdmissionAvailable(): void { + if (this._disposed || this._disposing) { + throw new Error("Cannot admit a session action because the session is disposing or disposed."); + } + if (this.unfinishedActionCount > 0 && this._sessionInputPumpSuspended) { + throw new Error("Cannot admit a session action while queued session input is suspended."); + } + } + private _admitSessionInput( action: QueuedSessionAction, options: { restore?: boolean; front?: boolean; wake?: boolean; immediatelyEligible?: boolean } = {}, ): { accepted: boolean; disposition: "starts_when_admitted" | "queued"; ticket?: ActionTicket } { + if (this._disposed || this._disposing) { + throw new Error("Cannot admit a session action because the session is disposing or disposed."); + } const coalescedOwner = options.restore ? undefined : this._coalescedFollowUpOwner(action); if (coalescedOwner) { if (action.agentMessageId !== coalescedOwner.agentMessageId) { @@ -5096,7 +5091,7 @@ export class AgentSession { return this._admitSessionInput(action).accepted; } - private _runtimeActivity(mandatoryCheckpointsComplete = true): RuntimeActivity { + private _runtimeActivity(): RuntimeActivity { return { lowerAgentRun: this.isStreaming, compaction: this.isCompacting, @@ -5106,7 +5101,6 @@ export class AgentSession { branchMutation: this._branchSummaryOperation !== undefined, schedulerPauseCount: this._queuedWorkPauses.size + (this._sessionInputPumpSuspended ? 1 : 0), disposing: this._disposed || this._disposing, - mandatoryCheckpointsComplete, }; } @@ -5117,6 +5111,18 @@ export class AgentSession { ); } + get hasPendingSessionWork(): boolean { + return this._actionStore.unfinishedActions().some((action) => { + const state = action.lifecycle.state; + return ( + state === "queued" || + state === "selected" || + state === "preparing" || + (state === "committing" && action.payload.kind === "turn" && !primaryDeliveryRecord(action).durable) + ); + }); + } + private _scheduleSessionInputPump(): void { if (this._sessionInputPumpSuspended || this._queuedWorkPauses.size > 0) return; if (this._disposed || this._disposing || this._sessionInputPumpRequested || !this._hasSelectableSessionInput()) { @@ -5133,183 +5139,195 @@ export class AgentSession { } private async _pumpSessionInputs(epoch: number): Promise { - this._pumpingSessionInput = true; let blocked = false; try { while (!this._disposed && !this._disposing && this._hasSelectableSessionInput()) { await this.agent.waitForIdle(); - const admission = await this._acquireTurnAdmission(); + const preselected = this._actionStore + .activeActions() + .find((action) => action.lifecycle.state === "selected"); + if (epoch !== this._sessionInputPumpEpoch) { + if (preselected) { + this._actionStore.rollback(preselected); + this._notifySessionInputCheckpointChange(); + this._emitQueueUpdate(); + } + return; + } + if (!this._hasCancelledDispatchCapture()) await this._agentEventQueue; + if (!preselected || preselected.payload.kind === "session_command") await this._waitForRefineIdle(); + const activity = this._runtimeActivity(); + const canSelectPreselectedTurn = + preselected?.payload.kind === "turn" && canSelectSessionAction({ ...activity, refinementApply: false }); + if ( + this._isSessionInputHandoffDeferred(epoch) || + (!canSelectPreselectedTurn && !canSelectSessionAction(activity)) + ) { + blocked = true; + this._notifySessionInputCheckpointChange(); + return; + } + const first = preselected ?? this._actionStore.selectFirst(); + if (!first) return; + if (first.payload.kind === "session_command") { + await this._executeSelectedSessionCommand(first, epoch); + return; + } + + const mode = first.delivery === "next_turn_boundary" ? this.steeringMode : this.followUpMode; + const actions: QueuedSessionAction[] = [first]; + while (!preselected && mode === "all") { + const next = this._actionStore.queuedActions(first.delivery)[0]; + if ( + !next || + next.payload.kind !== "turn" || + !turnExecutionPoliciesEqual(first.payload.executionPolicy, next.payload.executionPolicy) + ) { + break; + } + this._actionStore.selectFirst(); + actions.push(next); + } + if (epoch !== this._sessionInputPumpEpoch) { + for (const action of actions) this._actionStore.rollback(action); + return; + } + for (const action of actions) transitionSessionAction(action, { state: "preparing" }); + this._notifySessionInputCheckpointChange(); + this._emitQueueUpdate(); try { - await this._turnAdmissionContext.run(admission.owner, async () => { - const preselected = this._actionStore - .activeActions() - .find((action) => action.lifecycle.state === "selected"); - if (epoch !== this._sessionInputPumpEpoch) { - if (preselected) { - this._actionStore.rollback(preselected); - this._notifySessionInputCheckpointChange(); - this._emitQueueUpdate(); + await this._startPreparedTurnActions(actions, epoch); + for (const action of actions) { + if (action.lifecycle.state === "committing") { + const primary = primaryDeliveryRecord(action); + if (this.agent.state.messages.includes(primary.message)) { + primary.durable = true; + transitionSessionAction(action, { state: "running", execution: "agent_turn" }); } - return; - } - if (!this._hasCancelledDispatchCapture()) await this._agentEventQueue; - if (!preselected || preselected.payload.kind === "session_command") await this._waitForRefineIdle(); - const activity = this._runtimeActivity(); - const canSelectPreselectedTurn = - preselected?.payload.kind === "turn" && - canSelectSessionAction({ ...activity, refinementApply: false }); - if ( - this._isSessionInputHandoffDeferred(epoch) || - (!canSelectPreselectedTurn && !canSelectSessionAction(activity)) - ) { - blocked = true; - return; } - const first = preselected ?? this._actionStore.selectFirst(); - if (!first) return; - if (first.payload.kind === "session_command") { - await this._executeSelectedSessionCommand(first); - return; + if (action.lifecycle.state === "running") { + transitionSessionAction(action, { state: "completed" }); + this._actionStore.ticketFor(action).settleCompleted(); + this._settleAgentMessage(action.agentMessageId, "completion"); } - - const mode = first.delivery === "next_turn_boundary" ? this.steeringMode : this.followUpMode; - const actions: QueuedSessionAction[] = [first]; - while (!preselected && mode === "all") { - const next = this._actionStore.queuedActions(first.delivery)[0]; - if ( - !next || - next.payload.kind !== "turn" || - !turnExecutionPoliciesEqual(first.payload.executionPolicy, next.payload.executionPolicy) - ) { - break; + } + } catch (error) { + const transcript = this.agent.state.messages; + const delivered = new Set(transcript); + const undelivered: QueuedSessionAction[] = []; + for (const action of actions) { + if (action.payload.kind !== "turn" || action.lifecycle.state === "cancelled") continue; + for (const record of action.payload.records) record.durable ||= delivered.has(record.message); + action.payload.records = action.payload.records.filter((record) => { + if (record.role === "prefix") return !record.durable; + if (record.role === "next_turn") return record.durable; + return true; + }); + if (!primaryDeliveryRecord(action).durable) undelivered.push(action); + } + if (this._isDeferredSessionInputError(error, epoch)) { + for (const action of undelivered) { + if (action.lifecycle.state === "committing") { + this._actionStore.rollback(action, { dispatchSettled: true, transcript }); + } else if (action.lifecycle.state === "preparing" || action.lifecycle.state === "selected") { + this._actionStore.rollback(action); } - this._actionStore.selectFirst(); - actions.push(next); } - if (epoch !== this._sessionInputPumpEpoch) { - for (const action of actions) this._actionStore.rollback(action); - return; + if (undelivered.length > 0) this._emitQueueUpdate(); + blocked = epoch !== this._sessionInputPumpEpoch || this._isBusyForSessionInput("pump"); + if (blocked) return; + continue; + } + const terminalError = this._asError(error); + for (const action of actions) { + if (action.lifecycle.state === "cancelled") continue; + if (action.lifecycle.state !== "completed" && action.lifecycle.state !== "failed") { + transitionSessionAction(action, { state: "failed", error: terminalError }); } - for (const action of actions) transitionSessionAction(action, { state: "preparing" }); - this._notifySessionInputCheckpointChange(); - this._emitQueueUpdate(); - try { - await this._startPreparedTurnActions(actions, epoch, admission.release); - for (const action of actions) { - if (action.lifecycle.state === "committing") { - const primary = primaryDeliveryRecord(action); - if (this.agent.state.messages.includes(primary.message)) { - primary.durable = true; - transitionSessionAction(action, { state: "running", execution: "agent_turn" }); - } - } - if (action.lifecycle.state === "running") { - transitionSessionAction(action, { state: "completed" }); - this._actionStore.ticketFor(action).settleCompleted(); - this._settleAgentMessage(action.agentMessageId, "completion"); - } - } - } catch (error) { - const transcript = this.agent.state.messages; - const delivered = new Set(transcript); - const undelivered: QueuedSessionAction[] = []; - for (const action of actions) { - if (action.payload.kind !== "turn" || action.lifecycle.state === "cancelled") continue; - for (const record of action.payload.records) record.durable ||= delivered.has(record.message); - action.payload.records = action.payload.records.filter((record) => { - if (record.role === "prefix") return !record.durable; - if (record.role === "next_turn") return record.durable; - return true; - }); - if (!primaryDeliveryRecord(action).durable) undelivered.push(action); - } - if (this._isDeferredSessionInputError(error, epoch)) { - for (const action of undelivered) { - if (action.lifecycle.state === "committing") { - this._actionStore.rollback(action, { dispatchSettled: true, transcript }); - } else if (action.lifecycle.state === "preparing" || action.lifecycle.state === "selected") { - this._actionStore.rollback(action); - } - } - if (undelivered.length > 0) this._emitQueueUpdate(); - blocked = true; - return; - } - const terminalError = this._asError(error); - for (const action of actions) { - if (action.lifecycle.state === "cancelled") continue; - if (action.lifecycle.state !== "completed" && action.lifecycle.state !== "failed") { - transitionSessionAction(action, { state: "failed", error: terminalError }); - } - const ticket = this._actionStore.ticketFor(action); - if (undelivered.includes(action)) { - ticket.rejectDelivered(terminalError); - this._settleAgentMessage(action.agentMessageId, "delivery", terminalError); - } - this._settleAgentMessage(action.agentMessageId, "completion", terminalError); - ticket.settleCompleted(terminalError); - } - if (actions.some((action) => action.payload.kind !== "turn" || action.payload.queueVisible)) { - this._surfaceSessionInputError(error); - } - } finally { - for (const action of actions) { - const retainedCancelledDispatch = - action.lifecycle.state === "cancelled" && - action.payload.kind === "turn" && - action.payload.captureRunMessages !== undefined; - if ( - !retainedCancelledDispatch && - (action.lifecycle.state === "completed" || - action.lifecycle.state === "failed" || - action.lifecycle.state === "cancelled") - ) { - this._actionStore.releaseTerminal(action); - } - } - this._notifySessionInputCheckpointChange(); - this._emitQueueUpdate(); + const ticket = this._actionStore.ticketFor(action); + if (undelivered.includes(action)) { + ticket.rejectDelivered(terminalError); + this._settleAgentMessage(action.agentMessageId, "delivery", terminalError); } - }); + this._settleAgentMessage(action.agentMessageId, "completion", terminalError); + ticket.settleCompleted(terminalError); + } + if (actions.some((action) => action.payload.kind !== "turn" || action.payload.queueVisible)) { + this._surfaceSessionInputError(error); + } } finally { - admission.release(); + for (const action of actions) { + const retainedCancelledDispatch = + action.lifecycle.state === "cancelled" && + action.payload.kind === "turn" && + action.payload.captureRunMessages !== undefined; + if ( + !retainedCancelledDispatch && + (action.lifecycle.state === "completed" || + action.lifecycle.state === "failed" || + action.lifecycle.state === "cancelled") + ) { + this._actionStore.releaseTerminal(action); + } + } + this._notifySessionInputCheckpointChange(); + this._emitQueueUpdate(); } if (epoch !== this._sessionInputPumpEpoch || blocked) return; } } finally { - this._pumpingSessionInput = false; if (!blocked && epoch === this._sessionInputPumpEpoch && this._hasSelectableSessionInput()) { this._scheduleSessionInputPump(); } } } - private async _executeSelectedSessionCommand(action: QueuedSessionAction): Promise { + private async _executeSelectedSessionCommand(action: QueuedSessionAction, epoch: number): Promise { if (action.payload.kind !== "session_command") throw new Error("Expected a selected session command"); - transitionSessionAction(action, { state: "running", execution: "session_command" }); - this._notifySessionInputCheckpointChange(); - this._emitQueueUpdate(); + const input = action.payload; + const commitFence = await this._acquireSessionActionCommitFence(); try { - await this._executeQueuedSessionCommand(action); - transitionSessionAction(action, { state: "completed" }); - this._actionStore.ticketFor(action).settleDelivered({ status: "not_applicable" }); - this._actionStore.ticketFor(action).settleCompleted(); - this._settleAgentMessage(action.agentMessageId, "completion"); - } catch (error) { - const commandError = this._asError(error); - transitionSessionAction(action, { state: "failed", error: commandError }); - const ticket = this._actionStore.ticketFor(action); - ticket.rejectDelivered(commandError); - ticket.settleCompleted(commandError); - this._rejectAgentMessage(action.agentMessageId, commandError); + await this._sessionActionCommitContext.run(commitFence.owner, async () => { + const isCancelled = () => action.lifecycle.state === "cancelled"; + if (isCancelled()) return; + await this._waitForRefineIdle(); + if (isCancelled()) return; + if (this._isSessionInputHandoffDeferred(epoch) || !canSelectSessionAction(this._runtimeActivity())) { + this._actionStore.rollback(action); + this._notifySessionInputCheckpointChange(); + this._emitQueueUpdate(); + return; + } + transitionSessionAction(action, { state: "running", execution: "session_command" }); + this._notifySessionInputCheckpointChange(); + this._emitQueueUpdate(); + try { + this._appendDurableSessionCommandMessage(input.text, input.command, false); + this._actionStore.ticketFor(action).settleDelivered({ status: "not_applicable" }); + this._settleAgentMessage(action.agentMessageId, "delivery"); + await this._executeQueuedSessionCommand(action); + transitionSessionAction(action, { state: "completed" }); + this._actionStore.ticketFor(action).settleCompleted(); + this._settleAgentMessage(action.agentMessageId, "completion"); + } catch (error) { + const commandError = this._asError(error); + transitionSessionAction(action, { state: "failed", error: commandError }); + const ticket = this._actionStore.ticketFor(action); + ticket.rejectDelivered(commandError); + ticket.settleCompleted(commandError); + this._rejectAgentMessage(action.agentMessageId, commandError); + } finally { + this._actionStore.releaseTerminal(action); + this._notifySessionInputCheckpointChange(); + this._emitQueueUpdate(); + } + }); } finally { - this._actionStore.releaseTerminal(action); - this._notifySessionInputCheckpointChange(); - this._emitQueueUpdate(); + commitFence.release(); } } - private _isBusyForSessionInput(point: "preflight" | "handoff" | "pump"): boolean { + private _isBusyForSessionInput(point: "preflight" | "pump"): boolean { const externalBusy = this.isCompacting || this.isRetrying || this.isBashRunning; if (point === "pump") { return ( @@ -5321,9 +5339,7 @@ export class AgentSession { this._branchSummaryOperation !== undefined ); } - return ( - externalBusy || this._actionStore.unfinishedActions().length > 0 || (point === "handoff" && this.isStreaming) - ); + return externalBusy || this._actionStore.unfinishedActions().length > 0; } private _isSessionInputHandoffDeferred(epoch: number): boolean { @@ -5358,11 +5374,7 @@ export class AgentSession { } } - private async _startPreparedTurnActions( - actions: QueuedSessionAction[], - epoch: number, - releaseAdmission: () => void, - ): Promise { + private async _startPreparedTurnActions(actions: QueuedSessionAction[], epoch: number): Promise { let nextTurnMessages: CustomMessage[] = []; const activeTurns = () => actions.filter( @@ -5424,38 +5436,50 @@ export class AgentSession { restoreNextTurnContext(); return; } - if (this._isSessionInputHandoffDeferred(epoch)) { - throw new DeferredSessionInputError("Session input paused before handoff"); - } - if (this.isStreaming) throw new DeferredSessionInputError("Agent became active before session input handoff"); const { prepared, turns } = preparedTurn; - if (executionPolicy.nextTurnContextTiming === "commit") { - nextTurnMessages = this._takePendingNextTurnMessages(); - } - const contextRecords = nextTurnMessages.map((message) => - this._createDeliveryRecord(turns[0].id, "next_turn", message), - ); - turns[0].payload.records.unshift(...contextRecords); - const preparedMessages: AgentMessage[] = [...nextTurnMessages]; - for (const action of turns) { - preparedMessages.push(...actionPrefixMessages(action), primaryDeliveryRecord(action).message); - if (action.suppressAutonomousContinuation) { - this._markAutonomousContinuationSuppressed(primaryDeliveryRecord(action).message); - } + const commitFence = await this._acquireSessionActionCommitFence(); + let promptPromise: Promise; + try { + promptPromise = this._sessionActionCommitContext.run(commitFence.owner, () => { + if ( + this._isSessionInputHandoffDeferred(epoch) || + this.isStreaming || + turns.some((action) => action.lifecycle.state !== "preparing") + ) { + throw new DeferredSessionInputError("Agent became active before session input handoff"); + } + if (executionPolicy.nextTurnContextTiming === "commit") { + nextTurnMessages = this._takePendingNextTurnMessages(); + } + const contextRecords = nextTurnMessages.map((message) => + this._createDeliveryRecord(turns[0].id, "next_turn", message), + ); + const firstPrimaryIndex = turns[0].payload.records.indexOf(primaryDeliveryRecord(turns[0])); + turns[0].payload.records.splice(firstPrimaryIndex, 0, ...contextRecords); + const preparedMessages: AgentMessage[] = turns.flatMap((action) => + action.payload.records.map((record) => record.message), + ); + for (const action of turns) { + if (action.suppressAutonomousContinuation) { + this._markAutonomousContinuationSuppressed(primaryDeliveryRecord(action).message); + } + } + if (executionPolicy.runBeforeAgentStart) { + this._appendBeforeAgentStartMessages(preparedMessages, prepared?.result); + this._applyPreparedSystemPrompt(prepared, executionPolicy.preserveEmptyExtensionPrompt); + } else if (executionPolicy.nextTurnContextTiming !== "skip") { + this.agent.state.systemPrompt = this._baseSystemPrompt; + } + for (const action of turns) transitionSessionAction(action, { state: "committing" }); + this._notifySessionInputCheckpointChange(); + this._emitQueueUpdate(); + return turns.some((action) => action.suppressAutonomousContinuation) + ? this._runWithAutonomousContinuationSuppressed(() => this.agent.prompt(preparedMessages)) + : this.agent.prompt(preparedMessages); + }); + } finally { + commitFence.release(); } - if (executionPolicy.runBeforeAgentStart) { - this._appendBeforeAgentStartMessages(preparedMessages, prepared?.result); - this._applyPreparedSystemPrompt(prepared, executionPolicy.preserveEmptyExtensionPrompt); - } else if (executionPolicy.nextTurnContextTiming !== "skip") { - this.agent.state.systemPrompt = this._baseSystemPrompt; - } - for (const action of turns) transitionSessionAction(action, { state: "committing" }); - this._notifySessionInputCheckpointChange(); - this._emitQueueUpdate(); - const promptPromise = turns.some((action) => action.suppressAutonomousContinuation) - ? this._runWithAutonomousContinuationSuppressed(() => this.agent.prompt(preparedMessages)) - : this.agent.prompt(preparedMessages); - releaseAdmission(); await promptPromise; if (executionPolicy.completionIncludesRetryChain) await this.waitForRetry(); if (!this._hasCancelledDispatchCapture()) await this._agentEventQueue; @@ -5485,9 +5509,6 @@ export class AgentSession { private async _executeQueuedSessionCommand(action: QueuedSessionAction): Promise { if (action.payload.kind !== "session_command") throw new Error("Expected a session command action"); const input = action.payload; - this._appendDurableSessionCommandMessage(input.text, input.command, false); - this._actionStore.ticketFor(action).settleDelivered({ status: "not_applicable" }); - this._settleAgentMessage(action.agentMessageId, "delivery"); try { let resultText: string | undefined; switch (input.command.name) { @@ -5511,7 +5532,9 @@ export class AgentSession { await this._handleAutonomousSlashCommand(input.text); break; } - if (resultText) this._appendDurableSessionCommandMessage(resultText, input.command, true, false); + if (resultText) { + this._appendDurableSessionCommandMessage(resultText, input.command, true, false); + } } catch (error) { if (error instanceof CompactionSkippedError) return; const commandError = error instanceof Error ? error : new Error(String(error)); @@ -5610,7 +5633,7 @@ export class AgentSession { }); } } else if (options?.triggerTurn) { - const admission = await this._acquireSessionActionAdmission(); + const admissionFence = await this._acquireDirectTurnAdmissionFence(); try { const normalized = normalizeMessageContent(message.content); const immediatelyEligible = this._canStartSessionActionImmediately(); @@ -5621,12 +5644,11 @@ export class AgentSession { queueVisible: false, }); const result = this._admitSessionInput(action, { immediatelyEligible }); + admissionFence.release(); if (!result.ticket) return; - admission.release(); await result.ticket.completed; } finally { - admission.release(); - this._scheduleSessionInputPump(); + admissionFence.release(); } } else { this.agent.state.messages.push(appMessage); @@ -5907,26 +5929,24 @@ export class AgentSession { for (const resolve of waiters) resolve(); } - private _observeSessionActionRollback(action: QueuedSessionAction): { rolledBack: Promise; stop(): void } { - let resolveRollback = () => {}; - const rolledBack = new Promise((resolve) => { - resolveRollback = resolve; + private _observeSessionActionDeferral(action: QueuedSessionAction): { deferred: Promise; stop(): void } { + let resolveDeferral = () => {}; + const deferred = new Promise((resolve) => { + resolveDeferral = resolve; }); const check = () => { - if (action.lifecycle.state === "queued") resolveRollback(); + if (action.lifecycle.state === "queued") resolveDeferral(); else this._sessionInputCheckpointWaiters.add(check); }; this._sessionInputCheckpointWaiters.add(check); return { - rolledBack, + deferred, stop: () => this._sessionInputCheckpointWaiters.delete(check), }; } async waitForSessionInputCheckpoint(signal?: AbortSignal): Promise { - // An executing command was already shifted off the queue; the snapshot - // must wait for it or it vanishes from durable queued input. - while ( + const blocksCheckpoint = () => this._actionStore.activeActions().some((action) => { if (action.payload.kind === "session_command") { return action.lifecycle.state === "selected" || action.lifecycle.state === "running"; @@ -5936,32 +5956,40 @@ export class AgentSession { action.lifecycle.state === "preparing" || (action.lifecycle.state === "committing" && !primaryDeliveryRecord(action).durable) ); - }) - ) { - if (signal?.aborted) throw new Error("Update restart preparation cancelled"); - await new Promise((resolve, reject) => { - const onChange = () => { - cleanup(); - resolve(); - }; - const onAbort = () => { - cleanup(); - reject(new Error("Update restart preparation cancelled")); - }; - const cleanup = () => { - this._sessionInputCheckpointWaiters.delete(onChange); - signal?.removeEventListener("abort", onAbort); - }; - this._sessionInputCheckpointWaiters.add(onChange); - signal?.addEventListener("abort", onAbort, { once: true }); }); + while (true) { + while (blocksCheckpoint()) { + if (signal?.aborted) throw new Error("Update restart preparation cancelled"); + await new Promise((resolve, reject) => { + const onChange = () => { + cleanup(); + resolve(); + }; + const onAbort = () => { + cleanup(); + reject(new Error("Update restart preparation cancelled")); + }; + const cleanup = () => { + this._sessionInputCheckpointWaiters.delete(onChange); + signal?.removeEventListener("abort", onAbort); + }; + this._sessionInputCheckpointWaiters.add(onChange); + signal?.addEventListener("abort", onAbort, { once: true }); + if (signal?.aborted) onAbort(); + }); + } + const commitFence = await this._acquireSessionActionCommitFence(signal); + try { + if (blocksCheckpoint()) continue; + if (signal?.aborted) throw new Error("Update restart preparation cancelled"); + await waitForPromiseOrAbort(this._agentEventQueue, signal, "Update restart preparation cancelled"); + if (signal?.aborted) throw new Error("Update restart preparation cancelled"); + this.sessionManager.flushNow(); + return; + } finally { + commitFence.release(); + } } - if (signal?.aborted) throw new Error("Update restart preparation cancelled"); - // Handed-off inputs reach the transcript through the event queue; drain the - // current snapshot so the flush below persists them before the snapshot is taken. - await waitForPromiseOrAbort(this._agentEventQueue, signal, "Update restart preparation cancelled"); - if (signal?.aborted) throw new Error("Update restart preparation cancelled"); - this.sessionManager.flushNow(); } acquireQueuedWorkPause(): { release(): void } { @@ -5975,100 +6003,90 @@ export class AgentSession { if (released) return; released = true; this._queuedWorkPauses.delete(token); - if (this._queuedWorkPauses.size === 0) this._wakeQueuedWorkAdmissionWaiters(); this._notifySessionInputCheckpointChange(); this._scheduleSessionInputPump(); }, }; } - private _wakeQueuedWorkAdmissionWaiters(): void { - for (const resolve of this._queuedWorkAdmissionWaiters) resolve(); - this._queuedWorkAdmissionWaiters.clear(); - } - - private _assertSessionActionAdmissionAvailable(): void { - if (this._disposed || this._disposing) { - throw new Error("Cannot admit a session action because the session is disposing or disposed."); + private async _acquireDirectTurnAdmissionFence(signal?: AbortSignal): Promise<{ owner: symbol; release(): void }> { + const inheritedOwner = this._sessionActionCommitContext.getStore(); + if (inheritedOwner !== undefined && inheritedOwner === this._sessionActionCommitOwner) { + this._assertSessionActionAdmissionAvailable(); + return this._acquireSessionActionCommitFence(signal); } - if (this.unfinishedActionCount > 0 && this._sessionInputPumpSuspended) { - throw new Error("Cannot admit a session action while queued session input is suspended."); + const disposeSignal = this._sessionActionCommitDisposeAbortController.signal; + const waitSignal = signal ? AbortSignal.any([signal, disposeSignal]) : disposeSignal; + while (true) { + this._assertSessionActionAdmissionAvailable(); + if (this._queuedWorkPauses.size > 0) { + let wake = () => {}; + const pauseReleased = new Promise((resolve) => { + wake = resolve; + this._sessionInputCheckpointWaiters.add(resolve); + }); + try { + await waitForPromiseOrAbort(pauseReleased, waitSignal, "Update restart preparation cancelled"); + } catch (error) { + if (disposeSignal.aborted) { + throw new Error("Cannot admit a session action because the session is disposing or disposed."); + } + throw error; + } finally { + this._sessionInputCheckpointWaiters.delete(wake); + } + continue; + } + const fence = await this._acquireSessionActionCommitFence(signal); + try { + if (this._queuedWorkPauses.size === 0) { + this._assertSessionActionAdmissionAvailable(); + return fence; + } + } catch (error) { + fence.release(); + throw error; + } + fence.release(); } } - /** PR 5 deletion target: preparation still shares this lease with branch mutation. */ - private async _acquireTurnAdmission(signal?: AbortSignal): Promise<{ owner: symbol; release(): void }> { - const inheritedOwner = this._turnAdmissionContext.getStore(); - if (inheritedOwner !== undefined && inheritedOwner === this._turnAdmissionOwner) { + private async _acquireSessionActionCommitFence(signal?: AbortSignal): Promise<{ owner: symbol; release(): void }> { + const inheritedOwner = this._sessionActionCommitContext.getStore(); + if (inheritedOwner !== undefined && inheritedOwner === this._sessionActionCommitOwner) { return { owner: inheritedOwner, release: () => {} }; } - const previous = this._turnAdmissionTail; + const previous = this._sessionActionCommitTail; let resolve = () => {}; - this._turnAdmissionTail = new Promise((release) => { + this._sessionActionCommitTail = new Promise((release) => { resolve = release; }); + const disposeSignal = this._sessionActionCommitDisposeAbortController.signal; + const waitSignal = signal ? AbortSignal.any([signal, disposeSignal]) : disposeSignal; try { - await waitForPromptAdmission(previous, signal); + await waitForPromiseOrAbort(previous, waitSignal, "Update restart preparation cancelled"); } catch (error) { - // Keep this cancelled waiter in the FIFO chain until its predecessor - // releases; resolving immediately would let a later caller bypass the - // current admission owner. + // A cancelled waiter remains in the FIFO chain until its predecessor releases. void previous.then(resolve, resolve); + if (disposeSignal.aborted) { + throw new Error("Cannot admit a session action because the session is disposing or disposed."); + } throw error; } - const owner = Symbol("turn-admission"); - this._turnAdmissionOwner = owner; + const owner = Symbol("session-action-commit"); + this._sessionActionCommitOwner = owner; let released = false; return { owner, release: () => { if (released) return; released = true; - if (this._turnAdmissionOwner === owner) this._turnAdmissionOwner = undefined; + if (this._sessionActionCommitOwner === owner) this._sessionActionCommitOwner = undefined; resolve(); }, }; } - private async _acquireSessionActionAdmission( - options: { signal?: AbortSignal } = {}, - ): Promise<{ owner: symbol; release(): void }> { - while (true) { - this._assertSessionActionAdmissionAvailable(); - await this._waitForQueuedWorkAdmission(options.signal); - const admission = await this._acquireTurnAdmission(options.signal); - try { - this._assertSessionActionAdmissionAvailable(); - if (this._queuedWorkPauses.size > 0) { - admission.release(); - continue; - } - throwIfPromptAdmissionCancelled(options.signal); - return admission; - } catch (error) { - admission.release(); - throw error; - } - } - } - - private async _waitForQueuedWorkAdmission(signal?: AbortSignal): Promise { - while (this._queuedWorkPauses.size > 0) { - this._assertSessionActionAdmissionAvailable(); - let wake = () => {}; - const wait = new Promise((resolve) => { - wake = resolve; - this._queuedWorkAdmissionWaiters.add(resolve); - }); - try { - await waitForPromptAdmission(wait, signal); - } finally { - this._queuedWorkAdmissionWaiters.delete(wake); - } - this._assertSessionActionAdmissionAvailable(); - } - } - /** Resume the scheduler after requestAbort/abortForUpdateRestart suspended it; owned pause leases are unaffected. */ resumeQueuedWork(): boolean { this._sessionInputPumpSuspended = false; @@ -6081,7 +6099,7 @@ export class AgentSession { while (true) { const pump = this._sessionInputPump; await pump; - if (pump === this._sessionInputPump && !this._sessionInputPumpRequested && !this._pumpingSessionInput) return; + if (pump === this._sessionInputPump && !this._sessionInputPumpRequested) return; } } @@ -6097,10 +6115,12 @@ export class AgentSession { const pump = this._sessionInputPump; await pump; await this.agent.waitForIdle(); + const agentEventQueue = this._agentEventQueue; + await agentEventQueue; if ( pump === this._sessionInputPump && + agentEventQueue === this._agentEventQueue && !this._sessionInputPumpRequested && - !this._pumpingSessionInput && !this.agent.state.isStreaming && this.unfinishedActionCount === 0 ) { @@ -6137,18 +6157,6 @@ export class AgentSession { this._pendingNextTurnMessages.push(...messages.map((message) => cloneCustomMessage(message))); } - hasQueuedFollowUp(queueKey: string): boolean { - return this._actionStore - .unfinishedActions("when_run_idle") - .some( - (action) => - action.queueKey === queueKey && - (action.lifecycle.state === "queued" || - action.lifecycle.state === "selected" || - action.lifecycle.state === "preparing"), - ); - } - removeQueuedFollowUp(queueKey: string): boolean { const matching = this._actionStore .clearableActions() @@ -6969,7 +6977,7 @@ export class AgentSession { return; } // An empty queue is not idle while the scheduler still owns active work. - if (this.unfinishedActionCount > 0 || this._pumpingSessionInput || this._sessionInputPumpRequested) { + if (this.unfinishedActionCount > 0 || this._sessionInputPumpRequested) { this._scheduleSessionInputPump(); await this._sessionInputPump; if (this._postCompactionContinuationScheduled) { @@ -7733,9 +7741,7 @@ export class AgentSession { const resumeAfterFailure = () => { if ( reason === "requested" && - (shouldContinueAfterCompaction || - this.agent.hasQueuedMessages() || - this._actionStore.queuedActions().length > 0) + (shouldContinueAfterCompaction || this.agent.hasQueuedMessages() || this.hasPendingSessionWork) ) { this._schedulePostCompactionContinue(); } @@ -7772,7 +7778,7 @@ export class AgentSession { this._emit({ type: "compaction_end", reason, result, aborted: false, willRetry, customInstructions }); // Queued work lives in both the agent queues and the session-owned queues. - const hasQueuedMessages = this.agent.hasQueuedMessages() || this._actionStore.queuedActions().length > 0; + const hasQueuedMessages = this.agent.hasQueuedMessages() || this.hasPendingSessionWork; const willContinueAfterCompaction = willRetry || shouldContinueAfterCompaction || hasQueuedMessages; if (willRetry) { @@ -10024,17 +10030,18 @@ export class AgentSession { } const queuedWorkPause = this.acquireQueuedWorkPause(); - let admission: { owner: symbol; release(): void } | undefined; + let commitFence: { owner: symbol; release(): void } | undefined; try { - admission = await this._acquireTurnAdmission(); - return await this._turnAdmissionContext.run(admission.owner, async () => { + // Branch navigation and turn dispatch mutate the same transcript leaf. + commitFence = await this._acquireSessionActionCommitFence(); + return await this._sessionActionCommitContext.run(commitFence.owner, async () => { await this.agent.waitForIdle(); await this._agentEventQueue; return this._navigateTreeUnderPause(targetId, targetEntry, options); }); } finally { queuedWorkPause.release(); - admission?.release(); + commitFence?.release(); } } diff --git a/packages/coding-agent/src/core/cron-jobs.ts b/packages/coding-agent/src/core/cron-jobs.ts index 8e8e2254f..999e68659 100644 --- a/packages/coding-agent/src/core/cron-jobs.ts +++ b/packages/coding-agent/src/core/cron-jobs.ts @@ -87,6 +87,7 @@ export interface HeartbeatCronSessionActivity { isCompacting?: boolean; isRetrying?: boolean; isBashRunning: boolean; + hasPendingSessionWork: boolean; unfinishedActionCount: number; } @@ -1357,7 +1358,8 @@ export function shouldDeferHeartbeatCronJob(job: AgentCronJob, activity: Heartbe activity.isCompacting === true || activity.isRetrying === true || activity.isBashRunning || - activity.unfinishedActionCount > 0; + activity.hasPendingSessionWork || + (!activity.isStreaming && activity.unfinishedActionCount > 0); if (busyBesidesStreaming) { return true; } diff --git a/packages/coding-agent/src/core/session-action-store.ts b/packages/coding-agent/src/core/session-action-store.ts index db969f413..8ed497985 100644 --- a/packages/coding-agent/src/core/session-action-store.ts +++ b/packages/coding-agent/src/core/session-action-store.ts @@ -307,7 +307,6 @@ export interface RuntimeActivity { branchMutation: boolean; schedulerPauseCount: number; disposing: boolean; - mandatoryCheckpointsComplete: boolean; } export function canSelectSessionAction(activity: RuntimeActivity): boolean { @@ -322,7 +321,3 @@ export function canSelectSessionAction(activity: RuntimeActivity): boolean { !activity.disposing ); } - -export function shouldYieldLowerRun(store: ActionStore, activity: RuntimeActivity): boolean { - return activity.mandatoryCheckpointsComplete && store.queuedActions("next_turn_boundary").length > 0; -} diff --git a/packages/coding-agent/src/modes/agent-connection/daemon-agent-connection.ts b/packages/coding-agent/src/modes/agent-connection/daemon-agent-connection.ts index 1aa3c8246..a739407c4 100644 --- a/packages/coding-agent/src/modes/agent-connection/daemon-agent-connection.ts +++ b/packages/coding-agent/src/modes/agent-connection/daemon-agent-connection.ts @@ -568,20 +568,16 @@ export class DaemonAgentConnection implements AgentConnection { jobId: string, action: AgentHeartbeatManagementAction, ): Promise { - const hasCapability = this.client.supportsServerCapability("heartbeat_management"); - if (!hasCapability && this.client.hello?.protocol.version !== 3) { + if (!this.client.supportsServerCapability("heartbeat_management")) { throw new Error("Heartbeat management requires a newer Prime Agent daemon."); } try { - const command = { + const data = await this.requestData<{ heartbeat: AgentCronJob }>({ type: "heartbeat_manage", activeSessionId, jobId, action, - } as const; - const data = hasCapability - ? await this.requestData<{ heartbeat: AgentCronJob }>(command) - : await this.requestLegacyData<{ heartbeat: AgentCronJob }>(command); + }); return data.heartbeat; } catch (error) { if (isUnknownDaemonCommandError(error, "heartbeat_manage")) { @@ -1380,14 +1376,6 @@ export class DaemonAgentConnection implements AgentConnection { await this.requestData(command); } - private async requestLegacyData(command: DaemonCommandBody, timeoutMs?: number): Promise { - const response = await this.client.requestLegacy(command, timeoutMs); - if (!response.success) { - throw deserializeDaemonError(response); - } - return response.data as T; - } - private async requestData( command: DaemonCommandBody, timeoutMs?: number, diff --git a/packages/coding-agent/src/modes/daemon/daemon-client.ts b/packages/coding-agent/src/modes/daemon/daemon-client.ts index f87090f78..cffdb6978 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-client.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-client.ts @@ -328,15 +328,6 @@ export class DaemonClient { ); } - /** One-release compatibility path for preparing and stopping a v1 daemon. */ - async requestLegacy( - command: DaemonCommandBody, - timeoutMs = 30000, - options: DaemonClientRequestOptions = {}, - ): Promise { - return this.requestWire(command, timeoutMs, options); - } - async authenticateWorker(token: string, timeoutMs = 3000): Promise { const legacyAuthentication = { type: "worker_auth", token } as DaemonWorkerCommandBody; const response = await this.requestWire(legacyAuthentication, timeoutMs); diff --git a/packages/coding-agent/test/agent-connection-daemon.test.ts b/packages/coding-agent/test/agent-connection-daemon.test.ts index 0d8bf77cc..a4da5b6db 100644 --- a/packages/coding-agent/test/agent-connection-daemon.test.ts +++ b/packages/coding-agent/test/agent-connection-daemon.test.ts @@ -53,7 +53,6 @@ class FakeDaemonClient { promptError: Error | undefined; promptResponseError: string | undefined; cancelPromptAdmissionStatus: "cancelled" | "owned" | "unknown" = "owned"; - legacyHeartbeatCommandsSupported = false; serverCapabilities = new Set(); updateRestartSessions: Array> = []; hello: DaemonHello | undefined = { @@ -271,7 +270,7 @@ class FakeDaemonClient { data: { steering: ["aborted"], followUp: ["cleared"] }, }; case "heartbeats_list": - return this.legacyHeartbeatCommandsSupported || this.serverCapabilities.has("heartbeat_catalog") + return this.serverCapabilities.has("heartbeat_catalog") ? { type: "response", command: command.type, success: true, data: { heartbeats: [] } } : { type: "response", @@ -280,7 +279,7 @@ class FakeDaemonClient { error: "Unknown daemon command: heartbeats_list", }; case "heartbeat_manage": - return this.legacyHeartbeatCommandsSupported || this.serverCapabilities.has("heartbeat_management") + return this.serverCapabilities.has("heartbeat_management") ? { type: "response", command: command.type, @@ -467,14 +466,6 @@ class FakeDaemonClient { } } - async requestLegacy( - command: DaemonCommand, - timeoutMs = 30000, - options: DaemonClientRequestOptions = {}, - ): Promise { - return this.request(command, timeoutMs, options); - } - supportsServerCapability(capability: string): boolean { return this.serverCapabilities.has(capability); } @@ -965,43 +956,6 @@ describe("DaemonAgentConnection", () => { expect(fakeClient.requests).toEqual([]); }); - it("does not query the heartbeat catalog on a retained protocol-3 daemon", async () => { - const fakeClient = new FakeDaemonClient(); - fakeClient.hello = { - type: "daemon_hello", - socketPath: "/tmp/prime-agent.sock", - protocol: { ...DAEMON_PROTOCOL_INFO, version: 3 }, - clientId: "legacy-client", - serverCapabilities: [], - }; - fakeClient.legacyHeartbeatCommandsSupported = true; - const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-original"); - - await expect(connection.listHeartbeats()).resolves.toEqual([]); - await expect(connection.manageHeartbeat("active-original", "job-1", "pause")).resolves.toMatchObject({ - id: "job-1", - }); - expect(fakeClient.requests.map((request) => request.type)).toEqual(["heartbeat_manage"]); - }); - - it("degrades heartbeat commands missing from an older protocol-3 daemon", async () => { - const fakeClient = new FakeDaemonClient(); - fakeClient.hello = { - type: "daemon_hello", - socketPath: "/tmp/prime-agent.sock", - protocol: { ...DAEMON_PROTOCOL_INFO, version: 3 }, - clientId: "legacy-client", - serverCapabilities: [], - }; - const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-original"); - - await expect(connection.listHeartbeats()).resolves.toEqual([]); - await expect(connection.manageHeartbeat("active-original", "job-1", "pause")).rejects.toThrow( - "Heartbeat management requires a newer Prime Agent daemon.", - ); - expect(fakeClient.requests.map((request) => request.type)).toEqual(["heartbeat_manage"]); - }); - it("reattaches an open window to its restored session after an update restart", async () => { const fakeClient = new FakeDaemonClient(); fakeClient.emitCloseOnClose = true; diff --git a/packages/coding-agent/test/cron-jobs.test.ts b/packages/coding-agent/test/cron-jobs.test.ts index 0229a37d0..838dd688e 100644 --- a/packages/coding-agent/test/cron-jobs.test.ts +++ b/packages/coding-agent/test/cron-jobs.test.ts @@ -1352,6 +1352,7 @@ describe("shouldDeferHeartbeatCronJob", () => { shouldDeferHeartbeatCronJob(job, { isStreaming: true, isBashRunning: false, + hasPendingSessionWork: false, unfinishedActionCount: 0, }), ).toBe(true); @@ -1359,6 +1360,7 @@ describe("shouldDeferHeartbeatCronJob", () => { shouldDeferHeartbeatCronJob(job, { isStreaming: false, isBashRunning: true, + hasPendingSessionWork: false, unfinishedActionCount: 0, }), ).toBe(true); @@ -1366,6 +1368,7 @@ describe("shouldDeferHeartbeatCronJob", () => { shouldDeferHeartbeatCronJob(job, { isStreaming: false, isBashRunning: false, + hasPendingSessionWork: false, unfinishedActionCount: 1, }), ).toBe(true); @@ -1383,7 +1386,8 @@ describe("shouldDeferHeartbeatCronJob", () => { shouldDeferHeartbeatCronJob(job, { isStreaming: true, isBashRunning: false, - unfinishedActionCount: 0, + hasPendingSessionWork: false, + unfinishedActionCount: 1, }), ).toBe(false); } @@ -1398,6 +1402,7 @@ describe("shouldDeferHeartbeatCronJob", () => { isStreaming: true, isCompacting: true, isBashRunning: false, + hasPendingSessionWork: false, unfinishedActionCount: 0, }), ).toBe(true); @@ -1405,6 +1410,7 @@ describe("shouldDeferHeartbeatCronJob", () => { shouldDeferHeartbeatCronJob(job, { isStreaming: false, isBashRunning: true, + hasPendingSessionWork: false, unfinishedActionCount: 0, }), ).toBe(true); @@ -1412,6 +1418,7 @@ describe("shouldDeferHeartbeatCronJob", () => { shouldDeferHeartbeatCronJob(job, { isStreaming: false, isBashRunning: false, + hasPendingSessionWork: false, unfinishedActionCount: 1, }), ).toBe(true); @@ -1420,14 +1427,16 @@ describe("shouldDeferHeartbeatCronJob", () => { isStreaming: false, isRetrying: true, isBashRunning: false, + hasPendingSessionWork: false, unfinishedActionCount: 0, }), ).toBe(true); expect( shouldDeferHeartbeatCronJob(job, { - isStreaming: false, + isStreaming: true, isBashRunning: false, - unfinishedActionCount: 1, + hasPendingSessionWork: true, + unfinishedActionCount: 2, }), ).toBe(true); }); @@ -1436,7 +1445,7 @@ describe("shouldDeferHeartbeatCronJob", () => { expect( shouldDeferHeartbeatCronJob( { ...baseJob, source: "heartbeat" }, - { isStreaming: false, isBashRunning: false, unfinishedActionCount: 0 }, + { isStreaming: false, isBashRunning: false, hasPendingSessionWork: false, unfinishedActionCount: 0 }, ), ).toBe(false); }); @@ -1445,7 +1454,7 @@ describe("shouldDeferHeartbeatCronJob", () => { expect( shouldDeferHeartbeatCronJob( { ...baseJob, source: "cron" }, - { isStreaming: true, isBashRunning: true, unfinishedActionCount: 2 }, + { isStreaming: true, isBashRunning: true, hasPendingSessionWork: true, unfinishedActionCount: 2 }, ), ).toBe(false); }); diff --git a/packages/coding-agent/test/daemon-client.test.ts b/packages/coding-agent/test/daemon-client.test.ts index d43f3c39b..8520b90e1 100644 --- a/packages/coding-agent/test/daemon-client.test.ts +++ b/packages/coding-agent/test/daemon-client.test.ts @@ -384,25 +384,6 @@ describe("DaemonClient", () => { }); }); - it("keeps a raw one-release request path for v1 daemon handoff", async () => { - const client = new DaemonClient("/tmp/prime-agent.sock"); - const connect = client.connect(); - const socket = netMock.sockets[0]!; - socket.emit("connect"); - await connect; - - const response = client.requestLegacy({ type: "prepare_update_restart" }); - const command = JSON.parse(socket.writes[0]!.trim()) as { id: string; type: string }; - expect(command.type).toBe("prepare_update_restart"); - expect(command).not.toHaveProperty("protocol"); - socket.emit( - "data", - `${JSON.stringify({ id: command.id, type: "response", command: command.type, success: true })}\n`, - ); - await expect(response).resolves.toMatchObject({ success: true }); - client.close(); - }); - it("keeps durable command envelopes on the session-action protocol", async () => { const client = new DaemonClient("/tmp/prime-agent.sock"); const connect = client.connect(); diff --git a/packages/coding-agent/test/daemon-mode.test.ts b/packages/coding-agent/test/daemon-mode.test.ts index 78b50cd6b..7c15f5b55 100644 --- a/packages/coding-agent/test/daemon-mode.test.ts +++ b/packages/coding-agent/test/daemon-mode.test.ts @@ -4620,7 +4620,7 @@ describe("daemon mode helpers", () => { }, { name: "does not enqueue another heartbeat when one is already pending", - activity: { isStreaming: true, unfinishedActionCount: 1 }, + activity: { isStreaming: true, hasPendingSessionWork: true, unfinishedActionCount: 1 }, jobs: [{ id: "heartbeat-1", source: "heartbeat" }], acceptingAgentMessage: false, assertQueuedHeartbeatUntouched: true, @@ -4742,6 +4742,8 @@ describe("daemon mode helpers", () => { const sessionState = { isStreaming: false, isBashRunning: false, + hasPendingSessionWork: false, + unfinishedActionCount: 1, sessionActions: { queuedCount: 0, steering: [], followUps: [] }, }; const prompt = vi.fn(async (_message: string, options?: { preflightResult?: (didSucceed: boolean) => void }) => { @@ -5803,6 +5805,7 @@ type CronAdmissionActivity = Partial<{ isCompacting: boolean; isRetrying: boolean; isBashRunning: boolean; + hasPendingSessionWork: boolean; unfinishedActionCount: number; }>; @@ -5841,6 +5844,7 @@ function makeCronAdmissionFixture( isCompacting: false, isRetrying: false, isBashRunning: false, + hasPendingSessionWork: false, unfinishedActionCount: 0, ...activity, prompt, diff --git a/packages/coding-agent/test/session-action-store.test.ts b/packages/coding-agent/test/session-action-store.test.ts index 7dc46a78f..2df70f9dd 100644 --- a/packages/coding-agent/test/session-action-store.test.ts +++ b/packages/coding-agent/test/session-action-store.test.ts @@ -67,7 +67,6 @@ function activity(overrides: Partial = {}): RuntimeActivity { branchMutation: false, schedulerPauseCount: 0, disposing: false, - mandatoryCheckpointsComplete: true, ...overrides, }; } diff --git a/packages/coding-agent/test/suite/agent-session-action-races.test.ts b/packages/coding-agent/test/suite/agent-session-action-races.test.ts new file mode 100644 index 000000000..86c755870 --- /dev/null +++ b/packages/coding-agent/test/suite/agent-session-action-races.test.ts @@ -0,0 +1,377 @@ +import { fauxAssistantMessage } from "@earendil-works/pi-ai"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { CustomMessage } from "../../src/core/messages.js"; +import type { ActionStore, SessionAction } from "../../src/core/session-action-store.js"; +import { createHarness, getMessageText, getUserTexts, type Harness } from "./harness.js"; +import { createDeferred } from "./scheduling.js"; + +type ActionKind = "turn" | "command"; + +interface CommitFenceInternals { + _actionStore: ActionStore; + _refineInFlight?: Promise; + _scheduleSessionInputPump(): void; + _acquireDirectTurnAdmissionFence(signal?: AbortSignal): Promise<{ release(): void }>; + _acquireSessionActionCommitFence(signal?: AbortSignal): Promise<{ release(): void }>; + _promptInjectedMessage( + text: string, + message: { + role: "custom"; + customType: string; + content: string; + display: boolean; + details: Record; + timestamp: number; + }, + options?: { returnAfterAccepted?: boolean }, + ): Promise; +} + +function deliveredCount(harness: Harness, kind: ActionKind, text: string): number { + if (kind === "turn") return getUserTexts(harness).filter((candidate) => candidate === text).length; + return harness.session.messages.filter((message) => getMessageText(message) === text).length; +} + +function yieldToEventLoop(): Promise { + return new Promise((resolve) => setImmediate(resolve)); +} + +function createContextMessage(content: string): CustomMessage { + return { + role: "custom", + customType: "action-order-context", + content, + display: false, + timestamp: Date.now(), + }; +} + +describe("AgentSession action commit-fence races", () => { + const harnesses: Harness[] = []; + + afterEach(() => { + while (harnesses.length > 0) harnesses.pop()?.cleanup(); + }); + + it.each([ + ...["turn", "command"].flatMap((kind) => + (["pause", "clear", "restart", "dispose"] as const).map((interruption) => ({ + kind: kind as ActionKind, + interruption, + })), + ), + ])("settles one $kind exactly once when $interruption wins the commit fence", async ({ kind, interruption }) => { + const harness = await createHarness(); + harnesses.push(harness); + if (kind === "turn") harness.setResponses([fauxAssistantMessage("done")]); + const text = kind === "turn" ? "commit-fence turn" : "/autonomous status"; + const reached = createDeferred(); + const releaseFence = createDeferred(); + const internals = harness.session as unknown as CommitFenceInternals; + const acquireFence = internals._acquireSessionActionCommitFence.bind(internals); + internals._acquireSessionActionCommitFence = async () => { + if (internals._actionStore.unfinishedActions().length === 0) return acquireFence(); + reached.resolve(); + await releaseFence.promise; + return acquireFence(); + }; + + let completion: Promise; + if (kind === "turn") { + expect(await harness.session.followUp(text, undefined, { resumeIfIdle: true })).toBe(true); + const action = internals._actionStore.unfinishedActions()[0]; + if (!action) throw new Error("Expected an owned turn action"); + completion = internals._actionStore.ticketFor(action).ticket.completed; + } else { + completion = harness.session.promptAndWait(text); + } + let settlementCount = 0; + const outcome = completion.then( + () => { + settlementCount++; + return "resolved" as const; + }, + () => { + settlementCount++; + return "rejected" as const; + }, + ); + await reached.promise; + + let pause: { release(): void } | undefined; + if (interruption === "pause") pause = harness.session.acquireQueuedWorkPause(); + else if (interruption === "clear") { + expect(harness.session.clearQueue().followUp).toEqual([text]); + } else if (interruption === "restart") harness.session.abortForUpdateRestart(); + else harness.session.dispose(); + releaseFence.resolve(); + + if (interruption === "pause" || interruption === "restart") { + await harness.session.waitForSessionInputCheckpoint(); + expect(deliveredCount(harness, kind, text)).toBe(0); + expect(harness.session.getSessionActionRecoverySnapshot().actions).toHaveLength(1); + if (interruption === "pause") pause?.release(); + else harness.session.resumeQueuedWork(); + expect(await outcome).toBe("resolved"); + await harness.session.waitForIdle(); + expect(deliveredCount(harness, kind, text)).toBe(1); + expect(harness.session.getSessionActionRecoverySnapshot().actions).toHaveLength(0); + } else { + expect(await outcome).toBe("rejected"); + expect(deliveredCount(harness, kind, text)).toBe(0); + expect(harness.session.getSessionActionRecoverySnapshot().actions).toHaveLength(0); + } + await Promise.resolve(); + expect(settlementCount).toBe(1); + }); + + it("aborts a prompt waiting for the commit fence without leaking the fence", async () => { + const harness = await createHarness(); + harnesses.push(harness); + const internals = harness.session as unknown as CommitFenceInternals; + const heldFence = await internals._acquireSessionActionCommitFence(); + const controller = new AbortController(); + const prompt = harness.session.prompt("blocked prompt", { signal: controller.signal }); + let promptSettled = false; + let promptError: unknown; + void prompt.then( + () => { + promptSettled = true; + }, + (error: unknown) => { + promptSettled = true; + promptError = error; + }, + ); + + controller.abort(); + await vi.waitFor(() => { + expect(promptSettled).toBe(true); + expect(promptError).toMatchObject({ + name: "PromptAdmissionCancelledError", + message: "Prompt admission was cancelled.", + }); + }); + + const nextFencePromise = internals._acquireSessionActionCommitFence(); + heldFence.release(); + const nextFence = await nextFencePromise; + nextFence.release(); + }); + + it("rejects a prompt waiting behind tree navigation when the session is disposed", async () => { + const treeHookReached = createDeferred(); + const treeHookGate = createDeferred(); + const harness = await createHarness({ + extensionFactories: [ + (pi) => { + pi.on("session_before_tree", async () => { + treeHookReached.resolve(); + await treeHookGate.promise; + }); + }, + ], + }); + harnesses.push(harness); + harness.setResponses([fauxAssistantMessage("one"), fauxAssistantMessage("two")]); + await harness.session.prompt("one"); + const target = harness.sessionManager.getEntries().find((entry) => entry.type === "message"); + expect(target).toBeDefined(); + await harness.session.prompt("two"); + + const navigation = harness.session.navigateTree(target!.id, { summarize: false }); + await treeHookReached.promise; + const prompt = harness.session.prompt("blocked prompt"); + + harness.session.dispose(); + try { + await expect(prompt).rejects.toThrow( + "Cannot admit a session action because the session is disposing or disposed.", + ); + } finally { + treeHookGate.resolve(); + } + await expect(navigation).resolves.toMatchObject({ cancelled: false }); + }); + + it("does not restart a session command cancelled while waiting for the commit fence", async () => { + const harness = await createHarness(); + harnesses.push(harness); + const internals = harness.session as unknown as CommitFenceInternals; + const schedule = vi.spyOn(internals, "_scheduleSessionInputPump").mockImplementation(() => {}); + const text = "/autonomous status"; + const completion = harness.session.promptAndWait(text); + await vi.waitFor(() => expect(internals._actionStore.unfinishedActions()).toHaveLength(1)); + const heldFence = await internals._acquireSessionActionCommitFence(); + + schedule.mockRestore(); + internals._scheduleSessionInputPump(); + await vi.waitFor(() => expect(internals._actionStore.unfinishedActions()[0]?.lifecycle.state).toBe("selected")); + const rejection = expect(completion).rejects.toThrow("cleared before delivery"); + expect(harness.session.clearQueue().followUp).toEqual([text]); + heldFence.release(); + + await rejection; + await harness.session.waitForSessionInputIdle(); + expect(internals._actionStore.unfinishedActions()).toEqual([]); + }); + + it("rechecks refinement before running a command parked on the commit fence", async () => { + const harness = await createHarness(); + harnesses.push(harness); + const internals = harness.session as unknown as CommitFenceInternals; + const schedule = vi.spyOn(internals, "_scheduleSessionInputPump").mockImplementation(() => {}); + const command = harness.session.promptAndWait("/autonomous status"); + await vi.waitFor(() => expect(internals._actionStore.unfinishedActions()).toHaveLength(1)); + const heldFence = await internals._acquireSessionActionCommitFence(); + schedule.mockRestore(); + internals._scheduleSessionInputPump(); + await vi.waitFor(() => expect(internals._actionStore.unfinishedActions()[0]?.lifecycle.state).toBe("selected")); + + const refineGate = createDeferred(); + const refineInFlight = refineGate.promise.finally(() => { + if (internals._refineInFlight === refineInFlight) internals._refineInFlight = undefined; + }); + internals._refineInFlight = refineInFlight; + heldFence.release(); + await yieldToEventLoop(); + + expect(internals._actionStore.unfinishedActions()[0]?.lifecycle.state).toBe("selected"); + expect(harness.session.messages).toEqual([]); + + refineGate.resolve(); + await command; + expect(internals._actionStore.unfinishedActions()).toEqual([]); + }); + + it("releases the direct-turn fence when suspension wins after acquisition", async () => { + const harness = await createHarness(); + harnesses.push(harness); + const internals = harness.session as unknown as CommitFenceInternals; + await harness.session.followUp("queued work"); + + const admission = internals._acquireDirectTurnAdmissionFence(); + harness.session.requestAbort(); + await expect(admission).rejects.toThrow("Cannot admit a session action while queued session input is suspended."); + + let nextFence: { release(): void } | undefined; + let nextError: unknown; + void internals._acquireSessionActionCommitFence().then( + (fence) => { + nextFence = fence; + }, + (error: unknown) => { + nextError = error; + }, + ); + await yieldToEventLoop(); + expect(nextError).toBeUndefined(); + expect(nextFence).toBeDefined(); + nextFence?.release(); + }); + + it("preserves pause-held arrival order between injected and ordinary prompts", async () => { + const harness = await createHarness(); + harnesses.push(harness); + harness.setResponses([fauxAssistantMessage("first done"), fauxAssistantMessage("second done")]); + const internals = harness.session as unknown as CommitFenceInternals; + const pause = harness.session.acquireQueuedWorkPause(); + const injectedMessage = { + role: "custom" as const, + customType: "injected-order-probe", + content: "earlier injected prompt", + display: true, + details: {}, + timestamp: Date.now(), + }; + + const injected = internals._promptInjectedMessage("earlier injected prompt", injectedMessage, { + returnAfterAccepted: true, + }); + const ordinary = harness.session.promptUntilAccepted("later ordinary prompt"); + await yieldToEventLoop(); + pause.release(); + await Promise.all([injected, ordinary]); + await harness.session.waitForIdle(); + + expect( + harness.session.messages + .map((message) => getMessageText(message)) + .filter((text) => text === "earlier injected prompt" || text === "later ordinary prompt"), + ).toEqual(["earlier injected prompt", "later ordinary prompt"]); + }); + + it("keeps each prefix with its primary when dispatching an all-mode batch", async () => { + const deliveredMessages: string[] = []; + const harness = await createHarness(); + harnesses.push(harness); + harness.setResponses([ + (context) => { + deliveredMessages.push(...context.messages.map(getMessageText)); + return fauxAssistantMessage("done"); + }, + ]); + harness.session.setFollowUpMode("all"); + await harness.session.restoreFollowUpMessage("primary A", undefined, { + prefixMessages: [createContextMessage("prefix A")], + }); + await harness.session.restoreFollowUpMessage("primary B", undefined, { + prefixMessages: [createContextMessage("prefix B")], + }); + await harness.session.sendCustomMessage( + { customType: "shared-next-turn", content: "shared next turn", display: false }, + { deliverAs: "nextTurn" }, + ); + + harness.session.resumeQueuedWork(); + await harness.session.waitForIdle(); + + expect(deliveredMessages).toEqual(["prefix A", "shared next turn", "primary A", "prefix B", "primary B"]); + }); + + it("cancels an ordinary prompt while queued work is paused", async () => { + const harness = await createHarness(); + harnesses.push(harness); + const pause = harness.session.acquireQueuedWorkPause(); + const controller = new AbortController(); + const prompt = harness.session.promptUntilAccepted("cancel during pause", { signal: controller.signal }); + await yieldToEventLoop(); + + controller.abort(); + await expect(prompt).rejects.toMatchObject({ + name: "PromptAdmissionCancelledError", + message: "Prompt admission was cancelled.", + }); + pause.release(); + expect(getUserTexts(harness)).toEqual([]); + }); + + it("admits a reentrant prompt from a navigation hook without waiting on its own pause", async () => { + let promptFromHook: (() => Promise) | undefined; + const harness = await createHarness({ + extensionFactories: [ + (pi) => { + pi.on("session_before_tree", async () => { + if (!promptFromHook) throw new Error("Expected prompt hook to be initialized"); + await promptFromHook(); + }); + }, + ], + }); + harnesses.push(harness); + promptFromHook = () => harness.session.promptUntilAccepted("prompt from navigation hook"); + harness.setResponses([ + fauxAssistantMessage("one done"), + fauxAssistantMessage("two done"), + fauxAssistantMessage("hook done"), + ]); + await harness.session.prompt("one"); + const target = harness.sessionManager.getEntries().find((entry) => entry.type === "message"); + expect(target).toBeDefined(); + await harness.session.prompt("two"); + + await harness.session.navigateTree(target!.id, { summarize: false }); + await harness.session.waitForIdle(); + expect(getUserTexts(harness)).toContain("prompt from navigation hook"); + }); +}); diff --git a/packages/coding-agent/test/suite/agent-session-compaction.test.ts b/packages/coding-agent/test/suite/agent-session-compaction.test.ts index dc7e86590..54431e37c 100644 --- a/packages/coding-agent/test/suite/agent-session-compaction.test.ts +++ b/packages/coding-agent/test/suite/agent-session-compaction.test.ts @@ -316,6 +316,57 @@ describe("AgentSession compaction characterization", () => { } }); + it("defers post-compaction refine behind a preparing session action", async () => { + const preparationReached = vi.fn(); + let releasePreparation = () => {}; + const preparationGate = new Promise((resolve) => { + releasePreparation = resolve; + }); + const harness = await createHarness({ + settings: { compaction: { keepRecentTokens: 1 } }, + extensionFactories: [ + (pi) => { + pi.on("before_agent_start", async (event) => { + if (event.prompt !== "preparing across compaction") return; + preparationReached(); + await preparationGate; + }); + pi.on("session_before_compact", async (event) => ({ + compaction: { + summary: "summary from extension", + firstKeptEntryId: event.preparation.firstKeptEntryId, + tokensBefore: event.preparation.tokensBefore, + details: { source: "extension" }, + }, + })); + }, + ], + }); + harnesses.push(harness); + harness.setResponses([ + fauxAssistantMessage("one response"), + fauxAssistantMessage("two response"), + fauxAssistantMessage("prepared response"), + ]); + await harness.session.prompt("one"); + await harness.session.prompt("two"); + await harness.session.followUp("preparing across compaction", undefined, { resumeIfIdle: true }); + await vi.waitFor(() => expect(preparationReached).toHaveBeenCalledOnce()); + const internals = harness.session as unknown as SessionWithCompactionInternals & { + _cancelPostCompactionContinue(): void; + _scheduleAutoRefineAfterCompaction(willContinueAfterCompaction: boolean): void; + }; + const scheduleAutoRefineSpy = vi.spyOn(internals, "_scheduleAutoRefineAfterCompaction"); + try { + await internals._runAutoCompaction("requested", false); + expect(scheduleAutoRefineSpy).toHaveBeenCalledWith(true); + } finally { + releasePreparation(); + internals._cancelPostCompactionContinue(); + } + await harness.session.waitForIdle(); + }); + it("throws when compacting without a model", async () => { const harness = await createHarness(); harnesses.push(harness); diff --git a/packages/coding-agent/test/suite/agent-session-prompt.test.ts b/packages/coding-agent/test/suite/agent-session-prompt.test.ts index bbb26ef6b..0d173d494 100644 --- a/packages/coding-agent/test/suite/agent-session-prompt.test.ts +++ b/packages/coding-agent/test/suite/agent-session-prompt.test.ts @@ -47,47 +47,6 @@ describe("AgentSession prompt characterization", () => { } }); - it("cancels a prompt while direct-turn admission is paused", async () => { - const harness = await createHarness(); - harnesses.push(harness); - const pause = harness.session.acquireQueuedWorkPause(); - const controller = new AbortController(); - - const prompt = harness.session.prompt("cancel me", { signal: controller.signal }); - controller.abort(); - - await expect(prompt).rejects.toThrow("Prompt admission was cancelled."); - expect(getUserTexts(harness)).toEqual([]); - pause.release(); - }); - - it("releases action admission when cancellation lands immediately after acquisition", async () => { - const harness = await createHarness(); - harnesses.push(harness); - const controller = new AbortController(); - const internals = harness.session as unknown as { - _acquireSessionActionAdmission(options?: { - signal?: AbortSignal; - }): Promise<{ owner: object; release(): void }>; - }; - const acquire = internals._acquireSessionActionAdmission.bind(internals); - internals._acquireSessionActionAdmission = async (options) => { - const admission = await acquire(options); - controller.abort(); - return admission; - }; - - await expect(harness.session.prompt("cancel after acquire", { signal: controller.signal })).rejects.toThrow( - "Prompt admission was cancelled.", - ); - expect(getUserTexts(harness)).toEqual([]); - - internals._acquireSessionActionAdmission = acquire; - harness.setResponses([fauxAssistantMessage("recovered")]); - await harness.session.prompt("second"); - expect(getUserTexts(harness)).toEqual(["second"]); - }); - it("releases action admission when ownership commit throws", async () => { const harness = await createHarness(); harnesses.push(harness); @@ -181,6 +140,38 @@ describe("AgentSession prompt characterization", () => { await Promise.all([first, second]); }); + it("preserves idle prompt order while the first input handler is pending", async () => { + const firstInputReached = createDeferred(); + const firstInputGate = createDeferred(); + const inputOrder: string[] = []; + const harness = await createHarness({ + extensionFactories: [ + (pi) => { + pi.on("input", async (event) => { + inputOrder.push(event.text); + if (event.text === "first") { + firstInputReached.resolve(); + await firstInputGate.promise; + } + }); + }, + ], + }); + harnesses.push(harness); + harness.setResponses([fauxAssistantMessage("first done"), fauxAssistantMessage("second done")]); + + const first = harness.session.prompt("first"); + await firstInputReached.promise; + const second = harness.session.prompt("second"); + await new Promise((resolve) => setImmediate(resolve)); + expect(inputOrder).toEqual(["first"]); + + firstInputGate.resolve(); + await Promise.all([first, second]); + expect(inputOrder).toEqual(["first", "second"]); + expect(getUserTexts(harness)).toEqual(["first", "second"]); + }); + it("admits reentrant hook prompts without deadlocking the active action", async () => { let submitted = false; const harness = await createHarness({ @@ -981,36 +972,38 @@ stale post-hook extension instructions`, expect(harness.session.queuedActionCount).toBe(0); }); - it("queues accepted agent messages if the session becomes busy before handoff", async () => { + it("waits to accept agent messages while queued work is paused without leaking a waiter", async () => { const harness = await createHarness(); harnesses.push(harness); const agentPrompt = "Agent-to-agent message received.\nSource: agent_message\nTo: Target, active target, session session-target\nMessage id: agentmsg_handoff_busy\n\nqueue at handoff"; - let releaseRefine: (() => void) | undefined; - const refineGate = new Promise((resolve) => { - releaseRefine = resolve; - }); const sessionInternals = harness.session as unknown as { - _refineInFlight?: Promise; - _userBashRunning?: boolean; + _sessionInputCheckpointWaiters: Set<() => void>; }; - sessionInternals._refineInFlight = refineGate; + const pause = harness.session.acquireQueuedWorkPause(); + let acceptedSettled = false; + const accepted = harness.session + .acceptAgentMessagePrompt(agentPrompt, { + expandPromptTemplates: false, + streamingBehavior: "followUp", + queueIfBusy: true, + }) + .then(() => { + acceptedSettled = true; + }); + await new Promise((resolve) => setImmediate(resolve)); - const accepted = harness.session.acceptAgentMessagePrompt(agentPrompt, { - expandPromptTemplates: false, - streamingBehavior: "followUp", - queueIfBusy: true, - }); - await Promise.resolve(); - sessionInternals._userBashRunning = true; - sessionInternals._refineInFlight = undefined; - releaseRefine?.(); + expect(harness.session.getFollowUpMessages()).toEqual([]); + expect(acceptedSettled).toBe(false); + expect(sessionInternals._sessionInputCheckpointWaiters.size).toBe(1); + harness.setResponses([fauxAssistantMessage("delivered")]); + pause.release(); await accepted; - sessionInternals._userBashRunning = false; + await harness.session.waitForIdle(); - expect(harness.session.getFollowUpMessages()).toEqual([agentPrompt]); - expect(harness.getPendingResponseCount()).toBe(0); + expect(getUserTexts(harness)).toEqual([agentPrompt]); + expect(sessionInternals._sessionInputCheckpointWaiters.size).toBe(0); }); it("restores nextTurn context when handoff busy rejection cannot queue", async () => { @@ -1182,6 +1175,48 @@ stale post-hook extension instructions`, expect(harness.session.hasAcceptedPromptInFlight).toBe(false); }); + it("waitForIdle waits for cancelled dispatch cleanup in the session event queue", async () => { + const eventQueueReached = createDeferred(); + const eventQueueGate = createDeferred(); + const harness = await createHarness({ + extensionFactories: [ + (pi) => { + pi.on("agent_start", async () => { + eventQueueReached.resolve(); + await eventQueueGate.promise; + }); + }, + ], + }); + harnesses.push(harness); + harness.setResponses([fauxAssistantMessage("never delivered")]); + const dispatchGate = gateNextAgentStart(harness); + const prompt = + "Agent-to-agent message received.\nSource: agent_message\nTo: Target, active target, session session-target\nMessage id: agentmsg_idle_cleanup\n\ncancel me"; + const accepted = harness.session.acceptAgentMessagePrompt(prompt, { expandPromptTemplates: false }); + const rejected = expect(accepted).rejects.toThrow("cleared before delivery"); + await Promise.all([eventQueueReached.promise, dispatchGate.reached]); + + harness.session.clearQueuedUserMessagesMatching((text) => text === prompt); + let idle = false; + const waiting = harness.session.waitForIdle().then(() => { + idle = true; + }); + dispatchGate.release(); + await rejected; + await harness.session.agent.waitForIdle(); + await new Promise((resolve) => setImmediate(resolve)); + + const store = (harness.session as unknown as { _actionStore: { ownedActions(): readonly unknown[] } }) + ._actionStore; + expect(idle).toBe(false); + expect(store.ownedActions()).toHaveLength(1); + eventQueueGate.resolve(); + await waiting; + expect(store.ownedActions()).toHaveLength(0); + expect(harness.session.agent.state.errorMessage).toBeUndefined(); + }); + it("restores drained nextTurn messages when direct agent message acceptance fails before delivery", async () => { const harness = await createHarness(); harnesses.push(harness); @@ -1811,7 +1846,6 @@ stale post-hook extension instructions`, }); const flushNow = vi.spyOn(harness.sessionManager, "flushNow"); const controller = new AbortController(); - const removeEventListener = vi.spyOn(controller.signal, "removeEventListener"); const checkpoint = harness.session.waitForSessionInputCheckpoint(controller.signal); controller.abort(); @@ -1819,7 +1853,6 @@ stale post-hook extension instructions`, await expect(checkpoint).rejects.toThrow("Update restart preparation cancelled"); expect(queueDrained).toBe(false); expect(flushNow).not.toHaveBeenCalled(); - expect(removeEventListener).toHaveBeenCalledWith("abort", expect.any(Function)); extensionGate.resolve(); await prompt; await eventQueue; diff --git a/packages/coding-agent/test/suite/agent-session-queue.test.ts b/packages/coding-agent/test/suite/agent-session-queue.test.ts index 927246347..a4371f64b 100644 --- a/packages/coding-agent/test/suite/agent-session-queue.test.ts +++ b/packages/coding-agent/test/suite/agent-session-queue.test.ts @@ -11,6 +11,7 @@ import { createAgentSessionMessage, createAgentSessionMessagePrompt, } from "../../src/core/agent-messages.js"; +import { type AgentCronJob, shouldDeferHeartbeatCronJob } from "../../src/core/cron-jobs.js"; import { createSessionSlashCommandMessage } from "../../src/core/messages.js"; import { applyRefinementProposal, @@ -80,6 +81,24 @@ function agentPromptText(id: string, body: string): string { return `Agent-to-agent message received.\nSource: agent_message\nTo: Target, active target, session session-target\nMessage id: ${id}\n\n${body}`; } +function heartbeatJob(): AgentCronJob { + return { + id: "heartbeat-test", + status: "active", + source: "heartbeat", + activeSessionId: "active-test", + sessionId: "session-test", + sessionFile: "/tmp/session.jsonl", + cwd: "/tmp", + prompt: "check progress", + schedule: { kind: "interval", expression: "every 5m", intervalMs: 300_000 }, + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + nextRunAt: "2026-01-01T00:05:00.000Z", + runCount: 0, + }; +} + const skipReviewer = vi.fn(async () => ({ shouldRefine: true, rationale: "durable lesson" })); describe("AgentSession queue characterization", () => { @@ -627,6 +646,39 @@ describe("AgentSession queue characterization", () => { expect(navigated).toBe(true); }); + it("cancels a restart checkpoint while tree navigation owns the commit fence", async () => { + const treeEventReached = createDeferred(); + const treeEventGate = createDeferred(); + const harness = await createHarness({ + extensionFactories: [ + (pi) => { + pi.on("session_before_tree", async () => { + treeEventReached.resolve(); + await treeEventGate.promise; + }); + }, + ], + }); + harnesses.push(harness); + harness.setResponses([fauxAssistantMessage("one"), fauxAssistantMessage("two")]); + await harness.session.prompt("one"); + const target = harness.sessionManager.getEntries().find((entry) => entry.type === "message"); + expect(target).toBeDefined(); + await harness.session.prompt("two"); + + const navigation = harness.session.navigateTree(target!.id, { summarize: false }); + await treeEventReached.promise; + const controller = new AbortController(); + const checkpoint = harness.session.waitForSessionInputCheckpoint(controller.signal); + controller.abort(); + + await expect(checkpoint).rejects.toThrow("Update restart preparation cancelled"); + const nextCheckpoint = harness.session.waitForSessionInputCheckpoint(); + treeEventGate.resolve(); + await navigation; + await expect(nextCheckpoint).resolves.toBeUndefined(); + }); + it("keeps queued work paused until tree navigation events settle", async () => { const treeEvent = createDeferred(); const harness = await createHarness({ @@ -1254,6 +1306,24 @@ describe("AgentSession queue characterization", () => { expect(harness.session.messages).toEqual([]); }); + it("settles a visibly queued session command while an earlier action is preparing", async () => { + const hook = gatedHook({ prompt: "preparing turn" }); + const harness = await createHarness({ extensionFactories: [hook.factory] }); + harnesses.push(harness); + harness.setResponses([fauxAssistantMessage("done")]); + const pause = harness.session.acquireQueuedWorkPause(); + await harness.session.followUp("preparing turn", undefined, { resumeIfIdle: true }); + pause.release(); + await hook.reached; + + const command = harness.session.prompt("/goal status"); + await vi.waitFor(() => expect(harness.session.getFollowUpMessages()).toContain("/goal status")); + await expect(command).resolves.toBeUndefined(); + + hook.release(); + await harness.session.waitForIdle(); + }); + it("removes goal context while its steering handoff is still preparing", async () => { const hook = gatedHook({ prompt: "stale goal context" }); const harness = await createHarness({ extensionFactories: [hook.factory] }); @@ -1441,6 +1511,51 @@ describe("AgentSession queue characterization", () => { await harness.session.waitForSessionInputIdle(); }); + it("revalidates turn batches after acquiring the commit fence", async () => { + const harness = await createHarness(); + harnesses.push(harness); + harness.session.setFollowUpMode("all"); + harness.setResponses([fauxAssistantMessage("survivor delivered")]); + const pause = harness.session.acquireQueuedWorkPause(); + withStreaming(harness, true); + const removed = harness.session.promptAndWait("remove after preparation", { + streamingBehavior: "followUp", + followUpQueueKey: "remove", + }); + const removedOutcome = removed.then( + () => ({ error: undefined }), + (error: unknown) => ({ error }), + ); + const survivor = harness.session.promptAndWait("survive re-selection", { + streamingBehavior: "followUp", + followUpQueueKey: "survivor", + }); + const survivorOutcome = survivor.then( + () => ({ error: undefined }), + (error: unknown) => ({ error }), + ); + withStreaming(harness, false); + const internals = harness.session as unknown as { + _acquireSessionActionCommitFence(): Promise<{ release(): void }>; + }; + const fence = await internals._acquireSessionActionCommitFence(); + + pause.release(); + try { + await vi.waitFor(() => expect(harness.session.getSessionActionSnapshot().active?.phase).toBe("preparing")); + expect(harness.session.removeQueuedFollowUp("remove")).toBe(true); + } finally { + fence.release(); + } + await harness.session.waitForSessionInputIdle(); + expect(harness.session.getFollowUpMessages()).toEqual([]); + expect((await removedOutcome).error).toEqual( + expect.objectContaining({ message: "Queued agent message was cleared before delivery." }), + ); + expect((await survivorOutcome).error).toBeUndefined(); + expect(getUserTexts(harness)).toEqual(["survive re-selection"]); + }); + it("stops counting cleared preparing inputs as pending", async () => { const hook = gatedHook({ prompt: "cleared prompt" }); const harness = await createHarness({ extensionFactories: [hook.factory] }); @@ -1461,6 +1576,24 @@ describe("AgentSession queue characterization", () => { expect(getUserTexts(harness)).toEqual([]); }); + it("coalesces a same-key follow-up while a steering owner is preparing", async () => { + const hook = gatedHook({ prompt: "steering heartbeat" }); + const harness = await createHarness({ extensionFactories: [hook.factory] }); + harnesses.push(harness); + harness.setResponses([fauxAssistantMessage("done")]); + const pause = harness.session.acquireQueuedWorkPause(); + await harness.session.steer("steering heartbeat", undefined, { queueKey: "heartbeat", resumeIfIdle: true }); + pause.release(); + await hook.reached; + + expect(await harness.session.followUp("duplicate heartbeat", undefined, { queueKey: "heartbeat" })).toBe(false); + expect(harness.session.getFollowUpMessages()).toEqual([]); + + hook.release(); + await harness.session.waitForIdle(); + expect(getUserTexts(harness)).toEqual(["steering heartbeat"]); + }); + it("queues a same-key follow-up once the prior owner has handed off", async () => { const harness = await createHarness(); harnesses.push(harness); @@ -1603,6 +1736,31 @@ describe("AgentSession queue characterization", () => { expect(getUserTexts(harness)).toEqual(["kept"]); }); + it("restores next-turn context from cancelled actions in action order", async () => { + const harness = await createHarness(); + harnesses.push(harness); + const firstPrompt = agentPromptText("agentmsg_restore_first", "first"); + const secondPrompt = agentPromptText("agentmsg_restore_second", "second"); + withStreaming(harness, true); + await harness.session.sendCustomMessage( + { customType: "next-turn", content: "context A", display: true, details: {} }, + { deliverAs: "nextTurn" }, + ); + await harness.session.queueAgentMessagePrompt(firstPrompt, "followUp"); + await harness.session.sendCustomMessage( + { customType: "next-turn", content: "context B", display: true, details: {} }, + { deliverAs: "nextTurn" }, + ); + await harness.session.queueAgentMessagePrompt(secondPrompt, "followUp"); + + expect(harness.session.clearQueue().followUp).toEqual([firstPrompt, secondPrompt]); + expect(harness.session.getPendingNextTurnMessageSnapshots().map(getMessageText)).toEqual([ + "context A", + "context B", + ]); + withStreaming(harness, false); + }); + it("delivers next-turn context when the first preparing turn is cancelled", async () => { const firstPrompt = agentPromptText("agentmsg_cancel_first", "cancelled"); let cancelFirst: (() => void) | undefined; @@ -2095,33 +2253,7 @@ describe("AgentSession queue characterization", () => { expect(getAssistantTexts(harness)).toEqual(["custom done", "follow-up done"]); }); - it("rechecks queued-work pauses acquired at the action-admission boundary", async () => { - const harness = await createHarness(); - harnesses.push(harness); - harness.setResponses([fauxAssistantMessage("direct done")]); - const internals = harness.session as unknown as { - _acquireTurnAdmission(): Promise<{ owner: symbol; release(): void }>; - }; - const acquireTurnAdmission = internals._acquireTurnAdmission.bind(internals); - let pause: { release(): void } | undefined; - let first = true; - internals._acquireTurnAdmission = async () => { - const admission = await acquireTurnAdmission(); - if (first) { - first = false; - pause = harness.session.acquireQueuedWorkPause(); - } - return admission; - }; - const prompt = harness.session.prompt("direct"); - await new Promise((resolve) => setImmediate(resolve)); - expect(getUserTexts(harness)).toEqual([]); - pause?.release(); - await prompt; - expect(getUserTexts(harness)).toEqual(["direct"]); - }); - - it("rejects disposal while action admission waits behind a pause", async () => { + it("settles an action disposed while its queued-work pause is held", async () => { const harness = await createHarness(); harnesses.push(harness); const pause = harness.session.acquireQueuedWorkPause(); @@ -2333,6 +2465,117 @@ describe("AgentSession queue characterization", () => { expect(harness.sessionManager.getLeafId()).not.toBe(secondId); }); + it("defers steer heartbeats while a non-streaming session command is running", async () => { + const commandStarted = createDeferred(); + const commandGate = createDeferred(); + const harness = await createHarness(); + harnesses.push(harness); + vi.spyOn(harness.session, "refine").mockImplementation(async () => { + commandStarted.resolve(); + await commandGate.promise; + return emptyRefinementResult(); + }); + + const command = harness.session.prompt("/refine --local"); + await commandStarted.promise; + expect(harness.session.isStreaming).toBe(false); + expect(harness.session.unfinishedActionCount).toBe(1); + expect(harness.session.hasPendingSessionWork).toBe(false); + expect(shouldDeferHeartbeatCronJob(heartbeatJob(), harness.session)).toBe(true); + + commandGate.resolve(); + await command; + }); + + it("keeps a slow session command on one branch while concurrent navigation waits", async () => { + const commandStarted = createDeferred(); + const commandGate = createDeferred(); + const harness = await createHarness(); + harnesses.push(harness); + harness.setResponses([fauxAssistantMessage("one"), fauxAssistantMessage("two")]); + await harness.session.prompt("one"); + const target = harness.sessionManager.getEntries().find((entry) => entry.type === "message"); + expect(target).toBeDefined(); + await harness.session.prompt("two"); + vi.spyOn(harness.session, "refine").mockImplementation(async () => { + commandStarted.resolve(); + await commandGate.promise; + return emptyRefinementResult(); + }); + + const command = harness.session.prompt("/refine --local"); + await commandStarted.promise; + let navigationSettled = false; + const navigation = harness.session.navigateTree(target!.id, { summarize: false }).then(() => { + navigationSettled = true; + }); + await new Promise((resolve) => setImmediate(resolve)); + expect(navigationSettled).toBe(false); + + commandGate.resolve(); + await command; + await navigation; + const entries = harness.sessionManager.getEntries(); + const inputEntry = entries.find( + (entry) => + entry.type === "custom_message" && + entry.customType === "session_slash_command" && + entry.content === "/refine --local", + ); + const resultEntry = entries.find( + (entry) => entry.type === "custom_message" && entry.customType === "session_slash_command_result", + ); + expect(inputEntry).toBeDefined(); + expect(resultEntry).toBeDefined(); + expect(harness.sessionManager.getBranch(resultEntry!.id).map((entry) => entry.id)).toContain(inputEntry!.id); + }); + + it("allows a /compact extension hook to navigate without deadlocking on the commit fence", async () => { + let targetId: string | undefined; + let navigateFromContext: ((target: string) => Promise<{ cancelled: boolean }>) | undefined; + let navigated = false; + const harness = await createHarness({ + settings: { compaction: { keepRecentTokens: 1 } }, + extensionFactories: [ + (pi) => { + pi.registerCommand("capture-navigation", { + description: "Capture command context", + handler: async (_args, ctx) => { + navigateFromContext = (target) => ctx.navigateTree(target, { summarize: false }); + }, + }); + pi.on("session_before_compact", async () => { + if (!navigateFromContext) throw new Error("Expected captured extension command context"); + const result = await navigateFromContext(targetId!); + navigated = !result.cancelled; + return { cancel: true }; + }); + }, + ], + }); + harnesses.push(harness); + await harness.session.bindExtensions({ + commandContextActions: { + waitForIdle: () => harness.session.waitForIdle(), + newSession: async () => ({ cancelled: false }), + fork: async () => ({ cancelled: false }), + navigateTree: async (target, options) => harness.session.navigateTree(target, options), + switchSession: async () => ({ cancelled: false }), + reload: async () => {}, + }, + }); + harness.setResponses([fauxAssistantMessage("one"), fauxAssistantMessage("two")]); + await harness.session.prompt("one"); + targetId = harness.sessionManager.getEntries().find((entry) => entry.type === "message")?.id; + expect(targetId).toBeDefined(); + await harness.session.prompt("two"); + await harness.session.prompt("/capture-navigation"); + + await harness.session.prompt("/compact"); + + expect(navigated).toBe(true); + }); + it("waitForIdle yields to timers while queued work is paused", async () => { const harness = await createHarness(); harnesses.push(harness); @@ -2389,6 +2632,19 @@ describe("AgentSession queue characterization", () => { expect(getUserTexts(harness)).toEqual(["queued across abort"]); }); + it("admits a resumeIfIdle follow-up after abort suspends queued work", async () => { + const harness = await createHarness(); + harnesses.push(harness); + harness.setResponses([fauxAssistantMessage("queued done"), fauxAssistantMessage("wake done")]); + await harness.session.followUp("queued before abort"); + await harness.session.abort(); + + await expect(harness.session.followUp("wake after abort", undefined, { resumeIfIdle: true })).resolves.toBe(true); + await harness.session.waitForIdle(); + + expect(getUserTexts(harness)).toEqual(["queued before abort", "wake after abort"]); + }); + it("waitForIdle resolves when the queue is cleared while the pump is suspended", async () => { const harness = await createHarness(); harnesses.push(harness); @@ -2427,6 +2683,35 @@ describe("AgentSession queue characterization", () => { withStreaming(harness, false); }); + it("waitForIdle observes event-queue work chained while settling", async () => { + const harness = await createHarness(); + harnesses.push(harness); + const initialEvent = createDeferred(); + const chainedOperation = createDeferred(); + const internals = harness.session as unknown as { _agentEventQueue: Promise }; + let eventQueue: Promise; + eventQueue = initialEvent.promise.then(() => { + internals._agentEventQueue = eventQueue.then(() => chainedOperation.promise); + }); + internals._agentEventQueue = eventQueue; + let idle = false; + const waiting = harness.session.waitForIdle().then(() => { + idle = true; + }); + await new Promise((resolve) => setImmediate(resolve)); + + initialEvent.resolve(); + await eventQueue; + await new Promise((resolve) => setImmediate(resolve)); + try { + expect(idle).toBe(false); + } finally { + chainedOperation.resolve(); + } + await waiting; + expect(idle).toBe(true); + }); + it("waitForIdle observes a run that starts at its final idle boundary", async () => { const responseGate = createDeferred(); const waitForResponseStart = createDeferred(); diff --git a/packages/coding-agent/test/suite/regressions/4482-heartbeat-injected-prompt.test.ts b/packages/coding-agent/test/suite/regressions/4482-heartbeat-injected-prompt.test.ts index b4fe6b75c..394f860d7 100644 --- a/packages/coding-agent/test/suite/regressions/4482-heartbeat-injected-prompt.test.ts +++ b/packages/coding-agent/test/suite/regressions/4482-heartbeat-injected-prompt.test.ts @@ -3,7 +3,7 @@ import { fauxAssistantMessage, fauxToolCall, type Message } from "@earendil-work import { Container, type MarkdownTheme } from "@earendil-works/pi-tui"; import { Type } from "typebox"; import { afterEach, beforeAll, describe, expect, it, vi } from "vitest"; -import type { AgentCronJob } from "../../../src/core/cron-jobs.js"; +import { type AgentCronJob, shouldDeferHeartbeatCronJob } from "../../../src/core/cron-jobs.js"; import { createGoalContextMessage, type GoalState } from "../../../src/core/goals.js"; import { createHeartbeatPromptMessage, HEARTBEAT_PROMPT_CUSTOM_TYPE } from "../../../src/core/messages.js"; import { @@ -263,6 +263,133 @@ describe("ENG-4482 heartbeat injected prompt UI", () => { ).toBe(true); }); + it("orders a heartbeat after an earlier prompt with a slow input handler", async () => { + let markInputReached = () => {}; + const inputReached = new Promise((resolve) => { + markInputReached = resolve; + }); + let releaseInput = () => {}; + const inputGate = new Promise((resolve) => { + releaseInput = resolve; + }); + const harness = await createHarness({ + extensionFactories: [ + (pi) => { + pi.on("input", async (event) => { + if (event.text !== "ordinary first") return; + markInputReached(); + await inputGate; + }); + }, + ], + }); + harnesses.push(harness); + const providerOrder: string[] = []; + harness.setResponses([ + (context) => { + providerOrder.push(getMessageText(context.messages.at(-1))); + return fauxAssistantMessage("first done"); + }, + (context) => { + providerOrder.push(getMessageText(context.messages.at(-1))); + return fauxAssistantMessage("heartbeat done"); + }, + ]); + + const ordinary = harness.session.prompt("ordinary first"); + await inputReached; + const heartbeat = harness.session.promptHeartbeat(createHeartbeat()); + releaseInput(); + await Promise.all([ordinary, heartbeat]); + await harness.session.waitForIdle(); + + expect(providerOrder).toEqual(["ordinary first", createHeartbeat().prompt]); + }); + + it("waits for a queued-work pause before admitting a streaming heartbeat", async () => { + let markStarted = () => {}; + const started = new Promise((resolve) => { + markStarted = resolve; + }); + let releaseTurn = () => {}; + const turnGate = new Promise((resolve) => { + releaseTurn = resolve; + }); + const harness = await createHarness(); + harnesses.push(harness); + harness.setResponses([ + async () => { + markStarted(); + await turnGate; + return fauxAssistantMessage("original done"); + }, + fauxAssistantMessage("heartbeat done"), + ]); + + const originalTurn = harness.session.prompt("start"); + await started; + const pause = harness.session.acquireQueuedWorkPause(); + const heartbeat = harness.session.promptHeartbeat(createHeartbeat(), { streamingBehavior: "followUp" }); + await Promise.resolve(); + expect(harness.session.getSessionActionRecoverySnapshot().actions).toHaveLength(0); + + pause.release(); + await heartbeat; + expect(harness.session.getSessionActionRecoverySnapshot().actions).toHaveLength(1); + releaseTurn(); + await originalTurn; + await harness.session.waitForIdle(); + expect(harness.session.getSessionActionRecoverySnapshot().actions).toHaveLength(0); + }); + + it("preserves next-turn context order across queued heartbeat admission", async () => { + let markStarted = () => {}; + const started = new Promise((resolve) => { + markStarted = resolve; + }); + let releaseTurn = () => {}; + const turnGate = new Promise((resolve) => { + releaseTurn = resolve; + }); + const harness = await createHarness(); + harnesses.push(harness); + let deliveredOrder: string[] = []; + harness.setResponses([ + async () => { + markStarted(); + await turnGate; + return fauxAssistantMessage("original turn complete"); + }, + (context) => { + deliveredOrder = context.messages + .map(getMessageText) + .filter((text) => ["context A", "context B", createHeartbeat().prompt].includes(text)); + return fauxAssistantMessage("heartbeat handled"); + }, + ]); + + const originalTurn = harness.session.prompt("start"); + await started; + expect(harness.session.isStreaming).toBe(true); + expect(harness.session.unfinishedActionCount).toBe(1); + expect(harness.session.hasPendingSessionWork).toBe(false); + expect(shouldDeferHeartbeatCronJob(createHeartbeat(), harness.session)).toBe(false); + await harness.session.sendCustomMessage( + { customType: "next-turn", content: "context A", display: true, details: {} }, + { deliverAs: "nextTurn" }, + ); + await harness.session.promptHeartbeat(createHeartbeat(), { streamingBehavior: "followUp" }); + await harness.session.sendCustomMessage( + { customType: "next-turn", content: "context B", display: true, details: {} }, + { deliverAs: "nextTurn" }, + ); + releaseTurn(); + await originalTurn; + await harness.session.waitForIdle(); + + expect(deliveredOrder).toEqual(["context A", "context B", createHeartbeat().prompt]); + }); + it("returns raw text when clearing queued heartbeat prompts while previews remain labeled", async () => { let releaseToolExecution: (() => void) | undefined; const toolRelease = new Promise((resolve) => {