#!/usr/bin/env python3 import json import sqlite3 import sys import tempfile import unittest from pathlib import Path sys.path.insert(0, str(Path(__file__).parent)) import telemetry class TelemetryStoreTests(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory() self.db_path = Path(self.temporary.name) / "telemetry.sqlite" def tearDown(self): self.temporary.cleanup() def test_record_job_is_idempotent_and_completion_enriches_launch_row(self): telemetry.record_job( self.db_path, pipeline_run_id="run-1", slug="intent-a", job_id="job-1", role="builder", configured_profile={ "runtime": "codex", "model": "gpt-configured", "reasoning_effort": "low", }, recorded_at="2026-09-09T10:00:00Z", ) row = telemetry.record_job( self.db_path, pipeline_run_id="run-1", slug="intent-a", job_id="job-1", role="builder", actual_profile={ "runtime": "codex", "provider": "openai", "model": "gpt-actual", "reasoning_effort": "none", }, usage={ "input_tokens": 100, "cached_input_tokens": 60, "output_tokens": 20, "reasoning_tokens": 5, }, timing={"ttft_seconds": 1.5, "duration_seconds": 8.25}, iterations=7, outcome={"status": "committed", "success": True, "commit_hash": "abc123"}, session_id="session-1", ) self.assertEqual(row["configured_profile"]["model"], "gpt-configured") self.assertEqual(row["actual_profile"]["model"], "gpt-actual") self.assertEqual(row["usage"]["total_tokens"], 120) self.assertEqual(row["usage"]["reasoning_output_tokens"], 5) self.assertEqual(row["outcome"]["success"], True) connection = sqlite3.connect(self.db_path) try: self.assertEqual( connection.execute("SELECT COUNT(*) FROM job_telemetry").fetchone()[0], 1 ) self.assertEqual(connection.execute("PRAGMA journal_mode").fetchone()[0], "wal") finally: connection.close() def test_store_rejects_identity_without_storing_arbitrary_payload(self): with self.assertRaises(ValueError): telemetry.record_job( self.db_path, pipeline_run_id="", slug="intent-a", job_id="job-1", role="operator", outcome={"error": "secret error body", "status": "failed"}, ) telemetry.record_job( self.db_path, pipeline_run_id="run-1", slug="intent-a", job_id="job-1", role="operator", outcome={"error": "secret error body", "status": "failed", "error_kind": "timeout"}, ) raw = self.db_path.read_bytes() self.assertNotIn(b"secret error body", raw) def test_aggregate_and_slug_run_queries_are_json_safe(self): for role, job_id, slug, input_tokens in ( ("operator", "op-1", "parent", 10), ("builder", "build-1", "child-a", 20), ("reviewer", "review-1", "child-a", 30), ): telemetry.record_job( self.db_path, pipeline_run_id="run-split", slug=slug, job_id=job_id, role=role, usage={"input_tokens": input_tokens, "cached_input_tokens": 5, "output_tokens": 2}, timing={"duration_seconds": 1.25}, iterations=1, recorded_at="2026-09-09T10:00:00Z", ) result = telemetry.get_pipeline_run(self.db_path, "run-split") self.assertEqual(result["slugs"], ["child-a", "parent"]) self.assertEqual(result["totals"]["job_count"], 3) self.assertEqual(result["totals"]["input_tokens"], 60) self.assertEqual(result["totals"]["non_cached_input_tokens"], 45) self.assertEqual(result["totals"]["total_tokens"], 66) self.assertEqual(result["totals"]["iterations"], 3) json.dumps(result) runs = telemetry.get_slug_runs(self.db_path, "child-a") self.assertEqual([run["pipeline_run_id"] for run in runs], ["run-split"]) class TelemetryParserTests(unittest.TestCase): def test_parse_codex_cli_jsonl(self): output = "\n".join( json.dumps(event) for event in ( {"type": "thread.started", "thread_id": "thread-1"}, {"type": "item.completed", "item": {"type": "command_execution", "command": "private"}}, {"type": "item.completed", "item": {"type": "agent_message", "text": "private response"}}, { "type": "turn.completed", "usage": { "input_tokens": 100, "cached_input_tokens": 70, "output_tokens": 15, "reasoning_output_tokens": 4, }, }, ) ) parsed = telemetry.parse_codex_jsonl(output) self.assertEqual(parsed["session_id"], "thread-1") self.assertEqual(parsed["actual_profile"], {"runtime": "codex"}) self.assertEqual(parsed["usage"]["total_tokens"], 115) self.assertEqual(parsed["usage"]["cached_input_tokens"], 70) self.assertEqual(parsed["iterations"], 2) self.assertNotIn("response", json.dumps(parsed)) self.assertNotIn("private", json.dumps(parsed)) def test_parse_codex_rollout_uses_cumulative_usage_and_timestamps(self): output = "\n".join( json.dumps(event) for event in ( { "timestamp": "2026-09-09T10:00:00Z", "type": "session_meta", "payload": {"id": "session-rollout"}, }, { "timestamp": "2026-09-09T10:00:00.500Z", "type": "turn_context", "payload": {"model": "gpt-5.6-luna", "effort": "low"}, }, { "timestamp": "2026-09-09T10:00:03Z", "type": "item.completed", "item": {"type": "agent_message", "text": "ignored"}, }, { "timestamp": "2026-09-09T10:00:08Z", "type": "event_msg", "payload": { "type": "token_count", "info": { "total_token_usage": { "input_tokens": 80, "cached_input_tokens": 50, "output_tokens": 12, "reasoning_output_tokens": 3, "total_tokens": 92, } }, }, }, ) ) parsed = telemetry.parse_codex_jsonl(output) self.assertEqual(parsed["session_id"], "session-rollout") self.assertEqual(parsed["actual_profile"]["model"], "gpt-5.6-luna") self.assertEqual(parsed["actual_profile"]["reasoning_effort"], "low") self.assertEqual(parsed["timing"]["ttft_seconds"], 3.0) self.assertEqual(parsed["timing"]["model_duration_seconds"], 8.0) self.assertEqual(parsed["usage"]["total_tokens"], 92) def test_parse_claude_result_normalises_cache_and_model(self): output = json.dumps( { "type": "result", "subtype": "success", "is_error": False, "result": "private response", "session_id": "claude-session", "duration_ms": 8200, "duration_api_ms": 6100, "num_turns": 4, "total_cost_usd": 0.0123, "usage": { "input_tokens": 10, "cache_creation_input_tokens": 20, "cache_read_input_tokens": 70, "output_tokens": 15, }, "modelUsage": {"claude-sonnet-4-5": {"costUSD": 0.0123}}, } ) parsed = telemetry.parse_claude_output(output) self.assertEqual(parsed["actual_profile"]["model"], "claude-sonnet-4-5") self.assertEqual(parsed["usage"]["input_tokens"], 100) self.assertEqual(parsed["usage"]["cached_input_tokens"], 70) self.assertEqual(parsed["usage"]["cache_creation_input_tokens"], 20) self.assertEqual(parsed["usage"]["total_tokens"], 115) self.assertEqual(parsed["timing"]["duration_seconds"], 8.2) self.assertEqual(parsed["timing"]["model_duration_seconds"], 6.1) self.assertEqual(parsed["iterations"], 4) self.assertEqual(parsed["usage"]["cost_usd"], 0.0123) self.assertNotIn("private response", json.dumps(parsed)) if __name__ == "__main__": unittest.main()