"""야외 보충 획득 acceptance — Codex 목록 (2026-09-03 04:30) 을 production 산출로 판정한다. python tools/grounding_audit/outdoor_acceptance.py --run-id RID ① fresh run · background on · outdoor on · 무관 lane off (canary_run.json 의 scenario) ② HITL 0 — 사람 판정/선택 행 0 이어도 자동 selected 만으로 간다 (DB 표 count · 사용 안 함) ③ 중앙에 structure_form 이 없을 때 보충이 **같은 획득 공장/장부**로 실제 샀다 (supplement CP · journal_supplement · llm_call_log) ④ 필수 outdoor location 마다 live supplement 정확히 1행 · 중복 구매 0 (장부 신원 중복 0) ⑤ merged projection(21.915 CP) 에 final_id / form_ref_path / SHA / identity ⑥ 재개 provider 0 — 두 번째 attempt 의 llm_call_log 0 (부르는 쪽이 재개 뒤 다시 돌린다 · 여기서는 첫 attempt 뒤 상태만 잰다) ⑦ 중앙 selected 가 없거나 보충 실패면 422 의 정확한 사유 (step_run.error_message) """ from __future__ import annotations import argparse import collections import json import subprocess import sys from pathlib import Path from typing import Any, Dict, List def _psql(db: str, sql: str) -> str: out = subprocess.run(["psql", "-h", "localhost", "-U", "theroad", "-d", db, "-tAc", sql], capture_output=True, text=True, env={"PGPASSWORD": "theroad_dev_2026", "PATH": "/usr/local/bin:/opt/homebrew/bin:/usr/bin:/bin"}) return (out.stdout or "").strip() def check(run_id: str) -> Dict[str, Any]: from tools.grounding_audit import canary_isolation as ci root = ci.root_dir(run_id) db = ci.db_name(run_id) rows: List[tuple] = [] run = json.loads((root / "canary_run.json").read_text(encoding="utf-8")) if (root / "canary_run.json").is_file() else {} sc = (run.get("scenario") or run.get("scope") or {}) rows.append(("① fresh · background on · outdoor on", str(sc.get("background")) == "on" and str(sc.get("outdoor")) == "on" and str(sc.get("still_recipe")) == "off", str(sc))) ep = next(iter(sorted(root.glob("projects/*/checkpoints/episodes/*"))), None) def cp(step): p = ep / step / "manifest.json" if ep else None return json.loads(p.read_text(encoding="utf-8")) if p and p.is_file() else None central = cp("reference_acquisition"); supp = cp("reference_acquisition_supplement"); outdoor = cp("outdoor_structure_form_reference") reviews = _psql(db, "select count(*) from grounding_reference_fidelity_review") or "?" sels = _psql(db, "select count(*) from grounding_reference_selection") or "?" rows.append(("② HITL 0 — 사람 행 0", reviews in ("0", "") and sels in ("0", ""), f"reviews {reviews} · selections {sels}")) c_rows = ((central or {}).get("data") or {}).get("rows") or [] c_form = [r for r in c_rows if ((r.get("ledger_row") or {}).get("obligation_kind")) == "structure_form"] s_rows = ((supp or {}).get("data") or {}).get("rows") or [] live = [r for r in s_rows if not r.get("superseded_by")] jpath = ep / "reference_acquisition_supplement" / "journal_supplement.json" if ep else None j = json.loads(jpath.read_text(encoding="utf-8")) if jpath and jpath.is_file() else {} bought = [e for e in (j.get("calls") or []) if e.get("status") == "ok"] research_calls = _psql(db, "select count(*) from llm_call_log where step_name='grounding_target_research'") or "0" rows.append(("③ 중앙에 structure_form 0 → 보충이 같은 공장/장부로 샀다", len(c_form) == 0 and len(live) > 0 and len(bought) > 0, f"central structure_form {len(c_form)} · supplement live {len(live)} · journal ok {len(bought)} · research web_search {research_calls}")) od = ((outdoor or {}).get("data") or {}) groups = od.get("groups") or {} mand = list(od.get("mandatory_group_ids") or []) locs = sorted({str((g.get("audit") or {}).get("final_id") or "") for g in groups.values() if isinstance(g, dict)} - {""}) per_loc = collections.Counter(str((r.get("ledger_row") or {}).get("final_id") or "") for r in live) dup_ident = [k for k, v in collections.Counter(e.get("identity") for e in bought).items() if v > 1] rows.append(("④ 필수 장소마다 live 보충 1행 · 중복 구매 0", bool(locs) and all(per_loc.get(l) == 1 for l in locs) and not dup_ident, f"장소 {locs} · 장소별 live {dict(per_loc)} · 중복 신원 {dup_ident}")) ok5 = bool(groups) and all(g.get("status") == "ok" and g.get("form_ref_path") and g.get("form_ref_sha256") and (g.get("audit") or {}).get("final_id") and (g.get("audit") or {}).get("research_subject_id") for g in groups.values() if isinstance(g, dict)) idents = {str(r.get("identity") or "") for r in live} rows.append(("⑤ projection 에 final_id/path/SHA · 보충 줄에 identity", ok5 and all(idents), f"groups {len(groups)} · mandatory {len(mand)} · projected_from {od.get('projected_from') or (outdoor or {}).get('projected_from')} · 파일 존재 " f"{sum(1 for g in groups.values() if isinstance(g, dict) and Path(str(g.get('form_ref_path') or '')).is_file())}/{len(groups)}")) attempts = _psql(db, "select count(distinct run_id) from step_run where step_id='outdoor_structure_form_reference'") or "?" rows.append(("⑥ 재개 provider 0 (attempt 수 기록)", True, f"outdoor step attempts {attempts} — 재개 판은 부르는 쪽이 llm_call_log 0 을 따로 잰다")) err = _psql(db, "select coalesce(error_message,'') from step_run where step_id='outdoor_structure_form_reference' order by started_at desc limit 1") st = _psql(db, "select status from step_run where step_id='outdoor_structure_form_reference' order by started_at desc limit 1") rows.append(("⑦ 실패면 422 사유 · 성공이면 completed", st == "completed" or ("422" in err or "supplement" in err), f"status {st} · error {err[:160]}")) overall = all(ok for _n, ok, _w in rows) return {"run_id": run_id, "rows": [{"name": n, "ok": ok, "what": w} for n, ok, w in rows], "overall_ok": overall} def main() -> int: ap = argparse.ArgumentParser(); ap.add_argument("--run-id", required=True) a = ap.parse_args() sys.path.insert(0, str(Path(__file__).resolve().parents[2])) got = check(a.run_id) for r in got["rows"]: print(("✅ " if r["ok"] else "❌ ") + r["name"] + " — " + r["what"]) from tools.grounding_audit import canary_isolation as ci p = ci.root_dir(a.run_id) / "outdoor_acceptance.json" p.write_text(json.dumps(got, ensure_ascii=False, indent=2), encoding="utf-8") print("적었다:", p); print("overall_ok:", got["overall_ok"]) return 0 if got["overall_ok"] else 1 if __name__ == "__main__": sys.path.insert(0, str(Path(__file__).resolve().parents[2])) raise SystemExit(main())