File size: 4,496 Bytes
c27193b | 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 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | """Durable append-only autoresearch ledger (discipline 4).
`operator/runs.py` keeps a 200-event in-memory ring buffer for live SSE; its history
dies on restart. The ledger is the complementary durable record: one append-only
`attempts.jsonl` under the search's `log_root`, surviving restarts and aggregating
every researcher's attempts (the §5.3 wide-search property). It is also the dedup
source the guard consults — `has_commit` keeps the search from re-scoring a commit.
Mirrors the in-tree JSONL-ledger convention (`autotune/feedback.py`,
`eval/mei/history.py`): pydantic rows, `model_dump_json()` per line, malformed lines
skipped on read. Pure stdlib + pydantic; base-install importable.
"""
from __future__ import annotations
from datetime import UTC, datetime
from pathlib import Path
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field
LEDGER_FILENAME = "attempts.jsonl"
# Terminal statuses mirror the attempt runner: 124 → timed_out, nonzero → failed.
LedgerStatus = Literal["running", "completed", "timed_out", "failed"]
class AttemptMetric(BaseModel):
"""The measured scalar for one attempt (None until the attempt completes)."""
model_config = ConfigDict(extra="forbid", frozen=True)
name: str = Field(min_length=1)
value: float | None = None
mode: Literal["max", "min"] = "max"
class LedgerEntry(BaseModel):
"""One row in the autoresearch ledger. Frozen + extra=forbid for hygiene."""
model_config = ConfigDict(extra="forbid", frozen=True)
timestamp: str = Field(description="ISO-8601 UTC of the event.")
kind: Literal["attempt", "baseline"] = "attempt"
researcher: str = Field(min_length=1)
commit: str = Field(min_length=1, description="Short git commit the attempt measured.")
status: LedgerStatus
run_dir: str | None = None
metric: AttemptMetric | None = None
class Ledger:
"""Append-only ledger bound to ``log_root/attempts.jsonl``.
One shared aggregate file with a ``researcher`` field per row; the read API
filters in memory. Concurrent appends from future parallel researchers rely on
POSIX ``O_APPEND`` atomicity for sub-PIPE_BUF lines — the seam to revisit
(per-researcher files) when k8s researchers land.
"""
def __init__(self, log_root: Path) -> None:
self._path = Path(log_root) / LEDGER_FILENAME
@property
def path(self) -> Path:
return self._path
def append(
self,
*,
researcher: str,
commit: str,
status: LedgerStatus,
kind: Literal["attempt", "baseline"] = "attempt",
run_dir: str | None = None,
metric: AttemptMetric | dict[str, object] | None = None,
timestamp: str | None = None,
) -> LedgerEntry:
"""Append one row; creates the parent directory if needed."""
self._path.parent.mkdir(parents=True, exist_ok=True)
coerced = AttemptMetric.model_validate(metric) if isinstance(metric, dict) else metric
entry = LedgerEntry(
timestamp=timestamp or datetime.now(UTC).isoformat(),
kind=kind,
researcher=researcher,
commit=commit,
status=status,
run_dir=run_dir,
metric=coerced,
)
with self._path.open("a", encoding="utf-8") as fh:
fh.write(entry.model_dump_json() + "\n")
return entry
def read_all(self) -> list[LedgerEntry]:
"""Read every row, oldest-first; missing file → []; malformed lines skipped."""
if not self._path.exists():
return []
out: list[LedgerEntry] = []
for line in self._path.read_text(encoding="utf-8").splitlines():
stripped = line.strip()
if not stripped:
continue
try:
out.append(LedgerEntry.model_validate_json(stripped))
except ValueError:
continue
return out
def attempts_for(self, researcher: str) -> list[LedgerEntry]:
"""Every row for one researcher, oldest-first."""
return [e for e in self.read_all() if e.researcher == researcher]
def has_commit(self, researcher: str, commit: str) -> bool:
"""True iff this researcher already has a row for `commit` (dedup gate)."""
return any(e.researcher == researcher and e.commit == commit for e in self.read_all())
__all__ = ["LEDGER_FILENAME", "AttemptMetric", "Ledger", "LedgerEntry", "LedgerStatus"]
|