From e07cbdfce8a31c36b1e8ef341554aedcd5e37fe5 Mon Sep 17 00:00:00 2001 From: jenkins Date: Sun, 13 Sep 2026 20:41:00 -0500 Subject: [PATCH] hermes: recover confirmation for the published Soteria task --- .../release_soteria_publication_retry.py | 106 ++++++++++++++---- .../test_release_soteria_publication_retry.py | 89 +++++++++++++-- 2 files changed, 164 insertions(+), 31 deletions(-) diff --git a/services/hermes/scripts/release_soteria_publication_retry.py b/services/hermes/scripts/release_soteria_publication_retry.py index 97b9f006..5e705414 100755 --- a/services/hermes/scripts/release_soteria_publication_retry.py +++ b/services/hermes/scripts/release_soteria_publication_retry.py @@ -9,6 +9,7 @@ import sqlite3 from pathlib import Path from typing import Any +from gitea_api_policy import _draft_title import scm_broker_client import supervisor_state @@ -27,6 +28,8 @@ SOURCE_RAW_SHA = "13de550c15ecfdf6684022b07a823da1c51586f959e9d9d2b3aea67433a908 FAILED_RESULT_SHA = "0395eaed346eab364990de9b6d6c9a791c3d015a6b026f2473a9c80e8a1d2548" FAILED_EVENT_ID = 209 RETRY_AFTER = 1789346240 +CONFIRM_RUN = "11" +CONFIRM_EVENT_ID = 213 def _value(task: Any, name: str) -> Any: @@ -34,30 +37,30 @@ def _value(task: Any, name: str) -> Any: 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.""" +def _pool_guard(database: Path, run_id: str) -> None: + """Require one exact terminal mediator result and no competing assignment.""" 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), + (BOARD, CHILD, run_id), ).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), + (BOARD, CHILD, run_id), ).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") + raise ValueError("publication attempt is not the exact terminal run") if not isinstance(row[0], str) or not isinstance(row[1], str): - raise ValueError("run 10 evidence is malformed") + raise ValueError("publication attempt 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 + raise ValueError("publication attempt 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 @@ -74,11 +77,11 @@ def _pool_guard(database: Path) -> None: 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") + raise ValueError("publication attempt is not the retained rejected publication") -def _native_guard() -> None: - """Require event 209 to remain the private child’s current blocked outcome.""" +def _native_guard(run_id: str, event_id: int, status: str, event_kind: str) -> None: + """Require one exact native terminal event and private root/child chain.""" from hermes_cli import kanban_db with kanban_db.scoped_current_board(BOARD): @@ -98,19 +101,19 @@ def _native_guard() -> None: connection.close() if ( task is None - or _value(task, "status") != "blocked" + or _value(task, "status") != status 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") + or tuple(event or ()) != (event_id, int(run_id), event_kind) + or tuple(latest or ()) != (int(run_id), "blocked", "blocked") ): raise ValueError("native task changed after run 10") -def _receipt_guard() -> None: +def _receipt_guard(run_id: str) -> dict[str, Any]: """Verify the sealed source-8 receipt and its normalized title provenance.""" - receipt = supervisor_state.publication_retry(BOARD, CHILD, FAILED_RUN) + receipt = supervisor_state.publication_retry(BOARD, CHILD, run_id) if not isinstance(receipt, dict): raise ValueError("publication receipt is unavailable") source = receipt.get("source") @@ -134,7 +137,7 @@ def _receipt_guard() -> None: "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") + pending = (SOURCE_RUN, ORDINAL, run_id, "", 1, RETRY_AFTER, "9") released = (SOURCE_RUN, ORDINAL, "", "", 1, RETRY_AFTER, "9") if ( tuple(retry or ()) not in {pending, released} @@ -144,6 +147,7 @@ def _receipt_guard() -> None: or not provenance[1].startswith("normalized-title-from-sealed-receipt-sha256:") ): raise ValueError("publication retry state changed after run 10") + return receipt def _remote_guard() -> None: @@ -167,15 +171,41 @@ def _remote_guard() -> None: raise ValueError("pull request changed after run 10") +def _published_remote_guard(receipt: dict[str, Any]) -> None: + """Require the already-published PR head and prose to equal sealed evidence.""" + try: + pull = json.loads(scm_broker_client.read("/api/v1/repos/titan/soteria/pulls/3")) + except (json.JSONDecodeError, OSError) as error: + raise ValueError("published 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 + title, body = receipt.get("title"), receipt.get("body") + if ( + not isinstance(title, str) + or not isinstance(body, str) + or 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") != PRESERVED_HEAD + or not isinstance(base, dict) + or base.get("ref") != "main" + or pull.get("title") != _draft_title(title) + or pull.get("body") != body + ): + raise ValueError("published pull request changed after run 11") + + 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() + _pool_guard(pool_database, FAILED_RUN) + _native_guard(FAILED_RUN, FAILED_EVENT_ID, "blocked", "blocked") + _receipt_guard(FAILED_RUN) _remote_guard() with supervisor_state._connect(BOARD) as connection: changed = connection.execute( @@ -196,15 +226,49 @@ def release(pool_database: Path) -> bool: raise ValueError("publication retry state changed before CAS release") +def confirm_published_release(pool_database: Path) -> bool: + """Release run 11 only after its PR publish is independently visible. + + This intentionally does not reopen native work. The caller must separately + use the native triage specification API after a fixed mediator is deployed. + """ + _pool_guard(pool_database, CONFIRM_RUN) + _native_guard(CONFIRM_RUN, CONFIRM_EVENT_ID, "triage", "block_loop_detected") + receipt = _receipt_guard(CONFIRM_RUN) + _published_remote_guard(receipt) + 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, CONFIRM_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("published confirmation changed before CAS release") + + if __name__ == "__main__": import argparse import os parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--confirm-published", action="store_true") 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}") + if args.confirm_published: + confirm_published_release(args.pool_db) + print(f"released {BOARD}/{CHILD} published_run={CONFIRM_RUN} source_run={SOURCE_RUN}") + else: + release(args.pool_db) + print(f"released {BOARD}/{CHILD} failed_run={FAILED_RUN} source_run={SOURCE_RUN}") diff --git a/testing/tests/test_release_soteria_publication_retry.py b/testing/tests/test_release_soteria_publication_retry.py index 44efae3b..76b8b05c 100644 --- a/testing/tests/test_release_soteria_publication_retry.py +++ b/testing/tests/test_release_soteria_publication_retry.py @@ -31,10 +31,12 @@ def _receipt() -> dict: }, "head": release.PRESERVED_HEAD, "result_digest": release.RECEIPT_DIGEST, + "title": "Replace cache literals safely.", + "body": "Preserved validation evidence.", } -def _pool(tmp_path: Path, *, active: bool = False, digest: str | None = None) -> tuple[Path, str]: +def _pool(tmp_path: Path, *, active: bool = False, digest: str | None = None, run_id: str = release.FAILED_RUN) -> tuple[Path, str]: """Create only the terminal pool columns consumed by the release guard.""" database = tmp_path / "pool.db" connection = sqlite3.connect(database) @@ -50,7 +52,7 @@ def _pool(tmp_path: Path, *, active: bool = False, digest: str | None = None) -> 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, + (release.BOARD, release.CHILD, run_id, json.dumps({"scm_resume": resume}), encoded, digest or actual_digest, "finalized", 0, 1), ) if active: @@ -63,12 +65,12 @@ def _pool(tmp_path: Path, *, active: bool = False, digest: str | None = None) -> return database, actual_digest -def _native(monkeypatch, *, event=None, latest=None): +def _native(monkeypatch, *, event=None, latest=None, status="blocked", block_kind="capability"): """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"), + task=SimpleNamespace(status=status, current_run_id=None, block_kind=block_kind), event=event, latest=latest, ) @@ -118,9 +120,9 @@ def test_exact_historical_fixture_passes_all_release_guards(tmp_path, monkeypatc _receipt_state(monkeypatch, _receipt()) _remote(monkeypatch) - release._pool_guard(pool) - release._native_guard() - release._receipt_guard() + release._pool_guard(pool, release.FAILED_RUN) + release._native_guard(release.FAILED_RUN, release.FAILED_EVENT_ID, "blocked", "blocked") + release._receipt_guard(release.FAILED_RUN) release._remote_guard() @@ -138,12 +140,12 @@ def test_conflicting_evidence_fails_closed_without_a_release(tmp_path, monkeypat with pytest.raises(ValueError): if change in {"terminal_digest", "active"}: - release._pool_guard(pool) + release._pool_guard(pool, release.FAILED_RUN) elif change == "later_event": assert state.event[0] == 210 - release._native_guard() + release._native_guard(release.FAILED_RUN, release.FAILED_EVENT_ID, "blocked", "blocked") elif change == "changed_receipt": - release._receipt_guard() + release._receipt_guard(release.FAILED_RUN) else: release._remote_guard() @@ -189,3 +191,70 @@ def test_exact_cas_release_is_repeatable_and_preserves_budget_and_provenance(tmp connection.close() assert retry == ("", 1, release.RETRY_AFTER, "9") assert provenance[0] == release.SOURCE_RAW_SHA + + +def test_published_confirmation_requires_run11_triage_and_exact_pr_prose(tmp_path, monkeypatch): + """The post-push recovery accepts only the unchanged run-11 confirmation.""" + pool, digest = _pool(tmp_path, run_id=release.CONFIRM_RUN) + monkeypatch.setattr(release, "FAILED_RESULT_SHA", digest) + _native( + monkeypatch, + event=(release.CONFIRM_EVENT_ID, int(release.CONFIRM_RUN), "block_loop_detected"), + latest=(int(release.CONFIRM_RUN), "blocked", "blocked"), + status="triage", + ) + receipt = _receipt() + _receipt_state(monkeypatch, receipt, issued=release.CONFIRM_RUN) + pull = { + "state": "open", "merged": False, + "head": {"ref": release.BRANCH, "sha": release.PRESERVED_HEAD}, + "base": {"ref": "main"}, + "title": release._draft_title(receipt["title"]), "body": receipt["body"], + } + monkeypatch.setattr(release.scm_broker_client, "read", lambda _path: json.dumps(pull).encode()) + + release._pool_guard(pool, release.CONFIRM_RUN) + release._native_guard(release.CONFIRM_RUN, release.CONFIRM_EVENT_ID, "triage", "block_loop_detected") + release._published_remote_guard(release._receipt_guard(release.CONFIRM_RUN)) + + pull["body"] = "changed" + with pytest.raises(ValueError, match="published pull"): + release._published_remote_guard(receipt) + + +def test_published_confirmation_cas_is_repeatable_without_spending_budget(tmp_path, monkeypatch): + """A crash before native triage specification leaves one harmless rerun path.""" + state_db = tmp_path / "state.db" + connection = sqlite3.connect(state_db) + connection.execute( + "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)" + ) + connection.execute( + "INSERT INTO publication_retries VALUES(?,?,?,?,?,?,?,?,?)", + (release.BOARD, release.CHILD, release.SOURCE_RUN, 0, release.CONFIRM_RUN, "", 1, release.RETRY_AFTER, "9"), + ) + connection.commit() + connection.close() + + @contextmanager + def connect(_board): + connection = sqlite3.connect(state_db) + try: + yield connection + connection.commit() + finally: + connection.close() + + monkeypatch.setattr(release, "_pool_guard", lambda *_args: None) + monkeypatch.setattr(release, "_native_guard", lambda *_args: None) + monkeypatch.setattr(release, "_receipt_guard", lambda *_args: {}) + monkeypatch.setattr(release, "_published_remote_guard", lambda *_args: None) + monkeypatch.setattr(release.supervisor_state, "_connect", connect) + + assert release.confirm_published_release(tmp_path / "unused.db") is True + assert release.confirm_published_release(tmp_path / "unused.db") is False + connection = sqlite3.connect(state_db) + row = connection.execute("SELECT issued_run_id,reissue_count,retry_after,last_reissued_run_id FROM publication_retries").fetchone() + connection.close() + assert row == ("", 1, release.RETRY_AFTER, "9")