diff --git a/loopx/cli_runtime.py b/loopx/cli_runtime.py index 454a8ddf1d..e25a6db73e 100644 --- a/loopx/cli_runtime.py +++ b/loopx/cli_runtime.py @@ -173,12 +173,12 @@ def resolve_cli_registry( registry_was_configured = user_supplied_registry(raw_argv) or bool( os.environ.get("LOOPX_REGISTRY") ) - if args.command == "quota" and getattr(args, "quota_command", None) == "context-read": + if args.command == "quota" and getattr(args, "quota_command", None) in {"context-read", "replan-view"}: from .cli_commands.context_machine import has_explicit_context_read_source - # The reader owns capture integrity and native-session scope. It cannot - # reach Goal authority, so default registry conflicts are unrelated. - if has_explicit_context_read_source(args): + # Captured readers own source validation (including missing references) + # and native-session scope; neither reaches live Goal authority. + if args.quota_command == "replan-view" or has_explicit_context_read_source(args): return registry_path, registry_was_configured if str(args.registry) == HOST_GLOBAL_REGISTRY_SELECTOR: try: diff --git a/tests/control_plane/test_context_capture_registry_independence.py b/tests/control_plane/test_context_capture_registry_independence.py index 5c8c9331a1..3c4776469f 100644 --- a/tests/control_plane/test_context_capture_registry_independence.py +++ b/tests/control_plane/test_context_capture_registry_independence.py @@ -3,12 +3,14 @@ import json import os from pathlib import Path +import shlex import subprocess import sys import pytest from loopx import paths +from loopx.cli_runtime import HOST_GLOBAL_REGISTRY_SELECTOR from test_codex_context_lifecycle import project, setup @@ -93,3 +95,111 @@ def test_implicit_source_and_authority_commands_preserve_route_conflict(tmp_path result = run_cli("loopx.entrypoint", tmp_path, *command) assert result.returncode != 0 assert "Both default LoopX runtime roots contain state" in result.stdout + result.stderr + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_replan_stages_read_original_authority_capture_without_registry( + tmp_path, monkeypatch, conflicting_defaults, provider, +): + from canonical_authority_fixture import isolate_sqlite_runtime + from loopx.control_plane.effect_runtime import restart_effect_runtime + from test_replan_progress_native_cli import ( + AGENT, GOAL, TODO, TURN, call_progress_cli, make_progress_cli_fixture, + ) + + isolate_sqlite_runtime(tmp_path, monkeypatch) + fixture = make_progress_cli_fixture(tmp_path, provider, "advanced") + registry = json.loads(fixture["registry"].read_bytes()) + registry["goals"][0]["explore_graph"] = {"enabled": True} + fixture["registry"].write_text(json.dumps(registry)) + document = tmp_path / "outcome.json" + document.write_text(json.dumps({"outcome_id": "original-artifact", "owner_agent_id": AGENT, + "objective": "Validate the original artifact", "todo_ids": [TODO], + "requirements": [{"id": "retained-stop", "description": "Stop before publication."}], + "source": {"ref": "message:original-request", "summary": "Original accepted boundary"}})) + try: + revision = call_progress_cli(fixture, "goal-acceptance", "inspect", "--goal-id", GOAL)["provider_revision"] + call_progress_cli(fixture, "goal-acceptance", "refine", "--goal-id", GOAL, "--agent-id", AGENT, + "--document", str(document), "--expected-provider-revision", revision, + "--operation-id", "original-outcome", "--execute") + guard = call_progress_cli(fixture, "quota", "should-run", "--codex-app", + "--goal-id", GOAL, "--agent-id", AGENT, "--turn-instance-id", TURN, + "--todo-id", TODO, "--decision-output-dir", str(tmp_path / "capture")) + capture = tmp_path / "capture" / "decision.json" + source = json.loads(capture.read_bytes()) + identity = source["heartbeat_receipt"]["settlement_identity"] + assert (identity["goal_id"], identity["agent_id"], identity["turn_instance_id"]) == (GOAL, AGENT, TURN) + missing = tmp_path / "unrelated-missing-registry.json" + + def snapshot(): + return {str(p.relative_to(tmp_path)): p.read_bytes() + for root in (fixture["project"], fixture["runtime"], capture.parent) + for p in root.rglob("*") if p.is_file()} | {"registry": fixture["registry"].read_bytes()} + + before = snapshot() + for module in ("loopx.entrypoint", "loopx.cli"): + for route in ((), ("--registry", str(missing)), ("--registry", HOST_GLOBAL_REGISTRY_SELECTOR)): + def read(command): + argv = shlex.split(command) + result = run_cli(module, fixture["project"], *route, *argv[argv.index("quota"):]) + assert result.returncode == 0, result.stdout + result.stderr + value = json.loads(result.stdout) + assert value["ok"] and (value["goal_id"], value["agent_id"]) == (GOAL, AGENT) + assert value["obligation_id"] == source["replan_action_packet"]["obligation_id"] + return value + + review = read(guard["replan_action_packet"]["review"]) + writer = read(review["next"]["writeback"]) + assert writer["context_id"] == review["context_id"] + assert writer["source_hash"] == review["source_hash"] + assert writer["writeback"]["settlement_plan"]["identity"] == identity + index = review["acceptance_index"] + assert len(index["outcomes"]) == 1 + acceptance = read(index["read_action_template"].replace("{outcome_ref}", index["outcomes"][0]["outcome_ref"])) + assert acceptance["outcome"]["requirements"][0]["description"] == "Stop before publication." + explore = read(guard["replan_action_packet"]["review"].replace("--replan-stage review", "--replan-stage explore")) + assert explore["explore_basis"] == source["autonomous_replan_obligation"]["replan_context"].get("explore_basis") + assert snapshot() == before, "captured readers cannot admit, settle, repair routes or rewrite authority" + assert not missing.exists() + finally: + restart_effect_runtime(wait_for_exit=True) + + +@pytest.mark.parametrize("module", ["loopx.entrypoint", "loopx.cli"]) +def test_replan_capture_failures_belong_to_reader_not_default_route(tmp_path, conflicting_defaults, module): + from loopx.control_plane.quota.turn_envelope import _turn_envelope_result + from test_replan_host_context_projection import _quota_payload + + capture = tmp_path / "source.json" + source = _quota_payload() + source["replan_read_path"] = str(capture) + + def command(value, stage="review"): + capture.write_text(json.dumps(value)) + packet = _turn_envelope_result("replan_cli_view", payload=value)["replan_action_packet"] + argv = shlex.split(packet[stage]) + return argv[argv.index("quota"):] + + def rejected(argv, reason): + result = run_cli(module, tmp_path, *argv) + assert result.returncode == 1, result.stdout + result.stderr + value = json.loads(result.stdout) + assert not value["ok"] and reason in value["reason"], value + + argv = command(source) + rejected(["quota", "replan-view"], "requires --decision-file and --source-hash") + rejected(["quota", "replan-view", "--decision-file", str(capture)], "requires --decision-file and --source-hash") + rejected(["quota", "replan-view", "--source-hash", "sha256:" + "0" * 64], "requires --decision-file and --source-hash") + wrong = list(argv) + wrong[wrong.index("--source-hash") + 1] = "sha256:" + "0" * 64 + rejected(wrong, "source hash") + capture.write_text("{") + rejected(argv, "Expecting") + capture.unlink() + rejected(argv, "No such file") + source["autonomous_replan_obligation"]["replan_context"]["obligation_id"] = "different-obligation" + rejected(command(source), "matching decision context") + source = _quota_payload() + source["replan_read_path"] = str(capture) + source["interaction_contract"]["schema_version"] = "unsupported-contract-version" + rejected(command(source, "writeback"), "schema_version")