Resume a stream reliably

Survive disconnects and duplicates without losing or double-processing events.

Session streams are long-lived and delivery is at-least-once. A correct consumer needs three things: a durable cursor, deduplication by event ID, and settlement read from the log rather than from the connection.

The three rules

Order by seq

seq is the per-Session sequence. It is the cursor you persist and the order you process in.

Skip what you have

Delivery is at-least-once. In one Session stream the cursor filters replays; across unordered consumers, deduplicate on the globally unique id.

Settle on events

A closed connection is not a finished workflow. Read both the status event and its stop_reason before deciding to stop following.

Connecting and resuming

CHECKFU_API_KEY and CHECKFU_WORKSPACE_ID come from Get access; CHECKFU_SESSION_ID is the sess_… you are following. The TypeScript tab uses the TypeScript SDK with this client:

import { Checkfu } from "@checkfu/sdk"

const checkfu = new Checkfu({
  apiKey: process.env.CHECKFU_API_KEY,
})

On first connection the stream replays every persisted event, then stays live. On reconnect, send the last sequence you fully processed in Last-Event-ID:

cURL

curl --no-buffer \
  "https://api.checkfu.com/v1/sessions/$CHECKFU_SESSION_ID/events/stream" \
  --header "Authorization: Bearer $CHECKFU_API_KEY" \
  --header "Checkfu-Version: 2026-08-24" \
  --header "Last-Event-ID: 12"

Each SSE frame carries seq in id:, the event type in event:, and the full envelope in data:.

const TERMINAL_SESSION_EVENTS = new Set([
  "session.status_completed",
  "session.status_failed",
  "session.status_canceled",
])

const stopsWorkflowFollow = (event: {
  type: string
  payload?: { stop_reason?: string }
}) =>
  TERMINAL_SESSION_EVENTS.has(event.type) ||
  (event.type === "session.status_idle" &&
    event.payload?.stop_reason !== "requires_action")

const MAX_RETRY_AFTER_MS = 30_000

const retryAfterMs = (header: string | null): number | undefined => {
  if (header === null) return undefined
  const seconds = Number(header)
  const delay = Number.isFinite(seconds)
    ? seconds * 1000
    : Date.parse(header) - Date.now()
  if (!Number.isFinite(delay) || delay < 0) return undefined
  return Math.min(delay, MAX_RETRY_AFTER_MS)
}

class StreamRequestError extends Error {
  constructor(
    message: string,
    readonly retryable: boolean,
    readonly retryDelayMs?: number,
  ) {
    super(message)
  }
}

const isRetryableHttpStatus = (status: number) =>
  status === 429 || status === 503 || status >= 500

const openStream = async (
  sessionId: string,
  cursor: number,
): Promise<ReadableStream<Uint8Array>> => {
  const response = await fetch(
    `https://api.checkfu.com/v1/sessions/${sessionId}/events/stream`,
    {
      headers: {
        Authorization: `Bearer ${process.env.CHECKFU_API_KEY}`,
        "Checkfu-Version": "2026-08-24",
        "Last-Event-ID": String(cursor),
      },
    },
  )
  if (!response.ok) {
    const retryable = isRetryableHttpStatus(response.status)
    const delay =
      response.status === 429 || response.status === 503
        ? retryAfterMs(response.headers.get("Retry-After"))
        : undefined
    throw new StreamRequestError(
      `stream request failed with HTTP ${response.status}`,
      retryable,
      delay,
    )
  }
  if (response.body === null) {
    throw new StreamRequestError("stream response had no body", true)
  }
  return response.body
}

/** Parse an SSE body into event envelopes. Frames are separated by a blank line. */
async function* readFrames(body: ReadableStream<Uint8Array>) {
  const reader = body.pipeThrough(new TextDecoderStream()).getReader()
  let buffer = ""

  while (true) {
    const { done, value } = await reader.read()
    if (done) return
    buffer += value

    let boundary: number
    while ((boundary = buffer.indexOf("\n\n")) !== -1) {
      const frame = buffer.slice(0, boundary)
      buffer = buffer.slice(boundary + 2)

      const data = frame
        .split("\n")
        .filter((line) => line.startsWith("data:"))
        .map((line) => line.slice("data:".length).trim())
        .join("\n")

      if (data) yield JSON.parse(data)
    }
  }
}

export const follow = async (sessionId: string) => {
  let cursor = (await loadCursor(sessionId)) ?? 0
  let settled = false

  while (!settled) {
    try {
      const body = await openStream(sessionId, cursor)
      for await (const event of readFrames(body)) {
        // Replay after a reconnect re-sends events you already handled.
        // seq is monotonic within a Session, so the cursor is the whole filter.
        if (event.seq <= cursor) continue

        await handle(event)

        // Advance only after the side effect is durable, so a crash replays
        // the event rather than skipping it.
        cursor = event.seq
        await saveCursor(sessionId, cursor)

        if (stopsWorkflowFollow(event)) {
          settled = true
          break
        }
      }
      if (!settled) throw new StreamRequestError("stream ended", true)
    } catch (error) {
      if (error instanceof StreamRequestError && !error.retryable) throw error
      await backoff(error instanceof StreamRequestError ? error.retryDelayMs : undefined)
    }
  }
}

Two details carry the correctness:

• Persist the cursor after the side effect, never before. At-least-once delivery makes a replayed event harmless; a skipped one is lost permanently.
• Deduplicate on seq, not on a growing set of IDs. Within one Session stream seq is monotonic, so the cursor you already persist is a complete and bounded dedupe filter. Keep an ID set only where ordering is not guaranteed, such as across webhook deliveries, where id is the right key.

Live previews do not move the cursor

The stream is durable-only unless you repeat the event_deltas query parameter. An opted-in stream may also receive event_start and event_delta frames for live agent.message output, plus event_start for agent.thinking. These frames have no SSE id: line and no seq, are never replayed, and may be dropped. Do not save a cursor for them.

If the corresponding observation commits, the final persisted agent.message or agent.thinking event uses the preview’s event ID. Reconcile the temporary rendering to that event, perform durable side effects from the final envelope, and advance Last-Event-ID only from its numeric seq. A failed, interrupted, or lease-lost turn can leave no matching final event, so discard unmatched previews when the turn settles. A non-SDK client may also clear them at an observable HTTP reconnect; the SDK reconnects transparently, so recreating its iterator is the observable reconnect boundary. The TypeScript SDK enforces cursor neutrality automatically when you pass eventDeltas to events.stream.

Settlement

Use both the event type and stop_reason to decide whether your workflow follower can return:

Event	Stop following?	Meaning
session.status_idle with stop_reason: "requires_action"	No	One compute-closed Run parked and the admitted continuation is being awaited or started
Any other session.status_idle	Yes	The turn ended and the Session can be driven again
session.status_completed	Yes	Explicit successful terminal outcome
session.status_failed	Yes	Terminal failure
session.status_canceled	Yes	Explicitly canceled

session.status_idle settles one Run boundary; it does not end the Session. The requires_action variant is deliberately intermediate: stopping there can hide the continuation Run that follows an approval, custom-tool result, or outcome evaluation. session.status_waiting is not settlement at all: the Session is parked until the named action advances. See Handle an approval.

Do not infer settlement from a closed HTTP request or a dropped SSE connection. A connection can close for reasons that have nothing to do with the Run, and the Run keeps going. Read settlement from the log.

Backpressure

A stalled consumer does not lose events. Checkfu keeps at most one internal page in flight and lets the rest wait in the durable log, so a slow reader applies backpressure rather than dropping data. Reconnecting from Last-Event-ID resumes from that log.

This means a consumer that falls far behind is fine. It also means the stream is not a substitute for a queue: if your processing is genuinely slow, read with the list endpoint on your own schedule instead.

Catching up without streaming

GET /v1/sessions/{id}/events pages the same log. limit defaults to 100 and caps at 1000.

The list supports two mutually exclusive entry points. Pass a decimal event seq as after to begin immediately after that durable position—the same coordinate carried by SSE Last-Event-ID. Or start without after and continue an ordinary paginated walk by echoing each opaque next_page value as page. Never send after and page together, and never try to construct an opaque page token from a sequence.

Use it for backfill, reconciliation after downtime, or rebuilding a projection from scratch. The stream and the list return the same persisted envelopes.

Next steps

Events

The envelope, the full event catalog, and the drive events you can post.

Project a session

Fold the stream into named views: messages, tool calls, ActionApprovals, the subagent tree, and the context ledger.