tenant.openagents/omega

1"""Local, launch-time index of runs started from this machine.
2
3Every launch writes a tiny record mapping a run id to where it lives on the
4volume (namespace + experiment) plus a little metadata. This lets `status`,
5`logs`, `fetch`, `report`, and `rejudge` locate a run from just its id, so the
6operator no longer has to repeat `--namespace`/`--experiment-name`.
7
8It is deliberately best-effort: a missing, unreadable, or unwritable index never
9blocks a launch and never crashes a lookup. Commands that can't find a run in
10the index fall back to the explicit `--experiment-name`/`--namespace` flags with
11a friendly hint.
12"""
13
14from __future__ import annotations
15
16import json
17import os
18import tempfile
19from pathlib import Path
20from typing import Any
21
22INDEX_VERSION = 1
23MAX_ENTRIES = 500
24REQUIRED_FIELDS = ("run_id", "namespace", "experiment_name")
25OPTIONAL_FIELDS = (
26    "volume",
27    "agent_model",
28    "judge_preset",
29    "build_id",
30    "suite_id",
31    "kind",
32    "created_at",
33)
34
35
36def index_path() -> Path:
37    override = os.environ.get("AGENT_EVALS_RUN_INDEX")
38    if override:
39        return Path(override).expanduser()
40    cache_home = os.environ.get("XDG_CACHE_HOME")
41    base = Path(cache_home).expanduser() if cache_home else Path.home() / ".cache"
42    return base / "agent-evals" / "run-index.json"
43
44
45def _string_field(data: dict[str, Any], field: str) -> str | None:
46    value = data.get(field)
47    if value is None:
48        return None
49    value = str(value)
50    return value if value else None
51
52
53def _normalize_entry(data: dict[str, Any]) -> dict[str, Any] | None:
54    entry = {}
55    for field in REQUIRED_FIELDS:
56        value = _string_field(data, field)
57        if value is None:
58            return None
59        entry[field] = value
60    for field in OPTIONAL_FIELDS:
61        value = data.get(field)
62        if value is not None:
63            entry[field] = value
64    return entry
65
66
67def _load_entries() -> list[dict[str, Any]]:
68    try:
69        data = json.loads(index_path().read_text(encoding="utf-8"))
70    except (OSError, json.JSONDecodeError):
71        return []
72    raw_entries = data.get("runs") if isinstance(data, dict) else data
73    if not isinstance(raw_entries, list):
74        return []
75
76    entries = []
77    for raw_entry in raw_entries:
78        if not isinstance(raw_entry, dict):
79            continue
80        entry = _normalize_entry(raw_entry)
81        if entry is not None:
82            entries.append(entry)
83    return entries
84
85
86def _entry_from_request(run_request: dict[str, Any]) -> dict[str, Any] | None:
87    entry = _normalize_entry(
88        {
89            "run_id": run_request.get("run_id"),
90            "namespace": run_request.get("namespace"),
91            "experiment_name": run_request.get("experiment_name"),
92            "volume": run_request.get("volume_name"),
93            "agent_model": run_request.get("agent_model"),
94            "judge_preset": run_request.get("judge_preset"),
95            "build_id": run_request.get("build_id"),
96            "suite_id": run_request.get("suite_id"),
97            "kind": run_request.get("kind"),
98            "created_at": run_request.get("created_at"),
99        }
100    )
101    return entry
102
103
104def _write_entries(entries: list[dict[str, Any]]) -> None:
105    path = index_path()
106    payload = json.dumps({"version": INDEX_VERSION, "runs": entries}, indent=2) + "\n"
107    path.parent.mkdir(parents=True, exist_ok=True)
108    temporary_path: Path | None = None
109    try:
110        with tempfile.NamedTemporaryFile(
111            "w",
112            encoding="utf-8",
113            dir=path.parent,
114            prefix=f".{path.name}.",
115            suffix=".tmp",
116            delete=False,
117        ) as temporary_file:
118            temporary_path = Path(temporary_file.name)
119            temporary_file.write(payload)
120        os.replace(temporary_path, path)
121    except OSError:
122        if temporary_path is not None:
123            temporary_path.unlink(missing_ok=True)
124        raise
125
126
127def record_run(run_request: dict[str, Any]) -> None:
128    """Append (or refresh) the index entry for a launched run.
129
130    Accepts the same request dict the controller is spawned with. Silently does
131    nothing if the essential fields are missing or the index can't be written —
132    bookkeeping must never abort a launch.
133    """
134    entry = _entry_from_request(run_request)
135    if entry is None:
136        return
137
138    entries = [
139        item for item in _load_entries() if item.get("run_id") != entry["run_id"]
140    ]
141    entries.append(entry)
142    entries = entries[-MAX_ENTRIES:]
143    try:
144        _write_entries(entries)
145    except OSError:
146        pass
147
148
149def lookup(run_id: str | None) -> dict[str, Any] | None:
150    """Return the most recently recorded entry for `run_id`, if any."""
151    if not run_id:
152        return None
153    for entry in reversed(_load_entries()):
154        if entry["run_id"] == run_id:
155            return entry
156    return None
157
158
159def recent(limit: int = 20) -> list[dict[str, Any]]:
160    """Return up to `limit` most-recent entries, newest first."""
161    if limit <= 0:
162        return []
163    return list(reversed(_load_entries()[-limit:]))
164
165
166def most_recent() -> dict[str, Any] | None:
167    entries = recent(1)
168    return entries[0] if entries else None
169