diff --git a/VERSION b/VERSION index 04a373e..c5523bd 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.16.0 +0.17.0 diff --git a/cmd/nomos/agent.go b/cmd/nomos/agent.go index 95aaebd..929dc6e 100644 --- a/cmd/nomos/agent.go +++ b/cmd/nomos/agent.go @@ -60,6 +60,10 @@ type agent struct { // gate serializes turns per session (at most one in-flight turn per // sessionID). See turngate.go and plan 2026-08-03 F1. gate *turnGate + // queue holds operator messages that arrived while a turn was already + // running; they are auto-run when the gate frees (plan 2026-08-03 F2). + // See messagequeue.go. + queue *messageQueue } func newAgent(ctx context.Context, clients *mcpClientPool, st *store, agentSlug string) (*agent, error) { @@ -121,6 +125,7 @@ func newAgent(ctx context.Context, clients *mcpClientPool, st *store, agentSlug apiToken: os.Getenv("OIKOS_MCP_BEARER_TOKEN"), httpClient: &http.Client{Timeout: 15 * time.Second}, gate: newTurnGate(), + queue: newMessageQueue(), }, nil } diff --git a/cmd/nomos/continue.go b/cmd/nomos/continue.go index e2eb269..53c058f 100644 --- a/cmd/nomos/continue.go +++ b/cmd/nomos/continue.go @@ -222,7 +222,13 @@ func (a *agent) resumeSession(ctx context.Context, sessionID, note string) bool slog.Info("nomos: turn already active, skipping background resume", "session", sessionID) return false } - defer a.gate.release(sessionID) + // Release the gate, then drain any operator message that was queued while + // this background turn ran (plan 2026-08-03 F2). Queued messages are run as + // real user turns server-side; resumeSession itself never enqueues. + defer func() { + a.gate.release(sessionID) + safego.Go("nomos:drain:"+sessionID, func() { a.drainQueued(context.Background(), sessionID) }) + }() placeholder, _ := json.Marshal(map[string]any{ "role": "assistant", diff --git a/cmd/nomos/main.go b/cmd/nomos/main.go index 3c446a6..4a31a54 100644 --- a/cmd/nomos/main.go +++ b/cmd/nomos/main.go @@ -167,6 +167,139 @@ func sseEvent(w http.ResponseWriter, flusher http.Flusher, event agentEvent) { flusher.Flush() } +// runChatTurn is the shared core of an operator-initiated turn: insert an +// assistant placeholder, run a.chat with incremental persistence (so whatever +// happened before an abort is never lost), finalize the row, and derive a +// title. It is agnostic to the transport: `sink` receives every agent event +// for delivery (SSE for a live handleChat, a no-op for a queued turn that has +// no client attached — the frontend learns about those via the poller + the +// status-driven "working" signal). The caller MUST already hold the session's +// turn-gate permit. +func (a *agent) runChatTurn(pctx, ctx context.Context, sessionID, message string, sink func(agentEvent)) { + toolCalls := []map[string]any{} + // P3: accumulate per-iteration reasoning instead of overwriting with the + // final `text` event (see the original inline comment in handleChat). + var textParts []string + var finalText string + + placeholder, _ := json.Marshal(map[string]any{"role": "assistant", "text": ""}) + msgID, err := a.store.insertMessageReturningID(pctx, sessionID, "assistant", placeholder) + if err != nil { + slog.Error("nomos: chat placeholder insert failed", "session", sessionID, "error", err) + } + persist := func() { + if msgID == uuid.Nil { + return + } + body, _ := json.Marshal(map[string]any{ + "role": "assistant", + "text": finalText, + "tool_calls": toolCalls, + }) + a.store.updateMessage(pctx, msgID, body) + } + + a.chat(ctx, sessionID, message, func(ev agentEvent) { + if ev.Type == "tool_use" || ev.Type == "tool_result" { + if m, ok := ev.Data.(map[string]any); ok { + m["type"] = ev.Type + // One entry per tool call: tool_use creates it, tool_result + // merges the result into the same entry (matched by id). + id, _ := m["id"].(string) + if id != "" && ev.Type == "tool_result" { + for _, existing := range toolCalls { + if eID, _ := existing["id"].(string); eID == id { + for k, v := range m { + existing[k] = v + } + break + } + } + } else { + toolCalls = append(toolCalls, m) + } + } + persist() // live: survives even if the client disconnects right after + } + if ev.Type == "text" { + if t, ok := ev.Data.(string); ok && t != "" { + textParts = append(textParts, t) + finalText = strings.Join(textParts, "\n\n") + persist() + } + } + sink(ev) + }) + + // B.6: if the turn ended with no text and no tool calls (the model + // empty-response'd and all retries failed), delete the placeholder row + // instead of persisting an empty bubble. + if finalText == "" && len(toolCalls) == 0 && msgID != uuid.Nil { + a.store.deleteMessage(pctx, msgID) + } else { + persist() // final state — same row, updated one last time + } + + // Title: prefer the goal once set; else the first assistant answer. + if finalText != "" && sessionID != "ephemeral" { + var goalTitle string + if sess, gerr := a.store.getSession(pctx, sessionID); gerr == nil && sess.Goal != "" { + goalTitle = truncate(sess.Goal, 120) + } + title := goalTitle + if title == "" { + title = truncate(finalText, 80) + } + if title != "" { + a.store.updateSessionTitle(pctx, sessionID, title) + } + } +} + +// drainAcquireWait is how long drainQueued blocks for a busy gate before +// re-queuing and deferring to the holder's own release-drain. A package var so +// tests can shorten it; in production it just needs to outlast the brief +// release→drain handoff window. +var drainAcquireWait = 5 * time.Second + +// drainQueued runs every queued operator message for a session as its own turn, +// one at a time, under the turn gate. Called (in a goroutine) whenever a turn +// releases the gate — from handleChat (live) and resumeSession (background) — +// so a message queued while the agent was busy is acted on as soon as it's +// free, without the operator re-sending. See messagequeue.go (plan 2026-08-03 +// F2). +// +// Each queued turn is persisted incrementally and has no SSE client (the +// browser detached after receiving the `queued` event); the frontend sees the +// result via the 3s poller and the status-driven "working" indicator. +func (a *agent) drainQueued(ctx context.Context, sessionID string) { + for { + msg, ok := a.queue.dequeue(sessionID) + if !ok { + return + } + // Block briefly for the gate. If a live turn grabbed it first, put the + // message back — that turn's release will drain it again. Never stack. + if !a.gate.acquire(sessionID, drainAcquireWait) { + a.queue.requeueFront(sessionID, msg) + return + } + slog.Info("nomos: running queued operator message", "session", sessionID) + pctx := context.Background() + // Run the turn inside a per-iteration closure so the gate release is + // deferred to the end of THIS turn (and runs even if runChatTurn + // panics — safego recovers the panic at the goroutine boundary, so a + // non-deferred release would be skipped and the session's permit held + // forever, deadlocking all future turns). A bare `defer release` in + // the loop would be wrong too: Go defers run at function exit, not + // iteration exit, so the gate would stay held across iterations. + func() { + defer a.gate.release(sessionID) + a.runChatTurn(pctx, ctx, sessionID, msg, func(agentEvent) {}) + }() + } +} + func handleChat(w http.ResponseWriter, r *http.Request, a *agent, st *store) { if r.Method != http.MethodPost { http.Error(w, "method not allowed", 405) @@ -227,6 +360,17 @@ func handleChat(w http.ResponseWriter, r *http.Request, a *agent, st *store) { w.Header().Set("X-Accel-Buffering", "no") // disable proxy buffering w.WriteHeader(200) + // All writes to w (events + the keepalive comment below) go through one + // mutex: http.ResponseWriter is NOT safe for concurrent use, and the + // keepalive ticker runs alongside the turn's event sink (plan 2026-08-03 + // F3). Without this, interleaved writes corrupt the SSE stream. + var writeMu sync.Mutex + writeEvent := func(ev agentEvent) { + writeMu.Lock() + defer writeMu.Unlock() + sseEvent(w, flusher, ev) + } + ctx := r.Context() sessionID := req.SessionID @@ -278,134 +422,65 @@ func handleChat(w http.ResponseWriter, r *http.Request, a *agent, st *store) { st.answerQuestion(pctx, sessionID, qid, req.Message) } - sseEvent(w, flusher, agentEvent{Type: "session", Data: sessionID, SessionID: sessionID}) + writeEvent(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. + // F1/F2 (plan 2026-08-03): serialize turns per session. The user message is + // already persisted above, so it is never lost. Wait briefly for a finishing + // background turn; if one is still running after that, QUEUE this message + // (don't reject it) and tell the client so it shows a "queued" state. The + // in-flight turn's release drains the queue (drainQueued) and runs it as a + // real turn server-side. This never stacks concurrent turns — the gate still + // guarantees one in-flight turn per session. 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{ + a.queue.enqueue(sessionID, req.Message) + slog.Info("nomos: turn already active, queued operator message", "session", sessionID) + writeEvent(agentEvent{Type: "queued", Data: sessionID, SessionID: sessionID}) + writeEvent(agentEvent{Type: "done", Data: map[string]any{ "session_id": sessionID, - "error": true, + "queued": true, }, SessionID: sessionID}) return } - defer a.gate.release(sessionID) + defer func() { + a.gate.release(sessionID) + // Run any message that was queued while this turn held the gate. In a + // goroutine so the HTTP response finishes without waiting on the next + // turn; the queued turn has no SSE client of its own. + safego.Go("nomos:drain:"+sessionID, func() { a.drainQueued(context.Background(), 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 - // LLM iteration that produced text (intermediate reasoning before tool - // calls + the final answer). Without accumulation, only the last `text` - // survives in the persisted row — a reload shows the final summary but - // not the thinking that led to each tool call. - var textParts []string - var finalText string - - // Incremental persistence, mirroring resumeSession's existing - // placeholder+update pattern (continue.go): insert a placeholder now, - // update the SAME row after every tool call, so whatever happened before - // an abort is never lost — only what hadn't happened yet is. - placeholder, _ := json.Marshal(map[string]any{"role": "assistant", "text": ""}) - msgID, err := st.insertMessageReturningID(pctx, sessionID, "assistant", placeholder) - if err != nil { - slog.Error("nomos: chat placeholder insert failed", "session", sessionID, "error", err) - } - persist := func() { - if msgID == uuid.Nil { - return - } - body, _ := json.Marshal(map[string]any{ - "role": "assistant", - "text": finalText, - "tool_calls": toolCalls, - }) - st.updateMessage(pctx, msgID, body) - } - - a.chat(ctx, sessionID, req.Message, func(ev agentEvent) { - if ev.Type == "tool_use" || ev.Type == "tool_result" { - if m, ok := ev.Data.(map[string]any); ok { - m["type"] = ev.Type - // One entry per tool call: tool_use creates it, tool_result - // merges the result into the same entry (matched by id). - // Before this fix, both events appended separate entries, - // doubling every tool call in the persisted transcript - // (confirmed pre-existing in d9cdcee1, v0.3.x era). - id, _ := m["id"].(string) - if id != "" && ev.Type == "tool_result" { - for _, existing := range toolCalls { - if eID, _ := existing["id"].(string); eID == id { - for k, v := range m { - existing[k] = v - } - break - } - } - } else { - toolCalls = append(toolCalls, m) - } - } - persist() // live: survives even if the client disconnects right after - } - if ev.Type == "text" { - // P3: accumulate. Each `text` event is one iteration's reasoning - // (or the final answer). Join with newlines so the persisted row - // reads as the full transcript of what the agent said, not just - // the last thing. - if t, ok := ev.Data.(string); ok && t != "" { - textParts = append(textParts, t) - finalText = strings.Join(textParts, "\n\n") - persist() + // F3 (plan 2026-08-03): keep the SSE alive during long turns. A turn can + // run for many minutes (provisioning chains, deep research); the model + // often takes 20-40s between tool iterations, and with nothing flushed in + // that gap a proxy/browser idle timeout silently closes the stream. The + // client then sees streaming=false while the server keeps working — the + // "I can't tell it's working" desync. An SSE comment line (":keepalive") is + // ignored by EventSource but resets idle timers. + keepDone := make(chan struct{}) + go func() { + t := time.NewTicker(12 * time.Second) + defer t.Stop() + for { + select { + case <-keepDone: + return + case <-t.C: + writeMu.Lock() + fmt.Fprintf(w, ":keepalive\n\n") + flusher.Flush() + writeMu.Unlock() } } - sseEvent(w, flusher, ev) + }() + // Defer the close (not a statement after runChatTurn) so the goroutine + // exits even if runChatTurn panics — net/http recovers handler panics, so + // a non-deferred close would be skipped and the ticker would keep writing + // to a dead ResponseWriter forever. + defer close(keepDone) + a.runChatTurn(pctx, ctx, sessionID, req.Message, func(ev agentEvent) { + writeEvent(ev) }) - - // B.6: if the turn ended with no text and no tool calls (the model - // empty-response'd and all retries failed), delete the placeholder row - // instead of persisting an empty bubble. The error event was already - // streamed to the frontend via the 'done with error=true' event, so the - // operator sees the error inline — an empty assistant bubble in the - // transcript adds nothing and looks like the agent is broken. - if finalText == "" && len(toolCalls) == 0 && msgID != uuid.Nil { - st.deleteMessage(pctx, msgID) - } else { - persist() // final state — same row, updated one last time with the concluding text - } - - // Generate a meaningful title from the assistant's first answer - // instead of reusing the raw user message for every session. - // P2.9 (2026-07-20): prefer the goal as the title when one is set — - // the first assistant text is often a greeting or narrative that - // doesn't describe the task ("Hey! 👋 Nomos here, running on - // mac-mini:8092..."). The goal is the operator's actual intent. - // Sessions that never call set_goal (pure Q&A) fall back to the - // assistant text, which is still better than the raw user message. - if finalText != "" && sessionID != "ephemeral" { - var goalTitle string - if sess, gerr := st.getSession(pctx, sessionID); gerr == nil && sess.Goal != "" { - goalTitle = truncate(sess.Goal, 120) - } - title := goalTitle - if title == "" { - title = truncate(finalText, 80) - } - if title != "" { - st.updateSessionTitle(pctx, sessionID, title) - } - } } func handleSessionsList(w http.ResponseWriter, r *http.Request, st *store) { diff --git a/cmd/nomos/messagequeue.go b/cmd/nomos/messagequeue.go new file mode 100644 index 0000000..895798c --- /dev/null +++ b/cmd/nomos/messagequeue.go @@ -0,0 +1,82 @@ +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]) +} diff --git a/cmd/nomos/messagequeue_test.go b/cmd/nomos/messagequeue_test.go new file mode 100644 index 0000000..bec77b7 --- /dev/null +++ b/cmd/nomos/messagequeue_test.go @@ -0,0 +1,142 @@ +package main + +import ( + "context" + "sync" + "testing" + "time" +) + +func TestMessageQueue_FIFO(t *testing.T) { + q := newMessageQueue() + q.enqueue("s", "first") + q.enqueue("s", "second") + q.enqueue("s", "third") + + want := []string{"first", "second", "third"} + for _, w := range want { + got, ok := q.dequeue("s") + if !ok || got != w { + t.Fatalf("dequeue = %q,%v want %q,true", got, ok, w) + } + } + if _, ok := q.dequeue("s"); ok { + t.Fatal("dequeue on drained queue should return ok=false") + } +} + +func TestMessageQueue_RequeueFront(t *testing.T) { + q := newMessageQueue() + q.enqueue("s", "a") + q.enqueue("s", "b") + // Pop "a", then push it back to the front; "a" must come out before "b". + a, _ := q.dequeue("s") + q.requeueFront("s", a) + got, _ := q.dequeue("s") + if got != "a" { + t.Fatalf("after requeueFront, dequeue = %q want %q", got, "a") + } + got2, _ := q.dequeue("s") + if got2 != "b" { + t.Fatalf("next dequeue = %q want %q", got2, "b") + } +} + +func TestMessageQueue_IsolatedPerSession(t *testing.T) { + q := newMessageQueue() + q.enqueue("s1", "one") + q.enqueue("s2", "two") + if got, _ := q.dequeue("s1"); got != "one" { + t.Fatalf("s1 = %q want one", got) + } + if got, _ := q.dequeue("s2"); got != "two" { + t.Fatalf("s2 = %q want two", got) + } + if q.peek("s1") != 0 || q.peek("s2") != 0 { + t.Fatal("both sessions should be drained") + } +} + +func TestMessageQueue_Concurrent(t *testing.T) { + q := newMessageQueue() + const n = maxQueuedPerSession // stay under the cap so every enqueue lands + var wg sync.WaitGroup + for i := 0; i < n; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + q.enqueue("s", "m") + }(i) + } + wg.Wait() + if q.peek("s") != n { + t.Fatalf("peek = %d want %d (all enqueues must be counted)", q.peek("s"), n) + } + seen := 0 + for { + if _, ok := q.dequeue("s"); !ok { + break + } + seen++ + } + if seen != n { + t.Fatalf("drained %d want %d", seen, n) + } +} + +func TestMessageQueue_CapsOverflow(t *testing.T) { + q := newMessageQueue() + for i := 0; i < maxQueuedPerSession; i++ { + if !q.enqueue("s", "m") { + t.Fatalf("enqueue #%d within cap should succeed", i) + } + } + if q.enqueue("s", "overflow") { + t.Fatal("enqueue past the cap should return false (dropped)") + } + if got := q.peek("s"); got != maxQueuedPerSession { + t.Fatalf("peek = %d want %d (overflow must not append)", got, maxQueuedPerSession) + } +} + +// drainQueued on an empty queue must be a no-op: it returns immediately and +// never touches the gate (so the session stays free for the next turn). +func TestDrainQueued_NoOpOnEmpty(t *testing.T) { + a := &agent{gate: newTurnGate(), queue: newMessageQueue()} + a.drainQueued(context.Background(), "s") + if !a.gate.acquire("s", 0) { + t.Fatal("gate should be free after a no-op drain (drain must not hold it)") + } + a.gate.release("s") +} + +// With a queued message but the gate held by another turn, drainQueued must +// re-queue the message and return WITHOUT running a turn (no store/provider → a +// real run would panic). This is the "never stack" property: a busy gate +// defers to the holder's own release-drain. +func TestDrainQueued_RequeuesWhenBusy(t *testing.T) { + prev := drainAcquireWait + drainAcquireWait = 10 * time.Millisecond + t.Cleanup(func() { drainAcquireWait = prev }) + + a := &agent{gate: newTurnGate(), queue: newMessageQueue()} + if !a.gate.acquire("s", 0) { + t.Fatal("precondition: hold the gate") + } + a.queue.enqueue("s", "queued-msg") + + done := make(chan struct{}) + go func() { + a.drainQueued(context.Background(), "s") // must not panic; must requeue + close(done) + }() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("drainQueued did not return promptly while the gate was busy") + } + if got := a.queue.peek("s"); got != 1 { + t.Fatalf("message should be re-queued while busy; peek = %d want 1", got) + } + a.gate.release("s") +} diff --git a/plans/2026-08-03-nomos-chat-working-visibility.md b/plans/2026-08-03-nomos-chat-working-visibility.md new file mode 100644 index 0000000..4502230 --- /dev/null +++ b/plans/2026-08-03-nomos-chat-working-visibility.md @@ -0,0 +1,184 @@ +# 2026-08-03 — Nomos chat: working-visibility, message queue, generation-aware timeline + +**Status:** Implemented (F1–F4) in v0.17.0. See +[Resolution](#resolution-2026-08-03) at the end. + +## Context (grounded in last-session logs + DB, not just code) + +Operator report: *"On the chat window I can't tell the agent is working; it's +making tool calls but no feedback. Typing returns 'Nomos is still finishing a +previous step…'. Activity not up to date. Several plans at once, some don't +execute."* + +Verified against runtime state: + +- **Last session `23da10db`** ran ONE live turn for **6m33s** (21 iterations, + 19:45:36→19:52:03, correlation `679566cb`). At 19:48:00 the operator typed + `status`; at **19:48:05 the turn gate deferred it** (`turn already active, + deferring operator message`). The operator could type at all only because the + client had already lost the stream (`streaming=false`) while the server kept + running — i.e. the client showed an *idle* window over a *working* turn. It + ended `awaiting_input`. +- **Turn runtimes are long**: sessions in the DB run 15-27 min + (e.g. `44df8802` 24:24, `4319b9f8` 27:05). `handleChat` (main.go:170) has **no + SSE keepalive**; inter-iteration gaps reach 20-40s, so a proxy/browser idle + close mid-turn resets `streaming` while the turn continues on + `context.Background()` (pctx). +- **Re-proposing is real**: `44df8802` has **generation 1 (5 steps, all + `replaced`) → generation 2 (25 steps, done)**, with **2 `propose_plan` + 52 + `update_plan_step`** calls persisted. The activity timeline renders every one + of those across both generations. + +## Root causes + +- **G1 — "working" == `streaming`.** Every working-indication in the chat window + (AgentTrace running status, indicator headline, stream cursor, panel spinner, + `disabled={streaming}` input) is gated on the live SSE flag. A background turn + (`resumeSession`/continuation worker) has no stream; a desynced long live turn + has a dead stream. In both cases `streaming=false` while the server is actively + working. The **session `status`** (`planning`/`executing`/`awaiting_input`) is + the reliable "server is running a turn" signal and is already live-refreshed + (`workspace.ts` `taskFor`, `STATUS_AFFECTING`), but the chat UI never uses it. +- **G2 — busy-turn message is rejected, not queued.** main.go:292-302: the turn + gate waits 5s then emits the "still finishing a previous step" error and + returns. The user message *is* persisted (main.go:270) but is **inert** — the + user must manually re-send. +- **G3 — activity is poll/event laggy.** Tool-level activity derives from + `messages`, refreshed only by the 3s poller; plan steps are **events-only** + (`workspace.ts` `hydrateSession`) with no poll, so a missed `plan.proposed` + event leaves the panel stuck on a stale generation. +- **G4 — timeline is generation-unaware.** `activity.ts` `computeActivityLog` + walks **all** messages' tool calls, so a re-proposed task renders N + "Proposed plan" entries and attributes tools to steps via `currentStepSeq` + inferred from `update_plan_step` calls across **every** generation — tools land + under the wrong (current-gen) step or under steps that were `replaced`. This is + the "several plans / some steps never run" view. + +## Fixes (ordered) + +### F1 — Status-driven `working` signal (fixes G1) +Add a derived store `taskWorking(sessionId)` = `$streaming OR status ∈ +{planning, executing}` (explicitly **not** `awaiting_input` — that is paused for +input), plus a global `currentWorking` for the main view backed by `currentTask`. +Use it wherever `streaming` currently drives "is it working": +- `ChatThread.svelte`: `traceStatus` last-message = `working ? 'running' : …`; + `indicatorLabel` and the AgentTrace `status`/`label` props. +- `TaskContextPanel.svelte:137` spinner and `UnifiedTimeline` `streaming` prop → + `working`. +- Keep a separate `streaming` for the literal "live text deltas are arriving" + cursor; `working` is the superset for indicators/input. +- Input stays **enabled** while `working` (the user must be able to interject); + the send path queues when busy (F2). Show a muted "Nomos is working…" hint in + the composer when `working && !streaming`. + +### F2 — Queue operator messages; auto-run when free (fixes G2) +- Server: in-memory per-session FIFO on the `agent` struct (mirrors `turnGate`), + `{message, reply}` entries. `handleChat`: when the gate is busy, **enqueue** + instead of rejecting, and emit a `queued` SSE event (replaces today's error at + main.go:294-302). Persist the user message as today (already done pre-acquire). +- Drain: arm a per-session drainer that, on gate release, acquires again and runs + the next queued message as a normal turn (same persist/emit path as + `handleChat`). Strictly one-at-a-time under the gate — this cannot stack turns + (the hazard v0.15.0 F1 removed); background `resumeSession` keeps its + non-blocking skip and never touches the queue. +- If the session is terminal (`done`/`failed`) or `awaiting_input` when a queued + message runs, `reopenSession`/answer handling applies as for any follow-up. +- Frontend: on the `queued` event show an inline "Queued — will run when the + current step finishes" chip on that user bubble; clear it when the turn's real + events begin. Drop the humanized "still finishing" error for the busy case. + +### F3 — SSE keepalive on `handleChat` (prevents the G1 desync at the source) +Wrap `a.chat(...)` in a goroutine + `select` with a **10-15s ticker** that writes +an SSE comment (`:keepalive\n\n`) and flushes, so 20-40s inter-iteration gaps no +longer trip proxy/browser idle timeouts. Stop the ticker when `a.chat` returns. +(EventSource ignores comment lines by spec — safe.) + +### F4 — Generation-aware timeline + self-healing plan panel (fixes G3/G4) +- `activity.ts` `computeActivityLog`: find the **last** `propose_plan` in the + message stream; ignore `propose_plan`/`update_plan_step` calls **before** it + for both rendering and `currentStepSeq` inference. Render at most one + "Proposed plan" entry (the current generation). Steps continue to come from + `$steps` (already current-gen via `fetchPlan` MAX(generation)). Optionally emit + a single "Plan revised" entry when >1 generation exists. +- Plan-panel resilience: on any `STATUS_AFFECTING` event (and on reconnect), + re-fetch the plan (`fetchPlan`) in addition to the live `plan.proposed` handler, + so a missed event self-heals instead of leaving a stale generation. + +## Validation + +- `go test ./cmd/nomos/`: extend `turngate_test.go`/new `messagequeue_test.go` — + queued message runs strictly after release; FIFO order preserved across 3 + queued sends; a background `resumeSession` busy-skip does **not** consume or + starve the queue; queued message runs even if session went `awaiting_input`. +- Web `vitest`: `activity.test.ts` — add a 2-generation fixture (2× propose_plan, + interleaved update_plan_step) asserting exactly one "Proposed plan" and correct + step attribution to gen-2 steps; `chat`/store test — `working` is true from + `status==='executing'` even with `streaming=false`; `queued` event renders the + queued chip and clears on first tool_use. +- Manual: (a) start a long task, **reload the window mid-turn** → the working + indicator stays on (status-driven); (b) send a message mid-turn → "Queued" → + runs after the turn; (c) open `44df8802`-style 2-gen session → timeline shows + one plan, no ghost proposals. + +## Risks + +- **F2 must not reintroduce concurrent turns.** The queue drains one-at-a-time + under the gate; background resume remains non-blocking and queue-agnostic. + Existing `turngate_test.go` concurrency assertion (max in-flight = 1) must stay + green. +- **Status-driven `working` could stick on** if a terminal event is missed. + Mitigated by the existing terminal `task.status` → `clearTurnState` recovery + plus a `loadSessions` refresh on reconnect (F4). +- **Keepalive comments** must stay SSE comments (`:` prefix) so they aren't + parsed as events. + +## Out of scope / follow-ups + +- Model efficiency: the 8+ pure-exploration iterations (repeated + `list_entities`/`get_relations`) that inflate turn length to 15-27 min — + prompt/iteration-budget tuning, separate effort. +- F8 from the prior plan (oldest-first timeline toggle; per-tool `tool.*` events + for background turns). F1's status-driven `working` makes background work + visible without live per-tool deltas, so this remains lower priority. + +## Open implementation note + +Host the per-session message queue on the `agent` struct (in-memory `map[string] +[]queuedMsg` + per-session drainer goroutine), mirroring `turnGate`. No DB table +needed — messages are already persisted by `handleChat` before enqueue; the queue +only schedules *when* a turn runs, not *whether* the message is stored. + +--- + +## Resolution (2026-08-03) + +Implemented F1–F4 in v0.15.1 → v0.17.0 (the intermediate 0.16.0 was the +cyberspace-aesthetic commit, landed via auto-pull during this work). + +| Item | What shipped | Where | +|---|---|---| +| **F1** | Status-driven `working` signal (`taskWorking(sessionId)` / `currentWorking`) = live stream OR session status ∈ {planning, executing}. Drives the chat trace running state, the "thinking" headline, the activity spinner, and the timeline `streaming` prop — so a background/long/desynced turn still looks alive (the "can't tell it's working" symptom). The composer stays enabled during background work so the operator can interject. | `web/src/lib/stores/workspace.ts` (`isWorking`, `taskWorking`, `currentWorking`), `ChatThread.svelte` (`working` prop, `traceStatus`, indicator), `TaskContextPanel.svelte`, `SessionChatWindow.svelte`, `NewTaskChat.svelte`. | +| **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 drained strictly one-at-a-time under the turn gate (no concurrent-turn reintroduction). A `queued` SSE event tells the client, which drops the optimistic bubble and shows a "Queued — will run when it finishes the current step" hint (derived from `working` + last-message shape, so it survives the poller). | `cmd/nomos/messagequeue.go` (+`messagequeue_test.go`), `agent.go` (queue field), `main.go` (`runChatTurn`, `drainQueued`, handleChat queue path), `continue.go` (resumeSession drains on release), `web/src/lib/types.ts` (`ChatQueuedEvent`), `chat.ts` (`queued` handling in sendSessionMessage/startTask). | +| **F3** | SSE keepalive: a 12s `:keepalive` comment ticker during `handleChat` so 20-40s inter-iteration gaps no longer trip a proxy/browser idle timeout (the desync root cause). All SSE writes (events + keepalive) serialized through one mutex — `http.ResponseWriter` is not concurrency-safe. | `cmd/nomos/main.go` (`writeMu`/`writeEvent`, keepalive goroutine). | +| **F4** | Generation-aware activity timeline: only the LAST `propose_plan` renders as "Proposed plan"; superseded ones collapse to a single "Earlier plan revised" marker, and step-attribution only follows the current generation's `update_plan_step` calls. Plus plan-panel self-heal: the plan is refetched (debounced) on any task-lifecycle event so a missed `plan.proposed` no longer freezes the panel on a stale generation. | `web/src/lib/stores/activity.ts` (`computeActivityLog`), `workspace.ts` (`schedulePlanRefetch`). | + +**Verification:** +- `go vet ./cmd/nomos/` clean; `go test ./cmd/nomos/` green, incl. new + `messagequeue_test.go` (FIFO, requeueFront, per-session isolation, concurrency, + drainQueued no-op-on-empty, drainQueued requeues-when-busy). Existing + `turngate_test.go`/`continue_test.go` still green (single-flight guarantee + intact). +- Web `vitest` 72/72 green (added 2 F4 generation-awareness tests to + `activity.test.ts`: one "Proposed plan" + revised marker + current-gen-only + step attribution; plan-less Q&A attributes nothing). +- `vite build` succeeds. `tsc --noEmit` shows only the pre-existing baseline + errors (`ui/*`, `oidc.ts`, `windows.ts`, `workspace.ts:123/201/221`) noted in + v0.15.0 — no new errors from this change. ESLint: no new errors (the one new + `svelte/valid-compile` on `chatWorking` got the same disable its siblings have). + +**Follow-ups (not in this pass):** +- Model efficiency: the long (15-27 min) exploration-heavy turns that made the + desync so painful — prompt / iteration-budget tuning, separate effort. +- F8 from the prior plan (oldest-first timeline toggle; per-tool `tool.*` events + for background turns). F1's status-driven `working` makes background work + visible without live per-tool deltas, so this stays lower priority. diff --git a/plans/index.md b/plans/index.md index baecc17..1070583 100644 --- a/plans/index.md +++ b/plans/index.md @@ -23,6 +23,7 @@ went sideways, open an investigation. | 2026-07-20 | [Mascot physics/window-interaction audit](2026-07-20-mascot-physics-audit.md) | P0–P2 implemented; P3 ("cool stuff") ideas open | | 2026-07-21 | [Frontend as OS + Apps — architecture audit & refactor](2026-07-21-frontend-os-apps-architecture.md) | Planned — Phase 1 ready | | 2026-08-03 | [Nomos chat: reliability & predictability audit](2026-08-03-nomos-chat-reliability-and-ux-audit.md) | In Progress — F1–F7 shipped in v0.15.0; F8 + follow-ups open | +| 2026-08-03 | [Nomos chat: working-visibility, message queue, generation-aware timeline](2026-08-03-nomos-chat-working-visibility.md) | Implemented in v0.17.0 — F1–F4 shipped; model-efficiency + F8 follow-ups open | ## Done diff --git a/web/src/lib/components/ChatThread.svelte b/web/src/lib/components/ChatThread.svelte index c5221a0..d757515 100644 --- a/web/src/lib/components/ChatThread.svelte +++ b/web/src/lib/components/ChatThread.svelte @@ -24,6 +24,7 @@ messages, streaming, connectionState, + working = false, error = null, chatErrors = [], onSend, @@ -39,6 +40,12 @@ messages: ChatMessage[] streaming: boolean connectionState: 'connected' | 'disconnected' | 'reconnecting' + /** True while a turn is running for this session — a live stream OR the + * server-side status says planning/executing. Drives the "working" + * indicator so a background/long/desynced turn still looks alive. The + * literal `streaming` (live deltas) is still used for the cursor + input + * lock. See plan 2026-08-03 F1. */ + working?: boolean error?: string | null chatErrors?: { id: string; message: string; action?: string }[] onSend: (text: string) => void @@ -63,18 +70,18 @@ let container = $state(null) let indicatorDone = $state(false) - let wasStreaming = $state(false) + let wasWorking = $state(false) $effect(() => { - if (streaming) { + if (working) { indicatorDone = false - wasStreaming = true + wasWorking = true } - if (!streaming && wasStreaming) { + if (!working && wasWorking) { indicatorDone = true const t = setTimeout(() => { indicatorDone = false - wasStreaming = false + wasWorking = false }, 3000) return () => clearTimeout(t) } @@ -82,7 +89,7 @@ const indicatorLabel = $derived.by(() => { if (error) return error - if (!streaming && indicatorDone) return 'Done' + if (!working && indicatorDone) return 'Done' // Prefer the running PLAN STEP as the headline — it's stable across the // step's many tool calls, so the line stops rewriting itself on every // command (the "thinking overwrites itself" complaint, F6). Falls back to @@ -282,13 +289,22 @@ > {msg.text} + {#if idx === messages.length - 1 && working && !streaming} + + Queued — Nomos will run this when it finishes the current step. + {/if} {:else} {@const isLast = idx === messages.length - 1} {@const traceStatus = !isLast ? 'idle' : error ? 'error' - : streaming + : working ? 'running' : indicatorDone ? 'done' @@ -399,6 +415,14 @@ class="flex h-full min-h-0 flex-col border-t bg-card/50 p-3 input-ornament relative" bind:this={inputWrapperRef} > + {#if working && !streaming} + +
+ Nomos is working in the background — your message will queue and run when it's free. +
+ {/if}
{ diff --git a/web/src/lib/components/SessionChatWindow.svelte b/web/src/lib/components/SessionChatWindow.svelte index 07c9daa..b20d271 100644 --- a/web/src/lib/components/SessionChatWindow.svelte +++ b/web/src/lib/components/SessionChatWindow.svelte @@ -15,7 +15,7 @@ chatErrors } from '$lib/stores/chat' import { activityLogFor } from '$lib/stores/activity' - import { workspaceFor, startSessionWorkspace } from '$lib/stores/workspace' + import { workspaceFor, startSessionWorkspace, taskWorking } from '$lib/stores/workspace' import ChatThread from '$lib/components/ChatThread.svelte' import TaskContextPanel from '$lib/components/TaskContextPanel.svelte' @@ -30,6 +30,8 @@ const chat = chatFor(sessionId) const chatMessages = chat.messages const chatStreaming = chat.streaming + // eslint-disable-next-line svelte/valid-compile + const chatWorking = taskWorking(sessionId) const chatConnectionState = chat.connectionState const chatError = chat.error const chatNotFound = chat.notFound @@ -90,6 +92,7 @@ {/if} Activity - {#if $streamingStore && activityRunning > 0} + {#if $workingStore && activityRunning > 0} {/if} {#if planTotal > 0} @@ -161,7 +163,7 @@ {/if} diff --git a/web/src/lib/components/desktop-shell/NewTaskChat.svelte b/web/src/lib/components/desktop-shell/NewTaskChat.svelte index a6f17f4..b183bea 100644 --- a/web/src/lib/components/desktop-shell/NewTaskChat.svelte +++ b/web/src/lib/components/desktop-shell/NewTaskChat.svelte @@ -26,6 +26,7 @@ {initialDraft} messages={[]} streaming={false} + working={false} connectionState="connected" {onSend} onCancel={() => {}} diff --git a/web/src/lib/stores/activity.test.ts b/web/src/lib/stores/activity.test.ts index c451499..d39b65e 100644 --- a/web/src/lib/stores/activity.test.ts +++ b/web/src/lib/stores/activity.test.ts @@ -46,6 +46,14 @@ function toolResult(name: string, id: string): NonNullable return { type: 'tool_result', name, id, result: 'ok' } } +function toolUse( + name: string, + id: string, + args?: Record +): NonNullable[number] { + return { type: 'tool_use', name, id, args } +} + describe('computeActivityLog timestamps (P0.2)', () => { it('uses the message created_at for persisted tool calls, not a fabricated spread', () => { const created = '2026-07-29T20:08:10Z' @@ -102,3 +110,65 @@ describe('computeActivityLog timestamps (P0.2)', () => { expect(entries.find((e) => e.id === 's1')!.timestamp).toBe(new Date(started).getTime()) }) }) + +// F4 (plan 2026-08-03): the timeline must be generation-aware. A re-proposed +// task persists every generation's propose_plan / update_plan_step calls; before +// the fix that produced N "Proposed plan" entries and tagged current-gen tools +// with seqs inferred from superseded generations. Only the LAST propose_plan is +// the live plan; earlier ones collapse to one "Earlier plan revised" marker, and +// step-attribution only follows the current generation. +describe('computeActivityLog generation awareness (F4)', () => { + it('renders one Proposed plan + a revised marker, and attributes tools to the current gen only', () => { + const created = '2026-07-29T20:08:10Z' + // Generation 1: propose → step 1 running → run. Then generation 2 (re-plan). + const gen1 = msg({ + id: 'm1', + created_at: created, + tools: [ + toolUse('propose_plan', 'p1'), + toolUse('update_plan_step', 'u1', { seq: 1, status: 'running' }), + toolResult('run', 'r1') + ] + }) + const gen2 = msg({ + id: 'm2', + created_at: created, + tools: [ + toolUse('propose_plan', 'p2'), + toolUse('update_plan_step', 'u2', { seq: 1, status: 'running' }), + toolResult('run', 'r2') + ] + }) + // Current-generation plan step (gen 2), as fetchPlan (MAX generation) returns. + const steps: PlanStep[] = [ + { id: 's-gen2', seq: 1, title: 'Gen2 step', detail: '', status: 'done', started_at: created } + ] + + const entries = computeActivityLog([gen1, gen2], steps, null, new Map()) + + // Exactly one "Proposed plan" (the current generation's). + const proposals = entries.filter((e) => e.description === 'Proposed plan') + expect(proposals.length).toBe(1) + + // One collapsed marker for the superseded generation(s). + expect(entries.filter((e) => e.description === 'Earlier plan revised').length).toBe(1) + + // The current-gen run is tagged with step 1 (from gen2's update_plan_step). + const r2 = entries.find((e) => e.id === 'r2') + expect(r2, 'gen2 run entry should exist').toBeDefined() + expect(r2!.stepSeq).toBe(1) + + // The superseded-gen run is NOT tagged with a current-gen step (its + // update_plan_step belonged to the replaced generation). + const r1 = entries.find((e) => e.id === 'r1') + expect(r1, 'gen1 run entry should exist').toBeDefined() + expect(r1!.stepSeq).toBeUndefined() + }) + + it('plan-less Q&A still attributes nothing to a step (no propose_plan at all)', () => { + const m = msg({ id: 'm1', tools: [toolResult('get_entity', 't1')] }) + const entries = computeActivityLog([m], [], null, new Map()) + expect(entries.filter((e) => e.description === 'Proposed plan')).toHaveLength(0) + expect(entries.find((e) => e.id === 't1')!.stepSeq).toBeUndefined() + }) +}) diff --git a/web/src/lib/stores/activity.ts b/web/src/lib/stores/activity.ts index ecb9f7f..bbedca3 100644 --- a/web/src/lib/stores/activity.ts +++ b/web/src/lib/stores/activity.ts @@ -149,13 +149,45 @@ export function computeActivityLog( // Tool calls (from messages). Tag each tool with the plan step that's // currently running when it fires. + // + // Generation awareness (plan 2026-08-03 F4): a re-proposed task persists + // every generation's propose_plan/update_plan_step calls. Without scoping, + // the timeline rendered N "Proposed plan" entries and inferred + // currentStepSeq from superseded generations — tools landed under the wrong + // (current-gen) step and it read as "several plans, some never run." So: + // only the LAST propose_plan is the live plan; earlier ones collapse to a + // single "Earlier plan revised" marker, and update_plan_step step-tracking + // only applies to the current generation. + let lastPlanMi = -1 + let lastPlanTi = -1 + let planCount = 0 + for (let mi = 0; mi < $msgs.length; mi++) { + for (let ti = 0; ti < $msgs[mi].tools.length; ti++) { + if ($msgs[mi].tools[ti].name === 'propose_plan') { + planCount++ + lastPlanMi = mi + lastPlanTi = ti + } + } + } + const revised = planCount > 1 + let currentStepSeq = 0 let entryIdx = 0 + // No plan at all (plan-less Q&A) → treat the whole transcript as current. + let sawCurrentPlan = planCount === 0 + let emittedRevised = false for (let mi = 0; mi < $msgs.length; mi++) { const msgTs = tsOf($msgs[mi].created_at) - for (const t of $msgs[mi].tools) { - // Track current step from update_plan_step calls - if (t.type === 'tool_use' && t.name === 'update_plan_step') { + for (let ti = 0; ti < $msgs[mi].tools.length; ti++) { + const t = $msgs[mi].tools[ti] + const isLastPlan = mi === lastPlanMi && ti === lastPlanTi + if (isLastPlan) sawCurrentPlan = true + + // Track current step ONLY from the current generation's + // update_plan_step calls; a superseded generation's seqs would tag + // tools with the wrong (current-gen) step. + if (sawCurrentPlan && t.type === 'tool_use' && t.name === 'update_plan_step') { const s = typeof t.args?.seq === 'number' ? t.args.seq : undefined const status = typeof t.args?.status === 'string' ? t.args.status : undefined if (s && status === 'running') currentStepSeq = s @@ -163,6 +195,23 @@ export function computeActivityLog( currentStepSeq = 0 } + // Skip superseded-generation propose_plan entries; emit one collapsed + // "revised" marker so a re-proposal stays visible without reading as a + // second active plan. + if (t.name === 'propose_plan' && !isLastPlan) { + if (revised && !emittedRevised) { + emittedRevised = true + entries.push({ + id: `plan_revised_${mi}_${ti}`, + type: 'plan', + description: 'Earlier plan revised', + timestamp: freeze(`plan_revised_${mi}_${ti}`, msgTs), + status: 'done' + }) + } + continue + } + const label = toolActivityLabel(t) const stepTag = currentStepSeq > 0 ? currentStepSeq : undefined const id = t.id ?? `tool_${mi}_${entryIdx++}` diff --git a/web/src/lib/stores/chat.ts b/web/src/lib/stores/chat.ts index c35be65..ef8d2af 100644 --- a/web/src/lib/stores/chat.ts +++ b/web/src/lib/stores/chat.ts @@ -72,6 +72,21 @@ function mid(): string { return crypto.randomUUID() } +// dropOptimisticAssistantBubble removes the trailing empty assistant message +// that sendSessionMessage/startTask optimistically append — used when a turn is +// QUEUED behind an in-flight one (plan 2026-08-03 F2): no live assistant stream +// is attached, so the empty placeholder must go (otherwise it lingers as a +// blank bubble). Shared so the guard can't drift between the two call sites. +function dropOptimisticAssistantBubble(messages: Writable): void { + messages.update((ms) => { + const last = ms[ms.length - 1] + if (last && last.role === 'assistant' && last.text === '' && last.tools.length === 0) { + return ms.slice(0, -1) + } + return ms + }) +} + export const messages = writable([]) export const streaming = writable(false) export const connectionState = writable<'connected' | 'disconnected' | 'reconnecting'>('connected') @@ -657,6 +672,16 @@ export function sendSessionMessage(sessionId: string, text: string) { sessionId, (ev: ChatEvent) => { if (ev.type === 'session') return // sessionId is already known for a window + if (ev.type === 'queued') { + // This message was queued behind an in-flight turn (plan 2026-08-03 + // F2): no assistant stream is attached to this response. Drop the + // optimistic empty assistant bubble so the user message is the last + // thing on screen — the thread then shows a "Queued" hint while the + // session is working, and the poller surfaces the queued turn's + // result once it runs server-side. + dropOptimisticAssistantBubble(chat.messages) + return + } if (ev.type === 'tool_use') { const tr: ToolCallResult = { type: 'tool_use', @@ -791,6 +816,13 @@ export function startTask(text: string, onSession: (sessionId: string) => void): function apply(ev: ChatEvent) { const c = chat if (!c || !sessionId) return + if (ev.type === 'queued') { + // Defensive: a brand-new task won't normally queue (its session has no + // in-flight turn), but handle it symmetrically with sendSessionMessage — + // drop the optimistic empty assistant bubble. See plan 2026-08-03 F2. + dropOptimisticAssistantBubble(c.messages) + return + } if (ev.type === 'tool_use') { const tr: ToolCallResult = { type: 'tool_use', diff --git a/web/src/lib/stores/workspace.ts b/web/src/lib/stores/workspace.ts index 1e63efe..b9d3828 100644 --- a/web/src/lib/stores/workspace.ts +++ b/web/src/lib/stores/workspace.ts @@ -1,6 +1,6 @@ import { writable, derived, get, type Writable, type Readable } from 'svelte/store' import { liveEvents, subscribeEvents } from './events' -import { currentSession, sessions, loadSessions } from './chat' +import { currentSession, sessions, loadSessions, chatFor, streaming } from './chat' import { fetchPlan, fetchQuestions, @@ -271,10 +271,13 @@ export function startWorkspace(): () => void { if (!sid) return // Oldest-first application so ordering (e.g. plan.step.started before // .finished) is preserved. + let planAffecting = false for (const e of fresh.slice().reverse()) { applyEventTo(globalWorkspace, sid, e) applyHealthChangedTo(globalWorkspace, e) + if (e.correlation_id === sid && STATUS_AFFECTING.has(e.type)) planAffecting = true } + if (planAffecting) schedulePlanRefetch(globalWorkspace, sid) }) return () => { @@ -306,12 +309,62 @@ export function taskFor(sessionId: string): Readable { return derived(sessions, ($sessions) => $sessions.find((s) => s.id === sessionId) ?? null) } +// Session statuses where the server is actively running a turn for this task. +// Deliberately EXCLUDES `awaiting_input` (paused for the operator) and the +// terminal states (done/failed/abandoned). This is the reliable "the agent is +// working" truth that survives a dropped SSE stream or an autonomous/background +// turn (which has no chat stream at all) — see plan 2026-08-03 F1. +const ACTIVE_TURN_STATUS = new Set(['planning', 'executing']) + +function isWorking($streaming: boolean, $task: Session | null): boolean { + return $streaming || (!!$task && !!$task.status && ACTIVE_TURN_STATUS.has($task.status)) +} + +// taskWorking(sessionId): true while this session has a live stream OR its +// server-side status says a turn is running. Used by the chat window's +// "working" indicator, trace running state, and the activity spinner so a +// background/long/desynced turn still looks alive (the symptom: "can't tell +// the agent is working"). +export function taskWorking(sessionId: string): Readable { + const chat = chatFor(sessionId) + return derived([chat.streaming, taskFor(sessionId)], ([$s, $t]) => isWorking($s, $t)) +} + +// Global "current session" working signal for the main view's panel. +export const currentWorking = derived( + [streaming, currentTask], + ([$s, $t]) => isWorking($s, $t) +) + async function hydrateSession(ws: WorkspaceState, sessionId: string) { const [steps, qs] = await Promise.all([fetchPlan(sessionId), fetchQuestions(sessionId)]) ws.planSteps.set(steps) ws.questions.set(qs) } +// Self-heal for the plan panel (plan 2026-08-03 F4): plan steps are otherwise +// driven ONLY by live plan.proposed/plan.step.* events plus a one-time hydrate +// on mount. If an event is missed (window opened mid-turn, a brief events- +// stream gap), the panel freezes on a stale generation. Refetching the plan +// (current generation) on any task-lifecycle event makes it converge back to +// truth. Debounced per session since several of these land in one burst. +const planRefreshTimers = new Map>() +function schedulePlanRefetch(ws: WorkspaceState, sessionId: string) { + const existing = planRefreshTimers.get(sessionId) + if (existing) clearTimeout(existing) + planRefreshTimers.set( + sessionId, + setTimeout(async () => { + planRefreshTimers.delete(sessionId) + try { + ws.planSteps.set(await fetchPlan(sessionId)) + } catch { + // network blip — the next lifecycle event retries + } + }, 400) + ) +} + export function startSessionWorkspace(sessionId: string): () => void { const ws = workspaceFor(sessionId) const unsub = subscribeEvents() @@ -327,10 +380,13 @@ export function startSessionWorkspace(sessionId: string): () => void { if (maxId <= lastSeen) return const fresh = evs.filter((e) => e.id > lastSeen) lastSeen = maxId + let planAffecting = false for (const e of fresh.slice().reverse()) { applyEventTo(ws, sessionId, e) applyHealthChangedTo(ws, e) + if (e.correlation_id === sessionId && STATUS_AFFECTING.has(e.type)) planAffecting = true } + if (planAffecting) schedulePlanRefetch(ws, sessionId) }) return () => { diff --git a/web/src/lib/types.ts b/web/src/lib/types.ts index 62b01dd..a41d926 100644 --- a/web/src/lib/types.ts +++ b/web/src/lib/types.ts @@ -39,6 +39,14 @@ export interface ChatErrorEvent { data: string } +// The turn was queued behind an in-flight turn for this session (plan +// 2026-08-03 F2). No live assistant stream follows in this response; the +// queued turn runs server-side when the gate frees and the poller surfaces it. +export interface ChatQueuedEvent { + type: 'queued' + data: string // session id +} + export type ChatEvent = | ChatSessionEvent | ChatToolUseEvent @@ -47,6 +55,7 @@ export type ChatEvent = | ChatTextEvent | ChatDoneEvent | ChatErrorEvent + | ChatQueuedEvent // ---- Tool call result (merged from tool_use + tool_result SSE pairs) ----