@psci-labs/chat-runtime

Node agent runner for the Claude Agent SDK chat surface. Returns a Web-standard Request → Response handler with SSE streaming, MCP plumbing, and a pluggable persistence interface.

pnpm add @psci-labs/chat-runtime @anthropic-ai/claude-agent-sdk

@anthropic-ai/claude-agent-sdk is a peer dependency — pin the version your app wants and the runtime forwards through to it. @psci-labs/chat-protocol is a direct dependency (no need to install separately).

Persistence

import { InMemoryPersistence, type PersistenceAdapter } from '@psci-labs/chat-runtime';

const adapter: PersistenceAdapter = new InMemoryPersistence();

The interface (in src/persistence/adapter.ts) is intentionally narrow:

• saveCheckpoint({ threadId, sessionId, sdkVersion, state, metadata }) — sdkVersion is required so SDK shape drift fails at the boundary.
• loadCheckpoint(threadId) — returns the latest checkpoint for the thread, or null.
• appendMessage(threadId, sessionId, message) / listMessages(threadId, sessionId?) — replay history, optionally scoped to a single session.
• archiveSession(threadId, sessionId) — lifecycle hook; messages stay readable.

Reusable contract test

Every adapter is tested against the same suite via @psci-labs/chat-runtime/testing:

import { describe } from 'vitest';
import { runPersistenceAdapterContract } from '@psci-labs/chat-runtime/testing';
import { PostgresPersistence } from '../src/index.js';

describe('PostgresPersistence', () => {
  runPersistenceAdapterContract(() => new PostgresPersistence({ ...config }));
});

The factory must return a fresh empty adapter per call; the contract invokes it once per test case so suites stay isolated.

Auth (getUserContext)

The runtime is auth-agnostic. Apps inject a callback:

import { resolveUserContext, type GetUserContext } from '@psci-labs/chat-runtime';

const getUserContext: GetUserContext = async (req) => {
  const session = await getServerSession(req);
  return session ? { userId: session.user.id } : null;
};

Returning null is the contract for "unauthenticated" — the runtime answers 401 with { error: 'Unauthorized' }. Throwing from the callback is reserved for genuine bugs in the integration and surfaces as 500.

The shape returned is validated against the UserContext schema from @psci-labs/chat-protocol, so a misshapen context (empty userId, wrong attributes type) fails loudly at the boundary.

System prompt

Three input shapes — passed in via createAgentRunner({ systemPrompt }):

// Plain string — full custom prompt, SDK preset bypassed
systemPrompt: 'You are a billing-workflow specialist.'

// Bare preset — use the SDK's claude_code preset as-is
systemPrompt: { preset: 'claude_code' }

// Preset + per-request append (static or dynamic)
systemPrompt: {
  preset: 'claude_code',
  append: ({ userContext, threadId }) =>
    `User ${userContext.userId} on thread ${threadId}.`,
}

resolveSystemPrompt(input, ctx) translates these into the SDK's expected shape and is awaited per request, so the dynamic callback can do async work (e.g. read the user's tenant config from a DB).

Built-in MCP servers

Two SDK MCP servers ship with the runtime. Both are pluggable — apps inject the side-effect (notification delivery, user-question UI) via callbacks; the servers own the tool name, description, and Zod schema.

import { createNotificationsServer, createInteractionServer } from '@psci-labs/chat-runtime';

const notifications = createNotificationsServer({
  onNotify: async ({ type, title, body }) => {
    await pushNotificationService.send({
      /* ... */
    });
  },
});

const interaction = createInteractionServer({
  askUser: async (questions) => {
    // session manager wires this up — resolves when the user replies
    return await sessionManager.askUser(threadId, questions);
  },
});

Renderer keys (used by chat-ui to dispatch to a renderer):

• mcp__notifications__notify — type (progress/completed/error/info), title, body
• mcp__interaction__ask_user — array of questions; supports freeform / single-select / multi-select / yes-no

For unit tests, the handler is also exported as makeNotifyHandler / makeAskUserHandler so you can exercise the business logic without spinning up the SDK.

SSE encoder

import { encodeSSE } from '@psci-labs/chat-runtime';

const stream = encodeSSE(eventIterable, { signal: req.signal });
return new Response(stream, {
  headers: { 'content-type': 'text/event-stream' },
});

pull-based ReadableStream — backpressure is automatic. Custom formatter hook lets you set the SSE event: line when you want typed events. Stream cancellation calls iter.return?.() so the upstream agent can release resources promptly.

Session manager

The session manager is the runtime's brain. It owns:

• The active-sessions map keyed by threadId
• A MessagePump<UserPrompt> per session (queues follow-up messages while the agent is busy, drains as the SDK pulls)
• Lifecycle: start / continueSession / terminate / clearContext
• Persistence checkpointing at session boundaries (sdkVersion always recorded; state carries previous-session IDs and cost/turn metrics)
• respondToTool(threadId, answer) to resolve the mcp__interaction__ask_user waiter

The SDK call is injected via a runAgent: (opts) => AsyncIterable<SDKEvent> factory, so tests drive the manager with canned event sequences and the Phase 2E runner factory binds the real query() call (with per-session MCP wiring).

import { SessionManager, InMemoryPersistence } from '@psci-labs/chat-runtime';
import { query } from '@anthropic-ai/claude-agent-sdk';

const mgr = new SessionManager({
  persistence: new InMemoryPersistence(),
  sdkVersion: '0.1.77',
  runAgent: ({ prompts, signal, resumeSessionId }) =>
    query({ prompt: prompts, options: { signal, resume: resumeSessionId } }),
});

const events = await mgr.start({ threadId, userMessage: { text: 'Hello' } });
// pipe through encodeSSE → return as Response

HTTP handlers

Four standalone handlers, one per route. Each takes (req, ctx) where ctx carries the resolved userContext, threadId, sessionManager, and persistence. The router (Phase 2E) is responsible for parsing the URL and resolving auth before dispatching.

| Handler | Route | Behavior | | --------------- | -------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | handleStream | POST /threads/:id/stream | Body { message: { text } }. Returns text/event-stream (status 200). When the thread is mid-turn, the stream opens immediately and emits queue.entered while it waits its turn — there is no 409. 400 invalid body, 429 when the per-thread arbitration queue is full (maxQueueDepth exceeded). Forwards req.signal so both an active turn (abort the agent) and a queued turn (silently remove the slot) react. | | handleHistory | GET /threads/:id/history | Optional ?sessionId. Returns { messages: StoredMessage[] }. | | handleCancel | POST /threads/:id/cancel | Idempotent. terminate(threadId) → 200. | | handleClear | POST /threads/:id/clear | Idempotent. clearContext(threadId) → 200. The current session is archived; its sessionId is queued for the next session's previousSessionIds. |

Status codes are surfaced explicitly so the UI can react: 429 says "the per-thread queue is full, back off"; 400 says "your client sent something I can't parse." Auth-derived 401s come from the router, not the handlers.

Multi-user threads + arbitration

A single thread can receive messages from multiple users (think a sandbox shared between a broker and an inspector reviewing the same engagement). Rather than rejecting the second message with 409 Conflict, the runtime serializes turns through a per-thread FIFO queue (ThreadSerializer).

What this means for callers:

• POST /threads/:id/stream always returns 200 with an SSE body when the request is well-formed, even if another turn is active.
• The SSE stream emits these runtime events alongside the underlying SDKEvent envelopes:
  • { type: 'queue.entered', position, threadId } — your turn is waiting. position is the 1-based depth at enqueue time (1 = next in line).
  • { type: 'queue.advanced', position, threadId } — reserved for future progress signals; not currently emitted in v1 (positions don't re-advertise as predecessors finish — clients should treat turn.starting as the cue that the wait is over).
  • { type: 'turn.starting', threadId } — your turn has begun; the next events are the agent's stream.
  • { type: 'turn.rejected', reason: 'timeout', threadId } — your wait exceeded queueTimeoutMs. The SSE stream closes immediately after.
• HTTP 429 is returned synchronously (no SSE opened) when the queue is already full (maxQueueDepth waiting turns). Body: { "error": "queue_full" }.
• A client disconnect mid-wait (the Request's AbortSignal fires) removes the slot from the queue silently.

Defaults are configurable on createAgentRunner({ arbitration }):

createAgentRunner({
  // ...other options
  arbitration: {
    maxQueueDepth: 3, // default
    queueTimeoutMs: 600_000, // default — 10 minutes
  },
});

userId metadata on persisted messages

StoredMessage now carries an optional userId?: string. The runtime stamps every persisted user message with userContext.userId from the session that produced it; assistant messages do not carry a userId.

This is metadata only in v1 — the originating user id is never injected into the model prompt. Adapter implementers should treat userId as a write-through field (the in-memory, filesystem, and postgres adapters all do).

Putting it all together: createAgentRunner

// app/api/chat/[...path]/route.ts (Next.js App Router)
import { createAgentRunner } from '@psci-labs/chat-runtime';
import { getServerSession } from 'next-auth';

const runner = createAgentRunner({
  apiKey: process.env.ANTHROPIC_API_KEY,
  model: 'claude-sonnet-4-6',

  getUserContext: async (req) => {
    const session = await getServerSession();
    return session ? { userId: session.user.id } : null;
  },

  systemPrompt: { preset: 'claude_code', append: 'Use only SharePoint MCP tools.' },
  allowedTools: ['Read', 'Grep', 'Glob', 'WebSearch'],

  mcpServers: { sharepoint: sharepointMcpServer },

  onNotify: async ({ type, title, body, userContext, threadId }) => {
    await pushService.send(userContext.userId, { type, title, body, threadId });
  },
  onAskUser: async ({ questions, threadId }) => {
    return await ui.deliverAndAwait(threadId, questions);
  },
});

export const { GET, POST } = runner.toNextHandlers();

The factory:

• Defaults persistence to InMemoryPersistence if omitted
• Pins sdkVersion to the bundled SDK release (overridable via sdkVersion)
• Defaults disallowedTools to ['AskUserQuestion'] so the SDK's built-in AskUser stops shadowing the mcp__interaction__ask_user MCP renderer. Pass an explicit empty array (disallowedTools: []) to opt out and allow the SDK's version.
• Always registers the interaction MCP server. If you don't supply an onAskUser callback, the runtime supplies a default that resolves via the session manager's pendingToolResponse slot — wired to POST /threads/:id/respond. Apps with custom AskUser delivery (Slack-mediated, server-side automation, etc.) override the default by supplying their own onAskUser.
• Builds the per-session notifications MCP server only when onNotify is supplied
• Exposes sessionManager and persistence on the returned AgentRunner so apps can read history or call respondToTool outside the HTTP layer
• Accepts a runAgent override for tests and the future opencode/pi adapter

runner.handle(req) is also a Web-standard handler — works under any framework that exposes Request/Response (Hono, Fastify with @fastify/web-fetch, plain Node 22+ HTTP server, etc.).

Resuming sessions

To continue a thread across runner-process boundaries (e.g. Vercel function invocations, sandbox sleep/restore), the host app threads a resumeSessionId through POST /threads/:id/stream:

await fetch(`/api/chat/threads/${threadId}/stream`, {
  method: 'POST',
  headers: { 'content-type': 'application/json' },
  body: JSON.stringify({
    message: { text: userText },
    resumeSessionId: priorSessionId, // captured from a prior turn's stream events or history
  }),
});

The runner threads resumeSessionId into the Agent SDK's query({ resume }) call. If the SDK's local session store is warm (jsonl present on disk for this sessionId), the SDK rehydrates prior tool calls and assistant messages natively. If the session store is cold (fresh process, no jsonl), the runtime replays prior StoredMessage[] from the persistence adapter as a system-prompt summary — see Restore or replay below.

The host obtains the prior sessionId from:

• the system.init.session_id SSE event on the previous turn, or
• the sessionId field on entries returned by GET /threads/:id/history.

The runtime auto-persists every assistant message and tool-result user message into the configured PersistenceAdapter, stamped with the sessionId that produced it.

Restore or replay

When resumeSessionId arrives on POST /threads/:id/stream, the runtime asks the SessionResolver (an fs probe under the hood) which restoration path to take. The decision is emitted as a session.resolved event — the first SSE chunk every stream produces — so host telemetry can see at a glance what happened:

{
  type: 'session.resolved',
  path: 'warm' | 'cold' | 'fresh',
  sessionId: string | null, // the requested resume id, or null on fresh
  messageCount: number,     // # of stored messages the cold path replayed
}

Path semantics:

• warm — <sdkSessionRoot>/<sessionId>.jsonl exists. The runtime passes resumeSessionId straight through to query({ resume }); the SDK rehydrates from disk. No replay overhead.
• cold — the jsonl is missing (fresh process, evicted volume snapshot, sandbox restart) but the adapter has StoredMessage[] for the thread. The runtime renders the stored history as a compact User: … \nAssistant: … digest and appends it to the system prompt via the SDK's systemPrompt.append. resumeSessionId is not forwarded to the SDK (it would throw — the session doesn't exist there); a fresh query() is started with the seeded summary.
• fresh — no resumeSessionId was supplied, or cold path with zero stored messages (orphan thread). A console.warn fires on the orphan downgrade so the bookkeeping mismatch is visible in logs.

Why a system-prompt summary and not seeded input? The Agent SDK's query() prompt parameter is typed as string | AsyncIterable<SDKUserMessage> — assistant-role messages cannot be seeded through the input iterator. The summary digest captures the prior conversation in the way the SDK does support: text on the system prompt.

Overriding the SDK session root

The probe looks at <sdkSessionRoot>/<sessionId>.jsonl. By default this mirrors the Agent SDK's on-disk convention:

<CLAUDE_CONFIG_DIR or ~/.claude>/projects/<cwd-slug>/

where cwd-slug = cwd.replace(/[^a-zA-Z0-9]/g, '-'). If your host relocates the SDK store (sandbox volume, alt config dir), override on the runner factory:

createAgentRunner({
  // ...other options
  sdkSessionRoot: '/var/lib/sandbox/agent-sessions',
});

Without the override, the probe will always report cold and the runtime will replay-via-summary even when the SDK could have rehydrated natively.

Hosting in a Node process (Bun, in-sandbox, plain Node)

For non-Next.js consumers — e.g. a long-lived in-sandbox process behind a reverse proxy — boot a Node http server directly:

import { createAgentRunner } from '@psci-labs/chat-runtime';

const runner = createAgentRunner({
  getUserContext: () => ({ userId: process.env.USER_ID! }),
  // ...other options
});

const server = runner.toNodeServer();
server.listen(Number(process.env.PORT ?? 8787), () => {
  // eslint-disable-next-line no-console
  console.log(`chat-runtime listening on :${(server.address() as { port: number }).port}`);
});

For Bun, Hono, or other web-standard hosts, use runner.fetchHandler directly:

Bun.serve({ fetch: runner.fetchHandler });