driver.py

#!/usr/bin/env python3
"""
machine-driver — the dumb, tireless driver. Part 2 of "The Machine."
(See: Obsidian Vault/Systems/The Machine — Conformance Spec v1.md)

The driver is the DUMBEST part of the system, on purpose. It spends no model
tokens and makes no judgments. It reads a goal, picks the next pending task,
fires ONE fresh worker (the only place judgment is spent), checks the result
against REALITY (not the worker's say-so), records the outcome, and loops —
surviving restarts via an atomic checkpoint. The thing that runs for days is
THIS loop; the model runs in short bursts, one per task.

Iteration 1 (2026-06-14) hardens the skeleton toward Conformance Spec v1 WITHOUT
adding a single model token to the loop — every addition below is deterministic:
  - Box 0  CONTRACT GATE — a move may not begin unless the goal carries a contract
           whose target was ratified by someone OTHER than its proposer
           (ratified_by != proposed_by). Missing / self-ratified -> forced propose-only.
           (The Council is the natural ratifier here — independent judgment on the
           TARGET. Ground-truth tests stay the Box 4 verifier; a model panel is not.)
  - Box 4  PROOF — TWO artifacts, split cleanly (iteration 2, 2026-06-14):
           driver-log.jsonl = the hash-chained loop AUDIT (dispatch/requeue/halt);
           aar/<task-id>.json = one canonical, Ed25519-SIGNED AAR per resolved task,
           signed via our own agentscontrolplane.org signer. The signed AAR is the
           proof layer (verifier ≠ subject ⇒ AAR L2); the log is telemetry. Runs
           keyless (skips AAR) when no signing identity is configured.
  - Box 5  GOVERNOR — a goal-level budget (worker-runs / wall-clock). Breach -> HALT
           + operator alert. THIS is the control the 131-duplicate incident lacked.
           AUTONOMY — effective = min(operator mode, contract ceiling, verifier trust);
           NO verifier -> trust 0 -> propose-only (the keystone gate).
  - ESCALATION — block / quarantine / budget-halt fire an operator alert (Telegram),
           never a silent stdout line. The incident's defining failure was "no alert."

The loop stays dumb: all of the above is deterministic plumbing. The only judgment
in the whole system is on the far side of the `worker_cmd` shell-out.

Usage:  python3 driver.py goal.json
Env (optional): TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID  -> operator alerts.
"""

from __future__ import annotations  # keep type hints lazy so this runs on Python 3.7+

import hashlib
import json
import os
import shlex
import subprocess
import sys
import time
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from pathlib import Path

GOAL_PATH = Path(sys.argv[1] if len(sys.argv) > 1 else "goal.json")
DRIVER_LOG_PATH = GOAL_PATH.with_name("driver-log.jsonl")  # hash-chained loop audit (was aar.jsonl)
AAR_DIR = GOAL_PATH.with_name("aar")                       # one canonical SIGNED AAR per resolved task
MUTATION_ONLY_PATH = "all mutation via the gate"
DEFAULT_OVERRIDE_EFFECT_SLO_SECONDS = 5
DEFAULT_ACK_SLO_SECONDS = 60
WORKFORCE_HEALTH_SCHEMA = "frontier.machine.health.v1"
WORKFORCE_HEALTH_LAYERS = ("process", "scheduler", "execution", "governance")
WORKFORCE_HEALTH_TOP_LEVEL_FIELDS = {"schema_version", "deployment_id", "checked_at", "layers", "aggregate_policy"}
WORKFORCE_PROPOSE_ONLY_REASONS = {"missing_verifier", "stale_verifier", "unratified_contract"}
WORKFORCE_HALTED_REASONS = {"active_override", "no_ack_halt"}
WORKFORCE_BLOCKED_REASONS = {"auth_failed", "credit_exhausted", "scheduler_stalled", "worker_unavailable"}
WORKFORCE_ALLOWED_REASON_CODES = (
    WORKFORCE_PROPOSE_ONLY_REASONS |
    WORKFORCE_HALTED_REASONS |
    WORKFORCE_BLOCKED_REASONS |
    {"governance_gate_failed"}
)


def now() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def load() -> dict:
    return json.loads(GOAL_PATH.read_text())


def save(state: dict) -> None:
    # Atomic write — a crash mid-write must never corrupt the goal. This is what
    # makes the loop resumable: kill it any time, restart, it picks up the
    # pending tasks. This is a provider- and harness-neutral checkpoint pattern.
    tmp = GOAL_PATH.with_suffix(".tmp")
    tmp.write_text(json.dumps(state, indent=2))
    tmp.replace(GOAL_PATH)


def run(cmd: str, cwd: str) -> tuple[int, str]:
    # One fresh process = one fresh context. The driver never reasons here;
    # it just shells out. The intelligence is on the other side of this call.
    p = subprocess.run(cmd, cwd=cwd, shell=True, capture_output=True, text=True)
    return p.returncode, (p.stdout + p.stderr)


def run_argv(argv: list[str], cwd: str) -> tuple[int, str]:
    p = subprocess.run(argv, cwd=cwd, capture_output=True, text=True)
    return p.returncode, (p.stdout + p.stderr)


def alert(msg: str) -> None:
    # Escalation's outbound half. The incident cooked a GPU because nothing alerted;
    # a block / quarantine / budget-halt must reach a human, not just stdout.
    print(f"[{now()}] ALERT: {msg.splitlines()[0]}")
    bot, chat = os.environ.get("TELEGRAM_BOT_TOKEN"), os.environ.get("TELEGRAM_CHAT_ID")
    if not (bot and chat):
        return  # degrade silently when unconfigured — the stdout line still fired
    try:
        data = urllib.parse.urlencode({"chat_id": chat, "text": f"🤖 machine-driver\n{msg}"}).encode()
        urllib.request.urlopen(
            urllib.request.Request(f"https://api.telegram.org/bot{bot}/sendMessage", data=data), timeout=15)
    except Exception as e:
        print(f"[{now()}]    (alert send failed: {e})")


def alert_with_ack(state: dict, msg: str, *, event: str = "alert", ack_slo_seconds: int | None = None) -> dict:
    """Raise an operator alert and persist the pending acknowledgement obligation.

    The stdout/Telegram alert is the transport. The durable `pending_acks` row is the service
    obligation: an operator must acknowledge the event inside the declared SLO or the monitor can
    escalate on the next pass.
    """
    acknowledgement = {
        "id": hashlib.sha256(f"{event}|{msg}|{now()}".encode()).hexdigest()[:16],
        "event": event,
        "status": "pending",
        "ack_required": True,
        "ack_slo_seconds": ack_slo_seconds or DEFAULT_ACK_SLO_SECONDS,
        "created_at": now(),
        "message_sha256": hashlib.sha256(msg.encode()).hexdigest(),
    }
    state.setdefault("pending_acks", []).append(acknowledgement)
    alert(msg)
    return acknowledgement


def log_append(record: dict) -> str:
    # Driver-loop AUDIT (driver-log.jsonl): an append-only, hash-chained line per transition.
    # Each line carries the previous line's hash, so a tampered/missing record breaks the
    # chain. This is loop telemetry (dispatch/requeue/halt), NOT the proof layer — the
    # canonical *signed* AAR per resolved task is emit_aar() below (agentscontrolplane.org).
    prev = ""
    if DRIVER_LOG_PATH.exists():
        lines = DRIVER_LOG_PATH.read_text().splitlines()
        if lines:
            prev = json.loads(lines[-1]).get("hash", "")
    record = {"ts": now(), "prev_hash": prev, **record}
    record["hash"] = hashlib.sha256((prev + json.dumps(record, sort_keys=True)).encode()).hexdigest()[:16]
    with DRIVER_LOG_PATH.open("a") as f:
        f.write(json.dumps(record) + "\n")
    return record["hash"]


def idem_key(goal_id: str, task: dict) -> str:
    if task.get("idempotency_key"):
        return str(task["idempotency_key"])
    # idempotency_key = hash(goal, task, normalized intent). The persistent local store
    # rejects duplicate ACTIVE/DONE keys before the next worker side effect can run.
    return hashlib.sha256(f"{goal_id}|{task['id']}|{task['goal']}".encode()).hexdigest()[:16]


def _cfg_path(p) -> Path:
    """aar_tool / aar_priv resolution: '~' expands; a relative path anchors to goal.json's
    directory (NOT the driver's cwd) — the goal file is the config, so it is the anchor.
    Regression: 2026-07-07 stack-demo run, a repo-root-relative key path hit ENOENT because
    resolution silently depended on subprocess cwd."""
    q = Path(p).expanduser()
    return q if q.is_absolute() else (GOAL_PATH.resolve().parent / q).resolve()


def _read_json(path: Path, default):
    try:
        return json.loads(path.read_text())
    except FileNotFoundError:
        return default


def _write_json(path: Path, value: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(json.dumps(value, indent=2))
    tmp.replace(path)


def idempotency_store_path(state: dict) -> Path:
    return _cfg_path(state.get("idempotency_store") or state.get("idempotency_store_path") or "idempotency-store.json")


def claim_idempotency(state: dict, task: dict, key: str) -> tuple[bool, str]:
    """Store-level idempotency: reject duplicate active/done keys before a worker side effect."""
    path = idempotency_store_path(state)
    store = _read_json(path, {"keys": {}})
    keys = store.setdefault("keys", {})
    existing = keys.get(key)
    if existing and existing.get("status") in ("ACTIVE", "DONE"):
        existing_status = existing.get("status")
        reason = f"reject duplicate idempotency key already exists: {existing_status}"
        store.setdefault("rejections", []).append({"ts": now(), "key": key, "task": task["id"], "reason": reason})
        _write_json(path, store)
        return False, reason
    keys[key] = {"status": "ACTIVE", "task": task["id"], "claimed_at": now()}
    _write_json(path, store)
    return True, "claimed"


def finish_idempotency(state: dict, key: str, status: str) -> None:
    path = idempotency_store_path(state)
    store = _read_json(path, {"keys": {}})
    rec = store.setdefault("keys", {}).setdefault(key, {})
    rec.update({"status": status, "finished_at": now()})
    _write_json(path, store)


def _operator_override_cfg(state: dict) -> dict:
    cfg = state.get("operator_override") or {}
    if isinstance(cfg, str):
        cfg = {"path": cfg}
    if not isinstance(cfg, dict):
        cfg = {}
    if state.get("operator_override_path"):
        cfg.setdefault("path", state["operator_override_path"])
    return cfg


def operator_override_active(state: dict) -> dict | None:
    cfg = _operator_override_cfg(state)
    override = dict(cfg) if cfg.get("active") else None
    if cfg.get("path"):
        path = _cfg_path(cfg["path"])
        file_override = _read_json(path, None)
        if isinstance(file_override, dict) and file_override.get("active"):
            override = {**cfg, **file_override, "path": str(path)}
    if not override:
        return None
    override.setdefault("action", "halt")
    override.setdefault("override_effect_slo_seconds", cfg.get("override_effect_slo_seconds", DEFAULT_OVERRIDE_EFFECT_SLO_SECONDS))
    return override


def _parse_ts(value: str | None):
    if not value:
        return None
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        return None


def _seconds_since(value: str | None) -> float | None:
    parsed = _parse_ts(value)
    if parsed is None:
        return None
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    return (datetime.now(timezone.utc) - parsed).total_seconds()


def record_operator_override_effect(state: dict, override: dict, stage: str) -> None:
    slo = int(override.get("override_effect_slo_seconds") or DEFAULT_OVERRIDE_EFFECT_SLO_SECONDS)
    effect_seconds = _seconds_since(override.get("requested_at"))
    effect_slo_met = True if effect_seconds is None else effect_seconds <= slo
    state["operator_override_effective_at"] = now()
    state["operator_override_stage"] = stage
    log_append({
        "event": "override_effective",
        "stage": stage,
        "action": override.get("action", "halt"),
        "override_effect_slo_seconds": slo,
        "effect_seconds": None if effect_seconds is None else round(effect_seconds, 3),
        "effect_slo_met": effect_slo_met,
    })
    alert_with_ack(
        state,
        f"OPERATOR OVERRIDE effective at {stage}: action={override.get('action', 'halt')}",
        event="override_effective",
        ack_slo_seconds=int(override.get("ack_slo_seconds") or DEFAULT_ACK_SLO_SECONDS),
    )


def operator_override_checkpoint(state: dict, stage: str) -> bool:
    override = operator_override_active(state)
    if not override:
        return False
    record_operator_override_effect(state, override, stage)
    action = str(override.get("action", "halt")).lower()
    if action in ("halt", "stop", "kill"):
        save(state)
        return True
    state["_operator_override_forced_mode"] = "propose"
    save(state)
    return False


def runtime_health_manifest(state: dict) -> dict:
    return state.get("runtime_health_manifest") or {
        "independent_monitor": True,
        "components": [
            {"id": "driver", "last_success_at_field": "last_success_at", "staleness_threshold_seconds": 300,
             "ack_slo_seconds": DEFAULT_ACK_SLO_SECONDS, "owner": "operator"},
            {"id": "monitor", "last_success_at_field": "last_monitor_at", "staleness_threshold_seconds": 300,
             "ack_slo_seconds": DEFAULT_ACK_SLO_SECONDS, "owner": "operator"},
        ],
        "detectors": ["staleness", "duplicate_key", "budget_thermal"],
    }


def run_runtime_anomaly_detector(state: dict, detector: str) -> list[dict]:
    anomalies: list[dict] = []
    if detector == "duplicate_key":
        store = _read_json(idempotency_store_path(state), {"rejections": []})
        for rejection in store.get("rejections", []):
            anomalies.append({"detector": detector, "kind": "duplicate_key", "detail": rejection.get("reason")})
    if detector == "budget_thermal":
        budget = state.get("budget") or {}
        if budget.get("max_worker_runs") and state.get("runs", 0) >= budget["max_worker_runs"]:
            anomalies.append({"detector": detector, "kind": "budget_thermal", "detail": "worker-run budget exhausted"})
    return anomalies


def assess_workforce_health(state: dict, manifest: dict) -> dict | None:
    """Evaluate the canonical ``frontier.machine.health.v1`` contract without model logic.

    The input shape and aggregation intentionally match the plugin's bundled schema/checker.
    Provider-specific probes remain outside this generic driver; they publish canonical checks.
    Local operator overrides and unacknowledged alerts are added as fail-closed halt controls.
    """
    cfg = manifest.get("workforce_health")
    if not cfg:
        return None  # backward-compatible: legacy manifests did not declare this contract
    if cfg is True:
        cfg = {}
    if not isinstance(cfg, dict):
        cfg = {}
    evidence_field = cfg.get("evidence_field", "workforce_health_evidence")
    contract = state.get(evidence_field)
    errors = []
    failures = []
    blockers = []
    propose_only = []
    halted = []
    degraded = []
    control_failures = []

    def parse_contract_timestamp(value, path):
        if not isinstance(value, str):
            errors.append(f"{path} must be an ISO timestamp string")
            return None
        parsed = _parse_ts(value)
        if parsed is None:
            errors.append(f"{path} must be a valid ISO timestamp")
            return None
        if parsed.tzinfo is None:
            parsed = parsed.replace(tzinfo=timezone.utc)
        return parsed

    checked_at_value = contract.get("checked_at") if isinstance(contract, dict) else None
    checked_at = parse_contract_timestamp(checked_at_value, "checked_at")
    if not isinstance(contract, dict):
        errors.append("contract must be a JSON object")
        contract = {}
    for field in contract:
        if field not in WORKFORCE_HEALTH_TOP_LEVEL_FIELDS:
            errors.append(f"unexpected top-level field {field}")
    if not isinstance(contract.get("deployment_id"), str) or not contract.get("deployment_id", "").strip():
        errors.append("deployment_id must be a non-empty string")
    if contract.get("schema_version") != WORKFORCE_HEALTH_SCHEMA:
        errors.append(f"schema_version must be {WORKFORCE_HEALTH_SCHEMA}")
    contract_layers = contract.get("layers")
    if not isinstance(contract_layers, dict):
        errors.append("layers must be an object")
        contract_layers = {}
    else:
        for layer_name in contract_layers:
            if layer_name not in WORKFORCE_HEALTH_LAYERS:
                errors.append(f"unexpected layer {layer_name}")
    aggregate_policy = contract.get("aggregate_policy")
    if "aggregate_policy" in contract:
        if not isinstance(aggregate_policy, dict):
            errors.append("aggregate_policy must be an object")
        else:
            if aggregate_policy.get("status") is not None and aggregate_policy.get("status") != "fail_closed":
                errors.append("aggregate_policy.status must be fail_closed")
            for field in ("rule", "warning"):
                if field in aggregate_policy and not isinstance(aggregate_policy.get(field), str):
                    errors.append(f"aggregate_policy.{field} must be a string")

    layer_results = {}
    for layer_name in WORKFORCE_HEALTH_LAYERS:
        layer = contract_layers.get(layer_name)
        layer_failures = []
        layer_hard_failure = False
        layer_degraded = False
        if not isinstance(layer, dict):
            layer_failures.append("missing layer")
            blockers.append(f"{layer_name}: missing layer")
            layer_results[layer_name] = {"status": "fail", "failures": layer_failures}
            continue
        checks = layer.get("checks")
        if not isinstance(checks, list) or not checks:
            layer_failures.append("no checks")
            blockers.append(f"{layer_name}: no checks")
            layer_results[layer_name] = {"status": "fail", "failures": layer_failures}
            continue
        for index, check in enumerate(checks):
            prefix = f"{layer_name}.checks[{index}]"

            def structural_failure(message):
                layer_failures.append(message)
                errors.append(message)

            if not isinstance(check, dict):
                structural_failure(f"{prefix} must be an object")
                continue
            check_id = check.get("id")
            if not isinstance(check_id, str) or not check_id.strip():
                structural_failure(f"{prefix}.id missing")
            check_status = check.get("status")
            if check_status not in ("pass", "fail", "unknown"):
                structural_failure(f"{prefix}.status must be pass, fail, or unknown")
            if "critical" in check and not isinstance(check.get("critical"), bool):
                structural_failure(f"{prefix}.critical must be a boolean")
            critical = check.get("critical") is not False
            reason_code = check.get("reason_code") if isinstance(check.get("reason_code"), str) else None
            if "reason_code" in check and (reason_code is None or reason_code not in WORKFORCE_ALLOWED_REASON_CODES):
                structural_failure(f"{prefix}.reason_code is not recognized")
            if "degradation_code" in check and not isinstance(check.get("degradation_code"), str):
                structural_failure(f"{prefix}.degradation_code must be a string")
            if "evidence" in check and not isinstance(check.get("evidence"), str):
                structural_failure(f"{prefix}.evidence must be a string")

            def classify(message):
                nonlocal layer_hard_failure, layer_degraded
                rendered = f"{layer_name}: {message}"
                if reason_code in WORKFORCE_HALTED_REASONS:
                    halted.append(f"{rendered} ({reason_code})")
                    layer_hard_failure = True
                elif reason_code in WORKFORCE_BLOCKED_REASONS:
                    blockers.append(f"{rendered} ({reason_code})")
                    layer_hard_failure = True
                elif reason_code in WORKFORCE_PROPOSE_ONLY_REASONS:
                    propose_only.append(f"{rendered} ({reason_code})")
                    layer_hard_failure = True
                elif critical:
                    blockers.append(f"{rendered}{f' ({reason_code})' if reason_code else ''}")
                    layer_hard_failure = True
                else:
                    degraded.append(rendered)
                    layer_degraded = True

            if check_status in ("fail", "unknown"):
                message = f"{check_id or prefix} status {check_status or 'missing'}"
                layer_failures.append(message)
                classify(message)

            observed_at = parse_contract_timestamp(check.get("observed_at"), f"{prefix}.observed_at")
            stale_after = check.get("stale_after_seconds")
            if not isinstance(stale_after, int) or isinstance(stale_after, bool) or stale_after < 1:
                structural_failure(f"{prefix}.stale_after_seconds must be a positive integer")
            elif checked_at is not None and observed_at is not None:
                age_seconds = int((checked_at - observed_at).total_seconds() // 1)
                if age_seconds < 0:
                    future_message = f"{check_id or prefix} observed_at is after checked_at"
                    layer_failures.append(future_message)
                    classify(future_message)
                if age_seconds > stale_after:
                    stale_message = f"{check_id or prefix} stale by {age_seconds - stale_after}s"
                    layer_failures.append(stale_message)
                    classify(stale_message)
            if not isinstance(check.get("summary"), str) or not check.get("summary", "").strip():
                structural_failure(f"{prefix}.summary missing")

        failures.extend(f"{layer_name}: {failure}" for failure in layer_failures)
        layer_results[layer_name] = {
            "status": "fail" if layer_hard_failure else "degraded" if layer_degraded else "pass" if not layer_failures else "fail",
            "failures": layer_failures,
        }

    # Local Machine controls strengthen the portable contract without changing its shape.
    pending_ack = any(
        not isinstance(ack, dict) or ack.get("status") == "pending"
        for ack in state.get("pending_acks", [])
    )
    active_override = operator_override_active(state) is not None
    if active_override:
        message = "local: operator override active (active_override)"
        halted.append(message)
        control_failures.append("active_override")
    if pending_ack:
        message = "local: operator acknowledgement pending (no_ack_halt)"
        halted.append(message)
        control_failures.append("no_ack_halt")

    status = (
        "invalid" if errors else
        "halted" if halted else
        "blocked" if blockers else
        "propose_only" if propose_only else
        "degraded" if degraded else
        "pass"
    )
    can_mutate = status in ("pass", "degraded")
    disposition = "allowed" if can_mutate else "blocked" if status == "invalid" else status
    report = {
        "schema_version": WORKFORCE_HEALTH_SCHEMA,
        "status": status,
        "aggregate": "pass" if status == "pass" else status,
        "disposition": disposition,
        "can_mutate": can_mutate,
        "deployment_id": contract.get("deployment_id"),
        "checked_at": checked_at_value,
        "evidence_field": evidence_field,
        "layers": layer_results,
        "errors": errors,
        "failures": failures,
        "blockers": blockers,
        "propose_only": propose_only,
        "halted": halted,
        "degraded": degraded,
        "control_failures": control_failures,
        "rule": "process, scheduler, execution, and governance must all pass fresh critical checks; can_mutate is true on pass or degraded",
        "evidence_sha256": hashlib.sha256(
            json.dumps(contract, sort_keys=True, separators=(",", ":")).encode()
        ).hexdigest(),
    }
    state["last_workforce_health"] = report
    return report


def monitor_health_once() -> int:
    state = load()
    manifest = runtime_health_manifest(state)
    state["last_monitor_at"] = now()
    problems = 0
    for component in manifest.get("components", []):
        field = component.get("last_success_at_field", "last_success_at")
        observed = state.get(field)
        if component.get("id") == "monitor":
            observed = state["last_monitor_at"]
        age = _seconds_since(observed)
        threshold = int(component.get("staleness_threshold_seconds", 300))
        if age is None or age > threshold:
            problems += 1
            log_append({
                "event": "runtime_stale",
                "component": component.get("id"),
                "staleness_threshold_seconds": threshold,
                "observed_age_seconds": None if age is None else round(age, 3),
            })
            alert_with_ack(
                state,
                f"RUNTIME HEALTH STALE — {component.get('id')} heartbeat exceeded {threshold}s",
                event="runtime_stale",
                ack_slo_seconds=int(component.get("ack_slo_seconds") or DEFAULT_ACK_SLO_SECONDS),
            )
    for detector in manifest.get("detectors", []):
        for anomaly in run_runtime_anomaly_detector(state, detector):
            problems += 1
            log_append({"event": "runtime_anomaly", **anomaly})
            alert_with_ack(state, f"RUNTIME HEALTH ANOMALY — {anomaly['kind']}", event="runtime_anomaly")
    workforce = assess_workforce_health(state, manifest)
    if workforce is not None:
        log_append({
            "event": "workforce_health",
            "status": workforce["status"],
            "disposition": workforce["disposition"],
            "can_mutate": workforce["can_mutate"],
            "control_failures": workforce["control_failures"],
            "evidence_field": workforce["evidence_field"],
        })
        if not workforce["can_mutate"]:
            problems += 1
            failed = ", ".join(
                workforce["halted"] + workforce["blockers"] + workforce["propose_only"] +
                workforce["errors"] + workforce["control_failures"]
            )
            alert_with_ack(
                state,
                f"WORKFORCE HEALTH FAILED — {failed}; process liveness cannot override downstream failure",
                event="workforce_health_failed",
                ack_slo_seconds=int((manifest.get("workforce_health") or {}).get(
                    "ack_slo_seconds", DEFAULT_ACK_SLO_SECONDS
                )) if isinstance(manifest.get("workforce_health"), dict) else DEFAULT_ACK_SLO_SECONDS,
            )
    save(state)
    return 2 if problems else 0


def _git_snapshot_field(repo: str, args: list[str], fmt: str, *, include_output: bool = False) -> dict:
    """Capture one deterministic git observation without making AAR emission fragile.

    A non-repository, missing git binary, or other inspection failure is evidence too: the
    signed record carries an explicit unavailable/error state instead of silently omitting
    the field or aborting an otherwise valid keyless/legacy driver run.
    """
    try:
        p = subprocess.run(
            ["git", "-C", repo, *args], capture_output=True, check=False,
        )
    except OSError as exc:
        return {
            "state": "unavailable",
            "format": fmt,
            "error": {"kind": type(exc).__name__, "message": str(exc)},
        }
    if p.returncode != 0:
        return {
            "state": "error",
            "format": fmt,
            "error": {
                "exit_code": p.returncode,
                "stderr_sha256": hashlib.sha256(p.stderr).hexdigest(),
            },
        }
    field = {
        "state": "available",
        "format": fmt,
        "sha256": hashlib.sha256(p.stdout).hexdigest(),
        "byte_length": len(p.stdout),
    }
    if include_output:
        field["value"] = p.stdout.decode("utf-8", errors="replace").strip()
    return field


def repository_state_snapshot(repo: str, verification_target: str) -> dict:
    """Return the signed repository identity corresponding to working-tree verification.

    Porcelain ``-z`` output avoids locale/presentation ambiguity. ``git diff HEAD`` binds
    both staged and unstaged tracked changes, while status additionally binds untracked
    path state. All byte hashes are full SHA-256 values.
    """
    return {
        "target": verification_target,
        "head": _git_snapshot_field(repo, ["rev-parse", "--verify", "HEAD"], "git-object-id", include_output=True),
        "status": _git_snapshot_field(
            repo, ["status", "--porcelain=v1", "-z", "--untracked-files=all"],
            "git-status-porcelain-v1-z",
        ),
        "diff": _git_snapshot_field(
            repo, ["diff", "--binary", "HEAD", "--"], "git-diff-binary-head",
        ),
    }


def emit_aar(task: dict, vcode: int, vout: str, repo: str, verify_cmd, aar_cfg: dict):
    """Box-4 PROOF: one canonical, Ed25519-signed AAR per RESOLVED task — our own standard
    (agentscontrolplane.org), so the driver MUST emit it. Deterministic plumbing; no model
    token. If the signing identity/key isn't configured the driver runs KEYLESS: skip the AAR
    (driver-log.jsonl still records the loop). The verify step already gives `verifier ≠ subject`.
    """
    tool, priv = aar_cfg.get("aar_tool"), aar_cfg.get("aar_priv")
    subject, principal = aar_cfg.get("subject"), aar_cfg.get("principal")
    if not (tool and priv and subject and principal):
        return None  # keyless — no signing identity; loop audit still written
    tool, priv = _cfg_path(tool), _cfg_path(priv)
    AAR_DIR.mkdir(parents=True, exist_ok=True)
    verified = vcode == 0
    verification_target = aar_cfg.get("verification_target") or "working_tree"
    target_label = {
        "working_tree": "repository working tree",
        "head": "repository HEAD",
    }.get(verification_target, verification_target)
    source = f"{repo}#working-tree" if verification_target == "working_tree" else f"{repo}#{verification_target}"
    rec = {
        "aar": "0.02",
        "subject": subject,                                  # the worker — did:web:<org>:machine-driver
        "principal": principal,                              # the signing org — did:web:<org> (= sig.by)
        "task": {"id": task["id"], "claim": task["goal"]},
        "verdict": "verified" if verified else "rejected",
        "ground_truth": "confirmed" if verified else "contradicted",
        "reason": f"verify_cmd exited {vcode} against {target_label}",
        "checks": [{
            "source": source,
            "query": verify_cmd or "(no verify_cmd)",
            "observed_at": now(),
            "response_sha256": hashlib.sha256((vout or "").encode()).hexdigest(),
            "excerpt": (vout or "")[-300:],
        }],
        # verifier ≠ subject ⇒ AAR L2 (structural independence; same org ⇒ disclosed, not audit-grade)
        "verifier": {"id": aar_cfg.get("verifier") or f"{principal}:verifier", "independence": "same_principal"},
        "repository_state": repository_state_snapshot(repo, verification_target),
        "issued": now(),
    }
    if aar_cfg.get("evidence_binding") is not None:
        # JSON round-trip makes the signed record an immutable value-copy of the goal-level
        # binding rather than retaining an in-process reference to caller-owned data.
        rec["evidence_binding"] = json.loads(json.dumps(aar_cfg["evidence_binding"]))
    out = AAR_DIR / f"{task['id']}.json"
    out.write_text(json.dumps(rec, indent=2))
    # Pre-check the resolved paths so a config mistake alerts with the FULL path, not a
    # cryptic ENOENT from node. The unsigned record stays on disk either way (sign later).
    missing = [str(p) for p in (tool, priv) if not p.exists()]
    if missing:
        alert(f"AAR sign SKIPPED for task {task['id']}: missing {', '.join(missing)} — record left unsigned at {out}")
        return str(out)
    # sign with OUR OWN signer (eat our own cooking) — driver stays pure-Python, shells to node.
    rc, sout = run(
        f"node {shlex.quote(str(tool))} sign {shlex.quote(str(out))} --priv {shlex.quote(str(priv))}",
        str(GOAL_PATH.resolve().parent),
    )
    if rc != 0:
        alert(f"AAR sign FAILED for task {task['id']} (rc {rc}): {sout.strip().splitlines()[-1] if sout.strip() else '?'}")
    return str(out)


def next_pending(state: dict) -> dict | None:
    for t in state["tasks"]:
        if t.get("status", "pending") == "pending":
            return t
    return None


def contract_ratified(contract: dict) -> bool:
    # Box 0: a contract is INDEPENDENTLY ratified iff it exists and a THIRD party ratified
    # its target — ratified_by is present and is not the proposer. Missing / self-ratified
    # ⇒ not ratified. This is the predicate the durable-mutation hard-deny gates on.
    if not contract:
        return False
    return bool(contract.get("ratified_by")) and contract.get("ratified_by") != contract.get("proposed_by")


def effective_mode(state: dict, task: dict | None = None) -> tuple[str, str]:
    # Box 5 keystone (deterministic; no model token). The autonomy TIER granted is the
    # contract's autonomy_ceiling; the operator dial (commit/propose) and verifier trust
    # are binary admissions. Commit proceeds iff: operator asked commit AND a verifier is
    # present (trust 1) AND the granted ceiling clears the tier THIS action REQUIRES.
    # NO verifier (no verify_cmd) => trust 0 => propose-only, no matter what's asked.
    # autonomy_ceiling scale: 0=propose · 1+=commit-allowed (full dial is later).
    #
    # Δ1 (vNext Box 5) — reversibility is a TERM INSIDE the evaluated gate, not commentary:
    # a reversible action clears the baseline commit tier (>=1); an IRREVERSIBLE action must
    # clear the higher bar contract.irreversible_min_trust. A missing per-task `reversible`
    # defaults to False (treat as irreversible); a missing irreversible_min_trust defaults
    # to the STRICTEST reading (deny autonomous irreversible commit until the contract opts
    # in by raising the ceiling to meet it).
    asked = state.get("mode", "propose")
    override = operator_override_active(state)
    if override or state.get("_operator_override_forced_mode") == "propose":
        action = (override or {}).get("action", "propose")
        return "propose", f"operator_override active ({action})"
    contract = state.get("contract") or {}
    ceiling = contract.get("autonomy_ceiling", 0)
    trust = 1 if state.get("verify_cmd") else 0
    reversible = bool(task.get("reversible", False)) if task is not None else False
    if reversible:
        required = 1                                  # baseline commit tier
    else:
        irr = contract.get("irreversible_min_trust")  # higher bar for irreversible actions
        required = irr if isinstance(irr, int) and not isinstance(irr, bool) and irr >= 1 else float("inf")
    if asked == "commit" and trust >= 1 and ceiling >= required:
        note = "reversible" if reversible else f"irreversible · ceiling {ceiling}>={required}"
        return "commit", f"operator=commit · {note} · verifier present"
    if not trust:
        why = "no verifier -> trust 0"
    elif asked != "commit":
        why = "operator asked propose"
    elif ceiling < 1:
        why = "contract ceiling forbids commit"
    elif not reversible and ceiling < required:
        req = "strictest" if required == float("inf") else required
        why = f"irreversible action needs ceiling>={req} (have {ceiling})"
    else:
        why = "operator asked propose"
    return "propose", why


def apply_mutation(repo: str, task: dict, mode: str, state: dict) -> bool:
    """THE single named path to a durable mutation — every git commit routes through here,
    so "all mutation via the gate" is ONE auditable chokepoint (non-bypassable). It
    RE-CHECKS the Box-5 gate at the mutation site (defense in depth, not just where the
    decision was first taken) and enforces the Box-0 contract before committing. Returns
    True iff it committed; False if it REFUSED (caller leaves the diff in the working tree).

    Acknowledged residue, OUT OF SCOPE here: the pre-gate `worker_cmd` shell-out in main()'s
    WORKER step already edits the working tree BEFORE this gate runs; that pre-gate tree
    mutation is not yet routed through this chokepoint.
    """
    if mode != "commit":
        return False
    if operator_override_active(state):
        alert_with_ack(state, f"Operator override DENIED commit of {task['id']}.", event="gate_denied")
        log_append({"event": "commit_denied", "task": task["id"], "reason": "operator_override"})
        return False
    contract = state.get("contract") or {}
    goal_id = state.get("goal", "goal")
    # Re-evaluate the Box-5 gate HERE, at the mutation site (non-bypassable choke).
    gate_mode, _ = effective_mode(state, task)
    if gate_mode != "commit":
        alert_with_ack(state, f"Box-5 gate re-check at mutation site DENIED commit of {task['id']} on {goal_id!r}.",
                       event="gate_denied")
        log_append({"event": "commit_denied", "task": task["id"], "reason": "gate_recheck"})
        return False
    # Box 0 HARD-DENY (not a silent downgrade): a durable mutation requires an independently
    # ratified contract. Refuse loudly; do NOT quietly behave as propose.
    if not contract_ratified(contract):
        alert_with_ack(state, f"Box-0 HARD-DENY: refused durable mutation (commit) of {task['id']} on {goal_id!r} "
                       f"— contract not independently ratified "
                       f"(ratified_by={contract.get('ratified_by')!r}, proposed_by={contract.get('proposed_by')!r}).",
                       event="gate_denied")
        log_append({"event": "commit_denied", "task": task["id"], "reason": "contract_not_ratified"})
        return False
    add_code, add_out = run_argv(["git", "add", "-A"], repo)
    if add_code != 0:
        task["commit_error"] = add_out[-600:]
        alert_with_ack(state, f"git add failed for task {task['id']} on {goal_id!r}; commit not recorded.",
                       event="commit_failed")
        log_append({"event": "commit_failed", "task": task["id"], "phase": "git_add", "exit": add_code})
        return False
    commit_code, commit_out = run_argv(["git", "commit", "-m", f"driver: {task['id']}"], repo)
    if commit_code != 0:
        task["commit_error"] = commit_out[-600:]
        alert_with_ack(state, f"git commit failed for task {task['id']} on {goal_id!r}; commit not recorded.",
                       event="commit_failed")
        log_append({"event": "commit_failed", "task": task["id"], "phase": "git_commit", "exit": commit_code})
        return False
    log_append({"event": "commit", "task": task["id"]})
    return True


def main() -> int:
    state = load()
    repo = state["repo"]
    goal_id = state.get("goal", "goal")
    worker_cmd = state["worker_cmd"]
    verify_cmd = state.get("verify_cmd")
    max_attempts = state.get("max_attempts", 3)
    tick = state.get("tick_seconds", 1)
    budget = state.get("budget", {})
    contract = state.get("contract") or {}
    aar_cfg = {k: state.get(k) for k in (
        "aar_tool", "aar_priv", "subject", "principal", "verifier", "verification_target",
        "evidence_binding",
    )}

    # --- Box 0 gate: a DURABLE MUTATION may not proceed without a contract whose ratifier
    #     is not its proposer. Enforcement is a HARD-DENY at the mutation chokepoint
    #     (apply_mutation), NOT a silent downgrade. We surface the condition early here;
    #     a propose-mode run performs no durable mutation, so it proceeds normally even
    #     with no/unratified contract (the keyless / propose path stays live). ---
    proposed_by, ratified_by = contract.get("proposed_by"), contract.get("ratified_by")
    if not contract_ratified(contract):
        alert(f"Box-0 contract NOT independently ratified on {goal_id!r} "
              f"(ratified_by={ratified_by!r}, proposed_by={proposed_by!r}) — any durable mutation "
              f"(commit) will be HARD-DENIED until a third party ratifies; propose-mode still runs.")

    # Banner reflects the gate for the FIRST pending task (reversibility is per-task, so a
    # single goal-level line would mislead); the loop re-decides per task authoritatively.
    mode, why = effective_mode(state, next_pending(state))
    print(f"[{now()}] driver up — goal: {goal_id!r}  mode={mode} ({why})")
    log_append({"event": "driver_up", "goal": goal_id, "mode": mode,
                "ratified_by": ratified_by, "proposed_by": proposed_by})

    runs = state.get("runs", 0)
    t0 = time.monotonic()

    while True:
        # --- Box 5 governor: the control the 131-duplicate incident lacked. ---
        if budget.get("max_worker_runs") and runs >= budget["max_worker_runs"]:
            msg = f"BUDGET HALT — {runs} worker-runs hit max_worker_runs={budget['max_worker_runs']} on {goal_id!r}."
            log_append({"event": "budget_halt", "runs": runs})
            alert_with_ack(state, msg, event="budget_halt")
            save(state)
            print(f"[{now()}] {msg}")
            return 2
        if budget.get("max_wall_seconds") and (time.monotonic() - t0) >= budget["max_wall_seconds"]:
            msg = f"BUDGET HALT — wall-clock hit max_wall_seconds={budget['max_wall_seconds']} on {goal_id!r}."
            log_append({"event": "budget_halt", "wall_s": round(time.monotonic() - t0)})
            alert_with_ack(state, msg, event="budget_halt")
            save(state)
            print(f"[{now()}] {msg}")
            return 2
        if operator_override_checkpoint(state, "before_dispatch"):
            print(f"[{now()}] OPERATOR OVERRIDE HALT — no further worker dispatch.")
            return 3

        task = next_pending(state)
        if task is None:
            blocked = [t for t in state["tasks"] if t.get("status") == "blocked"]
            print(f"[{now()}] GOAL COMPLETE — {len(state['tasks'])} tasks, {len(blocked)} blocked/surfaced.")
            log_append({"event": "goal_complete", "tasks": len(state["tasks"]), "blocked": len(blocked)})
            return 0

        task["attempts"] = task.get("attempts", 0) + 1
        ikey = idem_key(goal_id, task)
        claimed, idem_reason = claim_idempotency(state, task, ikey)
        if not claimed:
            task["status"] = "done" if "DONE" in idem_reason else "blocked"
            task["reason"] = idem_reason
            log_append({"event": "duplicate_rejected", "task": task["id"], "idempotency_key": ikey, "reason": idem_reason})
            if task["status"] == "blocked":
                alert_with_ack(state, f"Duplicate ACTIVE idempotency key blocked task {task['id']}: {ikey}",
                               event="duplicate_rejected")
            save(state)
            continue
        task["status"] = "running"
        task["started"] = now()
        save(state)
        print(f"[{now()}] -> {task['id']} (attempt {task['attempts']}): {task['goal']}")
        log_append({"event": "dispatch", "task": task["id"], "attempt": task["attempts"], "idempotency_key": ikey})

        # 3. WORKER — fresh process; the only judgment in the whole loop.
        run(worker_cmd.format(task=task["goal"]), repo)
        runs += 1
        state["runs"] = runs
        if operator_override_checkpoint(state, "after_worker"):
            finish_idempotency(state, ikey, "HALTED")
            print(f"[{now()}] OPERATOR OVERRIDE HALT — worker result held before verification/mutation.")
            return 3

        # 4. VERIFY — ground truth. We trust the exit code, not a 'done' claim.
        #    No verifier => Box 4 trust 0 => cannot pass (fail-closed).
        if not verify_cmd:
            vcode, vout = 1, "(no verify_cmd — Box 4 trust 0; cannot verify against reality)"
        else:
            vcode, vout = run(verify_cmd, repo)
        task["verify_tail"] = vout[-600:]

        if vcode == 0:
            task["status"] = "done"
            task["done"] = now()
            state["last_success_at"] = now()           # observability heartbeat
            finish_idempotency(state, ikey, "DONE")
            # Δ1: evaluate the gate PER-TASK — reversibility is a per-task term, so the
            # autonomy decision can differ task to task even within one goal.
            task_mode, task_why = effective_mode(state, task)
            print(f"[{now()}]    ok  verified.")
            log_append({"event": "verified", "task": task["id"], "verify_exit": 0,
                        "verifier_id": verify_cmd, "idempotency_key": ikey, "autonomy_used": task_mode})
            emit_aar(task, vcode, vout, repo, verify_cmd, aar_cfg)   # signed AAR — verified/confirmed
            # 5. AUTONOMY — propose leaves the diff for review; commit proceeds, but ONLY
            #    through the single apply_mutation chokepoint (re-checks gate + Box 0).
            if task_mode == "commit":
                if apply_mutation(repo, task, task_mode, state):
                    print(f"[{now()}]    committed.")
                else:
                    if task.get("commit_error"):
                        task["status"] = "blocked"
                        task["reason"] = "commit_failed"
                        finish_idempotency(state, ikey, "BLOCKED")
                    print(f"[{now()}]    commit not recorded (gate denied or git failed) — diff left in the working tree.")
            else:
                print(f"[{now()}]    (propose mode — {task_why}; diff left for your review)")
        elif task["attempts"] >= max_attempts:
            task["status"] = "blocked"                 # terminal quarantine — surfaced, not re-queued
            task["reason"] = f"verify failed {max_attempts}x"
            finish_idempotency(state, ikey, "BLOCKED")
            log_append({"event": "quarantine", "task": task["id"], "verify_exit": vcode, "idempotency_key": ikey})
            alert_with_ack(state, f"task {task['id']} QUARANTINED on {goal_id!r} after {max_attempts} failed verifies — surfaced, not skipped.",
                           event="quarantine")
            emit_aar(task, vcode, vout, repo, verify_cmd, aar_cfg)   # signed AAR — rejected/contradicted
            print(f"[{now()}]    BLOCKED after {max_attempts} attempts — surfaced + alerted.")
        else:
            task["status"] = "pending"   # re-queue: a fresh worker, fresh context, tries again
            finish_idempotency(state, ikey, "RETRYABLE")
            log_append({"event": "requeue", "task": task["id"], "verify_exit": vcode})
            print(f"[{now()}]    verify failed — re-queueing for a fresh attempt.")

        save(state)
        time.sleep(tick)


if __name__ == "__main__":
    if len(sys.argv) > 2 and sys.argv[2] == "--monitor-once":
        raise SystemExit(monitor_health_once())
    raise SystemExit(main())