fix: validate plan reviewer rotation chains

Co-authored-by: Ouroboros <311266734+ouroboros-agent@users.noreply.github.com>

ouroboros/review_thread_continuity.py

@@ -24,10 +24,9 @@ def ensure_review_thread(
         "primaryHarness": route.route_id, "eligibleHarnesses": [route.route_id],
         "access": "readonly",
     }
-    # A thread turn must never fall back to the daemon's sticky/pool precedence
-    # by omission. ``None`` is the engine's typed default-subject pin; a string
-    # is the exact expected profile.
-    request["credentialProfileId"] = getattr(route, "profile_id", "") or None
+    profile_id = str(getattr(route, "profile_id", "") or "")
+    if profile_id:
+        request["credentialProfileId"] = profile_id
     return str(gateway.create_thread(request, idempotency_key=key).get("id") or "")
 
 
@@ -62,6 +61,7 @@ def profile_rotation_receipts(gateway: Any, run_id: str) -> list[dict]:
             else row
         )
         receipts.append({
+            "seq": row.get("seq"),
             "type": "route.profile.rotated",
             "from_profile_id": payload.get("from_profile_id"),
             "to_profile_id": payload.get("to_profile_id"),
@@ -75,22 +75,44 @@ def profile_rotation_receipts(gateway: Any, run_id: str) -> list[dict]:
 def profile_continuity_receipt(
     expected_profile: str, applied_profile: str, rotations: list[dict],
 ) -> dict:
-    """Bind expected and applied profiles, accepting only an engine rotation receipt."""
+    """Bind profiles through one complete ordered engine-owned rotation chain."""
     expected, applied = str(expected_profile or ""), str(applied_profile or "")
-    matched = next((
-        dict(row) for row in rotations if isinstance(row, dict)
-        and row.get("type") == "route.profile.rotated"
-        and str(row.get("from_profile_id") or "") == expected
-        and str(row.get("to_profile_id") or "") == applied
-        and str(row.get("reason") or "")
-    ), None)
-    status = "matched" if expected == applied else ("typed_rotation" if matched else "cannot_verify")
-    return {
-        "expected_profile": expected,
-        "applied_profile": applied,
-        "status": status,
-        "rotation_receipt": matched or {},
-    }
+    chain = [dict(row) for row in rotations if isinstance(row, dict)]
+
+    def result(status: str, reason: str, receipt: dict | None = None) -> dict:
+        return {
+            "expected_profile": expected,
+            "applied_profile": applied,
+            "status": status,
+            "verification_reason": reason,
+            "rotation_receipt": dict(receipt or {}),
+            "rotation_receipts": chain,
+        }
+
+    if not rotations:
+        return result(
+            "matched" if expected == applied else "cannot_verify",
+            "profile_matched" if expected == applied else "unexplained_profile_drift",
+        )
+    if len(chain) != len(rotations):
+        return result("cannot_verify", "rotation_event_malformed")
+    current, previous_seq = expected, None
+    for row in chain:
+        seq = row.get("seq")
+        source = str(row.get("from_profile_id") or "")
+        target = str(row.get("to_profile_id") or "")
+        if (row.get("type") != "route.profile.rotated"
+                or not isinstance(seq, int) or isinstance(seq, bool)
+                or not target or target == source or not str(row.get("reason") or "")):
+            return result("cannot_verify", "rotation_event_malformed")
+        if previous_seq is not None and seq <= previous_seq:
+            return result("cannot_verify", "rotation_event_order_invalid")
+        if source != current:
+            return result("cannot_verify", "rotation_chain_gap")
+        current, previous_seq = target, seq
+    if current != applied:
+        return result("cannot_verify", "rotation_terminal_mismatch")
+    return result("typed_rotation", "typed_rotation_chain", chain[-1])
 
 
 def review_thread_receipt(

tests/test_phase4_plan_review_continuity.py

@@ -360,6 +360,58 @@ def test_continued_thread_explicitly_repins_the_expected_profile() -> None:
     assert captured["request"]["credentialProfileId"] == "profile-a"
 
 
+def test_initial_unpinned_thread_omits_profile_but_its_turn_sends_explicit_null() -> None:
+    from types import SimpleNamespace
+
+    from ouroboros.gateways.claudexor import ClaudexorGateway, DaemonEndpoint
+    from ouroboros.review_thread_continuity import ensure_review_thread, start_review_thread_turn
+
+    captured = {}
+
+    class Custody:
+        @staticmethod
+        def idempotency_key(*parts):
+            return ":".join(str(part) for part in parts)
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        body = json.loads(request.content.decode("utf-8"))
+        if request.url.path == "/v2/threads":
+            # Installed ControlThreadCreateRequest: nonblank string or omission.
+            assert "credentialProfileId" not in body
+            captured["create"] = body
+            return httpx.Response(200, json={"id": "thread-1"})
+        if request.url.path == "/v2/threads/thread-1/turns":
+            # Installed ControlThreadTurnRequest: explicit null clears a sticky pin.
+            assert "credentialProfileId" in body and body["credentialProfileId"] is None
+            captured["turn"] = body
+            return httpx.Response(200, json={
+                "threadId": "thread-1", "turnId": "turn-1", "runId": "run-1",
+            })
+        return httpx.Response(404, json={"code": "not_found", "message": "no"})
+
+    gateway = ClaudexorGateway(DaemonEndpoint(host="127.0.0.1", port=1, token="token"))
+    gateway._client.close()
+    gateway._client = httpx.Client(
+        base_url="http://127.0.0.1:1", transport=httpx.MockTransport(handler),
+        headers={"Authorization": "Bearer token"},
+    )
+    try:
+        thread_id = ensure_review_thread(
+            gateway, Custody(), "", route=SimpleNamespace(route_id="claude", profile_id=""),
+            root="/repo", surface="plan_review", slot_id="s1", task_id="task-1",
+        )
+        start_review_thread_turn(
+            gateway, thread_id,
+            {"prompt": "review", "credentialProfileId": None, "_thread_id": thread_id},
+            idempotency_key="turn-key",
+        )
+    finally:
+        gateway.close()
+
+    assert thread_id == "thread-1"
+    assert captured["turn"]["credentialProfileId"] is None
+
+
 def test_profile_rotation_receipt_is_read_from_the_settled_run_events() -> None:
     from ouroboros.review_thread_continuity import profile_rotation_receipts
 
@@ -376,6 +428,7 @@ def test_profile_rotation_receipt_is_read_from_the_settled_run_events() -> None:
             }) + "\n").encode()
 
     assert profile_rotation_receipts(Gateway(), "run-2") == [{
+        "seq": 7,
         "type": "route.profile.rotated",
         "from_profile_id": "profile-a",
         "to_profile_id": "profile-b",
@@ -385,6 +438,64 @@ def test_profile_rotation_receipt_is_read_from_the_settled_run_events() -> None:
     }]
 
 
+def _rotation(seq, source, target, *, reason="vendor_limit_rejected"):
+    return {
+        "seq": seq, "type": "route.profile.rotated",
+        "from_profile_id": source, "to_profile_id": target,
+        "reason": reason, "attempt_id": f"attempt-{seq}", "resets_at": "later",
+    }
+
+
+@pytest.mark.parametrize(
+    ("applied", "rotations", "expected_status", "expected_reason"),
+    [
+        (
+            "profile-b", [_rotation(10, "profile-a", "profile-b")],
+            "typed_rotation", "typed_rotation_chain",
+        ),
+        (
+            "profile-c",
+            [_rotation(10, "profile-a", "profile-b"), _rotation(20, "profile-b", "profile-c")],
+            "typed_rotation", "typed_rotation_chain",
+        ),
+        (
+            "profile-c",
+            [_rotation(10, "profile-a", "profile-b"), _rotation(20, "profile-x", "profile-c")],
+            "cannot_verify", "rotation_chain_gap",
+        ),
+        (
+            "profile-c", [_rotation(10, "profile-a", "profile-b")],
+            "cannot_verify", "rotation_terminal_mismatch",
+        ),
+        (
+            "profile-c",
+            [_rotation(20, "profile-a", "profile-b"), _rotation(10, "profile-b", "profile-c")],
+            "cannot_verify", "rotation_event_order_invalid",
+        ),
+        (
+            "profile-b",
+            [{**_rotation(10, "profile-a", "profile-b"), "reason": ""}],
+            "cannot_verify", "rotation_event_malformed",
+        ),
+    ],
+    ids=("one-hop", "multi-hop", "broken-gap", "terminal-mismatch", "reordered", "malformed"),
+)
+def test_profile_continuity_folds_only_one_ordered_engine_rotation_chain(
+    applied, rotations, expected_status, expected_reason,
+) -> None:
+    from ouroboros.review_thread_continuity import profile_continuity_receipt
+
+    receipt = profile_continuity_receipt("profile-a", applied, rotations)
+
+    assert receipt["status"] == expected_status
+    assert receipt["verification_reason"] == expected_reason
+    if expected_status == "typed_rotation":
+        assert receipt["rotation_receipts"] == rotations
+        assert receipt["rotation_receipt"] == rotations[-1]
+    else:
+        assert receipt["rotation_receipt"] == {}
+
+
 def test_agent_session_continuation_passes_the_real_thread_id(monkeypatch, tmp_path) -> None:
     from ouroboros.review_execution import AgentSessionReviewExecutor, ReviewAssignment
     from ouroboros.review_substrate import ReviewRequest, ReviewSlot