232 lines
9.1 KiB
Python
232 lines
9.1 KiB
Python
#!/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()
|