diff --git a/ably-common b/ably-common index a78f596ba..5c2757846 160000 --- a/ably-common +++ b/ably-common @@ -1 +1 @@ -Subproject commit a78f596baa86f170a07c7b53910b213c54a0898d +Subproject commit 5c2757846ee8d7c7d8ca5fb04855a1512ddc4a3a diff --git a/docs/internals/client-session.md b/docs/internals/client-session.md index b4ad072cd..43a3a6ca9 100644 --- a/docs/internals/client-session.md +++ b/docs/internals/client-session.md @@ -66,7 +66,7 @@ All non-lifecycle messages pass through the codec decoder inside `applyWireMessa 1. `decoder.decode(rawMessage)` yields `{ inputs, outputs }` split by wire direction. 2. `tree.applyMessage({ inputs, outputs }, headers, serial)` — the Tree folds events into the owning Run's (or input node's) projection and emits an `output` event carrying the message's outputs. This is the single fan-out point for run outputs; consumers (the View, and the Vercel chat transport's per-run stream) subscribe to it. A wire-only carrier that decodes to no events and carries no `run-id` is skipped (the eventual reply run is created later by its run-start). -After the apply returns, `_handleMessage` calls `tree.emitAblyMessage(rawMsg)` so subscribers to `'ably-message'` can observe the raw wire — emitted _after_ the apply so View subscribers can already find the owning Run. Any error thrown while processing a message is caught and surfaced as a session `error` event (`SessionSubscriptionError`) rather than escaping the listener. +After the apply returns, `_handleMessage` calls `tree.emitAblyMessage(rawMsg)` so subscribers to `'ably-message'` can observe the raw wire — emitted _after_ the apply so View subscribers can already find the owning Run. Any error thrown while processing a message is caught and surfaced as a session `error` event (`SessionMessageProcessingFailed`) rather than escaping the listener. There is no separate observer-state map. The Tree's per-Run projection is the single source of truth for every Run (own or observer); the View extracts messages on demand via `codec.getMessages(run.projection)`. diff --git a/docs/reference/error-codes.md b/docs/reference/error-codes.md index 912926a06..08f1c261e 100644 --- a/docs/reference/error-codes.md +++ b/docs/reference/error-codes.md @@ -4,27 +4,30 @@ AI Transport uses `Ably.ErrorInfo` as its error type. Each error has a numeric ` ## Error codes -| Code | Name | Status | Description | Recovery | -| ------ | ---------------------------- | ------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| 40000 | `BadRequest` | 400 | The request was invalid | Check the request parameters | -| 40003 | `InvalidArgument` | 400 | An argument passed to a public method was invalid | Fix the argument value | -| 40033 | `OperationCancelled` | 400 | The operation was cancelled — the run was cancelled, the caller's abort signal fired, or the session began closing while the operation was in flight | Expected during cancellation; retry only if the cancel was not intended | -| 40160 | `InsufficientCapability` | 401 | The Ably channel rejected a publish for a capability reason (missing publish capability) | Grant the client publish capability on the channel | -| 50000 | `InternalError` | 500 | An internal invariant failed — the SDK or the Ably service behaved in a way the SDK cannot recover from or explain (e.g. a publish succeeded but returned no serial) | Not caused by caller input; check connectivity and report if it persists | -| 93002 | Mutable messages not enabled | 400 | The channel's namespace does not have the `mutableMessages` rule enabled, so AI Transport cannot append stream tokens. The first append fails with `Can only update/delete/append messages on channels with mutableMessages enabled`. The single most common AI Transport setup failure. | Enable the **Message annotations, updates, deletes, and appends** rule on the namespace. See [channel rules](https://ably.com/docs/ai-transport/getting-started/channel-rules). | -| 104000 | `EncoderRecoveryFailed` | 500 | Encoder recovery failed after flush — one or more `updateMessage` calls could not recover a failed append pipeline | Non-fatal; the message may be incomplete on the channel. Check network connectivity | -| 104001 | `SessionSubscriptionError` | 500 | The session's channel subscription failed — the subscribe/attach step failed, or a session-level subscription callback threw unexpectedly | Non-fatal when a callback threw; the session is still operational. Check error handler logic | -| 104002 | `CancelListenerError` | 500 | Cancel listener or `onCancel` hook threw while processing a cancel message | Non-fatal; check the `onCancel` hook implementation | -| 104003 | `RunLifecycleError` | 500 | A publish within a run failed (lifecycle event, message, or event) | Non-fatal; the run may not be fully visible to other clients. Check channel permissions | -| 104004 | `SessionClosed` | 400 | An operation was attempted on a session, view, or encoder that has already been closed | Create a new instance | -| 104005 | `SessionSendFailed` | 500 | A send failed: the core's channel publish failed, or the Vercel chat transport's agent-invocation POST failed (network error or non-2xx) | Check channel publish capability, or agent availability and endpoint URL | -| 104006 | `ChannelContinuityLost` | 500 | The Ably channel lost message continuity after its initial attach (FAILED, SUSPENDED, DETACHED, or re-attached with `resumed: false`) | Surfaced via `session.on('error')`. Check network connectivity and channel state | -| 104007 | `ChannelNotReady` | 400 | An operation was attempted but the channel is not in a usable state (not ATTACHED or ATTACHING) | Check the channel state and why it entered that state | -| 104008 | `StreamError` | 500 | An error occurred while piping a response stream to the channel — the source event stream threw (e.g. LLM provider rate limit, model error, network failure) or an underlying publish failed mid-stream. Also the fallback code when a run-end reports an error without a code on the wire | Surfaced via the agent's `onError`; the run ends with reason `error` | -| 104010 | `InputEventNotFound` | 504 | A fresh process adopting an open run via `adoptRun().load()` waited for that run's `ai-run-start` to be observed on the channel (live + bounded history scan) but `load()`'s `timeoutMs` lapsed (or history exhausted) without seeing it | Retryable — a workflow-ordering error where the open activity's run-start has not yet propagated; retry the adopting activity. Any history-fetch failure is the `cause` | -| 104011 | `HistoryFetchFailed` | 500 | A `channel.history()` page fetch failed after retries while paginating history (a `loadOlder` reveal, or the agent's `run.view` drain) | Surfaced on the client as the view's `loadError` (and `useView`'s `loadError`); retry the load and check network connectivity | +| Code | Name | Status | Description | Recovery | +| ------ | -------------------------------- | ------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| 40000 | `BadRequest` | 400 | The request was invalid | Check the request parameters | +| 40003 | `InvalidArgument` | 400 | An argument passed to a public method was invalid | Fix the argument value | +| 40033 | `OperationCancelled` | 400 | The operation was cancelled — the run was cancelled, the caller's abort signal fired, or the session began closing while the operation was in flight | Expected during cancellation; retry only if the cancel was not intended | +| 40160 | `InsufficientCapability` | 401 | The Ably channel rejected a publish for a capability reason (missing publish capability) | Grant the client publish capability on the channel | +| 50000 | `InternalError` | 500 | An internal invariant failed — the SDK or the Ably service behaved in a way the SDK cannot recover from or explain (e.g. a publish succeeded but returned no serial) | Not caused by caller input; check connectivity and report if it persists | +| 93002 | Mutable messages not enabled | 400 | The channel's namespace does not have the `mutableMessages` rule enabled, so AI Transport cannot append stream tokens. The first append fails with `Can only update/delete/append messages on channels with mutableMessages enabled`. The single most common AI Transport setup failure. | Enable the **Message annotations, updates, deletes, and appends** rule on the namespace. See [channel rules](https://ably.com/docs/ai-transport/getting-started/channel-rules). | +| 104000 | `EncoderRecoveryFailed` | 500 | Encoder recovery failed after flush — one or more `updateMessage` calls could not recover a failed append pipeline | Non-fatal; the message may be incomplete on the channel. Check network connectivity | +| 104001 | `SessionSubscriptionError` | 500 | The session could not subscribe to and attach its channel during `connect()` | Fatal for that attempt; nothing sends or receives. Call `connect()` again for a transient cause; fix capabilities or auth otherwise | +| 104002 | `RunCancelHandlerFailed` | 500 | The run's `onCancel` hook threw while the SDK was processing a cancel message | The run is **not** cancelled — the SDK never reached the abort. Fix the `onCancel` hook so it returns normally | +| 104003 | `RunLifecycleEventPublishFailed` | 500 | A lifecycle event publish failed, at either tier — a run's `ai-run-start` / `ai-run-suspend` / `ai-run-end`, or a step's `ai-step-start` / `ai-step-end` | The event is not on the channel, so clients do not observe that phase. Check the `cause` and channel publish capability | +| 104004 | `SessionClosed` | 400 | An operation was attempted on a session, view, or encoder that has already been closed | Create a new instance | +| 104005 | `SessionSendFailed` | 500 | A send failed: the core's channel publish failed, or the Vercel chat transport's agent-invocation POST failed (network error or non-2xx) | Check channel publish capability, or agent availability and endpoint URL | +| 104006 | `ChannelContinuityLost` | 500 | The Ably channel lost message continuity after its initial attach (FAILED, SUSPENDED, DETACHED, or re-attached with `resumed: false`) | Surfaced via `session.on('error')`. Check network connectivity and channel state | +| 104007 | `ChannelNotReady` | 400 | An operation was attempted but the channel is not in a usable state (not ATTACHED or ATTACHING) | Check the channel state and why it entered that state | +| 104008 | `StreamError` | 500 | An error occurred while piping a response stream to the channel — the source event stream threw (e.g. LLM provider rate limit, model error, network failure) or an underlying publish failed mid-stream. Also the fallback code when a run-end reports an error without a code on the wire | Surfaced via the agent's `onError`; the run ends with reason `error` | +| 104009 | `SessionMessageProcessingFailed` | 500 | Processing an inbound channel message threw — the codec folding it into session state, or a session-level subscription callback | Non-fatal; the subscription survives and the session keeps working. Inspect the `cause` and fix the handler | +| 104010 | `InputEventNotFound` | 504 | A fresh process adopting an open run via `adoptRun().load()` waited for that run's `ai-run-start` to be observed on the channel (live + bounded history scan) but `load()`'s `timeoutMs` lapsed (or history exhausted) without seeing it | Retryable — a workflow-ordering error where the open activity's run-start has not yet propagated; retry the adopting activity. Any history-fetch failure is the `cause` | +| 104011 | `HistoryFetchFailed` | 500 | A `channel.history()` page fetch failed after retries while paginating history (a `loadOlder` reveal, or the agent's `run.view` drain) | Surfaced on the client as the view's `loadError` (and `useView`'s `loadError`); retry the load and check network connectivity | +| 104012 | `RunSteerHandlerFailed` | 500 | The run's `onSteer` hook threw while the SDK was notifying it that a steering message folded into the run | The run is unaffected — the steering message already folded in, so only the notification failed. Fix the `onSteer` hook | +| 104013 | `RunCancelRoutingFailed` | 500 | Routing an inbound cancel message to its target run failed — not a fault in a developer-supplied hook | The cancel was neither honoured nor rejected and the run keeps running. Cancel again; inspect the `cause` | -Codes 40000, 40003, 40033, 40160, 50000, and 93002 are standard Ably error codes. Codes 104000–104999 are reserved for the AI Transport SDK (104009 is currently unused). +Codes 40000, 40003, 40033, 40160, 50000, and 93002 are standard Ably error codes. Codes 104000–104999 are reserved for the AI Transport SDK. ## Checking error codes diff --git a/scripts/validate-error-codes.ts b/scripts/validate-error-codes.ts index 1d0fc4ea5..692db4ce8 100644 --- a/scripts/validate-error-codes.ts +++ b/scripts/validate-error-codes.ts @@ -13,8 +13,19 @@ const __filename = fileURLToPath(import.meta.url); const __dirname = path.dirname(__filename); const ERRORS_JSON_PATH = path.join(__dirname, '../ably-common/protocol/errors.json'); +/** One registry entry, as generated into errors.json from `errors/codes/*.md`. */ +interface ErrorEntry { + /** The registry's canonical snake_case name for the code. */ + identifier: string; + /** Short human-readable title. */ + title: string; + /** One-paragraph description of the failure. */ + summary: string; +} + interface ErrorsJson { - [code: string]: string; + /** Registry entries keyed by numeric code. */ + codes: Record; } function main(): void { @@ -29,6 +40,14 @@ function main(): void { process.exit(1); } + const registry = errorsJson.codes; + // A pin predating the `codes` envelope yields no registry at all — say so, + // rather than reporting every code as missing. + if (typeof registry !== 'object') { + console.error(`No "codes" object in ${ERRORS_JSON_PATH}; the ably-common submodule may be stale or uninitialised`); + process.exit(1); + } + // Get all error codes from the enum const errorCodes = Object.values(ErrorCode).filter((value) => typeof value === 'number') as number[]; @@ -36,12 +55,12 @@ function main(): void { let hasErrors = false; const missingCodes: number[] = []; - const foundCodes: Array<{ code: number; message: string }> = []; + const foundCodes: Array<{ code: number; identifier: string }> = []; for (const code of errorCodes) { - const codeStr = code.toString(); - if (errorsJson[codeStr]) { - foundCodes.push({ code, message: errorsJson[codeStr] }); + const entry = registry[code.toString()]; + if (entry) { + foundCodes.push({ code, identifier: entry.identifier }); } else { missingCodes.push(code); hasErrors = true; @@ -51,8 +70,8 @@ function main(): void { // Print results if (foundCodes.length > 0) { console.log('Found codes:'); - for (const { code, message } of foundCodes) { - console.log(` ${code}: ${message}`); + for (const { code, identifier } of foundCodes) { + console.log(` ${code}: ${identifier}`); } console.log(); } diff --git a/src/core/transport/agent-session.ts b/src/core/transport/agent-session.ts index 60696f6b4..55ad3b231 100644 --- a/src/core/transport/agent-session.ts +++ b/src/core/transport/agent-session.ts @@ -43,6 +43,7 @@ import { locateInputEvent } from './input-event-locator.js'; import { evictOldestIfFull } from './internal/bounded-map.js'; import type { Invocation } from './invocation.js'; import { createLeafBranchSource } from './leaf-branch-source.js'; +import { publishLifecycleEvent } from './lifecycle-publish.js'; import { createMaterialisation } from './materialisation.js'; import type { RunManager } from './run-manager.js'; import { createRunManager } from './run-manager.js'; @@ -403,7 +404,7 @@ class DefaultAgentSession< 'error', new Ably.ErrorInfo( `unable to end run ${reg.runId} on session end; ${errorMessage(error)}`, - ErrorCode.RunLifecycleError, + ErrorCode.RunLifecycleEventPublishFailed, 500, errorCause(error), ), @@ -565,7 +566,7 @@ class DefaultAgentSession< } catch (error) { const errInfo = new Ably.ErrorInfo( `unable to process cancel for run ${runId}; onCancel handler threw: ${errorMessage(error)}`, - ErrorCode.CancelListenerError, + ErrorCode.RunCancelHandlerFailed, 500, errorCause(error), ); @@ -676,7 +677,7 @@ class DefaultAgentSession< this._handleCancelMessage(msg).catch((error: unknown) => { const errInfo = new Ably.ErrorInfo( `unable to route cancel message; ${errorMessage(error)}`, - ErrorCode.CancelListenerError, + ErrorCode.RunCancelRoutingFailed, 500, errorCause(error), ); @@ -882,7 +883,7 @@ class DefaultAgentSession< } catch (error) { const errInfo = new Ably.ErrorInfo( `unable to notify steer for run ${runId}; onSteer handler threw: ${errorMessage(error)}`, - ErrorCode.CancelListenerError, + ErrorCode.RunSteerHandlerFailed, 500, errorCause(error), ); @@ -1146,10 +1147,9 @@ class DefaultAgentSession< } /** - * Run a run-lifecycle publish (run-start / run-suspend / run-end) and wrap - * any failure as a `RunLifecycleError`, logging at error and rethrowing. - * Shared by start(), suspend(), and end() so the three publishes can't - * drift on the error code, message shape, or cause preservation. + * Run a run-lifecycle publish (run-start / run-suspend / run-end) through + * the shared lifecycle bracket, which the step-lifecycle publishes in + * {@link createRunStepWriter} also use. * @param phase - The lifecycle wire phase, used in the error message. * @param method - The Run method name, used in the log prefix. * @param publish - The RunManager publish to run. @@ -1159,18 +1159,7 @@ class DefaultAgentSession< method: 'start' | 'suspend' | 'end', publish: () => Promise, ): Promise => { - try { - await publish(); - } catch (error) { - const errInfo = new Ably.ErrorInfo( - `unable to publish ${phase} for run ${runId}; ${errorMessage(error)}`, - ErrorCode.RunLifecycleError, - 500, - errorCause(error), - ); - logger?.error(`Run.${method}(); failed to publish ${phase}`, { runId }); - throw errInfo; - } + await publishLifecycleEvent({ phase, method, runId, logger }, publish); }; // The shared run read-model (runId, status, error, whole-turn messages). diff --git a/src/core/transport/lifecycle-publish.ts b/src/core/transport/lifecycle-publish.ts new file mode 100644 index 000000000..8568b136c --- /dev/null +++ b/src/core/transport/lifecycle-publish.ts @@ -0,0 +1,58 @@ +import * as Ably from 'ably'; + +import { ErrorCode } from '../../errors.js'; +import type { Logger } from '../../logger.js'; +import { errorCause, errorMessage } from '../../utils.js'; + +/** + * A lifecycle wire event whose publish is bracketed by + * {@link publishLifecycleEvent}. Spans both tiers — the run's own lifecycle and + * the step lifecycle nested within it — because a failed step publish is the + * same class of failure as a failed run publish and surfaces identically. + */ +export type LifecyclePhase = 'run-start' | 'run-suspend' | 'run-end' | 'step-start' | 'step-end'; + +/** + * Options identifying the lifecycle publish being bracketed. + */ +export interface PublishLifecycleOptions { + /** The lifecycle wire phase, named in the error message. */ + phase: LifecyclePhase; + /** The method name to prefix the error log with (e.g. `start`, `openStep`). */ + method: string; + /** The run the event belongs to, named in the error message. */ + runId: string; + /** Logger for the failure; the phase and `runId` are logged with it. */ + logger?: Logger; + /** Extra structured context for the failure log (e.g. the step id). */ + logContext?: Record; +} + +/** + * Run a lifecycle publish and wrap any failure as a + * {@link ErrorCode.RunLifecycleEventPublishFailed}, logging at error and + * rethrowing. Every run- and step-lifecycle publish goes through here so they + * cannot drift on the error code, message shape, or cause preservation. + * @param options - Identifies the publish (see {@link PublishLifecycleOptions}). + * @param publish - The RunManager publish to run. + * @returns Whatever `publish` resolves with (the ACK serial, for the publishes that report one). + * @throws {@link Ably.ErrorInfo} with {@link ErrorCode.RunLifecycleEventPublishFailed} if `publish` rejects. + */ +export const publishLifecycleEvent = async ( + options: PublishLifecycleOptions, + publish: () => Promise, +): Promise => { + const { phase, method, runId, logger, logContext } = options; + try { + return await publish(); + } catch (error) { + const errInfo = new Ably.ErrorInfo( + `unable to publish ${phase} for run ${runId}; ${errorMessage(error)}`, + ErrorCode.RunLifecycleEventPublishFailed, + 500, + errorCause(error), + ); + logger?.error(`Run.${method}(); failed to publish ${phase}`, { runId, ...logContext }); + throw errInfo; + } +}; diff --git a/src/core/transport/run-step-writer.ts b/src/core/transport/run-step-writer.ts index 42fc8cf3f..126aba31b 100644 --- a/src/core/transport/run-step-writer.ts +++ b/src/core/transport/run-step-writer.ts @@ -24,6 +24,7 @@ import type { Logger } from '../../logger.js'; import { errorCause } from '../../utils.js'; import type { Codec, CodecInputEvent, CodecOutputEvent } from '../codec/types.js'; import { buildTransportHeaders } from './headers.js'; +import { publishLifecycleEvent } from './lifecycle-publish.js'; import { pipeStream } from './pipe-stream.js'; import type { RunManager, StepClientScopes } from './run-manager.js'; import type { DefaultTree } from './tree.js'; @@ -307,7 +308,10 @@ export const createRunStepWriter = < // inference runs. markOutputProduced fires per-pass in doPipe/doSend instead. // The steers to stamp are likewise drained per-pipe, not here. const scopes = stepScopes(stepClientId); - const stepStartSerial = await runManager.startStep(runId, stepId, scopes); + const stepStartSerial = await publishLifecycleEvent( + { phase: 'step-start', method: 'openStep', runId, logger, logContext: { stepId } }, + async () => runManager.startStep(runId, stepId, scopes), + ); getTree().applyStepLifecycle({ type: 'step-start', runId, @@ -352,7 +356,10 @@ export const createRunStepWriter = < return; } const scopes = stepScopes(stepClientId); - await runManager.endStep(runId, stepId, stepStartSerial, reason, scopes); + await publishLifecycleEvent( + { phase: 'step-end', method: 'closeStep', runId, logger, logContext: { stepId } }, + async () => runManager.endStep(runId, stepId, stepStartSerial, reason, scopes), + ); getTree().applyStepLifecycle({ type: 'step-end', runId, diff --git a/src/core/transport/session-support.ts b/src/core/transport/session-support.ts index 3684a7a46..0a9f126fb 100644 --- a/src/core/transport/session-support.ts +++ b/src/core/transport/session-support.ts @@ -104,15 +104,17 @@ export const subscribeAndAttach = async ( /** * Wrap a failure thrown while processing an inbound channel message as a - * `SessionSubscriptionError`, preserving the original as `cause`. Single source - * of truth for the message-processing error shape both sessions surface. + * `SessionMessageProcessingFailed`, preserving the original as `cause`. Single source + * of truth for the message-processing error shape both sessions surface. Kept + * distinct from the connect-time `SessionSubscriptionError`: the subscription + * survives this, so the session stays usable and the fix is in the handler. * @param error - The thrown value. * @returns The wrapped error. */ export const wrapMessageProcessingError = (error: unknown): Ably.ErrorInfo => new Ably.ErrorInfo( `unable to process channel message; ${errorMessage(error)}`, - ErrorCode.SessionSubscriptionError, + ErrorCode.SessionMessageProcessingFailed, 500, errorCause(error), ); diff --git a/src/core/transport/types/agent.ts b/src/core/transport/types/agent.ts index c77b93c2b..45292ddfb 100644 --- a/src/core/transport/types/agent.ts +++ b/src/core/transport/types/agent.ts @@ -254,7 +254,11 @@ export interface RunHooks { * - Stream failures in `pipe` — the underlying error is also returned on * `StreamResult.error`, but this callback delivers it wrapped as an * `Ably.ErrorInfo` (code `StreamError`) for standardized observability. - * - Failures in the `onCancel` handler. + * - A throw from the `onCancel` handler (code `RunCancelHandlerFailed`). The run + * is NOT cancelled: the SDK never reaches the abort. + * - A throw from the `onSteer` handler (code `RunSteerHandlerFailed`). The run is + * unaffected — the steering message has already folded in, so only the + * notification failed. * * Publish failures in `start` and `end` * are not delivered here — those methods reject their returned promise @@ -265,9 +269,9 @@ export interface RunHooks { * * Channel-wide events (e.g. continuity loss) are delivered via the * session-level {@link AgentSession.on}('error'), not here. A failure in the - * `onCancel` handler with no `onError` set falls back to that session emitter - * so it is never silently dropped; a `pipe` stream failure with no `onError` - * is always still available on {@link StreamResult.error}. + * `onCancel` or `onSteer` handler with no `onError` set falls back to that + * session emitter so it is never silently dropped; a `pipe` stream failure + * with no `onError` is always still available on {@link StreamResult.error}. */ onError?: (error: Ably.ErrorInfo) => void; @@ -316,6 +320,7 @@ export interface RunStep { * a second call is a no-op. Rejects if another step is already active on the * run (only one step may be open at a time), or if the run has ended. * @throws InvalidArgument if another step is active or the run has ended. + * @throws {Ably.ErrorInfo} `RunLifecycleEventPublishFailed` if the `ai-step-start` publish fails. */ start(): Promise; /** @@ -359,6 +364,7 @@ export interface RunStep { * auto-closes a still-open step, so a forgotten `end()` cannot strand * observers — but an explicit `end()` is clearer and lets you set the reason. * @param params - Optional {@link StepEndParams}; the reason is derived if omitted. + * @throws {Ably.ErrorInfo} `RunLifecycleEventPublishFailed` if the `ai-step-end` publish fails. */ end(params?: StepEndParams): Promise; } @@ -587,7 +593,7 @@ export interface OpenableRun; } diff --git a/src/errors.ts b/src/errors.ts index 94115e113..b0778147b 100644 --- a/src/errors.ts +++ b/src/errors.ts @@ -43,22 +43,27 @@ export enum ErrorCode { EncoderRecoveryFailed = 104000, /** - * The session's channel subscription failed — the subscribe/attach step - * failed, or a session-level subscription callback threw unexpectedly. + * The session could not subscribe to and attach its channel during + * `connect()`. Nothing sends or receives until the attach succeeds; whether a + * retry helps depends on the `cause` (a transient disconnect clears, a + * capability or auth rejection does not). */ SessionSubscriptionError = 104001, /** - * A run-scoped developer callback threw while the SDK invoked it — the - * `onCancel` hook processing a cancel message, or the `onSteer` hook - * notifying that a steering message folded into the run. + * The run's `onCancel` hook threw while the SDK was processing a cancel + * message. The SDK never reaches the abort, so the run is **not** cancelled. */ - CancelListenerError = 104002, + RunCancelHandlerFailed = 104002, /** - * A publish within a run failed (lifecycle event, message, or event). + * A lifecycle event publish failed, at either tier: a run's `ai-run-start` / + * `ai-run-suspend` / `ai-run-end`, or a step's `ai-step-start` / + * `ai-step-end`. The event is not on the channel, so clients do not observe + * the run or step entering that phase. The underlying publish failure is the + * `cause`. */ - RunLifecycleError = 104003, + RunLifecycleEventPublishFailed = 104003, /** * An operation was attempted on a session, view, or encoder that has already @@ -95,6 +100,14 @@ export enum ErrorCode { */ StreamError = 104008, + /** + * Processing an inbound channel message threw — the codec folding it into + * session state, or a session-level subscription callback. The subscription + * survives and the session keeps sending and receiving; only that one + * message's processing failed. The thrown value is the `cause`. + */ + SessionMessageProcessingFailed = 104009, + /** * A fresh process adopting an open run via {@link AdoptedRun.load} waited for * that run's `ai-run-start` to be observed on the channel — across the live @@ -113,6 +126,20 @@ export enum ErrorCode { * `cause` where available. */ HistoryFetchFailed = 104011, + + /** + * The run's `onSteer` hook threw while the SDK was notifying it that a + * steering message folded into the run. The steering message has already + * folded in by then, so the run is unaffected — only the notification failed. + */ + RunSteerHandlerFailed = 104012, + + /** + * Routing an inbound cancel message to its target run failed. Not a fault in + * a developer-supplied hook: the dispatch itself could not complete, so the + * cancel was neither honoured nor rejected and the run keeps running. + */ + RunCancelRoutingFailed = 104013, } /** diff --git a/test/core/transport/agent-session.test.ts b/test/core/transport/agent-session.test.ts index 5300163e1..e4009809e 100644 --- a/test/core/transport/agent-session.test.ts +++ b/test/core/transport/agent-session.test.ts @@ -3267,7 +3267,7 @@ describe('AgentSession', () => { failSession.on('error', onError); await failSession.connect(); const run = createRunFromOpts(failSession, { runId: 'run-1', onError }); - await expect(run.start()).rejects.toBeErrorInfoWithCode(ErrorCode.RunLifecycleError); + await expect(run.start()).rejects.toBeErrorInfoWithCode(ErrorCode.RunLifecycleEventPublishFailed); expect(onError).not.toHaveBeenCalled(); await failSession.detach(); }); @@ -3276,7 +3276,9 @@ describe('AgentSession', () => { const run = createRunFromOpts(session, { runId: 'run-1' }); await run.start(); vi.mocked(channel.publish).mockRejectedValueOnce(new Error('publish failed')); - await expect(run.end({ reason: 'complete' })).rejects.toBeErrorInfoWithCode(ErrorCode.RunLifecycleError); + await expect(run.end({ reason: 'complete' })).rejects.toBeErrorInfoWithCode( + ErrorCode.RunLifecycleEventPublishFailed, + ); }); it('pipe() calls onError when the stream errors', async () => { @@ -3337,7 +3339,147 @@ describe('AgentSession', () => { simulateCancel(channel, { [HEADER_RUN_ID]: 'run-1' }); await new Promise((r) => setTimeout(r, 5)); - expect(sessionOnError).toHaveBeenCalledWith(expect.toBeErrorInfo({ code: ErrorCode.CancelListenerError })); + expect(sessionOnError).toHaveBeenCalledWith(expect.toBeErrorInfo({ code: ErrorCode.RunCancelHandlerFailed })); + }); + + it('a cancel-dispatch failure surfaces as RunCancelRoutingFailed, not as a handler error', async () => { + // Distinct from RunCancelHandlerFailed: the dispatch itself could not + // complete, so the cancel was neither honoured nor rejected. Reached when + // the run's own onError throws while reporting an onCancel failure — the + // throw escapes the cancel handler and lands in the routing bracket. + const sessionOnError = vi.fn(); + session.on('error', sessionOnError); + const run = createRunFromOpts(session, { + runId: 'run-1', + // eslint-disable-next-line @typescript-eslint/require-await -- mock + onCancel: async () => { + throw new Error('handler boom'); + }, + onError: () => { + throw new Error('reporter boom'); + }, + }); + await run.start(); + + simulateCancel(channel, { [HEADER_RUN_ID]: 'run-1' }); + await new Promise((r) => setTimeout(r, 5)); + + expect(sessionOnError).toHaveBeenCalledWith( + expect.toBeErrorInfo({ + code: ErrorCode.RunCancelRoutingFailed, + statusCode: 500, + message: 'unable to route cancel message; reporter boom', + }), + ); + // The run was never aborted — the cancel did not take effect. + expect(run.abortSignal.aborted).toBe(false); + }); + + it('onSteer throws → RunSteerHandlerFailed on the run onError, and the run is unaffected', async () => { + // Distinct from RunCancelHandlerFailed: the steering message has already + // folded in by the time onSteer is notified, so only the notification + // failed. A functional decoder is needed so the delivered steer yields a + // real input event (the default mock decoder returns none). + const ch = createMockChannel(); + const functionalCodec = createMockCodec(); + functionalCodec.createDecoder = vi.fn(() => ({ + decode: (m: Ably.InboundMessage) => { + const hdrs = (m.extras as { ai?: { transport?: Record } } | undefined)?.ai?.transport ?? {}; + const id = hdrs[HEADER_CODEC_MESSAGE_ID] ?? 'unknown'; + return { + inputs: [{ kind: 'user-message' as const, message: { id, content: id } }], + outputs: [], + }; + }, + })); + const steerSession = createAgentSession({ + client: createMockClient(ch), + channelName: 'test-channel', + codec: functionalCodec, + }); + await steerSession.connect(); + + const runId = 'run-steer-throw'; + const invocationId = 'inv-steer-throw'; + const inputEventId = 'p-steer-throw'; + const onError = vi.fn(); + const run = createRunFromOpts(steerSession, { + runId, + invocationId, + inputEventId, + onError, + onSteer: () => { + throw new Error('steer hint boom'); + }, + }); + const startPromise = run.start(); + deliverInputEvent(ch, { + invocationId, + codecMessageId: 'id-1', + serial: 'serial-1', + inputEventId, + publisherClientId: 'user-a', + }); + await startPromise; + + // The steer folds into the open run, firing the throwing onSteer hint. + deliverInputEvent(ch, { + invocationId, + runId, + codecMessageId: 'id-2', + serial: 'serial-2', + inputEventId: 'p-id-2', + publisherClientId: 'user-a', + }); + + expect(onError).toHaveBeenCalledWith( + expect.toBeErrorInfo({ + code: ErrorCode.RunSteerHandlerFailed, + statusCode: 500, + message: `unable to notify steer for run ${runId}; onSteer handler threw: steer hint boom`, + }), + ); + // Only the notification failed: the steer still folded in, so the run + // still sees it as fresh input, and the run was not aborted. + expect(run.hasInput()).toBe(true); + expect(run.abortSignal.aborted).toBe(false); + + await steerSession.detach(); + }); + + it('step.start() throws RunLifecycleEventPublishFailed on a step-start publish failure', async () => { + // A step-lifecycle publish failure surfaces identically to a + // run-lifecycle one — same code, same message shape, cause preserved. + const run = createRunFromOpts(session, { runId: 'run-1' }); + await run.start(); + + const cause = new Ably.ErrorInfo('publish refused', 40160, 401); + vi.mocked(channel.publish).mockRejectedValueOnce(cause); + const step = run.createStep(); + + await expect(step.start()).rejects.toBeErrorInfo({ + code: ErrorCode.RunLifecycleEventPublishFailed, + statusCode: 500, + message: 'unable to publish step-start for run run-1; publish refused', + cause, + }); + }); + + it('step.end() throws RunLifecycleEventPublishFailed on a step-end publish failure', async () => { + const run = createRunFromOpts(session, { runId: 'run-1' }); + await run.start(); + const step = run.createStep(); + await step.start(); + + const cause = new Ably.ErrorInfo('publish refused', 40160, 401); + vi.mocked(channel.publish).mockRejectedValueOnce(cause); + + await expect(step.end()).rejects.toBeErrorInfo({ + code: ErrorCode.RunLifecycleEventPublishFailed, + statusCode: 500, + message: 'unable to publish step-end for run run-1; publish refused', + cause, + }); }); }); @@ -3533,7 +3675,7 @@ describe('AgentSession', () => { it('surfaces a per-run terminal publish failure via on(error) and still tears the rest down', async () => { // A run-end publish that fails on session.end() must NOT abort the teardown: - // it is logged + emitted as RunLifecycleError on the session emitter, and + // it is logged + emitted as RunLifecycleEventPublishFailed on the session emitter, and // the OTHER open runs are still ended and the channel still detached // (best-effort per run). const ch = createMockChannel(); @@ -3558,10 +3700,10 @@ describe('AgentSession', () => { await s.end(); // Both runs' terminal failures surfaced on the session emitter as - // RunLifecycleError (one per run), never silently dropped. + // RunLifecycleEventPublishFailed (one per run), never silently dropped. expect(onError).toHaveBeenCalledTimes(2); for (const [emitted] of onError.mock.calls) { - expect(emitted).toBeErrorInfoWithCode(ErrorCode.RunLifecycleError); + expect(emitted).toBeErrorInfoWithCode(ErrorCode.RunLifecycleEventPublishFailed); } // Teardown still completed: both controllers aborted and the channel detached. expect(run1.abortSignal.aborted).toBe(true); diff --git a/test/core/transport/lifecycle-publish.test.ts b/test/core/transport/lifecycle-publish.test.ts new file mode 100644 index 000000000..a47a393e6 --- /dev/null +++ b/test/core/transport/lifecycle-publish.test.ts @@ -0,0 +1,93 @@ +import '../../helper/expectations.js'; + +import * as Ably from 'ably'; +import { describe, expect, it, vi } from 'vitest'; + +import { publishLifecycleEvent } from '../../../src/core/transport/lifecycle-publish.js'; +import { ErrorCode } from '../../../src/errors.js'; +import { type LogHandler, LogLevel, makeLogger } from '../../../src/logger.js'; + +const silentLogger = makeLogger({ logLevel: LogLevel.Silent }); + +/** + * A publish that resolves with `serial`. + * @param serial - The ACK serial the publish reports. + * @returns The publish thunk. + */ +const publishOk = (serial: string) => vi.fn<() => Promise>().mockResolvedValue(serial); + +/** + * A publish that rejects with `error`. + * @param error - The failure the publish rejects with. + * @returns The publish thunk. + */ +const publishFails = (error: unknown) => vi.fn<() => Promise>().mockRejectedValue(error); + +describe('publishLifecycleEvent', () => { + it('returns the publish result untouched on success', async () => { + await expect( + publishLifecycleEvent( + { phase: 'step-start', method: 'openStep', runId: 'run-1', logger: silentLogger }, + publishOk('serial-1'), + ), + ).resolves.toBe('serial-1'); + }); + + it('wraps a publish failure as RunLifecycleEventPublishFailed, naming the phase and run', async () => { + const cause = new Ably.ErrorInfo('publish refused', 40160, 401); + await expect( + publishLifecycleEvent( + { phase: 'run-start', method: 'start', runId: 'run-1', logger: silentLogger }, + publishFails(cause), + ), + ).rejects.toBeErrorInfo({ + code: ErrorCode.RunLifecycleEventPublishFailed, + statusCode: 500, + message: 'unable to publish run-start for run run-1; publish refused', + cause, + }); + }); + + it('wraps a non-ErrorInfo failure, leaving no cause to preserve', async () => { + await expect( + publishLifecycleEvent( + { phase: 'run-end', method: 'end', runId: 'run-2', logger: silentLogger }, + publishFails(new Error('socket closed')), + ), + ).rejects.toBeErrorInfo({ + code: ErrorCode.RunLifecycleEventPublishFailed, + statusCode: 500, + message: 'unable to publish run-end for run run-2; socket closed', + }); + }); + + it('logs the failure at error with the run, and the caller-supplied context', async () => { + const handler = vi.fn(); + const logger = makeLogger({ logLevel: LogLevel.Error, logHandler: handler }); + + await expect( + publishLifecycleEvent( + { phase: 'step-end', method: 'closeStep', runId: 'run-1', logger, logContext: { stepId: 'step-7' } }, + publishFails(new Error('publish failed')), + ), + ).rejects.toBeErrorInfoWithCode(ErrorCode.RunLifecycleEventPublishFailed); + + expect(handler).toHaveBeenCalledWith( + 'Run.closeStep(); failed to publish step-end', + LogLevel.Error, + expect.objectContaining({ runId: 'run-1', stepId: 'step-7' }), + ); + }); + + it('does not log when the publish succeeds', async () => { + const handler = vi.fn(); + const logger = makeLogger({ logLevel: LogLevel.Error, logHandler: handler }); + + await publishLifecycleEvent( + { phase: 'run-suspend', method: 'suspend', runId: 'run-1', logger }, + publishOk('serial-1'), + ); + + expect(handler).not.toHaveBeenCalled(); + }); +}); diff --git a/test/core/transport/session-support.test.ts b/test/core/transport/session-support.test.ts index 5eeb73365..b21889579 100644 --- a/test/core/transport/session-support.test.ts +++ b/test/core/transport/session-support.test.ts @@ -266,10 +266,10 @@ describe('subscribeAndAttach', () => { }); describe('wrapMessageProcessingError', () => { - it('wraps a thrown value as a SessionSubscriptionError preserving the cause', () => { + it('wraps a thrown value as a SessionMessageProcessingFailed preserving the cause', () => { const cause = new Ably.ErrorInfo('boom', 50000, 500); expect(wrapMessageProcessingError(cause)).toBeErrorInfo({ - code: ErrorCode.SessionSubscriptionError, + code: ErrorCode.SessionMessageProcessingFailed, statusCode: 500, message: 'unable to process channel message; boom', cause, @@ -293,7 +293,7 @@ describe('handleWireMessage', () => { }, onError); expect(onError).toHaveBeenCalledOnce(); expect(onError.mock.calls[0]?.[0]).toBeErrorInfo({ - code: ErrorCode.SessionSubscriptionError, + code: ErrorCode.SessionMessageProcessingFailed, statusCode: 500, message: 'unable to process channel message; bad message', }); diff --git a/test/helper/run-from-opts.ts b/test/helper/run-from-opts.ts index 717f83ad3..54db02162 100644 --- a/test/helper/run-from-opts.ts +++ b/test/helper/run-from-opts.ts @@ -20,6 +20,7 @@ interface RunOpts { onCancelled?: (write: (event: TOutput) => Promise) => void | Promise; onCancel?: (request: CancelRequest) => Promise; onError?: (error: Ably.ErrorInfo) => void; + onSteer?: () => void; } /** @@ -59,6 +60,7 @@ export const createRunFromOpts =