From 77997a5545a99d0dc411971b1bd8e4499a3b74e1 Mon Sep 17 00:00:00 2001 From: Nigel-2405 Date: Tue, 27 Oct 2026 17:53:22 -0400 Subject: [PATCH 1/2] feat(inference)!: cut a model call that stops making progress, and give a deadline expiry one fresh call (#378) A provider that keeps a resident's connection open and never answers holds the resident's turn with no end (#268). On main, run on Bun 1.4.4 against a stand-in provider, a stream kept alive by SSE pings or keepalive comments, or trickling a character every 20 s, was still waiting after 45 minutes, and so was a non-streamed call whose body trickled. A call that went silent ended at Bun's own fetch timeout, 360 s, before the Anthropic adapter's 600 s watchdogs, and a turn sent it three times, so a resident waited about 18 minutes. A stalled compression holds the resident's next turn behind it. What changes: - A resident's streamed call, and an MCPL server's streamed inference request, get membrane 0.8.0's progress bounds (`progressBounds`), which apply where the adapter's stream reports content (Anthropic, Bedrock). Before content: 60 s from the send, or, once a thinking block has started, the later of that and the block's allowance, its `max_tokens` at 87 tokens a second. After content: 150 s from the last provider event, or the allowance when that event was a thinking block's start and the allowance is longer. - Every streamed call the framework makes also gets a flat `progressTimeoutMs` from its last provider event: the larger of a setup term plus that allowance and 1,862 s (`FLAT_MINIMUM_MS`). The setup term is the larger of the before-content and between-events bounds, 150 s by default, so the flat bound doesn't run out before a phase bound whatever values a host sets, as long as the request's `maxTokens` is what's sent. It's the progress bound on an adapter without progress bounds, such as OpenAI-compatible, where a keepalive comment doesn't renew it. - A whole non-streamed call gets `progressTimeoutMs` of the setup term plus its allowance from its send, or the larger of the setup term and the flat minimum when it names no `max_tokens`: compression's and the reply judge's calls, through the membrane wrapper every context manager and the judge already call through, and an MCPL server's non-streamed request. A caller that passes its own `progressTimeoutMs` keeps it; `complete()` doesn't read `progressBounds`, so a caller's are passed on beside the framework's bound. Where an adapter's `complete()` streams under the hood, such as membrane's OpenAI Responses adapter in subscription mode, membrane renews the bound at each parsed provider event. - A cut arrives from membrane as an `aborted` event with reason `timeout`; a turn's stream now reports it as an `error` whose `MembraneError` has type `timeout`, `retryable: false`, membrane's message naming the bound followed by the host setting for that bound ("(the host's setting for this bound: `inferenceBounds.progressBounds.beforeContentMs`)", picked by the message's opening; one membrane doesn't word that way is left as it is), and membrane's deadline mark (`Symbol.for('membrane.deadlineExpiry')`, read the same way, since 0.8.0 exports no reader). - `DefaultErrorPolicy` gives a deadline expiry its own schedule, `deadline`: 2 tries, 1 s apart. That covers a bound's cut and the expiries membrane throws marked (the Anthropic adapter's watchdogs, the SDK's request timeout, a runtime's fetch timeout), which went to the base schedule's 3 tries and stay retryable. For a deadline expiry the framework passes an `errorPolicy` the call's consecutive expiries as `attempt`, in place of the turn's failures, reset by a completed provider call, by another failure, at a turn's start and when the agent goes, so each cut call gets its fresh call whatever else the turn met. A refusal membrane sends again replays the same request and doesn't reset it. Settings with no `deadline` schedule give an expiry their base schedule, counted the same way. - The `[inference-failed]` note for a timeout says the model call timed out (or the turn's last call did, after rounds ran), and, when that call was a fresh call made for an earlier cut and a deadline ended it too, that the call was made again and timed out again, with no count, since another failure between two cuts resets the count ("sent" stays for the reply); the cause in parentheses is left out when it's only `timeout`. None of the bounds depends on how much the agent writes, so the note gives no advice about output: it closes with "If a reply is still owed from that turn, you can give it now.", since the agent reads it in a later turn. Other failures keep main's wording and advice. - An MCPL server's streamed request that ends early is answered with membrane's message beside the reason, so a cut names its bound, and for a cut the host setting for it too. - A turn sent again after its retries ran out (#846's re-arm) gets no fresh call for a cut before one of its rounds has run or had its tool calls cut off, as for any failure there. #846's check of whether a failure is transient reads membrane's deadline mark through the same isDeadlineExpiry, so a bound's cut and a thrown expiry stay out of the re-arm, as before. - `FrameworkConfig.inferenceBounds` sets the values: `progressBounds` (membrane's `beforeContentMs`, `betweenEventsMs`, `thinkTokensPerSecond`, which the flat and whole-call bounds derive from) and `flatMinimumMs`. `false` sets none of these bounds. One setting covers every agent. A value it can't use is refused at `create()`, naming the field under `inferenceBounds`, before the store opens. These bounds sit above the runtime's fetch timeout and the Anthropic adapter's watchdogs, which still end a silent connection first when a bound is raised past them. Where the values come from: membrane's README gives the progress bounds' measurements, on a production router's streamed Anthropic calls. The flat minimum is the longest silence between frames of any completed call in that router's OpenAI calls (the Responses API over a WebSocket), standing in for an OpenAI-compatible stream. The whole-call bound's setup term is five times the longest wait from the send to content or a thinking block's start in those calls (a wait that router cuts at 30 s), and its allowance is generation at membrane's 87 tokens a second; no completed streamed call there ran past it. Bedrock, other models, OpenAI-compatible streams and non-streamed calls weren't measured. Not changed: a stream that keeps sending content isn't ended by any bound, since content is progress. The direct Agent API (`Agent.runInference`) reads no framework setting and gets none of these bounds. Tests: test/inference-bounds.test.ts runs real turns and MCPL requests, and complete() calls through the membrane a resident's compression and its reply judge are handed, against a local stand-in provider with short bounds, asserting which bound ran out by membrane's message, and records the options each call is sent. It runs on a stepped clock: node:test's mocked timers and Date, with performance.now, which membrane's bounds compare against, read from the same Date, while the sockets stay real; the clock moves 5 ms per event-loop turn and not while a file or lookup request is in flight, so the file takes a few seconds rather than the minute and more its bounds take in real time. With main's sources for the four files this change wires, 49 of its 56 tests fail; the 7 that pass pin what doesn't change: a stream that keeps sending content, a caller's own deadline, `inferenceBounds: false` on a turn, on compression and on both kinds of MCPL request, and a host's settings with no deadline schedule. Four existing test cases used an `aborted` event with reason `timeout` to stand for an abort that isn't a stop, a path a timeout no longer takes: one each in end-turn-usage and tool-result-guard, and two through prose-drafts' shared helper. They now use reason `error`, which still takes it. The provider-hold test's hand-built framework gains the new map; the policy test that called an unmarked timeout a stream deadline now calls it an HTTP 408; the router test's aborted answer carries membrane's message. --- README.md | 30 + changelog.d/378-inference-bounds.added.md | 9 + changelog.d/378-inference-bounds.breaking.md | 38 + changelog.d/378-inference-bounds.changed.md | 16 + src/agent.ts | 5 + src/framework.ts | 111 ++- src/index.ts | 1 + src/inference-bounds.ts | 111 +++ src/mcpl/inference-router.ts | 13 +- src/types/framework.ts | 28 +- test/default-error-policy.test.ts | 4 +- test/end-turn-usage.test.ts | 2 +- test/inference-bounds.test.ts | 768 +++++++++++++++++++ test/inference-failure-bounds.test.ts | 2 +- test/mcpl-inference-router.test.ts | 4 +- test/prose-drafts.test.ts | 2 +- test/tool-result-guard.test.ts | 8 +- 17 files changed, 1105 insertions(+), 47 deletions(-) create mode 100644 changelog.d/378-inference-bounds.added.md create mode 100644 changelog.d/378-inference-bounds.breaking.md create mode 100644 changelog.d/378-inference-bounds.changed.md create mode 100644 src/inference-bounds.ts create mode 100644 test/inference-bounds.test.ts diff --git a/README.md b/README.md index 718d17c6a..8223dc4bf 100644 --- a/README.md +++ b/README.md @@ -543,6 +543,36 @@ static tools are listed like any server's. 5. **Budget restart**: If input tokens exceed `maxStreamTokens`, the stream is cancelled, context is recompressed, and a fresh stream starts. An agent whose model binds its thinking (`thinkingBinding.binds`) with `physicalWindowTokens` set skips this, and the window projection restarts it instead before a request the window can't hold 6. **Complete**: Final response saved to context, agent returns to idle +### Inference bounds + +A model call that stops making progress is cut, so an agent isn't held by a provider that has gone quiet, or that keeps its connection open with keepalives (an Anthropic stream's pings, SSE comments) and never answers. + +- **A resident's streamed call**, and an MCPL server's streamed inference request, get membrane's progress bounds where the adapter's stream reports content (`AnthropicAdapter`, `BedrockAdapter`). Before content, a call ends 60 s after its send, or, once a thinking block has started, at the later of that and the block's allowance, its `max_tokens` at 87 tokens a second. Once content has begun, it ends after 150 s without a provider event, or after the block's allowance when the last event was a thinking block's start and the allowance is longer. Membrane's README gives the rules whole. +- **Every streamed call the framework makes** also gets a flat bound from its last provider event: the larger of a setup term plus that allowance and 1,862 s (`FLAT_MINIMUM_MS`). The setup term is the larger of the before-content and between-events bounds, 150 s by default. So on an adapter with progress bounds the flat bound never runs out first. On one without, such as `OpenAICompatibleAdapter`, it's the only progress bound, and an SSE keepalive comment doesn't restart it, though a data frame that parses as an event does, a provider's own ping object included. +- **A whole non-streamed call**, a compression's, the reply judge's, or an MCPL server's non-streamed inference request, ends the setup term plus its allowance after its send, or, when it names no `max_tokens`, at the larger of the setup term and the flat minimum, unless its caller passed a `progressTimeoutMs` of its own. Its caller gets the `timeout` error. The event gate's own 20 s judge timeout ends the reply judge's call first at the defaults. Context-manager takes a cut compression call as a transient failure: the lane pauses and tries again, so compression that never lands within its bound shows as a lane that keeps pausing, not as a failed turn, and is a reason to raise the bound. On an adapter whose `complete()` streams under the hood, such as membrane's OpenAI Responses adapter in subscription mode, membrane renews the bound at each parsed provider event, so there it runs from the last event, as a stream's flat bound does. +- **A turn's call a bound cuts** fails as a `timeout`, and `DefaultErrorPolicy` gives it one fresh call, 1 s later, whatever other failures its turn had, except in a turn sent again after its retries ran out, before one of its rounds has run or had its tool calls cut off, whose failures get no retries. A fresh call that's cut too, with no other failure between, ends the turn, and its `[inference-failed]` note says the call timed out, that the call was made again and timed out again, and the reason. For a bound's cut the reason names the bound that ran out and the host's setting for it. A deadline expiry membrane throws (the Anthropic adapter's own watchdogs, the SDK's request timeout, a runtime's fetch timeout on a silent connection) counts the same toward the fresh call and the turn's end, and its reason is its own message, which names no setting. A bound's cut isn't `retryable`, so a host's own `errorPolicy` that retries only retryable errors gives it no fresh call; an expiry membrane throws stays `retryable`, as membrane made it. +- **A stream that keeps sending content** isn't ended by any of these, since content is progress. + +These bounds don't replace the deadlines below them: a silent connection still meets the runtime's own fetch timeout (360 s on Bun 1.4.4) and, on `AnthropicAdapter`, the adapter's 600 s first-event and idle watchdogs, so on a silent connection a bound raised past those is met by them first. + +The phase bounds, and a reason that names the bound that ran out, need the `Membrane` a host passes in to come from membrane 0.8.0 or later. An older one ignores `progressBounds`, so only the flat bound acts, and a cut's reason names no bound. + +The flat and whole-call bounds read the request's `maxTokens`. A `max_tokens` set through `providerParams` or a hook isn't read, where membrane's own allowance reads what's sent, so a larger one set that way can let the flat bound run out before membrane's. + +Where the values come from: membrane's README gives its progress bounds' measurements, on a production router's streamed Anthropic calls to `claude-opus-5-5` and `claude-haiku-5-5`. The flat minimum is the longest silence between frames of any completed call in that router's OpenAI calls (the Responses API over a WebSocket), standing in for an OpenAI-compatible stream, which wasn't measured. The whole-call bound's setup term is five times the longest wait from the send to content or a thinking block's start in those calls (a wait that router cuts at 30 s and sends again, so a longer one is unseen), and its allowance is generation at 87 tokens a second, the 1st percentile of Opus's thinking speed over its thinking blocks of 30 seconds or more. No completed streamed call in that record ran past the whole-call bound. Bedrock, other models (`claude-opus-4-6` among them), OpenAI-compatible streams and calls that aren't streamed, compression's among them, weren't measured, so a host whose calls run slower raises the bounds with `inferenceBounds`. Membrane's README also reports the wait for content growing with the prompt, so a resident near a full context window is the likeliest to meet the before-content bound first. + +`inferenceBounds` sets them: `progressBounds` takes membrane's three values, from which the flat and whole-call bounds also derive, and `flatMinimumMs` the flat bound's least. `false` sets none. One setting covers every agent, so a host with one slower model or adapter raises them for all of its agents. A stream cut's reason is membrane's message, naming the bound that ran out, followed by the host setting for that bound, as in ``(the host's setting for this bound: `inferenceBounds.progressBounds.beforeContentMs`)``. A whole call's caller gets membrane's message alone. Each reason's opening and its setting: + +| The reason begins | The setting | Default | +|---|---|---| +| `No content within …s of the send` | `progressBounds.beforeContentMs` | 60,000 ms | +| `No provider event within …s` | `progressBounds.betweenEventsMs` | 150,000 ms | +| `No content within …s of a thinking block's start` (or `No provider event within …s of a thinking block's start`) | `progressBounds.thinkTokensPerSecond`: a lower rate allows a slower think | 87 | +| `No provider progress within …ms`, on a stream | `flatMinimumMs`, or the setup term and allowance when they're larger | 1,862,000 ms | +| `No provider progress within …ms`, on a call that isn't streamed | the setup term (`beforeContentMs`, `betweenEventsMs`) and the allowance (`thinkTokensPerSecond`, or the request's `maxTokens`); the larger of the setup term and `flatMinimumMs` when the request names no `maxTokens` | 150 s plus `maxTokens` at 87 a second; 1,862,000 ms with no `maxTokens` | + +Each cut of a turn's call is traced as `inference:failed`, with its reason as `error`, and logged with the turn's inference log as a failed call, whether or not its fresh call then answers, so a host can count cuts a resident never sees and raise a bound before turns start failing. An MCPL server's request that a bound cuts is traced as `mcpl:inference_fail` with the reason as `error`, and a compression or reply-judge call's cut reaches only its caller, as the `timeout` error. + ### Abort / Cancellation ```typescript diff --git a/changelog.d/378-inference-bounds.added.md b/changelog.d/378-inference-bounds.added.md new file mode 100644 index 000000000..fc92397fa --- /dev/null +++ b/changelog.d/378-inference-bounds.added.md @@ -0,0 +1,9 @@ +- `FrameworkConfig.inferenceBounds` sets how long a model call may go without + progress: `progressBounds`, membrane's `beforeContentMs`, `betweenEventsMs` + and `thinkTokensPerSecond`, from which the flat and whole-call bounds also + derive, and `flatMinimumMs`, by default `FLAT_MINIMUM_MS` (exported). `false` + sets none. A value it can't use is refused at `create()`, naming the field + under `inferenceBounds`. One setting covers every agent, so a host with one + slower model raises the bounds for all of them. +- Settings passed to `new DefaultErrorPolicy(settings)` can carry a + `deadline` schedule for deadline expiries. diff --git a/changelog.d/378-inference-bounds.breaking.md b/changelog.d/378-inference-bounds.breaking.md new file mode 100644 index 000000000..44faa1003 --- /dev/null +++ b/changelog.d/378-inference-bounds.breaking.md @@ -0,0 +1,38 @@ +- **Hosts:** a model call that stops making progress is now cut + (agent-framework#378). On an adapter whose stream reports content + (`AnthropicAdapter`, `BedrockAdapter`), a resident's streamed call, and an + MCPL server's, gets membrane's progress bounds: before content, 60 s from + its send, or, once a thinking block has started, the later of that and the + block's allowance (its `max_tokens` at 87 tokens a second); after content, + 150 s from its last provider event, or the allowance when that event was a + thinking block's start and the allowance is longer. On every adapter a + streamed call also ends once the larger of a setup term (the larger of the + before-content and between-events bounds, 150 s) plus that allowance and + 1,862 s passes with no provider event; an SSE keepalive comment isn't one. + A compression's, the reply judge's and an MCPL server's non-streamed call + ends the setup term plus its allowance after its send, or at the larger of + the setup term and 1,862 s when it names no `max_tokens`, unless its caller + passed a `progressTimeoutMs` of its own; on an adapter whose `complete()` + streams under the hood, such as membrane's OpenAI Responses adapter in + subscription mode, that bound runs from the last provider event instead. + Before, a call ended only at the Anthropic adapter's own 600 s watchdogs, + the SDK's request timeout or the runtime's fetch timeout, which still apply, + and a call kept open with keepalives didn't end. A call that would have run + past one of these bounds and then completed now fails as a `timeout`. + **Migration:** `inferenceBounds` raises them, and `inferenceBounds: false` + sets none of them. The phase bounds act only when the `Membrane` passed in + comes from membrane 0.8.0 or later; with an older one only the flat bound + acts, so a host that builds its own `Membrane` raises its membrane to + `^0.8.0` with this release. +- **Hosts:** `DefaultErrorPolicy` gives a deadline expiry one fresh call, + 1 s later. An expiry is a bound's cut, the Anthropic adapter's watchdogs, + the SDK's request timeout or a runtime's fetch timeout; the last three were + thrown as retryable timeouts and got the base schedule's three tries. For a + deadline expiry the framework passes an `errorPolicy` the count of that + call's consecutive expiries as `attempt`, in place of the turn's failures, + so each call a deadline cuts gets its fresh call; a host's own policy reads + that count too. A quiet call on an adapter without progress bounds now ends + after two of the runtime's fetch timeouts (on Bun, 360 s each) rather than + three. **Migration:** settings passed to `new DefaultErrorPolicy(settings)` + with no `deadline` schedule give an expiry their base schedule, counted the + same way. diff --git a/changelog.d/378-inference-bounds.changed.md b/changelog.d/378-inference-bounds.changed.md new file mode 100644 index 000000000..b26de90bd --- /dev/null +++ b/changelog.d/378-inference-bounds.changed.md @@ -0,0 +1,16 @@ +- A turn that ends on a timeout says so in its `[inference-failed]` note: the + model call timed out, or its last call did, with the reason, which for a + bound's cut names the bound that ran out. When that call was a fresh call made for an earlier cut and a + deadline ended it too, the note says so: "The call was made again, and it timed out again." The cause in parentheses + is left out when it's only `timeout`. Its closing line says that if a reply + is still owed from that turn, it can be given now. Other failures keep their + wording and advice. +- An MCPL server's streamed inference request that ends early is answered + with the reason and, when membrane gives one, its message, such as the + bound that ran out: ``The inference stopped before it finished (timeout: + No content within 60s of the send: … (the host's setting for this bound: + `inferenceBounds.progressBounds.beforeContentMs`))``. +- A stream cut's reason, in a turn's error and note and in an MCPL server's + answer, ends by naming the host setting for the bound it hit, such as + ``(the host's setting for this bound: + `inferenceBounds.progressBounds.beforeContentMs`)``. diff --git a/src/agent.ts b/src/agent.ts index 08b1304a1..2b983fcae 100644 --- a/src/agent.ts +++ b/src/agent.ts @@ -2,6 +2,7 @@ import { NOTICE_TAG } from './notice-tags.js'; import { heartbeatToolAlias } from './heartbeat/heartbeat.js'; import type { Membrane, NormalizedMessage, NormalizedRequest, ContentBlock, ThinkingBindingEntry, YieldingStream } from '@animalabs/membrane'; import { isAbortedResponse } from '@animalabs/membrane'; +import { streamBounds, type InferenceBounds } from './inference-bounds.js'; import type { CacheWireReceipt, KvUnifiedRequestHooks } from './kv-unified-wire.js'; import { ToolResultGuard } from './tool-result-guard.js'; import { withoutTokenEstimates } from './thinking-token-stamp.js'; @@ -50,6 +51,9 @@ export interface ActivationRequestOptions { * pass it (its errorPolicy is their retry layer); a direct caller of * startStream keeps membrane's retries. */ membraneRetries?: boolean; + /** The framework's inferenceBounds, resolved: the stream gets their phase + * bounds and flat bound. None when left out. */ + inferenceBounds?: InferenceBounds | null; } /** A compile's render stats, as context-manager's `getRenderStats()` types @@ -2169,6 +2173,7 @@ export class Agent { // since membrane 0.7 its own schedules, ten minutes at a 529, would // otherwise run inside each of the caller's attempts, out of its sight. ...(options.membraneRetries === false ? { retry: false } : {}), + ...streamBounds(options.inferenceBounds ?? null, request.config.maxTokens), // Retry a content-policy refusal at the provider seam rather than at // the framework level: membrane replays the SAME request immediately // (cache-warm), where a framework-level requeue would recompile and diff --git a/src/framework.ts b/src/framework.ts index 3fc3f788e..92de9a734 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -11,6 +11,7 @@ import { JsStore } from '@animalabs/chronicle'; import type { Membrane, ContentBlock, NormalizedRequest, NormalizedResponse, YieldingStream, DetailedUsage, ToolParameter, ToolResult as MembraneToolResult, ToolResultContentBlock, StreamEvent } from '@animalabs/membrane'; import { MembraneError } from '@animalabs/membrane'; import { classifyFailure, retryDecision, MEMBRANE_RETRY_SETTINGS, type RetrySettings } from '@animalabs/membrane/retry'; +import { deadlineCut, isDeadlineExpiry, resolveInferenceBounds, wholeCallMs, type InferenceBounds } from './inference-bounds.js'; import { ContextManager, PassthroughStrategy, WindowedPassthroughStrategy, OverBudgetError, UncoveredDropError, stripUnpairedToolBlocks, continuesBody, MalformedContentError, malformedContentPlaces, CompressionWithdrawn } from '@animalabs/context-manager'; import type { CacheWireReceipt } from './kv-unified-wire.js'; import { SILENT_HEARTBEAT_SEPARATOR, type RequestOnlyPrompt, type RequestOnlyStamps } from './silent-heartbeat.js'; @@ -1184,14 +1185,9 @@ function waitAlertsOnItsOwn(wait: ProviderWait): boolean { return wait.until === null || wait.until - wait.setAt > PROVIDER_WAIT_ALERT_MS; } -// membrane 0.8.0 marks a deadline it reports as run out with this registered -// symbol (on a sent request, or the Anthropic SDK's timeout, which covers its -// connect timeout too), and doesn't export its reader. -const MEMBRANE_DEADLINE_EXPIRY: unique symbol = Symbol.for('membrane.deadlineExpiry'); - function isTransientFailure(error: Error): boolean { if (!isMembraneError(error) || error.retryable !== true) return false; - if ((error as { [MEMBRANE_DEADLINE_EXPIRY]?: unknown })[MEMBRANE_DEADLINE_EXPIRY] === true) return false; + if (isDeadlineExpiry(error)) return false; return error.type === 'server' || error.type === 'network' || error.type === 'timeout' || error.type === 'rate_limit'; } @@ -2288,9 +2284,15 @@ function hostCommandRequester(serverId: string, params: HostCommandParams): Oper * schedule, jittered, 10 s doubling to 5 min (by default 13 tries, where * membrane's settings have 7: about 30 min). * - rate limited (429): the base schedule with at least 5 tries. - * - any other retryable failure, a stream deadline on a sent request - * included (membrane leaves those to its caller): the base schedule (by - * default 3 tries, 1 s then 2 s). + * - a deadline expiry, which membrane leaves to its caller: a progress + * bound's cut (FrameworkConfig.inferenceBounds), the adapter's own + * watchdogs, the SDK's request timeout or a runtime's fetch timeout. The + * deadline schedule: by default 2 tries, 1 s apart, so one fresh call. For + * one, the framework's `attempt` counts that call's consecutive expiries, + * not the turn's other failures, so each call a deadline cuts gets its + * fresh call. Settings with no deadline schedule give it their base one. + * - any other retryable failure: the base schedule (by default 3 tries, 1 s + * then 2 s). * A provider's stated wait (`retryAfterMs`) is waited when it is longer than * the backoff, up to the schedule's ceiling, which by default is at least the * framework's own cap (PROVIDER_ACCELERATION_MAX_COOLDOWN_MS, ten minutes); @@ -2312,6 +2314,10 @@ export class DefaultErrorPolicy implements ErrorPolicy { } onInferenceError(error: Error, _agentName: string, attempt: number): ErrorAction { + if (isDeadlineExpiry(error)) { + const decision = retryDecision(this.settings.deadline ? 'deadline' : 'retryable', attempt + 1, undefined, this.settings); + return decision.retry ? { retry: true, delayMs: decision.delayMs } : { retry: false }; + } if (!isMembraneError(error)) { // Not a provider failure membrane classified: retried as any other // retryable failure, on the base schedule. @@ -2320,9 +2326,7 @@ export class DefaultErrorPolicy implements ErrorPolicy { } if (error.retryable === false) return { retry: false }; // The facts membrane classifies by, read off the error itself (from either - // copy of membrane, as isMembraneError takes one). No `deadline` fact: a - // stream deadline is a retryable timeout here, as it was before membrane - // 0.7, where membrane itself gives it no retry. + // copy of membrane, as isMembraneError takes one). const facts = error as { retryable?: boolean; type?: string; httpStatus?: number; message?: string }; // A failure that doesn't say is retried as an unknown one is; only false refuses (handled above). const cls = classifyFailure({ retryable: facts.retryable !== false, type: facts.type, httpStatus: facts.httpStatus, message: facts.message }); @@ -2416,13 +2420,21 @@ const PROVIDER_ACCELERATION_DEFAULT_COOLDOWN_MS = 65_000; * longest stated wait an ephemeral run sleeps in its turn: a longer one ends * the run instead (nonOwningProviderWait). */ const PROVIDER_ACCELERATION_MAX_COOLDOWN_MS = 10 * 60_000; +const FAILURE_ADVICE = 'If this recurs with the same cause, change approach rather than retrying identically ' + + '(e.g. drop an oversized attachment or an unsupported setting).'; +const TIMEOUT_ADVICE = 'If a reply is still owed from that turn, you can give it now.'; +const MADE_AGAIN = 'The call was made again, and it timed out again.'; const OVERLOADED_TURN_TRIES = 13; -const TURN_RETRY_SETTINGS: RetrySettings = Object.fromEntries( - Object.entries(MEMBRANE_RETRY_SETTINGS as RetrySettings).map(([cls, s]) => [cls, s && { - ...s, - ...(cls === 'overloaded' ? { maxAttempts: OVERLOADED_TURN_TRIES } : {}), - statedWaitCeilingMs: s.statedWaitCeilingMs === null ? null : Math.max(s.statedWaitCeilingMs, PROVIDER_ACCELERATION_MAX_COOLDOWN_MS), - }])) as RetrySettings; +const DEADLINE_TURN_TRIES = 2; +const TURN_RETRY_SETTINGS: RetrySettings = { + ...Object.fromEntries( + Object.entries(MEMBRANE_RETRY_SETTINGS as RetrySettings).map(([cls, s]) => [cls, s && { + ...s, + ...(cls === 'overloaded' ? { maxAttempts: OVERLOADED_TURN_TRIES } : {}), + statedWaitCeilingMs: s.statedWaitCeilingMs === null ? null : Math.max(s.statedWaitCeilingMs, PROVIDER_ACCELERATION_MAX_COOLDOWN_MS), + }])), + deadline: { maxAttempts: DEADLINE_TURN_TRIES, firstDelayMs: 1_000, multiplier: 1, maxDelayMs: 1_000, statedWaitCeilingMs: null }, +} as RetrySettings; const PROVIDER_ACCELERATION_JITTER_MS = 5_000; /** Longest delay one setTimeout can represent (2^31-1 ms, about 24.8 days); * a longer timer fires almost at once, so longer waits re-arm in steps. */ @@ -2472,6 +2484,11 @@ async function* turnStreamEvents(stream: YieldingStream, agentName: string): Asy console.error(`[stream] ${agentName}: the stream sent an event (${event.type}) after its ${decidedBy} event, which decided its turn; it and the rest of the stream decide nothing`); return; } + if (event.type === 'aborted' && event.reason === 'timeout') { + decidedBy = 'error'; + yield { type: 'error', error: deadlineCut(event.message) } as StreamEvent; + continue; + } if (event.type === 'complete' || event.type === 'error' || event.type === 'aborted') decidedBy = event.type; yield event; } @@ -3371,6 +3388,9 @@ export class AgentFramework { private providerAccelerationLastRecovery: Map = new Map(); private providerAccelerationDefaultCooldownMs = PROVIDER_ACCELERATION_DEFAULT_COOLDOWN_MS; private providerAccelerationJitterMs = PROVIDER_ACCELERATION_JITTER_MS; + private inferenceBounds: InferenceBounds | null = resolveInferenceBounds(undefined); + private readonly deadlineCuts = new Map(); + private readonly freshCallsCut = new Set(); private providerHoldHook: ProviderHoldHook | undefined; /** Provider waits (stated retry-after) per (agent, model), kept across restarts. */ private providerWaits: ProviderWaits | undefined; @@ -4059,6 +4079,7 @@ export class AgentFramework { static async create(config: FrameworkConfig): Promise { // Before anything opens or starts: a refused config must leave nothing behind. if (config.codeExecution != null) validateCodeExecutionConfig(config.codeExecution); + resolveInferenceBounds(config.inferenceBounds); assertFrameworkSwitches(config); for (const agentConfig of config.agents) assertAgentSwitches(agentConfig); assertAgentNamesUnique(config.agents); @@ -4254,6 +4275,7 @@ export class AgentFramework { ); framework.providerHoldHook = config.providerHold; framework.turnRearmMs = turnRearmDelays(config.exhaustedTurnRearmMs); + framework.inferenceBounds = resolveInferenceBounds(config.inferenceBounds); // Before any delivery. await framework.reconcileDiscordAwarenessAtStartup(); @@ -5885,8 +5907,17 @@ export class AgentFramework { const value = Reflect.get(obj, prop, receiver); if (prop === 'complete' && typeof value === 'function') { return (...args: unknown[]) => { - const model = (args[0] as { config?: { model?: unknown } } | undefined)?.config?.model; - return admit(() => Reflect.apply(value, this.membrane, args) as Promise, + const config = (args[0] as { config?: { model?: unknown; maxTokens?: unknown } } | undefined)?.config; + const model = config?.model; + const options = args[1] as { progressTimeoutMs?: unknown } | undefined; + const bounds = this.inferenceBounds; + const sent = bounds !== null && options?.progressTimeoutMs === undefined + ? [args[0], { + ...options, + progressTimeoutMs: wholeCallMs(bounds, typeof config?.maxTokens === 'number' ? config.maxTokens : undefined), + }, ...args.slice(2)] + : args; + return admit(() => Reflect.apply(value, this.membrane, sent) as Promise, typeof model === 'string' ? model : undefined); }; } @@ -10579,6 +10610,7 @@ export class AgentFramework { // deregistered, so disposal must release EventGate liveness itself. this.eventGate?.onInferenceEnded(agent.name); this.agents.delete(agent.name); + this.deadlineCuts.delete(agent.name); this.thinkingLossIncidents.forget(agent.name); this.reasoningLoss.forget(agent.name); // Its pending self-wake ends with it, as a fork's does. @@ -15382,6 +15414,7 @@ export class AgentFramework { this.activeStreams.delete(agentName); this.pendingAssistantBlocks.delete(agentName); this.agents.delete(agentName); + this.deadlineCuts.delete(agentName); this.thinkingLossIncidents.forget(agentName); this.reasoningLoss.forget(agentName); // Its pending self-wake ends with it, rather than wake a name that's gone. @@ -20265,6 +20298,7 @@ export class AgentFramework { attempt = 0, providerGateAlreadyHeld = false, ): Promise { + if (attempt === 0) this.deadlineCuts.delete(agent.name); const ownsProviderGate = !this.ephemeralRuns.has(agent.name) && !this.conversationAgentHomes.has(agent.name); // A stopping framework starts no turn, whoever owns provider admission @@ -21281,6 +21315,7 @@ export class AgentFramework { // This turn's retries are the framework's (errorPolicy below), not // membrane's inside each attempt. membraneRetries: false, + inferenceBounds: this.inferenceBounds, }); // Each later round this stream sends is noted as it goes (noteRoundSent). if (noteRequestSent) this.streamRequestNotes.set(stream, { note: noteRequestSent, rounds: [], stopped: false }); @@ -21463,8 +21498,12 @@ export class AgentFramework { return false; } + const cuts = isDeadlineExpiry(err) ? this.deadlineCuts.get(agent.name) ?? 0 : undefined; const action: ErrorAction = statedWait === 'fail' || (isRearmed(trigger) && !gotPartway(routing)) - ? { retry: false } : this.errorPolicy.onInferenceError(err, agent.name, attempt); + ? { retry: false } : this.errorPolicy.onInferenceError(err, agent.name, cuts ?? attempt); + if (action.retry && cuts !== undefined) this.deadlineCuts.set(agent.name, cuts + 1); + else this.deadlineCuts.delete(agent.name); + if (!action.retry && cuts !== undefined && cuts > 0) this.freshCallsCut.add(agent.name); if (action.retry) { // A policy's delay can be a provider's retry-after: wait all of it. await this.waitForRetry(action.delayMs, checkWindow, agent); @@ -23307,8 +23346,12 @@ export class AgentFramework { break; } + const cuts = isDeadlineExpiry(err) ? this.deadlineCuts.get(agent.name) ?? 0 : undefined; const action: ErrorAction = statedWait === 'fail' || (isRearmed(trigger) && !gotPartway(streamRouting)) - ? { retry: false } : this.errorPolicy.onInferenceError(err, agent.name, attempt); + ? { retry: false } : this.errorPolicy.onInferenceError(err, agent.name, cuts ?? attempt); + if (action.retry && cuts !== undefined) this.deadlineCuts.set(agent.name, cuts + 1); + else this.deadlineCuts.delete(agent.name); + if (!action.retry && cuts !== undefined && cuts > 0) this.freshCallsCut.add(agent.name); 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 @@ -23476,6 +23519,7 @@ export class AgentFramework { } case 'usage': { + this.deadlineCuts.delete(agent.name); // Cumulative across the stream's rounds: replace, never sum. latestUsage = event.usage; // The stream's accumulated fresh input, which is what the @@ -28542,6 +28586,7 @@ export class AgentFramework { const ended = this.exhaustedTurns?.get(agentName); this.exhaustedTurns?.delete(agentName); const rearmed = this.exhaustingRearms?.delete(agentName) === true; + const madeAgain = this.freshCallsCut?.delete(agentName) === true; // The reason is the failure's own message, and a provider can echo the // whole rejected request in it: one household's [inference-failed] // marker carried ~1.7 MB of its own request back into its context. Every @@ -28598,9 +28643,10 @@ export class AgentFramework { // (2) Agent-facing chronicle marker (no inference triggered → no loop). const agent = this.agents.get(agentName); if (agent && !(rearmed && !ended) && process.env.SUPPRESS_INFERENCE_FAILED_MARKER !== '1') { - const cause = [errorType, httpStatus !== undefined ? `HTTP ${httpStatus}` : undefined] - .filter((part): part is string => part !== undefined) - .join(', '); + const timedOut = errorType === 'timeout'; + const causeParts = [errorType, httpStatus !== undefined ? `HTTP ${httpStatus}` : undefined] + .filter((part): part is string => part !== undefined); + const cause = timedOut && causeParts.length === 1 ? '' : causeParts.join(', '); try { agent.getContextManager().addMessage( 'user', @@ -28608,17 +28654,17 @@ export class AgentFramework { type: 'text', text: (ended - ? `[${NOTICE_TAG.inferenceFailed}] Your previous turn did not complete: it ` + - `failed${cause ? ` (${suppliedValue(cause)})` : ''}.` + + ? `[${NOTICE_TAG.inferenceFailed}] Your previous turn did not complete: ` + + `${timedOut ? 'its last call timed out' : 'it failed'}${cause ? ` (${suppliedValue(cause)})` : ''}.` + + `${madeAgain ? ` ${MADE_AGAIN}` : ''}` + `${ended.roundsRan ? ' Its earlier rounds ran, and their tool calls ran as their results show.' : ''}` + `${ended.callsCutOff ? ` ${CALLS_CUT_OFF}` : ''}` + `${ended.liveSpeech ? ` ${LIVE_SPEECH_UNRECEIPTED}` : ''}` + `${ended.speechRouting ? ` ${SPEECH_STILL_ROUTING}` : ''} Reason: ` - : `[${NOTICE_TAG.inferenceFailed}] Your previous turn did not complete: the model ` + - `call failed${cause ? ` (${suppliedValue(cause)})` : ''} and produced no response, so nothing was sent. Reason: `) + - `${failureShown(markerReason)}. If this recurs with the same cause, change approach ` + - `rather than retrying identically (e.g. drop an oversized attachment ` + - `or an unsupported setting).`, + : `[${NOTICE_TAG.inferenceFailed}] Your previous turn did not complete: ` + + `the model call ${timedOut ? 'timed out' : 'failed'}${cause ? ` (${suppliedValue(cause)})` : ''} ` + + `and produced no response, so nothing was sent.${madeAgain ? ` ${MADE_AGAIN}` : ''} Reason: `) + + `${failureShown(markerReason)}. ${timedOut ? TIMEOUT_ADVICE : FAILURE_ADVICE}`, }], { system: true, kind: 'inference-failed', reason: markerReason, consecutive: streak, ...classification, ...cutAt(markerReason) }, ); @@ -28878,6 +28924,7 @@ export class AgentFramework { this.featureSetManager, inferenceRouting ?? null, (event) => this.emitTrace(event as { type: TraceEvent['type']; [key: string]: unknown }), + () => this.inferenceBounds, ); // Checkpoint manager (Step 8), built and loaded at the top diff --git a/src/index.ts b/src/index.ts index 8bad14e76..96d2f21d5 100644 --- a/src/index.ts +++ b/src/index.ts @@ -5,6 +5,7 @@ export * from './types/index.js'; export { AgentFramework, BudgetPreflightError, DefaultErrorPolicy, ProviderWaitNotSleptError, QuiesceBlockedError, ResumeBlockedError } from './framework.js'; export type { RuntimeSettingsPreview, HostModeStatus } from './framework.js'; export { Agent } from './agent.js'; +export { FLAT_MINIMUM_MS, type InferenceBoundsOptions } from './inference-bounds.js'; export { storedInjectionMarkOf, type StoredInjectionMark } from './stored-injections.js'; export type { StartStreamResult } from './agent.js'; export { ProcessQueueImpl } from './queue.js'; diff --git a/src/inference-bounds.ts b/src/inference-bounds.ts new file mode 100644 index 000000000..90f172e6e --- /dev/null +++ b/src/inference-bounds.ts @@ -0,0 +1,111 @@ +import { MembraneError } from '@animalabs/membrane'; +import { PROGRESS_BOUNDS, resolveProgressBounds, type ProgressBounds } from '@animalabs/membrane/progress-bounds'; + +/** FrameworkConfig.inferenceBounds: how long the framework lets a model call + * go without progress. */ +export interface InferenceBoundsOptions { + /** Membrane's phase bounds for a stream whose adapter reports content + * (Anthropic, Bedrock), and the values the flat and whole-call bounds + * derive from: the larger of `beforeContentMs` and `betweenEventsMs`, + * plus `max_tokens` at `thinkTokensPerSecond`. Each one given replaces + * membrane's default (PROGRESS_BOUNDS). */ + progressBounds?: Partial; + /** The least the flat bound on a stream can be, from its last provider + * event. Default FLAT_MINIMUM_MS. */ + flatMinimumMs?: number; +} + +export const FLAT_MINIMUM_MS = 1_862_000; + +export interface InferenceBounds { + readonly progressBounds: Readonly; + readonly flatMinimumMs: number; +} + +export function resolveInferenceBounds(option: false | InferenceBoundsOptions | undefined): InferenceBounds | null { + if (option === false) return null; + if (option !== undefined && (typeof option !== 'object' || option === null)) { + throw new TypeError(`[agent-framework] inferenceBounds must be false or an object, got ${option === null ? 'null' : typeof option}`); + } + const given = option?.progressBounds; + if (given !== undefined && (typeof given !== 'object' || given === null)) { + throw new TypeError(`[agent-framework] inferenceBounds.progressBounds must be an object of bounds, got ${given === null ? 'null' : typeof given}`); + } + let progressBounds: ProgressBounds; + try { + progressBounds = resolveProgressBounds(given ?? true) ?? { ...PROGRESS_BOUNDS }; + } catch (refused) { + const why = refused instanceof Error ? refused.message.replace(/^\[membrane\] /, '') : String(refused); + throw new (refused instanceof TypeError ? TypeError : RangeError)(`[agent-framework] inferenceBounds.${why}`); + } + const flatMinimumMs: unknown = option?.flatMinimumMs === undefined ? FLAT_MINIMUM_MS : option.flatMinimumMs; + if (typeof flatMinimumMs !== 'number' || !Number.isFinite(flatMinimumMs) || flatMinimumMs <= 0) { + throw new RangeError(`[agent-framework] inferenceBounds.flatMinimumMs must be a finite number above 0, got ${String(flatMinimumMs)}`); + } + return Object.freeze({ progressBounds: Object.freeze(progressBounds), flatMinimumMs }); +} + +function allowanceMs(bounds: Readonly, maxTokens: number | undefined): number { + return typeof maxTokens === 'number' && Number.isFinite(maxTokens) && maxTokens > 0 + ? Math.ceil(maxTokens / bounds.thinkTokensPerSecond) * 1000 + : 0; +} + +function setupMs(bounds: Readonly): number { + return Math.max(bounds.beforeContentMs, bounds.betweenEventsMs); +} + +export function streamBounds( + bounds: InferenceBounds | null, + maxTokens: number | undefined, +): { progressBounds?: Readonly; progressTimeoutMs?: number } { + if (bounds === null) return {}; + const phases = bounds.progressBounds; + return { + progressBounds: phases, + progressTimeoutMs: Math.max(setupMs(phases) + allowanceMs(phases, maxTokens), bounds.flatMinimumMs), + }; +} + +export function wholeCallMs(bounds: InferenceBounds, maxTokens: number | undefined): number { + if (!(typeof maxTokens === 'number' && Number.isFinite(maxTokens) && maxTokens > 0)) return Math.max(setupMs(bounds.progressBounds), bounds.flatMinimumMs); + return setupMs(bounds.progressBounds) + allowanceMs(bounds.progressBounds, maxTokens); +} + +// membrane 0.8.0 marks a deadline it reports as run out with this registered +// symbol (on a sent request, or the Anthropic SDK's timeout, which covers its +// connect timeout too) and exports no reader or marker for it. The one reader: +// the deadline class here and the transient-failure check of an exhausted +// turn's re-arm both use it. +const DEADLINE_EXPIRY = Symbol.for('membrane.deadlineExpiry'); + +export function isDeadlineExpiry(error: unknown): boolean { + return typeof error === 'object' && error !== null && (error as Record)[DEADLINE_EXPIRY] === true; +} + +/** + * Names the host setting for the bound a stream cut hit, read from the opening + * of membrane 0.8.0's message for it (its progress-bounds.ts and + * providers/utils.ts), so the reason an operator and the agent see names the + * host's setting for that bound, as the README's table maps them. A message it + * doesn't know is left as it is. + */ +export function withHostSetting(message: string): string { + const setting = message.includes('of a thinking block\'s start') ? 'progressBounds.thinkTokensPerSecond' + : message.startsWith('No content within ') ? 'progressBounds.beforeContentMs' + : message.startsWith('No provider event within ') ? 'progressBounds.betweenEventsMs' + : message.startsWith('No provider progress within ') ? 'flatMinimumMs' + : undefined; + return setting === undefined ? message : `${message} (the host's setting for this bound: \`inferenceBounds.${setting}\`)`; +} + +export function deadlineCut(message: string | undefined): MembraneError { + const error = new MembraneError({ + type: 'timeout', + retryable: false, + message: message === undefined ? 'The provider made no progress within the call\'s bound' : withHostSetting(message), + rawError: undefined, + }); + Object.defineProperty(error, DEADLINE_EXPIRY, { value: true }); + return error; +} diff --git a/src/mcpl/inference-router.ts b/src/mcpl/inference-router.ts index b58aeabde..1a1aab124 100644 --- a/src/mcpl/inference-router.ts +++ b/src/mcpl/inference-router.ts @@ -27,6 +27,7 @@ import type { FeatureSetManager } from './feature-set-manager.js'; import { McplFeatureSetError } from './feature-set-manager.js'; import type { HookOrchestrator } from './hook-orchestrator.js'; import { logValue } from './log-value.js'; +import { streamBounds, wholeCallMs, withHostSetting, type InferenceBounds } from '../inference-bounds.js'; // ============================================================================ // Responder interface @@ -193,6 +194,7 @@ export class InferenceRouter { featureSetManager: FeatureSetManager, routingPolicy: InferenceRoutingPolicy | null, emitTraceFn: (event: { type: string; [key: string]: unknown }) => void, + private readonly inferenceBounds: () => InferenceBounds | null = () => null, ) { this.membrane = membrane; this.hookOrchestrator = hookOrchestrator; @@ -324,7 +326,7 @@ export class InferenceRouter { // builder makes the result from it. const response = chunkRequestId != null ? await this.streamResponse(request, chunkRequestId, sendChunk!) - : await this.membrane.complete(request); + : await this.membrane.complete(request, this.wholeCallBound(request)); // 6. Answer const result = resultFrom(response, model); @@ -407,6 +409,11 @@ export class InferenceRouter { // Private: Streaming // ========================================================================== + private wholeCallBound(request: NormalizedRequest): { progressTimeoutMs?: number } { + const bounds = this.inferenceBounds(); + return bounds === null ? {} : { progressTimeoutMs: wholeCallMs(bounds, request.config.maxTokens) }; + } + /** * Run the inference as a stream, sending each visible text delta to the * server as an inference/chunk, and return membrane's response once the @@ -425,7 +432,7 @@ export class InferenceRouter { requestId: string | number, sendChunk: (params: InferenceChunkParams) => void, ): Promise { - const stream = this.membrane.streamYielding(request); + const stream = this.membrane.streamYielding(request, streamBounds(this.inferenceBounds(), request.config.maxTokens)); let index = 0; for await (const event of stream) { @@ -448,7 +455,7 @@ export class InferenceRouter { case 'error': throw event.error; case 'aborted': - throw new Error(`The inference stopped before it finished (${event.reason})`); + throw new Error(`The inference stopped before it finished (${event.reason}${event.message ? `: ${event.reason === 'timeout' ? withHostSetting(event.message) : event.message}` : ''})`); } } throw new Error('The inference stream ended without a result'); diff --git a/src/types/framework.ts b/src/types/framework.ts index be5ed5455..116db08c9 100644 --- a/src/types/framework.ts +++ b/src/types/framework.ts @@ -6,6 +6,7 @@ import type { ProcessEvent } from './events.js'; import type { McplServerConfig, InferenceRoutingPolicy } from '../mcpl/types.js'; import type { ConversationRouterConfig } from '../mcpl/conversation-router.js'; import type { GateOptions } from '../gate/types.js'; +import type { InferenceBoundsOptions } from '../inference-bounds.js'; // Re-export trace types export type { TraceEvent, TraceEventListener } from './trace.js'; @@ -197,6 +198,28 @@ export interface FrameworkConfig { /** Custom error policy */ errorPolicy?: ErrorPolicy; + /** + * How long a model call may go without progress before it is cut. A + * resident's streamed call, and an MCPL server's, gets membrane's progress + * bounds where its adapter's stream reports content (Anthropic, Bedrock), + * and on every adapter a flat bound from its last provider event: the + * larger of a setup term (the larger of `beforeContentMs` and + * `betweenEventsMs`) plus the request's `maxTokens` at + * `thinkTokensPerSecond`, and `flatMinimumMs`. An auxiliary call + * (compression, the reply judge) and an MCPL server's non-streamed call get + * a bound on the whole call from its send, the setup term plus that + * allowance (the larger of the setup term and `flatMinimumMs` when it + * names no `maxTokens`), unless its + * caller passed a `progressTimeoutMs` of its own; where an adapter's + * `complete()` streams under the hood, membrane renews it at each parsed + * provider event. A call a bound cuts fails + * with type `timeout`, and on an agent's turn DefaultErrorPolicy gives each + * such call one fresh call. One setting covers every agent. `false` sets + * none of these bounds. Default: membrane's PROGRESS_BOUNDS and + * FLAT_MINIMUM_MS. + */ + inferenceBounds?: false | InferenceBoundsOptions; + /** * Host verdict on a failed inference: "this cannot succeed until later". * Consulted before the error policy and before the built-in @@ -456,7 +479,10 @@ export interface InferencePolicy { */ export interface ErrorPolicy { /** - * Handle an inference error. + * Handle an inference error. `attempt` is how many times the turn's + * request has failed before this one; for a deadline expiry (a progress + * bound's cut, or one membrane marks), how many times that call's request + * has expired in a row before this one. */ onInferenceError( error: Error, diff --git a/test/default-error-policy.test.ts b/test/default-error-policy.test.ts index 05a33794f..c6483d316 100644 --- a/test/default-error-policy.test.ts +++ b/test/default-error-policy.test.ts @@ -41,8 +41,8 @@ describe('DefaultErrorPolicy on membrane\'s schedules', () => { assert.deepEqual(delays(policy, err({ type: 'server', retryable: true, httpStatus: 502 })), [1000, 2000]); }); - it('retries a stream deadline as a retryable failure, as before membrane 0.7, where membrane gives it none', () => { - assert.deepEqual(delays(policy, err({ type: 'timeout', retryable: true, message: 'SSE stream idle timeout' })), [1000, 2000]); + it('retries a timeout membrane doesn\'t mark as a deadline, such as an HTTP 408, on the base schedule', () => { + assert.deepEqual(delays(policy, err({ type: 'timeout', retryable: true, httpStatus: 408, message: 'Request Timeout' })), [1000, 2000]); }); it('retries a rate limit with no stated wait at least five tries', () => { diff --git a/test/end-turn-usage.test.ts b/test/end-turn-usage.test.ts index e55ea6282..f4f6e8ab3 100644 --- a/test/end-turn-usage.test.ts +++ b/test/end-turn-usage.test.ts @@ -464,7 +464,7 @@ describe('a stream counts its usage once, however it ends', () => { for (const [name, ending] of [ ['a stream error', { type: 'error', error: new MembraneError({ type: 'auth', retryable: false, message: 'key revoked', rawError: { status: 401 } }) } as StreamEvent], - ['an abort', { type: 'aborted', reason: 'timeout' } as StreamEvent], + ['an abort', { type: 'aborted', reason: 'error' } as StreamEvent], ['a thrown stream', new Error('socket hang up')], ] as const) { it(`${name} after a finished round counts that round once and logs it with the failure`, async () => { diff --git a/test/inference-bounds.test.ts b/test/inference-bounds.test.ts new file mode 100644 index 000000000..7d8c66f4d --- /dev/null +++ b/test/inference-bounds.test.ts @@ -0,0 +1,768 @@ +import { after, before, describe, it, mock } from 'node:test'; +import assert from 'node:assert/strict'; +import { createServer, type IncomingMessage, type Server, type ServerResponse } from 'node:http'; +import type { AddressInfo } from 'node:net'; +import { existsSync, mkdtempSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { AnthropicAdapter, Membrane, MembraneError, NativeFormatter, OpenAICompatibleAdapter } from '@animalabs/membrane'; +import { PROGRESS_BOUNDS, type ProgressBounds } from '@animalabs/membrane/progress-bounds'; +import { MEMBRANE_RETRY_SETTINGS, type RetrySettings } from '@animalabs/membrane/retry'; + +import { AgentFramework, DefaultErrorPolicy, FLAT_MINIMUM_MS } from '../src/index.js'; +import type { ErrorAction, EventResponse, Module, ModuleContext, ProcessEvent, ProcessState, ToolCall, ToolDefinition, ToolResult } from '../src/index.js'; +import type { InferenceRouter } from '../src/mcpl/inference-router.js'; + +type Play = + | 'whole' | 'error500' | 'error408' | 'tool-use' | 'refusal' + | 'quiet-before-content' | 'silent-after-start' | 'pings-after-content' | 'deltas-after-content' | 'think-then-pings' | 'content-then-think' + | 'openai-comments' | 'openai-headers-only' + | 'no-headers' | 'body-trickle'; + +const TICK_MS = 50; + +// The file runs on one stepped clock: node:test's mocked setTimeout, +// setInterval and Date, with performance.now (which membrane's bounds compare +// against, and mock.timers doesn't cover) read from the same mocked Date, from +// a whole millisecond. The sockets to the stand-in provider stay real. The +// clock moves STEP_MS once per turn of the real event loop, and not while a +// file or lookup request is in flight. The stand-in runs in this process over +// loopback, so reading what it writes takes loop turns, not wall time, and +// the shortest bound here is 60 turns: a verdict rests on loop turns, not on +// the machine's load. +// +// It's one clock for the file because a clock reset between tests left later +// turns with failed calls or tool rounds unable to stop, and a fractional +// performance.now base froze some runs of the file outright: both by runs, +// their causes unread. +const STEP_MS = 5; +const REAL_WORK = /^(FSReq|GetAddrInfo|GetNameInfo)/; +const realSetImmediate = setImmediate; +let clockRuns = false; +before(() => { + mock.timers.enable({ apis: ['setTimeout', 'setInterval', 'Date'], now: Date.now() }); + const dateBase = Date.now(); + const perfBase = Math.floor(performance.now()); + Object.defineProperty(performance, 'now', { configurable: true, writable: true, value: () => perfBase + (Date.now() - dateBase) }); + clockRuns = true; + const step = (): void => { + if (!clockRuns) return; + try { + if (!process.getActiveResourcesInfo().some((kind) => REAL_WORK.test(kind))) mock.timers.tick(STEP_MS); + } finally { + realSetImmediate(step); + } + }; + realSetImmediate(step); +}); +after(() => { + clockRuns = false; + mock.timers.reset(); + delete (performance as { now?: unknown }).now; +}); +const NOT_ENDED = Symbol('not ended'); +const within = (work: Promise, ms: number): Promise => { + let timer: ReturnType | undefined; + return Promise.race([work, new Promise((resolve) => { timer = setTimeout(() => resolve(NOT_ENDED), ms); })]) + .finally(() => clearTimeout(timer)); +}; +const until = async (check: () => boolean, what: string, ms = 8_000): Promise => { + const end = Date.now() + ms; + while (!check()) { + if (Date.now() > end) throw new Error(`timed out waiting for ${what}`); + await new Promise((resolve) => setTimeout(resolve, 10)); + } +}; +const SHORT = { progressBounds: { beforeContentMs: 300, betweenEventsMs: 300, thinkTokensPerSecond: 1_000_000 }, flatMinimumMs: 400 }; +const SHORT_WITH_RATE = (thinkTokensPerSecond: number) => ({ ...SHORT, progressBounds: { ...SHORT.progressBounds, thinkTokensPerSecond } }); + +const sse = (type: string, data: Record) => `event: ${type}\ndata: ${JSON.stringify({ type, ...data })}\n\n`; +const START = sse('message_start', { message: { id: 'msg_t', type: 'message', role: 'assistant', model: 't', content: [], stop_reason: null, stop_sequence: null, usage: { input_tokens: 10, output_tokens: 1 } } }); +const TEXT_START = sse('content_block_start', { index: 0, content_block: { type: 'text', text: '' } }); +const delta = (text: string) => sse('content_block_delta', { index: 0, delta: { type: 'text_delta', text } }); +const ending = (stopReason: string) => sse('content_block_stop', { index: 0 }) + sse('message_delta', { delta: { stop_reason: stopReason, stop_sequence: null }, usage: { output_tokens: 5 } }) + sse('message_stop', {}); +const OPENAI_CHUNK = `data: ${JSON.stringify({ id: 'c', object: 'chat.completion.chunk', created: 1, model: 't', choices: [{ index: 0, delta: { role: 'assistant', content: 'Thinking it over' }, finish_reason: null }] })}\n\n`; + +interface Provider { base: string; requests: () => number; arrivals: number[]; close: () => Promise } + +async function provider(plays: Play[]): Promise { + let n = 0; + const arrivals: number[] = []; + const timers = new Set>(); + const sockets = new Set(); + const server: Server = createServer((req: IncomingMessage, res: ServerResponse) => { + req.resume(); + req.on('end', () => { + if (String(req.url).includes('count_tokens')) { + res.writeHead(200, { 'content-type': 'application/json' }); + res.end(JSON.stringify({ input_tokens: 10 })); + return; + } + const play = plays[Math.min(n, plays.length - 1)]!; + n += 1; + arrivals.push(Date.now()); + const every = (write: () => void) => { + const timer = setInterval(write, TICK_MS); + timers.add(timer); + res.on('close', () => clearInterval(timer)); + }; + const stream = () => res.writeHead(200, { 'content-type': 'text/event-stream' }); + switch (play) { + case 'whole': + stream(); + res.end(START + TEXT_START + delta('The answer.') + ending('end_turn')); + return; + case 'refusal': + stream(); + res.end(START + TEXT_START + delta('I can') + ending('refusal')); + return; + case 'tool-use': + stream(); + res.end(START + + sse('content_block_start', { index: 0, content_block: { type: 'tool_use', id: `toolu_${n}`, name: 'look', input: {} } }) + + sse('content_block_delta', { index: 0, delta: { type: 'input_json_delta', partial_json: '{}' } }) + + ending('tool_use')); + return; + case 'error408': + res.writeHead(408, { 'content-type': 'application/json' }); + res.end(JSON.stringify({ type: 'error', error: { type: 'timeout_error', message: 'stand-in request timeout' } })); + return; + case 'error500': + res.writeHead(500, { 'content-type': 'application/json' }); + res.end(JSON.stringify({ type: 'error', error: { type: 'api_error', message: 'stand-in internal error' } })); + return; + case 'quiet-before-content': + stream(); + res.write(START); + every(() => res.write(sse('ping', {}))); + return; + case 'silent-after-start': + stream(); + res.write(START); + return; + case 'pings-after-content': + stream(); + res.write(START + TEXT_START + delta('Thinking it over')); + every(() => res.write(sse('ping', {}))); + return; + case 'deltas-after-content': + stream(); + res.write(START + TEXT_START + delta('Thinking it over')); + every(() => res.write(delta('.'))); + return; + case 'content-then-think': + stream(); + res.write(START + TEXT_START + delta('Thinking it over') + sse('content_block_stop', { index: 0 }) + sse('content_block_start', { index: 1, content_block: { type: 'thinking', thinking: '', signature: '' } })); + every(() => res.write(sse('ping', {}))); + return; + case 'think-then-pings': + stream(); + res.write(START + sse('content_block_start', { index: 0, content_block: { type: 'thinking', thinking: '', signature: '' } })); + every(() => res.write(sse('ping', {}))); + return; + case 'openai-comments': + stream(); + res.write(OPENAI_CHUNK); + every(() => res.write(': keepalive\n\n')); + return; + case 'openai-headers-only': + stream(); + res.flushHeaders(); + return; + case 'no-headers': + return; + case 'body-trickle': + res.writeHead(200, { 'content-type': 'application/json' }); + res.write('{'); + every(() => res.write(' ')); + return; + } + }); + }); + server.on('connection', (socket) => { sockets.add(socket); socket.on('close', () => sockets.delete(socket)); }); + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); + return { + base: `http://127.0.0.1:${(server.address() as AddressInfo).port}`, + requests: () => n, + arrivals, + close: async () => { + for (const timer of timers) clearInterval(timer); + for (const socket of sockets) socket.destroy(); + await new Promise((resolve) => server.close(() => resolve())); + }, + }; +} + +class Asker implements Module { + readonly name = 'asker'; + async start(_ctx: ModuleContext): Promise {} + async stop(): Promise {} + getTools(): ToolDefinition[] { return [{ name: 'look', description: 'Look around', inputSchema: { type: 'object', properties: {} } }]; } + async handleToolCall(_call: ToolCall): Promise { return { success: true, data: { saw: 'a quiet room' } }; } + async onProcess(event: ProcessEvent, _state: ProcessState): Promise { + if (event.type !== 'external-message') return {}; + return { addMessages: [{ participant: 'Antra', content: [{ type: 'text', text: String((event as { content?: unknown }).content) }] }], requestInference: true }; + } +} + +class Recording extends Membrane { + readonly sent: Array<{ kind: 'stream' | 'complete'; options: Record | undefined }> = []; + override streamYielding(...args: Parameters) { + this.sent.push({ kind: 'stream', options: args[1] as Record | undefined }); + return super.streamYielding(...args); + } + override complete(...args: Parameters) { + this.sent.push({ kind: 'complete', options: args[1] as Record | undefined }); + return super.complete(...args); + } +} + +class WatchdogsAt200ms extends Membrane { + override streamYielding(...args: Parameters) { + return super.streamYielding(args[0], { ...args[1], idleTimeoutMs: 200, firstEventTimeoutMs: 200 } as Parameters[1]); + } +} + +const adapterFor = (stand: Provider, openai = false) => openai + ? new OpenAICompatibleAdapter({ apiKey: 'test-key', baseURL: stand.base } as never) + : new AnthropicAdapter({ apiKey: 'test-key', baseURL: stand.base, cacheKeepalive: { enabled: false } } as never); + +async function framework(stand: Provider, options: { openai?: boolean; inferenceBounds?: unknown; agent?: Record; errors?: Error[]; membrane?: Membrane; retries?: RetrySettings; rearmMs?: number[] } = {}) { + const policy = new DefaultErrorPolicy(options.retries); + return AgentFramework.create({ + storePath: join(mkdtempSync(join(tmpdir(), 'af-bounds-')), 'store.chronicle'), + membrane: options.membrane ?? new Membrane(adapterFor(stand, options.openai)), + agents: [{ name: 'resident', model: options.openai ? 'gpt-test' : 'claude-opus-5-5', systemPrompt: 'You are a resident.', maxTokens: 4096, ...options.agent } as never], + modules: [new Asker()], + syncIntervalMs: 0, + maintenanceIntervalMs: 0, + ...(options.rearmMs ? { exhaustedTurnRearmMs: options.rearmMs } : {}), + errorPolicy: { + maxRetries: policy.maxRetries, + onInferenceError(error: Error, agentName: string, attempt: number): ErrorAction { + options.errors?.push(error); + return policy.onInferenceError(error, agentName, attempt); + }, + }, + ...('inferenceBounds' in options ? (options.inferenceBounds === undefined ? {} : { inferenceBounds: options.inferenceBounds as never }) : { inferenceBounds: SHORT }), + }); +} + +interface Turn { + requests: number; + arrivals: number[]; + traces: Array<{ type: string; error?: unknown }>; + markers: string[]; + errors: Error[]; + idle: boolean; + endedAt: number; +} + +async function turn(plays: Play[], options: { openai?: boolean; inferenceBounds?: unknown; agent?: Record; waitMs?: number; stopInFirstWait?: boolean; membrane?: (stand: Provider) => Membrane; retries?: RetrySettings; rearmMs?: number[]; untilRequests?: number } = {}): Promise { + const stand = await provider(plays); + const errors: Error[] = []; + const { membrane: makeMembrane, untilRequests, ...rest } = options; + const fw = await framework(stand, { ...rest, errors, ...(makeMembrane ? { membrane: makeMembrane(stand) } : {}) }); + const traces: Turn['traces'] = []; + fw.onTrace((event) => { + const type = (event as { type: string }).type; + if (type.startsWith('inference:') && type !== 'inference:tokens') traces.push({ type, error: (event as { error?: unknown }).error }); + }); + const error = console.error; + console.error = () => {}; + let idle = false; + let stoppedInWait: boolean | undefined; + let requestsWhenStopped: number | undefined; + try { + fw.start(); + fw.pushEvent({ type: 'external-message', source: 'test', content: 'Hello?', metadata: {} } as unknown as ProcessEvent); + if (options.stopInFirstWait) { + await until(() => traces.some((t) => t.type === 'inference:waiting'), 'the first wait to send a cut call again'); + requestsWhenStopped = stand.requests(); + stoppedInWait = Boolean(fw.abortInference('resident', 'the test stops it during its wait to retry')); + await fw.runUntilIdle(); + fw.pushEvent({ type: 'external-message', source: 'test', content: 'And now?', metadata: {} } as unknown as ProcessEvent); + } + if (untilRequests !== undefined) { + await until(() => stand.requests() >= untilRequests, `request ${untilRequests}`, 15_000); + await new Promise((resolve) => setTimeout(resolve, 3_000)); + } + idle = (await within(fw.runUntilIdle().then(() => 'idle' as const), options.waitMs ?? 15_000)) === 'idle'; + const endedAt = Date.now(); + const rows = (fw.getAgent('resident')!.getContextManager() as unknown as { getAllMessages(): Array<{ content: Array<{ type: string; text?: string }> }> }).getAllMessages(); + const markers = rows.flatMap((row) => row.content).flatMap((block) => block.type === 'text' && block.text?.startsWith('[inference-failed]') ? [block.text] : []); + return { requests: stand.requests(), arrivals: [...stand.arrivals], traces, markers, errors, idle, endedAt, stoppedInWait, requestsWhenStopped }; + } finally { + console.error = error; + await fw.stop(); + await stand.close(); + } +} + +const exhausted = (t: Turn) => t.traces.filter((e) => e.type === 'inference:exhausted'); +const completed = (t: Turn) => t.traces.filter((e) => e.type === 'inference:completed'); +const gap = (t: Turn, i: number) => t.arrivals[i + 1]! - t.arrivals[i]!; + +describe('inference bounds: a stalled turn call ends at its bound', () => { + it('an Anthropic stream with no content ends at beforeContentMs, gets one fresh call 1 s later, and the turn then ends as a timeout', async () => { + const t = await turn(['quiet-before-content']); + assert.equal(t.idle, true); + assert.equal(t.requests, 2); + assert.equal(exhausted(t).length, 1); + assert.match(String(exhausted(t)[0]!.error), /^No content within 0\.3s of the send/); + assert.match(String(exhausted(t)[0]!.error), /\(the host's setting for this bound: `inferenceBounds\.progressBounds\.beforeContentMs`\)$/); + assert.ok(gap(t, 0) >= 1_250 && gap(t, 0) < 1_450, `the fresh call came ${gap(t, 0)} ms after the first, its 0.3 s bound and the 1 s wait`); + }); + + it('an Anthropic stream that only pings after content ends at betweenEventsMs, twice', async () => { + const t = await turn(['pings-after-content']); + assert.equal(t.requests, 2); + assert.match(String(exhausted(t)[0]!.error), /^No provider event within 0\.3s/); + assert.match(String(exhausted(t)[0]!.error), /\(the host's setting for this bound: `inferenceBounds\.progressBounds\.betweenEventsMs`\)$/); + }); + + it('a thinking block that starts before content gets its allowance, max_tokens at thinkTokensPerSecond, not beforeContentMs', async () => { + const t = await turn(['think-then-pings'], { inferenceBounds: SHORT_WITH_RATE(2_048) }); + assert.equal(t.requests, 2); + assert.match(String(exhausted(t)[0]!.error), /^No content within 2s of a thinking block's start \(max_tokens 4096 at 2048 tokens a second\)/); + assert.match(String(exhausted(t)[0]!.error), /\(the host's setting for this bound: `inferenceBounds\.progressBounds\.thinkTokensPerSecond`\)$/); + assert.ok(gap(t, 0) >= 2_900 && gap(t, 0) < 3_100, `the fresh call came ${gap(t, 0)} ms after the first, its 2 s allowance and the 1 s wait`); + }); + + it('a thinking block that starts after content gets its allowance in place of betweenEventsMs when that is longer', async () => { + const t = await turn(['content-then-think'], { inferenceBounds: SHORT_WITH_RATE(2_048) }); + assert.equal(t.requests, 2); + assert.match(String(exhausted(t)[0]!.error), /^No provider event within 2s of a thinking block's start \(max_tokens 4096 at 2048 tokens a second\)/); + assert.match(String(exhausted(t)[0]!.error), /\(the host's setting for this bound: `inferenceBounds\.progressBounds\.thinkTokensPerSecond`\)$/); + assert.ok(gap(t, 0) >= 2_900 && gap(t, 0) < 3_100, `the fresh call came ${gap(t, 0)} ms after the first, its 2 s allowance and the 1 s wait`); + }); + + it('an OpenAI-compatible stream that sends only keepalive comments ends at the flat bound, twice', async () => { + const t = await turn(['openai-comments'], { openai: true }); + assert.equal(t.idle, true); + assert.equal(t.requests, 2); + assert.equal(exhausted(t).length, 1); + assert.match(String(exhausted(t)[0]!.error), /^No provider progress within 1300ms/); + assert.match(String(exhausted(t)[0]!.error), /\(the host's setting for this bound: `inferenceBounds\.flatMinimumMs`\)$/); + assert.ok(gap(t, 0) >= 2_250 && gap(t, 0) < 2_450, `the fresh call came ${gap(t, 0)} ms after the first, its 1.3 s flat bound and the 1 s wait`); + }); + + it('an OpenAI-compatible stream that sends its headers and no event ends at the flat bound from its send', async () => { + const t = await turn(['openai-headers-only'], { openai: true }); + assert.equal(t.requests, 2); + assert.match(String(exhausted(t)[0]!.error), /^No provider progress within 1300ms/); + }); + + it('a deadline expiry membrane throws on a turn gets one fresh call, and the turn then ends', async () => { + const t = await turn(['silent-after-start'], { + inferenceBounds: { ...SHORT, progressBounds: { ...SHORT.progressBounds, beforeContentMs: 5_000, betweenEventsMs: 5_000 } }, + membrane: (stand) => new WatchdogsAt200ms(adapterFor(stand)), + }); + assert.equal(t.requests, 2); + assert.equal(exhausted(t).length, 1); + assert.equal(t.errors.length, 2); + for (const thrown of t.errors) { + assert.ok(thrown instanceof MembraneError && thrown.type === 'timeout' && thrown.retryable === true, String(thrown)); + assert.equal((thrown as unknown as Record)[Symbol.for('membrane.deadlineExpiry')], true); + } + }); + + it('flatMinimumMs sets the flat bound\'s least', async () => { + const t = await turn(['openai-comments'], { openai: true, inferenceBounds: { ...SHORT, flatMinimumMs: 2_000 } }); + assert.equal(t.requests, 2); + assert.match(String(exhausted(t)[0]!.error), /^No provider progress within 2000ms/); + }); + + it('the flat bound takes beforeContentMs as its setup when that\'s the larger', async () => { + const t = await turn(['openai-comments'], { openai: true, inferenceBounds: { ...SHORT, progressBounds: { ...SHORT.progressBounds, beforeContentMs: 1_500 } } }); + assert.equal(t.requests, 2); + assert.match(String(exhausted(t)[0]!.error), /^No provider progress within 2500ms/); + }); + + it('a fresh call that answers completes the turn, with no failure marker', async () => { + const t = await turn(['quiet-before-content', 'whole']); + assert.equal(t.requests, 2); + assert.equal(completed(t).length, 1); + assert.equal(exhausted(t).length, 0); + assert.deepEqual(t.markers, []); + }); + + it('a cut gets its fresh call whatever other failures the turn had before it', async () => { + const t = await turn(['error500', 'error500', 'quiet-before-content', 'whole']); + assert.equal(t.requests, 4); + assert.equal(completed(t).length, 1); + }); + + it('a cut gets its fresh call after an earlier round\'s cut was answered by its own fresh call', async () => { + const t = await turn(['quiet-before-content', 'tool-use', 'quiet-before-content', 'whole']); + assert.equal(t.requests, 4); + assert.equal(completed(t).length, 1); + }); + + it('a cut gets its fresh call when another failure came between it and an earlier cut', async () => { + const t = await turn(['quiet-before-content', 'error500', 'quiet-before-content', 'whole']); + assert.equal(t.requests, 4); + assert.equal(completed(t).length, 1); + }); + + it('a fresh call refused, sent again by membrane and cut is the same request\'s second cut, so the turn ends', async () => { + const t = await turn(['quiet-before-content', 'refusal', 'quiet-before-content', 'whole'], { agent: { refusalHandling: { retries: 1 } } }); + assert.equal(t.requests, 3); + assert.equal(exhausted(t).length, 1); + assert.equal(completed(t).length, 0); + assert.equal(t.markers.length, 1); + assert.match(t.markers[0]!, /The call was made again, and it timed out again\./, 'membrane\'s re-send is inside the one fresh call'); + }); + + it('a cut counts among the turn\'s tries for its other failures, as every failure does', async () => { + const t = await turn(['quiet-before-content', 'error500', 'error500', 'whole']); + assert.equal(t.requests, 3); + assert.equal(exhausted(t).length, 1); + assert.match(String(exhausted(t)[0]!.error), /stand-in internal error/); + assert.equal(t.markers.length, 1); + assert.doesNotMatch(t.markers[0]!, /made again/, 'the turn ended on a 500, not on a fresh call that timed out'); + }); + + it('a turn sent again after its retries ran out gets no fresh call for a cut before a round of it has run, as for any failure', async () => { + const t = await turn(['error500', 'error500', 'error500', 'quiet-before-content', 'whole'], { rearmMs: [100], untilRequests: 4 }); + assert.equal(t.requests, 4, 'three 500s, then the re-armed send, cut once and not sent again'); + assert.equal(completed(t).length, 0); + assert.match(String(exhausted(t).at(-1)!.error), /^No content within 0\.3s of the send/); + }); + + it('a turn sent again after its retries ran out gets its fresh call for a cut once a round of it has run', async () => { + const t = await turn(['error500', 'error500', 'error500', 'tool-use', 'quiet-before-content', 'whole'], { rearmMs: [100], untilRequests: 6 }); + assert.equal(t.requests, 6, 'three 500s, the re-armed send\'s tool round, its cut call and that call\'s fresh call'); + assert.equal(completed(t).length, 1); + }); + + it('a turn stopped while it waits to send a cut call again leaves the next turn its own fresh call', async () => { + const t = await turn(['quiet-before-content', 'quiet-before-content', 'whole'], { stopInFirstWait: true }); + assert.equal(t.requestsWhenStopped, 1); + assert.equal(t.stoppedInWait, true); + assert.equal(t.requests, 3); + assert.equal(completed(t).length, 1); + assert.equal(exhausted(t).length, 0); + assert.deepEqual(t.markers, []); + }); + + it('a bound\'s cut reaches the error policy as a timeout that isn\'t retryable, carries membrane\'s deadline mark and names the bound', async () => { + const t = await turn(['quiet-before-content']); + assert.equal(t.errors.length, 2); + for (const cut of t.errors) { + assert.ok(cut instanceof MembraneError); + assert.equal(cut.type, 'timeout'); + assert.equal(cut.retryable, false); + assert.equal((cut as unknown as Record)[Symbol.for('membrane.deadlineExpiry')], true); + assert.match(cut.message, /^No content within 0\.3s of the send/); + } + }); + + it('a turn its cuts end tells the agent the call timed out, that the call was made again and timed out again, why, the host setting for that bound, and that a reply still owed can be given now', async () => { + const t = await turn(['quiet-before-content']); + assert.equal(t.markers.length, 1); + assert.match(t.markers[0]!, /^\[inference-failed\] Your previous turn did not complete: the model call timed out and produced no response, so nothing was sent\. The call was made again, and it timed out again\. Reason: No content within 0\.3s of the send/); + assert.match(t.markers[0]!, /\(the host's setting for this bound: `inferenceBounds\.progressBounds\.beforeContentMs`\)\. If a reply is still owed from that turn, you can give it now\.$/); + }); + + it('a host schedule that gives a deadline expiry two fresh calls says the call was made again, with no count', async () => { + const retries = { ...MEMBRANE_RETRY_SETTINGS, deadline: { maxAttempts: 3, firstDelayMs: 50, multiplier: 1, maxDelayMs: 50, statedWaitCeilingMs: null } } as RetrySettings; + const t = await turn(['quiet-before-content', 'quiet-before-content', 'quiet-before-content', 'whole'], { retries }); + assert.equal(t.requests, 3); + assert.equal(t.markers.length, 1); + assert.match(t.markers[0]!, /so nothing was sent\. The call was made again, and it timed out again\. Reason: /); + assert.doesNotMatch(t.markers[0]!, /once more|more times/); + }); + + it('a turn whose cuts come either side of another failure says the call was made again and timed out again, with no count', async () => { + const t = await turn(['quiet-before-content', 'error500', 'quiet-before-content', 'quiet-before-content', 'whole']); + assert.equal(t.requests, 4, 'a cut, its fresh call\'s 500, a retry cut, and that cut\'s fresh call cut too'); + assert.equal(t.markers.length, 1); + assert.match(t.markers[0]!, /so nothing was sent\. The call was made again, and it timed out again\. Reason: No content within 0\.3s of the send/); + assert.doesNotMatch(t.markers[0]!, /once more|more times/); + }); + + it('a timeout that got no fresh call, such as an HTTP 408, says neither that it was sent again nor more than its cause', async () => { + const t = await turn(['error408']); + assert.equal(t.requests, 3); + assert.equal(t.markers.length, 1); + assert.match(t.markers[0]!, /^\[inference-failed\] Your previous turn did not complete: the model call timed out \(timeout, HTTP 408\) and produced no response, so nothing was sent\. Reason: /); + assert.doesNotMatch(t.markers[0]!, /made again/); + assert.match(t.markers[0]!, /\. If a reply is still owed from that turn, you can give it now\.$/); + }); + + it('a turn cut after a tool round tells the agent its last call timed out and its earlier rounds ran', async () => { + const t = await turn(['tool-use', 'quiet-before-content']); + assert.equal(t.requests, 3); + assert.equal(t.markers.length, 1); + assert.match(t.markers[0]!, /^\[inference-failed\] Your previous turn did not complete: its last call timed out\. The call was made again, and it timed out again\. Its earlier rounds ran, and their tool calls ran as their results show\./); + assert.match(t.markers[0]!, /Reason: No content within 0\.3s of the send/); + assert.match(t.markers[0]!, /\. If a reply is still owed from that turn, you can give it now\.$/); + }); + + it('a stream that keeps sending content is not ended by any progress bound', async () => { + const t = await turn(['deltas-after-content'], { waitMs: 5_000, inferenceBounds: { ...SHORT, progressBounds: { ...SHORT.progressBounds, betweenEventsMs: 1_000 } } }); + assert.equal(t.idle, false); + assert.equal(t.requests, 1); + assert.ok(t.endedAt - t.arrivals[0]! >= 4_000, `the turn was still in its one call ${t.endedAt - t.arrivals[0]!} ms after it arrived, twice its 2 s flat bound`); + }); + + it('inferenceBounds: false sets none of these bounds on a turn', async () => { + let recording: Recording | undefined; + const t = await turn(['quiet-before-content'], { inferenceBounds: false, waitMs: 3_000, membrane: (stand) => (recording = new Recording(adapterFor(stand))) }); + assert.equal(t.idle, false); + assert.equal(t.requests, 1); + const stream = recording!.sent.find((s) => s.kind === 'stream')!; + assert.equal(stream.options?.progressBounds, undefined); + assert.equal(stream.options?.progressTimeoutMs, undefined); + }); +}); + +describe('inference bounds: what a host sets', () => { + for (const [given, field] of [ + [{ progressBounds: { beforeContentMs: -1 } }, 'inferenceBounds.progressBounds.beforeContentMs'], + [{ progressBounds: { betweenEventsMs: 'soon' } }, 'inferenceBounds.progressBounds.betweenEventsMs'], + [{ progressBounds: { thinkTokensPerSecond: 0 } }, 'inferenceBounds.progressBounds.thinkTokensPerSecond'], + [{ progressBounds: 5 }, 'inferenceBounds.progressBounds'], + [{ progressBounds: null }, 'inferenceBounds.progressBounds'], + [{ flatMinimumMs: 0 }, 'inferenceBounds.flatMinimumMs'], + [{ flatMinimumMs: -5 }, 'inferenceBounds.flatMinimumMs'], + [{ flatMinimumMs: Infinity }, 'inferenceBounds.flatMinimumMs'], + [{ flatMinimumMs: NaN }, 'inferenceBounds.flatMinimumMs'], + [{ flatMinimumMs: '5' }, 'inferenceBounds.flatMinimumMs'], + [{ flatMinimumMs: null }, 'inferenceBounds.flatMinimumMs'], + ['on', 'inferenceBounds'], + ] as const) { + it(`${JSON.stringify(given)} is refused at create, naming ${field}, before a store is opened`, async () => { + const storePath = join(mkdtempSync(join(tmpdir(), 'af-bounds-')), 'store.chronicle'); + await assert.rejects( + AgentFramework.create({ + storePath, + membrane: new Membrane(new AnthropicAdapter({ apiKey: 'k' } as never)), + agents: [{ name: 'resident', model: 'claude-opus-5-5', systemPrompt: 'x' } as never], + modules: [], + inferenceBounds: given as never, + }), + (error: Error) => error.message.startsWith(`[agent-framework] ${field}`), + ); + assert.equal(existsSync(storePath), false); + }); + } + + it('with no inferenceBounds a turn\'s stream is sent membrane\'s progress bounds and the 1,862 s flat minimum, and compression the whole-call bound', async () => { + const stand = await provider(['whole']); + const membrane = new Recording(adapterFor(stand)); + const fw = await framework(stand, { inferenceBounds: undefined, membrane }); + try { + fw.start(); + fw.pushEvent({ type: 'external-message', source: 'test', content: 'Hello?', metadata: {} } as unknown as ProcessEvent); + await fw.runUntilIdle(); + const compression = (fw.getAgent('resident')!.getContextManager() as unknown as { membrane: Membrane }).membrane; + await compression.complete({ messages: [{ participant: 'user', content: [{ type: 'text', text: 'Summarize.' }] }], config: { model: 'claude-opus-5-5', maxTokens: 16_000 } } as never, { formatter: new NativeFormatter() } as never) + .catch(() => 'the stand-in answers in a stream, which a non-streamed call can\'t read; the options it was sent are what this checks'); + } finally { + await fw.stop(); + await stand.close(); + } + const stream = membrane.sent.find((s) => s.kind === 'stream')!; + assert.deepEqual({ ...(stream.options!.progressBounds as ProgressBounds) }, { ...PROGRESS_BOUNDS }); + assert.equal(stream.options!.progressTimeoutMs, FLAT_MINIMUM_MS); + assert.equal(FLAT_MINIMUM_MS, 1_862_000); + const complete = membrane.sent.find((s) => s.kind === 'complete')!; + assert.equal(complete.options!.progressTimeoutMs, 334_000, '150 s, plus 16,000 tokens at 87 a second'); + }); +}); + +describe('inference bounds: a whole non-streamed call', () => { + async function auxiliary(play: Play, inferenceBounds: unknown, options: { callerOptions?: Record; maxTokens?: number | null; through?: 'compression' | 'judge' } = {}) { + const stand = await provider([play]); + const recording = new Recording(adapterFor(stand)); + const fw = await framework(stand, { inferenceBounds, membrane: recording }); + const membrane = options.through === 'judge' + ? (fw as unknown as { judgeMembraneFor(name: string): Membrane }).judgeMembraneFor('resident') + : (fw.getAgent('resident')!.getContextManager() as unknown as { membrane: Membrane }).membrane; + const maxTokens = options.maxTokens === undefined ? 4096 : options.maxTokens; + const sentAt = Date.now(); + try { + const ended = await within(membrane.complete({ + messages: [{ participant: 'user', content: [{ type: 'text', text: 'Summarize.' }] }], + config: { model: 'claude-opus-5-5', ...(maxTokens === null ? {} : { maxTokens }) }, + } as never, { formatter: new NativeFormatter(), ...options.callerOptions } as never), 8_000); + return { threw: ended === NOT_ENDED ? 'still waiting' : undefined as unknown, endedAfterMs: Date.now() - sentAt, requests: stand.requests(), sent: recording.sent.find((c) => c.kind === 'complete')?.options }; + } catch (error) { + return { threw: error, endedAfterMs: Date.now() - sentAt, requests: stand.requests(), sent: recording.sent.find((c) => c.kind === 'complete')?.options }; + } finally { + await fw.stop(); + await stand.close(); + } + } + // On the stepped clock a call's end is exact to a few steps, so the bound is + // checked by when the call ended as well as by membrane's message. + const timedOutWithin = ({ threw, endedAfterMs }: { threw: unknown; endedAfterMs: number }, ms: number) => { + assert.ok(threw instanceof MembraneError && threw.type === 'timeout', String(threw)); + assert.match(threw.message, new RegExp(`^No provider progress within ${ms}ms`)); + assert.ok(endedAfterMs >= ms && endedAfterMs < ms + 100, `the call ended ${endedAfterMs} ms after its send, against its ${ms} ms bound`); + }; + + for (const play of ['no-headers', 'body-trickle'] as const) { + it(`a resident's compression call to a provider that ${play === 'no-headers' ? 'never answers' : 'trickles its body'} ends at its setup plus its allowance, as a timeout that isn't retryable`, async () => { + const call = await auxiliary(play, SHORT); + timedOutWithin(call, 1_300); + assert.equal((call.threw as MembraneError).retryable, false); + }); + } + + it('the reply judge\'s call ends at the same bound', async () => { + const call = await auxiliary('no-headers', SHORT, { through: 'judge' }); + timedOutWithin(call, 1_300); + }); + + it('the allowance is max_tokens at thinkTokensPerSecond', async () => { + const call = await auxiliary('no-headers', SHORT_WITH_RATE(4_096), { maxTokens: 12_288 }); + timedOutWithin(call, 3_300); + }); + + it('the setup term is beforeContentMs when that\'s the larger', async () => { + const call = await auxiliary('no-headers', { ...SHORT, progressBounds: { ...SHORT.progressBounds, beforeContentMs: 1_500 } }); + timedOutWithin(call, 2_500); + }); + + it('a call that names no max_tokens ends at the larger of the setup term and the flat minimum', async () => { + const call = await auxiliary('no-headers', { ...SHORT, flatMinimumMs: 2_000 }, { maxTokens: null }); + timedOutWithin(call, 2_000); + const setupLarger = await auxiliary('no-headers', { ...SHORT, progressBounds: { ...SHORT.progressBounds, beforeContentMs: 1_500 } }, { maxTokens: null }); + timedOutWithin(setupLarger, 1_500); + }); + + it('a caller\'s own progressTimeoutMs is kept in place of the framework\'s', async () => { + const call = await auxiliary('no-headers', SHORT, { callerOptions: { progressTimeoutMs: 2_500 } }); + timedOutWithin(call, 2_500); + }); + + it('a caller\'s progressBounds, which complete() doesn\'t read, are passed on beside the framework\'s bound', async () => { + const call = await auxiliary('no-headers', SHORT, { callerOptions: { progressBounds: { beforeContentMs: 60_000 } } }); + const sent = call.sent; + timedOutWithin(call, 1_300); + assert.deepEqual(sent?.progressBounds, { beforeContentMs: 60_000 }); + assert.equal(sent?.progressTimeoutMs, 1_300); + }); + + it('inferenceBounds: false sets no bound on a compression call', async () => { + const { threw, requests, sent } = await auxiliary('no-headers', false); + assert.equal(requests, 1); + assert.equal(threw, 'still waiting', 'not ended in 8 s, past every bound SHORT would set'); + assert.equal(sent?.progressTimeoutMs, undefined); + }); +}); + +describe('inference bounds: an MCPL server\'s inference request', () => { + async function mcpl(play: Play, options: { stream: boolean; openai?: boolean; inferenceBounds?: unknown }) { + const stand = await provider([play]); + const recording = new Recording(adapterFor(stand, options.openai)); + const fw = await framework(stand, { openai: options.openai, membrane: recording, inferenceBounds: 'inferenceBounds' in options ? options.inferenceBounds : SHORT }); + await (fw as unknown as { initializeMcpl(servers: unknown[], routing: unknown): Promise }).initializeMcpl([], { default: options.openai ? 'gpt-test' : 'claude-opus-5-5' }); + const router = (fw as unknown as { inferenceRouter: InferenceRouter }).inferenceRouter; + (router as unknown as { featureSetManager: unknown }).featureSetManager = { validateInbound: () => {} }; + let answered: { result?: unknown; error?: string } = {}; + const sentAt = Date.now(); + let answeredAfterMs = -1; + try { + const ended = await within(router.handleInferenceRequest('a-server', { + featureSet: 'memory.consolidation', + stream: options.stream, + messages: [{ role: 'user', content: 'Summarize.' }], + preferences: { maxTokens: 4096 }, + } as never, { + requestId: 1, + respond: (result: unknown) => { answered = { result }; }, + respondError: (_code: number, message: string) => { answered = { error: message }; answeredAfterMs = Date.now() - sentAt; }, + sendChunk: () => {}, + } as never), 8_000); + if (ended === NOT_ENDED) answered = { error: 'still waiting' }; + } finally { + await fw.stop(); + await stand.close(); + } + return { answered, answeredAfterMs, requests: stand.requests(), sent: recording.sent.at(-1)?.options }; + } + + it('a non-streamed request ends at its whole-call bound, and the server is told the bound', async () => { + const { answered, answeredAfterMs, requests } = await mcpl('no-headers', { stream: false }); + assert.equal(requests, 1); + assert.match(String(answered.error), /^No provider progress within 1300ms/); + assert.ok(answeredAfterMs >= 1_300 && answeredAfterMs < 1_400, `answered ${answeredAfterMs} ms after the request`); + }); + + it('a streamed request on an adapter that reports content ends at its phase bound, and the server is told the bound', async () => { + const { answered, answeredAfterMs, requests } = await mcpl('quiet-before-content', { stream: true }); + assert.equal(requests, 1); + assert.ok(answeredAfterMs >= 300 && answeredAfterMs < 400, `answered ${answeredAfterMs} ms after the request`); + assert.match(String(answered.error), /^The inference stopped before it finished \(timeout: No content within 0\.3s of the send/); + assert.match(String(answered.error), /\(the host's setting for this bound: `inferenceBounds\.progressBounds\.beforeContentMs`\)\)$/); + }); + + it('a streamed request on an adapter that doesn\'t ends at the flat bound', async () => { + const { answered, answeredAfterMs, requests } = await mcpl('openai-comments', { stream: true, openai: true }); + assert.equal(requests, 1); + assert.ok(answeredAfterMs >= 1_300 && answeredAfterMs < 1_400, `answered ${answeredAfterMs} ms after the request`); + assert.match(String(answered.error), /^The inference stopped before it finished \(timeout: No provider progress within 1300ms/); + assert.match(String(answered.error), /\(the host's setting for this bound: `inferenceBounds\.flatMinimumMs`\)\)$/); + }); + + for (const stream of [false, true]) { + it(`inferenceBounds: false sets no bound on an MCPL ${stream ? 'streamed' : 'non-streamed'} request`, async () => { + const { answered, requests, sent } = await mcpl(stream ? 'quiet-before-content' : 'no-headers', { stream, inferenceBounds: false }); + assert.equal(requests, 1); + assert.equal(answered.error, 'still waiting', 'not ended in 8 s, past every bound SHORT would set'); + assert.equal(sent?.progressTimeoutMs, undefined); + assert.equal(sent?.progressBounds, undefined); + }); + } +}); + +describe('inference bounds: the deadline class', () => { + async function markedByMembrane(): Promise { + const stand = await provider(['silent-after-start']); + try { + const stream = new Membrane(adapterFor(stand)).streamYielding( + { messages: [{ participant: 'user', content: [{ type: 'text', text: 'hi' }] }], config: { model: 'claude-opus-5-5', maxTokens: 100 } } as never, + { retry: false, idleTimeoutMs: 200, firstEventTimeoutMs: 200 } as never, + ); + for await (const event of stream) { + if (event.type === 'error') return (event as { error: Error }).error; + if (event.type === 'complete' || event.type === 'aborted') break; + } + throw new Error('membrane ended the stream without an error'); + } finally { + await stand.close(); + } + } + const delays = (policy: DefaultErrorPolicy, error: Error) => { + const out: number[] = []; + for (let attempt = 0; attempt < 20; attempt++) { + const action = policy.onInferenceError(error, 'resident', attempt); + if (!action.retry) return out; + out.push(action.delayMs); + } + throw new Error('the policy never stopped'); + }; + + it('a deadline expiry membrane marks itself gets one fresh call, 1 s later', async () => { + const marked = await markedByMembrane(); + assert.ok(marked instanceof MembraneError && marked.type === 'timeout' && marked.retryable === true, String(marked)); + assert.deepEqual(delays(new DefaultErrorPolicy(), marked), [1_000]); + }); + + it('a host\'s own deadline schedule is used for one', async () => { + const marked = await markedByMembrane(); + const settings = { ...MEMBRANE_RETRY_SETTINGS, deadline: { maxAttempts: 3, firstDelayMs: 500, multiplier: 2, maxDelayMs: 5_000, statedWaitCeilingMs: null } } as RetrySettings; + assert.deepEqual(delays(new DefaultErrorPolicy(settings), marked), [500, 1_000]); + }); + + it('settings with no deadline schedule give one their base schedule', async () => { + const marked = await markedByMembrane(); + const { deadline: _none, ...settings } = MEMBRANE_RETRY_SETTINGS as RetrySettings; + assert.deepEqual(delays(new DefaultErrorPolicy(settings as RetrySettings), marked), [1_000, 2_000]); + assert.deepEqual(delays(new DefaultErrorPolicy(settings as RetrySettings), new MembraneError({ type: 'server', retryable: true, message: 'x', rawError: undefined })), [1_000, 2_000]); + }); +}); diff --git a/test/inference-failure-bounds.test.ts b/test/inference-failure-bounds.test.ts index 6fe63cdbe..2ad819e2f 100644 --- a/test/inference-failure-bounds.test.ts +++ b/test/inference-failure-bounds.test.ts @@ -475,7 +475,7 @@ test('a provider hold keeps a bounded reason, and a failed resume logs the bound const providerState = (fw: any) => Object.assign(fw, { ephemeralRuns: new Map(), conversationAgentHomes: new Map(), providerGates: new Map(), providerAccelerationCooldowns: new Map(), providerAccelerationRecoveries: new Map(), - providerAdmissionClosed: false, activeTurnTokens: new Map(), quiesced: false, + providerAdmissionClosed: false, activeTurnTokens: new Map(), quiesced: false, deadlineCuts: new Map(), }); const held = makeHarness(); providerState(held.fw); diff --git a/test/mcpl-inference-router.test.ts b/test/mcpl-inference-router.test.ts index 3a6b08ca0..5578fb36b 100644 --- a/test/mcpl-inference-router.test.ts +++ b/test/mcpl-inference-router.test.ts @@ -306,8 +306,8 @@ for (const mode of ['native', 'xml'] as const) { ); assert.strictEqual(answer.result, undefined); - assert.deepStrictEqual(answer.error, { code: -32603, message: 'The inference stopped before it finished (timeout)' }); - assert.deepStrictEqual(answer.sent, ['chunk:A summ', 'error:The inference stopped before it finished (timeout)']); + assert.deepStrictEqual(answer.error, { code: -32603, message: 'The inference stopped before it finished (timeout: no event for 120000ms)' }); + assert.deepStrictEqual(answer.sent, ['chunk:A summ', 'error:The inference stopped before it finished (timeout: no event for 120000ms)']); }); it('answers a refusal with the error, not end_turn', async () => { diff --git a/test/prose-drafts.test.ts b/test/prose-drafts.test.ts index 016f8aa4f..53f8cd5cd 100644 --- a/test/prose-drafts.test.ts +++ b/test/prose-drafts.test.ts @@ -326,7 +326,7 @@ function breakAfterToolRound( } if (how === 'abort' && !cancelled) { await opts.beforeFailure?.(); - yield { type: 'aborted', reason: 'timeout', partialContent: [{ type: 'text', text: PARTIAL }] } as unknown as StreamEvent; + yield { type: 'aborted', reason: 'error', partialContent: [{ type: 'text', text: PARTIAL }] } as unknown as StreamEvent; return; } while (!cancelled) await new Promise((resolve) => { wake = resolve; }); diff --git a/test/tool-result-guard.test.ts b/test/tool-result-guard.test.ts index c4f00270b..7a19451c1 100644 --- a/test/tool-result-guard.test.ts +++ b/test/tool-result-guard.test.ts @@ -2447,7 +2447,7 @@ const SETTLED_AS_IT_LEFT = ['staged', 'linked', 'withheld', 'annotated', 'stub_u // gone with it. Pinned for a failure whose stream ends only after the run's // caller has deregistered the agent, in both shapes, and for a run the // completion watchdog ends while its stream says nothing. -for (const how of ['an uncaused error', 'an abort whose reason is timeout', 'the completion watchdog'] as const) { +for (const how of ['an uncaused error', 'an abort whose reason is error', 'the completion watchdog'] as const) { test(`an ephemeral run that ends by ${how} with its batch pending settles the batch as it leaves`, async () => { const h = await harness([[calls('one')]]); if (how !== 'the completion watchdog') waitAfterTerminal(h.membrane, () => h.framework.getAgent('ephemeral') === null); @@ -2460,8 +2460,8 @@ for (const how of ['an uncaused error', 'an abort whose reason is timeout', 'the stream.processResponse = () => {}; if (how === 'an uncaused error') { stream.events.push({ type: 'error', error: new MembraneError({ type: 'auth', retryable: false, message: 'key revoked', rawError: { status: 401 } }) }); - } else if (how === 'an abort whose reason is timeout') { - stream.events.push({ type: 'aborted', reason: 'timeout' }); + } else if (how === 'an abort whose reason is error') { + stream.events.push({ type: 'aborted', reason: 'error' }); } }; let outcome = 'pending'; @@ -2482,7 +2482,7 @@ for (const how of ['an uncaused error', 'an abort whose reason is timeout', 'the assert.equal(created.agent.toolResultGuard.hasPending, false, 'the run left no batch pending'); } finally { await h.framework.stop(); } assert.match(outcome, how === 'the completion watchdog' ? /^rejected: Ephemeral agent "ephemeral" stalled/ - : how === 'an uncaused error' ? /^rejected: key revoked/ : /^rejected: Stream aborted: timeout/, 'the run ends as expected'); + : how === 'an uncaused error' ? /^rejected: key revoked/ : /^rejected: Stream aborted: error/, 'the run ends as expected'); assert.equal(spills, 1, "the batch's original is offered to the host's spill once"); const audit = await ephemeralBatchAudit(h.base); assert.deepEqual(audit.types, SETTLED_AS_IT_LEFT, 'the batch is settled once, and its stub says where its original is'); From 6ee150eb90209aa505a96748a6877541adcfed30 Mon Sep 17 00:00:00 2001 From: Nigel-2405 Date: Wed, 28 Oct 2026 23:13:16 -0400 Subject: [PATCH 2/2] fix(inference): clear an ended agent's deadline-cut count in its release releaseAgentRuntime is the release an agent gets when it's stopped, when its end cascades from a parent's, and when a fork is disposed. It now clears the agent's count of consecutive deadline cuts, so an agent that ends that way with a cut still counted leaves no count behind. An ephemeral run's end already cleared its own. disposeConversationAgent's own clear moves into the branch where nothing is live under the name, beside the other per-name state that branch forgets. test/inference-bounds.test.ts pins the release's clear: an agent with a count, stopped through the registry, keeps none. --- src/framework.ts | 3 ++- test/inference-bounds.test.ts | 24 ++++++++++++++++++++++++ 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/src/framework.ts b/src/framework.ts index 769c5e1d3..57cca74a7 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -11807,6 +11807,7 @@ export class AgentFramework { this.eventGate?.agentGone(name); this.agents.delete(name); this.agentConfigs.delete(name); + this.deadlineCuts.delete(name); this.thinkingLossIncidents.forget(name); this.reasoningLoss.forget(name); // main's manager release: the context manager an ephemeral run or a fork @@ -17720,7 +17721,6 @@ export class AgentFramework { private disposeConversationAgent(agentName: string, reason: AgentEndReason = 'idle-ttl'): void { const home = this.homeOf(agentName); const agent = this.agents.get(agentName); - this.deadlineCuts.delete(agentName); // The one release every ended agent gets (releaseAgentRuntime), then the // registry's end, whose outcome settles what the fork spawned. if (agent) { @@ -17728,6 +17728,7 @@ export class AgentFramework { } else { // Nothing live under the name: only what the process keeps by name. this.closingConversationAgents.delete(agentName); + this.deadlineCuts.delete(agentName); this.thinkingLossIncidents.forget(agentName); this.reasoningLoss.forget(agentName); // A continuation of its turn still on its way goes with it. diff --git a/test/inference-bounds.test.ts b/test/inference-bounds.test.ts index 7d8c66f4d..2bae16baa 100644 --- a/test/inference-bounds.test.ts +++ b/test/inference-bounds.test.ts @@ -765,4 +765,28 @@ describe('inference bounds: the deadline class', () => { assert.deepEqual(delays(new DefaultErrorPolicy(settings as RetrySettings), marked), [1_000, 2_000]); assert.deepEqual(delays(new DefaultErrorPolicy(settings as RetrySettings), new MembraneError({ type: 'server', retryable: true, message: 'x', rawError: undefined })), [1_000, 2_000]); }); + + it('an agent ended through the registry takes its count of deadline cuts with it', async () => { + const stand = await provider([]); + const agent = { model: 'claude-opus-5-5', systemPrompt: 'You are a resident.', maxTokens: 4096 }; + const fw = await AgentFramework.create({ + storePath: join(mkdtempSync(join(tmpdir(), 'af-bounds-')), 'store.chronicle'), + membrane: new Membrane(adapterFor(stand)), + agents: [{ name: 'resident', ...agent } as never, { name: 'second', ...agent } as never], + modules: [new Asker()], + syncIntervalMs: 0, + maintenanceIntervalMs: 0, + inferenceBounds: SHORT, + }); + try { + const counts = (fw as unknown as { deadlineCuts: Map }).deadlineCuts; + // Set as a cut that gets its fresh call sets it, so the release's clear is pinned. + counts.set('second', 1); + fw.stopAgent('second', { by: 'test' }); + assert.equal(counts.has('second'), false); + } finally { + await fw.stop(); + await stand.close(); + } + }); });