fix: task_message dispatch uses order, not priority boost

Author: clean6378-max-it

docs/architecture.md

@@ -75,7 +75,7 @@ In `utils/tool_dispatch.py`, tool results are classified through `_parse_tool_re
 
 When adding a new tool renderer:
 
-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).
+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_tool_dispatch_ordering.py tests/test_tool_dispatch_adversarial.py tests/test_jsonl_parser.py tests/test_real_session_fixtures.py -v`.
 

tests/test_real_session_fixtures.py

@@ -145,10 +145,10 @@ def test_task_completed_with_message_key_matches_task_message_first() -> None:
     """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 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.
+    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,8 +1,7 @@
 """Behavioral adversarial fixtures for ``_TOOL_RESULT_DISPATCH`` predicate overlap.
 
-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`` (highest priority).
+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
@@ -12,45 +11,32 @@
 
 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
-
-# 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",
-}
+from utils.tool_dispatch import _TOOL_RESULT_DISPATCH, _parse_tool_result, _winning_dispatch_entry
 
-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]
 
 
@@ -119,6 +105,19 @@ def test_task_retrieval_narrow_shape_without_task_message_keys() -> None:
     assert result.get("task_id") == "task-narrow"
 
 
+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:
@@ -146,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 priority invariants for ``_TOOL_RESULT_DISPATCH``.
+"""Structural overlap invariants for ``_TOOL_RESULT_DISPATCH``.
 
-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.
+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, ToolResultDispatchEntry
+from utils.tool_dispatch import (
+    _TOOL_RESULT_DISPATCH,
+    ToolResultDispatchEntry,
+    _winning_dispatch_entry,
+)
 
 Predicate = Callable[..., bool]
 
@@ -51,6 +56,34 @@
     "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 _entry_for(predicate: Predicate) -> ToolResultDispatchEntry:
     for entry in _TOOL_RESULT_DISPATCH:
@@ -60,19 +93,27 @@ def _entry_for(predicate: Predicate) -> ToolResultDispatchEntry:
 
 
 @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:
+    blob = OVERLAP_BLOBS[fixture_id]
     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}"
+    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}"
     )

utils/tool_dispatch.py

@@ -2,13 +2,15 @@
 
 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:
+Use ``priority`` when a shape must beat a **later** overlapping entry without
+moving registration order. Documented overlap exception:
 
-- ``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``).
+- ``plan`` (priority 1) over ``file_write`` (0) when both match — plan blobs may
+  carry ``filePath`` + ``content``.
+
+``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.
 
 To add a shape: append a ``ToolResultDispatchEntry`` with predicate, builder,
 and priority. If the new predicate overlaps an existing one, set priority higher
@@ -254,7 +256,6 @@ def _tool_result_build_user_input(tr: ToolResultDict, base: dict[str, object]) -
         "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