diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index fd27d94..9a0992c 100644 --- a/CONTRIBUTING.md +++ b/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). diff --git a/docs/architecture.md b/docs/architecture.md index b2ea3bb..4c164a5 100644 --- a/docs/architecture.md +++ b/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 diff --git a/tests/test_jsonl_parser.py b/tests/test_jsonl_parser.py index c927d85..052c9a3 100644 --- a/tests/test_jsonl_parser.py +++ b/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": [], diff --git a/tests/test_real_session_fixtures.py b/tests/test_real_session_fixtures.py index 291d166..15e2853 100644 --- a/tests/test_real_session_fixtures.py +++ b/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", diff --git a/tests/test_tool_dispatch_adversarial.py b/tests/test_tool_dispatch_adversarial.py index 95afe42..f181338 100644 --- a/tests/test_tool_dispatch_adversarial.py +++ b/tests/test_tool_dispatch_adversarial.py @@ -1,48 +1,29 @@ """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] = { @@ -50,6 +31,12 @@ "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) diff --git a/tests/test_tool_dispatch_ordering.py b/tests/test_tool_dispatch_ordering.py index 008f644..3e99dc4 100644 --- a/tests/test_tool_dispatch_ordering.py +++ b/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,7 +18,11 @@ 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] @@ -25,22 +30,22 @@ ( 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}" ) diff --git a/utils/tool_dispatch.py b/utils/tool_dispatch.py index 588e561..507a42c 100644 --- a/utils/tool_dispatch.py +++ b/utils/tool_dispatch.py @@ -1,18 +1,21 @@ """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. +Use ``priority`` when a shape must beat a **later** overlapping entry without +moving registration order. Documented overlap exception: -Notably ``task_message`` is broad (``task_id`` or ``message``) and sits before -``task_retrieval`` / ``task_completed`` / ``task_async``. +- ``plan`` (priority 1) over ``file_write`` (0) when both match — plan blobs may + carry ``filePath`` + ``content``. -To add a shape: append ``(pred, build)`` at the end, or insert only after -verifying predicates above would not steal intended matches. +``task_message`` stays at default priority and relies on registration order +(before ``task_retrieval`` / ``task_completed`` / ``task_async``) because its +predicate is broad (``task_id`` or ``message``) and must not beat earlier shapes. -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 +25,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 +37,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 +62,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) @@ -170,7 +189,8 @@ def _tool_result_build_task_retrieval( tr: ToolResultDict, base: dict[str, object] ) -> dict[str, object]: result = dict(base) - task_obj = tr["task"] if isinstance(tr["task"], dict) else {} + raw_task = tr.get("task") + task_obj = raw_task if isinstance(raw_task, dict) else {} result["result_type"] = "task" result["retrieval_status"] = tr.get("retrieval_status") result["task_id"] = task_obj.get("task_id") @@ -216,26 +236,56 @@ 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, + ), + 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), ) + +def _validate_dispatch_ids( + table: tuple[ToolResultDispatchEntry, ...], +) -> dict[str, int]: + """Fail fast on duplicate entry IDs and return the registration-order index.""" + order: dict[str, int] = {} + for index, entry in enumerate(table): + if entry.id in order: + raise ValueError( + f"duplicate ToolResultDispatchEntry id {entry.id!r} " + f"(indices {order[entry.id]} and {index})" + ) + order[entry.id] = index + return order + + +_DISPATCH_ORDER = _validate_dispatch_ids(_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 +353,42 @@ 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 + order = _validate_dispatch_ids(_TOOL_RESULT_DISPATCH) + return max( + matches, + key=lambda entry: (entry.priority, -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"