| """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" |
| |
| 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"] |
|
|
