"""재사용된 outlook canon 이 **이 에피소드** link 를 받는가 (#108, 2026-08-29). ## 무엇이 결함이었나 — 실측 episode 0aea12c2 `scene_image_pipeline` 7샷 중 5샷이 같은 줄로 죽었다: still_recipe S1sh8: outfit_assignments invalid: unknown outlook_id: 'O02' — fail-closed O01/O02 는 **멀쩡한 canon** 이었고 `character_outlook` 쌍(C01→O01, C02→O02)도 유효했다. 끊긴 자리는 하나였다 — `entity_episode_link` 에 이 에피소드 행이 없었다. `OutlookSyncService` 가 link INSERT 를 **`else`(새 canon) 갈래 안에만** 두어, 프로젝트에 이미 있던 outlook 이 update 갈래로 가면 link 를 못 받았다. load_episode_entity_dicts link 가 있는 canon 만 싣는다 → still_recipe_service short_to_uuid 에 'O01' 이 없다 → still_recipe _norm_uuid('O01') = None → unknown outlook_id `EntitySyncService` 의 같은 자리는 분기 **밖**에서 upsert 해 이 결함이 없다. ## 왜 fake DB 로 안 재는가 이 결함은 「query 한 번을 어느 갈래에 뒀는가」다. fake DB 는 내가 짠 갈래만 흉내 내므로 같이 틀린다. 그래서 **실물 PG** 에 옛 canon 을 심어 두고, sync 뒤 **`load_episode_entity_dicts` 가 그 outlook 을 싣는지**까지 본다 — 그것이 still_recipe 가 실제로 읽는 자리다. Lane: ``-m pg``. """ from __future__ import annotations import json import uuid from pathlib import Path import pytest from sqlalchemy import text as sql_text pytestmark = pytest.mark.pg def _seed(session, pid: str, ep_old: str, ep_new: str) -> None: uid = f"outlook-link-{uuid.uuid4()}" session.execute(sql_text( "INSERT INTO user_account (id, username, display_name, password_hash, " "role, is_active, created_at, updated_at) VALUES " "(:uid, :uname, 't', 'x', 'creator', 1, '2026-01-01', '2026-01-01')" ), {"uid": uid, "uname": f"u_{uid}"}) session.execute(sql_text( "INSERT INTO project_registry (id, name, created_by, created_at, updated_at) " "VALUES (:pid, 'outlook-link', :uid, '2026-01-01', '2026-01-01')" ), {"pid": pid, "uid": uid}) for n, (eid, title) in enumerate(((ep_old, "1화"), (ep_new, "2화")), start=1): session.execute(sql_text( "INSERT INTO episode (id, project_id, title, episode_number, " "source_filename, source_path, created_at, updated_at) VALUES " "(:eid, :pid, :t, :n, 'x.txt', 'x/x.txt', '2026-01-01', '2026-01-01')" ), {"eid": eid, "pid": pid, "t": title, "n": n}) session.commit() def _add_canon(session, pid: str, cid: str, short_id: str, name: str, etype: str) -> None: session.execute(sql_text( "INSERT INTO entity_canon (id, project_id, short_id, name, entity_type, " "description, stable_traits, metadata_json, t2i_prompt, status, " "created_at, updated_at) VALUES " "(:cid, :pid, :s, :n, :e, '', '[]', '{}', '', 'active', " "'2026-01-01', '2026-01-01')" ), {"cid": cid, "pid": pid, "s": short_id, "n": name, "e": etype}) def _link(session, pid: str, eid: str, cid: str) -> None: session.execute(sql_text( "INSERT INTO entity_episode_link (id, canon_id, project_id, episode_id) " "VALUES (:lid, :cid, :pid, :eid)" ), {"lid": str(uuid.uuid4()), "cid": cid, "pid": pid, "eid": eid}) def _write_cp(root: Path, pid: str, eid: str, step_id: str, payload: dict) -> None: d = root / pid / "checkpoints" / "episodes" / eid / step_id d.mkdir(parents=True, exist_ok=True) (d / "manifest.json").write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8") @pytest.fixture def scene(pg_session, tmp_path: Path, monkeypatch): """1화에서 만들어진 outlook 이 2화 체크포인트에 **다시** 나오는 상황.""" monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path)) pid = f"p-{uuid.uuid4()}" ep_old, ep_new = str(uuid.uuid4()), str(uuid.uuid4()) _seed(pg_session, pid, ep_old, ep_new) char_id, outlook_id = str(uuid.uuid4()), str(uuid.uuid4()) _add_canon(pg_session, pid, char_id, "C01", "정임", "character") _add_canon(pg_session, pid, outlook_id, "O01", "감색차장제복", "outlook") # 옛 에피소드에만 달려 있다 — 이것이 실측 episode 0aea12c2 의 모양이다. _link(pg_session, pid, ep_old, char_id) _link(pg_session, pid, ep_old, outlook_id) # 새 에피소드엔 인물만 달려 있다 (EntitySyncService 는 제 몫을 한다). _link(pg_session, pid, ep_new, char_id) pg_session.commit() # 실측 체크포인트(0aea12c2/outlook_phase3)의 키를 그대로 쓴다 — 키를 짐작하면 # `scene_assignments` 를 못 읽어 character_outlook 이 안 생기고, 그러면 # `_remove_orphan_outlooks` 가 O01 을 지워 결함과 다른 것을 재게 된다. _write_cp(tmp_path, pid, ep_new, "outlook_phase3", { "status": "completed", "data": { "outlooks": [{ "name": "감색차장제복", "description": "감색 차장 제복", "character_id": "C01", "is_shared": False, "short_id": "O01", }], "scene_assignments": [{ "scene_index": 1, "assignments": [{"character_id": "C01", "outlook_name": "감색차장제복", "outlook_id": "O01"}], }], "removed": [], "null_outlook_chars": [], }, }) return pid, ep_new, outlook_id def _sync(session, pid: str, eid: str): from app.services.checkpoint_sync import OutlookSyncService OutlookSyncService(session, pid, eid).sync_from_checkpoint() session.commit() def test_이미_있던_outlook_이_새_에피소드_link_를_받는다(pg_session, scene): pid, eid, outlook_id = scene _sync(pg_session, pid, eid) rows = pg_session.execute(sql_text( "SELECT count(*) FROM entity_episode_link " "WHERE canon_id = :cid AND episode_id = :eid" ), {"cid": outlook_id, "eid": eid}).fetchone() assert rows[0] == 1, "재사용된 outlook 에 이 에피소드 link 가 안 생겼다" def test_still_recipe_가_읽는_자리에_그_outlook_이_실린다(pg_session, scene): """★끝점 — leaf 가 아니라 `load_episode_entity_dicts` 출력으로 잰다.""" from app.services.scene_persistence_service import ScenePersistenceService pid, eid, _ = scene _sync(pg_session, pid, eid) dicts = ScenePersistenceService(pg_session, pid).load_episode_entity_dicts(eid) shorts = {d["short_id"] for d in dicts} assert "O01" in shorts, ( f"still_recipe 의 short_to_uuid 가 'O01' 을 못 만든다 — 실린 것: {sorted(shorts)}") def test_두_번_sync_해도_link_가_안_늘어난다(pg_session, scene): """upsert 지 append 가 아니다.""" pid, eid, outlook_id = scene _sync(pg_session, pid, eid) _sync(pg_session, pid, eid) n = pg_session.execute(sql_text( "SELECT count(*) FROM entity_episode_link " "WHERE canon_id = :cid AND episode_id = :eid" ), {"cid": outlook_id, "eid": eid}).fetchone()[0] assert n == 1, f"link 가 {n} 개로 늘었다" def test_옛_에피소드_link_는_그대로_남는다(pg_session, scene, ): """이 수정이 남의 에피소드를 건드리지 않는다.""" pid, eid, outlook_id = scene before = pg_session.execute(sql_text( "SELECT count(*) FROM entity_episode_link WHERE canon_id = :cid" ), {"cid": outlook_id}).fetchone()[0] _sync(pg_session, pid, eid) after = pg_session.execute(sql_text( "SELECT count(*) FROM entity_episode_link WHERE canon_id = :cid" ), {"cid": outlook_id}).fetchone()[0] assert after == before + 1, f"옛 link 가 흔들렸다 ({before} → {after})"