"""SCALE-MOCK runner for the 100-tenant/1200-call engineering baseline. The runner is opt-in because it holds 1200 mock calls for the requested wall clock duration. It reports protocol-journal limitations instead of claiming production media capacity. """ from __future__ import annotations import argparse import copy import json import time from pathlib import Path from typing import Any from agent_call.core import ( AgentCallService, InMemoryBroker, ServiceError, iso, utcnow, ) from scripts.acceptance import load_fixture ROOT = Path(__file__).resolve().parents[1] def build_profile(path: Path, duration_minutes: int) -> dict[str, Any]: try: profile = json.loads( (ROOT / "docs/contracts/mock-profile.json").read_text(encoding="utf-8") ) except (OSError, json.JSONDecodeError) as exc: raise RuntimeError("mock profile unavailable") from exc keys = [f"tenant-{index:03d}" for index in range(100)] profile["tenants"] = [ {"tenant_id": key, "tenant_key": f"{key}.key", "enabled": True} for key in keys ] profile["tasks"] = [ { "task_id": f"task-{index:03d}", "tenant_id": key, "state": "running", "revision": 1, } for index, key in enumerate(keys) ] profile["cells"] = [ { "cell_id": f"cell-{index:03d}", "capacity": 12, "media_capacity": 12, "ai_capacity": 12, "egress_pool_id": "egress-mock", "ari_mode": "mock", } for index in range(101) ] startup_window_seconds = max(120, duration_minutes * 60) profile["limits"].update( { "global_concurrency": 1200, "global_cps": 20, "tenant_concurrency": 12, "tenant_cps": 1, "max_queue_messages": 1000, "max_queue_bytes": 16 * 1024 * 1024, "pending_window_per_tenant": 16, "pending_window_global": 64, "max_unacked_per_tenant": 4, # Keep calls alive through warm-up and the required observation window. "hold_ms": (startup_window_seconds + duration_minutes * 60 + 5) * 1000, } ) try: path.parent.mkdir(parents=True, exist_ok=True) path.write_text( json.dumps(profile, ensure_ascii=False, indent=2) + "\n", encoding="utf-8" ) except (OSError, TypeError, ValueError) as exc: raise RuntimeError("scale profile could not be written") from exc return profile def command_for( template: dict[str, Any], index: int, tenant_index: int ) -> dict[str, Any]: body = copy.deepcopy(template) tenant_id = f"tenant-{tenant_index:03d}" body["command_id"] = f"scale-cmd-{index:04d}" body["trace_id"] = f"scale-trace-{index:04d}" body["tenant_id"] = tenant_id body["tenant_key"] = f"{tenant_id}.key" body["issued_at"] = iso(utcnow()) body["payload"]["execution_id"] = f"scale-exec-{index:04d}" body["payload"]["task_id"] = f"task-{tenant_index:03d}" body["payload"]["task_item_id"] = f"scale-item-{index:04d}" return body def dry_run(duration_minutes: int) -> dict[str, Any]: return { "status": "not_run", "tenants": 100, "calls": 1200, "calls_per_tenant": 12, "required_hold_minutes": duration_minutes, "cells": 101, "safe_capacity_after_one_cell_loss": 1200, "reason": "pass --run only on a dedicated SCALE-MOCK host", } def cleanup_load(svc: AgentCallService) -> dict[str, Any]: # The failed-cell probe is restored before a full control barrier is applied; # a barrier must not claim every cell applied while one cell is unreachable. svc.set_cell_health("cell-000", True) requested = 0 for tenant_index in range(100): tenant_id = f"tenant-{tenant_index:03d}" command_id = f"scale-stop-{tenant_index:03d}" svc.control_task( tenant_id, f"task-{tenant_index:03d}", { "command_id": command_id, "action": "stop", "expected_task_revision": 1, "reason": "SCALE-MOCK teardown", "active_call_policy": "hangup", }, command_id, allow_hangup=True, ) requested += 1 svc.apply_controls() active = nonterminal = pending = 0 deadline = time.monotonic() + 30 while time.monotonic() < deadline: svc.process_once() active = svc.store.one( "SELECT COUNT(*) AS n FROM calls WHERE call_state NOT IN ('ended','failed')" )["n"] nonterminal = svc.store.one( "SELECT COUNT(*) AS n FROM commands WHERE status IN ('accepted','waiting','executing','applying')" )["n"] pending = svc.store.one("SELECT COUNT(*) AS n FROM events WHERE published=0")[ "n" ] if active == 0 and nonterminal == 0 and pending == 0: applied = svc.store.one( "SELECT COUNT(*) AS n FROM commands WHERE command_id LIKE 'scale-stop-%' AND status='applied'" )["n"] return { "stop_commands_requested": requested, "stop_commands_applied": applied, "active_calls_after_cleanup": active, "nonterminal_commands_after_cleanup": nonterminal, "outbox_pending_after_cleanup": pending, "clean": True, } time.sleep(0.1) applied = svc.store.one( "SELECT COUNT(*) AS n FROM commands WHERE command_id LIKE 'scale-stop-%' AND status='applied'" )["n"] return { "stop_commands_requested": requested, "stop_commands_applied": applied, "active_calls_after_cleanup": active, "nonterminal_commands_after_cleanup": nonterminal, "outbox_pending_after_cleanup": pending, "clean": False, } def run(directory: Path, duration_minutes: int) -> dict[str, Any]: profile_path = directory / "profile.json" build_profile(profile_path, duration_minutes) broker = InMemoryBroker(max_messages=1000, max_bytes=16 * 1024 * 1024) svc = AgentCallService( db_path=directory / "scale.sqlite3", profile_path=profile_path, object_dir=directory / "objects", broker=broker, start_background=True, ) svc.set_cell_health("cell-000", False) template = load_fixture() started = time.monotonic() try: for offset in range(12): for tenant_index in range(100): body = command_for(template, tenant_index * 12 + offset, tenant_index) while True: try: svc.publish_execute(body) break except ServiceError as exc: if exc.code != "PUBLISH_RATE_LIMIT": raise time.sleep(1.05) warmup_deadline = started + max(120, duration_minutes * 60) active = 0 threshold_at = None while time.monotonic() < warmup_deadline: active = svc.store.one( "SELECT COUNT(*) AS n FROM calls WHERE call_state='answered'" )["n"] if active >= 1000: threshold_at = time.monotonic() break time.sleep(0.25) minimum_active = active observed_hold_seconds = 0.0 if threshold_at is not None: hold_until = threshold_at + duration_minutes * 60 while active >= 1000 and time.monotonic() < hold_until: active = svc.store.one( "SELECT COUNT(*) AS n FROM calls WHERE call_state='answered'" )["n"] minimum_active = min(minimum_active, active) time.sleep(1) observed_hold_seconds = max( 0.0, min(time.monotonic(), hold_until) - threshold_at ) result = { "status": "observed" if minimum_active >= 1000 and observed_hold_seconds >= duration_minutes * 60 else "fail", "tenants": 100, "submitted_calls": 1200, "answered_calls_at_end": active, "minimum_answered_calls": minimum_active, "one_cell_disabled": True, "required_hold_minutes": duration_minutes, "threshold_reached_seconds": None if threshold_at is None else round(threshold_at - started, 3), "observed_hold_seconds": round(observed_hold_seconds, 3), "continuous_media_evidence": False, "reason": "ProtocolJournal records mock lifecycle facts; it is not a real RTP capacity proof", } cleanup = cleanup_load(svc) result["cleanup"] = cleanup if not cleanup["clean"]: result["status"] = "fail" return result finally: svc.stop() def main() -> int: parser = argparse.ArgumentParser( description="run or declare the SCALE-MOCK baseline" ) parser.add_argument( "--run", action="store_true", help="hold the mock load instead of printing a preflight", ) parser.add_argument("--duration-minutes", type=int, default=60) parser.add_argument("--directory", type=Path, default=Path("data/scale-mock")) args = parser.parse_args() if args.duration_minutes < 1: raise SystemExit("--duration-minutes must be positive") result = ( run(args.directory, args.duration_minutes) if args.run else dry_run(args.duration_minutes) ) print(json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True)) return 0 if result["status"] != "fail" else 1 if __name__ == "__main__": raise SystemExit(main())