from __future__ import annotations import argparse import json from dataclasses import dataclass from datetime import datetime, timedelta, timezone from pathlib import Path from statistics import median from sqlalchemy import create_engine, text def _parse_iso(value: str | None) -> datetime | None: raw = str(value or "").strip() if not raw: return None try: parsed = datetime.fromisoformat(raw) except ValueError: return None if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=timezone.utc) return parsed.astimezone(timezone.utc) def _percentile(values: list[float], q: float) -> float | None: if not values: return None if len(values) == 1: return values[0] sorted_values = sorted(values) idx = (len(sorted_values) - 1) * q lo = int(idx) hi = min(lo + 1, len(sorted_values) - 1) if lo == hi: return sorted_values[lo] frac = idx - lo return sorted_values[lo] + (sorted_values[hi] - sorted_values[lo]) * frac @dataclass class CallTimeline: call_id: str started_at: datetime | None = None ended_at: datetime | None = None ended_reconciled: bool | None = None recording_ready_at: datetime | None = None recording_ready_reconciled: bool | None = None recording_uploaded_at: datetime | None = None interaction_id: str | None = None def start_to_end_seconds(self) -> float | None: if self.started_at and self.ended_at: return max((self.ended_at - self.started_at).total_seconds(), 0.0) return None def end_to_recording_ready_seconds(self) -> float | None: if self.ended_at and self.recording_ready_at: return max((self.recording_ready_at - self.ended_at).total_seconds(), 0.0) return None def end_to_recording_upload_seconds(self) -> float | None: if self.ended_at and self.recording_uploaded_at: return max((self.recording_uploaded_at - self.ended_at).total_seconds(), 0.0) return None def status(self) -> str: if self.started_at and not self.ended_at: return "active_no_end" if self.ended_at and not self.recording_uploaded_at: return "ended_no_recording_upload" if self.ended_at and self.recording_uploaded_at: return "complete" return "unknown" def main() -> int: parser = argparse.ArgumentParser( description="Track 10: report Asterisk voice lifecycle latency." ) parser.add_argument("--database-url", required=True) parser.add_argument("--since-hours", type=int, default=6) parser.add_argument("--since-minutes", type=int, default=0) parser.add_argument("--limit-events", type=int, default=5000) parser.add_argument("--max-started-to-ended-seconds", type=float, default=30.0) parser.add_argument("--max-ended-to-recording-seconds", type=float, default=45.0) parser.add_argument("--print-limit", type=int, default=15) parser.add_argument("--json-out", default="") parser.add_argument("--breach-mode", choices=["all", "direct"], default="direct") parser.add_argument("--fail-on-breach", action="store_true") args = parser.parse_args() if args.since_minutes and args.since_minutes > 0: cutoff = datetime.now(timezone.utc) - timedelta(minutes=args.since_minutes) else: cutoff = datetime.now(timezone.utc) - timedelta(hours=max(args.since_hours, 1)) cutoff_iso = cutoff.isoformat() engine = create_engine(args.database_url, future=True) calls: dict[str, CallTimeline] = {} started_from_asterisk: set[str] = set() with engine.begin() as conn: rows = conn.execute( text( """ SELECT call_id, interaction_id, event_type, payload_json, created_at FROM voice_events WHERE event_type IN ('call.started', 'call.ended', 'recording.ready') AND created_at >= :cutoff ORDER BY id DESC LIMIT :limit """ ), {"cutoff": cutoff_iso, "limit": max(args.limit_events, 100)}, ).mappings().all() for row in rows: call_id = str(row["call_id"] or "").strip() if not call_id: continue payload_raw = str(row["payload_json"] or "{}") try: payload = json.loads(payload_raw) except json.JSONDecodeError: payload = {} event_type = str(row["event_type"] or "") created_at = _parse_iso(str(row["created_at"] or "")) if created_at is None: continue timeline = calls.get(call_id) or CallTimeline(call_id=call_id) if event_type == "call.started": if payload.get("source") != "asterisk": continue started_from_asterisk.add(call_id) timeline.started_at = max(filter(None, [timeline.started_at, created_at])) elif event_type == "call.ended": if timeline.ended_at is None or created_at > timeline.ended_at: timeline.ended_at = created_at timeline.ended_reconciled = bool(payload.get("reconciled")) elif event_type == "recording.ready": if timeline.recording_ready_at is None or created_at > timeline.recording_ready_at: timeline.recording_ready_at = created_at timeline.recording_ready_reconciled = bool(payload.get("reconciled")) timeline.interaction_id = timeline.interaction_id or str(row["interaction_id"] or "").strip() or None calls[call_id] = timeline rec_rows = conn.execute( text( """ SELECT call_id, interaction_id, created_at FROM call_recordings WHERE created_at >= :cutoff ORDER BY id DESC LIMIT :limit """ ), {"cutoff": cutoff_iso, "limit": max(args.limit_events, 100)}, ).mappings().all() for row in rec_rows: call_id = str(row["call_id"] or "").strip() if call_id not in calls: continue created_at = _parse_iso(str(row["created_at"] or "")) if created_at is None: continue timeline = calls[call_id] if timeline.recording_uploaded_at is None or created_at > timeline.recording_uploaded_at: timeline.recording_uploaded_at = created_at timeline.interaction_id = timeline.interaction_id or str(row["interaction_id"] or "").strip() or None items = [calls[call_id] for call_id in started_from_asterisk if call_id in calls] items.sort(key=lambda x: x.started_at or datetime.min.replace(tzinfo=timezone.utc), reverse=True) start_end = [v for v in (item.start_to_end_seconds() for item in items) if v is not None] end_rec = [v for v in (item.end_to_recording_upload_seconds() for item in items) if v is not None] direct_items = [item for item in items if item.ended_at is not None and item.ended_reconciled is False] direct_start_end = [v for v in (item.start_to_end_seconds() for item in direct_items) if v is not None] direct_end_rec = [ v for v in (item.end_to_recording_upload_seconds() for item in direct_items) if v is not None ] ended_no_recording = [item for item in items if item.status() == "ended_no_recording_upload"] active_no_end = [item for item in items if item.status() == "active_no_end"] p95_start_end = _percentile(start_end, 0.95) p95_end_rec = _percentile(end_rec, 0.95) p95_direct_start_end = _percentile(direct_start_end, 0.95) p95_direct_end_rec = _percentile(direct_end_rec, 0.95) breach = False if args.breach_mode == "direct": left = p95_direct_start_end if p95_direct_start_end is not None else p95_start_end right = p95_direct_end_rec if p95_direct_end_rec is not None else p95_end_rec else: left = p95_start_end right = p95_end_rec if left is not None and left > args.max_started_to_ended_seconds: breach = True if right is not None and right > args.max_ended_to_recording_seconds: breach = True if ended_no_recording: breach = True report = { "generated_at": datetime.now(timezone.utc).isoformat(), "since_hours": args.since_hours, "since_minutes": args.since_minutes, "thresholds": { "max_started_to_ended_seconds": args.max_started_to_ended_seconds, "max_ended_to_recording_seconds": args.max_ended_to_recording_seconds, }, "summary": { "total_calls": len(items), "complete_calls": sum(1 for item in items if item.status() == "complete"), "active_no_end": len(active_no_end), "ended_no_recording_upload": len(ended_no_recording), "ended_reconciled_calls": sum(1 for item in items if item.ended_reconciled is True), "ended_direct_calls": sum(1 for item in items if item.ended_reconciled is False), "start_to_end": { "count": len(start_end), "median": median(start_end) if start_end else None, "p95": p95_start_end, }, "end_to_recording_upload": { "count": len(end_rec), "median": median(end_rec) if end_rec else None, "p95": p95_end_rec, }, "direct_start_to_end": { "count": len(direct_start_end), "median": median(direct_start_end) if direct_start_end else None, "p95": p95_direct_start_end, }, "direct_end_to_recording_upload": { "count": len(direct_end_rec), "median": median(direct_end_rec) if direct_end_rec else None, "p95": p95_direct_end_rec, }, "breach_mode": args.breach_mode, }, "worst_calls": [ { "call_id": item.call_id, "interaction_id": item.interaction_id, "status": item.status(), "started_at": item.started_at.isoformat() if item.started_at else None, "ended_at": item.ended_at.isoformat() if item.ended_at else None, "ended_reconciled": item.ended_reconciled, "recording_ready_at": item.recording_ready_at.isoformat() if item.recording_ready_at else None, "recording_ready_reconciled": item.recording_ready_reconciled, "recording_uploaded_at": item.recording_uploaded_at.isoformat() if item.recording_uploaded_at else None, "start_to_end_seconds": item.start_to_end_seconds(), "end_to_recording_upload_seconds": item.end_to_recording_upload_seconds(), } for item in items[: max(args.print_limit, 1)] ], } print( "[INFO] Track10 voice latency report: " f"calls={report['summary']['total_calls']} " f"complete={report['summary']['complete_calls']} " f"active_no_end={report['summary']['active_no_end']} " f"ended_no_recording_upload={report['summary']['ended_no_recording_upload']}" ) print( "[INFO] p95 started->ended=" f"{report['summary']['start_to_end']['p95']}s; " "p95 ended->recording_upload=" f"{report['summary']['end_to_recording_upload']['p95']}s" ) print( "[INFO] direct p95 started->ended=" f"{report['summary']['direct_start_to_end']['p95']}s; " "direct p95 ended->recording_upload=" f"{report['summary']['direct_end_to_recording_upload']['p95']}s" ) print(json.dumps(report, ensure_ascii=False, indent=2)) if args.json_out.strip(): out_path = Path(args.json_out).expanduser().resolve() out_path.parent.mkdir(parents=True, exist_ok=True) out_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") print(f"[INFO] report written: {out_path}") if args.fail_on_breach and breach: print("[FAIL] SLO breach detected for Track 10 criteria") return 1 return 0 if __name__ == "__main__": raise SystemExit(main())