| """Run one measured autoresearch attempt (discipline 1). |
| |
| Wraps the existing run path: enforce the git fence + dedup, append a `running` row, |
| shell the contract's `command` under a bounded wall-clock (identical per attempt so |
| scores are comparable), read the declared metric, capture the editable's diff for the |
| UI, and append the terminal row. Bounded teardown is a full process-group SIGTERM→ |
| SIGKILL so a timed-out training subprocess never orphans (exit 124 on timeout). |
| |
| The metric `source` is read from a conventional artifact the command writes into the |
| run dir — `metrics_jsonl` is the universal path; the eval-backed sources read the JSON |
| their (extra-gated) eval already produced, so this module stays base-install importable. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import json |
| import os |
| import signal |
| import subprocess |
| import time |
| from pathlib import Path |
|
|
| from pydantic import BaseModel, ConfigDict, Field |
|
|
| from mindxtrain.research.contract import AttemptContract |
| from mindxtrain.research.guard import assert_clean_and_committed, assert_new_commit, git, repo_root |
| from mindxtrain.research.ledger import AttemptMetric, Ledger, LedgerStatus |
|
|
| |
| _SOURCE_ARTIFACT: dict[str, tuple[str, str | None]] = { |
| "classroom_delta": ("classroom.json", "imprint_delta"), |
| "mei": ("mei.json", "composite"), |
| "bfcl": ("bfcl.json", None), |
| "tau_bench": ("tau_bench.json", None), |
| "held_out_loss": ("held_out_loss.json", None), |
| } |
| _TIMEOUT_EXIT = 124 |
|
|
|
|
| class AttemptResult(BaseModel): |
| """The outcome of one measured attempt.""" |
|
|
| model_config = ConfigDict(extra="forbid", frozen=True) |
|
|
| commit: str |
| value: float | None = None |
| status: LedgerStatus |
| run_dir: str |
| kind: str = Field(default="attempt") |
|
|
| def summary(self) -> str: |
| value = "n/a" if self.value is None else f"{self.value:.4f}" |
| return f"champion commit={self.commit} value={value} status={self.status}" |
|
|
|
|
| def _terminate_group(pid: int, *, grace_s: float = 3.0) -> None: |
| """SIGTERM the process group, wait briefly, then SIGKILL any survivors.""" |
| try: |
| pgid = os.getpgid(pid) |
| except OSError: |
| return |
| try: |
| os.killpg(pgid, signal.SIGTERM) |
| except OSError: |
| return |
| deadline = time.monotonic() + grace_s |
| while time.monotonic() < deadline: |
| try: |
| os.killpg(pgid, 0) |
| except OSError: |
| return |
| time.sleep(0.1) |
| try: |
| os.killpg(pgid, signal.SIGKILL) |
| except OSError: |
| pass |
|
|
|
|
| def run_logged(command: str, log_path: Path, timeout_s: float, *, cwd: Path | None = None) -> int: |
| """Run `command` (shell), tee output to `log_path`, return exit code (124 on timeout).""" |
| log_path.parent.mkdir(parents=True, exist_ok=True) |
| with log_path.open("w", encoding="utf-8") as log: |
| proc = subprocess.Popen( |
| command, |
| shell=True, |
| cwd=str(cwd) if cwd is not None else None, |
| stdout=log, |
| stderr=subprocess.STDOUT, |
| start_new_session=True, |
| text=True, |
| ) |
| try: |
| return proc.wait(timeout=timeout_s) |
| except subprocess.TimeoutExpired: |
| _terminate_group(proc.pid) |
| try: |
| proc.wait(timeout=5.0) |
| except subprocess.TimeoutExpired: |
| pass |
| return _TIMEOUT_EXIT |
|
|
|
|
| def _format_command(contract: AttemptContract, run_dir: Path, run_name: str) -> str: |
| return ( |
| contract.command.replace("{run_dir}", str(run_dir)) |
| .replace("{run_name}", run_name) |
| .replace("{attempt_timeout_minutes}", str(contract.attempt_timeout_minutes)) |
| ) |
|
|
|
|
| def _read_last_jsonl(path: Path, key: str) -> float | None: |
| if not path.exists(): |
| return None |
| last: object = None |
| for line in path.read_text(encoding="utf-8").splitlines(): |
| stripped = line.strip() |
| if not stripped: |
| continue |
| try: |
| last = json.loads(stripped) |
| except ValueError: |
| continue |
| if not isinstance(last, dict): |
| return None |
| val = last.get(key) |
| return float(val) if isinstance(val, (int, float)) else None |
|
|
|
|
| def read_metric(contract: AttemptContract, run_dir: Path) -> float | None: |
| """Resolve the contract's scalar from the artifact the command wrote; None if absent.""" |
| if contract.source == "metrics_jsonl": |
| return _read_last_jsonl(run_dir / "metrics.jsonl", contract.metric) |
| filename, fixed_key = _SOURCE_ARTIFACT[contract.source] |
| path = run_dir / filename |
| if not path.exists(): |
| return None |
| try: |
| data = json.loads(path.read_text(encoding="utf-8")) |
| except ValueError: |
| return None |
| if not isinstance(data, dict): |
| return None |
| val = data.get(fixed_key or contract.metric) |
| return float(val) if isinstance(val, (int, float)) else None |
|
|
|
|
| def _capture_diff(editable: list[Path], run_dir: Path, root: Path) -> None: |
| """Best-effort: write the just-committed edit to run_dir/code.diff for the UI.""" |
| try: |
| diff = git("diff", "HEAD~1", "HEAD", "--", *[str(p) for p in editable], cwd=root) |
| except Exception: |
| diff = "" |
| (run_dir / "code.diff").write_text(diff, encoding="utf-8") |
|
|
|
|
| def run_attempt( |
| contract: AttemptContract, |
| researcher: str, |
| run_dir: Path, |
| ledger: Ledger, |
| *, |
| root: Path | None = None, |
| kind: str = "attempt", |
| ) -> AttemptResult: |
| """Fence → dedup → run → measure → record. Returns the structured result.""" |
| root = root or repo_root() |
| run_dir = Path(run_dir) |
| run_dir.mkdir(parents=True, exist_ok=True) |
|
|
| commit = assert_clean_and_committed(contract.editable, root) |
| assert_new_commit(commit, researcher, ledger) |
| ledger.append( |
| researcher=researcher, commit=commit, status="running", |
| kind=kind, run_dir=str(run_dir), |
| ) |
|
|
| command = _format_command(contract, run_dir, run_dir.name) |
| code = run_logged( |
| command, run_dir / "logs.log", contract.attempt_timeout_minutes * 60.0, cwd=root |
| ) |
| value = read_metric(contract, run_dir) |
|
|
| if code == _TIMEOUT_EXIT: |
| status: LedgerStatus = "timed_out" |
| elif code == 0 and value is not None: |
| status = "completed" |
| else: |
| status = "failed" |
|
|
| _capture_diff(contract.editable, run_dir, root) |
| ledger.append( |
| researcher=researcher, commit=commit, status=status, |
| kind=kind, run_dir=str(run_dir), |
| metric=AttemptMetric(name=contract.metric, value=value, mode=contract.mode), |
| ) |
| return AttemptResult(commit=commit, value=value, status=status, run_dir=str(run_dir), kind=kind) |
|
|
|
|
| __all__ = ["AttemptResult", "read_metric", "run_attempt", "run_logged"] |
|
|
