tegwick
20e6f381f6
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a06ba0-10aa-7ea0-b20a-4f3fac39efe9
144 lines
4.4 KiB
Python
144 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from rein_aharness.metrics import (
|
|
external_metrics_dir,
|
|
record_execution,
|
|
record_external_execution,
|
|
regenerate_summary,
|
|
)
|
|
|
|
|
|
def test_record_execution_writes_jsonl_and_summary(tmp_path: Path) -> None:
|
|
path = record_execution(
|
|
tmp_path,
|
|
"coach",
|
|
success=True,
|
|
execution_time_s=12.5,
|
|
tokens=100,
|
|
committed=True,
|
|
head_after="abc123",
|
|
reason=None,
|
|
metadata={"task_title": "hello"},
|
|
)
|
|
assert path.is_file()
|
|
lines = path.read_text(encoding="utf-8").strip().splitlines()
|
|
assert len(lines) == 1
|
|
rec = json.loads(lines[0])
|
|
assert rec["agent"] == "coach"
|
|
assert rec["success"] is True
|
|
assert rec["tokens"] == 100
|
|
assert rec["harness"] == "rein-aharness"
|
|
assert rec["metadata"]["task_title"] == "hello"
|
|
|
|
summary = json.loads(
|
|
(tmp_path / ".kaizen" / "metrics" / "coach" / "summary.json").read_text()
|
|
)
|
|
assert summary["execution_count"] == 1
|
|
assert summary["success_rate"] == 1.0
|
|
assert summary["avg_execution_time_s"] == 12.5
|
|
|
|
|
|
def test_summary_aggregates_multiple(tmp_path: Path) -> None:
|
|
record_execution(tmp_path, "coach", success=True, execution_time_s=10)
|
|
record_execution(tmp_path, "coach", success=False, execution_time_s=20)
|
|
summary = json.loads(
|
|
(tmp_path / ".kaizen" / "metrics" / "coach" / "summary.json").read_text()
|
|
)
|
|
assert summary["execution_count"] == 2
|
|
assert summary["success_rate"] == 0.5
|
|
assert summary["avg_execution_time_s"] == 15.0
|
|
|
|
|
|
def test_regenerate_summary_empty() -> None:
|
|
assert regenerate_summary("x", [])["execution_count"] == 0
|
|
|
|
|
|
def test_external_metrics_are_durable_and_projection_ready(tmp_path: Path) -> None:
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
state_dir = tmp_path / "state"
|
|
|
|
path = record_external_execution(
|
|
repo,
|
|
"coach",
|
|
success=True,
|
|
committed=True,
|
|
head_after="abc123",
|
|
metadata={"repository_grant_id": "grant-1"},
|
|
session_id="transaction-1",
|
|
state_dir=state_dir,
|
|
)
|
|
|
|
directory = external_metrics_dir(repo, "coach", state_dir=state_dir)
|
|
assert path == directory / "executions.jsonl"
|
|
assert path.stat().st_mode & 0o777 == 0o600
|
|
assert directory.stat().st_mode & 0o777 == 0o700
|
|
record = json.loads(path.read_text(encoding="utf-8").strip())
|
|
assert record["metadata"]["repository_grant_id"] == "grant-1"
|
|
assert record["session_id"] == "transaction-1"
|
|
projection = json.loads(
|
|
(directory / "projection.json").read_text(encoding="utf-8")
|
|
)
|
|
assert projection["repository_name"] == "repo"
|
|
assert projection["target_relative_directory"] == ".kaizen/metrics/coach"
|
|
assert not (repo / ".kaizen").exists()
|
|
|
|
|
|
def test_external_metrics_deduplicate_transaction_replay(tmp_path: Path) -> None:
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
state_dir = tmp_path / "state"
|
|
|
|
for _attempt in range(2):
|
|
path = record_external_execution(
|
|
repo,
|
|
"coach",
|
|
success=True,
|
|
session_id="transaction-1",
|
|
state_dir=state_dir,
|
|
)
|
|
|
|
assert len(path.read_text(encoding="utf-8").splitlines()) == 1
|
|
summary = json.loads((path.parent / "summary.json").read_text(encoding="utf-8"))
|
|
assert summary["execution_count"] == 1
|
|
|
|
|
|
def test_external_metrics_refuse_an_incomplete_ledger(tmp_path: Path) -> None:
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
state_dir = tmp_path / "state"
|
|
directory = external_metrics_dir(repo, "coach", state_dir=state_dir)
|
|
directory.mkdir(parents=True)
|
|
ledger = directory / "executions.jsonl"
|
|
ledger.write_text('{"incomplete":true}', encoding="utf-8")
|
|
|
|
with pytest.raises(OSError, match="incomplete record"):
|
|
record_external_execution(
|
|
repo,
|
|
"coach",
|
|
success=True,
|
|
state_dir=state_dir,
|
|
)
|
|
|
|
assert ledger.read_text(encoding="utf-8") == '{"incomplete":true}'
|
|
|
|
|
|
def test_external_metrics_refuse_unsafe_projection_agent(tmp_path: Path) -> None:
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
state_dir = tmp_path / "state"
|
|
|
|
with pytest.raises(OSError, match="safe metrics projection"):
|
|
record_external_execution(
|
|
repo,
|
|
"../escape",
|
|
success=True,
|
|
state_dir=state_dir,
|
|
)
|
|
|
|
assert not state_dir.exists()
|