| from __future__ import annotations |
|
|
| import json |
| from pathlib import Path |
| import tempfile |
| import unittest |
|
|
| from agent_harness.telemetry import ( |
| EventWriter, |
| RunIdentity, |
| load_completed_or_archive_incomplete, |
| run_directory, |
| ) |
|
|
|
|
| def sample_identity() -> RunIdentity: |
| return RunIdentity( |
| experiment_id="E03", |
| task_id="TASK001", |
| harness_id="H000", |
| harness_hash="a" * 64, |
| model_id="M001", |
| model_key="qwen3.6-35b-a3b", |
| model_config_hash="b" * 64, |
| context_budget=65536, |
| seed=0, |
| repetition=0, |
| repository_sha="c" * 40, |
| code_revision="d" * 40, |
| ) |
|
|
|
|
| class TelemetryTests(unittest.TestCase): |
| def test_run_identity_is_deterministic(self) -> None: |
| self.assertEqual(sample_identity().run_id, sample_identity().run_id) |
| self.assertEqual(len(sample_identity().run_id), 20) |
|
|
| def test_writer_creates_manifest_and_ordered_events(self) -> None: |
| identity = sample_identity() |
| with tempfile.TemporaryDirectory() as temporary: |
| root = Path(temporary) |
| with EventWriter(root, identity, {"harness_id": "H000"}, {"id": "qwen"}) as writer: |
| writer.emit("run_started", {"order": 1}) |
| writer.emit("run_finished", {"resolved_at_1": False}) |
| writer.write_artifact("patch.diff", "") |
|
|
| directory = run_directory(root, identity) |
| manifest = json.loads((directory / "run_manifest.json").read_text()) |
| events = [json.loads(line) for line in (directory / "trajectory.jsonl").read_text().splitlines()] |
| self.assertEqual(manifest["run_id"], identity.run_id) |
| self.assertEqual([event["sequence"] for event in events], [0, 1]) |
| self.assertTrue((directory / "patch.diff").exists()) |
|
|
| with self.assertRaises(FileExistsError): |
| EventWriter(root, identity, {}, {}) |
|
|
| def test_artifact_paths_cannot_escape_run_directory(self) -> None: |
| with tempfile.TemporaryDirectory() as temporary: |
| with EventWriter(Path(temporary), sample_identity(), {}, {}) as writer: |
| with self.assertRaises(ValueError): |
| writer.write_artifact("../outside.txt", "unsafe") |
|
|
| def test_incomplete_attempt_is_archived_before_identical_retry(self) -> None: |
| with tempfile.TemporaryDirectory() as temporary: |
| root = Path(temporary) |
| identity = sample_identity() |
| incomplete = run_directory(root, identity) |
| incomplete.mkdir(parents=True) |
| (incomplete / "trajectory.jsonl").write_text( |
| "partial\n", encoding="utf-8" |
| ) |
| self.assertIsNone(load_completed_or_archive_incomplete(root, identity)) |
| self.assertFalse(incomplete.exists()) |
| archive_root = ( |
| root |
| / "infrastructure_attempts" |
| / identity.experiment_id |
| / identity.harness_id |
| / identity.task_id |
| ) |
| archives = list(archive_root.iterdir()) |
| self.assertEqual(len(archives), 1) |
| self.assertTrue((archives[0] / "trajectory.jsonl").exists()) |
| self.assertTrue((archives[0] / "archive_record.json").exists()) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
