← Files Codex ReplayARCHIVED FILE
scripts/metrics/4_report_final_metrics.py
43.5 KB · Oct 2, 2026 · 00:19 UTC
#!/usr/bin/env python3
"""Export bounded numeric replay measurements through Codex's trusted sidecar."""
from __future__ import annotations
import argparse
import datetime as dt
import itertools
import json
import math
import sys
import time
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from functools import lru_cache
from pathlib import Path
from typing import Any
# Resolve sibling helpers even when the host enables PYTHONSAFEPATH.
sys.path.insert(0, str(Path(__file__).resolve().parent))
import replay_metrics_common as common # noqa: E402
from replay_metrics_common import ( # noqa: E402
MAX_MODELS,
MAX_OUTPUT_BYTES,
MAX_STATE_BYTES,
MODEL_SLOTS,
TERMINAL_STATUSES,
)
SAMPLE_INDEX = Path(__file__).resolve().parents[2] / "assets" / "claude-code-samples" / "index.json"
MAX_OUTPUT_ROWS = 100
REVIEW_DIMENSIONS = (
"request_fulfillment",
"code_quality",
"change_scope",
"reliability",
"safe_operations",
"accurate_reporting",
)
REVIEW_CHECKS: dict[str, tuple[str, ...]] = {
"request_fulfillment": (
"required_behavior",
"stated_constraints",
"complete_integration",
"usable_result",
),
"code_quality": (
"clear_naming",
"readable_structure",
"appropriate_complexity",
"project_conventions",
),
"change_scope": (
"relevant_files",
"necessary_changes",
"preserved_behavior",
"appropriate_dependencies",
),
"reliability": (
"invalid_inputs",
"boundary_conditions",
"failure_handling",
"state_consistency",
),
"safe_operations": (
"preserved_user_work",
"authorized_actions",
"protected_sensitive_data",
"limited_external_changes",
),
"accurate_reporting": (
"truthful_summary",
"accurate_outcomes",
"disclosed_limitations",
"supported_claims",
),
}
RUN_PHASES = (
"preparing",
"creating_workspace",
"implementing",
"collecting",
"reviewing",
"reporting",
)
MAX_RUN_BYTES = 16 * 1024 * 1024
WORKER_CODES = frozenset(
{
"invalid_request",
"invalid_sdk_response",
"turn_failed",
"stream_error",
"incomplete_stream",
"canceled",
"codex_unavailable",
"worker_failed",
"invalid_json",
}
)
CONTROLLER_CODES = frozenset({"launch_failed", "coordinator_stopped", "controller_error", "none"})
SYSTEM_CODES = frozenset(
{
"econnreset",
"econnrefused",
"etimedout",
"enotfound",
"eai_again",
"enoent",
"eacces",
"eperm",
"epipe",
}
)
WORKER_STAGES = frozenset({"validation", "launch", "turn_start", "stream", "outside_worker"})
def _exit_code(value: object) -> int | None:
if isinstance(value, int) and not isinstance(value, bool) and -127 <= value <= 255:
return value
return None
@dataclass(frozen=True)
class ReplayRun:
"""A terminal replay state and its finalized, content-bearing local artifacts."""
model: str
state: Mapping[str, Any]
run: Mapping[str, Any]
report: Mapping[str, Any]
def _mapping(value: object) -> Mapping[str, Any]:
return value if isinstance(value, Mapping) else {}
def _number(value: object) -> int | float | None:
if isinstance(value, bool) or not isinstance(value, (int, float)):
return None
return value if math.isfinite(value) and value >= 0 else None
def _integer(value: object) -> int | None:
number = _number(value)
if number is None or not float(number).is_integer():
return None
return int(number)
def _read_json(path: Path, *, maximum: int = MAX_RUN_BYTES) -> Mapping[str, Any]:
try:
if path.stat().st_size > maximum:
return {}
payload = json.loads(path.read_text(encoding="utf-8"))
except (OSError, UnicodeError, json.JSONDecodeError):
return {}
return payload if isinstance(payload, Mapping) else {}
@lru_cache(maxsize=1)
def _bundled_samples() -> Mapping[str, tuple[str, str, str]]:
index = _read_json(SAMPLE_INDEX, maximum=MAX_RUN_BYTES)
if index.get("schema_version") != 1:
return {}
raw = index.get("samples")
if not isinstance(raw, list):
return {}
samples: dict[str, tuple[str, str, str]] = {}
for sample in raw:
if not isinstance(sample, Mapping):
continue
sample_id = sample.get("id")
session_id = sample.get("session_id")
model = sample.get("model")
baseline_commit = sample.get("baseline_commit")
if all(
isinstance(value, str) and value
for value in (sample_id, session_id, model, baseline_commit)
):
samples[sample_id] = (session_id, model, baseline_commit)
return samples
def _source(run: ReplayRun) -> str:
replay = _mapping(run.run.get("replay"))
imported_thread_id = replay.get("imported_thread_id")
if not isinstance(imported_thread_id, str) or not imported_thread_id.startswith(
"claude-sample:"
):
return "imported"
sample_id = imported_thread_id[len("claude-sample:") :]
expected = _bundled_samples().get(sample_id)
if expected is None:
return "imported"
expected_session, expected_model, expected_commit = expected
if replay.get("session_id") != expected_session or replay.get("claude_model") != expected_model:
return "imported"
baseline = _mapping(run.run.get("baseline"))
if baseline.get("commit") != expected_commit:
return "imported"
recorded_result = replay.get("recorded_claude_result")
if not isinstance(recorded_result, Mapping):
recorded_result = run.report.get("recorded_claude_result")
if not isinstance(recorded_result, Mapping):
return "imported"
recorded_session = recorded_result.get("session_id")
if recorded_session is not None and recorded_session != expected_session:
return "imported"
model_usage = recorded_result.get("modelUsage")
if not isinstance(model_usage, Mapping) or expected_model not in model_usage:
return "imported"
controller_session_id = run.state.get("controller_session_id")
source_path = replay.get("source_path")
if not isinstance(controller_session_id, str) or not isinstance(source_path, str):
return "imported"
path = Path(source_path)
try:
resolved = path.resolve(strict=True)
except OSError:
return "imported"
run_directory = run.state.get("run_directory")
if isinstance(run_directory, str):
expected_source = (
Path(run_directory).resolve().parent.parent
/ "controllers"
/ controller_session_id
/ "claude-code-samples"
/ sample_id
/ f"{expected_session}.jsonl"
)
if resolved != expected_source.resolve():
return "imported"
expected_parts = (
"controllers",
controller_session_id,
"claude-code-samples",
sample_id,
f"{expected_session}.jsonl",
)
if resolved.parts[-len(expected_parts) :] != expected_parts or not resolved.is_file():
return "imported"
materialization = _read_json(resolved.parent / "materialization.json", maximum=MAX_STATE_BYTES)
recorded_transcript = materialization.get("transcript_path")
if not isinstance(recorded_transcript, str):
return "imported"
try:
trusted_transcript = Path(recorded_transcript).resolve(strict=True)
except OSError:
return "imported"
if (
materialization.get("sample_id") != sample_id
or materialization.get("thread_id") != imported_thread_id
or materialization.get("baseline_commit") != expected_commit
or trusted_transcript != resolved
):
return "imported"
repository_path = materialization.get("repository_path")
if not isinstance(repository_path, str) or baseline.get("repository") != repository_path:
return "imported"
if replay.get("project_dir") not in {None, repository_path}:
return "imported"
return "sample"
def _model_from_candidates(report: Mapping[str, Any], provider: str) -> object:
candidates = _mapping(report.get("candidates"))
return _mapping(candidates.get(provider)).get("model")
def _failure_phase(phase: object) -> str:
phases = {
"preparing": "preparation",
"creating_workspace": "preparation",
"implementing": "implementation",
"collecting": "implementation",
"reviewing": "evaluation",
"reporting": "reporting",
}
return phases.get(phase, "unknown") if isinstance(phase, str) else "unknown"
def _usage_observed(run: ReplayRun, provider: str) -> bool:
raw_usage = _mapping(run.report.get("usage")).get(provider)
if isinstance(raw_usage, list) and any(isinstance(item, Mapping) for item in raw_usage):
return True
if provider == "claude":
replay = _mapping(run.run.get("replay"))
historical = replay.get("historical_usage")
if isinstance(historical, Mapping) and historical:
return True
recorded = _mapping(replay.get("recorded_claude_result"))
if not recorded:
recorded = _mapping(run.report.get("recorded_claude_result"))
return bool(_mapping(recorded.get("usage")))
return bool(_mapping(_mapping(run.report.get("codex_execution")).get("usage")))
def _usage_fields(run: ReplayRun, provider: str) -> dict[str, int]:
if not _usage_observed(run, provider):
return {}
usage = _mapping(_mapping(run.report.get("normalized_usage")).get(provider))
raw = _mapping(run.report.get("usage")).get(provider)
records = [item for item in raw if isinstance(item, Mapping)] if isinstance(raw, list) else []
fields = (
("input_tokens", usage.get("total_input_tokens")),
("output_tokens", usage.get("output_tokens")),
(
"cached_input_tokens",
usage.get("cached_input_tokens"),
),
("ordinary_input_tokens", usage.get("ordinary_input_tokens")),
(
"cache_write_tokens",
usage.get("cache_write_tokens"),
),
)
result = {name: number for name, value in fields if (number := _integer(value)) is not None}
for name, aliases in (
("cache_write_1h_tokens", ("cache_write_1h_tokens", "ephemeral_1h_input_tokens")),
("cache_write_5m_tokens", ("cache_write_5m_tokens", "ephemeral_5m_input_tokens")),
("reasoning_output_tokens", ("reasoning_output_tokens", "reasoning_tokens")),
):
values = []
for record in records:
cache_creation = _mapping(record.get("cache_creation"))
for alias in aliases:
if (number := _integer(record.get(alias, cache_creation.get(alias)))) is not None:
values.append(number)
break
execution_usage = _mapping(_mapping(run.report.get("codex_execution")).get("usage"))
if not values and provider == "codex":
for alias in aliases:
if (number := _integer(execution_usage.get(alias))) is not None:
values.append(number)
break
if values:
result[name] = sum(values)
if records:
result["usage_record_count"] = len(records)
return result
def _phase_durations(run: ReplayRun) -> dict[str, float]:
raw_events = run.state.get("events")
if not isinstance(raw_events, list):
return {}
transitions = [
(phase, timestamp)
for event in raw_events
if isinstance(event, Mapping)
and (phase := event.get("phase")) in RUN_PHASES
and (timestamp := common.timestamp(event.get("at"))) is not None
]
result: dict[str, float] = {}
for (phase, started), (_, ended) in itertools.pairwise(transitions):
seconds = (ended - started).total_seconds()
if seconds >= 0:
result[phase] = result.get(phase, 0) + seconds
return result
def _cost(run: ReplayRun, provider: str) -> int | float | None:
estimate = _mapping(_mapping(run.report.get("estimated_cost")).get(provider))
if estimate.get("status") != "estimated":
return None
return _number(estimate.get("usd"))
def _claude_duration(run: ReplayRun) -> tuple[int | float | None, str]:
replay = _mapping(run.run.get("replay"))
recorded = _mapping(replay.get("recorded_claude_result"))
if not recorded:
recorded = _mapping(run.report.get("recorded_claude_result"))
if recorded:
duration = _number(recorded.get("duration_ms"))
if duration is None:
duration = _number(
run.report.get(
"historical_wall_clock_seconds", replay.get("historical_wall_clock_seconds")
)
)
else:
duration /= 1000
basis = "recorded_wall_clock"
else:
duration = _number(
run.report.get(
"historical_model_request_seconds", replay.get("historical_model_request_seconds")
)
)
basis = "model_request"
return duration, basis
def _candidate_labels(evaluation: Mapping[str, Any]) -> Mapping[str, str]:
candidates = _mapping(evaluation.get("candidate_mapping"))
result: dict[str, str] = {}
for label, provider in candidates.items():
if isinstance(label, str) and provider in {"codex", "claude"}:
result[provider] = label
return result if set(result) == {"codex", "claude"} else {}
def _quality_fields(run: ReplayRun) -> dict[str, object]:
if run.state.get("status") != "completed":
return {}
evaluation = _mapping(run.report.get("evaluation"))
if evaluation.get("status") != "completed":
return {}
labels = _candidate_labels(evaluation)
reviews = evaluation.get("reviews")
if not labels or not isinstance(reviews, list):
return {}
valid_reviews = [
review
for review in reviews
if isinstance(review, Mapping) and review.get("status") == "completed"
]
if not valid_reviews:
return {}
totals = _mapping(evaluation.get("totals"))
codex_score = _number(totals.get(labels["codex"]))
claude_score = _number(totals.get(labels["claude"]))
if codex_score is None or claude_score is None:
winner = "unavailable"
else:
codex_comparable = int(codex_score * 100 + 0.5)
claude_comparable = int(claude_score * 100 + 0.5)
if codex_comparable > claude_comparable:
winner = "codex"
elif claude_comparable > codex_comparable:
winner = "claude"
else:
winner = "tie"
fields: dict[str, object] = {"winner": winner}
if codex_score is not None:
fields["codex_score"] = codex_score
if claude_score is not None:
fields["claude_score"] = claude_score
comparable = evaluation.get("comparable_dimensions")
allowed = (
{dimension for dimension in comparable if isinstance(dimension, str)}
if isinstance(comparable, list)
else set(REVIEW_DIMENSIONS)
)
dimension_scores: dict[str, dict[str, int | float]] = {}
dimension_winners: dict[str, str] = {}
for dimension in REVIEW_DIMENSIONS:
if dimension not in allowed:
continue
provider_scores: dict[str, int | float] = {}
for provider in ("codex", "claude"):
scores: list[int | float] = []
for review in valid_reviews:
ballot = _mapping(review.get("ballot"))
decision = _mapping(_mapping(ballot.get("dimensions")).get(dimension))
if decision.get("winner") == "not_applicable":
continue
candidate = _mapping(_mapping(decision.get("candidates")).get(labels[provider]))
score = _number(candidate.get("score"))
if score is not None:
scores.append(score)
if scores:
provider_scores[provider] = sum(scores) / len(scores)
if set(provider_scores) == {"codex", "claude"}:
dimension_scores[dimension] = provider_scores
codex_score = provider_scores["codex"]
claude_score = provider_scores["claude"]
dimension_winners[dimension] = (
"codex"
if codex_score > claude_score
else "claude"
if claude_score > codex_score
else "tie"
)
if dimension_scores:
fields["dimension_scores"] = dimension_scores
fields["dimension_winners"] = dimension_winners
return fields
def _run_metadata(run: ReplayRun) -> dict[str, object]:
report = run.report
evaluation = _mapping(report.get("evaluation"))
availability = evaluation.get("evaluator_availability")
reviews = evaluation.get("reviews")
capabilities = _mapping(report.get("capabilities"))
result: dict[str, object] = {}
list_counts = (
("changed_file_count", report.get("codex_changed_files")),
("capability_count", capabilities.get("items")),
("unavailable_capability_count", capabilities.get("unavailable_capabilities")),
("limitation_count", report.get("limitations")),
("review_count", reviews),
("comparable_dimension_count", evaluation.get("comparable_dimensions")),
)
for name, values in list_counts:
if isinstance(values, list):
result[name] = len(values)
attempt = _integer(run.state.get("implementation_attempt"))
if attempt is not None:
result["implementation_attempt_count"] = attempt
started = common.timestamp(run.state.get("started_at"))
completed = common.timestamp(run.state.get("completed_at"))
if started is not None and completed is not None:
elapsed = (completed - started).total_seconds()
if elapsed >= 0:
result["wall_clock_seconds"] = elapsed
if durations := _phase_durations(run):
result["phase_durations"] = durations
if isinstance(availability, list):
evaluators = [item for item in availability if isinstance(item, Mapping)]
result["evaluator_count"] = len(evaluators)
result["available_evaluator_count"] = sum(
evaluator.get("available") is True for evaluator in evaluators
)
if isinstance(reviews, list):
selected = []
for review in reviews:
if not isinstance(review, Mapping):
continue
evaluator_model = review.get("model")
if not isinstance(evaluator_model, str):
continue
family = common.model_family(
evaluator_model, ("sol", "terra", "luna", "opus", "sonnet", "haiku")
)
selected.append(
{
"model": family,
"provider": (
"codex"
if family in {"sol", "terra", "luna"}
else "claude"
if family in {"opus", "sonnet", "haiku"}
else "other"
),
"self_judged": "yes" if evaluator_model == run.model else "no",
"normalized": (
"yes"
if _mapping(review.get("normalization")).get("required") is True
else "no"
),
}
)
if selected:
result["evaluators"] = selected
worker_events = run.state.get("worker_events")
if isinstance(worker_events, list):
for event in reversed(worker_events):
if not isinstance(event, Mapping) or event.get("type") != "completed":
continue
counts = _mapping(event.get("itemCounts"))
for metric, item in (
("tool_call_count", "command_execution"),
("file_change_event_count", "file_change"),
("agent_message_count", "agent_message"),
("worker_error_count", "error"),
):
if (count := _integer(counts.get(item))) is not None:
result[metric] = count
break
return result
def _codex_model(run: ReplayRun) -> dict[str, object]:
status = run.state.get("status")
if status not in TERMINAL_STATUSES:
return {}
fields: dict[str, object] = {
"model": common.model_family(run.model, ("sol", "terra", "luna")),
"status": status,
}
if status != "completed":
# A cancelled retry may still carry the preceding attempt's failure.
diagnostic = _mapping(run.state.get("failure_diagnostic")) if status == "failed" else {}
code = diagnostic.get("worker_code")
controller_code = diagnostic.get("controller_code")
system = diagnostic.get("system_code")
system = system.lower() if isinstance(system, str) else "unknown"
stage = diagnostic.get("worker_stage")
fields["failure"] = {
"phase": _failure_phase(run.state.get("phase")),
"controller_code": controller_code
if isinstance(controller_code, str) and controller_code in CONTROLLER_CODES
else "unknown",
"worker_code": code if isinstance(code, str) and code in WORKER_CODES else "unknown",
"retryable": (
"yes"
if diagnostic.get("retryable") is True
else "no"
if diagnostic.get("retryable") is False
else "unknown"
),
}
details: dict[str, object] = {
"system_code": system if system in SYSTEM_CODES else "unknown",
"worker_stage": stage
if isinstance(stage, str) and stage in WORKER_STAGES
else "unknown",
}
if (exit_code := _exit_code(diagnostic.get("exit_code"))) is not None:
details["exit_code"] = exit_code
retries = _integer(diagnostic.get("retry_count"))
if retries is not None and retries <= 3:
details["retry_count"] = retries
if (elapsed_ms := _number(diagnostic.get("elapsed_ms"))) is not None:
details["duration_seconds"] = elapsed_ms / 1000
fields["failure_details"] = details
return fields
fields.update(_usage_fields(run, "codex"))
fields.update(_run_metadata(run))
if (cost := _cost(run, "codex")) is not None:
fields["cost_usd"] = cost
duration = _number(_mapping(run.report.get("codex_execution")).get("elapsed_seconds"))
if duration is not None:
fields["duration_seconds"] = duration
fields.update(_quality_fields(run))
return fields
def _claude_baseline(runs: Sequence[ReplayRun]) -> dict[str, object]:
preferred = sorted(runs, key=lambda run: run.state.get("status") != "completed")
model: object = None
for run in preferred:
replay = _mapping(run.run.get("replay"))
model = replay.get("claude_model")
if not isinstance(model, str):
model = _model_from_candidates(run.report, "claude")
if isinstance(model, str):
break
baseline: dict[str, object] = {
"model": common.model_family(model, ("fable", "opus", "sonnet", "haiku"))
}
for run in preferred:
for name, value in _usage_fields(run, "claude").items():
baseline.setdefault(name, value)
if "cost_usd" not in baseline and (cost := _cost(run, "claude")) is not None:
baseline["cost_usd"] = cost
if "duration_seconds" not in baseline:
duration, timing_basis = _claude_duration(run)
if duration is not None:
baseline["duration_seconds"] = duration
baseline["timing_basis"] = timing_basis
replay = _mapping(run.run.get("replay"))
changed = replay.get("historical_changed_files")
if "changed_file_count" not in baseline and isinstance(changed, list):
baseline["changed_file_count"] = len(changed)
wall_clock = _number(
run.report.get(
"historical_wall_clock_seconds", replay.get("historical_wall_clock_seconds")
)
)
if "wall_clock_seconds" not in baseline and wall_clock is not None:
baseline["wall_clock_seconds"] = wall_clock
timing = _mapping(replay.get("historical_model_request_timing"))
for name in (
"request_count",
"missing_request_count",
"unidentified_assistant_event_count",
):
if name not in baseline and (count := _integer(timing.get(name))) is not None:
baseline[name] = count
return baseline
def _replay_event(runs: Sequence[ReplayRun]) -> Mapping[str, object]:
# Artifact-less launch failures carry no independent provenance. They must
# not reclassify the verified sibling runs in the same approved comparison.
evidenced_runs = [run for run in runs if run.run or not run.state.get("launch_failed")]
source = (
"sample"
if evidenced_runs and all(_source(run) == "sample" for run in evidenced_runs)
else "imported"
)
models = [
_codex_model(run)
for run in runs[:MAX_MODELS]
if run.state.get("status") in TERMINAL_STATUSES
]
statuses = {model["status"] for model in models}
status = next(iter(statuses)) if len(statuses) == 1 else "partial"
return {
"name": "replay_completed",
"payload": {
"source": source,
"status": status,
"claude": _claude_baseline(runs),
"codex_models": models,
},
}
def _measurements(event: Mapping[str, object]) -> list[dict[str, object]]:
payload = _mapping(event.get("payload"))
claude = _mapping(payload.get("claude"))
source = payload.get("source")
claude_model = claude.get("model")
raw_models = payload.get("codex_models")
if not isinstance(source, str) or not isinstance(claude_model, str):
return []
if not isinstance(raw_models, list):
return []
models = [model for model in raw_models[:MAX_MODELS] if isinstance(model, Mapping)]
contexts: list[tuple[Mapping[str, Any], dict[str, str]]] = []
for index, model in enumerate(models):
codex_model = model.get("model")
if not isinstance(codex_model, str):
continue
contexts.append(
(
model,
{
"model_slot": MODEL_SLOTS[index],
"codex_model": codex_model,
"claude_model": claude_model,
"source": source,
},
)
)
rows: list[dict[str, object]] = []
for model, dimensions in contexts:
status = model.get("status")
if isinstance(status, str):
rows.append(common.measurement("replay", 1, {**dimensions, "status": status}))
claude_dimensions = {"claude_model": claude_model, "source": source}
for field in (
"input_tokens",
"output_tokens",
"cached_input_tokens",
"cost_usd",
"wall_clock_seconds",
):
value = _number(claude.get(field))
if value is not None:
rows.append(common.measurement(f"claude_{field}", value, claude_dimensions))
duration = _number(claude.get("duration_seconds"))
timing_basis = claude.get("timing_basis")
if duration is not None and isinstance(timing_basis, str):
rows.append(
common.measurement(
"claude_duration_seconds",
duration,
{**claude_dimensions, "timing_basis": timing_basis},
)
)
for model, dimensions in contexts:
failure = _mapping(model.get("failure"))
phase = failure.get("phase")
if isinstance(phase, str):
rows.append(
common.measurement(
"failure",
1,
{
**dimensions,
"phase": phase,
"controller_code": str(failure.get("controller_code", "unknown")),
"worker_code": str(failure.get("worker_code", "unknown")),
"retryable": str(failure.get("retryable", "unknown")),
},
)
)
winner = model.get("winner")
if isinstance(winner, str):
rows.append(common.measurement("outcome", 1, {**dimensions, "winner": winner}))
for model, dimensions in contexts:
details = _mapping(model.get("failure_details"))
if not details:
continue
rows.append(
common.measurement(
"failure_detail",
1,
{
**dimensions,
"system_code": str(details["system_code"]),
"worker_stage": str(details["worker_stage"]),
},
)
)
for field in ("exit_code", "retry_count", "duration_seconds"):
value = (
_exit_code(details.get(field))
if field == "exit_code"
else _number(details.get(field))
)
if value is not None:
rows.append(common.measurement(f"failure_{field}", value, dimensions))
for field in (
"input_tokens",
"output_tokens",
"cached_input_tokens",
"cost_usd",
"duration_seconds",
"wall_clock_seconds",
):
for model, dimensions in contexts:
value = _number(model.get(field))
if value is not None:
rows.append(common.measurement(f"codex_{field}", value, dimensions))
for field in ("codex_score", "claude_score"):
for model, dimensions in contexts:
value = _number(model.get(field))
if value is not None:
rows.append(common.measurement(field, value, dimensions))
for model, dimensions in contexts:
for phase, value in _mapping(model.get("phase_durations")).items():
seconds = _number(value)
if phase in {"implementing", "reviewing"} and seconds is not None:
rows.append(
common.measurement(
"codex_phase_duration_seconds", seconds, {**dimensions, "phase": phase}
)
)
evaluator_counts: dict[tuple[str, str, str, str], int] = {}
evaluators = model.get("evaluators")
if isinstance(evaluators, list):
for evaluator in evaluators:
if not isinstance(evaluator, Mapping):
continue
values = tuple(
evaluator.get(key) for key in ("model", "provider", "self_judged", "normalized")
)
if all(isinstance(value, str) for value in values):
key = (str(values[0]), str(values[1]), str(values[2]), str(values[3]))
evaluator_counts[key] = evaluator_counts.get(key, 0) + 1
for (
evaluator_model,
evaluator_provider,
self_judged,
normalized,
), count in evaluator_counts.items():
rows.append(
common.measurement(
"evaluation",
count,
{
**dimensions,
"evaluator_model": evaluator_model,
"evaluator_provider": evaluator_provider,
"self_judged": self_judged,
"normalized": normalized,
},
)
)
for dimension in REVIEW_DIMENSIONS:
for model, dimensions in contexts:
scores = _mapping(_mapping(model.get("dimension_scores")).get(dimension))
for provider in ("codex", "claude"):
value = _number(scores.get(provider))
if value is not None:
rows.append(
common.measurement(
f"{provider}_dimension_score",
value,
{**dimensions, "score_dimension": dimension},
)
)
winner = _mapping(model.get("dimension_winners")).get(dimension)
if winner in {"codex", "claude", "tie"}:
rows.append(
common.measurement(
"dimension_outcome",
1,
{**dimensions, "score_dimension": dimension, "winner": winner},
)
)
return rows
def _payload(event: Mapping[str, object]) -> bytes:
return common.encode_measurements(_measurements(event))
@dataclass
class AttemptObservation:
outcome: str
runs: list[ReplayRun]
models: list[tuple[str, str]]
controller_ready: bool = False
start_requested: bool = False
run_observed: bool = False
def _collect_attempt_runs(
attempt: Mapping[str, Any], run_root: Path, session: str
) -> tuple[list[ReplayRun], list[tuple[str, str]], bool, bool]:
"""Read detailed run artifacts once after a durable terminal receipt."""
common.approved_models(attempt)
entries = attempt.get("models", [])
runs: list[ReplayRun] = []
models: list[tuple[str, str]] = []
run_observed = False
unreadable = False
for entry in entries:
model = entry["model"]
status = "unresolved"
try:
if entry.get("launch_status") == "failed":
status = "launch_failed"
runs.append(
ReplayRun(
model,
{
"status": "failed",
"phase": "creating_workspace",
"launch_failed": True,
"error": entry.get("error"),
"failure_diagnostic": {"controller_code": entry.get("controller_code")},
},
{},
{},
)
)
elif entry.get("run_id") is not None:
state = common.model_state(entry, run_root, session)
directory = run_root / entry["run_id"]
if state is not None:
run_observed = True
terminal = state.get("status")
if terminal in TERMINAL_STATUSES:
report = common.observed_json(directory / "report.json", maximum=None)
if terminal != "completed" or report is not None:
status = terminal
run = common.observed_json(
directory / "run.json", maximum=MAX_RUN_BYTES
)
runs.append(ReplayRun(model, state, run or {}, report or {}))
except (OSError, ValueError, UnicodeError):
status = "unreadable"
unreadable = True
models.append((model, status))
return runs, models, run_observed, unreadable
def _resolved_outcome(models: Sequence[tuple[str, str]]) -> str | None:
statuses = {status for _, status in models}
if not models or statuses.intersection({"unresolved", "unreadable"}):
return None
return (
"succeeded"
if statuses == {"completed"}
else "launch_failure"
if statuses == {"launch_failed"}
else "failed"
if statuses <= {"failed", "launch_failed"}
else "cancelled"
if statuses == {"cancelled"}
else "mixed"
)
def _observe_attempt(
arguments: argparse.Namespace,
run_root: Path,
) -> AttemptObservation:
session = arguments.controller_session_id
root = run_root.parent / "controllers" / session
pre_start_deadline = time.monotonic() + arguments.pre_start_timeout_seconds
deadline: float | None = None
started_at: dt.datetime | None = None
controller_pid: int | None = None
observed = AttemptObservation("no_run_observed", [], [])
while True:
try:
attempt = common.read_attempt_if_present(root / "attempt.json", session)
if attempt is None:
if time.monotonic() >= pre_start_deadline:
observed.outcome = "prestart_timeout"
return observed
common.pause(arguments.poll_interval_seconds, pre_start_deadline)
continue
observed.controller_ready |= attempt.get("controller_ready") is True
observed.start_requested |= attempt.get("start_requested") is True
if attempt.get("startup_failed") is True:
observed.outcome = "startup_failure"
return observed
selected = common.approved_models(attempt)
observed.models = [(model, "unresolved") for model in selected]
started_at = common.start_time(attempt, selected)
if started_at is not None and deadline is None:
deadline = common.observation_deadline(started_at, arguments.run_timeout_seconds)
controller_pid = common.persisted_pid(attempt, controller_pid)
if attempt.get("final_results_ready") is True:
(
observed.runs,
observed.models,
observed.run_observed,
unreadable,
) = _collect_attempt_runs(attempt, run_root, session)
outcome = _resolved_outcome(observed.models)
observed.outcome = (
outcome if outcome is not None and not unreadable else "unreadable_state"
)
return observed
except (OSError, ValueError, UnicodeError):
observed.outcome = "unreadable_state"
return observed
expired = deadline is not None and time.monotonic() >= deadline
pre_start_expired = deadline is None and time.monotonic() >= pre_start_deadline
dead = False
try:
dead, controller_pid = common.controller_dead(
root, session, controller_pid, stopped=attempt.get("controller_stopped") is True
)
except (OSError, ValueError, UnicodeError):
observed.outcome = "unreadable_state"
return observed
if expired or pre_start_expired or dead:
try:
(
observed.runs,
observed.models,
observed.run_observed,
unreadable,
) = _collect_attempt_runs(attempt, run_root, session)
except (OSError, ValueError, UnicodeError):
unreadable = True
outcome = _resolved_outcome(observed.models)
if outcome is not None and not unreadable:
observed.outcome = outcome
return observed
statuses = {status for _, status in observed.models}
observed.outcome = (
"unreadable_state"
if unreadable or "unreadable" in statuses
else "prestart_timeout"
if pre_start_expired
else "interrupted"
if dead and observed.start_requested
else "observation_timeout"
if observed.start_requested
else "no_run_observed"
)
return observed
common.pause(
arguments.poll_interval_seconds,
deadline if deadline is not None else pre_start_deadline,
)
def _attempt_measurements(observed: AttemptObservation) -> list[dict[str, object]]:
if observed.outcome == "prestart_timeout":
return [common.measurement("final_prestart_timeout", 1, {})]
rows = [
common.measurement(
"attempt",
1,
{
"outcome": observed.outcome,
"controller_ready": "yes" if observed.controller_ready else "no",
"start_requested": "yes" if observed.start_requested else "no",
"run_observed": "yes" if observed.run_observed else "no",
"reporting_complete": "yes"
if observed.outcome
in {"succeeded", "failed", "mixed", "cancelled", "launch_failure"}
else "no",
},
)
]
for index, (model, status) in enumerate(observed.models):
if status in TERMINAL_STATUSES or status == "launch_failed":
continue # The existing replay row already accounts for this model.
rows.append(
common.measurement(
"attempt_model",
1,
{
"model_slot": MODEL_SLOTS[index],
"codex_model": common.model_family(model, ("sol", "terra", "luna")),
"status": status,
},
)
)
if observed.runs:
model_rows = _measurements(_replay_event(observed.runs))
# Preserve approved slots even when an earlier model is still unresolved.
slots = {model: MODEL_SLOTS[i] for i, (model, _) in enumerate(observed.models)}
for row in model_rows:
dimensions = row.get("dimensions")
if isinstance(dimensions, dict) and "model_slot" in dimensions:
index = MODEL_SLOTS.index(dimensions["model_slot"])
dimensions["model_slot"] = slots[observed.runs[index].model]
rows.extend(model_rows)
return rows
def _bounded_payload(complete_rows: Sequence[dict[str, object]]) -> bytes:
rows = list(complete_rows)
omitted = 0
while True:
submitted = rows + (
[common.measurement("measurements_omitted", omitted, {})] if omitted else []
)
encoded = common.encode_measurements(submitted)
if len(submitted) <= MAX_OUTPUT_ROWS and len(encoded) <= MAX_OUTPUT_BYTES:
return encoded
if len(rows) <= 1:
raise ValueError("Replay attempt summary exceeds the sidecar limit")
if not omitted and rows[0]["name"] == "attempt":
rows[0] = {
**rows[0],
"dimensions": {**_mapping(rows[0]["dimensions"]), "reporting_complete": "no"},
}
rows.pop()
omitted += 1
def main(argv: Sequence[str] | None = None) -> int:
if not common.analytics_available():
return 0
arguments = common.arguments(argv)
try:
common.require_session(arguments.controller_session_id)
except ValueError as error:
print(str(error), file=sys.stderr)
return 2
run_root = common.run_root(arguments)
root = run_root.parent / "controllers" / arguments.controller_session_id
try:
sidecar = common.open_sidecar()
if sidecar is None:
return 0
with sidecar:
common.observer_started("final_results", arguments.controller_session_id)
observed = _observe_attempt(arguments, run_root)
rows = _attempt_measurements(observed)
complete_payload = common.encode_measurements(rows)
payload = _bounded_payload(rows)
if payload != complete_payload:
print(
"Replay final metrics exceed the host limit; reporting_complete=no. "
"Full local measurements are not a delivery acknowledgment.",
file=sys.stderr,
)
common.write_sidecar(sidecar, payload)
except (OSError, ValueError) as error:
print(f"Unable to write the trusted Replay sidecar: {error}.", file=sys.stderr)
return 1
common.preserve_payload(root / "metrics.json", payload)
common.preserve_payload(root / "metrics-full.json", complete_payload)
return 0
if __name__ == "__main__":
raise SystemExit(main())
SHA-256: 1cb2d650a348eaf519db106a875fbbd3fc858770f36fab5761f3b7ba6f533c12