dtoro/oikos
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

feat(nomos): chat working-visibility, message queue, generation-aware timeline
Some checks failed
ci / build-test (push) Has been cancelled
Details
ci / docker-build (push) Has been cancelled
Details
ci / web (push) Has been cancelled
Details
Desktop App / Build Linux (amd64) (push) Has been cancelled
Details
Desktop App / Attach to Release (push) Has been cancelled
Details

Browse Source 
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
This commit is contained in:
authentik Default Admin
2026-08-03 22:34:14 +02:00
parent 757ef2f34b
commit 5b68bdc16c
17 changed files with 877 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
web/src/lib/components/ChatThread.svelte
View File

@@ -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<HTMLDivElement | null>(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}
</div>
{#if idx === messages.length - 1 && working && !streaming}
<!-- The last message is this user bubble and the agent is
working but not live-streaming → the message was queued
behind an in-flight turn (plan 2026-08-03 F2). It'll run
when the current step finishes. -->
<span class="px-1 text-[10px] text-muted-foreground"
>Queued — Nomos will run this when it finishes the current step.</span
>
{/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}
<!-- Background/autonomous turn in progress (no live stream to watch):
keep the composer open so the operator can queue a follow-up
(plan 2026-08-03 F1/F2). -->
<div class="mb-1 px-1 text-[10px] text-muted-foreground">
Nomos is working in the background — your message will queue and run when it's free.
</div>
{/if}
<form
class="relative mx-auto flex h-full w-full max-w-3xl"
onsubmit={(e) => {

5
web/src/lib/components/SessionChatWindow.svelte
View File

@@ -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 @@
<ChatThread
messages={$chatMessages}
streaming={$chatStreaming}
working={$chatWorking}
connectionState={$chatConnectionState}
error={$chatError}
chatErrors={$chatErrors}

12
web/src/lib/components/TaskContextPanel.svelte
View File

@@ -5,12 +5,14 @@
startWorkspace,
planSteps,
currentTask,
currentWorking,
touched,
healthDiffs,
workspaceFor,
taskFor
taskFor,
taskWorking
} from '$lib/stores/workspace'
import { streaming, messages, chatFor } from '$lib/stores/chat'
import { messages, chatFor } from '$lib/stores/chat'
import { activityLog, activityLogFor } from '$lib/stores/activity'
import SessionGraph from './SessionGraph.svelte'
import UnifiedTimeline from './UnifiedTimeline.svelte'
@@ -37,7 +39,7 @@
const healthDiffsStore = $derived(ws ? ws.healthDiffs : healthDiffs)
const taskStore = $derived(sessionId ? taskFor(sessionId) : currentTask)
const chat = $derived(sessionId ? chatFor(sessionId) : null)
const streamingStore = $derived(chat ? chat.streaming : streaming)
const workingStore = $derived(sessionId ? taskWorking(sessionId) : currentWorking)
const messagesStore = $derived(chat ? chat.messages : messages)
const activityLogStore = $derived(sessionId ? activityLogFor(sessionId) : activityLog)
@@ -134,7 +136,7 @@
class="size-3"
/>{/if}
<span>Activity</span>
{#if $streamingStore && activityRunning > 0}
{#if $workingStore && activityRunning > 0}
<Spinner class="size-3 text-primary" />
{/if}
{#if planTotal > 0}
@@ -161,7 +163,7 @@
<UnifiedTimeline
entries={$activityLogStore}
planSteps={$planStepsStore}
streaming={$streamingStore}
streaming={$workingStore}
/>
</div>
{/if}

1
web/src/lib/components/desktop-shell/NewTaskChat.svelte
View File

@@ -26,6 +26,7 @@
{initialDraft}
messages={[]}
streaming={false}
working={false}
connectionState="connected"
{onSend}
onCancel={() => {}}

70
web/src/lib/stores/activity.test.ts
View File

@@ -46,6 +46,14 @@ function toolResult(name: string, id: string): NonNullable<ChatMessage['tools']>
return { type: 'tool_result', name, id, result: 'ok' }
}
function toolUse(
name: string,
id: string,
args?: Record<string, unknown>
): NonNullable<ChatMessage['tools']>[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()
})
})

55
web/src/lib/stores/activity.ts
View File

@@ -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++}`

32
web/src/lib/stores/chat.ts
View File

@@ -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<ChatMessage[]>): 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<ChatMessage[]>([])
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',

58
web/src/lib/stores/workspace.ts
View File

@@ -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<Session | null> {
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<boolean> {
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<string, ReturnType<typeof setTimeout>>()
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 () => {

9
web/src/lib/types.ts
View File

@@ -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) ----