From bac472a4c53c324ed8aaeba11b488d8acbe03de1 Mon Sep 17 00:00:00 2001 From: 0xKT Date: Sun, 20 Sep 2026 23:16:47 +0800 Subject: [PATCH 1/6] fix(ui-web): subtitle a task node by agent, status, duration and tokens The node panel's subtitle carried the spawn's minted handle, a "usage not reported" fragment for every lane that reports none, and the tool and failure counts -- none of which the reader asked that line for. It now reads agent, status word, duration, and the token total only when the lane reported one; the tool count stays on the board card and in the process fold. The two catalogue keys nothing reads any more are removed, the TUI's generated copy follows, and the design note says what the header shows. Co-authored-by: Claude (claude-fable-5-1) --- .../2026-09-18-desk-tasks-list-design.md | 7 ++-- i18n/messages.json | 8 ----- ui-tui/src/i18n/messages.generated.ts | 4 --- ui-web/src/features/tasks/TasksPage.test.tsx | 22 ++++++++++--- ui-web/src/features/tasks/TasksPage.tsx | 32 +++++++------------ 5 files changed, 32 insertions(+), 41 deletions(-) diff --git a/docs/specs/2026-09-18-desk-tasks-list-design.md b/docs/specs/2026-09-18-desk-tasks-list-design.md index fae436032..a48768a33 100644 --- a/docs/specs/2026-09-18-desk-tasks-list-design.md +++ b/docs/specs/2026-09-18-desk-tasks-list-design.md @@ -147,9 +147,10 @@ truth after a reconnect or a terminal event. `features/tasks/` derives its types from `generated.ts`, reads through `tasks.list`, and draws the mock: the list row, the strip (three chips and an overflow), the running count on the tab, the pane (status bar with `interrupted` and `cancelled`, a stop action, the why -banner naming the failed node, file and diff chips, the board with tool and failure counts -and a lane per shared `(agent, instance)`), the node panel (header with tokens or "not -reported", context and order tabs, a chat dock for a stateful agent's instance). The desk's +banner naming the failed node, file and diff chips, the board with tool counts and a lane +per shared `(agent, instance)`), the node panel (header with the agent, its status word, the +duration and the token total when the lane reported one, context and order tabs, a chat dock +for a stateful agent's instance). The desk's deliverables and diff tabs gain a task-derived group beside the session-level rows; a task file's diff is built on click from the node's messages with the page's existing hunk builders. The offline page answers `tasks.list` from `src/rpc/fixtures/tasks.ts`. diff --git a/i18n/messages.json b/i18n/messages.json index a6a8b2c2d..d7b9ea627 100644 --- a/i18n/messages.json +++ b/i18n/messages.json @@ -7099,10 +7099,6 @@ "en": "{n} tokens", "zh": "{n} tokens" }, - "gui.tasks.tokens_none": { - "en": "usage not reported", - "zh": "不报用量" - }, "gui.tasks.no_output": { "en": "no output", "zh": "无输出" @@ -7111,10 +7107,6 @@ "en": "{n} tools", "zh": "{n} 次工具" }, - "gui.tasks.tools_failed_n": { - "en": "{n} failed", - "zh": "{n} 次失败" - }, "gui.tasks.stop": { "en": "Stop", "zh": "停止" diff --git a/ui-tui/src/i18n/messages.generated.ts b/ui-tui/src/i18n/messages.generated.ts index 384ccdc54..a2b9ad8f0 100644 --- a/ui-tui/src/i18n/messages.generated.ts +++ b/ui-tui/src/i18n/messages.generated.ts @@ -1983,10 +1983,8 @@ export const UI_TEXT: Record> = { 'gui.tasks.node_st_interrupted': 'Interrupted', 'gui.tasks.node_st_exception': 'Awaiting a decision', 'gui.tasks.tokens_n': '{n} tokens', - 'gui.tasks.tokens_none': 'usage not reported', 'gui.tasks.no_output': 'no output', 'gui.tasks.tools_n': '{n} tools', - 'gui.tasks.tools_failed_n': '{n} failed', 'gui.tasks.stop': 'Stop', 'gui.tasks.why_at': 'stuck at: ', 'gui.tasks.why_interrupted': @@ -3719,10 +3717,8 @@ export const UI_TEXT: Record> = { 'gui.tasks.node_st_interrupted': '中断', 'gui.tasks.node_st_exception': '待裁决', 'gui.tasks.tokens_n': '{n} tokens', - 'gui.tasks.tokens_none': '不报用量', 'gui.tasks.no_output': '无输出', 'gui.tasks.tools_n': '{n} 次工具', - 'gui.tasks.tools_failed_n': '{n} 次失败', 'gui.tasks.stop': '停止', 'gui.tasks.why_at': '挂在：', 'gui.tasks.why_interrupted': '这次运行没有跑完，也没有人在继续跑它。停住那一步之后的都没有派出去。', diff --git a/ui-web/src/features/tasks/TasksPage.test.tsx b/ui-web/src/features/tasks/TasksPage.test.tsx index a93232aa1..f1ffaeca2 100644 --- a/ui-web/src/features/tasks/TasksPage.test.tsx +++ b/ui-web/src/features/tasks/TasksPage.test.tsx @@ -570,22 +570,34 @@ describe('the node panel', () => { expect(document.querySelector('.tkspecid b')?.textContent).toBe('run-123') }) - it('says nothing about tools rather than a dash when the lane never reported a count', () => { + it('subtitle is the agent, the status word and the duration when the lane reported no usage', () => { const done = task({ id: 'a', kind: 'dag', status: 'completed', - nodes: [node({ node_id: 'n1', status: 'completed', started_at: 1000, ended_at: 2000, tool_call_count: null })], + nodes: [node({ + node_id: 'n1', status: 'completed', started_at: 1000, ended_at: 2000, + tokens_in: null, tokens_out: null, tool_call_count: 14, tool_failure_count: 1, + })], + }) + pick(done) + expect(document.querySelector('.tksub')?.textContent).toBe('raven · gui.tasks.node_st_completed · 1s') + }) + + it('subtitle adds the token total, grouped by thousands, when the lane reported usage', () => { + const done = task({ + id: 'a', kind: 'spawn', status: 'running', + nodes: [node({ node_id: 'n1', status: 'running', started_at: Date.now() - 110_000, tokens_in: 4000, tokens_out: 910 })], }) pick(done) - expect(document.querySelector('.tksub')?.textContent).not.toContain('—') + expect(document.querySelector('.tksub')?.textContent).toBe('raven · gui.tasks.node_st_running · 1m50s · gui.tasks.tokens_n {"n":"4,910"}') }) - it('sets the agent apart from the rest of the subtitle in its own ', () => { + it('sets the agent apart in its own , without the handle', () => { const done = task({ id: 'a', kind: 'dag', status: 'completed', nodes: [node({ node_id: 'n1', status: 'completed', agent: 'coder', instance: 'x1' })], }) pick(done) - expect(document.querySelector('.tksub b')?.textContent).toBe('coder @x1') + expect(document.querySelector('.tksub b')?.textContent).toBe('coder') }) describe('a cross-run dependency', () => { diff --git a/ui-web/src/features/tasks/TasksPage.tsx b/ui-web/src/features/tasks/TasksPage.tsx index 9f8bc954b..87af44f87 100644 --- a/ui-web/src/features/tasks/TasksPage.tsx +++ b/ui-web/src/features/tasks/TasksPage.tsx @@ -375,43 +375,33 @@ function useNodeRecord(row: TaskRow, node: TaskNode): RecordLoad { return { ...state, retry: () => setNonce((n) => n + 1) } } -function tokensText(node: TaskNode): string { - if (node.status === 'pending' || node.status === 'skipped') return '' - if (node.tokens_in == null && node.tokens_out == null) return t('gui.tasks.tokens_none') - return t('gui.tasks.tokens_n', { n: fmtN((node.tokens_in || 0) + (node.tokens_out || 0)) }) -} - /* Grouped by thousands, the way the prototype's own `fmtN` reads a token count -- "4,321 tokens" rather than "4321 tokens". */ const fmtN = (n: number): string => n.toLocaleString('en-US') /* The rest of the subtitle, after the agent (which the caller sets apart in - its own ``): status word, duration, tokens, tool count -- pushed only - when the fact is there to push. No dash for a missing tool count: a fact - this node's lane never reported is a fact this line says nothing about, - not a line that says "—". */ + its own ``): status word, duration, tokens -- each pushed only when the + fact is there to push. A lane that never reported usage is a fact this + line says nothing about, not a line that says "not reported"; the tool + count is the board card's and the process fold's, not this line's. */ function nodeSubtitleRest(node: TaskNode): string[] { const parts = [nodeStatusWord(node.status)] const dur = node.started_at ? formatDuration((node.ended_at ?? Date.now()) - node.started_at) : '' if (dur) parts.push(dur) - const tk = tokensText(node) - if (tk) parts.push(tk) - if (node.tool_call_count != null) { - parts.push(node.tool_failure_count - ? t('gui.tasks.tools_n', { n: node.tool_call_count }) + ' · ' + t('gui.tasks.tools_failed_n', { n: node.tool_failure_count }) - : t('gui.tasks.tools_n', { n: node.tool_call_count })) + if (node.tokens_in != null || node.tokens_out != null) { + parts.push(t('gui.tasks.tokens_n', { n: fmtN((node.tokens_in || 0) + (node.tokens_out || 0)) })) } if (node.status === 'completed' && node.has_output === false) parts.push(t('gui.tasks.no_output')) return parts } -/* The node panel head's subtitle line: the agent (+@instance) in its own - ``, the rest of the sentence plain after it -- the prototype sets the - agent apart the same way. */ +/* The node panel head's subtitle line: the agent in its own ``, the rest + of the sentence plain after it. The agent alone, without its handle: a + spawn's handle is a minted id (`TaskRow.handle`), and the work-order tab + already names the instance for the reader who wants it. */ function NodeSubtitle({ node }: { node: TaskNode }): JSX.Element { - const agent = node.agent + (node.instance ? ' @' + node.instance : '') const rest = nodeSubtitleRest(node) - return {agent}{rest.length ? ' · ' + rest.join(' · ') : ''} + return {node.agent}{rest.length ? ' · ' + rest.join(' · ') : ''} } /* Why a step nobody dispatched has nothing to read: skipped names the From a43c41b12a45549d06c2f1240ce01b3708174b20 Mon Sep 17 00:00:00 2001 From: 0xKT Date: Sun, 20 Sep 2026 23:21:27 +0800 Subject: [PATCH 2/6] fix(ui-web): keep reading a running spawn's record in the task node panel A spawn task's node panel read its record once, when opened, and never again: useNodeRecord re-read only on store.nodeVersion, which only a dag's node_updated event bumps, and its key held no status, so neither the steps a running spawn kept making nor its final answer ever reached a panel left open on it. The transcript's own spawn card already reads subagent.context on a one-second beat while the run is live; the panel now does the same for a running spawn (skipping a beat while a read is still out), and re-reads once on any status transition, which also covers a dag node settled by a run_completed frame that no node_updated announced. Co-authored-by: Claude (claude-fable-5-1) --- ui-web/src/features/tasks/TasksPage.test.tsx | 100 ++++++++++++++++++- ui-web/src/features/tasks/TasksPage.tsx | 39 ++++++-- 2 files changed, 128 insertions(+), 11 deletions(-) diff --git a/ui-web/src/features/tasks/TasksPage.test.tsx b/ui-web/src/features/tasks/TasksPage.test.tsx index f1ffaeca2..d7e16fedf 100644 --- a/ui-web/src/features/tasks/TasksPage.test.tsx +++ b/ui-web/src/features/tasks/TasksPage.test.tsx @@ -1,6 +1,6 @@ // @vitest-environment happy-dom import { act, cleanup, fireEvent, render } from '@testing-library/react' -import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { resetTranslator, setTranslator } from '../../i18n/t' import { setCurrent } from '../../lib/session' @@ -686,6 +686,104 @@ describe('the node panel', () => { await act(async () => { resolveSecond?.({ dispatch: 'go', steps: [], answer: 'second answer', outputTruncated: false }) }) expect(document.querySelector('.tkans')?.textContent).toBe('second answer') }) + + /* A spawn's lane sends no per-step event -- `subagent.status` moves on + pending, running and the terminal word only -- so a running spawn's + record is re-read on a beat, the transcript's spawn card's own + cadence, rather than left at whatever the first read saw. */ + it("a running spawn's record is re-read on a beat, so its steps keep arriving", async () => { + vi.useFakeTimers() + try { + const calls: string[] = [] + record = { dispatch: 'go', steps: [], answer: null, outputTruncated: false } + setSources({ + tasks: { ...source(), node: async () => { calls.push('fetch'); return record } }, + workspace: { shortPath: (p: string) => p, hostPlatform: () => 'mac', canBrowse: false, openPath: () => {} }, + }) + const running = task({ + id: 's1', kind: 'spawn', status: 'running', nodes: [node({ node_id: 's1', status: 'running' })], + }) + pick(running) + await act(async () => {}) + expect(calls).toEqual(['fetch']) + + record = { dispatch: 'go', steps: [{ kind: 'say', text: 'first step' }], answer: null, outputTruncated: false } + await act(async () => { vi.advanceTimersByTime(1000) }) + expect(calls).toEqual(['fetch', 'fetch']) + expect(document.querySelector('.tkprocb .tkans')?.textContent).toBe('first step') + + await act(async () => { vi.advanceTimersByTime(1000) }) + expect(calls).toEqual(['fetch', 'fetch', 'fetch']) + } finally { + vi.useRealTimers() + } + }) + + it('skips a beat while a read is still out, rather than stacking reads', async () => { + vi.useFakeTimers() + try { + let calls = 0 + setSources({ + tasks: { ...source(), node: () => { calls += 1; return new Promise(() => {}) } }, + workspace: { shortPath: (p: string) => p, hostPlatform: () => 'mac', canBrowse: false, openPath: () => {} }, + }) + const running = task({ + id: 's1', kind: 'spawn', status: 'running', nodes: [node({ node_id: 's1', status: 'running' })], + }) + pick(running) + await act(async () => {}) + expect(calls).toBe(1) + await act(async () => { vi.advanceTimersByTime(3000) }) + expect(calls).toBe(1) + } finally { + vi.useRealTimers() + } + }) + + it('a running dag node is not re-read on a beat: its node_updated events already do that', async () => { + vi.useFakeTimers() + try { + const calls: string[] = [] + setSources({ + tasks: { ...source(), node: async () => { calls.push('fetch'); return record } }, + workspace: { shortPath: (p: string) => p, hostPlatform: () => 'mac', canBrowse: false, openPath: () => {} }, + }) + const running = task({ + id: 'r1', kind: 'dag', status: 'running', nodes: [node({ node_id: 'n1', status: 'running' })], + }) + pick(running) + await act(async () => {}) + await act(async () => { vi.advanceTimersByTime(3000) }) + expect(calls).toEqual(['fetch']) + } finally { + vi.useRealTimers() + } + }) + + it("is re-read once when the spawn settles, so the answer lands without reopening the node", async () => { + const calls: string[] = [] + record = { dispatch: 'go', steps: [], answer: null, outputTruncated: false } + setSources({ + tasks: { ...source(), node: async () => { calls.push('fetch'); return record } }, + workspace: { shortPath: (p: string) => p, hostPlatform: () => 'mac', canBrowse: false, openPath: () => {} }, + }) + const running = task({ + id: 's1', kind: 'spawn', status: 'running', agent: 'raven', handle: 'h1', + nodes: [node({ node_id: 's1', status: 'running', instance: 'h1', started_at: 1000 })], + }) + store.set((prev) => ({ ...prev, rows: [running], loaded: true })) + pick(running) + await act(async () => {}) + expect(calls).toEqual(['fetch']) + + record = { dispatch: 'go', steps: [], answer: 'the answer', outputTruncated: false } + await act(async () => { + store.onSubagentStatus({ task_id: 't1', call_id: 's1', agent: 'raven', label: 'x', status: 'completed', ended_at: 2000 }) + }) + expect(calls).toEqual(['fetch', 'fetch']) + expect(document.querySelector('.tkanswer .tkans')?.textContent).toBe('the answer') + expect(document.querySelector('.tksub')?.textContent).toBe('raven · gui.tasks.node_st_completed · 1s') + }) }) describe('the context tab while the record is loading or failed', () => { diff --git a/ui-web/src/features/tasks/TasksPage.tsx b/ui-web/src/features/tasks/TasksPage.tsx index 87af44f87..961490d87 100644 --- a/ui-web/src/features/tasks/TasksPage.tsx +++ b/ui-web/src/features/tasks/TasksPage.tsx @@ -9,7 +9,7 @@ * flatter rendering of the same steps to fall out of step with the graph. */ -import { useEffect, useState, useSyncExternalStore } from 'react' +import { useEffect, useRef, useState, useSyncExternalStore } from 'react' import { Glyph, SendGlyph } from '../../components/Ico' import { t } from '../../i18n/t' @@ -340,19 +340,29 @@ interface RecordLoad { retry: () => void } +/* The beat a running spawn's record is re-read on: the one the transcript's + own spawn card reads `subagent.context` on (TranscriptPage.tsx), so the + two views of one run move together. */ +const SPAWN_READ_BEAT_MS = 1000 + /* Fetched once per (row, node) and shared by both tabs: the order tab's "instruction" is the same rendered prompt the context tab's dispatch is, and asking for it twice would be asking the gateway the same question twice for one screen. Never fetched for a step that has not been dispatched -- there is nothing yet to read. - Refetched on every live event that names this node (`store.nodeVersion`), - on top of the (row, node) identity: `dag.node_updated` fires once per tool - call while a node runs, not only on a status transition, so a node opened - mid-run keeps reading its own steps and its answer as they arrive rather - than freezing at the first read. A stale record is kept on screen through - a refetch rather than cleared back to `null` -- the reader is watching a - node run, not watching it flicker blank once a second. */ + Refetched on every live event that names this node (`store.nodeVersion`) + and on every status transition, on top of the (row, node) identity. + `dag.node_updated` fires once per tool call while a dag node runs, so a + dag node opened mid-run keeps reading its own steps and its answer as they + arrive rather than freezing at the first read. A spawn has no per-step + event -- `subagent.status` moves on pending, running and the terminal word + only -- so while one runs its record is re-read on a beat instead, skipping + a beat while a read is still out; the status key then covers the terminal + frame, so the answer lands without the reader closing and reopening the + node. A stale record is kept on screen through a refetch rather than + cleared back to `null` -- the reader is watching a node run, not watching + it flicker blank once a second. */ function useNodeRecord(row: TaskRow, node: TaskNode): RecordLoad { const dispatched = node.status !== 'pending' && node.status !== 'skipped' const version = useSyncExternalStore(store.subscribe, () => store.nodeVersion(row.kind, row.id, node.node_id)) @@ -360,18 +370,27 @@ function useNodeRecord(row: TaskRow, node: TaskNode): RecordLoad { const [state, setState] = useState<{ loading: boolean; record: NodeRecord | null; failed: boolean }>( { loading: dispatched, record: null, failed: false }, ) + const reading = useRef(false) useEffect(() => { if (!dispatched) { setState({ loading: false, record: null, failed: false }); return } setState((prev) => ({ loading: true, record: prev.record, failed: false })) let alive = true const src = store.source() if (!src) { setState((prev) => ({ loading: false, record: prev.record, failed: true })); return } + reading.current = true src.node(row, node) .then((r) => { if (alive) setState({ loading: false, record: r, failed: false }) }) .catch(() => { if (alive) setState((prev) => ({ loading: false, record: prev.record, failed: true })) }) - return () => { alive = false } + .finally(() => { if (alive) reading.current = false }) + return () => { alive = false; reading.current = false } // eslint-disable-next-line react-hooks/exhaustive-deps - }, [row.kind, row.id, node.node_id, dispatched, version, nonce]) + }, [row.kind, row.id, node.node_id, node.status, version, nonce]) + const beating = row.kind === 'spawn' && node.status === 'running' + useEffect(() => { + if (!beating) return + const beat = setInterval(() => { if (!reading.current) setNonce((n) => n + 1) }, SPAWN_READ_BEAT_MS) + return () => clearInterval(beat) + }, [beating]) return { ...state, retry: () => setNonce((n) => n + 1) } } From edb80add058dae8992f23fff93cf1bbab92ad641 Mon Sep 17 00:00:00 2001 From: 0xKT Date: Sun, 20 Sep 2026 23:47:47 +0800 Subject: [PATCH 3/6] fix(ui-web): no record beat without a source, and tests for when the beat stops Review follow-ups on the node panel's record beat. A pane with no tasks source would have re-run the read effect into its failed branch once a second, so the beat now requires a source. The subtitle comment no longer credits the process fold with a tool count it does not show, and the design note's node-context section says how often a record is re-read. Tests cover usage reported on one side only, the beat stopping when the spawn settles and when the panel closes, and the token line on the settled shape the server actually sends. Co-authored-by: Claude (claude-fable-5-1) --- .../2026-09-18-desk-tasks-list-design.md | 6 +- ui-web/src/features/tasks/TasksPage.test.tsx | 93 ++++++++++++++----- ui-web/src/features/tasks/TasksPage.tsx | 7 +- 3 files changed, 78 insertions(+), 28 deletions(-) diff --git a/docs/specs/2026-09-18-desk-tasks-list-design.md b/docs/specs/2026-09-18-desk-tasks-list-design.md index a48768a33..1f9901998 100644 --- a/docs/specs/2026-09-18-desk-tasks-list-design.md +++ b/docs/specs/2026-09-18-desk-tasks-list-design.md @@ -116,7 +116,11 @@ dispatch leaves no record and is withdrawn by the `subagent.status{cancelled}` f `dag.node(run_id, node, session_key)` and `subagent.context(id, session_id)` return `messages[]` in the `session.resume` shape; the tab draws them with the transcript's own -renderer, including the synthetic `role=console` row an in-flight `cli` lane emits. +renderer, including the synthetic `role=console` row an in-flight `cli` lane emits. The +record is re-read on every status transition and, for a dag node, on each `dag.node_updated` +frame; a spawn gets no per-step frame, so its record is re-read on a one-second beat while it +runs (a beat is skipped while a read is still out), the cadence the transcript's spawn card +already reads on. ### 5. Live updates diff --git a/ui-web/src/features/tasks/TasksPage.test.tsx b/ui-web/src/features/tasks/TasksPage.test.tsx index d7e16fedf..f41c0ba1a 100644 --- a/ui-web/src/features/tasks/TasksPage.test.tsx +++ b/ui-web/src/features/tasks/TasksPage.test.tsx @@ -582,13 +582,24 @@ describe('the node panel', () => { expect(document.querySelector('.tksub')?.textContent).toBe('raven · gui.tasks.node_st_completed · 1s') }) + /* A settled node: every lane writes its usage into the record when the run + finishes, so that is the shape the server sends the fragment on. */ it('subtitle adds the token total, grouped by thousands, when the lane reported usage', () => { const done = task({ - id: 'a', kind: 'spawn', status: 'running', - nodes: [node({ node_id: 'n1', status: 'running', started_at: Date.now() - 110_000, tokens_in: 4000, tokens_out: 910 })], + id: 'a', kind: 'spawn', status: 'completed', + nodes: [node({ node_id: 'n1', status: 'completed', started_at: 1000, ended_at: 111_000, tokens_in: 4000, tokens_out: 910 })], + }) + pick(done) + expect(document.querySelector('.tksub')?.textContent).toBe('raven · gui.tasks.node_st_completed · 1m50s · gui.tasks.tokens_n {"n":"4,910"}') + }) + + it('subtitle counts usage a lane reported on one side only', () => { + const done = task({ + id: 'a', kind: 'spawn', status: 'completed', + nodes: [node({ node_id: 'n1', status: 'completed', started_at: 1000, ended_at: 2000, tokens_in: 500, tokens_out: null })], }) pick(done) - expect(document.querySelector('.tksub')?.textContent).toBe('raven · gui.tasks.node_st_running · 1m50s · gui.tasks.tokens_n {"n":"4,910"}') + expect(document.querySelector('.tksub')?.textContent).toBe('raven · gui.tasks.node_st_completed · 1s · gui.tasks.tokens_n {"n":"500"}') }) it('sets the agent apart in its own , without the handle', () => { @@ -760,29 +771,61 @@ describe('the node panel', () => { } }) - it("is re-read once when the spawn settles, so the answer lands without reopening the node", async () => { - const calls: string[] = [] - record = { dispatch: 'go', steps: [], answer: null, outputTruncated: false } - setSources({ - tasks: { ...source(), node: async () => { calls.push('fetch'); return record } }, - workspace: { shortPath: (p: string) => p, hostPlatform: () => 'mac', canBrowse: false, openPath: () => {} }, - }) - const running = task({ - id: 's1', kind: 'spawn', status: 'running', agent: 'raven', handle: 'h1', - nodes: [node({ node_id: 's1', status: 'running', instance: 'h1', started_at: 1000 })], - }) - store.set((prev) => ({ ...prev, rows: [running], loaded: true })) - pick(running) - await act(async () => {}) - expect(calls).toEqual(['fetch']) + it('is re-read once when the spawn settles, so the answer lands without reopening the node, and the beat stops', async () => { + vi.useFakeTimers() + try { + const calls: string[] = [] + record = { dispatch: 'go', steps: [], answer: null, outputTruncated: false } + setSources({ + tasks: { ...source(), node: async () => { calls.push('fetch'); return record } }, + workspace: { shortPath: (p: string) => p, hostPlatform: () => 'mac', canBrowse: false, openPath: () => {} }, + }) + const running = task({ + id: 's1', kind: 'spawn', status: 'running', agent: 'raven', handle: 'h1', + nodes: [node({ node_id: 's1', status: 'running', instance: 'h1', started_at: 1000 })], + }) + store.set((prev) => ({ ...prev, rows: [running], loaded: true })) + pick(running) + await act(async () => {}) + expect(calls).toEqual(['fetch']) - record = { dispatch: 'go', steps: [], answer: 'the answer', outputTruncated: false } - await act(async () => { - store.onSubagentStatus({ task_id: 't1', call_id: 's1', agent: 'raven', label: 'x', status: 'completed', ended_at: 2000 }) - }) - expect(calls).toEqual(['fetch', 'fetch']) - expect(document.querySelector('.tkanswer .tkans')?.textContent).toBe('the answer') - expect(document.querySelector('.tksub')?.textContent).toBe('raven · gui.tasks.node_st_completed · 1s') + record = { dispatch: 'go', steps: [], answer: 'the answer', outputTruncated: false } + await act(async () => { + store.onSubagentStatus({ task_id: 't1', call_id: 's1', agent: 'raven', label: 'x', status: 'completed', ended_at: 2000 }) + }) + expect(calls).toEqual(['fetch', 'fetch']) + expect(document.querySelector('.tkanswer .tkans')?.textContent).toBe('the answer') + expect(document.querySelector('.tksub')?.textContent).toBe('raven · gui.tasks.node_st_completed · 1s') + + /* Settled: the beat has nothing left to follow. */ + await act(async () => { vi.advanceTimersByTime(3000) }) + expect(calls).toEqual(['fetch', 'fetch']) + } finally { + vi.useRealTimers() + } + }) + + it('stops the beat when the node panel closes', async () => { + vi.useFakeTimers() + try { + const calls: string[] = [] + setSources({ + tasks: { ...source(), node: async () => { calls.push('fetch'); return record } }, + workspace: { shortPath: (p: string) => p, hostPlatform: () => 'mac', canBrowse: false, openPath: () => {} }, + }) + const running = task({ + id: 's1', kind: 'spawn', status: 'running', nodes: [node({ node_id: 's1', status: 'running' })], + }) + const mounted = pick(running) + await act(async () => {}) + await act(async () => { vi.advanceTimersByTime(1000) }) + expect(calls).toEqual(['fetch', 'fetch']) + mounted.unmount() + await act(async () => { vi.advanceTimersByTime(3000) }) + expect(calls).toEqual(['fetch', 'fetch']) + } finally { + vi.useRealTimers() + } }) }) diff --git a/ui-web/src/features/tasks/TasksPage.tsx b/ui-web/src/features/tasks/TasksPage.tsx index 961490d87..dedbf5348 100644 --- a/ui-web/src/features/tasks/TasksPage.tsx +++ b/ui-web/src/features/tasks/TasksPage.tsx @@ -385,7 +385,9 @@ function useNodeRecord(row: TaskRow, node: TaskNode): RecordLoad { return () => { alive = false; reading.current = false } // eslint-disable-next-line react-hooks/exhaustive-deps }, [row.kind, row.id, node.node_id, node.status, version, nonce]) - const beating = row.kind === 'spawn' && node.status === 'running' + /* No source, no beat: with nothing to read from, a beat would only re-run + the effect into its failed branch once a second. */ + const beating = row.kind === 'spawn' && node.status === 'running' && !!store.source() useEffect(() => { if (!beating) return const beat = setInterval(() => { if (!reading.current) setNonce((n) => n + 1) }, SPAWN_READ_BEAT_MS) @@ -402,7 +404,8 @@ const fmtN = (n: number): string => n.toLocaleString('en-US') its own ``): status word, duration, tokens -- each pushed only when the fact is there to push. A lane that never reported usage is a fact this line says nothing about, not a line that says "not reported"; the tool - count is the board card's and the process fold's, not this line's. */ + count is the board card's, and the calls themselves are the process + fold's, so neither is this line's. */ function nodeSubtitleRest(node: TaskNode): string[] { const parts = [nodeStatusWord(node.status)] const dur = node.started_at ? formatDuration((node.ended_at ?? Date.now()) - node.started_at) : '' From b97a49c20c791bff6f87f9234d0c6bf39c1239e6 Mon Sep 17 00:00:00 2001 From: 0xKT Date: Mon, 21 Sep 2026 00:23:27 +0800 Subject: [PATCH 4/6] feat(*): read a running node's usage and tool counts off its live activity A node's tokens, tool counts and files reached tasks.list only from the record on disk, which the run writes when it finishes, so a running node read as reporting nothing until it settled. The reader now overlays those facts from the activity being collected for the node in this process -- the same in-memory account subagent.context and dag.node already serve a transcript from -- keyed by the record id for a spawn and by node_live_key for a dag node, and only for what the lane has reported so far; a lane that has not spoken keeps its null. On the page, each read of a running node's record also re-reads its row through tasks.list(kind, id), so the subtitle's token total moves with the record on the beat and on every node_updated frame. reconcile keeps a frame that landed while such a read was out, the rule refresh already follows, so a read started on the beat cannot put a settled row back to running. Co-authored-by: Claude (claude-fable-5-1) --- .../2026-09-18-desk-tasks-list-design.md | 8 +- raven/rpc/methods/tasks.py | 27 +++++- tests/test_rpc_tasks.py | 89 +++++++++++++++++++ ui-web/src/features/tasks/TasksPage.test.tsx | 23 +++++ ui-web/src/features/tasks/TasksPage.tsx | 6 ++ ui-web/src/features/tasks/store.test.ts | 26 ++++++ ui-web/src/features/tasks/store.ts | 16 +++- 7 files changed, 188 insertions(+), 7 deletions(-) diff --git a/docs/specs/2026-09-18-desk-tasks-list-design.md b/docs/specs/2026-09-18-desk-tasks-list-design.md index 1f9901998..b451358e6 100644 --- a/docs/specs/2026-09-18-desk-tasks-list-design.md +++ b/docs/specs/2026-09-18-desk-tasks-list-design.md @@ -80,7 +80,10 @@ stores it uncapped; a spawn's is the head of `.error.md`, or of `.out.md` when ` `tokens_in` / `tokens_out` (null = the lane cannot report; never zero), `tool_call_count` (null = zero or unreported, an acknowledged ambiguity of `as_meta`), `tool_failure_count`, `has_output`, `prompt_template`, `inputs`, `skills`, `mcps`, `files[]` -(`{path, op: write|edit, add, del, size}`). +(`{path, op: write|edit, add, del, size}`). While a node runs, the usage, the tool counts and +the files are read off the activity being collected for it in this process (the same +in-memory account `subagent.context` and `dag.node` serve a transcript from), since the record +on disk carries them only once the run finishes; what the lane has not reported yet stays null. ### 2. Status derivation @@ -120,7 +123,8 @@ renderer, including the synthetic `role=console` row an in-flight `cli` lane emi record is re-read on every status transition and, for a dag node, on each `dag.node_updated` frame; a spawn gets no per-step frame, so its record is re-read on a one-second beat while it runs (a beat is skipped while a read is still out), the cadence the transcript's spawn card -already reads on. +already reads on. Each such read of a running node also re-reads its row through +`tasks.list(kind, id)`, which is how the panel's token total moves during the run. ### 5. Live updates diff --git a/raven/rpc/methods/tasks.py b/raven/rpc/methods/tasks.py index 67f4a45e7..197967efb 100644 --- a/raven/rpc/methods/tasks.py +++ b/raven/rpc/methods/tasks.py @@ -20,8 +20,9 @@ from pathlib import Path from typing import TYPE_CHECKING, Any +from raven.agent.subagent import activity as run_activity from raven.agent.subagent.activity import merge_file_change -from raven.agent.subagent.dag_store import REGISTRY_FILENAME, RUNNING, UNRECORDED +from raven.agent.subagent.dag_store import REGISTRY_FILENAME, RUNNING, UNRECORDED, node_live_key from raven.agent.subagent.history import dag_root, nodes_root, session_history_root from raven.agent.subagent.instances import get_registry from raven.rpc.methods.instances import _graph_of @@ -114,6 +115,28 @@ def _files_of(source: dict[str, Any]) -> list[dict[str, Any]]: return folded +def _overlay_live(node: dict[str, Any], live: Any) -> None: + """Fill a running node's usage, tool counts and files from the activity + being collected for it in this process. + + The record on disk carries those only once the run finishes (``as_meta`` + is written by ``finish``), so without this a node reads as reporting + nothing for the whole of its run. Only what the lane has said so far is + taken: a lane that has not spoken keeps its null. + """ + if live is None: + return + for key in ("tokens_in", "tokens_out"): + if node[key] is None: + node[key] = _int_or_none(getattr(live, key, None)) + calls = getattr(live, "tool_calls", None) + if node["tool_call_count"] is None and calls: + node["tool_call_count"] = len(calls) + node["tool_failure_count"] = len(getattr(live, "tool_failures", None) or []) + if not node["files"]: + node["files"] = _files_of({"files": list(getattr(live, "files", None) or [])}) + + def _counts(statuses: list[str]) -> dict[str, int]: counts = {key: 0 for key in _COUNT_KEYS} for status in statuses: @@ -262,6 +285,7 @@ def _spawn_row(files: "_NodeFiles", agent_loop_factory: "AgentLoopFactory | None "prompt_template": None, "files": _files_of(meta), } + _overlay_live(node, run_activity.live(files.node_id)) task_summary = meta.get("task_summary") or meta.get("label") or _label_from_prompt(files) or None return { "id": files.node_id, @@ -422,6 +446,7 @@ def _dag_row( node["skills"] = list(gnode["skills"]) if gnode.get("mcps") is not None: node["mcps"] = list(gnode["mcps"]) + _overlay_live(node, run_activity.live(node_live_key(run_dir.name, nid))) nodes.append(node) statuses.append(status) if isinstance(started, int): diff --git a/tests/test_rpc_tasks.py b/tests/test_rpc_tasks.py index de2699f53..bcb7368e3 100644 --- a/tests/test_rpc_tasks.py +++ b/tests/test_rpc_tasks.py @@ -922,3 +922,92 @@ def register(self, name: str, handler: Any) -> None: tasks_mod.register_tasks_methods(_Dispatcher(), agent_loop_factory=None) # type: ignore[arg-type] assert set(handlers) == {"tasks.list"} assert await handlers["tasks.list"]({"session_key": ""}) == {"tasks": []} + + +# --------------------------------------------------------------------------- +# a running node's usage, counts and files, off the live activity +# --------------------------------------------------------------------------- + + +async def test_a_running_spawn_reads_usage_and_counts_off_the_live_activity(workspace: Path) -> None: + """The record carries usage only once ``finish`` writes ``as_meta``; while + the run is in flight the only account is the activity being collected in + this process, and the row reads that -- and reads nothing again once the + collecting block has closed without the record having been finished.""" + from raven.agent.subagent import activity as activity_mod + + session_dir = _session_dir(workspace) + await _claim_spawn_node(session_dir, "counting") + SpawnRecord.open( + session_dir, + task_id="counting", + task="count things", + meta=_spawn_meta(agent="Raven", handle="counting", task_summary="Counting"), + node_id="counting", + ) + loop = _loop_stub(live_spawn_handles=frozenset({("Raven", "counting")})) + + with activity_mod.collecting(live_key="counting"): + activity_mod.note_usage({"prompt_tokens": 40, "completion_tokens": 2}) + activity_mod.note_usage({"prompt_tokens": 10, "completion_tokens": 3}) + activity_mod.note_tool_call("exec") + activity_mod.note_tool_call("write_file") + activity_mod.note_tool_failure("exec") + activity_mod.note_file_change("notes/a.md", "write", 3, 0, 12) + node = (await tasks_list({"session_key": SESSION}, agent_loop_factory=_factory(loop)))["tasks"][0]["nodes"][0] + + assert node["status"] == "running" + assert node["tokens_in"] == 50 and node["tokens_out"] == 5 + assert node["tool_call_count"] == 2 and node["tool_failure_count"] == 1 + assert node["files"] == [{"path": "notes/a.md", "op": "write", "add": 3, "del": 0, "size": 12}] + + after = (await tasks_list({"session_key": SESSION}, agent_loop_factory=_factory(loop)))["tasks"][0]["nodes"][0] + assert after["tokens_in"] is None and after["tool_call_count"] is None and after["files"] == [] + + +async def test_a_lane_that_has_not_spoken_keeps_its_nulls_while_live(workspace: Path) -> None: + """A live activity with nothing reported yet is not "zero": the row says + nothing rather than a count the lane never gave.""" + from raven.agent.subagent import activity as activity_mod + + session_dir = _session_dir(workspace) + await _claim_spawn_node(session_dir, "quiet") + SpawnRecord.open( + session_dir, + task_id="quiet", + task="say nothing yet", + meta=_spawn_meta(agent="Raven", handle="quiet", task_summary="Quiet"), + node_id="quiet", + ) + loop = _loop_stub(live_spawn_handles=frozenset({("Raven", "quiet")})) + + with activity_mod.collecting(live_key="quiet"): + node = (await tasks_list({"session_key": SESSION}, agent_loop_factory=_factory(loop)))["tasks"][0]["nodes"][0] + + assert node["tokens_in"] is None and node["tokens_out"] is None + assert node["tool_call_count"] is None and node["tool_failure_count"] is None + assert node["files"] == [] + + +async def test_a_running_dag_node_reads_usage_off_the_live_activity(workspace: Path) -> None: + """The dag runner collects a node's activity under ``node_live_key``; the + row reads that key, so a running node's usage grows before the manifest + (which a finalized run writes) exists at all.""" + from raven.agent.subagent import activity as activity_mod + from raven.agent.subagent.dag_store import node_live_key + from raven.agent.subagent.instances import get_registry + + session_dir = _session_dir(workspace) + await _make_run(session_dir, RUN_ID, _GRAPH, ["n1", "n2"]) + await get_registry().upsert_dag_node(SESSION, RUN_ID, "n1", "Raven", "running") + loop = _loop_stub(live_run_ids=frozenset({RUN_ID})) + + with activity_mod.collecting(live_key=node_live_key(RUN_ID, "n1")): + activity_mod.note_usage({"input_tokens": 7, "output_tokens": 1}) + activity_mod.note_tool_call("read_file") + row = (await tasks_list({"session_key": SESSION}, agent_loop_factory=_factory(loop)))["tasks"][0] + + n1, n2 = row["nodes"] + assert n1["status"] == "running" and n1["tokens_in"] == 7 and n1["tokens_out"] == 1 + assert n1["tool_call_count"] == 1 and n1["tool_failure_count"] == 0 + assert n2["tokens_in"] is None and n2["tool_call_count"] is None diff --git a/ui-web/src/features/tasks/TasksPage.test.tsx b/ui-web/src/features/tasks/TasksPage.test.tsx index f41c0ba1a..f3a486f01 100644 --- a/ui-web/src/features/tasks/TasksPage.test.tsx +++ b/ui-web/src/features/tasks/TasksPage.test.tsx @@ -827,6 +827,29 @@ describe('the node panel', () => { vi.useRealTimers() } }) + + /* The row rides the same beat: a running node's usage grows on the + server with no frame to carry it, so the subtitle's token total is + read again with the record. */ + it("re-reads a running spawn's row on the beat, so its usage reaches the subtitle", async () => { + vi.useFakeTimers() + try { + const running = task({ + id: 's1', kind: 'spawn', status: 'running', nodes: [node({ node_id: 's1', status: 'running', started_at: 1000 })], + }) + rows = [running] + store.set((prev) => ({ ...prev, rows: [running], loaded: true })) + pick(running) + await act(async () => {}) + expect(document.querySelector('.tksub')?.textContent).not.toContain('gui.tasks.tokens_n') + + rows = [{ ...running, nodes: [node({ node_id: 's1', status: 'running', started_at: 1000, tokens_in: 1200, tokens_out: 34 })] }] + await act(async () => { vi.advanceTimersByTime(1000) }) + expect(document.querySelector('.tksub')?.textContent).toContain('gui.tasks.tokens_n {"n":"1,234"}') + } finally { + vi.useRealTimers() + } + }) }) describe('the context tab while the record is loading or failed', () => { diff --git a/ui-web/src/features/tasks/TasksPage.tsx b/ui-web/src/features/tasks/TasksPage.tsx index dedbf5348..0c3f13c66 100644 --- a/ui-web/src/features/tasks/TasksPage.tsx +++ b/ui-web/src/features/tasks/TasksPage.tsx @@ -378,6 +378,12 @@ function useNodeRecord(row: TaskRow, node: TaskNode): RecordLoad { const src = store.source() if (!src) { setState((prev) => ({ loading: false, record: prev.record, failed: true })); return } reading.current = true + /* The row too, while the node runs: its usage and tool counts grow on the + server as the lane reports them (`tasks.list` reads the live activity), + and no frame carries them -- so the subtitle's token total moves with + the record. Once the node settles, the terminal frame's own reconcile + brings the final copy. */ + if (node.status === 'running') void store.reconcile(row.kind, row.id) src.node(row, node) .then((r) => { if (alive) setState({ loading: false, record: r, failed: false }) }) .catch(() => { if (alive) setState((prev) => ({ loading: false, record: prev.record, failed: true })) }) diff --git a/ui-web/src/features/tasks/store.test.ts b/ui-web/src/features/tasks/store.test.ts index e6e994fa0..14ee34df9 100644 --- a/ui-web/src/features/tasks/store.test.ts +++ b/ui-web/src/features/tasks/store.test.ts @@ -250,6 +250,32 @@ describe('live event consumers', () => { expect(store.byKey('dag', 'r1')?.task_summary).not.toBe('reconciled') }) + + /* A row read started on a running node's beat can be answered after a + terminal frame has already moved the row; that answer is the older copy + and must not put the row back to running. */ + it('reconcile keeps a frame that landed while its read was out', async () => { + store.set((prev) => ({ ...prev, rows: [row({ id: 's1', kind: 'spawn', status: 'running' })], loaded: true })) + let release: (r: TaskRow | null) => void = () => {} + let reads = 0 + source.one = () => { + reads += 1 + /* The first read (the beat's) hangs; the frame's own reconcile, which + comes second, answers the settled row at once. */ + if (reads === 1) return new Promise((res) => { release = res }) + return Promise.resolve(row({ id: 's1', kind: 'spawn', status: 'completed', task_summary: 'settled' })) + } + + const onBeat = store.reconcile('spawn', 's1') + store.onSubagentStatus({ task_id: 't1', call_id: 's1', agent: 'raven', label: 'x', status: 'completed', ended_at: 2000 }) + await Promise.resolve() + await Promise.resolve() + release(row({ id: 's1', kind: 'spawn', status: 'running' })) + await onBeat + + expect(store.byKey('spawn', 's1')?.status).toBe('completed') + expect(store.byKey('spawn', 's1')?.task_summary).toBe('settled') + }) }) /* The read that fills the panel and the frames that move it race: the server diff --git a/ui-web/src/features/tasks/store.ts b/ui-web/src/features/tasks/store.ts index 62106c70e..8a38d6b16 100644 --- a/ui-web/src/features/tasks/store.ts +++ b/ui-web/src/features/tasks/store.ts @@ -213,18 +213,26 @@ export function setFold(nodeKey: string, fold: string, open: boolean): void { } /* The server's own read for one row, folded back over whatever a live event - already guessed. Used after every terminal live event and after a stop: a - frame carries no tokens, no files and no final error text, and a stop's own - answer is a bare `found` flag. */ -async function reconcile(kind: TaskKind, id: string): Promise { + already guessed. Used after every terminal live event, after a stop, and on + each read of a running node's record (`TasksPage.tsx`'s `useNodeRecord`): + a frame carries no tokens, no files and no final error text, a stop's own + answer is a bare `found` flag, and a running node's usage and counts grow + on the server with no frame to carry them. */ +export async function reconcile(kind: TaskKind, id: string): Promise { const src = source() if (!src) return const key = sessionCurrent() + const tick = liveTick const row = await src.one(kind, id).catch(() => null) /* Same guard as refresh: a row read for the conversation the reader has since left does not belong in the one they are looking at now, and a stop reconciled after the switch must not re-insert it either. */ if (!row || key !== sessionCurrent()) return + /* And the same rule as refresh for a frame that landed while the read was + out: the answer is then the older copy of this row, and the frame's own + reconcile brings the newer one a moment later. Without this a read + started on the beat could put a settled row back to running. */ + if ((liveAt.get(rowKey(row)) ?? 0) > tick) return const now = store.get().rows const at = now.findIndex((r) => r.kind === kind && r.id === id) patch({ rows: at >= 0 ? now.map((r, i) => (i === at ? row : r)) : [row, ...now] }) From 6578c175dd7695ca0cfbece4f6ba50c0bde7ce87 Mon Sep 17 00:00:00 2001 From: 0xKT Date: Mon, 21 Sep 2026 00:52:18 +0800 Subject: [PATCH 5/6] fix(*): keep a finished dag node's usage until the manifest, and gate the overlay Two holes in the live usage read. A dag node's account reaches disk with the run's manifest, written once the whole run is over, and collecting() drops the live entry the moment the node's block exits -- so a node that finished while its run was still going read as reporting nothing again, and the total the panel had shown while it ran vanished until the run ended. The runner now sets the node's account aside (activity.record_settled) as the node settles, tasks.list reads it there when the manifest has no entry yet, and the run forgets the set-aside copies once the manifest carries them, or when a hard stop means no manifest ever will. The overlay itself now applies to a running node only: the live index is keyed by a record id that is unique per conversation, so a finished spawn could have taken the numbers of another conversation's run under the same id. On the page, one row read is out at a time, so a dag node's per-tool-call frames do not stack tasks.list requests. Co-authored-by: Claude (claude-fable-5-1) --- .../2026-09-18-desk-tasks-list-design.md | 6 +- raven/agent/subagent/activity.py | 32 ++++++++- raven/agent/subagent/dag_runner.py | 11 ++++ raven/rpc/methods/tasks.py | 19 ++++-- tests/test_rpc_tasks.py | 66 +++++++++++++++++++ tests/test_subagent_activity.py | 13 ++++ tests/test_subagent_dag_runner.py | 62 +++++++++++++++++ ui-web/src/features/tasks/TasksPage.test.tsx | 38 +++++++++++ ui-web/src/features/tasks/TasksPage.tsx | 12 +++- 9 files changed, 248 insertions(+), 11 deletions(-) diff --git a/docs/specs/2026-09-18-desk-tasks-list-design.md b/docs/specs/2026-09-18-desk-tasks-list-design.md index b451358e6..ea44412df 100644 --- a/docs/specs/2026-09-18-desk-tasks-list-design.md +++ b/docs/specs/2026-09-18-desk-tasks-list-design.md @@ -83,7 +83,11 @@ stores it uncapped; a spawn's is the head of `.error.md`, or of `.out.md` when ` (`{path, op: write|edit, add, del, size}`). While a node runs, the usage, the tool counts and the files are read off the activity being collected for it in this process (the same in-memory account `subagent.context` and `dag.node` serve a transcript from), since the record -on disk carries them only once the run finishes; what the lane has not reported yet stays null. +on disk carries them only once the run finishes; what the lane has not reported yet stays null, +and a node that is not running takes nothing from that index (its key is a record id unique per +conversation only). A dag node that has finished while its run has not keeps the account the +runner set aside for it at its end (`activity.record_settled`) until the manifest is written, +so its usage does not vanish between the two. ### 2. Status derivation diff --git a/raven/agent/subagent/activity.py b/raven/agent/subagent/activity.py index daa8a2422..f429b0513 100644 --- a/raven/agent/subagent/activity.py +++ b/raven/agent/subagent/activity.py @@ -27,7 +27,7 @@ from __future__ import annotations import time -from collections.abc import Awaitable, Callable, Iterator +from collections.abc import Awaitable, Callable, Iterable, Iterator from contextlib import contextmanager from contextvars import ContextVar from dataclasses import dataclass, field @@ -205,6 +205,33 @@ def as_meta(self) -> dict[str, Any]: their ``collecting`` block.""" +_settled: dict[str, dict[str, Any]] = {} +"""The account of a run's node that has finished while its run has not. + +A dag node's account reaches disk with the run's manifest, written once the +whole run is over; between the node's own end (when ``collecting`` drops it from +``_live``) and that write, nothing else holds it, and a reader that showed the +node's usage while it ran would show nothing the moment it finished. The runner +records the account here as the node settles and forgets the run's entries once +the manifest is written.""" + + +def record_settled(key: str, meta: dict[str, Any]) -> None: + """Set aside a finished node's account under its live key.""" + _settled[key] = dict(meta) + + +def settled(key: str) -> dict[str, Any] | None: + """The account of a node that finished while its run has not, or None.""" + return _settled.get(key) + + +def forget_settled(keys: Iterable[str]) -> None: + """Drop the accounts a run set aside, once its manifest carries them.""" + for key in keys: + _settled.pop(key, None) + + _live_instances: dict[tuple[str, str, str], RunActivity] = {} """The same activities, addressed the way a *conversation* reader has to ask. @@ -595,9 +622,12 @@ def note_transcript(messages: list[dict[str, Any]] | None) -> None: "RunActivity", "collecting", "current", + "forget_settled", "set_transcript", "live", "live_instance", + "record_settled", + "settled", "merge_file_change", "note_alive", "note_closing", diff --git a/raven/agent/subagent/dag_runner.py b/raven/agent/subagent/dag_runner.py index 35bfe4efd..de46b31cf 100644 --- a/raven/agent/subagent/dag_runner.py +++ b/raven/agent/subagent/dag_runner.py @@ -683,6 +683,9 @@ async def run_dag( await _reap_carried(carried) _mark_stopped(status) await _record_outcome(store, status, cancelled=True) + # No manifest will carry them: a stopped run's set-aside accounts + # would otherwise outlive it for the life of the process. + activity.forget_settled(node_live_key(store.run_id, nid) for nid in status) # This run's own memory-record pollers, scheduled for nodes that had # already completed before this cancellation landed, would otherwise # keep polling with nothing left to reap them. @@ -1681,6 +1684,11 @@ async def _run_node( finally: stall_watch.cancel() node_activity[node.id] = did.as_meta() + # Set aside for the reader until the manifest carries + # it: `collecting` drops the live entry as this block + # exits, and the manifest is written once the whole + # run is over. + activity.record_settled(node_live_key(store.run_id, node.id), node_activity[node.id]) # The whole answer when the reply cap cut one: the in-context copy of # a terminal output is capped again on the way out (see # `_terminal_outputs`), and this file is what the reader, the next @@ -1958,6 +1966,9 @@ async def _finalize( summary = _tally(status) await store.write_manifest(manifest) + # The manifest now carries every node's account; the copies set aside for + # the nodes that finished before it was written have done their job. + activity.forget_settled(node_live_key(store.run_id, nid) for nid in by_id) await _record_outcome(store, status) return DagRunResult( run_id=store.run_id, diff --git a/raven/rpc/methods/tasks.py b/raven/rpc/methods/tasks.py index 197967efb..f19d7949f 100644 --- a/raven/rpc/methods/tasks.py +++ b/raven/rpc/methods/tasks.py @@ -124,7 +124,10 @@ def _overlay_live(node: dict[str, Any], live: Any) -> None: nothing for the whole of its run. Only what the lane has said so far is taken: a lane that has not spoken keeps its null. """ - if live is None: + # A node that is not running has an account of its own on disk (or none), + # and the live index is keyed by a record id that is unique per conversation + # only: another conversation's run under the same id must not fill it. + if live is None or node["status"] != "running": return for key in ("tokens_in", "tokens_out"): if node[key] is None: @@ -414,6 +417,10 @@ def _dag_row( continue nid = gnode["id"] entry = manifest.get(nid) if isinstance(manifest.get(nid), dict) else None + live_key = node_live_key(run_dir.name, nid) + # A node that finished while its run has not has no manifest entry yet; + # its account is what the runner set aside at the node's end. + account = entry if entry is not None else (run_activity.settled(live_key) or {}) status, started, ended = _dag_node_state(run_dir.name, nid, entry, registry_nodes, by_node) if status in _NOT_LIVE_PENDING and not live: status = "interrupted" @@ -421,7 +428,7 @@ def _dag_row( # Never ran: the registry stamps it with the moment the run was # finalized, which is not a clock this node ever had. started = ended = None - call_count, failure_count = _tool_counts(entry or {}) + call_count, failure_count = _tool_counts(account) node: dict[str, Any] = { "node_id": nid, "node_summary": gnode.get("node_summary") or None, @@ -432,13 +439,13 @@ def _dag_row( "started_at": started, "ended_at": ended, "error": _clip((entry or {}).get("error")), - "tokens_in": _int_or_none((entry or {}).get("tokens_in")), - "tokens_out": _int_or_none((entry or {}).get("tokens_out")), + "tokens_in": _int_or_none(account.get("tokens_in")), + "tokens_out": _int_or_none(account.get("tokens_out")), "tool_call_count": call_count, "tool_failure_count": failure_count, "has_output": _dag_has_output(nid, entry, registry_nodes.get(nid), nodes_dir), "prompt_template": gnode.get("prompt_template") or None, - "files": _files_of(entry or {}), + "files": _files_of(account), } if gnode.get("inputs") is not None: node["inputs"] = gnode["inputs"] @@ -446,7 +453,7 @@ def _dag_row( node["skills"] = list(gnode["skills"]) if gnode.get("mcps") is not None: node["mcps"] = list(gnode["mcps"]) - _overlay_live(node, run_activity.live(node_live_key(run_dir.name, nid))) + _overlay_live(node, run_activity.live(live_key)) nodes.append(node) statuses.append(status) if isinstance(started, int): diff --git a/tests/test_rpc_tasks.py b/tests/test_rpc_tasks.py index bcb7368e3..f98df53e6 100644 --- a/tests/test_rpc_tasks.py +++ b/tests/test_rpc_tasks.py @@ -1011,3 +1011,69 @@ async def test_a_running_dag_node_reads_usage_off_the_live_activity(workspace: P assert n1["status"] == "running" and n1["tokens_in"] == 7 and n1["tokens_out"] == 1 assert n1["tool_call_count"] == 1 and n1["tool_failure_count"] == 0 assert n2["tokens_in"] is None and n2["tool_call_count"] is None + + +async def test_a_dag_node_that_finished_mid_run_reads_the_account_the_runner_set_aside(workspace: Path) -> None: + """Between a node's end and the run's manifest, the node's account lives in + the settled index; the row reads it there, and reads nothing once the run + has forgotten it (the manifest carries it by then).""" + from raven.agent.subagent import activity as activity_mod + from raven.agent.subagent.dag_store import node_live_key + from raven.agent.subagent.instances import get_registry + + session_dir = _session_dir(workspace) + await _make_run(session_dir, RUN_ID, _GRAPH, ["n1", "n2"]) + await get_registry().upsert_dag_node(SESSION, RUN_ID, "n1", "Raven", "completed") + await get_registry().upsert_dag_node(SESSION, RUN_ID, "n2", "Raven", "running") + loop = _loop_stub(live_run_ids=frozenset({RUN_ID})) + key = node_live_key(RUN_ID, "n1") + activity_mod.record_settled( + key, + { + "tokens_in": 7, + "tokens_out": 1, + "tool_calls": ["read_file"], + "tool_failures": ["read_file"], + "files": [{"path": "a.md", "op": "write", "add": 1, "del": 0, "size": 5}], + }, + ) + try: + row = (await tasks_list({"session_key": SESSION}, agent_loop_factory=_factory(loop)))["tasks"][0] + finally: + activity_mod.forget_settled([key]) + + n1, n2 = row["nodes"] + assert n1["status"] == "completed" + assert n1["tokens_in"] == 7 and n1["tokens_out"] == 1 + assert n1["tool_call_count"] == 1 and n1["tool_failure_count"] == 1 + assert n1["files"] == [{"path": "a.md", "op": "write", "add": 1, "del": 0, "size": 5}] + assert n2["tokens_in"] is None and n2["tool_call_count"] is None + + after = (await tasks_list({"session_key": SESSION}, agent_loop_factory=_factory(loop)))["tasks"][0]["nodes"][0] + assert after["tokens_in"] is None and after["tool_call_count"] is None and after["files"] == [] + + +async def test_a_settled_spawn_ignores_a_live_activity_under_its_key(workspace: Path) -> None: + """The live index is keyed by a record id that is unique per conversation + only: a finished spawn must not take the numbers of another run collected + under the same id.""" + from raven.agent.subagent import activity as activity_mod + + session_dir = _session_dir(workspace) + await _claim_spawn_node(session_dir, "shared") + record = SpawnRecord.open( + session_dir, + task_id="shared", + task="done already", + meta=_spawn_meta(agent="Raven", handle="shared", task_summary="Shared"), + node_id="shared", + ) + record.finish(status="completed", output="done") + + with activity_mod.collecting(live_key="shared"): + activity_mod.note_usage({"prompt_tokens": 40, "completion_tokens": 2}) + activity_mod.note_tool_call("exec") + node = (await tasks_list({"session_key": SESSION}))["tasks"][0]["nodes"][0] + + assert node["status"] == "completed" + assert node["tokens_in"] is None and node["tool_call_count"] is None and node["files"] == [] diff --git a/tests/test_subagent_activity.py b/tests/test_subagent_activity.py index 65ef1324f..5e46bc165 100644 --- a/tests/test_subagent_activity.py +++ b/tests/test_subagent_activity.py @@ -150,3 +150,16 @@ def test_note_file_change_needs_a_collector_and_a_path() -> None: activity.note_file_change("", "write", 1, 0, 0) activity.note_file_change("kept.md", "edit", 2, 1, 9) assert did.files == [{"path": "kept.md", "op": "edit", "add": 2, "del": 1, "size": 9}] + + +def test_a_settled_account_is_kept_until_its_run_forgets_it(): + """A node's account outlives its collecting block only through this index: + the manifest that carries it is written when the whole run ends.""" + meta = {"tokens_in": 7, "tool_calls": ["read_file"]} + activity.record_settled("dag:r1:a", meta) + meta["tokens_in"] = 99 + assert activity.settled("dag:r1:a") == {"tokens_in": 7, "tool_calls": ["read_file"]}, ( + "a copy, not the runner's dict" + ) + activity.forget_settled(["dag:r1:a", "dag:r1:never-recorded"]) + assert activity.settled("dag:r1:a") is None diff --git a/tests/test_subagent_dag_runner.py b/tests/test_subagent_dag_runner.py index 73c37b437..2b5338d21 100644 --- a/tests/test_subagent_dag_runner.py +++ b/tests/test_subagent_dag_runner.py @@ -8618,3 +8618,65 @@ async def _announce(run_id, node_id, report, origin, **_): assert result.summary["completed"] == 0, "an unjudgeable node must not pass" assert result.summary["failed"] == 1 + + +class _SettledPeeker(_FakeExec): + """Node a reports usage; node b, which runs after it, reads the account a + set aside at its end -- the manifest that will carry it is not written yet.""" + + def __init__(self) -> None: + super().__init__() + self.run_id = "" + self.seen_settled: list[object] = [] + + async def run(self, task: str, **kw) -> str: + from raven.agent.subagent import activity + from raven.agent.subagent.dag_store import node_live_key + + if kw["task_id"] == "a": + activity.note_usage({"prompt_tokens": 5, "completion_tokens": 1}) + else: + self.seen_settled.append(activity.settled(node_live_key(self.run_id, "a"))) + return await super().run(task, **kw) + + +async def test_a_finished_nodes_account_is_set_aside_until_the_manifest_is_written() -> None: + """Between a node's end and the run's manifest a reader has nowhere else to + find the node's usage; the runner sets it aside, and the manifest takes it + over.""" + import raven.agent.subagent.dag_runner as runner_mod + from raven.agent.subagent import activity + from raven.agent.subagent.dag_store import node_live_key + + spec = parse_dag_spec( + { + "task_summary": "run the graph under test", + "nodes": [ + {"id": "a", "subagent": "x", "node_summary": "node a", "prompt_template": "hello"}, + {"id": "b", "subagent": "x", "node_summary": "node b", "prompt_template": "then", "depends_on": ["a"]}, + ], + } + ) + exec_ = _SettledPeeker() + mint = runner_mod.make_run_id + + def _mint() -> str: + exec_.run_id = mint() + return exec_.run_id + + runner_mod.make_run_id = _mint + try: + result = await run_dag( + spec, + resolve=_by_name({"x": exec_}), + backend=_InMemBackend(), + workdir="/w", + run_root="/hist/mas_dag", + nodes_root="/hist/nodes", + history_root="/hist", + ) + finally: + runner_mod.make_run_id = mint + + assert exec_.seen_settled == [{"tokens_in": 5, "tokens_out": 1}], "b saw a's account while the run was going" + assert activity.settled(node_live_key(result.run_id, "a")) is None, "and the manifest took it over" diff --git a/ui-web/src/features/tasks/TasksPage.test.tsx b/ui-web/src/features/tasks/TasksPage.test.tsx index f3a486f01..59c5a6027 100644 --- a/ui-web/src/features/tasks/TasksPage.test.tsx +++ b/ui-web/src/features/tasks/TasksPage.test.tsx @@ -850,6 +850,44 @@ describe('the node panel', () => { vi.useRealTimers() } }) + + it("re-reads a running dag node's row on node_updated, so its usage reaches the subtitle", async () => { + const running = task({ + id: 'r1', kind: 'dag', status: 'running', nodes: [node({ node_id: 'n1', status: 'running', started_at: 1000 })], + }) + rows = [running] + store.set((prev) => ({ ...prev, rows: [running], loaded: true })) + pick(running) + await act(async () => {}) + expect(document.querySelector('.tksub')?.textContent).not.toContain('gui.tasks.tokens_n') + + rows = [{ ...running, nodes: [node({ node_id: 'n1', status: 'running', started_at: 1000, tokens_in: 4000, tokens_out: 910 })] }] + await act(async () => { store.onNodeUpdated({ run_id: 'r1', node: 'n1', status: 'running', tool_call_id: 'c1' }) }) + expect(document.querySelector('.tksub')?.textContent).toContain('gui.tasks.tokens_n {"n":"4,910"}') + }) + + it('keeps one row read out at a time across node_updated frames', async () => { + let reads = 0 + let release: ((r: TaskRow | null) => void) | null = null + const running = task({ + id: 'r1', kind: 'dag', status: 'running', nodes: [node({ node_id: 'n1', status: 'running' })], + }) + setSources({ + tasks: { ...source(), one: () => { reads += 1; return new Promise((res) => { release = res }) } }, + workspace: { shortPath: (p: string) => p, hostPlatform: () => 'mac', canBrowse: false, openPath: () => {} }, + }) + store.set((prev) => ({ ...prev, rows: [running], loaded: true })) + pick(running) + await act(async () => {}) + expect(reads).toBe(1) + await act(async () => { store.onNodeUpdated({ run_id: 'r1', node: 'n1', status: 'running', tool_call_id: 'c1' }) }) + await act(async () => { store.onNodeUpdated({ run_id: 'r1', node: 'n1', status: 'running', tool_call_id: 'c2' }) }) + /* Two frames while the first row read is still out: no second read. */ + expect(reads).toBe(1) + await act(async () => { release?.(null) }) + await act(async () => { store.onNodeUpdated({ run_id: 'r1', node: 'n1', status: 'running', tool_call_id: 'c3' }) }) + expect(reads).toBe(2) + }) }) describe('the context tab while the record is loading or failed', () => { diff --git a/ui-web/src/features/tasks/TasksPage.tsx b/ui-web/src/features/tasks/TasksPage.tsx index 0c3f13c66..738c8c913 100644 --- a/ui-web/src/features/tasks/TasksPage.tsx +++ b/ui-web/src/features/tasks/TasksPage.tsx @@ -371,6 +371,7 @@ function useNodeRecord(row: TaskRow, node: TaskNode): RecordLoad { { loading: dispatched, record: null, failed: false }, ) const reading = useRef(false) + const reconciling = useRef(false) useEffect(() => { if (!dispatched) { setState({ loading: false, record: null, failed: false }); return } setState((prev) => ({ loading: true, record: prev.record, failed: false })) @@ -381,9 +382,14 @@ function useNodeRecord(row: TaskRow, node: TaskNode): RecordLoad { /* The row too, while the node runs: its usage and tool counts grow on the server as the lane reports them (`tasks.list` reads the live activity), and no frame carries them -- so the subtitle's token total moves with - the record. Once the node settles, the terminal frame's own reconcile - brings the final copy. */ - if (node.status === 'running') void store.reconcile(row.kind, row.id) + the record. One row read out at a time: a dag node's frames arrive once + per tool call, and stacking a read per frame would multiply requests + the way the record's own `reading` guard exists to prevent. Once the + node settles, the terminal frame's own reconcile brings the final copy. */ + if (node.status === 'running' && !reconciling.current) { + reconciling.current = true + void store.reconcile(row.kind, row.id).finally(() => { reconciling.current = false }) + } src.node(row, node) .then((r) => { if (alive) setState({ loading: false, record: r, failed: false }) }) .catch(() => { if (alive) setState((prev) => ({ loading: false, record: prev.record, failed: true })) }) From 2612503d0c4a7ea0b7d43e4a51a9ea4854f5664d Mon Sep 17 00:00:00 2001 From: 0xKT Date: Mon, 21 Sep 2026 01:10:39 +0800 Subject: [PATCH 6/6] fix(ui-web): re-read a dag node's row on its own terminal frame A dag node's terminal node_updated frame moved the node and bumped its record version, but re-read no row: the node panel's row read is for a running node, and the frame itself scheduled no reconcile. The account the runner sets aside as the node settles therefore stayed off the page until dag.run_completed -- with a sibling still running, the rest of the run. applyNodeUpdated now answers a refetch for a node's own terminal frame, the way the spawn and run frames already do, and onNodeUpdated applies it; a row read that was out when the frame landed is still dropped by the liveTick guard, and the frame's own reconcile is the replacement. Co-authored-by: Claude (claude-fable-5-1) --- .../2026-09-18-desk-tasks-list-design.md | 4 +- ui-web/src/features/tasks/TasksPage.test.tsx | 45 +++++++++++++++++++ ui-web/src/features/tasks/live.test.ts | 23 +++++++--- ui-web/src/features/tasks/live.ts | 19 ++++++-- ui-web/src/features/tasks/store.test.ts | 31 +++++++++++++ ui-web/src/features/tasks/store.ts | 8 ++-- 6 files changed, 117 insertions(+), 13 deletions(-) diff --git a/docs/specs/2026-09-18-desk-tasks-list-design.md b/docs/specs/2026-09-18-desk-tasks-list-design.md index ea44412df..f5cb67507 100644 --- a/docs/specs/2026-09-18-desk-tasks-list-design.md +++ b/docs/specs/2026-09-18-desk-tasks-list-design.md @@ -134,7 +134,9 @@ already reads on. Each such read of a running node also re-reads its row through `dag.run_started` builds a row (`status=running`; `nodes[].subagent` becomes `agent`; no timestamps, so `started_at` starts from the run id's prefix); `dag.node_updated` moves a -node (timestamps only on `running` / `completed` / `failed` / `exception`); +node (timestamps only on `running` / `completed` / `failed` / `exception`) and, on a node's +own terminal frame, re-reads the row so the account the runner set aside for the node reaches +the page before the run's own end; `dag.run_completed` ends a row (a hard stop sends `{stopped: true}` and no `files`); `dag.run_replanned` marks the old row `cancelled` with `replan`; `subagent.status` builds a spawn row at `pending` with `handle = instance ?? task_id`, and `call_id` from `running` diff --git a/ui-web/src/features/tasks/TasksPage.test.tsx b/ui-web/src/features/tasks/TasksPage.test.tsx index 59c5a6027..39e512241 100644 --- a/ui-web/src/features/tasks/TasksPage.test.tsx +++ b/ui-web/src/features/tasks/TasksPage.test.tsx @@ -888,6 +888,51 @@ describe('the node panel', () => { await act(async () => { store.onNodeUpdated({ run_id: 'r1', node: 'n1', status: 'running', tool_call_id: 'c3' }) }) expect(reads).toBe(2) }) + + /* The node's own terminal frame brings its final usage while a sibling + keeps the run going -- and a row read that was out when the frame + landed is the older copy, dropped rather than put over the frame. */ + it("shows a settled dag node's final usage while a sibling still runs, over a stale read that was out", async () => { + let reads = 0 + let release: ((r: TaskRow | null) => void) | null = null + const before = task({ + id: 'r1', kind: 'dag', status: 'running', + nodes: [node({ node_id: 'n1', status: 'running', started_at: 1000 }), node({ node_id: 'n2', status: 'pending' })], + }) + const after = task({ + id: 'r1', kind: 'dag', status: 'running', + nodes: [ + node({ node_id: 'n1', status: 'completed', started_at: 1000, ended_at: 2000, tokens_in: 4000, tokens_out: 910 }), + node({ node_id: 'n2', status: 'running', started_at: 2000 }), + ], + }) + setSources({ + tasks: { + ...source(), + one: () => { + reads += 1 + /* The first read (the panel's, on open) hangs; the frame's own + reconcile, which comes second, answers the settled row. */ + if (reads === 1) return new Promise((res) => { release = res }) + return Promise.resolve(after) + }, + }, + workspace: { shortPath: (p: string) => p, hostPlatform: () => 'mac', canBrowse: false, openPath: () => {} }, + }) + store.set((prev) => ({ ...prev, rows: [before], loaded: true })) + pick(before) + await act(async () => {}) + expect(reads).toBe(1) + + await act(async () => { store.onNodeUpdated({ run_id: 'r1', node: 'n1', status: 'completed', ended_at: 2000 }) }) + expect(reads).toBe(2) + expect(document.querySelector('.tksub')?.textContent).toBe('raven · gui.tasks.node_st_completed · 1s · gui.tasks.tokens_n {"n":"4,910"}') + + /* The stale read lands now, still saying n1 runs with no usage. */ + await act(async () => { release?.(before) }) + expect(document.querySelector('.tksub')?.textContent).toBe('raven · gui.tasks.node_st_completed · 1s · gui.tasks.tokens_n {"n":"4,910"}') + expect(store.byKey('dag', 'r1')?.status).toBe('running') + }) }) describe('the context tab while the record is loading or failed', () => { diff --git a/ui-web/src/features/tasks/live.test.ts b/ui-web/src/features/tasks/live.test.ts index 6c4db5162..0de4eb8fd 100644 --- a/ui-web/src/features/tasks/live.test.ts +++ b/ui-web/src/features/tasks/live.test.ts @@ -87,7 +87,7 @@ describe('applyNodeUpdated', () => { }] it('moves the named node and recomputes the row status', () => { - const rows = applyNodeUpdated(base, { run_id: 'r1', node: 'a', status: 'completed', ended_at: 2000 }) + const { rows } = applyNodeUpdated(base, { run_id: 'r1', node: 'a', status: 'completed', ended_at: 2000 }) const row = rows[0]! expect(row.nodes[0]).toMatchObject({ status: 'completed', ended_at: 2000 }) /* b is still pending, so the row as a whole is still running. */ @@ -95,18 +95,31 @@ describe('applyNodeUpdated', () => { }) it('leaves the timestamps alone when the frame carries none, as a skip does', () => { - const rows = applyNodeUpdated(base, { run_id: 'r1', node: 'b', status: 'skipped' }) + const { rows } = applyNodeUpdated(base, { run_id: 'r1', node: 'b', status: 'skipped' }) expect(rows[0]!.nodes[1]).toMatchObject({ status: 'skipped', started_at: null, ended_at: null }) }) it('replaying the same frame lands on the same rows', () => { const once = applyNodeUpdated(base, { run_id: 'r1', node: 'a', status: 'completed', ended_at: 2000 }) - const twice = applyNodeUpdated(once, { run_id: 'r1', node: 'a', status: 'completed', ended_at: 2000 }) - expect(twice).toEqual(once) + const twice = applyNodeUpdated(once.rows, { run_id: 'r1', node: 'a', status: 'completed', ended_at: 2000 }) + expect(twice.rows).toEqual(once.rows) }) it('does nothing for a run this page holds no row for', () => { - expect(applyNodeUpdated(base, { run_id: 'other', node: 'a', status: 'completed' })).toEqual(base) + const { rows, refetch } = applyNodeUpdated(base, { run_id: 'other', node: 'a', status: 'completed' }) + expect(rows).toEqual(base) + expect(refetch).toBeUndefined() + }) + + /* The frame carries no usage; the runner sets the node's account aside as + the node settles, and only a read brings it. */ + it("asks for a reconcile on a node's own terminal frame, and not before", () => { + expect(applyNodeUpdated(base, { run_id: 'r1', node: 'a', status: 'running', tool_call_id: 'c1' }).refetch).toBeUndefined() + /* `interrupted` is the reader's own word for a run the gateway lost, never + a frame's, so the wire type does not carry it. */ + for (const status of ['completed', 'failed', 'skipped', 'cancelled', 'exception'] as const) { + expect(applyNodeUpdated(base, { run_id: 'r1', node: 'a', status }).refetch).toEqual({ kind: 'dag', id: 'r1' }) + } }) }) diff --git a/ui-web/src/features/tasks/live.ts b/ui-web/src/features/tasks/live.ts index c93edfb79..7cbfbeb54 100644 --- a/ui-web/src/features/tasks/live.ts +++ b/ui-web/src/features/tasks/live.ts @@ -108,17 +108,30 @@ export function applyRunStarted(rows: readonly TaskRow[], p: RunStartedPayload): return [row, ...rows] } +/* The statuses a node's own frame can settle it into: its collecting block + has exited, so the account the runner set aside for it is final. */ +const NODE_SETTLED: ReadonlySet = new Set( + ['completed', 'failed', 'skipped', 'cancelled', 'interrupted', 'exception'], +) + /* Moves one node. Timestamps only where the event actually carries them -- a skipped/cancelled transition has none (contract §5.1) -- and the row's - own status and counts are recomputed from the whole node list every time. */ -export function applyNodeUpdated(rows: readonly TaskRow[], p: NodeUpdatedPayload): TaskRow[] { - return rows.map((row) => { + own status and counts are recomputed from the whole node list every time. + A node's own terminal frame asks for a reconcile: the frame carries no + usage, the runner sets the node's account aside as it settles, and only a + read brings that final total -- the run's own terminal frame reconciles + too, but with a sibling still running it can be minutes away. */ +export function applyNodeUpdated(rows: readonly TaskRow[], p: NodeUpdatedPayload): LiveResult { + const at = rows.findIndex((r) => r.kind === 'dag' && r.id === p.run_id) + if (at < 0) return { rows: [...rows] } + const next = rows.map((row) => { if (row.kind !== 'dag' || row.id !== p.run_id) return row const nodes = row.nodes.map((n) => (n.node_id === p.node ? { ...n, status: p.status, started_at: p.started_at ?? n.started_at, ended_at: p.ended_at ?? n.ended_at } : n)) return { ...row, nodes, counts: countsOf(nodes), status: deriveStatus(nodes) } }) + return { rows: next, refetch: NODE_SETTLED.has(p.status) ? { kind: 'dag', id: p.run_id } : undefined } } /* Ends the frame. `files` absent (or empty) is the hard-cancel shape -- diff --git a/ui-web/src/features/tasks/store.test.ts b/ui-web/src/features/tasks/store.test.ts index 14ee34df9..5d0d66dfc 100644 --- a/ui-web/src/features/tasks/store.test.ts +++ b/ui-web/src/features/tasks/store.test.ts @@ -234,6 +234,37 @@ describe('live event consumers', () => { expect(store.byKey('dag', 'r1')?.task_summary).toBe('reconciled') }) + /* A node's own terminal frame too: with a sibling still running the run's + terminal frame can be minutes away, and the settled node's final usage + is only on the server. */ + it("a node's terminal frame reconciles the row; a running frame does not", async () => { + const node = (over: Partial & Pick): TaskNode => ({ + agent: 'raven', depends_on: [], files: [], ...over, + }) + store.set((prev) => ({ + ...prev, + rows: [row({ id: 'r1', kind: 'dag', status: 'running', nodes: [node({ node_id: 'n1', status: 'running' }), node({ node_id: 'n2', status: 'pending' })] })], + loaded: true, + })) + let reads = 0 + source.one = async (kind, id) => { reads += 1; return rows.find((r) => r.kind === kind && r.id === id) || null } + rows = [row({ + id: 'r1', kind: 'dag', status: 'running', + nodes: [node({ node_id: 'n1', status: 'completed', tokens_in: 4000, tokens_out: 910 }), node({ node_id: 'n2', status: 'running' })], + })] + + store.onNodeUpdated({ run_id: 'r1', node: 'n1', status: 'running', tool_call_id: 'c1' }) + await Promise.resolve() + await Promise.resolve() + expect(reads).toBe(0) + + store.onNodeUpdated({ run_id: 'r1', node: 'n1', status: 'completed', ended_at: 2000 }) + await Promise.resolve() + await Promise.resolve() + expect(reads).toBe(1) + expect(store.byKey('dag', 'r1')?.nodes[0]).toMatchObject({ status: 'completed', tokens_in: 4000, tokens_out: 910 }) + }) + /* Independent of any React or effect timing: a terminal event schedules reconcile, the reader switches conversations before it answers, and the answer must not prepend or overwrite a row into the session now open. */ diff --git a/ui-web/src/features/tasks/store.ts b/ui-web/src/features/tasks/store.ts index 8a38d6b16..b640c22bf 100644 --- a/ui-web/src/features/tasks/store.ts +++ b/ui-web/src/features/tasks/store.ts @@ -89,8 +89,8 @@ export const byKey = (kind: TaskKind, id: string): TaskRow | null => * trip describes a row the answer cannot: a run dispatched in that window is * missing from it entirely, and a node the frame moved is stale in it. Neither * is recoverable afterwards -- `dag.run_started` fires once per run, and - * `applyNodeUpdated` schedules no reconcile, so whatever the answer writes - * stands until the next frame or the next read. + * `applyNodeUpdated` schedules a reconcile only on a node's own terminal frame, + * so whatever the answer writes stands until the next frame or the next read. * * A key stamped later than the tick a read captured is a row that read must * leave alone. A key stamped with no row behind it is one a frame RETIRED (a @@ -138,7 +138,7 @@ export async function refresh(): Promise { The answer is the richer copy -- it alone carries the tokens, the output files and the final error text -- but it may also be describing a node the reader has already watched finish, and nothing would correct that: - `dag.node_updated` schedules no reconcile. The other way costs nothing + a mid-run `dag.node_updated` schedules no reconcile. The other way costs nothing for long, because every terminal frame DOES schedule one (`apply`'s refetch), so the richer copy lands a moment later of its own accord. */ const fresher = (k: string): boolean => (liveAt.get(k) ?? 0) > tick @@ -250,7 +250,7 @@ export function onRunStarted(p: live.RunStartedPayload): void { liveRows(live.applyRunStarted(store.get().rows, p)) } export function onNodeUpdated(p: live.NodeUpdatedPayload): void { - liveRows(live.applyNodeUpdated(store.get().rows, p)) + apply(live.applyNodeUpdated(store.get().rows, p)) bumpNodeVersion('dag', p.run_id, p.node) } export function onRunCompleted(p: live.RunCompletedPayload): void {