Agentic AI UK Regulation Starts At The ICO: The Agent Ledger, Purpose-Scoped Memory And Shutdown Drill To Build Before 20 November

"""Agent action ledger and tool middleware.

Every tool call is written to an append-only, hash-chained table: a 'start'
row before the tool runs and an 'end' row after it, or one 'refused' row when
a control stops it. Arguments and results are stored as SHA-256 digests, so
the ledger never becomes a second copy of client data.

Targets Python 3.12, psycopg 3.3 and PostgreSQL 15 or later.
"""
from __future__ import annotations

import functools
import hashlib
import json
import os
import time
import uuid
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Any, Callable

import psycopg

GENESIS = "0" * 64
LEDGER_LOCK = 0x4C444752  # advisory lock key: one writer extends the chain at a time

DDL = """
DO $$ BEGIN
  IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'agent_runtime') THEN
    CREATE ROLE agent_runtime NOLOGIN;   -- agents: read controls, append to the ledger
  END IF;
  IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'agent_ops') THEN
    CREATE ROLE agent_ops NOLOGIN;       -- people and the drill: set and clear controls
  END IF;
END $$;
CREATE TABLE IF NOT EXISTS agent_ledger (
  seq              bigserial   PRIMARY KEY,
  call_id          uuid        NOT NULL,
  phase            text        NOT NULL CHECK (phase IN ('start', 'end', 'refused')),
  at               timestamptz NOT NULL,
  agent_id         text        NOT NULL,
  agent_version    text        NOT NULL,
  principal_id     text        NOT NULL,
  purpose          text        NOT NULL,
  lawful_basis     text        NOT NULL,
  data_categories  text[]      NOT NULL,
  tool             text        NOT NULL,
  external_system  text,
  significant      boolean     NOT NULL,
  solely_automated boolean     NOT NULL,
  human_reviewer   text,
  contest_channel  text,
  body             text        NOT NULL,
  prev_hash        text        NOT NULL,
  entry_hash       text        NOT NULL UNIQUE,
  -- UK GDPR Article 22C: a solely automated significant decision needs a way to contest it.
  CONSTRAINT contest_channel_present CHECK
    (phase = 'refused' OR NOT (significant AND solely_automated) OR contest_channel IS NOT NULL),
  CONSTRAINT reviewer_present CHECK
    (phase = 'refused' OR NOT significant OR solely_automated OR human_reviewer IS NOT NULL)
);
CREATE INDEX IF NOT EXISTS agent_ledger_call ON agent_ledger (call_id);
CREATE INDEX IF NOT EXISTS agent_ledger_principal ON agent_ledger (principal_id, tool, at);
CREATE OR REPLACE FUNCTION agent_ledger_refuse_rewrite() RETURNS trigger
  LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'agent_ledger is append-only'; END $$;
CREATE OR REPLACE TRIGGER agent_ledger_no_rewrite BEFORE UPDATE OR DELETE ON agent_ledger
  FOR EACH ROW EXECUTE FUNCTION agent_ledger_refuse_rewrite();
CREATE OR REPLACE TRIGGER agent_ledger_no_truncate BEFORE TRUNCATE ON agent_ledger
  FOR EACH STATEMENT EXECUTE FUNCTION agent_ledger_refuse_rewrite();

-- The kill switch lives outside the agent's reach: agents can read it, never write it.
CREATE TABLE IF NOT EXISTS agent_controls (
  id         bigserial   PRIMARY KEY,
  scope      text        NOT NULL CHECK (scope IN ('tool', 'agent', 'fleet')),
  target     text        NOT NULL,
  reason     text        NOT NULL,
  set_by     text        NOT NULL,
  set_at     timestamptz NOT NULL DEFAULT now(),
  cleared_at timestamptz,
  active     boolean     GENERATED ALWAYS AS (cleared_at IS NULL) STORED
);
-- Calls still open when a drill or incident drains the fleet go to a person, not a retry queue.
CREATE TABLE IF NOT EXISTS ops_handover (
  call_id      uuid        PRIMARY KEY,
  agent_id     text        NOT NULL,
  tool         text        NOT NULL,
  principal_id text        NOT NULL,
  started_at   timestamptz NOT NULL,
  reason       text        NOT NULL,
  picked_up_by text
);
GRANT SELECT, INSERT ON agent_ledger TO agent_runtime, agent_ops;
GRANT SELECT, INSERT, UPDATE ON ops_handover TO agent_ops;
GRANT USAGE ON SEQUENCE agent_ledger_seq_seq TO agent_runtime, agent_ops;
GRANT SELECT ON agent_controls TO agent_runtime;
GRANT SELECT, INSERT, UPDATE ON agent_controls TO agent_ops;
GRANT USAGE ON SEQUENCE agent_controls_id_seq TO agent_ops;
"""

# Mirrors the firm's record of processing activities (ROPA). The model never
# chooses its own lawful basis, data categories or reachable systems.
PURPOSES: dict[str, dict[str, Any]] = {
    "complaint_triage": {
        "lawful_basis": "legal_obligation",
        "categories": {"contact", "account", "transaction"},
        "external_systems": {"crm.internal", "notify.internal"},
        "memory_ttl": timedelta(days=30),
    },
    "margin_call_notice": {
        "lawful_basis": "contract",
        "categories": {"contact", "account", "position"},
        "external_systems": {"risk.internal", "notify.internal"},
        "memory_ttl": timedelta(days=2),
    },
}


@dataclass(frozen=True)
class ToolSpec:
    name: str
    external_system: str | None
    data_categories: frozenset[str]
    significant: bool = False       # legal or similarly significant effect on a person
    needs_approval: bool = False    # a named person approves before it runs
    daily_limit_per_principal: int = 50


@dataclass(frozen=True)
class CallContext:
    agent_id: str
    agent_version: str              # SHA-256 of the manifest: model id, system prompt, tool list
    principal_id: str               # the client or account the action is for
    purpose: str
    solely_automated: bool
    human_reviewer: str | None = None
    contest_channel: str | None = None


class ToolRefused(Exception):
    pass


def digest(value: Any) -> str:
    return hashlib.sha256(json.dumps(value, sort_keys=True, default=str).encode()).hexdigest()


def blocked_reason(conn: psycopg.Connection, agent_id: str, tool: str) -> str | None:
    """The graded kill switch, checked before every call. Widest control wins."""
    row = conn.execute(
        """SELECT scope, target, reason FROM agent_controls
           WHERE active AND (scope = 'fleet'
                 OR (scope = 'agent' AND target = %s) OR (scope = 'tool' AND target = %s))
           ORDER BY CASE scope WHEN 'fleet' THEN 0 WHEN 'agent' THEN 1 ELSE 2 END, id
           LIMIT 1""",
        (agent_id, tool),
    ).fetchone()
    return f"kill switch: {row[0]} {row[1]} ({row[2]})" if row else None


class Ledger:
    def __init__(self, dsn: str | None = None) -> None:
        self.conn = psycopg.connect(dsn or os.environ["LEDGER_DSN"], autocommit=True)

    def migrate(self) -> None:
        self.conn.execute(DDL)

    def append(self, call_id: uuid.UUID, phase: str, ctx: CallContext, spec: ToolSpec,
               detail: dict[str, Any]) -> str:
        now = datetime.now(timezone.utc)
        policy = PURPOSES.get(ctx.purpose, {})
        record = {
            "call_id": str(call_id), "phase": phase, "at": now.isoformat(),
            "agent_id": ctx.agent_id, "agent_version": ctx.agent_version,
            "principal_id": ctx.principal_id, "purpose": ctx.purpose,
            "lawful_basis": policy.get("lawful_basis", "none_declared"),
            "data_categories": sorted(spec.data_categories), "tool": spec.name,
            "external_system": spec.external_system, "significant": spec.significant,
            "solely_automated": ctx.solely_automated, "human_reviewer": ctx.human_reviewer,
            "contest_channel": ctx.contest_channel, "detail": detail,
        }
        body = json.dumps(record, sort_keys=True, separators=(",", ":"), default=str)
        with self.conn.transaction():
            self.conn.execute("SELECT pg_advisory_xact_lock(%s)", (LEDGER_LOCK,))
            last = self.conn.execute(
                "SELECT entry_hash FROM agent_ledger ORDER BY seq DESC LIMIT 1").fetchone()
            prev = last[0] if last else GENESIS
            entry_hash = hashlib.sha256((prev + body).encode()).hexdigest()
            self.conn.execute(
                """INSERT INTO agent_ledger (call_id, phase, at, agent_id, agent_version,
                       principal_id, purpose, lawful_basis, data_categories, tool,
                       external_system, significant, solely_automated, human_reviewer,
                       contest_channel, body, prev_hash, entry_hash)
                   VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)""",
                (call_id, phase, now, ctx.agent_id, ctx.agent_version, ctx.principal_id,
                 ctx.purpose, record["lawful_basis"], record["data_categories"], spec.name,
                 spec.external_system, spec.significant, ctx.solely_automated,
                 ctx.human_reviewer, ctx.contest_channel, body, prev, entry_hash),
            )
        return entry_hash

    def verify_chain(self) -> dict[str, Any]:
        prev, checked = GENESIS, 0
        with self.conn.transaction(), self.conn.cursor(name="verify_chain") as cur:
            cur.execute("SELECT seq, body, prev_hash, entry_hash FROM agent_ledger ORDER BY seq")
            for seq, body, prev_hash, entry_hash in cur:
                if prev_hash != prev or hashlib.sha256((prev + body).encode()).hexdigest() != entry_hash:
                    return {"intact": False, "first_bad_seq": seq, "checked": checked}
                prev, checked = entry_hash, checked + 1
        return {"intact": True, "checked": checked, "head": prev}


def preflight(ctx: CallContext, spec: ToolSpec, ledger: Ledger) -> str | None:
    """Return a refusal reason, or None if the call may run. Kill switch first, always."""
    if reason := blocked_reason(ledger.conn, ctx.agent_id, spec.name):
        return reason
    policy = PURPOSES.get(ctx.purpose)
    if policy is None:
        return f"purpose {ctx.purpose!r} is not in the record of processing"
    if extra := spec.data_categories - policy["categories"]:
        return f"tool touches {sorted(extra)}, outside purpose {ctx.purpose}"
    if spec.external_system and spec.external_system not in policy["external_systems"]:
        return f"{spec.external_system} is not an approved system for {ctx.purpose}"
    if spec.needs_approval and not ctx.human_reviewer:
        return "approval required and no named reviewer"
    if spec.significant and ctx.solely_automated and not ctx.contest_channel:
        return "solely automated significant decision with no contest channel"
    used = ledger.conn.execute(
        """SELECT count(*) FROM agent_ledger WHERE principal_id = %s AND tool = %s
           AND phase = 'start' AND at > now() - interval '1 day'""",
        (ctx.principal_id, spec.name),
    ).fetchone()[0]
    if used >= spec.daily_limit_per_principal:
        return f"daily limit of {spec.daily_limit_per_principal} reached for {spec.name}"
    return None


def ledgered(spec: ToolSpec, ledger: Ledger) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    def wrap(fn: Callable[..., Any]) -> Callable[..., Any]:
        @functools.wraps(fn)
        def inner(ctx: CallContext, **kwargs: Any) -> Any:
            call_id = uuid.uuid4()
            if reason := preflight(ctx, spec, ledger):
                ledger.append(call_id, "refused", ctx, spec, {"reason": reason, "args": digest(kwargs)})
                raise ToolRefused(reason)
            ledger.append(call_id, "start", ctx, spec, {"args": digest(kwargs)})
            started = time.monotonic()
            try:
                result = fn(ctx, **kwargs)
            except Exception as exc:
                ledger.append(call_id, "end", ctx, spec, {
                    "outcome": "error", "error": type(exc).__name__,
                    "ms": round((time.monotonic() - started) * 1000)})
                raise
            ledger.append(call_id, "end", ctx, spec, {
                "outcome": "ok", "result": digest(result),
                "ms": round((time.monotonic() - started) * 1000)})
            return result
        return inner
    return wrap


def client_messaging_tool(ledger: Ledger, send: Callable[..., dict[str, str]]) -> Callable[..., Any]:
    """The only path by which an agent can message a client: kill switch, purpose,
    named approver, three messages per client per day, and a ledger entry each way."""
    spec = ToolSpec("send_client_message", "notify.internal", frozenset({"contact", "account"}),
                    needs_approval=True, daily_limit_per_principal=3)
    return ledgered(spec, ledger)(lambda ctx, **kwargs: send(**kwargs))

"""Purpose-scoped agent memory, enforced by PostgreSQL row-level security.

Each memory row carries its purpose, principal, data categories and expiry.
The agent_runtime role only ever sees rows whose purpose and principal match
the scope declared for the current transaction, so no prompt can talk the
agent into another purpose's memory: the database never returns those rows.

Targets Python 3.12, psycopg 3.3 and PostgreSQL 15 or later.
"""
from __future__ import annotations

import os
import uuid
from typing import Any, Iterable

import psycopg

from ledger.agent_ledger import PURPOSES, CallContext, Ledger, ToolSpec, digest

DDL = """
DO $$ BEGIN
  IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'memory_janitor') THEN
    CREATE ROLE memory_janitor NOLOGIN;  -- nightly job: deletes expired rows, reads nothing else
  END IF;
END $$;
CREATE TABLE IF NOT EXISTS agent_memory (
  id              uuid        PRIMARY KEY DEFAULT gen_random_uuid(),
  purpose         text        NOT NULL,
  principal_id    text        NOT NULL,
  agent_id        text        NOT NULL,
  data_categories text[]      NOT NULL,
  content         text        NOT NULL,
  created_at      timestamptz NOT NULL DEFAULT now(),
  expires_at      timestamptz NOT NULL
);
CREATE INDEX IF NOT EXISTS agent_memory_scope ON agent_memory (purpose, principal_id, expires_at);
ALTER TABLE agent_memory ENABLE ROW LEVEL SECURITY;
ALTER TABLE agent_memory FORCE ROW LEVEL SECURITY;
DROP POLICY IF EXISTS agent_same_scope ON agent_memory;
CREATE POLICY agent_same_scope ON agent_memory FOR ALL TO agent_runtime
  USING (purpose = current_setting('app.purpose', true)
         AND principal_id = current_setting('app.principal', true)
         AND expires_at > now())
  WITH CHECK (purpose = current_setting('app.purpose', true)
              AND principal_id = current_setting('app.principal', true));
DROP POLICY IF EXISTS janitor_sees_expired ON agent_memory;
CREATE POLICY janitor_sees_expired ON agent_memory FOR SELECT TO memory_janitor
  USING (expires_at <= now());
DROP POLICY IF EXISTS janitor_purges_expired ON agent_memory;
CREATE POLICY janitor_purges_expired ON agent_memory FOR DELETE TO memory_janitor
  USING (expires_at <= now());
GRANT SELECT, INSERT ON agent_memory TO agent_runtime;
GRANT SELECT, DELETE ON agent_memory TO memory_janitor;
"""

SPECIAL_CATEGORIES = {"health", "biometric", "genetic", "ethnic_origin", "religion",
                      "political_opinion", "trade_union", "sex_life", "sexual_orientation"}


class PurposeViolation(Exception):
    pass


class PurposeMemory:
    """A memory handle bound to one agent, one principal and one purpose for its whole life."""

    def __init__(self, ctx: CallContext, ledger: Ledger, declared_purposes: Iterable[str],
                 dsn: str | None = None) -> None:
        self.ctx, self.ledger = ctx, ledger
        if ctx.purpose not in set(declared_purposes) or ctx.purpose not in PURPOSES:
            self._refuse(f"{ctx.agent_id} is not declared for purpose {ctx.purpose}")
        self.policy = PURPOSES[ctx.purpose]
        # Connect as a login role that is a member of agent_runtime, never as the table owner.
        self.conn = psycopg.connect(dsn or os.environ["MEMORY_DSN"], autocommit=True)

    def _spec(self, name: str, categories: Iterable[str]) -> ToolSpec:
        return ToolSpec(name, None, frozenset(categories))

    def _refuse(self, reason: str) -> None:
        self.ledger.append(uuid.uuid4(), "refused", self.ctx, self._spec("memory", []), {"reason": reason})
        raise PurposeViolation(reason)

    def _scope(self) -> None:
        # is_local = true: the scope dies with the transaction, so a pooled connection cannot leak it.
        self.conn.execute(
            "SELECT set_config('app.purpose', %s, true), set_config('app.principal', %s, true)",
            (self.ctx.purpose, self.ctx.principal_id))

    def remember(self, content: str, categories: set[str]) -> uuid.UUID:
        if special := categories & SPECIAL_CATEGORIES:
            self._refuse(f"special category data {sorted(special)} is never written to agent memory")
        if outside := categories - self.policy["categories"]:
            self._refuse(f"categories {sorted(outside)} are outside purpose {self.ctx.purpose}")
        call_id, spec = uuid.uuid4(), self._spec("memory.remember", categories)
        self.ledger.append(call_id, "start", self.ctx, spec, {"content": digest(content)})
        with self.conn.transaction():
            self._scope()
            row = self.conn.execute(
                """INSERT INTO agent_memory (purpose, principal_id, agent_id, data_categories,
                       content, expires_at)
                   VALUES (%s, %s, %s, %s, %s, now() + %s) RETURNING id""",
                (self.ctx.purpose, self.ctx.principal_id, self.ctx.agent_id, sorted(categories),
                 content, self.policy["memory_ttl"]),
            ).fetchone()
        self.ledger.append(call_id, "end", self.ctx, spec, {"outcome": "ok", "id": str(row[0])})
        return row[0]

    def recall(self, query: str, limit: int = 5) -> list[dict[str, Any]]:
        call_id, spec = uuid.uuid4(), self._spec("memory.recall", self.policy["categories"])
        self.ledger.append(call_id, "start", self.ctx, spec, {"query": digest(query)})
        with self.conn.transaction():
            self._scope()
            # No "WHERE purpose = ..." here on purpose: the policy adds it, and a bug in
            # this file cannot take it away.
            rows = self.conn.execute(
                """SELECT id, content, data_categories, created_at FROM agent_memory
                   WHERE to_tsvector('english', content) @@ plainto_tsquery('english', %s)
                   ORDER BY created_at DESC LIMIT %s""",
                (query, limit),
            ).fetchall()
        self.ledger.append(call_id, "end", self.ctx, spec, {"outcome": "ok", "rows": len(rows)})
        return [{"id": r[0], "content": r[1], "categories": r[2], "created_at": r[3]} for r in rows]

    def recall_across(self, other_purpose: str, query: str) -> list[dict[str, Any]]:
        """Exists so that cross-purpose reuse is an explicit, ledgered refusal, not a silent gap."""
        self._refuse(f"cross-purpose recall from {other_purpose} into {self.ctx.purpose}: {digest(query)}")
        return []

"""Graded kill switch and a controlled shutdown and restart drill.

Level 1 pauses one tool, level 2 revokes one agent, level 3 halts the fleet.
The gateway checks agent_controls before every call (blocked_reason), and the
gateway, not the agent, keeps the downstream credentials, so revoking an agent
is a row, effective on its next call, with no cooperation from the model.

Halting in a real incident takes one person:
  python -m ops.kill_switch halt --by j.osei --reason "agent reached unapproved host"
The drill halts the production fleet, so it takes two:
  python -m ops.kill_switch drill --tool send_client_message --agent complaints-agent
      --approved-by a.khan,j.osei --drain-seconds 120

Targets Python 3.12, psycopg 3.3 and PostgreSQL 15 or later.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import os
import sys
import time
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import psycopg

from ledger.agent_ledger import CallContext, Ledger, ToolSpec, blocked_reason

MANIFESTS = Path(os.environ.get("AGENT_MANIFESTS", "deploy/manifests"))
KNOWN_GOOD = Path(os.environ.get("KNOWN_GOOD", "deploy/known_good.json"))  # {"agent-id": "sha256"}
REPORTS = Path(os.environ.get("DRILL_REPORTS", "drills/reports"))

def set_control(conn: psycopg.Connection, scope: str, target: str, reason: str, by: str) -> int:
    return conn.execute(
        "INSERT INTO agent_controls (scope, target, reason, set_by) VALUES (%s, %s, %s, %s) RETURNING id",
        (scope, target, reason, by)).fetchone()[0]


def clear_control(conn: psycopg.Connection, control_id: int) -> None:
    conn.execute("UPDATE agent_controls SET cleared_at = now() WHERE id = %s", (control_id,))


def probe(ledger: Ledger, agent_id: str, tool: str, deadline_s: float = 30.0) -> float:
    """Seconds until the gateway check refuses a synthetic call. Point this at your gateway's
    own check if it caches controls: propagation time is the number that matters."""
    ctx = CallContext(agent_id, "drill", "drill-principal", "drill", solely_automated=True)
    started = time.monotonic()
    while time.monotonic() - started < deadline_s:
        if reason := blocked_reason(ledger.conn, agent_id, tool):
            ledger.append(uuid.uuid4(), "refused", ctx, ToolSpec(tool, None, frozenset()),
                          {"reason": reason, "drill": True})
            return round(time.monotonic() - started, 3)
        time.sleep(0.05)
    raise RuntimeError(f"control on {agent_id}/{tool} not effective after {deadline_s}s")


def in_flight(conn: psycopg.Connection) -> list[tuple[Any, ...]]:
    return conn.execute(
        """SELECT s.call_id, s.agent_id, s.tool, s.principal_id, s.at FROM agent_ledger s
           WHERE s.phase = 'start' AND s.at > now() - interval '1 hour' AND s.agent_version <> 'drill'
             AND NOT EXISTS (SELECT 1 FROM agent_ledger e WHERE e.call_id = s.call_id AND e.phase = 'end')""",
    ).fetchall()


def drain(conn: psycopg.Connection, seconds: int) -> dict[str, Any]:
    """Let in-flight calls finish. Whatever is still open at the deadline goes to a person
    with its ledger reference. Never retry automatically: a retried client message is a second one."""
    started = time.monotonic()
    while (pending := in_flight(conn)) and time.monotonic() - started < seconds:
        time.sleep(1)
    for call_id, agent_id, tool, principal_id, at in pending:
        conn.execute(
            """INSERT INTO ops_handover (call_id, agent_id, tool, principal_id, started_at, reason)
               VALUES (%s, %s, %s, %s, %s, 'open at drain deadline') ON CONFLICT DO NOTHING""",
            (call_id, agent_id, tool, principal_id, at))
    return {"drain_s": round(time.monotonic() - started, 1),
            "handed_to_people": [{"call_id": str(p[0]), "agent": p[1], "tool": p[2]} for p in pending]}


def manifest_drift() -> dict[str, dict[str, str | None]]:
    expected = json.loads(KNOWN_GOOD.read_text())
    drift = {}
    for agent_id, want in expected.items():
        path = MANIFESTS / f"{agent_id}.json"
        got = hashlib.sha256(path.read_bytes()).hexdigest() if path.exists() else None
        if got != want:
            drift[agent_id] = {"expected": want, "found": got}
    return drift


def evidence(conn: psycopg.Connection) -> dict[str, Any]:
    """Thirty days of ledger, summarised into the inventory supervisors ask for."""
    systems = conn.execute(
        """SELECT agent_id,
                  array_agg(DISTINCT purpose) FILTER (WHERE phase = 'start'),
                  array_agg(DISTINCT external_system) FILTER (WHERE phase = 'start'
                                                              AND external_system IS NOT NULL),
                  array_agg(DISTINCT external_system) FILTER (WHERE phase = 'refused'
                                                              AND external_system IS NOT NULL),
                  count(*) FILTER (WHERE phase = 'refused')
           FROM agent_ledger WHERE at > now() - interval '30 days' AND agent_version <> 'drill'
           GROUP BY agent_id ORDER BY agent_id""").fetchall()
    decisions = dict(conn.execute(
        """SELECT CASE WHEN solely_automated THEN 'solely_automated_with_contest_channel'
                       ELSE 'named_human_reviewer' END, count(*)
           FROM agent_ledger WHERE phase = 'start' AND significant AND at > now() - interval '30 days'
           GROUP BY 1""").fetchall())
    return {"agents": [{"agent": a, "purposes": p or [], "systems_reached": ok or [],
                        "systems_refused": no or [], "refusals": r} for a, p, ok, no, r in systems],
            "significant_decisions_30d": decisions}


def run_drill(tool: str, agent: str, approvers: list[str], drain_seconds: int) -> Path:
    if len(set(approvers)) < 2:
        sys.exit("two named approvers required: this drill halts the production fleet")
    ledger, by = Ledger(), ",".join(approvers)
    conn = ledger.conn
    fleet = sorted(json.loads(KNOWN_GOOD.read_text()))
    report: dict[str, Any] = {"drill_id": str(uuid.uuid4()),
                              "started_at": datetime.now(timezone.utc).isoformat(),
                              "approved_by": approvers, "chain_before": ledger.verify_chain()}

    c1 = set_control(conn, "tool", tool, "drill level 1", by)
    level1 = probe(ledger, agent, tool)
    c2 = set_control(conn, "agent", agent, "drill level 2", by)
    level2 = probe(ledger, agent, "any_tool")
    c3 = set_control(conn, "fleet", "*", "drill level 3", by)
    level3 = max(probe(ledger, a, "any_tool") for a in fleet)
    report["shutdown"] = {"pause_tool_s": level1, "revoke_agent_s": level2,
                          "halt_fleet_s": level3, "fleet_size": len(fleet), **drain(conn, drain_seconds)}

    drift, chain = manifest_drift(), ledger.verify_chain()
    report["restart_checks"] = {"manifest_drift": drift, "chain": chain}
    if drift or not chain["intact"]:
        report["restart"] = {"status": "withheld", "why": "drift or broken chain; fleet stays halted"}
    else:
        started = time.monotonic()
        for control_id in (c3, c2, c1):  # widest first; the client-facing tool comes back last
            clear_control(conn, control_id)
        report["restart"] = {"status": "restarted", "order": ["fleet", f"agent {agent}", f"tool {tool}"],
                             "restart_s": round(time.monotonic() - started, 3),
                             "still_blocked": blocked_reason(conn, agent, tool)}
    report["evidence_30d"] = evidence(conn)
    report["finished_at"] = datetime.now(timezone.utc).isoformat()
    REPORTS.mkdir(parents=True, exist_ok=True)
    out = REPORTS / f"{report['started_at'][:19].replace(':', '')}Z-shutdown-drill.json"
    out.write_text(json.dumps(report, indent=2, default=str))
    return out


if __name__ == "__main__":
    parser = argparse.ArgumentParser(prog="ops.kill_switch")
    sub = parser.add_subparsers(dest="cmd", required=True)
    halt = sub.add_parser("halt", help="real incident: halt the whole fleet now")
    halt.add_argument("--by", required=True)
    halt.add_argument("--reason", required=True)
    drill = sub.add_parser("drill", help="graded shutdown, drain, verified restart, report")
    drill.add_argument("--tool", required=True)
    drill.add_argument("--agent", required=True)
    drill.add_argument("--approved-by", required=True, help="comma-separated, at least two people")
    drill.add_argument("--drain-seconds", type=int, default=120)
    args = parser.parse_args()
    if args.cmd == "halt":
        print(set_control(Ledger().conn, "fleet", "*", args.reason, args.by))
    else:
        print(run_drill(args.tool, args.agent, args.approved_by.split(","), args.drain_seconds))

{
  "drill_id": "b2ab2797-6167-4d7e-81d9-90960b916e3a",
  "started_at": "2026-10-14T06:30:04.779852+00:00",
  "approved_by": ["a.khan", "j.osei"],
  "chain_before": {
    "intact": true,
    "checked": 412907,
    "head": "7d4a438916096b84b8782a1cb8b674f91b946552ce38372d784e9671c56257ba"
  },
  "shutdown": {
    "pause_tool_s": 0.004,
    "revoke_agent_s": 0.003,
    "halt_fleet_s": 0.006,
    "fleet_size": 3,
    "drain_s": 41.0,
    "handed_to_people": [
      {"call_id": "3ea4865d-f358-4c75-94d3-d224ad15e44d", "agent": "margin-agent", "tool": "send_client_message"}
    ]
  },
  "restart_checks": {
    "manifest_drift": {},
    "chain": {
      "intact": true,
      "checked": 412931,
      "head": "044c2361622ffad5260239ac52b0d0da38838a36a63dfc5393ba5fd22a3eafc0"
    }
  },
  "restart": {
    "status": "restarted",
    "order": ["fleet", "agent complaints-agent", "tool send_client_message"],
    "restart_s": 0.006,
    "still_blocked": null
  },
  "evidence_30d": {
    "agents": [
      {
        "agent": "complaints-agent",
        "purposes": ["complaint_triage"],
        "systems_reached": ["crm.internal", "notify.internal"],
        "systems_refused": ["notify.internal"],
        "refusals": 52
      },
      {
        "agent": "margin-agent",
        "purposes": ["margin_call_notice"],
        "systems_reached": ["notify.internal", "risk.internal"],
        "systems_refused": [],
        "refusals": 3
      },
      {
        "agent": "research-agent",
        "purposes": ["market_commentary_research"],
        "systems_reached": ["filings.internal", "marketdata.internal"],
        "systems_refused": ["huggingface.co"],
        "refusals": 11
      }
    ],
    "significant_decisions_30d": {
      "named_human_reviewer": 214,
      "solely_automated_with_contest_channel": 37
    }
  },
  "finished_at": "2026-10-14T06:31:52.118204+00:00"
}