feat(nomos): per-session turn serialization + chat reliability/UX fixes

The agent could run two turns for one session at once (a reconnect resumed while the live turn was still going), and their interleaved tool calls corrupted the activity panel, fabricated a confusing "parallel/nested" sequence, and made tasks feel stuck/never-ending. Several UX gaps compounded it. Turn serialization (F1): - turnGate: at most one in-flight turn per session. Background resume paths (continuation worker, idle sweep, answer-question, /resume, reconnect) skip non-blocking when busy; the live chat path waits briefly then bails cleanly instead of stacking a second turn. - resumeSession returns whether it ran; continueSession marks an execution "continued" only after a real run (review P0) so a busy-skip can't lose a finished-execution result. Idle nudge bumps only after delivery (P1). Connection state (F2/F3, web): - humanize/bucket raw errors ("model connection dropped..."); one surface per drop; a terminal task.status event clears stuck streaming/disconnected state and dismisses the connection toast. Reconnect no longer spawns turns. Streaming where you look (F4, web): - live command output in the global activity timeline and in the inline tool card (auto-opened, tail-pinned) -- not just the per-window rail. Other (web): artifact/knowledge deep links (F5); step-first stable "thinking" headline (F6); stable chat layout, no empty->content reflow (F7); lazy event sync (P2.2); reconnect skips a terminal session (P2.1). VERSION: 0.14.2 -> 0.15.0
2026-08-03 15:42:10 +02:00
parent bb05f215c6
commit 39e9227fdb
18 changed files with 1197 additions and 153 deletions
--- a/cmd/nomos/agent.go
+++ b/cmd/nomos/agent.go
@@ -57,6 +57,9 @@ type agent struct {
 	apiBase    string // oikos HTTP API base, derived from NOMOS_MCP_URL, for chat-assent approvals
 	apiToken   string // OIKOS_MCP_BEARER_TOKEN — api's combinedAuth requires it (no dev-open bypass)
 	httpClient *http.Client
+	// gate serializes turns per session (at most one in-flight turn per
+	// sessionID). See turngate.go and plan 2026-08-03 F1.
+	gate *turnGate
 }

 func newAgent(ctx context.Context, clients *mcpClientPool, st *store, agentSlug string) (*agent, error) {
@@ -117,6 +120,7 @@ func newAgent(ctx context.Context, clients *mcpClientPool, st *store, agentSlug
 		apiBase:    apiBase,
 		apiToken:   os.Getenv("OIKOS_MCP_BEARER_TOKEN"),
 		httpClient: &http.Client{Timeout: 15 * time.Second},
+		gate:       newTurnGate(),
 	}, nil
 }

--- a/cmd/nomos/continue.go
+++ b/cmd/nomos/continue.go
@@ -76,16 +76,20 @@ func (a *agent) processIdleSweep(ctx context.Context) {
 		s := s
 		if s.CompletionNudges == 0 {
 			safego.Go("nomos:idle-nudge:"+s.ID, func() {
-				if err := a.store.bumpCompletionNudge(ctx, s.ID); err != nil {
-					slog.Error("nomos: idle nudge bump failed", "session", s.ID, "error", err)
-					return
+				note := fmt.Sprintf("[System: this task ('%s') has been idle for %s with no complete_task call. "+
+					"If the goal is done (or can't be completed), call complete_task now with the outcome and a "+
+					"one-line summary. If you're still genuinely working through the plan, ignore this and continue.]",
+					s.Goal, idleTaskThreshold)
+				note = a.store.enrichResumeNote(ctx, s.ID, note)
+				// P1: only count the nudge if it actually delivered. resumeSession
+				// skips (returns false) when a turn is already active; bumping the
+				// counter anyway would make the next sweep auto-close a merely-busy
+				// session as "unanswered."
+				if a.resumeSession(ctx, s.ID, note) {
+					if err := a.store.bumpCompletionNudge(ctx, s.ID); err != nil {
+						slog.Error("nomos: idle nudge bump failed", "session", s.ID, "error", err)
+					}
 				}
-			note := fmt.Sprintf("[System: this task ('%s') has been idle for %s with no complete_task call. "+
-				"If the goal is done (or can't be completed), call complete_task now with the outcome and a "+
-				"one-line summary. If you're still genuinely working through the plan, ignore this and continue.]",
-				s.Goal, idleTaskThreshold)
-			note = a.store.enrichResumeNote(ctx, s.ID, note)
-			a.resumeSession(ctx, s.ID, note)
 			})
 			continue
 		}
@@ -163,7 +167,9 @@ func (a *agent) processContinuations(ctx context.Context) {
 				continue
 			}
 		}
-		a.store.markContinued(ctx, p.ExecID) // stamp first: a failure here must not cause a re-continue loop
+		// markContinued now happens inside continueSession, AFTER resumeSession
+		// actually runs (P0). Pre-marking here consumed the item even when
+		// resumeSession skipped on a busy session, losing the result.
 		safego.Go("nomos:continue-session:"+p.SessionID, func() { a.continueSession(ctx, p) })
 	}
 }
@@ -179,7 +185,18 @@ func (a *agent) processContinuations(ctx context.Context) {
 // something new to poll for.
 func (a *agent) continueSession(ctx context.Context, p pendingContinuation) {
 	slog.Info("nomos: auto-continuing session", "session", p.SessionID, "execution", p.ExecID, "status", p.Status)
-	a.resumeSession(ctx, p.SessionID, buildContinuationNote(p))
+	// P0 (plans/2026-08-03-nomos-chat-changes-review.md): mark the execution
+	// continued ONLY after the turn actually ran. resumeSession skips (returns
+	// false) when another turn is already active for this session; marking
+	// before that — as the old code did — consumed the item (continued_at set,
+	// never re-queued by pendingContinuations) and silently lost the result.
+	// On a skip, leave it pending so the next worker tick retries once the
+	// active turn frees the permit.
+	if !a.resumeSession(ctx, p.SessionID, buildContinuationNote(p)) {
+		slog.Info("nomos: continuation deferred — a turn is active; will retry next tick", "session", p.SessionID, "execution", p.ExecID)
+		return
+	}
+	a.store.markContinued(ctx, p.ExecID)
 }

 // resumeSession re-invokes the agent for a session with a system-injected note —
@@ -187,7 +204,26 @@ func (a *agent) continueSession(ctx context.Context, p pendingContinuation) {
 // (handleAnswerQuestion) — persisting progress LIVE (a placeholder row updated
 // in place as each tool call lands) so the frontend poller sees each step,
 // instead of total silence until the whole resume concludes.
-func (a *agent) resumeSession(ctx context.Context, sessionID, note string) {
+//
+// F1 (plan 2026-08-03): this is the single entry point for EVERY background
+// turn — the continuation worker, idle sweep, answer-question, /resume, and the
+// empty-message reconnect all funnel through here. It acquires the session's
+// turn permit non-blocking and SKIPS if a turn is already running. A duplicate
+// resume while a turn (live or background) is active is exactly the
+// interleaving that corrupted the activity panel and made tasks feel stuck.
+//
+// Returns whether the turn actually ran. Callers that mutate state before
+// resuming (the continuation worker's markContinued, the idle sweep's nudge
+// bump) MUST gate that mutation on a true return — otherwise a busy-skip leaves
+// the state changed but the work undone (lost continuation / false auto-close).
+// See plans/2026-08-03-nomos-chat-changes-review.md P0/P1.
+func (a *agent) resumeSession(ctx context.Context, sessionID, note string) bool {
+	if !a.gate.acquire(sessionID, 0) {
+		slog.Info("nomos: turn already active, skipping background resume", "session", sessionID)
+		return false
+	}
+	defer a.gate.release(sessionID)
+
 	placeholder, _ := json.Marshal(map[string]any{
 		"role": "assistant",
 		"text": "",
@@ -242,7 +278,7 @@ func (a *agent) resumeSession(ctx context.Context, sessionID, note string) {
 		if attempt > 0 {
 			select {
 			case <-cctx.Done():
-				return
+				return true // a turn ran on an earlier attempt; consume, don't re-loop
 			case <-time.After(time.Duration(2<<attempt) * time.Second): // 4s, 8s
 			}
 		}
@@ -314,9 +350,10 @@ func (a *agent) resumeSession(ctx context.Context, sessionID, note string) {
 			// No placeholder was inserted (rare), save directly.
 			a.store.saveMessage(context.Background(), sessionID, "assistant", body)
 		}
-		return // do not call persist() again — already persisted above
+		return true // do not call persist() again — already persisted above
 	}
 	persist() // final state — same row, updated one last time with the concluding text
+	return true
 }

 // buildContinuationNote frames the finished execution for the model: what
--- a/cmd/nomos/continue_test.go
+++ b/cmd/nomos/continue_test.go
@@ -1,6 +1,11 @@
 package main

-import "testing"
+import (
+	"context"
+	"testing"
+
+	"github.com/google/uuid"
+)

 func TestExtractExecutionIDs(t *testing.T) {
 	// Real tool-result phrasings that should yield an execution id.
@@ -37,3 +42,37 @@ func TestExtractExecutionIDs(t *testing.T) {
 		t.Errorf("expected de-dup to 1 id, got %v", ids)
 	}
 }
+
+// TestResumeSession_SkipsWhenBusy guards the P0 fix
+// (plans/2026-08-03-nomos-chat-changes-review.md): resumeSession must skip —
+// return false, body never executed — when a turn is already active for the
+// session. continueSession relies on this so it only marks a continuation
+// "continued" after a turn really ran (otherwise the result is lost: marked
+// continued, never re-queued by pendingContinuations).
+//
+// A minimal agent with only a gate is enough: if the body ever ran, chatWith
+// would dereference the nil provider and panic. Returning false cleanly proves
+// the body was skipped.
+func TestResumeSession_SkipsWhenBusy(t *testing.T) {
+	a := &agent{gate: newTurnGate()}
+	if !a.gate.acquire("sess", 0) {
+		t.Fatal("precondition: initial acquire should succeed on a free session")
+	}
+	ran := a.resumeSession(context.Background(), "sess", "note")
+	if ran {
+		t.Fatal("resumeSession must return false (skip) while a turn is active for the session")
+	}
+}
+
+// TestContinueSession_DefersWhenBusy guards the other half of P0: when the
+// session is busy, continueSession defers (leaves the execution pending for the
+// next worker tick) instead of running or marking it. It must return cleanly
+// without reaching resumeSession's body (nil provider → panic) or markContinued.
+func TestContinueSession_DefersWhenBusy(t *testing.T) {
+	a := &agent{gate: newTurnGate()}
+	if !a.gate.acquire("sess", 0) {
+		t.Fatal("precondition: initial acquire should succeed on a free session")
+	}
+	p := pendingContinuation{ExecID: uuid.New(), SessionID: "sess", Status: "completed"}
+	a.continueSession(context.Background(), p) // must not panic; must not run/mark
+}
--- a/cmd/nomos/main.go
+++ b/cmd/nomos/main.go
@@ -186,12 +186,22 @@ func handleChat(w http.ResponseWriter, r *http.Request, a *agent, st *store) {
 		return
 	}

-	// Empty message with an existing session = reconnect/resume. The
-	// frontend sends this after a dropped SSE stream to re-establish the
-	// connection and catch up on any auto-continuation work that happened
-	// while disconnected. Route into resumeSession so the agent sees a
-	// system note and reports current state.
+	// Empty message with an existing session = reconnect/resume. This path is
+	// defensive now — the frontend (post F2) recovers a dropped SSE via the
+	// poller + terminal task.status clearing, and no longer POSTs empty
+	// messages. If a client ever does, route into resumeSession so the agent
+	// reports current state — but SKIP a terminal session (done/failed/
+	// abandoned): there's nothing to resume, and running a "report state"
+	// turn there is just a spare turn the operator never asked for (P2.1).
 	if req.Message == "" && req.SessionID != "" {
+		if sess, err := st.getSession(context.Background(), req.SessionID); err == nil {
+			switch sess.Status {
+			case "done", "failed", "abandoned":
+				slog.Info("nomos: reconnect skipped — session already terminal", "session", req.SessionID, "status", sess.Status)
+				w.WriteHeader(202)
+				return
+			}
+		}
 		slog.Info("nomos: reconnect", "session", req.SessionID)
 		safego.Go("nomos:reconnect:"+req.SessionID, func() {
 			base := "[System: the operator's connection was re-established. The task may have progressed in the background.]"
@@ -200,7 +210,7 @@ func handleChat(w http.ResponseWriter, r *http.Request, a *agent, st *store) {
 		})
 		// Return 202 so the frontend doesn't try to consume an SSE stream
 		// from this POST — resumeSession writes to the DB directly and
-		// the poller (already running from handleDisconnect) picks it up.
+		// the poller picks it up.
 		w.WriteHeader(202)
 		return
 	}
@@ -270,6 +280,29 @@ func handleChat(w http.ResponseWriter, r *http.Request, a *agent, st *store) {

 	sseEvent(w, flusher, agentEvent{Type: "session", Data: sessionID, SessionID: sessionID})

+	// F1 (plan 2026-08-03): serialize turns per session. The user message is
+	// already persisted above, so even if we can't run this turn right now it
+	// isn't lost. Wait briefly for a finishing background turn (continuation /
+	// resume) so the common case is seamless; if one is still running after
+	// that, tell the operator to retry rather than spawning a second
+	// concurrent turn (the interleaving this gate exists to prevent). On
+	// success the permit is held until this handler returns (stream + post-
+	// processing done); background resumeSession callers skip while it's held.
+	const turnWait = 5 * time.Second
+	if !a.gate.acquire(sessionID, turnWait) {
+		slog.Info("nomos: turn already active, deferring operator message", "session", sessionID)
+		sseEvent(w, flusher, agentEvent{
+			Type: "error",
+			Data: "Nomos is still finishing a previous step. Your message was saved — give it a moment to finish, then send it again.",
+		})
+		sseEvent(w, flusher, agentEvent{Type: "done", Data: map[string]any{
+			"session_id": sessionID,
+			"error":      true,
+		}, SessionID: sessionID})
+		return
+	}
+	defer a.gate.release(sessionID)
+
 	toolCalls := []map[string]any{}
 	// P3: accumulate per-iteration reasoning instead of overwriting with
 	// the final `text` event. The agent loop emits a `text` event for each
--- a/cmd/nomos/turngate.go
+++ b/cmd/nomos/turngate.go
@@ -0,0 +1,90 @@
+package main
+
+import (
+	"sync"
+	"time"
+)
+
+// turnGate enforces at most one in-flight agent turn per session.
+//
+// Why this exists (plan 2026-08-03, F1): handleChat runs a turn in the HTTP
+// request goroutine, and every "resume" path (the empty-message reconnect,
+// the auto-continuation worker, the idle sweep, answer-question, the /resume
+// endpoint) launches ANOTHER goroutine running a full turn. Nothing prevented
+// two turns for the SAME session at once, so a network blip that triggered a
+// reconnect would spawn a duplicate resumeSession while the original turn was
+// still alive — their tool calls interleaved on the wire and in the persisted
+// transcript, which is the root cause behind the "parallel/nesting/sequence
+// is off" and "task didn't end / flaky" reports.
+//
+// Model: one permit (buffered-1 channel seeded with a single token) per
+// session id. Acquiring consumes the token; releasing puts it back.
+//   - Background/best-effort callers (resumeSession and everything it backs)
+//     use a non-blocking acquire and SKIP when busy — a duplicate nudge while a
+//     turn is already running adds nothing, and the continuation/idle tickers
+//     will retry on their own.
+//   - The live chat path (an operator message) waits briefly for a finishing
+//     background turn, then bails with an actionable error if still busy — see
+//     handleChat.
+//
+// The permits map grows one entry per session id seen. For this single-agent
+// homelab process that set is small and bounded by real sessions; cleanup is
+// intentionally omitted (a sweep would race with acquire/release and the
+// memory is negligible).
+type turnGate struct {
+	mu      sync.Mutex
+	permits map[string]chan struct{}
+}
+
+func newTurnGate() *turnGate {
+	return &turnGate{permits: make(map[string]chan struct{})}
+}
+
+// permit returns the single token-channel for sessionID, creating and seeding
+// it on first use. Creation is guarded so two concurrent first-callers for the
+// same id share one channel.
+func (g *turnGate) permit(sessionID string) chan struct{} {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	ch, ok := g.permits[sessionID]
+	if !ok {
+		ch = make(chan struct{}, 1)
+		ch <- struct{}{}
+		g.permits[sessionID] = ch
+	}
+	return ch
+}
+
+// acquire takes the session's permit. With wait <= 0 it is non-blocking
+// (returns false immediately if a turn is active). With wait > 0 it blocks up
+// to wait for the permit, returning false on timeout. Every true return MUST
+// be paired with exactly one release.
+func (g *turnGate) acquire(sessionID string, wait time.Duration) bool {
+	ch := g.permit(sessionID)
+	if wait <= 0 {
+		select {
+		case <-ch:
+			return true
+		default:
+			return false
+		}
+	}
+	t := time.NewTimer(wait)
+	defer t.Stop()
+	select {
+	case <-ch:
+		return true
+	case <-t.C:
+		return false
+	}
+}
+
+// release returns the session's permit. Idempotent: a release with no matching
+// acquire (or a double release) is a no-op rather than a blocking send.
+func (g *turnGate) release(sessionID string) {
+	ch := g.permit(sessionID)
+	select {
+	case ch <- struct{}{}:
+	default:
+	}
+}
--- a/cmd/nomos/turngate_test.go
+++ b/cmd/nomos/turngate_test.go
@@ -0,0 +1,114 @@
+package main
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestTurnGate_NonBlockingSkipsWhenBusy(t *testing.T) {
+	g := newTurnGate()
+	if !g.acquire("s1", 0) {
+		t.Fatal("first non-blocking acquire should succeed on a free session")
+	}
+	// A second non-blocking acquire (a background resume) must skip, not queue.
+	if g.acquire("s1", 0) {
+		t.Fatal("second non-blocking acquire should fail while a turn is active")
+	}
+	// A different session is independent.
+	if !g.acquire("s2", 0) {
+		t.Fatal("acquire on a different session should succeed")
+	}
+	g.release("s2")
+	g.release("s1")
+	// After release, the session is free again.
+	if !g.acquire("s1", 0) {
+		t.Fatal("acquire should succeed again after release")
+	}
+	g.release("s1")
+}
+
+func TestTurnGate_BlockingAcquireWaitsForRelease(t *testing.T) {
+	g := newTurnGate()
+	if !g.acquire("s1", 0) {
+		t.Fatal("first acquire should succeed")
+	}
+
+	got := make(chan bool, 1)
+	go func() { got <- g.acquire("s1", 2*time.Second) }()
+
+	select {
+	case <-got:
+		t.Fatal("blocking acquire should wait, not return before release")
+	case <-time.After(50 * time.Millisecond):
+		// expected: still waiting
+	}
+
+	g.release("s1")
+	select {
+	case ok := <-got:
+		if !ok {
+			t.Fatal("blocking acquire should succeed after release")
+		}
+	case <-time.After(time.Second):
+		t.Fatal("blocking acquire did not return after release")
+	}
+	g.release("s1")
+}
+
+func TestTurnGate_BlockingAcquireTimesOut(t *testing.T) {
+	g := newTurnGate()
+	g.acquire("s1", 0) // hold the permit
+
+	start := time.Now()
+	if g.acquire("s1", 60*time.Millisecond) {
+		t.Fatal("acquire should time out while permit is held")
+	}
+	if elapsed := time.Since(start); elapsed < 50*time.Millisecond {
+		t.Fatalf("acquire returned too fast (%v); expected to wait ~60ms", elapsed)
+	}
+	g.release("s1")
+}
+
+// TestTurnGate_SingleFlightConcurrent is the core F1 guarantee: many concurrent
+// background acquirers on the SAME session, exactly one runs at a time. This is
+// the property that prevents two turns interleaving tool calls.
+func TestTurnGate_SingleFlightConcurrent(t *testing.T) {
+	g := newTurnGate()
+	const n = 50
+	var inFlight, maxInFlight int64
+	var runs int64
+	var wg sync.WaitGroup
+	wg.Add(n)
+	start := make(chan struct{})
+	for i := 0; i < n; i++ {
+		go func() {
+			defer wg.Done()
+			<-start
+			if !g.acquire("shared", 0) { // background-style: skip if busy
+				return
+			}
+			defer g.release("shared")
+			cur := atomic.AddInt64(&inFlight, 1)
+			for {
+				m := atomic.LoadInt64(&maxInFlight)
+				if cur <= m || atomic.CompareAndSwapInt64(&maxInFlight, m, cur) {
+					break
+				}
+			}
+			atomic.AddInt64(&runs, 1)
+			time.Sleep(2 * time.Millisecond)
+			atomic.AddInt64(&inFlight, -1)
+		}()
+	}
+	close(start)
+	wg.Wait()
+
+	if maxInFlight != 1 {
+		t.Fatalf("max in-flight turns = %d, want 1 (turns must not overlap)", maxInFlight)
+	}
+	if runs == 0 {
+		t.Fatal("expected at least one turn to run")
+	}
+}