diff --git a/ouroboros/review_thread_continuity.py b/ouroboros/review_thread_continuity.py index b4d78c417..154499f98 100644 --- a/ouroboros/review_thread_continuity.py +++ b/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( diff --git a/tests/test_phase4_plan_review_continuity.py b/tests/test_phase4_plan_review_continuity.py index 57e8b9e5d..54ea811cc 100644 --- a/tests/test_phase4_plan_review_continuity.py +++ b/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