refactor: priority-based tool result dispatch selection

CONTRIBUTING.md

@@ -145,7 +145,7 @@ See [`docs/architecture.md`](docs/architecture.md) for data flow, export state m
 
 Claude Code assistant `tool_use` blocks carry a `name` string (e.g. `"Read"`, `"Bash"`). The browser coordinates that name across four sites; drift is caught by `tests/test_tool_dispatch_sync.py`.
 
-1. **`utils/tool_dispatch.py`** — add the name to `_FILE_ACTIVITY_HANDLERS` (`None` if no file/bash/web side effects); `KNOWN_TOOL_TYPES` is derived from its keys. If the tool has a distinct `toolUseResult` JSON shape, add `(predicate, builder)` to `_TOOL_RESULT_DISPATCH` (respect ordering — see module docstring and `tests/test_tool_dispatch_ordering.py`).
+1. **`utils/tool_dispatch.py`** — add the name to `_FILE_ACTIVITY_HANDLERS` (`None` if no file/bash/web side effects); `KNOWN_TOOL_TYPES` is derived from its keys. If the tool has a distinct `toolUseResult` JSON shape, add a `ToolResultDispatchEntry` to `_TOOL_RESULT_DISPATCH` (set `priority` when overlapping another predicate — see module docstring and `tests/test_tool_dispatch_ordering.py`).
 2. **`models/tool_results.py`** — add the name to `ToolNameLiteral` and, when the tool has a distinct result payload, add the TypedDict, type guard (`is_*_tool_result`), and union member on `ToolResultUnion`.
 3. **`utils/md_exporter.py`** — add an `elif name == "…"` branch in `_render_tool_use` (sync test parses these branches).
 4. **`static/js/render/registry.js`** — add a `TOOL_USE_RENDERERS` entry (and a `tool_use/*.js` renderer module).

docs/architecture.md

@@ -71,13 +71,13 @@
 
 ## Dispatch table
 
-In `utils/tool_dispatch.py`, tool results are classified through `_parse_tool_result`, a **predicate-ordered dispatch table** (not a simple `if tool_name == ...` chain). **Order is load-bearing**: the first matching predicate wins. Tests in `tests/test_jsonl_parser.py` and `tests/test_real_session_fixtures.py` guard ordering regressions.
+In `utils/tool_dispatch.py`, tool results are classified through `_parse_tool_result`, a **priority-based dispatch table**. Every matching predicate is considered; the entry with the highest `priority` wins (ties favor earlier registration). Overlap exceptions are explicit priorities, not tuple position — see `tests/test_tool_dispatch_ordering.py` and `tests/test_tool_dispatch_adversarial.py`.
 
 When adding a new tool renderer:
 
-1. Add a `(predicate, builder)` pair to `_TOOL_RESULT_DISPATCH` in `utils/tool_dispatch.py`, preserving existing predicate order unless you also update fixtures and ordering tests (`tests/test_jsonl_parser.py`, `tests/test_real_session_fixtures.py`). Order is **not** “specific before generic” in general — the first match wins. `is_task_message_tool_result` is the intentional broad-before-narrow exception (`task_id` or `message` before retrieval/completed/async).
+1. Add a `ToolResultDispatchEntry` to `_TOOL_RESULT_DISPATCH` in `utils/tool_dispatch.py`. Set `priority` higher than any overlapping predicate it must beat, and add a row to `ORDERING_INVARIANTS` in `tests/test_tool_dispatch_ordering.py` when overlaps exist. Documented exceptions: `plan` (priority 1) over `file_write` (0); `task_message` (1) over `task_retrieval` / `task_completed` / `task_async` (0).
 2. Add or extend a JSONL fixture under `tests/fixtures/` (especially for overlaps with existing predicates).
-3. Run `pytest tests/test_jsonl_parser.py tests/test_real_session_fixtures.py -v`.
+3. Run `pytest tests/test_tool_dispatch_ordering.py tests/test_tool_dispatch_adversarial.py tests/test_jsonl_parser.py tests/test_real_session_fixtures.py -v`.
 
 ## Export state machine
 

tests/test_jsonl_parser.py

@@ -241,7 +241,7 @@ def test_plan_result(self):
         assert r["result_type"] == "plan"
 
     def test_plan_with_content_not_classified_as_file_write(self):
-        """plan is registered before file_write in _TOOL_RESULT_DISPATCH."""
+        """plan outranks file_write on overlapping toolUseResult blobs."""
         r = _parse_tool_result(
             {
                 "plan": [],

tests/test_real_session_fixtures.py

@@ -13,7 +13,7 @@
 import pytest
 
 from utils.jsonl_parser import parse_session
-from utils.tool_dispatch import _TOOL_RESULT_DISPATCH, _parse_tool_result
+from utils.tool_dispatch import _TOOL_RESULT_DISPATCH, _parse_tool_result, _winning_dispatch_entry
 
 FIXTURES_DIR = os.path.join(os.path.dirname(__file__), "fixtures")
 
@@ -89,7 +89,7 @@ def test_real_session_minimal_has_bash_tool_result() -> None:
 
 
 def test_real_session_all_tool_types_covers_dispatch_predicates() -> None:
-    hit: set[int] = set()
+    hit: set[str] = set()
     path = _fixture_path("real_session_all_tool_types.jsonl")
     with open(path, encoding="utf-8") as f:
         for line in f:
@@ -100,14 +100,10 @@ def test_real_session_all_tool_types_covers_dispatch_predicates() -> None:
             tr = entry.get("toolUseResult")
             if not isinstance(tr, dict):
                 continue
-            matched = False
-            for i, (pred, _) in enumerate(_TOOL_RESULT_DISPATCH):
-                if pred(tr):
-                    hit.add(i)
-                    matched = True
-                    break
-            assert matched, f"toolUseResult matched no predicate: {list(tr.keys())}"
-    assert hit == set(range(len(_TOOL_RESULT_DISPATCH)))
+            winner = _winning_dispatch_entry(tr)
+            assert winner is not None, f"toolUseResult matched no predicate: {list(tr.keys())}"
+            hit.add(winner.id)
+    assert hit == {entry.id for entry in _TOOL_RESULT_DISPATCH}
 
 
 def test_real_session_nested_tools_has_sidechain_and_tool_use() -> None:

tests/test_tool_dispatch_adversarial.py

@@ -1,13 +1,14 @@
 """Behavioral adversarial fixtures for ``_TOOL_RESULT_DISPATCH`` predicate overlap.
 
-Structural tuple-position guards live in ``test_tool_dispatch_ordering.py``.
+Structural priority guards live in ``test_tool_dispatch_ordering.py``.
 These tests construct ``toolUseResult`` JSON that satisfies multiple predicates
-and assert the classified winner via ``_parse_tool_result`` (first match wins).
+and assert the classified winner via ``_parse_tool_result`` (highest priority).
 """
 
 from __future__ import annotations
 
 from collections.abc import Callable
+from dataclasses import replace
 
 import pytest
 
@@ -118,19 +119,19 @@ def test_task_retrieval_narrow_shape_without_task_message_keys() -> None:
     assert result.get("task_id") == "task-narrow"
 
 
-def test_inverted_plan_file_write_dispatch_misclassifies_overlap(
+def test_inverted_plan_file_write_priority_misclassifies_overlap(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    """Regression: swapping plan below file_write flips the overlap winner."""
-    table = list(_TOOL_RESULT_DISPATCH)
-    plan_idx = next(
-        i for i, (pred, _) in enumerate(table) if pred.__name__ == "is_plan_tool_result"
+    """Regression: giving file_write higher priority than plan flips the overlap winner."""
+    table = tuple(
+        replace(entry, priority=1)
+        if entry.id == "file_write"
+        else replace(entry, priority=0)
+        if entry.id == "plan"
+        else entry
+        for entry in _TOOL_RESULT_DISPATCH
     )
-    write_idx = next(
-        i for i, (pred, _) in enumerate(table) if pred.__name__ == "is_file_write_tool_result"
-    )
-    table[plan_idx], table[write_idx] = table[write_idx], table[plan_idx]
-    monkeypatch.setattr(tool_dispatch, "_TOOL_RESULT_DISPATCH", tuple(table))
+    monkeypatch.setattr(tool_dispatch, "_TOOL_RESULT_DISPATCH", table)
 
     result = _parse_tool_result(PLAN_FILE_WRITE_OVERLAP)
     assert result is not None

tests/test_tool_dispatch_ordering.py

@@ -1,8 +1,8 @@
-"""Structural ordering invariants for ``_TOOL_RESULT_DISPATCH``.
+"""Structural priority invariants for ``_TOOL_RESULT_DISPATCH``.
 
-First matching predicate wins; misordering silently misclassifies tool results.
-Invariants are declared as ``(before, after, reason)`` triples — add a row to
-``ORDERING_INVARIANTS`` when inserting a predicate that must sit above another.
+When multiple predicates match, the highest ``priority`` wins. Invariants are
+declared as ``(before, after, reason)`` triples — add a row to
+``ORDERING_INVARIANTS`` when a new predicate must outrank another on overlap.
 """
 
 from collections.abc import Callable
@@ -17,30 +17,30 @@
     is_task_message_tool_result,
     is_task_retrieval_tool_result,
 )
-from utils.tool_dispatch import _TOOL_RESULT_DISPATCH
+from utils.tool_dispatch import _TOOL_RESULT_DISPATCH, ToolResultDispatchEntry
 
 Predicate = Callable[..., bool]
 
 ORDERING_INVARIANTS: list[tuple[Predicate, Predicate, str]] = [
     (
         is_plan_tool_result,
         is_file_write_tool_result,
-        "plan blobs may carry filePath + content; plan must win before file_write",
+        "plan blobs may carry filePath + content; plan must outrank file_write",
     ),
     (
         is_task_message_tool_result,
         is_task_retrieval_tool_result,
-        "task_message is broad (task_id or message); must precede narrower task_retrieval",
+        "task_message is broad (task_id or message); must outrank task_retrieval",
     ),
     (
         is_task_message_tool_result,
         is_task_completed_tool_result,
-        "task_message is broad (task_id or message); must precede narrower task_completed",
+        "task_message is broad (task_id or message); must outrank task_completed",
     ),
     (
         is_task_message_tool_result,
         is_task_async_tool_result,
-        "task_message is broad (task_id or message); must precede narrower task_async",
+        "task_message is broad (task_id or message); must outrank task_async",
     ),
 ]
 
@@ -52,12 +52,10 @@
 ]
 
 
-def _predicate_index(predicate: Predicate) -> int:
-    for i, entry in enumerate(_TOOL_RESULT_DISPATCH):
-        pred = entry[0]
-        # Identity match: dispatch table must store bare function refs (not wrappers).
-        if pred is predicate:
-            return i
+def _entry_for(predicate: Predicate) -> ToolResultDispatchEntry:
+    for entry in _TOOL_RESULT_DISPATCH:
+        if entry.predicate is predicate:
+            return entry
     raise ValueError(f"predicate {predicate.__name__} not found in _TOOL_RESULT_DISPATCH")
 
 
@@ -71,10 +69,10 @@ def test_tool_dispatch_ordering_invariant(
     after: Predicate,
     reason: str,
 ) -> None:
-    before_idx = _predicate_index(before)
-    after_idx = _predicate_index(after)
-    assert before_idx < after_idx, (
-        f"_TOOL_RESULT_DISPATCH ordering violation: "
-        f"{before.__name__} (index {before_idx}) must precede "
-        f"{after.__name__} (index {after_idx}). Reason: {reason}"
+    before_entry = _entry_for(before)
+    after_entry = _entry_for(after)
+    assert before_entry.priority > after_entry.priority, (
+        f"_TOOL_RESULT_DISPATCH priority violation: "
+        f"{before.__name__} (priority {before_entry.priority}) must outrank "
+        f"{after.__name__} (priority {after_entry.priority}). Reason: {reason}"
     )

utils/tool_dispatch.py

@@ -1,18 +1,19 @@
 """Tool-result classification for Claude Code JSONL toolUseResult blobs.
 
-Dispatch registry: **first matching predicate wins** (legacy if/elif parity).
-Order is load-bearing — do not sort alphabetically or "more specific first"
-without replaying tests and real session fixtures.
+Dispatch registry: among **all matching predicates**, the entry with the highest
+``priority`` wins (ties favor earlier registration). Default priority is 0.
+Documented overlap exceptions use priority 1 so contributors set priority on
+new shapes instead of reasoning about full tuple position:
 
-Notably ``task_message`` is broad (``task_id`` or ``message``) and sits before
-``task_retrieval`` / ``task_completed`` / ``task_async``.
+- ``plan`` (1) over ``file_write`` (0) when both match — plan blobs may carry
+  ``filePath`` + ``content``.
+- ``task_message`` (1) over ``task_retrieval`` / ``task_completed`` /
+  ``task_async`` (0) — ``task_message`` is broad (``task_id`` or ``message``).
 
-To add a shape: append ``(pred, build)`` at the end, or insert only after
-verifying predicates above would not steal intended matches.
-
-Ordering invariants are enforced structurally by
-``tests/test_tool_dispatch_ordering.py`` — add a ``(before, after, reason)``
-tuple there when a new predicate must sit above another.
+To add a shape: append a ``ToolResultDispatchEntry`` with predicate, builder,
+and priority. If the new predicate overlaps an existing one, set priority higher
+than the shapes it must beat and add a row to ``ORDERING_INVARIANTS`` in
+``tests/test_tool_dispatch_ordering.py``.
 
 Predicates live in ``models.tool_results`` (single source of truth for narrowing).
 
@@ -22,8 +23,9 @@
    side effects); ``KNOWN_TOOL_TYPES`` is derived from its keys.
 2. Add the name to ``ToolNameLiteral`` in ``models/tool_results.py`` and, if the
    tool has a distinct ``toolUseResult`` JSON shape, add the TypedDict, predicate,
-   and ``(predicate, builder)`` pair in ``_TOOL_RESULT_DISPATCH`` (respect ordering
-   — see notes above and ``tests/test_tool_dispatch_ordering.py``).
+   and ``ToolResultDispatchEntry`` in ``_TOOL_RESULT_DISPATCH`` (set ``priority``
+   when overlapping another predicate — see notes above and
+   ``tests/test_tool_dispatch_ordering.py``).
 3. Add a Markdown branch in ``utils/md_exporter.py`` ``_render_tool_use``.
 4. Add ``TOOL_USE_RENDERERS`` entry in ``static/js/render/registry.js``.
 5. Run ``pytest tests/test_tool_dispatch_sync.py -v`` — it fails with the
@@ -33,7 +35,8 @@
 """
 
 from collections.abc import Callable
-from typing import Any, cast
+from dataclasses import dataclass
+from typing import Any
 
 from models.session import SessionMetadataBuilderDict
 from models.tool_results import (
@@ -57,6 +60,20 @@
     is_web_search_tool_result,
 )
 
+_DISPATCH_PRIORITY_DEFAULT = 0
+# Overlap winners: broad predicates that must beat narrower shapes on the same blob.
+_DISPATCH_PRIORITY_OVERLAP = 1
+
+
+@dataclass(frozen=True, slots=True)
+class ToolResultDispatchEntry:
+    """One toolUseResult classifier: predicate, builder, and overlap priority."""
+
+    id: str
+    predicate: Callable[[ToolResultDict], bool]
+    build: Callable[[ToolResultDict, dict[str, object]], dict[str, object]]
+    priority: int = _DISPATCH_PRIORITY_DEFAULT
+
 
 def _tool_result_build_bash(tr: ToolResultDict, base: dict[str, object]) -> dict[str, object]:
     result = dict(base)
@@ -216,26 +233,41 @@ def _tool_result_build_user_input(tr: ToolResultDict, base: dict[str, object]) -
     return result
 
 
-# Registry order is load-bearing (see module docstring).
-# ``plan`` before ``file_write``: plan blobs may carry ``filePath`` + ``content``.
-_TOOL_RESULT_DISPATCH = (
-    (is_bash_tool_result, _tool_result_build_bash),
-    (is_file_edit_tool_result, _tool_result_build_file_edit),
-    (is_plan_tool_result, _tool_result_build_plan),
-    (is_file_write_tool_result, _tool_result_build_file_write),
-    (is_glob_tool_result, _tool_result_build_glob),
-    (is_grep_tool_result, _tool_result_build_grep),
-    (is_read_tool_result, _tool_result_build_file_read),
-    (is_web_search_tool_result, _tool_result_build_web_search),
-    (is_web_fetch_tool_result, _tool_result_build_web_fetch),
-    (is_task_message_tool_result, _tool_result_build_task_message),
-    (is_task_retrieval_tool_result, _tool_result_build_task_retrieval),
-    (is_task_completed_tool_result, _tool_result_build_task_completed),
-    (is_task_async_tool_result, _tool_result_build_task_async),
-    (is_todo_write_tool_result, _tool_result_build_todo_write),
-    (is_user_input_tool_result, _tool_result_build_user_input),
+# Registration order is tie-break only when priorities are equal.
+_TOOL_RESULT_DISPATCH: tuple[ToolResultDispatchEntry, ...] = (
+    ToolResultDispatchEntry("bash", is_bash_tool_result, _tool_result_build_bash),
+    ToolResultDispatchEntry("file_edit", is_file_edit_tool_result, _tool_result_build_file_edit),
+    ToolResultDispatchEntry(
+        "plan",
+        is_plan_tool_result,
+        _tool_result_build_plan,
+        priority=_DISPATCH_PRIORITY_OVERLAP,
+    ),
+    ToolResultDispatchEntry("file_write", is_file_write_tool_result, _tool_result_build_file_write),
+    ToolResultDispatchEntry("glob", is_glob_tool_result, _tool_result_build_glob),
+    ToolResultDispatchEntry("grep", is_grep_tool_result, _tool_result_build_grep),
+    ToolResultDispatchEntry("read", is_read_tool_result, _tool_result_build_file_read),
+    ToolResultDispatchEntry("web_search", is_web_search_tool_result, _tool_result_build_web_search),
+    ToolResultDispatchEntry("web_fetch", is_web_fetch_tool_result, _tool_result_build_web_fetch),
+    ToolResultDispatchEntry(
+        "task_message",
+        is_task_message_tool_result,
+        _tool_result_build_task_message,
+        priority=_DISPATCH_PRIORITY_OVERLAP,
+    ),
+    ToolResultDispatchEntry(
+        "task_retrieval", is_task_retrieval_tool_result, _tool_result_build_task_retrieval
+    ),
+    ToolResultDispatchEntry(
+        "task_completed", is_task_completed_tool_result, _tool_result_build_task_completed
+    ),
+    ToolResultDispatchEntry("task_async", is_task_async_tool_result, _tool_result_build_task_async),
+    ToolResultDispatchEntry("todo_write", is_todo_write_tool_result, _tool_result_build_todo_write),
+    ToolResultDispatchEntry("user_input", is_user_input_tool_result, _tool_result_build_user_input),
 )
 
+_DISPATCH_ORDER = {entry.id: index for index, entry in enumerate(_TOOL_RESULT_DISPATCH)}
+
 # Claude Code assistant tool_use ``name`` values coordinated across parser file
 # activity, Markdown export, and the SPA ``TOOL_USE_RENDERERS`` map.
 # ``_FILE_ACTIVITY_HANDLERS`` is the single registry; ``KNOWN_TOOL_TYPES`` is derived.
@@ -303,27 +335,41 @@ def track_tool_file_activity(
         handler(tool_input, metadata)
 
 
+def _matching_dispatch_entries(tr: ToolResultDict) -> list[ToolResultDispatchEntry]:
+    return [entry for entry in _TOOL_RESULT_DISPATCH if entry.predicate(tr)]
+
+
+def _winning_dispatch_entry(tr: ToolResultDict) -> ToolResultDispatchEntry | None:
+    matches = _matching_dispatch_entries(tr)
+    if not matches:
+        return None
+    return max(
+        matches,
+        key=lambda entry: (entry.priority, -_DISPATCH_ORDER[entry.id]),
+    )
+
+
 def _parse_tool_result(
     tool_result: ToolResultUnion | None, slug: str | None = None
 ) -> dict[str, object] | None:
     """Figure out what kind of tool result this is (bash, file edit, glob, etc.)
     by looking at which keys are present, since the JSONL doesn't always tag them.
 
-    Classification uses ``_TOOL_RESULT_DISPATCH``: ordered ``(predicate, builder)``
-    pairs; the **first** predicate that matches wins (parity with the historical
-    ``if``/``elif`` chain — order is not strictly “specific before generic”).
+    Classification uses ``_TOOL_RESULT_DISPATCH``: every matching predicate is
+    considered; the entry with the highest ``priority`` wins (ties favor earlier
+    registration). Set ``priority`` above overlapping shapes instead of relying on
+    tuple position — see module docstring for documented overlap exceptions.
 
-    Append a new pair at the end to register a shape, or insert mid-table only
-    after checking interactions with broader predicates above (see notes on the
-    tuple)."""
+    Append a new ``ToolResultDispatchEntry`` to register a shape. If it overlaps an
+    existing predicate, raise its ``priority`` and add a row to
+    ``ORDERING_INVARIANTS`` in ``tests/test_tool_dispatch_ordering.py``."""
     if not is_tool_result_dict(tool_result):
         return None
 
     base: dict[str, object] = {"slug": slug}
-    for pred, build in _TOOL_RESULT_DISPATCH:
-        if pred(tool_result):
-            # Builders take ToolResultDict; cast after pred (heterogeneous tuple, no union narrow).
-            return build(cast(ToolResultDict, tool_result), base)
+    winner = _winning_dispatch_entry(tool_result)
+    if winner is not None:
+        return winner.build(tool_result, base)
 
     result = dict(base)
     result["result_type"] = "unknown"