"""A/B 두 arm — **격리 clone 둘**(DB·artifact·run_id 분리)에서 같은 샷을 다시 그린다.
arm A = 자동 선택된 고증 사진(sidecar)이 붙은 판 그대로 (HITL 0 · 사람 판정은 조건이 아니다).
arm B = clone 에서 **그 참조 줄들(예: LP01#detail·LP03#detail)만 날것 CP 에서 빼고** production helper 로
sidecar 를 다시 투영한 판 — 카드 본문·설정은 같고, 다른 것은 grounding 부착(bytes·role·required)뿐.
★2026-09-03: 리뷰 삭제가 아니다 — production 은 리뷰를 안 읽으므로 리뷰를 지워도 아무것도 안 바뀐다.
★원본 run(69e821758f3d)의 DB·장부·판정 표는 **한 글자도 안 건드린다**. clone 은 `CREATE DATABASE
… TEMPLATE …` + `projects/` 복사다. 두 arm 은 서로 다른 DB·artifact·run_id 라 병렬로 돈다.
★판정은 사람만 한다 — 이 도구는 판을 차리고(순서 가린 갤러리) 기록만 남긴다.
부명령:
clone --src RID --arm A|B [--ablate LP01#detail,LP03#detail] [--shots 1:1,2:4]
capture --run-id RID --shots 1:1,2:4 (무료 · provider 직전 payload)
compare --a RID --b RID --shots 1:1,2:4 (무료 · 텍스트 동일 + 차이는 grounding 부착뿐)
generate --run-id RID --shots 1:1,2:4 --repeats 2 --text-cap 40 --image-cap 12 (유료)
gallery --a RID --b RID --shots 1:1,2:4 --out DIR
"""
from __future__ import annotations
import argparse
import hashlib
import json
import os
import random
import shutil
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
KST = timezone(timedelta(hours=9))
AB_CONTRACT = "1.202609021900"
#: stage2a 와 같은 조건 — 두 arm 공통. process 앞에 박는다.
LANE_ENV = {"BACKGROUND_MODE": "off", "STILL_RECIPE_MODE": "off",
"OUTDOOR_LANE_PLAN_ENABLED": "false", "OUTDOOR_LANE_PIPE_ENABLED": "false",
"OUTDOOR_DIRECT_COMPOSE_ENABLED": "false", "OUTDOOR_MAP_CONTI_ENABLED": "false"}
ARM_SUFFIX = {"A": "aa", "B": "bb"}
def _now() -> str:
return datetime.now(KST).isoformat(timespec="seconds")
def _shots(spec: str) -> List[Tuple[int, int]]:
out = []
for x in str(spec).split(","):
x = x.strip()
if not x:
continue
a, b = x.split(":")
out.append((int(a), int(b)))
return out
def clone_rid(src: str, arm: str) -> str:
return f"{src}{ARM_SUFFIX[arm]}"
# ── clone ────────────────────────────────────────────────────────────
def clone_database(src_rid: str, dst_rid: str) -> Dict[str, Any]:
"""`CREATE DATABASE dst TEMPLATE src`. ★있으면 안 덮는다 · 원본에 연결이 있으면 선다."""
import psycopg2
from sqlalchemy.engine import make_url
from tools.grounding_audit import canary_isolation as ci
src_db, dst_db = ci.db_name(src_rid), ci.db_name(dst_rid)
admin = make_url(ci.template_url()).set(database="postgres")
con = psycopg2.connect(admin.render_as_string(hide_password=False))
con.autocommit = True
try:
cur = con.cursor()
cur.execute("SELECT 1 FROM pg_database WHERE datname = %s", (dst_db,))
if cur.fetchone():
raise RuntimeError(f"{dst_db} 가 이미 있다 — 덮지 않는다")
cur.execute("SELECT count(*) FROM pg_stat_activity WHERE datname = %s", (src_db,))
n = int(cur.fetchone()[0])
if n:
raise RuntimeError(f"원본 {src_db} 에 연결 {n}개 — TEMPLATE 복제는 연결 0 이어야 한다")
cur.execute(f'CREATE DATABASE "{dst_db}" TEMPLATE "{src_db}"')
cur.close()
finally:
con.close()
return {"src_db": src_db, "dst_db": dst_db}
def clone_artifacts(src_rid: str, dst_rid: str) -> Dict[str, Any]:
from tools.grounding_audit import canary_isolation as ci
src, dst = ci.root_dir(src_rid), ci.root_dir(dst_rid)
if dst.exists():
raise RuntimeError(f"{dst} 가 이미 있다 — 덮지 않는다")
shutil.copytree(src / "projects", dst / "projects")
return {"src_root": str(src), "dst_root": str(dst)}
def ablate_reviews(dst_rid: str, subjects: List[str]) -> int:
"""clone 의 판정 표에서 **그 대상만** 지운다. ★clone(arm B) 이름이 아니면 선다."""
import psycopg2
from tools.grounding_audit import canary_isolation as ci
if not dst_rid.endswith(ARM_SUFFIX["B"]):
raise RuntimeError(f"{dst_rid} 는 arm B clone 이 아니다 — 판정을 지우지 않는다")
con = psycopg2.connect(ci.db_url(dst_rid))
try:
cur = con.cursor()
cur.execute("SELECT current_database()")
now = str(cur.fetchone()[0])
if now != ci.db_name(dst_rid):
raise RuntimeError(f"붙고 보니 {now!r} 다 — clone 이 아니면 안 지운다")
cur.execute("DELETE FROM grounding_reference_fidelity_review WHERE research_subject_id = ANY(%s)",
(list(subjects),))
n = cur.rowcount
con.commit()
cur.close()
finally:
con.close()
return int(n)
def ablate_rows(cp: Dict[str, Any], subjects: List[str]) -> Dict[str, Any]:
"""★HITL 0 (2026-09-03): arm B 는 **참조 줄 제외**다 — production 은 리뷰를 안 읽으므로 리뷰를
지워도 아무것도 안 바뀐다. 그 대상의 줄을 CP 사본에서 빼면 sidecar 가 그 참조를 못 붙인다.
★없는 대상을 지목하면 선다."""
import copy
out = copy.deepcopy(cp)
rows = list(((out.get("data") or {}).get("rows")) or [])
have = {str(r.get("research_subject_id") or "") for r in rows}
missing = [s for s in subjects if s not in have]
if missing:
raise RuntimeError(f"CP 에 없는 대상을 뺐다고 적을 뻔했다: {missing}")
out["data"]["rows"] = [r for r in rows if str(r.get("research_subject_id") or "") not in set(subjects)]
out["data"]["ab_ablated_subjects"] = list(subjects)
return out
def reproject_sidecar(dst_rid: str, project_id: str, episode_id: str,
shots: List[Tuple[int, int]], *,
exclude_subjects: Optional[List[str]] = None) -> Dict[str, Any]:
"""clone 의 scene_detail 카드에 **production helper 로** sidecar 를 다시 투영한다.
★바뀌는 키가 grounding 부착(members·required) 밖이면 선다."""
from app.core.database import SessionLocal, register_models
from app.modules.pipeline import grounding_fidelity_review as fr
from app.modules.pipeline import grounding_reference_bundle as rb
from app.modules.pipeline import grounding_sidecar_writer as sw
from app.modules.pipeline.grounding_canonical_ref_inputs import _CheckpointReader
# ★HITL 0 (2026-09-03): production 과 같은 **날것 CP**. arm B 는 리뷰 삭제가 아니라 참조 줄 제외다.
cp_path = (Path(os.environ["PROJECTS_DIR"]) / project_id / "checkpoints" / "episodes"
/ episode_id / "reference_acquisition" / "manifest.json")
cp = json.loads(cp_path.read_text(encoding="utf-8"))
sw.assert_central_checkpoint(cp)
if exclude_subjects:
cp = ablate_rows(cp, list(exclude_subjects))
root = sw.default_reference_root()
p = (Path(os.environ["PROJECTS_DIR"]) / project_id / "checkpoints" / "episodes"
/ episode_id / "scene_detail" / "manifest.json")
d = json.loads(p.read_text(encoding="utf-8"))
allowed = {rb.RPC_MEMBERS_KEY, rb.RPC_REQUIRED_KEY}
diff: Dict[str, Any] = {}
for card in d["data"]["scenes"]:
key = (int(card.get("scene_index")), int(card.get("_shot_index") or 0))
if key not in shots:
continue
rpc = card["render_prompt_card"]
before = json.loads(json.dumps(rpc))
sw.write_for_shot(rpc, cp, card.get("visible_entities") or [],
content_sha_of=lambda r: sw.row_content_sha256(r, root=root),
coordinate_of=lambda r: sw.row_file_coordinate(r, root=root))
changed = sorted(k for k in set(before) | set(rpc) if before.get(k) != rpc.get(k))
if set(changed) - allowed:
raise RuntimeError(f"{key}: 허용 밖 키가 바뀌었다 {sorted(set(changed) - allowed)}")
diff[f"{key[0]}:{key[1]}"] = {
"changed_keys": changed,
"members_before": [(m.get("subject_final_id"), m.get("member_identity")) for m in before.get(rb.RPC_MEMBERS_KEY) or []],
"members_after": [(m.get("subject_final_id"), m.get("member_identity")) for m in rpc.get(rb.RPC_MEMBERS_KEY) or []],
"required_before": before.get(rb.RPC_REQUIRED_KEY), "required_after": rpc.get(rb.RPC_REQUIRED_KEY),
}
p.write_text(json.dumps(d, ensure_ascii=False), encoding="utf-8")
return diff
# ── capture / compare ───────────────────────────────────────────────
def capture(rid: str, project_id: str, episode_id: str, shots: List[Tuple[int, int]]) -> Dict[str, Any]:
from tools.grounding_audit import canary_exit_probe as probe
from tools.grounding_audit import canary_isolation as ci
out = {}
root = ci.root_dir(rid)
for si, sh in shots:
got = probe.run(run_id=rid, project_id=project_id, episode_id=episode_id,
scene_index=si, shot_index=sh, actor_id="canary-ab-capture")
(root / f"ab_capture_s{si}_sh{sh}.json").write_text(json.dumps(got, ensure_ascii=False, indent=2), encoding="utf-8")
parts = [p for c in got.get("captured") or [] for p in c["parts"]]
out[f"{si}:{sh}"] = {"stopped_at": got.get("stopped_at"),
"text_sha256": hashlib.sha256("\n".join(p.get("full_text") or p.get("head") or "" for p in parts if p["kind"] == "text").encode("utf-8")).hexdigest(),
"inline_shas": sorted(p["sha256"] for p in parts if p["kind"] == "inline"),
"text_parts": sum(1 for p in parts if p["kind"] == "text")}
return out
def compare(a: Dict[str, Any], b: Dict[str, Any], grounding_shas: List[str]) -> Dict[str, Any]:
out = {}
for k in a:
ta, tb = a[k]["text_sha256"], b[k]["text_sha256"]
ia, ib = set(a[k]["inline_shas"]), set(b[k]["inline_shas"])
out[k] = {"text_identical": ta == tb, "a_minus_b": sorted(ia - ib), "b_minus_a": sorted(ib - ia),
"a_minus_b_is_grounding_only": bool(ia - ib) and (ia - ib) <= set(grounding_shas),
"ok": ta == tb and not (ib - ia) and (ia - ib) <= set(grounding_shas)}
return out
# ── generate (유료) ─────────────────────────────────────────────────
def generate(rid: str, project_id: str, episode_id: str, shots: List[Tuple[int, int]], *,
repeats: int, text_cap: int, image_cap: int) -> Dict[str, Any]:
from tools.grounding_audit import canary_isolation as ci
from tools.grounding_audit.canary_image_budget import canary_image_scope
from tools.grounding_audit.canary_outbound_gates import canary_outbound_scope
from tools.grounding_audit.canary_text_budget import canary_text_scope
ci.assert_engine_is_canary(rid)
from app.core.database import SessionLocal, register_models
register_models()
from app.models.project import SceneStill
from app.services.scene_image_service import SceneImageService
root = ci.root_dir(rid)
ledger = root / "ab_ledger.json"
rows = json.loads(ledger.read_text(encoding="utf-8")) if ledger.is_file() else []
if any(r.get("status") == "running" for r in rows):
raise RuntimeError("열린 attempt 가 있다 — 얼마 썼는지 모르는 채 안 산다")
used_img = sum(int(r.get("image_used") or 0) for r in rows)
if used_img >= image_cap:
raise RuntimeError(f"이 arm 의 이미지 누계 {used_img} ≥ cap {image_cap}")
attempt = {"contract": AB_CONTRACT, "attempt_id": hashlib.sha256(f"{rid}:{_now()}".encode()).hexdigest()[:12],
"status": "running", "started_kst": _now(), "shots": [f"{a}:{b}" for a, b in shots],
"repeats": repeats, "text_cap": text_cap, "image_cap": image_cap - used_img, "images": []}
rows.append(attempt)
ledger.write_text(json.dumps(rows, ensure_ascii=False, indent=2), encoding="utf-8")
def _save():
ledger.write_text(json.dumps(rows, ensure_ascii=False, indent=2), encoding="utf-8")
db = SessionLocal()
try:
# ★actor 는 **있는 계정**이어야 한다 — activity_log 의 외래 열쇠. 실측 2026-09-02:
# 지어낸 "canary-ab" 로 첫 이미지를 만든 뒤 flush 에서 죽었다(이미지 1장 헛삼).
from app.models.catalog import UserAccount
actor = db.query(UserAccount).order_by(UserAccount.created_at).first()
if actor is None:
raise RuntimeError("user_account 가 비어 있다 — actor 없이 안 산다")
svc = SceneImageService(db, project_id, str(actor.id))
with canary_text_scope(cap=text_cap) as tb, \
canary_image_scope(cap=image_cap - used_img) as ib, \
canary_outbound_scope(search_cap=0, download_cap=0):
try:
for si, sh in shots:
still = (db.query(SceneStill).filter(SceneStill.episode_id == episode_id,
SceneStill.scene_index == si,
SceneStill.shot_index == sh).first())
if still is None:
raise RuntimeError(f"씬 {si} 샷 {sh} still 없음")
for r in range(repeats):
got = svc.generate_single_scene_image(str(still.id))
attempt["images"].append({"shot": f"{si}:{sh}", "repeat": r, "asset_id": got.get("id"),
"file_path": got.get("file_path"), "at": _now()})
attempt["text_used"] = int(tb.snapshot().get("used", 0))
attempt["image_used"] = int(ib.snapshot().get("used", 0))
_save()
attempt["status"] = "completed"
except BaseException as exc: # noqa: BLE001
attempt["status"] = "crashed"
attempt["error"] = str(exc)[:400]
raise
finally:
attempt["text_used"] = int(tb.snapshot().get("used", 0))
attempt["image_used"] = int(ib.snapshot().get("used", 0))
attempt["finished_kst"] = _now()
_save()
finally:
db.close()
return attempt
# ── gallery (순서 가림) ─────────────────────────────────────────────
def gallery(a_rid: str, b_rid: str, shots: List[Tuple[int, int]], out: Path, *, seed: int = 20260902) -> Dict[str, Any]:
from tools.grounding_audit import canary_isolation as ci
def _images(rid):
led = json.loads((ci.root_dir(rid) / "ab_ledger.json").read_text(encoding="utf-8"))
got: Dict[str, List[str]] = {}
root = ci.root_dir(rid)
for att in led:
for im in att.get("images") or []:
fp = im.get("file_path")
if fp:
# ★자산 경로는 run root 기준 상대 경로 — cwd 로 풀지 않는다
got.setdefault(im["shot"], []).append(str(fp if Path(fp).is_absolute() else root / fp))
return got
ia, ib = _images(a_rid), _images(b_rid)
out.mkdir(parents=True, exist_ok=True)
(out / "img").mkdir(exist_ok=True)
rng = random.Random(seed)
key: List[Dict[str, Any]] = []
html = ['', f'A/B 순서 가린 판정 · {a_rid[:-2] if a_rid.endswith(ARM_SUFFIX["A"]) else a_rid}',
'',
'
A/B 판정 (순서 가림) — 씬 1 샷 1 · 씬 2 샷 4
',
'
같은 카드·같은 설정·같은 인물 참조. 한쪽만 사람이 확인한 고증 사진(LP01 나무 미닫이문 / LP03)이 요청에 실렸다. '
'어느 쪽이 고증에서, 어느 쪽이 전체 품질에서 나은지 쌍마다 적어 주세요(왼쪽/오른쪽/차이 없음). 정답표는 페이지 밖 ab_key.json.
']
n = 0
for si, sh in shots:
k = f"{si}:{sh}"
pa, pb = ia.get(k, []), ib.get(k, [])
for i in range(min(len(pa), len(pb))):
n += 1
left_is_a = rng.random() < 0.5
left, right = (pa[i], pb[i]) if left_is_a else (pb[i], pa[i])
ln, rn = f"pair{n}_left.png", f"pair{n}_right.png"
shutil.copy(left, out / "img" / ln); shutil.copy(right, out / "img" / rn)
key.append({"pair": n, "shot": k, "repeat": i, "left": "A" if left_is_a else "B",
"left_file": left, "right_file": right})
html.append(f'