@helix-agents/agent-server

HTTP server for hosting agents remotely. Provides AgentServer for agent lifecycle management, HTTP adapters for routing, and SSE streaming utilities.

Installation

npm install @helix-agents/agent-server

AgentServer

The main class for hosting agents over HTTP.

Constructor

import { AgentServer } from '@helix-agents/agent-server';

const server = new AgentServer(config: AgentServerConfig);

AgentServerConfig

interface AgentServerConfig {
  /** Registry of agents this server can run. Keys are agent type identifiers. */
  agents: Record<string, AgentConfig<any, any>>;

  /** State store for persisting session state */
  stateStore: SessionStateStore;

  /** Stream manager for real-time event streaming */
  streamManager: StreamManager;

  /** Executor (any runtime) */
  executor: AgentExecutor;

  /**
   * OpenTelemetry for the server's requests (#179). With it, each HTTP request is a SERVER span
   * (`POST /start`, `GET /sse`, …) under the trace it carries, read through this option's propagator, and
   * the run a `/start` or `/resume` starts nests under that span. Without it: no extraction, no span, and
   * runs behave as before. See "Tracing" below.
   */
  telemetry?: TelemetryOptions;

  /**
   * Optional usage store. When wired, `GET /usage` serves
   * `getRollup(sessionId, { includeSubAgents: true })` — the recursive fold —
   * and it also backs the consumer's `getUsage()` fallback fold. This does
   * **not** attach `usage` to the SSE `end` frame: no `AgentServer`-hosted
   * executor (JS/Temporal/DBOS/Cloudflare-Workflows) puts `usage` on `end`
   * today — only the Cloudflare Durable Object producer does. When absent,
   * `/usage` responds 404 `FEATURE_UNAVAILABLE` and remote consumers degrade
   * gracefully (skip the usage fold with a warning).
   */
  usageStore?: UsageStore;

  /**
   * Optional PURE state mapper. Its presence makes this instance a PROJECTED
   * mount: applied producer-side to point-in-time full states only —
   * `getSnapshot().state` and the SSE `end` frame's `state` — and every
   * `state_patch` SSE frame is SUPPRESSED (root, descendants, and
   * `CustomStateStreamer` frames alike). All other frames pass through with
   * their original sequence numbers, so `fromSequence` resumption is
   * unaffected. Absent = raw mount.
   */
  projection?: (state: Record<string, unknown>) => Record<string, unknown>;

  /**
   * Optional usage-rollup mapper (e.g. tokens → billing units). Applied to
   * `GET /usage`'s `.rollup`, and (code path exists, but see note below) the
   * SSE `end` frame's `usage`. Absent = raw rollup. `usageProjection` alone
   * does NOT suppress `state_patch` frames — only `projection` does.
   *
   * Note: on this (`AgentServer`) producer the `end`-frame application is
   * unreachable in practice — as documented on `usageStore` above, no
   * `AgentServer`-hosted executor ever attaches `usage` to `end`, so there is
   * never a rollup for this mapper to project there. It only has an
   * observable effect on `GET /usage`.
   */
  usageProjection?: (rollup: UsageRollup) => Record<string, unknown>;

  /**
   * Optional authentication hook. Called before each HTTP operation; see
   * {@link AuthenticateHook}. If omitted, the constructor throws unless
   * `allowUnauthenticated: true` is also set.
   */
  authenticate?: AuthenticateHook;

  /**
   * Explicit opt-in for running without an `authenticate` hook. The
   * constructor emits a loud warning when set.
   */
  allowUnauthenticated?: boolean;

  /**
   * #177: what the HTTP `/start` and `/resume` apply of a body's client state:
   * `statePatch` is refused and `state` accepted by default; change either
   * (`{ state?, statePatch? }`) or decide per request with a hook. See
   * "Client state on /start and /resume" below.
   */
  clientState?: ClientStateHook | { state?: 'accept' | 'refuse'; statePatch?: 'accept' | 'refuse' };

  /** Optional structured logger (defaults to noop). */
  logger?: Logger;

  /** Whether to leak Zod issue details in 400 responses: an invalid payload's `details`, an `INVALID_RESULT`'s `issues` and detailed message (those two redacted with core `sanitizeToolErrorText`; defaults `false`). */
  verboseValidationErrors?: boolean;

  /** Optional human-readable service name surfaced in startup logs. */
  serviceName?: string;

  /**
   * Workspace provider kinds wired into the underlying executor. When
   * set, AgentServer validates at construction that every agent declaring
   * a workspace provider kind has a matching entry here.
   */
  workspaceProviderKinds?: ReadonlyArray<string>;

  /**
   * Phase 7 chat-handler orchestrator (Layer B). Optional — when set,
   * the `/chat` family of routes is wired in. When omitted, those
   * routes respond `501 NOT_IMPLEMENTED` and the legacy routes
   * remain fully operational. Wire by binding `handleChatStream` from
   * `@helix-agents/ai-sdk`:
   *
   * ```ts
   * import { handleChatStream } from '@helix-agents/ai-sdk';
   *
   * new AgentServer({
   *   ...,
   *   chatHandler: (params) => handleChatStream(
   *     { executor, stateStore, streamManager, agent, logger },
   *     params,
   *   ),
   * });
   * ```
   */
  chatHandler?: ChatStreamHandler;
}

Two mounts = two fidelities

Two AgentServer instances over the same stores — one with projection / usageProjection, one without, each with its own authenticate — give you a raw internal mount and a projected external mount. Point executor-to-executor consumers at a raw mount: a transport aimed at a projected mount folds the lossy projected view as the child's customState. See the Cross-Service Remote Agents guide.

Methods

startAgent

Start a new agent execution. Idempotent — returns the existing session if already running.

const response = await server.startAgent(request: RemoteStartRequest): Promise<RemoteStartResponse>;

Parameters:

interface RemoteStartRequest {
  sessionId: string; // Unique session identifier
  agentType: string; // Must match a key in the agents registry
  message: string | UserInputMessage[]; // Initial user message (string or structured array)
  state?: Record<string, unknown>; // Optional initial custom state
  statePatch?: JsonMergePatch; // #177: the turn's state patch (RFC 7396); the HTTP route gates it (clientState)
  metadata?: Record<string, string>; // Optional metadata
  requestKey?: string; // D-2: the request key (the HTTP route reads X-Helix-Request-Key)
}

Returns:

interface RemoteStartResponse {
  sessionId: string;
  streamId: string;
  runId: string;
  status?: 'started' | 'resumed' | 'attached'; // 'attached' only for a keyed duplicate
}

Errors:

• NOT_FOUND — Unknown agentType
• ALREADY_COMPLETED — Session already completed or failed (not for a keyed duplicate, see Request keys)
• A typed HelixError — the executor's execute() threw one, or returned a handle whose entry failed before its run-start commit (preCommitFailure, rethrown with helixErrorFromDetail; resolveErrorDetail(error) returns that detail verbatim)

resumeAgent

Resume an interrupted or paused agent execution.

const response = await server.resumeAgent(request: RemoteResumeRequest): Promise<RemoteStartResponse>;

Parameters:

interface RemoteResumeRequest {
  sessionId: string;
  message?: string | UserInputMessage[]; // Optional message to send on resume
  requestKey?: string; // D-2: the request key (the HTTP route reads X-Helix-Request-Key)
}

Errors:

• NOT_FOUND — Session not found
• ALREADY_COMPLETED — Session already completed or failed (not for a keyed duplicate, see Request keys)
• A typed HelixError — the executor's resume() threw one, or returned a handle whose entry failed before its run-start commit (preCommitFailure, rethrown with helixErrorFromDetail; resolveErrorDetail(error) returns that detail verbatim)

streamAgent

Subscribe to agent execution events. Returns an async iterable of transport events.

const events = server.streamAgent(
  sessionId: string,
  fromSequence?: number
): AsyncIterable<TransportEvent>;

Parameters:

• sessionId — Session to stream
• fromSequence — Skip events up to this sequence number (for reconnection)

Transport events:

type TransportEvent =
  | { type: 'chunk'; chunk: StreamChunk; sequence: number }
  | {
      type: 'end';
      status: 'completed' | 'paused' | 'interrupted';
      output?: unknown;
      state?: Record<string, unknown>;
      usage?: UsageRollup;
    }
  | { type: 'fail'; error: string; errorDetail?: ErrorDetail; recoverable: boolean }
  | { type: 'truncated'; floor: number; errorDetail: ErrorDetail };

The iterable always ends with one terminal event, taken from the stream's own info when the reader ends: ended → end with its end status (completed, or interrupted for a stopped run), paused → end { status: 'paused' }, failed → a non-recoverable fail with its errorDetail. A reader that ends while the stream is still active (or gone, for a session not known to be completed) did not see the end: it yields a recoverable fail with errorDetail.code transport_stream_drop, never a synthesized end { status: 'completed' }. The v1 error event is fail (upgrade guide).

getStatus

Get the current execution status for a session.

const status = await server.getStatus(sessionId: string): Promise<RemoteStatusResponse>;

Returns:

interface RemoteStatusResponse {
  sessionId: string;
  runId?: string;
  status: 'running' | 'completed' | 'failed' | 'interrupted' | 'paused';
  stepCount: number;
  output?: unknown;
  state?: Record<string, unknown>;
  error?: string; // sanitized
  errorDetail?: ErrorDetail; // message and cause sanitized, code kept
  isExecuting: boolean;
  streamId: string;
  latestSequence: number; // Current stream position from StreamManager.getStreamInfo()
}

latestSequence reflects the actual stream position from the stream manager. Clients can use this value as fromSequence when reconnecting to /sse to avoid replaying already-seen events.

error and errorDetail leave the server sanitized (core sanitizeErrorMessage / sanitizeErrorDetail, error text on outward surfaces); the session row keeps the raw text.

interruptAgent

Soft stop of the session's current run (the session can be resumed later). Goes through core stopRun and answers at once with a StopAnswer; it never waits for the run.

await server.interruptAgent(
  sessionId: string,
  reason?: string,
  delegation?: StopDelegation // CP-217: { dispatchId } | { parentToolCallId }
): Promise<StopAnswer>;

• { status: 'stopping', runId, kind } — recorded and signalled; the run is executing (HTTP 202).
• { status: 'interrupted', runId } — recorded and resolved by this call, e.g. a run paused on a client tool (HTTP 200).
• { status: 'not_stoppable', currentStatus } — no live run; nothing written (HTTP 409 state_not_stoppable).
• { status: 'target_mismatch' } — with delegation: the session serves another delegation (a dispatch its binding does not name, or a consult its run does not record); nothing written (HTTP 409 state_stop_target_mismatch). The remote-protocol /interrupt / /abort routes always pass one.

Errors: rejects with a HelixError state_session_not_found (HTTP 404) when the session does not exist.

abortAgent

Hard stop: the run ends failed with framework_cancelled. Same stopRun path and answers as interruptAgent ({ status: 'failed', runId } when resolved by the call).

With no reason, both record core's default (DEFAULT_INTERRUPT_REASON / DEFAULT_ABORT_REASON, stopReasonOrDefault), the same default every runtime uses.

await server.abortAgent(sessionId: string, reason?: string): Promise<StopAnswer>;

submitToolResult

Submit a result for a paused client-executed tool. The submit size limit is the hosted executor's own (DI-51): configure it on the executor (submitToolResultSchema: createSubmitToolResultSchema({ maxResultBytes })), never on the server. server.submitLimits reads it back (executor.submitLimits), and /submit-tool-result pre-checks against it: a result over it, or a body over maxResultBytes × 4, is core's typed 413 validation_payload_too_large ({ error, code, retryable: false, cause: 'tool_result' | 'request_body', limitBytes, sizeBytes? }); a body that is not a valid submit is a typed 400 validation_error with a fixed message (the Zod details only with verboseValidationErrors); a chunked body without Content-Length is a typed 411 (cause: 'length_required'). The executor's own 413 (an executor with no declared limits) is relayed as 413. A result the state store refuses as too large to hold (a batch whose stored pending map would exceed the store's limit) is core's typed 413 state_size_exceeded ({ error, code, retryable: false, cause, limitBytes, sizeBytes }), nothing stored; it used to be a 500. See the submit size limit.

const result = await server.submitToolResult(payload: SubmitToolResult): Promise<SubmitToolResultResponse>;

describeWorkspace

Operator-facing introspection of the agent's workspace registry for a session. Returns a discriminated union: { kind: 'ok', snapshot: EntrySnapshot | null }, { kind: 'no-runtime-support' }, or { kind: 'no-registry' }. Most consumers should use the HTTP GET /workspace?sessionId=… route instead (response { workspace: EntrySnapshot | null }, operation tag 'workspace').

Request keys

POST /start, POST /resume and POST /chat read the request key (D-2) from the X-Helix-Request-Key header. A malformed key (not 1–128 characters of [A-Za-z0-9._:-]) answers 400 { error, code: 'validation_error' } before anything else runs (after authentication), and the key is never echoed.

• /start and /resume pass the key to startAgent / resumeAgent (RemoteStartRequest.requestKey, RemoteResumeRequest.requestKey), which pass it on the executor call. When a run of the session already holds the key (getRunByRequestKey), the request is a repeat: startAgent skips its session-exists refusals (ALREADY_COMPLETED / ALREADY_RUNNING) and resumeAgent its completed / failed refusal. A repeated user send is answered from the run that holds the key, without calling the executor: core planKeyedSend treats one message as one request whichever route each attempt took (a first /start retried as /resume once the session exists is the same send). A run that holds the key and is still running goes to the executor instead (it attaches to a live run, or takes a dead owner's run over), as does a keyed /resume without a message. A duplicate whose run is live, or finished and still current, answers 200 { sessionId, streamId, runId, status: 'attached' } naming that run (its handle is not tracked; the original entry's is). A key whose run ended while a later run exists answers 409 { error, code: 'state_request_already_completed', runId }, and a key reused for other input 409 { error, code: 'state_request_key_conflict', runId } (core typedErrorResponse).
• /chat passes the key to the ChatStreamHandler as requestKey. A handler that hands its params to handleChatStream (as in the chatHandler example above) forwards it as params.requestKey (see keyed sends).

Only an executor that records keys dedupes them (runtime-js; supportsRequestKey: true). On CF Workflows, Temporal and DBOS the key is ignored with a warning and the request runs keyless, so the existing refusals apply.

Client-tool deadlines on serverless hosts

A serverless or frozen host does not advance a run at a client-tool deadline. The JS executor behind AgentServer fires a lapsed call's deadline from an in-process timer: at the deadline it expires the call (client_tool_timeout) and continues the run, with no request needed, but only while that process is running.

When the process exits after the suspension, or the platform freezes it between requests (a serverless function, a scaled-to-zero container), nothing fires. The run stays suspended until a request that reaches the executor recovers it:

• a /resume or a new message expires the lapsed call in its drain and continues;
• a submit re-arms the timer in the now-live process, when the executor was built with an agentRegistry.

/snapshot, /status and /sse read the store and do not touch the executor, so a client that only reconnects and reads does not re-arm it.

Host agent-server on a long-lived process if a run must advance on its own while no client is connected. See the JS runtime.

Durable stop model

A stop is recorded durably, then either the executing owner is signalled or the stop's terminal is committed at once. interruptAgent and abortAgent work whether the session is running on this replica, a different replica, or paused; the answer never waits for the run, and there is no 504 (interruptObservationDeadlineMs and INTERRUPT_DEADLINE_EXCEEDED were removed).

AgentServerError

Error class thrown by AgentServer methods.

class AgentServerError extends Error {
  code:
    | 'NOT_FOUND'
    | 'ALREADY_RUNNING'
    | 'ALREADY_COMPLETED'
    | 'INVALID_REQUEST'
    | 'INTERNAL_ERROR'
    | 'WORKSPACE_WIRING'
    | 'NOT_IMPLEMENTED'
    | 'FEATURE_UNAVAILABLE';
  /** On ALREADY_RUNNING: the live stream to attach to. */
  streamId?: string;
  /** The typed core ErrorCode behind it, when there is one; serialized as `errorCode`. */
  errorCode?: string;
}

A /start that passes the existence check but loses the executor's run-start commit to a concurrent /start (RunStartConflictError) answers 409 { code: 'ALREADY_RUNNING', streamId, errorCode: 'state_already_running' }, the same attachable response as the existence check.

HTTP status mapping:

Code	HTTP Status
NOT_FOUND	404
ALREADY_RUNNING	409
ALREADY_COMPLETED	409
INVALID_REQUEST	400
INTERNAL_ERROR	500
WORKSPACE_WIRING	500
NOT_IMPLEMENTED	501
validation_payload_too_large* (submit)	413
state_size_exceeded* (submit)	413
validation_error* (submit, also a 411)	400 / 411
UNAUTHORIZED*	401 / 403
framework_completion_strategy_unavailable*	400
framework_completion_schema_unsupported*	400

* Surfaced from the HTTP adapter directly rather than from AgentServerError.

The two framework_completion_* codes are the executor's early refusal of an outputSchema agent whose model the adapter cannot finish. /start, /resume and both submit-tool-result routes answer them as 400 { error, code } (the code is the core one, as on the Cloudflare DO), before anything is written.

/start and /resume entry failures. POST /start and POST /resume answer every typed entry failure, a HelixError thrown by the executor or a handle whose entry failed before its run-start commit (preCommitFailure, which startAgent / resumeAgent reject with helixErrorFromDetail), with core typedErrorResponse: exactly { error, code } (the core code, message scrubbed) with the core httpStatusForHelixError status, as the Cloudflare DO's /start does. A pre-commit failed handle is never answered as started: on /resume the caller would attach to the session's stream, which still holds the previous turn. A checkpoint id that names another session's checkpoint (state_checkpoint_foreign) is answered exactly as a missing one (404 state_checkpoint_not_found, the same message), and the foreign detail is logged at warn through the server's logger.

/start entry failure	HTTP Status
framework_agent_not_found (an AgentResolutionError miss)	404
state_session_not_found / state_checkpoint_not_found	404
A completion-support refusal or validation_error	400
A retryable error that left no session behind (nothing written)	503
A retryable error that left the session behind (or the session read failed)	409 / 500
A non-retryable state_* error (e.g. state_history_incomplete)	409
Any other typed error	500

/start only enters a session that does not exist, so a retryable error is 503 only while the failed entry left no session behind. Once the session exists, a re-sent /start is refused by startAgent's pre-check (ALREADY_COMPLETED / ALREADY_RUNNING), so the error gets its non-retryable row (409 for state_*, else 500). That covers an entry whose run committed (its input is in the log; a re-send would append it twice) and a pre-commit failure after a fresh session's createSession landed (runtime-js, CF Workflows).

/resume uses the same table, with "the session's current run moved during the call" (the resume committed a new run, so its message is in the log) in place of "left the session behind": a retryable error is 503 only while the current run did not move.

A failed session or run read behind either decision is answered as committed (never 503) and logged as a warn through the server's logger. A plain (non-HelixError) failure keeps the 500 { code: 'INTERNAL_ERROR' } envelope. On the other routes, any typed framework error other than the completion refusals keeps the 500 { code: 'INTERNAL_ERROR', errorCode } envelope.

Authentication

AuthenticateHook

Authentication hook called before every HTTP operation. Set via AgentServerConfig.authenticate; the server is fail-closed without it (constructor throws unless allowUnauthenticated: true is set).

type AuthenticateHook = (
  request: AgentHttpRequest,
  operation: AgentServerOperation
) => Promise<AuthenticateResult> | AuthenticateResult;

type AuthenticateResult = boolean | { error: string; status?: number };

true → request proceeds. false → 401 generic. { error, status? } → custom rejection (use status: 403 for "authenticated but not authorized").

A start or resume body may carry a child binding (a remote parent's delegation). It is caller-asserted: the server stores it on the child session it creates and checks later entries against it, but cannot verify it. Admit start / resume requests with a binding only from your parent services. The child and dispatch ids it names are unguessable but not secret. See The child binding on the protocol.

Client state on /start and /resume

A body can carry two kinds of client-supplied state, and AgentServerConfig.clientState is the one place the host decides what the HTTP routes apply. The defaults differ by risk:

• statePatch (/start, and /resume with a message): an RFC 7396 JSON Merge Patch of an EXISTING session's customState, applied to the turn's entry state and committed with its input (#177). The server checks it against the agent's stateSchema on the keys it touches, nothing more. Refused by default: a non-empty patch is the typed 400 validation_error, and nothing runs. An empty patch ({}) is no patch. A patch that is not a JSON object, or that carries a __proto__ key at any depth, is the typed 400 validation_error whatever the policy (core requestStatePatchSchema; the key is refused, never dropped).
• state (/start): the FRESH session's initial customState. Accepted by default: it only seeds a new session, and every runtime's remote sub-agent dispatch sends the call's input through it, so refusing it would break every cross-service producer.

Change either default with clientState: { state?: 'accept' | 'refuse', statePatch?: 'accept' | 'refuse' }, or decide per request with a ClientStateHook: (ctx: { operation: 'start' | 'resume'; sessionId; state?; statePatch?; request }) => { state?; statePatch? }. The hook sees both fields and returns what to apply (a field left out is not applied, and the entry runs without it). On /resume the hook's state is ignored: a resume never seeds state. Under every policy, a state that is not a JSON object is the typed 400 validation_error (naming state in detail), before the hook runs; malformed input is never "accepted".

Security note: body state seeds a fresh session's customState from the caller. If your agent keeps authorization-relevant keys (role, tier, tenant) in customState, filter state in the hook, or refuse it in authenticate. Accept a statePatch only from callers that may write every key it touches. The programmatic startAgent / resumeAgent are the host's own calls and are not gated. An accepted patch is still checked against the agent's stateSchema, its object-level refinements (.refine() / .superRefine()) included: a patch that breaks one is the typed 400 validation_error naming the statePatch and the refinement (see State).

AgentServerOperation

Stable operation tags passed to the hook. Names are NOT 1:1 with paths — they're preserved across releases so existing switches keep working.

type AgentServerOperation =
  | 'start'
  | 'resume'
  | 'stream' // GET /sse
  | 'status'
  | 'interrupt'
  | 'abort'
  | 'submit-tool-result'
  | 'workspace'
  // Cross-service remote-agent protocol routes
  | 'snapshot' // GET /snapshot — the checkpoint-pinned state baseline
  | 'usage' // GET /usage — the recursive usage rollup
  // Layer B chat routes (Phase 7 Task 7.8)
  | 'chat'
  | 'chat-stream'
  | 'chat-submit-tool-result'
  | 'chat-interrupt'
  | 'chat-abort';

'snapshot' gates a high-exfiltration surface (it serves the session's full customState) — gate it accordingly.

Tracing

With telemetry set, createHttpAdapter (and so createExpressAdapter) opens one SERVER span per request:

• It is named {METHOD} {route} (POST /start, GET /chat/{sessionId}/stream, else /*).
• It is parented on the inbound trace, read only through the configured propagator.
• It ends when the response is returned. An SSE body is not part of it.
• The run a /start or /resume starts is parented on it (ExecuteOptions.telemetryParent). runtime-js honours that per-call parent; Temporal, DBOS and CF Workflows ignore it for now.
• POST /chat passes it to the chat handler as telemetryParent. With the documented wiring (chatHandler: (params) => handleChatStream(deps, params)), the chat turn's run parents on it too, including a client-tool answer's caller_resumes continuation.

Without telemetry nothing is extracted and no span is opened.

Authenticating requests is the app's (authenticate), and so is deciding whether to trust a request's trace headers: a server open to untrusted callers lets them choose the trace its runs join.

HTTP Adapters

createHttpAdapter

Creates a generic HTTP handler from an AgentServer instance. Framework-agnostic.

import { createHttpAdapter } from '@helix-agents/agent-server';

const handler: AgentHttpHandler = createHttpAdapter(server);

AgentHttpHandler signature:

interface AgentHttpHandler {
  handleRequest(req: AgentHttpRequest): Promise<AgentHttpResponse>;
}

AgentHttpRequest:

interface AgentHttpRequest {
  method: string;
  url: string;
  headers: Record<string, string | string[] | undefined>;
  body?: unknown;
  query?: Record<string, string | string[] | undefined>;
}

AgentHttpResponse:

interface AgentHttpResponse {
  status: number;
  headers: Record<string, string>;
  body: string | ReadableStream<Uint8Array>;
}

Routes handled:

Layer	Method	Path	Action
A	POST	/start	Start agent execution
A	POST	/resume	Resume agent execution
A	GET	/sse	SSE event stream
A	GET	/status	Get execution status
A	POST	/interrupt	Soft stop (durable-flag protocol)
A	POST	/abort	Hard stop (durable-flag protocol)
A	POST	/submit-tool-result	Submit a client-executed tool result
A	GET	/workspace	Operator introspection of the workspace registry for a session
A	GET	/snapshot?sessionId=	Checkpoint-pinned RemoteSnapshotResponse (this session's customState + its aligned streamSequence)
A	GET	/usage?sessionId=	RemoteUsageResponse — getRollup({ includeSubAgents: true }); 404 FEATURE_UNAVAILABLE without a usageStore
B	POST	/chat	AI SDK v6-native chat stream (requires chatHandler)
B	GET	/chat/{sessionId}/stream	Reconnect to an active chat stream
B	POST	/chat/{sessionId}/submit-tool-result	Chat-route variant of /submit-tool-result (sessionId in path)
B	POST	/chat/{sessionId}/interrupt	Chat-route soft stop
B	POST	/chat/{sessionId}/abort	Chat-route hard stop

Layer B routes respond 501 NOT_IMPLEMENTED when the server is constructed without a chatHandler.

Submit answers. Both submit-tool-result routes parse the executor's answer with core submitToolResultResponseSchema() before answering (unknown keys are stripped; a malformed answer is a typed 500 whose errorCode is transport_error): 200 accepted / already_completed with their continuation, 404{ status: 'unknown_tool_call' }, or 404 { status: 'unknown_tool_call', reason: 'run_stopped' } when the call's run was stopped (WA MR2 §7.1: the runtime stored nothing and continues nothing; see the client-tools guide).

/snapshot is checkpoint-pinned. For a non-terminal session the handler serves the latest checkpoint's (state, streamSequence) — the only pairing written atomically (at step promote). On Node stores this is a bounded pointer-retry loop (loadState → getLatestCheckpoint → verify state.checkpointId === checkpoint.id); if it can't get a consistent view it degrades to streamSequence: -1 (snapshot-only) rather than hand out a misaligned replay origin. Terminal sessions, sessions with no checkpoint yet, and the degraded sentinel all return -1 — never derive a fromSequence from it. Unknown session → 404 with a RemoteAgentErrorResponse envelope (code: 'NOT_FOUND'), never null.

end.state is captured at the endStream call site, not re-loaded after the reader drains: a fast follow-up /resume can re-mutate (even re-complete) session state before the reader finishes, so live session state is untrustworthy for the terminal frame. end.state rides the end frame when the run loop's terminal persist committed; end.usage does not — no AgentServer-hosted executor attaches usage to the end frame today (computeTerminalUsage is a stub), so consumers always recover the rollup via getUsage(). When state is absent, consumers fall back to getSnapshot().

Frames: GET /sse speaks stream wire contract v2 (chunk { sequence, chunk }, end { status, output?, state?, usage? }, fail { error, errorDetail?, recoverable }, truncated { floor, errorDetail }), encoded with core streamFrameToSSE. Every body ends with exactly one terminal frame: when streamAgent ends without one, the handler writes a recoverable fail with errorDetail.code transport_stream_drop, and a thrown error is a fail that is recoverable when the error is retryable. Messages and details on fail are sanitized. A body that ends with no terminal frame (a proxy cut, a crash) is a drop: readers throw the retryable StreamDropError and re-attach from their cursor.

Truncation: a GET /sse whose fromSequence is below the stream's retention floor (getStreamInfo().oldestRetainedSequence) gets a single terminal truncated { floor, errorDetail } frame (errorDetail.code: 'TRUNCATED') instead of a silently gapped replay; the consumer refetches GET /snapshot and resumes from its streamSequence. fromSequence: 0 is a legitimate pinned origin and is gated the same way.

createExpressAdapter

Wraps an AgentHttpHandler as Express middleware.

import { createExpressAdapter } from '@helix-agents/agent-server';

const middleware = createExpressAdapter(handler: AgentHttpHandler);

app.use('/agents', middleware);

Features:

• Uses req.path (relative to mount point) for routing
• Handles both string and ReadableStream responses
• Pipes SSE streams with proper disconnect cleanup

SSE Utilities

createSSEStream

Creates a pull-based ReadableStream that emits SSE-formatted data from an async source. The consumer controls the pace — no unbounded buffering. Heartbeats are emitted as SSE comments when the source has no events within the heartbeat interval.

import { createSSEStream, type SSEEvent } from '@helix-agents/agent-server';

async function* events(): AsyncGenerator<SSEEvent> {
  yield { id: '1', event: 'chunk', data: '{"hello":"world"}' };
  yield { event: 'end', data: '{"done":true}' };
}
// To write the stream protocol's frames, map them with core `streamFrameToSSE(frame)`.

const stream: ReadableStream<Uint8Array> = createSSEStream(events(), {
  heartbeatIntervalMs: 15_000, // default
});

SSEEvent:

interface SSEEvent {
  id?: string;
  event: string;
  data: string;
}

The stream automatically sends : heartbeat comments every 15 seconds (configurable). The stream closes when the source is exhausted; errors are logged and the stream closes gracefully.

expiredSessionCleanup

Operator-driven sweep that scans the state store for sessions whose expiresAt has elapsed, prunes their workspace snapshots (best-effort), and CAS-transitions the session status to 'failed' with reason 'session_expired'. Not run automatically by the framework — wire it into a cron / Cloudflare Alarm / k8s CronJob yourself.

import { expiredSessionCleanup } from '@helix-agents/agent-server';

const summary = await expiredSessionCleanup({
  stateStore,
  workspaceProviders, // optional Map<string, WorkspaceProvider>
  logger, // optional structured logger (defaults to noop)
  pageSize: 200, // optional override (default 200)
  now: () => Date.now(), // optional clock override for tests
});

Returns:

interface ExpiredSessionCleanupResult {
  detected: number; // expired sessions detected
  marked: number; // successfully transitioned to 'failed'
  alreadyTerminal: number; // CAS rejected (already terminal)
  snapshotsDeleted: number; // snapshots successfully deleted
  errors: number; // per-session errors that were logged
}

Per-session failures (load error, snapshot error, CAS conflict) are logged but do not abort the loop — one bad session must not block cleanup of the rest.

HttpRemoteAgentTransport

Client-side transport for communicating with a remote AgentServer. Defined in @helix-agents/core.

import { HttpRemoteAgentTransport } from '@helix-agents/core';

Constructor

interface HttpTransportConfig {
  /** Base URL of the remote agent server */
  url: string;

  /** Static headers or async function returning headers */
  headers?:
    | Record<string, string>
    | (() => Record<string, string> | Promise<Record<string, string>>);

  /**
   * Maximum retry attempts (default: 3) for the reads and stop requests
   * (`/status`, `/snapshot`, `/usage`, `/interrupt`, `/abort`). The entries
   * `/start` and `/resume` are never re-sent (CP-95).
   */
  maxRetries?: number;

  /** Base delay between retries in ms (default: 1000) */
  retryBaseDelayMs?: number;

  /**
   * Custom fetch used for EVERY request this transport issues (POSTs, the
   * status/snapshot/usage GETs, and the SSE stream). Enables Cloudflare
   * service bindings, Miniflare `dispatchFetch` in tests, and Node-side
   * proxies. Defaults to `globalThis.fetch`, resolved lazily at call time.
   */
  fetch?: typeof globalThis.fetch;

  /**
   * OpenTelemetry for this transport's requests (#179). Each request is a CLIENT span named by its path
   * after `url` (`POST /start`, `GET /sse`, …), parented on the context the transport was pinned to
   * (`withTraceContext`), else the active one, and its trace goes out through the configured propagator.
   */
  telemetry?: TelemetryOptions;
}

withTraceContext(parent) returns the same transport with every request parented on parent (the original is unchanged). The runtimes pin it to the call's execute_tool span through core transportFor, so a remote child's run joins the caller's trace. transportFor pins nothing when the parent has no valid span (telemetry off).

Pinning works on a copy, and telemetry never changes a transport's behaviour. HttpRemoteAgentTransport copies itself. A subclass keeps its behaviour: by default it is not copied (its #private fields and constructor-held resources could not be), so it is not pinned and its requests parent on the active context. To keep trace pinning, override protected copyForTraceContext(): this to return a properly constructed instance with the same configuration. DOStubTransport works the same way.

Which trace a request carries:

• With a parent (the pinned span, or an active span): the CLIENT span's trace. It replaces an app-supplied traceparent (or any header the propagator writes) of the same name, in any case, so exactly one goes out.
• With no parent:
  • an app-supplied traceparent goes out as the app set it, and nothing is injected;
  • with none, the CLIENT span's own context goes out when it is valid (a recording root, through a configured provider), so the producer joins the consumer's trace;
  • with a no-op provider the context is invalid and nothing is written.

Methods

Implements the RemoteAgentTransport interface:

interface RemoteAgentTransport {
  start(request: RemoteStartRequest): Promise<RemoteStartResponse>;
  resume(request: RemoteResumeRequest): Promise<RemoteStartResponse>;
  stream(
    sessionId: string,
    options?: {
      fromSequence?: number;
      signal?: AbortSignal;
    }
  ): AsyncIterable<TransportEvent>;
  getStatus(sessionId: string): Promise<RemoteStatusResponse>;
  /** GET /snapshot — checkpoint-pinned state baseline. */
  getSnapshot(sessionId: string): Promise<RemoteSnapshotResponse>;
  /** GET /usage — recursive rollup (`includeSubAgents: true`). */
  getUsage(sessionId: string): Promise<RemoteUsageResponse>;
  /** A parent's stop of its child, naming the delegation (CP-217): { reason, target }. */
  interrupt(sessionId: string, stop: RemoteStopRequest): Promise<void>;
  abort(sessionId: string, stop: RemoteStopRequest): Promise<void>;
  /** POST /submit-tool-result — answer a pending call of the session's tree (CP-157; optional). */
  submitToolResult?(sessionId: string, submit: SubmitToolResult): Promise<SubmitToolResultResponse>;
}

Responses are Zod-validated at the boundary (remoteStartResponseSchema(), remoteStatusResponseSchema(), remoteSnapshotResponseSchema(), remoteUsageResponseSchema()); a schema mismatch throws RemoteProtocolValidationError rather than surfacing later as an undefined field access. Every non-2xx answer, on every route, goes through core remoteProtocolError (shared with the CF DOStubTransport; the entries start / resume through its wrapper remoteEntryProtocolError, see Retry Behavior):

• an envelope ALREADY_RUNNING / NOT_FOUND / FEATURE_UNAVAILABLE becomes RemoteAgentAlreadyRunningError / RemoteAgentNotFoundError / RemoteAgentFeatureUnavailableError;
• any other envelope with an errorCode, or any envelope answered 503, becomes a HelixError carrying the errorCode and cause, retryable exactly on a 503;
• a typed { error, code, cause?, retryable? } body (what POST /start and POST /resume answer a typed entry failure with) becomes a HelixError with that code and cause (helixErrorFromPeerBody: retryable from the body, else on a 503);
• anything else, a bare 404 included, stays a generic Error.

No code is inferred from the message.

Retry Behavior

The reads and stop requests (/status, /snapshot, /usage, /interrupt, /abort):

• 5xx errors — Retried with exponential backoff (baseDelay * 2^attempt), up to maxRetries
• 4xx errors — Not retried
• Network errors — Retried with backoff

The entries /start and /resume are sent exactly once (CP-95): a re-sent entry could start the remote run, or append the resume's message, twice.

• A thrown request, or a 5xx that is not the host's typed retryable answer (an untyped body such as a load balancer's 503 page, or a typed non-retryable one), rejects with the non-retryable RemoteEntryOutcomeUnknownError (transport_remote_entry_outcome_unknown, entry, status?): whether the run started is unknown, so check the remote session's status before acting.
• A typed retryable answer (a Helix host's 503 body, read with Zod) is the host's own typed retryable error: a host never answers "retryable" once the input is committed (DI-39). It is still not re-sent by the transport.
• A 4xx is a definite refusal, mapped as above.

ALREADY_RUNNING is an attach signal, not a failure: the shared dispatch catches RemoteAgentAlreadyRunningError and proceeds to stream the existing run (using the envelope's streamId when present). See Remote agents: entries are sent once.

SSE Parsing

The transport reads the /sse body with native ReadableStream parsing (not EventSource) and decodes it with core decodeTransportEvents (stream wire contract v2):

• chunk, end, fail and truncated frames are TransportEvents; a terminal frame ends the iteration
• a malformed frame is yielded as a recoverable fail
• caught-up, rewound, unknown events (the v1 error included) and comments such as : heartbeat are skipped
• a body that ends with no terminal frame throws the retryable StreamDropError (re-attach from your cursor; never re-send /start)
• AbortSignal cancels the stream; an aborted read ends quietly

A frame that cannot be parsed (bad JSON, or a chunk frame with no numeric sequence) now surfaces as a recoverable error event. It is no longer defaulted to sequence: 0 — that default was silently swallowed by the consumer's <= lastSequence dedup. Recoverable routes the consumer into drop-recovery (reconnect from the last good sequence); a persistently corrupt frame exhausts the retry budget and fails loudly.

createRemoteSubAgentTool

Factory function for creating a tool that delegates to a remote agent. Defined in @helix-agents/core.

import { createRemoteSubAgentTool } from '@helix-agents/core';

Signature

function createRemoteSubAgentTool<TInput, TOutput>(
  name: string,
  config: {
    description: string;
    inputSchema: z.ZodType<TInput>;
    outputSchema: z.ZodType<TOutput>;
    transport: RemoteAgentTransport;
    remoteAgentType: string;
    timeoutMs: number;
    streamRetries?: number;
    streamRetryBaseMs?: number;
    maxFoldStateBytes?: number;
    resumeDeadProducer?: boolean;
  }
): RemoteSubAgentTool<TInput, TOutput>;

Parameters

Parameter	Type	Description
name	string	Tool name (automatically prefixed with subagent__)
description	string	Description shown to the LLM
inputSchema	z.ZodType	Zod schema for tool input
outputSchema	z.ZodType	Zod schema for expected output
transport	RemoteAgentTransport	Transport instance for HTTP communication
remoteAgentType	string	Agent type identifier (must match server registry key)
timeoutMs	number	Maximum execution time in milliseconds (must be a finite positive number)
streamRetries	number (optional)	Max stream reconnection attempts on SSE drop (default: 3, range: 0-50)
streamRetryBaseMs	number (optional)	Base delay in ms between stream retries, doubled each attempt (default: 100)
maxFoldStateBytes	number (optional)	Max serialized size of the child's terminal customState folded into the parent. Over-cap: the state fold is skipped with a warning (output/usage unaffected). Default 262_144 (DEFAULT_MAX_FOLD_STATE_BYTES)
resumeDeadProducer	boolean (optional)	Opt into a single blind transport.resume(...) when the producer reports running + isExecuting: false with no stream progress. Default false — the dispatch fails fast with 'producer-dead' instead (a blind resume is wrong for multi-replica producers)

RemoteSubAgentTool

interface RemoteSubAgentTool<TInput, TOutput> extends Tool<TInput, TOutput> {
  _isRemoteSubAgent: true;
  _remoteConfig: {
    transport: RemoteAgentTransport;
    remoteAgentType: string;
    timeoutMs: number;
    streamRetries: number; // Default: 3
    streamRetryBaseMs: number; // Default: 100
  };
}

Input-to-Message Mapping

When the tool executes, the full tool input is JSON-serialized and sent as the message field in the RemoteStartRequest. The remote agent server is responsible for parsing the JSON and constructing the appropriate user message for its agent.

Runtime Behavior

All three runtimes provide first-class remote sub-agent support:

• JS Runtime — Intercepts via isRemoteSubAgentTool() for enhanced handling: stream proxying, SubSessionRef tracking with remote metadata, timeout management via AbortSignal, reconnection on resume
• Temporal Runtime — Dedicated executeRemoteSubAgentCall activity with deterministic session IDs, crash recovery via transport.getStatus(), heartbeat-based reconnection, stream proxying, and interrupt propagation
• Cloudflare Runtime — Dedicated executeRemoteSubAgentCall step with deterministic session IDs, crash recovery, stream proxying, interrupt propagation via abort-check interval, and timeout enforcement

The tool's built-in execute() method exists as a fallback but is not used by any of the pre-built runtimes.

Every runtime shares the same post-terminal fold (processRemoteSubAgentCompletion), so these behaviors are uniform:

• end.status gating — only status: 'completed' folds directly. A 'paused' or 'interrupted' end re-checks getStatus() instead of being treated as completed-with-null.
• Output re-validation — the child's terminal output is re-validated against this tool's outputSchema (validateRemoteOutput); a mismatch surfaces to the parent LLM as a structured tool error, and the parsed value (not the raw wire value) is what gets folded.
• childCustomState fold — the child's terminal customState reaches afterSubAgent as payload.childCustomState, sourced from end.state or, on recovery, transport.getSnapshot(); skipped with a warning when over maxFoldStateBytes.
• Usage fold — the child's recursive rollup reaches afterSubAgent as payload.remoteRollup (+ remoteRollupOrigin), sourced from end.usage or transport.getUsage(), and is embedded on the parent's SubAgentUsageEntry.

See the Cross-Service Remote Agents guide for the full consumer contract.

Types

All types are exported from @helix-agents/agent-server:

import type {
  AgentServerConfig,
  AgentHttpHandler,
  AgentHttpRequest,
  AgentHttpResponse,
  AgentServerOperation,
  AuthenticateHook,
  AuthenticateResult,
  ChatStreamHandler,
  ExpiredSessionCleanupDeps,
  ExpiredSessionCleanupResult,
} from '@helix-agents/agent-server';

Transport protocol types are exported from @helix-agents/core:

import type {
  RemoteStartRequest,
  RemoteResumeRequest,
  RemoteStartResponse,
  RemoteStatusResponse,
  RemoteSnapshotResponse,
  RemoteUsageResponse,
  RemoteAgentErrorResponse,
  TransportEvent,
  RemoteAgentTransport,
} from '@helix-agents/core';

See Also

• Remote Agents Guide — Concepts, patterns, and production setup
• Cross-Service Remote Agents Guide — Snapshot/usage across a service boundary, exposure mounts, reconstruction
• Remote protocol types — RemoteSnapshotResponse, RemoteUsageResponse, error codes, transport interface
• Remote Agents Example — Full working example
• Sub-Agents Guide — Local sub-agent orchestration