"""canary fixture 를 **production 스텝으로** 세우는 입구. ★문을 다 지난 뒤에만. Codex 결정 (2026-08-31) — ① 전용 fixture project. > 격리된 projects_dir/DB namespace, **실제 production step** 으로 scene_detail > 까지 생성. sidecar 는 production `write_sidecar` 한 벌로만 작성. > 원본 프로젝트/CP/DB 쓰기 **0**. ## 문 (순서가 계약이다) ①격리 `assert_isolated` — DB·디렉토리가 원본이 아니고 run_id 를 갖는다 ②DB `create_database` — 관리자 연결은 maintenance DB 여야 한다 ③schema `canary_alembic` — 별도 프로세스 · 연결 안에서 DB 확인 · head ④상한 `canary_text_scope(cap=)` — **모든 스레드**의 글 호출을 센다 ⑤부트스트랩 프로젝트·에피소드 (★프로젝트 이름 짓기에 **유료 1회**) ★이 파일은 **아무것도 안 산다.** 부르는 쪽이 `--live` 를 줘야 산다. ## 운영자가 줘야 하는 것 (코드에 안 박는다) THEROAD_CANARY_ADMIN_DSN maintenance DB(postgres) 관리자 연결 THEROAD_CANARY_TEMPLATE_URL 접속 정보의 본 (DB 이름은 바꿔 쓴다) """ from __future__ import annotations import os import sys import uuid from pathlib import Path from typing import Any, Dict, Optional BACKEND = Path(__file__).resolve().parents[2] sys.path.insert(0, str(BACKEND)) sys.path.insert(0, str(BACKEND / "tests")) from tools.grounding_audit import canary_isolation as ci # noqa: E402 def new_run_id() -> str: """★`RUN_ID_RE` 가 받는 꼴로만 만든다.""" return uuid.uuid4().hex[:12] def prepare_env(run_id: str, *, grounding_mode: Optional[str] = None) -> Dict[str, str]: """이 판이 쓸 환경. ★원본을 가리키는 것은 **안 넣는다**. Args: grounding_mode: 이 판이 돌 고증 모드. 주면 `GROUNDING_MODE` 를 **명시로** 박는다. ★★안 주면 바깥 `.env` 값이 그대로 들어온다 — 그러면 계획표는 한 모드를 적고 pipeline 은 다른 모드로 도는, 이미 한 번 겪은 자리가 된다(`visual_continuity_anchor` 실측 2026-09-01). ★값은 production resolver 로 **검사해서** 넣는다 — 오타면 여기서 선다. 조용히 legacy 로 안 떨어진다. """ root = ci.root_dir(run_id) (root / "projects").mkdir(parents=True, exist_ok=True) env = dict(os.environ) env["DATABASE_URL"] = ci.db_url(run_id) env["PROJECTS_DIR"] = str(root / "projects") env["THEROAD_CANARY_RUN_ID"] = run_id if grounding_mode is not None: from app.core.grounding_mode import resolve_grounding_mode env["GROUNDING_MODE"] = resolve_grounding_mode( {"grounding_mode": grounding_mode}) ci.assert_isolated(run_id, env=env) return env #: 쓸 수 있는 원고들. ★이름을 여기 한 번만 적는다. #: ★`canary_one_scene` 은 **배경 묶음 전용 회귀 fixture** 로 보존한다 #: (Codex 2026-09-02) — 지우지 않는다. FIXTURES = ("canary_one_scene", "period_episode", "modern_episode") DEFAULT_FIXTURE = "canary_one_scene" def load_fixture(name: str): """이름 → fixture 모듈. ★목록 밖이면 선다.""" import importlib if name not in FIXTURES: raise ValueError(f"모르는 원고 {name!r} — 있는 것 {FIXTURES}") return importlib.import_module(f"tests.grounding.fixtures.{name}") def manuscript_pdf(path: Path, *, fixture: str = DEFAULT_FIXTURE) -> Path: """fixture 원고를 **PDF 로** 만든다. ★무료 — PyMuPDF 다. `EpisodeService.create_episode` 가 PDF bytes 를 받으므로 production 입구를 그대로 쓰려면 PDF 가 있어야 한다. """ import pymupdf fx = load_fixture(fixture) if hasattr(fx, "assert_shape"): fx.assert_shape() doc = pymupdf.open() page = doc.new_page() # ★★글꼴을 잘못 고르면 **한글이 점으로 바뀐다** (실측: `china-s` 로 # 「이발소 앞」이 「.」 이 됐다). 원고가 통째로 새는 자리다. page.insert_textbox(pymupdf.Rect(48, 48, 548, 780), fx.manuscript(), fontsize=11, fontname="korea") path.parent.mkdir(parents=True, exist_ok=True) doc.save(str(path)) doc.close() return path def ensure_user(db: Any) -> Any: """canary 전용 사용자. ★원본 사용자를 안 쓴다.""" from app.models.catalog import UserAccount now = __import__("datetime").datetime.utcnow().isoformat() got = db.query(UserAccount).filter( UserAccount.username == "canary").first() if got is not None: return got got = UserAccount(id=uuid.uuid4().hex[:12], username="canary", display_name="canary", password_hash="x", role="creator", is_active=1, created_at=now, updated_at=now) db.add(got) db.commit() return got def bootstrap(run_id: str, *, live: bool, cap: Optional[int] = None, fixture: str = DEFAULT_FIXTURE) -> Dict[str, Any]: """④상한 안에서 ⑤프로젝트·에피소드를 만든다. ★`live` 아니면 안 산다. Args: cap: 부트스트랩 **전용** counted 상한. ★없으면 선다. ★★부트스트랩 비용은 pipeline 상한과 **섞지 않는다**. 제 상한·제 장부· **제 부모 trace** 로 끝내고, pipeline 은 새 예산으로 연다. ★★★**재개** (Codex BLOCK 2026-08-31) — 앞 판은 장부를 열기만 하고 매번 같은 신원을 uncertain 으로 **덮은 뒤 다시 샀다**. 이제 — ok 줄이 있으면 DB 와 대조하고 **되쓴다**(구매 0) uncertain/reserved **자동 terminal**(`unknown_not_retried`) · provider 0 · 사람에게 안 묻는다 계약이 바뀌었으면 사기 **전에** 선다 ★`generate_english_name` 은 실패를 잡아 romanization 으로 떨어지므로 「무료였다」로 읽지 않는다 — **예산 delta** 로 본다. """ from app.modules.llm.opik_trace import open_trace from app.modules.pipeline.grounding_chunk_journal import (STATUS_OK, ChunkJournal) from tools.grounding_audit import cc_runner as rr from tools.grounding_audit.canary_text_budget import (canary_text_scope, snapshot_of) if live and not cap: raise ci.IsolationRefused("상한 없이 사지 않는다 — `cap=` 을 줘야 한다") # ★`SessionLocal` 은 import 때 만들어지므로 **격리가 먼저**여야 한다. # 단독으로 부를 때를 위해 여기서도 본다 — 통합 입구는 이미 앞에서 걸었다. if live and "app.core.database" not in sys.modules: ci.assert_database_module_not_loaded() env = prepare_env(run_id) os.environ.update(env) root = ci.root_dir(run_id) pdf = manuscript_pdf(root / "manuscript.pdf", fixture=fixture) if not live: return {"run_id": run_id, "live": False, "pdf": str(pdf), "env_ok": True, "cap": cap, "note": "★안 샀다 — `--live` 와 `cap` 이 있어야 산다"} from app.core.database import SessionLocal from app.services.episode_service import EpisodeService from app.services.project_service import ProjectService jr = ChunkJournal(root / "_bootstrap_journal.json", contract={"kind": "canary_bootstrap", "cap": cap, "run_id": run_id}) if jr.contract_drifted(): raise ci.IsolationRefused( "부트스트랩 장부의 계약이 지금과 다르다 — 사기 전에 선다") ident = f"{run_id}:create_project" # ★①이미 산 판인가 — **DB 와 대조하고** 되쓴다 db = SessionLocal() try: hit = jr.get(ident) if hit and hit.get("project_id"): got = _resume(db, hit) if got is not None: jr.note_reuse() return {"run_id": run_id, "live": True, "cap": cap, **got, "reused": True, "pdf": str(pdf), "journal": str(jr.path), "★means": "앞 판 것을 되썼다 — **한 번도 안 샀다**"} # ★②앞 판이 「샀는지 모른다」로 끝났나 — **자동으로 끝낸다** prev = jr.entries.get(ident) or {} if prev.get("status") in ("uncertain", "reserved"): # ★★장부는 「모른다」인데 **DB 에 이미 있으면** 그 구매는 끝났다. # 다시 사지 않고 이어 쓴다 — 격리된 DB 라 남의 것일 수 없다. # ★★앞 판의 **감사 기록을 덮지 않는다** (Codex 2026-08-31) — # `trace_id` 와 앞 상태를 그대로 이고 가고, 앞 구매를 0으로 # 적지 않는다. 이번 복구가 **새로 산 것이 0**일 뿐이다. found = _reconcile_from_db(db) if found: eid = found.get("episode_id") if eid is None: # ★아직 없으면 **무료로** 만든다 (PDF 에서 글만 뽑는다) user = ensure_user(db) eid = str(EpisodeService( db, found["project_id"], str(user.id)).create_episode( episode_number=1, title="1화", pdf_file_bytes=pdf.read_bytes(), filename=pdf.name).id) made = "만들었다(무료)" else: made = "이미 있어 되썼다" prior = { "status": prev.get("status"), "trace_id": prev.get("trace_id"), "run_id": prev.get("run_id"), "evidence": "canary DB 에 프로젝트가 있다", "logical_dispatch_min": 1, "physical_attempts": ("미측정 — Opik·provider 로그로 " "센다. 0 으로 발명하지 않는다"), } jr.put(ident, {"project_id": found["project_id"], "episode_id": eid, "recovery_new_counted": 0, "prior_purchase": prior}, status=STATUS_OK, meta={"run_id": run_id, # ★앞 판의 결속 키를 **그대로** 이고 간다 "trace_id": prev.get("trace_id"), "prior_status": prev.get("status"), "why": "앞 판이 산 것을 DB 에서 찾아 이어 썼다"}) return {"run_id": run_id, "live": True, "cap": cap, "project_id": found["project_id"], "episode_id": eid, "reconciled": True, "episode": made, "recovery_new_counted": 0, "prior_purchase": prior, "trace_id": prev.get("trace_id"), "name_en": found.get("name_en"), "pdf": str(pdf), "journal": str(jr.path), "★means": ("**이번 복구가** 새로 산 것이 0 이다. " "앞 판의 구매는 `prior_purchase` 에 그대로 " "남아 있다 — 0 으로 덮지 않았다")} jr.put(ident, None, status="unknown_not_retried", meta={"run_id": run_id, "why": "앞 판이 답을 못 받았고 DB 에도 없다 — 재구매 0"}) return {"run_id": run_id, "live": True, "cap": cap, "journal_status": "unknown_not_retried", "bought": 0, "pdf": str(pdf), "journal": str(jr.path), "★means": ("앞 판이 샀는지 모르고 DB 에도 없다 — **다시 안 " "산다**. 사람에게 장부를 고치라고 안 한다")} if not jr.reserve(ident, cap=1): raise ci.IsolationRefused( f"같은 신원 {ident} 을 이미 누가 잡았다 — 두 번 안 산다") user = ensure_user(db) with canary_text_scope(cap=cap) as budget: before = dict(budget.snapshot()) # ★③부모 trace 가 안 열리면 **안 산다** with open_trace(name="canary_bootstrap", tags=[ci.TRACE_TAG], metadata={rr.ID_META_KEY: ident, rr.RUN_META_KEY: run_id}, thread_id=ci.TRACE_THREAD, input_data={"identity": ident}) as tr: tid = getattr(tr, "uid", None) if tr is not None else None if not tid: jr.release(ident) raise rr.TraceUnavailable( "부모 Opik trace 를 못 열었다 — 무엇을 샀는지 못 " "되짚는다. 자리를 놓고 provider 앞에서 선다") jr.put(ident, None, status="uncertain", meta={"run_id": run_id, "trace_id": tid}) proj = ProjectService(db).create_project( name="캐너리 검증 원고", description="참조 묶음 canary", creator_user=user) epi = EpisodeService(db, str(proj.id), str(user.id)).create_episode( episode_number=1, title="1화", pdf_file_bytes=pdf.read_bytes(), filename=pdf.name) after = dict(budget.snapshot()) spent = int(after["used"]) - int(before["used"]) jr.put(ident, {"project_id": str(proj.id), "episode_id": str(epi.id), "counted": spent}, status=STATUS_OK, meta={"run_id": run_id, "trace_id": tid}) snap = snapshot_of(budget) return {"run_id": run_id, "live": True, "cap": cap, "project_id": str(proj.id), "episode_id": str(epi.id), "bootstrap_budget": snap, "name_call_counted": spent, "trace_id": tid, "pdf": str(pdf), "journal": str(jr.path), "★means": ("이 수는 **부트스트랩 몫**이다. pipeline 은 새 " "예산으로 열고 그쪽 상한은 따로다")} finally: db.close() def main() -> int: if len(sys.argv) < 2 or sys.argv[1] not in ("--dry", "--live"): print(__doc__) return 2 rid = sys.argv[2] if len(sys.argv) > 2 else new_run_id() got = bootstrap(rid, live=sys.argv[1] == "--live", cap=int(sys.argv[3]) if len(sys.argv) > 3 else None) print(f"■ run {rid} · {got}") return 0 def _resume(db: Any, hit: Dict[str, Any]) -> Optional[Dict[str, Any]]: """장부의 성공 줄이 **DB 에 실제로 있나**. ★없으면 되쓰지 않는다.""" from app.models.catalog import ProjectRegistry from app.models.project import Episode pid, eid = str(hit.get("project_id") or ""), str(hit.get("episode_id") or "") if not pid or not eid: return None if db.query(ProjectRegistry).filter(ProjectRegistry.id == pid).first() \ is None: return None if db.query(Episode).filter(Episode.id == eid).first() is None: return None return {"project_id": pid, "episode_id": eid, # ★앞 판이 남긴 것을 **그대로** 옮긴다 — 없으면 「모른다」다 "name_call_counted": hit.get("counted"), "prior_purchase": hit.get("prior_purchase"), "recovery_new_counted": hit.get("recovery_new_counted")} def _reconcile_from_db(db: Any) -> Optional[Dict[str, Any]]: """★★장부는 「모른다」인데 **DB 에 이미 있다**면 그것이 답이다. 유료 호출이 나간 **뒤** 다음 걸음에서 죽으면 장부는 `uncertain` 으로 남는다. 그런데 canary DB 에 프로젝트가 있으면 그 구매는 **끝난 것**이다 — 다시 사지 않고 그것을 이어 쓴다 (실측 2026-08-31: `create_project` 는 성공했고 `create_episode` 서명이 틀려 죽었다). ★에피소드도 **같이 본다** (Codex 2026-08-31). 프로젝트만 보고 이어 가면 이미 있는 판에서 **둘째 에피소드를 만들려 든다**. Returns: `{"project_id", "episode_id"|None, "name_en"}`. `episode_id` 가 `None` 이면 아직 안 만든 것이니 부르는 쪽이 **무료로** 만든다. ★이 판단은 **격리된 canary DB** 라서 안전하다. 그 안에 프로젝트가 있다면 이 canary 가 만든 것 말고는 없다. 둘 이상이면 **짐작하지 않는다**. """ from app.models.catalog import ProjectRegistry from app.models.project import Episode rows = db.query(ProjectRegistry).all() if len(rows) != 1: return None # ★둘 이상이면 짐작하지 않는다 pid = str(rows[0].id) eps = db.query(Episode).filter(Episode.project_id == pid).all() if len(eps) > 1: return None # ★여럿이면 어느 것인지 모른다 return {"project_id": pid, "episode_id": (str(eps[0].id) if eps else None), "name_en": getattr(rows[0], "name_en", "")} if __name__ == "__main__": raise SystemExit(main())