Server

experimental-a2/server

handler(entry)

Normalizes a handler entry so you can compose an existing handler without branching on its shape. For ordinary use, define handlers inline in createServer({ handlers }).

• Accepts a function or { handler, lane?, abortOn? } and returns the object shape: a function becomes { handler: entry }, an object is returned unchanged, and both keep their original types.
• Does not bind a contract; createServer({ handlers }) checks the match. Annotate a new handler's context, as below, or pass an already typed function.
• Spreading the result keeps lane and abortOn; set either after the spread to override it. Changing a built-in AI handler's lane can separate it from the handlers it coordinates with.
• A composition runs in one handler attempt. Work after the original resolves still runs before its returned events commit, a throw fails the composed attempt, and a retry can run both again.

import * as a2 from 'experimental-a2'
import { createServer, handler, type HandlerContext } from 'experimental-a2/server'
import { z } from 'zod'

const orders = a2.contract({
  name: 'composed-orders',
  events: {
    created: z.object({ shopId: z.string() }),
    notified: z.object({ shopId: z.string() }),
  },
})

const created = handler({
  lane: 'orders',
  handler: async (ctx: HandlerContext<typeof orders.events, 'created'>) => {
    return orders.batch({ type: 'notified', payload: ctx.event.payload })
  },
})

export const ordersServer = createServer({
  contract: orders,
  handlers: {
    created: {
      ...created,
      handler: async (ctx) => {
        const result = await created.handler(ctx)
        // Your additional work goes here:
        // await notifyShop(ctx.event.payload, { idempotencyKey: ctx.event.id })
        return result
      },
    },
  },
})

createServer(options)

createServer(options: {
  contract: Contract // the vocabulary this server implements
  store?: A2Store    // default: sqlite in dev, memory in tests, required in prod
  scheduler?: A2Scheduler
  telemetry?: A2Telemetry // optional instrumentation; see experimental-a2/otel
  presence?: { ttlMs?: number } // presence expiry policy; default 60s
  handlers?: {
    [type]:
      | Handler
      | {
          abortOn?: AbortSpec
          lane?: string | ((ctx: LaneContext) => string)
          handler: Handler
        }
  }
}): A2Server

type Handler = (
  ctx: Context,
) => Promise<void | AppendInput | readonly AppendInput[]>

Implements a contract: binds its vocabulary to storage and reactions. experimental-a2/server is the only entry point that can reach a store backend; under the browser condition its exports resolve to a loud error. See Handlers.

generate: {
  abortOn: {
    cancelled: (event, trigger, { attempt }) =>
      event.payload.of === `${trigger.id}:${attempt}`,
  },
  handler: async (ctx) => { /* ... */ },
}

• Handlers are concurrent by default. A2 starts every eligible event once its append commits.
• A handler returns nothing, one event, or an array. A2 marks the trigger processed and appends the returned batch in one atomic store operation; when the handler throws, the returned events never exist.
• Handlers are optional per event type. An event type without one settles in its append transaction with no dispatch attempt, and starts no drain or scheduler arm when the session has no older pending handler work. See Events without handlers.

The context every handler receives:

ctx.session shares the server.session(id) surface; handler-specific behavior is noted under each method. Its state(reducer) folds through the trigger's index by default (see session.state), and its append takes a name:

ctx.session.append(name: string, ...events: AppendInput[]): Promise<Event[]>

• It commits immediately, so its events may run while the handler is still active unless a lane orders them. See Append what should happen now.
• The name is the idempotency key, scoped to the trigger, for events whose id is omitted: each such id derives from the trigger id, the name, and the event's position. The same name and generated-id batch returns the existing rows on retry; a changed batch conflicts with them.
• An explicit id wins, so different triggering events can converge on one fact. Root server.session(id).append(...events) takes no name.

server.fetch(request, options?)

server.fetch(
  request: Request,
  options?: {
    authorize?: (operation: A2Operation) => boolean | Promise<boolean>
    upgradeWebSocket?: (
      attach: (socket: A2Socket) => void,
    ) => Response | Promise<Response>
    multiplexWebSocket?: boolean
  },
): Promise<Response>

type A2Operation =
  | {
      type: 'stream'
      sessionId: string
      startAfter: number
      transport: 'http' | 'websocket'
    }
  | {
      type: 'history'
      sessionId: string
      gte: number
      lte: number
      transport: 'http'
    }
  | {
      type: 'push'
      sessionId: string
      events: readonly {
        type: ContractEventType | string
        payload: unknown
        id?: string
      }[]
      presence?: {
        participant: string
        values: Readonly<Record<string, unknown>>
        seen?: number
        at?: number
      }
      transport: 'http' | 'websocket'
    }

The bound Fetch API route for this server: an unprotected route exports it directly as GET and POST. See Routes.

A contract that declares presence streams it and accepts presence pushes automatically.

• Authentication stays in the route, where headers, cookies, and framework context are available. Capture the authenticated principal in authorize.
• authorize is an ingress policy for this fetch call, not a global server policy: direct session.append, handler appends, scheduled appends, and server-side setPresence bypass it.
• By default the sessionId and index query parameters bind one session to the socket. A2 authorizes the stream before upgrading, and frames can't select another session.
• A multiplexed socket runs authorize for each session subscribe and each event or presence push. Authenticate the physical WebSocket GET before calling server.fetch, and validate Origin when browser cookies authenticate it (handshakes skip the CORS preflight). Enable multiplexing only when that request context may access every session authorize accepts.
• A denied subscribe receives unsubscribed and leaves other sessions alone. A denied event push receives a FORBIDDEN ack. Denied presence is silently dropped, because it is fire-and-forget.
• SSE and WebSocket lifetimes use the platform's ambient invocation deadline, and background work uses its ambient waitUntil. Neither is an option.

server.session(id)

server.session(id: string): Session

A handle on one instance of the contract: the unit of log ordering, lane keys, recovery, state, and live sync. Creating it does no I/O; nothing loads until you append, read, or stream. session.id is the id passed in. See Sessions.

It has every session method, plus deadLetters() and redrive(), which ctx.session doesn't.

server.drain(sessionId)

server.drain(sessionId: string): Promise<{ settled: boolean }>

Claims every eligible event and runs their handlers concurrently. settled means no actionable or live-claimed work remains. Past the invocation's deadline it claims nothing and returns settled: false. A dead-lettered event can block later work in its lane while other lanes continue, until session.redrive() resolves it. Scheduler callbacks use it; you'll rarely call it yourself. The public result stays this simple boolean.

A2Store

Custom store adapters implement these atomic drain methods:

// Custom server store adapter:
type EventCause = {
  index: number
  attempt: number
  batchSize?: number // present on named handler appends
}

type AppendEvent = {
  type: string
  payload: unknown
  id?: string
  cause?: EventCause
  lane?: string
  settled?: true // internal: no handler is registered for this event type
}

type ReturnedEvent = AppendEvent & { id: string }

type StoredEvent = Event & {
  cause: EventCause | null
  lane: string | null
  processedAt: Date | null
  processedByAttempt: number | null
  returnedEventIds: string[] | null
  firstClaimedAt: Date | null
  lastClaimedAt: Date | null
  attemptCount: number
  claimHolder: string | null
  claimExpiresAt: Date | null
  failureCount: number
  lastFailedAt: Date | null
  lastFailedAttempt: number | null
  lastError: string | null
  failedAt: Date | null
  clearIndex?: number // on claimed events, once the session has been cleared
}

type StoreAppendResult = {
  events: StoredEvent[]
  hasPending: boolean
}

type StoreClaimAvailableResult =
  | { outcome: 'claimed'; events: StoredEvent[] }
  | { outcome: 'busy'; retryAt: Date }
  | { outcome: 'settled' }

type CompleteAttemptResult =
  | { outcome: 'completed'; events: StoredEvent[] }
  | { outcome: 'superseded' }

interface A2Store {
  append(
    sessionId: string,
    events: AppendEvent[],
    options?: { originIndex?: number },
  ): Promise<StoreAppendResult>

  // optional together; required for session.clear()
  clear?(options: {
    sessionId: string
    ifLastIndex?: number
    cause?: EventCause
  }): Promise<{ cleared: boolean; index: number; clearIndex: number }>
  reclaim?(sessionId: string): Promise<{ deleted: number; done: boolean }>

  // optional together; required for session.deadLetters() and redrive()
  deadLetters?(sessionId: string): Promise<StoredEvent[]>
  redrive?(options: {
    sessionId: string
    index: number
    action: 'retry' | 'skip'
  }): Promise<{ redriven: boolean; hasPending: boolean }>

  claimAvailable(options: {
    sessionId: string
    holder: string
    ttlMs: number
    expiresAtMs?: number
    excludeIndexes?: readonly number[]
    maxFailures?: number
  }): Promise<StoreClaimAvailableResult>

  renewClaims(options: {
    sessionId: string
    holder: string
    claims: readonly { index: number; attempt: number }[]
    ttlMs: number
    expiresAtMs?: number
  }): Promise<{ renewed: number[]; superseded: number[] }>

  completeAttempt(options: {
    sessionId: string
    index: number
    attempt: number
    events: ReturnedEvent[]
  }): Promise<CompleteAttemptResult>

  failAttempt(options: {
    sessionId: string
    index: number
    attempt: number
    error: string
    maxFailures: number
    retryAfterMs?: number
  }): Promise<{
    outcome: 'failed' | 'dead_lettered' | 'superseded'
    failureCount: number
  }>
}

readHistory(sessionId, options) is an optional capability with the same exclusive afterIndex and inclusive throughIndex bounds as read(). Without it, session.history() falls back to read().

A store that supports clearing treats every event at or below the latest marker as superseded, whether or not its row still exists. Each check belongs inside the operation's existing atomic step:

• claimAvailable never claims it; lane readiness at append and completion ignores it, so a dead-lettered lane head stops blocking.
• renewClaims reports it in superseded; completeAttempt and failAttempt return superseded.
• An append whose cause names it throws SUPERSEDED_ATTEMPT; one whose originIndex is at or below the marker throws SESSION_CLEARED.
• Reads start at the marker; readState returns its index as clearIndex, and claimed events carry it as clearIndex.
• In append and completeAttempt, an event id whose row is below this session's marker is free to reuse.

Without clear and reclaim, session.clear() and session.schedule.clear() throw a TypeError before writing anything. Without deadLetters and redrive, so do session.deadLetters() and session.redrive(). All built-in stores support both.

Stored fields:

• attemptCount counts claims, including abandoned ones. failureCount counts failed attempts: caught failures, and lapsed claims a later claim took over. Ten dead-letter the event. See Guarantees.
• cause names the event and handler attempt whose ctx.session.append or return value first persisted the child; null means a top-level event.
• lane is the session-scoped serialized group, resolved before append.
• returnedEventIds makes a lost completion acknowledgment recoverable without accepting a partial child batch.
• Lifecycle timestamps are adapter clock values for their atomic operations, not exact database commit times. Built-in adapters persist all these fields inside their existing atomic operations, with no extra round trip.

A2Scheduler

Custom scheduler adapters implement two methods:

Application routes call scheduler.handler(...servers) on the configured adapter. See Schedulers.

SchedulerTask is SchedulerDrainTask | SchedulerAppendTask | SchedulerClearTask. These task types, plus ScheduledEvent, are exported from experimental-a2/server.

Drain tasks wake pending work and carry no application payload. Append tasks carry the events for session.schedule; originIndex is the scheduling handler's event, absent for root schedules. Clear tasks carry the condition for session.schedule.clear.

Deliver an authenticated, normalized append or clear task through the server helpers:

import {
  deliverSchedulerAppend,
  deliverSchedulerClear,
  type DrainableServer,
  type SchedulerAppendTask,
  type SchedulerClearTask,
} from 'experimental-a2/server'

export async function deliver(
  server: DrainableServer,
  task: SchedulerAppendTask | SchedulerClearTask,
) {
  if (task.kind === 'clear') await deliverSchedulerClear(server, task)
  else await deliverSchedulerAppend(server, task)
}

Authenticate the transport and validate the task envelope first. Each helper verifies that the task and server contract match.

• deliverSchedulerAppend revalidates the carried event inputs and uses the ordinary top-level append. The target server must have a scheduler. When the append leaves pending handler work, it resolves only after a recovery watchdog is durable, so a send failure keeps the delivery retryable. A task whose originIndex was cleared resolves without appending.
• deliverSchedulerClear runs clear({ ifLastIndex }) and resolves after its deletion finishes. A store failure rejects, so the provider redelivers and the retry finishes deleting.

Session

session.append(...events)

session.append(
  ...events: Array<{ type: string; payload: unknown; id?: string }>
): Promise<Event[]>

The only way to move a session forward. See Guarantees.

• Validates payloads against the contract's schemas before writing anything.
• A multi-event append is atomic: all-or-nothing, consecutive positions, one transaction.
• An explicit id makes it idempotent across retries; re-sending an identical batch returns the original rows.
• Events pushed through server.fetch arrive through this same append.
• Only event types with registered handlers are dispatched; the append itself settles the rest. An unhandled append still starts session healing when older handled work is pending. The store reports that pending state in the atomic append, so the decision needs no follow-up read.

session.append.dispatch(...events)

session.append.dispatch(
  ...events: Array<{ type: string; payload: unknown; id?: string }>
): Promise<Event[]>

Commits the same atomic batch, then sends pending work to the configured scheduler instead of starting an inline drain. See Dispatch in a fresh invocation.

• Awaits acceptance of an immediate scheduler message. A batch that leaves no pending handler work sends none.
• Throws before writing when the server has no scheduler.
• If the send fails, the events are already durable: give them explicit IDs and retry the same dispatch.
• It picks how the session wakes, not which worker runs: an already active drain may claim the new work first.
• Exists only on a server session's top-level append. ctx.session.append runs inside an active drain, where its immediate and returned children become eligible, so it has no dispatch.

session.schedule(name, timing, ...events)

session.schedule(
  name: string,
  timing: ScheduleTiming, // { delay: ScheduleDelay } | { at: Date }
  ...events: Array<{ type: string; payload: unknown; id?: string }>
): Promise<void>

Appends the events later through the configured scheduler, from root and handler sessions. Resolves after provider acceptance, not after append. name must be non-empty, with at least one event. ScheduleDelay, ScheduleTiming, and SessionSchedule are exported from experimental-a2/server. See Timers for stale-timer guards, provider limits, and deployment compatibility.

• The template type checks the unit. At runtime the number must be an unsigned base-10 decimal without leading zeros, finite, and above zero; 0s, signs, exponents, malformed durations, invalid dates, and both fields at once are rejected.
• The target is the earliest useful delivery time. Adapters encode it in whole-second slots; an already-due target is sent immediately.
• The task id derives from the contract, session, and name. Handler tasks add the trigger id and, once the session has been cleared, the clear, so a trigger id reused in a new lifetime schedules a new task. Omitted event ids derive from the task id and position; explicit ids win.
• Reuse a name only for identical timing and events. Provider deduplication is an optimization; stable event ids make later duplicate deliveries converge in the store.
• Payloads are snapshotted into plain JSON, then validated before sending. Encode dates as ISO strings and classes as plain data. undefined, bigint, functions, symbols, cycles, accessors, and non-plain prototypes are rejected; shared references become independent values.
• Delivery validates the snapshotted schema input through an ordinary top-level append, outside any handler's completion transaction, so a schema transform produces the stored payload from the original input. The events have cause: null and run through handlers, lanes, state, and streams normally.
• A handler task belongs to its trigger's lifetime: if the session is cleared before delivery, it is acknowledged without appending. A root task joins whichever lifetime exists at delivery, and its name identifies one task across lifetimes, so put something from the new lifetime in the name when scheduling again after a clear.
• There is no cancellation, reschedule, cron, or timer-listing API.

session.clear(options?)

session.clear(options?: { ifLastIndex?: number }): Promise<{
  cleared: boolean
  index: number
}>

Ends the session's current lifetime: one atomic step checks the condition and appends a reserved clear marker at the next index. The session id stays usable and indexes never restart. See Clearing a session.

ifLastIndex is a non-negative safe integer. The first matching rule decides, and index is always the head after the call:

• Every event at or below the marker is superseded: never dispatched again, never blocking its lane. Its running handler's appends throw SUPERSEDED_ATTEMPT and its state reads throw SESSION_CLEARED rather than fold the new lifetime. Exception: after a clear from another instance, a read bounded at the handler's own event may return that event's state from memory until its ctx.signal aborts.
• Pushes and appends after the marker start the new lifetime.
• clear() then deletes the old events, their bookkeeping and event-id records, and cached state below the marker in short batches, resolving when they're gone. Other work on the session runs between batches. There is no background sweep.
• If deletion fails, the call rejects with the session already cleared; any later clear of the session, including a retry, finishes the deletion.
• Old event ids can be used again. Presence is not cleared.
• On ctx.session, the clear is fenced like the handler's appends and ends the handler's own attempt: ctx.signal aborts with SUPERSEDED_ATTEMPT and returned events are discarded. Its deletion keeps running under the platform's waitUntil.
• All built-in stores support clearing.

session.schedule.clear(name, options)

session.schedule.clear(
  name: string,
  options: ({ delay: ScheduleDelay } | { at: Date }) & {
    ifLastIndex: number
  },
): Promise<void>

Publishes a conditional clear to the configured scheduler and resolves after provider acceptance. name is non-empty and ifLastIndex is required; timing works as in session.schedule, with the same provider limits. See Scheduling a clear.

• The task id derives from the contract, session, name, and condition, plus the trigger id and any earlier clear inside a handler. Reuse a name only for the same timing.
• Delivery runs clear({ ifLastIndex }) and acknowledges after the deletion finishes. No event is appended; a stale or duplicate delivery changes nothing. Indexes never repeat, so a condition from an older lifetime never matches a newer one.
• ifLastIndex: ctx.event.index matches only while the trigger is still the head. The handler's own appends and returned events move the head, so a handler that writes should use the index of its last ctx.session.append. Actor handlers always commit state, so they can't use the trigger's index.
• A delay short enough to deliver while the handler still runs can match before its writes land; the clear then ends the handler's attempt and discards its returned events.
• Each delivery attempt resumes the deletion. Vercel Queues redelivers until retention ends. QStash retries up to its retries budget; a deletion that outlasts them leaves the message in QStash's dead-letter queue, the rest of the old history still invisible, and any later clear of the session finishes it.

session.deadLetters()

session.deadLetters(): Promise<DeadLetter[]>

type DeadLetter = Event & {
  lane: string | null
  attemptCount: number
  failureCount: number
  lastError: string | null
  failedAt: Date
}

The session's dead-lettered events, oldest first, each with why it failed. See Resolve a dead letter.

• failureCount counts the attempts that spent its budget: caught failures, and attempts that ended before they finished. attemptCount counts every claim. lastError is the last error a handler threw, stringified, or null when every failed attempt ended before it finished.
• Events at or below the latest clear marker are never listed.
• Only on server.session(id). DeadLetter is exported from experimental-a2/server.

session.redrive(index, options)

session.redrive(
  index: number,
  options: { action: 'retry' | 'skip' },
): Promise<{ redriven: boolean }>

Resolves the dead-lettered event at index in one atomic store operation.

• index is a positive safe integer and action is required; anything else throws a TypeError before any I/O.
• redriven is false, with nothing written, when the event isn't dead-lettered: still pending, already processed, already redriven, missing, or at or below the latest clear marker. Calling it again is safe.
• Afterwards, unfinished work wakes the session as an append does: an inline drain, plus a scheduler watchdog when one is configured, joined for up to schedulerArmTimeoutMs. A failed send doesn't reject the call; it marks the a2.redrive span and leaves later appends, server.drain(), or another redrive to wake the session.
• A retried event keeps its attemptCount and last failure, so the next run's ctx.attempt keeps counting. A skipped event keeps failedAt and has no processedByAttempt.
• Only on server.session(id). SessionRedriveOptions and SessionRedriveResult are exported from experimental-a2/server.

session.history()

session.history(options?: { gte?: number; lte?: number }): Promise<Event[]>

Raw events, oldest first, without dispatch metadata. See History.

• gte and lte are inclusive, non-negative safe-integer indexes; omit one to leave that end open. lte: 0 returns []; gte > lte throws.
• After a clear, history starts after the marker; the marker itself is not returned.

session.state(reducer, options?)

session.state(
  reducer: Reducer<S>,
  options?: { through?: number | 'latest' },
): Promise<{ state: S; index: number; clearIndex?: number }>

The log folded through one committed prefix. index is its exact boundary, an append-order cursor rather than a handler-settlement marker; hand it to the client to resume there. See State.

• Omitting through reads the latest committed frontier on a root session, and through ctx.event.index on ctx.session, inclusive and stable across concurrent appends and retries. In a handler, use { through: 'latest' } only when later committed events are intentionally part of the decision.
• A numeric through is an inclusive, non-negative safe integer; through: 0 returns the reducer's initial state.
• A latest read and a following append are not atomic. See State for joins that converge.
• Never dispatches handlers or waits for pending work. Snapshot write-back uses platform waitUntil, and its failures never fail the completed read: they go to a2.snapshot telemetry or the default console sink, without snapshot contents.
• After a clear, the result includes clearIndex, the latest marker, and state folds from there; forward the whole result when hydrating a client. A positive through below clearIndex throws SESSION_CLEARED; through: 0 returns the seed without clearIndex.
• On ctx.session, every state read throws SESSION_CLEARED once the trigger has been cleared, including { through: 'latest' }. Handler history() and stream() are not fenced.
• Default and numeric handler reads retain their exact folded checkpoint until the trigger completes or dead-letters and no other unfinished handler uses it. Handler { through: 'latest' } reads and root reads retain nothing. There are no retention options. A numeric read behind the latest snapshot can use the greatest eligible retained checkpoint. See Snapshots are a cache.
• Operational snapshot-read failures keep existing A2 errors and turn others into STORE_UNAVAILABLE with the original error as cause, without falling back to individual reads or raw history. Cache misses and stateSchema rejections rebuild from the log.

session.stream(options?)

session.stream(options?: { startAfter?: number }): AsyncIterable<Event | SessionCleared>
session.stream(options: {
  startAfter?: number
  presence: true
}): AsyncIterable<Event | SessionCleared | PresencePatch | PresenceSnapshot>

A live feed of the session's events. Server-side only; server.fetch exposes it over SSE and WebSocket. Subscribing never dispatches handlers.

• startAfter is a non-negative safe integer; startAfter: 20 begins with event 21.
• A clear arrives in order as a SessionCleared marker (type: 'a2.cleared'). A stream that opens below the latest marker begins with it. A running stream delivers the log in order, so events from before the clear can arrive ahead of the marker. Reset when it arrives. Clearing never closes a stream.
• With presence: true, the feed first yields one PresenceSnapshot, { snapshot }: the current pruned map with each field's value, seen, and at. Live PresencePatch values, { participant, values, seen, at }, then interleave with events; values is typed by the contract's presence schemas, and null marks a cleared field. Over SSE both are named frames, which clients that don't know them skip.
• The return type widens only under the literal presence: true. The option exists only on sessions of contracts that declare presence; elsewhere it is a type error, not a runtime throw.

session.setPresence(patch)

session.setPresence(patch: {
  participant: string
  values: { [field]: payload | null } // partial; null clears a field
  seen?: number
  at?: number // the sender's LWW stamp, epoch ms; receipt time when absent
}): Promise<void>

Validates each field against the contract's presence schemas, then broadcasts. No log row, no dispatch, no scheduler arm; history() and state() never see it. See Presence.

• Last writer wins per participant per field, ordered by at, the sender's epoch-ms stamp; the browser client stamps its sends automatically. Without at, receipt time stands in: a stampless caller is a single writer, so receipt order is its send order.
• A bad payload throws INVALID_PAYLOAD, an unknown field throws UNKNOWN_PRESENCE_FIELD, and a non-finite or negative at throws a TypeError.
• seen is the sender's stream frontier, which view code uses to reconcile presence with the fold. See Reconciling the planes.
• Values expire by backend TTL when a participant goes silent (default 60 seconds; see presence.ttlMs). Expiry counts on the storage clock, so a hostile stamp can only vandalize its own field, which still expires on schedule.
• The patch is the push envelope's presence sibling; server.fetch's push lane forwards it whole to session.setPresence(presence).
• ctx.session.setPresence(...) is the same operation; seen defaults to the triggering event's index.

session.presence()

session.presence(): Promise<{
  [participant: string]: {
    [field: string]: { value: unknown; seen: number; at: Date }
  }
}>

The current presence map, expired values pruned: a point-in-time read for server renders. Live consumers use stream({ presence: true }). Each entry's at is the sender's stamp for when the value was set, the last-writer-wins comparator; how long a value stays painted is view logic, decided at render time.