diff --git a/services/hermes/kustomization.yaml b/services/hermes/kustomization.yaml index c1ca53cf..c4685cb9 100644 --- a/services/hermes/kustomization.yaml +++ b/services/hermes/kustomization.yaml @@ -112,6 +112,7 @@ configMapGenerator: - supervisor_lineage.py=scripts/supervisor_lineage.py - supervisor_state.py=scripts/supervisor_state.py - publication_retry.py=scripts/publication_retry.py + - publication_retry_recovery.py=scripts/publication_retry_recovery.py - scm_resume_bootstrap.py=scripts/scm_resume_bootstrap.py - bootstrap_soteria_publication_retry.py=scripts/bootstrap_soteria_publication_retry.py - seed_legacy_scm_roots.py=scripts/seed_legacy_scm_roots.py diff --git a/services/hermes/scripts/bootstrap_soteria_publication_retry.py b/services/hermes/scripts/bootstrap_soteria_publication_retry.py index e9d580b3..8d00d71b 100644 --- a/services/hermes/scripts/bootstrap_soteria_publication_retry.py +++ b/services/hermes/scripts/bootstrap_soteria_publication_retry.py @@ -14,6 +14,7 @@ from typing import Any import supervisor_state from execution_pool_protocol import derive_ordinal_key, payload_digest, read_key, sign_envelope +from publication_retry_recovery import normalize BOARD = "soteria" @@ -118,6 +119,52 @@ def _native_guard() -> supervisor_state.Lineage: return child["lineage"] +def _run9_guard(pool_database: Path) -> None: + """Require run 9's exact blocked retry before its one bounded release.""" + from hermes_cli import kanban_db + + source = sqlite3.connect(f"file:{pool_database}?mode=ro", uri=True) + try: + row = source.execute( + "SELECT payload_json,result_json,state,worker_ordinal FROM assignments " + "WHERE board=? AND task_id=? AND run_id=?", (BOARD, CHILD, "9") + ).fetchone() + later = source.execute( + "SELECT 1 FROM assignments WHERE board=? AND task_id=? AND run_id<>? " + "AND state IN ('assigned','running','result')", (BOARD, CHILD, "9") + ).fetchone() + finally: + source.close() + try: + payload, result = (json.loads(row[0]), json.loads(row[1])) if row else ({}, {}) + except (TypeError, json.JSONDecodeError) as error: + raise ValueError("run 9 terminal evidence is malformed") from error + structured = result.get("structured") if isinstance(result, dict) else None + if ( + row is None or row[2:] != ("finalized", ORDINAL) or later is not None + or not isinstance(payload, dict) or not isinstance(payload.get("scm_resume"), dict) + or not isinstance(structured, dict) or structured.get("status") != "blocked" + or result.get("scm_submission") is not None + ): + raise ValueError("run 9 is not the exact blocked publication retry") + 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) + 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 latest is None or tuple(latest) != (9, "blocked", "blocked") + or not isinstance(parents, (list, tuple, set)) or ROOT not in {str(value) for value in parents} + ): + raise ValueError("native task is not the exact blocked publication retry") + + def bootstrap(receipt_path: Path, pool_database: Path) -> str: """Persist one receipt only after the independent native and pool guards agree.""" receipt = _receipt(receipt_path) @@ -140,10 +187,20 @@ def bootstrap(receipt_path: Path, pool_database: Path) -> str: return raw_sha -def signed_assignment(pool_database: Path, key_file: Path) -> bytes: - """Attest the verified retained assignment with the current coordinator key.""" +def normalize_reissue(receipt_path: Path, pool_database: Path) -> str: + """Normalize the rejected metadata then consume run 9's one retry budget.""" + receipt = _receipt(receipt_path) + _assignment, raw_sha, _attempt = _pool_record(pool_database) + _run9_guard(pool_database) + sealed_sha = normalize(BOARD, CHILD, "9", raw_sha, receipt) + if not supervisor_state.reissue_publication_retry(BOARD, CHILD, "9"): + raise ValueError("publication retry budget is unavailable") + return sealed_sha + + +def _signed_assignment(pool_database: Path, key_file: Path) -> bytes: + """Sign the exact retained assignment with ordinal zero's authority.""" assignment, _raw_sha, attempt = _pool_record(pool_database) - _native_guard() key = derive_ordinal_key(read_key(key_file), ORDINAL) return json.dumps(sign_envelope(key, "assignment", { "board": BOARD, "task_id": CHILD, "run_id": RUN_ID, @@ -151,20 +208,42 @@ def signed_assignment(pool_database: Path, key_file: Path) -> bytes: }, assignment), separators=(",", ":"), sort_keys=True).encode() +def signed_assignment(pool_database: Path, key_file: Path) -> bytes: + """Attest run 8 only while it remains the native terminal run.""" + _native_guard() + return _signed_assignment(pool_database, key_file) + + +def signed_normalized_assignment(pool_database: Path, key_file: Path) -> bytes: + """Re-attest run 8 only after the exact run-9 metadata block.""" + _run9_guard(pool_database) + return _signed_assignment(pool_database, key_file) + + def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--receipt-file", type=Path) + parser.add_argument("--normalize-reissue-receipt-file", type=Path) parser.add_argument("--pool-db", type=Path, default=Path(os.environ.get("HERMES_HOME", "/opt/data")) / "execution-pool/assignments.db") parser.add_argument("--emit-assignment", action="store_true") + parser.add_argument("--emit-normalized-assignment", action="store_true") parser.add_argument("--key-file", type=Path, default=Path(os.environ.get("HERMES_EXECUTION_POOL_KEY_FILE", "/runtime-access/execution-pool-key"))) args = parser.parse_args() - if args.emit_assignment == (args.receipt_file is not None): + selected = sum((args.emit_assignment, args.emit_normalized_assignment, args.receipt_file is not None, args.normalize_reissue_receipt_file is not None)) + if selected != 1: raise SystemExit("select exactly one bootstrap action") if args.emit_assignment: print(signed_assignment(args.pool_db, args.key_file).decode()) return 0 - bootstrap(args.receipt_file, args.pool_db) - print(f"bootstrapped {BOARD}/{CHILD} run={RUN_ID} head={HEAD[:12]}") + if args.emit_normalized_assignment: + print(signed_normalized_assignment(args.pool_db, args.key_file).decode()) + return 0 + if args.receipt_file: + bootstrap(args.receipt_file, args.pool_db) + print(f"bootstrapped {BOARD}/{CHILD} run={RUN_ID} head={HEAD[:12]}") + return 0 + sealed_sha = normalize_reissue(args.normalize_reissue_receipt_file, args.pool_db) + print(f"normalized {BOARD}/{CHILD} run=9 sealed_sha={sealed_sha[:12]}") return 0 diff --git a/services/hermes/scripts/execution_pool_client.py b/services/hermes/scripts/execution_pool_client.py index 99bc322e..f2af1cd1 100644 --- a/services/hermes/scripts/execution_pool_client.py +++ b/services/hermes/scripts/execution_pool_client.py @@ -4,6 +4,7 @@ from __future__ import annotations import json +import logging import os import threading import urllib.error @@ -40,6 +41,64 @@ PORT = int(os.environ.get("HERMES_EXECUTION_CLIENT_PORT", "9009")) RESULT_FIELDS = frozenset( {"status", "summary", "changed_files", "tests_run", "artifacts", "findings", "blockers"} ) +LOG = logging.getLogger(__name__) +COORDINATOR_REJECTION_CATEGORIES = { + "assignment": ( + "assignment is unknown or stale", "worker ordinal does not own this assignment", + "assignment attempt is stale", + ), + "lease": ("assignment lease expired", "Kanban run no longer owns this worker"), + "state": ("assignment is no longer running", "assignment cannot accept a result"), + "delivery": ("delivery identifier was reused", "conflicting result for completed delivery"), +} + + +class CoordinatorRejected(ProtocolError): + """A coordinator conflict reduced to one non-sensitive fixed category.""" + + def __init__(self, category: str): + self.category = category + super().__init__(f"coordinator rejected request: {category}") + + +def _coordinator_rejection(error: urllib.error.HTTPError) -> CoordinatorRejected: + """Read one bounded coordinator conflict body and retain no free-form detail.""" + try: + raw = error.read(MAX_WIRE_BYTES + 1) + value = json.loads(raw) if len(raw) <= MAX_WIRE_BYTES else {} + detail = value.get("error") if isinstance(value, dict) else None + except (OSError, UnicodeDecodeError, json.JSONDecodeError): + detail = None + for category, messages in COORDINATOR_REJECTION_CATEGORIES.items(): + if detail in messages: + return CoordinatorRejected(category) + return CoordinatorRejected("policy") + + +def _rejection_category(operation: str, error: Exception) -> str: + """Classify local mediator rejections without logging request or error text.""" + if isinstance(error, CoordinatorRejected): + return f"coordinator-{error.category}" + if isinstance(error, urllib.error.URLError): + return "transport" + if isinstance(error, OSError): + return "transport" + if not isinstance(error, ProtocolError): + return "input" + if operation != "resume": + return "policy" + message = str(error) + if message.startswith("SCM resume source") or message.startswith("SCM resume artifact"): + return "source" + if message.startswith("SCM resume evidence") or message.startswith("SCM resume metadata"): + return "evidence" + if message.startswith("preserved SCM workspace") or message.startswith("preserved SCM baseline"): + return "workspace" + if "lease" in message: + return "lease" + if message.startswith("SCM resume publication"): + return "scm" + return "policy" def _binding(value: dict[str, Any]) -> dict[str, Any]: @@ -85,8 +144,13 @@ class ClientBoundary: method="POST", headers={"Content-Type": "application/json", "Cache-Control": "no-store"}, ) - with urllib.request.urlopen(request, timeout=60) as response: - body = response.read(MAX_WIRE_BYTES + 1) + try: + with urllib.request.urlopen(request, timeout=60) as response: + body = response.read(MAX_WIRE_BYTES + 1) + except urllib.error.HTTPError as error: + if error.code == 409: + raise _coordinator_rejection(error) from error + raise if len(body) > MAX_WIRE_BYTES: raise ProtocolError("coordinator response exceeds the wire limit") return verify_envelope(self.key, json.loads(body)) @@ -153,7 +217,10 @@ class ClientBoundary: def checkpoint() -> None: if failures: - raise ProtocolError("SCM publication lease was lost") from failures[0] + error = failures[0] + if isinstance(error, CoordinatorRejected): + raise error + raise ProtocolError("SCM publication lease was lost") from error return checkpoint, lambda: (stop.set(), thread.join(timeout=1)) @@ -307,6 +374,7 @@ def handler_factory(boundary: ClientBoundary) -> type[BaseHTTPRequestHandler]: ) def do_POST(self) -> None: # noqa: N802 + operation = "unknown" try: length = int(self.headers.get("Content-Length", "0")) request = ( @@ -317,17 +385,23 @@ def handler_factory(boundary: ClientBoundary) -> type[BaseHTTPRequestHandler]: allowed = {"operation", "binding", "payload", "title", "body"} if not request or set(request) - allowed: raise ProtocolError("invalid local mediator request") - operation = str(request.get("operation") or "") + requested = str(request.get("operation") or "") routes = { "poll": boundary.poll, "heartbeat": lambda: boundary.heartbeat(request), "resume": lambda: boundary.resume(request), "finish": lambda: boundary.finish(request), } - if operation not in routes: + if requested not in routes: raise ProtocolError("unsupported local mediator operation") + operation = requested self._reply(200, routes[operation]()) except (ProtocolError, OSError, ValueError, urllib.error.URLError) as error: + LOG.warning( + "mediator_rejected operation=%s category=%s", + operation, + _rejection_category(operation, error), + ) self._reply(409, {"error": str(error)[:2000]}) def log_message(self, _format: str, *_arguments: Any) -> None: diff --git a/services/hermes/scripts/execution_pool_worker.py b/services/hermes/scripts/execution_pool_worker.py index c6a2aa03..30d08ccd 100644 --- a/services/hermes/scripts/execution_pool_worker.py +++ b/services/hermes/scripts/execution_pool_worker.py @@ -381,7 +381,7 @@ def execute(assignment: dict[str, Any]) -> None: "finish", binding=binding, payload=result_payload, - title=truncate_utf8(str(structured.get("summary") or f"Hermes task {binding['task_id']}"), 512), + title=truncate_utf8(str(structured.get("summary") or f"Hermes task {binding['task_id']}")[:240], 507), body=json.dumps(structured, indent=2, sort_keys=True)[:12000], ) if not response.get("ack", {}).get("accepted"): diff --git a/services/hermes/scripts/publication_retry_recovery.py b/services/hermes/scripts/publication_retry_recovery.py new file mode 100644 index 00000000..1397f09e --- /dev/null +++ b/services/hermes/scripts/publication_retry_recovery.py @@ -0,0 +1,84 @@ +#!/usr/bin/env python3 +"""Fail-closed one-time normalization of a mediator publication receipt.""" + +from __future__ import annotations + +import hashlib +import json +from typing import Any + +from execution_pool_protocol import truncate_utf8 +from publication_retry import receipt_digest +import supervisor_state + + +def canonical_title(summary: Any) -> str: + """Apply the broker's character and UTF-8-byte title limits.""" + if not isinstance(summary, str) or not (text := summary.strip()): + raise ValueError("publication retry summary is invalid") + return truncate_utf8(text[:240], 507) + + +def normalize( + board: str, child_task_id: str, run_id: str, raw_result_sha256: str, receipt: Any, +) -> str: + """Replace only a rejected title on the exact pending mediator receipt.""" + if not isinstance(receipt, dict) or not isinstance(raw_result_sha256, str) or len(raw_result_sha256) != 64: + raise ValueError("publication retry normalization input is invalid") + with supervisor_state._connect(board) as connection: + row = connection.execute( + "SELECT receipt_json,issued_run_id,resolved_run_id,reissue_count,last_reissued_run_id FROM publication_retries " + "WHERE board=? AND child_task_id=?", (board, child_task_id) + ).fetchone() + provenance = connection.execute( + "SELECT raw_result_sha256,reconstruction FROM publication_retry_provenance " + "WHERE board=? AND child_task_id=?", (board, child_task_id) + ).fetchone() + if row is None or provenance is None or provenance[0] != raw_result_sha256: + raise ValueError("publication retry provenance is unavailable") + pending = (run_id, "", 0, "") + released = ("", "", 1, run_id) + if row[1:] not in {pending, released}: + raise ValueError("publication retry is not the exact pending run") + try: + old = json.loads(row[0]) + except (TypeError, json.JSONDecodeError) as error: + raise ValueError("sealed publication receipt is malformed") from error + binding = old.get("source") if isinstance(old, dict) else None + if not isinstance(binding, dict): + raise ValueError("sealed publication receipt is malformed") + old, _ordinal, _source_run = supervisor_state._retry_receipt( + board, child_task_id, binding, old, path=None + ) + new, _ordinal, _source_run = supervisor_state._retry_receipt( + board, child_task_id, binding, receipt, path=None + ) + unchanged = ("source", "baseline_sha", "head", "body", "structured") + if any(old[name] != new[name] for name in unchanged): + raise ValueError("publication retry evidence changed during normalization") + expected = canonical_title(old["structured"].get("summary")) + if new["title"] != expected or new["result_digest"] != receipt_digest( + new["structured"], new["title"], new["body"] + ): + raise ValueError("publication retry title normalization is invalid") + encoded = json.dumps(new, separators=(",", ":"), sort_keys=True) + prefix = "normalized-title-from-sealed-receipt-sha256:" + if row[0] == encoded: + sealed_sha = provenance[1][len(prefix):] if isinstance(provenance[1], str) else "" + if not isinstance(provenance[1], str) or not provenance[1].startswith(prefix) or len(sealed_sha) != 64: + raise ValueError("publication retry normalization provenance conflicts") + return sealed_sha + sealed_sha = hashlib.sha256(row[0].encode()).hexdigest() + marker = prefix + sealed_sha + changed = connection.execute( + "UPDATE publication_retries SET receipt_json=? WHERE board=? AND child_task_id=? " + "AND receipt_json=? AND issued_run_id=? AND resolved_run_id='' AND reissue_count=0", + (encoded, board, child_task_id, row[0], run_id), + ).rowcount + if changed != 1: + raise ValueError("publication retry changed during normalization") + connection.execute( + "UPDATE publication_retry_provenance SET reconstruction=? WHERE board=? AND child_task_id=?", + (marker, board, child_task_id), + ) + return sealed_sha diff --git a/services/hermes/scripts/scm_resume_bootstrap.py b/services/hermes/scripts/scm_resume_bootstrap.py index 21f43666..b44c46d7 100755 --- a/services/hermes/scripts/scm_resume_bootstrap.py +++ b/services/hermes/scripts/scm_resume_bootstrap.py @@ -88,7 +88,7 @@ def bootstrap(key: bytes, raw_assignment: Any) -> dict[str, Any]: boundary = Boundary(key) assignment = boundary.verify(raw_assignment) structured = _completed_result(assignment) - title = truncate_utf8(structured["summary"].strip(), 512) + title = truncate_utf8(structured["summary"].strip()[:240], 507) body = json.dumps(structured, indent=2, sort_keys=True) return boundary.resume_artifact(assignment, {"title": title, "body": body}, structured) diff --git a/testing/tests/test_hermes_execution_pool_mediator.py b/testing/tests/test_hermes_execution_pool_mediator.py index 30ab7386..ac61459d 100644 --- a/testing/tests/test_hermes_execution_pool_mediator.py +++ b/testing/tests/test_hermes_execution_pool_mediator.py @@ -2,6 +2,7 @@ from __future__ import annotations +import io import json import subprocess import sys @@ -221,6 +222,43 @@ def http_request(handler, path, *, body=None): server.server_close() +@pytest.mark.parametrize( + ("detail", "category"), + [ + ("assignment lease expired", "coordinator-lease"), + ("assignment is no longer running", "coordinator-state"), + ("unrecognized coordinator error", "coordinator-policy"), + ], +) +def test_client_reduces_coordinator_conflicts_to_fixed_categories(monkeypatch, detail, category): + """A coordinator conflict never propagates its response detail to worker logs.""" + error = urllib.error.HTTPError( + "http://coordinator", 409, "Conflict", {}, io.BytesIO(json.dumps({"error": detail}).encode()) + ) + monkeypatch.setattr(client.urllib.request, "urlopen", lambda *_args, **_kwargs: (_ for _ in ()).throw(error)) + with pytest.raises(client.CoordinatorRejected) as rejected: + client.ClientBoundary(KEY)._post("/v1/heartbeat", assignment()) + assert f"coordinator-{rejected.value.category}" == category + + +def test_mediator_logs_only_fixed_rejection_operation_and_category(caplog): + """Resume failures retain an actionable category without request or error text.""" + class Boundary: + poll = staticmethod(lambda: {"assignment": None}) + heartbeat = staticmethod(lambda _request: {}) + finish = staticmethod(lambda _request: {}) + + @staticmethod + def resume(_request): + raise client.CoordinatorRejected("lease") + + body = protocol.canonical_json({"operation": "resume", "binding": binding(), "payload": {}}) + with caplog.at_level("WARNING"): + status, response = http_request(client.handler_factory(Boundary()), "/v1/client", body=body) + assert status == 409 and response == {"error": "coordinator rejected request: lease"} + assert "mediator_rejected operation=resume category=coordinator-lease" in caplog.text + + def test_model_api_exposes_only_gated_state_machine_operations(): class Boundary: poll = staticmethod(lambda: {"assignment": None}) diff --git a/testing/tests/test_hermes_execution_pool_submission.py b/testing/tests/test_hermes_execution_pool_submission.py index 4804bae5..bfb52c68 100644 --- a/testing/tests/test_hermes_execution_pool_submission.py +++ b/testing/tests/test_hermes_execution_pool_submission.py @@ -15,6 +15,7 @@ SCM_SCRIPTS = ROOT / "services/hermes/scm-common/scripts" sys.path[:0] = [str(SCRIPTS), str(SCM_SCRIPTS)] import execution_pool_client as client # noqa: E402 +import gitea_api_policy as gitea_policy # noqa: E402 import execution_pool_protocol as protocol # noqa: E402 import execution_pool_scm as scm # noqa: E402 import scm_resume_bootstrap as resume_bootstrap # noqa: E402 @@ -479,10 +480,36 @@ def test_mediator_bootstrap_truncates_multibyte_title_at_the_byte_limit(tmp_path resume_bootstrap.bootstrap(KEY, exact) - assert len(seen["request"]["title"].encode()) <= 512 + assert len(seen["request"]["title"]) <= 240 + assert len(seen["request"]["title"].encode()) <= 507 + assert gitea_policy._draft_title(seen["request"]["title"]) == "WIP: " + seen["request"]["title"] assert seen["structured"]["summary"] == completed["summary"] +@pytest.mark.parametrize("summary", ["x" * 300, "界" * 300, "Ж" * 300]) +def test_mediator_bootstrap_titles_are_accepted_by_gitea_draft_policy(tmp_path, monkeypatch, summary): + """Resume titles leave room for Gitea's draft prefix in both limits.""" + exact = protocol.sign_envelope( + KEY, "assignment", binding(), payload(root_task_id="t_deadbeef", continuation_kind="repair") + ) + log = tmp_path / "session-state" / "metis" / "t_deadbeef" / "42.log" + log.parent.mkdir(parents=True) + completed = {**RESULT, "summary": summary} + log.write_bytes(protocol.canonical_json({"structured_output": completed}) + b"\n") + monkeypatch.setattr(resume_bootstrap, "ROOT", tmp_path) + seen = {} + monkeypatch.setattr( + scm.Boundary, "resume_artifact", + lambda _self, _assignment, request, _structured: seen.update(request=request) or {}, + ) + + resume_bootstrap.bootstrap(KEY, exact) + + title = seen["request"]["title"] + assert len(title) <= 240 and len(title.encode()) <= 507 + assert gitea_policy._draft_title(title) == f"WIP: {title}" + + def test_mediator_bootstrap_rejects_a_symlinked_log_ancestor(tmp_path, monkeypatch): exact = protocol.sign_envelope(KEY, "assignment", binding(), payload()) outside = tmp_path / "outside" @@ -512,6 +539,20 @@ def test_publication_lease_signs_the_exact_binding_and_stops_cleanly(): assert {name: verified[name] for name in binding()} == binding() +def test_publication_lease_preserves_a_fixed_coordinator_conflict_category(): + """An initial coordinator conflict remains observable at the mediator boundary.""" + boundary = client.ClientBoundary(KEY, FailingSCM(protocol.ProtocolError("unused"))) + boundary._post = lambda *_args: (_ for _ in ()).throw(client.CoordinatorRejected("lease")) + + checkpoint, close = boundary._publication_lease(binding()) + try: + with pytest.raises(client.CoordinatorRejected, match="lease") as rejected: + checkpoint() + assert rejected.value.category == "lease" + finally: + close() + + def test_publication_lease_renews_during_a_slow_scm_step_without_sleeping(monkeypatch): boundary = client.ClientBoundary(KEY, FailingSCM(protocol.ProtocolError("unused"))) calls = [] diff --git a/testing/tests/test_hermes_execution_pool_worker_execute.py b/testing/tests/test_hermes_execution_pool_worker_execute.py index 05b5add4..ac79f94f 100644 --- a/testing/tests/test_hermes_execution_pool_worker_execute.py +++ b/testing/tests/test_hermes_execution_pool_worker_execute.py @@ -11,11 +11,13 @@ import pytest ROOT = Path(__file__).parents[2] SCRIPTS = ROOT / "services/hermes/scripts" -sys.path.insert(0, str(SCRIPTS)) +SCM_SCRIPTS = ROOT / "services/hermes/scm-common/scripts" +sys.path[:0] = [str(SCRIPTS), str(SCM_SCRIPTS)] from cli_lane_config import ProcessResult, Route # noqa: E402 import execution_pool_protocol as protocol # noqa: E402 import execution_pool_worker as worker # noqa: E402 +import gitea_api_policy as gitea_policy # noqa: E402 from testing.tests.test_hermes_execution_pool_worker_v2 import assignment # noqa: E402 @@ -120,6 +122,23 @@ def test_execute_completed_clean_result_refreshes_and_finishes_exact_run( assert state["terminal_at"] > 0 and state["baseline_sha"] == "a" * 40 +@pytest.mark.parametrize("summary", ["x" * 300, "界" * 300, "Ж" * 300]) +def test_execute_titles_are_accepted_by_gitea_draft_policy(tmp_path, monkeypatch, summary): + """Ordinary worker publication preserves Gitea's character and byte limits.""" + _exact, _workspace, calls, _providers, _routes, _refreshed = prepare_execute( + tmp_path, monkeypatch, results=[completed_result(structured={ + "status": "completed", "summary": summary, "changed_files": [], + "tests_run": [], "artifacts": [], "findings": [], "blockers": [], + })] + ) + + worker.execute(_exact) + + title = calls[-1][1]["title"] + assert len(title) <= 240 and len(title.encode()) <= 507 + assert gitea_policy._draft_title(title) == "WIP: " + title + + def test_execute_capacity_fallback_changes_provider_and_preserves_handoff( tmp_path, monkeypatch ): diff --git a/testing/tests/test_hermes_legacy_scm_roots.py b/testing/tests/test_hermes_legacy_scm_roots.py index 09f86ee0..c8924881 100644 --- a/testing/tests/test_hermes_legacy_scm_roots.py +++ b/testing/tests/test_hermes_legacy_scm_roots.py @@ -20,6 +20,7 @@ seed = _load("seed_legacy_scm_roots") retry = sys.modules["publication_retry"] bootstrap = _load("bootstrap_soteria_publication_retry") protocol = sys.modules["execution_pool_protocol"] +recovery = _load("publication_retry_recovery") class NativeKanban: @@ -158,6 +159,98 @@ def test_publication_bootstrap_assignment_uses_ordinal_authority(tmp_path, monke protocol.derive_ordinal_key(master, 1), envelope, expected_kind="assignment" ) + seen = [] + monkeypatch.setattr(bootstrap, "_run9_guard", lambda _pool: seen.append("run9")) + normalized = json.loads(bootstrap.signed_normalized_assignment(tmp_path / "pool.db", key_file)) + assert seen == ["run9"] + assert protocol.verify_envelope(derived, normalized, expected_kind="assignment")["payload"] == assignment + + +def test_publication_retry_normalization_preserves_all_evidence_except_title(tmp_path, monkeypatch): + """The broker-cap correction is fenced to run 9 and one sealed receipt.""" + board, root_id, child_id, baseline, head = ( + "soteria", "t_root", "t_child", "a" * 40, "b" * 40 + ) + lineage = state.Lineage(root_id, "hermes-repair/cache", "https://scm.bstein.dev/titan/soteria/pulls/3", "soteria", "main") + monkeypatch.setattr(state, "KANBAN_ROOT", tmp_path / "boards") + state.record_submission(board, root_id, lineage, baseline) + state.record_child(board, child_id, root_id, root_id, "repair", baseline, "replace cache literals") + structured = { + "status": "completed", "summary": "界" * 300, + "changed_files": ["internal/k8s/job_manifests.go"], "tests_run": [], + "artifacts": [], "findings": [], "blockers": [], + } + source = { + "board": board, "task_id": child_id, "run_id": "8", "worker_ordinal": 0, + "attempt": 1, "root_task_id": root_id, "repo_url": "https://scm.bstein.dev/titan/soteria.git", + "branch": lineage.branch, "base_branch": "main", + } + old_title, body = "界" * 170, json.dumps(structured, sort_keys=True) + old = {"source": source, "baseline_sha": baseline, "head": head, "title": old_title, "body": body, + "structured": structured, "result_digest": retry.receipt_digest(structured, old_title, body)} + binding = {name: source[name] for name in ("board", "task_id", "run_id", "worker_ordinal", "attempt")} + state.record_publication_retry(board, child_id, binding, old) + state.record_publication_retry_provenance(board, child_id, "c" * 64) + state.issue_publication_retry(board, child_id, "9") + title = recovery.canonical_title(structured["summary"]) + fixed = {**old, "title": title, "result_digest": retry.receipt_digest(structured, title, body)} + + sealed = recovery.normalize(board, child_id, "9", "c" * 64, fixed) + assert len(title) == 169 and len(title.encode()) == 507 and len(sealed) == 64 + assert state.publication_retry(board, child_id, "9")["title"] == title + assert recovery.normalize(board, child_id, "9", "c" * 64, fixed) == sealed + + changed = {**fixed, "body": body + "!"} + changed["result_digest"] = retry.receipt_digest(structured, title, changed["body"]) + with pytest.raises(ValueError, match="evidence changed"): + recovery.normalize(board, child_id, "9", "c" * 64, changed) + + assert state.reissue_publication_retry(board, child_id, "9") is True + assert recovery.normalize(board, child_id, "9", "c" * 64, fixed) == sealed + monkeypatch.setattr(bootstrap, "BOARD", board) + monkeypatch.setattr(bootstrap, "CHILD", child_id) + monkeypatch.setattr(bootstrap, "_receipt", lambda _path: fixed) + monkeypatch.setattr(bootstrap, "_pool_record", lambda _path: ({}, "c" * 64, 1)) + monkeypatch.setattr(bootstrap, "_run9_guard", lambda _path: None) + assert bootstrap.normalize_reissue(tmp_path / "receipt.json", tmp_path / "pool.db") == sealed + assert bootstrap.normalize_reissue(tmp_path / "receipt.json", tmp_path / "pool.db") == sealed + + +def test_normalized_assignment_accepts_real_shaped_native_run_row(tmp_path, monkeypatch): + """The one-time run-9 guard handles sqlite Row values without weakening it.""" + database = tmp_path / "pool.db" + with __import__("sqlite3").connect(database) as connection: + connection.execute( + "CREATE TABLE assignments(board,task_id,run_id,payload_json,result_json,state,worker_ordinal)" + ) + connection.execute( + "INSERT INTO assignments VALUES(?,?,?,?,?,?,?)", ( + bootstrap.BOARD, bootstrap.CHILD, "9", json.dumps({"scm_resume": {}}), + json.dumps({"structured": {"status": "blocked"}, "scm_submission": None}), + "finalized", bootstrap.ORDINAL, + ) + ) + + class NativeRow: + def __iter__(self): + return iter((9, "blocked", "blocked")) + + class Connection: + def execute(self, _sql, _args): + return SimpleNamespace(fetchone=lambda: NativeRow()) + + def close(self): + return None + + kanban = SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=lambda **_kwargs: Connection(), + get_task=lambda _connection, _task: {"status": "blocked", "current_run_id": None}, + parent_ids=lambda _connection, _task: [bootstrap.ROOT], + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=kanban)) + bootstrap._run9_guard(database) + def test_publication_retry_is_bound_once_and_never_falls_back_to_a_model(tmp_path, monkeypatch): """A mediator receipt can power one fresh ordinal-pinned publication only."""