tenant.openagents/omega

1from __future__ import annotations
2
3import json
4import os
5import tempfile
6import unittest
7from pathlib import Path
8from unittest.mock import patch
9
10from zed_eval import run_index
11
12
13class RunIndexTests(unittest.TestCase):
14    def with_index_path(self, path: Path):
15        return patch.dict(os.environ, {"AGENT_EVALS_RUN_INDEX": str(path)})
16
17    def test_record_lookup_and_recent(self) -> None:
18        with tempfile.TemporaryDirectory() as temporary_directory:
19            path = Path(temporary_directory) / "nested" / "run-index.json"
20            with self.with_index_path(path):
21                run_index.record_run(
22                    {
23                        "run_id": "run-1",
24                        "namespace": "alice",
25                        "experiment_name": "rf",
26                        "volume_name": "custom-volume",
27                        "agent_model": "sonnet-4.6",
28                        "created_at": "2026-01-01T00:00:00+00:00",
29                    }
30                )
31                run_index.record_run(
32                    {
33                        "run_id": "run-2",
34                        "namespace": "alice",
35                        "experiment_name": "qna",
36                        "volume_name": None,
37                    }
38                )
39
40                self.assertEqual(run_index.lookup("run-1")["volume"], "custom-volume")
41                self.assertEqual(
42                    [entry["run_id"] for entry in run_index.recent(2)],
43                    ["run-2", "run-1"],
44                )
45                self.assertEqual(run_index.most_recent()["run_id"], "run-2")
46                self.assertNotIn("volume", run_index.lookup("run-2"))
47
48    def test_record_refreshes_existing_run(self) -> None:
49        with tempfile.TemporaryDirectory() as temporary_directory:
50            path = Path(temporary_directory) / "run-index.json"
51            with self.with_index_path(path):
52                run_index.record_run(
53                    {
54                        "run_id": "run-1",
55                        "namespace": "alice",
56                        "experiment_name": "rf",
57                    }
58                )
59                run_index.record_run(
60                    {
61                        "run_id": "run-1",
62                        "namespace": "bob",
63                        "experiment_name": "tw",
64                    }
65                )
66
67                entries = run_index.recent(10)
68                self.assertEqual(len(entries), 1)
69                self.assertEqual(entries[0]["namespace"], "bob")
70                self.assertEqual(entries[0]["experiment_name"], "tw")
71
72    def test_corrupt_or_malformed_index_is_ignored(self) -> None:
73        with tempfile.TemporaryDirectory() as temporary_directory:
74            path = Path(temporary_directory) / "run-index.json"
75            with self.with_index_path(path):
76                path.write_text("not json")
77                self.assertIsNone(run_index.lookup("run-1"))
78                self.assertEqual(run_index.recent(), [])
79
80                path.write_text(
81                    json.dumps(
82                        {
83                            "runs": [
84                                {"run_id": "missing-location"},
85                                ["not", "a", "dict"],
86                                {
87                                    "run_id": "run-1",
88                                    "namespace": "alice",
89                                    "experiment_name": "rf",
90                                },
91                            ]
92                        }
93                    )
94                )
95                self.assertIsNone(run_index.lookup("missing-location"))
96                self.assertEqual(run_index.lookup("run-1")["experiment_name"], "rf")
97
98    def test_caps_entries(self) -> None:
99        with tempfile.TemporaryDirectory() as temporary_directory:
100            path = Path(temporary_directory) / "run-index.json"
101            with self.with_index_path(path), patch.object(run_index, "MAX_ENTRIES", 2):
102                for index in range(3):
103                    run_index.record_run(
104                        {
105                            "run_id": f"run-{index}",
106                            "namespace": "alice",
107                            "experiment_name": "rf",
108                        }
109                    )
110
111                self.assertEqual(
112                    [entry["run_id"] for entry in run_index.recent(10)],
113                    ["run-2", "run-1"],
114                )
115                self.assertIsNone(run_index.lookup("run-0"))
116
117    def test_zero_limit_returns_no_entries(self) -> None:
118        with tempfile.TemporaryDirectory() as temporary_directory:
119            path = Path(temporary_directory) / "run-index.json"
120            with self.with_index_path(path):
121                run_index.record_run(
122                    {
123                        "run_id": "run-1",
124                        "namespace": "alice",
125                        "experiment_name": "rf",
126                    }
127                )
128                self.assertEqual(run_index.recent(0), [])
129
130
131if __name__ == "__main__":
132    unittest.main()
133