"""Daily canary — single-vs-batch reference contract regression check. spec: docs/superpowers/specs/2026-05-08-single-batch-reference-contract-design.md §5.3 plan: docs/superpowers/plans/2026-05-08-single-batch-reference-contract-implementation.md Task 6 Read-only. PID 80f62523 의 5 NG stills 의 latest llm_call_log entry 의 actual ref labels 를 fixture 의 expected 와 비교. 회귀 발견 시 alert (exit 1). 두 검증 모드: - 기본: actual labels 가 expected refs 모두 포함하는지 (정상 generate path) - expect_fail_fast: RPC lookup + dry-run validate_attached_refs 로 fail-fast 가 발동하는지 + 메시지/코드 매칭 (silent-skip 봉쇄 검증). fail-fast 시 새 llm_call_log 가 생성되지 않으므로 actual stale 은 정상. 가드: - 실제 이미지 재생성 X (read-only DB query) - DB 수정 X - API 호출 X - Task 1~5 production 검증 후 사용 Usage: PYTHONPATH=. .venv/bin/python scripts/canary_single_vs_batch_refs.py """ from __future__ import annotations import json import sys from pathlib import Path def main() -> int: backend_dir = Path(__file__).resolve().parent.parent sys.path.insert(0, str(backend_dir)) fixture_path = backend_dir / "tests" / "fixtures" / "c01_zero_gate_regression.json" if not fixture_path.exists(): print(f"FIXTURE MISSING: {fixture_path}", file=sys.stderr) return 1 fixture = json.loads(fixture_path.read_text(encoding="utf-8")) pid = fixture["project_id"] eid = fixture["episode_id"] # Lazy import — fixture path 검증 후 from app.models import project, catalog # noqa: F401 (model registration) from app.core.database import SessionLocal from app.core.ref_contract_validator import ( RefContractError, validate_attached_refs, ) from app.services.scene_generation_coordinator import lookup_render_prompt_card from sqlalchemy import text failures: list[tuple[str, str]] = [] fail_fast_pass: list[tuple[str, str]] = [] with SessionLocal() as db: for entry in fixture["stills"]: sid = entry["still_id"] label = entry["label"] row = db.execute( text(""" SELECT reference_image_ids FROM llm_call_log WHERE project_id = :pid AND episode_id = :eid AND operation_type IN ('single_scene_image_gen', 'scene_image_gen') AND metadata_json::text LIKE :sid_pat ORDER BY created_at DESC LIMIT 1 """), {"pid": pid, "eid": eid, "sid_pat": f'%"still_id": "{sid}"%'}, ).first() if not row: actual: list[str] = [] latest_log_present = False else: latest_log_present = True try: actual = json.loads(row[0] or "[]") except Exception as exc: failures.append((label, f"reference_image_ids parse 실패: {exc}")) continue # ────────────────────────────────────────────────────────────── # expect_fail_fast 분기 — RPC lookup + dry-run validate # ────────────────────────────────────────────────────────────── if entry.get("expect_fail_fast"): expected_code = entry.get("expected_error_code", "ref_contract.violation") expected_msg = entry.get("expected_error_message_contains", "") # 1. RPC lookup (lookup 실패 자체도 fail-fast 형태) try: rpc = lookup_render_prompt_card( project_id=pid, episode_id=eid, scene_index=entry["scene_index"], shot_index=entry["shot_index"], ) except RefContractError as exc: if exc.code == expected_code and expected_msg in str(exc): fail_fast_pass.append(( label, f"RPC lookup fail-fast (code={exc.code}): {exc}", )) continue failures.append(( label, f"expect_fail_fast: RPC lookup raised wrong error " f"code={exc.code} msg={exc}", )) continue # 2. dry-run validate_attached_refs — stale actual labels 로 시뮬 stale_labeled_refs = [(lbl, b"") for lbl in actual] try: validate_attached_refs( rpc, stale_labeled_refs, prompt="", is_close_framing=bool(entry.get("is_close_framing")), ) except RefContractError as exc: if exc.code == expected_code and expected_msg in str(exc): stale_note = entry.get("actual_refs_stale_note") or "" fail_fast_pass.append(( label, f"dry-run validate fail-fast: {exc}" + (f" | stale_note={stale_note}" if stale_note else ""), )) continue failures.append(( label, f"expect_fail_fast: validate raised but mismatch " f"code={exc.code} msg={exc} (expected_code={expected_code} " f"expected_msg_contains={expected_msg!r})", )) continue # validate did NOT raise — fail-fast 회귀 failures.append(( label, f"expect_fail_fast: validate_attached_refs did NOT raise — " f"fail-fast regression (latest_log_present={latest_log_present} " f"stale_actual={actual})", )) continue # ────────────────────────────────────────────────────────────── # 기본 분기 — actual labels 가 expected 포함하는지 # ────────────────────────────────────────────────────────────── if not latest_log_present: failures.append((label, "no llm_call_log entry — generate-image 호출 안 됨?")) continue # S21 은 P0-5 후속 (prop carry) 이라 expected_actual_ref_labels_p0 사용 expected = ( entry.get("expected_actual_ref_labels") or entry.get("expected_actual_ref_labels_p0") or [] ) missing = [e for e in expected if e not in actual] if missing: failures.append(( label, f"expected refs missing: {missing} | actual={actual}", )) if fail_fast_pass: print(f"CANARY fail-fast PASS: {len(fail_fast_pass)} stills") for label, note in fail_fast_pass: print(f" {label}: {note}") if failures: print(f"CANARY FAILURE: {len(failures)} stills regressed") for label, reason in failures: print(f" {label}: {reason}") return 1 print( f"CANARY OK: {len(fixture['stills'])} stills contract preserved " f"({len(fail_fast_pass)} fail-fast / " f"{len(fixture['stills']) - len(fail_fast_pass)} normal-generate)" ) return 0 if __name__ == "__main__": sys.exit(main())