File size: 3,355 Bytes
d61821a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 | 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()
|