"""E72 cross-host shared mission state over PostgreSQL.

This is the E72 answer to the E71 BLOCKED gate "cross-host shared state". It adds a transactional shared-state
backend (PostgreSQL) and a coordinator/node protocol that runs on physically separate hosts. It does not create a
new authority path: every dispatched operation still carries an Ed25519-signed grant and a single-use ticket using
the same CAIN domains as E68 (`cain66.child.grant.v1`, `cain68.op.ticket.v1`). PostgreSQL stores mission state,
leases, fencing tokens and idempotent results. Consensus is not claimed: this is a single PostgreSQL primary, so
the database is a single point of failure and the cluster is *not* Byzantine fault tolerant.

Measured properties that this module can actually demonstrate:
  * two coordinator processes on two hosts, one elected by a PostgreSQL lease with a monotonic fencing token
  * a worker on host B executing a task while a different worker on host A independently recomputes it
  * lease expiry and reclaim, with a higher fencing token rejecting a stale result
  * idempotent result submission (same task/attempt/kind inserted once)
  * failover: kill the leader, the other coordinator acquires leadership and continues the same mission
  * reboot: pending leases are reclaimed after the lease deadline; expired grants are refused after recovery
"""
from __future__ import annotations

import hashlib
import json
import os
import threading
import time
import urllib.error
import urllib.request
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Dict, List, Mapping, Optional, Sequence

try:  # pg8000 is a pure-python driver; it is imported lazily so offline verifiers do not need it
    import pg8000.dbapi as pgdb  # type: ignore
except Exception:  # noqa: BLE001
    pgdb = None  # type: ignore

from .. import canon as C
from .. import ops

SCHEMA = "cain72.crosshost.v1"
TICKET_DEFAULT_TTL = 45.0
GRANT_DEFAULT_TTL = 900.0


class SharedStateError(RuntimeError):
    pass


def _utc(ts: Optional[float] = None) -> datetime:
    return datetime.fromtimestamp(time.time() if ts is None else ts, tz=timezone.utc)


def _iso(ts: Optional[float] = None) -> str:
    return _utc(ts).strftime("%Y-%m-%dT%H:%M:%S.%fZ")


def _json(value: Any, default: Any = None) -> Any:
    """pg8000 returns json/jsonb as text on some builds and as objects on others. Normalise both."""
    if value is None:
        return default
    if isinstance(value, (dict, list)):
        return value
    if isinstance(value, (bytes, bytearray)):
        value = value.decode()
    if isinstance(value, str):
        try:
            return json.loads(value)
        except Exception:  # noqa: BLE001
            return default
    return value


def _epoch(value: Any) -> float:
    if value is None:
        return 0.0
    if isinstance(value, datetime):
        return value.replace(tzinfo=value.tzinfo or timezone.utc).timestamp()
    try:
        return float(value)
    except (TypeError, ValueError):
        return 0.0


class Pg:
    """PostgreSQL wrapper.

    A single connection guarded by an RLock is used deliberately. Coordination traffic is low-volume, and the
    naive thread-local design leaked one connection per HTTP handler thread (PostgreSQL max_connections=100),
    which is itself a reliability defect. Serialising database work removes the leak and keeps transactions
    correctly scoped. Long-running worker execution never happens inside this lock.
    """

    def __init__(self, dsn: Optional[str] = None):
        self.dsn = dsn or os.environ.get("CAIN72_PG", "")
        if not self.dsn:
            raise SharedStateError("CAIN72_PG is not set")
        self._lock = threading.RLock()
        self._conn = None

    def _parse(self) -> Dict[str, Any]:
        raw = self.dsn
        if raw.startswith("postgresql://"):
            raw = raw[len("postgresql://"):]
        auth, _, hostpart = raw.rpartition("@")
        user, _, password = auth.partition(":")
        hostport, _, db = hostpart.partition("/")
        host, _, port = hostport.partition(":")
        return {"user": user, "password": password, "host": host or "127.0.0.1",
                "port": int(port or 5432), "database": db or "postgres"}

    def conn(self):
        if pgdb is None:
            raise SharedStateError("pg8000 is not installed")
        if self._conn is None:
            c = pgdb.connect(timeout=8, **self._parse())
            c.autocommit = False
            self._conn = c
        return self._conn

    def _reconnect(self) -> None:
        try:
            if self._conn is not None:
                self._conn.close()
        except Exception:  # noqa: BLE001
            pass
        self._conn = None

    def execute(self, sql: str, args: Sequence[Any] = ()):
        with self._lock:
            try:
                c = self.conn()
                cur = c.cursor()
                cur.execute(sql, tuple(args))
                return cur
            except Exception:
                if not getattr(self, "_in_tx", False):
                    try:
                        c.rollback()
                    except Exception:  # noqa: BLE001
                        pass
                    if not self._alive():
                        self._reconnect()
                raise

    def _alive(self) -> bool:
        try:
            cur = self.conn().cursor()
            cur.execute("SELECT 1")
            cur.fetchall()
            return True
        except Exception:  # noqa: BLE001
            return False

    def query(self, sql: str, args: Sequence[Any] = ()) -> List[Dict[str, Any]]:
        with self._lock:
            cur = self.execute(sql, args)
            cols = [d[0] for d in (cur.description or [])]
            rows = cur.fetchall() or []
            if not getattr(self, "_in_tx", False):
                self.conn().commit()
            return [dict(zip(cols, r)) for r in rows]

    def one(self, sql: str, args: Sequence[Any] = ()) -> Optional[Dict[str, Any]]:
        rows = self.query(sql, args)
        return rows[0] if rows else None

    def run(self, sql: str, args: Sequence[Any] = ()) -> None:
        with self._lock:
            self.execute(sql, args)
            if not getattr(self, "_in_tx", False):
                self.conn().commit()

    def tx(self):
        return _Tx(self)


class _Tx:
    def __init__(self, pg: Pg):
        self.pg = pg

    def __enter__(self):
        self.pg._lock.acquire()
        try:
            self.pg._in_tx = True
            self.pg.execute("BEGIN")
        except Exception:
            self.pg._in_tx = False
            self.pg._lock.release()
            raise
        return self.pg

    def __exit__(self, exc_type, exc, tb):
        try:
            if exc_type is None:
                self.pg.conn().commit()
            else:
                self.pg.conn().rollback()
        finally:
            self.pg._in_tx = False
            self.pg._lock.release()
        return False


def init_schema(pg: Pg) -> None:
    stmts = [
        "CREATE TABLE IF NOT EXISTS nodes(host_id text PRIMARY KEY, pub text NOT NULL, slots int NOT NULL DEFAULT 1,"
        " meta jsonb NOT NULL DEFAULT '{}'::jsonb, state text NOT NULL DEFAULT 'ACTIVE',"
        " registered_at timestamptz NOT NULL DEFAULT now(), last_seen timestamptz NOT NULL DEFAULT now())",
        "CREATE TABLE IF NOT EXISTS leaders(id int PRIMARY KEY, instance_id text, endpoint text,"
        " lease_expires timestamptz, fencing bigint NOT NULL DEFAULT 0, updated_at timestamptz)",
        "INSERT INTO leaders(id, fencing) VALUES(1,0) ON CONFLICT (id) DO NOTHING",
        "CREATE TABLE IF NOT EXISTS missions(mission_id text PRIMARY KEY, tenant text NOT NULL, objective text NOT NULL,"
        " state text NOT NULL, spec jsonb NOT NULL, mission_grant jsonb, budget jsonb NOT NULL DEFAULT '{}'::jsonb,"
        " deadline timestamptz, created_at timestamptz NOT NULL DEFAULT now(),"
        " updated_at timestamptz NOT NULL DEFAULT now(), completed_at timestamptz)",
        "CREATE TABLE IF NOT EXISTS tasks(task_id text PRIMARY KEY, mission_id text NOT NULL REFERENCES missions(mission_id),"
        " seq int NOT NULL, kind text NOT NULL, operation text NOT NULL, capability text NOT NULL,"
        " params jsonb NOT NULL, depends jsonb NOT NULL DEFAULT '[]'::jsonb, state text NOT NULL,"
        " attempt int NOT NULL DEFAULT 0, max_attempts int NOT NULL DEFAULT 3, timeout_s double precision NOT NULL DEFAULT 60,"
        " verify_method text NOT NULL DEFAULT 'recompute', idem_key text UNIQUE, fencing bigint NOT NULL DEFAULT 0,"
        " lease_owner text, lease_host text, lease_expires timestamptz, result jsonb, result_sig text,"
        " result_digest text, result_at timestamptz, verifier_id text, verifier_host text, verify_digest text,"
        " verified_at timestamptz, error text, created_at timestamptz NOT NULL DEFAULT now(),"
        " updated_at timestamptz NOT NULL DEFAULT now())",
        "CREATE INDEX IF NOT EXISTS tasks_ready_idx ON tasks(mission_id, state, capability)",
        "CREATE TABLE IF NOT EXISTS tickets(ticket_id text PRIMARY KEY, mission_id text NOT NULL, task_id text NOT NULL,"
        " attempt int NOT NULL, worker_id text NOT NULL, host_id text NOT NULL, operation text NOT NULL,"
        " params_digest text NOT NULL, fencing bigint NOT NULL, issued_at timestamptz NOT NULL,"
        " expires_at timestamptz NOT NULL, used_at timestamptz, decision text NOT NULL DEFAULT 'ALLOW')",
        "CREATE TABLE IF NOT EXISTS results(task_id text NOT NULL, attempt int NOT NULL, kind text NOT NULL,"
        " worker_id text NOT NULL, host_id text NOT NULL, digest text NOT NULL, body jsonb NOT NULL,"
        " sig text NOT NULL, created_at timestamptz NOT NULL DEFAULT now(),"
        " PRIMARY KEY(task_id, attempt, kind))",
        "CREATE TABLE IF NOT EXISTS grant_usage(grant_id text PRIMARY KEY, ops_used bigint NOT NULL DEFAULT 0)",
        "CREATE TABLE IF NOT EXISTS nonces(nonce text PRIMARY KEY, seen_at timestamptz NOT NULL DEFAULT now())",
        "CREATE TABLE IF NOT EXISTS events(seq bigserial PRIMARY KEY, at timestamptz NOT NULL DEFAULT now(),"
        " type text NOT NULL, mission_id text, host_id text, body jsonb NOT NULL, prev_hash text, hash text)",
        "CREATE TABLE IF NOT EXISTS faults(id bigserial PRIMARY KEY, at timestamptz NOT NULL DEFAULT now(),"
        " kind text NOT NULL, detail jsonb NOT NULL)",
    ]
    for s in stmts:
        pg.run(s)


# ----------------------------------------------------------------------------- authority


@dataclass
class Authority:
    key: Any
    pub: str
    kid: str

    @classmethod
    def load(cls, path: str) -> "Authority":
        raw = open(path, encoding="utf-8").read().strip()
        key = C.load_priv(raw)
        return cls(key=key, pub=C.pub_b64(key), kid=C.key_id(C.pub_b64(key)))

    def issue_grant(self, *, mission_id: str, child_id: str, capabilities: Sequence[str],
                    max_ops: int, ttl_s: float = GRANT_DEFAULT_TTL) -> Dict[str, Any]:
        now = time.time()
        body = {"grant_id": C.rid("g72", 6), "mission_id": mission_id, "child_id": child_id,
                "parent_id": "operator", "capabilities": [c.upper() for c in capabilities],
                "side_effects": ["local_artifact"], "max_ops": int(max_ops), "ops_used": 0, "depth": 1,
                "issued_at": now, "expires_at": now + float(ttl_s), "parent_pub": self.pub,
                "parent_grant_id": None, "schema": "cain72.crosshost.grant.v1"}
        body["signature"] = C.sign(self.key, C.D_GRANT, body)
        return {"body": body, "kid": self.kid, "pub": self.pub}

    def issue_ticket(self, *, mission_id: str, task_id: str, attempt: int, fencing: int,
                     worker_id: str, host_id: str, operation: str, params_digest: str,
                     stage: str = "EXECUTE", ttl_s: float = TICKET_DEFAULT_TTL) -> Dict[str, Any]:
        now = time.time()
        body = {"ticket_id": C.rid("t72", 8), "task_id": task_id, "attempt": int(attempt),
                "mission_id": mission_id, "worker_id": worker_id, "host_id": host_id,
                "grant_id": None, "operation": operation, "capability": "WORK",
                "side_effect": "local_artifact", "params_digest": params_digest, "stage": stage,
                "fencing": int(fencing), "issued_at": now, "expires_at": now + float(ttl_s),
                "schema": "cain72.crosshost.ticket.v1"}
        return {"body": body, "sig": C.sign(self.key, C.D_TICKET, body), "kid": self.kid}


def verify_grant_offline(grant: Mapping[str, Any], pub: str, *, now: Optional[float] = None) -> List[str]:
    now = time.time() if now is None else float(now)
    body = grant.get("body") or grant
    reasons: List[str] = []
    if not C.verify(pub, str(body.get("signature") or grant.get("signature") or ""), C.D_GRANT, {
        k: v for k, v in body.items() if k != "signature"}):
        reasons.append("BAD_GRANT_SIGNATURE")
    if now > float(body.get("expires_at") or 0):
        reasons.append("GRANT_EXPIRED")
    return reasons


def verify_ticket(ticket: Mapping[str, Any], pub: str, *, now: Optional[float] = None) -> List[str]:
    now = time.time() if now is None else float(now)
    body = ticket.get("body") or {}
    reasons: List[str] = []
    if not C.verify(pub, str(ticket.get("sig") or ""), C.D_TICKET, body):
        reasons.append("BAD_TICKET_SIGNATURE")
    if now > float(body.get("expires_at") or 0):
        reasons.append("TICKET_EXPIRED")
    return reasons


# ----------------------------------------------------------------------------- shared state


class SharedState:
    def __init__(self, pg: Pg, *, authority: Authority):
        self.pg = pg
        self.authority = authority

    # ---------------------------------------------------------------- events
    def event(self, type_: str, *, mission_id: str = "", host_id: str = "", body: Optional[Mapping[str, Any]] = None) -> int:
        payload = dict(body or {})
        with self.pg.tx() as tx:
            tx.execute("SELECT pg_advisory_xact_lock(720072)")
            prev = tx.one("SELECT hash FROM events ORDER BY seq DESC LIMIT 1")
            prev_hash = str(prev["hash"]) if prev else ""
            h = hashlib.sha256((prev_hash + C.sha({"type": type_, "mission_id": mission_id,
                                                   "host_id": host_id, "body": payload})).encode()).hexdigest()
            row = tx.one("INSERT INTO events(type,mission_id,host_id,body,prev_hash,hash) "
                         "VALUES(%s,%s,%s,%s::jsonb,%s,%s) RETURNING seq",
                         (type_, mission_id, host_id, json.dumps(payload, sort_keys=True), prev_hash, h))
        return int(row["seq"]) if row else 0

    def event_chain(self) -> Dict[str, Any]:
        rows = self.pg.query("SELECT seq,type,mission_id,host_id,body,prev_hash,hash FROM events ORDER BY seq")
        prev = ""
        bad: List[int] = []
        for r in rows:
            h = hashlib.sha256((prev + C.sha({"type": r["type"], "mission_id": r["mission_id"] or "",
                                              "host_id": r["host_id"] or "",
                                              "body": _json(r["body"], {})})).encode()).hexdigest()
            if h != r["hash"] or (r["prev_hash"] or "") != prev:
                bad.append(int(r["seq"]))
            prev = r["hash"]
        return {"events": len(rows), "ok": not bad, "bad_sequences": bad, "head": prev}

    # ---------------------------------------------------------------- nodes
    def enroll_node(self, host_id: str, pub: str, *, slots: int = 1, meta: Optional[Mapping[str, Any]] = None) -> Dict[str, Any]:
        self.pg.run("INSERT INTO nodes(host_id,pub,slots,meta) VALUES(%s,%s,%s,%s::jsonb) "
                    "ON CONFLICT (host_id) DO UPDATE SET pub=EXCLUDED.pub, slots=EXCLUDED.slots, meta=EXCLUDED.meta",
                    (host_id, pub, int(slots), json.dumps(dict(meta or {}), sort_keys=True)))
        self.event("NODE_ENROLLED", host_id=host_id, body={"slots": slots, "meta": dict(meta or {})})
        return {"host_id": host_id, "enrolled": True}

    def node_pub(self, host_id: str) -> Optional[str]:
        r = self.pg.one("SELECT pub,state FROM nodes WHERE host_id=%s", (host_id,))
        if not r or r["state"] != "ACTIVE":
            return None
        return str(r["pub"])

    def heartbeat(self, host_id: str) -> None:
        self.pg.run("UPDATE nodes SET last_seen=now() WHERE host_id=%s", (host_id,))

    def node_health(self) -> List[Dict[str, Any]]:
        return self.pg.query("SELECT host_id,state,slots,last_seen,"
                             "EXTRACT(EPOCH FROM (now()-last_seen)) AS age_s FROM nodes ORDER BY host_id")

    # ---------------------------------------------------------------- leadership
    def acquire_leadership(self, instance_id: str, endpoint: str, *, ttl_s: float = 6.0) -> Dict[str, Any]:
        with self.pg.tx() as tx:
            tx.execute("SELECT pg_advisory_xact_lock(720073)")
            row = tx.one("SELECT instance_id,lease_expires,fencing,endpoint FROM leaders WHERE id=1 FOR UPDATE")
            now = time.time()
            current = str(row["instance_id"] or "") if row else ""
            expires = _epoch(row["lease_expires"]) if row else 0.0
            fencing = int(row["fencing"] or 0) if row else 0
            if current == instance_id and expires > now:
                tx.execute("UPDATE leaders SET lease_expires=%s, endpoint=%s, updated_at=now() WHERE id=1",
                           (_utc(now + ttl_s), endpoint))
                return {"leader": True, "fencing": fencing, "expires_at": now + ttl_s}
            if current and expires > now:
                return {"leader": False, "fencing": fencing, "leader_instance": current,
                        "leader_endpoint": row["endpoint"] if row else None}
            fencing += 1
            tx.execute("UPDATE leaders SET instance_id=%s, endpoint=%s, lease_expires=%s, fencing=%s,"
                       " updated_at=now() WHERE id=1",
                       (instance_id, endpoint, _utc(now + ttl_s), fencing))
        self.event("LEADER_ACQUIRED", body={"instance_id": instance_id, "endpoint": endpoint, "fencing": fencing})
        return {"leader": True, "fencing": fencing, "expires_at": now + ttl_s}

    def release_leadership(self, instance_id: str) -> None:
        self.pg.run("UPDATE leaders SET lease_expires=now(), updated_at=now() WHERE id=1 AND instance_id=%s",
                    (instance_id,))

    def leadership(self) -> Optional[Dict[str, Any]]:
        r = self.pg.one("SELECT instance_id,endpoint,lease_expires,fencing FROM leaders WHERE id=1")
        if not r:
            return None
        return {"instance_id": r["instance_id"], "endpoint": r["endpoint"],
                "expires_at": _epoch(r["lease_expires"]), "fencing": int(r["fencing"] or 0),
                "active": _epoch(r["lease_expires"]) > time.time()}

    # ---------------------------------------------------------------- missions
    def create_mission(self, *, mission_id: str, tenant: str, objective: str,
                       tasks: Sequence[Mapping[str, Any]], grant: Mapping[str, Any],
                       budget: Optional[Mapping[str, Any]] = None, deadline_s: float = 3600.0) -> Dict[str, Any]:
        now = time.time()
        spec = {"objective": objective, "tasks": list(tasks), "budget": dict(budget or {})}
        digest = C.sha(spec)
        with self.pg.tx() as tx:
            tx.execute("INSERT INTO missions(mission_id,tenant,objective,state,spec,mission_grant,budget,deadline)"
                       " VALUES(%s,%s,%s,'RUNNING',%s::jsonb,%s::jsonb,%s::jsonb,%s)",
                       (mission_id, tenant, objective, json.dumps(spec, sort_keys=True),
                        json.dumps(grant, sort_keys=True), json.dumps(dict(budget or {}), sort_keys=True),
                        _utc(now + float(deadline_s))))
            ids: Dict[str, str] = {}
            for i, t in enumerate(tasks):
                tid = str(t.get("task_id") or f"t-{i:05d}")
                deps = [ids.get(str(d), str(d)) for d in (t.get("depends") or [])]
                idem = str(t.get("idem_key") or C.sha({"mission": mission_id, "i": i, "task": t}))
                state = "PENDING" if deps else "READY"
                tx.execute("INSERT INTO tasks(task_id,mission_id,seq,kind,operation,capability,params,depends,state,"
                           "max_attempts,timeout_s,verify_method,idem_key) VALUES(%s,%s,%s,%s,%s,%s,%s::jsonb,%s::jsonb,"
                           "%s,%s,%s,%s,%s)",
                           (tid, mission_id, i, str(t.get("kind") or "EXECUTE"), str(t.get("operation") or "hash_text"),
                            str(t.get("capability") or "WORK").upper(),
                            json.dumps(dict(t.get("params") or {}), sort_keys=True),
                            json.dumps(deps), state, int(t.get("max_attempts") or 3),
                            float(t.get("timeout_s") or 60), str(t.get("verify") or "recompute"), idem))
                ids[tid] = tid
        self.event("MISSION_CREATED", mission_id=mission_id,
                   body={"tasks": len(tasks), "spec_digest": digest, "tenant": tenant})
        return {"mission_id": mission_id, "tasks": len(tasks), "spec_digest": digest}

    def mission(self, mission_id: str) -> Optional[Dict[str, Any]]:
        m = self.pg.one("SELECT mission_id,tenant,objective,state,spec,mission_grant,budget,deadline,created_at,"
                        "updated_at,completed_at FROM missions WHERE mission_id=%s", (mission_id,))
        if not m:
            return None
        counts = self.pg.query("SELECT state,COUNT(*) n FROM tasks WHERE mission_id=%s GROUP BY state", (mission_id,))
        m["task_states"] = {r["state"]: int(r["n"]) for r in counts}
        m["tasks"] = sum(m["task_states"].values())
        m["verified"] = m["task_states"].get("VERIFIED", 0)
        m["deadline_at"] = _epoch(m["deadline"])
        return m

    def list_missions(self) -> List[Dict[str, Any]]:
        rows = self.pg.query("SELECT mission_id,state FROM missions ORDER BY created_at")
        return [self.mission(str(r["mission_id"])) for r in rows]

    def _promote_ready(self, mission_id: str) -> int:
        rows = self.pg.query(
            "SELECT t.task_id FROM tasks t WHERE t.mission_id=%s AND t.state='PENDING' AND NOT EXISTS ("
            " SELECT 1 FROM jsonb_array_elements_text(t.depends) d WHERE NOT EXISTS ("
            "  SELECT 1 FROM tasks p WHERE p.task_id=d.value AND p.state='VERIFIED'))", (mission_id,))
        n = 0
        for r in rows:
            self.pg.run("UPDATE tasks SET state='READY', updated_at=now() WHERE task_id=%s AND state='PENDING'",
                        (r["task_id"],))
            n += 1
        if n:
            self.pg.run("UPDATE missions SET updated_at=now() WHERE mission_id=%s", (mission_id,))
        return n

    # ---------------------------------------------------------------- task lifecycle
    def claim(self, *, host_id: str, worker_id: str, capabilities: Sequence[str],
              kinds: Sequence[str] = ("EXECUTE", "VERIFY"), lease_s: float = 30.0,
              mission_id: Optional[str] = None, exclude_host: Optional[str] = None) -> Optional[Dict[str, Any]]:
        caps = [c.upper() for c in capabilities]
        kinds = tuple(k.upper() for k in kinds)
        with self.pg.tx() as tx:
            tx.execute("SELECT pg_advisory_xact_lock(720074)")
            # expire leases first
            tx.execute("UPDATE tasks SET state=CASE WHEN kind='EXECUTE' THEN 'READY' ELSE 'VERIFY_READY' END,"
                       " lease_owner=NULL, lease_host=NULL, lease_expires=NULL, updated_at=now(),"
                       " error='LEASE_EXPIRED' WHERE state IN ('LEASED','VERIFY_LEASED') AND lease_expires < now()")
            sql = ("SELECT t.* FROM tasks t JOIN missions m ON m.mission_id=t.mission_id "
                   "WHERE m.state='RUNNING' AND t.capability = ANY(%s)"
                   " AND ((t.kind='EXECUTE' AND t.state='READY') OR t.state='VERIFY_READY')"
                   " AND (t.lease_expires IS NULL OR t.lease_expires < now())"
                   " AND (t.state<>'VERIFY_READY' OR COALESCE(("
                   "  SELECT r.host_id FROM results r WHERE r.task_id=t.task_id AND r.kind='EXECUTE'"
                   "  ORDER BY r.created_at DESC LIMIT 1), '') <> %s)")
            args: List[Any] = [caps, host_id]
            if mission_id:
                sql += " AND t.mission_id=%s"
                args.append(mission_id)
            if exclude_host:
                sql += " AND (t.lease_host IS NULL OR t.lease_host <> %s)"
                args.append(exclude_host)
            sql += " ORDER BY t.seq LIMIT 1 FOR UPDATE SKIP LOCKED"
            row = tx.one(sql, args)
            if not row:
                return None
            if str(row["kind"]) == "EXECUTE" and row["result_digest"] and str(row["state"]) == "READY":
                # the execution already happened; do not re-execute it on another host (that destroys independence)
                tx.execute("UPDATE tasks SET state='VERIFY_READY', updated_at=now() WHERE task_id=%s",
                           (row["task_id"],))
                return {"terminal": True, "task_id": row["task_id"], "reason": "ALREADY_EXECUTED"};
            g = tx.one("SELECT mission_grant FROM missions WHERE mission_id=%s", (row["mission_id"],))
            gb = _json((g or {}).get("mission_grant"), {}) or {}
            gb = gb.get("body") if isinstance(gb, dict) and "body" in gb else gb
            reasons = verify_grant_offline(gb, str(gb.get("parent_pub") or ""))
            if "WORK" not in [str(x).upper() for x in gb.get("capabilities") or []]:
                reasons.append("CAPABILITY_NOT_GRANTED")
            if reasons:
                tx.execute("UPDATE tasks SET state='BLOCKED', error=%s, updated_at=now() WHERE task_id=%s",
                           ("GRANT_REFUSED:" + ",".join(reasons), row["task_id"]))
                return {"terminal": True, "task_id": row["task_id"], "reason": "GRANT_REFUSED",
                        "reasons": reasons}
            used = tx.one("INSERT INTO grant_usage(grant_id,ops_used) VALUES(%s,1)"
                          " ON CONFLICT (grant_id) DO UPDATE SET ops_used=grant_usage.ops_used+1"
                          " RETURNING ops_used", (gb.get("grant_id"),))
            if used and int(used["ops_used"]) > int(gb.get("max_ops") or 0):
                tx.execute("UPDATE tasks SET state='BLOCKED', error='OP_BUDGET_EXHAUSTED',"
                           " updated_at=now() WHERE task_id=%s", (row["task_id"],))
                return {"terminal": True, "task_id": row["task_id"], "reason": "OP_BUDGET_EXHAUSTED"}
            attempt = int(row["attempt"] or 0) + 1
            if attempt > int(row["max_attempts"] or 3):
                tx.execute("UPDATE tasks SET state='FAILED', error='MAX_ATTEMPTS', updated_at=now() WHERE task_id=%s",
                           (row["task_id"],))
                return {"terminal": True, "task_id": row["task_id"]}
            fencing = int(row["fencing"] or 0) + 1
            stage = "VERIFY" if str(row["state"]) == "VERIFY_READY" else "EXECUTE"
            new_state = "VERIFY_LEASED" if stage == "VERIFY" else "LEASED"
            tx.execute("UPDATE tasks SET state=%s, attempt=%s, fencing=%s, lease_owner=%s, lease_host=%s,"
                       " lease_expires=%s, updated_at=now(), error=NULL WHERE task_id=%s",
                       (new_state, attempt, fencing, worker_id, host_id, _utc(time.time() + lease_s),
                        row["task_id"]))
            task = dict(row)
            task.update({"state": new_state, "attempt": attempt, "fencing": fencing, "lease_owner": worker_id,
                         "lease_host": host_id, "lease_expires": time.time() + lease_s})
        params_digest = C.sha(task["params"] if isinstance(task["params"], dict) else json.loads(task["params"]))
        ticket = self.authority.issue_ticket(mission_id=str(task["mission_id"]), task_id=str(task["task_id"]),
                                             attempt=attempt, fencing=fencing, worker_id=worker_id,
                                             host_id=host_id, operation=str(task["operation"]),
                                             params_digest=params_digest, stage=stage)
        self.pg.run("INSERT INTO tickets(ticket_id,mission_id,task_id,attempt,worker_id,host_id,operation,"
                    "params_digest,fencing,issued_at,expires_at) VALUES(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)",
                    (ticket["body"]["ticket_id"], task["mission_id"], task["task_id"], attempt, worker_id, host_id,
                     task["operation"], params_digest, fencing, _utc(ticket["body"]["issued_at"]),
                     _utc(ticket["body"]["expires_at"])))
        self.event("TASK_LEASED", mission_id=str(task["mission_id"]), host_id=host_id,
                   body={"task_id": task["task_id"], "kind": task["kind"], "attempt": attempt, "fencing": fencing})
        grant_row = self.pg.one("SELECT mission_grant FROM missions WHERE mission_id=%s", (task["mission_id"],))
        return {"task_id": str(task["task_id"]), "mission_id": str(task["mission_id"]),
                "kind": stage, "operation": str(task["operation"]),
                "capability": str(task["capability"]), "params": _json(task["params"], {}),
                "attempt": int(attempt), "fencing": int(fencing), "timeout_s": float(task["timeout_s"] or 60),
                "verify_method": str(task["verify_method"] or "recompute"),
                "ticket": ticket, "mission_grant": _json(grant_row["mission_grant"], {})}

    def submit_execute(self, *, task_id: str, attempt: int, fencing: int, worker_id: str, host_id: str,
                       ticket: Mapping[str, Any], payload: Mapping[str, Any], sig: str) -> Dict[str, Any]:
        digest = C.sha({k: v for k, v in dict(payload).items() if k != "_node"})
        with self.pg.tx() as tx:
            tx.execute("SELECT pg_advisory_xact_lock(720075)")
            t = tx.one("SELECT * FROM tasks WHERE task_id=%s FOR UPDATE", (task_id,))
            if not t:
                return {"accepted": False, "reason": "UNKNOWN_TASK"}
            if t["state"] not in ("LEASED", "VERIFY_LEASED"):
                return {"accepted": False, "reason": "TASK_NOT_LEASED:" + str(t["state"])}
            stage = "VERIFY" if t["state"] == "VERIFY_LEASED" else "EXECUTE"
            if int(t["fencing"]) != int(fencing):
                return {"accepted": False, "reason": f"STALE_FENCING:{t['fencing']}!={fencing}"}
            if int(t["attempt"]) != int(attempt):
                return {"accepted": False, "reason": "ATTEMPT_MISMATCH"}
            if str(t["lease_owner"]) != worker_id:
                return {"accepted": False, "reason": "LEASE_OWNER_MISMATCH"}
            tb = dict(ticket.get("body") or {})
            used_row = tx.one("SELECT used_at FROM tickets WHERE ticket_id=%s", (tb["ticket_id"],)) \
                if tb.get("ticket_id") else None
            if used_row and used_row["used_at"]:
                return {"accepted": False, "reason": "TICKET_REPLAY"}
            public = self.node_pub(host_id)
            if not public or not C.verify(public, sig, C.D_RESULT, {"task_id": task_id, "attempt": attempt,
                                                                   "fencing": fencing, "digest": digest,
                                                                   "worker_id": worker_id, "host_id": host_id}):
                return {"accepted": False, "reason": "BAD_RESULT_SIGNATURE"}
            kind = stage
            tx.execute("INSERT INTO results(task_id,attempt,kind,worker_id,host_id,digest,body,sig)"
                       " VALUES(%s,%s,%s,%s,%s,%s,%s::jsonb,%s) ON CONFLICT DO NOTHING",
                       (task_id, attempt, kind, worker_id, host_id, digest,
                        json.dumps(dict(payload), sort_keys=True), sig))
            if tb.get("ticket_id"):
                tx.execute("UPDATE tickets SET used_at=now() WHERE ticket_id=%s", (tb["ticket_id"],))
            if stage == "EXECUTE":
                tx.execute("UPDATE tasks SET state='VERIFY_READY', result=%s::jsonb, result_sig=%s, result_digest=%s,"
                           " result_at=now(), lease_owner=NULL, lease_host=NULL, lease_expires=NULL, updated_at=now()"
                           " WHERE task_id=%s",
                           (json.dumps(dict(payload), sort_keys=True), sig, digest, task_id))
            else:
                producer = tx.one("SELECT digest FROM results WHERE task_id=%s AND kind='EXECUTE'"
                                  " ORDER BY created_at DESC LIMIT 1", (task_id,))
                producer_host = tx.one("SELECT host_id FROM results WHERE task_id=%s AND kind='EXECUTE'"
                                       " ORDER BY created_at DESC LIMIT 1", (task_id,))
                independent = bool(producer_host) and str(producer_host["host_id"]) != host_id
                match = bool(producer) and str(producer["digest"]) == digest and independent
                if not match:
                    why = "DIGEST_MISMATCH" if producer and str(producer["digest"]) != digest else "NOT_INDEPENDENT"
                    final = int(t["attempt"] or 0) >= int(t["max_attempts"] or 3)
                    tx.execute("UPDATE tasks SET state=CASE WHEN %s THEN 'FAILED' ELSE 'VERIFY_READY' END,"
                               " error=%s, verifier_id=%s, verifier_host=%s, verify_digest=%s, updated_at=now()"
                               " WHERE task_id=%s", (final, why, worker_id, host_id, digest, task_id))
                    return {"accepted": False, "reason": "VERIFY_MISMATCH"}
                tx.execute("UPDATE tasks SET state='VERIFIED', verifier_id=%s, verifier_host=%s,"
                           " verify_digest=%s, verified_at=now(), lease_owner=NULL, lease_host=NULL,"
                           " lease_expires=NULL, updated_at=now() WHERE task_id=%s",
                           (worker_id, host_id, digest, task_id))
                tx.execute("UPDATE missions SET updated_at=now() WHERE mission_id=%s", (t["mission_id"],))
        self.event("TASK_VERIFIED" if kind == "VERIFY" else "TASK_EXECUTED",
                   mission_id=str(t["mission_id"]), host_id=host_id,
                   body={"task_id": task_id, "attempt": attempt, "fencing": fencing, "digest": digest})
        if kind == "VERIFY":
            self._promote_ready(str(t["mission_id"]))
            self._maybe_complete(str(t["mission_id"]))
        return {"accepted": True, "digest": digest, "kind": kind}

    def _maybe_complete(self, mission_id: str) -> None:
        m = self.mission(mission_id)
        if m and m["task_states"] and set(m["task_states"]) <= {"VERIFIED"}:
            self.pg.run("UPDATE missions SET state='COMPLETED', completed_at=now(), updated_at=now()"
                        " WHERE mission_id=%s AND state='RUNNING'", (mission_id,))
            self.event("MISSION_COMPLETED", mission_id=mission_id,
                       body={"tasks": m["tasks"], "verified": m["verified"]})

    def grant_authorized(self, *, mission_id: str, capability: str, now: Optional[float] = None) -> Dict[str, Any]:
        now = time.time() if now is None else float(now)
        grant = self.pg.one("SELECT mission_grant FROM missions WHERE mission_id=%s", (mission_id,))
        if not grant:
            return {"ok": False, "reasons": ["UNKNOWN_MISSION"]}
        g = _json(grant["mission_grant"], {})
        body = g.get("body") if isinstance(g, dict) and "body" in g else g
        pub = str(body.get("parent_pub") or g.get("pub") or "")
        reasons = verify_grant_offline(body, pub, now=now)
        if capability.upper() not in [str(x).upper() for x in body.get("capabilities") or []]:
            reasons.append("CAPABILITY_NOT_GRANTED")
        used = self.pg.one("SELECT ops_used FROM grant_usage WHERE grant_id=%s", (body.get("grant_id"),))
        if used and int(used["ops_used"]) >= int(body.get("max_ops") or 0):
            reasons.append("OP_BUDGET_EXHAUSTED")
        return {"ok": not reasons, "reasons": reasons, "grant_id": body.get("grant_id")}

    def consume_grant_op(self, mission_id: str, *, amount: int = 1) -> bool:
        grant = self.pg.one("SELECT mission_grant FROM missions WHERE mission_id=%s", (mission_id,))
        if not grant:
            return False
        g = _json(grant["mission_grant"], {})
        body = g.get("body") if isinstance(g, dict) and "body" in g else g
        gid = str(body.get("grant_id"))
        max_ops = int(body.get("max_ops") or 0)
        with self.pg.tx() as tx:
            r = tx.one("INSERT INTO grant_usage(grant_id,ops_used) VALUES(%s,%s)"
                       " ON CONFLICT (grant_id) DO UPDATE SET ops_used=grant_usage.ops_used+%s"
                       " RETURNING ops_used", (gid, amount, amount))
            return int(r["ops_used"]) <= max_ops

    # ---------------------------------------------------------------- evidence
    def export(self, out_dir: str) -> Dict[str, Any]:
        out = os.path.abspath(out_dir)
        os.makedirs(out, exist_ok=True)
        data = {
            "schema": SCHEMA,
            "generated_at": _iso(),
            "nodes": self.node_health(),
            "leaders": self.pg.query("SELECT instance_id,endpoint,lease_expires,fencing,updated_at FROM leaders"),
            "missions": self.pg.query("SELECT * FROM missions ORDER BY created_at"),
            "tasks": self.pg.query("SELECT * FROM tasks ORDER BY mission_id,seq"),
            "results": self.pg.query("SELECT * FROM results ORDER BY task_id,attempt,kind"),
            "tickets": self.pg.query("SELECT ticket_id,mission_id,task_id,attempt,worker_id,host_id,operation,"
                                     "params_digest,fencing,issued_at,expires_at,used_at FROM tickets ORDER BY issued_at"),
            "grant_usage": self.pg.query("SELECT * FROM grant_usage ORDER BY grant_id"),
            "events": self.pg.query("SELECT seq,at,type,mission_id,host_id,body,prev_hash,hash FROM events ORDER BY seq"),
            "faults": self.pg.query("SELECT * FROM faults ORDER BY id"),
        }
        # datetimes / decimals -> strings/floats
        from decimal import Decimal

        def clean(v):
            if isinstance(v, Decimal):
                return float(v)
            if isinstance(v, datetime):
                return v.replace(tzinfo=v.tzinfo or timezone.utc).astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ")
            if isinstance(v, dict):
                return {k: clean(x) for k, x in v.items()}
            if isinstance(v, list):
                return [clean(x) for x in v]
            return v
        data = clean(data)
        data["chain"] = self.event_chain()
        manifest = {"schema": SCHEMA, "files": {}, "chain_head": data["chain"]["head"],
                    "mission_ids": [m["mission_id"] for m in data["missions"]]}
        for name in ("state",):
            path = os.path.join(out, f"{name}.json")
            with open(path, "w", encoding="utf-8") as f:
                json.dump(data, f, indent=1, sort_keys=True)
            manifest["files"][f"{name}.json"] = hashlib.sha256(open(path, "rb").read()).hexdigest()
        manifest["sig"] = C.sign(self.authority.key, C.D_BUNDLE, manifest)
        manifest["kid"] = self.authority.kid
        manifest["authority_pub"] = self.authority.pub
        with open(os.path.join(out, "manifest.json"), "w", encoding="utf-8") as f:
            json.dump(manifest, f, indent=1, sort_keys=True)
        return manifest


# ----------------------------------------------------------------------------- node runtime


class CrossHostNode:
    """A node process. Holds no authority: it verifies a signed ticket and reports a signed result."""

    def __init__(self, *, host_id: str, key_path: str, coordinators: Sequence[str],
                 authority_pub: str, workdir: str, slots: int = 1, worker_id: Optional[str] = None,
                 capabilities: Sequence[str] = ("WORK",), kinds: Sequence[str] = ("EXECUTE", "VERIFY")):
        self.host_id = host_id
        self.worker_id = worker_id or f"{host_id}:w1"
        self.key = C.load_priv(open(key_path, encoding="utf-8").read().strip())
        self.pub = C.pub_b64(self.key)
        self.coordinators = [c.rstrip("/") for c in coordinators]
        self.authority_pub = authority_pub
        self.workdir = os.path.abspath(workdir)
        os.makedirs(self.workdir, exist_ok=True)
        self.slots = int(slots)
        self.capabilities = [c.upper() for c in capabilities]
        self.kinds = [k.upper() for k in kinds]
        self.stop = threading.Event()
        self.executed = 0
        self.verified = 0
        self.refused = 0
        self.errors: List[str] = []
        self.journal_path = os.path.join(self.workdir, "node-journal.jsonl")
        self.seen_tickets = self._load_seen()
        self.coord_index = 0
        self.leader_endpoint: Optional[str] = None

    def _load_seen(self) -> set:
        seen = set()
        if os.path.exists(self.journal_path):
            for line in open(self.journal_path, encoding="utf-8"):
                try:
                    r = json.loads(line)
                    if r.get("ticket_id"):
                        seen.add(r["ticket_id"])
                except Exception:  # noqa: BLE001
                    pass
        return seen

    def _journal(self, rec: Mapping[str, Any]) -> None:
        with open(self.journal_path, "a", encoding="utf-8") as f:
            f.write(json.dumps(dict(rec), sort_keys=True) + "\n")
            f.flush()
            os.fsync(f.fileno())

    def _post(self, base: str, kind: str, body: Mapping[str, Any], *, timeout: float = 60.0) -> Dict[str, Any]:
        body = dict(body)
        body.setdefault("nonce", os.urandom(12).hex())
        body.setdefault("at", time.time())
        msg = {"kind": kind, "body": body, "host_id": self.host_id, "pub": self.pub}
        msg["sig"] = C.sign(self.key, "cain72.crosshost.node.v1", msg)
        req = urllib.request.Request(base + "/v1/" + kind, data=json.dumps(msg).encode(),
                                     headers={"Content-Type": "application/json"}, method="POST")
        try:
            with urllib.request.urlopen(req, timeout=timeout) as r:
                return json.loads(r.read().decode() or "{}")
        except urllib.error.HTTPError as e:
            try:
                return json.loads(e.read().decode() or "{}")
            except Exception:  # noqa: BLE001
                raise
        except Exception as e:  # noqa: BLE001
            raise RuntimeError(f"{type(e).__name__}: {e}") from e

    def register(self) -> bool:
        for base in self.coordinators:
            try:
                out = self._post(base, "register", {"slots": self.slots, "at": time.time(),
                                                    "worker_id": self.worker_id,
                                                    "capabilities": self.capabilities})
                if out.get("ok"):
                    self.leader_endpoint = base
                    return True
            except Exception as e:  # noqa: BLE001
                self.errors.append(f"register {base}: {e}")
        return False

    def run_once(self) -> bool:
        bases = ([self.leader_endpoint] if self.leader_endpoint else []) + self.coordinators
        bases = list(dict.fromkeys(b for b in bases if b))
        for base in bases:
            try:
                out = self._post(base, "poll", {"capacity": self.slots, "at": time.time(),
                                                "capabilities": self.capabilities, "kinds": self.kinds})
            except Exception as e:  # noqa: BLE001
                self.errors.append(f"poll {base}: {e}")
                continue
            if out.get("leader") is False and out.get("leader_endpoint"):
                self.leader_endpoint = str(out["leader_endpoint"]).rstrip("/")
                continue
            task = out.get("task")
            if not task:
                return False
            self.leader_endpoint = base
            self._handle(base, task)
            return True
        return False

    def _handle(self, base: str, task: Mapping[str, Any]) -> None:
        ticket = task.get("ticket") or {}
        tb = ticket.get("body") or {}
        reasons = verify_ticket(ticket, self.authority_pub)
        if tb.get("host_id") != self.host_id or tb.get("worker_id") != self.worker_id:
            reasons.append("TICKET_NOT_FOR_NODE")
        if tb.get("ticket_id") in self.seen_tickets:
            reasons.append("TICKET_REPLAY")
        grant = task.get("mission_grant") or {}
        gb = grant.get("body") if "body" in grant else grant
        reasons += verify_grant_offline(gb, str(gb.get("parent_pub") or grant.get("pub") or self.authority_pub))
        if "WORK" not in [str(x).upper() for x in gb.get("capabilities") or []]:
            reasons.append("GRANT_CAPABILITY")
        params = dict(task.get("params") or {})
        if C.sha(params) != tb.get("params_digest"):
            reasons.append("PARAMS_DIGEST_MISMATCH")
        if reasons:
            self.refused += 1
            self._journal({"at": time.time(), "task_id": task.get("task_id"), "refused": reasons})
            try:
                self._post(base, "refuse", {"task_id": task.get("task_id"), "attempt": task.get("attempt"),
                                            "fencing": task.get("fencing"), "reasons": reasons})
            except Exception:  # noqa: BLE001
                pass
            return
        t0 = time.time()
        try:
            payload = ops.run_local(str(task.get("operation")), params)
            error = None
        except Exception as e:  # noqa: BLE001
            payload = {"error": f"{type(e).__name__}: {e}"}
            error = "OPERATION_FAULT"
        elapsed = time.time() - t0
        digest = C.sha(payload)
        body = {"task_id": task.get("task_id"), "attempt": int(task.get("attempt")),
                "fencing": int(task.get("fencing")), "digest": digest,
                "worker_id": self.worker_id, "host_id": self.host_id}
        sig = C.sign(self.key, C.D_RESULT, body)
        payload = dict(payload)
        payload["_node"] = {"host_id": self.host_id, "worker_id": self.worker_id, "elapsed_s": round(elapsed, 6),
                            "at": time.time(), "error": error}
        out = self._post(base, "result", {"task_id": task.get("task_id"), "attempt": int(task.get("attempt")),
                                          "fencing": int(task.get("fencing")), "digest": digest,
                                          "payload": payload, "sig": sig,
                                          "ticket": ticket})
        self.seen_tickets.add(str(tb.get("ticket_id")))
        self._journal({"at": time.time(), "ticket_id": tb.get("ticket_id"), "task_id": task.get("task_id"),
                       "kind": task.get("kind"), "digest": digest, "accepted": out.get("accepted"),
                       "reason": out.get("reason"), "elapsed_s": elapsed})
        if out.get("accepted"):
            if str(task.get("kind")) == "EXECUTE":
                self.executed += 1
            else:
                self.verified += 1
        else:
            self.refused += 1

    def run(self, duration_s: float = 60.0, idle_sleep: float = 0.2) -> Dict[str, Any]:
        start = time.time()
        registered = self.register()
        while not self.stop.is_set() and time.time() - start < duration_s:
            try:
                did = self.run_once()
            except Exception as e:  # noqa: BLE001
                self.errors.append(f"run_once: {e}")
                did = False
            if not did:
                time.sleep(idle_sleep)
        return {"host_id": self.host_id, "worker_id": self.worker_id, "registered": registered,
                "executed": self.executed, "verified": self.verified, "refused": self.refused,
                "errors": self.errors[-20:], "elapsed_s": round(time.time() - start, 3)}