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 without relying on registration order, and add a row to `ORDERING_INVARIANTS` in `tests/test_tool_dispatch_ordering.py` when overlaps exist. Documented exception: `plan` (priority 1) over `file_write` (0). Broad predicates like `task_message` use registration order instead of elevated priority.
 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,12 +142,13 @@ 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
-    known false-positive surface.
+    key and is registered before the narrower task predicates, so registration order
+    picks it on overlap. Future tool shapes that add ``message`` for status text
+    (e.g. web-fetch) would be misclassified as task unless dispatch is refined —
+    this test locks that known false-positive surface.
     """
     tr = {
         "agentId": "agent-sanitized",

tests/test_tool_dispatch_adversarial.py

@@ -1,55 +1,42 @@
 """Behavioral adversarial fixtures for ``_TOOL_RESULT_DISPATCH`` predicate overlap.
 
-Structural tuple-position 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).
+Overlap blobs and invariant tables live in ``test_tool_dispatch_ordering.py``.
+These tests assert classified output via ``_parse_tool_result``.
 """
 
 from __future__ import annotations
 
 from collections.abc import Callable
+from dataclasses import replace
 
 import pytest
 
-from tests.test_tool_dispatch_ordering import ORDERING_INVARIANT_IDS, ORDERING_INVARIANTS
+from models.tool_results import is_file_write_tool_result, is_task_message_tool_result
+from tests.test_tool_dispatch_ordering import (
+    ORDERING_INVARIANT_IDS,
+    ORDERING_INVARIANTS,
+    OVERLAP_BLOBS,
+)
 from utils import tool_dispatch
-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
 
-# Overlap blobs: keys chosen so both predicates in each ORDERING_INVARIANTS pair match.
-PLAN_FILE_WRITE_OVERLAP: dict[str, object] = {
-    "plan": {"name": "sprint-plan", "steps": ["index", "search", "render"]},
-    "filePath": ".cursor/plans/week28.md",
-    "content": "Plan body that would also satisfy file_write.",
-}
-
-TASK_MESSAGE_RETRIEVAL_OVERLAP: dict[str, object] = {
-    "task_id": "task-overlap-retrieval",
-    "message": "polling retrieval",
-    "retrieval_status": "found",
-    "task": {"task_id": "task-overlap-retrieval", "description": "subagent scan"},
-}
-
-TASK_MESSAGE_COMPLETED_OVERLAP: dict[str, object] = {
-    "agentId": "agent-overlap-completed",
-    "totalDurationMs": 3200,
-    "totalTokens": 900,
-    "status": "completed",
-    "message": "task finished",
-}
-
-TASK_MESSAGE_ASYNC_OVERLAP: dict[str, object] = {
-    "agentId": "agent-overlap-async",
-    "isAsync": True,
-    "description": "explore auth handlers",
-    "message": "task launched",
-}
+PLAN_FILE_WRITE_OVERLAP = OVERLAP_BLOBS["plan_before_file_write"]
+TASK_MESSAGE_RETRIEVAL_OVERLAP = OVERLAP_BLOBS["task_message_before_task_retrieval"]
+TASK_MESSAGE_COMPLETED_OVERLAP = OVERLAP_BLOBS["task_message_before_task_completed"]
+TASK_MESSAGE_ASYNC_OVERLAP = OVERLAP_BLOBS["task_message_before_task_async"]
 
 # Narrow retrieval shape: only the downstream predicate matches (no task_id/message).
 TASK_RETRIEVAL_NARROW: dict[str, object] = {
     "retrieval_status": "pending",
     "task": {"task_id": "task-narrow", "description": "wait for result"},
 }
 
+FILE_WRITE_TASK_MESSAGE_OVERLAP: dict[str, object] = {
+    "message": "status line on a write result",
+    "filePath": "/tmp/example.txt",
+    "content": "file body",
+}
+
 AssertWinner = Callable[[dict[str, object]], None]
 
 
@@ -118,19 +105,32 @@ 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_file_write_beats_task_message_when_both_match() -> None:
+    """Regression: broad task_message must not outrank earlier shapes via priority."""
+    blob = FILE_WRITE_TASK_MESSAGE_OVERLAP
+    assert is_file_write_tool_result(blob)
+    assert is_task_message_tool_result(blob)
+    winner = _winning_dispatch_entry(blob)
+    assert winner is not None
+    assert winner.id == "file_write"
+    result = _parse_tool_result(blob)
+    assert result is not None
+    assert result["result_type"] == "file_write"
+
+
+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"
-    )
-    write_idx = next(
-        i for i, (pred, _) in enumerate(table) if pred.__name__ == "is_file_write_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
     )
-    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
@@ -145,3 +145,4 @@ def test_ordering_invariants_have_adversarial_coverage() -> None:
     assert len(ORDERING_INVARIANTS) == len(ORDERING_INVARIANT_IDS)
     assert len(ORDERING_INVARIANT_IDS) == len(_INVARIANT_BEHAVIOR)
     assert set(_INVARIANT_BEHAVIOR.keys()) == set(ORDERING_INVARIANT_IDS)
+    assert set(OVERLAP_BLOBS.keys()) == set(ORDERING_INVARIANT_IDS)

tests/test_tool_dispatch_ordering.py

@@ -1,8 +1,9 @@
-"""Structural ordering invariants for ``_TOOL_RESULT_DISPATCH``.
+"""Structural overlap 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; equal priority
+favors earlier registration. Invariants are declared as ``(before, after, reason)``
+triples with a shared overlap blob — add a row to ``ORDERING_INVARIANTS`` when a
+new predicate must outrank another on overlap.
 """
 
 from collections.abc import Callable
@@ -17,30 +18,34 @@
     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,
+    _winning_dispatch_entry,
+)
 
 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",
     ),
 ]
 
@@ -51,30 +56,64 @@
     "task_message_before_task_async",
 ]
 
+# Overlap blobs: keys chosen so both predicates in each ORDERING_INVARIANTS pair match.
+OVERLAP_BLOBS: dict[str, dict[str, object]] = {
+    "plan_before_file_write": {
+        "plan": {"name": "sprint-plan", "steps": ["index", "search", "render"]},
+        "filePath": ".cursor/plans/week28.md",
+        "content": "Plan body that would also satisfy file_write.",
+    },
+    "task_message_before_task_retrieval": {
+        "task_id": "task-overlap-retrieval",
+        "message": "polling retrieval",
+        "retrieval_status": "found",
+        "task": {"task_id": "task-overlap-retrieval", "description": "subagent scan"},
+    },
+    "task_message_before_task_completed": {
+        "agentId": "agent-overlap-completed",
+        "totalDurationMs": 3200,
+        "totalTokens": 900,
+        "status": "completed",
+        "message": "task finished",
+    },
+    "task_message_before_task_async": {
+        "agentId": "agent-overlap-async",
+        "isAsync": True,
+        "description": "explore auth handlers",
+        "message": "task launched",
+    },
+}
+
 
-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")
 
 
 @pytest.mark.parametrize(
-    "before,after,reason",
-    ORDERING_INVARIANTS,
+    "before,after,reason,fixture_id",
+    [
+        (*row, invariant_id)
+        for row, invariant_id in zip(ORDERING_INVARIANTS, ORDERING_INVARIANT_IDS, strict=True)
+    ],
     ids=ORDERING_INVARIANT_IDS,
 )
 def test_tool_dispatch_ordering_invariant(
     before: Predicate,
     after: Predicate,
     reason: str,
+    fixture_id: 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}"
+    blob = OVERLAP_BLOBS[fixture_id]
+    before_entry = _entry_for(before)
+    after_entry = _entry_for(after)
+    assert before(blob), f"{fixture_id}: fixture no longer matches {before.__name__}"
+    assert after(blob), f"{fixture_id}: fixture no longer matches {after.__name__}"
+    winner = _winning_dispatch_entry(blob)
+    assert winner is not None
+    assert winner.id == before_entry.id, (
+        f"_TOOL_RESULT_DISPATCH overlap violation: expected {before_entry.id!r} "
+        f"to beat {after_entry.id!r} on {fixture_id}. Reason: {reason}"
     )