dtoro
5b68bdc16c
Make background/long/desynced turns visible and queueable, fixing the four
symptoms that survived the v0.15.0 chat reliability pass.
F1 - status-driven working signal (workspace.ts taskWorking/currentWorking =
streaming OR status in {planning,executing}). Drives the chat trace, indicator,
and activity spinner so a turn with no live stream (background resume, a dropped
SSE, an idle-close mid long turn) still looks alive.
F2 - operator messages sent during an in-flight turn are now QUEUED and
auto-run when the gate frees, replacing the "still finishing a previous step...
send it again" rejection. Per-session in-memory FIFO (messagequeue.go, capped at
20) drained one-at-a-time under the turn gate; a `queued` SSE event drives a
"Queued" hint. drainQueued releases via a per-iteration deferred closure so a
runChatTurn panic can't deadlock the session's gate.
F3 - SSE keepalive (12s `:keepalive` comment) in handleChat so 20-40s
inter-iteration gaps no longer trip a proxy/browser idle close (the desync root
cause). All SSE writes serialized through one mutex.
F4 - generation-aware activity timeline (only the last propose_plan renders;
superseded ones collapse to one "Earlier plan revised" marker; step-attribution
follows only the current generation) + debounced plan refetch on lifecycle
events so a missed plan.proposed self-heals.
Verified against the last session (23da10db: 6m33s turn, operator "status"
deferred at 19:48:05). go test ./cmd/nomos/ green (new messagequeue tests);
web vitest 72/72 (new F4 generation tests); vite build clean.
VERSION: 0.16.0 -> 0.17.0
83 lines
3.0 KiB
Go
83 lines
3.0 KiB
Go
package main
|
|
|
|
import (
|
|
"log/slog"
|
|
"sync"
|
|
)
|
|
|
|
// maxQueuedPerSession caps a session's queue. A held turn plus unbounded
|
|
// enqueues would grow memory without limit; an operator nudging a long
|
|
// autonomous turn realistically queues only a handful, so a generous cap is
|
|
// pure insurance. Overflow drops the newest enqueue and logs (the message is
|
|
// already persisted in the DB by handleChat before enqueue, so it isn't lost
|
|
// from the transcript — it just won't auto-run).
|
|
const maxQueuedPerSession = 20
|
|
|
|
// messageQueue holds operator messages that arrived while a turn was already
|
|
// running for a session. Plan 2026-08-03 (F2): instead of rejecting the
|
|
// operator's message with "Nomos is still finishing a previous step… send it
|
|
// again", the message is queued and auto-run when the in-flight turn releases
|
|
// the session's turn-gate permit.
|
|
//
|
|
// The queue only schedules WHEN a turn runs, not WHETHER the message is stored
|
|
// — handleChat persists the user message before acquiring the gate, so a queued
|
|
// message is already in the transcript; this just makes sure a turn eventually
|
|
// acts on it.
|
|
//
|
|
// Draining is strictly one-at-a-time under the turn gate (see drainQueued in
|
|
// main.go), so this cannot stack concurrent turns — the exact hazard the gate
|
|
// itself exists to prevent. Background resumeSession callers never touch this
|
|
// queue; they keep their non-blocking skip.
|
|
type messageQueue struct {
|
|
mu sync.Mutex
|
|
queue map[string][]string
|
|
}
|
|
|
|
func newMessageQueue() *messageQueue {
|
|
return &messageQueue{queue: map[string][]string{}}
|
|
}
|
|
|
|
// enqueue appends a message to the back of the session's FIFO. Returns false
|
|
// (and logs) if the session is already at maxQueuedPerSession — the caller's
|
|
// message is already persisted in the DB, so this only skips auto-running it.
|
|
func (q *messageQueue) enqueue(sessionID, msg string) bool {
|
|
q.mu.Lock()
|
|
defer q.mu.Unlock()
|
|
if len(q.queue[sessionID]) >= maxQueuedPerSession {
|
|
slog.Warn("nomos: message queue full; dropping auto-run for operator message", "session", sessionID, "cap", maxQueuedPerSession)
|
|
return false
|
|
}
|
|
q.queue[sessionID] = append(q.queue[sessionID], msg)
|
|
return true
|
|
}
|
|
|
|
// dequeue pops the next message from the front of the session's FIFO. Returns
|
|
// ok=false when empty.
|
|
func (q *messageQueue) dequeue(sessionID string) (string, bool) {
|
|
q.mu.Lock()
|
|
defer q.mu.Unlock()
|
|
xs := q.queue[sessionID]
|
|
if len(xs) == 0 {
|
|
return "", false
|
|
}
|
|
m := xs[0]
|
|
q.queue[sessionID] = xs[1:]
|
|
return m, true
|
|
}
|
|
|
|
// requeueFront pushes a message back to the front — used when a drainer popped
|
|
// a message but lost the race for the gate to a live turn; that turn's own
|
|
// release will drain it again.
|
|
func (q *messageQueue) requeueFront(sessionID, msg string) {
|
|
q.mu.Lock()
|
|
defer q.mu.Unlock()
|
|
q.queue[sessionID] = append([]string{msg}, q.queue[sessionID]...)
|
|
}
|
|
|
|
// peek reports the queued depth for a session (test/diagnostic helper).
|
|
func (q *messageQueue) peek(sessionID string) int {
|
|
q.mu.Lock()
|
|
defer q.mu.Unlock()
|
|
return len(q.queue[sessionID])
|
|
}
|