Realtime

Every run has a durable timeline: an append-only, ordered record of its status transitions and ctx.log lines. runs.watch streams it live, so you follow a run as it executes instead of polling GET /runs/{id} in a loop.

Watching a run

runs.watch(id) is an async generator of timeline frames. It replays the run's history on connect, tails new frames as they land, and ends on its own when the run reaches a terminal state.

import { createClient } from "@duraton/sdk/client";

const duraton = createClient({
  url: process.env.DURATON_URL!,
  apiKey: process.env.DURATON_API_KEY,
});

for await (const frame of duraton.runs.watch(runId)) {
  switch (frame.kind) {
    case "run_status":
      console.log("run ->", frame.status);
      break;
    case "step_status":
      console.log("  step", frame.name, "->", frame.status);
      break;
    case "log":
      console.log("  log", frame.level, frame.message);
      break;
  }
}

from duraton.client import AsyncDuratonClient

async with AsyncDuratonClient() as dx:
    async for frame in dx.runs.watch(run_id):
        if frame.kind == "run_status":
            print("run ->", frame.status)
        elif frame.kind == "step_status":
            print("  step", frame.name, "->", frame.status)
        elif frame.kind == "log":
            print("  log", frame.level, frame.message)

import (
	"context"
	"fmt"

	"duraton.dev/sdk-go/client"
)

dx := client.New(client.Options{})

err := dx.Runs.Watch(context.Background(), runID, client.WatchOptions{}, func(f client.TimelineFrame) bool {
	switch f.Kind {
	case client.TimelineRunStatus:
		fmt.Println("run ->", f.Status)
	case client.TimelineStepStatus:
		fmt.Println("  step", f.Name, "->", f.Status)
	case client.TimelineLog:
		fmt.Println("  log", f.Level, f.Message)
	}
	return true
})

A frame is a discriminated union on kind; narrowing on kind gives you the right fields:

Status transitions and log lines append to that one stream, ordered by a per-run seq. Each transition's timeline row is written in the same transaction as the status change, so the stream cannot disagree with the run.

Reading log history

To read the persisted ctx.log lines without opening a stream - a one-shot dump, a report, paging a long run - use runs.logs:

const lines = await duraton.runs.logs(runId);                                    // oldest first
const more = await duraton.runs.logs(runId, { from: lines.at(-1)?.seq, limit: 200 });

from duraton.client import LogsOptions

lines = await dx.runs.logs(run_id)                                    # oldest first
more = await dx.runs.logs(run_id, LogsOptions(from_seq=lines[-1].seq, limit=200))

lines, err := dx.Runs.Logs(context.Background(), runID, client.LogsOptions{Limit: 200}) // oldest first

Each line carries seq, ts, level, message, fields?, scope, and attempt. Pass the last seq you saw as from to page forward. This is the same data the log frames carry, served as history from GET /runs/{id}/logs.

Resuming after a drop

Every frame carries a monotonic seq. Remember the last one you saw and reconnect past it:

let last = 0;
for await (const frame of duraton.runs.watch(runId, { from: last })) {
  last = frame.seq;
  // ...handle frame
}

from duraton.client import WatchOptions

last = 0
async for frame in dx.runs.watch(run_id, WatchOptions(from_seq=last)):
    last = frame.seq
    # ...handle frame

var last int64
dx.Runs.Watch(context.Background(), runID, client.WatchOptions{From: last}, func(f client.TimelineFrame) bool {
	last = f.Seq
	// ...handle frame
	return true
})

{ from } replays strictly after that seq - no gaps, no duplicates.

How it streams

runs.watch opens a Server-Sent Events connection to GET /runs/{id}/stream over native fetch - no extra dependency. It replays every row after ?from=, then delivers new rows as they commit.

A frame is published only after its row commits, so nothing is pushed that is not already durable. The live push is a latency optimization over that durable record, not the record itself: a push that never arrives has not lost the row, and the next read or { from } reconnect returns it. What a dropped frame costs you is push latency, not a frame.

The project-wide stream

runs.watchAll (GET /runs/stream) tails run_status transitions across the whole project - the read behind a live runs list. It is best-effort: there is no global cursor across runs, so it tails by timestamp and a frame is a "refetch" signal rather than a lossless log; a missed frame self-corrects on the next transition.

for await (const frame of duraton.runs.watchAll()) {
  console.log(frame.runId, "->", frame.status); // a run_status frame; pass { signal } to stop
}

Use runs.watch when you need one run's exact, ordered timeline, and runs.watchAll when you only need to know the project had activity.

Filtering the project-wide stream

runs.watchFiltered narrows the firehose to only the run_status frames matching a filter, so a multi-tenant view subscribes to exactly the slice it renders instead of receiving every run's transitions and filtering client-side. It hits the same GET /runs/stream endpoint with query params, and keeps the same best-effort semantics as watchAll (each frame is a refetch signal, not a lossless log).

The filter axes are the same as runs.list, combined with AND:

Omitting every axis is equivalent to runs.watchAll.

// Only running charges for one tenant, live.
for await (const frame of duraton.runs.watchFiltered({
  app: "billing",
  workflow: "charge",
  status: "running",
  tags: { tenant: "acme" },
})) {
  console.log(frame.runId, "->", frame.status); // pass { signal } to stop
}

Status is read from the frame itself; app, workflow, and tags are run attributes the engine resolves once per run id and caches for the life of the connection (a run's app, workflow, and tags are fixed at creation), so a busy workspace costs one lookup per distinct run, not per frame.

Backpressure and fanout

Each subscriber is served by its own connection with a bounded send buffer; a subscriber that falls behind drops frames rather than growing an unbounded queue, so one slow client can never stall the engine or another subscriber. Because the durable timeline is the source of truth, a dropped push is only lost latency: the next transition (or a periodic refetch) re-surfaces the run's current state. Fanout cost scales with the number of subscribers times the transition rate; the server-side filter keeps each subscriber's delivered bytes proportional to its slice, not the whole project.

Validating fanout under real concurrency (many subscribers, high transition rate) is a deploy-time step against a running engine with a load harness - it is not exercised by the unit/integration tests, which cover the filter matching and the slow-consumer drop policy in isolation.

Durable transitions across the project

runs.watchAll and runs.watchFiltered are deliberately lossy: they carry only run_status frames and treat each one as a refetch signal, so they are perfect for a live list but useless if you need to observe every step of every run without missing one. runs.watchTransitions is the durable counterpart - the project-wide, step-level, resumable transition log. It is the managed equivalent of an in-process onStepTransition hook, except it survives a reconnect.

It hits the same GET /runs/stream endpoint as watchAll/watchFiltered, with two additions:

• kinds - which transition kinds to receive. The SDK method defaults to both run_status and step_status (TRANSITION_KINDS), sent as ?kinds=run_status,step_status. Only these two kinds are carried project-wide; log and ai_chunk stay on the per-run runs.watch to keep the firehose bounded. (The raw endpoint defaults to run_status only when ?kinds= is omitted, which is why watchAll is unaffected - it never sends the param.)
• since - an RFC3339 timestamp cursor. The stream resumes from that position instead of tailing from connect time, so a reconnect picks up exactly where you left off. Omit it to tail from now.

It also takes the same optional app / workflow / status / tags narrowing axes as watchFiltered (AND across tags).

At-least-once, dedupe on (runId, seq)

Unlike the per-run runs.watch, there is no single global cursor across runs - the engine tails the workspace timeline by timestamp and re-scans a short overlap window on each poll, because rows can commit slightly out of order. That makes delivery at-least-once: on a reconnect (and around the since boundary) a frame you already saw can arrive again. Combine each frame's runId with its per-run seq into an idempotency key and dedupe on it:

import { createClient } from "@duraton/sdk/client";

const duraton = createClient({
  url: process.env.DURATON_URL!,
  apiKey: process.env.DURATON_API_KEY,
});

const seen = new Set<string>();
let cursor: string | undefined; // persist this to resume across restarts

for await (const frame of duraton.runs.watchTransitions({
  kinds: ["run_status", "step_status"], // the default; narrow to one if you like
  since: cursor,
  app: "billing", // optional: same axes as watchFiltered
})) {
  const key = `${frame.runId}:${frame.seq}`;
  if (seen.has(key)) continue; // at-least-once -> dedupe
  seen.add(key);
  cursor = frame.ts; // an RFC3339 ts; feed back as `since` on reconnect

  switch (frame.kind) {
    case "run_status":
      console.log(frame.runId, "run ->", frame.status);
      break;
    case "step_status":
      console.log(frame.runId, "step", frame.name, "->", frame.status);
      break;
  }
}

Bounded replay: since has a max lookback

A since cursor cannot resume from arbitrarily far back. The stream bounds how much history it will replay, so a stale or wrong cursor - a process that reconnects after a long outage still holding an old persisted since - cannot silently replay weeks of history as if it just happened and re-trigger every side effect on the way.

The clamp is never silent. When since predates the bound, the very first frame on the connection names both the cursor you asked for and the one the stream actually resumed from:

for await (const frame of duraton.runs.watchTransitions({ since: cursor })) {
  if (frame.kind === "cursor_clamped") {
    console.warn(
      `cursor ${frame.requestedSince} predates the ${frame.maxLookbackMs}ms bound; ` +
        `resumed from ${frame.effectiveSince} instead - some history was skipped`,
    );
    continue;
  }
  // ...
}

A cursor this stale usually means the consumer was down far longer than expected, or persisted a since from the wrong stream. Deeper history is still reachable - the bound applies only to the live, resumable firehose - by paging a specific run's timeline with runs.logs or reading run state directly with runs.list/runs.get.

Catch-up vs live: the historical flag

Every run_status and step_status frame from the project-wide stream carries historical: true while the frame is catch-up replay from your since cursor, false once the stream has caught up to "now" as of connect time. It flips at most once per connection, from true to false, and stays false for the rest of the stream.

Use it to keep local state current from every frame while limiting side effects to genuinely new activity:

for await (const frame of duraton.runs.watchTransitions({ since: cursor })) {
  if (frame.kind === "cursor_clamped") continue;
  applyToLocalState(frame); // always keep local state current, replay included

  if (frame.historical) continue; // already handled the first time it happened
  notifySlack(frame); // side effects only for frames that are actually live
}

watchFiltered vs watchTransitions

Both narrow the same GET /runs/stream endpoint; they differ in what a frame means:

Reach for watchFiltered when a frame just tells you "something changed, refetch". Reach for watchTransitions when you must act on each transition and cannot afford to drop one across a reconnect.

In the console

The run-detail view tails GET /runs/{id}/stream, so status and steps update live; a step's ctx.log lines appear under a Logs tab on that step, and handler-level lines under a Logs tab on the run.