feat: queued follow-up messages in composer (#58) #287

Merged

dries merged 2 commits from feat/queued-follow-up-messages into main

2026-07-12 01:34:24 +02:00

dries commented

2026-07-11 14:00:51 +02:00

Owner

Closes #58.

Server-side, shared, race-free follow-up message queue. Prompts submitted while a session is mid-turn are queued and auto-sent one per turn when the session goes idle.

Design

Server-side, not frontend — the queue lives in state.db (migration v21), shared across every connected client and surviving a client moving machines. A frontend-only queue would show phantom messages to other clients watching the same session.
Race-free by construction — enqueue is unconditional (no busy check → no check-then-act race); flush is the sole send gate, serialized per session and driven by the existing session.idle SSE edge. One message drains per idle edge, oldest first (one follow-up per turn).
Remote-transparent — the queue row carries the compound platform id, so flush routes back over gRPC for remote sessions.

What's included

Backend

internal/queuesvc — enqueue / flush (per-session lock) + List/Remove/Move.
internal/state — v21 queued_message table + CRUD (append, head, list, delete, transactional reorder).
Composer POST /message enqueues; session.idle flushes. Loops/MCP still send directly (programmatic, not composer).
REST: GET/DELETE /api/session/{id}/queue{/qmid}, POST .../move (ownership-scoped).
ocman.queue.updated broadcast carrying the session's full list.

Frontend

useMessageQueue renders the list purely from server state (initial fetch + queue.updated broadcasts).
Composer renders the queue under the footer with remove / reorder controls, via a dedicated QueuedMessages leaf component.

Reliability & live-update fixes

Follow-up round hardening the live-update path so the list stays correct without a refresh:

http.Flusher on the status-recorder middleware — the wrapper implemented http.Hijacker (for WebSockets) but not http.Flusher, so the global /api/events SSE handler failed its w.(http.Flusher) check and returned before subscribing. No client ever subscribed → broadcasts reached zero subscribers → the UI only updated on refresh. (Per-session /api/session/{id}/events worked only because it bypasses that path.) This was the root cause of "new queued messages don't appear until refresh."
Reliable delivery — queue.updated is coalesced in the broadcast hub: on a full subscriber buffer the latest snapshot is parked (last-write-wins) instead of dropped, so the freshest list is never lost under load.
Load-on-reconnect — useMessageQueue reloads from the endpoint on every SSE (re)connect (mirroring the conversation SSE), with a monotonic seq guard so a slow reload can't clobber a fresher broadcast.
No spurious clears — the hook only resets the list on an actual session change, not on every effect re-run (e.g. platform resolving), which previously wiped a just-shown enqueue.
Blocking composer send — the composer disables its send controls until the POST completes, so the broadcast has landed by the time it re-enables; no optimistic queue state needed.
Sweep backstop fix — a force-queued message is no longer marked drained at enqueue time, which had disarmed the Sweep and could strand a message whose session.idle edge never arrived.
Dropped a vite proxy proxyRes data listener that put SSE responses into flowing mode and consumed the bytes.

Tests

Go: state CRUD, queuesvc (order, one-per-idle, send-error-retry, ownership guards, notify-on-every-mutation), server route + SSE-wire integration, broadcast coalescing, statusRecorder Flush/Hijack delegation.
Frontend: useGlobalEvents queue routing, useMessageQueue (broadcast apply, reconnect reload, seq guard, effect-rerun survival), QueuedMessages component.
All green: go test, go vet, platform-branching check, tsc -b, pnpm lint (0 errors), pnpm test (1480 pass), coverage ratchet ✅.

Closes #58. Server-side, shared, race-free follow-up message queue. Prompts submitted while a session is mid-turn are queued and auto-sent one per turn when the session goes idle. ## Design - **Server-side, not frontend** — the queue lives in `state.db` (migration v21), shared across every connected client and surviving a client moving machines. A frontend-only queue would show phantom messages to other clients watching the same session. - **Race-free by construction** — enqueue is unconditional (no busy check → no check-then-act race); flush is the sole send gate, serialized per session and driven by the existing `session.idle` SSE edge. One message drains per idle edge, oldest first (one follow-up per turn). - **Remote-transparent** — the queue row carries the compound platform id, so flush routes back over gRPC for remote sessions. ## What's included **Backend** - `internal/queuesvc` — enqueue / flush (per-session lock) + List/Remove/Move. - `internal/state` — v21 `queued_message` table + CRUD (append, head, list, delete, transactional reorder). - Composer `POST /message` enqueues; `session.idle` flushes. Loops/MCP still send directly (programmatic, not composer). - REST: GET/DELETE `/api/session/{id}/queue{/qmid}`, POST `.../move` (ownership-scoped). - `ocman.queue.updated` broadcast carrying the session's full list. **Frontend** - `useMessageQueue` renders the list purely from server state (initial fetch + `queue.updated` broadcasts). - Composer renders the queue under the footer with remove / reorder controls, via a dedicated `QueuedMessages` leaf component. ## Reliability & live-update fixes Follow-up round hardening the live-update path so the list stays correct without a refresh: - **`http.Flusher` on the status-recorder middleware** — the wrapper implemented `http.Hijacker` (for WebSockets) but not `http.Flusher`, so the global `/api/events` SSE handler failed its `w.(http.Flusher)` check and returned before subscribing. No client ever subscribed → broadcasts reached zero subscribers → the UI only updated on refresh. (Per-session `/api/session/{id}/events` worked only because it bypasses that path.) This was the root cause of "new queued messages don't appear until refresh." - **Reliable delivery** — `queue.updated` is coalesced in the broadcast hub: on a full subscriber buffer the latest snapshot is parked (last-write-wins) instead of dropped, so the freshest list is never lost under load. - **Load-on-reconnect** — `useMessageQueue` reloads from the endpoint on every SSE (re)connect (mirroring the conversation SSE), with a monotonic seq guard so a slow reload can't clobber a fresher broadcast. - **No spurious clears** — the hook only resets the list on an actual session change, not on every effect re-run (e.g. `platform` resolving), which previously wiped a just-shown enqueue. - **Blocking composer send** — the composer disables its send controls until the POST completes, so the broadcast has landed by the time it re-enables; no optimistic queue state needed. - **Sweep backstop fix** — a force-queued message is no longer marked drained at enqueue time, which had disarmed the Sweep and could strand a message whose `session.idle` edge never arrived. - Dropped a vite proxy `proxyRes` data listener that put SSE responses into flowing mode and consumed the bytes. ## Tests - Go: state CRUD, queuesvc (order, one-per-idle, send-error-retry, ownership guards, notify-on-every-mutation), server route + SSE-wire integration, broadcast coalescing, `statusRecorder` Flush/Hijack delegation. - Frontend: `useGlobalEvents` queue routing, `useMessageQueue` (broadcast apply, reconnect reload, seq guard, effect-rerun survival), `QueuedMessages` component. - All green: `go test`, `go vet`, platform-branching check, `tsc -b`, `pnpm lint` (0 errors), `pnpm test` (1480 pass), coverage ratchet ✅.

forgejo-actions commented

2026-07-11 14:08:46 +02:00

Coverage ratchet: ✅ pass

Suite	Baseline	This PR	Δ
go	72.10%	72.40%	+0.30	✅
frontend	62.24%	62.39%	+0.15	✅

Tolerance: -0.1%. Baseline stored on gh-pages.

### Coverage ratchet: ✅ pass | Suite | Baseline | This PR | Δ | | |---|---|---|---|---| | go | 72.10% | 72.40% | +0.30 | ✅ | | frontend | 62.24% | 62.39% | +0.15 | ✅ | _Tolerance: -0.1%. Baseline stored on `gh-pages`._