"""`short_id` 를 **프로젝트가 발급**한다 (2026-09-04). ## 무엇이 결함이었나 `entity_steps._assign_short_ids` 와 `outlook_steps` 가 목록 안 **위치**를 번호로 썼다 (`f"{prefix}{i:02d}"`). 그래서 화가 달라도 늘 `01` 부터 시작했고, `entity_canon` 의 유일성이 `(project_id, short_id)` 라 2화의 `C01` 이 1화의 행을 찾아 덮었다. 실측 da049582 — 1화 CP C01=민수 · 2·3화 CP C01=정임 · 지금 DB C01=정임 ## 여기서 재는 것 1. 화를 넘어 **이어서** 발급하는가 (덮어쓰기가 구조적으로 불가능한가) 2. `O00` 을 **절대 안 내는가** (Null Outlook 예약값) 3. **지워진 번호를 다시 안 내는가** — DB 최대값만 보면 낸다. 실측에서 1화의 `O03 회색외투` 는 DB 에 없지만 1화 체크포인트에는 살아 있다. 4. 상한 999 에서 **서는가** (소비자 정규식이 세 자리까지만 받는다) 5. 이미 신원이 있는 줄은 **안 건드리는가** Lane: ``-m pg`` — 장부의 원자적 증가가 PostgreSQL 동작이라 fake 로는 못 잰다. """ 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) -> None: uid = f"alloc-{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, 'alloc', :uid, '2026-01-01', '2026-01-01')" ), {"pid": pid, "uid": uid}) session.commit() def _canon(session, pid: str, short_id: str, name: str, etype: str) -> str: cid = str(uuid.uuid4()) 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}) return cid def _episode(session, pid: str, eid: str, num: int) -> None: session.execute(sql_text( "INSERT INTO episode (id, project_id, title, episode_number, " "source_filename, source_path, created_at, updated_at) VALUES " "(:e, :p, :t, :n, 'x', 'x', 'x', 'x')"), {"e": eid, "p": pid, "t": f"{num}화", "n": num}) 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 (:l, :c, :p, :e)"), {"l": str(uuid.uuid4()), "c": cid, "p": pid, "e": eid}) @pytest.fixture def proj(pg_session, tmp_path: Path, monkeypatch): monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path)) pid = f"p-{uuid.uuid4()}" _seed(pg_session, pid) return pid, tmp_path def test_화를_넘어_이어서_발급한다(pg_session, proj): """★핵심 — 1화가 C01~C03 을 썼으면 2화는 C04 부터다.""" from app.core.entity_identity import reserve_short_ids pid, _ = proj for n, name in ((1, "민수"), (2, "연희"), (3, "노인")): _canon(pg_session, pid, f"C{n:02d}", name, "character") pg_session.commit() got = reserve_short_ids(pg_session, pid, "character", 2) assert got == ["C04", "C05"], f"1화 번호를 다시 냈다: {got}" def test_같은_번호를_두_번_내지_않는다(pg_session, proj): from app.core.entity_identity import reserve_short_ids pid, _ = proj first = reserve_short_ids(pg_session, pid, "location", 3) second = reserve_short_ids(pg_session, pid, "location", 3) assert set(first) & set(second) == set(), f"겹쳤다: {first} / {second}" assert second == ["L04", "L05", "L06"], second def test_지워진_번호를_체크포인트에서_보고_다시_안_낸다(pg_session, proj): """★DB 최대값만 보면 O03 을 다시 낸다 — 1화 CP 에는 그 번호가 살아 있다.""" from app.core.entity_identity import reserve_short_ids pid, root = proj # DB 에는 O01·O02 만 남아 있다 (O03 은 앞 판의 삭제로 사라졌다). _canon(pg_session, pid, "O01", "감색차장제복", "outlook") _canon(pg_session, pid, "O02", "작업점퍼", "outlook") pg_session.commit() # 그런데 1화 체크포인트에는 O03 이 그대로 있다. ep1 = str(uuid.uuid4()) d = root / pid / "checkpoints" / "episodes" / ep1 / "outlook_phase3" d.mkdir(parents=True) (d / "manifest.json").write_text(json.dumps({ "status": "completed", "data": {"outlooks": [ {"short_id": "O01", "name": "남색작업복"}, {"short_id": "O02", "name": "노란우비"}, {"short_id": "O03", "name": "회색외투"}, ]}, }, ensure_ascii=False), encoding="utf-8") got = reserve_short_ids(pg_session, pid, "outlook", 1) assert got == ["O04"], ( f"체크포인트에 살아 있는 O03 을 다시 냈다: {got} — " "DB 최대값만 보면 이렇게 된다") def test_O00_은_절대_안_나온다(pg_session, proj): """Null Outlook 예약값. 발급기가 내면 그 자리들이 실체를 sentinel 로 읽는다.""" from app.core.entity_identity import NULL_OUTLOOK_SHORT_ID, reserve_short_ids pid, _ = proj got = reserve_short_ids(pg_session, pid, "outlook", 5) assert NULL_OUTLOOK_SHORT_ID not in got, f"예약값이 나왔다: {got}" assert got[0] == "O01", f"첫 번호가 O01 이 아니다: {got}" def test_O00_만_있어도_다음은_O01(pg_session, proj): """seed 계산에서 `O00`(=0)은 무시돼야 한다.""" from app.core.entity_identity import NULL_OUTLOOK_SHORT_ID, reserve_short_ids pid, _ = proj _canon(pg_session, pid, NULL_OUTLOOK_SHORT_ID, "Null Outlook", "outlook") pg_session.commit() assert reserve_short_ids(pg_session, pid, "outlook", 1) == ["O01"] def test_상한을_넘으면_선다(pg_session, proj): """소비자 정규식이 `[CLPO]\\d{2,3}` 이라 1000 은 못 쓴다. 조용히 넘지 않는다.""" from app.core.entity_identity import (SHORT_ID_MAX, ShortIdExhausted, reserve_short_ids) pid, _ = proj _canon(pg_session, pid, f"P{SHORT_ID_MAX}", "마지막소품", "prop") pg_session.commit() with pytest.raises(ShortIdExhausted): reserve_short_ids(pg_session, pid, "prop", 1) def test_100번째부터는_세_자리(pg_session, proj): from app.core.entity_identity import reserve_short_ids pid, _ = proj _canon(pg_session, pid, "L99", "아흔아홉", "location") pg_session.commit() assert reserve_short_ids(pg_session, pid, "location", 2) == ["L100", "L101"] def test_이미_신원이_있는_줄은_안_건드린다(pg_session, proj): """앞 화에서 물려받은 short_id 를 덮으면 이어 붙이기가 무너진다.""" from app.core.entity_identity import assign_short_ids pid, _ = proj rows = [{"name": "정임", "short_id": "C07"}, {"name": "최씨"}, {"name": "노인", "short_id": ""}] out = assign_short_ids(pg_session, pid, "character", rows) assert out[0]["short_id"] == "C07", "물려받은 신원을 덮었다" assert out[1]["short_id"] == "C01" assert out[2]["short_id"] == "C02" def test_location_과_location_part_는_따로_센다(pg_session, proj): """접두표가 prefix-free 가 아니다 — `L` 과 `LP`. 한 통에 세면 어긋난다.""" from app.core.entity_identity import reserve_short_ids pid, _ = proj _canon(pg_session, pid, "LP07", "흙바닥", "location_part") pg_session.commit() assert reserve_short_ids(pg_session, pid, "location", 1) == ["L01"], ( "LP07 을 L 것으로 읽었다") assert reserve_short_ids(pg_session, pid, "location_part", 1) == ["LP08"] # ── 끝점: **스텝이** 그 발급기를 쓰는가 ────────────────────────────── def test_리스팅_스텝이_앞_화_번호를_이어서_받는다(pg_session, proj, monkeypatch): """★★모듈만 재면 스텝이 옛 방식을 그대로 써도 초록이다 — 실제로 그랬다. 프로덕션이 부르는 것은 **Step 클래스**다. 1화가 C01~C03 을 쓴 프로젝트에서 2화 리스팅을 돌려 `C04` 부터 나오는지 본다. """ import json from app.core.steps import STEP_CLASSES from app.modules.pipeline import entity_lister as el pid, root = proj # ★명부는 **화 범위**다 — 앞 화(1화)에 링크가 있어야 2화가 본다. ep1, eid = str(uuid.uuid4()), str(uuid.uuid4()) _episode(pg_session, pid, ep1, 1) _episode(pg_session, pid, eid, 2) for n, name in ((1, "민수"), (2, "연희"), (3, "노인")): _link(pg_session, pid, ep1, _canon(pg_session, pid, f"C{n:02d}", name, "character")) pg_session.commit() for step, data in (("visual_world_rules", {"era": "E", "region": "R"}), ("shot_validator", {"scenes": [ {"scene_index": 1, "scene_heading": "h", "shots": [{"shot_index": 1, "description": "d"}]}]})): d = root / pid / "checkpoints" / "episodes" / eid / step d.mkdir(parents=True, exist_ok=True) (d / "manifest.json").write_text( json.dumps({"status": "completed", "data": data}), encoding="utf-8") # ★모델은 안 부른다 — 재는 것은 「번호를 어디서 받나」다. monkeypatch.setattr(el, "call_structured", lambda **kw: { "characters": [{"name": "정임", "shot_count": 2, "prior_short_id": "NEW"}]}) monkeypatch.setattr(el, "load_prompt", lambda *a, **k: "p") monkeypatch.setattr(el, "load_schema", lambda *a, **k: { "type": "object", "properties": {"characters": {"type": "array", "items": { "type": "object", "properties": {"name": {"type": "string"}, "shot_count": {"type": "integer"}}, "required": ["name", "shot_count"], "additionalProperties": False}}}, "required": ["characters"], "additionalProperties": False}) runner = STEP_CLASSES["entity_all_character"]( step_id="entity_all_character", project_id=pid, episode_id=eid, db=pg_session, project_config={}) out = runner._execute() got = out["data"]["characters"][0]["short_id"] assert got == "C04", ( f"스텝이 앞 화 번호를 이어받지 않았다: {got} — " "화마다 C01 부터 다시 매기면 앞 화 행을 덮는다") def test_리스팅_스텝이_앞_화_명부를_프롬프트에_싣는다(pg_session, proj, monkeypatch): """★명부가 실제로 **나가는지**를 본다 — 만들어 놓고 안 붙이면 소용없다.""" import json from app.core.steps import STEP_CLASSES from app.modules.pipeline import entity_lister as el pid, root = proj ep1, eid = str(uuid.uuid4()), str(uuid.uuid4()) _episode(pg_session, pid, ep1, 1) _episode(pg_session, pid, eid, 2) _link(pg_session, pid, ep1, _canon(pg_session, pid, "C01", "민수", "character")) pg_session.commit() for step, data in (("visual_world_rules", {"era": "E", "region": "R"}), ("shot_validator", {"scenes": [ {"scene_index": 1, "scene_heading": "h", "shots": [{"shot_index": 1, "description": "d"}]}]})): d = root / pid / "checkpoints" / "episodes" / eid / step d.mkdir(parents=True, exist_ok=True) (d / "manifest.json").write_text( json.dumps({"status": "completed", "data": data}), encoding="utf-8") seen = {} def _capture(**kw): seen.update(kw) return {"characters": [{"name": "민수", "shot_count": 2, "prior_short_id": "C01"}]} monkeypatch.setattr(el, "call_structured", _capture) monkeypatch.setattr(el, "load_prompt", lambda *a, **k: "p") monkeypatch.setattr(el, "load_schema", lambda *a, **k: { "type": "object", "properties": {"characters": {"type": "array", "items": { "type": "object", "properties": {"name": {"type": "string"}, "shot_count": {"type": "integer"}}, "required": ["name", "shot_count"], "additionalProperties": False}}}, "required": ["characters"], "additionalProperties": False}) runner = STEP_CLASSES["entity_all_character"]( step_id="entity_all_character", project_id=pid, episode_id=eid, db=pg_session, project_config={}) out = runner._execute() assert "C01" in seen["user_prompt"], "앞 화 명부가 프롬프트에 안 실렸다" assert "민수" in seen["user_prompt"] enum = (seen["response_schema"]["properties"]["characters"]["items"] ["properties"]["prior_short_id"]["enum"]) assert enum == ["C01", "NEW"], f"허용 목록이 명부 밖으로 넓다: {enum}" assert out["data"]["characters"][0]["short_id"] == "C01", ( "앞 화 신원을 물려받지 않았다")