fix(todo): live task state via revisioned snapshots and a dedicated todo.updated event

Salvaged from PR #97815 by @itsflownium, slimmed to the schema-free core: - TodoStore gains a monotonic in-memory revision; the todo tool result returns it so clients can reject stale updates - tui_gateway emits a dedicated todo.updated full-snapshot event that bypasses optional tool-progress display settings - session resume/activate responses attach the authoritative todo snapshot; renderer restores it with revision arbitration - desktop store tracks per-session revisions and rejects regressions The session_todo_state DB table from the original PR is intentionally dropped: canonical todo tool results already persist in conversation history, so resume paths derive the snapshot from the stored transcript instead of a parallel store.
2026-11-06 13:26:40 +00:00 · 2026-08-29 18:40:51 -07:00
1 parent cf692532cf
commit 393af4a310
@@ -4,6 +4,7 @@ import { flashPetActivity, setPetActivity } from '@/store/pet'
 import { pruneDelegateFallbackSubagents, upsertSubagent } from '@/store/subagents'
 import { reportMcpToolResult } from '@/store/suggestion-providers/repair'
 import { invalidateSkillSuggestionIndex } from '@/store/suggestion-providers/skill'
+import { restoreSessionTodosFromSnapshot } from '@/store/todos'
 import { recordToolDiff } from '@/store/tool-diffs'
 import { setSessionDraftingTool } from '@/store/tool-drafting'
 import { notifyWorkspaceChanged, toolChangedPath, toolMayMutateFiles } from '@/store/workspace-events'
@@ -17,6 +18,14 @@ export function handleToolEvent(ctx: GatewayEventContext): boolean {
  const { deps, event, payload, sessionId, isActiveEvent, occurredAt } = ctx
  const { flushQueuedDeltas, nativeSubagentSessionsRef, sessionInterrupted, updateSessionState, upsertToolCall } = deps

+  if (event.type === 'todo.updated') {
+    if (sessionId && !sessionInterrupted(sessionId)) {
+      restoreSessionTodosFromSnapshot(sessionId, payload, true)
+    }
+
+    return true
+  }
+
  if (event.type === 'tool.generating') {
    // Announced while the model is still emitting the call's JSON, so it
    // carries a name and nothing else — no id, no args. Materializing a row
@@ -23,7 +23,7 @@ import {
  generatedImageEchoSources,
  stripGeneratedImageEchoes
 } from '@/lib/generated-images'
-import { nextTodosFromToolEvent } from '@/lib/todos'
+import { nextTodosFromToolEvent, parseTodoRevision } from '@/lib/todos'
 import { dispatchNativeNotification } from '@/store/native-notifications'
 import { isDiskFullErrorMessage, notifyError } from '@/store/notifications'
 import { broadcastSessionsChanged } from '@/store/session-sync'
@@ -465,7 +465,7 @@ export function useMessageStream({
        const todos = nextTodosFromToolEvent($todosBySession.get()[sessionId] ?? [], payload)

        if (todos) {
-          setSessionTodos(sessionId, todos)
+          setSessionTodos(sessionId, todos, parseTodoRevision(payload))
        }
      }

@@ -56,4 +56,18 @@ describe('useMessageStream turn-end todo cleanup', () => {

    expect($todosBySession.get()[SID]).toBeUndefined()
  })
+
+  it('applies a dedicated todo snapshot immediately', () => {
+    mountStream()
+
+    act(() =>
+      stream.handleEvent({
+        payload: { revision: 3, todos: [todo('live', 'in_progress')] },
+        session_id: SID,
+        type: 'todo.updated'
+      })
+    )
+
+    expect($todosBySession.get()[SID]?.[0]?.id).toBe('live')
+  })
 })
@@ -118,6 +118,7 @@ import {
 import { broadcastSessionsChanged } from '@/store/session-sync'
 import { forgetSessionUnread } from '@/store/session-unread'
 import { $archivedSessions } from '@/store/sidebar-archive'
+import { restoreSessionTodosFromSnapshot } from '@/store/todos'
 import {
  dropTranscriptTail,
  dropTranscriptTailEverywhere,
@@ -1173,6 +1174,8 @@ export function useSessionActions({
                  ? false
                  : resolveResumedBusy(activated.running ?? cachedViewState.busy, Boolean(latestCachedState?.busy))

+              restoreSessionTodosFromSnapshot(cachedRuntimeId, activated.todo_state, running)
+
              const activatedTurnStartedAt =
                typeof activated.turn_started_at === 'number' && activated.turn_started_at > 0
                  ? activated.turn_started_at * 1000
@@ -1613,6 +1616,8 @@ export function useSessionActions({
          Boolean(sessionStateByRuntimeIdRef.current.get(resumed.session_id)?.busy)
        )

+        restoreSessionTodosFromSnapshot(resumed.session_id, resumed.todo_state, resumedRunning)
+
        // Crash-survivable turn progress: fold a journaled in-flight tail
        // (persisted by use-session-state-cache while the turn streamed;
        // survives renderer/app death) back onto the restored transcript. The
@@ -71,6 +71,7 @@ export type GatewayEventPayload = {
  inline_diff?: string
  duration_s?: number
  todos?: unknown
+  revision?: number
  model?: string
  provider?: string
  reasoning_effort?: string
@@ -1,6 +1,6 @@
 import { describe, expect, it } from 'vitest'

-import { latestSessionTodos, mergeTodoItems, nextTodosFromToolEvent, parseTodoPatch, parseTodos, todoTree } from './todos'
+import { latestSessionTodos, mergeTodoItems, nextTodosFromToolEvent, parseTodoPatch, parseTodoRevision, parseTodos, todoTree } from './todos'

 describe('todoTree', () => {
  it('orders parents before children with depths', () => {
@@ -79,6 +79,19 @@ describe('parseTodos', () => {
  })
 })

+describe('parseTodoRevision', () => {
+  it('parses direct and wrapped revisions', () => {
+    expect(parseTodoRevision({ revision: 3 })).toBe(3)
+    expect(parseTodoRevision({ result: '{"revision":4}' })).toBe(4)
+  })
+
+  it('rejects invalid revisions', () => {
+    expect(parseTodoRevision({ revision: -1 })).toBeNull()
+    expect(parseTodoRevision({ revision: 1.5 })).toBeNull()
+    expect(parseTodoRevision({ revision: '3' })).toBeNull()
+  })
+})
+
 describe('latestSessionTodos', () => {
  const todoPart = (todos: unknown, extra: Record<string, unknown> = {}) => ({
    type: 'tool-call',
@@ -203,6 +203,32 @@ export function nextTodosFromToolEvent(
  return parseTodos(args)
 }

+function parseRevision(value: unknown, depth: number): null | number {
+  if (depth > 2) {
+    return null
+  }
+
+  if (typeof value === 'string' && value.trim()) {
+    try {
+      return parseRevision(JSON.parse(value), depth + 1)
+    } catch {
+      return null
+    }
+  }
+
+  if (!isRecord(value)) {
+    return null
+  }
+
+  if (typeof value.revision === 'number' && Number.isSafeInteger(value.revision) && value.revision >= 0) {
+    return value.revision
+  }
+
+  return Object.hasOwn(value, 'result') ? parseRevision(value.result, depth + 1) : null
+}
+
+export const parseTodoRevision = (value: unknown): null | number => parseRevision(value, 0)
+
 /** Latest parseable todo list from one message's aui content parts (tool-call
 *  parts named `todo`; live parts carry `todos`, hydrated ones args/result). */
 export function todosFromMessageContent(content: unknown): null | TodoItem[] {
@@ -3,9 +3,11 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import type { TodoItem } from '@/lib/todos'

 import {
+  $todoRevisionsBySession,
  $todosBySession,
  clearActiveSessionTodos,
  clearSessionTodos,
+  restoreSessionTodosFromSnapshot,
  setSessionTodos,
  todosForHydration
 } from './todos'
@@ -103,3 +105,33 @@ describe('todosForHydration (stale-active guard on restore)', () => {
    expect(todosForHydration(null)).toBeNull()
  })
 })
+
+describe('revisioned snapshots', () => {
+  beforeEach(() => {
+    vi.useFakeTimers()
+    clearSessionTodos('s1')
+  })
+
+  afterEach(() => {
+    clearSessionTodos('s1')
+    vi.useRealTimers()
+  })
+
+  it('rejects a snapshot older than the latest live update', () => {
+    setSessionTodos('s1', [todo('new', 'in_progress')], 5)
+    setSessionTodos('s1', [todo('old', 'pending')], 4)
+
+    expect($todosBySession.get().s1?.[0]?.id).toBe('new')
+    expect($todoRevisionsBySession.get().s1).toBe(5)
+  })
+
+  it('restores an active snapshot only while the session is running', () => {
+    const snapshot = { revision: 7, todos: [todo('active', 'in_progress')] }
+
+    restoreSessionTodosFromSnapshot('s1', snapshot, false)
+    expect($todosBySession.get().s1).toBeUndefined()
+
+    restoreSessionTodosFromSnapshot('s1', snapshot, true)
+    expect($todosBySession.get().s1?.[0]?.id).toBe('active')
+  })
+})
@@ -2,7 +2,7 @@ import { atom, computed } from 'nanostores'

 import { keyedTimeouts } from '@/lib/keyed-timeouts'
 import { stableRecord } from '@/lib/stable-array'
-import type { TodoItem } from '@/lib/todos'
+import { parseTodoRevision, parseTodos, type TodoItem } from '@/lib/todos'

 import { $sessions, lineageAliases } from './session'
 import { $sessionStates } from './session-states'
@@ -17,6 +17,7 @@ import { $sessionStates } from './session-states'
 *   above the composer forever.
 */
 export const $todosBySession = atom<Record<string, TodoItem[]>>({})
+export const $todoRevisionsBySession = atom<Record<string, number>>({})

 export const todoListActive = (todos: readonly TodoItem[]) =>
  todos.some(t => t.status === 'pending' || t.status === 'in_progress')
@@ -69,30 +70,64 @@ export function todosForHydration(todos: readonly TodoItem[] | null): TodoItem[]
 const FINISHED_LINGER_MS = 4_000
 const clearTimers = keyedTimeouts()

-export function setSessionTodos(sid: string, todos: TodoItem[]) {
+function acceptRevision(sid: string, revision?: null | number): boolean {
+  const revisions = $todoRevisionsBySession.get()
+  const current = revisions[sid]
+
+  if (revision == null) {
+    return current == null
+  }
+
+  if (current != null && revision < current) {
+    return false
+  }
+
+  if (current !== revision) {
+    $todoRevisionsBySession.set({ ...revisions, [sid]: revision })
+  }
+
+  return true
+}
+
+export function setSessionTodos(sid: string, todos: TodoItem[], revision?: null | number) {
  if (!sid) {
    return
  }

+  if (!acceptRevision(sid, revision)) {
+    return
+  }
+
  clearTimers.cancel(sid)
  $todosBySession.set({ ...$todosBySession.get(), [sid]: todos })

  if (!todoListActive(todos)) {
-    clearTimers.schedule(sid, FINISHED_LINGER_MS, () => clearSessionTodos(sid))
+    clearTimers.schedule(sid, FINISHED_LINGER_MS, () => dropSessionTodos(sid, false))
  }
 }

-export function clearSessionTodos(sid: string) {
+function dropSessionTodos(sid: string, forgetRevision: boolean) {
  clearTimers.cancel(sid)

  const map = $todosBySession.get()

-  if (!(sid in map)) {
-    return
+  if (sid in map) {
+    const { [sid]: _drop, ...rest } = map
+    $todosBySession.set(rest)
  }

-  const { [sid]: _drop, ...rest } = map
-  $todosBySession.set(rest)
+  if (forgetRevision) {
+    const revisions = $todoRevisionsBySession.get()
+
+    if (sid in revisions) {
+      const { [sid]: _drop, ...rest } = revisions
+      $todoRevisionsBySession.set(rest)
+    }
+  }
+}
+
+export function clearSessionTodos(sid: string) {
+  dropSessionTodos(sid, true)
 }

 // Drop a still-active todo list (any pending/in_progress item) — used at turn
@@ -107,5 +142,25 @@ export function clearActiveSessionTodos(sid: string) {
    return
  }

-  clearSessionTodos(sid)
+  dropSessionTodos(sid, false)
+}
+
+/** Apply a session.resume/activate or todo.updated full snapshot. Idle
+ * sessions keep the existing stale-active guard; running sessions restore the
+ * active plan because the backend has proved that turn is still live. */
+export function restoreSessionTodosFromSnapshot(sid: string, snapshot: unknown, running: boolean) {
+  const todos = parseTodos(snapshot)
+
+  if (!sid || todos === null) {
+    return
+  }
+
+  const revision = parseTodoRevision(snapshot)
+  const visible = running ? todos : todosForHydration(todos)
+
+  if (visible !== null) {
+    setSessionTodos(sid, visible, revision)
+  } else if (acceptRevision(sid, revision)) {
+    dropSessionTodos(sid, false)
+  }
 }
@@ -683,6 +683,12 @@ export interface SessionResumeResponse {
  session_key?: string
  started_at?: number
  status?: string
+  /** Latest full task snapshot. Revisions let the renderer reject a response
+   * that raced with a newer live update. */
+  todo_state?: {
+    revision?: number
+    todos?: unknown
+  }
  /** Epoch seconds the current turn started, or null when idle. */
  turn_started_at?: number | null
 }
@@ -14,6 +14,7 @@ export type GatewayEventName =
  | 'tool.progress'
  | 'tool.complete'
  | 'tool.generating'
+  | 'todo.updated'
  | 'clarify.request'
  | 'approval.request'
  | 'sudo.request'
@@ -4790,6 +4790,7 @@ class AIAgent:

        # Walk history backwards to find the most recent todo tool response
        last_todo_response = None
+        last_todo_revision = 0
        for idx in range(len(history) - 1, -1, -1):
            msg = history[idx]
            if msg.get("role") != "tool":
@@ -4815,15 +4816,32 @@ class AIAgent:
                data = json.loads(content)
                if "todos" in data and isinstance(data["todos"], list):
                    last_todo_response = data["todos"]
+                    last_todo_revision = data.get("revision", 1)
                    break
            except (json.JSONDecodeError, TypeError):
                continue

-        if last_todo_response:
-            # Replay the items into the store (replace mode)
-            self._todo_store.write(last_todo_response, merge=False)
-            if not self.quiet_mode:
-                self._vprint(f"{self.log_prefix}📋 Restored {len(last_todo_response)} todo item(s) from history")
+        if last_todo_response is not None:
+            # Restore only when history carries a newer revision than the
+            # store already holds (a live store re-hydrated in place must not
+            # be rolled back by older history). Sessions that predate
+            # revisions default to 1 so they still hydrate. Empty lists
+            # matter: they are an authoritative clear after an earlier
+            # non-empty plan.
+            current_revision = int(
+                self._todo_store.snapshot().get("revision", 0) or 0
+            )
+            try:
+                history_revision = max(0, int(last_todo_revision or 0))
+            except (TypeError, ValueError):
+                history_revision = 1
+            if history_revision > current_revision:
+                self._todo_store.restore(
+                    last_todo_response,
+                    revision=history_revision,
+                )
+                if not self.quiet_mode:
+                    self._vprint(f"{self.log_prefix}📋 Restored {len(last_todo_response)} todo item(s) from history")
        _set_interrupt(False)

    @classmethod
@@ -904,6 +904,56 @@ class TestHydrateTodoStore:
            agent._hydrate_todo_store(history)
        assert not agent._todo_store.has_items()

+    def test_newer_live_revision_wins_over_history(self, agent):
+        agent._todo_store.restore(
+            [{"id": "db", "content": "Current", "status": "in_progress"}],
+            revision=5,
+        )
+        history = [
+            self._assistant_todo_call(),
+            {
+                "role": "tool",
+                "tool_call_id": "c1",
+                "content": json.dumps(
+                    {
+                        "todos": [
+                            {"id": "old", "content": "Old", "status": "pending"}
+                        ],
+                        "revision": 4,
+                    }
+                ),
+            },
+        ]
+
+        with patch("run_agent._set_interrupt"):
+            agent._hydrate_todo_store(history)
+
+        assert agent._todo_store.snapshot()["revision"] == 5
+        assert agent._todo_store.read()[0]["id"] == "db"
+
+    def test_history_recovers_newer_snapshot(self, agent):
+        history = [
+            self._assistant_todo_call(),
+            {
+                "role": "tool",
+                "tool_call_id": "c1",
+                "content": json.dumps(
+                    {
+                        "todos": [
+                            {"id": "new", "content": "Recovered", "status": "pending"}
+                        ],
+                        "revision": 2,
+                    }
+                ),
+            },
+        ]
+
+        with patch("run_agent._set_interrupt"):
+            agent._hydrate_todo_store(history)
+
+        assert agent._todo_store.snapshot()["revision"] == 2
+        assert agent._todo_store.read()[0]["id"] == "new"
+



@@ -130,6 +130,7 @@ class TestTodoToolFunction:
        result = json.loads(todo_tool(store=store))
        assert result["summary"]["total"] == 1
        assert result["summary"]["pending"] == 1
+        assert result["revision"] == 1


    def test_no_store_returns_error(self):
@@ -137,6 +138,30 @@ class TestTodoToolFunction:
        assert "error" in result


+class TestTodoStoreSnapshots:
+    def test_revision_only_advances_when_state_changes(self):
+        store = TodoStore()
+        items = [{"id": "1", "content": "Task", "status": "pending"}]
+
+        store.write(items)
+        first = store.snapshot()
+        store.write(items)
+
+        assert first["revision"] == 1
+        assert store.snapshot() == first
+
+    def test_restore_adopts_a_trusted_revision(self):
+        store = TodoStore()
+        store.restore(
+            [{"id": "1", "content": "Task", "status": "pending"}], revision=7
+        )
+
+        assert store.snapshot()["revision"] == 7
+
+        store.write([{"id": "1", "content": "Task", "status": "completed"}])
+        assert store.snapshot()["revision"] == 8
+
+
 class TestTodoStoreBounds:
    """Bounds on persisted todo state (GHSA-5g4g-6jrg-mw3g hardening).

@@ -0,0 +1,80 @@
+"""Todo snapshots bypass optional tool-progress display settings."""
+
+import json
+
+import tui_gateway.server as server
+
+
+def test_todo_completion_always_emits_snapshot_and_compat_event(monkeypatch):
+    sid = "todo-state-test"
+    events = []
+    session = {
+        "agent": None,
+        "edit_snapshots": {},
+        "tool_started_at": {},
+        "tool_progress_mode": "off",
+    }
+    monkeypatch.setitem(server._sessions, sid, session)
+    monkeypatch.setattr(server, "_tool_progress_enabled", lambda _sid: False)
+    monkeypatch.setattr(server, "_tool_lifecycle_required_for_ui", lambda _name: False)
+    monkeypatch.setattr(
+        server,
+        "_emit",
+        lambda event, event_sid, payload=None: events.append(
+            (event, event_sid, payload)
+        ),
+    )
+
+    state = {
+        "todos": [{"id": "1", "content": "Work", "status": "in_progress"}],
+        "revision": 9,
+    }
+    server._on_tool_complete(sid, "call-1", "todo", {}, json.dumps(state))
+
+    assert [event[0] for event in events] == ["tool.complete", "todo.updated"]
+    assert events[-1] == ("todo.updated", sid, state)
+    assert session["todo_state"] == state
+
+
+def test_non_todo_completion_stays_suppressed_when_progress_is_off(monkeypatch):
+    sid = "ordinary-tool-test"
+    events = []
+    monkeypatch.setitem(
+        server._sessions,
+        sid,
+        {
+            "agent": None,
+            "edit_snapshots": {},
+            "tool_started_at": {},
+            "tool_progress_mode": "off",
+        },
+    )
+    monkeypatch.setattr(server, "_tool_progress_enabled", lambda _sid: False)
+    monkeypatch.setattr(server, "_tool_lifecycle_required_for_ui", lambda _name: False)
+    monkeypatch.setattr(server, "_emit", lambda *args: events.append(args))
+
+    server._on_tool_complete(sid, "call-1", "terminal", {}, "ok")
+
+    assert events == []
+
+
+def test_live_snapshot_prefers_the_highest_revision():
+    class Store:
+        @staticmethod
+        def snapshot():
+            return {"todos": [], "revision": 4}
+
+    class Agent:
+        _todo_store = Store()
+
+    session = {
+        "agent": Agent(),
+        "todo_state": {
+            "todos": [{"id": "1", "content": "Current", "status": "pending"}],
+            "revision": 5,
+        },
+    }
+
+    payload = server._attach_todo_state({}, session)
+
+    assert payload["todo_state"]["revision"] == 5
@@ -2,10 +2,11 @@
 """
 Todo Tool Module - Planning & Task Management

-Provides an in-memory task list the agent uses to decompose complex tasks,
-track progress, and maintain focus across long conversations. The state
-lives on the AIAgent instance (one per session) and is re-injected into
-the conversation after context compression events.
+Provides an in-memory, revisioned task list the agent uses to decompose
+complex tasks, track progress, and maintain focus across long conversations.
+The state lives on the AIAgent instance (one per session), is re-injected into
+the conversation after context compression events, and every write bumps a
+monotonic revision so UI clients can reject stale updates.

 Design:
 - Single `todo` tool: provide `todos` param to write, omit to read
@@ -15,7 +16,7 @@ Design:
 """

 import json
-from typing import Dict, Any, List, Optional
+from typing import Any, Dict, List, Optional


 # Valid status values for todo items
@@ -56,6 +57,7 @@ class TodoStore:

    def __init__(self):
        self._items: List[Dict[str, str]] = []
+        self._revision = 0

    def write(self, todos: List[Dict[str, Any]], merge: bool = False) -> List[Dict[str, str]]:
        """
@@ -66,6 +68,7 @@ class TodoStore:
            merge: if False, replace the entire list. If True, update
                   existing items by id and append new ones.
        """
+        before = self.read()
        if not merge:
            # Replace mode: new list entirely
            self._items = self._normalize_order(
@@ -113,6 +116,8 @@ class TodoStore:
        if len(self._items) > MAX_TODO_ITEMS:
            self._items = self._items[:MAX_TODO_ITEMS]
        self._sanitize_parents(self._items)
+        if self._items != before:
+            self._revision += 1
        return self.read()

    def read(self) -> List[Dict[str, str]]:
@@ -123,6 +128,26 @@ class TodoStore:
        """Check if there are any items in the list."""
        return bool(self._items)

+    def snapshot(self) -> Dict[str, Any]:
+        """Return the full state clients can reconcile atomically."""
+        return {"todos": self.read(), "revision": self._revision}
+
+    def restore(
+        self,
+        todos: List[Dict[str, Any]],
+        *,
+        revision: Any = 0,
+    ) -> List[Dict[str, str]]:
+        """Restore a trusted snapshot without manufacturing a new revision."""
+        self._items = self._normalize_order(
+            [self._validate(t) for t in self._dedupe_by_id(todos)]
+        )[:MAX_TODO_ITEMS]
+        try:
+            self._revision = max(0, int(revision or 0))
+        except (TypeError, ValueError):
+            self._revision = 0
+        return self.read()
+
    def format_for_injection(self) -> Optional[str]:
        """
        Render the todo list for post-compression injection.
@@ -330,6 +355,7 @@ def todo_tool(

    return json.dumps({
        "todos": items,
+        "revision": store.snapshot()["revision"],
        "summary": {
            "total": len(items),
            "pending": pending,
@@ -473,17 +473,20 @@ def _(rid, params: dict) -> dict:
                    history = live.get("history") or []
                    return _ok(
                        rid,
-                        {
-                            "session_id": live_sid,
-                            "stored_session_id": str(live.get("session_key") or ""),
-                            "message_count": len(history),
-                            "messages": [] if omit_messages else _history_to_messages(history),
-                            "info": {
-                                "model": _resolve_model(),
-                                "lazy": True,
-                                "profile_name": profile or "",
+                        _attach_todo_state(
+                            {
+                                "session_id": live_sid,
+                                "stored_session_id": str(live.get("session_key") or ""),
+                                "message_count": len(history),
+                                "messages": [] if omit_messages else _history_to_messages(history),
+                                "info": {
+                                    "model": _resolve_model(),
+                                    "lazy": True,
+                                    "profile_name": profile or "",
+                                },
                            },
-                        },
+                            live,
+                        ),
                    )

                # Stranded-session adoption (#93296 follow-up): before session
@@ -562,6 +565,11 @@ def _(rid, params: dict) -> dict:
                target = tip
                found = db.get_session(target) or found

+        # A full snapshot is cheap to read and makes every resume path an
+        # authoritative recovery point for task UI state, including deferred
+        # agent builds and reconnects that missed the live event.
+        resume_todo_state = _read_persisted_todo_state(db, target)
+
        # Every interactive resume path materializes the model history, even when
        # omit_messages suppresses the response copy. Count the complete lineage
        # before any reopen/history read so a runaway transcript cannot exhaust
@@ -682,6 +690,7 @@ def _(rid, params: dict) -> dict:
                close_on_disconnect=is_truthy_value(params.get("close_on_disconnect", False)),
                profile_home=profile_home,
                lazy=True,
+                todo_state=resume_todo_state,
            )
            if (live := _claim_or_reuse_live(sid, target, record, lease)) is not None:
                return _reuse_live_response(*live)
@@ -704,19 +713,22 @@ def _(rid, params: dict) -> dict:
            messages = [] if omit_messages else _history_to_messages(display_history)
            return _ok(
                rid,
-                {
-                    "session_id": sid,
-                    "resumed": target,
-                    "message_count": len(display_history) if omit_messages else len(messages),
-                    "messages": messages,
-                    "messages_omitted": omit_messages,
-                    "info": _lazy_resume_info(cwd, profile=profile),
-                    "inflight": None,
-                    "running": child_running,
-                    "session_key": target,
-                    "started_at": record["created_at"],
-                    "status": "streaming" if child_running else "idle",
-                },
+                _attach_todo_state(
+                    {
+                        "session_id": sid,
+                        "resumed": target,
+                        "message_count": len(display_history) if omit_messages else len(messages),
+                        "messages": messages,
+                        "messages_omitted": omit_messages,
+                        "info": _lazy_resume_info(cwd, profile=profile),
+                        "inflight": None,
+                        "running": child_running,
+                        "session_key": target,
+                        "started_at": record["created_at"],
+                        "status": "streaming" if child_running else "idle",
+                    },
+                    record,
+                ),
            )

        # Desktop can ask for a bounded acknowledgement and hydrate the display
@@ -750,6 +762,7 @@ def _(rid, params: dict) -> dict:
                profile_home=profile_home,
                model_override=overrides.get("model_override"),
                resume_runtime_overrides=overrides or None,
+                todo_state=resume_todo_state,
            )
            record["resume_history_ready"] = threading.Event()
            record["resume_hydrating"] = True
@@ -765,24 +778,27 @@ def _(rid, params: dict) -> dict:
            _schedule_session_cap_enforcement()
            return _ok(
                rid,
-                {
-                    "session_id": sid,
-                    "resumed": target,
-                    "message_count": record["resume_message_count"],
-                    "messages": [],
-                    "hydrating": True,
-                    "info": _lazy_resume_info(
-                        cwd,
-                        model=model_override.get("model") or "",
-                        provider=overrides.get("provider_override") or "",
-                        profile=profile,
-                    ),
-                    "inflight": None,
-                    "running": False,
-                    "session_key": target,
-                    "started_at": record["created_at"],
-                    "status": "resuming",
-                },
+                _attach_todo_state(
+                    {
+                        "session_id": sid,
+                        "resumed": target,
+                        "message_count": record["resume_message_count"],
+                        "messages": [],
+                        "hydrating": True,
+                        "info": _lazy_resume_info(
+                            cwd,
+                            model=model_override.get("model") or "",
+                            provider=overrides.get("provider_override") or "",
+                            profile=profile,
+                        ),
+                        "inflight": None,
+                        "running": False,
+                        "session_key": target,
+                        "started_at": record["created_at"],
+                        "status": "resuming",
+                    },
+                    record,
+                ),
            )

        # Cold resume default: register the live session and read its stored
@@ -846,6 +862,7 @@ def _(rid, params: dict) -> dict:
                profile_home=profile_home,
                model_override=overrides.get("model_override"),
                resume_runtime_overrides=overrides or None,
+                todo_state=resume_todo_state,
            )
            if (live := _claim_or_reuse_live(sid, target, record, lease)) is not None:
                return _reuse_live_response(*live)
@@ -875,7 +892,7 @@ def _(rid, params: dict) -> dict:
            }
            if auto_continue is not None:
                payload["auto_continue"] = auto_continue
-            return _ok(rid, payload)
+            return _ok(rid, _attach_todo_state(payload, record))

        # Build the agent OUTSIDE the lock — _make_agent can block for seconds
        # (MCP discovery, prompt/skill build, AIAgent construction). Holding
@@ -1077,7 +1094,7 @@ def _(rid, params: dict) -> dict:
    }
    if auto_continue is not None:
        payload["auto_continue"] = auto_continue
-    return _ok(rid, payload)
+    return _ok(rid, _attach_todo_state(payload, session))


@method("session.cwd.set")
@@ -3263,6 +3263,7 @@ def _start_agent_build(sid: str, session: dict) -> None:
            # Session DB row deferred to first run_conversation() call.
            # pending_title applied post-first-message (see cli.exec handler).
            current["agent"] = agent
+            _session_todo_state(current)
            # Baseline for the per-turn config sync; the profile home
            # override is still active here.
            current["config_model_seen"] = _config_model_target()
@@ -7603,6 +7604,92 @@ def _tool_summary(name: str, result: str, duration_s: float | None) -> str | Non
    return f"{text}{suffix}" if text else None


+def _normalize_todo_state(value: object) -> dict | None:
+    """Return a client-safe full todo snapshot or ``None`` when malformed."""
+    if not isinstance(value, dict) or not isinstance(value.get("todos"), list):
+        return None
+    try:
+        revision = max(0, int(value.get("revision") or 0))
+    except (TypeError, ValueError):
+        return None
+    return {"todos": list(value["todos"]), "revision": revision}
+
+
+def _session_todo_state(session: dict) -> dict | None:
+    """Return the newest live/cached todo snapshot for a runtime session."""
+    cached = _normalize_todo_state(session.get("todo_state"))
+    live = None
+    agent = session.get("agent")
+    store = getattr(agent, "_todo_store", None)
+    snapshot = getattr(store, "snapshot", None)
+    if callable(snapshot):
+        try:
+            live = _normalize_todo_state(snapshot())
+        except Exception:
+            logger.debug("failed to read live todo state", exc_info=True)
+
+    if live is not None and (
+        cached is None or live["revision"] >= cached["revision"]
+    ):
+        cached = live
+    if cached is not None:
+        session["todo_state"] = cached
+    return cached
+
+
+def _attach_todo_state(payload: dict, session: dict) -> dict:
+    """Attach the authoritative todo snapshot to a session response."""
+    state = _session_todo_state(session)
+    if state is not None:
+        payload["todo_state"] = state
+    return payload
+
+
+def _read_persisted_todo_state(db, session_id: str) -> dict | None:
+    """Derive the latest todo snapshot from the session's stored transcript.
+
+    Used by resume paths that answer before an AIAgent (and its live
+    TodoStore) exists. The canonical todo tool results already persist in the
+    session DB as ordinary tool messages, so the latest one paired with an
+    assistant ``todo`` tool call IS the durable snapshot — no side table.
+    """
+    getter = getattr(db, "get_messages_as_conversation", None)
+    if not callable(getter):
+        return None
+    try:
+        from tools.todo_tool import MAX_TODO_RESULT_CHARS
+
+        raw = getter(session_id, include_ancestors=True)
+        history: list = list(raw) if isinstance(raw, list) else []
+        todo_call_ids: set[str] = set()
+        for msg in history:
+            for call in msg.get("tool_calls") or []:
+                if (call.get("function") or {}).get("name") == "todo":
+                    cid = call.get("id")
+                    if cid:
+                        todo_call_ids.add(cid)
+        for msg in reversed(history):
+            if msg.get("role") != "tool":
+                continue
+            if msg.get("tool_call_id") not in todo_call_ids:
+                continue
+            content = msg.get("content", "")
+            if (
+                not isinstance(content, str)
+                or len(content) > MAX_TODO_RESULT_CHARS
+                or '"todos"' not in content
+            ):
+                continue
+            try:
+                return _normalize_todo_state(json.loads(content))
+            except Exception:
+                continue
+        return None
+    except Exception:
+        logger.debug("failed to read persisted todo state", exc_info=True)
+        return None
+
+
 def _on_tool_start(sid: str, tool_call_id: str, name: str, args: dict):
    session = _sessions.get(sid)
    if session is not None:
@@ -7659,13 +7746,15 @@ def _on_tool_complete(sid: str, tool_call_id: str, name: str, args: dict, result
        result_text = _tool_result_text(result)
        if result_text:
            payload["result_text"] = result_text
+    todo_state = None
    if name == "todo":
-        try:
-            data = json.loads(result)
-            if isinstance(data, dict) and isinstance(data.get("todos"), list):
-                payload["todos"] = data.get("todos")
-        except Exception:
-            pass
+        todo_state = _normalize_todo_state(payload.get("result"))
+        if todo_state is not None:
+            payload.update(todo_state)
+            if session is not None:
+                cached = _normalize_todo_state(session.get("todo_state"))
+                if cached is None or todo_state["revision"] >= cached["revision"]:
+                    session["todo_state"] = todo_state
    try:
        from agent.display import render_edit_diff_with_delta

@@ -7680,8 +7769,18 @@ def _on_tool_complete(sid: str, tool_call_id: str, name: str, args: dict, result
            payload["inline_diff"] = "\n".join(rendered)
    except Exception:
        pass
-    if _tool_progress_enabled(sid) or payload.get("inline_diff") or _tool_lifecycle_required_for_ui(name):
+    if (
+        _tool_progress_enabled(sid)
+        or payload.get("inline_diff")
+        or _tool_lifecycle_required_for_ui(name)
+        or name == "todo"
+    ):
        _emit("tool.complete", sid, payload)
+    # Task state is application data, not optional tool-progress chrome. A
+    # dedicated full-snapshot event lets every client reconcile immediately
+    # without interpreting provider text or partial merge arguments.
+    if todo_state is not None:
+        _emit("todo.updated", sid, todo_state)


 def _on_tool_progress(
@@ -8872,6 +8971,7 @@ def _init_session(
            # session (stdio for Ink, JSON-RPC WS for the dashboard sidebar).
            "transport": current_transport() or _stdio_transport,
        }
+        _session_todo_state(_sessions[sid])
    _init_owns_db = False
    if session_db is not None:
        db = session_db
@@ -10244,6 +10344,7 @@ def _deferred_session_record(
    lazy: bool = False,
    model_override=None,
    resume_runtime_overrides: dict | None = None,
+    todo_state: dict | None = None,
 ) -> dict:
    """A live-session record whose AIAgent is built later (lazy watch / cold
    resume) — _init_session's shape minus the agent."""
@@ -10280,6 +10381,7 @@ def _deferred_session_record(
        "source": source,
        "tool_progress_mode": _load_tool_progress_mode(),
        "tool_started_at": {},
+        "todo_state": todo_state,
        "transport": current_transport() or _stdio_transport,
    }

@@ -10709,7 +10811,7 @@ def _live_session_payload(
        payload["pending_approval"] = approval
    if clarify := _pending_clarify_request_payload(sid):
        payload["pending_clarify"] = clarify
-    return payload
+    return _attach_todo_state(payload, session)


 def _main_runtime_from_agent(agent) -> dict | None: