From 88d2e0f1745c57c31790129bbf5f8695df2893b4 Mon Sep 17 00:00:00 2001
From: YZJF <195568136+YZJF@users.noreply.github.com>
Date: Thu, 17 Sep 2026 17:37:52 +0800
Subject: [PATCH] test(handoff): characterize the #4311 peer-dispatch
 obligations for M2

#4312 implemented a same-Goal dispatch broker for #4311 and was closed as
superseded by the merged capable-manager semantic-handoff RFC (#4330). The
defect is still real, but the accepted direction is one typed Core
collaboration request/assessment/result contract, so the acceptance
obligations #4312 recorded as migration inputs are characterized here against
the shipped manager-context inbox and the shipped claim-scope projection
instead of reintroducing a parallel broker.

Pinned against shipped code:

- O1 one executor-excluded independent_handoff successor reads as selectable
  for a same-Goal peer and as non-selectable for the origin agent, both from
  the same claim-scope projection
- O2 an empty eligible-peer set fails closed on dispatch instead of projecting
  a user gate
- O3 and O5 a repeated dispatch replays one durable entry and never resets a
  decision the receiver already recorded
- O4 delivery, read and decision read back intact from the store after a
  restart
- O6 a conflicting acknowledgement is rejected rather than overwritten
- O7 a dispatch is filed under the recipient scope hash, so another agent in
  the same Goal cannot reach it
- O8 dispatch and successor handoff identities are pure functions of their
  tuples, so a replay is never a redispatch

O9 (canonical Todo claim readback) is already pinned by
test_links_use_core_state_and_do_not_copy_progress and is only recorded in the
ledger to keep the retained set complete.

No production code changes.

Refs #4311, #4312, #4330

Signed-off-by: YZJF <195568136+YZJF@users.noreply.github.com>
---
 ...est_manager_handoff_m2_characterization.py | 404 ++++++++++++++++++
 ...anager_handoff_m2_characterization_v0.json | 106 +++++
 2 files changed, 510 insertions(+)
 create mode 100644 tests/control_plane/test_manager_handoff_m2_characterization.py
 create mode 100644 tests/fixtures/control_plane/manager_handoff_m2_characterization_v0.json

diff --git a/tests/control_plane/test_manager_handoff_m2_characterization.py b/tests/control_plane/test_manager_handoff_m2_characterization.py
new file mode 100644
index 0000000000..017fd881bf
--- /dev/null
+++ b/tests/control_plane/test_manager_handoff_m2_characterization.py
@@ -0,0 +1,404 @@
+"""Characterize the #4311 peer-dispatch obligations retained for the M2 handoff contract.
+
+#4312 implemented a same-Goal dispatch broker and was closed as superseded by the
+merged capable-manager semantic-handoff RFC (#4330). The defect in #4311 remains
+real, so the acceptance obligations #4312 recorded as migration inputs are pinned
+here against the shipped manager-context inbox and the shipped claim-scope
+projection. This is a characterization / compatibility baseline for the M2
+cutover, not an implementation of the new collaboration contract.
+
+The obligation ledger lives in
+``tests/fixtures/control_plane/manager_handoff_m2_characterization_v0.json``;
+every obligation that claims an ``asserted_by`` test must have one.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from loopx.capabilities.manager_context import (
+    _hash,
+    _root,
+    acknowledge,
+    authority,
+    deliver,
+    pending,
+)
+from loopx.capabilities.manager_context.tracking import query, record_read
+from loopx.control_plane.todos.quota_selection import project_quota_planning
+
+FIXTURE_PATH = (
+    Path(__file__).resolve().parents[1]
+    / "fixtures"
+    / "control_plane"
+    / "manager_handoff_m2_characterization_v0.json"
+)
+
+
+def _load() -> dict:
+    return json.loads(FIXTURE_PATH.read_text(encoding="utf-8"))
+
+
+@pytest.fixture
+def ledger() -> dict:
+    return _load()
+
+
+@pytest.fixture
+def scenario(tmp_path, ledger):
+    """A synthetic Goal with a monitor owner, one peer, and a solo control Goal."""
+
+    data = ledger["scenario"]
+    registry = tmp_path / "registry.json"
+    registry.write_text(
+        json.dumps(
+            {
+                "goals": [
+                    {
+                        "id": data["goal_id"],
+                        "repo": str(tmp_path),
+                        "coordination": {
+                            "registered_agents": data["registered_agents"]
+                        },
+                    },
+                    {
+                        "id": data["solo_goal_id"],
+                        "repo": str(tmp_path),
+                        "coordination": {
+                            "registered_agents": data["solo_registered_agents"]
+                        },
+                    },
+                ]
+            }
+        ),
+        encoding="utf-8",
+    )
+    session = {"session_id": "manager-session", "channel_id": "manager"}
+    turn = {
+        "client_turn_id": "heartbeat-one",
+        "origin": "web",
+        "message": "Route the monitor successor to an agent that may execute it.",
+    }
+    return {
+        "root": tmp_path,
+        "registry": registry,
+        "session": session,
+        "turn": turn,
+        "data": data,
+        "peer_request": {
+            "goal_id": data["goal_id"],
+            "agent_id": data["peer_agent"],
+        },
+        "origin_request": {
+            "goal_id": data["goal_id"],
+            "agent_id": data["monitor_owner"],
+        },
+    }
+
+
+def _planning(item: dict, agent_id: str) -> dict:
+    return project_quota_planning(
+        {},
+        all_open_items=[item],
+        source_open_count=1,
+        agent_identity={"agent_id": agent_id, "agent_model": "peer_v1"},
+        filter_user_gate_blocks_agent=False,
+        available_capabilities=None,
+    )
+
+
+def test_obligation_ledger_is_complete_public_safe_and_bound(ledger):
+    assert ledger["schema_version"] == "manager_handoff_m2_characterization_v0"
+    assert ledger["public_safe"] is True and ledger["synthetic_only"] is True
+    for flag in (
+        "contains_credentials",
+        "contains_provider_payloads",
+        "contains_private_locators",
+    ):
+        assert ledger[flag] is False, flag
+    ids = [row["id"] for row in ledger["obligations"]]
+    assert len(ids) == len(set(ids)), "obligation ids must be unique"
+    for row in ledger["obligations"]:
+        assert row["retained_from"], row["id"]
+        assert row["invariant"], row["id"]
+        covered = row.get("asserted_by") or row.get("covered_by_existing")
+        assert covered, f"{row['id']} is not bound to any test"
+        if row.get("asserted_by"):
+            assert row["asserted_by"] in globals(), (
+                f"{row['id']} points at missing test {row['asserted_by']}"
+            )
+
+
+def test_executor_excluded_successor_reads_dispatchable_for_a_peer_and_excluded_for_the_origin(
+    scenario,
+):
+    """O1: one successor, two readings, one projection."""
+
+    item = dict(scenario["data"]["successor"])
+    todo_id = item["todo_id"]
+    origin = _planning(item, scenario["data"]["monitor_owner"])
+    peer = _planning(item, scenario["data"]["peer_agent"])
+
+    assert [row["todo_id"] for row in origin["lanes"]["open_items"]] == []
+    assert origin["lanes"]["claim_scope"]["selectable_open_count"] == 0
+    assert origin["lanes"]["claim_scope"]["executor_excluded_self_count"] == 1
+    assert [
+        row["todo_id"]
+        for row in origin["lanes"]["claim_scope"]["executor_excluded_self_items"]
+    ] == [todo_id]
+
+    assert [row["todo_id"] for row in peer["lanes"]["open_items"]] == [todo_id]
+    assert peer["lanes"]["claim_scope"]["selectable_open_count"] == 1
+    assert peer["lanes"]["claim_scope"]["unclaimed_open_count"] == 1
+    assert peer["lanes"]["claim_scope"]["executor_excluded_self_count"] == 0
+
+
+def test_no_eligible_peer_fails_closed_without_inventing_a_user_gate(scenario):
+    """O2: an empty eligible-peer set is a typed failure, never a user gate."""
+
+    root, registry = scenario["root"], scenario["registry"]
+    session, turn = scenario["session"], scenario["turn"]
+    solo = scenario["data"]["solo_goal_id"]
+    owner = scenario["data"]["monitor_owner"]
+    peer = scenario["data"]["peer_agent"]
+
+    grant = authority(root, registry, session, turn)
+    same_goal = [row for row in grant["targets"] if row["goal_id"] == solo]
+    assert same_goal == [{"goal_id": solo, "agent_id": owner}]
+    assert [row["agent_id"] for row in same_goal if row["agent_id"] != owner] == []
+
+    with pytest.raises(ValueError, match="not authorized or registered"):
+        deliver(
+            root,
+            registry,
+            session=session,
+            turn=turn,
+            request={"goal_id": solo, "agent_id": peer},
+        )
+
+    empty = authority(root, registry, session, turn)
+    assert "user_gate" not in empty and "gate" not in empty
+    assert empty["mode"] == "context_only"
+
+
+def test_duplicate_dispatch_replays_one_durable_entry(scenario):
+    """O3: a repeated heartbeat dispatch replays instead of redispatching."""
+
+    root, registry = scenario["root"], scenario["registry"]
+    request = scenario["peer_request"]
+    first = deliver(
+        root,
+        registry,
+        session=scenario["session"],
+        turn=scenario["turn"],
+        request=request,
+    )
+    second = deliver(
+        root,
+        registry,
+        session=scenario["session"],
+        turn=scenario["turn"],
+        request=request,
+    )
+
+    assert second["request_id"] == first["request_id"]
+    assert first["replayed"] is False and second["replayed"] is True
+    folder = _root(root) / "entries" / _hash(request)
+    assert sorted(path.name for path in folder.glob("*.json")) == [
+        first["request_id"] + ".json"
+    ]
+    assert len(pending(root, request["goal_id"], request["agent_id"])["items"]) == 1
+
+
+def test_dispatched_read_claimed_chain_survives_a_restart(scenario):
+    """O4: delivery, read and decision all read back from the store after restart."""
+
+    root, registry = scenario["root"], scenario["registry"]
+    request = scenario["peer_request"]
+    delivered = deliver(
+        root,
+        registry,
+        session=scenario["session"],
+        turn=scenario["turn"],
+        request=request,
+    )
+    request_id = delivered["request_id"]
+
+    record_read(root, pending(root, request["goal_id"], request["agent_id"])["items"])
+    acknowledge(
+        root,
+        request["goal_id"],
+        request["agent_id"],
+        request_id,
+        "adopt",
+        "take the successor; the monitor transition is material",
+    )
+
+    # A restart re-reads the same runtime root with no in-process state.
+    row = query(root, registry, goal_ids=[request["goal_id"]], owner_scope=True)[
+        "rows"
+    ][0]
+    assert row["request_id"] == request_id
+    assert row["delivery"]["status"] == "delivered"
+    assert row["read"]["status"] == "supplied_to_receiver"
+    assert row["decision"]["status"] == "adopt"
+    assert row["warnings"] == []
+
+
+def test_redispatch_after_decision_keeps_the_recorded_decision(scenario):
+    """O5: a later delivery never overwrites a recorded decision."""
+
+    root, registry = scenario["root"], scenario["registry"]
+    request = scenario["peer_request"]
+    delivered = deliver(
+        root,
+        registry,
+        session=scenario["session"],
+        turn=scenario["turn"],
+        request=request,
+    )
+    request_id = delivered["request_id"]
+    acknowledge(
+        root, request["goal_id"], request["agent_id"], request_id, "adopt", "first"
+    )
+
+    replayed = deliver(
+        root,
+        registry,
+        session=scenario["session"],
+        turn=scenario["turn"],
+        request=request,
+    )
+    assert replayed["request_id"] == request_id and replayed["replayed"] is True
+
+    row = query(root, registry, goal_ids=[request["goal_id"]], owner_scope=True)[
+        "rows"
+    ][0]
+    assert row["decision"]["status"] == "adopt"
+    assert row["decision"]["reason"] == "first"
+
+
+def test_conflicting_acknowledgement_is_rejected_not_overwritten(scenario):
+    """O6: conflicting claim history stays visible instead of becoming latest state."""
+
+    root, registry = scenario["root"], scenario["registry"]
+    request = scenario["peer_request"]
+    request_id = deliver(
+        root,
+        registry,
+        session=scenario["session"],
+        turn=scenario["turn"],
+        request=request,
+    )["request_id"]
+    acknowledge(
+        root, request["goal_id"], request["agent_id"], request_id, "adopt", "first"
+    )
+
+    with pytest.raises(ValueError, match="already recorded"):
+        acknowledge(
+            root,
+            request["goal_id"],
+            request["agent_id"],
+            request_id,
+            "reject",
+            "second",
+        )
+
+    row = query(root, registry, goal_ids=[request["goal_id"]], owner_scope=True)[
+        "rows"
+    ][0]
+    assert row["decision"]["status"] == "adopt"
+
+
+def test_peer_dispatch_is_invisible_to_the_origin_inbox(scenario):
+    """O7: sharing a Goal is not sharing a handoff."""
+
+    root, registry = scenario["root"], scenario["registry"]
+    peer_request = scenario["peer_request"]
+    request_id = deliver(
+        root,
+        registry,
+        session=scenario["session"],
+        turn=scenario["turn"],
+        request=peer_request,
+    )["request_id"]
+
+    assert pending(root, peer_request["goal_id"], peer_request["agent_id"])["items"]
+    assert (
+        pending(
+            root,
+            scenario["origin_request"]["goal_id"],
+            scenario["data"]["monitor_owner"],
+        )["items"]
+        == []
+    )
+    # The entry is filed under the recipient's scope hash, so the origin identity
+    # cannot reach it at all; isolation is structural, not a filtered read.
+    assert not (
+        _root(root)
+        / "entries"
+        / _hash(scenario["origin_request"])
+        / (request_id + ".json")
+    ).exists()
+    with pytest.raises((OSError, ValueError)):
+        acknowledge(
+            root,
+            scenario["origin_request"]["goal_id"],
+            scenario["data"]["monitor_owner"],
+            request_id,
+            "adopt",
+            "wrong identity",
+        )
+
+
+def test_dispatch_and_handoff_identities_are_deterministic(scenario):
+    """O8: tuple-derived identities are stable, so a replay is never a redispatch."""
+
+    root, registry = scenario["root"], scenario["registry"]
+    request = scenario["peer_request"]
+    first = deliver(
+        root,
+        registry,
+        session=scenario["session"],
+        turn=scenario["turn"],
+        request=request,
+    )["request_id"]
+    assert first == _hash(
+        [
+            authority(root, registry, scenario["session"], scenario["turn"])[
+                "source_id"
+            ],
+            request,
+        ]
+    )
+    assert (
+        deliver(
+            root,
+            registry,
+            session=scenario["session"],
+            turn=scenario["turn"],
+            request=request,
+        )["request_id"]
+        == first
+    )
+
+    item = dict(scenario["data"]["successor"])
+    first_note = _planning(item, scenario["data"]["monitor_owner"])["lanes"][
+        "claim_scope"
+    ]["executor_excluded_self_items"][0]["handoff_note"]["handoff_id"]
+    assert (
+        _planning(dict(item), scenario["data"]["monitor_owner"])["lanes"][
+            "claim_scope"
+        ]["executor_excluded_self_items"][0]["handoff_note"]["handoff_id"]
+        == first_note
+    )
+    successor = dict(item, todo_id="todo_1b7d0c4e5a92")
+    assert (
+        _planning(successor, scenario["data"]["monitor_owner"])["lanes"]["claim_scope"][
+            "executor_excluded_self_items"
+        ][0]["handoff_note"]["handoff_id"]
+        != first_note
+    )
diff --git a/tests/fixtures/control_plane/manager_handoff_m2_characterization_v0.json b/tests/fixtures/control_plane/manager_handoff_m2_characterization_v0.json
new file mode 100644
index 0000000000..b7e40771e6
--- /dev/null
+++ b/tests/fixtures/control_plane/manager_handoff_m2_characterization_v0.json
@@ -0,0 +1,106 @@
+{
+  "schema_version": "manager_handoff_m2_characterization_v0",
+  "title": "Retained peer-dispatch handoff obligations for the capable-manager semantic-handoff M2 contract",
+  "public_safe": true,
+  "synthetic_only": true,
+  "contains_credentials": false,
+  "contains_provider_payloads": false,
+  "contains_private_locators": false,
+  "sources": [
+    "#4311 - dispatch executor-excluded handoffs instead of repolling origin monitor",
+    "#4312 closing note - obligations retained for M2 characterization / compatibility fixtures",
+    "#4330 RFC capable manager and semantic work handoff, section 8 legacy migration table"
+  ],
+  "why_a_fixture": [
+    "#4312 is closed as superseded by the merged capable-manager semantic-handoff RFC (#4330).",
+    "The defect in #4311 remains real, but the accepted direction is one typed Core collaboration",
+    "request/assessment/result contract. The obligations below are the ones #4312 recorded as",
+    "migration inputs; they are characterized here against the shipped manager-context inbox so the",
+    "M2 cutover has a compatibility baseline instead of re-deriving behaviour from a closed branch."
+  ],
+  "scenario": {
+    "goal_id": "research",
+    "solo_goal_id": "research-solo",
+    "monitor_owner": "monitor-origin",
+    "peer_agent": "peer-reviewer",
+    "registered_agents": ["monitor-origin", "peer-reviewer"],
+    "solo_registered_agents": ["monitor-origin"],
+    "successor": {
+      "todo_id": "todo_9f3c1a7b2e04",
+      "text": "[P0] Independently review the monitor transition before delivery.",
+      "task_class": "advancement_task",
+      "continuation_policy": "independent_handoff",
+      "excluded_agents": ["monitor-origin"],
+      "claimed_by": null,
+      "role": "agent",
+      "required_capabilities": ["shell"]
+    }
+  },
+  "obligations": [
+    {
+      "id": "O1",
+      "name": "dispatchable_by_peer_vs_unclaimed_but_executor_excluded",
+      "retained_from": "#4311 acceptance: frontier projection distinguishes the two states",
+      "invariant": "One unclaimed independent_handoff successor that excludes the origin agent reads as selectable for a same-goal peer and as executor-excluded, non-selectable for the origin agent. The two readings come from the same projection, not two sources of truth.",
+      "asserted_by": "test_executor_excluded_successor_reads_dispatchable_for_a_peer_and_excluded_for_the_origin"
+    },
+    {
+      "id": "O2",
+      "name": "no_eligible_peer_is_typed_and_not_a_user_gate",
+      "retained_from": "#4311 acceptance: typed blocker without inventing a user gate",
+      "invariant": "When the Goal registers no agent other than the excluded origin, the derived eligible-peer set is empty and a dispatch attempt fails closed with an authorization error. No user gate, queued placeholder or invented owner is projected.",
+      "asserted_by": "test_no_eligible_peer_fails_closed_without_inventing_a_user_gate"
+    },
+    {
+      "id": "O3",
+      "name": "duplicate_ingress_does_not_redispatch",
+      "retained_from": "#4312: duplicate ingress and deterministic dispatch identity",
+      "invariant": "Repeating the same heartbeat-bound dispatch for the same recipient yields the same request id, reports replayed, and leaves exactly one durable entry. The dispatch identity is derived, never re-emitted.",
+      "asserted_by": "test_duplicate_dispatch_replays_one_durable_entry"
+    },
+    {
+      "id": "O4",
+      "name": "dispatched_read_claimed_survives_restart",
+      "retained_from": "#4312: dispatched -> read -> claimed relation after restart",
+      "invariant": "After the peer reads and acknowledges, a fresh read from the same runtime root still reports the complete delivery/read/decision chain. A restart observes the recorded decision rather than re-offering the same dispatch as pending work.",
+      "asserted_by": "test_dispatched_read_claimed_chain_survives_a_restart"
+    },
+    {
+      "id": "O5",
+      "name": "redispatch_after_claim_does_not_reset_the_decision",
+      "retained_from": "#4312: duplicate ingress reconciliation",
+      "invariant": "A repeated dispatch after the receiver already decided replays the same entry and leaves the recorded decision intact; the decision is never overwritten by a later delivery.",
+      "asserted_by": "test_redispatch_after_decision_keeps_the_recorded_decision"
+    },
+    {
+      "id": "O6",
+      "name": "stale_claim_conflict_is_terminal",
+      "retained_from": "#4312: stale claim conflict",
+      "invariant": "A second, conflicting acknowledgement for one request is rejected instead of silently overwriting the first decision. Conflicting claim history stays observable and never becomes a fabricated latest state.",
+      "asserted_by": "test_conflicting_acknowledgement_is_rejected_not_overwritten"
+    },
+    {
+      "id": "O7",
+      "name": "same_goal_agent_identity_isolation",
+      "retained_from": "#4311 acceptance: same-Goal agent identity isolation",
+      "invariant": "A dispatch addressed to one agent is invisible to another agent's inbox, and an acknowledgement under the wrong identity fails closed. Sharing a Goal is not sharing a handoff.",
+      "asserted_by": "test_peer_dispatch_is_invisible_to_the_origin_inbox"
+    },
+    {
+      "id": "O8",
+      "name": "deterministic_legacy_alias",
+      "retained_from": "#4330 section 8: preserve the tuple-derived dispatch identity as a legacy alias, never redispatch",
+      "invariant": "The dispatch request id is a pure function of the source identity and the exact recipient, and the successor handoff id is a pure function of the successor tuple. Equal inputs give equal ids across independent projections; different inputs give different ids.",
+      "asserted_by": "test_dispatch_and_handoff_identities_are_deterministic"
+    },
+    {
+      "id": "O9",
+      "name": "canonical_todo_claim_readback",
+      "retained_from": "#4312: canonical Todo claim readback before acknowledgement",
+      "invariant": "Linking a context request to a Core Todo requires the Todo to belong to the receiving agent, read back from the canonical Todo authority rather than from a copied projection.",
+      "asserted_by": null,
+      "covered_by_existing": "tests/test_manager_context_tracking.py::test_links_use_core_state_and_do_not_copy_progress",
+      "note": "Already pinned against the shipped tracking seam; re-characterizing it here would duplicate coverage."
+    }
+  ]
+}