From 3c66bc57e6e588bb57eb299a62facfedc50bd195 Mon Sep 17 00:00:00 2001 From: "duanjialing.777" Date: Mon, 28 Sep 2026 05:15:18 +0800 Subject: [PATCH 1/2] test(goal): add Instance-qualified continuity end-to-end coverage Verify that Goal recreation retires the old instance identity, rejects stale session bindings/unbindings, and allows new bindings through the real source-session lifecycle entrypoints. 10 test cases cover: - Old instance resolution as "retired" after recreation - New instance auto-generation and "current" resolution - Stale session bind/unbind rejection with stale_goal_instance error - New session binding success and current resolution - Old session ids recorded in retired_session_ids - Recreation idempotency (replayed=true on replay) - Registration idempotency (replayed=true on replay) - Session binding idempotency - operation_id conflict detection for different sessions Design owner: source-session lifetime (source_session_lifetime.ts). Qualification: goal-immutability-coherence-defense-v0.md. Signed-off-by: Duang777 Signed-off-by: duanjialing.777 --- ...test_goal_instance_qualified_continuity.py | 444 ++++++++++++++++++ 1 file changed, 444 insertions(+) create mode 100644 tests/control_plane/test_goal_instance_qualified_continuity.py diff --git a/tests/control_plane/test_goal_instance_qualified_continuity.py b/tests/control_plane/test_goal_instance_qualified_continuity.py new file mode 100644 index 000000000..668df467c --- /dev/null +++ b/tests/control_plane/test_goal_instance_qualified_continuity.py @@ -0,0 +1,444 @@ +"""Instance-qualified continuity: Goal recreation enforces exact instance identity. + +Retire Goal A through the authorized lifecycle, create same-name Goal B, then +deliver A's delayed session binding / claim through the real entrypoints. +Verify no mutation or execution authority leaks into B. + +Design owner: source-session lifetime (source_session_lifetime.ts). +Qualification: goal-immutability-coherence-defense-v0.md, Instance-qualified +continuity slice. +""" + +from __future__ import annotations + +from pathlib import Path +from uuid import uuid4 + +import pytest + +from loopx.control_plane.effect_runtime import EffectRuntimeRejected +from loopx.control_plane.goals.source_session_binding import ( + SessionBindingRequest, + commit_project_session_binding, + commit_project_session_unbinding, + resolve_source_session_project, +) +from loopx.control_plane.goals.source_session_recreation import ( + RecreateGoalRequest, + recreate_goal_instance, +) +from loopx.control_plane.goals.source_session_registration import ( + FreshSourceSessionRegistration, + register_fresh_source_session_project, +) +from loopx.control_plane.projects.registry_codec import ( + load_project_registry, +) + + +def _fresh_registration(tmp_path: Path, goal_id: str) -> FreshSourceSessionRegistration: + root = tmp_path / goal_id + root.mkdir(parents=True, exist_ok=True) + runtime_root = root / "runtime" + return FreshSourceSessionRegistration( + registry_path=root / ".loopx" / "registry.json", + runtime_root=runtime_root, + operation_id=uuid4().hex, + project_id=f"proj-{goal_id}", + goal_id=goal_id, + objective="Verify instance-qualified continuity after recreation.", + non_goals=["No real effects."], + acceptance=["Old bindings are retired.", "New bindings succeed."], + unknowns=[], + next_effect="noop", + stop_condition="test completes", + project_record={"project_id": f"proj-{goal_id}", "display_name": goal_id}, + goal_record={ + "id": goal_id, + "display_name": goal_id, + "status": "active", + "project_id": f"proj-{goal_id}", + "quota": {"compute": 1, "allowed_slots": 1, "spent_slots": 0}, + }, + state_file=root / "GOAL_STATE.md", + ) + + +def _registration_and_ids( + tmp_path: Path, + goal_id: str, +) -> tuple[dict, str, str]: + """Register a fresh source-session Goal and return (result, goal_id, goal_instance_id).""" + registration = _fresh_registration(tmp_path, goal_id) + result = register_fresh_source_session_project(registration) + assert result["ok"] is True + goal_ref = result["goal_ref"] + return result, goal_ref["goal_id"], goal_ref["goal_instance_id"] + + +def _load_registry(result: dict) -> dict: + """Load the registry dict from the path recorded in a service result.""" + return load_project_registry(Path(result["registry"])) + + +# --------------------------------------------------------------------------- +# Core instance-qualified continuity +# --------------------------------------------------------------------------- + + +def test_recreation_retires_old_instance_and_allows_new(tmp_path: Path): + """After recreation, old goal_instance_id is retired; new one is current.""" + goal_id = "continuity-goal" + result, gid, old_instance = _registration_and_ids(tmp_path, goal_id) + + # goal_instance_id in the request is the CURRENT instance (CAS check); + # the new instance is auto-generated. + rec_req = RecreateGoalRequest( + registry_path=Path(result["registry"]), + goal_id=gid, + goal_instance_id=old_instance, + operation_id=uuid4().hex, + ) + recreation = recreate_goal_instance(rec_req) + assert recreation["ok"] is True + assert recreation["retired_goal_ref"]["goal_instance_id"] == old_instance + new_instance = recreation["goal_ref"]["goal_instance_id"] + assert new_instance != old_instance + + # Old instance resolves as retired. + registry = _load_registry(recreation) + old_resolution = resolve_source_session_project( + registry=registry, + registry_path=Path(result["registry"]), + goal_id=gid, + goal_instance_id=old_instance, + session_id=None, + ) + assert old_resolution["resolution"] == "retired", ( + f"old instance {old_instance} must resolve as retired, got {old_resolution['resolution']}" + ) + + # New instance resolves as current. + new_resolution = resolve_source_session_project( + registry=registry, + registry_path=Path(result["registry"]), + goal_id=gid, + goal_instance_id=new_instance, + session_id=None, + ) + assert new_resolution["resolution"] == "current", ( + f"new instance {new_instance} must resolve as current, got {new_resolution['resolution']}" + ) + + +def test_stale_session_bind_rejected_after_recreation(tmp_path: Path): + """A session bound to the old instance cannot rebind after recreation.""" + goal_id = "stale-bind-goal" + result, gid, old_instance = _registration_and_ids(tmp_path, goal_id) + + # Bind session S1 to old instance. + bind_req = SessionBindingRequest( + registry_path=Path(result["registry"]), + session_id="s1", + goal_id=gid, + goal_instance_id=old_instance, + operation_id=uuid4().hex, + ) + bind_result = commit_project_session_binding(bind_req) + assert bind_result["ok"] is True + assert bind_result["changed"] is True + + # Recreate Goal to new instance. + rec_req = RecreateGoalRequest( + registry_path=Path(result["registry"]), + goal_id=gid, + goal_instance_id=old_instance, + operation_id=uuid4().hex, + ) + recreation = recreate_goal_instance(rec_req) + assert recreation["ok"] is True + + # Attempting to bind S1 to old instance must fail. + with pytest.raises(ValueError, match="stale_goal_instance"): + commit_project_session_binding( + SessionBindingRequest( + registry_path=Path(result["registry"]), + session_id="s1", + goal_id=gid, + goal_instance_id=old_instance, + operation_id=uuid4().hex, + ) + ) + + +def test_new_session_can_bind_after_recreation(tmp_path: Path): + """After recreation, a new session can bind to the new instance.""" + goal_id = "new-bind-goal" + result, gid, old_instance = _registration_and_ids(tmp_path, goal_id) + + # Bind session S1 to old instance. + bind_req = SessionBindingRequest( + registry_path=Path(result["registry"]), + session_id="s1", + goal_id=gid, + goal_instance_id=old_instance, + operation_id=uuid4().hex, + ) + assert commit_project_session_binding(bind_req)["ok"] is True + + # Recreate. + rec_req = RecreateGoalRequest( + registry_path=Path(result["registry"]), + goal_id=gid, + goal_instance_id=old_instance, + operation_id=uuid4().hex, + ) + recreation = recreate_goal_instance(rec_req) + assert recreation["ok"] is True + new_instance = recreation["goal_ref"]["goal_instance_id"] + + # New session S2 binds to new instance successfully. + new_bind = commit_project_session_binding( + SessionBindingRequest( + registry_path=Path(result["registry"]), + session_id="s2", + goal_id=gid, + goal_instance_id=new_instance, + operation_id=uuid4().hex, + ) + ) + assert new_bind["ok"] is True + assert new_bind["changed"] is True + + # Verify S2 resolves as current. + registry = _load_registry(new_bind) + resolution = resolve_source_session_project( + registry=registry, + registry_path=Path(result["registry"]), + goal_id=gid, + goal_instance_id=new_instance, + session_id="s2", + ) + assert resolution["resolution"] == "current" + + +def test_recreation_retires_old_session_bindings(tmp_path: Path): + """Recreation removes session bindings tied to the old instance.""" + goal_id = "retire-bindings-goal" + result, gid, old_instance = _registration_and_ids(tmp_path, goal_id) + + # Bind two sessions to old instance. + for sid in ("s1", "s2"): + assert commit_project_session_binding( + SessionBindingRequest( + registry_path=Path(result["registry"]), + session_id=sid, + goal_id=gid, + goal_instance_id=old_instance, + operation_id=uuid4().hex, + ) + )["ok"] is True + + # Recreate. + rec_req = RecreateGoalRequest( + registry_path=Path(result["registry"]), + goal_id=gid, + goal_instance_id=old_instance, + operation_id=uuid4().hex, + ) + recreation = recreate_goal_instance(rec_req) + assert recreation["ok"] is True + new_instance = recreation["goal_ref"]["goal_instance_id"] + + # Both old session IDs are recorded as retired. + assert set(recreation["retired_session_ids"]) == {"s1", "s2"} + + # Neither old session can resolve as current with new instance. + registry = _load_registry(recreation) + for sid in ("s1", "s2"): + resolution = resolve_source_session_project( + registry=registry, + registry_path=Path(result["registry"]), + goal_id=gid, + goal_instance_id=new_instance, + session_id=sid, + ) + assert resolution["resolution"] != "current", ( + f"retired session {sid} must not resolve as current" + ) + + +# --------------------------------------------------------------------------- +# Unbind continuity +# --------------------------------------------------------------------------- + + +def test_stale_unbind_rejected_after_recreation(tmp_path: Path): + """An old-instance unbind is rejected after recreation.""" + goal_id = "stale-unbind-goal" + result, gid, old_instance = _registration_and_ids(tmp_path, goal_id) + + # Bind session S1. + assert commit_project_session_binding( + SessionBindingRequest( + registry_path=Path(result["registry"]), + session_id="s1", + goal_id=gid, + goal_instance_id=old_instance, + operation_id=uuid4().hex, + ) + )["ok"] is True + + # Recreate. + rec_req = RecreateGoalRequest( + registry_path=Path(result["registry"]), + goal_id=gid, + goal_instance_id=old_instance, + operation_id=uuid4().hex, + ) + assert recreate_goal_instance(rec_req)["ok"] is True + + # Unbinding S1 with old instance must be rejected. + with pytest.raises(ValueError, match="stale_goal_instance"): + commit_project_session_unbinding( + SessionBindingRequest( + registry_path=Path(result["registry"]), + session_id="s1", + goal_id=gid, + goal_instance_id=old_instance, + operation_id=uuid4().hex, + ) + ) + + +# --------------------------------------------------------------------------- +# Idempotency +# --------------------------------------------------------------------------- + + +def test_recreation_is_idempotent(tmp_path: Path): + """Replaying the same recreation returns the same receipt.""" + goal_id = "idempotent-goal" + result, gid, old_instance = _registration_and_ids(tmp_path, goal_id) + + operation_id = uuid4().hex + rec_req = RecreateGoalRequest( + registry_path=Path(result["registry"]), + goal_id=gid, + goal_instance_id=old_instance, + operation_id=operation_id, + ) + + first = recreate_goal_instance(rec_req) + assert first["ok"] is True + assert not first["replayed"] + + # Replay with same operation_id returns replayed result. + second = recreate_goal_instance(rec_req) + assert second["ok"] is True + assert second["replayed"] is True + assert second["goal_ref"] == first["goal_ref"] + assert second["retired_goal_ref"] == first["retired_goal_ref"] + + +def test_recreation_rejects_stale_instance_after_recreation(tmp_path: Path): + """After recreation, the old instance is retired; further recreation with it must fail.""" + goal_id = "stale-recreation-goal" + result, gid, old_instance = _registration_and_ids(tmp_path, goal_id) + + operation_id = uuid4().hex + rec_a = RecreateGoalRequest( + registry_path=Path(result["registry"]), + goal_id=gid, + goal_instance_id=old_instance, + operation_id=operation_id, + ) + assert recreate_goal_instance(rec_a)["ok"] is True + + # Trying to recreate again with the same old instance (now retired) must fail. + with pytest.raises(ValueError, match="stale_goal_instance"): + recreate_goal_instance( + RecreateGoalRequest( + registry_path=Path(result["registry"]), + goal_id=gid, + goal_instance_id=old_instance, + operation_id=uuid4().hex, + ) + ) + + +# --------------------------------------------------------------------------- +# Registration is idempotent +# --------------------------------------------------------------------------- + + +def test_registration_is_idempotent(tmp_path: Path): + """Re-registering the same fresh project replays, does not recreate.""" + goal_id = "registration-idempotent" + registration = _fresh_registration(tmp_path, goal_id) + + first = register_fresh_source_session_project(registration) + assert first["ok"] is True + assert first["changed"] is True + assert not first["replayed"] + + second = register_fresh_source_session_project(registration) + assert second["ok"] is True + assert not second["changed"] + assert second["replayed"] is True + assert second["goal_ref"] == first["goal_ref"] + + +# --------------------------------------------------------------------------- +# Session binding idempotency +# --------------------------------------------------------------------------- + + +def test_session_binding_is_idempotent(tmp_path: Path): + """Rebinding the same session with the same operation_id replays.""" + goal_id = "bind-idempotent" + result, gid, instance = _registration_and_ids(tmp_path, goal_id) + + req = SessionBindingRequest( + registry_path=Path(result["registry"]), + session_id="s1", + goal_id=gid, + goal_instance_id=instance, + operation_id=uuid4().hex, + ) + + first = commit_project_session_binding(req) + assert first["ok"] is True + assert first["changed"] is True + assert not first["replayed"] + + second = commit_project_session_binding(req) + assert second["ok"] is True + assert second["replayed"] is True + # changed mirrors the original receipt; the key marker is replayed=True + + +def test_session_binding_rejects_operation_id_with_different_session(tmp_path: Path): + """Reusing operation_id with different session_id must conflict.""" + goal_id = "bind-conflict" + result, gid, instance = _registration_and_ids(tmp_path, goal_id) + + operation_id = uuid4().hex + req_s1 = SessionBindingRequest( + registry_path=Path(result["registry"]), + session_id="s1", + goal_id=gid, + goal_instance_id=instance, + operation_id=operation_id, + ) + assert commit_project_session_binding(req_s1)["ok"] is True + + req_s2 = SessionBindingRequest( + registry_path=Path(result["registry"]), + session_id="s2", + goal_id=gid, + goal_instance_id=instance, + operation_id=operation_id, + ) + with pytest.raises((ValueError, EffectRuntimeRejected)): + commit_project_session_binding(req_s2) \ No newline at end of file From 7e2b8f5de6f0cbc11c692b6256b3c2c323773d23 Mon Sep 17 00:00:00 2001 From: "duanjialing.777" Date: Mon, 28 Sep 2026 09:40:23 +0800 Subject: [PATCH 2/2] test(goal): consolidate instance continuity coverage Address the exact-head review by removing the duplicate 444-line suite and extending the existing CLI lifetime owner. The focused path now proves multiple binding retirement, byte-for-byte successor invariance after stale bind and unbind attempts, absence of stale receipts, and a valid successor bind after both denials. Signed-off-by: duanjialing.777 --- .../test_source_session_lifetime.py | 53 ++- ...test_goal_instance_qualified_continuity.py | 444 ------------------ 2 files changed, 51 insertions(+), 446 deletions(-) delete mode 100644 tests/control_plane/test_goal_instance_qualified_continuity.py diff --git a/tests/cli_commands/test_source_session_lifetime.py b/tests/cli_commands/test_source_session_lifetime.py index 5817e8a3a..7fdaf593d 100644 --- a/tests/cli_commands/test_source_session_lifetime.py +++ b/tests/cli_commands/test_source_session_lifetime.py @@ -594,7 +594,7 @@ def test_bind_and_unbind_commit_exact_receipts( ] -def test_recreation_retires_bindings_and_fences_stale_bind( +def test_recreation_retires_bindings_fences_stale_operations_and_allows_successor( tmp_path: Path, capsys: pytest.CaptureFixture[str], ) -> None: @@ -608,6 +608,15 @@ def test_recreation_retires_bindings_and_fences_stale_bind( ) assert main(bind_arguments) == 0 capsys.readouterr() + second_bind_arguments = _binding_arguments( + registry_path, + operation="bind", + goal_instance_id=instance_a, + operation_id="bind-session-b", + session_id="session-b", + ) + assert main(second_bind_arguments) == 0 + capsys.readouterr() recreate_arguments = _recreation_arguments( registry_path, goal_instance_id=instance_a, @@ -633,7 +642,10 @@ def test_recreation_retires_bindings_and_fences_stale_bind( ] assert registry["lifetime_receipts"][-1] == recreated["receipt"] assert registry["session_receipts"][-1]["operation"] == "retire_bindings" - assert registry["session_receipts"][-1]["session_ids"] == ["session-a"] + assert registry["session_receipts"][-1]["session_ids"] == [ + "session-a", + "session-b", + ] before_replay = registry_path.read_bytes() assert main(recreate_arguments) == 0 @@ -669,6 +681,43 @@ def test_recreation_retires_bindings_and_fences_stale_bind( assert "stale_goal_instance" in stale_bind_rejection["error"] assert registry_path.read_bytes() == before_replay + stale_unbind = _binding_arguments( + registry_path, + operation="unbind", + goal_instance_id=instance_a, + operation_id="unbind-after-recreate", + ) + assert main(stale_unbind) == 1 + stale_unbind_rejection = json.loads(capsys.readouterr().out) + assert "stale_goal_instance" in stale_unbind_rejection["error"] + assert registry_path.read_bytes() == before_replay + + successor_goal_before = _registry_payload(registry_path)["goals"][0] + successor_bind = _binding_arguments( + registry_path, + operation="bind", + goal_instance_id=instance_b, + operation_id="bind-successor-session", + session_id="successor-session", + ) + assert main(successor_bind) == 0 + successor_bound = json.loads(capsys.readouterr().out) + registry = _registry_payload(registry_path) + assert successor_bound["changed"] is True + assert registry["goals"][0] == successor_goal_before + assert registry["session_bindings"] == [ + { + "session_id": "successor-session", + "foreground_goal_ref": recreated["goal_ref"], + } + ] + assert registry["session_receipts"][-1] == successor_bound["receipt"] + assert not any( + receipt.get("operation_id") + in {"bind-after-recreate", "unbind-after-recreate"} + for receipt in registry["session_receipts"] + ) + def test_paused_bind_cannot_cross_recreation_aba( tmp_path: Path, diff --git a/tests/control_plane/test_goal_instance_qualified_continuity.py b/tests/control_plane/test_goal_instance_qualified_continuity.py deleted file mode 100644 index 668df467c..000000000 --- a/tests/control_plane/test_goal_instance_qualified_continuity.py +++ /dev/null @@ -1,444 +0,0 @@ -"""Instance-qualified continuity: Goal recreation enforces exact instance identity. - -Retire Goal A through the authorized lifecycle, create same-name Goal B, then -deliver A's delayed session binding / claim through the real entrypoints. -Verify no mutation or execution authority leaks into B. - -Design owner: source-session lifetime (source_session_lifetime.ts). -Qualification: goal-immutability-coherence-defense-v0.md, Instance-qualified -continuity slice. -""" - -from __future__ import annotations - -from pathlib import Path -from uuid import uuid4 - -import pytest - -from loopx.control_plane.effect_runtime import EffectRuntimeRejected -from loopx.control_plane.goals.source_session_binding import ( - SessionBindingRequest, - commit_project_session_binding, - commit_project_session_unbinding, - resolve_source_session_project, -) -from loopx.control_plane.goals.source_session_recreation import ( - RecreateGoalRequest, - recreate_goal_instance, -) -from loopx.control_plane.goals.source_session_registration import ( - FreshSourceSessionRegistration, - register_fresh_source_session_project, -) -from loopx.control_plane.projects.registry_codec import ( - load_project_registry, -) - - -def _fresh_registration(tmp_path: Path, goal_id: str) -> FreshSourceSessionRegistration: - root = tmp_path / goal_id - root.mkdir(parents=True, exist_ok=True) - runtime_root = root / "runtime" - return FreshSourceSessionRegistration( - registry_path=root / ".loopx" / "registry.json", - runtime_root=runtime_root, - operation_id=uuid4().hex, - project_id=f"proj-{goal_id}", - goal_id=goal_id, - objective="Verify instance-qualified continuity after recreation.", - non_goals=["No real effects."], - acceptance=["Old bindings are retired.", "New bindings succeed."], - unknowns=[], - next_effect="noop", - stop_condition="test completes", - project_record={"project_id": f"proj-{goal_id}", "display_name": goal_id}, - goal_record={ - "id": goal_id, - "display_name": goal_id, - "status": "active", - "project_id": f"proj-{goal_id}", - "quota": {"compute": 1, "allowed_slots": 1, "spent_slots": 0}, - }, - state_file=root / "GOAL_STATE.md", - ) - - -def _registration_and_ids( - tmp_path: Path, - goal_id: str, -) -> tuple[dict, str, str]: - """Register a fresh source-session Goal and return (result, goal_id, goal_instance_id).""" - registration = _fresh_registration(tmp_path, goal_id) - result = register_fresh_source_session_project(registration) - assert result["ok"] is True - goal_ref = result["goal_ref"] - return result, goal_ref["goal_id"], goal_ref["goal_instance_id"] - - -def _load_registry(result: dict) -> dict: - """Load the registry dict from the path recorded in a service result.""" - return load_project_registry(Path(result["registry"])) - - -# --------------------------------------------------------------------------- -# Core instance-qualified continuity -# --------------------------------------------------------------------------- - - -def test_recreation_retires_old_instance_and_allows_new(tmp_path: Path): - """After recreation, old goal_instance_id is retired; new one is current.""" - goal_id = "continuity-goal" - result, gid, old_instance = _registration_and_ids(tmp_path, goal_id) - - # goal_instance_id in the request is the CURRENT instance (CAS check); - # the new instance is auto-generated. - rec_req = RecreateGoalRequest( - registry_path=Path(result["registry"]), - goal_id=gid, - goal_instance_id=old_instance, - operation_id=uuid4().hex, - ) - recreation = recreate_goal_instance(rec_req) - assert recreation["ok"] is True - assert recreation["retired_goal_ref"]["goal_instance_id"] == old_instance - new_instance = recreation["goal_ref"]["goal_instance_id"] - assert new_instance != old_instance - - # Old instance resolves as retired. - registry = _load_registry(recreation) - old_resolution = resolve_source_session_project( - registry=registry, - registry_path=Path(result["registry"]), - goal_id=gid, - goal_instance_id=old_instance, - session_id=None, - ) - assert old_resolution["resolution"] == "retired", ( - f"old instance {old_instance} must resolve as retired, got {old_resolution['resolution']}" - ) - - # New instance resolves as current. - new_resolution = resolve_source_session_project( - registry=registry, - registry_path=Path(result["registry"]), - goal_id=gid, - goal_instance_id=new_instance, - session_id=None, - ) - assert new_resolution["resolution"] == "current", ( - f"new instance {new_instance} must resolve as current, got {new_resolution['resolution']}" - ) - - -def test_stale_session_bind_rejected_after_recreation(tmp_path: Path): - """A session bound to the old instance cannot rebind after recreation.""" - goal_id = "stale-bind-goal" - result, gid, old_instance = _registration_and_ids(tmp_path, goal_id) - - # Bind session S1 to old instance. - bind_req = SessionBindingRequest( - registry_path=Path(result["registry"]), - session_id="s1", - goal_id=gid, - goal_instance_id=old_instance, - operation_id=uuid4().hex, - ) - bind_result = commit_project_session_binding(bind_req) - assert bind_result["ok"] is True - assert bind_result["changed"] is True - - # Recreate Goal to new instance. - rec_req = RecreateGoalRequest( - registry_path=Path(result["registry"]), - goal_id=gid, - goal_instance_id=old_instance, - operation_id=uuid4().hex, - ) - recreation = recreate_goal_instance(rec_req) - assert recreation["ok"] is True - - # Attempting to bind S1 to old instance must fail. - with pytest.raises(ValueError, match="stale_goal_instance"): - commit_project_session_binding( - SessionBindingRequest( - registry_path=Path(result["registry"]), - session_id="s1", - goal_id=gid, - goal_instance_id=old_instance, - operation_id=uuid4().hex, - ) - ) - - -def test_new_session_can_bind_after_recreation(tmp_path: Path): - """After recreation, a new session can bind to the new instance.""" - goal_id = "new-bind-goal" - result, gid, old_instance = _registration_and_ids(tmp_path, goal_id) - - # Bind session S1 to old instance. - bind_req = SessionBindingRequest( - registry_path=Path(result["registry"]), - session_id="s1", - goal_id=gid, - goal_instance_id=old_instance, - operation_id=uuid4().hex, - ) - assert commit_project_session_binding(bind_req)["ok"] is True - - # Recreate. - rec_req = RecreateGoalRequest( - registry_path=Path(result["registry"]), - goal_id=gid, - goal_instance_id=old_instance, - operation_id=uuid4().hex, - ) - recreation = recreate_goal_instance(rec_req) - assert recreation["ok"] is True - new_instance = recreation["goal_ref"]["goal_instance_id"] - - # New session S2 binds to new instance successfully. - new_bind = commit_project_session_binding( - SessionBindingRequest( - registry_path=Path(result["registry"]), - session_id="s2", - goal_id=gid, - goal_instance_id=new_instance, - operation_id=uuid4().hex, - ) - ) - assert new_bind["ok"] is True - assert new_bind["changed"] is True - - # Verify S2 resolves as current. - registry = _load_registry(new_bind) - resolution = resolve_source_session_project( - registry=registry, - registry_path=Path(result["registry"]), - goal_id=gid, - goal_instance_id=new_instance, - session_id="s2", - ) - assert resolution["resolution"] == "current" - - -def test_recreation_retires_old_session_bindings(tmp_path: Path): - """Recreation removes session bindings tied to the old instance.""" - goal_id = "retire-bindings-goal" - result, gid, old_instance = _registration_and_ids(tmp_path, goal_id) - - # Bind two sessions to old instance. - for sid in ("s1", "s2"): - assert commit_project_session_binding( - SessionBindingRequest( - registry_path=Path(result["registry"]), - session_id=sid, - goal_id=gid, - goal_instance_id=old_instance, - operation_id=uuid4().hex, - ) - )["ok"] is True - - # Recreate. - rec_req = RecreateGoalRequest( - registry_path=Path(result["registry"]), - goal_id=gid, - goal_instance_id=old_instance, - operation_id=uuid4().hex, - ) - recreation = recreate_goal_instance(rec_req) - assert recreation["ok"] is True - new_instance = recreation["goal_ref"]["goal_instance_id"] - - # Both old session IDs are recorded as retired. - assert set(recreation["retired_session_ids"]) == {"s1", "s2"} - - # Neither old session can resolve as current with new instance. - registry = _load_registry(recreation) - for sid in ("s1", "s2"): - resolution = resolve_source_session_project( - registry=registry, - registry_path=Path(result["registry"]), - goal_id=gid, - goal_instance_id=new_instance, - session_id=sid, - ) - assert resolution["resolution"] != "current", ( - f"retired session {sid} must not resolve as current" - ) - - -# --------------------------------------------------------------------------- -# Unbind continuity -# --------------------------------------------------------------------------- - - -def test_stale_unbind_rejected_after_recreation(tmp_path: Path): - """An old-instance unbind is rejected after recreation.""" - goal_id = "stale-unbind-goal" - result, gid, old_instance = _registration_and_ids(tmp_path, goal_id) - - # Bind session S1. - assert commit_project_session_binding( - SessionBindingRequest( - registry_path=Path(result["registry"]), - session_id="s1", - goal_id=gid, - goal_instance_id=old_instance, - operation_id=uuid4().hex, - ) - )["ok"] is True - - # Recreate. - rec_req = RecreateGoalRequest( - registry_path=Path(result["registry"]), - goal_id=gid, - goal_instance_id=old_instance, - operation_id=uuid4().hex, - ) - assert recreate_goal_instance(rec_req)["ok"] is True - - # Unbinding S1 with old instance must be rejected. - with pytest.raises(ValueError, match="stale_goal_instance"): - commit_project_session_unbinding( - SessionBindingRequest( - registry_path=Path(result["registry"]), - session_id="s1", - goal_id=gid, - goal_instance_id=old_instance, - operation_id=uuid4().hex, - ) - ) - - -# --------------------------------------------------------------------------- -# Idempotency -# --------------------------------------------------------------------------- - - -def test_recreation_is_idempotent(tmp_path: Path): - """Replaying the same recreation returns the same receipt.""" - goal_id = "idempotent-goal" - result, gid, old_instance = _registration_and_ids(tmp_path, goal_id) - - operation_id = uuid4().hex - rec_req = RecreateGoalRequest( - registry_path=Path(result["registry"]), - goal_id=gid, - goal_instance_id=old_instance, - operation_id=operation_id, - ) - - first = recreate_goal_instance(rec_req) - assert first["ok"] is True - assert not first["replayed"] - - # Replay with same operation_id returns replayed result. - second = recreate_goal_instance(rec_req) - assert second["ok"] is True - assert second["replayed"] is True - assert second["goal_ref"] == first["goal_ref"] - assert second["retired_goal_ref"] == first["retired_goal_ref"] - - -def test_recreation_rejects_stale_instance_after_recreation(tmp_path: Path): - """After recreation, the old instance is retired; further recreation with it must fail.""" - goal_id = "stale-recreation-goal" - result, gid, old_instance = _registration_and_ids(tmp_path, goal_id) - - operation_id = uuid4().hex - rec_a = RecreateGoalRequest( - registry_path=Path(result["registry"]), - goal_id=gid, - goal_instance_id=old_instance, - operation_id=operation_id, - ) - assert recreate_goal_instance(rec_a)["ok"] is True - - # Trying to recreate again with the same old instance (now retired) must fail. - with pytest.raises(ValueError, match="stale_goal_instance"): - recreate_goal_instance( - RecreateGoalRequest( - registry_path=Path(result["registry"]), - goal_id=gid, - goal_instance_id=old_instance, - operation_id=uuid4().hex, - ) - ) - - -# --------------------------------------------------------------------------- -# Registration is idempotent -# --------------------------------------------------------------------------- - - -def test_registration_is_idempotent(tmp_path: Path): - """Re-registering the same fresh project replays, does not recreate.""" - goal_id = "registration-idempotent" - registration = _fresh_registration(tmp_path, goal_id) - - first = register_fresh_source_session_project(registration) - assert first["ok"] is True - assert first["changed"] is True - assert not first["replayed"] - - second = register_fresh_source_session_project(registration) - assert second["ok"] is True - assert not second["changed"] - assert second["replayed"] is True - assert second["goal_ref"] == first["goal_ref"] - - -# --------------------------------------------------------------------------- -# Session binding idempotency -# --------------------------------------------------------------------------- - - -def test_session_binding_is_idempotent(tmp_path: Path): - """Rebinding the same session with the same operation_id replays.""" - goal_id = "bind-idempotent" - result, gid, instance = _registration_and_ids(tmp_path, goal_id) - - req = SessionBindingRequest( - registry_path=Path(result["registry"]), - session_id="s1", - goal_id=gid, - goal_instance_id=instance, - operation_id=uuid4().hex, - ) - - first = commit_project_session_binding(req) - assert first["ok"] is True - assert first["changed"] is True - assert not first["replayed"] - - second = commit_project_session_binding(req) - assert second["ok"] is True - assert second["replayed"] is True - # changed mirrors the original receipt; the key marker is replayed=True - - -def test_session_binding_rejects_operation_id_with_different_session(tmp_path: Path): - """Reusing operation_id with different session_id must conflict.""" - goal_id = "bind-conflict" - result, gid, instance = _registration_and_ids(tmp_path, goal_id) - - operation_id = uuid4().hex - req_s1 = SessionBindingRequest( - registry_path=Path(result["registry"]), - session_id="s1", - goal_id=gid, - goal_instance_id=instance, - operation_id=operation_id, - ) - assert commit_project_session_binding(req_s1)["ok"] is True - - req_s2 = SessionBindingRequest( - registry_path=Path(result["registry"]), - session_id="s2", - goal_id=gid, - goal_instance_id=instance, - operation_id=operation_id, - ) - with pytest.raises((ValueError, EffectRuntimeRejected)): - commit_project_session_binding(req_s2) \ No newline at end of file