diff --git a/loopx/control_plane/quota/void_commit.py b/loopx/control_plane/quota/void_commit.py index 652b903e76..fcdd74079f 100644 --- a/loopx/control_plane/quota/void_commit.py +++ b/loopx/control_plane/quota/void_commit.py @@ -22,6 +22,7 @@ QUOTA_VOID_COMMIT_STATUSES = frozenset( {"preview", "not_found", "written", "replayed", "repaired", "conflict"} ) +_USE_CURRENT_INDEX_DIGEST = object() _ECMASCRIPT_TRIM_CHARS = ( "\u0009\u000a\u000b\u000c\u000d\u0020\u00a0\u1680" "\u2000\u2001\u2002\u2003\u2004\u2005\u2006\u2007\u2008\u2009\u200a" @@ -108,11 +109,27 @@ def _void_result( return result -def _result_payload(result: Mapping[str, Any]) -> dict[str, Any]: +def _result_payload( + result: Mapping[str, Any], + *, + include_index_digest: bool, +) -> dict[str, Any]: payload = result.get("payload") if not isinstance(payload, Mapping): raise RuntimeError("TypeScript quota void commit omitted its payload") - return dict(payload) + projected = dict(payload) + if include_index_digest: + if "index_digest" not in result: + raise RuntimeError( + "TypeScript quota void commit omitted its index digest" + ) + index_digest = result["index_digest"] + if index_digest is not None and not isinstance(index_digest, str): + raise RuntimeError( + "TypeScript quota void commit index digest must be a string or null" + ) + projected["expected_index_digest"] = index_digest + return projected def _runtime_root(status_payload: Mapping[str, Any]) -> Path: @@ -139,6 +156,7 @@ def commit_quota_slot_void( _operation: str = "commit", registry_path: Path | None = None, goal_ref: Mapping[str, Any] | None = None, + _expected_index_digest: str | None | object = _USE_CURRENT_INDEX_DIGEST, ) -> dict[str, Any]: """Execute one TypeScript-owned quota void transaction.""" @@ -164,7 +182,11 @@ def commit_quota_slot_void( if registry_path is None and goal_ref is None: if execute: with exclusive_file_lock(index_path, operation="quota_void_commit"): - params["expected_index_digest"] = quota_spend_index_digest(index_path) + params["expected_index_digest"] = ( + quota_spend_index_digest(index_path) + if _expected_index_digest is _USE_CURRENT_INDEX_DIGEST + else _expected_index_digest + ) result = _void_result(params, expected_goal_id=safe_goal_id) else: params["expected_index_digest"] = quota_spend_index_digest(index_path) @@ -178,13 +200,17 @@ def commit_quota_slot_void( operation="quota_void_commit", lock_legacy_index=execute, ) as source_admission: - params["expected_index_digest"] = quota_spend_index_digest(index_path) + params["expected_index_digest"] = ( + quota_spend_index_digest(index_path) + if _expected_index_digest is _USE_CURRENT_INDEX_DIGEST + else _expected_index_digest + ) if goal_ref is not None: params["goal_ref"] = dict(goal_ref) if source_admission is not None: params["source_admission"] = dict(source_admission) result = _void_result(params, expected_goal_id=safe_goal_id) - return _result_payload(result) + return _result_payload(result, include_index_digest=not execute) def build_quota_slot_void_preview_for_decision( @@ -253,6 +279,17 @@ def record_quota_slot_void_from_preview( before = preview.get("before") if not isinstance(before, Mapping): raise ValueError("quota void preview has no before decision") + expected_index_digest: str | None | object = _USE_CURRENT_INDEX_DIGEST + if execute: + if "expected_index_digest" not in preview: + raise ValueError("quota void preview does not include its index basis") + expected_index_digest = preview["expected_index_digest"] + if expected_index_digest is not None and not isinstance( + expected_index_digest, str + ): + raise ValueError( + "quota void preview index basis must be a string or null" + ) return commit_quota_slot_void( status_payload, goal_id=safe_goal_id, @@ -265,4 +302,5 @@ def record_quota_slot_void_from_preview( reason_summary=reason_summary, registry_path=registry_path, goal_ref=goal_ref, + _expected_index_digest=expected_index_digest, ) diff --git a/tests/control_plane/test_quota_void_commit_runtime.py b/tests/control_plane/test_quota_void_commit_runtime.py index 8190f1cae7..9729733123 100644 --- a/tests/control_plane/test_quota_void_commit_runtime.py +++ b/tests/control_plane/test_quota_void_commit_runtime.py @@ -162,7 +162,16 @@ def _write_source_registry( transaction.commit(payload) +def _index_digest(index_path: Path) -> str | None: + try: + content = index_path.read_bytes() + except FileNotFoundError: + return None + return "sha256:" + hashlib.sha256(content).hexdigest() + + def _preview(runtime_root: Path) -> dict[str, Any]: + index_path = runtime_root / "goals" / GOAL_ID / "runs" / "index.jsonl" return { "ok": True, "mode": "void-slot", @@ -189,6 +198,7 @@ def _preview(runtime_root: Path) -> dict[str, Any]: "spend event is inside the same accounting window." ), "classification": "quota_slot_voided", + "expected_index_digest": _index_digest(index_path), } @@ -380,13 +390,14 @@ def test_source_void_stamps_the_current_goal_ref( goal_ref=goal_ref_b, ) assert preview["goal_ref"] == goal_ref_b + assert preview["expected_index_digest"] == _index_digest(index_path) assert index_path.read_text(encoding="utf-8").count("\n") == 1 - written = commit_quota_slot_void( + written = record_quota_slot_void_from_preview( + preview, {"runtime_root": str(runtime_root)}, goal_id=GOAL_ID, - voided_run_generated_at=TARGET_AT, - before=_decision(2), + render_markdown=render_quota_slot_preview_markdown, execute=True, source="heartbeat", registry_path=registry_path, @@ -530,9 +541,14 @@ def test_independent_invocations_against_one_target_append_twice( runtime_root = tmp_path / "runtime" index_path, _target_path = _write_spend_target(runtime_root) status = {"runtime_root": str(runtime_root)} - preview = _preview(runtime_root) def append_once() -> dict[str, Any]: + preview = build_quota_slot_void_preview_for_decision( + status, + goal_id=GOAL_ID, + voided_run_generated_at=TARGET_AT, + before=_decision(2), + ) return record_quota_slot_void_from_preview( preview, status, @@ -558,6 +574,138 @@ def append_once() -> dict[str, Any]: assert persisted["quota_event"]["voided_run_generated_at"] == TARGET_AT +def test_stale_preview_rejects_an_intervening_index_write( + tmp_path: Path, +) -> None: + runtime_root = tmp_path / "runtime" + index_path, _target_path = _write_spend_target(runtime_root) + status = {"runtime_root": str(runtime_root)} + preview = build_quota_slot_void_preview_for_decision( + status, + goal_id=GOAL_ID, + voided_run_generated_at=TARGET_AT, + before=_decision(2), + ) + with index_path.open("a", encoding="utf-8") as stream: + stream.write( + json.dumps( + { + "generated_at": "2026-08-31T09:02:00+08:00", + "goal_id": GOAL_ID, + "classification": "state_refreshed", + } + ) + + "\n" + ) + + with pytest.raises( + ValueError, + match="compare-and-swap precondition failed", + ): + record_quota_slot_void_from_preview( + preview, + status, + goal_id=GOAL_ID, + render_markdown=render_quota_slot_preview_markdown, + execute=True, + source="heartbeat", + reason_summary=REASON, + ) + + rows = [ + json.loads(line) + for line in index_path.read_text(encoding="utf-8").splitlines() + ] + assert [row["classification"] for row in rows] == [ + "quota_slot_spent", + "state_refreshed", + ] + + +def test_explicit_null_preview_basis_conflicts_when_an_index_appears( + tmp_path: Path, +) -> None: + runtime_root = tmp_path / "runtime" + preview = _preview(runtime_root) + assert preview["expected_index_digest"] is None + index_path, _target_path = _write_spend_target(runtime_root) + before = index_path.read_bytes() + + with pytest.raises( + ValueError, + match="compare-and-swap precondition failed", + ): + record_quota_slot_void_from_preview( + preview, + {"runtime_root": str(runtime_root)}, + goal_id=GOAL_ID, + render_markdown=render_quota_slot_preview_markdown, + execute=True, + source="heartbeat", + reason_summary=REASON, + ) + + assert index_path.read_bytes() == before + + +@pytest.mark.parametrize("preview_change", ["missing", "malformed"]) +def test_execute_from_preview_rejects_invalid_index_basis_before_runtime( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + preview_change: str, +) -> None: + runtime_root = tmp_path / "runtime" + index_path, _target_path = _write_spend_target(runtime_root) + preview = _preview(runtime_root) + if preview_change == "missing": + preview.pop("expected_index_digest") + message = "does not include its index basis" + else: + preview["expected_index_digest"] = 7 + message = "index basis must be a string or null" + before = index_path.read_bytes() + calls = 0 + + def unexpected_call(_method: str, _params: dict[str, Any]) -> object: + nonlocal calls + calls += 1 + raise AssertionError("invalid preview basis reached the TypeScript runtime") + + monkeypatch.setattr(void_commit, "effect_runtime_result", unexpected_call) + + with pytest.raises(ValueError, match=message): + record_quota_slot_void_from_preview( + preview, + {"runtime_root": str(runtime_root)}, + goal_id=GOAL_ID, + render_markdown=render_quota_slot_preview_markdown, + execute=True, + ) + + assert calls == 0 + assert index_path.read_bytes() == before + + +def test_non_executing_record_accepts_a_legacy_preview_without_index_basis( + tmp_path: Path, +) -> None: + runtime_root = tmp_path / "runtime" + index_path, _target_path = _write_spend_target(runtime_root) + preview = _preview(runtime_root) + preview.pop("expected_index_digest") + + result = record_quota_slot_void_from_preview( + preview, + {"runtime_root": str(runtime_root)}, + goal_id=GOAL_ID, + render_markdown=render_quota_slot_preview_markdown, + execute=False, + ) + + assert result["dry_run"] is True + assert result["expected_index_digest"] == _index_digest(index_path) + + def test_each_python_facade_uses_one_typed_runtime_request( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, @@ -658,6 +806,18 @@ def call(method: str, params: dict[str, Any]) -> dict[str, Any]: }, "payload effect_id mismatch", ), + ( + lambda result: { + key: value + for key, value in result.items() + if key != "index_digest" + }, + "omitted its index digest", + ), + ( + lambda result: {**result, "index_digest": 7}, + "index digest must be a string or null", + ), ], ) def test_python_facade_rejects_mismatched_typed_results( @@ -826,7 +986,7 @@ def unexpected_call(_method: str, _params: dict[str, Any]) -> object: assert not (tmp_path / "goals").exists() -def test_execute_holds_legacy_index_lock_and_sends_expected_digest( +def test_direct_execute_reads_digest_while_holding_legacy_index_lock( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -852,21 +1012,77 @@ def call(method: str, params: dict[str, Any]) -> dict[str, Any]: captured.update(params) return _typed_result(params, runtime_root) + def digest(path: Path) -> str | None: + assert path == index_path + timeline.append("digest") + return expected_digest + monkeypatch.setattr(void_commit, "exclusive_file_lock", lock) + monkeypatch.setattr(void_commit, "quota_spend_index_digest", digest) monkeypatch.setattr(void_commit, "effect_runtime_result", call) - payload = record_quota_slot_void_from_preview( - _preview(runtime_root), + payload = commit_quota_slot_void( {"runtime_root": str(runtime_root)}, goal_id=GOAL_ID, - render_markdown=render_quota_slot_preview_markdown, + voided_run_generated_at=TARGET_AT, + before=_decision(2), execute=True, source="heartbeat", reason_summary=REASON, ) assert payload["appended"] is True - assert timeline == ["lock-enter", "runtime", "lock-exit"] + assert timeline == ["lock-enter", "digest", "runtime", "lock-exit"] assert captured["execute"] is True assert captured["expected_index_digest"] == expected_digest assert str(captured["effect_id"]).startswith("quota-void:") + + +def test_execute_from_preview_holds_lock_and_forwards_original_index_basis( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + runtime_root = tmp_path / "runtime" + index_path, _target_path = _write_spend_target(runtime_root) + preview = _preview(runtime_root) + preview_basis = "sha256:preview-basis" + preview["expected_index_digest"] = preview_basis + timeline: list[str] = [] + captured: dict[str, Any] = {} + + @contextmanager + def lock(path: Path, **kwargs: Any) -> Iterator[Path]: + assert path == index_path + assert kwargs["operation"] == "quota_void_commit" + timeline.append("lock-enter") + try: + yield path.with_name(f"{path.name}.lock") + finally: + timeline.append("lock-exit") + + def unexpected_digest(_index_path: Path) -> str | None: + raise AssertionError("execute-from-preview reread the index digest") + + def call(method: str, params: dict[str, Any]) -> dict[str, Any]: + assert method == "quota.void.commit" + timeline.append("runtime") + captured.update(params) + return _typed_result(params, runtime_root) + + monkeypatch.setattr(void_commit, "exclusive_file_lock", lock) + monkeypatch.setattr(void_commit, "quota_spend_index_digest", unexpected_digest) + monkeypatch.setattr(void_commit, "effect_runtime_result", call) + + payload = record_quota_slot_void_from_preview( + preview, + {"runtime_root": str(runtime_root)}, + goal_id=GOAL_ID, + render_markdown=render_quota_slot_preview_markdown, + execute=True, + source="heartbeat", + reason_summary=REASON, + ) + + assert payload["appended"] is True + assert timeline == ["lock-enter", "runtime", "lock-exit"] + assert captured["expected_index_digest"] == preview_basis