#!/usr/bin/env python3
"""R03 isolated minimum Council core.

This module is deliberately not an installer or daemon.  Its only CLI action is
an internal-temporary-directory self-test.  It implements the deterministic
policy core that later relay and Telegram adapters must call; it has no network,
subprocess, service-control, sudo, or arbitrary-host-path interface.
"""
from __future__ import annotations

import argparse
import ast
import base64
from collections import deque
from dataclasses import dataclass
import hashlib
import hmac
import json
import os
from pathlib import Path
import re
import secrets
import shutil
import sqlite3
import stat
import tempfile
from typing import Any, Callable, Iterable

SCHEMA = "AI_COUNCIL_R03_ISOLATED_CORE_V1"
ROOM_SCHEMA = "RoomManifestV1"
DATA_CLASS = "SYNTHETIC_PUBLIC_ONLY"
ROOM_IDS = ("ENTRY", "BENCH", "COMMONS", "JOB", "ENGINEERING", "REVIEW", "APPROVAL", "INCIDENT")
ROOM_PREFIXES = {
    "ENTRY": "[ARC/ENTRY/v1]", "BENCH": "[ARC/BENCH/v1]",
    "COMMONS": "[ARC/COMMONS/v1]", "JOB": "[ARC/JOB/v1]",
    "ENGINEERING": "[ARC/ENG/v1]", "REVIEW": "[ARC/REVIEW/v1]",
    "APPROVAL": "[ARC/APPROVAL/v1]", "INCIDENT": "[ARC/INCIDENT/v1]",
}
TABLES = frozenset({
    "schema_meta", "agent_slots", "enrollments", "sessions", "room_manifests",
    "room_admissions", "replay_nonces", "jobs", "checkpoints", "workspace_objects",
    "test_receipts", "reviews", "changeplans", "audit_events",
    "enrollment_flags", "session_flags",
})
VALIDATORS = (
    "TEXT_EXACT_V1", "JSON_STRICT_PARSE_V1", "PYTHON_AST_PARSE_V1",
    "UNIFIED_DIFF_STRUCTURE_V1", "FROZEN_ENGINEERING_FIXTURE_V1",
)
REJECTED_DATA_CLASSES = (
    "PRIVATE", "PRODUCTION", "SECRET", "CREDENTIAL", "MARKET_LIVE", "VOICE_RAW",
    "PERSONAL_DATA", "UNKNOWN",
)
HEX64 = re.compile(r"[0-9a-f]{64}\Z")
HEX32 = re.compile(r"[0-9a-f]{32}\Z")
ID = re.compile(r"[A-Za-z0-9_.:-]{1,64}\Z")
STORAGE_NAME = re.compile(r"[0-9a-f]{32}\Z")


MESSAGE_MAX_CHARS = 4000
MESSAGE_MAX_BYTES = 12000


class CouncilError(RuntimeError):
    """Fail-closed policy or state rejection."""


def require(condition: bool, message: str) -> None:
    if not condition:
        raise CouncilError(message)


def canonical_json(value: Any) -> bytes:
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")


def sha256(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def b64u(data: bytes) -> str:
    return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")


def keyed(master: bytes, label: bytes, value: bytes) -> str:
    return hmac.new(master, label + b"\0" + value, hashlib.sha256).hexdigest()


def read_json(path: Path) -> dict[str, Any]:
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeError, json.JSONDecodeError) as error:
        raise CouncilError(f"invalid JSON: {path.name}: {type(error).__name__}") from error
    require(isinstance(value, dict), f"JSON object required: {path.name}")
    return value


@dataclass(frozen=True)
class SessionCredential:
    session_id: str
    agent_id: str
    key_b64u: str
    expires_at: int
    slot: int
    read_only: bool

    def key_bytes(self) -> bytes:
        padding = "=" * ((4 - len(self.key_b64u) % 4) % 4)
        return base64.urlsafe_b64decode(self.key_b64u + padding)


@dataclass(frozen=True)
class SessionContext:
    session_id: str
    agent_id: str
    agent_digest: str
    slot: int
    expires_at: int
    read_only: int


class DeterministicTokens:
    """Test-only deterministic token source; never selected by production callers."""

    def __init__(self) -> None:
        self.counter = 0

    def __call__(self, size: int) -> bytes:
        out = b""
        while len(out) < size:
            self.counter += 1
            out += hashlib.sha256(f"R03-TOKEN-{self.counter}".encode()).digest()
        return out[:size]


class MutableClock:
    def __init__(self, value: int) -> None:
        self.value = value

    def __call__(self) -> int:
        return self.value

    def advance(self, seconds: int) -> None:
        self.value += seconds


class CouncilCore:
    def __init__(
        self,
        state_root: Path,
        manifest_dir: Path,
        policy: dict[str, Any],
        master_key: bytes,
        clock: Callable[[], int],
        token_bytes: Callable[[int], bytes] = secrets.token_bytes,
    ) -> None:
        require(len(master_key) >= 32, "master key must be at least 32 bytes")
        self.policy = policy
        self.master_key = master_key
        self.clock = clock
        self.token_bytes = token_bytes
        self._validate_policy()
        self.state_root = self._prepare_root(state_root)
        self.workspace_root = self.state_root / "workspaces"
        self.database_path = self.state_root / "state.sqlite3"
        require(not self.database_path.is_symlink(), "database cannot be symlink")
        self._database_was_created = not self.database_path.exists()
        if self._database_was_created:
            descriptor = os.open(
                self.database_path,
                os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0),
                0o600,
            )
            os.close(descriptor)
        database_info = self.database_path.lstat()
        require(stat.S_ISREG(database_info.st_mode) and database_info.st_nlink == 1 and
                database_info.st_uid == os.geteuid(),
                "database must be one owned regular file")
        require(stat.S_IMODE(database_info.st_mode) == 0o600,
                "database mode must already be 0600")
        # R03-A2 serialises all relay and gateway-ingress calls with one
        # application lock.  check_same_thread=False permits those trusted
        # adapter threads to use this connection; CouncilCore itself remains
        # deliberately non-concurrent and must stay behind that lock.
        self.db = sqlite3.connect(self.database_path, timeout=5, check_same_thread=False)
        try:
            self.db.row_factory = sqlite3.Row
            self.db.execute("PRAGMA foreign_keys=ON")
            self.db.execute("PRAGMA busy_timeout=5000")
            # Validate existing state before any persistent journal-mode change.
            if not self._database_was_created:
                self._validate_existing_schema()
                self._validate_workspace_root()
            self.db.execute("PRAGMA journal_mode=WAL")
            self.db.execute("PRAGMA synchronous=FULL")
            if self._database_was_created:
                self._create_schema()
                self.workspace_root.mkdir(mode=0o700)
            os.chmod(self.database_path, 0o600)
            self._validate_workspace_root()
            self.manifests = self._load_manifests(manifest_dir)
        except BaseException:
            self.db.close()
            raise
        self.events: deque[dict[str, Any]] = deque()
        self.event_bytes = 0
        # Bodies are intentionally restart-cleared, but cursors must never
        # move backwards or a resumed client could miss all new live events.
        self.next_event_sequence = int(self.db.execute(
            "SELECT COALESCE(MAX(sequence), 0) + 1 FROM audit_events"
        ).fetchone()[0])

    def _validate_policy(self) -> None:
        expected = {
            "schema": "AI_COUNCIL_R03_CORE_POLICY_V1",
            "version": 1,
            "mode": "ISOLATED_CORE_ONLY_NOT_DEPLOYED",
            "pilot_result": "PROVISIONAL_PILOT",
            "max_agent_capacity": 12,
            "initial_active_agents": 3,
            "inactive_reserved_slots": 9,
            "slot_activation": "OWNER_ONLY_FUTURE_EXACT_PLAN",
            "room_ids": list(ROOM_IDS),
            "manifest_directory": "room_manifests/v1",
            "allowed_data_class": DATA_CLASS,
            "rejected_data_classes": list(REJECTED_DATA_CLASSES),
            "enrollment_ttl_seconds": 2147483647,
            "session_ttl_seconds": 2147483647,
            "request_clock_skew_seconds": 60,
            "nonce_ttl_seconds": 300,
            "admission_ttl_seconds": 2147483647,
            "job_ttl_seconds": 2147483647,
            "event_ring_max_events": 256,
            "event_ring_max_bytes": 2097152,
            "workspace_ttl_seconds": 2147483647,
            "workspace_total_bytes": 8388608,
            "workspace_max_objects": 128,
            "workspace_single_object_bytes": 1048576,
            "max_active_runs": 6,
            "validator_allowlist": list(VALIDATORS),
            "message_body_persisted": False,
            "arbitrary_command_or_path_api": False,
            "network_listener_implemented": False,
            "gateway_adapter_implemented": False,
            "target_install_authorized": False,
            "target_ubuntu_changes": 0,
        }
        require(set(self.policy) == set(expected) | {"manifest_sha256", "sqlite_schema_sha256"},
                "policy field set")
        for field, value in expected.items():
            require(self.policy[field] == value, f"policy value: {field}")
        manifest_hashes = self.policy["manifest_sha256"]
        require(isinstance(manifest_hashes, dict) and set(manifest_hashes) == set(ROOM_IDS) and
                all(isinstance(value, str) and HEX64.fullmatch(value) for value in manifest_hashes.values()),
                "policy manifest hashes")
        require(bool(HEX64.fullmatch(str(self.policy["sqlite_schema_sha256"]))),
                "SQLite schema hash")

    @staticmethod
    def _prepare_root(path: Path) -> Path:
        require(path.is_absolute(), "state root must be absolute")
        require(not path.is_symlink(), "state root cannot be symlink")
        if not path.exists():
            path.mkdir(mode=0o700, parents=True)
        require(path.is_dir() and not path.is_symlink(), "invalid state root")
        root_info = path.stat()
        require(stat.S_IMODE(root_info.st_mode) == 0o700 and root_info.st_uid == os.geteuid(),
                "state root must be owned with mode 0700")
        resolved = path.resolve(strict=True)
        require(resolved == path, "state root must be canonical")
        return resolved

    def _validate_workspace_root(self) -> None:
        require(self.workspace_root.is_dir() and not self.workspace_root.is_symlink(),
                "workspace root must already be one directory")
        workspace_info = self.workspace_root.stat()
        require(stat.S_IMODE(workspace_info.st_mode) == 0o700 and
                workspace_info.st_uid == os.geteuid(),
                "workspace root must be owned with mode 0700")

    def _create_schema(self) -> None:
        self.db.executescript(
            """
            PRAGMA application_id = 1095910211;
            PRAGMA user_version = 1;
            CREATE TABLE IF NOT EXISTS schema_meta (
              singleton INTEGER PRIMARY KEY CHECK(singleton = 1),
              schema_version INTEGER NOT NULL CHECK(schema_version = 1),
              created_at INTEGER NOT NULL
            );
            CREATE TABLE IF NOT EXISTS agent_slots (
              slot INTEGER PRIMARY KEY CHECK(slot BETWEEN 1 AND 12),
              enabled INTEGER NOT NULL CHECK(enabled IN (0,1)),
              agent_id_digest TEXT UNIQUE,
              prompt_id_digest TEXT UNIQUE,
              status TEXT NOT NULL,
              created_at INTEGER NOT NULL,
              last_seen_at INTEGER
            );
            CREATE TABLE IF NOT EXISTS enrollments (
              enrollment_digest TEXT PRIMARY KEY,
              slot INTEGER NOT NULL REFERENCES agent_slots(slot),
              prompt_id_digest TEXT NOT NULL,
              expires_at INTEGER NOT NULL,
              consumed_at INTEGER
            );
            CREATE TABLE IF NOT EXISTS sessions (
              session_id TEXT PRIMARY KEY,
              agent_id_digest TEXT NOT NULL,
              slot INTEGER NOT NULL REFERENCES agent_slots(slot),
              issued_at INTEGER NOT NULL,
              expires_at INTEGER NOT NULL,
              revoked_at INTEGER
            );
            CREATE TABLE IF NOT EXISTS enrollment_flags (
              enrollment_digest TEXT PRIMARY KEY REFERENCES enrollments(enrollment_digest),
              read_only INTEGER NOT NULL CHECK(read_only IN (0,1))
            );
            CREATE TABLE IF NOT EXISTS session_flags (
              session_id TEXT PRIMARY KEY REFERENCES sessions(session_id),
              read_only INTEGER NOT NULL CHECK(read_only IN (0,1))
            );
            CREATE TABLE IF NOT EXISTS room_manifests (
              room_id TEXT NOT NULL,
              version INTEGER NOT NULL,
              manifest_sha256 TEXT NOT NULL,
              installed_at INTEGER NOT NULL,
              PRIMARY KEY(room_id, version)
            );
            CREATE TABLE IF NOT EXISTS room_admissions (
              session_id TEXT NOT NULL REFERENCES sessions(session_id),
              room_id TEXT NOT NULL,
              manifest_sha256 TEXT NOT NULL,
              comprehension_sha256 TEXT NOT NULL,
              issued_at INTEGER NOT NULL,
              expires_at INTEGER NOT NULL,
              revoked_at INTEGER,
              PRIMARY KEY(session_id, room_id, manifest_sha256)
            );
            CREATE TABLE IF NOT EXISTS replay_nonces (
              session_id TEXT NOT NULL REFERENCES sessions(session_id),
              nonce_digest TEXT NOT NULL,
              expires_at INTEGER NOT NULL,
              PRIMARY KEY(session_id, nonce_digest)
            );
            CREATE TABLE IF NOT EXISTS jobs (
              job_id TEXT PRIMARY KEY,
              task_contract_sha256 TEXT NOT NULL,
              data_class TEXT NOT NULL CHECK(data_class = 'SYNTHETIC_PUBLIC_ONLY'),
              status TEXT NOT NULL,
              lease_agent_digest TEXT,
              fencing_token INTEGER NOT NULL DEFAULT 0,
              checkpoint_version INTEGER NOT NULL DEFAULT 0,
              created_at INTEGER NOT NULL,
              expires_at INTEGER NOT NULL
            );
            CREATE TABLE IF NOT EXISTS checkpoints (
              job_id TEXT NOT NULL REFERENCES jobs(job_id),
              version INTEGER NOT NULL,
              attempt_id TEXT NOT NULL,
              phase TEXT NOT NULL,
              artifact_set_sha256 TEXT NOT NULL,
              test_receipt_sha256 TEXT,
              plan_sha256 TEXT,
              created_at INTEGER NOT NULL,
              PRIMARY KEY(job_id, version)
            );
            CREATE TABLE IF NOT EXISTS workspace_objects (
              object_id TEXT PRIMARY KEY,
              kind TEXT NOT NULL CHECK(kind IN ('RUN','OBJECT')),
              run_id TEXT NOT NULL,
              owner_agent_digest TEXT NOT NULL,
              relative_storage_name TEXT NOT NULL UNIQUE,
              expected_sha256 TEXT,
              actual_sha256 TEXT,
              size_bytes INTEGER NOT NULL DEFAULT 0,
              state TEXT NOT NULL,
              expires_at INTEGER NOT NULL,
              deleted_at INTEGER
            );
            CREATE TABLE IF NOT EXISTS test_receipts (
              receipt_id TEXT PRIMARY KEY,
              job_id TEXT NOT NULL REFERENCES jobs(job_id),
              validator_id TEXT NOT NULL,
              input_set_sha256 TEXT NOT NULL,
              result_sha256 TEXT NOT NULL,
              passed INTEGER NOT NULL CHECK(passed IN (0,1)),
              created_at INTEGER NOT NULL
            );
            CREATE TABLE IF NOT EXISTS reviews (
              review_id TEXT PRIMARY KEY,
              job_id TEXT NOT NULL REFERENCES jobs(job_id),
              author_agent_digest TEXT NOT NULL,
              reviewer_agent_digest TEXT NOT NULL,
              artifact_set_sha256 TEXT NOT NULL,
              decision TEXT NOT NULL,
              review_sha256 TEXT NOT NULL,
              created_at INTEGER NOT NULL,
              CHECK(author_agent_digest <> reviewer_agent_digest)
            );
            CREATE TABLE IF NOT EXISTS changeplans (
              plan_id TEXT PRIMARY KEY,
              job_id TEXT NOT NULL REFERENCES jobs(job_id),
              plan_sha256 TEXT NOT NULL UNIQUE,
              target_scope_sha256 TEXT NOT NULL,
              nonce_digest TEXT NOT NULL,
              expires_at INTEGER NOT NULL,
              owner_decision TEXT,
              decided_at INTEGER,
              consumed_at INTEGER
            );
            CREATE TABLE IF NOT EXISTS audit_events (
              sequence INTEGER PRIMARY KEY AUTOINCREMENT,
              event_type TEXT NOT NULL,
              actor_digest TEXT,
              object_digest TEXT,
              result TEXT NOT NULL,
              occurred_at INTEGER NOT NULL,
              event_sha256 TEXT NOT NULL UNIQUE
            );
            """
        )
        require(self._sqlite_schema_sha256() == self.policy["sqlite_schema_sha256"],
                "created SQLite schema hash mismatch")
        now = self.now()
        self.db.execute(
            "INSERT OR IGNORE INTO schema_meta VALUES(1,1,?)", (now,)
        )
        for slot in range(1, self.policy["max_agent_capacity"] + 1):
            enabled = 1 if slot <= self.policy["initial_active_agents"] else 0
            self.db.execute(
                "INSERT OR IGNORE INTO agent_slots(slot,enabled,status,created_at) VALUES(?,?,?,?)",
                (slot, enabled, "AVAILABLE" if enabled else "RESERVED_DISABLED", now),
            )
        self.db.commit()

    def _sqlite_schema_sha256(self) -> str:
        rows = [
            dict(row) for row in self.db.execute(
                "SELECT type,name,tbl_name,sql FROM sqlite_master "
                "WHERE sql IS NOT NULL AND name NOT LIKE 'sqlite_%' ORDER BY type,name"
            )
        ]
        return sha256(canonical_json(rows))

    def _validate_existing_schema(self) -> None:
        require(self.db.execute("PRAGMA application_id").fetchone()[0] == 1095910211,
                "unknown database application id")
        require(self.db.execute("PRAGMA user_version").fetchone()[0] == 1,
                "unsupported database schema version")
        tables = {
            row[0] for row in self.db.execute("SELECT name FROM sqlite_master WHERE type='table'")
            if not row[0].startswith("sqlite_")
        }
        require(tables == TABLES, "partial or unknown database table set")
        require(self._sqlite_schema_sha256() == self.policy["sqlite_schema_sha256"],
                "existing SQLite schema hash mismatch")
        require(self.db.execute("PRAGMA integrity_check").fetchone()[0] == "ok",
                "database integrity check failed")
        metadata = self.db.execute(
            "SELECT schema_version FROM schema_meta WHERE singleton=1"
        ).fetchone()
        require(metadata is not None and metadata[0] == 1, "database schema metadata mismatch")
        slots = self.db.execute("SELECT slot,enabled FROM agent_slots ORDER BY slot").fetchall()
        pairs = [(row["slot"], row["enabled"]) for row in slots]
        active = sum(1 for _slot, enabled in pairs if enabled == 1)
        require(pairs == [(slot, 1 if slot <= active else 0) for slot in range(1, 13)],
                "agent slot capacity mismatch")

    def _validate_manifest_body(self, room: str, body: Any) -> None:
        required = {
            "purpose", "prefix", "allowed", "prohibited", "tools", "read_scope",
            "write_scope", "data_class", "quota", "ttl", "approval", "retention",
            "output", "risk_ceiling", "comprehension",
        }
        require(isinstance(body, dict) and set(body) == required, f"manifest body fields: {room}")
        require(body["prefix"] == ROOM_PREFIXES[room], f"manifest prefix: {room}")
        for field in ("purpose", "approval", "retention", "output", "risk_ceiling"):
            require(isinstance(body[field], str) and body[field], f"manifest {field}: {room}")
        for field in ("allowed", "prohibited", "tools", "read_scope", "write_scope"):
            require(isinstance(body[field], list) and all(isinstance(item, str) and item for item in body[field]),
                    f"manifest {field}: {room}")
        require(body["allowed"] and body["prohibited"] and body["tools"], f"manifest action/tool set: {room}")
        require(len(body["tools"]) == len(set(body["tools"])), f"duplicate manifest tool: {room}")
        require(body["data_class"] == DATA_CLASS, f"manifest data class: {room}")
        require(body["quota"] == {
            "message_chars": 2000, "single_object_bytes": 1048576,
            "workspace_bytes": 8388608, "workspace_objects": 128,
        }, f"manifest quota: {room}")
        require(body["ttl"] == {
            "admission_seconds": 3600, "workspace_seconds": 3600, "session_seconds": 86400,
        }, f"manifest ttl: {room}")
        questions = body["comprehension"]
        require(isinstance(questions, list) and len(questions) == 2, f"manifest comprehension: {room}")
        identifiers = set()
        for item in questions:
            require(isinstance(item, dict) and set(item) == {"id", "question", "answer"},
                    f"manifest comprehension fields: {room}")
            require(all(isinstance(item[field], str) and item[field] for field in item),
                    f"manifest comprehension value: {room}")
            identifiers.add(item["id"])
        require(len(identifiers) == 2, f"manifest comprehension IDs: {room}")

    def _load_manifests(self, manifest_dir: Path) -> dict[str, dict[str, Any]]:
        require(manifest_dir.is_dir() and not manifest_dir.is_symlink(), "manifest directory")
        found: dict[str, dict[str, Any]] = {}
        for room in ROOM_IDS:
            path = manifest_dir / f"{room.lower()}.json"
            require(path.is_file() and not path.is_symlink(), f"manifest missing: {room}")
            value = read_json(path)
            require(value.get("schema") == ROOM_SCHEMA, f"manifest schema: {room}")
            require(value.get("room_id") == room and value.get("version") == 1, f"manifest identity: {room}")
            content_hash = value.get("content_sha256")
            unsigned = dict(value)
            unsigned.pop("content_sha256", None)
            require(content_hash == sha256(canonical_json(unsigned)), f"manifest hash: {room}")
            require(self.policy.get("manifest_sha256", {}).get(room) == content_hash,
                    f"policy manifest hash: {room}")
            body = value.get("body")
            self._validate_manifest_body(room, body)
            found[room] = value
            installed = self.db.execute(
                "SELECT manifest_sha256 FROM room_manifests WHERE room_id=? AND version=1",
                (room,),
            ).fetchone()
            if installed is None:
                self.db.execute(
                    "INSERT INTO room_manifests VALUES(?,?,?,?)",
                    (room, 1, content_hash, self.now()),
                )
            else:
                require(installed["manifest_sha256"] == content_hash,
                        f"installed manifest hash mismatch: {room}")
        self.db.commit()
        require(set(found) == set(ROOM_IDS), "manifest set")
        return found

    def close(self) -> None:
        self.db.commit()
        self.db.close()

    def now(self) -> int:
        value = int(self.clock())
        require(value >= 0, "clock")
        return value

    def _agent_digest(self, agent_id: str) -> str:
        require(bool(ID.fullmatch(agent_id)), "invalid agent id")
        return keyed(self.master_key, b"agent", agent_id.encode())

    def _prompt_digest(self, prompt_id: str) -> str:
        require(bool(ID.fullmatch(prompt_id)), "invalid prompt id")
        return keyed(self.master_key, b"prompt", prompt_id.encode())

    def _audit(self, event_type: str, actor: str | None, object_digest: str | None, result: str) -> None:
        now = self.now()
        sequence = int(self.db.execute(
            "SELECT COALESCE(MAX(sequence), 0) + 1 FROM audit_events"
        ).fetchone()[0])
        event = {
            "sequence": sequence,
            "event_type": event_type,
            "actor_digest": actor,
            "object_digest": object_digest,
            "result": result,
            "occurred_at": now,
        }
        self.db.execute(
            "INSERT INTO audit_events(sequence,event_type,actor_digest,object_digest,result,occurred_at,event_sha256) VALUES(?,?,?,?,?,?,?)",
            (sequence, event_type, actor, object_digest, result, now, sha256(canonical_json(event))),
        )

    def create_enrollment(self, slot: int, prompt_id: str, ttl_seconds: int = 2147483647, read_only: bool = False) -> str:
        require(1 <= slot <= self.policy["max_agent_capacity"], "slot out of range")
        require(1 <= ttl_seconds <= 2147483647, "enrollment ttl")
        row = self.db.execute("SELECT * FROM agent_slots WHERE slot=?", (slot,)).fetchone()
        require(row is not None and row["enabled"] == 1, "slot is not enabled")
        require(row["agent_id_digest"] is None, "slot already bound")
        prompt_digest = self._prompt_digest(prompt_id)
        require(self.db.execute(
            "SELECT 1 FROM agent_slots WHERE prompt_id_digest=?", (prompt_digest,)
        ).fetchone() is None, "prompt id already bound")
        require(self.db.execute(
            "SELECT 1 FROM enrollments WHERE prompt_id_digest=? AND consumed_at IS NULL AND expires_at>=?",
            (prompt_digest, self.now()),
        ).fetchone() is None, "prompt id already has active enrollment")
        code = b64u(self.token_bytes(24))
        digest = keyed(self.master_key, b"enrollment", code.encode())
        self.db.execute(
            "INSERT INTO enrollments VALUES(?,?,?,?,NULL)",
            (digest, slot, prompt_digest, self.now() + ttl_seconds),
        )
        if read_only:
            self.db.execute("INSERT INTO enrollment_flags VALUES(?,1)", (digest,))
        self._audit("ENROLLMENT_CREATED", None, digest, "PASS")
        self.db.commit()
        return code

    def _session_key(self, session_id: str, agent_digest: str, expires_at: int) -> bytes:
        value = f"{session_id}\0{agent_digest}\0{expires_at}".encode()
        return hmac.new(self.master_key, b"ARC-SESSION-V1\0" + value, hashlib.sha256).digest()

    def consume_enrollment(self, code: str, agent_id: str, prompt_id: str) -> SessionCredential:
        digest = keyed(self.master_key, b"enrollment", code.encode())
        row = self.db.execute("SELECT * FROM enrollments WHERE enrollment_digest=?", (digest,)).fetchone()
        require(row is not None, "invalid enrollment")
        require(row["consumed_at"] is None and row["expires_at"] >= self.now(), "enrollment expired or consumed")
        require(row["prompt_id_digest"] == self._prompt_digest(prompt_id), "prompt binding mismatch")
        slot = self.db.execute("SELECT * FROM agent_slots WHERE slot=?", (row["slot"],)).fetchone()
        require(slot is not None and slot["enabled"] == 1 and slot["agent_id_digest"] is None, "slot unavailable")
        agent_digest = self._agent_digest(agent_id)
        require(self.db.execute("SELECT 1 FROM agent_slots WHERE agent_id_digest=?", (agent_digest,)).fetchone() is None,
                "agent id collision")
        session_id = self.token_bytes(16).hex()
        expires_at = self.now() + self.policy["session_ttl_seconds"]
        self.db.execute("BEGIN IMMEDIATE")
        try:
            cursor = self.db.execute(
                "UPDATE enrollments SET consumed_at=? WHERE enrollment_digest=? AND consumed_at IS NULL",
                (self.now(), digest),
            )
            require(cursor.rowcount == 1, "enrollment consume conflict")
            self.db.execute(
                "UPDATE agent_slots SET agent_id_digest=?,prompt_id_digest=?,status='ACTIVE',last_seen_at=? WHERE slot=?",
                (agent_digest, row["prompt_id_digest"], self.now(), row["slot"]),
            )
            self.db.execute(
                "INSERT INTO sessions VALUES(?,?,?,?,?,NULL)",
                (session_id, agent_digest, row["slot"], self.now(), expires_at),
            )
            flag = self.db.execute(
                "SELECT 1 FROM enrollment_flags WHERE enrollment_digest=?", (digest,)
            ).fetchone()
            if flag is not None:
                self.db.execute("INSERT INTO session_flags VALUES(?,1)", (session_id,))
                self.db.execute("DELETE FROM enrollment_flags WHERE enrollment_digest=?", (digest,))
            self._audit("ENROLLMENT_CONSUMED", agent_digest, digest, "PASS")
            self.db.commit()
        except BaseException:
            self.db.rollback()
            raise
        key = self._session_key(session_id, agent_digest, expires_at)
        return SessionCredential(session_id, agent_id, b64u(key), expires_at, int(row["slot"]), bool(flag is not None))

    @staticmethod
    def canonical_request(
        method: str,
        path: str,
        body: bytes,
        timestamp: int,
        nonce_hex: str,
        agent_id: str,
        session_id: str,
        content_type: str = "application/json",
    ) -> bytes:
        require(method in {"GET", "POST", "PUT"}, "method")
        require(path.startswith("/v1/") and "?" not in path and "#" not in path, "path")
        require(content_type == "application/json", "content type")
        require(bool(HEX32.fullmatch(nonce_hex)), "nonce")
        fields = (
            "ARC-HMAC-V1", method, path, "", content_type, sha256(body), str(timestamp), nonce_hex,
            agent_id, session_id, "",
        )
        return "\n".join(fields).encode()

    @classmethod
    def sign_request(
        cls, credential: SessionCredential, method: str, path: str, body: bytes,
        timestamp: int, nonce_hex: str,
    ) -> str:
        canonical = cls.canonical_request(
            method, path, body, timestamp, nonce_hex, credential.agent_id, credential.session_id
        )
        return b64u(hmac.new(credential.key_bytes(), canonical, hashlib.sha256).digest())

    def authenticate_request(
        self, session_id: str, agent_id: str, method: str, path: str, body: bytes,
        timestamp: int, nonce_hex: str, mac_b64u: str,
    ) -> SessionContext:
        require(abs(self.now() - int(timestamp)) <= self.policy["request_clock_skew_seconds"], "request expired")
        row = self.db.execute("SELECT * FROM sessions WHERE session_id=?", (session_id,)).fetchone()
        require(row is not None and row["revoked_at"] is None and row["expires_at"] >= self.now(), "invalid session")
        agent_digest = self._agent_digest(agent_id)
        require(agent_digest == row["agent_id_digest"], "invalid session")
        canonical = self.canonical_request(method, path, body, timestamp, nonce_hex, agent_id, session_id)
        expected = b64u(hmac.new(self._session_key(session_id, agent_digest, row["expires_at"]), canonical, hashlib.sha256).digest())
        require(hmac.compare_digest(expected, mac_b64u), "invalid authentication")
        self.db.execute("DELETE FROM replay_nonces WHERE expires_at < ?", (self.now(),))
        nonce_digest = keyed(self.master_key, b"nonce", bytes.fromhex(nonce_hex))
        try:
            self.db.execute(
                "INSERT INTO replay_nonces VALUES(?,?,?)",
                (session_id, nonce_digest, self.now() + self.policy["nonce_ttl_seconds"]),
            )
        except sqlite3.IntegrityError as error:
            self.db.rollback()
            raise CouncilError("nonce replay") from error
        self.db.execute("UPDATE agent_slots SET last_seen_at=? WHERE slot=?", (self.now(), row["slot"]))
        flag = self.db.execute(
            "SELECT 1 FROM session_flags WHERE session_id=?", (session_id,)
        ).fetchone()
        self._audit("REQUEST_AUTHENTICATED", agent_digest, sha256(canonical), "PASS")
        self.db.commit()
        return SessionContext(session_id, agent_id, agent_digest, int(row["slot"]), int(row["expires_at"]), 1 if flag else 0)

    def get_manifest(self, room_id: str) -> dict[str, Any]:
        require(room_id in self.manifests, "unknown room")
        return json.loads(json.dumps(self.manifests[room_id], ensure_ascii=False))

    def _require_active_context(self, context: SessionContext) -> None:
        row = self.db.execute("SELECT * FROM sessions WHERE session_id=?", (context.session_id,)).fetchone()
        require(row is not None and row["revoked_at"] is None and row["expires_at"] >= self.now() and
                row["agent_id_digest"] == context.agent_digest and row["slot"] == context.slot and
                self._agent_digest(context.agent_id) == context.agent_digest,
                "active session required")

    def ack_room(self, context: SessionContext, room_id: str, manifest_sha256: str, answers: dict[str, str]) -> int:
        self._require_active_context(context)
        require(room_id in self.manifests, "unknown room")
        manifest = self.manifests[room_id]
        require(manifest_sha256 == manifest["content_sha256"], "manifest hash mismatch")
        questions = manifest["body"]["comprehension"]
        expected = {item["id"]: item["answer"] for item in questions}
        require(answers == expected, "comprehension failed")
        expires_at = self.now() + self.policy["admission_ttl_seconds"]
        answer_hash = sha256(canonical_json(answers))
        self.db.execute(
            "INSERT OR REPLACE INTO room_admissions VALUES(?,?,?,?,?,?,NULL)",
            (context.session_id, room_id, manifest_sha256, answer_hash, self.now(), expires_at),
        )
        self._audit("ROOM_ACK", context.agent_digest, manifest_sha256, "PASS")
        self.db.commit()
        return expires_at

    def _require_admission(self, context: SessionContext, room_id: str, tool: str | None = None) -> None:
        self._require_active_context(context)
        manifest = self.manifests.get(room_id)
        require(manifest is not None, "unknown room")
        row = self.db.execute(
            "SELECT * FROM room_admissions WHERE session_id=? AND room_id=? AND manifest_sha256=?",
            (context.session_id, room_id, manifest["content_sha256"]),
        ).fetchone()
        require(row is not None and row["revoked_at"] is None and row["expires_at"] >= self.now(), "room admission required")
        if tool is not None:
            require(tool in manifest["body"]["tools"], "tool not allowed in room")

    @staticmethod
    def _require_data_class(value: str) -> None:
        require(value == DATA_CLASS, "data class not allowed")

    def publish_message(self, context: SessionContext, room_id: str, text: str, data_class: str) -> dict[str, Any]:
        require(not context.read_only, "read-only session")
        self._require_data_class(data_class)
        self._require_admission(context, room_id, "message.publish")
        encoded = text.encode("utf-8")
        require(1 <= len(text) <= MESSAGE_MAX_CHARS and len(encoded) <= MESSAGE_MAX_BYTES, "message size")
        event = {
            "sequence": self.next_event_sequence,
            "room_id": room_id,
            "sender": context.agent_id,
            "text": text,
            "data_class": data_class,
            "created_at": self.now(),
        }
        event["event_sha256"] = sha256(canonical_json(event))
        self.next_event_sequence += 1
        self.events.append(event)
        self.event_bytes += len(encoded)
        while len(self.events) > self.policy["event_ring_max_events"] or self.event_bytes > self.policy["event_ring_max_bytes"]:
            removed = self.events.popleft()
            self.event_bytes -= len(removed["text"].encode("utf-8"))
        self._audit("LIVE_MESSAGE", context.agent_digest, event["event_sha256"], "PASS")
        self.db.commit()
        return dict(event)

    def publish_message_idempotent(
        self, context: SessionContext, room_id: str, text: str,
        data_class: str, idempotency_key: str,
    ) -> dict[str, Any]:
        """Publish an Agent message once without persisting its body.

        Only a keyed digest of the client key plus request/event hashes is
        retained.  A same-key/same-request retry cannot re-send to Telegram;
        same-key/different-request is rejected.
        """
        require(bool(HEX32.fullmatch(idempotency_key)), "idempotency key")
        require(not context.read_only, "read-only session")
        self._require_data_class(data_class)
        self._require_admission(context, room_id, "message.publish")
        encoded = text.encode("utf-8")
        require(1 <= len(text) <= MESSAGE_MAX_CHARS and len(encoded) <= MESSAGE_MAX_BYTES, "message size")
        request_hash = sha256(canonical_json({
            "room_id": room_id,
            "text_sha256": sha256(encoded),
            "data_class": data_class,
        }))
        key_digest = keyed(
            self.master_key, b"message-idempotency",
            (context.agent_digest + "\0" + idempotency_key).encode(),
        )
        previous = self.db.execute(
            "SELECT result FROM audit_events WHERE event_type='LIVE_MESSAGE_IDEMPOTENCY' AND actor_digest=? AND object_digest=?",
            (context.agent_digest, key_digest),
        ).fetchone()
        if previous is not None:
            stored_request, separator, event_hash = str(previous["result"]).partition(":")
            require(separator == ":" and stored_request == request_hash and bool(HEX64.fullmatch(event_hash)),
                    "idempotency key reused with different request")
            live = next((item for item in self.events if item["event_sha256"] == event_hash), None)
            return {
                "duplicate": True,
                "sequence": live["sequence"] if live is not None else None,
                "event_sha256": event_hash,
            }
        event = {
            "sequence": self.next_event_sequence,
            "room_id": room_id,
            "sender": context.agent_id,
            "text": text,
            "data_class": data_class,
            "created_at": self.now(),
        }
        event["event_sha256"] = sha256(canonical_json(event))
        self.db.execute("BEGIN IMMEDIATE")
        try:
            self._audit("LIVE_MESSAGE", context.agent_digest, event["event_sha256"], "PASS")
            self._audit(
                "LIVE_MESSAGE_IDEMPOTENCY", context.agent_digest, key_digest,
                request_hash + ":" + event["event_sha256"],
            )
            self.db.commit()
        except BaseException:
            self.db.rollback()
            raise
        self.next_event_sequence += 1
        self.events.append(event)
        self.event_bytes += len(encoded)
        while len(self.events) > self.policy["event_ring_max_events"] or self.event_bytes > self.policy["event_ring_max_bytes"]:
            removed = self.events.popleft()
            self.event_bytes -= len(removed["text"].encode("utf-8"))
        return {**event, "duplicate": False}

    def ingest_trusted_gateway_message(
        self, gateway_event_sha256: str, room_id: str, text: str, data_class: str,
    ) -> dict[str, Any]:
        """Append one Owner message accepted by the peer-checked Unix adapter.

        This is a trusted embedding interface, never an HTTP route.  The
        gateway supplies only a one-way digest of Telegram routing metadata;
        raw chat, user, update and message identifiers never enter this DB.
        An existing digest is an at-most-once duplicate and is not re-added to
        the restart-cleared live ring.
        """
        require(bool(HEX64.fullmatch(gateway_event_sha256)), "gateway event digest")
        require(room_id in self.manifests, "unknown room")
        self._require_data_class(data_class)
        encoded = text.encode("utf-8")
        require(1 <= len(text) <= MESSAGE_MAX_CHARS and len(encoded) <= MESSAGE_MAX_BYTES, "message size")
        previous = self.db.execute(
            "SELECT sequence FROM audit_events WHERE event_type='GATEWAY_LIVE_MESSAGE' AND object_digest=?",
            (gateway_event_sha256,),
        ).fetchone()
        if previous is not None:
            return {
                "duplicate": True,
                "gateway_event_sha256": gateway_event_sha256,
            }
        event = {
            "sequence": self.next_event_sequence,
            "room_id": room_id,
            "sender": "OWNER",
            "text": text,
            "data_class": data_class,
            "created_at": self.now(),
        }
        event["event_sha256"] = sha256(canonical_json(event))
        self.next_event_sequence += 1
        self.events.append(event)
        self.event_bytes += len(encoded)
        while len(self.events) > self.policy["event_ring_max_events"] or self.event_bytes > self.policy["event_ring_max_bytes"]:
            removed = self.events.popleft()
            self.event_bytes -= len(removed["text"].encode("utf-8"))
        self._audit("GATEWAY_LIVE_MESSAGE", None, gateway_event_sha256, "PASS")
        self.db.commit()
        return {
            "duplicate": False,
            "gateway_event_sha256": gateway_event_sha256,
            "event_sha256": event["event_sha256"],
            "sequence": event["sequence"],
        }

    def poll_events(self, context: SessionContext, after_sequence: int) -> list[dict[str, Any]]:
        require(after_sequence >= 0, "event cursor")
        output = []
        for event in self.events:
            if event["sequence"] <= after_sequence:
                continue
            try:
                self._require_admission(context, event["room_id"])
            except CouncilError:
                continue
            output.append(dict(event))
        return output

    def create_job(self, task_contract: bytes, data_class: str) -> str:
        self._require_data_class(data_class)
        require(1 <= len(task_contract) <= 65536, "task contract size")
        job_id = self.token_bytes(16).hex()
        digest = sha256(task_contract)
        self.db.execute(
            "INSERT INTO jobs VALUES(?,?,?,'OPEN',NULL,0,0,?,?)",
            (job_id, digest, data_class, self.now(), self.now() + self.policy["job_ttl_seconds"]),
        )
        self._audit("JOB_CREATED", None, digest, "PASS")
        self.db.commit()
        return job_id

    def lease_job(self, context: SessionContext, job_id: str) -> int:
        self._require_admission(context, "JOB", "job.lease")
        row = self.db.execute("SELECT * FROM jobs WHERE job_id=?", (job_id,)).fetchone()
        require(row is not None and row["expires_at"] >= self.now(), "job unavailable")
        if row["status"] == "LEASED" and row["lease_agent_digest"] == context.agent_digest:
            return int(row["fencing_token"])
        require(row["status"] == "OPEN", "job already leased")
        token = int(row["fencing_token"]) + 1
        self.db.execute(
            "UPDATE jobs SET status='LEASED',lease_agent_digest=?,fencing_token=? WHERE job_id=?",
            (context.agent_digest, token, job_id),
        )
        self._audit("JOB_LEASED", context.agent_digest, job_id, "PASS")
        self.db.commit()
        return token

    @staticmethod
    def _optional_hash(value: str | None, name: str) -> str | None:
        if value is not None:
            require(bool(HEX64.fullmatch(value)), f"invalid {name}")
        return value

    def checkpoint(
        self, context: SessionContext, job_id: str, fencing_token: int, expected_version: int,
        attempt_id: str, phase: str, artifact_set_sha256: str,
        test_receipt_sha256: str | None = None, plan_sha256: str | None = None,
    ) -> int:
        self._require_admission(context, "JOB", "checkpoint.write")
        require(bool(ID.fullmatch(attempt_id)) and bool(ID.fullmatch(phase)), "checkpoint identity")
        require(bool(HEX64.fullmatch(artifact_set_sha256)), "artifact hash")
        self._optional_hash(test_receipt_sha256, "test receipt hash")
        self._optional_hash(plan_sha256, "plan hash")
        row = self.db.execute("SELECT * FROM jobs WHERE job_id=?", (job_id,)).fetchone()
        require(row is not None and row["lease_agent_digest"] == context.agent_digest, "not job lease holder")
        require(row["fencing_token"] == fencing_token, "stale fencing token")
        require(row["checkpoint_version"] == expected_version, "checkpoint CAS conflict")
        version = expected_version + 1
        self.db.execute("BEGIN IMMEDIATE")
        try:
            current = self.db.execute("SELECT * FROM jobs WHERE job_id=?", (job_id,)).fetchone()
            require(current is not None and current["lease_agent_digest"] == context.agent_digest and
                    current["fencing_token"] == fencing_token and
                    current["checkpoint_version"] == expected_version,
                    "checkpoint state changed")
            self.db.execute(
                "INSERT INTO checkpoints VALUES(?,?,?,?,?,?,?,?)",
                (job_id, version, attempt_id, phase, artifact_set_sha256, test_receipt_sha256, plan_sha256, self.now()),
            )
            cursor = self.db.execute(
                "UPDATE jobs SET checkpoint_version=? WHERE job_id=? AND checkpoint_version=?",
                (version, job_id, expected_version),
            )
            require(cursor.rowcount == 1, "checkpoint CAS conflict")
            self._audit("CHECKPOINT", context.agent_digest, artifact_set_sha256, "PASS")
            self.db.commit()
        except BaseException:
            self.db.rollback()
            raise
        return version

    def _run_row(self, context: SessionContext, run_id: str) -> sqlite3.Row:
        require(bool(HEX32.fullmatch(run_id)), "invalid run id")
        row = self.db.execute(
            "SELECT * FROM workspace_objects WHERE object_id=? AND kind='RUN' AND run_id=?",
            (run_id, run_id),
        ).fetchone()
        require(row is not None and row["owner_agent_digest"] == context.agent_digest and
                row["deleted_at"] is None and row["expires_at"] >= self.now(), "run unavailable")
        return row

    def _run_path(self, row: sqlite3.Row) -> Path:
        name = row["relative_storage_name"]
        require(bool(STORAGE_NAME.fullmatch(name)), "invalid run storage")
        path = self.workspace_root / name
        require(path.parent.resolve(strict=True) == self.workspace_root and path.is_dir() and not path.is_symlink(),
                "invalid run directory")
        run_info = path.stat()
        require(run_info.st_uid == os.geteuid() and stat.S_IMODE(run_info.st_mode) == 0o700,
                "run directory must be owned with mode 0700")
        return path

    def create_workspace_run(self, context: SessionContext, job_id: str, fencing_token: int) -> str:
        self._require_admission(context, "ENGINEERING", "workspace.create")
        job = self.db.execute("SELECT * FROM jobs WHERE job_id=?", (job_id,)).fetchone()
        require(job is not None and job["lease_agent_digest"] == context.agent_digest and
                job["fencing_token"] == fencing_token, "workspace job scope")
        active = self.db.execute(
            "SELECT COUNT(*) FROM workspace_objects WHERE kind='RUN' AND deleted_at IS NULL AND expires_at>=?",
            (self.now(),),
        ).fetchone()[0]
        require(active < self.policy["max_active_runs"], "active run limit")
        run_id = self.token_bytes(16).hex()
        storage = self.token_bytes(16).hex()
        path = self.workspace_root / storage
        require(not path.exists() and not path.is_symlink(), "run path collision")
        path.mkdir(mode=0o700)
        os.chmod(path, 0o700)
        self.db.execute(
            "INSERT INTO workspace_objects VALUES(?, 'RUN', ?, ?, ?, NULL, NULL, 0, 'ACTIVE', ?, NULL)",
            (run_id, run_id, context.agent_digest, storage, self.now() + self.policy["workspace_ttl_seconds"]),
        )
        self._audit("WORKSPACE_CREATED", context.agent_digest, run_id, "PASS")
        self.db.commit()
        return run_id

    def allocate_object(self, context: SessionContext, run_id: str, expected_sha256: str | None = None) -> str:
        run = self._run_row(context, run_id)
        self._optional_hash(expected_sha256, "expected object hash")
        count = self.db.execute(
            "SELECT COUNT(*) FROM workspace_objects WHERE run_id=? AND kind='OBJECT' AND deleted_at IS NULL",
            (run_id,),
        ).fetchone()[0]
        require(count < self.policy["workspace_max_objects"], "workspace object count")
        object_id = self.token_bytes(16).hex()
        storage = self.token_bytes(16).hex()
        self.db.execute(
            "INSERT INTO workspace_objects VALUES(?, 'OBJECT', ?, ?, ?, ?, NULL, 0, 'ALLOCATED', ?, NULL)",
            (object_id, run_id, context.agent_digest, storage, expected_sha256, run["expires_at"]),
        )
        self.db.commit()
        return object_id

    def _object_row(self, context: SessionContext, object_id: str) -> sqlite3.Row:
        require(bool(HEX32.fullmatch(object_id)), "invalid object id")
        row = self.db.execute(
            "SELECT * FROM workspace_objects WHERE object_id=? AND kind='OBJECT'", (object_id,)
        ).fetchone()
        require(row is not None and row["owner_agent_digest"] == context.agent_digest and
                row["deleted_at"] is None and row["expires_at"] >= self.now(), "object unavailable")
        self._run_row(context, row["run_id"])
        return row

    def _object_path(self, context: SessionContext, row: sqlite3.Row) -> Path:
        run = self._run_row(context, row["run_id"])
        directory = self._run_path(run)
        name = row["relative_storage_name"]
        require(bool(STORAGE_NAME.fullmatch(name)), "invalid object storage")
        path = directory / name
        require(path.parent.resolve(strict=True) == directory, "object path escape")
        return path

    def write_object(self, context: SessionContext, object_id: str, data: bytes) -> str:
        row = self._object_row(context, object_id)
        require(row["state"] == "ALLOCATED", "object is not writable")
        require(len(data) <= self.policy["workspace_single_object_bytes"], "single object quota")
        total = self.db.execute(
            "SELECT COALESCE(SUM(size_bytes),0) FROM workspace_objects WHERE run_id=? AND kind='OBJECT' AND deleted_at IS NULL",
            (row["run_id"],),
        ).fetchone()[0]
        require(total + len(data) <= self.policy["workspace_total_bytes"], "workspace byte quota")
        actual = sha256(data)
        require(row["expected_sha256"] is None or row["expected_sha256"] == actual, "object expected hash mismatch")
        path = self._object_path(context, row)
        require(not path.exists() and not path.is_symlink(), "object path already exists")
        temporary = path.with_name("." + path.name + ".tmp-" + self.token_bytes(8).hex())
        flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0)
        descriptor = os.open(temporary, flags, 0o600)
        try:
            with os.fdopen(descriptor, "wb") as handle:
                handle.write(data)
                handle.flush()
                os.fsync(handle.fileno())
            os.chmod(temporary, 0o600)
            os.replace(temporary, path)
            directory_fd = os.open(path.parent, os.O_RDONLY)
            try:
                os.fsync(directory_fd)
            finally:
                os.close(directory_fd)
        finally:
            try:
                temporary.unlink()
            except FileNotFoundError:
                pass
        information = path.lstat()
        require(stat.S_ISREG(information.st_mode) and information.st_nlink == 1 and
                information.st_uid == os.geteuid() and stat.S_IMODE(information.st_mode) == 0o600,
                "unsafe written object")
        self.db.execute(
            "UPDATE workspace_objects SET actual_sha256=?,size_bytes=?,state='READY' WHERE object_id=?",
            (actual, len(data), object_id),
        )
        self._audit("OBJECT_WRITTEN", context.agent_digest, actual, "PASS")
        self.db.commit()
        return actual

    def read_object(self, context: SessionContext, object_id: str) -> bytes:
        row = self._object_row(context, object_id)
        require(row["state"] == "READY", "object not ready")
        path = self._object_path(context, row)
        flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
        try:
            descriptor = os.open(path, flags)
        except OSError as error:
            raise CouncilError("object open rejected") from error
        try:
            before = os.fstat(descriptor)
            require(stat.S_ISREG(before.st_mode) and before.st_nlink == 1 and
                    before.st_uid == os.geteuid() and stat.S_IMODE(before.st_mode) == 0o600,
                    "unsafe object metadata")
            with os.fdopen(descriptor, "rb", closefd=False) as handle:
                data = handle.read(self.policy["workspace_single_object_bytes"] + 1)
            after = os.fstat(descriptor)
            require((before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns) ==
                    (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns),
                    "object changed while reading")
        finally:
            os.close(descriptor)
        require(len(data) == row["size_bytes"] and sha256(data) == row["actual_sha256"], "object integrity")
        return data

    def create_io_challenge(self, context: SessionContext, run_id: str) -> dict[str, Any]:
        nonce = self.token_bytes(16).hex()
        input_data = ("ARC-LIVE-IO-CHALLENGE-V1\nnonce=" + nonce + "\n").encode()
        expected_output = ("ARC-LIVE-IO-RESPONSE-V1:" + nonce + "\n").encode()
        input_id = self.allocate_object(context, run_id, sha256(input_data))
        self.write_object(context, input_id, input_data)
        output_id = self.allocate_object(context, run_id, sha256(expected_output))
        return {
            "input_object_id": input_id,
            "output_object_id": output_id,
            "expected_output": expected_output,
            "expected_output_sha256": sha256(expected_output),
        }

    def verify_io_challenge(self, context: SessionContext, output_object_id: str) -> str:
        row = self._object_row(context, output_object_id)
        require(row["expected_sha256"] is not None and row["state"] == "READY", "challenge output not ready")
        data = self.read_object(context, output_object_id)
        require(sha256(data) == row["expected_sha256"], "challenge output mismatch")
        self._audit("LIVE_IO_VERIFIED", context.agent_digest, row["expected_sha256"], "PASS")
        self.db.commit()
        return row["expected_sha256"]

    def run_validator(
        self, context: SessionContext, job_id: str, validator_id: str,
        object_ids: list[str], expected_sha256: str | None = None,
    ) -> dict[str, Any]:
        self._require_admission(context, "ENGINEERING", "validator.run")
        job = self.db.execute("SELECT * FROM jobs WHERE job_id=?", (job_id,)).fetchone()
        require(job is not None and job["lease_agent_digest"] == context.agent_digest, "validator job scope")
        require(validator_id in self.policy["validator_allowlist"], "validator not allowlisted")
        values = [self.read_object(context, item) for item in object_ids]
        input_hashes = [sha256(item) for item in values]
        passed = False
        detail: dict[str, Any] = {"validator_id": validator_id, "input_hashes": input_hashes}
        try:
            if validator_id == "TEXT_EXACT_V1":
                require(len(values) == 1 and expected_sha256 is not None, "TEXT_EXACT arguments")
                passed = input_hashes[0] == expected_sha256
            elif validator_id == "JSON_STRICT_PARSE_V1":
                require(len(values) == 1, "JSON arguments")
                parsed = json.loads(values[0].decode("utf-8"))
                passed = isinstance(parsed, (dict, list))
            elif validator_id == "PYTHON_AST_PARSE_V1":
                require(len(values) == 1, "AST arguments")
                ast.parse(values[0].decode("utf-8"), filename="opaque-object.py")
                passed = True
            elif validator_id == "UNIFIED_DIFF_STRUCTURE_V1":
                require(len(values) == 1, "diff arguments")
                text = values[0].decode("utf-8")
                passed = (text.startswith("--- a/") and "\n+++ b/" in text and "\n@@ " in text and
                          "GIT binary patch" not in text and "\x00" not in text)
            elif validator_id == "FROZEN_ENGINEERING_FIXTURE_V1":
                require(len(values) == 2, "fixture arguments")
                before = b"def answer():\n    return 1\n"
                after = b"def answer():\n    return 2\n"
                passed = values[0] == before and values[1] == after
            else:
                raise CouncilError("validator dispatch")
        except (UnicodeError, json.JSONDecodeError, SyntaxError, CouncilError):
            passed = False
        detail["passed"] = passed
        input_set_hash = sha256(canonical_json(input_hashes))
        result_hash = sha256(canonical_json(detail))
        receipt_id = self.token_bytes(16).hex()
        self.db.execute(
            "INSERT INTO test_receipts VALUES(?,?,?,?,?,?,?)",
            (receipt_id, job_id, validator_id, input_set_hash, result_hash, int(passed), self.now()),
        )
        self._audit("VALIDATOR", context.agent_digest, result_hash, "PASS" if passed else "FAIL")
        self.db.commit()
        return {"receipt_id": receipt_id, "validator_id": validator_id, "passed": passed,
                "input_set_sha256": input_set_hash, "result_sha256": result_hash}

    def submit_review(
        self, context: SessionContext, job_id: str, artifact_set_sha256: str,
        decision: str, rows: dict[str, dict[str, str]],
    ) -> str:
        self._require_admission(context, "REVIEW", "review.submit")
        require(bool(HEX64.fullmatch(artifact_set_sha256)), "review artifact hash")
        require(decision in {"APPROVE_STAGING", "REJECT"}, "review decision")
        job = self.db.execute("SELECT * FROM jobs WHERE job_id=?", (job_id,)).fetchone()
        require(job is not None and job["lease_agent_digest"] is not None, "review job")
        require(job["lease_agent_digest"] != context.agent_digest, "candidate self-review")
        required_rows = {"correctness", "scope", "security", "tests", "rollback", "non_target_preservation"}
        require(set(rows) == required_rows, "review rows")
        for value in rows.values():
            require(set(value) == {"result", "evidence_sha256"} and value["result"] in {"PASS", "FAIL"} and
                    bool(HEX64.fullmatch(value["evidence_sha256"])), "review row value")
        review = {
            "job_id": job_id,
            "artifact_set_sha256": artifact_set_sha256,
            "decision": decision,
            "rows": rows,
            "reviewer": context.agent_digest,
        }
        review_hash = sha256(canonical_json(review))
        review_id = self.token_bytes(16).hex()
        self.db.execute(
            "INSERT INTO reviews VALUES(?,?,?,?,?,?,?,?)",
            (review_id, job_id, job["lease_agent_digest"], context.agent_digest,
             artifact_set_sha256, decision, review_hash, self.now()),
        )
        self._audit("REVIEW", context.agent_digest, review_hash, "PASS")
        self.db.commit()
        return review_hash

    @staticmethod
    def _safe_remove_tree(root: Path) -> None:
        require(root.is_dir() and not root.is_symlink(), "cleanup root")
        directories: list[Path] = []
        files_to_remove: list[Path] = []
        # Validate the complete tree before deleting the first byte.
        for directory, subdirs, files in os.walk(root, topdown=True, followlinks=False):
            current = Path(directory)
            current_info = current.lstat()
            require(stat.S_ISDIR(current_info.st_mode) and current_info.st_uid == os.geteuid() and
                    stat.S_IMODE(current_info.st_mode) == 0o700, "cleanup unsafe directory")
            directories.append(current)
            for name in subdirs:
                item_info = (current / name).lstat()
                require(stat.S_ISDIR(item_info.st_mode) and item_info.st_uid == os.geteuid() and
                        stat.S_IMODE(item_info.st_mode) == 0o700,
                        "cleanup unsafe child directory")
            for name in files:
                item = current / name
                information = item.lstat()
                require(stat.S_ISREG(information.st_mode) and information.st_nlink == 1 and
                        information.st_uid == os.geteuid() and
                        stat.S_IMODE(information.st_mode) == 0o600,
                        "cleanup unsafe file")
                files_to_remove.append(item)
        for item in files_to_remove:
            item.unlink()
        for directory in reversed(directories):
            directory.rmdir()

    def sweep_expired_workspaces(self) -> int:
        rows = self.db.execute(
            "SELECT * FROM workspace_objects WHERE kind='RUN' AND deleted_at IS NULL AND expires_at < ?",
            (self.now(),),
        ).fetchall()
        removed = 0
        for row in rows:
            path = self._run_path(row)
            tombstone = self.workspace_root / ("expired-" + row["relative_storage_name"])
            require(not tombstone.exists() and not tombstone.is_symlink(), "cleanup tombstone collision")
            path.rename(tombstone)
            self._safe_remove_tree(tombstone)
            self.db.execute(
                "UPDATE workspace_objects SET state='DELETED',deleted_at=? WHERE run_id=?",
                (self.now(), row["run_id"]),
            )
            self._audit("WORKSPACE_DELETED", row["owner_agent_digest"], row["run_id"], "PASS")
            removed += 1
        self.db.commit()
        return removed


def expected_answers(manifest: dict[str, Any]) -> dict[str, str]:
    return {item["id"]: item["answer"] for item in manifest["body"]["comprehension"]}


def run_self_test(policy_path: Path, expected_path: Path) -> dict[str, Any]:
    policy = read_json(policy_path)
    manifest_dir = policy_path.parent / policy["manifest_directory"]
    expected = read_json(expected_path)
    tests: list[str] = []

    def passed(name: str, value: bool = True) -> None:
        require(value, f"self-test failed: {name}")
        tests.append(name)

    def rejected(name: str, call: Callable[[], Any]) -> None:
        try:
            call()
        except (CouncilError, sqlite3.IntegrityError):
            tests.append(name)
        else:
            raise CouncilError(f"self-test expected rejection: {name}")

    clock = MutableClock(1_788_667_200)
    tokens = DeterministicTokens()
    master = hashlib.sha256(b"R03-ISOLATED-SELF-TEST-MASTER").digest()
    unique_event_marker = "R03-EVENT-BODY-MUST-NOT-BE-IN-SQLITE-91f5"
    unique_task_marker = b"R03-TASK-BODY-MUST-NOT-BE-IN-SQLITE-c0d2"
    with tempfile.TemporaryDirectory(prefix="ai-council-r03-core-", dir="/tmp") as temp:
        unknown_root = Path(temp) / "unknown-state"
        unknown_root.mkdir(mode=0o700)
        unknown_database = unknown_root / "state.sqlite3"
        unknown_connection = sqlite3.connect(unknown_database)
        unknown_connection.execute("CREATE TABLE foreign_application(value TEXT)")
        unknown_connection.commit()
        unknown_connection.close()
        os.chmod(unknown_database, 0o600)
        unknown_before = (sha256(unknown_database.read_bytes()), sorted(item.name for item in unknown_root.iterdir()))
        rejected(
            "unknown_database_rejected_before_initialization",
            lambda: CouncilCore(unknown_root, manifest_dir, policy, master, clock, tokens),
        )
        unknown_after = (sha256(unknown_database.read_bytes()), sorted(item.name for item in unknown_root.iterdir()))
        passed("unknown_database_rejection_is_zero_change", unknown_before == unknown_after)
        root = Path(temp) / "state"
        core = CouncilCore(root, manifest_dir, policy, master, clock, tokens)
        try:
            passed("policy_schema", policy["schema"] == "AI_COUNCIL_R03_CORE_POLICY_V1")
            passed("max_agent_capacity_twelve", policy["max_agent_capacity"] == 12)
            passed("initial_active_agents_three", core.db.execute("SELECT COUNT(*) FROM agent_slots WHERE enabled=1").fetchone()[0] == 3)
            passed("inactive_reserved_slots_nine", core.db.execute("SELECT COUNT(*) FROM agent_slots WHERE enabled=0").fetchone()[0] == 9)
            tables = {row[0] for row in core.db.execute("SELECT name FROM sqlite_master WHERE type='table'") if not row[0].startswith("sqlite_")}
            passed("schema_fourteen_tables", len(tables) == 14)
            passed("sqlite_schema_hash_pinned", core._sqlite_schema_sha256() == policy["sqlite_schema_sha256"])
            forbidden_columns = {"message_body", "prompt_body", "response_body", "telegram_token", "raw_payload"}
            actual_columns = {row[1] for table in tables for row in core.db.execute(f"PRAGMA table_info({table})")}
            passed("schema_has_no_body_or_secret_columns", not (forbidden_columns & actual_columns))
            passed("eight_room_manifests_loaded", set(core.manifests) == set(ROOM_IDS))
            passed("room_manifest_hashes_exact", all(item["content_sha256"] == sha256(canonical_json({k:v for k,v in item.items() if k != "content_sha256"})) for item in core.manifests.values()))
            rejected("disabled_slot_four_rejected", lambda: core.create_enrollment(4, "PROMPT-04"))
            code1 = core.create_enrollment(1, "PROMPT-01")
            passed("active_slot_enrollment_created", bool(code1))
            rejected("duplicate_prompt_enrollment_rejected", lambda: core.create_enrollment(2, "PROMPT-01"))
            cred1 = core.consume_enrollment(code1, "AGENT-01", "PROMPT-01")
            passed("enrollment_consumed_once", cred1.slot == 1)
            rejected("enrollment_replay_rejected", lambda: core.consume_enrollment(code1, "AGENT-X", "PROMPT-01"))
            passed("session_ttl_twenty_four_hours", cred1.expires_at - clock() == 86400)
            code2 = core.create_enrollment(2, "PROMPT-02")
            cred2 = core.consume_enrollment(code2, "AGENT-02", "PROMPT-02")
            body = canonical_json({"data_class": DATA_CLASS})
            nonce = "00112233445566778899aabbccddeeff"
            mac = core.sign_request(cred1, "POST", "/v1/messages", body, clock(), nonce)
            ctx1 = core.authenticate_request(cred1.session_id, cred1.agent_id, "POST", "/v1/messages", body, clock(), nonce, mac)
            passed("valid_hmac_request_authenticated", ctx1.slot == 1)
            bad_nonce = "10112233445566778899aabbccddeeff"
            rejected("wrong_hmac_rejected", lambda: core.authenticate_request(cred1.session_id, cred1.agent_id, "POST", "/v1/messages", body, clock(), bad_nonce, "invalid"))
            stale_nonce = "20112233445566778899aabbccddeeff"
            stale_mac = core.sign_request(cred1, "POST", "/v1/messages", body, clock() - 61, stale_nonce)
            rejected("stale_timestamp_rejected", lambda: core.authenticate_request(cred1.session_id, cred1.agent_id, "POST", "/v1/messages", body, clock() - 61, stale_nonce, stale_mac))
            rejected("malformed_nonce_rejected", lambda: core.sign_request(cred1, "POST", "/v1/messages", body, clock(), "bad"))
            rejected("nonce_replay_rejected", lambda: core.authenticate_request(cred1.session_id, cred1.agent_id, "POST", "/v1/messages", body, clock(), nonce, mac))
            # Obtain a fresh authenticated context for Agent 2.
            nonce2 = "30112233445566778899aabbccddeeff"
            mac2 = core.sign_request(cred2, "POST", "/v1/messages", body, clock(), nonce2)
            ctx2 = core.authenticate_request(cred2.session_id, cred2.agent_id, "POST", "/v1/messages", body, clock(), nonce2, mac2)
            commons = core.get_manifest("COMMONS")
            rejected("wrong_manifest_hash_rejected", lambda: core.ack_room(ctx1, "COMMONS", "0"*64, expected_answers(commons)))
            wrong_answers = dict(expected_answers(commons)); wrong_answers[next(iter(wrong_answers))] = "wrong"
            rejected("wrong_comprehension_rejected", lambda: core.ack_room(ctx1, "COMMONS", commons["content_sha256"], wrong_answers))
            core.ack_room(ctx1, "COMMONS", commons["content_sha256"], expected_answers(commons))
            passed("exact_manifest_ack_admitted")
            rejected("unacked_agent_message_rejected", lambda: core.publish_message(ctx2, "COMMONS", "x", DATA_CLASS))
            event = core.publish_message(ctx1, "COMMONS", unique_event_marker, DATA_CLASS)
            passed("synthetic_live_message_published", event["sequence"] == 1)
            rejected("private_payload_rejected", lambda: core.publish_message(ctx1, "COMMONS", "private", "PRIVATE"))
            core.ack_room(ctx2, "COMMONS", commons["content_sha256"], expected_answers(commons))
            passed("admitted_agent_polls_live_event", core.poll_events(ctx2, 0)[0]["text"] == unique_event_marker)
            core.db.execute("PRAGMA wal_checkpoint(TRUNCATE)")
            persisted = core.database_path.read_bytes()
            wal = core.database_path.with_name(core.database_path.name + "-wal")
            if wal.exists(): persisted += wal.read_bytes()
            passed("live_message_body_not_persisted", unique_event_marker.encode() not in persisted)
            for index in range(260):
                core.publish_message(ctx1, "COMMONS", f"ring-{index}", DATA_CLASS)
            passed("event_ring_bounded_to_256", len(core.events) == 256)
            job_manifest = core.get_manifest("JOB")
            for context in (ctx1, ctx2):
                core.ack_room(context, "JOB", job_manifest["content_sha256"], expected_answers(job_manifest))
            job = core.create_job(unique_task_marker, DATA_CLASS)
            passed("job_stores_task_hash_only", core.db.execute("SELECT task_contract_sha256 FROM jobs WHERE job_id=?", (job,)).fetchone()[0] == sha256(unique_task_marker))
            fence = core.lease_job(ctx1, job)
            passed("job_lease_fencing_token", fence == 1)
            artifact_hash = sha256(b"artifact-set")
            version = core.checkpoint(ctx1, job, fence, 0, "ATTEMPT-1", "EDIT", artifact_hash)
            passed("checkpoint_cas_success", version == 1)
            rejected("stale_checkpoint_cas_rejected", lambda: core.checkpoint(ctx1, job, fence, 0, "ATTEMPT-1", "EDIT", artifact_hash))
            rejected("stale_fencing_token_rejected", lambda: core.checkpoint(ctx1, job, 0, 1, "ATTEMPT-1", "EDIT", artifact_hash))
            rejected("other_agent_cannot_lease_job", lambda: core.lease_job(ctx2, job))
            core.db.execute("PRAGMA wal_checkpoint(TRUNCATE)")
            persisted = core.database_path.read_bytes()
            passed("task_body_not_persisted", unique_task_marker not in persisted)
            engineering = core.get_manifest("ENGINEERING")
            for context in (ctx1, ctx2):
                core.ack_room(context, "ENGINEERING", engineering["content_sha256"], expected_answers(engineering))
            run_id = core.create_workspace_run(ctx1, job, fence)
            passed("workspace_run_opaque_id", bool(HEX32.fullmatch(run_id)))
            object_id = core.allocate_object(ctx1, run_id)
            data = b'{"ok":true}\n'
            core.write_object(ctx1, object_id, data)
            passed("object_atomic_write_readback", core.read_object(ctx1, object_id) == data)
            object_row = core._object_row(ctx1, object_id)
            object_path = core._object_path(ctx1, object_row)
            hardlink_path = object_path.with_name(object_path.name + ".hardlink-test")
            os.link(object_path, hardlink_path)
            rejected("hardlinked_object_read_rejected", lambda: core.read_object(ctx1, object_id))
            hardlink_path.unlink()
            rejected("path_traversal_identifier_rejected", lambda: core.read_object(ctx1, "../../etc/passwd"))
            rejected("object_second_write_rejected", lambda: core.write_object(ctx1, object_id, data))
            challenge = core.create_io_challenge(ctx1, run_id)
            input_data = core.read_object(ctx1, challenge["input_object_id"])
            passed("live_io_challenge_read", input_data.startswith(b"ARC-LIVE-IO-CHALLENGE-V1"))
            core.write_object(ctx1, challenge["output_object_id"], challenge["expected_output"])
            passed("live_io_exact_write_readback_verified", core.verify_io_challenge(ctx1, challenge["output_object_id"]) == challenge["expected_output_sha256"])
            rejected("cross_agent_object_read_rejected", lambda: core.read_object(ctx2, object_id))
            oversized = core.allocate_object(ctx1, run_id)
            rejected("single_object_quota_rejected", lambda: core.write_object(ctx1, oversized, b"x"*(policy["workspace_single_object_bytes"]+1)))
            symlink_object = core.allocate_object(ctx1, run_id)
            symlink_row = core._object_row(ctx1, symlink_object)
            symlink_path = core._object_path(ctx1, symlink_row)
            symlink_path.symlink_to("/dev/null")
            rejected("symlink_object_rejected", lambda: core.write_object(ctx1, symlink_object, b"x"))
            symlink_path.unlink()
            valid_json = core.allocate_object(ctx1, run_id); core.write_object(ctx1, valid_json, b'{"a":1}')
            passed("json_validator_pass", core.run_validator(ctx1, job, "JSON_STRICT_PARSE_V1", [valid_json])["passed"] is True)
            invalid_json = core.allocate_object(ctx1, run_id); core.write_object(ctx1, invalid_json, b'{bad')
            passed("json_validator_fail_closed", core.run_validator(ctx1, job, "JSON_STRICT_PARSE_V1", [invalid_json])["passed"] is False)
            valid_py = core.allocate_object(ctx1, run_id); core.write_object(ctx1, valid_py, b'x = 1\n')
            passed("python_ast_validator_pass", core.run_validator(ctx1, job, "PYTHON_AST_PARSE_V1", [valid_py])["passed"] is True)
            invalid_py = core.allocate_object(ctx1, run_id); core.write_object(ctx1, invalid_py, b'def broken(:\n')
            passed("python_ast_validator_fail_closed", core.run_validator(ctx1, job, "PYTHON_AST_PARSE_V1", [invalid_py])["passed"] is False)
            diff_obj = core.allocate_object(ctx1, run_id); core.write_object(ctx1, diff_obj, b'--- a/x\n+++ b/x\n@@ -1 +1 @@\n-a\n+b\n')
            passed("diff_structure_validator_pass", core.run_validator(ctx1, job, "UNIFIED_DIFF_STRUCTURE_V1", [diff_obj])["passed"] is True)
            bad_diff = core.allocate_object(ctx1, run_id); core.write_object(ctx1, bad_diff, b'not a diff\n')
            passed("diff_structure_validator_fail_closed", core.run_validator(ctx1, job, "UNIFIED_DIFF_STRUCTURE_V1", [bad_diff])["passed"] is False)
            before_obj = core.allocate_object(ctx1, run_id); core.write_object(ctx1, before_obj, b'def answer():\n    return 1\n')
            after_obj = core.allocate_object(ctx1, run_id); core.write_object(ctx1, after_obj, b'def answer():\n    return 2\n')
            passed("frozen_engineering_fixture_pass", core.run_validator(ctx1, job, "FROZEN_ENGINEERING_FIXTURE_V1", [before_obj, after_obj])["passed"] is True)
            passed("frozen_engineering_fixture_wrong_order_fails", core.run_validator(ctx1, job, "FROZEN_ENGINEERING_FIXTURE_V1", [after_obj, before_obj])["passed"] is False)
            while True:
                used = int(core.db.execute(
                    "SELECT COALESCE(SUM(size_bytes),0) FROM workspace_objects WHERE run_id=? AND kind='OBJECT'",
                    (run_id,),
                ).fetchone()[0])
                remaining = policy["workspace_total_bytes"] - used
                if remaining == 0:
                    break
                quota_object = core.allocate_object(ctx1, run_id)
                core.write_object(ctx1, quota_object, b"q" * min(remaining, policy["workspace_single_object_bytes"]))
            overflow_object = core.allocate_object(ctx1, run_id)
            rejected("workspace_total_byte_quota_rejected", lambda: core.write_object(ctx1, overflow_object, b"x"))
            audit_rows = core.db.execute("SELECT COUNT(*),COUNT(DISTINCT event_sha256) FROM audit_events").fetchone()
            passed("duplicate_content_audit_events_remain_unique", audit_rows[0] == audit_rows[1])
            current_objects = int(core.db.execute(
                "SELECT COUNT(*) FROM workspace_objects WHERE run_id=? AND kind='OBJECT'", (run_id,)
            ).fetchone()[0])
            for _ in range(policy["workspace_max_objects"] - current_objects):
                core.allocate_object(ctx1, run_id)
            rejected("workspace_object_count_quota_rejected", lambda: core.allocate_object(ctx1, run_id))
            review_manifest = core.get_manifest("REVIEW")
            for context in (ctx1, ctx2):
                core.ack_room(context, "REVIEW", review_manifest["content_sha256"], expected_answers(review_manifest))
            rows = {name:{"result":"PASS","evidence_sha256":sha256(name.encode())} for name in ("correctness","scope","security","tests","rollback","non_target_preservation")}
            rejected("candidate_self_review_rejected", lambda: core.submit_review(ctx1, job, artifact_hash, "APPROVE_STAGING", rows))
            review_hash = core.submit_review(ctx2, job, artifact_hash, "APPROVE_STAGING", rows)
            passed("independent_review_accepted", bool(HEX64.fullmatch(review_hash)))
            clock.advance(policy["admission_ttl_seconds"] + 1)
            rejected("expired_room_admission_rejected", lambda: core.publish_message(ctx1, "COMMONS", "late", DATA_CLASS))
            removed = core.sweep_expired_workspaces()
            passed("expired_workspace_swept", removed == 1)
            run_row = core.db.execute("SELECT * FROM workspace_objects WHERE object_id=?", (run_id,)).fetchone()
            passed("workspace_deletion_receipt_recorded", run_row["deleted_at"] == clock())
            passed("workspace_payload_removed", not list(core.workspace_root.iterdir()))
            passed("controller_has_no_network_or_subprocess_api", "socket" not in globals() and "subprocess" not in globals())
            passed("target_install_authority_false", policy["target_install_authorized"] is False)
            passed("initial_pilot_result_provisional", policy["pilot_result"] == "PROVISIONAL_PILOT")
        finally:
            core.close()
        # Reopen proves SQLite metadata durability while the in-memory event ring resets.
        reopened = CouncilCore(root, manifest_dir, policy, master, clock, tokens)
        try:
            passed("restart_preserves_metadata_checkpoint", reopened.db.execute("SELECT checkpoint_version FROM jobs WHERE job_id=?", (job,)).fetchone()[0] == 1)
            passed("restart_clears_live_message_bodies", len(reopened.events) == 0)
            passed("database_integrity_check_ok", reopened.db.execute("PRAGMA integrity_check").fetchone()[0] == "ok")
        finally:
            reopened.close()
    passed("temporary_state_and_workspace_removed", not Path(temp).exists())
    receipt = {
        "schema": "AI_COUNCIL_R03_ISOLATED_CORE_SELF_TEST_V1",
        "status": "PASS",
        "test_count": len(tests),
        "tests": tests,
        "max_agent_capacity": 12,
        "initial_active_agents": 3,
        "inactive_reserved_slots": 9,
        "room_manifest_count": 8,
        "sqlite_table_count": 14,
        "sqlite_schema_sha256": policy["sqlite_schema_sha256"],
        "allowed_data_class": DATA_CLASS,
        "message_body_persisted": False,
        "workspace_ttl_seconds": 3600,
        "network_calls": 0,
        "subprocess_calls": 0,
        "production_writes": 0,
        "target_ubuntu_changes": 0,
        "gateway_source_changes": 0,
        "pilot_result": "PROVISIONAL_PILOT",
        "temporary_artifacts_after_cleanup": 0,
    }
    require(canonical_json(receipt) == canonical_json(expected), "expected receipt mismatch")
    return receipt


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    sub = parser.add_subparsers(dest="command", required=True)
    test = sub.add_parser("self-test", help="run only in an internal /tmp temporary directory")
    test.add_argument("--policy", required=True, type=Path)
    test.add_argument("--expected", required=True, type=Path)
    return parser


def main() -> int:
    args = build_parser().parse_args()
    try:
        if args.command == "self-test":
            result = run_self_test(args.policy, args.expected)
            print(json.dumps(result, ensure_ascii=False, indent=2))
        return 0
    except (CouncilError, OSError, ValueError, KeyError, sqlite3.Error) as error:
        print(f"R03_CORE_ERROR: {error}", file=os.sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())