diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py index fd61da5e19..d16fec0dc0 100644 --- a/loopx/cli_commands/quota.py +++ b/loopx/cli_commands/quota.py @@ -221,7 +221,7 @@ def _dispatch_quota_turn_start_hooks( runtime_root_arg: str | None, ) -> tuple[dict[str, object] | None, bool]: _export("validate_quota_command_context_request")(args) - if args.quota_command != "should-run": + if args.quota_command != "should-run" or getattr(args, "dry_run", False): return None, False from .lark_inbox import dispatch_goal_lark_turn_start_hooks @@ -351,6 +351,25 @@ def _prepare_quota_result_context( return payload, activation +def _observe_live_quota_result( + args: argparse.Namespace, + payload: Mapping[str, object], + *, + context: QuotaCommandContext | None, + registry_path: Path, + heartbeat_turn_id: str | None, + started_at: int, +) -> None: + preview = args.quota_command == "should-run" and getattr(args, "dry_run", False) + if context is None or preview: + return + observe_quota_result( + args, payload, registry_path=registry_path, runtime_root=context.runtime_root, + turn_id=_effective_spend_turn_instance_id(payload, heartbeat_turn_id=heartbeat_turn_id), + started_at=started_at, + ) + + def _emit_scheduler_observation_result( payload: dict[str, object], args: argparse.Namespace, @@ -362,12 +381,10 @@ def _emit_scheduler_observation_result( print_payload: PrintPayload, ) -> int: """Emit the bounded observer without entering native capture or settlement.""" - if context is not None: - observe_quota_result( - args, payload, registry_path=registry_path, runtime_root=context.runtime_root, - turn_id=_effective_spend_turn_instance_id(payload, heartbeat_turn_id=heartbeat_turn_id), - started_at=usage_quota_started, - ) + _observe_live_quota_result( + args, payload, context=context, registry_path=registry_path, + heartbeat_turn_id=heartbeat_turn_id, started_at=usage_quota_started, + ) succeeded = payload.get("ok") is True print_payload(project_scheduler_observation(payload), args.format, _quota_renderer(args, failure=not succeeded)) @@ -414,12 +431,10 @@ def _emit_quota_result( cli["settlement_plan"], registry_path=registry_path, runtime_root=context.runtime_root if context is not None else None, ) - if context is not None: - observe_quota_result( - args, payload, registry_path=registry_path, runtime_root=context.runtime_root, - turn_id=_effective_spend_turn_instance_id(payload, heartbeat_turn_id=heartbeat_turn_id), - started_at=usage_quota_started, - ) + _observe_live_quota_result( + args, payload, context=context, registry_path=registry_path, + heartbeat_turn_id=heartbeat_turn_id, started_at=usage_quota_started, + ) if budgeted_spend: assert context is not None return _emit_bounded_settlement(payload, args, runtime_root=context.runtime_root, @@ -432,7 +447,7 @@ def _emit_quota_result( if capture_directory is not None and payload.get("replan_action_packet"): payload = {**payload, "replan_read_path": str(capture_directory / "decision.json"), "replan_read_argv": reader_argv} - if args.quota_command == "should-run" and context is not None: + if args.quota_command == "should-run" and not getattr(args, "dry_run", False) and context is not None: # Capture the one liveness observation with the decision that produced # it. Presentation and evidence recovery must not record it again. _export("attach_host_poll_receipt")( @@ -875,7 +890,8 @@ def handle_quota_command( runtime_root_arg=runtime_root_arg, error=exc, ) - if _export("should_log_quota")(args.quota_command, payload): + preview = args.quota_command == "should-run" and getattr(args, "dry_run", False) + if not preview and _export("should_log_quota")(args.quota_command, payload): spend_turn_instance_id = _effective_spend_turn_instance_id( payload, heartbeat_turn_id=heartbeat_turn_id, diff --git a/loopx/cli_commands/quota_context.py b/loopx/cli_commands/quota_context.py index e7b84085f9..c69859b269 100644 --- a/loopx/cli_commands/quota_context.py +++ b/loopx/cli_commands/quota_context.py @@ -363,7 +363,8 @@ def prepare_quota_command_context( agent_lane_id=args.agent_id, include_presentation_projections=False, ) - if bool(getattr(args, "write_projection_cache", False)): + preview = command == "should-run" and getattr(args, "dry_run", False) + if not preview and bool(getattr(args, "write_projection_cache", False)): cache_metadata = write_status_projection_cache( registry_path=registry_path, runtime_root=runtime_root, diff --git a/tests/capabilities/test_pr_wait_observation.py b/tests/capabilities/test_pr_wait_observation.py index 42f48b3590..6a4f642f01 100644 --- a/tests/capabilities/test_pr_wait_observation.py +++ b/tests/capabilities/test_pr_wait_observation.py @@ -169,6 +169,84 @@ def fetch(reference, **kwargs): assert not (project / "host-started").exists() +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_quota_should_run_dry_run_is_read_only(tmp_path, monkeypatch, provider): + from loopx.capabilities.issue_fix import pr_lifecycle + from loopx import usage_goal + from loopx.control_plane.quota.host_poll_receipts import receipt_path_for_goal + from loopx.rollout_event_log import load_rollout_events, rollout_event_log_path + from tests.test_loopx_turn_driver import _write_live_fixture + + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_live_fixture(tmp_path, provider=provider) + config = json.loads(registry.read_text()) + goal = config["goals"][0] + goal["available_capabilities"] = ["external_evidence_poll"] + registry.write_text(json.dumps(config)) + goal_id = goal["id"] + agent_id = "codex-fixture" + state = project / goal["state_file"] + todo_id = add_goal_todo( + registry_path=registry, goal_id=goal_id, role="agent", claimed_by=agent_id, + text="Continue when the exact dependency merges", status="deferred", resume_when="pr_merged:owner/repo#42", + )["todo_id"] + monkeypatch.chdir(project) + calls = [] + usage = [] + + def fetch(reference, **kwargs): + calls.append(reference) + return {"url": "https://github.com/owner/repo/pull/42", "state": "MERGED", + "mergedAt": "2026-10-01T00:00:00Z"} + + monkeypatch.setattr(pr_lifecycle, "fetch_github_pr_lifecycle_payload", fetch) + monkeypatch.setattr(usage_goal, "observe_quota_cycle", lambda **event: usage.append(event["phase"])) + argv = [ + "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "quota", "should-run", "--goal-id", goal_id, "--agent-id", agent_id, + "--codex-app", "--app-automation-current-rrule", "FREQ=MINUTELY;INTERVAL=30", + "--available-capability", "external_evidence_poll", "--context-delivery", "off", + "--record-host-poll", "--write-projection-cache", + ] + events_path = rollout_event_log_path(runtime, goal_id) + host_receipt = receipt_path_for_goal(goal, state) + assert host_receipt is not None and not host_receipt.exists() + before = list_goal_todos(registry_path=registry, goal_id=goal_id, todo_id=todo_id) + assert before["todo"]["resume_ready"] is False + events_before = load_rollout_events(events_path) + + code, preview = _invoke_turn([*argv, "--dry-run"]) + assert code == 0 and preview.get("should_run") is True, list(preview) + assert calls == [] and usage == [], "dry run dispatched an executing observation" + assert _observation(preview) is None and "host_poll_receipt" not in preview + assert "rollout_event" not in preview and not host_receipt.exists() + assert load_rollout_events(events_path) == events_before + after_preview = list_goal_todos(registry_path=registry, goal_id=goal_id, todo_id=todo_id) + assert after_preview["authority_read"]["provider_revision"] == before["authority_read"]["provider_revision"] + assert after_preview["todo"]["resume_ready"] is False + + code, refused = _invoke_turn([*argv, "--dry-run", "--turn-instance-id", "preview-should-not-admit"]) + assert code == 1 and refused["reason"] == "turn-scoped `quota should-run` cannot use --dry-run" + assert not calls and not usage and not host_receipt.exists() + assert load_rollout_events(events_path) == events_before + assert "status_projection_cache" not in preview and not (runtime / "status-projection-cache").exists() + + code, executed = _invoke_turn(argv) + assert code == 0 and executed["should_run"] is True, executed.get("reason") + assert len(calls) == 1 and usage == ["start"] and _observation(executed)["merged_count"] == 1 + assert executed["host_poll_receipt"]["poll_count"] == 1 + assert executed["status_projection_cache"]["written"] is True + assert list((runtime / "status-projection-cache").glob("*.json")) + assert json.loads(host_receipt.read_bytes())["poll_count"] == 1 + after_run = list_goal_todos(registry_path=registry, goal_id=goal_id, todo_id=todo_id)["todo"] + assert after_run["resume_ready"] is True + for key in ("status", "text", "claimed_by", "resume_when"): + assert after_run[key] == before["todo"][key] + kinds = [event["event_kind"] for event in load_rollout_events(events_path)[len(events_before):]] + assert kinds.count("pr_merge") == 1 and kinds.count("quota_should_run") == 1 + assert not {"quota_spend", "todo_complete"}.intersection(kinds) + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) @pytest.mark.parametrize("mode", ["absent-capability", "dry-run", "plan"]) def test_read_only_and_unconfigured_turns_do_not_observe(tmp_path, monkeypatch, provider, mode):