# Board events and confirmed stream shapes

`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:

```elixir
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:

```json
[
  {
    "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:

```json
{"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:

```json
[
  {"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

```elixir
{: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.