Chat wire protocol

Topic: chat:v1:app:{appKey}:thread:{threadId}

A thread is a Durable Object of type: "thread". Its messages slot is the authoritative append log; stream:{runId} slots hold live token rollups; run:{runId} slots hold run state. The DO log_id is the thread's monotonic cursor — the chat equivalent of a channel serial.

The channel does two things: it runs the DO transactions for send / edit / cancel etc., and it relays every broadcast on the thread's DO topic (durable:{appId}:thread:{threadId}) to the client. So change, batch, and the run:* lifecycle events all arrive by the same relay path.

v1 is frozen. See Versioning.

Principals

The token carries chat:{verb}:{threadId} capabilities; the channel translates the ones matching this thread into the short verbs above. A push whose required verb the principal lacks is rejected with { "reason": "forbidden: missing {verb}" }.

See Chat authentication and Authentication.

Join

Client

["1", "1", "chat:v1:app:pk_live_x:thread:t-abc", "phx_join", {
  "auth": "pk_live_x.eyJ...body...",
  "client_id": "user-42",
  "after_log_id": 1500
}]

client_id from the token wins; the join value is a fallback for the fast-path token, which may omit it. after_log_id replays missed log entries as change events before the snapshot.

Agent

{ "secret": "sk_live_x", "agent_id": "assistant", "after_log_id": 0 }

Join reply

{ "status": "ok", "response": { "socket_id": "abc-123" } }

Errors: "invalid token", "forbidden: missing chat:subscribe:{threadId}", "invalid secret", "auth required", "app_key mismatch", "invalid topic. Expected: chat:v1:app:{app_key}:thread:{thread_id}".

Server → client

snapshot

Sent on join, after any after_log_id replay. Last 50 messages + a collapsed runs map + any live stream: slots.

["1", null, "chat:v1:app:pk_live_x:thread:t-abc", "snapshot", {
  "state": {
    "messages": [ /* ...last 50 */ ],
    "runs": {
      "run-abc": {
        "runId": "run-abc",
        "status": "active",
        "ownerClientId": "user-42",
        "startedAt": 1738500002000,
        "endedAt": null,
        "suspendedTool": null
      }
    },
    "stream:run-abc": "Hello, how can I",
    "last_message_id": "msg-050"
  },
  "log_id": 1500,
  "has_older": true
}

change — single slot

{ "key": "stream:run-abc", "value": "Hello, how can I help you today?", "previous": "Hello, how can I help", "log_id": 1501 }

batch — atomic multi-slot

Emitted by GC compaction when a run ends. Four changes in one log entry:

{
  "changes": [
    { "key": "messages",        "value": [ /* ... */ ], "previous": [ /* ... */ ] },
    { "key": "run:run-abc",     "value": { "status": "complete", "endedAt": 1738500010000 }, "previous": { "status": "active" } },
    { "key": "stream:run-abc",  "value": null, "previous": "Hello, how can I help you today?" },
    { "key": "last_message_id", "value": "msg-051", "previous": "msg-050" }
  ],
  "log_id": 1502
}

stream:{runId} → null means the run is finalised: drop the in-progress streaming bubble, render the finalised message now in messages.

run:start / run:end / run:suspend / run:resume

Lifecycle signals, broadcast on the thread topic alongside the state changes. run:end is always immediately followed by its companion batch.

{ "runId": "run-abc", "ownerClientId": "user-42", "startedAt": 1738500002000 }

{ "runId": "run-abc", "reason": "complete", "endedAt": 1738500010000 }

{ "runId": "run-abc", "toolCallId": "call-xyz", "toolName": "executeTransfer", "args": { "amount": 500 } }

{ "runId": "run-abc" }

reason ∈ complete | cancelled | error.

view:update

Sent to a single client after its own branch:select — the re-projected message list for that client's branch choices.

{ "messages": [ /* projected */ ], "selections": { "msg-002": "msg-004" } }

Client → server

send — new user message

Requires publish:user_message. The client mints messageId and runId as UUIDs; runId comes back in the reply so the client can wake the agent (POST { threadId, runId } to your agent endpoint) without waiting for run:start.

["1", "5", "chat:v1:app:pk_live_x:thread:t-abc", "send", {
  "messageId": "msg-051",
  "runId": "run-def",
  "text": "Where is my refund?",
  "parentId": "msg-050"
}]

Reply: { "status": "ok", "response": { "messageId": "msg-051", "runId": "run-def", "serial": 1503, "status": "ok" } } (status: "duplicate" if messageId already exists — same reply shape, no re-append).

cancel

Requires publish:cancel_own (own run) or publish:cancel_any.

{ "runId": "run-abc" }

Validates ownerClientId on the run: slot, then runs cancel compaction (reason: "cancelled"). Reply { "status": "ok" }, or error "not_found" / "already_ended" / "forbidden".

regenerate

Requires publish:user_message. Forks the tree at an assistant message.

{ "messageId": "msg-002", "runId": "run-new", "parentId": "msg-001" }

Reply: { "messageId": "msg-002", "runId": "run-new", "serial": 1504, "status": "ok" }.

edit

Requires publish:user_message. Forks the tree at a user message.

{ "messageId": "msg-001", "text": "Where is my refund for #4821?", "newMessageId": "msg-052", "runId": "run-ghi" }

Reply: { "messageId": "msg-052", "runId": "run-ghi", "serial": 1505, "status": "ok" }.

tool_approval

Requires publish:tool_approval. Decides a suspended tool call; forwarded to the agent over an internal run topic.

{ "runId": "run-abc", "toolCallId": "call-xyz", "approved": true }

Reply: { "status": "ok" }.

load_older

Requires subscribe. Page backwards through messages (archive stubs expanded transparently).

{ "beforeMessageId": "msg-001", "limit": 50 }

Reply: { "messages": [ /* oldest-first */ ], "has_older": true }, or error "archive_unavailable".

Conversation tree (ADR-0010)

All require subscribe.

The tree is the full branching message graph; a view is one linear path through it, chosen per-client by branch:select selections. See Chat concepts.

Agent → server

All require the agent credential (secret + agent_id at join).

run:start

Requires publish:run_lifecycle. Transitions run:{runId} pending → active, broadcasts run:start.

{ "runId": "run-abc", "ownerClientId": "user-42" }

run:token

Requires publish:token. Overwrites stream:{runId} with the full accumulated content so far. Buffered (~40 ms rollup) then written to the WAL, emitting a change on the stream: slot.

{ "runId": "run-abc", "content": "Hello, how can I help you today?" }

run:end

Requires publish:run_lifecycle. Flushes the token buffer, then one atomic GC compaction transaction: read stream: → append finalised message to messages → set run: status → delete stream: → advance last_message_id. Emits run:end then batch. Idempotent — a second run:end on a non-active run replies "already_ended" and does nothing.

{ "runId": "run-abc", "reason": "complete" }

run:suspend / run:resume

Requires publish:run_lifecycle. Suspend parks the run for human approval (no GC — stream: persists); resume un-parks it.

{ "runId": "run-abc", "toolCallId": "call-xyz", "toolName": "executeTransfer", "args": { "amount": 500 } }

{ "runId": "run-abc" }

Every agent push replies { "status": "ok" } or { "status": "error", "response": { "reason": "..." } }.

Data shapes

RunInfo (run:{runId} slot)

{
  runId:         string
  ownerClientId: string
  status:        "pending" | "active" | "suspended" | "complete" | "cancelled" | "error"
  startedAt:     number       // unix ms
  endedAt:       number | null
  suspendedTool: { toolCallId: string, toolName: string, args: object } | null
}

pending — the send push created the slot before the agent called run:start. A background sweeper cancels pending / active runs older than the run timeout.

Message (WAL messages entry)

{
  messageId:     string          // client-minted UUID
  runId:         string
  ownerClientId: string          // client_id of the human who triggered the run
  role:          "user" | "assistant"
  content:       string
  parentId:      string | null
  forkOf:        string | null   // set on edit / regenerate
  msgRegenerate: string | null
  status:        "complete" | "cancelled" | "error"
  createdAt:     number
}

GC compaction invariants

• stream:{runId} exists iff run:{runId}.status is active or suspended. A terminal run never has a stream: slot.
• run:end is emitted before its companion batch.
• The GC batch is atomic — all four slot changes arrive together or not at all.
• A crashed agent leaves a dangling stream: slot; the 60 s sweeper cancels active runs past the timeout and runs cancel compaction.

Sequencing

• log_id is monotonic per thread and total-ordered across subscribers.
• snapshot reflects state at its log_id; live events start at log_id + 1.
• after_log_id on join replays every entry > after_log_id in order before the snapshot.
• run:end always precedes its batch.

See Chat protocol mapping for how this maps to the Ably AI Transport surface that @pingerchips/ai exposes.