rogee/agent-call
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
d4fa33477d5ce30132da8b291e60f12583b81bd2
agent-call/agent_call/core.py
T

2691 lines
103 KiB
Python
Raw Blame History

"""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 os
import queue
import secrets
import sqlite3
import threading
import time
import uuid
import wave
from collections.abc import Callable, Iterator
from contextlib import contextmanager
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, cast
from .mocks import MockAI, MockARI, MockRTP, MockSIPPeer, ProtocolJournal
try:
from jsonschema import Draft202012Validator
except ImportError: # pragma: no cover - the Docker image installs it
Draft202012Validator = 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 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 commands (
command_id TEXT PRIMARY KEY,
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,
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
);
CREATE INDEX IF NOT EXISTS commands_sched_idx ON commands(status, tenant_id, accepted_at);
CREATE TABLE IF NOT EXISTS executions (
execution_id TEXT PRIMARY KEY,
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,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS calls (
call_id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
execution_id TEXT NOT NULL UNIQUE,
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,
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
);
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 UNIQUE 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 TABLE IF NOT EXISTS events (
event_id TEXT PRIMARY KEY,
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
);
CREATE INDEX IF NOT EXISTS events_outbox_idx ON events(published, next_attempt_at, occurred_at);
CREATE TABLE IF NOT EXISTS replays (
replay_id TEXT PRIMARY KEY,
command_id TEXT NOT NULL UNIQUE,
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,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
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 (
event_id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
applied_at TEXT NOT NULL,
body_json TEXT NOT NULL
);
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 (
command_id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
body_json TEXT NOT NULL,
body_hash TEXT NOT NULL,
status TEXT NOT NULL,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS leases (
lease_name TEXT PRIMARY KEY,
owner_id TEXT NOT NULL,
expires_at TEXT NOT NULL
);
"""
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 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)
con.execute(
"INSERT OR IGNORE INTO schema_migrations(version, applied_at) VALUES(1, ?)",
(iso(),),
)
finally:
con.close()
@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 — sql is paired with bound params and only callers in this module can supply it.
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 — sql is paired with bound params and only callers in this module can supply it.
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) -> None:
self._queues: dict[str, queue.Queue[dict[str, Any]]] = {}
self.max_messages = max_messages
self._routes: dict[str, str] = {}
self._lock = threading.Lock()
self.outage = False
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._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)
)
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 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")
message = {
"exchange": exchange,
"routing_key": route,
"body": body,
"message_id": message_id,
"headers": headers or {},
}
try:
self._queues[target].put_nowait(message)
except queue.Full as exc:
raise BrokerError("queue is full; publication is retained") from exc
self.published.append(message)
def ack(self, message: dict[str, Any]) -> None:
return
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:
return target.get_nowait()
except queue.Empty:
return None
def ready(self) -> bool:
return not self.outage
class PikaBroker:
"""RabbitMQ adapter. Import and connection are optional in local unit tests."""
def __init__(self, url: str) -> 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
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.events.v1", exchange_type="topic", durable=True
)
self.channel.queue_declare(queue="agent-call.saas.events.v1", durable=True)
self.channel.queue_bind(
queue="agent-call.saas.events.v1",
exchange="agent-call.events.v1",
routing_key="agent-call.#",
)
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": 10000}
)
self.channel.queue_bind(
queue=name, exchange="agent-call.commands.v1", routing_key=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 {},
"_delivery_tag": method.delivery_tag,
}
except Exception as exc:
raise BrokerError(f"RabbitMQ consume failed: {type(exc).__name__}") from exc
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 ready(self) -> bool: # pragma: no cover
return bool(self.connection.is_open and self.channel.is_open)
class MockObjectStore:
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")
def path_for(self, upload_id: str) -> Path:
return self.root / f"{upload_id}.part"
def final_path_for(self, upload_id: str) -> Path:
return self.root / upload_id
def token(self, upload_id: str, tenant_id: str) -> str:
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")
path = self.path_for(upload_id)
path.write_bytes(data)
def verify_and_promote(
self, upload_id: str, expected_size: int, expected_checksum: str
) -> str:
path = self.path_for(upload_id)
if not path.exists():
raise ValidationError("OBJECT_MISSING", "object was not uploaded")
data = path.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",
)
final = self.final_path_for(upload_id)
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,
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.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_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)
else:
broker = InMemoryBroker(
_integer(
self.profile.get("limits", {}).get("max_queue_messages", 1000)
)
)
self.broker = broker
self.profile.setdefault("provider_modes", {})["rabbitmq"] = (
"rabbit" if isinstance(broker, PikaBroker) else "memory"
)
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
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 self.mode == "real" and any(value != "real" for value in providers.values()):
raise ConfigurationError(
"real mode refuses mock or test providers; no silent fallback is allowed"
)
seen: set[str] = set()
for tenant in self.profile.get("tenants", []):
tenant_key = tenant["tenant_key"]
if tenant_key in seen:
raise ConfigurationError("tenant_key must be unique")
seen.add(tenant_key)
routing_key(tenant_key)
queue_name(tenant_key)
if not self.profile.get("cells"):
raise ConfigurationError("at least one cell is required")
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", []):
self.broker.declare_tenant(tenant["tenant_key"])
self.broker.declare_events()
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)
def _run_loop(self) -> None:
while not self._stop.is_set():
try:
self.process_once()
except Exception 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:
return self.clock().astimezone(timezone.utc)
def _limits(self) -> dict[str, Any]:
return self.profile.get("limits", {})
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
):
return route
raise ValidationError(
"ROUTE_POLICY_NOT_ALLOWED", "route policy is not configured"
)
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).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_event(self, body: dict[str, Any]) -> None:
if Draft202012Validator is not None:
errors = sorted(
Draft202012Validator(self.mq_schema).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"])
task = self.store.one(
"SELECT * 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",
)
route = routing_key(tenant["tenant_key"])
body_hash = digest(body)
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()
foreign = con.execute(
"SELECT 1 FROM published_commands WHERE command_id = ?",
(body["command_id"],),
).fetchone()
if foreign and existing is None:
raise ConflictError("IDEMPOTENCY_CONFLICT", "command id is unavailable")
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": route,
}
else:
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",
iso(self._now()),
),
)
try:
self.broker.publish(
"agent-call.commands.v1", route, body, body["command_id"]
)
except Exception:
with self.store.tx() as con:
con.execute(
"UPDATE published_commands SET status = 'retained' WHERE command_id = ?",
(body["command_id"],),
)
raise
with self.store.tx() as con:
con.execute(
"UPDATE published_commands SET status = 'broker_confirmed' WHERE command_id = ?",
(body["command_id"],),
)
return {
"command_id": body["command_id"],
"status": "broker_confirmed",
"routing_key": route,
}
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:
continue
try:
message = self.broker.consume(queue_name(tenant["tenant_key"]))
except BrokerError:
break
if not message:
continue
body: Any = message.get("body", {})
body_map = body if isinstance(body, dict) else {}
try:
self.accept_execute(body)
accepted += 1
with self.store.tx() as con:
con.execute(
"UPDATE published_commands SET status = 'accepted' WHERE command_id = ?",
(body_map.get("command_id"),),
)
self.broker.ack(message)
except ServiceError as exc:
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"),
canonical(body),
exc.code,
iso(self._now()),
),
)
if body_map.get("command_id"):
con.execute(
"UPDATE published_commands SET status = 'rejected' WHERE command_id = ?",
(body_map["command_id"],),
)
self.broker.ack(message)
self._cursor = (self._cursor + max_per_tick) % len(self._tenant_order)
return accepted
def accept_execute(self, body: dict[str, Any]) -> dict[str, Any]:
self._validate_message(body)
tenant = self._tenant(body["tenant_id"], body["tenant_key"])
payload = body["payload"]
task = self.store.one(
"SELECT * FROM tasks WHERE task_id = ?", (payload["task_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()
foreign = con.execute(
"SELECT 1 FROM commands WHERE command_id = ?",
(body["command_id"],),
).fetchone()
if foreign and existing is None:
raise ConflictError("IDEMPOTENCY_CONFLICT", "command id is unavailable")
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 payload["agent_version_id"] not in {
x["agent_version_id"]
for x in self.profile.get("agents", [])
if x.get("immutable", True)
}:
status, reason = "rejected", "AGENT_VERSION_NOT_ALLOWED"
execution = con.execute(
"SELECT * FROM executions WHERE execution_id = ?",
(payload["execution_id"],),
).fetchone()
if (
execution
and execution["tenant_id"] != tenant["tenant_id"]
or 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, created_at) VALUES(?,?,?,?,?,?,?,?)",
(
payload["execution_id"],
tenant["tenant_id"],
body["command_id"],
payload["task_id"],
payload["task_item_id"],
digest(payload),
"accepted",
iso(now),
),
)
elif execution:
con.execute(
"UPDATE executions SET status = 'duplicate' WHERE execution_id = ?",
(payload["execution_id"],),
)
row = con.execute(
"SELECT * FROM commands WHERE command_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"],
"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]:
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"],
"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 _claim_lease(self) -> bool:
now = self._now()
expires = now + timedelta(seconds=2)
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) VALUES('scheduler', ?, ?) ON CONFLICT(lease_name) DO UPDATE SET owner_id=excluded.owner_id, expires_at=excluded.expires_at",
(self.instance_id, iso(expires)),
)
return True
def process_once(self) -> int:
if not self._claim_lease():
return 0
self.ingest_commands()
self.apply_controls()
scheduled = self.schedule_once()
self.dispatch_outbox()
self.consume_saas_events()
self.process_replays()
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) -> tuple[str, dict[str, Any]] | None:
for cell in self.profile.get("cells", []):
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"],),
)
if _integer(row["n"]) < _integer(cell.get("capacity", 1)):
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, command_id: str, reason: str) -> None:
with self.store.tx() as con:
row = con.execute(
"SELECT * FROM commands WHERE command_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 command_id=?",
(now, reason, version, now, command_id),
)
updated = con.execute(
"SELECT * FROM commands WHERE command_id = ?", (command_id,)
).fetchone()
self._command_event_tx(con, updated, reason)
def _reject_command(self, command_id: str, reason: str) -> None:
with self.store.tx() as con:
row = con.execute(
"SELECT * FROM commands WHERE command_id = ?", (command_id,)
).fetchone()
if row is None or row["status"] in {
"completed",
"failed",
"rejected",
"applied",
}:
return
version = row["aggregate_version"] + 1
now = iso(self._now())
con.execute(
"UPDATE commands SET status='rejected', reason_code=?, aggregate_version=?, updated_at=? WHERE command_id=?",
(reason, version, now, command_id),
)
updated = con.execute(
"SELECT * FROM commands WHERE command_id = ?", (command_id,)
).fetchone()
self._command_event_tx(con, updated, reason)
def schedule_once(self) -> int:
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
now = self._now()
if (
row["admission_deadline"]
and parse_time(row["admission_deadline"]) <= now
):
self._reject_command(row["command_id"], "ADMISSION_TIMEOUT")
continue
task = self.store.one(
"SELECT * FROM tasks WHERE task_id = ?", (row["task_id"],)
)
if task is None or task["state"] != "running":
self._reject_command(
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(row["command_id"], "CONCURRENCY")
continue
if not self._cps_available(tenant_id):
self._mark_waiting(row["command_id"], "CPS")
continue
cell = self._resource_cell()
if cell is None:
self._mark_waiting(row["command_id"], "CELL_CAPACITY")
continue
self._reserve_call(row["command_id"], row, cell)
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]]
) -> None:
cell_id, cell_cfg = cell
payload = _loads(row["payload_json"])["payload"]
route = self._route_config(payload["route_policy_id"])
prefix = str(route.get("dial_prefix", ""))
target_uri = (
payload["callee"]
if prefix and 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"],
)
call_id = new_id("call")
attempt_id = new_id("attempt")
now = iso(self._now())
with self.store.tx() as con:
current = con.execute(
"SELECT * FROM commands WHERE command_id = ?", (command_id,)
).fetchone()
if current is None or current["status"] not in {"accepted", "waiting"}:
return
version = current["aggregate_version"] + 1
con.execute(
"UPDATE commands SET status='executing', call_id=?, waiting_since=NULL, wait_reason_code=NULL, aggregate_version=?, updated_at=? WHERE command_id=?",
(call_id, version, now, command_id),
)
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, 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")
),
"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 execution_id=?",
(call_id, current["execution_id"]),
)
updated = con.execute(
"SELECT * FROM commands WHERE command_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"],
"call_state": "queued",
"reason_code": None,
},
now,
)
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()
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
self._update_call_state(call_id, "answered", started=True)
self.mock_sip.answered(call_id)
self.mock_rtp.packet(call_id, "outbound", 1)
self._sleep(_integer(self._limits().get("hold_ms", 0)))
if scenario == "ari_disconnect":
self._finish_call(call_id, "unknown", "ARI_DISCONNECTED_RECONCILING")
return
turns = _integer(self._limits().get("turns", 2))
for turn in range(turns):
if self._call_is_ended(call_id):
return
self._add_transcript(
call_id,
turn,
"customer",
f"customer turn {turn + 1}",
"not_applicable",
)
playback = (
"cancelled"
if scenario == "ai_timeout" and turn == turns - 1
else "playback_confirmed"
)
self._add_transcript(
call_id, turn, "agent", f"agent turn {turn + 1}", playback
)
self._finish_call(
call_id, "completed", "CALL_COMPLETED", upload_scenario=scenario
)
except Exception:
if not self._call_is_ended(call_id):
self._finish_call(call_id, "failed", "EXECUTOR_EXCEPTION")
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
) -> None:
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()
revision = (previous["revision"] + 1) if previous else 1
if previous and previous["revision"] >= 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, revision, 1, 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,
revision,
{
"call_id": call_id,
"segment_id": segment_id,
"turn_index": turn,
"speaker": speaker,
"text": text,
"revision": revision,
"is_final": True,
"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,
) -> 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 command_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 command_id=?",
(command_status, reason, command_version, now, row["command_id"]),
)
con.execute(
"UPDATE executions SET status=?, call_id=? WHERE execution_id=?",
(command_status, call_id, row["execution_id"]),
)
updated = con.execute(
"SELECT * FROM commands WHERE command_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,
)
self.mock_sip.bye(call_id, reason)
self.mock_ari.media_closed(call_id)
self.mock_ari.cleanup(call_id)
if outcome == "completed":
self._record_and_upload(call_id, upload_scenario or "success")
def _record_and_upload(self, call_id: str, scenario: str) -> None:
data = self._wav_bytes()
checksum = hashlib.sha256(data).hexdigest()
call = self.store.one("SELECT * FROM calls WHERE call_id = ?", (call_id,))
if call is None:
return
recording_id = new_id("rec")
upload_id = new_id("upload")
now = iso(self._now())
duration_ms = max(1, _integer(self._limits().get("turns", 2)) * 100)
path = str(self.object_store.path_for(upload_id))
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, upload_id, created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)",
(
recording_id,
call_id,
call["tenant_id"],
"audio/wav",
len(data),
"SHA-256",
checksum,
1,
16000,
duration_ms,
"uploading",
upload_id,
now,
),
)
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,
recording_id,
call["tenant_id"],
len(data),
checksum,
iso(
self._now()
+ timedelta(seconds=self._limits().get("upload_ttl_s", 300))
),
"created",
path,
now,
),
)
if scenario == "upload_missing":
self._recording_failed(recording_id, "OBJECT_MISSING")
return
token = self.object_store.token(upload_id, call["tenant_id"])
upload_data = data
if scenario == "upload_bad_checksum":
upload_data = data[:-1] + bytes([data[-1] ^ 1])
try:
self.object_store.put(upload_id, call["tenant_id"], upload_data, token)
self._complete_upload(
call["tenant_id"],
upload_id,
recording_id,
checksum=checksum,
size_bytes=len(data),
)
except ServiceError as exc:
self._recording_failed(recording_id, exc.code)
def _wav_bytes(self) -> bytes:
output = io.BytesIO()
with wave.open(output, "wb") as wav:
wav.setnchannels(1)
wav.setsampwidth(2)
wav.setframerate(16000)
wav.writeframes(
b"\0\0" * max(1600, _integer(self._limits().get("turns", 2)) * 1600)
)
return output.getvalue()
def _recording_failed(self, recording_id: str, code: str) -> None:
with self.store.tx() as con:
recording = con.execute(
"SELECT * FROM recordings WHERE recording_id = ?", (recording_id,)
).fetchone()
if recording is None or recording["status"] in {"verified", "failed"}:
return
con.execute(
"UPDATE recordings SET status='failed', failure_code=? WHERE recording_id=?",
(code, recording_id),
)
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 call_id = ?",
(recording["call_id"],),
).fetchone()
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]
) -> 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:
raise ValidationError(
"SCHEMA_INVALID", "upload request fields do not match the contract"
)
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()
if (
recording["size_bytes"] != request["size_bytes"]
or recording["checksum"] != request["checksum"]
):
raise ConflictError(
"RECORDING_METADATA_CONFLICT", "recording metadata is immutable"
)
upload = self.store.one(
"SELECT * FROM uploads WHERE recording_id = ?", (request["recording_id"],)
)
if upload:
status = 200
upload_id = upload["upload_id"]
expires = upload["expires_at"]
else:
upload_id = new_id("upload")
expires = iso(
self._now() + timedelta(seconds=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 recording_id=?",
(upload_id, request["recording_id"]),
)
status = 201
return status, {
"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"],
},
}
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 parse_time(upload["expires_at"]) <= self._now():
raise ServiceError("UPLOAD_EXPIRED", "upload session has expired", 410)
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 upload_id=?", (upload_id,)
)
def _complete_upload(
self,
tenant_id: str,
upload_id: str,
recording_id: str,
checksum: str,
size_bytes: int,
) -> 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()
if parse_time(upload["expires_at"]) <= self._now():
raise ServiceError("UPLOAD_EXPIRED", "upload session has expired", 410)
if (
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")
raise
now = iso(self._now())
with self.store.tx() as con:
con.execute(
"UPDATE uploads SET status='verified', completed_at=? WHERE upload_id=?",
(now, upload_id),
)
con.execute(
"UPDATE recordings SET status='verified', oss_id=?, verified_at=? WHERE recording_id=?",
(oss_id, now, recording_id),
)
recording = con.execute(
"SELECT * FROM recordings WHERE recording_id = ?", (recording_id,)
).fetchone()
call = con.execute(
"SELECT command_id FROM calls WHERE call_id = ?",
(recording["call_id"],),
).fetchone()
tenant = con.execute(
"SELECT tenant_key FROM tenants WHERE tenant_id = ?", (tenant_id,)
).fetchone()["tenant_key"]
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": now,
}
def complete_upload(
self, tenant_id: str, upload_id: str, request: dict[str, Any]
) -> 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"
):
raise ValidationError(
"SCHEMA_INVALID", "complete request fields do not match the contract"
)
return self._complete_upload(
tenant_id,
upload_id,
request["recording_id"],
request["checksum"],
request["size_bytes"],
)
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 event_id=?",
(attempts, retry_at, 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 event_id=?",
(iso(self._now()), 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.broker.ack(message)
continue
try:
self._validate_event(body)
with self.store.tx() as con:
exists = con.execute(
"SELECT 1 FROM inbox WHERE event_id = ?", (event_id,)
).fetchone()
if not exists:
con.execute(
"INSERT INTO inbox(event_id, tenant_id, applied_at, body_json) VALUES(?,?,?,?)",
(
event_id,
body["tenant_id"],
iso(self._now()),
canonical(body),
),
)
applied += 1
con.execute(
"UPDATE events SET saas_applied=1 WHERE event_id=?", (event_id,)
)
self.broker.ack(message)
except (KeyError, ServiceError, sqlite3.IntegrityError):
with self.store.tx() as con:
con.execute(
"INSERT INTO dead_letters(tenant_key, routing_key, body_json, reason_code, created_at) VALUES(?,?,?,?,?)",
(
body.get("tenant_key"),
message.get("routing_key"),
canonical(body),
"INBOX_INVALID",
iso(self._now()),
),
)
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
for row in rows:
body = _loads(row["payload_json"])
action = body["action"]
policy = body.get("active_call_policy", "drain")
if action == "stop" and policy == "hangup":
calls = self.store.all(
"SELECT call_id FROM calls WHERE task_id=? AND call_state NOT IN ('ended','failed')",
(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 task_id=?", (row["task_id"],)
).fetchone()
if task is None:
continue
now = iso(self._now())
con.execute(
"UPDATE commands SET status='applied', applied_task_revision=?, task_state=?, aggregate_version=aggregate_version+1, updated_at=? WHERE command_id=?",
(task["revision"], task["state"], now, row["command_id"]),
)
updated = con.execute(
"SELECT * FROM commands WHERE command_id=?", (row["command_id"],)
).fetchone()
self._command_event_tx(con, updated, "CONTROL_APPLIED")
affected = con.execute(
"SELECT * FROM commands WHERE task_id=? AND status IN ('accepted','waiting')",
(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 command_id=?",
(now, affected_row["command_id"]),
)
con.execute(
"UPDATE executions SET status='rejected' WHERE command_id=?",
(affected_row["command_id"],),
)
barrier_row = con.execute(
"SELECT * FROM commands WHERE command_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:
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]:
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 (
not isinstance(body["expected_task_revision"], 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"]:
raise ValidationError("SCHEMA_INVALID", "reason is required")
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 = digest(body)
existing = self.store.one(
"SELECT * FROM commands WHERE command_id=? AND tenant_id=?",
(body["command_id"], tenant_id),
)
foreign = self.store.one(
"SELECT 1 FROM commands WHERE command_id=?", (body["command_id"],)
)
if foreign and existing is None:
raise ConflictError("IDEMPOTENCY_CONFLICT", "command id is unavailable")
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" and 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 task_id=? AND revision=?",
(new_state, now, 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, task_id, requested_task_revision, task_state, accepted_at, trace_id, aggregate_version, updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
(
body["command_id"],
tenant_id,
self.store.one(
"SELECT tenant_key FROM tenants WHERE tenant_id=?", (tenant_id,)
)["tenant_key"],
"task.control",
"applying",
canonical(body),
payload_hash,
task_id,
body["expected_task_revision"],
new_state,
now,
body["command_id"],
1,
now,
),
)
row = con.execute(
"SELECT * FROM commands WHERE command_id=?", (body["command_id"],)
).fetchone()
self._command_event_tx(con, row, "CONTROL_ACCEPTED")
return self._control_response(
self.store.one(
"SELECT * FROM commands WHERE command_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 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 (
set(body) != {"command_id", "reason"}
or body["command_id"] != idempotency_key
):
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),
)
foreign = self.store.one(
"SELECT 1 FROM commands WHERE command_id=?", (body["command_id"],)
)
if foreign and existing is None:
raise ConflictError("IDEMPOTENCY_CONFLICT", "command id is unavailable")
if existing:
if existing["payload_hash"] != digest(body):
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.store.all(
"SELECT occurred_at FROM events WHERE tenant_id=? ORDER BY occurred_at LIMIT 1",
(tenant_id,),
)
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, 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),
digest(body),
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 command_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:
rows = self.store.all(
"SELECT * FROM replays WHERE status='accepted' ORDER BY created_at LIMIT 10"
)
count = 0
for replay in rows:
if replay["source_type"] == "call":
candidates = self.store.all(
"SELECT * FROM events WHERE tenant_id=? AND occurred_at <= ? ORDER BY occurred_at, event_id",
(replay["tenant_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 = 0
try:
for event in candidates:
body = _loads(event["body_json"])
if (
replay["source_type"] == "call"
and body.get("payload", {}).get("call_id")
!= replay["source_id"]
and body.get("aggregate_id") != replay["source_id"]
):
continue
self.broker.publish(
"agent-call.events.v1",
f"agent-call.{event['event_type']}",
body,
event["event_id"],
{"x-replay-id": replay["replay_id"]},
)
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())),
)
sent += 1
except BrokerError:
with self.store.tx() as con:
con.execute(
"UPDATE replays SET status='failed', failure_code='BROKER_UNAVAILABLE', updated_at=? WHERE replay_id=?",
(iso(self._now()), replay["replay_id"]),
)
continue
with self.store.tx() as con:
con.execute(
"UPDATE replays SET status='completed', sent_count=?, updated_at=? WHERE replay_id=?",
(sent, iso(self._now()), replay["replay_id"]),
)
con.execute(
"UPDATE commands SET status='completed', reason_code='REPLAY_SENT', aggregate_version=aggregate_version+1, updated_at=? WHERE command_id=?",
(iso(self._now()), replay["command_id"]),
)
updated = con.execute(
"SELECT * FROM commands WHERE command_id=?", (replay["command_id"],)
).fetchone()
self._command_event_tx(con, updated, "REPLAY_SENT")
count += sent
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 attempt_id, attempt_number, route_policy_id, caller_profile_id, target_uri, status, started_at, ended_at FROM attempts WHERE call_id=? ORDER BY attempt_number",
(call_id,),
)
]
segments = [
dict(row)
for row in self.store.all(
"SELECT segment_id, turn_index, speaker, text, revision, is_final, playback_state, updated_at FROM transcripts WHERE call_id=? ORDER BY turn_index, speaker",
(call_id,),
)
]
recordings = [
dict(row)
for row in self.store.all(
"SELECT recording_id, content_type, size_bytes, checksum_algorithm, checksum, channels, sample_rate_hz, duration_ms, status, oss_id, failure_code, verified_at FROM recordings WHERE call_id=? ORDER BY created_at",
(call_id,),
)
]
event_rows = self.store.all(
"SELECT published, broker_status, saas_applied FROM events WHERE tenant_id=? AND (aggregate_id=? OR payload_json LIKE ?)",
(tenant_id, call_id, f'%"call_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"],
"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"]
mode = self.mode.replace('"', "")
providers = self.profile.get("provider_modes", {})
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_mode_info gauge",
f'agent_call_mode_info{{mode="{mode}"}} 1',
]
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'
)
return "\n".join(lines) + "\n"
def ready(self) -> bool:
try:
self.store.one("SELECT 1")
return bool(self.broker.ready())
except Exception:
return False
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 ('accepted','waiting','executing','applying') LIMIT 1"
):
self.dispatch_outbox()
self.consume_saas_events()
return
time.sleep(0.01)
raise TimeoutError("mock service did not become idle")
Reference in New Issue View Git Blame Copy Permalink