hermes: publish self-contained Git packs through the SCM broker
This commit is contained in:
jenkins
parent
82cc28b383
commit
33e6f84a64
@ -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
|
||||
|
||||
@ -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)
|
||||
|
||||
|
||||
|
||||
@ -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:
|
||||
|
||||
210
services/hermes/scripts/release_soteria_publication_retry.py
Executable file
210
services/hermes/scripts/release_soteria_publication_retry.py
Executable file
@ -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}")
|
||||
@ -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)
|
||||
scan.validate_receive_pack(io.BytesIO(b"zzzz"), TOKEN, FORMS)
|
||||
|
||||
@ -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"
|
||||
|
||||
191
testing/tests/test_release_soteria_publication_retry.py
Normal file
191
testing/tests/test_release_soteria_publication_retry.py
Normal file
@ -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
|
||||
Loading…
x
Reference in New Issue
Block a user