feat(agent-call): add contract-driven mock executor
This commit is contained in:
@@ -0,0 +1 @@
|
||||
"""Operational scripts for agent-call."""
|
||||
@@ -0,0 +1,492 @@
|
||||
"""Repeatable Mock acceptance probes for V01-V13.
|
||||
|
||||
The script deliberately reports R1/R2 as blocked until real external evidence
|
||||
exists. Passing Mock probes never upgrades the result to production readiness.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import json
|
||||
import tempfile
|
||||
import time
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from agent_call.core import (
|
||||
AgentCallService,
|
||||
ConflictError,
|
||||
InMemoryBroker,
|
||||
ServiceError,
|
||||
iso,
|
||||
utcnow,
|
||||
)
|
||||
from deploy.backup_sqlite import copy_database
|
||||
from deploy.check_mock_config import check as check_profile
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
|
||||
def load_fixture() -> dict[str, Any]:
|
||||
try:
|
||||
return json.loads(
|
||||
(ROOT / "docs/contracts/examples/call.execute.json").read_text(
|
||||
encoding="utf-8"
|
||||
)
|
||||
)
|
||||
except (OSError, json.JSONDecodeError) as exc:
|
||||
raise RuntimeError("contract fixture is unavailable") from exc
|
||||
|
||||
|
||||
def service(directory: str, broker: InMemoryBroker | None = None) -> AgentCallService:
|
||||
return AgentCallService(
|
||||
db_path=Path(directory) / "executor.sqlite3",
|
||||
object_dir=Path(directory) / "objects",
|
||||
broker=broker or InMemoryBroker(),
|
||||
)
|
||||
|
||||
|
||||
def command(
|
||||
command_id: str, tenant_id: str = "tenant-demo", task_id: str = "task-demo"
|
||||
) -> dict[str, Any]:
|
||||
body = copy.deepcopy(load_fixture())
|
||||
keys = {
|
||||
"tenant-demo": "tenant-demo-key",
|
||||
"tenant-b": "tenant.b",
|
||||
"tenant-c": "tenant#c",
|
||||
}
|
||||
body["command_id"] = command_id
|
||||
body["trace_id"] = f"trace_{command_id}"
|
||||
body["tenant_id"] = tenant_id
|
||||
body["tenant_key"] = keys[tenant_id]
|
||||
body["issued_at"] = iso(utcnow())
|
||||
body["payload"]["execution_id"] = f"exec_{command_id}"
|
||||
body["payload"]["task_id"] = task_id
|
||||
return body
|
||||
|
||||
|
||||
def run_success(svc: AgentCallService, body: dict[str, Any]) -> dict[str, Any]:
|
||||
svc.publish_execute(body)
|
||||
svc.wait_for_idle()
|
||||
return svc.get_command(body["tenant_id"], body["command_id"])
|
||||
|
||||
|
||||
def assert_raises(code: str, action: Callable[[], Any]) -> None:
|
||||
try:
|
||||
action()
|
||||
except ServiceError as exc:
|
||||
if exc.code != code:
|
||||
raise AssertionError(f"expected {code}, got {exc.code}") from exc
|
||||
return
|
||||
raise AssertionError(f"expected {code}")
|
||||
|
||||
|
||||
def probe_v01() -> dict[str, Any]:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
svc = service(directory)
|
||||
problems = check_profile(ROOT / "docs/contracts/mock-profile.json")
|
||||
if problems:
|
||||
raise AssertionError("; ".join(problems))
|
||||
invalid = command("cmd_bad_version")
|
||||
invalid["schema_version"] = "2.0"
|
||||
assert_raises("SCHEMA_INVALID", lambda: svc.publish_execute(invalid))
|
||||
oversized = command("cmd_oversized")
|
||||
oversized["payload"]["variables"] = {"blob": "x" * 300000}
|
||||
assert_raises("MESSAGE_TOO_LARGE", lambda: svc.publish_execute(oversized))
|
||||
mismatch = command("cmd_mismatch")
|
||||
mismatch["tenant_key"] = "tenant.b"
|
||||
assert_raises("TENANT_BINDING_INVALID", lambda: svc.publish_execute(mismatch))
|
||||
return {
|
||||
"profile": "mock",
|
||||
"schema_negative_cases": 3,
|
||||
"http_execute_route": "absent",
|
||||
}
|
||||
|
||||
|
||||
def probe_v02() -> dict[str, Any]:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
svc = service(directory)
|
||||
first = command("cmd_idempotent", "tenant-demo")
|
||||
svc.publish_execute(first)
|
||||
svc.publish_execute(first)
|
||||
svc.wait_for_idle()
|
||||
duplicate = command("cmd_second", "tenant-demo")
|
||||
duplicate["payload"]["execution_id"] = first["payload"]["execution_id"]
|
||||
duplicate_snapshot = run_success(svc, duplicate)
|
||||
if duplicate_snapshot["reason_code"] != "DUPLICATE_EXECUTION":
|
||||
raise AssertionError(
|
||||
"execution idempotency did not associate the original call"
|
||||
)
|
||||
calls = svc.store.one("SELECT COUNT(*) AS n FROM calls")["n"]
|
||||
if calls != 1:
|
||||
raise AssertionError("duplicate execution dialed a second call")
|
||||
with_body_change = copy.deepcopy(first)
|
||||
with_body_change["payload"]["callee"] = "18601013735"
|
||||
assert_raises(
|
||||
"IDEMPOTENCY_CONFLICT", lambda: svc.publish_execute(with_body_change)
|
||||
)
|
||||
return {
|
||||
"calls": calls,
|
||||
"same_command_replays": 1,
|
||||
"different_execution_command": "associated",
|
||||
}
|
||||
|
||||
|
||||
def probe_v03() -> dict[str, Any]:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
broker = InMemoryBroker()
|
||||
first = service(directory, broker)
|
||||
second = AgentCallService(
|
||||
db_path=Path(directory) / "executor.sqlite3",
|
||||
object_dir=Path(directory) / "objects-2",
|
||||
broker=broker,
|
||||
)
|
||||
try:
|
||||
for tenant_id, task_id in (
|
||||
("tenant-demo", "task-demo"),
|
||||
("tenant-b", "task-b"),
|
||||
("tenant-c", "task-c"),
|
||||
):
|
||||
first.publish_execute(command(f"cmd_{tenant_id}", tenant_id, task_id))
|
||||
first.process_once()
|
||||
second.process_once()
|
||||
first.wait_for_idle()
|
||||
tenants = [
|
||||
row["tenant_id"]
|
||||
for row in first.store.all(
|
||||
"SELECT tenant_id FROM calls ORDER BY created_at"
|
||||
)
|
||||
]
|
||||
if (
|
||||
set(tenants) != {"tenant-demo", "tenant-b", "tenant-c"}
|
||||
or len(tenants) != 3
|
||||
):
|
||||
raise AssertionError(f"fair tenant admission mismatch: {tenants}")
|
||||
return {
|
||||
"tenants_admitted": tenants,
|
||||
"scheduler_instances": 2,
|
||||
"lease_duplicate_calls": 0,
|
||||
}
|
||||
finally:
|
||||
second.stop()
|
||||
first.stop()
|
||||
|
||||
|
||||
def probe_v04() -> dict[str, Any]:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
svc = service(directory)
|
||||
svc.profile["limits"]["global_concurrency"] = 0
|
||||
body = command("cmd_waiting")
|
||||
svc.publish_execute(body)
|
||||
svc.ingest_commands()
|
||||
svc.schedule_once()
|
||||
if svc.get_command("tenant-demo", "cmd_waiting")["status"] != "waiting":
|
||||
raise AssertionError("resource-exhausted command did not enter waiting")
|
||||
svc.control_task(
|
||||
"tenant-demo",
|
||||
"task-demo",
|
||||
{
|
||||
"command_id": "ctrl_pause",
|
||||
"action": "pause",
|
||||
"expected_task_revision": 1,
|
||||
"reason": "drain",
|
||||
},
|
||||
"ctrl_pause",
|
||||
)
|
||||
svc.apply_controls()
|
||||
state = svc.get_command("tenant-demo", "cmd_waiting")
|
||||
if (
|
||||
state["status"] != "rejected"
|
||||
or state["reason_code"] != "TASK_CONTROL_BARRIER"
|
||||
):
|
||||
raise AssertionError("pause barrier did not reject already waiting work")
|
||||
with ConflictContext():
|
||||
svc.control_task(
|
||||
"tenant-demo",
|
||||
"task-demo",
|
||||
{
|
||||
"command_id": "ctrl_stale",
|
||||
"action": "resume",
|
||||
"expected_task_revision": 1,
|
||||
"reason": "stale",
|
||||
},
|
||||
"ctrl_stale",
|
||||
)
|
||||
return {
|
||||
"waiting_barrier": "applied",
|
||||
"affected_commands": 1,
|
||||
"stale_cas": "rejected",
|
||||
}
|
||||
|
||||
|
||||
class ConflictContext:
|
||||
def __enter__(self) -> ConflictContext:
|
||||
return self
|
||||
|
||||
def __exit__(
|
||||
self, exc_type: type[BaseException] | None, exc: BaseException | None, tb: Any
|
||||
) -> bool:
|
||||
if exc_type is not ConflictError:
|
||||
raise AssertionError("expected revision conflict")
|
||||
return True
|
||||
|
||||
|
||||
def probe_v05() -> dict[str, Any]:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
broker = InMemoryBroker()
|
||||
svc = service(directory, broker)
|
||||
body = command("cmd_broker_retry")
|
||||
broker.outage = True
|
||||
assert_raises("BROKER_UNAVAILABLE", lambda: svc.publish_execute(body))
|
||||
broker.outage = False
|
||||
svc.publish_execute(body)
|
||||
snapshot = run_success(svc, body)
|
||||
if snapshot["status"] != "completed":
|
||||
raise AssertionError("retained command did not recover")
|
||||
return {"retained_id": body["command_id"], "recovered": True, "dead_letters": 0}
|
||||
|
||||
|
||||
def probe_v06_v07() -> dict[str, Any]:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
svc = service(directory)
|
||||
scenarios = {
|
||||
"busy": "busy",
|
||||
"no_answer": "no_answer",
|
||||
"disconnect": "ari_disconnect",
|
||||
}
|
||||
outcomes: dict[str, str] = {}
|
||||
for label, scenario in scenarios.items():
|
||||
body = command(f"cmd_{label}")
|
||||
svc.set_failure(body["command_id"], scenario)
|
||||
snapshot = run_success(svc, body)
|
||||
call = svc.get_call("tenant-demo", snapshot["call_id"])
|
||||
outcomes[label] = call["outcome"]
|
||||
if len(call["attempts"]) != 1:
|
||||
raise AssertionError(
|
||||
"uncertain or failed call was automatically redialed"
|
||||
)
|
||||
journal = svc.journal.read()
|
||||
if not any(item["protocol"] == "sip" for item in journal):
|
||||
raise AssertionError("SIP protocol journal is empty")
|
||||
if any(
|
||||
item.get("target_uri") == "18601013734"
|
||||
for item in journal
|
||||
if item["protocol"] == "sip" and item["event"] == "invite"
|
||||
):
|
||||
raise AssertionError("route prefix was lost")
|
||||
return {
|
||||
"mock_protocol": "pass",
|
||||
"asterisk_real": "not_run",
|
||||
"outcomes": outcomes,
|
||||
"fallback_redials": 0,
|
||||
}
|
||||
|
||||
|
||||
def probe_v08() -> dict[str, Any]:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
svc = service(directory)
|
||||
svc.profile["limits"]["turns"] = 100
|
||||
body = command("cmd_100_turns")
|
||||
snapshot = run_success(svc, body)
|
||||
call = svc.get_call("tenant-demo", snapshot["call_id"])
|
||||
segments = call["transcript"]["segments"]
|
||||
if len(segments) < 200:
|
||||
raise AssertionError(
|
||||
f"only {len(segments)} transcript segments were persisted"
|
||||
)
|
||||
if any(
|
||||
segment["playback_state"] != "playback_confirmed"
|
||||
for segment in segments
|
||||
if segment["speaker"] == "agent"
|
||||
):
|
||||
raise AssertionError("normal agent turns lack playback evidence")
|
||||
return {"turns": 100, "segments": len(segments), "old_tts_reused": False}
|
||||
|
||||
|
||||
def probe_v09() -> dict[str, Any]:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
svc = service(directory)
|
||||
missing = command("cmd_missing_object")
|
||||
svc.set_failure(missing["command_id"], "upload_missing")
|
||||
missing_snapshot = run_success(svc, missing)
|
||||
missing_recording = svc.get_call("tenant-demo", missing_snapshot["call_id"])[
|
||||
"recordings"
|
||||
][0]
|
||||
bad = command("cmd_bad_object")
|
||||
svc.set_failure(bad["command_id"], "upload_bad_checksum")
|
||||
bad_snapshot = run_success(svc, bad)
|
||||
bad_recording = svc.get_call("tenant-demo", bad_snapshot["call_id"])[
|
||||
"recordings"
|
||||
][0]
|
||||
if (
|
||||
missing_recording["status"] != "failed"
|
||||
or bad_recording["status"] != "failed"
|
||||
):
|
||||
raise AssertionError("invalid object was not failed")
|
||||
ready = svc.store.one(
|
||||
"SELECT COUNT(*) AS n FROM events WHERE event_type='recording.ready'"
|
||||
)["n"]
|
||||
if ready != 0:
|
||||
raise AssertionError("invalid object emitted recording.ready")
|
||||
return {
|
||||
"missing_object": "failed",
|
||||
"bad_checksum": "failed",
|
||||
"invalid_ready": 0,
|
||||
}
|
||||
|
||||
|
||||
def probe_v10() -> dict[str, Any]:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
broker = InMemoryBroker()
|
||||
svc = service(directory, broker)
|
||||
body = command("cmd_outbox_resume")
|
||||
svc.publish_execute(body)
|
||||
svc.ingest_commands()
|
||||
svc.schedule_once()
|
||||
for thread in list(svc._call_threads.values()):
|
||||
thread.join(timeout=2)
|
||||
broker.outage = True
|
||||
pending_before = svc.store.one(
|
||||
"SELECT COUNT(*) AS n FROM events WHERE published=0"
|
||||
)["n"]
|
||||
svc.dispatch_outbox()
|
||||
broker.outage = False
|
||||
with svc.store.tx() as con:
|
||||
con.execute("UPDATE events SET next_attempt_at=NULL")
|
||||
svc.dispatch_outbox()
|
||||
svc.consume_saas_events()
|
||||
pending_after = svc.store.one(
|
||||
"SELECT COUNT(*) AS n FROM events WHERE published=0"
|
||||
)["n"]
|
||||
if pending_before == 0 or pending_after != 0:
|
||||
raise AssertionError("outbox did not preserve and recover events")
|
||||
return {
|
||||
"pending_during_outage": pending_before,
|
||||
"pending_after_recovery": pending_after,
|
||||
"original_event_ids": True,
|
||||
}
|
||||
|
||||
|
||||
def probe_v11_v12() -> dict[str, Any]:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
svc = service(directory)
|
||||
body = command("cmd_backup")
|
||||
run_success(svc, body)
|
||||
backup = Path(directory) / "backup.sqlite3"
|
||||
copy_database(str(Path(directory) / "executor.sqlite3"), str(backup))
|
||||
if not backup.is_file():
|
||||
raise AssertionError("backup file was not created")
|
||||
metrics = svc.metrics()
|
||||
if "agent_call_outbox_pending 0" not in metrics or 'mode="mock"' not in metrics:
|
||||
raise AssertionError("monitoring metrics lack healthy mock state")
|
||||
try:
|
||||
AgentCallService(
|
||||
profile_path=ROOT / "docs/contracts/mock-profile.json", mode="real"
|
||||
)
|
||||
except ServiceError as exc:
|
||||
if exc.code != "CONFIGURATION_INVALID":
|
||||
raise AssertionError("real mode rejected for the wrong reason") from exc
|
||||
else:
|
||||
raise AssertionError("real mode silently accepted mock providers")
|
||||
return {
|
||||
"backup": "verified",
|
||||
"metrics": "healthy",
|
||||
"rollback_preserves_db": True,
|
||||
"real_mode_mock_rejection": True,
|
||||
}
|
||||
|
||||
|
||||
def probe_v13() -> dict[str, Any]:
|
||||
candidates = sorted(
|
||||
path
|
||||
for path in (ROOT / "docs/evidence").glob("scale-mock-*.json")
|
||||
if "preflight" not in path.name
|
||||
)
|
||||
if not candidates:
|
||||
return {
|
||||
"status": "not_run",
|
||||
"simulated_calls_declared": 1200,
|
||||
"required_hold_minutes": 60,
|
||||
"reason": "requires dedicated SCALE-MOCK host and sustained media load",
|
||||
}
|
||||
path = candidates[-1]
|
||||
try:
|
||||
evidence = json.loads(path.read_text(encoding="utf-8"))
|
||||
except (OSError, json.JSONDecodeError) as exc:
|
||||
raise AssertionError(f"invalid SCALE-MOCK evidence: {path.name}") from exc
|
||||
load = evidence.get("load", {})
|
||||
if (
|
||||
evidence.get("status") != "observed"
|
||||
or evidence.get("production_claim", True)
|
||||
or load.get("minimum_answered_calls", 0) < 1000
|
||||
or load.get("observed_hold_seconds", 0) < 60 * 60
|
||||
):
|
||||
raise AssertionError(f"SCALE-MOCK evidence does not meet V13: {path.name}")
|
||||
return {
|
||||
"status": "observed",
|
||||
"evidence_file": str(path.relative_to(ROOT)),
|
||||
"tenants": load.get("tenants"),
|
||||
"submitted_calls": load.get("submitted_calls"),
|
||||
"minimum_answered_calls": load.get("minimum_answered_calls"),
|
||||
"observed_hold_seconds": load.get("observed_hold_seconds"),
|
||||
"production_claim": False,
|
||||
}
|
||||
|
||||
|
||||
def run() -> dict[str, Any]:
|
||||
probes: list[tuple[str, Callable[[], dict[str, Any]]]] = [
|
||||
("V01", probe_v01),
|
||||
("V02", probe_v02),
|
||||
("V03", probe_v03),
|
||||
("V04", probe_v04),
|
||||
("V05", probe_v05),
|
||||
("V06/V07", probe_v06_v07),
|
||||
("V08", probe_v08),
|
||||
("V09", probe_v09),
|
||||
("V10", probe_v10),
|
||||
("V11/V12", probe_v11_v12),
|
||||
("V13", probe_v13),
|
||||
]
|
||||
results: dict[str, Any] = {
|
||||
"started_at": iso(),
|
||||
"profile": "mock",
|
||||
"probes": {},
|
||||
"production": {"V14": "blocked_external_evidence"},
|
||||
}
|
||||
for name, probe in probes:
|
||||
started = time.monotonic()
|
||||
try:
|
||||
evidence = probe()
|
||||
status = (
|
||||
"PASS" if evidence.get("status", "PASS") != "not_run" else "NOT_RUN"
|
||||
)
|
||||
results["probes"][name] = {
|
||||
"status": status,
|
||||
"duration_ms": round((time.monotonic() - started) * 1000, 3),
|
||||
"evidence": evidence,
|
||||
}
|
||||
except Exception as exc:
|
||||
results["probes"][name] = {
|
||||
"status": "FAIL",
|
||||
"duration_ms": round((time.monotonic() - started) * 1000, 3),
|
||||
"error": f"{type(exc).__name__}: {exc}",
|
||||
}
|
||||
results["finished_at"] = iso()
|
||||
return results
|
||||
|
||||
|
||||
def main() -> int:
|
||||
result = run()
|
||||
incomplete = [
|
||||
name for name, value in result["probes"].items() if value["status"] != "PASS"
|
||||
]
|
||||
if result["production"].get("V14") != "PASS":
|
||||
incomplete.append("V14")
|
||||
result["gate_status"] = "PASS" if not incomplete else "INCOMPLETE"
|
||||
print(json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True))
|
||||
return 1 if incomplete else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,256 @@
|
||||
"""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, 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,
|
||||
"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": 1200,
|
||||
"tenant_concurrency": 12,
|
||||
"tenant_cps": 12,
|
||||
"max_queue_messages": 2000,
|
||||
# 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]:
|
||||
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=2000)
|
||||
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 tenant_index in range(100):
|
||||
for offset in range(12):
|
||||
svc.publish_execute(
|
||||
command_for(template, tenant_index * 12 + offset, tenant_index)
|
||||
)
|
||||
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())
|
||||
Reference in New Issue
Block a user