acdl/core/hitl_gates.py

"""HITL pre-execution attestation gates (REQ-108, D-084).

Records the approver identity (`gitea.actor` / `github.actor`) to the
DynamoDB outbox for the contractId (attribute `approver_qa` /
`approver_prod` / `approver_dr`), runs the separation-of-duties check on
prod, invokes the 8-concern attestation matrix for the target env, and
returns (ok, reason). Dev skips (autonomous). `scripts/run_platform.sh`
calls `attest` before apply for qa/prod/dr.
"""

import os
import sys
from typing import Optional, Tuple

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from core.metrics.event_envelope import make_event, append_event
from core.metrics.decision_ledger import append as ledger_append


def _approver_attr(env: str) -> str:
    return {"qa": "approver_qa", "prod": "approver_prod", "dr": "approver_dr"}.get(env, "")


def attest(contract_id: str, env: str, approver: str,
           evidence: Optional[dict] = None,
           outbox_client=None) -> Tuple[bool, str]:
    """Attest a promotion gate for the given environment.

    Args:
        contract_id: the contract UUID.
        env: dev/qa/prod/dr.
        approver: the approver's username (`gitea.actor` / `github.actor`).
        evidence: optional operator-supplied evidence artifacts (for the
            attestation matrix operator-supplied concerns).
        outbox_client: optional moto-mocked DynamoDB outbox client for tests.

    Returns:
        (ok, reason). ok=False means block the promotion.
    """
    if env == "dev":
        return (True, "dev autonomous (no HITL gate)")

    if not approver:
        return (False, f"no approver identity for {env} (GITHUB_ACTOR/GITEA_ACTOR unset)")

    attr = _approver_attr(env)
    if not attr:
        return (False, f"unknown environment: {env}")

    # Record the approver to the outbox.
    if outbox_client is not None:
        outbox_client.put_approver(contract_id, attr, approver)

    # Run the separation-of-duties check on prod.
    if env == "prod":
        from core.separation_of_duties import check as sod_check, route_halt_artifact
        ok, reason = sod_check(outbox_client, contract_id, approver)
        if not ok:
            route_halt_artifact(contract_id, reason, oncall_client=None)
            return (False, reason)

    # Run the 8-concern attestation matrix.
    from core.attestation_matrix import check as matrix_check
    ok, reason = matrix_check(env, evidence or {})
    if not ok:
        return (False, reason)

    # Emit attestation.recorded event to the Decision Ledger (D-132).
    try:
        run_id = os.environ.get("NOVA_RUN_ID", f"attest-{contract_id[:8]}")
        attestation_data = {
            "approver": approver,
            "environment": env,
            "concerns": reason,
            "result": "pass",
            "contract_id": contract_id,
        }
        attestation_event = make_event("nova.attestation.recorded", run_id, env, attestation_data,
                                         contract_id=contract_id, actor_type="human-attestation",
                                         actor_id=approver)
        append_event(attestation_event)
        ledger_append(attestation_event)
    except Exception:
        pass  # metrics emission must never break the attestation gate

    return (True, f"{env} attested by {approver}")


def approver_from_env() -> Optional[str]:
    """Read the approver identity from the environment."""
    return os.environ.get("GITHUB_ACTOR") or os.environ.get("GITEA_ACTOR")


if __name__ == "__main__":
    # CLI: hitl_gates.py <contract_id> <env> [evidence.json]
    if len(sys.argv) < 3:
        print("usage: hitl_gates.py <contract_id> <env> [evidence.json]", file=sys.stderr)
        sys.exit(2)
    _cid = sys.argv[1]
    _env = sys.argv[2]
    _evidence = {}
    if len(sys.argv) >= 4 and os.path.isfile(sys.argv[3]):
        import json
        with open(sys.argv[3]) as f:
            _evidence = json.load(f)
    _approver = approver_from_env() or ""
    ok, reason = attest(_cid, _env, _approver, _evidence)
    if ok:
        print(f"HITL PASS: {reason}")
        sys.exit(0)
    else:
        print(f"HITL BLOCK: {reason}", file=sys.stderr)
        sys.exit(1)