1#!/usr/bin/env python3
2"""Daily ops for the autonomous research loop.
3 
4Runs the deterministic gates, the benchmark harness, and writes a dated
5snapshot report. Designed to be idempotent: running twice in a day overwrites
6the same dated snapshot but appends to benchmark history each time.
7"""
8 
9from __future__ import annotations
10 
11import argparse
12import json
13import subprocess
14import sys
15from pathlib import Path
16from typing import Any
17 
18from loop_common import (
19    QUEUE_DIR,
20    REPORTS_DIR,
21    RESEARCHER,
22    ROOT,
23    SNAPSHOTS_DIR,
24    categorize_runs,
25    load_config,
26    load_run_state,
27    read_jsonl,
28    today_utc,
29    utc_now,
30)
31 
32 
33def run_subprocess(cmd: list[str]) -> dict[str, Any]:
34    completed = subprocess.run(cmd, cwd=ROOT, capture_output=True, text=True, check=False)
35    parsed: Any = None
36    if completed.stdout.strip():
37        try:
38            parsed = json.loads(completed.stdout)
39        except json.JSONDecodeError:
40            parsed = None
41    return {
42        "exit_code": completed.returncode,
43        "passed": completed.returncode == 0,
44        "stdout_json": parsed,
45        "stderr": completed.stderr.strip(),
46    }
47 
48 
49def run_repo_validation() -> dict[str, Any]:
50    return run_subprocess([sys.executable, str(RESEARCHER / "scripts" / "validate_repo.py"), "--json"])
51 
52 
53def run_benchmarks() -> dict[str, Any]:
54    return run_subprocess([sys.executable, str(RESEARCHER / "scripts" / "run_benchmarks.py"), "--record", "--json"])
55 
56 
57def run_activation_cases() -> dict[str, Any]:
58    return run_subprocess([sys.executable, str(RESEARCHER / "scripts" / "check_activation_cases.py"), "--json"])
59 
60 
61def run_run_validations() -> dict[str, Any]:
62    buckets = categorize_runs()
63    results: list[dict[str, Any]] = []
64    for run_dir in buckets["active"] + buckets["parked"]:
65        result = run_subprocess(
66            [
67                sys.executable,
68                str(RESEARCHER / "scripts" / "validate_run.py"),
69                "--run-dir",
70                str(run_dir),
71                "--json",
72            ]
73        )
74        results.append({"run_id": run_dir.name, **result})
75    return {
76        "checked": len(results),
77        "passing": sum(1 for record in results if record.get("passed")),
78        "results": results,
79    }
80 
81 
82def queue_snapshot() -> dict[str, Any]:
83    buckets = categorize_runs()
84    inbox = read_jsonl(QUEUE_DIR / "inbox.jsonl")
85    parked = read_jsonl(QUEUE_DIR / "parked.jsonl")
86    done = read_jsonl(QUEUE_DIR / "done.jsonl")
87    quarantine = read_jsonl(QUEUE_DIR / "quarantine.jsonl")
88    return {
89        "inbox": len(inbox),
90        "active": len(buckets["active"]),
91        "parked": len(parked),
92        "closed": len(buckets["closed"]),
93        "done_ledger": len(done),
94        "quarantine": len(quarantine),
95        "unknown": len(buckets["unknown"]),
96    }
97 
98 
99def claims_due_for_review() -> list[dict[str, Any]]:
100    claims_path = RESEARCHER / "claims" / "index.jsonl"
101    if not claims_path.exists():
102        return []
103    today = today_utc()
104    due: list[dict[str, Any]] = []
105    for line in claims_path.read_text(encoding="utf-8").splitlines():
106        if not line.strip():
107            continue
108        try:
109            record = json.loads(line)
110        except json.JSONDecodeError:
111            continue
112        if record.get("volatility") == "high":
113            last_reviewed = record.get("last_reviewed", "")
114            if last_reviewed < today:
115                due.append({"claim_id": record.get("claim_id"), "owning_skill": record.get("owning_skill"), "last_reviewed": last_reviewed})
116    return due
117 
118 
119def render_snapshot(
120    timestamp: str,
121    repo: dict[str, Any],
122    activation: dict[str, Any],
123    benchmarks: dict[str, Any],
124    run_validations: dict[str, Any],
125    queue: dict[str, Any],
126    claims: list[dict[str, Any]],
127) -> str:
128    lines = [
129        f"# Researcher Snapshot {timestamp}",
130        "",
131        "## Queue",
132        f"- inbox: {queue['inbox']}",
133        f"- active runs: {queue['active']}",
134        f"- parked runs: {queue['parked']}",
135        f"- closed runs (state): {queue['closed']}",
136        f"- done ledger entries: {queue['done_ledger']}",
137        f"- quarantined sources: {queue['quarantine']}",
138        f"- unknown run state: {queue['unknown']}",
139        "",
140        "## Gates",
141        f"- repo validation: {'passed' if repo.get('passed') else 'failed'}",
142        f"- activation cases: {'passed' if activation.get('passed') else 'failed'}",
143        f"- benchmark harness: {'passed' if benchmarks.get('passed') else 'failed'}",
144        f"- per-run validations: {run_validations['passing']}/{run_validations['checked']} passing",
145        "",
146        "## Volatile Claims Due For Review",
147    ]
148    if not claims:
149        lines.append("- none")
150    else:
151        for claim in claims:
152            lines.append(
153                f"- {claim['claim_id']} ({claim['owning_skill']}) last reviewed {claim['last_reviewed'] or 'never'}"
154            )
155    lines.extend(
156        [
157            "",
158            "## Parked Runs",
159        ]
160    )
161    parked = read_jsonl(QUEUE_DIR / "parked.jsonl")
162    if not parked:
163        lines.append("- none")
164    else:
165        for record in parked:
166            lines.append(f"- {record.get('run_id')}: {record.get('reason')} (parked {record.get('parked_at')})")
167    return "\n".join(lines) + "\n"
168 
169 
170def daily(config: dict[str, Any]) -> dict[str, Any]:
171    timestamp = utc_now()
172    today = today_utc()
173    repo = run_repo_validation()
174    activation = run_activation_cases()
175    benchmarks = run_benchmarks()
176    run_validations = run_run_validations()
177    queue = queue_snapshot()
178    claims = claims_due_for_review()
179 
180    snapshot_text = render_snapshot(timestamp, repo, activation, benchmarks, run_validations, queue, claims)
181    snapshot_path = SNAPSHOTS_DIR / f"{today}.md"
182    snapshot_path.parent.mkdir(parents=True, exist_ok=True)
183    snapshot_path.write_text(snapshot_text, encoding="utf-8")
184 
185    summary = {
186        "timestamp": timestamp,
187        "snapshot": str(snapshot_path.relative_to(ROOT)),
188        "passing": all([repo.get("passed"), activation.get("passed"), benchmarks.get("passed")]),
189        "queue": queue,
190        "claims_due_for_review": len(claims),
191        "run_validations": {
192            "checked": run_validations["checked"],
193            "passing": run_validations["passing"],
194        },
195    }
196    return summary
197 
198 
199def main() -> int:
200    parser = argparse.ArgumentParser(description="Run daily ops for the autonomous research loop")
201    parser.add_argument("--json", action="store_true")
202    args = parser.parse_args()
203 
204    config = load_config()
205    result = daily(config)
206    if args.json:
207        print(json.dumps(result, indent=2))
208    else:
209        print(
210            f"daily: snapshot={result['snapshot']} "
211            f"queue=inbox{result['queue']['inbox']}/active{result['queue']['active']}/parked{result['queue']['parked']} "
212            f"all_gates_pass={result['passing']}"
213        )
214    return 0 if result["passing"] else 1
215 
216 
217if __name__ == "__main__":
218    sys.exit(main())
219