diff --git a/README.md b/README.md index 371607f58..d0384c066 100644 --- a/README.md +++ b/README.md @@ -432,6 +432,8 @@ The non-streaming path uses `AbortSignal` forwarded to `membrane.stream()`, whic A turn whose stream is cancelled this way, or by `agent.cancelStream()`, ends as a stop, whether the stream reports the cancel as `aborted`, as an `error`, or by just closing: one `inference:aborted` carrying the caller's reason (`user` when none was given), a `[turn-interrupted]` note in the agent's context, and never `inference:exhausted`, which is for failed inferences. A stream that membrane aborts on its own, at a deadline or on an error (reason `timeout` or `error`), is a failed inference. `stop()` and an operator's quiesce abandon end a turn as a stop too, with their own reasons, `shutdown` and `quiesce_abandoned`, and write no note. A stopped stream still writes its inference-log row, with `success: false`. +A caller's stop and an operator's quiesce abandon also reach a turn that has no live stream: one running its modules' `gatherContext`, its MCPL hooks or its compile, one waiting to retry a failed inference or compile (`abortInference()` and an operator's abandon end the wait, and a stop through `Agent.cancelStream()` waits for it to run out), and one whose tool-result guard is waiting to retry after a refusal, which holds only the round's finished stream. The turn ends at its next check, before it sends anything more, as a stop on a live stream ends it, and a turn whose compile fails after the stop ends at the failure, before a provider hold, the error policy's retry or `inference:exhausted`. A caller's stop leaves the note saying the turn was stopped before it sent anything or before it finished, and an operator's abandon leaves none. `abortInference()` returns `true` for such a turn. In the guard's wait, `stop()` ends the turn as a stop with reason `shutdown`, and an operator's abandon ends it as the abandon, even when a caller's stop came first; with both, it ends as `stop()`'s. `stop()` drops a turn in its hooks, its compile or a wait to retry before it sends anything, as before, and drops one whose compile fails while it runs, once a caller's stop has reached it, rather than ending it as that stop. While the failure of a turn's stream is reported (`inference:failed`), the agent no longer holds that stream, so `abortInference()` from a listener reports `false`. A failure to build a turn's request is reported while the turn's check is still ahead, so there `abortInference()` reports `true`, and the turn ends at that check. A stop made where no check of a turn is ahead stops nothing, and doesn't end the agent's next turn: while the agent is idle, while a turn finishes its last delivery, while a budget restart or a non-streaming turn's tool results wait to start the turn's next call, and while a provider-capacity hold or a provider's stated wait has parked a turn until the hold ends. + ## API Server WebSocket server for external clients (UIs, scripts, other agents). diff --git a/changelog.d/stop-without-stream.fixed.md b/changelog.d/stop-without-stream.fixed.md new file mode 100644 index 000000000..45367a068 --- /dev/null +++ b/changelog.d/stop-without-stream.fixed.md @@ -0,0 +1,68 @@ +- **A stop reaches a turn that has no live stream.** A turn can be alive + with no stream to carry a stop: while it runs its modules' + `gatherContext`, its MCPL hooks or its compile, while it waits to retry a + failed inference or a failed compile, and while the tool-result guard + waits for its speech to go out before retrying a refused round, when it + holds only the round's finished stream. In the first two, + `Agent.cancelStream()` had nothing to cancel, the framework's + `abortInference()` returned `false`, and an operator's quiesce abandon + left the turn running. In the guard's wait, the round's finished stream + was cancelled and `abortInference()` returned `true`, but the retry was + sent anyway. The turn went on to send its request or its retry, and + answered, and an abandoned one did so while the host was quiesced. Now + it ends at its next check, before it sends anything more. + - **What a host sees:** one `inference:aborted` with the stop's reason, + instead of a turn that goes on. The agent's `[turn-interrupted]` note + says the turn was stopped before it sent anything, or before it + finished; an operator's abandon writes no note. The inference log gets + the stop's row, as for a stop on a live stream: + `Stream cancelled ()`, or `Stream aborted: quiesce_abandoned` + for an operator's abandon. The framework's `abortInference()` returns + `true` for such a turn. `abortInference()` and an operator's abandon end + a turn's wait to retry, however long, so the turn ends at once; a stop + through `Agent.cancelStream()` ends it when the wait runs out. An MCPL + server granted `inferenceLifecycle` sees a failed inference whose retry + such a stop cancels end `failed`, as when `stop()` cancels it. In the + tool-result guard's wait, `stop()` and an operator's abandon end the + turn as their own stop even when a caller's stop came first, and as + `stop()`'s when both reach it. `host:quiesce`'s `unabandonable` lists a + turn that the abandon hasn't ended when its grace runs out: one with no + stream and no check ahead, such as a puppet tool turn, or one that + hasn't reached its check yet, the tool-result guard's wait for its + speech included. A turn whose stream the abandon cancelled while it was + still streaming isn't listed, even one still running. + - **A failure's report:** a turn's stream that failed or threw no longer + holds the agent while its `inference:failed` is delivered, nor does a + compiled stream held for a provider's wait. A listener sees the agent + idle, and `abortInference()` from a listener reports `false`: no stream + or check is there to take the stop. A stop through + `Agent.cancelStream()` made then ends the turn at the check after its + wait to retry, if the error policy retries it, and otherwise ends + nothing. A failure to build a turn's request is reported while the + turn's check is still ahead: `abortInference()` from a listener reports + `true` there, and the turn ends at that check. + - **A stop followed by a failure:** when a turn's compile fails after a + stop reached it, the turn ends at the failure: its `inference:failed`, + then its `inference:aborted`. It isn't held by a provider hold, and the + error policy doesn't retry it. If by then the agent was replaced, or the + framework is stopping or was stopped and started again, the turn is + dropped before it sends anything, as at the check after the compile. A + failure that would have exhausted the turn emits no + `inference:exhausted`, the event from which the hard-down alert's + failure streak, the over-budget drain and the poison-history breaker + count failures, so none of them counts it. + - **A stop made where no check of a turn is ahead ends nothing:** the + agent's next turn runs, as after a host's interrupt of every agent. + That covers an idle agent, a turn finishing its last delivery, a budget + restart or a non-streaming turn's tool results waiting to start the + turn's next call, and a turn a provider-capacity hold or a provider's + stated wait has parked. +- **A start whose framework was stopped and started again since it began is + dropped before it retries a failed compile.** Its `inference:aborted` + reads `a retry waiting while the framework was stopped and started again + is dropped before it sends anything`; before, the retry compiled and was + sent in the new lifecycle. A start whose hooks ran across a `stop()` and a + `start()` is dropped before its compile too, as `a turn assembled while + the framework was stopped and started again …`, where before the check + after its compile dropped it. Either way a stop kept for the turn goes + with it, as for a turn dropped while `stop()` runs. diff --git a/src/agent.ts b/src/agent.ts index c3dc0931c..5983f64e3 100644 --- a/src/agent.ts +++ b/src/agent.ts @@ -1613,21 +1613,42 @@ export class Agent { * the only provenance a host-side cancel has. */ private _pendingCancel: { reason?: string } | undefined; + /** The most recent cancelStream() made while this agent held no stream: + * during a turn's hooks or compile, while it waits to retry, or while no + * check of a turn is ahead at all. The framework collects it at the check + * ahead of it, which ends the turn, and the next start drops one no check + * collected, so a stop made while none was ahead, such as a host's + * interrupt of every agent, ends nothing. */ + private _stopWithoutStream: { reason?: string } | undefined; + /** * Cancel the stream this agent's state holds (streaming, waiting on its * tools, or ready at a tool boundary with every result in), and reset to - * idle. `reason` is provenance for the framework's inference:aborted trace - * and marker metadata (e.g. 'zombie_reclaim', 'subagent_cancel'); it never - * reaches the provider. + * idle. With no stream, the stop is kept for the turn in progress, which + * the framework ends at its next check, before it sends anything more, if + * one is still ahead; otherwise the next start drops it. `reason` is + * provenance for the framework's inference:aborted trace and marker + * metadata (e.g. 'zombie_reclaim', 'subagent_cancel'); it never reaches + * the provider. */ cancelStream(reason?: string): void { if ('stream' in this._state && this._state.stream) { this._pendingCancel = { reason }; this._state.stream.cancel(); + } else { + this._stopWithoutStream = { reason }; } this._state = { status: 'idle' }; } + /** Collect (and clear) the cancelStream() made while this agent held no + * stream, if one was made since the last collection. */ + takeStopWithoutStream(): { reason?: string } | undefined { + const s = this._stopWithoutStream; + this._stopWithoutStream = undefined; + return s; + } + /** Collect (and clear) the cancel that ended the current stream, if this * side issued one. The stream driver calls it once, while the stream * still owns the agent, at the stream's terminal event (`aborted` diff --git a/src/framework.ts b/src/framework.ts index 68a607d54..98bf1adf4 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -925,6 +925,22 @@ function ephemeralRunStopped(name: string, observed?: string): Error { * Its inference:failed came first, and this says no retry follows it. */ const RETRY_CANCELLED_BY_STOP = 'a retry waiting while the framework was stopping is dropped before it sends anything'; +/** The reason on the inference:aborted that ends a turn whose start waited to + * retry a failed compile across a stop() and a start(): the retry would + * revive the earlier lifecycle's turn in the next one. */ +const RETRY_ACROSS_RESTART = 'a retry waiting while the framework was stopped and started again is dropped before it sends anything'; + +/** One frame's window while it holds a turn with no stream and one of its + * checks is ahead (AgentFramework.checkAhead). `abandoned` is an operator's + * quiesce abandon that reached the turn there, which the frame's check + * collects; it goes with the window. `wake`, set while the frame waits to + * retry, ends that wait, so the check after it comes at once; called once + * the wait is over, it does nothing. */ +interface CheckWindow { + abandoned: boolean; + wake?: () => void; +} + function isTurnContinuation(reason: string): boolean { return reason === 'context_budget_restart' || reason === 'tool_results_ready' || reason === 'tool_result_guard_retry'; @@ -1267,9 +1283,15 @@ export interface HostModeStatus { /** Context writes (module events, api message.send) withheld because the * host is quiesced. Persisted while quiesced and flushed by resume(). */ deferredWrites: number; - /** Agents whose turn `abandon` could NOT cancel: the turn token is held but - * no stream exists yet (still in hooks/compile) or ever will (puppet tool - * turns). Present only after an abandon that left such turns behind. */ + /** Agents whose turn `abandon` had not ended when its grace ran out: a + * turn token held with no stream to cancel and no check ahead to stop it + * at (a puppet tool turn, a store surgery's reservation, a turn finishing + * its last delivery), or a turn with no stream that hasn't reached its + * next check, such as one in a hook that hasn't returned. A turn whose + * stream the abandon cancelled isn't listed, even one still running when + * the grace runs out, as one whose tool-result guard waits on its speech + * may be. Present only after an abandon that left such turns behind, while + * the host stays quiesced. */ unabandonable?: string[]; } @@ -2347,6 +2369,28 @@ export class AgentFramework { return this.activeTurnTriggers.get(agentName); } + /** + * The window of the frame that holds an agent's turn with no stream to + * carry a stop and one of its checks ahead: a start, from its entry until + * it hands its turn to a stream or ends (its checks after the hooks, after + * the compile, at a failure and after a wait to retry its compile), + * driveStream's wait to retry a failed inference (its check after the + * wait), and the tool-result guard's wait for the agent's speech (its + * check after the wait, which the window lets the abandon reach once a + * caller's cancelStream() has left the agent holding no stream). Each + * frame opens a window of its own and closes it only while it + * is still the agent's, so a start that ends after a later frame opened + * one, such as its retry's stream waiting to retry, leaves that one open. + * An operator's abandon marks the window, and the frame's check collects + * the mark, which goes with the window. abortInference() and the abandon + * count a turn with no stream only here; a stop made anywhere else, as + * while an agent is idle, a turn finishes its last delivery or a provider + * hold has parked it, has no check to end it, and the next start drops + * it. Keyed by the Agent, so a replacement agent under the same name + * never shares one. + */ + private checkAhead: Map = new Map(); + /** Structured calls executed in the current logical turn. Unlike the * per-stream local this survives framework retries and budget restarts. */ private logicalTurnToolCalls: WeakMap = new WeakMap(); @@ -4126,13 +4170,18 @@ export class AgentFramework { * Wait an error policy's retry delay, all of it: the delay can be a * provider's retry-after, and a single timer past 2^31-1 ms fires almost at * once. Steps through longer waits; stop() wakes it so a shutting-down - * framework is never left with a wait parked for days. + * framework is never left with a wait parked for days. The frame's window + * (checkWindow) carries a waker that ends the whole wait: a stop + * abortInference() records, or an operator's abandon, wakes it. */ - private async waitForRetry(ms: number): Promise { + private async waitForRetry(ms: number, checkWindow: CheckWindow): Promise { // An infinite delay waits until stop(), like any wait too long to end; a // delay that isn't a number, or is 0 or less, waits none. let remaining = ms > 0 ? ms : 0; - while (remaining > 0 && !this.providerAdmissionClosed) { + // stop() wakes this step through retryWaitWakers and ends the rest by + // closing admission; the window's waker ends the rest itself. + let ended = false; + while (remaining > 0 && !ended && !this.providerAdmissionClosed) { const step = Math.min(remaining, MAX_TIMER_DELAY_MS); remaining -= step; await new Promise((resolve) => { @@ -4140,6 +4189,7 @@ export class AgentFramework { const done = (): void => { clearTimeout(timer); wakers.delete(done); resolve(); }; const timer = setTimeout(done, step); wakers.add(done); + checkWindow.wake = () => { ended = true; done(); }; }); } } @@ -4797,7 +4847,10 @@ export class AgentFramework { } /** - * Abort an in-flight inference for an agent. + * Abort an in-flight inference for an agent. True when there was one to + * stop: a stream, or a turn with no stream and one of its checks ahead (its + * hooks or compile, a wait to retry, the tool-result guard's wait), which + * ends at that check, before it sends anything more. */ abortInference(agentName: string, reason?: string): boolean { const agent = this.agents.get(agentName); @@ -4816,6 +4869,17 @@ export class AgentFramework { if (result && !hadStream) { this.emitTrace({ type: 'inference:aborted', agentName, reason, durationMs: result.durationMs }); } + // A turn with no stream and a check still ahead (checkAhead) ends at that + // check, before it sends anything more, and reports the stop there. One + // waiting to retry reaches its check now, not when the wait runs out: + // the stop is reported taken, so the turn ends at once. A turn finishing + // its last delivery has no check ahead, and nothing to stop. + const checkWindow = this.checkAhead.get(agent); + if (!result && checkWindow) { + agent.cancelStream(reason); + checkWindow.wake?.(); + return true; + } return !!result; } @@ -4848,8 +4912,13 @@ export class AgentFramework { /** The stream's own log writer (driveStream's logStream), so the * terminal's entry carries the rounds the stream finished, once. */ log: (entry: Omit) => void; + /** A stop that reached the turn where no stream was running: 'sending' + * for a start that opens its turn, which has sent nothing, and + * 'finishing' for a start that continues a turn and for the guard's + * wait. The note says which, instead of a cut-off stream. */ + before?: 'sending' | 'finishing'; }): void { - const { agent, myStreamId, startTime, compiledRequest, reason, closes, log } = args; + const { agent, myStreamId, startTime, compiledRequest, reason, closes, log, before } = args; const durationMs = Date.now() - startTime; this.abortAgentScript(agent.name, 'stream cancelled'); agent.reset(); @@ -4894,10 +4963,15 @@ export class AgentFramework { 'user', [{ type: 'text', - text: - `[turn-interrupted] Your previous turn was cancelled mid-stream — ` + - `a deliberate stop, not a failure. Whatever you were still ` + - `producing when it stopped was cut off there.`, + text: before === 'sending' + ? `[turn-interrupted] Your previous turn was stopped before it sent anything — ` + + `a deliberate stop, not a failure.` + : before === 'finishing' + ? `[turn-interrupted] Your previous turn was stopped before it finished — ` + + `a deliberate stop, not a failure.` + : `[turn-interrupted] Your previous turn was cancelled mid-stream — ` + + `a deliberate stop, not a failure. Whatever you were still ` + + `producing when it stopped was cut off there.`, }], { system: true, kind: 'turn-interrupted', reason }, ); @@ -5925,6 +5999,7 @@ export class AgentFramework { const agent = this.agents.get(agentName); const state = agent?.state; const stream = state && 'stream' in state ? state.stream : undefined; + const checkWindow = agent ? this.checkAhead.get(agent) : undefined; if (agent && stream) { console.error(`[host-mode] abandoning in-flight turn for ${agentName}`); this.frameworkCancelledStreams.set( @@ -5932,11 +6007,28 @@ export class AgentFramework { 'quiesce_abandoned', ); stream.cancel(); + // A frame that waits with a window while the agent still holds a + // stream (the tool-result guard's wait, which holds the round's + // finished stream) is marked too, so the grace listing names the + // turn if it hasn't reached its check when the grace runs out. + if (checkWindow) checkWindow.abandoned = true; + } else if (checkWindow) { + // A turn with no stream and a check ahead (its hooks or compile, a + // wait to retry, the tool-result guard's wait once a caller's stop + // has taken its finished stream): the abandon marks its frame's + // window, that check collects the mark, and the turn ends there. A turn + // waiting to retry reaches the check after its wait now rather than + // when the wait runs out, as stop() wakes it, with the rest of a long + // wait ended too. + console.error(`[host-mode] abandoning in-flight turn for ${agentName}: it has no stream, so it ends at its next check`); + checkWindow.abandoned = true; + checkWindow.wake?.(); } else { - // Token held, no stream to cancel: the turn is between dequeue and - // stream registration (hooks/compile), or it is a puppet tool turn, - // which never has one. Say so — silence here reads as "abandon acted - // and teardown is slow", which is a different operator decision. + // Token held, no check ahead to stop it at: a puppet tool turn or a + // store surgery's reservation, which never has a stream, or a turn + // finishing its last delivery. Say so — silence here reads as + // "abandon acted and teardown is slow", which is a different + // operator decision. unabandonable.push(agentName); console.error( `[host-mode] cannot abandon turn for ${agentName}: turn token held with no stream ` + @@ -5951,7 +6043,20 @@ export class AgentFramework { while (this.quiesced && !this.queue?.isClosed && this.activeTurnTokens.size > 0 && Date.now() < grace) { await new Promise((resolve) => setTimeout(resolve, 250)); } - // Only turns still alive after the grace are worth reporting. + // Only turns still alive after the grace are worth reporting. A turn the + // abandon marked that hasn't reached its next check yet, such as one in a + // hook that hasn't returned, is still running too, so it is reported + // with them. + for (const [marked, checkWindow] of this.checkAhead) { + const name = marked.name; + if (checkWindow.abandoned && this.activeTurnTokens.has(name) && !unabandonable.includes(name)) { + console.error(`[host-mode] the abandon of ${name}'s turn waits for its next check, which it hasn't reached`); + unabandonable.push(name); + } + } + // A resume during the grace has cleared the list, and a serving host + // lists nothing. + if (!this.quiesced) return []; this.lastUnabandonable = unabandonable.filter((name) => this.activeTurnTokens.has(name)); return this.lastUnabandonable; } @@ -6537,8 +6642,8 @@ export class AgentFramework { 'a resume, so to end a turn that won\'t, quiesce with abandon: true before resuming. ' : 'Quiesce again once they have landed, or quiesce with abandon: true to end the turns they ' + 'wait for, then resume so they land, and quiesce again. ') + - 'The abandon\'s answer lists a turn it can\'t end as unabandonable, and its writes land once it has ' + - 'ended and the host is serving.', + 'The abandon\'s answer lists a turn it hasn\'t ended within its grace as unabandonable, and its writes ' + + 'land once it has ended and the host is serving.', ); } if (outcome !== 'nothing') this.deferredWritesCaughtUp(); @@ -15218,9 +15323,18 @@ export class AgentFramework { const turnToken = this.nextTurnToken++; this.activeTurnTokens.set(agent.name, turnToken); this.activeTurnTriggers.set(agent.name, trigger); + // A stop kept with no stream ends the turn at a check of the frame it + // found (checkAhead), which collects it. One still here when a start + // begins ends nothing: it was made where no check was ahead, as a host's + // interrupt of every agent is, or its frame ended without reaching one, + // as a withdrawn turn does. Drop it before anything awaits. This start's + // own checks are ahead now, in a window of its own. + agent.takeStopWithoutStream(); + const checkWindow: CheckWindow = { abandoned: false }; + this.checkAhead.set(agent, checkWindow); let tokenHandedOff = false; try { - tokenHandedOff = await this.beginAgentTurn(agent, trigger, attempt, turnToken, ownsProviderGate); + tokenHandedOff = await this.beginAgentTurn(agent, trigger, attempt, turnToken, ownsProviderGate, checkWindow); } catch (error) { // A throw outside the stream's own failure handling (the checkpoint, // locus resolution, the typing indicator) drops this start's trigger, @@ -15246,6 +15360,11 @@ export class AgentFramework { if (this.activeTurnTokens.get(agent.name) === turnToken) this.eventGate?.onInferenceEnded(agent.name); throw error; } finally { + // This start's checks are behind it, whether it handed its turn to a + // stream or the turn ended or parked here. A window a later frame of + // the agent has opened since, such as its retry's stream waiting to + // retry, stays open. + if (this.checkAhead.get(agent) === checkWindow) this.checkAhead.delete(agent); if (!tokenHandedOff && this.activeTurnTokens.get(agent.name) === turnToken) { this.activeTurnTokens.delete(agent.name); this.activeTurnTriggers.delete(agent.name); @@ -15288,6 +15407,78 @@ export class AgentFramework { this.emitTrace({ type: 'inference:aborted', agentName, reason, durationMs: now - startedAt }); } + /** + * The stop no stream will carry, collected at a check of the frame whose + * window it is (a start's checks before it sends anything, and the checks + * at a failure and after a wait to retry): an operator's quiesce abandon of + * the turn (the window's mark), else a caller's cancelStream() or + * abortInference() made while the agent held no stream, since the start + * that dropped an older one at its entry. When both are there, the + * framework's own stop is the one that ends the turn, as at a stream's + * terminal, and the caller's goes with it. + */ + private takeStartStop(agent: Agent, checkWindow: CheckWindow): + { kind: 'quiesce_abandoned' } | { kind: 'caller'; reason?: string } | undefined { + const callerStop = agent.takeStopWithoutStream(); + if (checkWindow.abandoned) return { kind: 'quiesce_abandoned' }; + return callerStop ? { kind: 'caller', reason: callerStop.reason } : undefined; + } + + /** + * End a start that a stop reached before it sent anything, through the + * terminal the same stop gets on a live stream: a caller's stop as + * settleDeliberateCancel ends a cancelled stream, with the note saying the + * turn was stopped before it sent anything (or, for a start that continues + * a turn, before it finished), and a quiesce abandon as + * settleFrameworkCancel ends one. Each settles the turn, an ephemeral run + * included, and emits the one inference:aborted. + */ + private endStoppedStart( + agent: Agent, + stop: { kind: 'quiesce_abandoned' } | { kind: 'caller'; reason?: string }, + routing: TurnRouting, + turnStartedAt: number, + compiledRequest: NormalizedRequest | undefined, + /** This start opened its turn, so nothing of the turn was sent, and the + * note says the turn was stopped before it sent anything; a start that + * continues one (a restart or a retry) may follow rounds already sent, + * so the note says it was stopped before it finished. */ + opensTurn: boolean, + ): void { + console.error(`[turn] ${agent.name}: ${stop.kind === 'caller' ? 'a stop' : 'a quiesce abandon'} reached the turn's start before it ${opensTurn ? 'sent anything' : 'finished'}; it ends there`); + this.channelRegistry?.stopTyping(); + // A start that continues a turn (a budget or guard restart) follows a + // frame that kept the turn's tool-result batch, and the guard's recovery, + // for its successor. The terminals below let a pending batch go; a + // recovery with no batch left ends here too. + agent.toolResultGuard.recovering = false; + // No stream ran, so there is no stream's logStream to write through and + // no usage to carry: the stop's entry is written directly, as the only + // entry of a request that was never sent. A stop before a compile made a + // request, or after one failed, has none to record, so the entry says + // that rather than a stream's note. + const requestId = `${agent.name}-${turnStartedAt}-${Math.random().toString(36).slice(2, 8)}`; + const common = { + agent, + myStreamId: agent.streamId, + startTime: turnStartedAt, + compiledRequest, + closes: routing, + log: (entry: Omit) => + this.logInference({ + timestamp: turnStartedAt, agentName: agent.name, requestId, ...entry, + request: compiledRequest ?? { note: 'no request: the turn was stopped before a compile made one' }, + }), + }; + if (stop.kind === 'caller') { + this.settleDeliberateCancel({ ...common, reason: stop.reason ?? 'user', before: opensTurn ? 'sending' : 'finishing' }); + } else { + this.settleFrameworkCancel({ ...common, kind: 'quiesce_abandoned' }); + } + const last = this.lastInferenceAt.get(agent.name); + if (last?.startedAt === turnStartedAt) this.lastInferenceAt.set(agent.name, { ...last, endedAt: Date.now() }); + } + /** * Body of startAgentStream, split out so the caller's try/finally owns the * turn token unconditionally. Returns true iff the token was handed off to @@ -15299,6 +15490,9 @@ export class AgentFramework { attempt: number, turnToken: number, ownsProviderGate: boolean, + /** This start's window (checkAhead), which its checks read for an + * operator's abandon. */ + checkWindow: CheckWindow, ): Promise { const lifecycle = this.lifecycleGeneration; const resumesTurn = resumesLogicalTurn(trigger?.reason); @@ -15599,9 +15793,14 @@ export class AgentFramework { // An ephemeral watchdog may dispose this Agent while hooks/compile await, // and stop() may close provider admission. Do not create a provider - // stream after its generation lost ownership, or once admission closed. - if (this.agents.get(agent.name) !== agent || this.providerAdmissionClosed) { - const why = this.agents.get(agent.name) !== agent ? 'the agent was replaced or removed' : 'the framework was stopping'; + // stream after its generation lost ownership, or once admission closed, + // or once a stop() and a start() have begun another lifecycle, as after + // the compile: these come before a stop kept for the turn, which goes + // with it. + if (this.agents.get(agent.name) !== agent || this.providerAdmissionClosed || lifecycle !== this.lifecycleGeneration) { + const why = this.agents.get(agent.name) !== agent ? 'the agent was replaced or removed' + : this.providerAdmissionClosed ? 'the framework was stopping' + : 'the framework was stopped and started again'; const reason = `a turn assembled while ${why} is dropped before it sends anything`; console.error(`[turn] ${agent.name}: ${reason}`); this.endDroppedTurn(agent.name, routing); @@ -15610,6 +15809,12 @@ export class AgentFramework { this.abortDroppedTurn(agent.name, turnToken, reason, turnStartedAt); return false; } + // A stop made while the hooks ran had no stream to cancel. + const stopAfterHooks = this.takeStartStop(agent, checkWindow); + if (stopAfterHooks) { + this.endStoppedStart(agent, stopAfterHooks, routing, turnStartedAt, undefined, opensTurn); + return false; + } // The subconscious's standing dispositions ride a fixed system-position // injection (antra, #tuneout-talk): never repeated in its timeline, and @@ -15661,6 +15866,18 @@ export class AgentFramework { this.abortDroppedTurn(agent.name, turnToken, reason, turnStartedAt); return false; } + // A stop made while the compile ran had no stream to cancel. The stream + // the compile made hasn't been iterated, so cancelling it sends nothing. + const stopAfterCompile = this.takeStartStop(agent, checkWindow); + if (stopAfterCompile) { + stream.cancel(); + // As above: the compile took the agent's stream id from any frame + // that started this one. + this.activeStreams.delete(agent.name); + this.pendingAssistantBlocks.delete(agent.name); + this.endStoppedStart(agent, stopAfterCompile, routing, turnStartedAt, compiledRequest, opensTurn); + return false; + } // A conversation agent takes no provider gate, so its own compression // can record a wait while this turn compiles, after the last boundary // checked. Check its waits again here, where nothing has been sent, and @@ -15670,12 +15887,18 @@ export class AgentFramework { if (wait && this.holdForProviderWait(agent, wait, trigger ? [trigger] : [])) { console.error(`[provider-wait] agent=${agent.name} model=${agent.model} binds after the compile ` + `(recorded while the turn compiled); turn held, nothing sent`); + // The stream the compile made won't be sent: it's cancelled, and + // this start's window closes, before the hold is reported, so a stop + // made while inference:failed is delivered finds neither, and the + // hold's later start drops a cancelStream() made then, as for any + // parked turn. + stream.cancel(); + agent.cancelStream(); + if (this.checkAhead.get(agent) === checkWindow) this.checkAhead.delete(agent); // The turn emitted inference:started before its compile; every turn // that does gets one terminal event, which listeners (a typing loop) // rely on, as the failure sites' holds emit inference:failed. this.emitTrace({ type: 'inference:failed', agentName: agent.name, error: providerWaitText(agent.name, agent.model, wait) }); - stream.cancel(); - agent.cancelStream(); this.activeStreams.delete(agent.name); this.pendingAssistantBlocks.delete(agent.name); this.channelRegistry?.stopTyping(); @@ -15692,6 +15915,10 @@ export class AgentFramework { else this.turnPrompts.delete(trigger); } + // The turn has a live stream from here, which carries a stop: this + // start's checks are behind it, so its window closes now, before + // driveStream can report a stream that fails at once. + if (this.checkAhead.get(agent) === checkWindow) this.checkAhead.delete(agent); const handle = this.driveStream( agent, stream, @@ -15722,6 +15949,34 @@ export class AgentFramework { }); agent.reset(); + // A stop made while the compile ran had no stream to cancel, and the + // compile's failure skipped the check after it; a stop made while the + // failure is reported, with this start's window still open, is kept + // the same way. Either ends the turn here, as at that check, before a + // hold parks the turn, a retry waits or the failure exhausts it. (The + // hooks fail open, and nothing after the compile throws here, so this + // catch never holds a compiled request.) As at the checks after the + // hooks and after the compile, a replaced agent or a stopping framework + // comes first: the turn is dropped as theirs, and the stop goes with + // it, and so does a framework stopped and started again, as after the + // compile. A failed start with no stop goes on below as before. + const stopAtFailure = this.takeStartStop(agent, checkWindow); + if (stopAtFailure) { + if (this.agents.get(agent.name) !== agent || this.providerAdmissionClosed || lifecycle !== this.lifecycleGeneration) { + const why = this.agents.get(agent.name) !== agent ? 'the agent was replaced or removed' + : this.providerAdmissionClosed ? 'the framework was stopping' + : 'the framework was stopped and started again'; + const reason = `a turn whose request couldn't be built while ${why} is dropped before it sends anything`; + console.error(`[turn] ${agent.name}: ${reason}`); + this.endDroppedTurn(agent.name, routing); + this.eventGate?.onInferenceEnded(agent.name); + this.abortDroppedTurn(agent.name, turnToken, reason, turnStartedAt); + return false; + } + this.endStoppedStart(agent, stopAtFailure, routing, turnStartedAt, undefined, opensTurn); + return false; + } + if (ownsProviderGate && this.holdProviderAcceleration(agent, err, trigger)) { // Capacity scheduling is not poisoned history. The cooldown timer owns // one later fresh compile; no immediate retry or exhaustion marker. @@ -15739,20 +15994,32 @@ export class AgentFramework { const action: ErrorAction = statedWait === 'fail' ? { retry: false } : this.errorPolicy.onInferenceError(err, agent.name, attempt); if (action.retry) { // A policy's delay can be a provider's retry-after: wait all of it. - await this.waitForRetry(action.delayMs); - if (this.providerAdmissionClosed) { + await this.waitForRetry(action.delayMs, checkWindow); + if (this.providerAdmissionClosed || lifecycle !== this.lifecycleGeneration) { // stop() ended the wait: that cancels the retry, it does not hasten - // it. The turn ends as one dropped before it sends anything does: - // a terminal event saying why, after the inference:failed above, - // and its record's end. Nothing is written (the store is closing); + // it. A stop() and a start() since this start began cancel it too, + // as after the compile: the retry would revive the earlier + // lifecycle's turn in the next one, and a stop kept for the turn + // goes with it. The turn ends as one dropped before it sends + // anything does: a terminal event saying why, after the + // inference:failed above, and its record's end. Nothing is written; // this frame still owns its token, since no retry frame replaced it. - this.abortDroppedTurn(agent.name, turnToken, RETRY_CANCELLED_BY_STOP, turnStartedAt); + this.abortDroppedTurn(agent.name, turnToken, + this.providerAdmissionClosed ? RETRY_CANCELLED_BY_STOP : RETRY_ACROSS_RESTART, turnStartedAt); if (this.activeTurnTokens.get(agent.name) === turnToken) { this.activeTurnTokens.delete(agent.name); this.activeTurnTriggers.delete(agent.name); } return false; } + // A stop made during the wait reached no stream: it ends the turn at + // this check, before the retry's start, as a stop during the hooks + // does at theirs. + const stopAfterWait = this.takeStartStop(agent, checkWindow); + if (stopAfterWait) { + this.endStoppedStart(agent, stopAfterWait, routing, turnStartedAt, undefined, opensTurn); + return false; + } // The retry re-enters startAgentStream, which replaces this frame's // turn token with its own; cleanup belongs to the innermost frame. await this.startAgentStream(agent, trigger, attempt + 1); @@ -16595,12 +16862,48 @@ export class AgentFramework { reason: 'tool_result_guard', inputTokens: agent.lastStreamInputTokens, budget: agent.maxStreamTokens, }); + // The wait has its own window (checkAhead), so a stop reaches + // the turn even once a caller's cancelStream() has left the + // agent holding no stream: the abandon marks the window, and + // the check below collects the mark. + const guardWindow: CheckWindow = { abandoned: false }; + this.checkAhead.set(agent, guardWindow); await turnSpeechChain; + if (this.checkAhead.get(agent) === guardWindow) this.checkAhead.delete(agent); if (this.agents.get(agent.name) !== agent || agent.streamId !== myStreamId) { generationLost = true; lifecyclePhase = 'aborted'; return; } + // The stream has completed, so a stop made during the wait + // reached no stream event: it ends the turn here instead of + // the retry. The framework's own stop comes first, as at a + // stream's terminal, even when a caller's came before it, and + // stop() before an abandon: stop() by its record on the stream + // or by having closed admission, since a caller's stop that + // took the stream leaves it nothing to record on, and the + // abandon by its record or by its mark on the window. Which + // of them came first, and where a caller's stop fell between + // them, then changes nothing. + const recorded = this.frameworkCancelledStreams.get(`${agent.name}:${myStreamId}`); + const frameworkStop = recorded === 'shutdown' || this.providerAdmissionClosed ? 'shutdown' + : recorded === 'quiesce_abandoned' || guardWindow.abandoned ? 'quiesce_abandoned' : undefined; + const callerStop = agent.takeCancel(); + if (frameworkStop === 'shutdown' || frameworkStop === 'quiesce_abandoned') { + lifecyclePhase = 'aborted'; + this.settleFrameworkCancel({ + agent, myStreamId, startTime, compiledRequest, kind: frameworkStop, closes: streamRouting, log: logStream, + }); + return; + } + if (callerStop) { + lifecyclePhase = 'aborted'; + this.settleDeliberateCancel({ + agent, myStreamId, startTime, compiledRequest, + reason: callerStop.reason ?? 'user', closes: streamRouting, log: logStream, before: 'finishing', + }); + return; + } agent.reset(); preserveEventGateForSuccessor = true; lifecyclePhase = 'aborted'; @@ -17188,6 +17491,15 @@ export class AgentFramework { // A retry is an inference of its own, with its own `started`, and // a stop that cancels the retry leaves this failure standing. lifecyclePhase = 'failed'; + // The stream has failed and won't go on: the agent stops holding + // it before the failure is reported. A stop made while + // inference:failed is delivered then finds no stream and, until a + // wait to retry opens its window, no window: abortInference() + // reports false, and a cancelStream() is kept for the check after + // that wait, or dropped by the next start, as any stop with no + // check ahead is. + if (agent.streamId === myStreamId) agent.reset(); + this.emitTrace({ type: 'inference:failed', agentName: agent.name, @@ -17206,7 +17518,6 @@ export class AgentFramework { if (agent.streamId !== myStreamId) break; this.abortAgentScript(agent.name, 'stream error'); - agent.reset(); if (ownsProviderGate && this.holdProviderAcceleration(agent, err, trigger)) { if (this.agents.get(agent.name) === agent && agent.streamId === myStreamId) this.eventGate?.onInferenceEnded(agent.name); @@ -17220,12 +17531,38 @@ export class AgentFramework { const action: ErrorAction = statedWait === 'fail' ? { retry: false } : this.errorPolicy.onInferenceError(err, agent.name, attempt); if (action.retry) { + // A stop made during the wait reaches no stream: the check after + // the wait ends the turn, as the guard's check after its wait + // does. Until then, that check is ahead, in a window of its own + // (checkAhead). + const retryWindow: CheckWindow = { abandoned: false }; + this.checkAhead.set(agent, retryWindow); // A policy's delay can be a provider's retry-after: wait all of it. - await this.waitForRetry(action.delayMs); + await this.waitForRetry(action.delayMs, retryWindow); + if (this.checkAhead.get(agent) === retryWindow) this.checkAhead.delete(agent); // stop() ending the wait cancels the retry (this frame's finally // clears the token); it is never permission to retry at once. - if (!this.providerAdmissionClosed) await this.startAgentStream(agent, trigger, attempt + 1); - else { + if (!this.providerAdmissionClosed) { + const stop = this.agents.get(agent.name) === agent && agent.streamId === myStreamId + ? this.takeStartStop(agent, retryWindow) : undefined; + if (stop) { + // The stop cancels the retry and ends the turn; the failed + // inference's phase stays `failed`, as when stop() cancels + // the retry. + if (stop.kind === 'quiesce_abandoned') { + this.settleFrameworkCancel({ + agent, myStreamId, startTime, compiledRequest, kind: 'quiesce_abandoned', closes: streamRouting, log: logStream, + }); + } else { + this.settleDeliberateCancel({ + agent, myStreamId, startTime, compiledRequest, + reason: stop.reason ?? 'user', closes: streamRouting, log: logStream, before: 'finishing', + }); + } + return; + } + await this.startAgentStream(agent, trigger, attempt + 1); + } else { // The turn ends here, and its inference:failed doesn't say so, // since a retry usually follows one: an inference:aborted // says why none does, as beginAgentTurn's cancelled retry @@ -17543,6 +17880,10 @@ export class AgentFramework { // fresh compile; every other throw keeps the ordinary exhausted path. const err = error instanceof Error ? error : new Error(String(error)); const durationMs = Date.now() - startTime; + // The stream won't go on: the agent stops holding it before the + // failure is reported, as at the stream's error event, so a stop made + // while inference:failed is delivered finds no stream. + agent.reset(); if (ownsProviderGate && this.holdProviderAcceleration(agent, err, trigger)) { this.emitTrace({ type: 'inference:failed', agentName: agent.name, error: err.message, stack: err.stack }); logStream({ success: false, diff --git a/src/types/trace.ts b/src/types/trace.ts index 94b5e99d6..2e13479ad 100644 --- a/src/types/trace.ts +++ b/src/types/trace.ts @@ -674,7 +674,7 @@ export type TraceEvent = parkedAdmissions?: number; /** True when undrained turns were force-cancelled (abandon). */ abandoned?: boolean; - /** Agents whose turn abandon could not cancel (token held, no stream). */ + /** Agents whose turn the abandon had not ended when its grace ran out: a token held with no stream and no check ahead (a puppet tool turn, a store surgery's reservation, a turn finishing its last delivery), or a turn that hadn't reached its next check, the tool-result guard's wait for its speech included. A turn whose stream the abandon cancelled while it was still streaming isn't listed, even one still running when the grace ran out. */ unabandonable?: string[]; }) | (TraceEventBase & { diff --git a/test/deferred-write-failures.test.ts b/test/deferred-write-failures.test.ts index 898c87a4e..59494d72d 100644 --- a/test/deferred-write-failures.test.ts +++ b/test/deferred-write-failures.test.ts @@ -286,7 +286,7 @@ test('a queue over the cap is refused by a quiesce() without abandon, with the h message: `quiesce refused: the queue's 1 deferred context write(s) come to more than 4194304 characters of JSON, ` + `the cap, so ${h.queuePath} can't hold them, and a restart would lose them. Quiesce again once they have landed, ` + 'or quiesce with abandon: true to end the turns they wait for, then resume so they land, and quiesce again. ' + - "The abandon's answer lists a turn it can't end as unabandonable, and its writes land once it has ended and " + + "The abandon's answer lists a turn it hasn't ended within its grace as unabandonable, and its writes land once it has ended and " + 'the host is serving.', }); assert.equal(h.framework.getHostModeStatus().quiesced, false, 'the host is still serving'); @@ -302,7 +302,7 @@ test('a queue over the cap is refused by a quiesce() without abandon, with the h message: `quiesce refused: the queue's 2 deferred context write(s) come to more than 4194304 characters of JSON, ` + `the cap, so ${h.queuePath} can't hold them, and a restart would lose them. Resume so they land, then quiesce ` + 'again. A write held for a turn still alive stays queued through a resume, so to end a turn that won\'t, ' + - "quiesce with abandon: true before resuming. The abandon's answer lists a turn it can't end as unabandonable, " + + "quiesce with abandon: true before resuming. The abandon's answer lists a turn it hasn't ended within its grace as unabandonable, " + 'and its writes land once it has ended and the host is serving.', }); assert.equal(h.framework.getHostModeStatus().quiesced, true, 'still quiesced'); diff --git a/test/inference-lifecycle-failed.test.ts b/test/inference-lifecycle-failed.test.ts index 911c1bf81..82cb2651c 100644 --- a/test/inference-lifecycle-failed.test.ts +++ b/test/inference-lifecycle-failed.test.ts @@ -161,6 +161,67 @@ test('a stream that fails, whose retry stop() cancels, ends failed', { timeout: } }); +test('a stream that fails, whose retry a caller\'s stop cancels during its wait, ends failed', { timeout: 60_000 }, async () => { + let waiting!: () => void; + const inDelay = new Promise((resolve) => { waiting = resolve; }); + const retryLater: ErrorPolicy = { + maxRetries: 1, + onInferenceError: (_error, _agent, attempt) => { + waiting(); + return attempt < 1 ? { retry: true, delayMs: 60_000 } : { retry: false }; + }, + }; + const { framework, membrane, traces, phases, teardown } = await frameworkOver([failure()], retryLater); + try { + framework.start(); + framework.pushEvent({ type: 'external-message', source: 'test', content: 'hello', metadata: {} }); + await inDelay; + assert.equal(framework.abortInference('resident', 'user'), true, 'abortInference reports the stop'); + for (let i = 0; i < 500 && !(traces.some((t) => t.type === 'inference:aborted') && framework.getHostModeStatus().activeTurns === 0); i++) { + await new Promise((resolve) => setTimeout(resolve, 10)); + } + assert.equal(membrane.calls.length, 1, 'the retry was cancelled, not sent'); + assert.ok( + traces.some((t) => t.type === 'inference:aborted' && t.reason === 'user'), + `the caller's stop ended the turn: ${JSON.stringify(traces.filter((t) => t.type.startsWith('inference:')))}`, + ); + assert.deepEqual(phases(), [['started', 'failed']]); + } finally { + await teardown(); + } +}); + +test('a stream that fails, whose retry an operator\'s abandon cancels during its wait, ends failed', { timeout: 60_000 }, async () => { + let waiting!: () => void; + const inDelay = new Promise((resolve) => { waiting = resolve; }); + const retryLater: ErrorPolicy = { + maxRetries: 1, + onInferenceError: (_error, _agent, attempt) => { + waiting(); + return attempt < 1 ? { retry: true, delayMs: 60_000 } : { retry: false }; + }, + }; + const { framework, membrane, traces, phases, teardown } = await frameworkOver([failure()], retryLater); + try { + framework.start(); + framework.pushEvent({ type: 'external-message', source: 'test', content: 'hello', metadata: {} }); + await inDelay; + await framework.quiesce({ timeoutMs: 50, abandon: true }); + for (let i = 0; i < 500 && !(traces.some((t) => t.type === 'inference:aborted') && framework.getHostModeStatus().activeTurns === 0); i++) { + await new Promise((resolve) => setTimeout(resolve, 10)); + } + assert.equal(membrane.calls.length, 1, 'the retry was cancelled, not sent'); + assert.ok( + traces.some((t) => t.type === 'inference:aborted' && t.reason === 'quiesce_abandoned'), + `the abandon ended the turn: ${JSON.stringify(traces.filter((t) => t.type.startsWith('inference:')))}`, + ); + assert.deepEqual(phases(), [['started', 'failed']]); + await framework.resume(); + } finally { + await teardown(); + } +}); + test('a stream whose 429 holds the turn for the provider\'s wait ends failed', async () => { const rateLimited = new MembraneError({ type: 'rate_limit', retryable: true, httpStatus: 429, retryAfterMs: 60_000, diff --git a/test/provider-waits.test.ts b/test/provider-waits.test.ts index c4516117c..f97cb0ee1 100644 --- a/test/provider-waits.test.ts +++ b/test/provider-waits.test.ts @@ -918,7 +918,7 @@ test('a retry wait past one timer waits in steps of at most one timer, and only await withStoreDir(async (path) => { const log = silence(); const { fw } = await framework(path, new WaitingMembrane(0, undefined)); - const stepping = fw as unknown as { waitForRetry(ms: number): Promise }; + const stepping = fw as unknown as { waitForRetry(ms: number, checkWindow: { abandoned: boolean }): Promise }; t.mock.timers.enable({ apis: ['setTimeout'] }); const mocked = globalThis.setTimeout; const delays: number[] = []; @@ -928,7 +928,7 @@ test('a retry wait past one timer waits in steps of at most one timer, and only }) as unknown as typeof setTimeout; try { let done = false; - const waiting = stepping.waitForRetry(ONE_TIMER_MS + 1_000).then(() => { done = true; }); + const waiting = stepping.waitForRetry(ONE_TIMER_MS + 1_000, { abandoned: false }).then(() => { done = true; }); t.mock.timers.tick(ONE_TIMER_MS); for (let i = 0; i < 20; i++) await Promise.resolve(); assert.equal(done, false, "still waiting after one timer's worth"); @@ -2136,6 +2136,32 @@ test("a conversation agent's turn is held, sending nothing, when its own wait is }); }); +test("a stop made while that hold is reported finds nothing to stop: abortInference() from a listener of its inference:failed reports false", async (t) => { + await withStoreDir(async (path) => { + holdClock(t); + const log = silence(); + let record: (() => void) | undefined; + const membrane = new WaitDuringCompile(() => { record?.(); record = undefined; }); + const { fw, internal } = await framework(path, membrane); + record = () => internal.providerWaits.set('resident', 'zz-model', 400, 'zz recorded while the turn compiled'); + // The compiled stream is cancelled, and the start's window closed, before + // the hold is reported: a host's listener of that report finds no turn + // it could stop, rather than a stop it takes and the hold then loses. + let taken: boolean | undefined; + fw.onTrace((event) => { + if (event.type === 'inference:failed' && (event as { agentName?: string }).agentName === 'resident' && taken === undefined) { + taken = fw.abortInference('resident', 'user'); + } + }); + try { + await conversationTurn(fw, internal as NonOwning, 'zz-first'); + assert.equal(taken, false, 'abortInference took no stop'); + assert.equal(membrane.sent, 0, 'nothing is sent into the wait'); + assert.equal(internal.providerAccelerationCooldowns.get('resident')?.heldRequests.length, 1, 'the hold has the turn'); + } finally { await fw.stop(); log.restore(); } + }); +}); + test('a release can name a disposed conversation agent whose wait the listing shows, and a name with neither is refused', async () => { await withStoreDir(async (path) => { const log = silence(); diff --git a/test/stop-without-stream.test.ts b/test/stop-without-stream.test.ts new file mode 100644 index 000000000..cf9c13ff5 --- /dev/null +++ b/test/stop-without-stream.test.ts @@ -0,0 +1,1686 @@ +/** + * A stop that lands while a turn is alive but has no live stream to cancel: + * before its stream exists (here a module's gatherContext, or the compile, + * holds it), while it waits to retry a failed inference or compile, or + * while the tool-result guard waits on the agent's speech to retry after a + * refusal. Agent.cancelStream(), abortInference() and a quiesce's abandon + * end the turn at its next check, + * or where its hooks or compile fail, before it sends anything more, + * through the terminal the same stop gets on a live stream. A stop made + * where no check of a turn is ahead ends nothing. + */ + +import { describe, it } from 'node:test'; +import assert from 'node:assert'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { + Membrane as RealMembrane, + MembraneError, + NativeFormatter, + type ContentBlock, + type Membrane, + type NormalizedRequest, + type NormalizedResponse, + type ProviderAdapter, + type ProviderRequest, + type ProviderResponse, + type StreamEvent, + type YieldingStream, + type YieldingStreamOptions, +} from '@animalabs/membrane'; +import type { ContextInjection } from '@animalabs/context-manager'; +import type { Agent, EventResponse, Module, ModuleContext, ProcessEvent, ToolResult, TraceEvent } from '../src/index.js'; +import { AgentFramework } from '../src/index.js'; +import { MockMembrane, MockYieldingStream, createMockResponse } from './helpers/mock-membrane.js'; + +function deferred(): { promise: Promise; resolve: (value: T) => void } { + let resolve!: (value: T) => void; + const promise = new Promise((r) => { resolve = r; }); + return { promise, resolve }; +} + +/** `promise`, or a failure naming what was awaited once `ms` have passed. */ +async function within(label: string, promise: Promise, ms = 5_000): Promise { + let timer: ReturnType | undefined; + try { + return await Promise.race([ + promise, + new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(`timed out after ${ms} ms waiting for ${label}`)), ms); + }), + ]); + } finally { + clearTimeout(timer); + } +} + +async function waitFor(label: string, predicate: () => boolean, timeoutMs = 5_000): Promise { + const deadline = Date.now() + timeoutMs; + while (!predicate()) { + if (Date.now() > deadline) throw new Error(`timed out after ${timeoutMs} ms waiting for ${label}`); + await new Promise((r) => setTimeout(r, 10)); + } +} + +/** Resolves once a quiesce's abandon has gone through the framework's turns, + * cancelling, marking or listing each, before its grace begins. The + * abandon does that before its first await, so it's done when the call + * returns. */ +function abandonReached(framework: AgentFramework): Promise { + const internal = framework as unknown as { abandonActiveTurns: () => Promise }; + const abandon = internal.abandonActiveTurns.bind(framework); + const reached = deferred(); + internal.abandonActiveTurns = () => { + const grace = abandon(); + reached.resolve(); + return grace; + }; + return reached.promise; +} + +/** Answers a message with a wake. Its gatherContext waits on the gate that + * `hold()` arms for the next call, which keeps that turn alive with no + * stream. */ +class GatedContextModule implements Module { + readonly name = 'gated'; + private gate: Promise | null = null; + private entered = deferred(); + async start(ctx: ModuleContext): Promise { ctx.registerSpeechHandler('*'); } + async stop(): Promise {} + async onAgentSpeech(): Promise {} + getTools() { + return [{ name: 'read', description: 'Read a result', inputSchema: { type: 'object' as const, properties: {} } }]; + } + async handleToolCall(): Promise { return { success: true, data: 'payload' }; } + async onProcess(event: ProcessEvent): Promise { + if (event.type === 'external-message') { + return { + addMessages: [{ participant: 'User', content: [{ type: 'text', text: String((event as { content: unknown }).content) }] }], + requestInference: true, + }; + } + return {}; + } + /** Hold the next gatherContext until `release()`; `entered` settles once it waits. */ + hold(): { entered: Promise; release: () => void } { + const released = deferred(); + this.gate = released.promise; + this.entered = deferred(); + return { entered: this.entered.promise, release: () => released.resolve() }; + } + async gatherContext(): Promise { + const gate = this.gate; + this.gate = null; + if (gate) { + this.entered.resolve(); + await gate; + } + return []; + } +} + +/** Membrane itself over a provider that answers every request. `requests` + * are the requests sent, which membrane makes only once a stream is read. */ +function answeringMembrane(): { membrane: RealMembrane; requests: ProviderRequest[] } { + const requests: ProviderRequest[] = []; + const adapter: ProviderAdapter = { + name: 'test', + usageCacheConvention: 'cache-excluded', + supportsModel: () => true, + complete: async () => { throw new Error('unexpected complete'); }, + stream: async (request: ProviderRequest): Promise => { + requests.push(request); + return { + content: [{ type: 'text', text: 'Answered.' }], + stopReason: 'end_turn', + usage: { inputTokens: 10, outputTokens: 5 }, + model: 'test-model', + } as ProviderResponse; + }, + }; + return { membrane: new RealMembrane(adapter, { formatter: new NativeFormatter() }), requests }; +} + +/** A stream whose only event is `error`. */ +class ErrorStream implements YieldingStream { + isWaitingForTools = false; + pendingToolCallIds: string[] = []; + toolDepth = 0; + isCancelled = false; + constructor(private readonly error: Error) {} + provideToolResults(): void { throw new Error('not waiting'); } + cancel(): void { this.isCancelled = true; } + async *[Symbol.asyncIterator](): AsyncIterator { + yield { type: 'error', error: this.error } as StreamEvent; + } +} + +/** Queue a budget restart's trigger for `prime`, as the scheduler sees one. */ +function queueRestart(framework: AgentFramework): void { + (framework as unknown as { pendingRequests: Array> }).pendingRequests.push({ + agentName: 'prime', reason: 'context_budget_restart', source: 'framework', timestamp: Date.now(), + }); +} + +/** Fails its first request on a spent quota, a 429 a host's providerHold + * parks, and answers the rest. `calls` are the streams made. */ +class QuotaOnceMembrane extends MockMembrane { + override streamYielding(request: NormalizedRequest): YieldingStream { + this.calls.push(request); + if (this.calls.length === 1) { + return new ErrorStream(new MembraneError({ + type: 'rate_limit', retryable: true, httpStatus: 429, message: 'quota spent', + rawError: { status: 429 }, rawRequest: request, + })); + } + return new MockYieldingStream([createMockResponse([{ type: 'text', text: 'Answered.' }])]); + } +} + +const abortReasons = (traces: TraceEvent[]): Array => + traces.filter((t) => t.type === 'inference:aborted').map((t) => (t as { reason?: string }).reason); +const otherTerminals = (traces: TraceEvent[]): string[] => + traces.map((t) => t.type).filter((t) => t === 'inference:completed' || t === 'inference:failed' || t === 'inference:exhausted'); + +/** The `[turn-interrupted]` notes in an agent's window. */ +function notes(agent: Agent): string[] { + return (agent.getContextManager().queryMessages({}).messages as unknown as Array<{ content: Array<{ type: string; text?: string }> }>) + .flatMap((m) => m.content.flatMap((b) => b.type === 'text' && (b.text ?? '').startsWith('[turn-interrupted]') ? [b.text!] : [])); +} + +/** A turn of `prime` held with no stream: in its hooks, or with + * `holdCompile`, at its compile once its hooks have run. */ +async function aliveWithoutStream(opts: { holdCompile?: boolean } = {}) { + const tempDir = mkdtempSync(join(tmpdir(), 'af-stop-without-stream-')); + const { membrane, requests } = answeringMembrane(); + const module = new GatedContextModule(); + const framework = await AgentFramework.create({ + storePath: join(tempDir, 'store'), + membrane, + agents: [{ name: 'prime', model: 'test-model', systemPrompt: 'You are prime.', allowedTools: 'all' }], + modules: [module], + syncIntervalMs: 0, + }); + const traces: TraceEvent[] = []; + framework.onTrace((event) => traces.push(event)); + const agent = framework.getAgent('prime')!; + let compiles = 0; + const compileEntered = deferred(); + const compileReleased = deferred(); + const compile = agent.startStreamWithInjections.bind(agent); + agent.startStreamWithInjections = (async (...args: Parameters) => { + compiles++; + return compile(...args); + }) as typeof agent.startStreamWithInjections; + // `holdCompile` holds the compile at its own await on the tool-result + // guard's annotations: the compile is under way, and the agent holds no + // stream until it returns. + const guard = agent.toolResultGuard; + const annotations = guard.afterAnnotations.bind(guard); + guard.afterAnnotations = (async (read: () => Promise) => { + if (opts.holdCompile && compiles === 1) { + compileEntered.resolve(); + await compileReleased.promise; + } + return annotations(read); + }) as typeof guard.afterAnnotations; + const hooks = opts.holdCompile ? null : module.hold(); + framework.start(); + framework.pushEvent({ type: 'external-message', source: 'test', content: 'hello', metadata: {}, triggerInference: true } as unknown as ProcessEvent); + await within(opts.holdCompile ? "the turn's compile" : "the turn's hooks", hooks ? hooks.entered : compileEntered.promise); + assert.equal(framework.getHostModeStatus().activeTurns, 1, 'the turn is alive'); + assert.ok(!('stream' in agent.state) || !agent.state.stream, 'and has no stream'); + return { + framework, agent, traces, requests, module, + compiles: () => compiles, + release: () => { hooks?.release(); compileReleased.resolve(); }, + turnEnded: () => framework.getHostModeStatus().activeTurns === 0 && agent.state.status === 'idle', + cleanup: async () => { + hooks?.release(); + compileReleased.resolve(); + await within('stop() in cleanup', framework.stop()).catch(() => {}); + rmSync(tempDir, { recursive: true, force: true }); + }, + }; +} + +/** The inference-log rows a stop wrote: the terminal's row, which reads as a + * stop on a live stream's does. */ +function stopRows(framework: AgentFramework, agentName: string, reason: string) { + const text = reason === 'quiesce_abandoned' || reason === 'shutdown' ? `Stream aborted: ${reason}` : `Stream cancelled (${reason})`; + return framework.queryInferenceLogs({ agentName }).entries.map((e) => e.entry).filter((entry) => entry.error === text); +} + +function endedAsStop(h: { framework: AgentFramework; traces: TraceEvent[]; requests: ProviderRequest[]; agent: Agent }, reason: string, note: RegExp | null): void { + assert.equal(h.requests.length, 0, 'nothing was sent'); + assert.deepEqual(abortReasons(h.traces), [reason], "one inference:aborted, with the stop's reason"); + assert.deepEqual(otherTerminals(h.traces), [], 'and no other terminal'); + const rows = h.framework.queryInferenceLogs({ agentName: h.agent.name }).entries.map((e) => e.entry); + assert.equal(rows.length, 1, 'the inference log has one row for the turn'); + assert.equal(stopRows(h.framework, h.agent.name, reason).length, 1, "and it's the stop's"); + assert.equal(rows[0].success, false); + assert.equal(rows[0].tokenUsage, undefined, 'with no usage, since nothing was sent'); + if (note) assert.ok(notes(h.agent).some((n) => note.test(n)), `the agent's note says ${note}`); + else assert.deepEqual(notes(h.agent), [], 'no note'); +} + +describe('a stop before the turn has a stream', () => { + for (const withStop of [true, false]) it(`a turn whose framework is stopped and started again during its hooks is dropped before its compile${withStop ? ', a stop made during them included, with no note' : ''}`, async () => { + const h = await aliveWithoutStream(); + try { + if (withStop) assert.equal(h.framework.abortInference('prime', 'user'), true, 'abortInference reports the stop'); + // As a stop() and a start() do: each advances the lifecycle, and + // admission is open again. + (h.framework as unknown as { lifecycleGeneration: number }).lifecycleGeneration += 2; + h.release(); + await waitFor('the turn to end', h.turnEnded); + assert.deepEqual(abortReasons(h.traces), ['a turn assembled while the framework was stopped and started again is dropped before it sends anything'], 'the turn was dropped'); + assert.equal(h.compiles(), 0, 'before its compile'); + assert.equal(h.requests.length, 0, 'nothing was sent'); + assert.deepEqual(notes(h.agent), [], 'and no [turn-interrupted] note was written'); + } finally { + await h.cleanup(); + } + }); + + it("a host's cancelStream() while the turn's hooks run ends the turn before its compile", async () => { + const h = await aliveWithoutStream(); + try { + h.agent.cancelStream(); + h.release(); + await waitFor('the turn to end', h.turnEnded); + endedAsStop(h, 'user', /stopped before it sent anything/); + assert.equal(h.compiles(), 0, 'the turn ended at its check after the hooks'); + assert.deepEqual(h.framework.queryInferenceLogs({ agentName: 'prime' }).entries[0].entry.request, + { note: 'no request: the turn was stopped before a compile made one' }, 'its row says no request was compiled'); + await within('stop() to finish', h.framework.stop()); + } finally { + await h.cleanup(); + } + }); + + it('abortInference() while the hooks run reports the stop, which ends the turn', async () => { + const h = await aliveWithoutStream(); + try { + assert.equal(h.framework.abortInference('prime', 'operator stop'), true, 'abortInference reports the stop'); + h.release(); + await waitFor('the turn to end', h.turnEnded); + endedAsStop(h, 'operator stop', /stopped before it sent anything/); + } finally { + await h.cleanup(); + } + }); + + it('a stop during the compile ends the turn before its stream is read', async () => { + const h = await aliveWithoutStream({ holdCompile: true }); + try { + h.agent.cancelStream('subagent_cancel'); + h.release(); + await waitFor('the turn to end', h.turnEnded); + assert.equal(h.compiles(), 1, 'the compile ran'); + endedAsStop(h, 'subagent_cancel', /stopped before it sent anything/); + const request = h.framework.queryInferenceLogs({ agentName: 'prime' }).entries[0].entry.request as { messages?: unknown[] }; + assert.ok(Array.isArray(request.messages), 'its row records the request the compile made, never sent'); + } finally { + await h.cleanup(); + } + }); + + it("a quiesce's abandon while the hooks run ends the turn, which sends nothing while quiesced", async () => { + const h = await aliveWithoutStream(); + try { + const reached = abandonReached(h.framework); + const quiescing = h.framework.quiesce({ timeoutMs: 50, abandon: true }); + await within('the abandon to reach the turn', reached); + h.release(); + const status = await within('the quiesce to return', quiescing); + assert.equal(status.quiesced, true); + assert.equal(status.unabandonable, undefined, 'the turn was not left unabandonable'); + endedAsStop(h, 'quiesce_abandoned', null); + const quiesce = h.traces.find((t) => t.type === 'host:quiesce') as { unabandonable?: string[] } | undefined; + assert.equal(quiesce?.unabandonable, undefined, 'its trace lists no unabandonable turn'); + await h.framework.resume(); + } finally { + await h.cleanup(); + } + }); + + it("a quiesce's abandon during the compile ends the turn before its stream is read", async () => { + const h = await aliveWithoutStream({ holdCompile: true }); + try { + const reached = abandonReached(h.framework); + const quiescing = h.framework.quiesce({ timeoutMs: 50, abandon: true }); + await within('the abandon to reach the turn', reached); + h.release(); + const status = await within('the quiesce to return', quiescing); + assert.equal(status.unabandonable, undefined, 'the turn was not left unabandonable'); + assert.equal(h.compiles(), 1, 'the compile ran'); + endedAsStop(h, 'quiesce_abandoned', null); + await h.framework.resume(); + } finally { + await h.cleanup(); + } + }); + + it("a caller's stop and a quiesce's abandon at the same start end it as the framework's stop", async () => { + const h = await aliveWithoutStream(); + try { + h.agent.cancelStream(); + const reached = abandonReached(h.framework); + const quiescing = h.framework.quiesce({ timeoutMs: 50, abandon: true }); + await within('the abandon to reach the turn', reached); + h.release(); + const status = await within('the quiesce to return', quiescing); + assert.equal(status.unabandonable, undefined, 'the turn was not left unabandonable'); + endedAsStop(h, 'quiesce_abandoned', null); + await h.framework.resume(); + } finally { + await h.cleanup(); + } + }); + + it("a resume during the abandon's grace leaves the serving host listing nothing, and the turn still ends at its check", async () => { + const h = await aliveWithoutStream(); + try { + const reached = abandonReached(h.framework); + const quiescing = h.framework.quiesce({ timeoutMs: 50, abandon: true }); + await within('the abandon to reach the turn', reached); + await within('the resume', h.framework.resume()); + await within('the quiesce to return', quiescing); + const status = h.framework.getHostModeStatus(); + assert.equal(status.quiesced, false, 'the host is serving'); + assert.equal(status.unabandonable, undefined, 'and lists no turn as unabandonable'); + h.release(); + await waitFor('the turn to end', h.turnEnded); + // The resume doesn't take the abandon back: the hooks' check collects + // the mark, as a stream the abandon cancelled still ends. + endedAsStop(h, 'quiesce_abandoned', null); + } finally { + await h.cleanup(); + } + }); + + it("a quiesce's abandon of a turn whose hooks outlast the abandon's grace lists it, and still ends it when they return", async () => { + const h = await aliveWithoutStream(); + try { + const reached = abandonReached(h.framework); + const quiescing = h.framework.quiesce({ timeoutMs: 50, abandon: true }); + await within('the abandon to reach the turn', reached); + // The abandon waits up to 10 s for the turns it reached to end. + const status = await within('the quiesce to return', quiescing, 15_000); + assert.deepEqual(status.unabandonable, ['prime'], 'the turn still in its hooks is listed'); + const quiesce = h.traces.find((t) => t.type === 'host:quiesce') as { unabandonable?: string[] } | undefined; + assert.deepEqual(quiesce?.unabandonable, ['prime'], 'and so is its trace'); + h.release(); + await waitFor('the turn to end', h.turnEnded); + endedAsStop(h, 'quiesce_abandoned', null); + await h.framework.resume(); + } finally { + await h.cleanup(); + } + }); + + /** `prime`'s turn waiting to retry an inference that failed, with no + * stream, until `delayMs` has passed. The provider answers the retry. */ + async function retryWaiting(delayMs: number) { + const tempDir = mkdtempSync(join(tmpdir(), 'af-stop-without-stream-')); + const requests: ProviderRequest[] = []; + // The provider fails until the framework's error policy decides to retry. + let failing = true; + const adapter: ProviderAdapter = { + name: 'test', + usageCacheConvention: 'cache-excluded', + supportsModel: () => true, + complete: async () => { throw new Error('unexpected complete'); }, + stream: async (request: ProviderRequest): Promise => { + requests.push(request); + if (failing) throw new Error('provider overloaded'); + return { content: [{ type: 'text', text: 'Answered.' }], stopReason: 'end_turn', usage: { inputTokens: 10, outputTokens: 5 }, model: 'test-model' } as ProviderResponse; + }, + }; + const retrying = deferred(); + const framework = await AgentFramework.create({ + storePath: join(tempDir, 'store'), + membrane: new RealMembrane(adapter, { formatter: new NativeFormatter() }), + agents: [{ name: 'prime', model: 'test-model', systemPrompt: 'You are prime.', allowedTools: 'all' }], + modules: [new GatedContextModule()], + syncIntervalMs: 0, + errorPolicy: { + maxRetries: 1, + onInferenceError: (_error, _agentName, attempt) => { + if (attempt > 0) return { retry: false }; + failing = false; + retrying.resolve(); + return { retry: true, delayMs }; + }, + }, + }); + const traces: TraceEvent[] = []; + framework.onTrace((event) => traces.push(event)); + framework.start(); + const agent = framework.getAgent('prime')!; + framework.pushEvent({ type: 'external-message', source: 'test', content: 'hello', metadata: {}, triggerInference: true } as unknown as ProcessEvent); + await within('the failed inference to be retried', retrying.promise); + assert.ok(!('stream' in agent.state) || !agent.state.stream, 'the turn waits with no stream'); + return { + framework, agent, traces, requests, + sentBeforeTheStop: requests.length, + turnEnded: () => framework.getHostModeStatus().activeTurns === 0 && agent.state.status === 'idle', + cleanup: async () => { + await within('stop() in cleanup', framework.stop()).catch(() => {}); + rmSync(tempDir, { recursive: true, force: true }); + }, + }; + } + + it("a stop while the turn waits to retry a failed inference ends it before the retry is sent, writes its row after the stream's, and leaves no check ahead", async () => { + const r = await retryWaiting(300); + try { + r.agent.cancelStream(); + await waitFor('the turn to end', r.turnEnded); + assert.equal(r.requests.length, r.sentBeforeTheStop, 'the retry was not sent'); + assert.ok(!r.traces.some((t) => t.type === 'inference:completed'), 'and nothing was answered'); + assert.deepEqual(abortReasons(r.traces), ['user'], "one inference:aborted, with the stop's reason"); + assert.ok(notes(r.agent).some((n) => /stopped before it finished/.test(n)), "the agent's note says the turn was stopped before it finished"); + // The stop's row comes after the failed stream's own, written through + // the same stream's writer, so both carry the stream's request id. + const rows = r.framework.queryInferenceLogs({ agentName: 'prime' }).entries + .slice().sort((a, b) => a.sequence - b.sequence).map((e) => e.entry); + const stops = stopRows(r.framework, 'prime', 'user'); + assert.equal(rows.length, 2, "the inference log has the failed stream's row and the stop's"); + assert.equal(stops.length, 1, "one is the stop's"); + assert.deepEqual(rows[1], stops[0], "the stop's row comes second"); + assert.equal(rows[0].success, false, "after the failed stream's row"); + assert.equal(rows[1].requestId, rows[0].requestId, "through the stream's writer, under the stream's request id"); + assert.equal(rows[1].tokenUsage, undefined, 'and carries no usage'); + assert.equal(r.framework.abortInference('prime', 'again'), false, 'and no check of the ended turn is ahead'); + await within('stop() to finish', r.framework.stop()); + } finally { + await r.cleanup(); + } + }); + + it('abortInference() while the turn waits to retry a failed inference ends it at once, long before the wait would run out', async () => { + // A minute's wait: the turn ends within the test's bound only if the + // stop ends the wait. + const r = await retryWaiting(60_000); + try { + assert.equal(r.framework.abortInference('prime', 'user'), true, 'abortInference reports the stop'); + await waitFor('the turn to end', r.turnEnded); + assert.equal(r.requests.length, r.sentBeforeTheStop, 'the retry was not sent'); + assert.deepEqual(abortReasons(r.traces), ['user'], "one inference:aborted, with the stop's reason"); + await within('stop() to finish', r.framework.stop()); + } finally { + await r.cleanup(); + } + }); + + it("abortInference() wakes only its own turn's wait to retry: another agent's wait, with no stop, stays asleep", async () => { + const tempDir = mkdtempSync(join(tmpdir(), 'af-stop-without-stream-')); + // Every request fails, and each agent's failed inference waits a minute + // to retry. + const sentBy = { prime: 0, second: 0 }; + const adapter: ProviderAdapter = { + name: 'test', + usageCacheConvention: 'cache-excluded', + supportsModel: () => true, + complete: async () => { throw new Error('unexpected complete'); }, + stream: async (request: ProviderRequest): Promise => { + sentBy[JSON.stringify(request).includes('You are second.') ? 'second' : 'prime']++; + throw new Error('provider overloaded'); + }, + }; + const waiting = { prime: deferred(), second: deferred() }; + const module = new GatedContextModule(); + const framework = await AgentFramework.create({ + storePath: join(tempDir, 'store'), + membrane: new RealMembrane(adapter, { formatter: new NativeFormatter() }), + agents: [ + { name: 'prime', model: 'test-model', systemPrompt: 'You are prime.', allowedTools: 'all' }, + { name: 'second', model: 'test-model', systemPrompt: 'You are second.', allowedTools: 'all' }, + ], + modules: [module], + syncIntervalMs: 0, + errorPolicy: { + maxRetries: 1, + onInferenceError: (_error, agentName, attempt) => { + if (attempt > 0) return { retry: false }; + waiting[agentName as 'prime' | 'second'].resolve(); + return { retry: true, delayMs: 60_000 }; + }, + }, + }); + const traces: TraceEvent[] = []; + framework.onTrace((event) => traces.push(event)); + try { + framework.start(); + for (const name of ['prime', 'second']) { + framework.getAgent(name)!.getContextManager().addMessage('user', [{ type: 'text', text: 'hello' }]); + framework.nudgeAgent(name, 'test'); + } + await within("both agents' failed inferences to be retried", Promise.all([waiting.prime.promise, waiting.second.promise])); + // A retry that started would reach its hooks, which hold. + const nextHooks = module.hold(); + let retryStarted = false; + void nextHooks.entered.then(() => { retryStarted = true; }); + const sentBefore = { ...sentBy }; + const startsOfSecond = () => traces.filter((t) => t.type === 'inference:started' && (t as { agentName?: string }).agentName === 'second').length; + const secondStartsBefore = startsOfSecond(); + const checkAhead = (framework as unknown as { checkAhead: Map }).checkAhead; + const second = framework.getAgent('second')!; + const secondWindow = checkAhead.get(second); + assert.ok(secondWindow, 'second waits to retry, in its window'); + assert.equal(framework.abortInference('prime', 'user'), true, "abortInference reports prime's stop"); + // A wake is a call, and the wait it ends leaves its window in the + // microtasks after it; one delivered on a timer with no delay does so + // before a timer set now fires. + await new Promise((resolve) => setTimeout(resolve, 0)); + assert.equal(checkAhead.get(second), secondWindow, "second's wait wasn't woken: it still holds its window"); + await waitFor("prime's turn to end", () => abortReasons(traces).length === 1); + assert.deepEqual(abortReasons(traces), ['user'], "prime's turn ended as the stop"); + assert.equal(framework.getHostModeStatus().activeTurns, 1, "second's turn is still running"); + assert.equal(startsOfSecond(), secondStartsBefore, "and its retry hasn't started"); + assert.equal(retryStarted, false, "nor reached its hooks"); + assert.deepEqual(sentBy, sentBefore, 'neither retry was sent'); + nextHooks.release(); + await within('stop() to finish', framework.stop()); + } finally { + await within('stop() in cleanup', framework.stop()).catch(() => {}); + rmSync(tempDir, { recursive: true, force: true }); + } + }); + + it("a quiesce's abandon while the turn waits to retry a failed inference ends it, and the retry isn't sent while quiesced", async () => { + // The wait has no end: the turn ends within the abandon's 10 s grace + // only if the abandon ends the wait, and the quiesce's answer says so. + const r = await retryWaiting(Number.POSITIVE_INFINITY); + try { + const quiescing = r.framework.quiesce({ timeoutMs: 50, abandon: true }); + const status = await within('the quiesce to return', quiescing, 15_000); + assert.equal(status.quiesced, true); + assert.equal(status.unabandonable, undefined, 'the turn was not left unabandonable'); + assert.ok(r.turnEnded(), 'the turn has ended'); + assert.equal(r.requests.length, r.sentBeforeTheStop, 'the retry was not sent'); + assert.ok(!r.traces.some((t) => t.type === 'inference:completed'), 'and nothing was answered'); + assert.deepEqual(abortReasons(r.traces), ['quiesce_abandoned'], "one inference:aborted, with the abandon's reason"); + assert.deepEqual(notes(r.agent), [], 'no note'); + await r.framework.resume(); + } finally { + await r.cleanup(); + } + }); + + /** A turn whose first inference fails, with a policy that retries at + * once, and a listener of that inference:failed. */ + async function failedInferenceWithListener(onFailed: (framework: AgentFramework, agent: Agent) => unknown) { + const tempDir = mkdtempSync(join(tmpdir(), 'af-stop-without-stream-')); + const requests: ProviderRequest[] = []; + let failing = true; + const adapter: ProviderAdapter = { + name: 'test', + usageCacheConvention: 'cache-excluded', + supportsModel: () => true, + complete: async () => { throw new Error('unexpected complete'); }, + stream: async (request: ProviderRequest): Promise => { + requests.push(request); + if (failing) throw new Error('provider overloaded'); + return { content: [{ type: 'text', text: 'Answered.' }], stopReason: 'end_turn', usage: { inputTokens: 10, outputTokens: 5 }, model: 'test-model' } as ProviderResponse; + }, + }; + const policy = { asked: 0 }; + const framework = await AgentFramework.create({ + storePath: join(tempDir, 'store'), + membrane: new RealMembrane(adapter, { formatter: new NativeFormatter() }), + agents: [{ name: 'prime', model: 'test-model', systemPrompt: 'You are prime.', allowedTools: 'all' }], + modules: [new GatedContextModule()], + syncIntervalMs: 0, + errorPolicy: { + maxRetries: 1, + onInferenceError: (_error, _agentName, attempt) => { + policy.asked++; + if (attempt > 0) return { retry: false }; + failing = false; + return { retry: true, delayMs: 0 }; + }, + }, + }); + const traces: TraceEvent[] = []; + const agent = framework.getAgent('prime')!; + // A host's listener, called in the trace's own delivery. + let answer: unknown; + let called = false; + framework.onTrace((event) => { + traces.push(event); + if (event.type === 'inference:failed' && !called) { called = true; answer = onFailed(framework, agent); } + }); + framework.start(); + framework.pushEvent({ type: 'external-message', source: 'test', content: 'hello', metadata: {}, triggerInference: true } as unknown as ProcessEvent); + await waitFor('the turn to end', () => traces.some((t) => t.type === 'inference:aborted' || t.type === 'inference:completed') + && framework.getHostModeStatus().activeTurns === 0 && agent.state.status === 'idle'); + return { + framework, agent, traces, requests, policy, answer: () => answer, + cleanup: async () => { + await within('stop() in cleanup', framework.stop()).catch(() => {}); + rmSync(tempDir, { recursive: true, force: true }); + }, + }; + } + + it("abortInference() from a listener of a failed inference's inference:failed reports false: the agent holds no stream and no check is ahead yet", async () => { + const r = await failedInferenceWithListener((framework) => framework.abortInference('prime', 'user')); + try { + assert.equal(r.answer(), false, 'abortInference took no stop'); + assert.equal(r.requests.length, 2, 'the retry was sent'); + assert.ok(r.traces.some((t) => t.type === 'inference:completed'), 'and answered'); + assert.deepEqual(abortReasons(r.traces), [], 'nothing was aborted'); + } finally { + await r.cleanup(); + } + }); + + it("cancelStream() from a listener of a failed inference's inference:failed ends the turn at the check after the wait to retry, before the retry is sent", async () => { + const r = await failedInferenceWithListener((_framework, agent) => agent.cancelStream()); + try { + assert.equal(r.requests.length, 1, 'the retry was not sent'); + assert.equal(r.policy.asked, 1, 'the error policy decided to retry'); + assert.ok(!r.traces.some((t) => t.type === 'inference:completed'), 'nothing was answered'); + assert.deepEqual(abortReasons(r.traces), ['user'], "one inference:aborted, with the stop's reason"); + assert.ok(notes(r.agent).some((n) => /stopped before it finished/.test(n)), "the agent's note says the turn was stopped before it finished"); + // The stop's row comes after the failed stream's own, through the same + // stream's writer. + const rows = r.framework.queryInferenceLogs({ agentName: 'prime' }).entries + .slice().sort((a, b) => a.sequence - b.sequence).map((e) => e.entry); + assert.equal(rows.length, 2, "the inference log has the failed stream's row and the stop's"); + assert.equal(rows[1].error, stopRows(r.framework, 'prime', 'user')[0]?.error, "the stop's row comes second"); + assert.equal(rows[1].requestId, rows[0].requestId, "through the stream's writer, under the stream's request id"); + } finally { + await r.cleanup(); + } + }); + + it("abortInference() from a listener of the inference:failed of a stream that threw reports false, and the turn ends as exhausted", async () => { + const tempDir = mkdtempSync(join(tmpdir(), 'af-stop-without-stream-')); + // The stream throws as it is read, rather than reporting an error event. + class ThrowingMembrane extends MockMembrane { + override streamYielding(request: NormalizedRequest): YieldingStream { + this.calls.push(request); + const stream = new MockYieldingStream([]); + Object.defineProperty(stream, Symbol.asyncIterator, { value: () => ({ next: async () => { throw new Error('the stream threw'); } }) }); + return stream; + } + } + const membrane = new ThrowingMembrane(); + const framework = await AgentFramework.create({ + storePath: join(tempDir, 'store'), membrane: membrane.asMembrane(), + agents: [{ name: 'prime', model: 'test-model', systemPrompt: 'You are prime.', allowedTools: 'all' }], + modules: [new GatedContextModule()], syncIntervalMs: 0, + }); + const traces: TraceEvent[] = []; + let taken: boolean | undefined; + framework.onTrace((event) => { + traces.push(event); + if (event.type === 'inference:failed' && taken === undefined) taken = framework.abortInference('prime', 'user'); + }); + try { + framework.start(); + framework.pushEvent({ type: 'external-message', source: 'test', content: 'hello', metadata: {}, triggerInference: true } as unknown as ProcessEvent); + await waitFor('the turn to end', () => traces.some((t) => t.type === 'inference:exhausted' || t.type === 'inference:aborted') + && framework.getHostModeStatus().activeTurns === 0); + assert.equal(taken, false, 'abortInference took no stop'); + assert.ok(traces.some((t) => t.type === 'inference:exhausted'), 'the turn ended as exhausted'); + assert.deepEqual(abortReasons(traces), [], 'not as a stop'); + await within('stop() to finish', framework.stop()); + } finally { + await within('stop() in cleanup', framework.stop()).catch(() => {}); + rmSync(tempDir, { recursive: true, force: true }); + } + }); + + it('a stop while the compile runs ends the turn when the compile fails, before the error policy waits to retry it', async () => { + const tempDir = mkdtempSync(join(tmpdir(), 'af-stop-without-stream-')); + const { membrane, requests } = answeringMembrane(); + const framework = await AgentFramework.create({ + storePath: join(tempDir, 'store'), membrane, + agents: [{ name: 'prime', model: 'test-model', systemPrompt: 'You are prime.', allowedTools: 'all' }], + modules: [new GatedContextModule()], syncIntervalMs: 0, + // A minute's wait to retry: the turn ends within the test's bound only + // if the stop ends it at the failure, before the wait. + errorPolicy: { maxRetries: 1, onInferenceError: (_error, _agentName, attempt) => (attempt > 0 ? { retry: false } : { retry: true, delayMs: 60_000 }) }, + }); + const traces: TraceEvent[] = []; + framework.onTrace((event) => traces.push(event)); + const agent = framework.getAgent('prime')!; + // The first compile waits at its await on the guard's annotations, and + // then fails. + let compiles = 0; + const compileEntered = deferred(); + const compileReleased = deferred(); + const guard = agent.toolResultGuard; + const annotations = guard.afterAnnotations.bind(guard); + guard.afterAnnotations = (async (read: () => Promise) => { + if (++compiles === 1) { + compileEntered.resolve(); + await compileReleased.promise; + throw new Error('the compile failed'); + } + return annotations(read); + }) as typeof guard.afterAnnotations; + try { + framework.start(); + framework.pushEvent({ type: 'external-message', source: 'test', content: 'hello', metadata: {}, triggerInference: true } as unknown as ProcessEvent); + await within("the turn's compile", compileEntered.promise); + agent.cancelStream(); + compileReleased.resolve(); + await waitFor('the turn to end', () => traces.some((t) => t.type === 'inference:aborted' || t.type === 'inference:completed') + && framework.getHostModeStatus().activeTurns === 0); + assert.ok(traces.some((t) => t.type === 'inference:failed'), "the compile's failure was reported"); + assert.equal(compiles, 1, 'no retry compiled'); + assert.equal(requests.length, 0, 'nothing was sent'); + assert.deepEqual(abortReasons(traces), ['user'], "one inference:aborted, with the stop's reason"); + assert.ok(!traces.some((t) => t.type === 'inference:exhausted'), 'the turn ended as a stop, not as exhausted'); + assert.ok(notes(agent).some((n) => /stopped before it sent anything/.test(n)), "the agent's note says the turn was stopped before it sent anything"); + await within('stop() to finish', framework.stop()); + } finally { + compileReleased.resolve(); + await within('stop() in cleanup', framework.stop()).catch(() => {}); + rmSync(tempDir, { recursive: true, force: true }); + } + }); + + it("abortInference() from a listener of a failed compile's inference:failed reports the stop, which ends the turn there, before the retry", async () => { + const tempDir = mkdtempSync(join(tmpdir(), 'af-stop-without-stream-')); + const { membrane, requests } = answeringMembrane(); + const framework = await AgentFramework.create({ + storePath: join(tempDir, 'store'), membrane, + agents: [{ name: 'prime', model: 'test-model', systemPrompt: 'You are prime.', allowedTools: 'all' }], + modules: [new GatedContextModule()], syncIntervalMs: 0, + errorPolicy: { maxRetries: 1, onInferenceError: (_error, _agentName, attempt) => (attempt > 0 ? { retry: false } : { retry: true, delayMs: 60_000 }) }, + }); + const traces: TraceEvent[] = []; + // A host's listener, called in the report's own delivery. The start's + // check is still ahead of it. + let taken: boolean | undefined; + framework.onTrace((event) => { + traces.push(event); + if (event.type === 'inference:failed' && taken === undefined) taken = framework.abortInference('prime', 'user'); + }); + const agent = framework.getAgent('prime')!; + let compiles = 0; + const guard = agent.toolResultGuard; + const annotations = guard.afterAnnotations.bind(guard); + guard.afterAnnotations = (async (read: () => Promise) => { + if (++compiles === 1) throw new Error('the compile failed'); + return annotations(read); + }) as typeof guard.afterAnnotations; + try { + framework.start(); + framework.pushEvent({ type: 'external-message', source: 'test', content: 'hello', metadata: {}, triggerInference: true } as unknown as ProcessEvent); + await waitFor('the turn to end', () => traces.some((t) => t.type === 'inference:aborted' || t.type === 'inference:completed') + && framework.getHostModeStatus().activeTurns === 0); + assert.equal(taken, true, 'abortInference reports the stop'); + assert.equal(compiles, 1, 'no retry compiled'); + assert.equal(requests.length, 0, 'nothing was sent'); + assert.deepEqual(abortReasons(traces), ['user'], "one inference:aborted, with the stop's reason"); + assert.ok(!traces.some((t) => t.type === 'inference:exhausted'), 'the turn ended as a stop, not as exhausted'); + assert.ok(notes(agent).some((n) => /stopped before it sent anything/.test(n)), "the agent's note says the turn was stopped before it sent anything"); + await within('stop() to finish', framework.stop()); + } finally { + await within('stop() in cleanup', framework.stop()).catch(() => {}); + rmSync(tempDir, { recursive: true, force: true }); + } + }); + + for (const stopDuringWait of [true, false]) it(`a start waiting to retry a failed compile across a stop() and a start() is dropped before it sends anything${stopDuringWait ? ', with a stop made during the wait' : ''}`, async () => { + const tempDir = mkdtempSync(join(tmpdir(), 'af-stop-without-stream-')); + const { membrane, requests } = answeringMembrane(); + const retrying = deferred(); + const framework = await AgentFramework.create({ + storePath: join(tempDir, 'store'), membrane, + agents: [{ name: 'prime', model: 'test-model', systemPrompt: 'You are prime.', allowedTools: 'all' }], + modules: [new GatedContextModule()], syncIntervalMs: 0, + errorPolicy: { + maxRetries: 1, + onInferenceError: (_error, _agentName, attempt) => { + if (attempt > 0) return { retry: false }; + retrying.resolve(); + // A stop made during the wait wakes it; otherwise it runs out. + return { retry: true, delayMs: stopDuringWait ? 60_000 : 300 }; + }, + }, + }); + const traces: TraceEvent[] = []; + framework.onTrace((event) => traces.push(event)); + const agent = framework.getAgent('prime')!; + // The first compile fails after a stop() and a start(), which each + // advance the lifecycle and leave admission open again; no stop is kept + // for the turn by then. + let compiles = 0; + const guard = agent.toolResultGuard; + const annotations = guard.afterAnnotations.bind(guard); + guard.afterAnnotations = (async (read: () => Promise) => { + if (++compiles === 1) { + (framework as unknown as { lifecycleGeneration: number }).lifecycleGeneration += 2; + throw new Error('the compile failed'); + } + return annotations(read); + }) as typeof guard.afterAnnotations; + try { + framework.start(); + framework.pushEvent({ type: 'external-message', source: 'test', content: 'hello', metadata: {}, triggerInference: true } as unknown as ProcessEvent); + await within('the failed compile to be retried', retrying.promise); + if (stopDuringWait) assert.equal(framework.abortInference('prime', 'user'), true, 'abortInference reports the stop'); + await waitFor('the turn to end', () => traces.some((t) => t.type === 'inference:aborted' || t.type === 'inference:completed' || t.type === 'inference:exhausted') + && framework.getHostModeStatus().activeTurns === 0); + assert.deepEqual(abortReasons(traces), ['a retry waiting while the framework was stopped and started again is dropped before it sends anything'], 'the retry was dropped'); + assert.equal(compiles, 1, 'no retry compiled'); + assert.equal(requests.length, 0, 'nothing was sent'); + assert.ok(!traces.some((t) => t.type === 'inference:completed' || t.type === 'inference:exhausted'), 'no other terminal'); + assert.deepEqual(notes(agent), [], 'and no [turn-interrupted] note was written'); + } finally { + await within('stop() in cleanup', framework.stop()).catch(() => {}); + rmSync(tempDir, { recursive: true, force: true }); + } + }); + + for (const how of ['stopping', 'replaced', 'restarted'] as const) it(`a stop during a compile that then fails ${how === 'stopping' ? 'while stop() runs' : how === 'replaced' ? 'after its agent was replaced' : 'after the framework was stopped and started again'} drops the turn as ${how === 'stopping' ? "stop()'s" : how === 'replaced' ? "the replacement's" : "the earlier lifecycle's"}, before it sends anything, with no note`, async () => { + const tempDir = mkdtempSync(join(tmpdir(), 'af-stop-without-stream-')); + const { membrane, requests } = answeringMembrane(); + const framework = await AgentFramework.create({ + storePath: join(tempDir, 'store'), membrane, + agents: [{ name: 'prime', model: 'test-model', systemPrompt: 'You are prime.', allowedTools: 'all' }], + modules: [new GatedContextModule()], syncIntervalMs: 0, + gate: { config: { policies: [], default: 'always' } }, + errorPolicy: { maxRetries: 1, onInferenceError: (_error, _agentName, attempt) => (attempt > 0 ? { retry: false } : { retry: true, delayMs: 60_000 }) }, + }); + const gate = (framework as unknown as { eventGate: { inferenceDiagnostics(): { inferring: string[] } } }).eventGate; + const traces: TraceEvent[] = []; + // stop() disposes the gate, which clears it: the gate is read as the + // drop is reported, after the drop's own release. + let inferringAtDrop: string[] | undefined; + framework.onTrace((event) => { + traces.push(event); + if (event.type === 'inference:aborted' && inferringAtDrop === undefined) inferringAtDrop = [...gate.inferenceDiagnostics().inferring]; + }); + const agent = framework.getAgent('prime')!; + const noteWrites: string[] = []; + // The first compile waits at its await on the guard's annotations, and + // then fails. + let compiles = 0; + const compileEntered = deferred(); + const compileReleased = deferred(); + const guard = agent.toolResultGuard; + const annotations = guard.afterAnnotations.bind(guard); + guard.afterAnnotations = (async (read: () => Promise) => { + if (++compiles === 1) { + compileEntered.resolve(); + await compileReleased.promise; + throw new Error('the compile failed'); + } + return annotations(read); + }) as typeof guard.afterAnnotations; + // The store closes with stop(), so a note is caught as it is written. + const cm = agent.getContextManager(); + const addMessage = cm.addMessage.bind(cm); + cm.addMessage = ((participant: string, content: Array<{ type: string; text?: string }>, ...rest: unknown[]) => { + for (const block of content) if (block.type === 'text' && (block.text ?? '').startsWith('[turn-interrupted]')) noteWrites.push(block.text!); + return (addMessage as (...args: unknown[]) => unknown)(participant, content, ...rest); + }) as typeof cm.addMessage; + try { + framework.start(); + framework.pushEvent({ type: 'external-message', source: 'test', content: 'hello', metadata: {}, triggerInference: true } as unknown as ProcessEvent); + await within("the turn's compile", compileEntered.promise); + agent.cancelStream(); + let stopping: Promise | undefined; + if (how === 'stopping') { + stopping = framework.stop(); + await waitFor('stop() to close provider admission', () => (framework as unknown as { providerAdmissionClosed: boolean }).providerAdmissionClosed); + } else if (how === 'replaced') { + // As a replacement under the agent's name does. + (framework as unknown as { agents: Map }).agents.set('prime', Object.create(agent)); + } else { + // As a stop() and a start() do: each advances the lifecycle, and + // admission is open again. + (framework as unknown as { lifecycleGeneration: number }).lifecycleGeneration += 2; + } + compileReleased.resolve(); + await waitFor('the turn to end', () => traces.some((t) => t.type === 'inference:aborted' || t.type === 'inference:exhausted')); + if (stopping) await within('stop() to finish', stopping); + const why = how === 'stopping' ? 'the framework was stopping' : how === 'replaced' ? 'the agent was replaced or removed' : 'the framework was stopped and started again'; + assert.ok(traces.some((t) => t.type === 'inference:failed'), "the compile's failure was reported"); + assert.equal(compiles, 1, 'no retry compiled'); + assert.equal(requests.length, 0, 'nothing was sent'); + assert.deepEqual(abortReasons(traces), [`a turn whose request couldn't be built while ${why} is dropped before it sends anything`], 'the turn was dropped'); + assert.ok(!traces.some((t) => t.type === 'inference:exhausted'), 'not exhausted'); + assert.deepEqual(noteWrites, [], 'and no [turn-interrupted] note was written'); + assert.deepEqual(inferringAtDrop, [], 'and the drop closed the turn at the event gate'); + } finally { + compileReleased.resolve(); + await within('stop() in cleanup', framework.stop()).catch(() => {}); + rmSync(tempDir, { recursive: true, force: true }); + } + }); + + it("abortInference() reports a stop while the retry of a failed compile waits to retry its own failed stream, and the turn ends there", async () => { + const tempDir = mkdtempSync(join(tmpdir(), 'af-stop-without-stream-')); + // The first stream fails as soon as it's read. + class FailingFirstStream extends MockMembrane { + override streamYielding(request: NormalizedRequest): YieldingStream { + this.calls.push(request); + if (this.calls.length === 1) return new ErrorStream(new Error('provider overloaded')); + return new MockYieldingStream([createMockResponse([{ type: 'text', text: 'Answered.' }])]); + } + } + const membrane = new FailingFirstStream(); + const retrying = deferred(); + const framework = await AgentFramework.create({ + storePath: join(tempDir, 'store'), membrane: membrane.asMembrane(), + agents: [{ name: 'prime', model: 'test-model', systemPrompt: 'You are prime.', allowedTools: 'all' }], + modules: [new GatedContextModule()], syncIntervalMs: 0, + // The failed compile is retried at once; the retry's failed stream + // waits to retry. + errorPolicy: { + maxRetries: 2, + onInferenceError: (_error, _agentName, attempt) => { + if (attempt === 0) return { retry: true, delayMs: 0 }; + if (attempt === 1) { retrying.resolve(); return { retry: true, delayMs: 300 }; } + return { retry: false }; + }, + }, + }); + const traces: TraceEvent[] = []; + framework.onTrace((event) => traces.push(event)); + const agent = framework.getAgent('prime')!; + let compiles = 0; + const guard = agent.toolResultGuard; + const annotations = guard.afterAnnotations.bind(guard); + guard.afterAnnotations = (async (read: () => Promise) => { + if (++compiles === 1) throw new Error('the compile failed'); + return annotations(read); + }) as typeof guard.afterAnnotations; + try { + framework.start(); + framework.pushEvent({ type: 'external-message', source: 'test', content: 'hello', metadata: {}, triggerInference: true } as unknown as ProcessEvent); + // The wait's window opened as the policy decided. By the time the test + // resumes here, with no timer run since, the starts that retried the + // compile have ended: a start that closed the agent's window, rather + // than only its own, has closed the wait's by now, and the stop isn't + // taken. + await within("the retry's failed stream to be retried", retrying.promise); + assert.equal(framework.abortInference('prime', 'user'), true, 'abortInference reports the stop'); + await waitFor('the turn to end', () => traces.some((t) => t.type === 'inference:aborted' || t.type === 'inference:completed') + && framework.getHostModeStatus().activeTurns === 0); + assert.equal(membrane.calls.length, 1, "the stream's retry was not sent"); + assert.deepEqual(abortReasons(traces), ['user'], "one inference:aborted, with the stop's reason"); + await within('stop() to finish', framework.stop()); + } finally { + await within('stop() in cleanup', framework.stop()).catch(() => {}); + rmSync(tempDir, { recursive: true, force: true }); + } + }); + + it('a stop while the compile runs ends the turn when the compile fails, before a provider hold can park it', async () => { + const tempDir = mkdtempSync(join(tmpdir(), 'af-stop-without-stream-')); + const { membrane, requests } = answeringMembrane(); + // The host would hold the turn on the compile's failure: without the stop, + // the hold parks it, and its start runs when the hold ends. + let holdsAsked = 0; + const framework = await AgentFramework.create({ + storePath: join(tempDir, 'store'), membrane, + agents: [{ name: 'prime', model: 'test-model', systemPrompt: 'You are prime.', allowedTools: 'all' }], + modules: [new GatedContextModule()], syncIntervalMs: 0, maintenanceIntervalMs: 0, + providerHold: () => (holdsAsked++ === 0 ? { holdMs: 300, reason: 'quota spent' } : undefined), + }); + const traces: TraceEvent[] = []; + framework.onTrace((event) => traces.push(event)); + const agent = framework.getAgent('prime')!; + // The first compile waits at its await on the guard's annotations, and + // then fails. + let compiles = 0; + const compileEntered = deferred(); + const compileReleased = deferred(); + const guard = agent.toolResultGuard; + const annotations = guard.afterAnnotations.bind(guard); + guard.afterAnnotations = (async (read: () => Promise) => { + if (++compiles === 1) { + compileEntered.resolve(); + await compileReleased.promise; + throw new Error('quota spent'); + } + return annotations(read); + }) as typeof guard.afterAnnotations; + try { + framework.start(); + framework.pushEvent({ type: 'external-message', source: 'test', content: 'hello', metadata: {}, triggerInference: true } as unknown as ProcessEvent); + await within("the turn's compile", compileEntered.promise); + agent.cancelStream(); + compileReleased.resolve(); + await waitFor('the turn to end', () => traces.some((t) => t.type === 'inference:aborted' || t.type === 'inference:completed') + && framework.getHostModeStatus().activeTurns === 0); + assert.deepEqual(abortReasons(traces), ['user'], "one inference:aborted, with the stop's reason"); + assert.equal(holdsAsked, 0, 'no provider hold was asked for'); + assert.equal(compiles, 1, 'the turn was not compiled again'); + assert.equal(requests.length, 0, 'nothing was sent'); + await within('stop() to finish', framework.stop()); + } finally { + compileReleased.resolve(); + await within('stop() in cleanup', framework.stop()).catch(() => {}); + rmSync(tempDir, { recursive: true, force: true }); + } + }); + + it("a quiesce's abandon during a compile that then fails ends the turn at the failure, and the agent's next turn runs", async () => { + const tempDir = mkdtempSync(join(tmpdir(), 'af-stop-without-stream-')); + const { membrane, requests } = answeringMembrane(); + const framework = await AgentFramework.create({ + storePath: join(tempDir, 'store'), membrane, + agents: [{ name: 'prime', model: 'test-model', systemPrompt: 'You are prime.', allowedTools: 'all' }], + modules: [new GatedContextModule()], syncIntervalMs: 0, + errorPolicy: { maxRetries: 0, onInferenceError: () => ({ retry: false }) }, + }); + const traces: TraceEvent[] = []; + framework.onTrace((event) => traces.push(event)); + const agent = framework.getAgent('prime')!; + // The first compile waits at its await on the guard's annotations, and + // then fails, short of its check after the compile. + let compiles = 0; + const compileEntered = deferred(); + const compileReleased = deferred(); + const guard = agent.toolResultGuard; + const annotations = guard.afterAnnotations.bind(guard); + guard.afterAnnotations = (async (read: () => Promise) => { + if (++compiles === 1) { + compileEntered.resolve(); + await compileReleased.promise; + throw new Error('the compile failed'); + } + return annotations(read); + }) as typeof guard.afterAnnotations; + try { + framework.start(); + framework.pushEvent({ type: 'external-message', source: 'test', content: 'hello', metadata: {}, triggerInference: true } as unknown as ProcessEvent); + await within("the turn's compile", compileEntered.promise); + const reached = abandonReached(framework); + const quiescing = framework.quiesce({ timeoutMs: 50, abandon: true }); + await within('the abandon to reach the turn', reached); + compileReleased.resolve(); + const status = await within('the quiesce to return', quiescing); + assert.deepEqual(abortReasons(traces), ['quiesce_abandoned'], "the turn ended at its failure, as the abandon's stop"); + assert.ok(!traces.some((t) => t.type === 'inference:exhausted'), 'not as exhausted'); + assert.equal(status.unabandonable, undefined, 'and the abandon left nothing behind'); + await framework.resume(); + framework.pushEvent({ type: 'external-message', source: 'test', content: 'again', metadata: {}, triggerInference: true } as unknown as ProcessEvent); + await waitFor('the next turn to be answered', () => traces.some((t) => t.type === 'inference:completed')); + assert.equal(requests.length, 1, 'the next turn sent its request'); + assert.deepEqual(abortReasons(traces), ['quiesce_abandoned'], 'and nothing ended it as a stop'); + await within('stop() to finish', framework.stop()); + } finally { + compileReleased.resolve(); + await within('stop() in cleanup', framework.stop()).catch(() => {}); + rmSync(tempDir, { recursive: true, force: true }); + } + }); + + it('a stop while a failed compile waits to retry ends the turn before the retry compiles', async () => { + const tempDir = mkdtempSync(join(tmpdir(), 'af-stop-without-stream-')); + const { membrane, requests } = answeringMembrane(); + const retrying = deferred(); + const framework = await AgentFramework.create({ + storePath: join(tempDir, 'store'), membrane, + agents: [{ name: 'prime', model: 'test-model', systemPrompt: 'You are prime.', allowedTools: 'all' }], + modules: [new GatedContextModule()], syncIntervalMs: 0, + errorPolicy: { + maxRetries: 1, + onInferenceError: (_error, _agentName, attempt) => { + if (attempt > 0) return { retry: false }; + retrying.resolve(); + return { retry: true, delayMs: 300 }; + }, + }, + }); + const traces: TraceEvent[] = []; + framework.onTrace((event) => traces.push(event)); + const agent = framework.getAgent('prime')!; + // The first compile fails, and the error policy waits to retry it. + let compiles = 0; + const guard = agent.toolResultGuard; + const annotations = guard.afterAnnotations.bind(guard); + guard.afterAnnotations = (async (read: () => Promise) => { + if (++compiles === 1) throw new Error('the compile failed'); + return annotations(read); + }) as typeof guard.afterAnnotations; + try { + framework.start(); + framework.pushEvent({ type: 'external-message', source: 'test', content: 'hello', metadata: {}, triggerInference: true } as unknown as ProcessEvent); + await within('the failed compile to be retried', retrying.promise); + agent.cancelStream(); + await waitFor('the turn to end', () => traces.some((t) => t.type === 'inference:aborted' || t.type === 'inference:completed') + && framework.getHostModeStatus().activeTurns === 0); + assert.equal(compiles, 1, "the retry's compile never ran"); + assert.equal(requests.length, 0, 'nothing was sent'); + assert.deepEqual(abortReasons(traces), ['user'], "one inference:aborted, with the stop's reason"); + assert.ok(notes(agent).some((n) => /stopped before it sent anything/.test(n)), "the agent's note says the turn was stopped before it sent anything"); + await within('stop() to finish', framework.stop()); + } finally { + await within('stop() in cleanup', framework.stop()).catch(() => {}); + rmSync(tempDir, { recursive: true, force: true }); + } + }); + + it('an ephemeral run stopped before its stream rejects with the stop, long before its watchdog', async () => { + const tempDir = mkdtempSync(join(tmpdir(), 'af-stop-without-stream-')); + const { membrane, requests } = answeringMembrane(); + const module = new GatedContextModule(); + const framework = await AgentFramework.create({ + storePath: join(tempDir, 'store'), membrane, agents: [], modules: [module], syncIntervalMs: 0, + }); + const traces: TraceEvent[] = []; + framework.onTrace((event) => traces.push(event)); + try { + const { agent, contextManager } = await framework.createEphemeralAgent({ + name: 'worker', model: 'test-model', systemPrompt: 'Do the task.', allowedTools: 'all', + }); + contextManager.addMessage('user', [{ type: 'text', text: 'go' }]); + const hooks = module.hold(); + // A watchdog far past this test's own bound: a run that settles within + // the bound was settled by the stop, not by the watchdog. + const run = framework.runEphemeralToCompletion(agent, contextManager, { idleTimeoutMs: 600_000, idlePollMs: 50 }); + const outcome = run.then(() => 'resolved', (error: unknown) => `rejected: ${error instanceof Error ? error.message : String(error)}`); + framework.start(); + await within("the run's hooks", hooks.entered); + agent.cancelStream('subagent_cancel'); + hooks.release(); + assert.equal(await within('the run to settle', outcome, 10_000), 'rejected: Stream cancelled', 'the run rejects with the stop'); + assert.equal(requests.length, 0, 'nothing was sent'); + assert.deepEqual(abortReasons(traces), ['subagent_cancel'], "one inference:aborted, with the stop's reason"); + await within('stop() to finish', framework.stop()); + } finally { + await within('stop() in cleanup', framework.stop()).catch(() => {}); + rmSync(tempDir, { recursive: true, force: true }); + } + }); +}); + +describe('a stop while no turn is running', () => { + it("ends nothing: the agent's next turn runs, as after a host's interrupt of every agent", async () => { + const tempDir = mkdtempSync(join(tmpdir(), 'af-stop-without-stream-')); + const { membrane, requests } = answeringMembrane(); + const framework = await AgentFramework.create({ + storePath: join(tempDir, 'store'), + membrane, + agents: [{ name: 'prime', model: 'test-model', systemPrompt: 'You are prime.', allowedTools: 'all' }], + modules: [new GatedContextModule()], + syncIntervalMs: 0, + }); + const traces: TraceEvent[] = []; + framework.onTrace((event) => traces.push(event)); + try { + framework.start(); + const agent = framework.getAgent('prime')!; + agent.cancelStream(); + assert.equal(framework.abortInference('prime', 'operator stop'), false, 'with no turn running, nothing is aborted'); + framework.pushEvent({ type: 'external-message', source: 'test', content: 'hello', metadata: {}, triggerInference: true } as unknown as ProcessEvent); + await waitFor('the turn to be answered', () => traces.some((t) => t.type === 'inference:completed')); + assert.equal(requests.length, 1, 'the turn sent its request'); + assert.deepEqual(abortReasons(traces), [], 'nothing was aborted'); + assert.deepEqual(notes(agent), [], 'no note'); + await within('stop() to finish', framework.stop()); + } finally { + await within('stop() in cleanup', framework.stop()).catch(() => {}); + rmSync(tempDir, { recursive: true, force: true }); + } + }); + + it('a message sent after a stop during a provider-capacity hold is answered when the hold ends', async () => { + // A host hold parks the turn after its first request fails on a spent + // quota: no turn is running while it waits, and when it ends, one start + // takes every wake held for the agent. A stop made during the hold must + // not cost the message sent after it its answer. + const tempDir = mkdtempSync(join(tmpdir(), 'af-stop-without-stream-')); + const membrane = new QuotaOnceMembrane(); + let holds = 0; + const framework = await AgentFramework.create({ + storePath: join(tempDir, 'store'), + membrane: membrane.asMembrane(), + agents: [{ name: 'prime', model: 'test-model', systemPrompt: 'You are prime.', allowedTools: 'all' }], + modules: [new GatedContextModule()], + syncIntervalMs: 0, + maintenanceIntervalMs: 0, + providerHold: () => (holds++ === 0 ? { holdMs: 500, reason: 'quota spent' } : undefined), + }); + const traces: TraceEvent[] = []; + framework.onTrace((event) => traces.push(event)); + const cooldowns = (framework as unknown as { providerAccelerationCooldowns: Map }).providerAccelerationCooldowns; + try { + framework.start(); + const agent = framework.getAgent('prime')!; + framework.pushEvent({ type: 'external-message', source: 'test', content: 'hello', metadata: {}, triggerInference: true } as unknown as ProcessEvent); + await waitFor('the hold to park the turn', () => cooldowns.has('prime') && framework.getHostModeStatus().activeTurns === 0); + assert.equal(membrane.calls.length, 1, 'the first request failed on the quota'); + assert.equal(framework.abortInference('prime', 'user'), false, 'abortInference() reports no turn to stop while the hold parks it'); + agent.cancelStream(); + framework.pushEvent({ type: 'external-message', source: 'test', content: 'a message sent after the stop', metadata: {}, triggerInference: true } as unknown as ProcessEvent); + await waitFor('the message sent after the stop to be answered', + () => traces.some((t) => t.type === 'inference:completed'), 10_000); + assert.ok(membrane.calls.some((request) => JSON.stringify(request.messages).includes('a message sent after the stop')), + 'a request carrying the message sent after the stop reached the provider'); + await within('stop() to finish', framework.stop()); + } finally { + await within('stop() in cleanup', framework.stop()).catch(() => {}); + rmSync(tempDir, { recursive: true, force: true }); + } + }); + + it("a stop made while no check is ahead doesn't end a restart queued later, as a hold replays the trigger it held", async () => { + const tempDir = mkdtempSync(join(tmpdir(), 'af-stop-without-stream-')); + const { membrane, requests } = answeringMembrane(); + const framework = await AgentFramework.create({ + storePath: join(tempDir, 'store'), membrane, + agents: [{ name: 'prime', model: 'test-model', systemPrompt: 'You are prime.', allowedTools: 'all' }], + modules: [new GatedContextModule()], syncIntervalMs: 0, + }); + const traces: TraceEvent[] = []; + framework.onTrace((event) => traces.push(event)); + try { + framework.start(); + framework.getAgent('prime')!.cancelStream(); + // A budget restart's trigger queued with no check of a turn ahead, as + // a provider hold's release queues the trigger it held, whatever its + // reason. + queueRestart(framework); + await waitFor('the restart to be answered', () => traces.some((t) => t.type === 'inference:completed')); + assert.equal(requests.length, 1, 'the restart sent its request'); + assert.deepEqual(abortReasons(traces), [], 'nothing was aborted'); + await within('stop() to finish', framework.stop()); + } finally { + await within('stop() in cleanup', framework.stop()).catch(() => {}); + rmSync(tempDir, { recursive: true, force: true }); + } + }); +}); + +// The guard's refusal retry: a round whose result the provider refuses is +// withheld, and the turn is retried once the agent's earlier speech has gone +// out. A channel that holds the round's post keeps the retry waiting. + +const answer = (text: string) => createMockResponse([{ type: 'text', text }]); +const refused = () => ({ + ...createMockResponse([{ type: 'text', text: 'discard-this-partial-output' }], 'refusal'), + raw: { request: {}, response: { stop_details: { category: 'test-category' } } }, +}) as NormalizedResponse; +const postThenRead = () => createMockResponse([ + { type: 'text', text: '>>#cafe Reading.' }, + { type: 'tool_use', id: 'one', name: 'gated--read', input: {} }, +] as ContentBlock[], 'tool_use'); + +/** Scripted streams; `requests` counts the streams made, one per request. */ +class ScriptMembrane { + readonly requests: NormalizedRequest[] = []; + constructor(private readonly scripts: NormalizedResponse[][]) {} + streamYielding(request: NormalizedRequest, _options: YieldingStreamOptions = {}) { + this.requests.push(request); + const script = this.scripts.shift(); + if (!script) throw new Error('unexpected extra inference'); + return new MockYieldingStream(script); + } + asMembrane(): Membrane { return this as unknown as Membrane; } +} + +/** A channel registry with one channel, `#cafe`, whose first post waits for + * `released`. `posted` collects each post's text. */ +function firstPostHeld(posted: string[], released: Promise): unknown { + // Speech reaches the registry through deliverSpeech, or through + // routeSpeech, which reduces its outcome to the confirmed delivery. + const deliverSpeech = async (_agent: string, text: string, channelId: string | null) => { + posted.push(text.trim()); + if (posted.length === 1) await released; + return { status: 'delivered' as const, destination: { serverId: 'stub', channelId: channelId ?? '' }, at: Date.now() }; + }; + return new Proxy({ + resolveProseTarget: (spec: string) => spec === '#cafe' ? { channelId: 'chan-cafe' } : { error: `no channel ${spec}` }, + deliverSpeech, + routeSpeech: async (agent: string, text: string, channelId: string) => { + const outcome = await deliverSpeech(agent, text, channelId); + return { delivered: true, channelId: outcome.destination.channelId }; + }, + getDefaultPublishChannel: () => null, + isChannelOpen: () => true, + getDescriptor: () => undefined, + getChannelTools: () => [], + }, { get: (target, key: string) => key in target ? (target as Record)[key] : () => undefined }); +} + +/** `assistant`'s guard retry, waiting on its round's held post. */ +async function guardRetryWaiting() { + const dir = mkdtempSync(join(tmpdir(), 'af-stop-without-stream-guard-')); + const membrane = new ScriptMembrane([[postThenRead(), refused()], [answer('>>#cafe continued')]]); + const module = new GatedContextModule(); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane: membrane.asMembrane(), + agents: [{ name: 'assistant', model: 'test', systemPrompt: 'system', toolResultGuard: true, proseRouting: 'explicit' }], + modules: [module], + syncIntervalMs: 0, + }); + const posted: string[] = []; + const postReleased = deferred(); + (framework as unknown as { channelRegistry: unknown }).channelRegistry = firstPostHeld(posted, postReleased.promise); + const traces: TraceEvent[] = []; + framework.onTrace((event) => traces.push(event)); + const agent = framework.getAgent('assistant')!; + framework.start(); + framework.pushEvent({ type: 'external-message', source: 'test', content: 'read', metadata: {} } as unknown as ProcessEvent); + await waitFor("the guard's retry to wait on the round's post", () => + traces.some((t) => t.type === 'inference:stream_restarted' && (t as { reason?: string }).reason === 'tool_result_guard')); + assert.equal(membrane.requests.length, 1, 'the retry has not started'); + assert.deepEqual(posted, ['Reading.'], "the round's post is held"); + return { + framework, agent, traces, membrane, module, posted, + releasePost: () => postReleased.resolve(), + turnEnded: () => framework.getHostModeStatus().activeTurns === 0 && agent.state.status === 'idle', + cleanup: async () => { + postReleased.resolve(); + await within('stop() in cleanup', framework.stop()).catch(() => {}); + rmSync(dir, { recursive: true, force: true }); + }, + }; +} + +/** `writer`: where the stop's row was written. At the guard's check after + * its wait, through the refused round's stream's writer; at a check of the + * retry's own start, directly, under a request id of the start's own. */ +function endedWithoutRetry(g: Awaited>, reason: string, note: RegExp | null, writer: 'round' | 'start' = 'round'): void { + assert.equal(g.membrane.requests.length, 1, 'the retry was not sent'); + assert.deepEqual(abortReasons(g.traces), [reason], "one inference:aborted, with the stop's reason"); + assert.deepEqual(otherTerminals(g.traces), [], 'and no other terminal'); + assert.ok(!g.posted.includes('continued'), 'nothing more was posted'); + assert.equal(g.agent.toolResultGuard.recovering, false, 'the guard is not left recovering'); + const stops = stopRows(g.framework, g.agent.name, reason); + assert.equal(stops.length, 1, "the inference log has the stop's row"); + assert.equal(stops[0].tokenUsage, undefined, 'which carries no usage'); + // The stop's row comes after the refused round's own. + const rows = g.framework.queryInferenceLogs({ agentName: g.agent.name }).entries + .slice().sort((a, b) => a.sequence - b.sequence).map((e) => e.entry); + const refused = rows.findIndex((row) => row.error === 'Tool output withheld by the guard'); + const stopAt = rows.findIndex((row) => row.error === stops[0].error); + assert.ok(refused >= 0, "the inference log has the refused round's own row"); + assert.ok(stopAt > refused, "and the stop's row comes after it"); + if (writer === 'round') { + assert.equal(rows[stopAt].requestId, rows[refused].requestId, "through the round's stream's writer, under its request id"); + } else { + assert.notEqual(rows[stopAt].requestId, rows[refused].requestId, "written directly by the retry's start, under its own request id"); + } + assert.equal(g.framework.abortInference(g.agent.name, 'user'), false, 'and no check of the turn is left ahead'); + if (note) assert.ok(notes(g.agent).some((n) => note.test(n)), `the agent's note says ${note}`); + else assert.deepEqual(notes(g.agent), [], 'no note'); +} + +describe("a stop while the guard's refusal retry waits", () => { + it("a host's cancelStream() ends the turn instead of the retry", async () => { + const g = await guardRetryWaiting(); + try { + g.agent.cancelStream(); + g.releasePost(); + await waitFor('the turn to end', g.turnEnded); + endedWithoutRetry(g, 'user', /stopped before it finished/); + } finally { + await g.cleanup(); + } + }); + + it('abortInference() reports the stop once, and the retry is not sent', async () => { + const g = await guardRetryWaiting(); + try { + assert.equal(g.framework.abortInference('assistant', 'operator stop'), true, 'abortInference reports the stop'); + g.releasePost(); + await waitFor('the turn to end', g.turnEnded); + endedWithoutRetry(g, 'operator stop', /stopped before it finished/); + } finally { + await g.cleanup(); + } + }); + + it("a quiesce's abandon ends the turn, and the retry doesn't run while quiesced", async () => { + const g = await guardRetryWaiting(); + try { + const reached = abandonReached(g.framework); + const quiescing = g.framework.quiesce({ timeoutMs: 50, abandon: true }); + await within('the abandon to reach the turn', reached); + g.releasePost(); + const status = await within('the quiesce to return', quiescing); + assert.equal(status.quiesced, true); + endedWithoutRetry(g, 'quiesce_abandoned', null); + await g.framework.resume(); + } finally { + await g.cleanup(); + } + }); + + it('stop() ends the turn as a shutdown', async () => { + const g = await guardRetryWaiting(); + // The store closes with stop(), so the rows are read as they are written. + const rows: Array<{ error?: string; tokenUsage?: unknown; requestId?: string }> = []; + const internal = g.framework as unknown as { logInference: (entry: { error?: string; tokenUsage?: unknown; requestId?: string }) => void }; + const write = internal.logInference.bind(g.framework); + internal.logInference = (entry) => { rows.push(entry); write(entry); }; + // The refused round's own row, written before the wait. + const refusedRow = g.framework.queryInferenceLogs({ agentName: g.agent.name }).entries + .map((e) => e.entry).find((row) => row.error === 'Tool output withheld by the guard'); + try { + const stopping = g.framework.stop(); + await waitFor('stop() to cancel the finished stream', () => g.agent.state.status === 'idle'); + g.releasePost(); + await within('stop() to finish', stopping); + assert.ok(refusedRow, "the refused round's own row was written before the stop"); + assert.equal(g.membrane.requests.length, 1, 'the retry was not sent'); + assert.deepEqual(abortReasons(g.traces), ['shutdown'], 'one inference:aborted, as a shutdown'); + assert.deepEqual(otherTerminals(g.traces), [], 'and no other terminal'); + assert.ok(!g.posted.includes('continued'), 'nothing more was posted'); + const shutdownRows = rows.filter((row) => row.error === 'Stream aborted: shutdown'); + assert.equal(shutdownRows.length, 1, "the stop's row was written"); + assert.equal(shutdownRows[0].tokenUsage, undefined, "with no usage: the refused round's own row carried it"); + assert.equal(shutdownRows[0].requestId, refusedRow!.requestId, "through the round's stream's writer, under its request id"); + } finally { + await g.cleanup(); + } + }); + + it("a caller's cancelStream() and then a quiesce's abandon end the turn as the abandon", async () => { + const g = await guardRetryWaiting(); + try { + // The caller's stop takes the finished stream, so the abandon finds + // none: it reaches the turn through the wait's window. + g.agent.cancelStream(); + const reached = abandonReached(g.framework); + const quiescing = g.framework.quiesce({ timeoutMs: 50, abandon: true }); + await within('the abandon to reach the turn', reached); + g.releasePost(); + const status = await within('the quiesce to return', quiescing); + assert.equal(status.unabandonable, undefined, 'the turn ended within the grace'); + endedWithoutRetry(g, 'quiesce_abandoned', null); + await g.framework.resume(); + } finally { + await g.cleanup(); + } + }); + + it("a caller's cancelStream() and then stop() end the turn as a shutdown", async () => { + const g = await guardRetryWaiting(); + try { + // The caller's stop takes the finished stream, so stop() marks none. + g.agent.cancelStream(); + const stopping = g.framework.stop(); + await waitFor('stop() to close provider admission', () => (g.framework as unknown as { providerAdmissionClosed: boolean }).providerAdmissionClosed); + g.releasePost(); + await within('stop() to finish', stopping); + assert.equal(g.membrane.requests.length, 1, 'the retry was not sent'); + assert.deepEqual(abortReasons(g.traces), ['shutdown'], 'one inference:aborted, as a shutdown'); + assert.deepEqual(otherTerminals(g.traces), [], 'and no other terminal'); + } finally { + await g.cleanup(); + } + }); + + it("a caller's cancelStream(), an abandon and then stop() end the turn as a shutdown", async () => { + const g = await guardRetryWaiting(); + try { + g.agent.cancelStream(); + const reached = abandonReached(g.framework); + const quiescing = g.framework.quiesce({ timeoutMs: 50, abandon: true }); + void quiescing.catch(() => {}); + await within('the abandon to reach the turn', reached); + // With no stream left to record on, stop() reaches the turn by closing + // admission, which the check reads before the abandon's mark. + const stopping = g.framework.stop(); + await waitFor('stop() to close provider admission', () => (g.framework as unknown as { providerAdmissionClosed: boolean }).providerAdmissionClosed); + g.releasePost(); + await within('stop() to finish', stopping); + assert.equal(g.membrane.requests.length, 1, 'the retry was not sent'); + assert.deepEqual(abortReasons(g.traces), ['shutdown'], 'one inference:aborted, as a shutdown'); + } finally { + await g.cleanup(); + } + }); + + it("an abandon, a caller's cancelStream() and then stop() end the turn as a shutdown", async () => { + const g = await guardRetryWaiting(); + try { + // The abandon records itself on the finished stream, which the agent + // still holds; the caller's stop then takes the stream, so stop() + // finds none to record on. + const reached = abandonReached(g.framework); + const quiescing = g.framework.quiesce({ timeoutMs: 50, abandon: true }); + void quiescing.catch(() => {}); + await within('the abandon to reach the turn', reached); + g.agent.cancelStream(); + const stopping = g.framework.stop(); + await waitFor('stop() to close provider admission', () => (g.framework as unknown as { providerAdmissionClosed: boolean }).providerAdmissionClosed); + g.releasePost(); + await within('stop() to finish', stopping); + assert.equal(g.membrane.requests.length, 1, 'the retry was not sent'); + assert.deepEqual(abortReasons(g.traces), ['shutdown'], 'one inference:aborted, as a shutdown'); + } finally { + await g.cleanup(); + } + }); + + it("an abandon of the wait while it still holds the finished stream lists the turn if it outlasts the grace, and still ends it as the abandon", async () => { + const g = await guardRetryWaiting(); + try { + const reached = abandonReached(g.framework); + const quiescing = g.framework.quiesce({ timeoutMs: 50, abandon: true }); + await within('the abandon to reach the turn', reached); + // The round's post stays held through the abandon's ten-second grace. + const status = await within('the quiesce to return', quiescing, 15_000); + assert.deepEqual(status.unabandonable, ['assistant'], 'the turn still waiting is listed'); + g.releasePost(); + await waitFor('the turn to end', g.turnEnded); + endedWithoutRetry(g, 'quiesce_abandoned', null); + await g.framework.resume(); + } finally { + await g.cleanup(); + } + }); + + it("a stop while the retry's own hooks run ends the turn before the retry is sent, and ends the guard's recovery", async () => { + const g = await guardRetryWaiting(); + try { + const hooks = g.module.hold(); + g.releasePost(); + await within("the retry's hooks", hooks.entered); + g.agent.cancelStream(); + hooks.release(); + await waitFor('the turn to end', g.turnEnded); + endedWithoutRetry(g, 'user', /stopped before it finished/, 'start'); + } finally { + await g.cleanup(); + } + }); + + it('with no stop, the retry is sent once the post has gone out', async () => { + const g = await guardRetryWaiting(); + try { + g.releasePost(); + await waitFor('the retry to be answered', () => g.traces.some((t) => t.type === 'inference:completed')); + assert.equal(g.membrane.requests.length, 2, 'the retry was sent'); + assert.deepEqual(abortReasons(g.traces), [], 'nothing was aborted'); + await waitFor('the retry to post', () => g.posted.includes('continued')); + } finally { + await g.cleanup(); + } + }); +}); + +/** `assistant`'s turn with its answer in and its stream let go, holding its + * token while its one post goes out, held until `release()`. */ +async function finishingTurn() { + const dir = mkdtempSync(join(tmpdir(), 'af-stop-without-stream-finishing-')); + const membrane = new ScriptMembrane([[answer('>>#cafe Done.')], [answer('>>#cafe Again.')]]); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane: membrane.asMembrane(), + agents: [{ name: 'assistant', model: 'test', systemPrompt: 'system', proseRouting: 'explicit' }], + modules: [new GatedContextModule()], + syncIntervalMs: 0, + }); + const posted: string[] = []; + const postReleased = deferred(); + (framework as unknown as { channelRegistry: unknown }).channelRegistry = firstPostHeld(posted, postReleased.promise); + const traces: TraceEvent[] = []; + framework.onTrace((event) => traces.push(event)); + const agent = framework.getAgent('assistant')!; + framework.start(); + framework.pushEvent({ type: 'external-message', source: 'test', content: 'first', metadata: {} } as unknown as ProcessEvent); + await waitFor('the turn to wait on its post', () => posted.length === 1 && agent.state.status === 'idle' + && framework.getHostModeStatus().activeTurns === 1); + const completed = () => traces.filter((t) => t.type === 'inference:completed').length; + return { + framework, agent, traces, membrane, completed, + release: () => postReleased.resolve(), + /** Push the agent's next message and wait for its turn's answer. */ + nextTurnAnswered: async () => { + framework.pushEvent({ type: 'external-message', source: 'test', content: 'second', metadata: {} } as unknown as ProcessEvent); + await waitFor('the next turn to be answered', () => completed() === 2); + assert.equal(membrane.requests.length, 2, 'the next turn sent its request'); + assert.deepEqual(abortReasons(traces), [], 'nothing was aborted'); + }, + cleanup: async () => { + postReleased.resolve(); + await within('stop() in cleanup', framework.stop()).catch(() => {}); + rmSync(dir, { recursive: true, force: true }); + }, + }; +} + +describe('a turn finishing its last delivery, with no check ahead', () => { + it("a quiesce's abandon leaves nothing to end a later turn: the turn completes, and the agent's next turn runs", async () => { + const f = await finishingTurn(); + try { + const reached = abandonReached(f.framework); + const quiescing = f.framework.quiesce({ timeoutMs: 50, abandon: true }); + await within('the abandon to reach the turn', reached); + f.release(); + const status = await within('the quiesce to return', quiescing); + assert.equal(status.unabandonable, undefined, 'the turn ended within the grace'); + assert.equal(f.completed(), 1, 'the turn completed'); + await f.framework.resume(); + await f.nextTurnAnswered(); + } finally { + await f.cleanup(); + } + }); + + it("a stop ends nothing: abortInference() reports none, and the agent's next turn runs", async () => { + const f = await finishingTurn(); + try { + assert.equal(f.framework.abortInference('assistant', 'operator stop'), false, 'no check of the turn is ahead'); + f.agent.cancelStream(); + f.release(); + await waitFor('the turn to complete', () => f.completed() === 1 && f.framework.getHostModeStatus().activeTurns === 0); + await f.nextTurnAnswered(); + } finally { + await f.cleanup(); + } + }); +});