* feat(opencode): support mid-turn steering Register the per-message Queue/Steer RPC on the OpenCode remote launcher so a waiting-queue row can be delivered into the active turn as a concurrent ACP session/prompt (soft inject, no cancel, no mode/model/effort mutation), reusing the existing reservation/delivery-state protocol shared with Codex and Cursor ACP. The steer gate is closed synchronously by every cancelling or terminal path (normal-prompt finally, abort, switch/exit, teardown, stderr stall cancel), each steer operation is tracked from the moment its reservation is taken until completion plus all restore/consumed bookkeeping settles, and the foreground prompt drains those operations — abort-aware through the original turn signal — before ready, thinking=false or the next batch. steerability is published through the existing agentState.steeringActive flag so the web's queued Steer button is reachable. Gate the feature on the shared steering flavors (now including opencode) and update the hub's supported-flavor wording. * fix(opencode): acknowledge native steering acceptance * fix(opencode): preserve steering delivery outcomes * docs(opencode): remove obsolete abort-race comment * fix(opencode): preserve steering resolution ownership * fix(opencode): isolate ambiguous steering attempts * fix(opencode): compare resolved steering settings * fix(opencode): reconcile steering receipts after dispatch * fix(opencode): distinguish pending steering delivery outcomes * fix(opencode): restore receipts after successful native responses Track native input requests separately from detached steer operations. Resume receipt correlation after inputs quarantined by cancellation return their own successful responses, while keeping uncertain writes on completion-based confirmation. * fix(opencode): bound receipt proof to unambiguous native inputs * feat(opencode): steer v2 inputs through native inbox * fix(opencode): preserve v2 restoration and cancellation ownership * docs(opencode): describe v2 inbox steering confirmation * fix(steer): expire Claude confirmation with its owning worker
18 KiB
Message pagination, windowing, and optimistic sends
Audience: Implementers of native HAPI clients (iOS / Android). This page specifies the message paging protocol (GET /api/sessions/:id/messages), the epoch reset contract, the tail-sync loop, recommended client windowing, and the optimistic-send / cancel lifecycle. Companion pages: sse, messages, rest.
Source of truth: shared/src/apiTypes.ts (MessagesQuerySchema, MessagesResponse, SendMessageRequestSchema), hub/src/web/routes/messages.ts, hub/src/sync/messageService.ts, hub/src/store/messages.ts, reference client web/src/lib/message-window-store.ts + web/src/lib/messages.ts.
Position key
Messages are ordered by a compound position, not by seq alone:
position = (at, seq) where at = invokedAt ?? createdAt
Ascending by at, ties broken by seq. Rationale: a queued user message sits at its createdAt until the agent consumes it, at which point invokedAt is stamped and the row moves forward to its invocation position. seq (per-session insert counter) alone would freeze queued rows at enqueue order. Every cursor in this protocol is therefore a (seq, at) pair — both halves are always required together.
GET /api/sessions/:id/messages
Query parameters (MessagesQuerySchema; all numbers coerced from strings):
| Param | Type | Constraint |
|---|---|---|
limit |
int | 1–200. Default 50 when omitted; the web reference uses 200 for full latest/reset, after, and before pages, and 20 for the first latest-tail refresh on cached re-entry. |
beforeSeq + beforeAt |
int + int | Page strictly older than this position. Pairwise required. |
afterSeq + afterAt |
int + int | Page strictly newer than this position. Pairwise required. |
untilSeq + untilAt |
int + int | Inclusive snapshot head for a catch-up loop. Pairwise required; requires an after cursor. |
epoch |
int ≥ 0 | Client's cached epoch. Requires an after cursor. |
Validation rules (violations are 400 {"error":"Invalid query","issues":…}):
beforeAt⇄beforeSeq,afterAt⇄afterSeq,untilAt⇄untilSeqmust each be provided together.beforeandafterare mutually exclusive.untilandepochare only valid alongsideafter.
Session errors: 404 not found, 403 foreign namespace (see errors).
Response shape
type MessagesResponse = {
messages: DecryptedMessage[] // ascending display order
page: {
direction: 'latest' | 'before' | 'after'
limit: number
epoch: number // server's current epoch for this session
reset: boolean // true ⇒ discard your window, this page replaces it
nextBeforeSeq: number | null // cursor for the next OLDER page
nextBeforeAt: number | null
nextAfterSeq: number | null // cursor for the next NEWER page
nextAfterAt: number | null
snapshotHeadSeq: number | null // newest position at snapshot time
snapshotHeadAt: number | null
hasMore: boolean // more rows exist in the requested direction
}
}
latest (no cursor)
Newest limit rows by position, plus — out of band — every uninvoked local user message (queued rows, including future-scheduled ones), so a fresh client still sees the queued bar even when those rows fall outside the page. The out-of-band rows are pinned to every latest response and do not affect the cursor: nextBefore* anchors to the oldest row of the position-ordered page proper. hasMore = at least one row exists before that. If a page contains only server-side-filtered rows (see messages), the hub auto-advances to older pages until it can return something or history is exhausted.
before
Rows strictly older than the cursor. nextBefore* = oldest row of this page; hasMore = at least one row older than that. The response also carries the current epoch — compare it to your cached one (see below).
after
Rows strictly newer than the cursor, bounded by an inclusive snapshot head = min(until, currentHead) (or whichever exists). This keeps a catch-up loop from chasing messages appended while it runs. Responses:
- Client
epoch≠ server epoch ⇒ the server ignores the cursor and returns the latest page withreset: true(direction: 'latest'). - Snapshot head ≤ cursor ⇒ empty page,
hasMore: false,nextAfter*echoes the cursor. - Otherwise:
nextAfter*= last row's position,hasMore=nextAfter < snapshotHead.
Epoch
epoch is a per-session monotonic counter (message_epochs table, starts at 0) that is bumped whenever history changes in a way that invalidates composite cursors (hub/src/store/messages.ts):
- a new row lands before the current head position (out-of-order insert, e.g. transcript import with an earlier timestamp);
- a queued message is deleted (cancel);
- rewind / history replace (
replaceSessionMessagesFrom); - messages are copied/merged between sessions (both sides), fork hydration.
Client contract:
afterrequest — always send your cachedepoch. On mismatch the server answers with the latest page andreset: true; discard the entire local window and replace it with that page.beforerequest — the response'spage.epochmay differ from your cached one; if it does, your cursors are meaningless: drop cursor state, flag the window for a latest reset, and run a fresh tail sync (web:fetchOlderMessages→epoch-resetoutcome).- A structural change is also announced live via the
messages-invalidatedSSE event — on the open session, clear the window and tail-sync from scratch.
Tail-sync loop
Run after connect, after an SSE resume: 'gap' handshake, on session open, and when told to (messages-invalidated). Reference: runTailSync in web/src/lib/message-window-store.ts.
- No usable state (no newest cursor, no cached epoch, or a reset is pending):
- For a genuinely cold window with no cursor and no pending structural reset, request
GET …/messages?limit=20(latest) so the newest usable conversation content can paint quickly. - For a pending structural reset, request the normal full latest page with
limit=200. - Replace/merge into the window, store
page.epoch,nextBefore*(older-page cursor) andsnapshotHead*(newest cursor). Done.
- For a genuinely cold window with no cursor and no pending structural reset, request
- Cached re-entry with a usable cursor:
GET …/messages?limit=20(latest), replace the stale server window while preserving queued/concurrent rows, then exposenextBefore*for older-history loading. Done. - Have cursor + epoch outside activation: loop
GET …/messages?afterSeq&afterAt&epoch[&untilSeq&untilAt]&limit=200, whereafterstarts at your newest cursor anduntilis thesnapshotHead*captured from the first response of the loop (fixes the target so the loop terminates).page.resetordirection: 'latest'⇒ replace the window with this page; stop.- Otherwise merge the rows, advance
after = nextAfter*, update the newest cursor tomax(current, nextAfter); stop whenhasMoreis false. - Guard: if
nextAfterdid not advance past the previous cursor, abort with an error (protocol violation, do not spin).
New live rows keep arriving via the SSE message-received event; ingest them and advance the newest cursor to max(current, incoming position). Only run one tail sync at a time per session; if events force another (e.g. a reset was flagged mid-loop), queue a trailing run.
Client windowing (normative recommendation)
Constants from the web reference (web/src/lib/message-window-store.ts):
| Constant | Value | Meaning |
|---|---|---|
INITIAL_PAGE_SIZE |
20 | Request size for the cold latest page used to prioritize first paint. |
PAGE_SIZE |
200 | Request size for ordinary latest/reset, forward, and older-page fetches. |
CACHED_REENTRY_PAGE_SIZE |
20 | First latest-tail page for a cached session re-entry. |
VISIBLE_WINDOW_SIZE |
400 | Max regular rows kept in tail mode (following live bottom). |
HISTORY_WINDOW_SIZE |
600 | Max regular rows kept in history mode (user scrolled back). |
OLDER_LOAD_WINDOW_SIZE |
800 | Temporary cap while an older page is being merged (prepend). |
AGENT_RUN_WINDOW_SIZE |
800 | Separate trim bucket for codex agent-run-* rows so background-agent traces don't evict chat. |
Rules:
- Tail mode trims from the top (oldest dropped). Dropping rows ⇒ set
hasMore: trueand recompute the older-page cursor from the oldest kept row. - History mode trims from the bottom (newest dropped). Dropping newest rows means your window no longer reaches the tail ⇒ flag "latest reset required": on returning to tail mode, discard cursors and fetch a fresh latest page rather than trusting stale ones.
- Queued rows are never trimmed (user messages with
invokedAt === null, see below) — they are re-merged after every trim. - Persist the window (messages + cursors + epoch) per session for instant cold-start rendering; a genuinely cold window and cached re-activation request a small latest page for first paint, while structural resets request the ordinary full latest page and reconcile. Another client may have advanced the session by many pages; cached re-activation keeps the returned older cursor available for on-demand history loading.
Optimistic sends
Send: POST /api/sessions/:id/messages (see constraints below). Reference: web/src/hooks/mutations/useSendMessage.ts, mergeMessages in web/src/lib/messages.ts.
Lifecycle:
- Generate a client-side
localIdand append an optimistic row:{id: localId, seq: null, localId, invokedAt: null, scheduledAt, createdAt: now, status:'sending', content: {role:'user', content:{type:'text', text, attachments?}, meta:{deliveryMode}}}. A row is optimistic iffid === localId. - On POST success: status →
queuedif the session is currently thinking, elsesent. On failure: drop the row and restore the composer (or keep it asfailedwith a retry affordance when attachments are involved). - Echo: the hub emits
message-receivedcarrying the stored row (serverid, realseq, samelocalId). Merging a stored row whoselocalIdmatches an optimistic row replaces the optimistic one, preserving the client-sidestatusand any already-knowninvokedAtthe server row lacks. Fallback when nolocalIdecho matches: drop an optimisticsentrow when a server user message lands within 10 s of the same position. messages-consumed {localIds, invokedAt}(SSE): stampinvokedAt, cleardeliveryState/deliveryPending, and flip status tosenton matching rows (skipfailedones). This is what moves a message out of the queued bar and into the thread at its invocation position.messages-indeterminate {localIds}(SSE): a native dispatch or queue mutation has an unknown outcome. KeepinvokedAt: null, markdeliveryState:'indeterminate', and exclude the row from automatic replay. Retry/Cancel are explicit resolution actions and may remain unavailable until the native outcome can be reconciled. An optionaldispatching:trueidentifies a live dispatch still awaiting confirmation. PreservedeliveryState:'indeterminate'and setdeliveryPending:true; while the session is active and thinking, show confirmation pending without offering another Steer/Retry. This pending state can outlive Steer admission while the current request drains. Without the flag, cleardeliveryPendingand retain explicit unknown-outcome resolution. Neither state proves invocation.messages-requeued {localIds}(SSE): an explicit Retry restored normal queue delivery; cleardeliveryStateanddeliveryPending.message-cancelled {messageId, localId?}(SSE): remove the row (match either id).
Queued semantics: a user message is "queued" iff invokedAt === null strictly, deliveryState !== 'indeterminate', and status !== 'failed'. An indeterminate row remains visible in the unresolved-delivery bar but is not eligible for automatic delivery. undefined means already-invoked (rows from pre-V8 hubs omit the field) — only rows explicitly carrying null belong in the queued bar. Server-side, a send without a localId gets a hub-minted one, so every REST send follows the queue/ack path; only agent-originated rows without a localId are stamped invoked at insert.
Queued-state recovery
After a reconnect whose handshake said resume: 'gap' (an ok resume replayed the consume/cancel events already), the consumed/cancelled events for your queued rows may have been lost. Reference: web/src/lib/queued-state-reconciliation.ts.
- Finish a tail sync.
- Collect candidate
localIds: user rows withinvokedAt === null, excluding optimistic rows stillsending/failed. POST /api/sessions/:id/messages/queued-statewith{"localIds": […]}(max 1000 per call; batch above that) →{queuedLocalIds: string[], indeterminateLocalIds: string[], dispatchingLocalIds?: string[], invokedLocalMessages: [{localId, invokedAt}]}.- Apply
invokedLocalMessagesexactly likemessages-consumed; markindeterminateLocalIdsas unresolved delivery. Retain both queued and indeterminate rows; drop only candidates absent from all three result groups. The optionaldispatchingLocalIdsis a subset ofindeterminateLocalIds: these rows are awaiting confirmation, not deleted or already invoked. Treat the remaining indeterminate rows as genuinely unknown. Nonterminal snapshots must not overwrite a newer SSE or local transition that occurred during the request.
Send constraints
POST /api/sessions/:id/messages body (SendMessageRequestSchema):
| Field | Type | Rules |
|---|---|---|
text |
string | Required (route also accepts empty text when attachments is non-empty). |
localId |
string | Optional but required for scheduledAt. When omitted the hub mints one; send your own to reconcile an optimistic row with its stored copy. |
attachments |
AttachmentMetadata[] |
Optional. Not allowed with scheduledAt. |
scheduledAt |
epoch ms | Optional. Must be ≤ now + 7 days; requires localId; no attachments; never steer. |
deliveryMode |
'queue' | 'steer' |
Optional, default queue. steer is honored only for Pi-flavor sessions and never for scheduled sends — the hub silently normalizes everything else to queue, and deferred/replayed delivery (reconnect backfill, retries, scheduled release) always degrades steer to queue. |
Response {"ok": true}. Sending to an inactive session returns 409 {"error":"Session is inactive","code":"session_inactive"} — resume/reopen first, and note the resumed session may have a different id (migrate drafts and re-target, see rest).
Cancel and steer
Cancel: DELETE /api/sessions/:id/messages/:messageId — :messageId may be the server id or the localId. Response union (CancelMessageResponseSchema):
| Response | Meaning | Client action |
|---|---|---|
{"status":"cancelled","localId":string|null} |
Row deleted (or already gone). Bumps the epoch. | Remove the row. |
{"status":"invoked","message":DecryptedMessage} |
Too late — the agent consumed it before the cancel landed. | Ingest the returned message as the authoritative row (correct invokedAt, status sent); do not resurrect the queued snapshot. |
{"status":"busy","localId":string} |
Native delivery/removal is unresolved; cancellation cannot be confirmed. | Restore the row as indeterminate; reconcile queued state before allowing Retry/Cancel. |
Other subscribers learn the same outcome via message-cancelled / messages-consumed SSE events.
Steer a queued message into the current turn: POST /api/sessions/:id/messages/:messageId/steer → SteerQueuedMessageResponseSchema. Unlike the send-time deliveryMode option above, this endpoint supports Claude (remote mode), Pi, Codex, OpenCode v1/v2, and Cursor ACP sessions (isSteeringSupportedForSession in shared/src/modes.ts, together with the agent's steeringActive state). OpenCode v2 requires access to its experimental native REST inbox. Durable admission keeps the row pending; an exact native inbox-delivered event or projected-input readback confirms consumption. If admission or cancellation cannot be verified, the row remains indeterminate without automatic resubmission. It rejects all scheduled messages, and rejects terminal-controlled sessions unless they advertise concurrentClients.
| Response | Client action |
|---|---|
{"status":"steered","localId"} |
Keep the uninvoked row until confirmed by messages-consumed or readback; show confirmation pending when deliveryPending:true. If REST beats the pending SSE, retain a provisional confirmation hold and reconcile queued state before allowing another action. This does not stamp invokedAt; a newer unknown/consumed state wins over a stale readback. |
{"status":"invoked","message"} |
Already consumed — ingest the message. |
{"status":"failed","error","localId":string|null} |
Surface the error. Do not infer delivery state from this alone; reconcile before retrying when the native outcome is unknown. |
Retry indeterminate delivery: POST /api/sessions/:id/messages/:messageId/retry
is user-initiated only. retried or already-queued means normal queue delivery;
invoked carries the authoritative message; not-found means the row is gone.
retry-unavailable leaves the row unresolved: the hub could not prove that
retrying would avoid duplicate work. Reconcile instead of automatically retrying.
OpenCode confirmation ownership ends when its worker exits, expires, or is replaced (machine/PID/process-start marker). The hub moves only that session's uninvoked dispatching rows to unknown delivery; queued and consumed rows remain unchanged. Closing Steer admission alone does not end an existing delivery. After a hub restart, stale persisted steeringActive is not sufficient to show a live confirmation wait. Unknown rows are never automatically replayed or stamped invoked; later positive consumption may still reconcile them.