Streams

arbe persists every thread as a durable stream — the append-only product log for that thread. Durable Streams give append-only byte storage; arbe adds a thread entity, payload contract, permission-checked proxy, and dispatch semantics.

Today arbe has one product stream family: arbe-thread-{threadId}. Each item is an ArbeThreadEntry: { id, ts, authorId?, payload }. id is the dedup key. ts is the persisted timestamp in unix ms. authorId is the acting or authoring agent when there is one; Pi-runtime and system-authored lifecycle entries may omit it. Payloads split into three groups:

chat — human/bot authored thread messages
pi.* — Pi transcript payloads decoded from runtime events (pi.assistant, pi.tool_result, pi.compaction); a few old threads also hold typing text (pi.message_start, pi.message_update, pi.chunk), which parses and is never shown
signal.* — arbe-owned lifecycle and control facts (signal.entry.*, signal.thread.*, signal.house.*, signal.environment.*, signal.files.*, signal.thread_director.*, signal.dispatch.*)

Canonical storage keeps durable facts, not adapter transcripts. OpenAI/Anthropic/Pi user and assistant roles are projection concerns at runtime edges. Thread history should still render directly from canonical entries: chat entries are content, signal entries narrate state, and a failed dispatch is recorded as signal.dispatch.failed instead of pretending the initial write rolled back.

Sandbox AI traces

child thread → PostHog session
Pi invocation → trace
model call    → generation
 tool run     → child span

Each invocation gets a fresh trace, including --continue. Pi may reuse its own session id, so it isn’t the trace id.

The mirror pairs tool results with the model call that requested them and measures execution time. PostHog receives model usage, stop reasons, allowlisted tool names, timing, and errors—never prompts, answers, arguments, results, call ids, names, or emails. Missing or invalid trace data is dropped without dropping valid spend.

Code: packages/core/schemas/sandbox-ai.ts, packages/core/pi/live-entries.ts, packages/core/pi/coding-run-usage.ts.

Client boundary

@arbe/streams/client owns transport. It exposes two separate layers:

createDurableStreamClient(baseUrl, options) talks directly to Durable Streams paths with service auth. Route handlers use it with Bearer DURABLE_STREAMS_SECRET for create, delete, append, batch append, read, read-from-offset, and raw stream handles.
createScopedStreamClient(options) talks to arbe’s app-facing proxy where callers know a scope/thread id, not a stream path or service secret. It owns scope URL construction, append helpers, long-poll tailing, offset tracking, abort-aware waits, and durable-stream protocol headers.

Do not merge these APIs. The low-level client must not learn arbe permissions, record lookup, thread lifecycle, or scoped URL rules. The scoped client must not receive the durable-stream service secret. @arbe/core/client may expose convenience methods such as postMessage() and tailThreadStream(), but those delegate to @arbe/streams instead of reimplementing tail loops.

Thread recency rides the client

Thread lists sort on the denormalized threads.last_entry_ts, and entries live on streams — so the bump is owned by the seam where stream-write capability is handed out, not by each producer (arbe-b78c). Server-side clients are minted through createRecencyTrackingStreamClient (@arbe/core/thread-recency), which wraps appends to arbe-thread-* streams with a guarded, throttled bump; sandbox writers without the master secret ride the /api/stream/:name JWT proxy, which fires the same bump. A new producer inherits recency correctness because it never sees an unwrapped client.

The raw constructor must therefore appear only in the wrapper, the sandbox writers (which post through the proxy), one-off scripts, and tests. Check with:

ast-grep -l ts -p 'createBearerDurableStreamClient($$$)' packages apps | grep -v -e packages/core/thread-recency.ts -e packages/sandbox/src -e scripts/ -e test

Any hit outside that allowlist is a bug: its appends will silently leave last_entry_ts stale.

Proxy and offsets

Browsers, CLI, and JS clients read through /api/threads/:id/stream; the app checks membership and injects the durable-stream secret upstream. Offsets stay opaque — store them, never parse or synthesize them (rules and sentinels in durable streams). Long-poll readers resume from stream-next-offset; stream-up-to-date says whether an empty response reached the tail.

createScopedStreamClient starts tailing from the caller’s fromOffset, retries transient long-poll failures, throws on auth failures, and races body reads against abort signals so CLI follow commands exit promptly.

Entry identity

Identity fields are semantic, not interchangeable:

authorId answers who authored content. Use it on chat/message content.
actorId answers who caused a workflow/system transition when the payload needs an explicit causal id.
agentId answers which agent record a signal is about.

Who may set authorId and which kinds each caller may append is decided in who may write what. Current thread entries use authorId; many signal.* payloads rely on it. signal.entry.deleted is the canonical deletion tombstone: its payload entryId names the deleted chat entry and the signal entry’s authorId names the deleting actor. If a future signal needs to distinguish author, actor, and subject, add explicit payload fields rather than overloading authorId.

Rules: thread history stays human-facing and directly renderable; transcript projection never enters storage; pi payloads stay at the pi boundary and arbe-owned payloads are schema-validated; no generic actor envelopes, no user.*/assistant.* variants, no span.* unless it is durable product history rather than tracing exhaust.

Deletion projection

Deleting a message never mutates durable history. DELETE /api/threads/:id/entries/:entryId appends signal.entry.deleted; raw reads and debug tails retain both entries, while normal chat projection removes the targeted chat. Only the original author or a house owner may delete, and only chat entries are valid targets. Dispatch history also omits tombstoned chat on later turns; a tombstone cannot undo dispatch that already ran.

Dispatch interaction

POST /api/threads/:id/entries appends the chat entry, then wakes the thread director; a bot reply is just another unread entry. Append-first matters: if the wake fails, the user message stays and a failure signal is appended. The log tells the truth.

Current view

What is happening right now is folded from the stream, never stored on its own: reduceThreadNow (packages/core/thread-now.ts) turns entries into who is working (a reply or a coding run), its phase and round, the tool calls with no result yet, the messages still waiting or taken in mid-reply, spend for the current reply or run, and why nobody speaks when that has a reason. The fold uses entry order only. Its inputs are signals already in the transcript; the one added for it is signal.dispatch.steered, which a reply appends when it takes messages in between tool rounds, so a steered message never looks unanswered while the reply runs.

GET /api/threads/:id/now folds a snapshot and returns it with the stream offset it reached. A normal thread folds the director’s cached tail and then everything past its decided offset, so a message still owed to a second mentioned bot and the reason the last pass ended both survive the offset moving on. A message stays waiting until the director’s pass over it ends with nobody picked; a pick alone doesn’t settle it. A coding run reads from the start its claim stored on the thread row (run_offset) and folds from its own running marker; a run claimed before starts were stored shows only its run state. Clients then tail /stream from that offset and apply the same reducer per entry. The response also carries status: the thread’s record plus one per bot that acted since the last message and one per live delegated run, derived by deriveThreadStatus from the same view; a new message resets the bots (a delegated run’s report doesn’t: it is that run’s result, held on T/<childId>), a bot’s result stands until then, and the thread takes the highest of blocked > working > error > done > idle. Working needs live evidence: when nothing has been written for longer than a reply may run (REPLY_TURN_CAP_MS, 10 minutes), a reply or a waiting message has nobody behind it and reads idle, or error if its attempts failed; a coding run is left to its own 90s silence rule. arbe thread now [--follow] and arbe thread entries read (which follows until nothing is working or waiting) do exactly that, and so does the web app.

The thread’s own record is also stored on its row (threads.status), so GET /api/threads and the house rollup never fold a stream. trackThreadStatus (@arbe/core/thread-status-sync) rides the same append client as the recency bump, plus the sandbox stream proxy: after an entry that can move the record (chat, director decision and outcome, dispatch lifecycle and phase, run state, pi_failed, orphaned, child_finished) it runs the /now read for the thread and writes the record if it changed, then appends a house threads event. Tool calls and assistant text don’t trigger it, so a stored working msg can lag /now mid-turn; the state doesn’t. A write is guarded by status_rev: the row is read before the stream and the update lands only if the revision is unchanged, else it reads again, so a stale fold never overwrites a newer one. Drift is repaired three ways: /now writes the record back whenever it disagrees, backstage’s 60s prune sweep recomputes every thread stored as working or blocked, and packages/core/scripts/backfill-thread-status.ts recomputes every live thread (or one house with --house) by hand. A house’s status (GET /api/houses[/:id]) is computed on read from its live, unarchived threads’ records by house_thread_statuses: the top state by the same order, the thread record that speaks for it, and a count per state.

Rendering in www

The browser tails the stream through @arbe/core/client. Chat.svelte renders the entries directly (ChatEntries.svelte, with pure helpers in apps/www/src/lib/chat-transforms.ts), and reads who is working from the thread’s current view (apps/www/src/lib/thread-now.svelte.ts, one follower per open thread). The renderer’s design lives in chat UI.

Code: packages/streams/client.ts, packages/core/entries.ts, packages/core/schemas/stream-entries/{envelope,thread}.ts, packages/core/schemas/thread.ts (threadStreamId), apps/www/src/routes/api/threads/[id]/stream/.
See durable streams, dispatch, threads, and the API.