Release v0.51.261 — Release IC (stage-r11) (#3616)

## Release v0.51.261 — Release IC (stage-r11)

Live Todos panel via an explicit `todo_state` SSE contract.

### Fixed
| Issue/PR | Author | Fix |
|----------|--------|-----|
| #3373 follow-up (#3454) | @v2psv | The Todos side panel now tracks `todo` tool state **live during an active run** instead of staying stale until settle / rolling back on a mid-stream reload. A dedicated `todo_state` SSE event sends a full, redacted, idempotent snapshot on todo-tool completion (no more truncated `tool_complete.preview`); the same `api.todo_state` parser feeds live + cold-load; live snapshots persist into INFLIGHT so reload/reattach restores the panel; cold-load vs INFLIGHT reconciled by timestamp (incl. the `coldTs===0` compressed-session edge); legacy reverse-scan kept as fallback for old servers. |

### Gate
- Full pytest suite: **7692 passed, 0 failed**
- ESLint: CLEAN · ruff: CLEAN · browser-smoke: CLEAN
- Codex (regression): **SAFE TO SHIP** — verified the new `todo_state` SSE handler composes with existing dispatch (no double-subscribe), INFLIGHT persistence is cleared on terminal/cancel (composes with discard_session + turn-journal), timestamp reconciliation can't let a stale local snapshot win, redaction holds, the legacy reverse-scan fallback still works with no double-render, and the `models.py` change is todo-scoped (no CLI-classification interaction).

Co-authored-by: v2psv <v2psv@users.noreply.github.com>

CHANGELOG.md

@@ -3,6 +3,11 @@
 
 ## [Unreleased]
 
+## [v0.51.261] — 2026-06-04 — Release IC (stage-r11 — live Todos panel via a todo_state SSE contract)
+
+### Fixed
+- **The Todos side panel now tracks `todo` tool state live during an active run**, instead of staying stale until the turn settled and occasionally rolling back on a reload/SSE-reattach mid-stream. A dedicated `todo_state` SSE event emits a full, redacted, idempotent todo snapshot when the `todo` tool completes (no longer relying on the truncated `tool_complete.preview`); the same `api.todo_state` parser feeds both live emission and cold-load hydration; live snapshots persist into INFLIGHT so reload/reattach restores the panel before the next event; and cold-load vs INFLIGHT snapshots are reconciled by timestamp (including the `coldTs===0` compressed-session edge). The legacy reverse-scan fallback is kept for old servers / upgrade windows. (#3373 follow-up, @v2psv)
+
 ## [v0.51.260] — 2026-06-04 — Release IB (stage-r8 — un-held safety fixes + cron/TTS/mcp/state-toast batch)
 
 ### Fixed

api/models.py

@@ -137,6 +137,14 @@ def _persisted_session_ids_snapshot() -> frozenset[str]:
     return ids
 
 
+def _session_dir_has_persisted_session_files() -> bool:
+    """Return True when the current session dir has at least one session JSON file."""
+    try:
+        return any(not p.name.startswith('_') for p in SESSION_DIR.glob('*.json'))
+    except Exception:
+        return False
+
+
 def _rebuild_session_index_background() -> None:
     try:
         _write_session_index(updates=None)
@@ -2825,6 +2833,8 @@ def all_sessions(diag=None):
             with LOCK:
                 in_memory_ids = set(SESSIONS.keys())
             persisted_ids = _persisted_session_ids_snapshot()
+            if not index and _session_dir_has_persisted_session_files():
+                raise ValueError("empty session index while session files exist")
             index = [
                 s for s in index
                 if (
@@ -2839,6 +2849,8 @@ def all_sessions(diag=None):
                     )
                 )
             ]
+            if not index and _session_dir_has_persisted_session_files():
+                raise ValueError("session index has no live rows while session files exist")
             backfilled = []
             for i, s in enumerate(index):
                 if 'last_message_at' not in s:

api/streaming.py

@@ -43,6 +43,7 @@ from api.helpers import redact_session_data, _redact_text
 from api.compression_anchor import is_context_compression_marker, visible_messages_for_anchor
 from api.metering import meter
 from api.run_journal import RunJournalWriter
+from api.todo_state import emit_todo_state
 from api.turn_journal import append_turn_journal_event_for_stream
 from api.usage import prompt_cache_hit_percent
 from api.models import (
@@ -4996,6 +4997,35 @@ def _run_agent_streaming(
                         'duration': cb_kwargs.get('duration'),
                         'is_error': bool(cb_kwargs.get('is_error', False)),
                     })
+                    # Mirror the todo tool's in-memory state into a
+                    # dedicated SSE event so the Todos panel can update
+                    # in real-time without waiting for the turn to
+                    # settle. The helper guards on name=='todo', sends
+                    # the full snapshot (idempotent under SSE replay)
+                    # and swallows internal errors so emission never
+                    # breaks tool delivery. Prefer the structured
+                    # `result` kwarg from modern Hermes builds; fall
+                    # back to the truncated `preview` only when the
+                    # callback was invoked without one (older builds).
+                    #
+                    # Graceful degradation on old builds: `preview` is a
+                    # truncated snippet, so its JSON is usually unparseable.
+                    # parse_todo_tool_result() then returns None and NO
+                    # todo_state event is emitted — live panel updates are
+                    # silently unavailable on pre-`result` builds. This is
+                    # intended: the panel still hydrates via cold-load on the
+                    # next session GET; it just won't update mid-stream.
+                    emit_todo_state(
+                        put,
+                        name=name,
+                        function_result=(
+                            cb_kwargs.get('result')
+                            if cb_kwargs.get('result') is not None
+                            else preview
+                        ),
+                        session_id=session_id,
+                        stream_id=stream_id,
+                    )
                     _tool_stats = meter().get_stats()
                     _tool_stats['session_id'] = session_id
                     _tool_stats['usage'] = _live_usage_snapshot()
@@ -5064,6 +5094,20 @@ def _run_agent_streaming(
                             'tid': tool_call_id,
                             'is_error': False,
                         })
+                        # Mirror the todo tool's in-memory state into
+                        # a dedicated SSE event so the Todos panel can
+                        # update in real-time without waiting for the
+                        # turn to settle. See the legacy path above
+                        # for the contract; the helper handles the
+                        # name guard, payload shape, and swallow-all
+                        # error policy.
+                        emit_todo_state(
+                            put,
+                            name=name,
+                            function_result=function_result,
+                            session_id=session_id,
+                            stream_id=stream_id,
+                        )
                     _tool_stats = meter().get_stats()
                     _tool_stats['session_id'] = session_id
                     _tool_stats['usage'] = _live_usage_snapshot()

api/todo_state.py

@@ -1,30 +1,91 @@
-"""Derive settled todo snapshots for session GET responses.
+"""Derive ``todo_state`` snapshots from tool results and settled session messages.
 
-The browser's Todos panel currently reconstructs state by reverse-scanning
-loaded tool messages. That works only when the latest todo tool result is inside
-the returned message window. This helper gives ``/api/session`` a compact,
-explicit ``todo_state`` sidecar derived from the full settled transcript.
+The ``todo`` tool's in-memory store lives on the per-session AIAgent. The
+WebUI bridge needs to mirror that state to the browser in two situations:
 
-The detector intentionally mirrors ``run_agent.AIAgent._hydrate_todo_store``:
-walk messages newest-first and use the first tool message whose JSON content has
-a ``todos`` list. Empty ``todos`` is a valid snapshot so a cleared task list does
-not fall through to an older non-empty write.
+1. **Live**: when the agent calls ``todo`` mid-stream, ``api.streaming``
+   emits a dedicated ``todo_state`` SSE event so the Todos panel updates
+   without waiting for the turn to finish. See :func:`emit_todo_state`.
+
+2. **Cold-load**: when the browser opens a session (no live stream), the
+   session GET handler attaches ``todo_state`` derived from the most
+   recent ``role='tool'`` message whose JSON content carries a ``todos``
+   list. See :func:`attach_todo_state`.
+
+Both paths normalize through :func:`_normalize_snapshot` so the frontend
+has a single deserialization contract:
+
+    {
+        "todos":   [{"id": ..., "content": ..., "status": ...}, ...],
+        "summary": {"total": N, "pending": N, "in_progress": N,
+                    "completed": N, "cancelled": N},
+        "version": 1,
+    }
+
+Live SSE payloads add ``session_id``, ``stream_id``, ``source`` and ``ts``
+on top so the frontend can filter cross-session events and ignore
+out-of-order replays.
+
+**Detection symmetry with the agent.** The cold-load helper deliberately
+uses the same loose detector as ``run_agent.AIAgent._hydrate_todo_store``
+(``role='tool'`` + JSON content with ``todos: list``). If a future change
+tightens or relaxes that detector, mirror it here so the WebUI panel
+never disagrees with the agent's in-memory ``TodoStore``.
+
+**Multimodal tool results.** Some tools return content as a list of
+OpenAI/Anthropic content parts rather than a JSON string. The ``todo``
+tool always returns a JSON string, so list-shaped content cannot be a
+todo write — :func:`derive_todo_state` skips them by design.
+
+This module is **side-effect free** by design — it only parses data and
+calls a caller-supplied ``put`` callable for SSE. Routing/event-shape
+decisions live here so the call sites stay one-liners.
 """
 
 from __future__ import annotations
 
 import json
 import logging
-from typing import Any, Iterable, Optional, Sequence
+import time
+from typing import Any, Callable, Iterable, Optional, Sequence
+
 
 logger = logging.getLogger(__name__)
 
+
+# Bumped when the on-wire payload shape changes in a non-additive way.
+# Additive fields (e.g. timestamps, tags) keep VERSION at 1.
 VERSION = 1
+
+# Single source of truth for the SSE event name and the session GET
+# payload key. Any current or future caller must reuse these so a
+# rename only happens in one place.
+EVENT_NAME = "todo_state"
 PAYLOAD_KEY = "todo_state"
 
 
 def _normalize_snapshot(data: Any) -> Optional[dict]:
-    """Return the canonical todo snapshot shape, or ``None`` for non-todo data."""
+    """Return a normalized snapshot dict, or ``None`` if the payload is invalid.
+
+    Accepts the canonical ``{"todos": [...], "summary": {...}}`` shape
+    produced by ``tools.todo_tool.todo_tool``. Anything else returns
+    ``None`` so callers can fall through to legacy paths or skip
+    emission.
+
+    The detector is intentionally loose so it stays symmetric with the
+    agent's hydration logic — see the module docstring.
+
+    **Empty list is a valid snapshot.** ``todos == []`` returns a normal
+    snapshot (not ``None``), so the latest write wins even when it cleared
+    the list. This is deliberately symmetric with the agent: its
+    ``_hydrate_todo_store`` (run_agent.py) breaks at the most-recent todo
+    message and, because ``if last_todo_response:`` is falsy for ``[]``,
+    leaves its TodoStore empty — i.e. agent shows empty, panel shows empty.
+    Do NOT reintroduce a ``len(todos) > 0`` guard here or in the frontend
+    fallback (``_legacyTodosFromMessages``): that was the pre-Phase-2
+    behavior that kept scanning past an empty write to an older non-empty
+    list, diverging from the agent and showing a stale "cleared" list.
+    """
     if not isinstance(data, dict):
         return None
     todos = data.get("todos")
@@ -41,16 +102,86 @@ def _normalize_snapshot(data: Any) -> Optional[dict]:
 
 
 def parse_todo_tool_result(function_result: Any) -> Optional[dict]:
-    """Parse a todo tool result JSON string or pre-parsed dict into a snapshot."""
+    """Parse a fresh ``todo`` tool call result into a snapshot dict.
+
+    The agent's ``todo`` handler returns a JSON string; this helper
+    accepts either that string or an already-parsed dict (defensive —
+    future callers may deserialize earlier in the pipeline).
+
+    Returns ``None`` on any parse/shape failure so the caller can
+    swallow the error without breaking the tool delivery path.
+    """
     data: Any = function_result
     if isinstance(function_result, str):
         try:
             data = json.loads(function_result)
-        except (TypeError, ValueError):
+        except (ValueError, TypeError):
             return None
     return _normalize_snapshot(data)
 
 
+def derive_todo_state(messages: Optional[Iterable[dict]]) -> Optional[dict]:
+    """Derive the latest todo snapshot from settled conversation history.
+
+    Mirrors the agent-side ``_hydrate_todo_store`` logic: walk messages
+    in reverse, return the first ``role='tool'`` message whose JSON
+    content carries a ``todos`` list. Returns ``None`` when no such
+    message is found (fresh session, or a session that never invoked
+    ``todo``).
+
+    Multimodal tool results — ``content`` as a list of content parts
+    rather than a JSON string — are skipped intentionally. The ``todo``
+    tool always returns a string, so list-shaped content cannot be a
+    todo write; non-string ``content`` is therefore correct to ignore.
+
+    The fast-path string check (``'"todos"' in content``) avoids parsing
+    JSON for every tool result — most sessions have many non-todo tool
+    calls but at most a handful of todo writes.
+    """
+    if not messages:
+        return None
+    # ``reversed`` works on ``list`` and ``tuple`` natively; for any
+    # other iterable (e.g. a generator) we materialize once. Routes
+    # always pass a list, so this branch is normally a no-op.
+    if not isinstance(messages, (list, tuple)):
+        messages = list(messages)
+    for idx in range(len(messages) - 1, -1, -1):
+        msg = messages[idx]
+        if not isinstance(msg, dict) or msg.get("role") != "tool":
+            continue
+        content = msg.get("content", "")
+        if not isinstance(content, str) or '"todos"' not in content:
+            continue
+        try:
+            data = json.loads(content)
+        except (ValueError, TypeError):
+            continue
+        snapshot = _normalize_snapshot(data)
+        if snapshot is not None:
+            # Carry a timestamp so the frontend can reconcile cold-load
+            # vs. INFLIGHT snapshots by recency.
+            #
+            # Primary source: this message's own ``timestamp``. But a
+            # todo tool message can lose its timestamp during context
+            # compression/rebuild — the on-disk message ends up with
+            # ``timestamp=None``. If we emit a snapshot with no ``ts``,
+            # the frontend reads coldTs=0 and a STALE-but-timestamped
+            # INFLIGHT snapshot wins the recency comparison, so the panel
+            # renders a historical todo list. This is the latest-by-
+            # POSITION snapshot, so it must never lose recency to an
+            # earlier list. When this message has no usable timestamp,
+            # fall back to the max timestamp seen anywhere at or before
+            # this position — guaranteeing cold ts >= any earlier todo
+            # write's ts.
+            ts_val = _message_ts_float(msg.get("timestamp"))
+            if ts_val <= 0:
+                ts_val = _max_timestamp_through(messages, idx)
+            if ts_val > 0:
+                snapshot["ts"] = ts_val
+            return snapshot
+    return None
+
+
 def _message_ts_float(ts_raw: Any) -> float:
     """Coerce a message ``timestamp`` field to a positive float, or 0.0."""
     try:
@@ -59,55 +190,122 @@ def _message_ts_float(ts_raw: Any) -> float:
         return 0.0
 
 
-def _max_timestamp_through(messages: Sequence[Any], upto_idx: int) -> float:
-    """Return the largest valid timestamp at or before ``upto_idx``."""
+def _max_timestamp_through(messages: "Sequence[Any]", upto_idx: int) -> float:
+    """Largest valid ``timestamp`` among messages[0:upto_idx+1].
+
+    Used as a recency floor when the latest todo message itself lost its
+    timestamp during compression/rebuild. Scanning only up to the todo's
+    position keeps the floor causally correct — it never borrows a
+    timestamp from a message that came after the todo write.
+    """
     best = 0.0
     end = min(upto_idx, len(messages) - 1)
     for i in range(end, -1, -1):
-        msg = messages[i]
-        if not isinstance(msg, dict):
+        m = messages[i]
+        if not isinstance(m, dict):
             continue
-        best = max(best, _message_ts_float(msg.get("timestamp")))
+        ts = _message_ts_float(m.get("timestamp"))
+        if ts > best:
+            best = ts
     return best
 
 
-def derive_todo_state(messages: Optional[Iterable[dict]]) -> Optional[dict]:
-    """Derive the latest settled todo snapshot from conversation history.
+def _redact_snapshot(snapshot: dict) -> dict:
+    """Redact credential-shaped text from a todo snapshot before it leaves the process.
 
-    Returns ``None`` when the session has no todo writes. Malformed or
-    non-string tool contents are skipped so unrelated tool results never break
-    session loading.
+    The live SSE path (:func:`emit_todo_state`) does NOT pass through
+    ``redact_session_data`` (api/helpers.py) the way the cold-load session
+    GET response does, so emission must redact the same content that path
+    would — otherwise the live Todos panel (and the run-journal replay that
+    persists every SSE event) becomes a redaction bypass for any credential
+    an agent wrote into a todo item's ``content``. The live event also
+    carries the FULL untruncated todos, a wider exposure surface than the
+    truncated ``preview`` the sibling ``tool``/``tool_complete`` events send.
+
+    ``_redact_value`` is imported lazily to keep the dependency direction
+    one-way (helpers must never import todo_state) and to avoid paying the
+    import cost on the cold-load path, which redacts via ``redact_session_data``.
+    The redaction setting is read once per SSE snapshot and threaded through the
+    recursive helper so nested strings do not reload settings.json individually.
+
+    Returns a new, redacted snapshot. Raises on failure so the caller fails
+    closed (no emission) rather than leaking an unredacted payload.
     """
-    if not messages:
-        return None
-    if not isinstance(messages, (list, tuple)):
-        messages = list(messages)
+    from typing import cast
+    from api.config import load_settings
+    from api.helpers import _redact_value
+    _enabled = bool(load_settings().get("api_redact_enabled", True))
+    # ``_redact_value`` preserves container shape (dict in → dict out); the
+    # cast narrows its broad recursive union back to dict for the type checker.
+    return cast(dict, _redact_value(snapshot, _enabled=_enabled))
 
-    for idx in range(len(messages) - 1, -1, -1):
-        msg = messages[idx]
-        if not isinstance(msg, dict) or msg.get("role") != "tool":
-            continue
-        content = msg.get("content", "")
-        if not isinstance(content, str) or '"todos"' not in content:
-            continue
-        snapshot = parse_todo_tool_result(content)
+
+def emit_todo_state(
+    put: Callable[[str, dict], Any],
+    *,
+    name: Optional[str],
+    function_result: Any,
+    session_id: Optional[str],
+    stream_id: Optional[str],
+    source: str = "tool",
+) -> bool:
+    """Emit a ``todo_state`` SSE event when ``name == 'todo'``.
+
+    Returns ``True`` if an event was emitted, ``False`` otherwise.
+    Always swallows internal errors — emission must never break tool
+    delivery, which is the caller's primary contract.
+
+    Args:
+        put: streaming queue callback; signature ``put(event, data)``.
+        name: tool name from the callback. Skipped when not ``'todo'``.
+        function_result: raw tool result (JSON string or dict).
+        session_id: tag so the frontend can filter cross-session events.
+        stream_id: tag so SSE replay can dedupe by stream.
+        source: emission origin tag. ``'tool'`` for live tool calls;
+                future callers may use ``'compression-refresh'`` etc.
+
+    The full snapshot is always sent — idempotent re-application is safe
+    under SSE replay through the run journal. The snapshot is redacted
+    before emission (see :func:`_redact_snapshot`); if redaction fails the
+    event is dropped (fail-closed) rather than leaking an unredacted payload.
+    """
+    if name != "todo":
+        return False
+    try:
+        snapshot = parse_todo_tool_result(function_result)
         if snapshot is None:
-            continue
-
-        ts_val = _message_ts_float(msg.get("timestamp"))
-        if ts_val <= 0:
-            ts_val = _max_timestamp_through(messages, idx)
-        if ts_val > 0:
-            snapshot["ts"] = ts_val
-        return snapshot
-    return None
+            return False
+        snapshot = _redact_snapshot(snapshot)
+        put(EVENT_NAME, {
+            "session_id": session_id,
+            "stream_id": stream_id,
+            "source": source,
+            "ts": time.time(),
+            **snapshot,
+        })
+        return True
+    except Exception:
+        # Per-call debug logging — a flood would mean the queue is
+        # broken, in which case the rest of the stream is already dead.
+        # Redaction failure also lands here and correctly drops the event.
+        logger.debug("todo_state emit failed (name=%s)", name, exc_info=True)
+        return False
 
 
-def attach_todo_state(payload: dict, messages: Optional[Iterable[dict]]) -> bool:
-    """Attach ``todo_state`` to a session payload when one can be derived.
+def attach_todo_state(
+    payload: dict,
+    messages: Optional[Iterable[dict]],
+) -> bool:
+    """Attach a derived ``todo_state`` snapshot to a session GET response.
 
-    Mutates ``payload`` in place. Errors are swallowed deliberately: a malformed
-    historical tool message must not make ``/api/session`` fail.
+    Mutates ``payload`` in place when a snapshot can be derived.
+    Returns ``True`` if attached, ``False`` otherwise. Always swallows
+    errors — a malformed sidecar must never break the session GET
+    response.
+
+    The caller is responsible for any higher-level gating
+    (e.g. ``load_messages``); this helper is a no-op on empty/``None``
+    ``messages`` so callers can hand it whatever message list they have.
     """
     if not messages:
         return False

static/messages.js

@@ -938,6 +938,8 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
       messages:inflight.messages||[],
       uploaded:inflight.uploaded||[...uploaded],
       toolCalls:inflight.toolCalls||[],
+      todos:Array.isArray(inflight.todos)?inflight.todos:S.todos,
+      todoStateMeta:inflight.todoStateMeta||S.todoStateMeta||null,
     });
   }
   function snapshotLiveTurn(){
@@ -1920,6 +1922,47 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
       scrollIfPinned();
     });
 
+    // Phase 2: dedicated `todo_state` event carries a full snapshot of
+    // the upstream TodoStore.  We treat it as the single source of truth
+    // for the Todos panel — never merge, always replace.  The handler
+    // is intentionally cheap: parse, validate, write S.todos, mirror to
+    // INFLIGHT, schedule a RAF render.  Out-of-order events are filtered
+    // by ts; SSE journal replay is idempotent because snapshots are full.
+    // Cross-session protection mirrors every other live listener:
+    // payload.session_id must match activeSid or the event is dropped.
+    source.addEventListener('todo_state',e=>{
+      let d;
+      try{ d=JSON.parse(e.data||'{}'); }catch(_){ return; }
+      if(!d||typeof d!=='object') return;
+      // Cross-session double check: payload.session_id is the SSE-side
+      // filter (some legacy emissions omit it), and S.session.session_id
+      // is the UI-side filter (a late event that arrives after the user
+      // already navigated to another session must not pollute S.todos).
+      // Both must agree with activeSid before we touch global state.
+      if(d.session_id&&d.session_id!==activeSid) return;
+      if(!S.session||S.session.session_id!==activeSid) return;
+      if(!Array.isArray(d.todos)) return;
+      const incomingTs=Number(d.ts)||0;
+      const currentTs=(S.todoStateMeta&&Number(S.todoStateMeta.ts))||0;
+      // Strictly older snapshots are discarded; equal-ts events still
+      // apply so a compression-source refresh can land on the same
+      // second as the tool emit it follows.
+      if(incomingTs&&currentTs&&incomingTs<currentTs) return;
+      S.todos=d.todos;
+      S.todoStateMeta={
+        ts:incomingTs||(Date.now()/1000),
+        source:String(d.source||'tool'),
+        version:Number(d.version)||1,
+      };
+      const inflight=INFLIGHT[activeSid];
+      if(inflight){
+        inflight.todos=S.todos;
+        inflight.todoStateMeta=S.todoStateMeta;
+      }
+      if(typeof persistInflightState==='function') persistInflightState();
+      if(typeof scheduleTodosRefresh==='function') scheduleTodosRefresh();
+    });
+
     source.addEventListener('approval',e=>{
       const d=JSON.parse(e.data);
       showApprovalForSession(activeSid, d, 1);
@@ -2093,6 +2136,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
           const _prevCacheWrite=(S.session&&S.session.cache_write_tokens)||0;
           S.session=d.session;S.messages=_carryForwardEphemeralTurnFields(S.messages||[], d.session.messages||[]);if(typeof _messagesTruncated!=='undefined')_messagesTruncated=!!d.session._messages_truncated;
           S.messages=_filterRecoveryControlMessages(S.messages || []);
+          if(typeof _hydrateTodosFromSession==='function') _hydrateTodosFromSession(S.session);
           if(S.session&&S.session.session_id){
             try{localStorage.setItem('hermes-webui-session',S.session.session_id);}catch(_){}
             if(typeof _setActiveSessionUrl==='function') _setActiveSessionUrl(S.session.session_id);
@@ -2522,6 +2566,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
             S.session=data.session;
             const _nextMsgs3018=(data.session.messages||[]).filter(m=>m&&m.role);
             S.messages=_carryForwardEphemeralTurnFields(S.messages||[], _nextMsgs3018);
+            if(typeof _hydrateTodosFromSession==='function') _hydrateTodosFromSession(S.session);
             clearLiveToolCards();if(!assistantText)removeThinking();
             _markSessionViewed(activeSid, data.session.message_count ?? S.messages.length);
             renderMessages({preserveScroll:true});
@@ -2540,7 +2585,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
       _setActivePaneIdleIfOwner();
     });
 
-    for(const _runJournalEventName of ['token','interim_assistant','reasoning','tool','tool_complete','approval','clarify','state_saved','title','title_status','context_status','goal','goal_continue','done','stream_end','pending_steer_leftover','compressing','compressed','metering','apperror','warning','error','cancel']){
+    for(const _runJournalEventName of ['token','interim_assistant','reasoning','tool','tool_complete','todo_state','approval','clarify','state_saved','title','title_status','context_status','goal','goal_continue','done','stream_end','pending_steer_leftover','compressing','compressed','metering','apperror','warning','error','cancel']){
       source.addEventListener(_runJournalEventName,_rememberRunJournalCursor);
     }
   }
@@ -2619,6 +2664,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
         const _nextMsgs3018=(session.messages||[]).filter(m=>m&&m.role);
         S.messages=_carryForwardEphemeralTurnFields(S.messages||[], _nextMsgs3018);
         S.messages=_filterRecoveryControlMessages(S.messages || []);
+        if(typeof _hydrateTodosFromSession==='function') _hydrateTodosFromSession(S.session);
         if(S.session&&S.session.session_id){
           try{localStorage.setItem('hermes-webui-session',S.session.session_id);}catch(_){}
           if(typeof _setActiveSessionUrl==='function') _setActiveSessionUrl(S.session.session_id);

static/panels.js

@@ -2666,34 +2666,69 @@ async function loadKanbanTask(taskId){
   } catch(e) { showToast(t('kanban_unavailable') + ': ' + (e.message || e), 'error'); }
 }
 
+// Phase 2: Single-source-of-truth render.
+//
+// Reads `S.todos` (set by the `todo_state` SSE listener, INFLIGHT
+// restore, or session cold-load — see _hydrateTodosFromSession in
+// ui.js).  When `S.todoStateMeta` is null we have never seen an
+// explicit signal and fall through to the legacy reverse-scan over
+// settled tool messages — this keeps the panel populated against
+// pre-Phase-1 servers and during the upgrade window.
+//
+// The render is short-circuited via `_todosLastRenderedHash` (defined
+// in ui.js): repeated emissions that yield identical DOM are no-ops.
+// Coalescing of bursty live updates happens upstream in
+// scheduleTodosRefresh().
 function loadTodos() {
   const panel = $('todoPanel');
   if (!panel) return;
 
-  const sessionTodoState = S.session && S.session.todo_state;
   let todos;
-  if (sessionTodoState && Array.isArray(sessionTodoState.todos)) {
-    todos = sessionTodoState.todos;
+  if (S.todoStateMeta) {
+    todos = Array.isArray(S.todos) ? S.todos : [];
   } else {
     todos = _legacyTodosFromMessages();
   }
 
   if (!todos.length) {
+    if (typeof _todosLastRenderedHash !== 'undefined' && _todosLastRenderedHash === '__empty__') return;
     panel.innerHTML = `<div style="color:var(--muted);font-size:12px;padding:4px 0">${esc(t('todos_no_active'))}</div>`;
+    if (typeof _todosLastRenderedHash !== 'undefined') _todosLastRenderedHash = '__empty__';
     return;
   }
+
+  if (typeof _todosHash === 'function' && typeof _todosLastRenderedHash !== 'undefined') {
+    const hash = _todosHash(todos);
+    if (hash === _todosLastRenderedHash) return;
+    _todosLastRenderedHash = hash;
+  }
+
   const statusIcon = {pending:li('square',14), in_progress:li('loader',14), completed:li('check',14), cancelled:li('x',14)};
   const statusColor = {pending:'var(--muted)', in_progress:'var(--blue)', completed:'rgba(100,200,100,.8)', cancelled:'rgba(200,100,100,.5)'};
-  panel.innerHTML = todos.map(todo => `
+  // Single innerHTML join is the cheapest correct way to materialize
+  // ~10–50 leaf nodes.  All user-controlled content goes through esc().
+  panel.innerHTML = todos.map(td => `
     <div style="display:flex;align-items:flex-start;gap:10px;padding:6px 0;border-bottom:1px solid var(--border);">
-      <span style="font-size:14px;display:inline-flex;align-items:center;flex-shrink:0;margin-top:1px;color:${statusColor[todo.status]||'var(--muted)'}">${statusIcon[todo.status]||li('square',14)}</span>
+      <span style="font-size:14px;display:inline-flex;align-items:center;flex-shrink:0;margin-top:1px;color:${statusColor[td.status]||'var(--muted)'}">${statusIcon[td.status]||li('square',14)}</span>
       <div style="flex:1;min-width:0">
-        <div style="font-size:13px;color:${todo.status==='completed'?'var(--muted)':todo.status==='in_progress'?'var(--text)':'var(--text)'};${todo.status==='completed'?'text-decoration:line-through;opacity:.5':''};line-height:1.4">${esc(todo.content)}</div>
-        <div style="font-size:10px;color:var(--muted);margin-top:2px;opacity:.6">${esc(todo.id)} · ${esc(todo.status)}</div>
+        <div style="font-size:13px;color:${td.status==='completed'?'var(--muted)':td.status==='in_progress'?'var(--text)':'var(--text)'};${td.status==='completed'?'text-decoration:line-through;opacity:.5':''};line-height:1.4">${esc(td.content)}</div>
+        <div style="font-size:10px;color:var(--muted);margin-top:2px;opacity:.6">${esc(td.id)} · ${esc(td.status)}</div>
       </div>
     </div>`).join('');
 }
 
+// Legacy fallback: reverse-scan settled tool messages for the most
+// recent {"todos":[...]} payload.  Used only when no `todo_state`
+// signal has been seen for the current session — primarily during
+// upgrade windows where the server has not yet been redeployed with
+// Phase 1.  Once Phase 1 is universally deployed and a stabilization
+// period has passed, this can be removed (Phase 3).
+//
+// Variable name `sourceMessages` is preserved verbatim from the
+// original loadTodos() implementation so the matching regression
+// test (R-todo-survive-refresh in tests/test_regressions.py) keeps
+// catching any future refactor that drops the raw-session-messages
+// fallback. See the test for the exact contract.
 function _legacyTodosFromMessages() {
   const sourceMessages = (S.session && Array.isArray(S.session.messages) && S.session.messages.length) ? S.session.messages : S.messages;
   if (!Array.isArray(sourceMessages)) return [];
@@ -2707,7 +2742,7 @@ function _legacyTodosFromMessages() {
     if (!content || content.indexOf('"todos"') < 0) continue;
     try {
       const d = JSON.parse(content);
-      if (d && Array.isArray(d.todos) && d.todos.length) return d.todos;
+      if (d && Array.isArray(d.todos)) return d.todos;
     } catch (_) {}
   }
   return [];

static/sessions.js

@@ -611,6 +611,7 @@ async function newSession(flash, options={}){
     const data=await api('/api/session/new',{method:'POST',body:JSON.stringify(reqBody)});
     S.session=data.session;S.messages=data.session.messages||[];
     if(_sessionSourceFilter==='cli') _sessionSourceFilter='webui';
+    if(typeof _hydrateTodosFromSession==='function') _hydrateTodosFromSession(S.session);
     S.lastUsage={...(data.session.last_usage||{})};
     if(!(options&&options.worktree)) _rememberNewChatDraftSession(S.session);
     if(flash)S.session._flash=true;
@@ -802,6 +803,7 @@ async function loadSession(sid){
   // Stale response? A newer loadSession() call has already started (#1060).
   if (_loadingSessionId !== sid) return;
   S.session=data.session;
+  if(typeof _hydrateTodosFromSession==='function') _hydrateTodosFromSession(S.session);
   S.session._modelResolutionDeferred=true;
   S.lastUsage={...(data.session.last_usage||{})};
   // Reset scroll-direction tracker on session switch so the new chat's
@@ -853,6 +855,13 @@ async function loadSession(sid){
         messages:Array.isArray(stored.messages)&&stored.messages.length?stored.messages:[],
         uploaded:Array.isArray(stored.uploaded)?stored.uploaded:[],
         toolCalls:Array.isArray(stored.toolCalls)?stored.toolCalls:[],
+        // Phase 2: restore the live todo snapshot from persisted INFLIGHT
+        // so the panel does not flicker to empty when a mid-stream
+        // browser reload reattaches before the next `todo_state` event
+        // fires.  Both fields are optional; missing values fall back to
+        // cold-load via session.todo_state.
+        todos:Array.isArray(stored.todos)?stored.todos:null,
+        todoStateMeta:stored.todoStateMeta||null,
         reattach:true,
       };
     }
@@ -872,6 +881,8 @@ async function loadSession(sid){
     if(_mergePendingSessionMessage(S.session,S.messages)){
       INFLIGHT[sid].messages=S.messages;
     }
+    // Refresh todos from cold-load or persisted INFLIGHT before painting.
+    if(typeof _hydrateTodosFromSession==='function') _hydrateTodosFromSession(S.session);
     S.busy=true;
     // appendLiveToolCard() is guarded by S.activeStreamId; restore it before
     // replaying persisted live tools so the compact Activity count survives
@@ -1569,9 +1580,31 @@ async function _ensureMessagesLoaded(sid) {
   S.messages = msgs;
   if(S.session&&S.session.session_id===sid){
     S.session.message_count=Number(data.session.message_count || msgs.length);
-    if(Object.prototype.hasOwnProperty.call(data.session,'todo_state')) S.session.todo_state=data.session.todo_state;
-    else delete S.session.todo_state;
     S.lastUsage={...(data.session.last_usage||S.lastUsage||{})};
+    // Phase 2: the messages=1 response carries the canonical cold-load
+    // `todo_state` snapshot, derived server-side from the FULL untruncated
+    // message list (api/routes.py + api/todo_state.py). The earlier
+    // messages=0 fetch in loadSession() does not include this field —
+    // attach_todo_state is gated on `load_messages`. Without applying it
+    // here, long sessions whose latest todo write falls outside the
+    // _INITIAL_MSG_LIMIT tail would lose the panel on refresh: the
+    // legacy reverse-scan in _legacyTodosFromMessages() can only see the
+    // tail S.messages, while the authoritative snapshot was already
+    // computed by the server and is sitting in this very response.
+    // _hydrateTodosFromSession is idempotent and picks newer of
+    // cold-load vs INFLIGHT by timestamp, so calling it again here is
+    // safe even when an INFLIGHT snapshot was already restored.
+    if(data.session.todo_state !== undefined){
+      S.session.todo_state = data.session.todo_state;
+    }else{
+      delete S.session.todo_state;
+    }
+    if(typeof _hydrateTodosFromSession === 'function'){
+      _hydrateTodosFromSession(S.session);
+    }
+    if(typeof scheduleTodosRefresh === 'function'){
+      scheduleTodosRefresh();
+    }
     _setSessionViewedCount(sid, Number(S.session.message_count || msgs.length));
   }
 }
@@ -2156,6 +2189,7 @@ function _renderBatchActionBar(){
       ids.forEach(_clearHandoffStorageForSession);
       if(S.session&&ids.includes(S.session.session_id)){
         S.session=null;S.messages=[];S.entries=[];localStorage.removeItem('hermes-webui-session');
+        if(typeof _hydrateTodosFromSession==='function') _hydrateTodosFromSession(null);
         const remaining=await api('/api/sessions');
         if(remaining.sessions&&remaining.sessions.length){await loadSession(remaining.sessions[0].session_id);}
         else{$('msgInner').innerHTML='';$('emptyState').style.display='';}
@@ -5225,6 +5259,7 @@ async function deleteSession(sid, beforeDelete=null){
   }
   if(S.session&&S.session.session_id===sid){
     S.session=null;S.messages=[];S.entries=[];
+    if(typeof _hydrateTodosFromSession==='function') _hydrateTodosFromSession(null);
     localStorage.removeItem('hermes-webui-session');
     // load the most recent remaining session, or show blank if none left
     const remaining=await api('/api/sessions');

static/ui.js

@@ -1,4 +1,11 @@
-const S={session:null,messages:[],entries:[],busy:false,pendingFiles:[],toolCalls:[],activeStreamId:null,currentDir:'.',activeProfile:'default',showHiddenWorkspaceFiles:false};
+// `todos` is the single source of truth for the Todos panel.  Any update
+// goes through the `todo_state` SSE event (live) or session.todo_state
+// (cold-load).  `todoStateMeta` doubles as a sentinel: while it is null
+// no explicit signal has been seen, so loadTodos() falls back to the
+// legacy reverse-scan over S.messages — that keeps new clients working
+// against old servers (Phase 1 may not yet be deployed everywhere).
+// See api/todo_state.py for the wire contract.
+const S={session:null,messages:[],entries:[],busy:false,pendingFiles:[],toolCalls:[],activeStreamId:null,currentDir:'.',activeProfile:'default',showHiddenWorkspaceFiles:false,todos:[],todoStateMeta:null};
 
 function assistantDisplayName(){
   if(S.activeProfile&&S.activeProfile!=='default') return S.activeProfile.charAt(0).toUpperCase()+S.activeProfile.slice(1);
@@ -4797,11 +4804,20 @@ function _compactInflightState(state){
   const limits=_getInflightStateLimits();
   const messages=Array.isArray(state.messages)?state.messages.slice(-limits.messages):[];
   const toolCalls=Array.isArray(state.toolCalls)?state.toolCalls.slice(-limits.toolCalls):[];
+  // Phase 2: persist the live todo snapshot so reload / SSE reattach
+  // restores the panel without waiting for the next live `todo` write.
+  // The list is bounded by the agent (typically <20 items) and each
+  // item is small, so no per-list cap is needed beyond the existing
+  // stringChars truncation in _truncateInflightValue.
+  const todos=Array.isArray(state.todos)?state.todos:null;
+  const todoStateMeta=(state.todoStateMeta&&typeof state.todoStateMeta==='object')?state.todoStateMeta:null;
   return _truncateInflightValue({
     streamId:state.streamId||null,
     messages,
     uploaded:Array.isArray(state.uploaded)?state.uploaded.slice(-20):[],
     toolCalls,
+    todos,
+    todoStateMeta,
   }, limits.stringChars);
 }
 function _writeInflightStateMap(all){
@@ -4863,6 +4879,154 @@ function clearInflightState(sid){
   }catch(_){ }
 }
 
+// ─── Todo state: single source of truth + render scheduling ─────────────────
+//
+// Three concerns live together so they can share state cleanly:
+//
+//   1. _todosHash(items)  — cheap content fingerprint; skips re-render when
+//      a snapshot would paint the same DOM.  Used both as a short-circuit
+//      and as the hash that compares "rendered vs current" snapshots.
+//
+//   2. scheduleTodosRefresh() — coalesces multiple `todo_state` events that
+//      land in the same animation frame into a single loadTodos() call.
+//      Skips work entirely when the panel is not active.
+//
+//   3. _hydrateTodosFromSession(session) — applies cold-load todo_state
+//      from the session GET payload, or clears the panel when neither a
+//      cold-load nor an INFLIGHT signal is available.  Called at every
+//      `S.session = ...` settle point so cross-session navigation never
+//      leaves a stale list visible.
+//
+// The hash is keyed on (id, content, status); the render itself uses
+// `esc()` for any user-controlled string, so XSS surface is the same as
+// any other innerHTML path in this file.
+let _todosLastRenderedHash=null;
+let _todosRenderRafId=0;
+
+function _todosHash(items){
+  if(!Array.isArray(items)) return '';
+  // String concat outperforms JSON.stringify on small arrays in V8 (no
+  // intermediate object allocation) and is exact enough — the field set
+  // matches what the renderer reads, so any visible change in DOM
+  // implies a hash change.  Field separators (\x1f, \x1e) are control
+  // chars unlikely to appear in real todo content, so collisions across
+  // boundaries are not realistic.
+  let h=items.length+'|';
+  for(let i=0;i<items.length;i++){
+    const t=items[i]||{};
+    h+=String(t.id==null?'':t.id)+'\x1f'+String(t.content==null?'':t.content)+'\x1f'+String(t.status==null?'':t.status)+'\x1e';
+  }
+  return h;
+}
+
+function _todosPanelIsActive(){
+  if(typeof document==='undefined') return false;
+  const panel=document.getElementById('panelTodos');
+  return !!(panel&&panel.classList&&panel.classList.contains('active'));
+}
+
+function scheduleTodosRefresh(){
+  // Idempotent: many `todo_state` events fire on each tool result, but
+  // only the latest snapshot needs to paint.  RAF lets us coalesce
+  // without timer drift.
+  if(_todosRenderRafId) return;
+  if(typeof requestAnimationFrame!=='function'){
+    if(typeof loadTodos==='function') loadTodos();
+    return;
+  }
+  _todosRenderRafId=requestAnimationFrame(()=>{
+    _todosRenderRafId=0;
+    if(!_todosPanelIsActive()) return;
+    if(typeof loadTodos==='function') loadTodos();
+  });
+}
+
+function _resetTodosRenderCache(){
+  // Clear after every cross-session navigation so the next render is
+  // never short-circuited against a hash from a different session.
+  _todosLastRenderedHash=null;
+}
+
+function _hydrateTodosFromSession(session){
+  // Three input cases, three deterministic outcomes:
+  //   a) cold-load AND inflight both present  → pick newer by ts so a
+  //      stale cold-load from the session GET cannot regress a fresher
+  //      INFLIGHT snapshot persisted from a still-running stream
+  //      (avoids visible rollback on reload).
+  //   b) only one of cold-load / inflight is present  → use it.
+  //   c) neither  → reset to empty + sentinel so loadTodos() falls
+  //      through to the legacy reverse-scan or paints the empty state.
+  const sid=(session&&session.session_id)||'';
+  const inflight=(typeof INFLIGHT==='object'&&INFLIGHT&&sid)?INFLIGHT[sid]:null;
+  const cold=session&&session.todo_state;
+  const coldOk=!!(cold&&Array.isArray(cold.todos));
+  const inflightOk=!!(inflight&&Array.isArray(inflight.todos)&&inflight.todoStateMeta);
+  const coldTs=coldOk?(Number(cold.ts)||0):0;
+  const inflightTs=inflightOk?(Number(inflight.todoStateMeta&&inflight.todoStateMeta.ts)||0):0;
+  // Whether a live stream currently owns this session. This is the signal
+  // that disambiguates a ts-less cold-load (see below); it comes from the
+  // session GET payload (mirrors sessions.js `S.session.active_stream_id`).
+  const streamActive=!!(session&&session.active_stream_id);
+  if(coldOk&&inflightOk){
+    // Reconcile the server's settled cold-load snapshot against the
+    // locally-persisted INFLIGHT snapshot.
+    //
+    // coldTs===0 means the cold-load carries NO usable timestamp, so we
+    // cannot order it against INFLIGHT by recency. A todo tool message can
+    // legitimately lose its `timestamp` during context compression/rebuild
+    // (the on-disk message ends up timestamp=None), and derive_todo_state
+    // (api/todo_state.py) then returns the correct latest-by-POSITION todos
+    // but omits `ts`. The tie-break depends on who owns the INFLIGHT tail:
+    //
+    //   - stream ACTIVE → INFLIGHT is the live tail. The most recent todo
+    //     write may still be in flight and not yet settled into the message
+    //     list derive_todo_state scans, so a ts-less cold-load can be an
+    //     OLDER (pre-latest-write) view. Letting cold win here rolls the
+    //     panel back to a stale list, and since the stream may have just
+    //     ended on that very write there is no guaranteed forward SSE event
+    //     to self-heal. So prefer INFLIGHT. If cold is in fact newer, the
+    //     reattach replay (sessions.js attachLiveStream, reconnecting) re-
+    //     emits the journaled `todo_state` events which reconcile forward by
+    //     ts, so any transient discrepancy corrects itself.
+    //
+    //   - stream IDLE → INFLIGHT is leftover from a finished/crashed stream
+    //     (idle sessions purge it shortly after, sessions.js), and there is
+    //     no replay to correct anything. The settled cold-load is the
+    //     authoritative latest-by-position view, so prefer cold. This also
+    //     preserves the original fix for the "shows an old todo list" bug,
+    //     where a stale prior-turn INFLIGHT must not beat a ts-less cold-load.
+    //
+    // When coldTs>0 the original recency rule stands: strict ">", and on a
+    // tie prefer INFLIGHT for the freshest in-tab edits.
+    const coldWins=(coldTs===0)?(!streamActive):(coldTs>inflightTs);
+    if(coldWins){
+      S.todos=cold.todos;
+      S.todoStateMeta={
+        ts:coldTs,
+        source:'cold-load',
+        version:Number(cold.version)||1,
+      };
+    }else{
+      S.todos=inflight.todos;
+      S.todoStateMeta=inflight.todoStateMeta;
+    }
+  }else if(coldOk){
+    S.todos=cold.todos;
+    S.todoStateMeta={
+      ts:coldTs,
+      source:'cold-load',
+      version:Number(cold.version)||1,
+    };
+  }else if(inflightOk){
+    S.todos=inflight.todos;
+    S.todoStateMeta=inflight.todoStateMeta;
+  }else{
+    S.todos=[];
+    S.todoStateMeta=null;
+  }
+  _resetTodosRenderCache();
+}
+
 function snapshotLiveTurnHtmlForSession(sid){
   // Keep the DOM snapshot memory-only. Persisted INFLIGHT state intentionally
   // stores structured stream state, not outerHTML, so a hard reload still uses

tests/test_streaming_todo_state_static.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+
+
+STREAMING_PY = Path(__file__).parent.parent / "api" / "streaming.py"
+
+
+def _emit_todo_state_calls() -> list[ast.Call]:
+    tree = ast.parse(STREAMING_PY.read_text(encoding="utf-8"))
+    calls: list[ast.Call] = []
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "emit_todo_state":
+            calls.append(node)
+    return calls
+
+
+def test_streaming_imports_todo_state_emitter():
+    tree = ast.parse(STREAMING_PY.read_text(encoding="utf-8"))
+
+    assert any(
+        isinstance(node, ast.ImportFrom)
+        and node.module == "api.todo_state"
+        and any(alias.name == "emit_todo_state" for alias in node.names)
+        for node in ast.walk(tree)
+    )
+
+
+def test_streaming_emits_todo_state_on_both_tool_callback_shapes():
+    calls = _emit_todo_state_calls()
+
+    assert len(calls) >= 2
+    for call in calls[:2]:
+        kw_names = {kw.arg for kw in call.keywords}
+        assert {"name", "function_result", "session_id", "stream_id"}.issubset(kw_names)
+
+
+def test_streaming_prefers_full_tool_result_when_available():
+    src = STREAMING_PY.read_text(encoding="utf-8")
+
+    assert "cb_kwargs.get('result')" in src
+    assert "else preview" in src
+    assert "function_result=function_result" in src

tests/test_todo_live_frontend_static.py

@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+
+
+REPO_ROOT = Path(__file__).parent.parent
+
+
+def _read_static(path: str) -> str:
+    return (REPO_ROOT / path).read_text(encoding="utf-8")
+
+
+def test_frontend_state_and_inflight_storage_include_todo_snapshot():
+    ui = _read_static("static/ui.js")
+    messages = _read_static("static/messages.js")
+
+    assert "todos:[],todoStateMeta:null" in ui
+    assert "const todos=Array.isArray(state.todos)?state.todos:null" in ui
+    assert "todoStateMeta" in ui[ui.find("function _compactInflightState"):ui.find("function _writeInflightStateMap")]
+    assert "todos:Array.isArray(inflight.todos)?inflight.todos:S.todos" in messages
+    assert "todoStateMeta:inflight.todoStateMeta||S.todoStateMeta||null" in messages
+
+
+def test_frontend_todo_state_listener_is_registered_and_journaled():
+    messages = _read_static("static/messages.js")
+
+    assert "source.addEventListener('todo_state'" in messages
+    assert "if(d.session_id&&d.session_id!==activeSid) return;" in messages
+    assert "if(!S.session||S.session.session_id!==activeSid) return;" in messages
+    assert "if(incomingTs&&currentTs&&incomingTs<currentTs) return;" in messages
+    assert "inflight.todos=S.todos" in messages
+    assert "'todo_state','approval'" in messages
+
+
+def test_hydrate_todos_from_session_reconciles_cold_and_inflight_snapshots():
+    ui = _read_static("static/ui.js")
+    start = ui.find("function _hydrateTodosFromSession(session)")
+    end = ui.find("function snapshotLiveTurnHtmlForSession")
+
+    assert start != -1
+    assert end != -1
+    block = ui[start:end]
+    assert "const cold=session&&session.todo_state;" in block
+    assert "const streamActive=!!(session&&session.active_stream_id);" in block
+    assert "const coldWins=(coldTs===0)?(!streamActive):(coldTs>inflightTs);" in block
+    assert "S.todos=cold.todos" in block
+    assert "S.todos=inflight.todos" in block
+    assert "S.todoStateMeta=null" in block
+    assert "_resetTodosRenderCache();" in block
+
+
+@pytest.mark.skipif(shutil.which("node") is None, reason="node is required for frontend handler behavior test")
+def test_todo_state_listener_replaces_snapshot_filters_session_and_rejects_older_ts(tmp_path):
+    script = r'''
+const fs = require('fs');
+const src = fs.readFileSync(process.argv[2], 'utf8');
+function extractTodoStateHandler(source) {
+  const start = source.indexOf("source.addEventListener('todo_state'");
+  if (start < 0) throw new Error('todo_state listener not found');
+  const arrow = source.indexOf('=>', start);
+  const bodyStart = source.indexOf('{', arrow);
+  let depth = 0;
+  for (let i = bodyStart; i < source.length; i++) {
+    const ch = source[i];
+    if (ch === '{') depth++;
+    else if (ch === '}') {
+      depth--;
+      if (depth === 0) return source.slice(bodyStart + 1, i);
+    }
+  }
+  throw new Error('todo_state listener body not closed');
+}
+const body = extractTodoStateHandler(src);
+let S = {session:{session_id:'s1'}, todos:[], todoStateMeta:null};
+let INFLIGHT = {s1:{messages:[]}};
+let activeSid = 's1';
+let persistCalls = 0;
+function persistInflightState(){ persistCalls++; }
+let refreshCalls = 0;
+function scheduleTodosRefresh(){ refreshCalls++; }
+const handler = new Function('e','S','INFLIGHT','activeSid','persistInflightState','scheduleTodosRefresh', body);
+function fire(payload){ handler({data: JSON.stringify(payload)}, S, INFLIGHT, activeSid, persistInflightState, scheduleTodosRefresh); }
+function assert(cond, msg){ if(!cond) throw new Error(msg); }
+
+fire({session_id:'other', todos:[{id:'x', content:'wrong', status:'pending'}], ts:20});
+assert(S.todos.length === 0, 'cross-session event must be ignored');
+assert(persistCalls === 0 && refreshCalls === 0, 'ignored event must not persist or refresh');
+
+fire({session_id:'s1', todos:[{id:'a', content:'current', status:'in_progress'}], ts:10, version:1});
+assert(S.todos[0].id === 'a', 'valid event replaces S.todos');
+assert(S.todoStateMeta.ts === 10, 'valid event stores timestamp');
+assert(INFLIGHT.s1.todos[0].id === 'a', 'valid event mirrors into INFLIGHT');
+assert(INFLIGHT.s1.todoStateMeta.ts === 10, 'valid event mirrors meta into INFLIGHT');
+assert(persistCalls === 1 && refreshCalls === 1, 'valid event persists and schedules refresh');
+
+fire({session_id:'s1', todos:[{id:'old', content:'old', status:'pending'}], ts:5, version:1});
+assert(S.todos[0].id === 'a', 'older event must not roll back S.todos');
+assert(persistCalls === 1 && refreshCalls === 1, 'older event must not persist or refresh');
+
+handler({data:'not json'}, S, INFLIGHT, activeSid, persistInflightState, scheduleTodosRefresh);
+assert(S.todos[0].id === 'a', 'malformed event must be swallowed');
+'''
+    script_path = tmp_path / "todo_listener_test.js"
+    script_path.write_text(script, encoding="utf-8")
+    result = subprocess.run(
+        ["node", str(script_path), str(REPO_ROOT / "static" / "messages.js")],
+        cwd=REPO_ROOT,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+    assert result.returncode == 0, result.stderr or result.stdout

tests/test_todo_panel_cold_load_static.py

@@ -4,15 +4,17 @@ from pathlib import Path
 REPO_ROOT = Path(__file__).parent.parent
 
 
-def test_ensure_messages_loaded_copies_session_todo_state_sidecar():
+def test_ensure_messages_loaded_hydrates_session_todo_state_sidecar():
     src = (REPO_ROOT / "static" / "sessions.js").read_text(encoding="utf-8")
 
-    assert "Object.prototype.hasOwnProperty.call(data.session,'todo_state')" in src
-    assert "S.session.todo_state=data.session.todo_state" in src
-    assert "else delete S.session.todo_state" in src
+    assert "if(data.session.todo_state !== undefined)" in src
+    assert "S.session.todo_state = data.session.todo_state" in src
+    assert "delete S.session.todo_state" in src
+    assert "_hydrateTodosFromSession(S.session)" in src
+    assert "scheduleTodosRefresh()" in src
 
 
-def test_load_todos_prefers_session_todo_state_before_legacy_scan():
+def test_load_todos_renders_single_source_of_truth_before_legacy_scan():
     src = (REPO_ROOT / "static" / "panels.js").read_text(encoding="utf-8")
     start = src.find("function loadTodos()")
     end = src.find("function _legacyTodosFromMessages()")
@@ -21,11 +23,10 @@ def test_load_todos_prefers_session_todo_state_before_legacy_scan():
     assert end != -1
     load_todos = src[start:end]
 
-    assert "const sessionTodoState = S.session && S.session.todo_state;" in load_todos
-    assert "sessionTodoState && Array.isArray(sessionTodoState.todos)" in load_todos
-    assert "todos = sessionTodoState.todos;" in load_todos
+    assert "if (S.todoStateMeta)" in load_todos
+    assert "todos = Array.isArray(S.todos) ? S.todos : [];" in load_todos
     assert "todos = _legacyTodosFromMessages();" in load_todos
-    assert load_todos.find("todos = sessionTodoState.todos;") < load_todos.find("todos = _legacyTodosFromMessages();")
+    assert load_todos.find("todos = Array.isArray(S.todos) ? S.todos : [];") < load_todos.find("todos = _legacyTodosFromMessages();")
 
 
 def test_legacy_todos_fallback_still_uses_raw_session_messages():

tests/test_todo_state_emission.py

@@ -0,0 +1,93 @@
+import json
+
+from api import helpers
+from api.todo_state import EVENT_NAME, emit_todo_state
+
+
+def _payload(todos):
+    return json.dumps({"todos": todos, "summary": {"total": len(todos)}})
+
+
+def test_emit_todo_state_emits_full_snapshot_with_session_tags():
+    events = []
+
+    emitted = emit_todo_state(
+        lambda event, data: events.append((event, data)),
+        name="todo",
+        function_result=_payload([{"id": "1", "content": "ship", "status": "pending"}]),
+        session_id="session-1",
+        stream_id="stream-1",
+    )
+
+    assert emitted is True
+    assert len(events) == 1
+    event, data = events[0]
+    assert event == EVENT_NAME
+    assert data["session_id"] == "session-1"
+    assert data["stream_id"] == "stream-1"
+    assert data["source"] == "tool"
+    assert data["todos"][0]["content"] == "ship"
+    assert data["summary"]["total"] == 1
+    assert data["version"] == 1
+    assert isinstance(data["ts"], float)
+
+
+def test_emit_todo_state_ignores_non_todo_or_bad_payload():
+    events = []
+
+    assert emit_todo_state(lambda event, data: events.append((event, data)), name="web_search", function_result=_payload([]), session_id="s", stream_id="st") is False
+    assert emit_todo_state(lambda event, data: events.append((event, data)), name="todo", function_result="not json", session_id="s", stream_id="st") is False
+    assert emit_todo_state(lambda event, data: events.append((event, data)), name="todo", function_result='{"todos":"not-list"}', session_id="s", stream_id="st") is False
+    assert events == []
+
+
+def test_emit_todo_state_redacts_before_sse(monkeypatch):
+    events = []
+
+    def fake_redact(value, *, _enabled=None):
+        redacted = dict(value)
+        redacted["todos"] = [{"id": "1", "content": "[REDACTED]", "status": "pending"}]
+        return redacted
+
+    monkeypatch.setattr(helpers, "_redact_value", fake_redact)
+
+    emitted = emit_todo_state(
+        lambda event, data: events.append((event, data)),
+        name="todo",
+        function_result=_payload([{"id": "1", "content": "secret token", "status": "pending"}]),
+        session_id="s",
+        stream_id="st",
+    )
+
+    assert emitted is True
+    assert events[0][1]["todos"][0]["content"] == "[REDACTED]"
+
+
+def test_emit_todo_state_threads_redaction_enabled_flag_once(monkeypatch):
+    events = []
+    settings_calls = []
+    redaction_enabled_args = []
+
+    def fake_load_settings():
+        settings_calls.append("load_settings")
+        return {"api_redact_enabled": False}
+
+    def fake_redact(value, *, _enabled=None):
+        redaction_enabled_args.append(_enabled)
+        return dict(value)
+
+    monkeypatch.setattr("api.config.load_settings", fake_load_settings)
+    monkeypatch.setattr(helpers, "_redact_value", fake_redact)
+
+    emitted = emit_todo_state(
+        lambda event, data: events.append((event, data)),
+        name="todo",
+        function_result=_payload([{"id": "1", "content": "ordinary task", "status": "pending"}]),
+        session_id="s",
+        stream_id="st",
+    )
+
+    assert emitted is True
+    assert len(events) == 1
+    assert settings_calls == ["load_settings"]
+    assert redaction_enabled_args == [False]