Durable Session recipes

Process-local start, Flow Sessions, Signal subscriptions, PostgreSQL and Convex durability, reconnect, recovery, testing, and Devtools.

These recipes use the current public API only. Replace nativeModel(...) with your adapter's language-model constructor.

Process-local start

Use memory Runtime storage for local demos and unit tests. Still configure a Thread RecordStore; Sessions register a real Thread owner.

import {
  config,
  createWorkHost,
  prompt,
  session,
} from "@use-crux/core";
import { agent } from "@use-crux/core/agent";
import {
  createRuntimeProgram,
  createRuntimeWorker,
  inMemoryRuntimeStore,
  node,
} from "@use-crux/core/runtime";
import { inMemoryRecordStore } from "@use-crux/core/storage";
import { aiSdk } from "@use-crux/ai";
import { z } from "zod";

const model = aiSdk(nativeModel("nebula-text-v2"));
export const support = agent({
  id: "support",
  model,
  prompt: prompt({
    input: z.object({ message: z.string() }),
    output: z.object({ reply: z.string() }),
    prompt: ({ input }) => input.message,
  }),
});

const store = inMemoryRuntimeStore();
const records = inMemoryRecordStore();
config({ storage: { records } });

const program = createRuntimeProgram({
  targets: [support],
  generationModels: [model],
  transports: [],
});

const host = createWorkHost({
  runtime: node({ store, namespace: "local", autoStartMaintenance: false }),
  program,
});
const worker = createRuntimeWorker({
  runtime: node({ store, namespace: "local", autoStartMaintenance: false }),
  program,
  pollIntervalMs: 5,
});

const conversation = await host.run(() =>
  session(support, { key: "demo-1" }),
);
const turn = await conversation.send({ message: "Hello" });
const reply = await turn.result();

Stop the worker and dispose the host in finally blocks during tests.

Flow Session with Signal subscription

import { createWorkHost, session } from "@use-crux/core";
import { flow } from "@use-crux/core/flow";
import { signal } from "@use-crux/core/signal";
import { createRuntimeProgram, node } from "@use-crux/core/runtime";
import { z } from "zod";

export const orderPaid = signal({
  id: "order.paid",
  schema: z.object({ orderId: z.string() }),
});

export const checkout = flow(
  "checkout",
  async (scope, input: { orderId: string }) => {
    // Flow may wait for a Signal; Session subscription gates Session-owned waiters.
    return { ok: true as const, orderId: input.orderId };
  },
);

const program = createRuntimeProgram({
  targets: [checkout],
  generationModels: [],
  transports: [],
});
const host = createWorkHost({
  runtime: node({ namespace: "orders" }),
  program,
});

const order = await host.run(() => session(checkout, { key: "order-42" }));
await order.subscribe(orderPaid.when({ orderId: "order-42" }));
const turn = await order.send({ orderId: "order-42" });
const result = await turn.result();

List and deactivate with order.subscriptions() and subscription.unsubscribe(). Close drains represented work and deactivates subscriptions at the barrier: await order.close().

Lifecycle and stream reconnect

const conversation = await host.run(() =>
  session(support, { key: "customer-42" }),
);

let lastCursor: string | undefined;
for await (const event of conversation.stream()) {
  lastCursor = event.cursor;
  if (event.type === "ingress.accepted") {
    // safe input id + cursor only
  }
}

// later reconnect
for await (const event of conversation.stream({ after: lastCursor })) {
  // resumes strictly after lastCursor, or snapshot + earliest retained if expired
}

const child = await conversation.fork();
await conversation.close();
await conversation.delete(); // only after close/kill

PostgreSQL durable deployment

Durable Agent Sessions need both stores against the same database:

Runtime storage (postgres() from @use-crux/postgres/runtime) for Session control, ingress, activation, checkpoints, Work, and wakes.
Thread RecordStore (postgresRecordStore() from @use-crux/postgres) for the Session-owned Thread head and linearizable owner registration.

Configuring only the Node durable Runtime guide is not enough: without the RecordStore, owner registration and Thread publication cannot commit.

crux.config.ts

import { config } from "@use-crux/core";
import { node } from "@use-crux/core/runtime";
import { postgres } from "@use-crux/postgres/runtime";
import { postgresRecordStore } from "@use-crux/postgres";

const records = postgresRecordStore(); // DATABASE_URL by default

export default config({
  storage: { records },
  runtime: node({
    store: postgres(),
    namespace: "production",
  }),
});

Apply both setups, generate the program, then run the worker:

crux setup --apply
crux runtime generate
crux runtime worker

Application process:

import { createWorkHost, getSession, session } from "@use-crux/core";
import { node } from "@use-crux/core/runtime";
import { postgres } from "@use-crux/postgres/runtime";
import { postgresRecordStore } from "@use-crux/postgres";
import { config } from "@use-crux/core";
import { runtimeProgram } from "./.crux/generated/runtime/program";
import { support } from "./src/agents/support";

config({ storage: { records: postgresRecordStore() } });

const host = createWorkHost({
  runtime: node({ store: postgres(), namespace: "production" }),
  program: runtimeProgram,
});

export async function openSupport(customerId: string) {
  return host.run(() => session(support, { key: `customer:${customerId}` }));
}

export async function reconnectSupport(customerId: string) {
  return host.run(() => getSession(support, `customer:${customerId}`));
}

Rules:

Run one execution worker per PostgreSQL store + Runtime namespace.
Package the same generated program with app and worker images.
Authenticate before using customer keys; the key is not an auth token.
Call records.setup.apply() and Runtime store.setup.apply() (or crux setup --apply) before accepting traffic.

Convex durable deployment

Convex owns Runtime state, component records, and the scheduler. Sessions use the Convex Runtime store and Convex RecordStore through the same component.

crux.config.ts

import { config } from "@use-crux/core";
import { convex } from "@use-crux/convex/runtime";

export default config({
  runtime: convex(),
});

Generate Convex runtime targets so exported Agents appear in the program and target executor. App code opens Sessions inside a Convex-bound host boundary (createCruxConvex(...).run(ctx, target, fn) or the equivalent Work host for your deployment), not from an unbound Next.js process configured with runtime: convex().

// Inside a Convex action / host-bound boundary
const conversation = await host.run(() =>
  session(support, { key: `customer:${customerId}` }),
);
const turn = await conversation.send({ message });
return { sessionId: conversation.id, inputId: turn.id, cursor: turn.cursor };

Use Convex setup/deploy for component tables. Keep target work under the lease TTL or raise leaseTtlMs. See the Convex Runtime guide.

Reconnect and join a result

Persist the business key (and optionally Session / input ids for support tools). Reopen later without creating a second Session:

const conversation = await host.run(() =>
  getSession(support, "customer-42"),
);
const status = await conversation.status();
const history = await conversation.thread.read();

// If you retained a turn handle from the accepting request:
const output = await previousTurn.result();

result() may be called immediately after acceptance. It waits while the worker runs, then returns the exact retained Agent output. After restart, reconstruct the host against the same store/namespace and call getSession again; completed turns remain joinable through retained Work identity.

Recovery and inspection

const inspection = await conversation.inspect();

if (inspection.recovery) {
  // code: SESSION_TURN_RESULT_ARTIFACT_UNAVAILABLE
  console.error(inspection.recovery.nextStep);
}

for (const input of inspection.inputs) {
  // id, cursor, state, workId?, checkpointPrepared, delivery?
}

Recovery replays journaled preparation and Thread basis. It does not re-run provider callbacks or republish owner-Thread messages. Use inspection and Devtools when:

pending inputs or work do not drain;
a turn is blocked after a worker crash;
you need Thread revision / checkpoint coverage without reading payloads.

Testing and faults

Drive the worker yourself. Prefer the shared conformance factory for adapter work:

import { runSessionConformanceTests } from "@use-crux/core/runtime/testing";

runSessionConformanceTests({
  name: "memory",
  createHarness: async (law) => createMemorySessionHarness(law),
});

Harness seams may inject only:

isolated store / records setup;
independent host and worker reconstruction;
deterministic pending-Work attempt control;
real after-checkpoint / after-thread-publication fault boundaries;
owner / receipt observations.

Do not use sleeps, synthetic success labels, or a second Session worker.

Devtools and observability

Surface	What you see	What you never see
Index Catalog	Authored `session` / `getSession` calls, Agent/Flow target, literal/dynamic key, identity, observed subscribe/stream/stats/lifecycle usage, Signal subscription lineage	Runtime key hashes as private lookup material; match payloads
Runs / `session.turn`	State, cursors, Work lineage, Thread revision, fork lineage, active subscriptions, checkpoint/recovery, Work + ingress stats	Prompts, inputs, outputs, reasoning, Tool args, credentials
Runtime Bridge `SessionRuntimeReadModel`	Closed JSON-safe projection for operators (same ports/ledger as Runtime)	Sealed request ids, provider-native objects

Evidence opens only when an observability sink is active. Production OTel and Devtools reuse the same privacy boundary; there is no alternate Session telemetry endpoint or payload viewer.

Durable Session recipes

On this page