Keep quota should-run previews read-only

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,

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,

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):