atlas-iac/scripts/ops/hermes_handoff_acceptance.py

#!/usr/bin/env python3
"""Fail-closed acceptance harness for the Hermes full handoff.

Read-only by default. The harness collects evidence from two genuinely separate
places — an external operator vantage and the Hermes agent probing itself from
inside its own pod — classifies every check PASS / FAIL / NOT_RUN /
NOT_APPLICABLE, and emits machine-readable JSON alongside a short human summary.

Any mandatory FAIL or NOT_RUN is a NO_GO, and so is a harness-level problem: an
unreachable vantage, a catalog whose evidence no longer exists, an expired
deadline, or two vantages that turn out to be the same principal. Nothing here
rounds an absence of evidence up to a pass.

    scripts/ops/hermes_handoff_acceptance.py --help

The release runbook lists every required freshness and lineage input.

Mutation lives behind a separate arming flag, needs an exact confirmation
phrase and a unique ref, refuses protected push targets before any network
call, and verifies its own cleanup. A default run reports those checks NOT_RUN.
"""

from __future__ import annotations

import argparse
import datetime as dt
import json
import sys
from collections.abc import Sequence
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

from hermes_handoff_catalog import Targets  # noqa: E402
from hermes_handoff_ephemeral import (  # noqa: E402
    CONFIRMATION,
    ArmingError,
    ArmRequest,
    preflight,
    run_armed,
)
from hermes_handoff_exec import (  # noqa: E402
    DEFAULT_COMMAND_TIMEOUT,
    DEFAULT_RUN_DEADLINE,
    Runner,
)
from hermes_handoff_model import GO, Report, unscreened_fields, utc_now  # noqa: E402
from hermes_handoff_policy import (  # noqa: E402
    ARMED,
    EXPECTED_BASE,
    EXPECTED_REMOTE,
    EXPECTED_REPO,
    READ_ONLY,
)
from hermes_handoff_redaction import DEFAULT_MAX_BYTES  # noqa: E402
from hermes_handoff_run import (  # noqa: E402
    build_catalog,
    build_report,
    resolve_vantages,
    unavailable_report,
    validate_catalog,
    validate_targets,
)

EXIT_GO = 0
EXIT_NO_GO = 1
EXIT_USAGE = 2


def dependency_head(value: str) -> tuple[int, str]:
    """Parse one `PR=SHA` dependency-head binding."""
    try:
        number, sha = value.split("=", 1)
        parsed = int(number)
    except (TypeError, ValueError) as exc:
        raise argparse.ArgumentTypeError(
            "dependency head must be PR=40-character-SHA"
        ) from exc
    if (
        parsed <= 0
        or len(sha) != 40
        or any(character not in "0123456789abcdef" for character in sha)
    ):
        raise argparse.ArgumentTypeError(
            "dependency head must be PR=lowercase-40-character-SHA"
        )
    return (parsed, sha)


def build_parser() -> argparse.ArgumentParser:
    """Return the command-line interface for the harness."""
    parser = argparse.ArgumentParser(
        prog="hermes_handoff_acceptance",
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("--namespace", default="hermes", help="Hermes namespace.")
    parser.add_argument(
        "--context", help="Kubernetes context for the operator vantage."
    )
    parser.add_argument("--kubeconfig", help="Kubeconfig for the operator vantage.")
    parser.add_argument(
        "--output", type=Path, help="Write the JSON report here instead of stdout."
    )
    parser.add_argument(
        "--json-only", action="store_true", help="Suppress the human summary."
    )
    parser.add_argument(
        "--command-timeout",
        type=float,
        default=DEFAULT_COMMAND_TIMEOUT,
        help="Per-command seconds.",
    )
    parser.add_argument(
        "--deadline",
        type=float,
        default=DEFAULT_RUN_DEADLINE,
        help="Whole-run seconds.",
    )
    parser.add_argument(
        "--max-output-bytes",
        type=int,
        default=DEFAULT_MAX_BYTES,
        help="Per-capture byte budget.",
    )
    parser.add_argument(
        "--baseline-commit",
        default=Targets.baseline_commit,
        help="Commit main must descend from.",
    )
    parser.add_argument(
        "--dependency-pr",
        type=int,
        action="append",
        default=None,
        help="Pull request that must be merged before this release is certifiable (repeatable).",
    )
    parser.add_argument(
        "--dependency-head",
        type=dependency_head,
        action="append",
        default=None,
        metavar="PR=SHA",
        help="Exact current dependency PR head (repeat for every dependency).",
    )
    parser.add_argument(
        "--remote-main-sha", default="", help="Exact observed origin/main SHA."
    )
    parser.add_argument(
        "--reviewed-head-sha", default="", help="Exact reviewed PR #19 head SHA."
    )
    parser.add_argument(
        "--agent-image",
        default="",
        help="Exact agent :git-<source>-build-<n>@sha256 identity.",
    )
    parser.add_argument(
        "--build-sha", default="", help="Exact source SHA used for the image build."
    )
    parser.add_argument(
        "--deployment-revision", default="", help="Exact running Deployment revision."
    )
    parser.add_argument(
        "--expected-suspension",
        action="append",
        default=None,
        metavar="NAMESPACE/NAME",
        help="Flux object allowed to be suspended (repeatable).",
    )
    parser.add_argument(
        "--node-count",
        type=int,
        default=3,
        help="Nodes the fleet must cover (fixed at 3).",
    )
    parser.add_argument(
        "--pool-replicas", type=int, default=3, help="Expected pool worker count."
    )
    parser.add_argument(
        "--chat-config-revision",
        default="",
        help="Expected chat pod-template config revision.",
    )
    parser.add_argument(
        "--chat-ordinal",
        type=int,
        default=0,
        help="Chat tenant ordinal that carries the Telegram identity.",
    )
    parser.add_argument(
        "--pool-worker-env",
        action="append",
        default=None,
        help="Environment variable the pool workers must declare (repeatable).",
    )
    parser.add_argument(
        "--max-evidence-age",
        type=int,
        default=86_400,
        help="Seconds before provider and routing evidence is treated as stale.",
    )
    armed = parser.add_argument_group("ephemeral mutation mode")
    armed.add_argument(
        "--arm-ephemeral-push",
        action="store_true",
        help="Permit one ephemeral branch push and draft pull request, then remove both.",
    )
    armed.add_argument(
        "--confirm", default="", help=f"Exact confirmation phrase: {CONFIRMATION!r}."
    )
    armed.add_argument(
        "--ephemeral-token", default="", help="Unique 8-64 character ref suffix."
    )
    return parser


def targets_from_args(arguments: argparse.Namespace, now: dt.datetime) -> Targets:
    """Return the catalog targets a parsed command line describes."""
    defaults = Targets()
    return Targets(
        namespace=arguments.namespace,
        baseline_commit=arguments.baseline_commit,
        dependency_pull_requests=tuple(
            arguments.dependency_pr or defaults.dependency_pull_requests
        ),
        dependency_heads=tuple(arguments.dependency_head or ()),
        remote_main_sha=arguments.remote_main_sha,
        reviewed_head_sha=arguments.reviewed_head_sha,
        agent_image=arguments.agent_image,
        build_sha=arguments.build_sha,
        deployment_revision=arguments.deployment_revision,
        expected_suspensions=tuple(arguments.expected_suspension or ()),
        node_count=arguments.node_count,
        pool_replicas=arguments.pool_replicas,
        chat_config_revision=arguments.chat_config_revision,
        chat_ordinal=arguments.chat_ordinal,
        pool_worker_env=tuple(arguments.pool_worker_env or defaults.pool_worker_env),
        max_evidence_age_seconds=arguments.max_evidence_age,
        now=now,
    )


def emit(report: Report, arguments: argparse.Namespace, stream=None) -> list[str]:
    """Write the report, refusing to publish one that still looks credential-shaped."""
    payload = report.as_dict()
    offenders = unscreened_fields(payload)
    if offenders:
        payload = {
            "harness": payload["harness"],
            "decision": "NO_GO",
            "harness_errors": [
                "the rendered report failed its own credential screening at "
                + ", ".join(offenders[:8])
            ],
        }
    rendered = json.dumps(payload, indent=2, sort_keys=True, default=str)
    if arguments.output:
        arguments.output.parent.mkdir(parents=True, exist_ok=True)
        arguments.output.write_text(rendered + "\n", encoding="utf-8")
    else:
        print(rendered, file=stream or sys.stdout)
    if not arguments.json_only:
        print(report.render_summary(), file=sys.stderr)
    return offenders


def _armed_results(arguments: argparse.Namespace, runner: Runner, vantages) -> list:
    request = ArmRequest(
        repo=EXPECTED_REPO,
        remote=EXPECTED_REMOTE,
        token=arguments.ephemeral_token,
        confirmation=arguments.confirm,
        base=EXPECTED_BASE,
        expected_head=arguments.reviewed_head_sha,
        expected_base_sha=arguments.remote_main_sha,
    )
    preflight(request, EXPECTED_REPO, Path.cwd())
    vantage = vantages["operator"]
    return run_armed(runner, vantage, request)


def main(argv: Sequence[str] | None = None) -> int:
    """Run the acceptance sweep and return the process exit status."""
    arguments = build_parser().parse_args(argv)
    now = dt.datetime.now(dt.timezone.utc)
    started_at = utc_now(lambda: now)
    targets = targets_from_args(arguments, now)
    mode = ARMED if arguments.arm_ephemeral_push else READ_ONLY

    specs = build_catalog(targets)
    startup_problems = [*validate_targets(targets), *validate_catalog(specs)]
    if arguments.arm_ephemeral_push:
        try:
            preflight(
                ArmRequest(
                    repo=EXPECTED_REPO,
                    remote=EXPECTED_REMOTE,
                    token=arguments.ephemeral_token,
                    confirmation=arguments.confirm,
                    base=EXPECTED_BASE,
                    expected_head=arguments.reviewed_head_sha,
                    expected_base_sha=arguments.remote_main_sha,
                ),
                EXPECTED_REPO,
                Path.cwd(),
            )
        except ArmingError as exc:
            startup_problems.append(f"arming refused: {exc}")
    if startup_problems:
        report = unavailable_report(
            mode, started_at, "; ".join(startup_problems), targets
        )
        emit(report, arguments)
        return EXIT_NO_GO

    try:
        runner = Runner(
            mode=mode,
            command_timeout=arguments.command_timeout,
            deadline_seconds=arguments.deadline,
            max_bytes=arguments.max_output_bytes,
        )
    except ValueError as exc:
        report = unavailable_report(
            mode, started_at, f"invalid execution bound: {exc}", targets
        )
        emit(report, arguments)
        return EXIT_USAGE
    vantages, records = resolve_vantages(
        runner, targets, arguments.kubeconfig, arguments.context
    )
    report = build_report(runner, targets, specs, vantages, records, mode, started_at)
    if arguments.arm_ephemeral_push:
        if report.decision != GO:
            report.harness_errors.append(
                "armed mutation was not started because read-only acceptance did not GO"
            )
        else:
            armed_results = _armed_results(arguments, runner, vantages)
            armed_ids = {result.spec.id for result in armed_results}
            report.results = [
                result for result in report.results if result.spec.id not in armed_ids
            ]
            report.results.extend(armed_results)
            report.results.sort(key=lambda result: result.spec.id)
            report.finished_at = utc_now()
    offenders = emit(report, arguments)
    if offenders:
        return EXIT_NO_GO
    return EXIT_GO if report.decision == GO else EXIT_NO_GO


if __name__ == "__main__":  # pragma: no cover - exercised through main()
    raise SystemExit(main())