T3.Board.Thread contains thread_id, title, state, last_turn_at,
last_message_snippet and host. Keep the returned cache outside LiveView's wire
handling and broadcast only the rows your UI needs:
rows = T3.Board.apply(%{}, chunk_values, host: "workstation")
rows = T3.Board.apply(rows, next_chunk_values, host: "workstation")
board_rows = Map.values(rows)The cache is keyed by {host, thread_id}. Choose a stable host label per T3
installation. Shell status is retained as a string; archived rows use
"archived". A thread projection uses its newest run by ordinal, or "idle"
when no run exists. Thread metadata events preserve the existing run state.
last_turn_at is the latest available run/turn activity timestamp (requested,
work-started, started or completed); shell rows use the corresponding latest-run
summary timestamps. This is activity time, not a promise that the run completed.
Timestamps remain ISO 8601 wire strings. Message snippets collapse whitespace and
keep at most 160 graphemes; override with snippet_length: 240.
Use apply/3 for stateful folding: it preserves values absent from sparse events,
replaces only the matching host's cache on an authoritative shell snapshot and
removes deleted threads. normalize/2 returns individual {:upsert, row} and
{:delete, thread_id} operations for custom consumers, but a partial upsert can
contain nil fields and does not itself express authoritative cache replacement.
Live read-only probe
On 2026-11-05, an installed local T3 Code 0.1.1 server returned the following shapes over Effect RPC, using this SDK on Elixir 1.20.5 / OTP 29.1.2. Only field names and redacted examples are recorded here; the probe did not persist real message bodies, titles, IDs or credentials. The short-lived read-only environment session was revoked afterwards.
Both streams deliver {:t3_rpc, subscription, {:chunk, values}}, where values
is an array of stream items, not an array of thread rows.
orchestration.subscribeThread takes %{"threadId" => thread_id}. Its initial
chunk included an ordinary full snapshot:
[
{
"kind": "snapshot",
"snapshotSequence": 123,
"projection": {
"thread": {"id": "redacted", "title": "Example", "archivedAt": null, "deletedAt": null},
"runs": [],
"messages": [{"id": "redacted", "role": "user", "text": "Example message", "createdAt": "2026-11-05T00:00:00.000Z", "updatedAt": "2026-11-05T00:00:00.000Z"}],
"providerTurns": [],
"attempts": [], "nodes": [], "subagents": [],
"providerSessions": [], "providerThreads": [], "runtimeRequests": [],
"plans": [], "turnItems": [], "visibleTurnItems": [],
"checkpointScopes": [], "checkpoints": [],
"contextHandoffs": [], "contextTransfers": [],
"updatedAt": "2026-11-05T00:00:00.000Z"
}
}
]The example omits unneeded fields within records; it is a shape illustration,
not a complete upstream schema. orchestration.getThreadProjection takes the
same %{"threadId" => thread_id} and returns the projection object directly.
Pass that object to T3.Board.apply/3 for a per-thread catch-up.
The pinned contract defines subsequent domain events as:
{"kind":"event","sequence":124,"event":{"id":"redacted","threadId":"redacted","occurredAt":"2026-11-05T00:01:00.000Z","type":"message.updated","payload":{"id":"redacted","threadId":"redacted","text":"Updated message","updatedAt":"2026-11-05T00:01:00.000Z"}}}Unlike the initial snapshots, this illustrative update envelope was verified
against the pinned schema rather than generated by mutating a live thread.
The normalizer handles thread.* metadata, thread.deleted, run.created,
run.updated, provider-turn.updated and message.updated; unrelated/unknown
records are ignored. Events must be applied in stream order. Internal __meta__
bookkeeping retains the newest run ID/ordinal and message ID/creation time so a
late completion of an older run or an edit to a historical message cannot replace
the current summary. Keep this metadata when caching a row; it is excluded from
Inspect and is not a display field. Shell summaries omit run ordinals and message
creation times, so the reducer falls back to the known run ID/request time and
message ID/update time until a full thread projection supplies those fields.
Use the shell stream for a board-wide current-status view.
orchestration.subscribeShell takes %{} and initially returned:
[
{"kind":"snapshot","snapshot":{"schemaVersion":2,"snapshotSequence":123,"projects":[],"threads":[{"id":"redacted","title":"Example","status":"idle","latestVisibleMessage":null}],"archivedThreads":[]}},
{"kind":"snapshot","resolvedRepositoryIdentityRoots":["/redacted"],"snapshot":{"schemaVersion":2,"snapshotSequence":123,"projects":[],"threads":[],"archivedThreads":[]}}
]The second item is a project-only enrichment snapshot, not an empty board.
apply/3 ignores this marked snapshot. Unmarked snapshots are authoritative;
thread updates have kind: "thread.updated" and a thread record; removals have
kind: "thread.removed", threadId and location ("active" or "archive").
An optional "synchronized" item is emitted when requesting completion markers.
Subscribe and acknowledge
{:ok, rpc} = T3.RPC.start_link(url: rpc_url, token: environment_token)
{:ok, subscription} = T3.RPC.subscribe(rpc, "orchestration.subscribeThread", %{"threadId" => thread_id})
receive do
{:t3_rpc, ^subscription, {:chunk, values}} ->
rows = T3.Board.apply(%{}, values, host: "workstation")
:ok = T3.RPC.ack(rpc, subscription)
Map.values(rows)
{:t3_rpc, ^subscription, {:error, error}} ->
{:error, error}
{:t3_rpc, ^subscription, {:done, result}} ->
{:done, result}
after
5_000 -> :timeout
end
:ok = T3.RPC.unsubscribe(rpc, subscription)
:ok = T3.RPC.close(rpc)Acknowledge each chunk once after consumption, then continue receiving in your service's loop. The low-level transport does not reconnect automatically. Do not enable compact ID/body/turn-item or deferred-work options for this thin normalizer; those formats require T3's additional projection-restoration logic.