diff --git a/services/hermes/agent-configmap.yaml b/services/hermes/agent-configmap.yaml index 4ed408bc..47b463eb 100644 --- a/services/hermes/agent-configmap.yaml +++ b/services/hermes/agent-configmap.yaml @@ -364,8 +364,8 @@ data: branch. Existing-ref updates are rejected unless a coordinator-issued continuation has validated the root lineage and worker grant; protected refs, deletion, and force-push are always rejected. The broker inflates and scans every pushed object, so thin - packs are rejected; always push with `git push --no-thin` so the pack - is self-contained. For bounded + packs are rejected; use `git -c pack.window=0 -c pack.depth=0 push --no-thin` + so the broker receives a self-contained pack. For bounded repository/pull-request evidence or to create a review-ready draft PR, load `$manage-atlas-pull-requests` and use `/opt/scm/gitea_api.py`. The client carries no repository credential; a separate least-authority diff --git a/services/hermes/scm-common/scripts/scm_broker_server.py b/services/hermes/scm-common/scripts/scm_broker_server.py index 19b4e691..4d0abf17 100644 --- a/services/hermes/scm-common/scripts/scm_broker_server.py +++ b/services/hermes/scm-common/scripts/scm_broker_server.py @@ -29,9 +29,33 @@ def validate_ascii_headers(headers, *, maximum_count: int, maximum_bytes: int, e raise error("SCM request has duplicate Content-Length") +def _rejection_category(phase: str, error: BaseException) -> str: + """Reduce known pack-policy failures to fixed, non-sensitive categories.""" + if error.__class__.__name__ != "PolicyError": + return "upstream" if error.__class__.__name__ in {"TimeoutError", "URLError"} else "io" + if phase != "pack": + return "policy" + message = str(error) + if message == "Git command does not match its task grant": + return "grant-command" + if message == "Git update does not prove fast-forward ancestry": + return "ancestry" + if message == "Git delta base is outside the pack; push full packs": + return "thin-pack" + if "runtime credential material" in message: + return "credential" + if "credential-shaped content" in message: + return "content" + if "ref command" in message or "limited to namespaced" in message: + return "ref" + if "receive-pack" in message or "Git request has" in message: + return "framing" + return "policy" + + def log_rejection(phase: str, error: BaseException) -> None: """Log a fixed operational category without request material or error text.""" - category = "policy" if error.__class__.__name__ == "PolicyError" else "upstream" if error.__class__.__name__ in {"TimeoutError", "URLError"} else "io" + category = _rejection_category(phase, error) logging.warning("scm_rejected phase=%s category=%s", phase, category) diff --git a/services/hermes/scripts/execution_pool_scm.py b/services/hermes/scripts/execution_pool_scm.py index e31dc4ab..2c36973c 100644 --- a/services/hermes/scripts/execution_pool_scm.py +++ b/services/hermes/scripts/execution_pool_scm.py @@ -1,8 +1,6 @@ #!/usr/bin/env python3 """Assignment-bound Git gate routed exclusively through the PR14 SCM broker.""" - from __future__ import annotations - import json import hashlib import os @@ -21,7 +19,6 @@ from scm_task_grants import ZERO_SHA, sign_grant from execution_pool_project import ATLAS_REPO, validate_branch from execution_pool_protocol import ProtocolError, atomic_json, canonical_json, verify_envelope - WORKSPACE_ROOT = Path(os.environ.get("HERMES_WORKER_ROOT", "/workspace")) SCM_ROOT = Path(os.environ.get("HERMES_SCM_STATE_ROOT", "/scm-state")) ORDINAL = int(os.environ.get("HERMES_WORKER_ORDINAL", "-1")) @@ -29,6 +26,10 @@ BROKER_ORIGIN = scm_broker_client.BROKER_ORIGIN.rstrip("/") MAX_STATUS_BYTES = 4 * 1024 * 1024 MAX_PULL_PAGES = 20 +def _broker_push_args(grant: str, target: str) -> tuple[str, ...]: + """Build a self-contained receive-pack push for the broker scanner.""" + return ("-c", f"http.extraHeader=X-Hermes-Task-Grant: {grant}", "-c", "pack.window=0", "-c", "pack.depth=0", "push", "--no-thin", "hermes-broker", f"HEAD:refs/heads/{target}") + def _push_failure(error: RuntimeError) -> str: """Return bounded remediation without reflecting Git headers or grant text.""" @@ -404,8 +405,7 @@ class Boundary: if checkpoint: checkpoint() _run( - "-c", f"http.extraHeader=X-Hermes-Task-Grant: {grant}", - "push", "--no-thin", "hermes-broker", f"HEAD:refs/heads/{target}", + *_broker_push_args(grant, target), cwd=destination, timeout=900, ) except RuntimeError as error: @@ -491,7 +491,7 @@ class Boundary: if remote != head: if checkpoint: checkpoint() - _run("-c", f"http.extraHeader=X-Hermes-Task-Grant: {grant}", "push", "--no-thin", "hermes-broker", f"HEAD:refs/heads/{target}", cwd=destination, timeout=900) + _run(*_broker_push_args(grant, target), cwd=destination, timeout=900) if checkpoint: checkpoint() if checkpoint: diff --git a/services/hermes/scripts/release_soteria_publication_retry.py b/services/hermes/scripts/release_soteria_publication_retry.py new file mode 100755 index 00000000..97b9f006 --- /dev/null +++ b/services/hermes/scripts/release_soteria_publication_retry.py @@ -0,0 +1,210 @@ +#!/usr/bin/env python3 +"""Release Soteria #3's failed publication run after a reviewed mediator fix.""" + +from __future__ import annotations + +import hashlib +import json +import sqlite3 +from pathlib import Path +from typing import Any + +import scm_broker_client +import supervisor_state + + +BOARD = "soteria" +CHILD = "t_4095fe0d" +ROOT = "t_c7c42600" +SOURCE_RUN = "8" +FAILED_RUN = "10" +ORDINAL = 0 +BRANCH = "hermes-repair/sonar-AZ9pTqVcN0JrBQvDGDs3" +REMOTE_HEAD = "51133b559ad62f324e45bc61587900f08156b733" +PRESERVED_HEAD = "f21833e78768e7e4045a6304e311989196846ae4" +RECEIPT_DIGEST = "32f99b7ff27b2251e55cfe6fe2ff38fd3054bb781a8cc5e06b010ef855eb6c82" +SOURCE_RAW_SHA = "13de550c15ecfdf6684022b07a823da1c51586f959e9d9d2b3aea67433a9085d" +FAILED_RESULT_SHA = "0395eaed346eab364990de9b6d6c9a791c3d015a6b026f2473a9c80e8a1d2548" +FAILED_EVENT_ID = 209 +RETRY_AFTER = 1789346240 + + +def _value(task: Any, name: str) -> Any: + """Read one scalar from either native task representation.""" + return task.get(name) if isinstance(task, dict) else getattr(task, name, None) + + +def _pool_guard(database: Path) -> None: + """Require exactly run 10's preserved, unpublished mediator result.""" + connection = sqlite3.connect(f"file:{database}?mode=ro", uri=True) + try: + row = connection.execute( + "SELECT payload_json,result_json,result_digest,state,worker_ordinal,attempt " + "FROM assignments WHERE board=? AND task_id=? AND run_id=?", + (BOARD, CHILD, FAILED_RUN), + ).fetchone() + active = connection.execute( + "SELECT 1 FROM assignments WHERE board=? AND task_id=? AND run_id<>? " + "AND state IN ('assigned','running','result')", + (BOARD, CHILD, FAILED_RUN), + ).fetchone() + finally: + connection.close() + if row is None or active is not None or tuple(row[3:]) != ("finalized", ORDINAL, 1): + raise ValueError("run 10 is not the exact terminal publication attempt") + if not isinstance(row[0], str) or not isinstance(row[1], str): + raise ValueError("run 10 evidence is malformed") + try: + payload, result = json.loads(row[0]), json.loads(row[1]) + except json.JSONDecodeError as error: + raise ValueError("run 10 evidence is malformed") from error + resume = payload.get("scm_resume") if isinstance(payload, dict) else None + structured = result.get("structured") if isinstance(result, dict) else None + source = resume.get("source") if isinstance(resume, dict) else None + if ( + hashlib.sha256(row[1].encode()).hexdigest() != FAILED_RESULT_SHA + or row[2] != FAILED_RESULT_SHA + or not isinstance(source, dict) + or source.get("run_id") != SOURCE_RUN + or source.get("worker_ordinal") != ORDINAL + or resume.get("head") != PRESERVED_HEAD + or not isinstance(structured, dict) + or structured.get("status") != "blocked" + or result.get("returncode") != 1 + or result.get("capacity_failure") is not False + or result.get("scm_submission") is not None + ): + raise ValueError("run 10 is not the retained rejected publication") + + +def _native_guard() -> None: + """Require event 209 to remain the private child’s current blocked outcome.""" + from hermes_cli import kanban_db + + with kanban_db.scoped_current_board(BOARD): + connection = kanban_db.connect(board=BOARD) + try: + task = kanban_db.get_task(connection, CHILD) + parents = kanban_db.parent_ids(connection, CHILD) + event = connection.execute( + "SELECT id,run_id,kind FROM task_events WHERE task_id=? ORDER BY id DESC LIMIT 1", + (CHILD,), + ).fetchone() + latest = connection.execute( + "SELECT id,status,outcome FROM task_runs WHERE task_id=? ORDER BY id DESC LIMIT 1", + (CHILD,), + ).fetchone() + finally: + connection.close() + if ( + task is None + or _value(task, "status") != "blocked" + or _value(task, "current_run_id") is not None + or _value(task, "block_kind") != "capability" + or ROOT not in {str(parent) for parent in parents} + or tuple(event or ()) != (FAILED_EVENT_ID, int(FAILED_RUN), "blocked") + or tuple(latest or ()) != (int(FAILED_RUN), "blocked", "blocked") + ): + raise ValueError("native task changed after run 10") + + +def _receipt_guard() -> None: + """Verify the sealed source-8 receipt and its normalized title provenance.""" + receipt = supervisor_state.publication_retry(BOARD, CHILD, FAILED_RUN) + if not isinstance(receipt, dict): + raise ValueError("publication receipt is unavailable") + source = receipt.get("source") + if ( + receipt.get("head") != PRESERVED_HEAD + or receipt.get("result_digest") != RECEIPT_DIGEST + or not isinstance(source, dict) + or source.get("board") != BOARD + or source.get("task_id") != CHILD + or source.get("root_task_id") != ROOT + or source.get("run_id") != SOURCE_RUN + or source.get("worker_ordinal") != ORDINAL + ): + raise ValueError("publication receipt changed after normalization") + with supervisor_state._connect(BOARD) as connection: + retry = connection.execute( + "SELECT source_run_id,source_ordinal,issued_run_id,resolved_run_id,reissue_count,retry_after,last_reissued_run_id " + "FROM publication_retries WHERE board=? AND child_task_id=?", (BOARD, CHILD) + ).fetchone() + provenance = connection.execute( + "SELECT raw_result_sha256,reconstruction FROM publication_retry_provenance " + "WHERE board=? AND child_task_id=?", (BOARD, CHILD) + ).fetchone() + pending = (SOURCE_RUN, ORDINAL, FAILED_RUN, "", 1, RETRY_AFTER, "9") + released = (SOURCE_RUN, ORDINAL, "", "", 1, RETRY_AFTER, "9") + if ( + tuple(retry or ()) not in {pending, released} + or provenance is None + or provenance[0] != SOURCE_RAW_SHA + or not isinstance(provenance[1], str) + or not provenance[1].startswith("normalized-title-from-sealed-receipt-sha256:") + ): + raise ValueError("publication retry state changed after run 10") + + +def _remote_guard() -> None: + """Require the continuing pull request to remain open at its old broker head.""" + try: + pull = json.loads(scm_broker_client.read("/api/v1/repos/titan/soteria/pulls/3")) + except (json.JSONDecodeError, OSError) as error: + raise ValueError("pull request metadata is unavailable") from error + head = pull.get("head") if isinstance(pull, dict) else None + base = pull.get("base") if isinstance(pull, dict) else None + if ( + not isinstance(pull, dict) + or pull.get("state") != "open" + or pull.get("merged") is not False + or not isinstance(head, dict) + or head.get("ref") != BRANCH + or head.get("sha") != REMOTE_HEAD + or not isinstance(base, dict) + or base.get("ref") != "main" + ): + raise ValueError("pull request changed after run 10") + + +def release(pool_database: Path) -> bool: + """CAS-release run 10’s issue fence without changing its retry budget/history. + + A retry after a crash before native unblock accepts the already-released + state. Any other state is evidence that another actor changed the card. + """ + _pool_guard(pool_database) + _native_guard() + _receipt_guard() + _remote_guard() + with supervisor_state._connect(BOARD) as connection: + changed = connection.execute( + "UPDATE publication_retries SET issued_run_id='' WHERE board=? AND child_task_id=? " + "AND source_run_id=? AND source_ordinal=? AND issued_run_id=? AND resolved_run_id='' " + "AND reissue_count=1 AND retry_after=? AND last_reissued_run_id='9'", + (BOARD, CHILD, SOURCE_RUN, ORDINAL, FAILED_RUN, RETRY_AFTER), + ).rowcount + if changed == 1: + return True + with supervisor_state._connect(BOARD) as connection: + row = connection.execute( + "SELECT source_run_id,source_ordinal,issued_run_id,resolved_run_id,reissue_count,retry_after,last_reissued_run_id " + "FROM publication_retries WHERE board=? AND child_task_id=?", (BOARD, CHILD) + ).fetchone() + if tuple(row or ()) == (SOURCE_RUN, ORDINAL, "", "", 1, RETRY_AFTER, "9"): + return False + raise ValueError("publication retry state changed before CAS release") + + +if __name__ == "__main__": + import argparse + import os + + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--pool-db", type=Path, + default=Path(os.environ.get("HERMES_HOME", "/opt/data")) / "execution-pool/assignments.db", + ) + args = parser.parse_args() + release(args.pool_db) + print(f"released {BOARD}/{CHILD} failed_run={FAILED_RUN} source_run={SOURCE_RUN}") diff --git a/testing/tests/test_hermes_receive_pack_scan.py b/testing/tests/test_hermes_receive_pack_scan.py index 7d17fa6f..bc46fe6d 100644 --- a/testing/tests/test_hermes_receive_pack_scan.py +++ b/testing/tests/test_hermes_receive_pack_scan.py @@ -4,11 +4,14 @@ from __future__ import annotations import hashlib import io +import subprocess +import sys import zlib import pytest from testing.tests.test_hermes_scm_broker_support import ( + ROOT, _blob_pack, _load, _object_entry, @@ -16,6 +19,9 @@ from testing.tests.test_hermes_scm_broker_support import ( _receive_command, ) +sys.path.insert(0, str(ROOT / "services/hermes/scripts")) +import execution_pool_scm as scm # noqa: E402 + ZERO = b"0" * 40 COMMIT = b"1" * 40 TOKEN = "runtime-sentinel" @@ -119,6 +125,42 @@ def test_thin_pack_deltas_are_rejected_outright(): scan.validate_receive_pack(_request(_pack_of([entry])), TOKEN, FORMS) +def test_native_git_delta_free_update_pack_is_self_contained(tmp_path): + """The worker's scoped push settings produce a broker-scannable pack.""" + scan = _load("receive_pack_scan") + + def git(*args, input=None): + completed = subprocess.run( + ("git", *args), cwd=tmp_path, input=input, capture_output=True, + check=False, + ) + assert completed.returncode == 0 + return completed.stdout + + git("init") + git("config", "user.email", "hermes@atlas.invalid") + git("config", "user.name", "Hermes") + tracked = tmp_path / "tracked.txt" + tracked.write_text("base\n" + "same text\n" * 1024) + git("add", "tracked.txt") + git("commit", "-m", "base") + old = git("rev-parse", "HEAD").strip() + tracked.write_text("updated\n" + "same text\n" * 1024) + git("commit", "-am", "update") + head = git("rev-parse", "HEAD").strip() + push_args = scm._broker_push_args("grant", "hermes/scan") + config_args = push_args[:push_args.index("push")] + pack = git( + *config_args, "pack-objects", + "--stdout", "--revs", input=head + b"\n^" + old + b"\n", + ) + request = _receive_command(old, head, b"refs/heads/hermes/scan", pack) + assert scan.validate_receive_pack( + request, TOKEN, FORMS, + expected=(old.decode(), head.decode(), "hermes/scan"), + ) == (old.decode(), head.decode(), "hermes/scan") + + def test_trailing_bytes_after_the_pack_are_rejected(): scan = _load("receive_pack_scan") body = _request(_pack_of([])) + b"extra" @@ -214,4 +256,4 @@ def test_validation_rewinds_the_stream_for_upstream_forwarding(): scan.validate_receive_pack(stream, TOKEN, FORMS) assert stream.tell() == 0 with pytest.raises(scan.PolicyError): - scan.validate_receive_pack(io.BytesIO(b"zzzz"), TOKEN, FORMS) \ No newline at end of file + scan.validate_receive_pack(io.BytesIO(b"zzzz"), TOKEN, FORMS) diff --git a/testing/tests/test_hermes_scm_server_coverage.py b/testing/tests/test_hermes_scm_server_coverage.py index 7a358640..7bee7813 100644 --- a/testing/tests/test_hermes_scm_server_coverage.py +++ b/testing/tests/test_hermes_scm_server_coverage.py @@ -40,6 +40,32 @@ def test_absolute_reader_supports_idle_limit_newline_eof_and_attribute_delegatio assert reader.readline() == b"" +@pytest.mark.parametrize( + ("phase", "detail", "expected"), + [ + ("pack", "Git command does not match its task grant", "grant-command"), + ("pack", "Git update does not prove fast-forward ancestry", "ancestry"), + ("pack", "Git delta base is outside the pack; push full packs", "thin-pack"), + ("pack", "Git push contains runtime credential material", "credential"), + ("pack", "Git push contains credential-shaped content", "content"), + ("pack", "Git receive-pack command framing is invalid", "framing"), + ("pack", "Git push is limited to namespaced feature branches", "ref"), + ("control", "arbitrary policy detail", "policy"), + ], +) +def test_rejection_log_uses_fixed_pack_categories_without_error_text( + monkeypatch, phase, detail, expected +): + module = _load("scm_broker_server") + policy = type("PolicyError", (Exception,), {}) + observed = [] + monkeypatch.setattr(module.logging, "warning", lambda message, *args: observed.append((message, args))) + + module.log_rejection(phase, policy(detail)) + + assert observed == [("scm_rejected phase=%s category=%s", (phase, expected))] + + def _handler_type(module): class StubHandler: request_version = "HTTP/1.1" diff --git a/testing/tests/test_release_soteria_publication_retry.py b/testing/tests/test_release_soteria_publication_retry.py new file mode 100644 index 00000000..44efae3b --- /dev/null +++ b/testing/tests/test_release_soteria_publication_retry.py @@ -0,0 +1,191 @@ +"""Fail-closed coverage for the one-time Soteria publication release.""" + +from __future__ import annotations + +import json +import hashlib +import sqlite3 +import sys +from contextlib import contextmanager, nullcontext +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from testing.tests.test_hermes_cli_support import _load + + +ROOT = Path(__file__).parents[2] +sys.path.insert(0, str(ROOT / "services/hermes/scm-common/scripts")) +release = _load("release_soteria_publication_retry") + + +def _receipt() -> dict: + return { + "source": { + "board": release.BOARD, + "task_id": release.CHILD, + "root_task_id": release.ROOT, + "run_id": release.SOURCE_RUN, + "worker_ordinal": release.ORDINAL, + }, + "head": release.PRESERVED_HEAD, + "result_digest": release.RECEIPT_DIGEST, + } + + +def _pool(tmp_path: Path, *, active: bool = False, digest: str | None = None) -> tuple[Path, str]: + """Create only the terminal pool columns consumed by the release guard.""" + database = tmp_path / "pool.db" + connection = sqlite3.connect(database) + connection.execute( + "CREATE TABLE assignments(board,task_id,run_id,payload_json,result_json,result_digest,state,worker_ordinal,attempt)" + ) + resume = {"source": {"run_id": release.SOURCE_RUN, "worker_ordinal": 0}, "head": release.PRESERVED_HEAD} + result = { + "structured": {"status": "blocked"}, "returncode": 1, + "capacity_failure": False, "scm_submission": None, + } + encoded = json.dumps(result, separators=(",", ":")) + actual_digest = hashlib.sha256(encoded.encode()).hexdigest() + connection.execute( + "INSERT INTO assignments VALUES(?,?,?,?,?,?,?,?,?)", + (release.BOARD, release.CHILD, release.FAILED_RUN, json.dumps({"scm_resume": resume}), encoded, + digest or actual_digest, "finalized", 0, 1), + ) + if active: + connection.execute( + "INSERT INTO assignments VALUES(?,?,?,?,?,?,?,?,?)", + (release.BOARD, release.CHILD, "11", "{}", "{}", "", "assigned", 0, 1), + ) + connection.commit() + connection.close() + return database, actual_digest + + +def _native(monkeypatch, *, event=None, latest=None): + """Install the actual scalar contract expected from the native adapter.""" + event = event or (release.FAILED_EVENT_ID, int(release.FAILED_RUN), "blocked") + latest = latest or (int(release.FAILED_RUN), "blocked", "blocked") + state = SimpleNamespace( + task=SimpleNamespace(status="blocked", current_run_id=None, block_kind="capability"), + event=event, latest=latest, + ) + + class Connection: + def execute(self, statement, _args): + row = state.event if "task_events" in statement else state.latest + return SimpleNamespace(fetchone=lambda: row) + + def close(self): + return None + + native = SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), connect=lambda **_kwargs: Connection(), + get_task=lambda _connection, _task_id: state.task, + parent_ids=lambda _connection, _task_id: [release.ROOT], + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=native)) + monkeypatch.setitem(sys.modules, "hermes_cli.kanban_db", native) + return state + + +def _receipt_state(monkeypatch, receipt: dict, *, issued: str = release.FAILED_RUN): + """Supply the exact sidecar rows without accepting arbitrary metadata.""" + retry = (release.SOURCE_RUN, 0, issued, "", 1, release.RETRY_AFTER, "9") + provenance = (release.SOURCE_RAW_SHA, "normalized-title-from-sealed-receipt-sha256:" + "a" * 64) + + class Connection: + def execute(self, statement, _args): + row = retry if "publication_retries" in statement else provenance + return SimpleNamespace(fetchone=lambda: row) + + monkeypatch.setattr(release.supervisor_state, "publication_retry", lambda *_args: receipt) + monkeypatch.setattr(release.supervisor_state, "_connect", lambda _board: nullcontext(Connection())) + + +def _remote(monkeypatch, *, head: str = release.REMOTE_HEAD): + """Return a scalar broker PR response without contacting a live broker.""" + value = {"state": "open", "merged": False, "head": {"ref": release.BRANCH, "sha": head}, "base": {"ref": "main"}} + monkeypatch.setattr(release.scm_broker_client, "read", lambda _path: json.dumps(value).encode()) + + +def test_exact_historical_fixture_passes_all_release_guards(tmp_path, monkeypatch): + """The copied run-10/native/receipt fixture admits only its reviewed form.""" + pool, digest = _pool(tmp_path) + monkeypatch.setattr(release, "FAILED_RESULT_SHA", digest) + _native(monkeypatch) + _receipt_state(monkeypatch, _receipt()) + _remote(monkeypatch) + + release._pool_guard(pool) + release._native_guard() + release._receipt_guard() + release._remote_guard() + + +@pytest.mark.parametrize("change", ["later_event", "changed_receipt", "changed_head", "terminal_digest", "active"]) +def test_conflicting_evidence_fails_closed_without_a_release(tmp_path, monkeypatch, change): + """A later action or altered evidence cannot pass the operator preconditions.""" + receipt = _receipt() + pool, digest = _pool(tmp_path, active=change == "active", digest="f" * 64 if change == "terminal_digest" else None) + monkeypatch.setattr(release, "FAILED_RESULT_SHA", digest if change != "terminal_digest" else "e" * 64) + state = _native(monkeypatch, event=(210, None, "blocked") if change == "later_event" else None) + if change == "changed_receipt": + receipt["result_digest"] = "e" * 64 + _receipt_state(monkeypatch, receipt) + _remote(monkeypatch, head="d" * 40 if change == "changed_head" else release.REMOTE_HEAD) + + with pytest.raises(ValueError): + if change in {"terminal_digest", "active"}: + release._pool_guard(pool) + elif change == "later_event": + assert state.event[0] == 210 + release._native_guard() + elif change == "changed_receipt": + release._receipt_guard() + else: + release._remote_guard() + + +def test_exact_cas_release_is_repeatable_and_preserves_budget_and_provenance(tmp_path, monkeypatch): + """A crash before native unblock can retry the CAS without spending another budget.""" + state_db = tmp_path / "state.db" + connection = sqlite3.connect(state_db) + connection.executescript( + "CREATE TABLE publication_retries(board,child_task_id,source_run_id,source_ordinal,issued_run_id," + "resolved_run_id,reissue_count,retry_after,last_reissued_run_id);" + "CREATE TABLE publication_retry_provenance(board,child_task_id,raw_result_sha256,reconstruction);" + ) + connection.execute( + "INSERT INTO publication_retries VALUES(?,?,?,?,?,?,?,?,?)", + (release.BOARD, release.CHILD, release.SOURCE_RUN, 0, release.FAILED_RUN, "", 1, release.RETRY_AFTER, "9"), + ) + connection.execute( + "INSERT INTO publication_retry_provenance VALUES(?,?,?,?)", + (release.BOARD, release.CHILD, release.SOURCE_RAW_SHA, "normalized-title-from-sealed-receipt-sha256:" + "a" * 64), + ) + connection.commit() + connection.close() + + @contextmanager + def connect(_board): + connection = sqlite3.connect(state_db) + try: + yield connection + connection.commit() + finally: + connection.close() + + for name in ("_pool_guard", "_native_guard", "_receipt_guard", "_remote_guard"): + monkeypatch.setattr(release, name, lambda *_args: None) + monkeypatch.setattr(release.supervisor_state, "_connect", connect) + + assert release.release(tmp_path / "unused.db") is True + assert release.release(tmp_path / "unused.db") is False + connection = sqlite3.connect(state_db) + retry = connection.execute("SELECT issued_run_id,reissue_count,retry_after,last_reissued_run_id FROM publication_retries").fetchone() + provenance = connection.execute("SELECT raw_result_sha256,reconstruction FROM publication_retry_provenance").fetchone() + connection.close() + assert retry == ("", 1, release.RETRY_AFTER, "9") + assert provenance[0] == release.SOURCE_RAW_SHA