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
 
@@ -105,7 +105,7 @@ The UI is a **hash-routed** SPA with ES modules under `static/js/`:
 
 - `app.js` — routing and boot
 - `projects.js`, `sessions.js`, `search.js`, `export.js` — route handlers
-- `render/registry.js` — **tool dispatch registry** for session UI: `TOOL_USE_RENDERERS` and `TOOL_RESULT_RENDERERS` map tool name / `result_type` → render function (one module per type under `render/tool_use/` and `render/tool_result/`). Parallels backend `utils/tool_dispatch.py` (backend uses ordered predicates; frontend uses direct key lookup + fallback).
+- `render/registry.js` — **tool dispatch registry** for session UI: `TOOL_USE_RENDERERS` and `TOOL_RESULT_RENDERERS` map tool name / `result_type` → render function (one module per type under `render/tool_use/` and `render/tool_result/`). Parallels backend `utils/tool_dispatch.py` (backend uses priority-based predicate matching; frontend uses direct key lookup + fallback).
 - `static/tool_types.json` — generated manifest of backend tool-use names (`python scripts/gen_tool_types_manifest.py` from `KNOWN_TOOL_TYPES`). Fetched non-blocking at boot by `render/tool_types_manifest.js` (`void initToolTypesManifest()` in `app.js`), which cross-checks `TOOL_USE_RENDERERS` and logs `console.warn` on drift.
 - `shared/markdown.js` — markdown + **DOMPurify** sanitization (do not render raw LLM HTML)
 - `shared/state.js`, `shared/utils.js`, `shared/theme.js` — shared UI state and helpers

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:
@@ -146,11 +142,12 @@ def test_task_retrieval_not_misclassified_as_task_message() -> None:
 
 
 def test_task_completed_with_message_key_matches_task_message_first() -> None:
-    """Legacy dispatch: broad task_message runs before task_completed when ``message`` present.
+    """task_message outranks task_completed when ``message`` key is present.
 
     ``is_task_message_tool_result`` matches any dict with a ``message`` or ``task_id``
-    key. Future tool shapes that add ``message`` for status text (e.g. web-fetch) would
-    be misclassified as task until dispatch order is refined — this test locks that
+    key and holds priority 1, beating the narrower task predicates at priority 0.
+    Future tool shapes that add ``message`` for status text (e.g. web-fetch) would
+    be misclassified as task unless given higher priority — this test locks that
     known false-positive surface.
     """
     tr = {

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}"
     )