Server

experimental-a2/server

consumer(options)

consumer(options: {
  contract: Contract // the contract whose events it reads
  name: string // its identity, unique per server
  types?: readonly EventType[] // default: every type
  lane?: string | ((ctx: LaneContext) => string | null) // default: concurrent
  backfill?: boolean // default false; needs a string lane, and no abortOn
  abortOn?: AbortSpec
}): { run(fn: (ctx: Context) => Promise<void | AppendInput | readonly AppendInput[]>): Consumer }

Defines a named reader of every session's log. .run() takes the function and returns the Consumer you pass to createServer({ consumers }) or session.consume(). In consumers, the same options without contract, and run, define one inline on the server's contract. See Consumers.

// consumer options: a cancel that names one recovered run
abortOn: {
  cancelled: (event, trigger, { attempt }) =>
    event.payload.of === `${trigger.id}:${attempt}`,
},

• run receives the context below, with ctx.attempt counting claims of this event for this consumer. Returned events commit with the cursor's advance; ctx.session.append commits immediately. Both derive ids of events without an id from the trigger and the consumer's name.
• An invalid name, an empty or unknown type list, or backfill with a non-string lane or with abortOn throws a TypeError before any I/O. createServer throws on two consumers with one name or a consumer of another contract.
• Consumer, ConsumerOptions, ConsumerLane, ConsumerResult, and AbortSpec are exported from experimental-a2/server.

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
  consumers?: readonly (Consumer | InlineConsumer)[] // see consumer(options)
}): A2Server

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 Consumers. The options type is ServerOptions, exported from experimental-a2/server.

• A2 starts every eligible event once its append commits. A consumer without a lane runs its events concurrently.
• run returns nothing, one event, or an array. A2 moves the consumer's cursor past the trigger and appends the returned batch in one atomic store operation; when run throws, the returned events never exist.
• An event no consumer takes starts no work: no dispatch attempt, and no drain or scheduler arm when the session has no older pending work. See Which events it takes.

The context every consumer's run receives, ConsumerContext:

ConsumerContext, ConsumerAppend, and LaneContext are exported from experimental-a2/server. ctx.session shares the server.session(id) surface; consumer-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 run 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 consumer's name, the append's 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, consumer 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 for each consumer and runs them 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 keep the log and one cursor per (session, consumer, lane). These are the dispatch methods:

// Custom server store adapter:
type EventCause = {
  consumer: string
  lane: string | null // null: a concurrent consumer's cursor of one event
  index: number
  attempt: number
  batchSize?: number // present on named consumer appends
}

type EventTarget = { consumer: string; lane: string | null }

type AppendEvent = {
  type: string
  payload: unknown
  id?: string
  cause?: EventCause
  targets?: readonly EventTarget[] // lane-function and concurrent consumers
}

type ReturnedEvent = AppendEvent & { id: string }

type StoredEvent = Event & { cause: EventCause | null }

type HeadRef = { consumer: string; lane: string | null; index: number }

type HeadDispatch = {
  consumer: string
  lane: 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
}

type HeadEvent = StoredEvent &
  HeadDispatch & {
    clearIndex?: number // on claimed heads, once the session has been cleared
  }

type StoredCursor = HeadDispatch & { position: number; head: number | null }

type StoreConsumer = {
  name: string
  // a consumer with one string lane: its cursor reads the log by type
  log?: { lane: string; types: readonly string[] | null; backfill: boolean }
}

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

type StoreClaimAvailableResult =
  | { outcome: 'claimed'; events: HeadEvent[]; head?: number } // head: the session's head when the claim was made
  | { outcome: 'busy'; retryAt: Date }
  | { outcome: 'settled' }

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

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

  cursors(sessionId: string): Promise<StoredCursor[]>

  // for session.consume(): create a missing log cursor (at the clear marker
  // with backfill, else at the head), give an idle one its next event,
  // leave one with a head alone; returns it and the session's head
  wakeLog(options: {
    sessionId: string
    consumer: StoreConsumer & { log: NonNullable<StoreConsumer['log']> }
  }): Promise<{ cursor: StoredCursor; head: number }>

  // 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<HeadEvent[]>
  redrive?(options: {
    sessionId: string
    consumer: string
    index: number
    action: 'retry' | 'skip'
    consumers?: readonly StoreConsumer[]
  }): Promise<{ redriven: boolean; hasPending: boolean }>

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

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

  completeAttempt(options: {
    sessionId: string
    consumer: string
    lane: string | null
    index: number
    attempt: number
    events: ReturnedEvent[]
    consumers?: readonly StoreConsumer[]
  }): Promise<CompleteAttemptResult>

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

A cursor's head is the next event it runs: the lowest unfinished event of its lane, the event itself for a concurrent cursor, or for a log cursor the next event after its position whose type it takes. The clear marker is never a head. A head is eligible when no live claim or retry hold owns it, it is not dead-lettered, and it lies above the latest clear marker. Moving a cursor past its head resets the head's attempt and failure fields and releases its snapshot pins. consumers names the consumers the caller serves: work for any other consumer is neither pending nor claimed. Omitted, every cursor counts.

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, so a dead-lettered lane head stops blocking; a new event in that lane becomes its head.
• 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 heads 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 of the current head, including abandoned ones. failureCount counts its failed attempts: caught failures, and lapsed claims a later claim took over. Ten dead-letter the head. Both start over when the cursor moves on. See Guarantees.
• cause names the consumer, lane, event, and attempt whose ctx.session.append or return value first persisted the child; null means a top-level event.
• lane is the cursor's lane, resolved before append; null is a concurrent cursor of one event.
• A completion's recorded child ids make a lost 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 consumer's trigger, 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 consumer 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. Ids are unique per session: the same id in another session is another event.
• Events pushed through server.fetch arrive through this same append.
• Only events some consumer takes are dispatched; the append itself settles the rest. An append no consumer takes still starts session healing when older 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 consumer 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.append.ifLastIndex(index, ...events)

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

Commits the same atomic batch as session.append, only while index is still the session's head: the index of its last event. See Appending from what you read.

• The store checks the head in the same atomic step as the write, so of two writers that read the same head, at most one lands.
• When the head is anything else, nothing is written and nothing runs: it throws an A2Error with code LAST_INDEX_MISMATCH, and details is { index }, the head the store found.
• A batch whose ids are all in the session already returns its rows, whatever the head is now. Give each event an id unique to this writer's decision and stable across its retries, such as a request id the client generates. Ids built from index alone are shared by every writer that read it, so they fit only when each of them would append the same events.
• Ids that conflict with committed events, in a batch that mixes them with fresh ones or doesn't match the rows they name, throw LAST_INDEX_MISMATCH once the head has moved past index, as when another writer's batch took them first. While the head is still index, they throw PARTIAL_DUPLICATE_BATCH.
• A clear's marker is the head after it, so an index read before a clear never matches after it.
• index is a non-negative safe integer, and the batch has at least one event; otherwise it throws a TypeError before writing.
• Exists only on a server session's top-level append, like 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 consumer 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. A consumer's tasks add the trigger id, the consumer's name, 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 consumer's completion transaction, so a schema transform produces the stored payload from the original input. The events have cause: null and run through consumers, lanes, state, and streams normally.
• A consumer's 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. A run still working on one has its 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 run'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 run's appends and ends the run'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, the consumer's name, and any earlier clear inside a consumer. 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 consumer's own appends and returned events move the head, so a consumer 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 run still works can match before its writes land; the clear then ends that 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 & {
  consumer: string
  lane: string | null
  attemptCount: number
  failureCount: number
  lastError: string | null
  failedAt: Date
}

The session's dead letters in event order, then by consumer, each with why it failed. One event can be dead for one consumer and done for another. 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 run 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(target, options)

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

Resolves the consumer's dead-lettered event at index in one atomic store operation. Every dead letter names its consumer, A2's own included.

• consumer is a consumer's name, 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 head keeps its attemptCount and last failure, so the next run's ctx.attempt keeps counting. A skip moves the consumer's cursor on, and its next head starts with fresh counts.
• Only on server.session(id). SessionRedriveOptions and SessionRedriveResult are exported from experimental-a2/server.

session.consume(consumer, options?)

session.consume(
  consumer: Consumer,
  options?: { through?: number; signal?: AbortSignal },
): Promise<{ index: number }>

Runs a consumer with a string lane on this session, here, until it has handled every event it takes through through: by default the session's head when the call starts, and never more than it. Returns the index it has handled every event through, at least through. See Run a consumer now.

• The consumer's cursor is created on the first call when it's missing: at the start of the log with backfill, otherwise at the head. An idle cursor also finds events appended by deployments that don't run the consumer. A session with nothing after its latest clear, or never written, gets no cursor: the call writes nothing.
• The server's own consumer shares its cursor with background drains: a head another invocation holds, or one waiting out a failure's backoff, is waited for. Nothing past through runs here; it carries on in the background. A different definition under one of the server's consumer names throws a TypeError.
• A consumer the server doesn't run is claimed only by consume, and keeps its place between calls.
• A dead-lettered head throws CONSUMER_DEAD_LETTERED. signal stops the wait and throws its reason, even while an event runs here; that run finishes on its own.
• ctx.session.consume of a consumer that is waiting on the running one throws a TypeError, unless that consumer has handled everything through through already.
• A consumer without a string lane, one of another contract, a through that isn't a non-negative safe integer, or ctx.session.consume of the running consumer itself throws a TypeError before any I/O.
• On server.session(id) and ctx.session. SessionConsumeOptions and SessionConsumeResult 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 consumer-progress 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 consumer, 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 read and a following append are not atomic. To append only if nothing has landed since the read, pass its index to session.append.ifLastIndex. In a consumer, see State for joins that converge.
• Never runs consumers 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' }. Its history() and stream() are not fenced.
• Default and numeric reads on ctx.session retain their exact folded checkpoint until the trigger completes or dead-letters and no other unfinished run uses it. Its { 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 runs consumers.

• 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.
• null clears a field. The store keeps the clear's at until the field's TTL runs out, so an older value that reaches it later, such as one sent through another server instance, still loses.
• 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 views 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.