"""Small, durable mock-first call executor.

The module intentionally keeps the state machine in one place. SQLite is used as
an actual relational store for local acceptance; RabbitMQ is an optional
publisher/consumer adapter selected by configuration. No HTTP endpoint accepts
an execution command.
"""

from __future__ import annotations

import hashlib
import importlib
import io
import json
import math
import os
import queue
import re
import secrets
import shutil
import sqlite3
import struct
import threading
import time
import uuid
import wave
from collections.abc import Callable, Iterator
from contextlib import contextmanager, suppress
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, cast

from .ai_runtime import AIConfigError, ConversationEngine, normalize_agent_config
from .ai_runtime import config_digest as ai_config_digest
from .mocks import MockAI, MockARI, MockRTP, MockSIPPeer, ProtocolJournal
from .sip_readonly import SipReadonlyClient, SipReadonlyError

try:
    from jsonschema import Draft202012Validator, FormatChecker
except ImportError:  # pragma: no cover - the Docker image installs it
    Draft202012Validator = None
    FormatChecker = None


ROOT = Path(__file__).resolve().parents[1]
DEFAULT_PROFILE = ROOT / "docs" / "contracts" / "mock-profile.json"
DEFAULT_MQ_SCHEMA = ROOT / "docs" / "contracts" / "mq.schema.json"


class ServiceError(Exception):
    def __init__(
        self, code: str, detail: str, status: int = 400, retryable: bool = False
    ) -> None:
        super().__init__(detail)
        self.code = code
        self.detail = detail
        self.status = status
        self.retryable = retryable


class BrokerError(ServiceError):
    def __init__(self, detail: str = "message broker unavailable") -> None:
        super().__init__("BROKER_UNAVAILABLE", detail, 503, True)


class ConfigurationError(ServiceError):
    def __init__(self, detail: str) -> None:
        super().__init__("CONFIGURATION_INVALID", detail, 500, False)


class NotFoundError(ServiceError):
    def __init__(self, detail: str = "resource not found") -> None:
        super().__init__("NOT_FOUND", detail, 404, False)


class ConflictError(ServiceError):
    def __init__(self, code: str, detail: str) -> None:
        super().__init__(code, detail, 409, False)


class ValidationError(ServiceError):
    def __init__(self, code: str, detail: str) -> None:
        super().__init__(code, detail, 422, False)


def utcnow() -> datetime:
    return datetime.now(timezone.utc)


def iso(value: datetime | None = None) -> str:
    value = value or utcnow()
    return (
        value.astimezone(timezone.utc)
        .isoformat(timespec="milliseconds")
        .replace("+00:00", "Z")
    )


def parse_time(value: str) -> datetime:
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except (TypeError, ValueError) as exc:
        raise ValidationError("INVALID_TIME", "timestamp must be RFC3339") from exc
    if parsed.tzinfo is None:
        raise ValidationError("INVALID_TIME", "timestamp must include a timezone")
    return parsed.astimezone(timezone.utc)


def new_id(prefix: str) -> str:
    return f"{prefix}_{uuid.uuid4().hex}"


def canonical(value: Any) -> str:
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))


def digest(value: Any) -> str:
    return hashlib.sha256(canonical(value).encode("utf-8")).hexdigest()


def routing_key(tenant_key: str) -> str:
    value = f"agent-call.tenant.{tenant_key}.call.execute"
    if len(value.encode("utf-8")) > 255:
        raise ValidationError(
            "TENANT_KEY_UNSUPPORTED",
            "tenant routing key exceeds the 255-byte broker limit; command is retained and not sent",
        )
    return value


def queue_name(tenant_key: str) -> str:
    value = f"agent-call.executor.{tenant_key}.v1"
    if len(value.encode("utf-8")) > 255:
        raise ValidationError(
            "TENANT_KEY_UNSUPPORTED", "tenant queue exceeds the broker limit"
        )
    return value


def _json(value: Any) -> str:
    return canonical(value)


def _loads(value: str | None, default: Any = None) -> Any:
    if value is None:
        return default
    try:
        return json.loads(value)
    except (TypeError, json.JSONDecodeError) as exc:
        raise ValidationError("INVALID_JSON", "stored JSON is invalid") from exc


def _load_json_file(path: Path) -> Any:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise ConfigurationError(f"cannot load contract file: {path.name}") from exc


def _integer(value: Any) -> int:
    try:
        return int(value)
    except (TypeError, ValueError) as exc:
        raise ValidationError("INVALID_INTEGER", "numeric value is invalid") from exc


SCHEMA_SQL = """
PRAGMA foreign_keys = ON;
CREATE TABLE IF NOT EXISTS schema_migrations (
  version INTEGER PRIMARY KEY,
  applied_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS readiness_probe (
  probe_id INTEGER PRIMARY KEY CHECK(probe_id = 1),
  touched_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS tenants (
  tenant_id TEXT PRIMARY KEY,
  tenant_key TEXT NOT NULL UNIQUE,
  enabled INTEGER NOT NULL DEFAULT 1,
  max_concurrency INTEGER NOT NULL DEFAULT 2,
  max_cps INTEGER NOT NULL DEFAULT 1,
  created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS tasks (
  task_id TEXT PRIMARY KEY,
  tenant_id TEXT NOT NULL REFERENCES tenants(tenant_id),
  state TEXT NOT NULL,
  revision INTEGER NOT NULL,
  updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS agent_versions (
  agent_version_id TEXT NOT NULL,
  tenant_id TEXT NOT NULL REFERENCES tenants(tenant_id),
  config_json TEXT NOT NULL,
  content_sha256 TEXT NOT NULL,
  immutable INTEGER NOT NULL DEFAULT 1 CHECK(immutable = 1),
  created_at TEXT NOT NULL,
  published_at TEXT NOT NULL,
  created_by TEXT NOT NULL,
  PRIMARY KEY(tenant_id, agent_version_id)
);
CREATE TABLE IF NOT EXISTS commands (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  command_id TEXT NOT NULL,
  tenant_id TEXT NOT NULL,
  tenant_key TEXT NOT NULL,
  command_type TEXT NOT NULL,
  status TEXT NOT NULL,
  payload_json TEXT NOT NULL,
  payload_hash TEXT NOT NULL,
  operation_type TEXT NOT NULL DEFAULT '',
  operation_target TEXT NOT NULL DEFAULT '',
  task_id TEXT,
  execution_id TEXT,
  call_id TEXT,
  trace_id TEXT,
  issued_at TEXT,
  not_after TEXT,
  admission_deadline TEXT,
  accepted_at TEXT,
  waiting_since TEXT,
  requested_task_revision INTEGER,
  applied_task_revision INTEGER,
  task_state TEXT,
  reason_code TEXT,
  wait_reason_code TEXT,
  snapshot_cutoff TEXT,
  source_type TEXT,
  source_id TEXT,
  aggregate_version INTEGER NOT NULL DEFAULT 1,
  updated_at TEXT NOT NULL,
  UNIQUE(tenant_id, command_id)
);
CREATE INDEX IF NOT EXISTS commands_sched_idx ON commands(tenant_id, status, accepted_at);
CREATE TABLE IF NOT EXISTS executions (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  execution_id TEXT NOT NULL,
  tenant_id TEXT NOT NULL,
  command_id TEXT NOT NULL,
  task_id TEXT NOT NULL,
  task_item_id TEXT NOT NULL,
  payload_hash TEXT NOT NULL,
  status TEXT NOT NULL,
  call_id TEXT,
  agent_version_id TEXT,
  created_at TEXT NOT NULL,
  UNIQUE(tenant_id, execution_id)
);
CREATE TABLE IF NOT EXISTS calls (
  call_id TEXT PRIMARY KEY,
  tenant_id TEXT NOT NULL,
  execution_id TEXT NOT NULL,
  command_id TEXT NOT NULL,
  task_id TEXT NOT NULL,
  task_item_id TEXT NOT NULL,
  callee TEXT NOT NULL,
  cell_id TEXT,
  trunk_id TEXT,
  egress_pool_id TEXT,
  agent_version_id TEXT,
  agent_config_sha256 TEXT,
  ai_metrics_json TEXT,
  call_state TEXT NOT NULL,
  call_version INTEGER NOT NULL DEFAULT 1,
  outcome TEXT,
  reason_code TEXT,
  started_at TEXT,
  ended_at TEXT,
  duration_ms INTEGER,
  created_at TEXT NOT NULL,
  UNIQUE(tenant_id, execution_id)
);
CREATE INDEX IF NOT EXISTS calls_active_idx ON calls(tenant_id, call_state);
CREATE TABLE IF NOT EXISTS attempts (
  attempt_id TEXT PRIMARY KEY,
  call_id TEXT NOT NULL REFERENCES calls(call_id),
  attempt_number INTEGER NOT NULL,
  route_policy_id TEXT NOT NULL,
  caller_profile_id TEXT NOT NULL,
  target_uri TEXT NOT NULL,
  status TEXT NOT NULL,
  started_at TEXT,
  ended_at TEXT,
  created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS transcripts (
  segment_id TEXT PRIMARY KEY,
  call_id TEXT NOT NULL REFERENCES calls(call_id),
  turn_index INTEGER NOT NULL,
  speaker TEXT NOT NULL,
  text TEXT NOT NULL,
  revision INTEGER NOT NULL,
  is_final INTEGER NOT NULL,
  playback_state TEXT NOT NULL,
  updated_at TEXT NOT NULL,
  UNIQUE(call_id, turn_index, speaker)
);
CREATE TABLE IF NOT EXISTS recordings (
  recording_id TEXT PRIMARY KEY,
  call_id TEXT NOT NULL REFERENCES calls(call_id),
  tenant_id TEXT NOT NULL,
  content_type TEXT NOT NULL,
  size_bytes INTEGER NOT NULL,
  checksum_algorithm TEXT NOT NULL,
  checksum TEXT NOT NULL,
  channels INTEGER NOT NULL,
  sample_rate_hz INTEGER NOT NULL,
  duration_ms INTEGER NOT NULL,
  status TEXT NOT NULL,
  oss_id TEXT,
  upload_id TEXT,
  failure_code TEXT,
  created_at TEXT NOT NULL,
  verified_at TEXT
);
CREATE TABLE IF NOT EXISTS uploads (
  upload_id TEXT PRIMARY KEY,
  recording_id TEXT NOT NULL REFERENCES recordings(recording_id),
  tenant_id TEXT NOT NULL,
  expected_size INTEGER NOT NULL,
  expected_checksum TEXT NOT NULL,
  expires_at TEXT NOT NULL,
  status TEXT NOT NULL,
  object_path TEXT NOT NULL,
  created_at TEXT NOT NULL,
  completed_at TEXT
);
CREATE INDEX IF NOT EXISTS uploads_recording_idx ON uploads(tenant_id, recording_id, created_at);
CREATE TABLE IF NOT EXISTS events (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  event_id TEXT NOT NULL,
  tenant_id TEXT NOT NULL,
  tenant_key TEXT NOT NULL,
  trace_id TEXT NOT NULL,
  event_type TEXT NOT NULL,
  aggregate_type TEXT NOT NULL,
  aggregate_id TEXT NOT NULL,
  aggregate_version INTEGER NOT NULL,
  occurred_at TEXT NOT NULL,
  payload_json TEXT NOT NULL,
  body_json TEXT NOT NULL,
  published INTEGER NOT NULL DEFAULT 0,
  broker_status TEXT NOT NULL DEFAULT 'pending',
  broker_attempts INTEGER NOT NULL DEFAULT 0,
  published_at TEXT,
  next_attempt_at TEXT,
  saas_applied INTEGER,
  replay_of TEXT,
  UNIQUE(tenant_id, event_id)
);
CREATE INDEX IF NOT EXISTS events_outbox_idx ON events(published, next_attempt_at, occurred_at);
CREATE TABLE IF NOT EXISTS replays (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  replay_id TEXT NOT NULL UNIQUE,
  command_id TEXT NOT NULL,
  tenant_id TEXT NOT NULL,
  source_type TEXT NOT NULL,
  source_id TEXT NOT NULL,
  snapshot_cutoff TEXT NOT NULL,
  status TEXT NOT NULL,
  sent_count INTEGER NOT NULL DEFAULT 0,
  failure_code TEXT,
  retry_attempts INTEGER NOT NULL DEFAULT 0,
  next_attempt_at TEXT,
  created_at TEXT NOT NULL,
  updated_at TEXT NOT NULL,
  UNIQUE(tenant_id, command_id)
);
CREATE TABLE IF NOT EXISTS replay_items (
  replay_id TEXT NOT NULL REFERENCES replays(replay_id),
  event_id TEXT NOT NULL,
  published_at TEXT NOT NULL,
  PRIMARY KEY(replay_id, event_id)
);
CREATE TABLE IF NOT EXISTS inbox (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  event_id TEXT NOT NULL,
  tenant_id TEXT NOT NULL,
  applied_at TEXT NOT NULL,
  body_json TEXT NOT NULL,
  UNIQUE(tenant_id, event_id)
);
CREATE TABLE IF NOT EXISTS dead_letters (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  tenant_key TEXT,
  routing_key TEXT,
  body_json TEXT NOT NULL,
  reason_code TEXT NOT NULL,
  created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS published_commands (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  command_id TEXT NOT NULL,
  tenant_id TEXT NOT NULL,
  body_json TEXT NOT NULL,
  body_hash TEXT NOT NULL,
  status TEXT NOT NULL,
  retry_attempts INTEGER NOT NULL DEFAULT 0,
  next_attempt_at TEXT,
  last_error_code TEXT,
  created_at TEXT NOT NULL,
  UNIQUE(tenant_id, command_id)
);
CREATE TABLE IF NOT EXISTS control_barriers (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  command_id TEXT NOT NULL,
  tenant_id TEXT NOT NULL,
  cell_id TEXT NOT NULL,
  status TEXT NOT NULL,
  applied_at TEXT,
  UNIQUE(tenant_id, command_id, cell_id)
);
CREATE TABLE IF NOT EXISTS storage_operations (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  tenant_id TEXT NOT NULL,
  actor_id TEXT NOT NULL,
  operation_type TEXT NOT NULL,
  target_id TEXT NOT NULL,
  idempotency_key TEXT NOT NULL,
  payload_hash TEXT NOT NULL,
  status INTEGER NOT NULL,
  response_json TEXT NOT NULL,
  created_at TEXT NOT NULL,
  updated_at TEXT NOT NULL,
  UNIQUE(tenant_id, actor_id, operation_type, target_id, idempotency_key)
);
CREATE TABLE IF NOT EXISTS leases (
  lease_name TEXT PRIMARY KEY,
  owner_id TEXT NOT NULL,
  expires_at TEXT NOT NULL,
  generation INTEGER NOT NULL DEFAULT 0
);
"""


class Store:
    def __init__(self, path: str | os.PathLike[str]) -> None:
        requested = str(path)
        self._shared_memory = requested == ":memory:"
        self.path = (
            f"file:agent_call_{uuid.uuid4().hex}?mode=memory&cache=shared"
            if self._shared_memory
            else requested
        )
        self._keeper: sqlite3.Connection | None = None
        if self._shared_memory:
            self._keeper = sqlite3.connect(
                self.path, uri=True, timeout=10, isolation_level=None
            )
        else:
            Path(self.path).parent.mkdir(parents=True, exist_ok=True)
        self._init_lock = threading.Lock()
        self.init_schema()

    def close(self) -> None:
        if self._keeper is not None:
            self._keeper.close()
            self._keeper = None

    def connect(self) -> sqlite3.Connection:
        con = sqlite3.connect(
            self.path, uri=self._shared_memory, timeout=10, isolation_level=None
        )
        con.row_factory = sqlite3.Row
        con.execute("PRAGMA foreign_keys = ON")
        con.execute("PRAGMA busy_timeout = 10000")
        return con

    def init_schema(self) -> None:
        with self._init_lock:
            con = self.connect()
            try:
                con.executescript(SCHEMA_SQL)
                version_row = con.execute(
                    "SELECT MAX(version) AS version FROM schema_migrations"
                ).fetchone()
                version = int(version_row["version"] or 0)
                if version < 2 and not self._has_column(con, "commands", "id"):
                    self._migrate_identity_keys(con)
                if version < 4:
                    self._add_compatibility_columns(con)
                con.execute(
                    "INSERT OR IGNORE INTO schema_migrations(version, applied_at) VALUES(3, ?)",
                    (iso(),),
                )
                con.execute(
                    "INSERT OR IGNORE INTO schema_migrations(version, applied_at) VALUES(4, ?)",
                    (iso(),),
                )
            finally:
                con.close()

    @staticmethod
    def _has_column(con: sqlite3.Connection, table: str, column: str) -> bool:
        table_info_sql = {
            "commands": "PRAGMA table_info(commands)",
            "executions": "PRAGMA table_info(executions)",
            "calls": "PRAGMA table_info(calls)",
            "events": "PRAGMA table_info(events)",
            "replays": "PRAGMA table_info(replays)",
            "replay_items": "PRAGMA table_info(replay_items)",
            "inbox": "PRAGMA table_info(inbox)",
            "published_commands": "PRAGMA table_info(published_commands)",
            "leases": "PRAGMA table_info(leases)",
        }
        return any(
            row["name"] == column
            for row in con.execute(table_info_sql[table]).fetchall()
        )

    def _migrate_identity_keys(self, con: sqlite3.Connection) -> None:
        # The first release used globally unique command/execution/event IDs.  Keep
        # the facts, but move uniqueness to (tenant_id, public_id) so independent
        # tenants may safely reuse their own idempotency namespace.
        tables = (
            "commands",
            "executions",
            "calls",
            "events",
            "replays",
            "replay_items",
            "inbox",
            "published_commands",
        )
        con.execute("PRAGMA foreign_keys=OFF")
        con.execute("DROP INDEX IF EXISTS commands_sched_idx")
        rename_sql = {
            "commands": "ALTER TABLE commands RENAME TO _legacy_commands",
            "executions": "ALTER TABLE executions RENAME TO _legacy_executions",
            "calls": "ALTER TABLE calls RENAME TO _legacy_calls",
            "events": "ALTER TABLE events RENAME TO _legacy_events",
            "replays": "ALTER TABLE replays RENAME TO _legacy_replays",
            "replay_items": "ALTER TABLE replay_items RENAME TO _legacy_replay_items",
            "inbox": "ALTER TABLE inbox RENAME TO _legacy_inbox",
            "published_commands": "ALTER TABLE published_commands RENAME TO _legacy_published_commands",
        }
        for table in tables:
            con.execute(rename_sql[table])
        con.executescript(SCHEMA_SQL)
        # SCHEMA_SQL enables foreign keys; keep them off while the disposable
        # tables are rolled back and the legacy names are restored.
        con.execute("PRAGMA foreign_keys=OFF")
        committed = False
        con.execute("BEGIN")
        try:
            con.execute(
                """INSERT INTO commands(
                   command_id, tenant_id, tenant_key, command_type, status,
                   payload_json, payload_hash, operation_type, operation_target,
                   task_id, execution_id, call_id, trace_id, issued_at, not_after,
                   admission_deadline, accepted_at, waiting_since,
                   requested_task_revision, applied_task_revision, task_state,
                   reason_code, wait_reason_code, snapshot_cutoff, source_type,
                   source_id, aggregate_version, updated_at)
                   SELECT command_id, tenant_id, tenant_key, command_type, status,
                   payload_json, payload_hash, command_type,
                   COALESCE(task_id, source_id, ''), task_id, execution_id, call_id,
                   trace_id, issued_at, not_after, admission_deadline, accepted_at,
                   waiting_since, requested_task_revision, applied_task_revision,
                   task_state, reason_code, wait_reason_code, snapshot_cutoff,
                   source_type, source_id, aggregate_version, updated_at
                   FROM _legacy_commands"""
            )
            con.execute(
                """INSERT INTO executions(
                   execution_id, tenant_id, command_id, task_id, task_item_id,
                   payload_hash, status, call_id, created_at)
                   SELECT execution_id, tenant_id, command_id, task_id, task_item_id,
                   payload_hash, status, call_id, created_at
                   FROM _legacy_executions"""
            )
            con.execute(
                """INSERT INTO calls(
                   call_id, tenant_id, execution_id, command_id, task_id,
                   task_item_id, callee, cell_id, trunk_id, egress_pool_id,
                   call_state, call_version, outcome, reason_code, started_at,
                   ended_at, duration_ms, created_at)
                   SELECT call_id, tenant_id, execution_id, command_id, task_id,
                   task_item_id, callee, cell_id, trunk_id, egress_pool_id,
                   call_state, call_version, outcome, reason_code, started_at,
                   ended_at, duration_ms, created_at
                   FROM _legacy_calls"""
            )
            con.execute(
                """INSERT INTO events(
                   event_id, tenant_id, tenant_key, trace_id, event_type,
                   aggregate_type, aggregate_id, aggregate_version, occurred_at,
                   payload_json, body_json, published, broker_status,
                   broker_attempts, published_at, next_attempt_at, saas_applied,
                   replay_of)
                   SELECT event_id, tenant_id, tenant_key, trace_id, event_type,
                   aggregate_type, aggregate_id, aggregate_version, occurred_at,
                   payload_json, body_json, published, broker_status,
                   broker_attempts, published_at, next_attempt_at, saas_applied,
                   replay_of
                   FROM _legacy_events"""
            )
            con.execute(
                """INSERT INTO replays(
                   replay_id, command_id, tenant_id, source_type, source_id,
                   snapshot_cutoff, status, sent_count, failure_code, created_at,
                   updated_at)
                   SELECT replay_id, command_id, tenant_id, source_type, source_id,
                   snapshot_cutoff, status, sent_count, failure_code, created_at,
                   updated_at
                   FROM _legacy_replays"""
            )
            con.execute(
                "INSERT INTO replay_items(replay_id, event_id, published_at) "
                "SELECT replay_id, event_id, published_at FROM _legacy_replay_items"
            )
            con.execute(
                "INSERT INTO inbox(event_id, tenant_id, applied_at, body_json) "
                "SELECT event_id, tenant_id, applied_at, body_json FROM _legacy_inbox"
            )
            con.execute(
                """INSERT INTO published_commands(
                   command_id, tenant_id, body_json, body_hash, status, created_at)
                   SELECT command_id, tenant_id, body_json, body_hash, status,
                   created_at FROM _legacy_published_commands"""
            )
            con.execute("COMMIT")
            committed = True
        except Exception:
            con.execute("ROLLBACK")
            raise
        finally:
            if committed:
                for table in tables:
                    con.execute(f"DROP TABLE IF EXISTS _legacy_{table}")
            else:
                # Keep the original tables usable if the copy failed.  The new
                # schema is disposable because no migration version was saved.
                for table in reversed(tables):
                    con.execute(f"DROP TABLE IF EXISTS {table}")
                for table in tables:
                    con.execute(f"ALTER TABLE _legacy_{table} RENAME TO {table}")
            con.execute("PRAGMA foreign_keys=ON")

    def _add_compatibility_columns(self, con: sqlite3.Connection) -> None:
        additions = {
            "executions": (("agent_version_id", "TEXT"),),
            "calls": (
                ("agent_version_id", "TEXT"),
                ("agent_config_sha256", "TEXT"),
                ("ai_metrics_json", "TEXT"),
            ),
            "published_commands": (
                ("retry_attempts", "INTEGER NOT NULL DEFAULT 0"),
                ("next_attempt_at", "TEXT"),
                ("last_error_code", "TEXT"),
            ),
            "replays": (
                ("retry_attempts", "INTEGER NOT NULL DEFAULT 0"),
                ("next_attempt_at", "TEXT"),
            ),
            "leases": (("generation", "INTEGER NOT NULL DEFAULT 0"),),
        }
        alter_sql = {
            (
                "executions",
                "agent_version_id",
            ): "ALTER TABLE executions ADD COLUMN agent_version_id TEXT",
            (
                "calls",
                "agent_version_id",
            ): "ALTER TABLE calls ADD COLUMN agent_version_id TEXT",
            (
                "calls",
                "agent_config_sha256",
            ): "ALTER TABLE calls ADD COLUMN agent_config_sha256 TEXT",
            (
                "calls",
                "ai_metrics_json",
            ): "ALTER TABLE calls ADD COLUMN ai_metrics_json TEXT",
            (
                "published_commands",
                "retry_attempts",
            ): "ALTER TABLE published_commands ADD COLUMN retry_attempts INTEGER NOT NULL DEFAULT 0",
            (
                "published_commands",
                "next_attempt_at",
            ): "ALTER TABLE published_commands ADD COLUMN next_attempt_at TEXT",
            (
                "published_commands",
                "last_error_code",
            ): "ALTER TABLE published_commands ADD COLUMN last_error_code TEXT",
            (
                "replays",
                "retry_attempts",
            ): "ALTER TABLE replays ADD COLUMN retry_attempts INTEGER NOT NULL DEFAULT 0",
            (
                "replays",
                "next_attempt_at",
            ): "ALTER TABLE replays ADD COLUMN next_attempt_at TEXT",
            (
                "leases",
                "generation",
            ): "ALTER TABLE leases ADD COLUMN generation INTEGER NOT NULL DEFAULT 0",
        }
        for table, columns in additions.items():
            for column, _definition in columns:
                if not self._has_column(con, table, column):
                    con.execute(alter_sql[(table, column)])

    @contextmanager
    def tx(self) -> Iterator[sqlite3.Connection]:
        con = self.connect()
        con.execute("BEGIN IMMEDIATE")
        try:
            yield con
            con.commit()
        except Exception:
            con.rollback()
            raise
        finally:
            con.close()

    def one(self, sql: str, params: tuple[Any, ...] = ()) -> sqlite3.Row:
        con = self.connect()
        try:
            # pi-lens-ignore: python-sql-injection
            return cast(sqlite3.Row, con.execute(sql, params).fetchone())
        finally:
            con.close()

    def all(self, sql: str, params: tuple[Any, ...] = ()) -> list[sqlite3.Row]:
        con = self.connect()
        try:
            # pi-lens-ignore: python-sql-injection
            return list(con.execute(sql, params).fetchall())
        finally:
            con.close()


class InMemoryBroker:
    """A strict direct-exchange broker used for deterministic acceptance tests."""

    def __init__(self, max_messages: int = 0, max_bytes: int = 0) -> None:
        self._queues: dict[str, queue.Queue[dict[str, Any]]] = {}
        self._queue_bytes: dict[str, int] = {}
        self.max_messages = max_messages
        self.max_bytes = max_bytes
        self._routes: dict[str, str] = {}
        self._lock = threading.Lock()
        self.outage = False
        self.blocked = False
        self.fail_next: str | None = None
        self.fail_after: int | None = None
        self._successful_publishes = 0
        self.published: list[dict[str, Any]] = []

    def declare_tenant(self, tenant_key: str) -> None:
        name = queue_name(tenant_key)
        route = routing_key(tenant_key)
        with self._lock:
            self._queues.setdefault(name, queue.Queue(maxsize=self.max_messages))
            self._queue_bytes.setdefault(name, 0)
            self._routes[route] = name

    def declare_events(self) -> None:
        with self._lock:
            self._queues.setdefault(
                "agent-call.saas.events.v1", queue.Queue(maxsize=self.max_messages)
            )
            self._queue_bytes.setdefault("agent-call.saas.events.v1", 0)

    def publish(
        self,
        exchange: str,
        route: str,
        body: dict[str, Any],
        message_id: str | None = None,
        headers: dict[str, str] | None = None,
    ) -> None:
        if self.outage:
            raise BrokerError()
        with self._lock:
            if self.blocked:
                raise BrokerError("broker is blocked; publication is retained")
            if self.fail_next:
                reason = self.fail_next
                self.fail_next = None
                raise BrokerError(reason)
            if (
                self.fail_after is not None
                and self._successful_publishes >= self.fail_after
            ):
                raise BrokerError("publisher confirmation was lost")
            if exchange == "agent-call.commands.v1":
                target = self._routes.get(route)
            else:
                target = (
                    "agent-call.saas.events.v1"
                    if route.startswith("agent-call.")
                    else None
                )
            if target is None or target not in self._queues:
                raise BrokerError("mandatory publish was unroutable")
            encoded_size = len(canonical(body).encode("utf-8"))
            if (
                self.max_bytes
                and self._queue_bytes[target] + encoded_size > self.max_bytes
            ):
                raise BrokerError("queue byte limit reached; publication is retained")
            message = {
                "exchange": exchange,
                "routing_key": route,
                "body": body,
                "message_id": message_id,
                "headers": headers or {},
                "queue_name": target,
                "_size": encoded_size,
            }
            try:
                self._queues[target].put_nowait(message)
            except queue.Full as exc:
                raise BrokerError("queue is full; publication is retained") from exc
            self._queue_bytes[target] += encoded_size
            self._successful_publishes += 1
            self.published.append(message)

    def ack(self, message: dict[str, Any]) -> None:
        return

    def reject(self, message: dict[str, Any]) -> None:
        return

    @staticmethod
    def _message_size(message: dict[str, Any]) -> int:
        try:
            return max(0, int(message.get("_size", 0)))
        except (TypeError, ValueError):
            return 0

    def requeue(self, message: dict[str, Any]) -> None:
        name = message.get("queue_name")
        if not name:
            raise BrokerError("message has no queue context")
        with self._lock:
            target = self._queues.get(name)
            if target is None:
                raise BrokerError("queue is not declared")
            try:
                target.put_nowait(message)
            except queue.Full as exc:
                raise BrokerError("queue is full while requeuing") from exc
            self._queue_bytes[name] += self._message_size(message)

    def consume(self, name: str) -> dict[str, Any] | None:
        if self.outage:
            raise BrokerError()
        with self._lock:
            target = self._queues.get(name)
            if target is None:
                raise BrokerError("queue is not declared")
            try:
                message = target.get_nowait()
            except queue.Empty:
                return None
            self._queue_bytes[name] = max(
                0, self._queue_bytes[name] - self._message_size(message)
            )
            return message

    def depth(self, name: str) -> int:
        with self._lock:
            target = self._queues.get(name)
            return target.qsize() if target is not None else 0

    def queue_bytes(self, name: str) -> int:
        with self._lock:
            return self._queue_bytes.get(name, 0)

    def has_route(self, route: str) -> bool:
        with self._lock:
            return route in self._routes

    def ready(self) -> bool:
        return not self.outage and not self.blocked


class PikaBroker:
    """RabbitMQ adapter. Import and connection are optional in local unit tests."""

    def __init__(
        self, url: str, max_messages: int = 1000, max_bytes: int = 16 * 1024 * 1024
    ) -> None:
        try:
            pika = importlib.import_module("pika")
        except ImportError as exc:  # pragma: no cover
            raise ConfigurationError(
                "BROKER_MODE=rabbit requires the pika package"
            ) from exc
        self.pika = pika
        self.max_messages = max_messages
        self.max_bytes = max_bytes
        self._declared_routes: set[str] = set()
        try:
            self.connection = pika.BlockingConnection(pika.URLParameters(url))
            self.channel = self.connection.channel()
            self.channel.confirm_delivery()
            self.channel.basic_qos(prefetch_count=1)
            self.channel.exchange_declare(
                exchange="agent-call.commands.v1", exchange_type="direct", durable=True
            )
            self.channel.exchange_declare(
                exchange="agent-call.dead.v1", exchange_type="direct", durable=True
            )
            self.channel.exchange_declare(
                exchange="agent-call.events.v1", exchange_type="topic", durable=True
            )
            for queue, dead_route in (
                ("agent-call.commands.dead.v1", "agent-call.commands.dead"),
                ("agent-call.events.dead.v1", "agent-call.events.dead"),
            ):
                self.channel.queue_declare(
                    queue=queue,
                    durable=True,
                    arguments={
                        "x-max-length": self.max_messages,
                        "x-max-length-bytes": self.max_bytes,
                        "x-overflow": "reject-publish",
                    },
                )
                self.channel.queue_bind(
                    queue=queue,
                    exchange="agent-call.dead.v1",
                    routing_key=dead_route,
                )
            self.channel.queue_declare(
                queue="agent-call.saas.events.v1",
                durable=True,
                arguments={
                    "x-max-length": self.max_messages,
                    "x-max-length-bytes": self.max_bytes,
                    "x-overflow": "reject-publish",
                    "x-dead-letter-exchange": "agent-call.dead.v1",
                    "x-dead-letter-routing-key": "agent-call.events.dead",
                },
            )
            self.channel.queue_bind(
                queue="agent-call.saas.events.v1",
                exchange="agent-call.events.v1",
                routing_key="agent-call.#",
            )
            self._blocked = False
            add_blocked = getattr(
                self.connection, "add_on_connection_blocked_callback", None
            )
            if callable(add_blocked):
                add_blocked(
                    lambda _connection, _reason: setattr(self, "_blocked", True)
                )
                add_unblocked = getattr(
                    self.connection, "add_on_connection_unblocked_callback", None
                )
                if callable(add_unblocked):
                    add_unblocked(lambda _connection: setattr(self, "_blocked", False))
        except Exception as exc:  # pragma: no cover
            raise BrokerError(
                f"RabbitMQ connection failed: {type(exc).__name__}"
            ) from exc

    def declare_tenant(self, tenant_key: str) -> None:  # pragma: no cover
        name = queue_name(tenant_key)
        route = routing_key(tenant_key)
        self.channel.queue_declare(
            queue=name,
            durable=True,
            arguments={
                "x-max-length": self.max_messages,
                "x-max-length-bytes": self.max_bytes,
                "x-overflow": "reject-publish",
                "x-dead-letter-exchange": "agent-call.dead.v1",
                "x-dead-letter-routing-key": "agent-call.commands.dead",
            },
        )
        self.channel.queue_bind(
            queue=name, exchange="agent-call.commands.v1", routing_key=route
        )
        self._declared_routes.add(route)

    def declare_events(self) -> None:  # pragma: no cover
        return

    def publish(
        self,
        exchange: str,
        route: str,
        body: dict[str, Any],
        message_id: str | None = None,
        headers: dict[str, str] | None = None,
    ) -> None:  # pragma: no cover
        try:
            props = self.pika.BasicProperties(
                delivery_mode=2,
                content_type="application/json",
                message_id=message_id,
                headers=headers or {},
            )
            confirmed = self.channel.basic_publish(
                exchange=exchange,
                routing_key=route,
                body=canonical(body).encode("utf-8"),
                properties=props,
                mandatory=True,
            )
            if confirmed is not None and not confirmed:
                raise BrokerError("publisher confirmation was negative")
        except BrokerError:
            raise
        except Exception as exc:
            raise BrokerError(f"RabbitMQ publish failed: {type(exc).__name__}") from exc

    def consume(self, name: str) -> dict[str, Any] | None:  # pragma: no cover
        try:
            method, props, body = self.channel.basic_get(queue=name, auto_ack=False)
            if method is None or props is None or body is None:
                return None
            try:
                parsed = _loads(body.decode("utf-8"))
            except (ServiceError, UnicodeDecodeError):
                parsed = {"_invalid_payload": True}
            return {
                "body": parsed,
                "message_id": props.message_id,
                "headers": props.headers or {},
                "routing_key": getattr(method, "routing_key", None),
                "queue_name": name,
                "_delivery_tag": method.delivery_tag,
            }
        except Exception as exc:
            raise BrokerError(f"RabbitMQ consume failed: {type(exc).__name__}") from exc

    def requeue(self, message: dict[str, Any]) -> None:  # pragma: no cover
        tag = message.get("_delivery_tag")
        if tag is None:
            return
        try:
            self.channel.basic_nack(tag, requeue=True)
        except Exception as exc:
            raise BrokerError("RabbitMQ requeue failed") from exc

    def depth(self, name: str) -> int:  # pragma: no cover
        try:
            return int(
                self.channel.queue_declare(
                    queue=name, passive=True
                ).method.message_count
            )
        except Exception as exc:
            raise BrokerError("RabbitMQ queue depth failed") from exc

    def has_route(self, route: str) -> bool:  # pragma: no cover
        return route in self._declared_routes

    def ack(self, message: dict[str, Any]) -> None:  # pragma: no cover
        tag = message.get("_delivery_tag")
        if tag is None:
            return
        try:
            self.channel.basic_ack(tag)
        except Exception as exc:
            raise BrokerError("RabbitMQ acknowledgement failed") from exc

    def reject(self, message: dict[str, Any]) -> None:  # pragma: no cover
        tag = message.get("_delivery_tag")
        if tag is None:
            return
        try:
            self.channel.basic_nack(tag, requeue=False)
        except Exception as exc:
            raise BrokerError("RabbitMQ rejection failed") from exc

    def ready(self) -> bool:  # pragma: no cover
        return bool(
            self.connection.is_open and self.channel.is_open and not self._blocked
        )


class MockObjectStore:
    _id_pattern = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")

    def __init__(self, root: str | os.PathLike[str], secret: str | None = None) -> None:
        self.root = Path(root)
        self.root.mkdir(parents=True, exist_ok=True)
        seed = secret or os.environ.get("MOCK_UPLOAD_SECRET") or self.root.as_posix()
        self.secret = seed.encode("utf-8")

    @classmethod
    def _check_id(cls, upload_id: str) -> None:
        if not isinstance(upload_id, str) or not cls._id_pattern.fullmatch(upload_id):
            raise ValidationError("UPLOAD_ID_INVALID", "upload id is invalid")

    def path_for(self, upload_id: str) -> Path:
        self._check_id(upload_id)
        return self.root / f"{upload_id}.part"

    def final_path_for(self, upload_id: str) -> Path:
        self._check_id(upload_id)
        return self.root / upload_id

    def has_final(self, upload_id: str) -> bool:
        return self.final_path_for(upload_id).is_file()

    def token(self, upload_id: str, tenant_id: str) -> str:
        self._check_id(upload_id)
        return hashlib.sha256(
            self.secret + b":" + tenant_id.encode() + b":" + upload_id.encode()
        ).hexdigest()

    def put(self, upload_id: str, tenant_id: str, data: bytes, token: str) -> None:
        if not secrets.compare_digest(token, self.token(upload_id, tenant_id)):
            raise ValidationError("UPLOAD_TOKEN_INVALID", "upload token is invalid")
        if not isinstance(data, bytes):
            raise ValidationError("OBJECT_BODY_INVALID", "object body must be bytes")
        self.path_for(upload_id).write_bytes(data)

    def verify_and_promote(
        self, upload_id: str, expected_size: int, expected_checksum: str
    ) -> str:
        final = self.final_path_for(upload_id)
        path = self.path_for(upload_id)
        candidate = final if final.exists() else path
        if not candidate.exists():
            raise ValidationError("OBJECT_MISSING", "object was not uploaded")
        data = candidate.read_bytes()
        actual = hashlib.sha256(data).hexdigest()
        if len(data) != expected_size or actual != expected_checksum:
            raise ValidationError(
                "OBJECT_VERIFICATION_FAILED",
                "size or checksum does not match the upload contract",
            )
        if candidate == path:
            os.replace(path, final)
        return f"oss://mock/{upload_id}"


class AgentCallService:
    def __init__(
        self,
        db_path: str | os.PathLike[str] = ":memory:",
        profile_path: str | os.PathLike[str] = DEFAULT_PROFILE,
        object_dir: str | os.PathLike[str] | None = None,
        broker: InMemoryBroker | PikaBroker | None = None,
        mode: str | None = None,
        clock: Callable[[], datetime] = utcnow,
        sip_readonly: SipReadonlyClient | Any | None = None,
        start_background: bool = False,
    ) -> None:
        self.clock = clock
        self.profile_path = Path(profile_path)
        self.profile = _load_json_file(self.profile_path)
        self.mq_schema = _load_json_file(DEFAULT_MQ_SCHEMA)
        self.mode = mode or self.profile.get("mode", "mock")
        self._validate_profile()
        self.sip_readonly = sip_readonly
        self.store = Store(db_path)
        self.instance_id = new_id("instance")
        object_root = (
            Path(object_dir)
            if object_dir
            else (
                Path(db_path).parent / "objects"
                if str(db_path) != ":memory:"
                else Path("data") / "agent-call-objects"
            )
        )
        self.object_root = object_root
        self.object_store = MockObjectStore(
            object_root, os.environ.get("MOCK_UPLOAD_SECRET")
        )
        self.journal = ProtocolJournal(object_root / "protocol.jsonl")
        self.mock_sip = MockSIPPeer(self.journal)
        self.mock_ari = MockARI(self.journal)
        self.mock_rtp = MockRTP(self.journal)
        self.mock_ai = MockAI(self.journal)
        if broker is None:
            broker_mode = self.profile.get("provider_modes", {}).get(
                "rabbitmq", "memory"
            )
            if broker_mode == "rabbit" or os.environ.get("BROKER_MODE") == "rabbit":
                broker_url = os.environ.get("RABBITMQ_URL")
                if not broker_url:
                    raise ConfigurationError("BROKER_MODE=rabbit requires RABBITMQ_URL")
                broker = PikaBroker(
                    broker_url,
                    max_messages=_integer(
                        self._limits().get("max_queue_messages", 1000)
                    ),
                    max_bytes=_integer(
                        self._limits().get("max_queue_bytes", 16 * 1024 * 1024)
                    ),
                )
            else:
                broker = InMemoryBroker(
                    _integer(self._limits().get("max_queue_messages", 1000)),
                    _integer(self._limits().get("max_queue_bytes", 16 * 1024 * 1024)),
                )
        self.broker = broker
        self.profile.setdefault("provider_modes", {})["rabbitmq"] = (
            "rabbit" if isinstance(broker, PikaBroker) else "memory"
        )
        self._provider_status = {
            str(component): str(provider)
            for component, provider in self.profile.get("provider_modes", {}).items()
        }
        self._unrouteable_tenants: set[str] = set()
        self._seed()
        self._tenant_order = [
            item["tenant_id"] for item in self.profile.get("tenants", [])
        ]
        self._cell_health = {
            item["cell_id"]: bool(item.get("enabled", True))
            for item in self.profile.get("cells", [])
        }
        self._cursor = 0
        self._failures: dict[str, str] = {}
        self._call_threads: dict[str, threading.Thread] = {}
        self._stop = threading.Event()
        self._background: threading.Thread | None = None
        self._last_error: str | None = None
        self._last_wall_clock: datetime | None = None
        self._last_monotonic: float | None = None
        self._clock_jump = False
        self._recover_durable_state()
        if start_background:
            self.start()

    def _validate_profile(self) -> None:
        if self.mode not in {"mock", "mixed", "real"}:
            raise ConfigurationError("mode must be mock, mixed, or real")
        providers = self.profile.get("provider_modes", {})
        if not isinstance(providers, dict):
            raise ConfigurationError("provider_modes must be an object")
        mock_values = {
            "saas": "mock",
            "database": "sqlite",
            "rabbitmq": "memory",
            "sip": "mock",
            "asterisk": "mock",
            "asr": "mock",
            "llm": "mock",
            "tts": "mock",
            "oss": "mock",
            "cloud": "fake-cli",
        }
        if self.mode == "real":
            raise ConfigurationError(
                "real mode is unavailable until every real provider adapter is configured; mock fallback is forbidden"
            )
        if self.mode == "mixed":
            raise ConfigurationError(
                "mixed mode requires explicit real adapters; this build has none"
            )
        unknown = set(providers) - set(mock_values)
        if unknown:
            raise ConfigurationError(f"unknown provider components: {sorted(unknown)}")
        missing = set(mock_values) - set(providers)
        if missing:
            raise ConfigurationError(f"provider modes are missing: {sorted(missing)}")
        for component, expected in mock_values.items():
            configured = providers[component]
            if configured != expected and not (
                component == "rabbitmq" and configured == "rabbit"
            ):
                raise ConfigurationError(
                    f"mock mode refuses provider {component}={configured!r}"
                )
        seen_ids: set[str] = set()
        seen_keys: set[str] = set()
        for tenant in self.profile.get("tenants", []):
            tenant_id = tenant.get("tenant_id")
            tenant_key = tenant.get("tenant_key")
            if not isinstance(tenant_id, str) or not tenant_id:
                raise ConfigurationError("tenant_id must be a non-empty string")
            if not isinstance(tenant_key, str) or not tenant_key:
                raise ConfigurationError("tenant_key must be a non-empty string")
            if tenant_id in seen_ids or tenant_key in seen_keys:
                raise ConfigurationError(
                    "tenant identifiers and tenant_key values must be unique"
                )
            seen_ids.add(tenant_id)
            seen_keys.add(tenant_key)
        agents = self.profile.get("agents", [])
        if not isinstance(agents, list) or not agents:
            raise ConfigurationError("at least one immutable agent version is required")
        seen_agent_ids: set[str] = set()
        for agent in agents:
            agent_id = (
                agent.get("agent_version_id") if isinstance(agent, dict) else None
            )
            if (
                not isinstance(agent_id, str)
                or not agent_id
                or agent_id in seen_agent_ids
            ):
                raise ConfigurationError("agent_version_id values must be unique")
            try:
                normalize_agent_config(agent, agent_id)
            except AIConfigError as exc:
                raise ConfigurationError(
                    f"invalid agent version {agent_id}: {exc.detail}"
                ) from exc
            seen_agent_ids.add(agent_id)
        if not self.profile.get("cells"):
            raise ConfigurationError("at least one cell is required")
        for cell in self.profile["cells"]:
            if _integer(cell.get("capacity", 0)) < 1:
                raise ConfigurationError("cell capacity must be positive")

    def _seed(self) -> None:
        now = iso(self.clock())
        limits = self.profile.get("limits", {})
        with self.store.tx() as con:
            for tenant in self.profile.get("tenants", []):
                con.execute(
                    "INSERT OR IGNORE INTO tenants(tenant_id, tenant_key, enabled, max_concurrency, max_cps, created_at) VALUES(?,?,?,?,?,?)",
                    (
                        tenant["tenant_id"],
                        tenant["tenant_key"],
                        _integer(tenant.get("enabled", True)),
                        limits.get("tenant_concurrency", 2),
                        limits.get("tenant_cps", 1),
                        now,
                    ),
                )
            for task in self.profile.get("tasks", []):
                con.execute(
                    "INSERT OR IGNORE INTO tasks(task_id, tenant_id, state, revision, updated_at) VALUES(?,?,?,?,?)",
                    (
                        task["task_id"],
                        task["tenant_id"],
                        task.get("state", "running"),
                        task.get("revision", 1),
                        now,
                    ),
                )
            for tenant in self.profile.get("tenants", []):
                for agent in self.profile.get("agents", []):
                    agent_id = agent["agent_version_id"]
                    normalized = normalize_agent_config(agent, agent_id)
                    con.execute(
                        "INSERT OR IGNORE INTO agent_versions(agent_version_id, tenant_id, config_json, content_sha256, immutable, created_at, published_at, created_by) VALUES(?,?,?,?,?,?,?,?)",
                        (
                            agent_id,
                            tenant["tenant_id"],
                            canonical(normalized),
                            ai_config_digest(normalized),
                            1,
                            now,
                            now,
                            "profile",
                        ),
                    )
        for tenant in self.profile.get("tenants", []):
            try:
                self.broker.declare_tenant(tenant["tenant_key"])
            except ValidationError:
                # Keep an over-budget tenant in durable state.  Its original
                # command is retained instead of being truncated or rerouted.
                self._unrouteable_tenants.add(tenant["tenant_id"])
        self.broker.declare_events()

    def _recover_durable_state(self) -> None:
        """Fence unfinished calls on restart; never infer that a channel ended."""
        with self.store.tx() as con:
            calls = con.execute(
                "SELECT * FROM calls WHERE call_state NOT IN ('ended', 'failed', 'reconciling')"
            ).fetchall()
            now = iso(self._now())
            for call in calls:
                con.execute(
                    "UPDATE calls SET call_state='reconciling', reason_code='RECOVERY_RECONCILING', call_version=call_version+1 WHERE call_id=?",
                    (call["call_id"],),
                )
                con.execute(
                    "UPDATE commands SET status='reconciling', reason_code='RECOVERY_RECONCILING', updated_at=? WHERE tenant_id=? AND command_id=? AND status NOT IN ('completed','failed','rejected','applied')",
                    (now, call["tenant_id"], call["command_id"]),
                )
                con.execute(
                    "UPDATE executions SET status='reconciling' WHERE tenant_id=? AND execution_id=? AND status NOT IN ('completed','failed','rejected')",
                    (call["tenant_id"], call["execution_id"]),
                )

    def start(self) -> None:
        if self._background and self._background.is_alive():
            return
        self._stop.clear()
        self._background = threading.Thread(
            target=self._run_loop, name="agent-call-loop", daemon=True
        )
        self._background.start()

    def stop(self) -> None:
        self._stop.set()
        deadline = time.monotonic() + 2
        if self._background:
            self._background.join(timeout=max(0, deadline - time.monotonic()))
        for thread in list(self._call_threads.values()):
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                break
            thread.join(timeout=remaining)
        self.store.close()

    def _run_loop(self) -> None:
        while not self._stop.is_set():
            try:
                self.process_once()
            except (
                ServiceError,
                OSError,
                RuntimeError,
                TypeError,
                ValueError,
                KeyError,
                sqlite3.DatabaseError,
            ) as exc:
                # The next tick retries durable outbox and waiting work. Do not log
                # payloads, credentials, or raw broker diagnostics.
                self._last_error = type(exc).__name__
            self._stop.wait(0.02)

    def _now(self) -> datetime:
        now = self.clock().astimezone(timezone.utc)
        monotonic = time.monotonic()
        if self._last_wall_clock is not None and self._last_monotonic is not None:
            wall_delta = (now - self._last_wall_clock).total_seconds()
            monotonic_delta = monotonic - self._last_monotonic
            if abs(wall_delta - monotonic_delta) > 2:
                self._clock_jump = True
        self._last_wall_clock = now
        self._last_monotonic = monotonic
        return now

    def _limits(self) -> dict[str, Any]:
        return self.profile.get("limits", {})

    def publish_agent_version(
        self,
        tenant_id: str,
        agent_version_id: str,
        config: dict[str, Any],
        actor_id: str = "internal",
    ) -> dict[str, Any]:
        self._tenant(tenant_id)
        if not isinstance(config, dict):
            raise ValidationError(
                "AI_CONFIG_INVALID", "agent version config must be an object"
            )
        if config.get("agent_version_id") not in {None, agent_version_id}:
            raise ValidationError(
                "AI_VERSION_ID_MISMATCH",
                "config agent_version_id does not match the publish request",
            )
        try:
            normalized = normalize_agent_config(config, agent_version_id)
        except AIConfigError as exc:
            raise ValidationError(exc.code, exc.detail) from exc
        content_sha256 = ai_config_digest(normalized)
        now = iso(self._now())
        existing = self.store.one(
            "SELECT * FROM agent_versions WHERE tenant_id=? AND agent_version_id=?",
            (tenant_id, agent_version_id),
        )
        if existing:
            if existing["content_sha256"] != content_sha256:
                raise ConflictError(
                    "AI_VERSION_IMMUTABLE_CONFLICT",
                    "published agent_version_id cannot be overwritten with different content",
                )
            return {
                "tenant_id": tenant_id,
                "agent_version_id": agent_version_id,
                "status": "reused",
                "immutable": True,
                "content_sha256": content_sha256,
            }
        with self.store.tx() as con:
            con.execute(
                "INSERT INTO agent_versions(agent_version_id, tenant_id, config_json, content_sha256, immutable, created_at, published_at, created_by) VALUES(?,?,?,?,?,?,?,?)",
                (
                    agent_version_id,
                    tenant_id,
                    canonical(normalized),
                    content_sha256,
                    1,
                    now,
                    now,
                    actor_id,
                ),
            )
        return {
            "tenant_id": tenant_id,
            "agent_version_id": agent_version_id,
            "status": "published",
            "immutable": True,
            "content_sha256": content_sha256,
        }

    def get_agent_version(
        self, tenant_id: str, agent_version_id: str
    ) -> dict[str, Any]:
        self._tenant(tenant_id)
        row = self.store.one(
            "SELECT * FROM agent_versions WHERE tenant_id=? AND agent_version_id=?",
            (tenant_id, agent_version_id),
        )
        if row is None:
            raise NotFoundError("agent version not found")
        raw_config = _loads(row["config_json"], {})
        if (
            row["immutable"] != 1
            or raw_config.get("agent_version_id") != agent_version_id
        ):
            raise ConfigurationError("stored agent version identity is invalid")
        try:
            config = normalize_agent_config(raw_config, agent_version_id)
        except (AIConfigError, ServiceError) as exc:
            raise ConfigurationError("stored agent version is invalid") from exc
        if ai_config_digest(config) != row["content_sha256"]:
            raise ConfigurationError("stored agent version digest does not match")
        return {
            "tenant_id": tenant_id,
            "agent_version_id": agent_version_id,
            "immutable": True,
            "content_sha256": row["content_sha256"],
            "created_at": row["created_at"],
            "published_at": row["published_at"],
            "created_by": row["created_by"],
            "config": config,
        }

    def _agent_config_for(
        self, tenant_id: str, agent_version_id: str
    ) -> dict[str, Any]:
        return cast(
            dict[str, Any],
            self.get_agent_version(tenant_id, agent_version_id)["config"],
        )

    def _conversation_engine(self, config: dict[str, Any]) -> ConversationEngine:
        return ConversationEngine(config, journal=self.journal)

    @staticmethod
    def _operation_hash(operation: str, target: str, body: dict[str, Any]) -> str:
        return digest({"operation": operation, "target": target, "body": body})

    def _route_config(self, route_policy_id: str) -> dict[str, Any]:
        for route in self.profile.get("routes", []):
            if route.get("route_policy_id") == route_policy_id and route.get(
                "allowed", True
            ):
                break
        else:
            raise ValidationError(
                "ROUTE_POLICY_NOT_ALLOWED", "route policy is not configured"
            )
        if self.sip_readonly is None:
            return route
        snapshot = self.sip_readonly.get_trunk(route["trunk_id"])
        config = snapshot.get("config")
        enabled = config.get("enabled") if isinstance(config, dict) else None
        if not isinstance(config, dict) or not isinstance(enabled, bool) or not enabled:
            raise SipReadonlyError(
                "SIP_TRUNK_NOT_PUBLISHED", "SIP Trunk is not published"
            )
        dynamic = dict(route)
        for field in (
            "codec_profile",
            "caller_ids",
            "dial_prefix",
            "egress_pool_id",
            "max_concurrency",
            "max_cps",
            "sip",
        ):
            if field in config:
                dynamic[field] = config[field]
        dynamic["trunk_revision"] = snapshot.get("revision")
        return dynamic

    def _tenant(self, tenant_id: str, tenant_key: str | None = None) -> sqlite3.Row:
        row = self.store.one("SELECT * FROM tenants WHERE tenant_id = ?", (tenant_id,))
        if (
            row is None
            or not row["enabled"]
            or (tenant_key is not None and row["tenant_key"] != tenant_key)
        ):
            raise ValidationError(
                "TENANT_BINDING_INVALID",
                "tenant_id and tenant_key are not an authorized pair",
            )
        return row

    def _validate_message(self, body: dict[str, Any]) -> None:
        if not isinstance(body, dict):
            raise ValidationError("SCHEMA_INVALID", "MQ message must be a JSON object")
        if len(canonical(body).encode("utf-8")) > self._limits().get(
            "max_mq_bytes", 262144
        ):
            raise ValidationError(
                "MESSAGE_TOO_LARGE", "message exceeds the MQ body limit"
            )
        if Draft202012Validator is not None:
            errors = sorted(
                Draft202012Validator(
                    self.mq_schema,
                    format_checker=FormatChecker() if FormatChecker else None,
                ).iter_errors(body),
                key=lambda error: list(error.path),
            )
            if errors:
                path = ".".join(str(part) for part in errors[0].path) or "body"
                raise ValidationError("SCHEMA_INVALID", f"invalid MQ message at {path}")
        else:
            required = {
                "schema_version",
                "command_type",
                "command_id",
                "tenant_id",
                "tenant_key",
                "trace_id",
                "issued_at",
                "not_after",
                "payload",
            }
            if not required.issubset(body):
                raise ValidationError(
                    "SCHEMA_INVALID", "required MQ fields are missing"
                )

    def _validate_command_times(self, body: dict[str, Any]) -> None:
        try:
            issued_at = parse_time(body["issued_at"])
            not_after = parse_time(body["not_after"])
        except (TypeError, ValueError, OverflowError, KeyError) as exc:
            raise ValidationError(
                "SCHEMA_INVALID", "command timestamps are invalid"
            ) from exc
        now = self._now()
        if issued_at > now + timedelta(seconds=5) or not_after <= issued_at:
            raise ValidationError(
                "SCHEMA_INVALID", "command timestamp window is invalid"
            )
        if not_after <= now:
            raise ConflictError("ADMISSION_EXPIRED", "command is past not_after")

    def _validate_event(self, body: dict[str, Any]) -> None:
        if Draft202012Validator is not None:
            errors = sorted(
                Draft202012Validator(
                    self.mq_schema,
                    format_checker=FormatChecker() if FormatChecker else None,
                ).iter_errors(body),
                key=lambda error: list(error.path),
            )
            if errors:
                raise ConfigurationError(
                    "service generated an event outside mq.schema.json"
                )

    def publish_execute(self, body: dict[str, Any]) -> dict[str, Any]:
        """SaaS-side test publisher. The production ingress is the broker queue."""
        self._validate_message(body)
        tenant = self._tenant(body["tenant_id"], body["tenant_key"])
        body_hash = digest(body)
        existing_before_admission = self.store.one(
            "SELECT * FROM published_commands WHERE command_id=? AND tenant_id=?",
            (body["command_id"], tenant["tenant_id"]),
        )
        if existing_before_admission:
            if existing_before_admission["body_hash"] != body_hash:
                raise ConflictError(
                    "IDEMPOTENCY_CONFLICT",
                    "command_id was already published with a different body",
                )
            if existing_before_admission["status"] not in {"pending", "retained"}:
                return {
                    "command_id": body["command_id"],
                    "status": existing_before_admission["status"],
                    "routing_key": routing_key(tenant["tenant_key"]),
                }
        self._validate_command_times(body)
        if existing_before_admission:
            task = self.store.one(
                "SELECT state FROM tasks WHERE task_id=? AND tenant_id=?",
                (body["payload"]["task_id"], tenant["tenant_id"]),
            )
            if task is None:
                raise ValidationError(
                    "TASK_TENANT_MISMATCH", "task is not owned by the command tenant"
                )
            if task["state"] != "running":
                raise ConflictError(
                    "TASK_PUBLISH_BLOCKED",
                    "SaaS persistent task block prevents new publication",
                )
        rate_limited = False
        with self.store.tx() as con:
            existing = con.execute(
                "SELECT * FROM published_commands WHERE command_id = ? AND tenant_id = ?",
                (body["command_id"], tenant["tenant_id"]),
            ).fetchone()
            if existing:
                if existing["body_hash"] != body_hash:
                    raise ConflictError(
                        "IDEMPOTENCY_CONFLICT",
                        "command_id was already published with a different body",
                    )
                if existing["status"] not in {"pending", "retained"}:
                    return {
                        "command_id": body["command_id"],
                        "status": existing["status"],
                        "routing_key": routing_key(tenant["tenant_key"]),
                    }
                if (
                    existing["last_error_code"] == "PUBLISH_RATE_LIMIT"
                    and existing["next_attempt_at"]
                    and parse_time(existing["next_attempt_at"]) > self._now()
                ):
                    raise ServiceError(
                        "PUBLISH_RATE_LIMIT",
                        "tenant publication rate limit reached",
                        429,
                        True,
                    )
            else:
                task = con.execute(
                    "SELECT * FROM tasks WHERE task_id=? AND tenant_id=?",
                    (body["payload"]["task_id"], tenant["tenant_id"]),
                ).fetchone()
                if task is None:
                    raise ValidationError(
                        "TASK_TENANT_MISMATCH",
                        "task is not owned by the command tenant",
                    )
                if task["state"] != "running":
                    raise ConflictError(
                        "TASK_PUBLISH_BLOCKED",
                        "SaaS persistent task block prevents new publication",
                    )
                created_at = iso(self._now())
                con.execute(
                    "INSERT INTO published_commands(command_id, tenant_id, body_json, body_hash, status, created_at) VALUES(?,?,?,?,?,?)",
                    (
                        body["command_id"],
                        tenant["tenant_id"],
                        canonical(body),
                        body_hash,
                        "pending",
                        created_at,
                    ),
                )
                publish_rate = _integer(self._limits().get("tenant_publish_rate", 0))
                if publish_rate:
                    recent = con.execute(
                        "SELECT COUNT(*) AS n FROM published_commands WHERE tenant_id=? AND created_at>=?",
                        (tenant["tenant_id"], iso(self._now() - timedelta(seconds=1))),
                    ).fetchone()["n"]
                    if _integer(recent) > publish_rate:
                        con.execute(
                            "UPDATE published_commands SET status='retained', last_error_code='PUBLISH_RATE_LIMIT', next_attempt_at=? WHERE tenant_id=? AND command_id=?",
                            (
                                iso(self._now() + timedelta(seconds=1)),
                                tenant["tenant_id"],
                                body["command_id"],
                            ),
                        )
                        rate_limited = True
        if rate_limited:
            raise ServiceError(
                "PUBLISH_RATE_LIMIT", "tenant publication rate limit reached", 429, True
            )
        try:
            route = routing_key(tenant["tenant_key"])
            self.broker.publish(
                "agent-call.commands.v1",
                route,
                body,
                body["command_id"],
                {"tenant_id": tenant["tenant_id"], "tenant_key": tenant["tenant_key"]},
            )
        except ValidationError as exc:
            with self.store.tx() as con:
                con.execute(
                    "UPDATE published_commands SET status='retained', last_error_code=? WHERE tenant_id=? AND command_id=?",
                    (exc.code, tenant["tenant_id"], body["command_id"]),
                )
            raise
        except Exception as exc:
            with self.store.tx() as con:
                attempts = (
                    con.execute(
                        "SELECT retry_attempts FROM published_commands WHERE tenant_id=? AND command_id=?",
                        (tenant["tenant_id"], body["command_id"]),
                    ).fetchone()["retry_attempts"]
                    + 1
                )
                con.execute(
                    "UPDATE published_commands SET status='retained', retry_attempts=?, next_attempt_at=?, last_error_code=? WHERE tenant_id=? AND command_id=?",
                    (
                        attempts,
                        iso(
                            self._now()
                            + timedelta(seconds=min(60, 2 ** min(attempts, 5)))
                        ),
                        type(exc).__name__,
                        tenant["tenant_id"],
                        body["command_id"],
                    ),
                )
            raise
        with self.store.tx() as con:
            con.execute(
                "UPDATE published_commands SET status='broker_confirmed', next_attempt_at=NULL, last_error_code=NULL WHERE tenant_id=? AND command_id=?",
                (tenant["tenant_id"], body["command_id"]),
            )
        return {
            "command_id": body["command_id"],
            "status": "broker_confirmed",
            "routing_key": route,
        }

    def _tenant_window_full(self, tenant_id: str, tenant_key: str) -> bool:
        limits = self._limits()
        tenant_limit = _integer(limits.get("pending_window_per_tenant", 16))
        global_limit = _integer(limits.get("pending_window_global", 256))
        tenant_row = self.store.one(
            "SELECT COUNT(*) AS n FROM commands WHERE tenant_id=? AND status NOT IN ('completed','failed','rejected','applied')",
            (tenant_id,),
        )
        global_row = self.store.one(
            "SELECT COUNT(*) AS n FROM commands WHERE status NOT IN ('completed','failed','rejected','applied')"
        )
        queue_depth = getattr(self.broker, "depth", lambda _name: 0)(
            queue_name(tenant_key)
        )
        return (
            _integer(tenant_row["n"]) >= tenant_limit
            or _integer(global_row["n"]) >= global_limit
            or queue_depth >= _integer(limits.get("max_unacked_per_tenant", 32))
        )

    def _validate_delivery(
        self,
        body: dict[str, Any],
        tenant: sqlite3.Row,
        message: dict[str, Any],
    ) -> None:
        expected_route = routing_key(tenant["tenant_key"])
        expected_queue = queue_name(tenant["tenant_key"])
        if message.get("queue_name") != expected_queue:
            raise ValidationError(
                "TENANT_ROUTE_MISMATCH", "message arrived on the wrong queue"
            )
        if message.get("routing_key") != expected_route:
            raise ValidationError(
                "TENANT_ROUTE_MISMATCH",
                "message routing key is not bound to the tenant",
            )
        headers = message.get("headers") or {}
        if not isinstance(headers, dict):
            raise ValidationError(
                "TENANT_ROUTE_MISMATCH", "message headers are invalid"
            )
        if "tenant_id" in headers and headers["tenant_id"] != tenant["tenant_id"]:
            raise ValidationError(
                "TENANT_ROUTE_MISMATCH", "tenant_id header does not match the queue"
            )
        if "tenant_key" in headers and headers["tenant_key"] != tenant["tenant_key"]:
            raise ValidationError(
                "TENANT_ROUTE_MISMATCH", "tenant_key header does not match the queue"
            )
        if message.get("message_id") not in {None, body.get("command_id")}:
            raise ValidationError(
                "MESSAGE_ID_MISMATCH", "message_id does not match command_id"
            )
        if (
            body.get("tenant_id") != tenant["tenant_id"]
            or body.get("tenant_key") != tenant["tenant_key"]
        ):
            raise ValidationError(
                "TENANT_BINDING_INVALID", "message tenant does not match the queue"
            )

    def _dead_letter(self, message: dict[str, Any], body: Any, reason: str) -> None:
        body_map = body if isinstance(body, dict) else {}
        try:
            encoded = canonical(body)
        except (TypeError, ValueError):
            encoded = "{}"
        with self.store.tx() as con:
            con.execute(
                "INSERT INTO dead_letters(tenant_key, routing_key, body_json, reason_code, created_at) VALUES(?,?,?,?,?)",
                (
                    body_map.get("tenant_key"),
                    message.get("routing_key"),
                    encoded,
                    reason,
                    iso(self._now()),
                ),
            )

    def ingest_commands(self, max_per_tick: int | None = None) -> int:
        max_per_tick = (
            max_per_tick
            if max_per_tick is not None
            else max(1, len(self._tenant_order))
        )
        if not self._tenant_order:
            return 0
        accepted = 0
        for offset in range(min(max_per_tick, len(self._tenant_order))):
            tenant_id = self._tenant_order[
                (self._cursor + offset) % len(self._tenant_order)
            ]
            tenant = self.store.one(
                "SELECT * FROM tenants WHERE tenant_id = ?", (tenant_id,)
            )
            if tenant is None or tenant_id in self._unrouteable_tenants:
                continue
            try:
                if self._tenant_window_full(tenant_id, tenant["tenant_key"]):
                    continue
                message = self.broker.consume(queue_name(tenant["tenant_key"]))
            except (BrokerError, ValidationError):
                break
            if not message:
                continue
            body: Any = message.get("body", {})
            try:
                body_map = body if isinstance(body, dict) else {}
                self._validate_delivery(body_map, tenant, message)
                self.accept_execute(body_map, delivery_context=message)
                accepted += 1
                with self.store.tx() as con:
                    con.execute(
                        "UPDATE published_commands SET status='accepted' WHERE tenant_id=? AND command_id=?",
                        (tenant_id, body_map.get("command_id")),
                    )
                self.broker.ack(message)
            except ServiceError as exc:
                self._dead_letter(message, body, exc.code)
                body_map = body if isinstance(body, dict) else {}
                if body_map.get("command_id"):
                    with self.store.tx() as con:
                        con.execute(
                            "UPDATE published_commands SET status='rejected', last_error_code=? WHERE tenant_id=? AND command_id=?",
                            (exc.code, tenant_id, body_map["command_id"]),
                        )
                reject = getattr(self.broker, "reject", None)
                if callable(reject):
                    reject(message)
                else:
                    self.broker.ack(message)
            except (
                OSError,
                RuntimeError,
                TypeError,
                ValueError,
                KeyError,
                sqlite3.DatabaseError,
            ):
                # Unknown persistence/broker errors must remain deliverable; an
                # invalid message is handled above and sent to the dead-letter log.
                with suppress(BrokerError):
                    self.broker.requeue(message)
                break
        self._cursor = (self._cursor + max_per_tick) % len(self._tenant_order)
        return accepted

    def accept_execute(
        self,
        body: dict[str, Any],
        delivery_context: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        self._validate_message(body)
        self._validate_command_times(body)
        tenant = self._tenant(body["tenant_id"], body["tenant_key"])
        if delivery_context is not None:
            self._validate_delivery(body, tenant, delivery_context)
        payload = body["payload"]
        task = self.store.one(
            "SELECT * FROM tasks WHERE task_id = ? AND tenant_id = ?",
            (payload["task_id"], tenant["tenant_id"]),
        )
        if task is None or task["tenant_id"] != tenant["tenant_id"]:
            raise ValidationError(
                "TASK_TENANT_MISMATCH", "task is not owned by the command tenant"
            )
        command_hash = digest(body)
        now = self._now()
        not_after = parse_time(body["not_after"])
        parse_time(body["issued_at"])
        deadline = min(
            not_after,
            now + timedelta(seconds=self._limits().get("admission_window_s", 30)),
        )
        with self.store.tx() as con:
            existing = con.execute(
                "SELECT * FROM commands WHERE command_id = ? AND tenant_id = ?",
                (body["command_id"], tenant["tenant_id"]),
            ).fetchone()
            if existing:
                if existing["payload_hash"] != command_hash:
                    raise ConflictError(
                        "IDEMPOTENCY_CONFLICT",
                        "command_id was already accepted with a different body",
                    )
                return self._command_snapshot(existing)
            reason: str | None = None
            status = "accepted"
            if payload["task_revision"] != task["revision"]:
                status, reason = "rejected", "STALE_TASK_REVISION"
            elif task["state"] != "running":
                status, reason = "rejected", f"TASK_{task['state'].upper()}"
            elif not_after <= now:
                status, reason = "rejected", "COMMAND_EXPIRED"
            elif payload["route_policy_id"] not in {
                x["route_policy_id"]
                for x in self.profile.get("routes", [])
                if x.get("allowed", True)
            }:
                status, reason = "rejected", "ROUTE_POLICY_NOT_ALLOWED"
            elif payload["caller_profile_id"] not in {
                x["caller_profile_id"]
                for x in self.profile.get("caller_profiles", [])
                if x.get("allowed", True)
            }:
                status, reason = "rejected", "CALLER_PROFILE_NOT_ALLOWED"
            elif (
                con.execute(
                    "SELECT 1 FROM agent_versions WHERE tenant_id=? AND agent_version_id=?",
                    (tenant["tenant_id"], payload["agent_version_id"]),
                ).fetchone()
                is None
            ):
                status, reason = "rejected", "AGENT_VERSION_NOT_ALLOWED"
            execution = con.execute(
                "SELECT * FROM executions WHERE execution_id = ? AND tenant_id = ?",
                (payload["execution_id"], tenant["tenant_id"]),
            ).fetchone()
            if execution and execution["payload_hash"] != digest(payload):
                status, reason = "rejected", "EXECUTION_ID_CONFLICT"
            elif execution:
                status, reason = "completed", "DUPLICATE_EXECUTION"
            con.execute(
                """INSERT INTO commands(command_id, tenant_id, tenant_key, command_type, status,
                   payload_json, payload_hash, task_id, execution_id, call_id, trace_id, issued_at,
                   not_after, admission_deadline, accepted_at, reason_code, task_state, updated_at)
                   VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
                (
                    body["command_id"],
                    tenant["tenant_id"],
                    tenant["tenant_key"],
                    body["command_type"],
                    status,
                    canonical(body),
                    command_hash,
                    payload["task_id"],
                    payload["execution_id"],
                    execution["call_id"] if execution else None,
                    body["trace_id"],
                    body["issued_at"],
                    body["not_after"],
                    iso(deadline),
                    iso(now),
                    reason,
                    task["state"],
                    iso(now),
                ),
            )
            if status == "accepted":
                con.execute(
                    "INSERT INTO executions(execution_id, tenant_id, command_id, task_id, task_item_id, payload_hash, status, agent_version_id, created_at) VALUES(?,?,?,?,?,?,?,?,?)",
                    (
                        payload["execution_id"],
                        tenant["tenant_id"],
                        body["command_id"],
                        payload["task_id"],
                        payload["task_item_id"],
                        digest(payload),
                        "accepted",
                        payload["agent_version_id"],
                        iso(now),
                    ),
                )
            row = con.execute(
                "SELECT * FROM commands WHERE tenant_id = ? AND command_id = ?",
                (tenant["tenant_id"], body["command_id"]),
            ).fetchone()
            self._command_event_tx(con, row, reason or "ACCEPTED")
            return self._command_snapshot(row)

    def _command_event_tx(
        self, con: sqlite3.Connection, row: sqlite3.Row, reason: str | None = None
    ) -> None:
        payload = {
            "command_id": row["command_id"],
            "command_type": row["command_type"],
            "status": row["status"],
            "reason_code": reason or row["reason_code"],
            "task_id": row["task_id"],
            "execution_id": row["execution_id"],
            "call_id": row["call_id"],
            "agent_version_id": _loads(row["payload_json"], {})
            .get("payload", {})
            .get("agent_version_id"),
            "requested_task_revision": row["requested_task_revision"],
            "applied_task_revision": row["applied_task_revision"],
            "wait_reason_code": row["wait_reason_code"],
        }
        self._event_tx(
            con,
            row["tenant_id"],
            row["tenant_key"],
            row["trace_id"] or row["command_id"],
            "command.result",
            "command",
            row["command_id"],
            row["aggregate_version"],
            payload,
            row["accepted_at"] or iso(self._now()),
        )

    def _event_tx(
        self,
        con: sqlite3.Connection,
        tenant_id: str,
        tenant_key: str,
        trace_id: str,
        event_type: str,
        aggregate_type: str,
        aggregate_id: str,
        version: int,
        payload: dict[str, Any],
        occurred_at: str | None = None,
        replay_of: str | None = None,
    ) -> dict[str, Any]:
        body = {
            "schema_version": "1.0",
            "event_id": new_id("evt"),
            "event_type": event_type,
            "tenant_id": tenant_id,
            "tenant_key": tenant_key,
            "trace_id": trace_id,
            "occurred_at": occurred_at or iso(self._now()),
            "aggregate_type": aggregate_type,
            "aggregate_id": aggregate_id,
            "aggregate_version": version,
            "payload": payload,
        }
        self._validate_event(body)
        con.execute(
            """INSERT INTO events(event_id, tenant_id, tenant_key, trace_id, event_type,
               aggregate_type, aggregate_id, aggregate_version, occurred_at, payload_json, body_json, replay_of)
               VALUES(?,?,?,?,?,?,?,?,?,?,?,?)""",
            (
                body["event_id"],
                tenant_id,
                tenant_key,
                trace_id,
                event_type,
                aggregate_type,
                aggregate_id,
                version,
                body["occurred_at"],
                canonical(payload),
                canonical(body),
                replay_of,
            ),
        )
        return body

    def _command_snapshot(self, row: sqlite3.Row | dict[str, Any]) -> dict[str, Any]:
        payload = _loads(row["payload_json"], {}).get("payload", {})
        return {
            "command_id": row["command_id"],
            "command_type": row["command_type"],
            "tenant_id": row["tenant_id"],
            "tenant_key": row["tenant_key"],
            "task_id": row["task_id"],
            "execution_id": row["execution_id"],
            "call_id": row["call_id"],
            "agent_version_id": payload.get("agent_version_id"),
            "status": row["status"],
            "reason_code": row["reason_code"],
            "wait_reason_code": row["wait_reason_code"],
            "accepted_at": row["accepted_at"],
            "waiting_since": row["waiting_since"],
            "admission_deadline": row["admission_deadline"],
            "requested_task_revision": row["requested_task_revision"],
            "applied_task_revision": row["applied_task_revision"],
            "task_state": row["task_state"],
            "updated_at": row["updated_at"],
            "aggregate_version": row["aggregate_version"],
        }

    def _lease_ttl(self) -> int:
        return max(1, _integer(self._limits().get("scheduler_lease_ttl_s", 10)))

    def retry_published_commands(self, limit: int = 20) -> int:
        now = self._now()
        rows = self.store.all(
            "SELECT * FROM published_commands WHERE status IN ('pending','retained') AND (next_attempt_at IS NULL OR next_attempt_at<=?) ORDER BY created_at LIMIT ?",
            (iso(now), limit),
        )
        retried = 0
        for row in rows:
            if row["last_error_code"] == "TENANT_KEY_UNSUPPORTED":
                continue
            try:
                self.publish_execute(_loads(row["body_json"]))
                retried += 1
            except ServiceError:
                continue
        return retried

    def recover_uploads(self, limit: int = 20) -> int:
        rows = self.store.all(
            "SELECT * FROM uploads WHERE status IN ('created','uploaded') ORDER BY created_at LIMIT ?",
            (limit,),
        )
        recovered = 0
        for upload in rows:
            recording = self.store.one(
                "SELECT * FROM recordings WHERE recording_id=? AND tenant_id=?",
                (upload["recording_id"], upload["tenant_id"]),
            )
            if recording is None:
                continue
            try:
                if self.object_store.has_final(upload["upload_id"]):
                    self._complete_upload(
                        upload["tenant_id"],
                        upload["upload_id"],
                        upload["recording_id"],
                        checksum=upload["expected_checksum"],
                        size_bytes=upload["expected_size"],
                        recovery=True,
                    )
                    recovered += 1
            except ServiceError:
                continue
            if parse_time(upload["expires_at"]) <= self._now():
                with self.store.tx() as con:
                    con.execute(
                        "UPDATE uploads SET status='expired' WHERE tenant_id=? AND upload_id=? AND status IN ('created','uploaded')",
                        (upload["tenant_id"], upload["upload_id"]),
                    )
        return recovered

    def _claim_lease(self) -> bool:
        if self._clock_jump:
            return False
        now = self._now()
        expires = now + timedelta(seconds=self._lease_ttl())
        with self.store.tx() as con:
            row = con.execute(
                "SELECT * FROM leases WHERE lease_name='scheduler'"
            ).fetchone()
            if (
                row
                and row["owner_id"] != self.instance_id
                and parse_time(row["expires_at"]) > now
            ):
                return False
            con.execute(
                """INSERT INTO leases(lease_name, owner_id, expires_at, generation)
                   VALUES('scheduler', ?, ?, 1)
                   ON CONFLICT(lease_name) DO UPDATE SET
                     owner_id=excluded.owner_id, expires_at=excluded.expires_at,
                     generation=leases.generation+1""",
                (self.instance_id, iso(expires)),
            )
        return True

    def _lease_owned_tx(
        self, con: sqlite3.Connection, now: datetime | None = None
    ) -> bool:
        now = now or self._now()
        row = con.execute(
            "SELECT * FROM leases WHERE lease_name='scheduler'"
        ).fetchone()
        return bool(
            row
            and row["owner_id"] == self.instance_id
            and parse_time(row["expires_at"]) > now
        )

    def process_once(self) -> int:
        if not self._claim_lease():
            return 0
        self.retry_published_commands()
        self.ingest_commands()
        self.apply_controls()
        scheduled = self.schedule_once()
        self.dispatch_outbox()
        self.consume_saas_events()
        self.process_replays()
        self.recover_uploads()
        return scheduled

    def _active_counts(self, tenant_id: str | None = None) -> tuple[int, int]:
        if tenant_id:
            row = self.store.one(
                "SELECT COUNT(*) AS n FROM calls WHERE call_state NOT IN ('ended', 'failed') AND tenant_id = ?",
                (tenant_id,),
            )
        else:
            row = self.store.one(
                "SELECT COUNT(*) AS n FROM calls WHERE call_state NOT IN ('ended', 'failed')"
            )
        return _integer(row["n"]), 0

    def _resource_cell(
        self, route: dict[str, Any] | None = None
    ) -> tuple[str, dict[str, Any]] | None:
        egress_pool_id = route.get("egress_pool_id") if route else None
        for cell in self.profile.get("cells", []):
            if egress_pool_id and cell.get("egress_pool_id") != egress_pool_id:
                continue
            if not self._cell_health.get(cell["cell_id"], False):
                continue
            row = self.store.one(
                "SELECT COUNT(*) AS n FROM calls WHERE cell_id = ? AND call_state NOT IN ('ended','failed')",
                (cell["cell_id"],),
            )
            capacity = min(
                _integer(cell.get("capacity", 1)),
                _integer(cell.get("media_capacity", cell.get("capacity", 1))),
                _integer(cell.get("ai_capacity", cell.get("capacity", 1))),
            )
            if _integer(row["n"]) < capacity:
                return cell["cell_id"], cell
        return None

    def set_cell_health(self, cell_id: str, healthy: bool) -> None:
        if cell_id not in self._cell_health:
            raise ValidationError("CELL_NOT_FOUND", "cell is not configured")
        self._cell_health[cell_id] = healthy

    def _cps_available(self, tenant_id: str) -> bool:
        now = self._now()
        since = iso(now - timedelta(seconds=1))
        global_row = self.store.one(
            "SELECT COUNT(*) AS n FROM attempts WHERE created_at >= ?", (since,)
        )
        tenant_row = self.store.one(
            "SELECT COUNT(*) AS n FROM attempts a JOIN calls c ON c.call_id = a.call_id WHERE c.tenant_id = ? AND a.created_at >= ?",
            (tenant_id, since),
        )
        return _integer(global_row["n"]) < _integer(
            self._limits().get("global_cps", 3)
        ) and _integer(tenant_row["n"]) < _integer(self._limits().get("tenant_cps", 1))

    def _mark_waiting(self, tenant_id: str, command_id: str, reason: str) -> None:
        with self.store.tx() as con:
            row = con.execute(
                "SELECT * FROM commands WHERE tenant_id=? AND command_id=?",
                (tenant_id, command_id),
            ).fetchone()
            if row is None or row["status"] not in {"accepted", "waiting"}:
                return
            if row["status"] == "waiting" and row["wait_reason_code"] == reason:
                return
            version = row["aggregate_version"] + 1
            now = iso(self._now())
            con.execute(
                "UPDATE commands SET status='waiting', waiting_since=COALESCE(waiting_since, ?), wait_reason_code=?, aggregate_version=?, updated_at=? WHERE tenant_id=? AND command_id=?",
                (now, reason, version, now, tenant_id, command_id),
            )
            updated = con.execute(
                "SELECT * FROM commands WHERE tenant_id=? AND command_id=?",
                (tenant_id, command_id),
            ).fetchone()
            self._command_event_tx(con, updated, reason)

    def _reject_command(self, tenant_id: str, command_id: str, reason: str) -> None:
        with self.store.tx() as con:
            row = con.execute(
                "SELECT * FROM commands WHERE tenant_id=? AND command_id=?",
                (tenant_id, command_id),
            ).fetchone()
            if row is None or row["status"] in {
                "completed",
                "failed",
                "rejected",
                "applied",
            }:
                return
            now = iso(self._now())
            con.execute(
                "UPDATE commands SET status='rejected', reason_code=?, aggregate_version=aggregate_version+1, updated_at=? WHERE tenant_id=? AND command_id=?",
                (reason, now, tenant_id, command_id),
            )
            updated = con.execute(
                "SELECT * FROM commands WHERE tenant_id=? AND command_id=?",
                (tenant_id, command_id),
            ).fetchone()
            self._command_event_tx(con, updated, reason)

    def schedule_once(self) -> int:
        if not self._claim_lease():
            return 0
        if not self.readiness()["ready"]:
            return 0
        scheduled = 0
        if not self._tenant_order:
            return scheduled
        order = [
            self._tenant_order[(self._cursor + index) % len(self._tenant_order)]
            for index in range(len(self._tenant_order))
        ]
        for tenant_id in order:
            row = self.store.one(
                "SELECT * FROM commands WHERE tenant_id = ? AND status IN ('accepted','waiting') ORDER BY accepted_at, command_id LIMIT 1",
                (tenant_id,),
            )
            if row is None:
                continue
            payload = _loads(row["payload_json"])["payload"]
            try:
                route = self._route_config(payload["route_policy_id"])
            except SipReadonlyError as exc:
                self._mark_waiting(
                    tenant_id,
                    row["command_id"],
                    "SIP_DIRECTORY" if exc.retryable else exc.code,
                )
                continue
            except ValidationError as exc:
                self._reject_command(tenant_id, row["command_id"], exc.code)
                continue
            now = self._now()
            if (
                row["admission_deadline"]
                and parse_time(row["admission_deadline"]) <= now
            ):
                self._reject_command(tenant_id, row["command_id"], "ADMISSION_TIMEOUT")
                continue
            task = self.store.one(
                "SELECT * FROM tasks WHERE task_id = ? AND tenant_id = ?",
                (row["task_id"], tenant_id),
            )
            if task is None or task["state"] != "running":
                self._reject_command(
                    tenant_id,
                    row["command_id"],
                    f"TASK_{task['state'].upper() if task else 'MISSING'}",
                )
                continue
            active_tenant = self._active_counts(tenant_id)[0]
            active_global = self._active_counts()[0]
            tenant_limit = _integer(
                self.store.one(
                    "SELECT max_concurrency FROM tenants WHERE tenant_id = ?",
                    (tenant_id,),
                )["max_concurrency"]
            )
            if active_tenant >= tenant_limit or active_global >= _integer(
                self._limits().get("global_concurrency", 6)
            ):
                self._mark_waiting(tenant_id, row["command_id"], "CONCURRENCY")
                continue
            if not self._cps_available(tenant_id):
                self._mark_waiting(tenant_id, row["command_id"], "CPS")
                continue
            cell = self._resource_cell(route)
            if cell is None:
                self._mark_waiting(tenant_id, row["command_id"], "CELL_CAPACITY")
                continue
            try:
                reserved = self._reserve_call(row["command_id"], row, cell, route)
            except ValidationError as exc:
                self._reject_command(tenant_id, row["command_id"], exc.code)
                continue
            if not reserved:
                self._mark_waiting(tenant_id, row["command_id"], "RESOURCE_RACE")
                continue
            scheduled += 1
        self._cursor = (self._cursor + 1) % len(self._tenant_order)
        return scheduled

    def _reserve_call(
        self,
        command_id: str,
        row: sqlite3.Row,
        cell: tuple[str, dict[str, Any]],
        route: dict[str, Any] | None = None,
    ) -> bool:
        cell_id, cell_cfg = cell
        payload = _loads(row["payload_json"])["payload"]
        route = route or self._route_config(payload["route_policy_id"])
        prefix = str(route.get("dial_prefix", ""))
        target_uri = (
            payload["callee"]
            if payload["callee"].startswith(prefix)
            else prefix + payload["callee"]
        )
        caller = next(
            (
                item.get("display", "")
                for item in self.profile.get("caller_profiles", [])
                if item.get("caller_profile_id") == payload["caller_profile_id"]
            ),
            payload["caller_profile_id"],
        )
        if route.get("caller_ids") and caller not in route["caller_ids"]:
            raise ValidationError(
                "CALLER_NOT_ALLOWED", "caller is not allowed by the published Trunk"
            )
        try:
            agent_config = self._agent_config_for(
                row["tenant_id"], payload["agent_version_id"]
            )
        except NotFoundError as exc:
            raise ValidationError(
                "AGENT_VERSION_NOT_ALLOWED",
                "agent version is not available for this tenant",
            ) from exc
        agent_config_sha256 = ai_config_digest(agent_config)
        call_id = new_id("call")
        attempt_id = new_id("attempt")
        now_dt = self._now()
        now = iso(now_dt)
        with self.store.tx() as con:
            if not self._lease_owned_tx(con, now_dt):
                return False
            current = con.execute(
                "SELECT * FROM commands WHERE tenant_id=? AND command_id=?",
                (row["tenant_id"], command_id),
            ).fetchone()
            task = con.execute(
                "SELECT * FROM tasks WHERE tenant_id=? AND task_id=?",
                (row["tenant_id"], row["task_id"]),
            ).fetchone()
            if (
                current is None
                or current["status"] not in {"accepted", "waiting"}
                or task is None
                or task["state"] != "running"
            ):
                return False
            tenant_active = con.execute(
                "SELECT COUNT(*) AS n FROM calls WHERE tenant_id=? AND call_state NOT IN ('ended','failed')",
                (row["tenant_id"],),
            ).fetchone()["n"]
            global_active = con.execute(
                "SELECT COUNT(*) AS n FROM calls WHERE call_state NOT IN ('ended','failed')"
            ).fetchone()["n"]
            tenant_limit = con.execute(
                "SELECT max_concurrency FROM tenants WHERE tenant_id=?",
                (row["tenant_id"],),
            ).fetchone()["max_concurrency"]
            if _integer(tenant_active) >= _integer(tenant_limit) or _integer(
                global_active
            ) >= _integer(self._limits().get("global_concurrency", 6)):
                return False
            cell_active = con.execute(
                "SELECT COUNT(*) AS n FROM calls WHERE cell_id=? AND call_state NOT IN ('ended','failed')",
                (cell_id,),
            ).fetchone()["n"]
            cell_capacity = min(
                _integer(cell_cfg.get("capacity", 1)),
                _integer(cell_cfg.get("media_capacity", cell_cfg.get("capacity", 1))),
                _integer(cell_cfg.get("ai_capacity", cell_cfg.get("capacity", 1))),
            )
            if _integer(cell_active) >= cell_capacity:
                return False
            route_active = con.execute(
                "SELECT COUNT(*) AS n FROM calls WHERE trunk_id=? AND call_state NOT IN ('ended','failed')",
                (route.get("trunk_id", "trunk-mock"),),
            ).fetchone()["n"]
            if _integer(route_active) >= _integer(route.get("max_concurrency", 10**9)):
                return False
            since = iso(now_dt - timedelta(seconds=1))
            global_cps = con.execute(
                "SELECT COUNT(*) AS n FROM attempts WHERE created_at>=?", (since,)
            ).fetchone()["n"]
            tenant_cps = con.execute(
                "SELECT COUNT(*) AS n FROM attempts a JOIN calls c ON c.call_id=a.call_id WHERE c.tenant_id=? AND a.created_at>=?",
                (row["tenant_id"], since),
            ).fetchone()["n"]
            route_cps = con.execute(
                "SELECT COUNT(*) AS n FROM attempts WHERE route_policy_id=? AND created_at>=?",
                (payload["route_policy_id"], since),
            ).fetchone()["n"]
            if (
                _integer(global_cps) >= _integer(self._limits().get("global_cps", 3))
                or _integer(tenant_cps) >= _integer(self._limits().get("tenant_cps", 1))
                or _integer(route_cps) >= _integer(route.get("max_cps", 10**9))
            ):
                return False
            version = current["aggregate_version"] + 1
            changed = con.execute(
                "UPDATE commands SET status='executing', call_id=?, waiting_since=NULL, wait_reason_code=NULL, aggregate_version=?, updated_at=? WHERE tenant_id=? AND command_id=? AND status IN ('accepted','waiting')",
                (call_id, version, now, row["tenant_id"], command_id),
            ).rowcount
            if changed != 1:
                return False
            con.execute(
                "INSERT INTO calls(call_id, tenant_id, execution_id, command_id, task_id, task_item_id, callee, cell_id, trunk_id, egress_pool_id, agent_version_id, agent_config_sha256, call_state, created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
                (
                    call_id,
                    current["tenant_id"],
                    current["execution_id"],
                    command_id,
                    current["task_id"],
                    payload["task_item_id"],
                    payload["callee"],
                    cell_id,
                    route.get("trunk_id", "trunk-mock"),
                    route.get(
                        "egress_pool_id", cell_cfg.get("egress_pool_id", "egress-mock")
                    ),
                    payload["agent_version_id"],
                    agent_config_sha256,
                    "queued",
                    now,
                ),
            )
            con.execute(
                "INSERT INTO attempts(attempt_id, call_id, attempt_number, route_policy_id, caller_profile_id, target_uri, status, created_at) VALUES(?,?,?,?,?,?,?,?)",
                (
                    attempt_id,
                    call_id,
                    1,
                    payload["route_policy_id"],
                    payload["caller_profile_id"],
                    target_uri,
                    "reserved",
                    now,
                ),
            )
            con.execute(
                "UPDATE executions SET status='executing', call_id=? WHERE tenant_id=? AND execution_id=?",
                (call_id, current["tenant_id"], current["execution_id"]),
            )
            updated = con.execute(
                "SELECT * FROM commands WHERE tenant_id=? AND command_id=?",
                (current["tenant_id"], command_id),
            ).fetchone()
            self._command_event_tx(con, updated, "EXECUTING")
            self._event_tx(
                con,
                current["tenant_id"],
                current["tenant_key"],
                current["trace_id"],
                "call.status",
                "call",
                call_id,
                1,
                {
                    "call_id": call_id,
                    "execution_id": current["execution_id"],
                    "task_id": current["task_id"],
                    "agent_version_id": payload["agent_version_id"],
                    "agent_config_sha256": agent_config_sha256,
                    "call_state": "queued",
                    "reason_code": None,
                },
                now,
            )
        try:
            if not self._call_permitted(call_id, cell_id):
                self._finish_call(
                    call_id, "failed", "ADMISSION_REVOKED", notify_protocol=False
                )
                return True
            self.mock_sip.invite(call_id, payload["callee"], target_uri, caller)
            self.mock_ari.channel_created(call_id, cell_id)
            self.mock_ari.bridge_ready(call_id)
            thread = threading.Thread(
                target=self._run_call,
                args=(call_id,),
                name=f"call-{call_id}",
                daemon=True,
            )
            self._call_threads[call_id] = thread
            thread.start()
        except (
            OSError,
            RuntimeError,
            TypeError,
            ValueError,
            KeyError,
            sqlite3.DatabaseError,
            ServiceError,
        ):
            self._finish_call(
                call_id, "failed", "PROTOCOL_SETUP_FAILED", notify_protocol=False
            )
        return True

    def _call_permitted(self, call_id: str, cell_id: str) -> bool:
        with self.store.tx() as con:
            if not self._lease_owned_tx(con):
                return False
            row = con.execute(
                """SELECT c.call_state, c.tenant_id, cmd.status, t.state
                   FROM calls c JOIN commands cmd ON cmd.tenant_id=c.tenant_id AND cmd.command_id=c.command_id
                   JOIN tasks t ON t.tenant_id=c.tenant_id AND t.task_id=c.task_id
                   WHERE c.call_id=? AND c.cell_id=?""",
                (call_id, cell_id),
            ).fetchone()
            return bool(
                row
                and row["call_state"] == "queued"
                and row["status"] == "executing"
                and row["state"] == "running"
                and self._cell_health.get(cell_id, False)
            )

    def set_failure(self, command_or_call_id: str, scenario: str) -> None:
        allowed = set(self.profile.get("failure_scenarios", []))
        if scenario not in allowed:
            raise ValidationError(
                "SCENARIO_NOT_ALLOWED", "failure scenario is not in the mock profile"
            )
        self._failures[command_or_call_id] = scenario

    def _scenario_for(self, call_id: str) -> str:
        row = self.store.one(
            "SELECT command_id FROM calls WHERE call_id = ?", (call_id,)
        )
        return self._failures.get(
            call_id, self._failures.get(row["command_id"] if row else "", "success")
        )

    def _sleep(self, milliseconds: int = 0) -> None:
        if milliseconds > 0:
            time.sleep(milliseconds / 1000)

    def _call_is_ended(self, call_id: str) -> bool:
        row = self.store.one(
            "SELECT call_state FROM calls WHERE call_id = ?", (call_id,)
        )
        return row is None or row["call_state"] in {"ended", "failed"}

    def _run_call(self, call_id: str) -> None:
        scenario = self._scenario_for(call_id)
        try:
            self._update_call_state(call_id, "dialing")
            self._update_attempt(call_id, "dialing", started=True)
            self._sleep(1 if scenario == "clock_jump" else 0)
            if self._call_is_ended(call_id):
                return
            self._update_call_state(call_id, "ringing")
            self.mock_sip.ringing(call_id)
            self.mock_rtp.packet(call_id, "inbound", 1)
            if scenario in {"busy", "no_answer"}:
                self._finish_call(
                    call_id,
                    "busy" if scenario == "busy" else "no_answer",
                    scenario.upper(),
                )
                return
            self._sleep(1 if scenario == "ai_timeout" else 0)
            if self._call_is_ended(call_id):
                return
            if scenario == "ai_timeout":
                self._finish_call(call_id, "failed", "AI_TIMEOUT")
                return
            self._update_call_state(call_id, "answered", started=True)
            self.mock_sip.answered(call_id)
            self.mock_rtp.packet(call_id, "outbound", 1)
            self._ensure_recording(call_id)
            self._sleep(_integer(self._limits().get("hold_ms", 0)))
            if scenario == "customer_silent":
                self._finish_call(call_id, "failed", "CUSTOMER_SILENT")
                return
            if scenario == "ari_disconnect":
                self._mark_reconciling(call_id, "ARI_DISCONNECTED_RECONCILING")
                return
            call = self.store.one("SELECT * FROM calls WHERE call_id=?", (call_id,))
            if call is None:
                return
            agent_config = self._agent_config_for(
                call["tenant_id"], call["agent_version_id"] or "agent_v1"
            )
            conversation = self._conversation_engine(agent_config)
            ai_metrics: dict[str, Any] = {
                "mode": "mock",
                "turns": 0,
                "discarded_late_chunks": 0,
                "llm_first_token_ms": [],
                "tts_first_audio_ms": [],
            }
            opening = agent_config["conversation"].get("opening", "")
            if opening:
                opening_result = conversation.speak(opening, f"opening_{call_id}")
                self._add_transcript(
                    call_id,
                    -1,
                    "agent",
                    opening,
                    opening_result["playback_state"],
                )
            turns = _integer(self._limits().get("turns", 2))
            for turn in range(turns):
                if self._call_is_ended(call_id):
                    return
                customer_text = f"customer turn {turn + 1}"
                self._add_transcript(
                    call_id, turn, "customer", customer_text, "not_applicable"
                )
                result = conversation.run_text(customer_text)
                if result.get("status") != "completed":
                    self._finish_call(
                        call_id,
                        "failed",
                        str(result.get("reason_code", "AI_GENERATION_FAILED")),
                    )
                    return
                playback = (
                    result.get("segments", [{}])[-1].get("playback_state", "sent")
                    if result.get("segments")
                    else "sent"
                )
                self._add_transcript(
                    call_id, turn, "agent", str(result.get("text", "")), playback
                )
                ai_metrics["turns"] += 1
                ai_metrics["discarded_late_chunks"] += int(
                    result.get("discarded_late_chunks", 0)
                )
                for metric in ("llm_first_token_ms", "tts_first_audio_ms"):
                    value = result.get(metric)
                    if value is not None:
                        ai_metrics[metric].append(value)
                with self.store.tx() as con:
                    con.execute(
                        "UPDATE calls SET ai_metrics_json=? WHERE call_id=?",
                        (canonical(ai_metrics), call_id),
                    )
            self._finish_call(
                call_id, "completed", "CALL_COMPLETED", upload_scenario=scenario
            )
        except (
            OSError,
            RuntimeError,
            TypeError,
            ValueError,
            KeyError,
            sqlite3.DatabaseError,
            ServiceError,
        ):
            if not self._call_is_ended(call_id):
                self._finish_call(call_id, "failed", "EXECUTOR_EXCEPTION")

    def _mark_reconciling(self, call_id: str, reason: str) -> None:
        now = iso(self._now())
        with self.store.tx() as con:
            row = con.execute(
                "SELECT * FROM calls WHERE call_id=?", (call_id,)
            ).fetchone()
            if row is None or row["call_state"] in {"ended", "failed"}:
                return
            version = row["call_version"] + 1
            con.execute(
                "UPDATE calls SET call_state='reconciling', call_version=?, reason_code=? WHERE call_id=?",
                (version, reason, call_id),
            )
            con.execute(
                "UPDATE commands SET status='reconciling', reason_code=?, aggregate_version=aggregate_version+1, updated_at=? WHERE tenant_id=? AND command_id=? AND status NOT IN ('completed','failed','rejected','applied')",
                (reason, now, row["tenant_id"], row["command_id"]),
            )
            con.execute(
                "UPDATE executions SET status='reconciling' WHERE tenant_id=? AND execution_id=? AND status NOT IN ('completed','failed','rejected')",
                (row["tenant_id"], row["execution_id"]),
            )
            tenant = con.execute(
                "SELECT tenant_key FROM tenants WHERE tenant_id=?", (row["tenant_id"],)
            ).fetchone()["tenant_key"]
            self._event_tx(
                con,
                row["tenant_id"],
                tenant,
                row["command_id"],
                "call.status",
                "call",
                call_id,
                version,
                {
                    "call_id": call_id,
                    "execution_id": row["execution_id"],
                    "call_state": "reconciling",
                    "reason_code": reason,
                },
                now,
            )

    def reconcile_call(
        self,
        call_id: str,
        ended: bool,
        outcome: str = "completed",
        reason: str = "RECONCILED",
    ) -> dict[str, Any]:
        if not ended:
            row = self.store.one("SELECT * FROM calls WHERE call_id=?", (call_id,))
            if row is None:
                raise NotFoundError()
            return self.get_call(row["tenant_id"], call_id)
        self._finish_call(call_id, outcome, reason)
        row = self.store.one("SELECT * FROM calls WHERE call_id=?", (call_id,))
        if row is None:
            raise NotFoundError()
        return self.get_call(row["tenant_id"], call_id)

    def _update_attempt(self, call_id: str, status: str, started: bool = False) -> None:
        now = iso(self._now())
        with self.store.tx() as con:
            row = con.execute(
                "SELECT * FROM attempts WHERE call_id = ? ORDER BY attempt_number DESC LIMIT 1",
                (call_id,),
            ).fetchone()
            if row is None:
                return
            con.execute(
                "UPDATE attempts SET status=?, started_at=COALESCE(started_at, ?), ended_at=? WHERE attempt_id=?",
                (
                    status,
                    now if started else row["started_at"],
                    now if status in {"ended", "failed"} else None,
                    row["attempt_id"],
                ),
            )

    def _update_call_state(
        self,
        call_id: str,
        state: str,
        outcome: str | None = None,
        reason: str | None = None,
        started: bool = False,
    ) -> None:
        now = iso(self._now())
        with self.store.tx() as con:
            row = con.execute(
                "SELECT * FROM calls WHERE call_id = ?", (call_id,)
            ).fetchone()
            if row is None or row["call_state"] in {"ended", "failed"}:
                return
            version = row["call_version"] + 1
            started_at = row["started_at"] or (
                now if started or state == "answered" else None
            )
            con.execute(
                "UPDATE calls SET call_state=?, call_version=?, outcome=COALESCE(?, outcome), reason_code=COALESCE(?, reason_code), started_at=? WHERE call_id=?",
                (state, version, outcome, reason, started_at, call_id),
            )
            self._event_tx(
                con,
                row["tenant_id"],
                self.store.one(
                    "SELECT tenant_key FROM tenants WHERE tenant_id = ?",
                    (row["tenant_id"],),
                )["tenant_key"],
                row["command_id"],
                "call.status",
                "call",
                call_id,
                version,
                {
                    "call_id": call_id,
                    "execution_id": row["execution_id"],
                    "call_state": state,
                    "outcome": outcome,
                    "reason_code": reason,
                },
                now,
            )

    def _add_transcript(
        self,
        call_id: str,
        turn: int,
        speaker: str,
        text: str,
        playback: str,
        revision: int | None = None,
        is_final: bool = True,
    ) -> None:
        if not isinstance(text, str) or not text:
            raise ValidationError("TRANSCRIPT_INVALID", "transcript text is required")
        if type(is_final) is not bool:
            raise ValidationError("TRANSCRIPT_INVALID", "is_final must be boolean")
        try:
            final_value = int(is_final)
        except (TypeError, ValueError, OverflowError) as exc:
            raise ValidationError("TRANSCRIPT_INVALID", "is_final is invalid") from exc
        now = iso(self._now())
        segment_id = f"seg_{call_id}_{turn}_{speaker}"
        with self.store.tx() as con:
            call = con.execute(
                "SELECT * FROM calls WHERE call_id=?", (call_id,)
            ).fetchone()
            if call is None or call["call_state"] in {"ended", "failed"}:
                return
            previous = con.execute(
                "SELECT * FROM transcripts WHERE segment_id=?", (segment_id,)
            ).fetchone()
            if (
                previous
                and previous["text"] == text
                and bool(previous["is_final"]) == is_final
                and previous["playback_state"] == playback
            ):
                return
            next_revision = (
                _integer(revision)
                if revision is not None
                else (previous["revision"] + 1 if previous else 1)
            )
            if next_revision < 1:
                raise ValidationError("TRANSCRIPT_INVALID", "revision must be positive")
            if previous and previous["is_final"]:
                if previous["text"] != text:
                    raise ConflictError(
                        "TRANSCRIPT_FINAL_CONFLICT",
                        "a final transcript segment cannot be overwritten",
                    )
                if not is_final:
                    return
            if previous and next_revision <= previous["revision"]:
                return
            con.execute(
                """INSERT INTO transcripts(segment_id, call_id, turn_index, speaker, text, revision, is_final, playback_state, updated_at)
                   VALUES(?,?,?,?,?,?,?,?,?)
                   ON CONFLICT(segment_id) DO UPDATE SET text=excluded.text, revision=excluded.revision,
                     is_final=excluded.is_final, playback_state=excluded.playback_state, updated_at=excluded.updated_at""",
                (
                    segment_id,
                    call_id,
                    turn,
                    speaker,
                    text,
                    next_revision,
                    final_value,
                    playback,
                    now,
                ),
            )
            tenant = con.execute(
                "SELECT tenant_key FROM tenants WHERE tenant_id=?", (call["tenant_id"],)
            ).fetchone()["tenant_key"]
            self._event_tx(
                con,
                call["tenant_id"],
                tenant,
                call["command_id"],
                "transcript.updated",
                "transcript_segment",
                segment_id,
                next_revision,
                {
                    "call_id": call_id,
                    "segment_id": segment_id,
                    "turn_index": turn,
                    "speaker": speaker,
                    "text": text,
                    "revision": next_revision,
                    "is_final": is_final,
                    "playback_state": playback,
                },
                now,
            )
        self.mock_ai.turn(call_id, turn, speaker, playback)

    def _finish_call(
        self,
        call_id: str,
        outcome: str,
        reason: str,
        upload_scenario: str | None = None,
        notify_protocol: bool = True,
    ) -> None:
        now_dt = self._now()
        now = iso(now_dt)
        with self.store.tx() as con:
            row = con.execute(
                "SELECT * FROM calls WHERE call_id = ?", (call_id,)
            ).fetchone()
            if row is None or row["call_state"] in {"ended", "failed"}:
                return
            version = row["call_version"] + 1
            start = parse_time(row["started_at"]) if row["started_at"] else now_dt
            duration_ms = max(0, _integer((now_dt - start).total_seconds() * 1000))
            state = "failed" if outcome == "failed" else "ended"
            con.execute(
                "UPDATE calls SET call_state=?, call_version=?, outcome=?, reason_code=?, ended_at=?, duration_ms=? WHERE call_id=?",
                (state, version, outcome, reason, now, duration_ms, call_id),
            )
            con.execute(
                "UPDATE attempts SET status=?, ended_at=? WHERE call_id=?",
                ("failed" if outcome == "failed" else "ended", now, call_id),
            )
            command = con.execute(
                "SELECT * FROM commands WHERE tenant_id=? AND command_id=?",
                (row["tenant_id"], row["command_id"]),
            ).fetchone()
            command_status = (
                "failed" if outcome == "failed" or outcome == "unknown" else "completed"
            )
            command_version = command["aggregate_version"] + 1
            con.execute(
                "UPDATE commands SET status=?, reason_code=?, aggregate_version=?, updated_at=? WHERE tenant_id=? AND command_id=?",
                (
                    command_status,
                    reason,
                    command_version,
                    now,
                    row["tenant_id"],
                    row["command_id"],
                ),
            )
            con.execute(
                "UPDATE executions SET status=?, call_id=? WHERE tenant_id=? AND execution_id=?",
                (command_status, call_id, row["tenant_id"], row["execution_id"]),
            )
            updated = con.execute(
                "SELECT * FROM commands WHERE tenant_id=? AND command_id=?",
                (row["tenant_id"], row["command_id"]),
            ).fetchone()
            self._command_event_tx(con, updated, reason)
            tenant = con.execute(
                "SELECT tenant_key FROM tenants WHERE tenant_id = ?",
                (row["tenant_id"],),
            ).fetchone()["tenant_key"]
            self._event_tx(
                con,
                row["tenant_id"],
                tenant,
                row["command_id"],
                "call.finished",
                "call",
                call_id,
                version,
                {
                    "call_id": call_id,
                    "execution_id": row["execution_id"],
                    "outcome": outcome,
                    "reason_code": reason,
                    "duration_ms": duration_ms,
                },
                now,
            )
        if notify_protocol:
            self.mock_sip.bye(call_id, reason)
            self.mock_ari.media_closed(call_id)
            self.mock_ari.cleanup(call_id)
        if outcome == "completed" or (row["started_at"] and outcome == "failed"):
            self._record_and_upload(call_id, upload_scenario or "success")

    def _ensure_recording(self, call_id: str) -> str | None:
        call = self.store.one("SELECT * FROM calls WHERE call_id=?", (call_id,))
        if call is None or not call["started_at"]:
            return None
        existing = self.store.one(
            "SELECT recording_id FROM recordings WHERE tenant_id=? AND call_id=? ORDER BY created_at LIMIT 1",
            (call["tenant_id"], call_id),
        )
        if existing:
            return existing["recording_id"]
        recording_id = new_id("rec")
        with self.store.tx() as con:
            con.execute(
                """INSERT INTO recordings(
                   recording_id, call_id, tenant_id, content_type, size_bytes,
                   checksum_algorithm, checksum, channels, sample_rate_hz,
                   duration_ms, status, created_at)
                   VALUES(?,?,?,?,?,?,?,?,?,?,?,?)""",
                (
                    recording_id,
                    call_id,
                    call["tenant_id"],
                    "audio/wav",
                    0,
                    "SHA-256",
                    "",
                    1,
                    16000,
                    0,
                    "recording",
                    iso(self._now()),
                ),
            )
        return recording_id

    def _record_and_upload(self, call_id: str, scenario: str) -> None:
        call = self.store.one("SELECT * FROM calls WHERE call_id=?", (call_id,))
        if call is None or not call["started_at"]:
            return
        existing = self.store.one(
            "SELECT * FROM recordings WHERE tenant_id=? AND call_id=? ORDER BY created_at LIMIT 1",
            (call["tenant_id"], call_id),
        )
        if existing and existing["status"] == "verified":
            return
        duration_ms = max(
            1,
            _integer(
                call["duration_ms"] or _integer(self._limits().get("turns", 2)) * 100
            ),
        )
        data = self._wav_bytes(duration_ms)
        checksum = hashlib.sha256(data).hexdigest()
        recording_id = (
            existing["recording_id"] if existing else self._ensure_recording(call_id)
        )
        if recording_id is None:
            return
        with self.store.tx() as con:
            con.execute(
                """UPDATE recordings SET content_type='audio/wav', size_bytes=?,
                   checksum_algorithm='SHA-256', checksum=?, channels=1,
                   sample_rate_hz=16000, duration_ms=?, status='uploading',
                   failure_code=NULL WHERE tenant_id=? AND recording_id=?""",
                (len(data), checksum, duration_ms, call["tenant_id"], recording_id),
            )
        if scenario == "upload_missing":
            self._recording_failed(recording_id, "OBJECT_MISSING", call["tenant_id"])
            return
        request = {
            "recording_id": recording_id,
            "call_id": call_id,
            "content_type": "audio/wav",
            "size_bytes": len(data),
            "checksum_algorithm": "SHA-256",
            "checksum": checksum,
            "channels": 1,
            "sample_rate_hz": 16000,
            "duration_ms": duration_ms,
        }
        try:
            _, session = self.create_upload(
                call["tenant_id"], request, f"executor:{call_id}", "executor"
            )
            upload_id = session["upload_id"]
            token = session["required_headers"]["X-Mock-Upload-Token"]
            upload_data = data
            if scenario == "upload_bad_checksum":
                upload_data = data[:-1] + bytes([data[-1] ^ 1])
            self.put_upload(upload_id, token, upload_data)
            self.complete_upload(
                call["tenant_id"],
                upload_id,
                {
                    "recording_id": recording_id,
                    "size_bytes": len(data),
                    "checksum_algorithm": "SHA-256",
                    "checksum": checksum,
                },
                f"executor-complete:{call_id}",
                "executor",
            )
        except ServiceError as exc:
            self._recording_failed(recording_id, exc.code, call["tenant_id"])
        except (OSError, RuntimeError, TypeError, ValueError):
            self._recording_failed(
                recording_id, "OBJECT_UPLOAD_FAILED", call["tenant_id"]
            )

    def _wav_bytes(self, duration_ms: int | None = None) -> bytes:
        try:
            duration_ms = max(
                1,
                _integer(duration_ms or _integer(self._limits().get("turns", 2)) * 100),
            )
            frames = max(1, int(16000 * duration_ms / 1000))
            output = io.BytesIO()
            with wave.open(output, "wb") as wav:
                wav.setnchannels(1)
                wav.setsampwidth(2)
                wav.setframerate(16000)
                samples = bytearray()
                for index in range(frames):
                    seconds = index / 16000
                    value = 11000 * math.sin(
                        2 * math.pi * 440 * seconds
                    ) + 7000 * math.sin(2 * math.pi * 660 * seconds)
                    samples.extend(
                        struct.pack("<h", max(-32768, min(32767, int(value))))
                    )
                wav.writeframes(bytes(samples))
            return output.getvalue()
        except (OSError, OverflowError, TypeError, ValueError, struct.error) as exc:
            raise ServiceError(
                "RECORDING_GENERATION_FAILED", "recording could not be generated", 500
            ) from exc

    def _recording_failed(self, recording_id: str, code: str, tenant_id: str) -> None:
        with self.store.tx() as con:
            recording = con.execute(
                "SELECT * FROM recordings WHERE tenant_id=? AND recording_id=?",
                (tenant_id, recording_id),
            ).fetchone()
            if recording is None or recording["status"] in {"verified", "failed"}:
                return
            changed = con.execute(
                "UPDATE recordings SET status='failed', failure_code=? WHERE tenant_id=? AND recording_id=? AND status NOT IN ('verified','failed')",
                (code, recording["tenant_id"], recording_id),
            ).rowcount
            tenant = con.execute(
                "SELECT tenant_key FROM tenants WHERE tenant_id = ?",
                (recording["tenant_id"],),
            ).fetchone()["tenant_key"]
            call = con.execute(
                "SELECT command_id FROM calls WHERE tenant_id=? AND call_id=?",
                (recording["tenant_id"], recording["call_id"]),
            ).fetchone()
            if not changed or call is None:
                return
            self._event_tx(
                con,
                recording["tenant_id"],
                tenant,
                call["command_id"],
                "recording.failed",
                "recording",
                recording_id,
                1,
                {
                    "recording_id": recording_id,
                    "call_id": recording["call_id"],
                    "failure_code": code,
                },
                iso(self._now()),
            )

    def create_upload(
        self,
        tenant_id: str,
        request: dict[str, Any],
        idempotency_key: str | None = None,
        actor_id: str = "internal",
    ) -> tuple[int, dict[str, Any]]:
        self._tenant(tenant_id)
        required = {
            "recording_id",
            "call_id",
            "content_type",
            "size_bytes",
            "checksum_algorithm",
            "checksum",
            "channels",
            "sample_rate_hz",
            "duration_ms",
        }
        if (
            set(request) != required
            or request.get("content_type") != "audio/wav"
            or request.get("checksum_algorithm") != "SHA-256"
        ):
            raise ValidationError(
                "SCHEMA_INVALID", "upload request fields do not match the contract"
            )
        if (
            type(request.get("size_bytes")) is not int
            or request["size_bytes"] < 1
            or request["size_bytes"]
            > _integer(self._limits().get("max_recording_bytes", 64 * 1024 * 1024))
        ):
            raise ValidationError(
                "SCHEMA_INVALID", "size_bytes is outside the allowed range"
            )
        if (
            type(request.get("channels")) is not int
            or request["channels"] != 1
            or type(request.get("sample_rate_hz")) is not int
            or not 8000 <= request["sample_rate_hz"] <= 48000
            or type(request.get("duration_ms")) is not int
            or not 1 <= request["duration_ms"] <= 3600000
        ):
            raise ValidationError(
                "SCHEMA_INVALID", "audio metadata is outside the allowed range"
            )
        if not isinstance(request.get("checksum"), str) or not re.fullmatch(
            r"[0-9a-f]{64}", request["checksum"]
        ):
            raise ValidationError(
                "SCHEMA_INVALID", "checksum must be a SHA-256 hex digest"
            )
        idempotency_key = idempotency_key or f"internal:{request['recording_id']}"
        operation_hash = self._operation_hash(
            "recording.create", request["recording_id"], request
        )
        existing_operation = self.store.one(
            "SELECT * FROM storage_operations WHERE tenant_id=? AND actor_id=? AND operation_type=? AND target_id=? AND idempotency_key=?",
            (
                tenant_id,
                actor_id,
                "recording.create",
                request["recording_id"],
                idempotency_key,
            ),
        )
        if existing_operation:
            if existing_operation["payload_hash"] != operation_hash:
                raise ConflictError(
                    "IDEMPOTENCY_CONFLICT",
                    "upload request was already submitted with a different body",
                )
            return existing_operation["status"], _loads(
                existing_operation["response_json"]
            )
        recording = self.store.one(
            "SELECT * FROM recordings WHERE recording_id=? AND tenant_id=?",
            (request["recording_id"], tenant_id),
        )
        if recording is None or recording["call_id"] != request["call_id"]:
            raise NotFoundError()
        metadata = (
            "content_type",
            "size_bytes",
            "checksum_algorithm",
            "checksum",
            "channels",
            "sample_rate_hz",
            "duration_ms",
        )
        if any(recording[field] != request[field] for field in metadata):
            raise ConflictError(
                "RECORDING_METADATA_CONFLICT", "recording metadata is immutable"
            )
        upload = self.store.one(
            "SELECT * FROM uploads WHERE tenant_id=? AND recording_id=? ORDER BY created_at DESC LIMIT 1",
            (tenant_id, request["recording_id"]),
        )
        if upload and (
            upload["status"] == "verified"
            or (
                parse_time(upload["expires_at"]) > self._now()
                and upload["status"] in {"created", "uploaded"}
            )
        ):
            status, upload_id, expires = 200, upload["upload_id"], upload["expires_at"]
        else:
            status = 201
            upload_id = new_id("upload")
            expires = iso(
                self._now()
                + timedelta(seconds=_integer(self._limits().get("upload_ttl_s", 300)))
            )
            path = str(self.object_store.path_for(upload_id))
            with self.store.tx() as con:
                con.execute(
                    "INSERT INTO uploads(upload_id, recording_id, tenant_id, expected_size, expected_checksum, expires_at, status, object_path, created_at) VALUES(?,?,?,?,?,?,?,?,?)",
                    (
                        upload_id,
                        request["recording_id"],
                        tenant_id,
                        request["size_bytes"],
                        request["checksum"],
                        expires,
                        "created",
                        path,
                        iso(self._now()),
                    ),
                )
                con.execute(
                    "UPDATE recordings SET upload_id=?, status='uploading' WHERE tenant_id=? AND recording_id=?",
                    (upload_id, tenant_id, request["recording_id"]),
                )
        response = {
            "upload_id": upload_id,
            "recording_id": request["recording_id"],
            "expires_at": expires,
            "upload_method": "PUT",
            "upload_url": f"/_mock/oss/{upload_id}",
            "required_headers": {
                "X-Mock-Upload-Token": self.object_store.token(upload_id, tenant_id)
            },
            "constraints": {
                "size_bytes": request["size_bytes"],
                "checksum_algorithm": "SHA-256",
                "checksum": request["checksum"],
            },
        }
        cleanup_upload_id: str | None = None
        with self.store.tx() as con:
            inserted = con.execute(
                """INSERT OR IGNORE INTO storage_operations(tenant_id, actor_id, operation_type, target_id, idempotency_key, payload_hash, status, response_json, created_at, updated_at)
                   VALUES(?,?,?,?,?,?,?,?,?,?)""",
                (
                    tenant_id,
                    actor_id,
                    "recording.create",
                    request["recording_id"],
                    idempotency_key,
                    operation_hash,
                    status,
                    canonical(response),
                    iso(self._now()),
                    iso(self._now()),
                ),
            ).rowcount
            if not inserted:
                existing_operation = con.execute(
                    "SELECT * FROM storage_operations WHERE tenant_id=? AND actor_id=? AND operation_type=? AND target_id=? AND idempotency_key=?",
                    (
                        tenant_id,
                        actor_id,
                        "recording.create",
                        request["recording_id"],
                        idempotency_key,
                    ),
                ).fetchone()
                if (
                    existing_operation is None
                    or existing_operation["payload_hash"] != operation_hash
                ):
                    raise ConflictError(
                        "IDEMPOTENCY_CONFLICT",
                        "upload request was already submitted with a different body",
                    )
                winner = _loads(existing_operation["response_json"])
                if upload_id != winner["upload_id"]:
                    con.execute(
                        "DELETE FROM uploads WHERE tenant_id=? AND upload_id=? AND recording_id=? AND status='created'",
                        (tenant_id, upload_id, request["recording_id"]),
                    )
                    con.execute(
                        "UPDATE recordings SET upload_id=? WHERE tenant_id=? AND recording_id=?",
                        (winner["upload_id"], tenant_id, request["recording_id"]),
                    )
                    cleanup_upload_id = upload_id
                result = (existing_operation["status"], winner)
            else:
                result = (status, response)
        if cleanup_upload_id:
            with suppress(OSError):
                self.object_store.path_for(cleanup_upload_id).unlink()
        return result

    def put_upload(self, upload_id: str, token: str, data: bytes) -> None:
        upload = self.store.one("SELECT * FROM uploads WHERE upload_id=?", (upload_id,))
        if upload is None:
            raise NotFoundError()
        if not isinstance(data, bytes) or len(data) > _integer(
            self._limits().get("max_recording_bytes", 64 * 1024 * 1024)
        ):
            raise ValidationError(
                "OBJECT_BODY_INVALID", "upload body is outside the allowed range"
            )
        if parse_time(upload["expires_at"]) <= self._now():
            raise ServiceError("UPLOAD_EXPIRED", "upload session has expired", 410)
        if upload["status"] == "verified" or self.object_store.has_final(upload_id):
            raise ConflictError(
                "UPLOAD_ALREADY_FINALIZED", "verified upload objects are immutable"
            )
        if (
            upload["status"] == "uploaded"
            and self.object_store.path_for(upload_id).is_file()
        ):
            existing = self.object_store.path_for(upload_id).read_bytes()
            if (
                len(existing) == len(data)
                and hashlib.sha256(existing).hexdigest()
                == hashlib.sha256(data).hexdigest()
            ):
                return
            raise ConflictError(
                "UPLOAD_ALREADY_RECEIVED",
                "upload session already contains different bytes",
            )
        self.object_store.put(upload_id, upload["tenant_id"], data, token)
        with self.store.tx() as con:
            con.execute(
                "UPDATE uploads SET status='uploaded' WHERE tenant_id=? AND upload_id=?",
                (upload["tenant_id"], upload_id),
            )

    def _complete_upload(
        self,
        tenant_id: str,
        upload_id: str,
        recording_id: str,
        checksum: str,
        size_bytes: int,
        recovery: bool = False,
    ) -> dict[str, Any]:
        upload = self.store.one(
            "SELECT * FROM uploads WHERE upload_id=? AND tenant_id=?",
            (upload_id, tenant_id),
        )
        if upload is None or upload["recording_id"] != recording_id:
            raise NotFoundError()
        recording = self.store.one(
            "SELECT * FROM recordings WHERE recording_id=? AND tenant_id=?",
            (recording_id, tenant_id),
        )
        if recording is None:
            raise NotFoundError()
        if upload["status"] == "verified" and recording["status"] == "verified":
            return {
                "upload_id": upload_id,
                "recording_id": recording_id,
                "status": "verified",
                "oss_id": recording["oss_id"],
                "verified_at": recording["verified_at"],
            }
        if parse_time(upload["expires_at"]) <= self._now() and not (
            recovery and self.object_store.has_final(upload_id)
        ):
            raise ServiceError("UPLOAD_EXPIRED", "upload session has expired", 410)
        if (
            type(size_bytes) is not int
            or not isinstance(checksum, str)
            or size_bytes != upload["expected_size"]
            or checksum != upload["expected_checksum"]
        ):
            raise ValidationError(
                "RECORDING_METADATA_CONFLICT",
                "complete request does not match the reserved metadata",
            )
        try:
            oss_id = self.object_store.verify_and_promote(
                upload_id, upload["expected_size"], upload["expected_checksum"]
            )
        except ValidationError:
            self._recording_failed(
                recording_id, "OBJECT_VERIFICATION_FAILED", tenant_id
            )
            raise
        now = iso(self._now())
        with self.store.tx() as con:
            con.execute(
                "UPDATE uploads SET status='verified', completed_at=? WHERE tenant_id=? AND upload_id=?",
                (now, tenant_id, upload_id),
            )
            changed = con.execute(
                "UPDATE recordings SET status='verified', oss_id=?, verified_at=?, failure_code=NULL WHERE tenant_id=? AND recording_id=? AND status!='verified'",
                (oss_id, now, tenant_id, recording_id),
            ).rowcount
            recording = con.execute(
                "SELECT * FROM recordings WHERE tenant_id=? AND recording_id=?",
                (tenant_id, recording_id),
            ).fetchone()
            call = con.execute(
                "SELECT command_id FROM calls WHERE tenant_id=? AND call_id=?",
                (tenant_id, recording["call_id"]),
            ).fetchone()
            tenant = con.execute(
                "SELECT tenant_key FROM tenants WHERE tenant_id=?", (tenant_id,)
            ).fetchone()["tenant_key"]
            if changed:
                self._event_tx(
                    con,
                    tenant_id,
                    tenant,
                    call["command_id"],
                    "recording.ready",
                    "recording",
                    recording_id,
                    1,
                    {
                        "recording_id": recording_id,
                        "call_id": recording["call_id"],
                        "oss_id": oss_id,
                        "size_bytes": recording["size_bytes"],
                        "checksum": recording["checksum"],
                    },
                    now,
                )
        return {
            "upload_id": upload_id,
            "recording_id": recording_id,
            "status": "verified",
            "oss_id": oss_id,
            "verified_at": recording["verified_at"],
        }

    def complete_upload(
        self,
        tenant_id: str,
        upload_id: str,
        request: dict[str, Any],
        idempotency_key: str | None = None,
        actor_id: str = "internal",
    ) -> dict[str, Any]:
        required = {"recording_id", "size_bytes", "checksum_algorithm", "checksum"}
        allowed = required | {"etag"}
        if (
            not required.issubset(request)
            or set(request) - allowed
            or request.get("checksum_algorithm") != "SHA-256"
            or type(request.get("size_bytes")) is not int
            or not isinstance(request.get("checksum"), str)
        ):
            raise ValidationError(
                "SCHEMA_INVALID", "complete request fields do not match the contract"
            )
        idempotency_key = idempotency_key or f"internal:{upload_id}"
        operation_hash = self._operation_hash("recording.complete", upload_id, request)
        existing = self.store.one(
            "SELECT * FROM storage_operations WHERE tenant_id=? AND actor_id=? AND operation_type=? AND target_id=? AND idempotency_key=?",
            (tenant_id, actor_id, "recording.complete", upload_id, idempotency_key),
        )
        if existing:
            if existing["payload_hash"] != operation_hash:
                raise ConflictError(
                    "IDEMPOTENCY_CONFLICT",
                    "upload completion was already submitted with a different body",
                )
            return _loads(existing["response_json"])
        response = self._complete_upload(
            tenant_id,
            upload_id,
            request["recording_id"],
            request["checksum"],
            request["size_bytes"],
        )
        with self.store.tx() as con:
            inserted = con.execute(
                "INSERT OR IGNORE INTO storage_operations(tenant_id, actor_id, operation_type, target_id, idempotency_key, payload_hash, status, response_json, created_at, updated_at) VALUES(?,?,?,?,?,?,?,?,?,?)",
                (
                    tenant_id,
                    actor_id,
                    "recording.complete",
                    upload_id,
                    idempotency_key,
                    operation_hash,
                    200,
                    canonical(response),
                    iso(self._now()),
                    iso(self._now()),
                ),
            ).rowcount
            if not inserted:
                existing = con.execute(
                    "SELECT * FROM storage_operations WHERE tenant_id=? AND actor_id=? AND operation_type=? AND target_id=? AND idempotency_key=?",
                    (
                        tenant_id,
                        actor_id,
                        "recording.complete",
                        upload_id,
                        idempotency_key,
                    ),
                ).fetchone()
                if existing is None or existing["payload_hash"] != operation_hash:
                    raise ConflictError(
                        "IDEMPOTENCY_CONFLICT",
                        "upload completion was already submitted with a different body",
                    )
                return _loads(existing["response_json"])
        return response

    def dispatch_outbox(self, limit: int = 100) -> int:
        now = iso(self._now())
        rows = self.store.all(
            "SELECT * FROM events WHERE published=0 AND (next_attempt_at IS NULL OR next_attempt_at <= ?) ORDER BY occurred_at, event_id LIMIT ?",
            (now, limit),
        )
        sent = 0
        for row in rows:
            body = _loads(row["body_json"])
            try:
                self.broker.publish(
                    "agent-call.events.v1",
                    f"agent-call.{row['event_type']}",
                    body,
                    row["event_id"],
                )
            except BrokerError:
                with self.store.tx() as con:
                    attempts = row["broker_attempts"] + 1
                    retry_at = iso(
                        self._now() + timedelta(seconds=min(60, 2 ** min(attempts, 5)))
                    )
                    con.execute(
                        "UPDATE events SET broker_attempts=?, broker_status='failed', next_attempt_at=? WHERE tenant_id=? AND event_id=?",
                        (attempts, retry_at, row["tenant_id"], row["event_id"]),
                    )
                continue
            with self.store.tx() as con:
                con.execute(
                    "UPDATE events SET published=1, broker_status='broker_confirmed', published_at=?, broker_attempts=broker_attempts+1 WHERE tenant_id=? AND event_id=?",
                    (iso(self._now()), row["tenant_id"], row["event_id"]),
                )
            sent += 1
        return sent

    def consume_saas_events(self, limit: int = 100) -> int:
        applied = 0
        for _ in range(limit):
            try:
                message = self.broker.consume("agent-call.saas.events.v1")
            except BrokerError:
                break
            if not message:
                break
            body_raw: Any = message.get("body", {})
            body = body_raw if isinstance(body_raw, dict) else {}
            event_id = body.get("event_id")
            if not event_id:
                self._dead_letter(message, body, "INBOX_INVALID")
                reject = getattr(self.broker, "reject", None)
                if callable(reject):
                    reject(message)
                else:
                    self.broker.ack(message)
                continue
            try:
                self._validate_event(body)
                tenant = self._tenant(body["tenant_id"], body["tenant_key"])
                expected_route = f"agent-call.{body['event_type']}"
                if message.get("routing_key") not in {None, expected_route}:
                    raise ValidationError(
                        "EVENT_ROUTE_MISMATCH", "event routing key is invalid"
                    )
                if message.get("message_id") not in {None, event_id}:
                    raise ValidationError(
                        "EVENT_ID_MISMATCH", "event message id is invalid"
                    )
                if body["event_type"] == "recording.ready":
                    recording_payload = body.get("payload", {})
                    recording = self.store.one(
                        "SELECT status, oss_id FROM recordings WHERE tenant_id=? AND recording_id=?",
                        (tenant["tenant_id"], recording_payload.get("recording_id")),
                    )
                    if (
                        recording is None
                        or recording["status"] != "verified"
                        or recording["oss_id"] != recording_payload.get("oss_id")
                    ):
                        raise ValidationError(
                            "ASSET_NOT_VERIFIED",
                            "recording.ready is not independently verified",
                        )
                with self.store.tx() as con:
                    exists = con.execute(
                        "SELECT 1 FROM inbox WHERE tenant_id=? AND event_id=?",
                        (tenant["tenant_id"], event_id),
                    ).fetchone()
                    if not exists:
                        con.execute(
                            "INSERT INTO inbox(event_id, tenant_id, applied_at, body_json) VALUES(?,?,?,?)",
                            (
                                event_id,
                                tenant["tenant_id"],
                                iso(self._now()),
                                canonical(body),
                            ),
                        )
                        applied += 1
                    con.execute(
                        "UPDATE events SET saas_applied=1 WHERE tenant_id=? AND event_id=?",
                        (tenant["tenant_id"], event_id),
                    )
                self.broker.ack(message)
            except (KeyError, ServiceError, sqlite3.IntegrityError) as exc:
                self._dead_letter(message, body, getattr(exc, "code", "INBOX_INVALID"))
                reject = getattr(self.broker, "reject", None)
                if callable(reject):
                    reject(message)
                else:
                    self.broker.ack(message)
        return applied

    def apply_controls(self) -> int:
        rows = self.store.all(
            "SELECT * FROM commands WHERE command_type='task.control' AND status='applying' ORDER BY accepted_at"
        )
        count = 0
        cell_ids = [str(cell["cell_id"]) for cell in self.profile.get("cells", [])]
        for row in rows:
            body = _loads(row["payload_json"])
            action = body["action"]
            policy = body.get("active_call_policy", "drain")
            if not all(self._cell_health.get(cell_id, False) for cell_id in cell_ids):
                continue
            if action == "stop" and policy == "hangup":
                calls = self.store.all(
                    "SELECT call_id FROM calls WHERE tenant_id=? AND task_id=? AND call_state NOT IN ('ended','failed')",
                    (row["tenant_id"], row["task_id"]),
                )
                for call in calls:
                    self._finish_controlled_call(call["call_id"])
            with self.store.tx() as con:
                task = con.execute(
                    "SELECT * FROM tasks WHERE tenant_id=? AND task_id=?",
                    (row["tenant_id"], row["task_id"]),
                ).fetchone()
                if task is None:
                    continue
                now = iso(self._now())
                for cell_id in cell_ids:
                    con.execute(
                        "INSERT INTO control_barriers(command_id, tenant_id, cell_id, status, applied_at) VALUES(?,?,?,?,?) ON CONFLICT(tenant_id, command_id, cell_id) DO UPDATE SET status=excluded.status, applied_at=excluded.applied_at",
                        (row["command_id"], row["tenant_id"], cell_id, "applied", now),
                    )
                con.execute(
                    "UPDATE commands SET status='applied', applied_task_revision=?, task_state=?, aggregate_version=aggregate_version+1, updated_at=? WHERE tenant_id=? AND command_id=? AND status='applying'",
                    (
                        task["revision"],
                        task["state"],
                        now,
                        row["tenant_id"],
                        row["command_id"],
                    ),
                )
                updated = con.execute(
                    "SELECT * FROM commands WHERE tenant_id=? AND command_id=?",
                    (row["tenant_id"], row["command_id"]),
                ).fetchone()
                self._command_event_tx(con, updated, "CONTROL_APPLIED")
                affected = con.execute(
                    "SELECT * FROM commands WHERE tenant_id=? AND task_id=? AND status IN ('accepted','waiting')",
                    (row["tenant_id"], row["task_id"]),
                ).fetchall()
                for affected_row in affected:
                    con.execute(
                        "UPDATE commands SET status='rejected', reason_code='TASK_CONTROL_BARRIER', aggregate_version=aggregate_version+1, updated_at=? WHERE tenant_id=? AND command_id=?",
                        (now, affected_row["tenant_id"], affected_row["command_id"]),
                    )
                    con.execute(
                        "UPDATE executions SET status='rejected' WHERE tenant_id=? AND execution_id=?",
                        (affected_row["tenant_id"], affected_row["execution_id"]),
                    )
                    barrier_row = con.execute(
                        "SELECT * FROM commands WHERE tenant_id=? AND command_id=?",
                        (affected_row["tenant_id"], affected_row["command_id"]),
                    ).fetchone()
                    self._command_event_tx(con, barrier_row, "TASK_CONTROL_BARRIER")
            count += 1
        return count

    def _finish_controlled_call(self, call_id: str) -> None:
        row = self.store.one("SELECT call_state FROM calls WHERE call_id=?", (call_id,))
        if row is not None and row["call_state"] == "reconciling":
            return
        self._finish_call(call_id, "failed", "CONTROL_STOPPED")

    def control_task(
        self,
        tenant_id: str,
        task_id: str,
        body: dict[str, Any],
        idempotency_key: str,
        allow_hangup: bool = False,
    ) -> dict[str, Any]:
        tenant = self._tenant(tenant_id)
        required = {"command_id", "action", "expected_task_revision", "reason"}
        allowed = required | {"active_call_policy"}
        if not required.issubset(body) or set(body) - allowed:
            raise ValidationError(
                "SCHEMA_INVALID", "control request fields do not match the contract"
            )
        if body["action"] not in {"pause", "resume", "stop"}:
            raise ValidationError("SCHEMA_INVALID", "control action is invalid")
        if (
            type(body["expected_task_revision"]) is not int
            or body["expected_task_revision"] < 1
        ):
            raise ValidationError(
                "SCHEMA_INVALID", "expected_task_revision must be a positive integer"
            )
        if (
            not isinstance(body["reason"], str)
            or not body["reason"]
            or len(body["reason"]) > 512
        ):
            raise ValidationError(
                "SCHEMA_INVALID", "reason must contain 1..512 characters"
            )
        if body["action"] != "stop" and "active_call_policy" in body:
            raise ValidationError(
                "SCHEMA_INVALID", "active_call_policy is only valid for stop"
            )
        if body.get("active_call_policy", "drain") not in {"drain", "hangup"}:
            raise ValidationError("SCHEMA_INVALID", "active_call_policy is invalid")
        if body["command_id"] != idempotency_key:
            raise ConflictError(
                "IDEMPOTENCY_KEY_MISMATCH", "Idempotency-Key must equal command_id"
            )
        if (
            body["action"] == "stop"
            and body.get("active_call_policy") == "hangup"
            and not allow_hangup
        ):
            raise ServiceError(
                "SCOPE_REQUIRED", "hangup control requires outbound.hangup scope", 403
            )
        task = self.store.one(
            "SELECT * FROM tasks WHERE task_id=? AND tenant_id=?", (task_id, tenant_id)
        )
        if task is None:
            raise NotFoundError()
        payload_hash = self._operation_hash("task.control", task_id, body)
        existing = self.store.one(
            "SELECT * FROM commands WHERE command_id=? AND tenant_id=?",
            (body["command_id"], tenant_id),
        )
        if existing:
            if existing["payload_hash"] != payload_hash:
                raise ConflictError(
                    "IDEMPOTENCY_CONFLICT",
                    "control command was already submitted with a different body",
                )
            return self._control_response(existing)
        if body["expected_task_revision"] != task["revision"]:
            raise ConflictError(
                "TASK_REVISION_CONFLICT",
                "expected_task_revision does not match current task revision",
            )
        if body["action"] == "resume":
            applying = self.store.one(
                "SELECT 1 FROM commands WHERE tenant_id=? AND task_id=? AND command_type='task.control' AND status='applying' LIMIT 1",
                (tenant_id, task_id),
            )
            if applying:
                raise ConflictError(
                    "CONTROL_IN_PROGRESS",
                    "the previous control barrier has not applied",
                )
            if task["state"] != "paused":
                raise ConflictError(
                    "TASK_NOT_RESUMABLE", "only a paused task can be resumed"
                )
        if body["action"] in {"pause", "stop"} and task["state"] == "stopped":
            raise ConflictError("TASK_STOPPED", "stopped task cannot be changed")
        new_state = {"pause": "paused", "resume": "running", "stop": "stopped"}[
            body["action"]
        ]
        now = iso(self._now())
        with self.store.tx() as con:
            con.execute(
                "UPDATE tasks SET state=?, revision=revision+1, updated_at=? WHERE tenant_id=? AND task_id=? AND revision=?",
                (new_state, now, tenant_id, task_id, body["expected_task_revision"]),
            )
            if con.execute("SELECT changes()").fetchone()[0] != 1:
                raise ConflictError(
                    "TASK_REVISION_CONFLICT", "task changed concurrently"
                )
            con.execute(
                """INSERT INTO commands(
                   command_id, tenant_id, tenant_key, command_type, status,
                   payload_json, payload_hash, operation_type, operation_target,
                   task_id, requested_task_revision, task_state, accepted_at,
                   trace_id, aggregate_version, updated_at)
                   VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
                (
                    body["command_id"],
                    tenant_id,
                    tenant["tenant_key"],
                    "task.control",
                    "applying",
                    canonical(body),
                    payload_hash,
                    "task.control",
                    task_id,
                    task_id,
                    body["expected_task_revision"],
                    new_state,
                    now,
                    body["command_id"],
                    1,
                    now,
                ),
            )
            row = con.execute(
                "SELECT * FROM commands WHERE tenant_id=? AND command_id=?",
                (tenant_id, body["command_id"]),
            ).fetchone()
            self._command_event_tx(con, row, "CONTROL_ACCEPTED")
        return self._control_response(
            self.store.one(
                "SELECT * FROM commands WHERE tenant_id=? AND command_id=?",
                (tenant_id, body["command_id"]),
            )
        )

    def _control_response(self, row: sqlite3.Row) -> dict[str, Any]:
        return {
            "command_id": row["command_id"],
            "tenant_id": row["tenant_id"],
            "tenant_key": row["tenant_key"],
            "task_id": row["task_id"],
            "status": "accepted",
            "requested_task_revision": row["requested_task_revision"],
            "accepted_at": row["accepted_at"],
        }

    def _events_for_call(
        self, tenant_id: str, call_id: str, cutoff: str | None = None
    ) -> list[sqlite3.Row]:
        sql = "SELECT * FROM events WHERE tenant_id=?"
        params: list[Any] = [tenant_id]
        if cutoff is not None:
            sql += " AND occurred_at<=?"
            params.append(cutoff)
        sql += " ORDER BY occurred_at, event_id"
        rows = self.store.all(sql, tuple(params))
        matched: list[sqlite3.Row] = []
        for row in rows:
            if row["aggregate_type"] == "call" and row["aggregate_id"] == call_id:
                matched.append(row)
                continue
            payload = _loads(row["payload_json"], {})
            if isinstance(payload, dict) and payload.get("call_id") == call_id:
                matched.append(row)
        return matched

    def create_replay(
        self,
        tenant_id: str,
        source_type: str,
        source_id: str,
        body: dict[str, Any],
        idempotency_key: str,
    ) -> dict[str, Any]:
        self._tenant(tenant_id)
        if source_type not in {"call", "command"}:
            raise ValidationError("SCHEMA_INVALID", "replay source type is invalid")
        if (
            set(body) != {"command_id", "reason"}
            or body["command_id"] != idempotency_key
            or not isinstance(body["reason"], str)
            or not body["reason"]
            or len(body["reason"]) > 512
        ):
            raise ValidationError(
                "SCHEMA_INVALID",
                "replay request requires command_id and reason matching Idempotency-Key",
            )
        existing = self.store.one(
            "SELECT * FROM commands WHERE command_id=? AND tenant_id=?",
            (body["command_id"], tenant_id),
        )
        operation_hash = self._operation_hash(
            "replay", f"{source_type}:{source_id}", body
        )
        if existing:
            if existing["payload_hash"] != operation_hash:
                raise ConflictError(
                    "IDEMPOTENCY_CONFLICT",
                    "replay command was already submitted with a different body",
                )
            return {
                "command_id": body["command_id"],
                "status": "accepted",
                "snapshot_cutoff": existing["snapshot_cutoff"],
            }
        if source_type == "call":
            source = self.store.one(
                "SELECT * FROM calls WHERE call_id=? AND tenant_id=?",
                (source_id, tenant_id),
            )
        else:
            source = self.store.one(
                "SELECT * FROM commands WHERE command_id=? AND tenant_id=?",
                (source_id, tenant_id),
            )
        if source is None:
            raise NotFoundError()
        cutoff = iso(self._now())
        retention = self._now() - timedelta(
            seconds=self._limits().get("replay_retention_s", 604800)
        )
        rows = (
            self._events_for_call(tenant_id, source_id, cutoff)
            if source_type == "call"
            else self.store.all(
                "SELECT occurred_at FROM events WHERE tenant_id=? AND aggregate_id=? AND occurred_at<=? ORDER BY occurred_at LIMIT 1",
                (tenant_id, source_id, cutoff),
            )
        )
        if rows and parse_time(rows[0]["occurred_at"]) < retention:
            raise ServiceError(
                "REPLAY_EXPIRED", "source is outside the replay retention window", 410
            )
        replay_id = new_id("replay")
        tenant = self.store.one(
            "SELECT tenant_key FROM tenants WHERE tenant_id=?", (tenant_id,)
        )["tenant_key"]
        now = iso(self._now())
        with self.store.tx() as con:
            con.execute(
                """INSERT INTO commands(
                   command_id, tenant_id, tenant_key, command_type, status,
                   payload_json, payload_hash, operation_type, operation_target,
                   snapshot_cutoff, source_type, source_id, accepted_at, trace_id,
                   aggregate_version, updated_at)
                   VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
                (
                    body["command_id"],
                    tenant_id,
                    tenant,
                    "replay",
                    "accepted",
                    canonical(body),
                    operation_hash,
                    "replay",
                    f"{source_type}:{source_id}",
                    cutoff,
                    source_type,
                    source_id,
                    now,
                    body["command_id"],
                    1,
                    now,
                ),
            )
            con.execute(
                "INSERT INTO replays(replay_id, command_id, tenant_id, source_type, source_id, snapshot_cutoff, status, created_at, updated_at) VALUES(?,?,?,?,?,?,?,?,?)",
                (
                    replay_id,
                    body["command_id"],
                    tenant_id,
                    source_type,
                    source_id,
                    cutoff,
                    "accepted",
                    now,
                    now,
                ),
            )
            row = con.execute(
                "SELECT * FROM commands WHERE tenant_id=? AND command_id=?",
                (tenant_id, body["command_id"]),
            ).fetchone()
            self._command_event_tx(con, row, "REPLAY_ACCEPTED")
        return {
            "command_id": body["command_id"],
            "status": "accepted",
            "snapshot_cutoff": cutoff,
        }

    def process_replays(self) -> int:
        now = iso(self._now())
        rows = self.store.all(
            "SELECT * FROM replays WHERE status='accepted' AND (next_attempt_at IS NULL OR next_attempt_at<=?) ORDER BY created_at LIMIT 10",
            (now,),
        )
        count = 0
        max_attempts = max(1, _integer(self._limits().get("max_replay_attempts", 6)))
        for replay in rows:
            tenant_row = self.store.one(
                "SELECT tenant_key FROM tenants WHERE tenant_id=?",
                (replay["tenant_id"],),
            )
            if tenant_row is None:
                continue
            tenant_key = tenant_row["tenant_key"]
            if replay["source_type"] == "call":
                candidates = self._events_for_call(
                    replay["tenant_id"], replay["source_id"], replay["snapshot_cutoff"]
                )
            else:
                candidates = self.store.all(
                    "SELECT * FROM events WHERE tenant_id=? AND aggregate_id=? AND occurred_at<=? ORDER BY occurred_at, event_id",
                    (
                        replay["tenant_id"],
                        replay["source_id"],
                        replay["snapshot_cutoff"],
                    ),
                )
            sent = _integer(replay["sent_count"])
            already = {
                item["event_id"]
                for item in self.store.all(
                    "SELECT event_id FROM replay_items WHERE replay_id=?",
                    (replay["replay_id"],),
                )
            }
            try:
                for event in candidates:
                    if event["event_id"] in already:
                        continue
                    self.broker.publish(
                        "agent-call.events.v1",
                        f"agent-call.{event['event_type']}",
                        _loads(event["body_json"]),
                        event["event_id"],
                        {
                            "x-replay-id": replay["replay_id"],
                            "tenant_id": replay["tenant_id"],
                            "tenant_key": tenant_key,
                        },
                    )
                    with self.store.tx() as con:
                        con.execute(
                            "INSERT OR IGNORE INTO replay_items(replay_id, event_id, published_at) VALUES(?,?,?)",
                            (replay["replay_id"], event["event_id"], iso(self._now())),
                        )
                    already.add(event["event_id"])
                    sent += 1
            except BrokerError:
                attempts = _integer(replay["retry_attempts"]) + 1
                status = "failed" if attempts >= max_attempts else "accepted"
                with self.store.tx() as con:
                    updated_at = iso(self._now())
                    next_attempt = (
                        None
                        if status == "failed"
                        else iso(
                            self._now()
                            + timedelta(seconds=min(60, 2 ** min(attempts, 5)))
                        )
                    )
                    con.execute(
                        "UPDATE replays SET status=?, failure_code='BROKER_UNAVAILABLE', retry_attempts=?, sent_count=?, next_attempt_at=?, updated_at=? WHERE tenant_id=? AND replay_id=?",
                        (
                            status,
                            attempts,
                            sent,
                            next_attempt,
                            updated_at,
                            replay["tenant_id"],
                            replay["replay_id"],
                        ),
                    )
                    if status == "failed":
                        con.execute(
                            "UPDATE commands SET status='failed', reason_code='REPLAY_BROKER_EXHAUSTED', aggregate_version=aggregate_version+1, updated_at=? WHERE tenant_id=? AND command_id=?",
                            (updated_at, replay["tenant_id"], replay["command_id"]),
                        )
                        failed_command = con.execute(
                            "SELECT * FROM commands WHERE tenant_id=? AND command_id=?",
                            (replay["tenant_id"], replay["command_id"]),
                        ).fetchone()
                        self._command_event_tx(con, failed_command, "REPLAY_FAILED")
                continue
            with self.store.tx() as con:
                con.execute(
                    "UPDATE replays SET status='completed', failure_code=NULL, sent_count=?, next_attempt_at=NULL, updated_at=? WHERE tenant_id=? AND replay_id=?",
                    (sent, iso(self._now()), replay["tenant_id"], replay["replay_id"]),
                )
                con.execute(
                    "UPDATE commands SET status='completed', reason_code='REPLAY_SENT', aggregate_version=aggregate_version+1, updated_at=? WHERE tenant_id=? AND command_id=?",
                    (iso(self._now()), replay["tenant_id"], replay["command_id"]),
                )
                updated = con.execute(
                    "SELECT * FROM commands WHERE tenant_id=? AND command_id=?",
                    (replay["tenant_id"], replay["command_id"]),
                ).fetchone()
                self._command_event_tx(con, updated, "REPLAY_SENT")
            count += sent - _integer(replay["sent_count"])
        return count

    def emit_opt_out(self, tenant_id: str, call_id: str, contact_ref: str) -> None:
        call = self.store.one(
            "SELECT * FROM calls WHERE call_id=? AND tenant_id=?", (call_id, tenant_id)
        )
        if call is None:
            raise NotFoundError()
        tenant = self.store.one(
            "SELECT tenant_key FROM tenants WHERE tenant_id=?", (tenant_id,)
        )["tenant_key"]
        with self.store.tx() as con:
            self._event_tx(
                con,
                tenant_id,
                tenant,
                call["command_id"],
                "contact.opt_out",
                "call",
                call_id,
                call["call_version"] + 1,
                {"call_id": call_id, "contact_ref": contact_ref, "source": "mock"},
            )

    def get_command(self, tenant_id: str, command_id: str) -> dict[str, Any]:
        row = self.store.one(
            "SELECT * FROM commands WHERE command_id=? AND tenant_id=?",
            (command_id, tenant_id),
        )
        if row is None:
            raise NotFoundError()
        return self._command_snapshot(row)

    def get_call(self, tenant_id: str, call_id: str) -> dict[str, Any]:
        call = self.store.one(
            "SELECT * FROM calls WHERE call_id=? AND tenant_id=?", (call_id, tenant_id)
        )
        if call is None:
            raise NotFoundError()
        attempts = [
            dict(row)
            for row in self.store.all(
                """SELECT a.attempt_id, a.attempt_number, a.route_policy_id,
                   a.caller_profile_id, a.target_uri, a.status, a.started_at, a.ended_at
                   FROM attempts a JOIN calls c ON c.call_id=a.call_id
                   WHERE a.call_id=? AND c.tenant_id=? ORDER BY a.attempt_number""",
                (call_id, tenant_id),
            )
        ]
        segments = [
            dict(row)
            for row in self.store.all(
                """SELECT tr.segment_id, tr.turn_index, tr.speaker, tr.text,
                   tr.revision, tr.is_final, tr.playback_state, tr.updated_at
                   FROM transcripts tr JOIN calls c ON c.call_id=tr.call_id
                   WHERE tr.call_id=? AND c.tenant_id=? ORDER BY tr.turn_index, tr.speaker""",
                (call_id, tenant_id),
            )
        ]
        recordings = [
            dict(row)
            for row in self.store.all(
                """SELECT r.recording_id, r.content_type, r.size_bytes,
                   r.checksum_algorithm, r.checksum, r.channels, r.sample_rate_hz,
                   r.duration_ms, r.status, r.oss_id, r.failure_code, r.verified_at
                   FROM recordings r JOIN calls c ON c.call_id=r.call_id
                   WHERE r.call_id=? AND c.tenant_id=? ORDER BY r.created_at""",
                (call_id, tenant_id),
            )
        ]
        event_rows = self._events_for_call(tenant_id, call_id)
        confirmed = sum(1 for row in event_rows if row["published"])
        pending = sum(
            1
            for row in event_rows
            if not row["published"] or row["broker_status"] == "failed"
        )
        applied_values = [row["saas_applied"] for row in event_rows if row["published"]]
        saas_applied: bool | None = (
            None
            if not applied_values or any(value is None for value in applied_values)
            else all(bool(value) for value in applied_values)
        )
        return {
            "call_id": call["call_id"],
            "execution_id": call["execution_id"],
            "task_id": call["task_id"],
            "task_item_id": call["task_item_id"],
            "agent_version_id": call["agent_version_id"],
            "agent_config_sha256": call["agent_config_sha256"],
            "ai": _loads(call["ai_metrics_json"], {}),
            "call_state": call["call_state"],
            "call_version": call["call_version"],
            "outcome": call["outcome"],
            "reason_code": call["reason_code"],
            "started_at": call["started_at"],
            "ended_at": call["ended_at"],
            "duration_ms": call["duration_ms"],
            "attempts": attempts,
            "transcript": {"segments": segments},
            "recordings": recordings,
            "delivery": {
                "events_total": len(event_rows),
                "broker_confirmed": confirmed,
                "broker_pending": pending,
                "saas_applied": saas_applied,
            },
            "snapshot_at": iso(self._now()),
        }

    def metrics(self) -> str:
        active = self._active_counts()[0]
        pending = self.store.one("SELECT COUNT(*) AS n FROM events WHERE published=0")[
            "n"
        ]
        dlq = self.store.one("SELECT COUNT(*) AS n FROM dead_letters")["n"]
        waiting = self.store.one(
            "SELECT COUNT(*) AS n FROM commands WHERE status='waiting'"
        )["n"]
        recording_failures = self.store.one(
            "SELECT COUNT(*) AS n FROM recordings WHERE status='failed'"
        )["n"]
        mode = self.mode.replace('"', "")
        providers = self._provider_status
        try:
            ready_value = 1 if self.ready() else 0
        except (OSError, BrokerError, sqlite3.DatabaseError, TypeError, ValueError):
            ready_value = 0
        clock_value = 1 if self._clock_jump else 0
        try:
            disk = shutil.disk_usage(self.object_root)
            disk_ratio = disk.used / disk.total if disk.total else 1.0
        except (OSError, TypeError, ZeroDivisionError):
            disk_ratio = 1.0
        lines = [
            "# HELP agent_call_active_calls Active calls including queued and ringing calls.",
            "# TYPE agent_call_active_calls gauge",
            f"agent_call_active_calls {active}",
            "# TYPE agent_call_outbox_pending gauge",
            f"agent_call_outbox_pending {pending}",
            "# TYPE agent_call_waiting_commands gauge",
            f"agent_call_waiting_commands {waiting}",
            "# TYPE agent_call_dead_letters_total counter",
            f"agent_call_dead_letters_total {dlq}",
            "# TYPE agent_call_recording_failures_total counter",
            f"agent_call_recording_failures_total {recording_failures}",
            "# TYPE agent_call_mode_info gauge",
            f'agent_call_mode_info{{mode="{mode}"}} 1',
            "# TYPE agent_call_ready gauge",
            f"agent_call_ready {ready_value}",
            "# TYPE agent_call_clock_jump gauge",
            f"agent_call_clock_jump {clock_value}",
            "# TYPE agent_call_unknown_calls gauge",
            f"agent_call_unknown_calls {self.readiness()['unknown_calls']}",
            "# TYPE agent_call_disk_usage_ratio gauge",
            f"agent_call_disk_usage_ratio {disk_ratio}",
        ]
        for component, provider_mode in sorted(providers.items()):
            safe_component = component.replace('"', "")
            safe_mode = str(provider_mode).replace('"', "")
            lines.append(
                f'agent_call_provider_mode_info{{component="{safe_component}",mode="{safe_mode}"}} 1'
            )
        for tenant in self.profile.get("tenants", []):
            try:
                depth = getattr(self.broker, "depth", lambda _name: 0)(
                    queue_name(tenant["tenant_key"])
                )
            except (BrokerError, ValidationError):
                depth = 0
            tenant_label = str(tenant["tenant_id"]).replace('"', "")
            lines.append(
                f'agent_call_queue_depth{{tenant_id="{tenant_label}"}} {depth}'
            )
        return "\n".join(lines) + "\n"

    def readiness(self) -> dict[str, Any]:
        checks: dict[str, bool] = {
            "database_read": False,
            "database_write": False,
            "broker": False,
            "routes": not self._unrouteable_tenants,
            "cells": bool(self._cell_health) and any(self._cell_health.values()),
            "providers": self.mode == "mock"
            and all(
                value in {"mock", "sqlite", "memory", "fake-cli", "rabbit"}
                for value in self._provider_status.values()
            ),
            "clock": not self._clock_jump,
            "reconciliation": False,
            "disk": False,
        }
        try:
            checks["database_read"] = self.store.one("SELECT 1") is not None
            with self.store.tx() as con:
                con.execute(
                    "INSERT INTO readiness_probe(probe_id, touched_at) VALUES(1, ?) ON CONFLICT(probe_id) DO UPDATE SET touched_at=excluded.touched_at",
                    (iso(self._now()),),
                )
                con.execute("DELETE FROM readiness_probe WHERE probe_id=1")
            checks["database_write"] = True
        except (OSError, sqlite3.DatabaseError):
            pass
        try:
            checks["broker"] = bool(self.broker.ready())
            has_route = getattr(self.broker, "has_route", None)
            if callable(has_route):
                checks["routes"] = checks["routes"] and all(
                    bool(has_route(routing_key(tenant["tenant_key"])))
                    for tenant in self.profile.get("tenants", [])
                )
            else:
                checks["routes"] = False
        except (BrokerError, ValidationError, OSError):
            checks["broker"] = False
        try:
            usage = shutil.disk_usage(self.object_root)
            ratio = usage.used / usage.total if usage.total else 1.0
            checks["disk"] = ratio < float(self._limits().get("disk_stop_pct", 0.95))
        except (OSError, TypeError, ValueError, ZeroDivisionError):
            checks["disk"] = False
        unknown = self.store.one(
            "SELECT COUNT(*) AS n FROM calls WHERE call_state='reconciling'"
        )["n"]
        checks["reconciliation"] = _integer(unknown) == 0
        return {
            "ready": all(checks.values()),
            "checks": checks,
            "unknown_calls": _integer(unknown),
        }

    def ready(self) -> bool:
        return bool(self.readiness()["ready"])

    def wait_for_idle(self, timeout: float = 5.0) -> None:
        end = time.monotonic() + timeout
        while time.monotonic() < end:
            self.process_once()
            live = [
                thread for thread in self._call_threads.values() if thread.is_alive()
            ]
            if (
                not live
                and not self.store.one(
                    "SELECT 1 FROM commands WHERE status IN ('pending','retained','broker_confirmed','accepted','waiting','executing','applying','reconciling') LIMIT 1"
                )
                and not self.store.one(
                    "SELECT 1 FROM published_commands WHERE status IN ('pending','retained','broker_confirmed') LIMIT 1"
                )
            ):
                self.dispatch_outbox()
                self.consume_saas_events()
                return
            time.sleep(0.01)
        raise TimeoutError("mock service did not become idle")