# Opik 기록 체계화 구현 계획 > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** Opik 기록에 계층(주행 → 작업 단위 → 호출)을 세우고 태그를 축으로 갈라, 「주행 훑기 · 프롬프트 감사 · 샷 계보 · 최종 샷 영향」 넷을 물을 수 있게 만든다. > **축 배선 현황 (2026-08-24 정정)** — 축 **여섯**을 정의하되 이번 판이 실제로 > 값을 싣는 것은 **다섯**(`step:` `op:` `model:` `provider:` `status:`)이다. > `kind:` 는 **후속**이다 — `op` 문자열을 글자로 갈라 유도하는 것은 이 저장소가 > 금지한 방식이라(글자·substring 으로 의미를 판단하지 않는다) 호출자나 구조화된 > registry 가 명시해야 하고, 그것은 호출 사이트 전수 배선이라 별건이다. > `build_axis_tags` 에 축은 이미 있어 값만 실으면 된다. → 아래 「후속 항목」. > **개정 (2026-08-23)** — Codex 코드 리뷰 6건(BLOCK)을 전부 file:line 으로 > 확인해 반영했다. 가장 큰 것은 **샷 경계가 프로덕션 본류에 없었다**는 것 > (→ Task 7-B·7-C 신설). 나머지: cine 승패 판별(Task 11) · 반환 경로 전수 > (Task 12) · v1 이중 계산(Task 15) · CLI 계약(Task 14) · 진단 오염(Task 16). **Architecture:** litellm 의 opik 통합은 `metadata["opik"]` 에서 `project_name` · `current_span_data` · `tags` · `thread_id` 넷만 읽는다. 그중 `current_span_data` 에 부모 trace_id 를 주면 litellm 이 trace 를 만들지 않고 **span 만 우리 trace 밑에 붙인다**. 이것을 지렛대로 삼아, 이미 코드에 있는 두 경계(`StepRunner` = 스텝, `generation_context` = 샷)에서 trace 를 열고 닫는다. 새 경계는 만들지 않는다. **Tech Stack:** Python 3.12 · FastAPI · pydantic-settings · litellm(opik 콜백) · opik SDK 1.10.45 · PostgreSQL · pytest **설계 SOT:** `docs/superpowers/specs/2026-08-23-opik-trace-taxonomy-design.md` ## Global Constraints 이 절의 요구는 **모든 태스크에 암묵적으로 포함**된다. - **설정 이름**: `OPIK_TRACE_V2_ENABLED` (pydantic 필드 `opik_trace_v2_enabled: bool = False`). **기본 OFF.** - **OFF 경로는 바이트 동일**: 설정이 꺼져 있으면 Opik 로 나가는 payload 가 지금과 한 바이트도 달라선 안 된다. 모든 신규 분기는 `if settings.opik_trace_v2_enabled:` 안쪽에만 둔다. - **기록 실패는 전부 non-fatal**: 기록 코드에서 새어 나온 예외가 본 작업을 죽여선 안 된다. `except Exception` 으로 삼키고 `logger.debug`/`logger.warning` 만 남긴다. 단 **시험 하네스의 자체 점검은 예외** — 거기서는 fail-fast 한다(Task 1). - **지문 금지**: `compute_input_fingerprint`(`backend/app/modules/pipeline/multiroll_select.py:671`)와 그 `extra` 기여 항목(`backend/app/services/still_recipe_service.py:802~912`)에 **기록 관련 값을 절대 넣지 않는다.** 넣으면 완주 판 넷이 전량 재생성된다. - **trace/span id 는 UUIDv7**: Opik 서버가 uuid4 를 `400 "Trace id must be a version 7 UUID"` 로 거부한다(2026-08-23 실측, `backend/opik_uid_probe.py`). - **시험은 돈을 못 쓴다**: `backend/tests/conftest.py` 가 `THEROAD_TEST_BLOCK_OUTBOUND=1`·`LLM_MAX_RETRIES=0` 을 기본으로 건다. 새 시험은 실제 provider 를 부르지 않는다 — 전부 가짜(fake)나 monkeypatch 로 짠다. - **★진단·확인도 프로덕션에 쓰지 않는다**: `record_provider_call` 은 `to_db=True` 가 기본이라 부르는 순간 프로덕션 `llm_call_log` 와 Opik 에 **진짜 행**이 남는다. 원인을 재려고 손으로 한 번 부르는 것도 오염이다. 손으로 확인할 때는 `DATABASE_URL`(→`theroad_test`)과 `OPIK_PROJECT_NAME`(→`…-test`)을 갈아입히고, **쓰기 전에 `assert` 로 갈렸는지 확인**한다. - **ContextVar 는 셋이다**: budget · generation_context · **trace**. 병렬 자리(`multiroll_select.py:1385` 등)에서 하나라도 빠뜨리면 **그 축만 조용히 무너진다.** 새 병렬 자리를 만들거나 고칠 때 셋을 함께 본다. - **fake 를 구현과 같은 함수로 만들지 않는다**: 같이 틀리면 mock 을 검증하게 된다. - **시험 실행**: `cd backend && ./.venv/bin/pytest <경로> -v` - **커밋 메시지**: 한국어. 개발 용어(commit·push·branch·staged·span·trace 등)는 원어 그대로. ### metadata 표준 칸 — 이번에 싣는 것과 안 싣는 것 spec ⑦ 의 한 벌 중 **이번에 배선하는 칸**: `run_tag` · `project_id` · `episode_id` · `project_name` · `episode_title` · `thread_id` · `step` · `scene_index` · `shot_index` · `still_id` · `entity_id` · `shot_run_uid` **안 싣는 칸과 이유**: | 칸 | 왜 안 싣나 | |---|---| | `op` · `kind` | **태그로 싣는다**(`op:` · `kind:`). metadata 에 또 두면 두 벌이 되고 어느 쪽이 진짜인지 갈린다. ★단 `kind:` 는 이번 판에서 **값을 안 싣는다** — 위 「축 배선 현황」 참조 | | `attempt` | 부를 수 있는 자리가 호출자마다 달라, 지금 배선하면 빠뜨린 자리가 조용히 빈다. 재시도는 `status:retry` 태그로 보인다 | | `prompt_version` | spec ⑦ 이 「부를 수 있는 자리에서만」이라 했다. 그 자리를 만드는 것은 별건이다 — 없는 값을 지어내지 않는다 | ★셋 다 **나중에 보태기 쉬운 모양**으로 둔다(`build_axis_tags` 에 `op`·`kind` 축이 이미 있다). ## File Structure | 파일 | 책임 | |---|---| | `backend/app/modules/llm/opik_trace.py` **(신규)** | trace 계층의 단일 창구 — UUIDv7 생성, trace scope(ContextVar), 축 태그 조립, 신원 metadata 조립. 다른 모듈은 전부 여기만 부른다 | | `backend/app/core/config.py` | `opik_trace_v2_enabled` 설정 한 줄 | | `backend/app/modules/llm/llm_client.py:121-145` | `_build_opik_metadata` — litellm 이 읽는 네 키를 올바로 채운다 | | `backend/app/core/step_runner.py:1598-1636`, `1253`, `1375` | `build_opik_metadata` 를 thread_id 에피소드 단위로. 스텝 trace 열고 닫기 | | `backend/app/services/image_capture/context.py:83-131` | `generation_context` 에서 샷 trace 열고 닫기 | | `backend/app/modules/llm/image_tracer.py:124-200` | `ImageTracer.log` — 부모가 있으면 trace 를 새로 만들지 않고 span 만 | | `backend/app/modules/pipeline/multiroll_select.py:1340-1360`, `1513-1532` | `records.json` 의 `refs`·`roll_refs` 에 `asset_id` 칸 | | `backend/app/services/still_recipe_service.py:2819-2880`, `4470-4600` | `ref_role_metadata` 조립·전달, `cine_source_sel` **직접 입력 기술**(자산 연결 아님 — Task 11 이 좁혔다) | | `backend/tests/conftest.py:45-77` | 시험용 Opik 프로젝트 강제 + 자체 점검 | | `tools/opik_prompt_audit/audit/fetch.py:59-100` | trace → span 으로 자료원 이관 | | `tools/shot_influence.py` **(신규)** | `still_id` 하나로 세 곳을 합쳐 출력 | --- # 단계 1 — 시험 격리 (설정 없음, 바로 나간다) ### Task 1: 시험이 프로덕션 Opik 에 못 쓰게 가른다 **Files:** - Modify: `backend/tests/conftest.py:77` (PROJECTS_DIR 설정 바로 뒤) - Test: `backend/tests/unit/test_opik_test_project_isolation.py` (신규) **Interfaces:** - Consumes: 없음 - Produces: 시험 실행 중 `settings.opik_project_name == "theroad-scene-lab-test"` 보장 **배경:** 돈 가드(`backend/tests/netprobe.py`)는 집 안(사설) 주소를 **일부러 통과**시킨다 — 요금이 없으니 옳다. 그런데 자체 호스팅 Opik 이 집 안 주소라 시험이 프로덕션 감사 데이터에 쓴다. 표집 4,200건 중 657건(15.6%, 전체 환산 약 2,600건)이 시험 기록이었다. **막지 않고 가른다** — 막으면 자체 호스팅 Opik 이 죽는다(2026-08-20 에 실제로 겪음). - [x] **Step 1: 실패하는 시험을 쓴다** ```python # backend/tests/unit/test_opik_test_project_isolation.py """시험은 프로덕션 Opik 프로젝트에 쓰면 안 된다. 돈 가드는 집 안 주소를 통과시킨다(요금이 없으니 옳다). 그래서 자체 호스팅 Opik 은 시험에서도 실제로 쓰기가 된다 — 감사 데이터가 오염된다. 막는 대신 **가른다**: 시험은 별도 프로젝트로 간다. """ TEST_PROJECT = "theroad-scene-lab-test" def test_settings_use_test_opik_project(): from app.core.config import settings assert settings.opik_project_name == TEST_PROJECT, ( f"시험이 {settings.opik_project_name!r} 에 쓰고 있다 — " f"프로덕션 감사 데이터가 오염된다" ) def test_env_var_is_set_for_child_readers(): """`_init_opik` 과 `ImageTracer` 는 env 를 되읽는다 — 거기도 시험 이름.""" import os assert os.environ.get("OPIK_PROJECT_NAME") == TEST_PROJECT ``` - [x] **Step 2: 실패를 확인한다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_opik_test_project_isolation.py -v` Expected: FAIL — `AssertionError: 시험이 'theroad-scene-lab' 에 쓰고 있다` - [x] **Step 3: conftest 에 env 를 건다** `backend/tests/conftest.py` 의 `os.environ["PROJECTS_DIR"] = ...` (77줄) 바로 다음에 넣는다. **자리가 중요하다** — conftest 는 app 을 env 설정 뒤에 import 한다(112줄 주석 「env 설정 후라야 안전」). 실측으로 env 가 `.env` 보다 우선함을 확인했다. ```python # 시험 기록을 프로덕션 감사 데이터에서 가른다 (2026-08-23). # 돈 가드(netprobe)는 집 안 주소를 통과시킨다 — 요금이 없으니 옳다. 그런데 # 자체 호스팅 Opik 이 집 안 주소라, 시험이 프로덕션 프로젝트에 실제로 쓴다. # 실측: 프로덕션 trace 표집 4,200건 중 657건(15.6%)이 시험 기록이었다 # (prompt='xxxx'·'p'·'SAMPLE prompt'). 막으면 자체 호스팅 Opik 이 죽으므로 # (2026-08-20 실측) **막지 않고 가른다**. # ★setdefault 가 아니라 강제 대입이다 — 기계 .env 의 값이 이기면 안 된다. os.environ["OPIK_PROJECT_NAME"] = "theroad-scene-lab-test" ``` - [x] **Step 4: 시험이 통과하는지 본다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_opik_test_project_isolation.py -v` Expected: PASS (2건) - [x] **Step 5: 자체 점검을 conftest 에 심는다** 값이 조용히 프로덕션으로 돌아가는 것이 이 조치가 막으려는 바로 그 실패다. `import pytest` 뒤, `pytest_configure` 가 있으면 그 안에, 없으면 새로 만든다. ```python def pytest_configure(config) -> None: # noqa: ARG001 """Opik 프로젝트가 시험용인지 확인 — 아니면 시험 자체를 세운다. ★fail-fast 다. 기록 실패를 non-fatal 로 삼키는 프로덕션 규약과 반대인데, 여기서 조용히 넘어가면 프로덕션 감사 데이터가 오염되는 것을 아무도 모른다. """ from app.core.config import settings if settings.opik_project_name != "theroad-scene-lab-test": raise RuntimeError( f"REFUSING test run: opik_project_name=" f"{settings.opik_project_name!r} — 시험이 프로덕션 Opik 프로젝트에 " f"쓰려 한다. conftest 의 OPIK_PROJECT_NAME 대입이 무력화됐다." ) ``` ★기존 `pytest_configure` 가 이미 있으면 **새로 만들지 말고 그 안에 이 블록을 보탠다** — 같은 이름 두 개면 뒤엣것이 앞엣것을 덮는다. - [x] **Step 6: 시험 한 바퀴로 회귀 확인 (관문 6)** 프로덕션 프로젝트의 trace 수를 시험 전후로 잰다. ```bash cd /Users/manta/Documents/Projects/TheRoad-I1 count() { curl -s -H "Comet-Workspace: default" \ "http://192.168.133.87:5173/api/v1/private/traces?project_id=019fffa9-b33c-7717-8228-efccb876dc67&page=1&size=1" \ | python3 -c "import json,sys; print(json.load(sys.stdin)['total'])"; } BEFORE=$(count) (cd backend && ./.venv/bin/pytest tests/ -q 2>&1 | tail -5) sleep 5 AFTER=$(count) echo "프로덕션 trace: $BEFORE → $AFTER (증가 $((AFTER-BEFORE)))" ``` Expected: **증가 0**. 그리고 실패 건수가 기준선 27건과 같아야 한다(늘었으면 이 변경이 뭔가를 깨뜨린 것). - [x] **Step 7: commit** ```bash git add backend/tests/conftest.py backend/tests/unit/test_opik_test_project_isolation.py git commit -m "fix(tests): 시험 Opik 기록을 프로덕션 프로젝트에서 가른다 돈 가드는 집 안 주소를 통과시킨다(요금 없음 — 옳다). 그런데 자체 호스팅 Opik 이 집 안 주소라 시험이 프로덕션 감사 데이터에 실제로 썼다. 표집 4,200건 중 657건(15.6%, 전체 환산 약 2,600건)이 시험 기록이었다. 막지 않고 가른다 — 막으면 자체 호스팅 Opik 이 죽는다(2026-08-20 실측). OPIK_PROJECT_NAME 을 시험용으로 강제하고, 값이 프로덕션으로 되돌아가면 pytest_configure 에서 fail-fast 한다." ``` --- # 단계 2 — 계층·이름·태그·metadata (설정 뒤, 한 덩어리) ### Task 2: 설정 + UUIDv7 + 축 태그 — `opik_trace.py` 의 뼈대 **Files:** - Create: `backend/app/modules/llm/opik_trace.py` - Modify: `backend/app/core/config.py:921` 부근 (`still_fix_ref_gate_enabled` 옆) - Test: `backend/tests/unit/test_opik_trace_primitives.py` (신규) **Interfaces:** - Consumes: 없음 - Produces: - `new_trace_uid() -> str` — UUIDv7 문자열 - `build_axis_tags(*, step: Optional[str] = None, op: Optional[str] = None, kind: Optional[str] = None, model: Optional[str] = None, provider: Optional[str] = None, status: Optional[str] = None) -> List[str]` - `settings.opik_trace_v2_enabled: bool` - [x] **Step 1: 실패하는 시험을 쓴다** ```python # backend/tests/unit/test_opik_trace_primitives.py """trace uid 와 축 태그 — Opik 계층의 두 원재료.""" import uuid import pytest from app.modules.llm.opik_trace import build_axis_tags, new_trace_uid def test_uid_is_uuid_version_7(): """Opik 서버가 uuid4 를 400 으로 거부한다(2026-08-23 실측). "Trace id must be a version 7 UUID" """ for _ in range(20): u = uuid.UUID(new_trace_uid()) assert u.version == 7, f"version={u.version} — Opik 이 거부한다" def test_uid_is_unique(): assert len({new_trace_uid() for _ in range(500)}) == 500 def test_uid_is_time_ordered(): """UUIDv7 은 시간순이다 — 정렬하면 만든 순서가 된다.""" made = [new_trace_uid() for _ in range(50)] assert made == sorted(made) def test_axis_tags_carry_prefixes(): tags = build_axis_tags( step="scene_image_pipeline", op="still_recipe_judge", kind="judge", model="x-ai/grok-4.6", provider="openrouter", ) assert tags == [ "step:scene_image_pipeline", "op:still_recipe_judge", "kind:judge", "model:x-ai/grok-4.6", "provider:openrouter", ] def test_axis_tags_skip_empty_axes(): assert build_axis_tags(step="a") == ["step:a"] assert build_axis_tags(step="a", op=None, model="") == ["step:a"] def test_axis_tags_never_emit_bare_names(): """접두사 없는 태그가 하나라도 나오면 축이 다시 섞인다.""" tags = build_axis_tags(step="s", op="o", kind="k", model="m", provider="p", status="retry") assert all(":" in t for t in tags) @pytest.mark.parametrize("bad", [" ", "\t", None, ""]) def test_axis_tags_treat_blank_as_absent(bad): assert build_axis_tags(step="s", op=bad) == ["step:s"] ``` - [x] **Step 2: 실패를 확인한다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_opik_trace_primitives.py -v` Expected: FAIL — `ModuleNotFoundError: No module named 'app.modules.llm.opik_trace'` - [x] **Step 3: 설정을 추가한다** `backend/app/core/config.py` 의 `still_fix_ref_gate_enabled: bool = False` 아래에 넣는다. ```python # Opik 기록 체계화 v2 (2026-08-23) — 계층(주행>작업 단위>호출)·축 태그· # 신원 metadata. 설계=docs/superpowers/specs/2026-08-23-opik-trace- # taxonomy-design.md # ★기본 OFF. 꺼져 있으면 Opik payload 가 지금과 바이트 동일하다. opik_trace_v2_enabled: bool = False ``` - [x] **Step 4: 모듈을 만든다** ```python # backend/app/modules/llm/opik_trace.py """Opik 기록 계층의 단일 창구 — uid·축 태그·trace scope. litellm 의 opik 통합이 `metadata["opik"]` 에서 읽는 키는 넷뿐이다: `project_name` · `current_span_data` · `tags` · `thread_id`. `trace_name` 은 litellm 소스에 없다 — trace 이름은 항상 `response_obj["object"]`("chat.completion")로 못박혀 있다. 그래서 이름 있는 trace 는 **우리가 만들고**, litellm 호출에는 `current_span_data={"trace_id": <우리 것>}` 을 실어 **span 만** 붙게 한다. 설계: docs/superpowers/specs/2026-08-23-opik-trace-taxonomy-design.md """ from __future__ import annotations import logging import os import time from typing import List, Optional logger = logging.getLogger(__name__) #: 태그 축 — 이름 순서가 곧 표시 순서다. _AXIS_ORDER = ("step", "op", "kind", "model", "provider", "status") def new_trace_uid() -> str: """Opik trace/span id — **UUIDv7 이어야 한다**. 2026-08-23 실측: uuid4 를 주면 서버가 `400 "Trace id must be a version 7 UUID"` 로 거부한다. litellm 도 같은 방식(`litellm.integrations.opik.utils.create_uuid7`)을 쓴다. 그쪽 함수를 빌리지 않는 이유: litellm 내부 경로라 판이 바뀌면 조용히 사라진다 — 기록이 죽는 자리를 남의 사정에 걸지 않는다. """ ns = time.time_ns() sixteen_secs = 16_000_000_000 t1, rest1 = divmod(ns, sixteen_secs) t2, rest2 = divmod(rest1 << 16, sixteen_secs) t3, _ = divmod(rest2 << 12, sixteen_secs) t3 |= 7 << 12 # version 7 seq = int.from_bytes(os.urandom(2), "big") & 0x3FFF t4 = (2 << 14) | seq # variant 0b10 rand = os.urandom(6) return f"{t1:>08x}-{t2:>04x}-{t3:>04x}-{t4:>04x}-{rand.hex()}" def build_axis_tags( *, step: Optional[str] = None, op: Optional[str] = None, kind: Optional[str] = None, model: Optional[str] = None, provider: Optional[str] = None, status: Optional[str] = None, ) -> List[str]: """축을 접두사로 못박은 태그 목록. 지금은 스텝·모델·제공자·프로젝트·에피소드·상태가 **한 자루**에 섞여 (실측 81종) 태그를 봐도 그게 무슨 축인지 모른다. 접두사를 붙이면 축이 갈린다. ★프로젝트 이름·에피소드 제목은 여기에 **안 넣는다** — 한글이고 카디널리티가 커진다. 그것은 metadata 로 간다. ★litellm 이 span 태그에 제공자 이름을 **맨 이름으로 덧붙인다** (`extract_tags` 의 `tags.append(custom_llm_provider)`). 그것은 못 막는다. trace 태그는 우리가 전부 만드므로 깨끗하다. """ values = {"step": step, "op": op, "kind": kind, "model": model, "provider": provider, "status": status} out: List[str] = [] for axis in _AXIS_ORDER: v = values.get(axis) if v is None: continue text = str(v).strip() if not text: continue tag = f"{axis}:{text}" if tag not in out: out.append(tag) return out ``` - [x] **Step 5: 시험이 통과하는지 본다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_opik_trace_primitives.py -v` Expected: PASS (10건 — parametrize 4건 포함) - [x] **Step 6: uid 가 진짜 Opik 에 먹히는지 실물로 본다** 시험은 형식만 본다. 서버가 받는지는 따로 확인한다. ```bash cd backend && ./.venv/bin/python -c " import json,urllib.request from app.modules.llm.opik_trace import new_trace_uid uid=new_trace_uid() body={'id':uid,'project_name':'theroad-probe-uid','name':'probe:task2', 'start_time':'2026-08-23T00:00:00.000000Z','end_time':'2026-08-23T00:00:01.000000Z', 'input':{},'output':{}} r=urllib.request.Request('http://192.168.133.87:5173/api/v1/private/traces', data=json.dumps(body).encode(), headers={'Comet-Workspace':'default','Content-Type':'application/json'},method='POST') print('HTTP', urllib.request.urlopen(r,timeout=20).status, uid) " ``` Expected: `HTTP 201` (400 이면 uid 생성이 잘못된 것) - [x] **Step 7: commit** ```bash git add backend/app/modules/llm/opik_trace.py backend/app/core/config.py \ backend/tests/unit/test_opik_trace_primitives.py git commit -m "feat(opik): trace uid(UUIDv7)·축 태그 원재료 + v2 설정(기본 OFF) Opik 서버는 uuid4 를 400 'Trace id must be a version 7 UUID' 로 거부한다 (2026-08-23 실측). litellm 의 create_uuid7 을 빌리지 않고 직접 만든다 — litellm 내부 경로라 판이 바뀌면 기록이 조용히 죽는다. 태그는 step:/op:/kind:/model:/provider:/status: 여섯 축을 접두사로 못박는다. 프로젝트 이름·에피소드 제목은 태그에서 뺀다(한글·카디널리티) — metadata 로." ``` --- ### Task 3: trace scope — ContextVar 로 부모를 나른다 **Files:** - Modify: `backend/app/modules/llm/opik_trace.py` - Test: `backend/tests/unit/test_opik_trace_scope.py` (신규) **Interfaces:** - Consumes: `new_trace_uid`, `build_axis_tags` (Task 2) - Produces: - `TraceHandle` — `dataclass(frozen=True)` with `uid: str`, `name: str`, `thread_id: Optional[str]` - `current_trace() -> Optional[TraceHandle]` - `open_trace(*, name, tags, metadata, thread_id, input_data=None) -> ContextManager[Optional[TraceHandle]]` - `bind_trace(handle) -> Token` / `reset_trace(token)` — worker thread 명시 전파용 - `finish_trace(handle, *, output=None, error=None)` — 밖에서 닫을 때 **설계 근거:** `generation_context` 가 `contextvars.ContextVar` 를 쓴다(`backend/app/services/image_capture/context.py:42`). 같은 그릇이라야 중첩·복원이 어긋나지 않는다 — `llm_client` 의 thread-local 을 쓰면 안 된다. - [x] **Step 1: 실패하는 시험을 쓴다** ```python # backend/tests/unit/test_opik_trace_scope.py """trace scope — 중첩·복원·꺼짐·실패 삼킴.""" import pytest from app.modules.llm.opik_trace import current_trace, open_trace @pytest.fixture(autouse=True) def _v2_on(monkeypatch): from app.core import config monkeypatch.setattr(config.settings, "opik_trace_v2_enabled", True) @pytest.fixture(autouse=True) def _no_real_client(monkeypatch): """진짜 Opik 클라이언트를 만들지 않는다 — 시험은 망을 안 탄다. ★구현과 같은 함수로 fake 를 만들지 않는다. 여기서는 payload 를 받아 적기만 하는 아주 단순한 대역을 쓴다. """ sent = [] class _FakeTrace: def __init__(self, **kw): self.kw = kw sent.append(("trace", kw)) def end(self, **kw): sent.append(("end", kw)) def update(self, **kw): sent.append(("update", kw)) class _FakeClient: def trace(self, **kw): return _FakeTrace(**kw) from app.modules.llm import opik_trace monkeypatch.setattr(opik_trace, "_get_client", lambda: _FakeClient()) return sent def test_no_trace_outside_scope(): assert current_trace() is None def test_scope_sets_and_restores(): with open_trace(name="step:a", tags=["step:a"], metadata={}, thread_id="t1") as h: assert h is not None assert current_trace() is h assert current_trace().name == "step:a" assert current_trace() is None def test_inner_scope_wins_and_outer_returns(): with open_trace(name="step:a", tags=[], metadata={}, thread_id="t1") as a: with open_trace(name="still:S1sh1", tags=[], metadata={}, thread_id="t1") as b: assert current_trace() is b assert b.uid != a.uid assert current_trace() is a def test_scope_restores_on_exception(): with pytest.raises(ValueError): with open_trace(name="step:a", tags=[], metadata={}, thread_id=None): raise ValueError("본 작업이 터졌다") assert current_trace() is None def test_disabled_yields_none_and_sends_nothing(monkeypatch, _no_real_client): from app.core import config monkeypatch.setattr(config.settings, "opik_trace_v2_enabled", False) with open_trace(name="step:a", tags=[], metadata={}, thread_id="t") as h: assert h is None assert current_trace() is None assert _no_real_client == [] def test_client_failure_is_non_fatal(monkeypatch): """기록이 터져도 본 작업은 산다.""" from app.modules.llm import opik_trace def _boom(): raise RuntimeError("Opik 서버가 죽었다") monkeypatch.setattr(opik_trace, "_get_client", _boom) with open_trace(name="step:a", tags=[], metadata={}, thread_id="t") as h: assert h is None # 부모가 없을 뿐 assert current_trace() is None def test_bind_and_reset_for_worker_threads(): from app.modules.llm.opik_trace import bind_trace, reset_trace with open_trace(name="step:a", tags=[], metadata={}, thread_id="t") as parent: token = bind_trace(parent) try: assert current_trace() is parent finally: reset_trace(token) ``` - [x] **Step 2: 실패를 확인한다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_opik_trace_scope.py -v` Expected: FAIL — `ImportError: cannot import name 'current_trace'` - [x] **Step 3: scope 를 구현한다** `backend/app/modules/llm/opik_trace.py` 아래에 이어 붙인다. ```python import threading from contextlib import contextmanager from contextvars import ContextVar, Token from dataclasses import dataclass from typing import Any, Dict, Iterator @dataclass(frozen=True) class TraceHandle: """지금 열려 있는 trace 의 손잡이. frozen — scope 안에서 안 바뀐다.""" uid: str name: str thread_id: Optional[str] = None _trace_ctx: ContextVar[Optional[TraceHandle]] = ContextVar( "opik_trace_ctx", default=None ) _client = None _client_lock = threading.Lock() def _get_client(): """Opik SDK 클라이언트 — 프로젝트 이름을 명시해서 만든다. ★인자 없이 만들면 기본 프로젝트로 쌓여 텍스트 호출과 갈린다 (2026-08-07 에 실제로 그래서 「이미지 기록이 통째로 없다」고 잘못 판단했다). """ global _client if _client is not None: return _client with _client_lock: if _client is None: import opik from app.core.config import settings if settings.opik_url_override: os.environ.setdefault( "OPIK_URL_OVERRIDE", settings.opik_url_override) os.environ.setdefault( "OPIK_WORKSPACE", settings.opik_workspace) _client = opik.Opik(project_name=settings.opik_project_name) return _client def current_trace() -> Optional[TraceHandle]: """지금 열려 있는 trace (없으면 None).""" return _trace_ctx.get() def bind_trace(handle: Optional[TraceHandle]) -> Token: """worker thread 명시 전파용 — 호출자가 reset_trace 로 되돌린다.""" return _trace_ctx.set(handle) def reset_trace(token: Token) -> None: _trace_ctx.reset(token) #: uid → SDK trace 객체. finish 할 때 되찾는다. #: ★`open_trace` 가 성공·예외 양쪽에서 반드시 `finish_trace` 를 불러 pop 하므로 #: 누수되지 않는다. `bind_trace` 로만 세운 handle 은 여기 없다 — 그 경우 #: `update_trace_output` 은 조용히 넘어간다(부모를 만든 쪽이 닫는다). _LIVE: Dict[str, Any] = {} def finish_trace( handle: Optional[TraceHandle], *, output: Optional[Dict[str, Any]] = None, error: Optional[str] = None, ) -> None: """trace 를 닫는다. 실패는 삼킨다.""" if handle is None: return try: live = _LIVE.pop(handle.uid, None) if live is None: return payload: Dict[str, Any] = {} if output is not None: payload["output"] = output if error: payload["error_info"] = { "exception_type": "PipelineError", "message": str(error)[:500], "traceback": "", } if payload: live.update(**payload) live.end() except Exception as exc: # noqa: BLE001 — 기록은 본 작업을 안 막는다 logger.debug("finish_trace 실패 (non-fatal): %s", exc) @contextmanager def open_trace( *, name: str, tags: List[str], metadata: Dict[str, Any], thread_id: Optional[str], input_data: Optional[Dict[str, Any]] = None, ) -> Iterator[Optional[TraceHandle]]: """이름 있는 trace 를 열고 scope 에 세운다. 설정이 꺼져 있으면 **아무것도 안 하고 None 을 준다** — 그 경우 하위 호출은 지금처럼 각자 trace 를 만든다(바이트 동일). 실패해도 None 을 줄 뿐 예외를 안 낸다 — 부모가 없으면 하위가 홀로 설 뿐이고, 그것이 기록 없는 것보다 낫다. """ from app.core.config import settings if not getattr(settings, "opik_trace_v2_enabled", False): yield None return handle: Optional[TraceHandle] = None try: uid = new_trace_uid() live = _get_client().trace( id=uid, name=name, tags=list(tags), metadata=dict(metadata), thread_id=thread_id, input=input_data or {}, ) _LIVE[uid] = live handle = TraceHandle(uid=uid, name=name, thread_id=thread_id) except Exception as exc: # noqa: BLE001 logger.debug("open_trace 실패 (non-fatal): %s", exc) handle = None if handle is None: yield None return token = _trace_ctx.set(handle) try: yield handle except Exception as exc: finish_trace(handle, error=str(exc)) raise else: finish_trace(handle) finally: _trace_ctx.reset(token) ``` - [x] **Step 4: 시험이 통과하는지 본다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_opik_trace_scope.py -v` Expected: PASS (8건) - [x] **Step 5: commit** ```bash git add backend/app/modules/llm/opik_trace.py backend/tests/unit/test_opik_trace_scope.py git commit -m "feat(opik): trace scope — ContextVar 로 부모를 나른다 generation_context 가 ContextVar 를 쓰므로 같은 그릇으로 맞춘다. thread-local 을 쓰면 중첩·복원이 어긋난다. 설정이 꺼져 있으면 None 을 주고 아무것도 안 보낸다(바이트 동일). 클라이언트가 터져도 None 일 뿐 — 부모 없이 홀로 서는 것이 기록 없는 것보다 낫다." ``` --- ### Task 4: litellm 이 읽는 네 키를 올바로 채운다 **Files:** - Modify: `backend/app/modules/llm/llm_client.py:121-145` (`_build_opik_metadata`) - Test: `backend/tests/unit/test_opik_metadata_keys.py` (신규) **Interfaces:** - Consumes: `current_trace` (Task 3) - Produces: `_build_opik_metadata(step, opik_metadata)` 가 v2 에서 `thread_id` · `current_span_data` 를 채운 dict 를 준다 **배경 (실측):** litellm 이 `metadata["opik"]` 에서 읽는 키는 `project_name` · `current_span_data` · `tags` · `thread_id` 넷뿐이다(`litellm/integrations/opik/opik_payload_builder/api.py`). 우리가 지금 싣는 `trace_name` 은 litellm 소스에 **0회** 등장하는 죽은 키이고, `session_id` 도 litellm 은 안 본다(`thread_id` 를 본다). 그래서 텍스트 호출은 thread 가 통째로 없다. - [x] **Step 1: 실패하는 시험을 쓴다** ```python # backend/tests/unit/test_opik_metadata_keys.py """litellm 이 실제로 읽는 키만 값을 갖는다. litellm/integrations/opik/opik_payload_builder/api.py 가 읽는 키: project_name · current_span_data · tags · thread_id `trace_name` 은 litellm 소스에 없다 — trace 이름은 response_obj["object"] ("chat.completion")로 못박혀 있다. """ import pytest from app.modules.llm.llm_client import _build_opik_metadata, set_opik_context @pytest.fixture(autouse=True) def _clean(): set_opik_context(None) yield set_opik_context(None) def test_litellm_honored_keys_are_the_contract(): """이 시험이 깨지면 litellm 판이 바뀐 것이다 — 배선을 다시 봐야 한다.""" import inspect from litellm.integrations.opik.opik_payload_builder import api src = inspect.getsource(api) assert 'opik_metadata.get("thread_id")' in src assert 'opik_metadata.get("current_span_data")' in src assert "trace_name" not in src, "litellm 이 trace_name 을 읽기 시작했다" def test_v1_keeps_session_id_untouched(monkeypatch): """설정 OFF 면 지금 모양 그대로 — 바이트 동일.""" from app.core import config monkeypatch.setattr(config.settings, "opik_trace_v2_enabled", False) set_opik_context({"session_id": "run-1", "tags": ["scene_detail"]}) md = _build_opik_metadata("scene_detail")["opik"] assert md["session_id"] == "run-1" assert "thread_id" not in md assert "current_span_data" not in md def test_v2_moves_session_to_thread_id(monkeypatch): from app.core import config monkeypatch.setattr(config.settings, "opik_trace_v2_enabled", True) set_opik_context({"thread_id": "ep-abc", "tags": ["step:scene_detail"]}) md = _build_opik_metadata("scene_detail")["opik"] assert md["thread_id"] == "ep-abc" def test_v2_attaches_parent_when_trace_open(monkeypatch): from app.core import config from app.modules.llm import opik_trace monkeypatch.setattr(config.settings, "opik_trace_v2_enabled", True) handle = opik_trace.TraceHandle(uid="0190-parent", name="still:S1sh1") token = opik_trace.bind_trace(handle) try: md = _build_opik_metadata("still_recipe")["opik"] assert md["current_span_data"] == {"trace_id": "0190-parent"} finally: opik_trace.reset_trace(token) def test_v2_without_parent_has_no_span_data(monkeypatch): """부모가 없으면 litellm 이 지금처럼 자기 trace 를 만든다 — 기록이 산다.""" from app.core import config monkeypatch.setattr(config.settings, "opik_trace_v2_enabled", True) md = _build_opik_metadata("scene_detail")["opik"] assert "current_span_data" not in md ``` - [x] **Step 2: 실패를 확인한다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_opik_metadata_keys.py -v` Expected: `test_v2_*` 3건 FAIL (`KeyError: 'current_span_data'` 등), 나머지 PASS - [x] **Step 3: `_build_opik_metadata` 를 고친다** `backend/app/modules/llm/llm_client.py:145` 의 `return metadata` **바로 앞**에 넣는다. ```python # ── v2 (2026-08-23): litellm 이 실제로 읽는 키만 채운다 ────────────── # litellm 이 metadata["opik"] 에서 읽는 것은 넷뿐이다: # project_name · current_span_data · tags · thread_id # 우리가 싣던 `trace_name` 은 litellm 소스에 0회 — 죽은 키였다. # `session_id` 도 litellm 은 안 본다(thread_id 를 본다) — 그래서 텍스트 # 호출은 thread 가 통째로 없었다(실측: 76%가 None). try: from app.core.config import settings if getattr(settings, "opik_trace_v2_enabled", False): from app.modules.llm.opik_trace import current_trace opik_block = metadata["opik"] # 부모 trace 가 열려 있으면 litellm 은 trace 를 만들지 않고 # span 만 붙인다 → `chat.completion` 이 사라진다. parent = current_trace() if parent is not None: opik_block["current_span_data"] = {"trace_id": parent.uid} # 죽은 키는 내보내지 않는다 — trace metadata 만 더럽힌다. opik_block.pop("trace_name", None) opik_block.pop("session_id", None) except Exception as exc: # noqa: BLE001 — 기록은 본 작업을 안 막는다 logger.debug("_build_opik_metadata v2 배선 실패 (non-fatal): %s", exc) return metadata ``` - [x] **Step 4: 시험이 통과하는지 본다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_opik_metadata_keys.py -v` Expected: PASS (6건) - [x] **Step 5: commit** ```bash git add backend/app/modules/llm/llm_client.py backend/tests/unit/test_opik_metadata_keys.py git commit -m "fix(opik): litellm 이 읽는 키를 올바로 채운다 (v2) litellm 은 metadata['opik'] 에서 project_name·current_span_data·tags· thread_id 넷만 읽는다. 우리가 싣던 trace_name 은 litellm 소스에 0회 등장하는 죽은 키였고, session_id 도 litellm 은 안 본다(thread_id 를 본다) — 그래서 텍스트 호출의 thread 가 76% None 이었다. 부모 trace 가 열려 있으면 current_span_data 로 붙여 litellm 이 trace 를 안 만들게 한다. 설정 OFF 면 지금 모양 그대로." ``` --- ### Task 5: thread_id 를 에피소드 단위로 — 재개해도 안 쪼개진다 **Files:** - Modify: `backend/app/core/step_runner.py:1598-1636` (`build_opik_metadata`) - Modify: `backend/app/modules/llm/opik_trace.py` (`episode_thread_id` 추가) - Test: `backend/tests/unit/test_opik_thread_grouping.py` (신규) **Interfaces:** - Consumes: `build_axis_tags` (Task 2) - Produces: `episode_thread_id(*, project_name: str, episode_title: str, episode_id: str) -> str` **배경 (실측):** `build_opik_context`(`backend/app/services/analysis_dispatch_service.py:47`)는 부를 때마다 `run_tag` 를 새로 만든다(`ts` = 현재 시각, `uuid4()[:8]`). `backend/app/api/v1/steps.py:253` 이 **단일 스텝 호출마다** 이것을 부른다. 215샷 주행을 30번 재개하면 thread 가 31개로 갈린다. 키 이름만 고쳐선 「주행 하나를 통째로」가 안 된다. - [x] **Step 1: 실패하는 시험을 쓴다** ```python # backend/tests/unit/test_opik_thread_grouping.py """주행 묶음 — 재개해도 하나여야 한다.""" import pytest from app.modules.llm.opik_trace import episode_thread_id def test_thread_id_is_stable_across_calls(): """같은 에피소드면 몇 번을 불러도 같은 값 — 재개가 쪼개지 않는다.""" a = episode_thread_id(project_name="마지막 임무", episode_title="1회", episode_id="f5372927-bbec-405d-ad2c-100d587f5373") b = episode_thread_id(project_name="마지막 임무", episode_title="1회", episode_id="f5372927-bbec-405d-ad2c-100d587f5373") assert a == b assert "f5372927" in a def test_thread_id_differs_per_episode(): a = episode_thread_id(project_name="p", episode_title="1회", episode_id="aaaaaaaa-0000-0000-0000-000000000000") b = episode_thread_id(project_name="p", episode_title="2회", episode_id="bbbbbbbb-0000-0000-0000-000000000000") assert a != b def test_thread_id_survives_missing_names(): """이름이 비어도 값이 나온다 — 묶음이 사라지는 것이 최악이다.""" t = episode_thread_id(project_name="", episode_title="", episode_id="cccccccc-0000-0000-0000-000000000000") assert t assert "cccccccc" in t ``` 그리고 `StepRunner` 쪽: ```python def _runner(**ctx): from app.core.step_runner import StepRunner r = StepRunner.__new__(StepRunner) r.project_id = "proj-1" r.episode_id = "f5372927-bbec-405d-ad2c-100d587f5373" r.step_id = "scene_detail" r.opik_context = ctx return r def test_v1_metadata_unchanged(monkeypatch): """설정 OFF 면 지금 그대로 — session_id·맨 이름 태그.""" from app.core import config monkeypatch.setattr(config.settings, "opik_trace_v2_enabled", False) m = _runner(run_tag="RT-1", project_name="마지막 임무", episode_title="1회").build_opik_metadata() assert m["session_id"] == "RT-1" assert m["tags"][0] == "scene_detail" assert "마지막 임무" in m["tags"] def test_v2_uses_thread_id_and_axis_tags(monkeypatch): from app.core import config monkeypatch.setattr(config.settings, "opik_trace_v2_enabled", True) m = _runner(run_tag="RT-1", project_name="마지막 임무", episode_title="1회").build_opik_metadata() assert "session_id" not in m assert "trace_name" not in m assert m["thread_id"].endswith("f5372927") # 축 태그만 — 한글 이름은 태그에서 빠지고 metadata 로 간다 assert m["tags"] == ["step:scene_detail"] assert "마지막 임무" not in m["tags"] assert m["project_name"] == "마지막 임무" assert m["episode_title"] == "1회" # run_tag 는 버리지 않는다 — 한 dispatch 를 따로 볼 때 쓴다 assert m["run_tag"] == "RT-1" def test_v2_thread_id_survives_without_run_tag(monkeypatch): """run_tag 없이 부른 스텝도 같은 주행에 묶여야 한다. 지금은 run_tag 가 있어야만 session_id 를 실어서, 그것 없이 도는 스텝은 묶음 밖으로 떨어졌다. """ from app.core import config monkeypatch.setattr(config.settings, "opik_trace_v2_enabled", True) m = _runner(project_name="마지막 임무", episode_title="1회").build_opik_metadata() assert m["thread_id"].endswith("f5372927") ``` - [x] **Step 2: 실패를 확인한다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_opik_thread_grouping.py -v` Expected: FAIL — `ImportError: cannot import name 'episode_thread_id'` - [x] **Step 3: `episode_thread_id` 를 만든다** `backend/app/modules/llm/opik_trace.py` 에 이어 붙인다. ```python def episode_thread_id( *, project_name: str, episode_title: str, episode_id: str ) -> str: """주행 묶음 키 — **에피소드 단위**로 고정한다. 지금은 `build_opik_context` 가 부를 때마다 run_tag 를 새로 만들고 (`analysis_dispatch_service.py:47`), 단일 스텝 API 가 그것을 스텝마다 부른다(`api/v1/steps.py:253`). 215샷 주행을 30번 재개하면 thread 가 31개로 갈린다. 에피소드에서 바로 만들면 재개·재기동·단일 스텝 호출이 전부 같은 값이 되고, 저장할 것도 없다. 한 dispatch 를 따로 보고 싶으면 metadata 의 `run_tag` 로 가른다. """ head = "_".join(x for x in (project_name or "", episode_title or "") if x) tail = (episode_id or "")[:8] or "unknown" return f"{head}_{tail}" if head else tail ``` - [x] **Step 4: `build_opik_metadata` 에 v2 분기를 넣는다** `backend/app/core/step_runner.py:1636` 의 `return meta` **앞**에 넣는다. 기존 v1 코드는 손대지 않는다. ```python # ── v2 (2026-08-23): 키 이름을 litellm 에 맞추고 축을 가른다 ── try: from app.core.config import settings if getattr(settings, "opik_trace_v2_enabled", False): from app.modules.llm.opik_trace import ( build_axis_tags, episode_thread_id) # 죽은 키를 뺀다. litellm 은 trace_name 을 안 읽고 # session_id 도 안 본다(thread_id 를 본다). meta.pop("trace_name", None) meta.pop("session_id", None) meta.pop("project_name_tag", None) meta.pop("episode_title_tag", None) meta["thread_id"] = episode_thread_id( project_name=ctx.get("project_name", ""), episode_title=ctx.get("episode_title", ""), episode_id=self.episode_id or "", ) # 이름은 태그가 아니라 metadata 로 — 한글이고 카디널리티가 크다. if ctx.get("project_name"): meta["project_name"] = ctx["project_name"] if ctx.get("episode_title"): meta["episode_title"] = ctx["episode_title"] if ctx.get("run_tag"): meta["run_tag"] = ctx["run_tag"] # 태그는 축만. extra_tags 는 status 축으로 받는다 # (호출자가 "retry"·"gpt_fallback" 같은 낮은 카디널리티만 준다). axis = build_axis_tags(step=self.step_id) for t in (extra_tags or []): tag = f"status:{t}" if tag not in axis: axis.append(tag) meta["tags"] = axis except Exception as exc: # noqa: BLE001 logger.debug("build_opik_metadata v2 실패 (non-fatal): %s", exc) return meta ``` - [x] **Step 5: 시험이 통과하는지 본다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_opik_thread_grouping.py -v` Expected: PASS (7건) - [x] **Step 6: commit** ```bash git add backend/app/core/step_runner.py backend/app/modules/llm/opik_trace.py \ backend/tests/unit/test_opik_thread_grouping.py git commit -m "fix(opik): thread_id 를 에피소드 단위로 — 재개해도 안 쪼개진다 build_opik_context 가 부를 때마다 run_tag 를 새로 만들고 단일 스텝 API 가 스텝마다 그것을 부른다. 215샷을 30번 재개하면 thread 가 31개로 갈린다. 에피소드에서 바로 만들면 재개·재기동·단일 스텝이 전부 같은 값이 되고 저장할 것도 없다. 한 dispatch 는 metadata.run_tag 로 가른다. 태그에서 한글 프로젝트·에피소드 이름을 빼 metadata 로 내렸다(카디널리티)." ``` --- ### Task 6: 스텝 경계에서 trace 를 연다 **Files:** - Modify: `backend/app/core/step_runner.py:1241-1253`, `1374-1375` - Test: `backend/tests/unit/test_opik_step_trace.py` (신규) **Interfaces:** - Consumes: `open_trace` (Task 3), `build_opik_metadata` (Task 5) - Produces: 스텝 실행 동안 `current_trace().name == f"step:{step_id}"` - [x] **Step 1: 실패하는 시험을 쓴다** ```python # backend/tests/unit/test_opik_step_trace.py """스텝 경계에서 trace 가 열리고 닫힌다.""" import pytest from app.modules.llm.opik_trace import current_trace @pytest.fixture def _fake_client(monkeypatch): sent = [] class _T: def __init__(self, **kw): sent.append(kw) def end(self, **kw): pass def update(self, **kw): pass class _C: def trace(self, **kw): return _T(**kw) from app.modules.llm import opik_trace monkeypatch.setattr(opik_trace, "_get_client", lambda: _C()) return sent def test_step_scope_opens_named_trace(monkeypatch, _fake_client): from app.core import config from app.modules.llm.opik_trace import open_trace monkeypatch.setattr(config.settings, "opik_trace_v2_enabled", True) seen = {} with open_trace(name="step:scene_detail", tags=["step:scene_detail"], metadata={"thread_id": "ep_f5372927"}, thread_id="ep_f5372927"): seen["name"] = current_trace().name assert seen["name"] == "step:scene_detail" assert current_trace() is None assert _fake_client[0]["name"] == "step:scene_detail" assert _fake_client[0]["thread_id"] == "ep_f5372927" def test_step_trace_helper_is_noop_when_disabled(monkeypatch, _fake_client): from app.core import config from app.core.step_runner import _step_trace_scope monkeypatch.setattr(config.settings, "opik_trace_v2_enabled", False) class _R: step_id = "scene_detail" def build_opik_metadata(self): return {"tags": ["scene_detail"]} with _step_trace_scope(_R()): assert current_trace() is None assert _fake_client == [] ``` - [x] **Step 2: 실패를 확인한다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_opik_step_trace.py -v` Expected: FAIL — `ImportError: cannot import name '_step_trace_scope'` - [x] **Step 3: 헬퍼를 만들고 `_execute_and_finalize` 를 감싼다** `backend/app/core/step_runner.py` 의 모듈 상단(클래스 밖)에 헬퍼를 둔다. ```python from contextlib import contextmanager @contextmanager def _step_trace_scope(runner): """스텝 하나를 Opik trace 하나로 연다. 설정이 꺼져 있으면 아무것도 안 한다(바이트 동일). 여는 데 실패해도 본 스텝은 그대로 돈다 — `open_trace` 가 None 을 줄 뿐이다. """ from app.modules.llm.opik_trace import open_trace meta = dict(runner.build_opik_metadata() or {}) tags = list(meta.pop("tags", []) or []) thread_id = meta.get("thread_id") with open_trace( name=f"step:{runner.step_id}", tags=tags, metadata=meta, thread_id=thread_id, ): yield ``` `_execute_and_finalize` 의 본문을 이 scope 로 감싼다. `set_opik_context` 는 **그대로 둔다** — 그것은 하위 호출이 읽는 별개 통로다. ```python self._update_step_run_strict("running") logger.info( "Step %s started (run_id=%s, model=%s)", self.step_id, self.run_id, self._resolve_model(), ) set_opik_context(self.build_opik_metadata()) with _step_trace_scope(self): return self._execute_and_finalize_inner(execute_mode) ``` 기존 `try: ... finally: set_opik_context(None)` 본문 전체를 `_execute_and_finalize_inner(self, execute_mode)` 라는 메서드로 그대로 옮긴다. **옮기기만 한다 — 한 줄도 바꾸지 않는다.** - [x] **Step 4: 시험이 통과하는지 본다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_opik_step_trace.py -v` Expected: PASS (2건) - [x] **Step 5: 회귀 — StepRunner 시험 전체를 태운다** Run: `cd backend && ./.venv/bin/pytest tests/ -q -k "step_runner or step_run or resume" 2>&1 | tail -8` Expected: 실패 건수가 이 태스크 전과 같다 (늘었으면 옮기다 뭔가 바뀐 것) - [x] **Step 6: commit** ```bash git add backend/app/core/step_runner.py backend/tests/unit/test_opik_step_trace.py git commit -m "feat(opik): 스텝 경계에서 trace 를 연다 StepRunner 의 claim~해제 구간이 이미 스텝 경계다 — 새 경계를 안 만든다. _execute_and_finalize 본문을 _execute_and_finalize_inner 로 그대로 옮기고 _step_trace_scope 로 감쌌다(옮기기만, 내용 변경 0). set_opik_context 는 그대로 둔다 — 하위 호출이 읽는 별개 통로다." ``` --- ### Task 7: 샷 경계에서 trace 를 연다 **Files:** - Modify: `backend/app/services/image_capture/context.py:83-131` - Test: `backend/tests/services/image_capture/test_shot_trace_scope.py` (신규) **Interfaces:** - Consumes: `open_trace`, `build_axis_tags` (Task 2·3) - Produces: `generation_context` 안에서 `current_trace().name` 이 `still: · ` 또는 `stage:` **설계 근거:** `generation_context` 는 `still_id · scene_index · shot_index · entity_id · stage` 를 들고 있고 worker 전파기(`bind_current_generation_context`)까지 있다. **그릇으로는 이것이 맞다.** > ★**이 태스크만으로는 프로덕션 본류를 못 덮는다.** `generation_context` 는 > 14곳에서 열리지만 **`still_recipe_service.py` 에는 0건**이다 — > 샷을 찍는 본류에 capture scope 가 아예 없다. 실측: still-recipe 계열 Opik > trace **529건 전수에 `still_id` 가 없다(100%)**. > > 그래서 **Task 7-B 가 이 태스크와 한 몸으로 나가야 한다** — 실제 샷 루프에 > scope 를 새로 넣는 일. 7 만 하고 7-B 를 빠뜨리면 모든 샷이 스텝 trace 를 > 부모로 삼아 **uid 가 전부 같아진다**(Task 12 가 깨진다). - [x] **Step 1: 실패하는 시험을 쓴다** ```python # backend/tests/services/image_capture/test_shot_trace_scope.py """샷 경계 = generation_context. 이미 있는 경계를 쓴다.""" import pytest from app.modules.llm.opik_trace import current_trace @pytest.fixture def _fake_client(monkeypatch): sent = [] class _T: def __init__(self, **kw): sent.append(kw) def end(self, **kw): pass def update(self, **kw): pass class _C: def trace(self, **kw): return _T(**kw) from app.modules.llm import opik_trace monkeypatch.setattr(opik_trace, "_get_client", lambda: _C()) return sent @pytest.fixture(autouse=True) def _v2(monkeypatch): from app.core import config monkeypatch.setattr(config.settings, "opik_trace_v2_enabled", True) def test_shot_scope_opens_named_trace(monkeypatch, _fake_client, tmp_path): from app.services.image_capture.context import generation_context monkeypatch.setenv("PROJECTS_DIR", str(tmp_path)) with generation_context("proj-1", "ep-1", stage="still_recipe", still_id="abcdef01-2222-3333-4444-555555555555", scene_index=12, shot_index=3): h = current_trace() assert h is not None assert h.name == "still:abcdef01 · still_recipe" assert current_trace() is None kw = _fake_client[0] assert "step:still_recipe" in kw["tags"] assert kw["metadata"]["still_id"] == "abcdef01-2222-3333-4444-555555555555" assert kw["metadata"]["scene_index"] == 12 assert kw["metadata"]["shot_index"] == 3 assert kw["metadata"]["project_id"] == "proj-1" def test_scope_without_still_id_uses_stage(_fake_client, monkeypatch, tmp_path): from app.services.image_capture.context import generation_context monkeypatch.setenv("PROJECTS_DIR", str(tmp_path)) with generation_context("proj-1", "ep-1", stage="background_render"): assert current_trace().name == "stage:background_render" def test_nested_shot_scope_wins(_fake_client, monkeypatch, tmp_path): """스텝 trace 안에 샷 trace 가 열리면 안쪽이 부모다.""" from app.modules.llm.opik_trace import open_trace from app.services.image_capture.context import generation_context monkeypatch.setenv("PROJECTS_DIR", str(tmp_path)) with open_trace(name="step:scene_image_pipeline", tags=[], metadata={}, thread_id="ep_x") as step_h: with generation_context("p", "e", stage="still_recipe", still_id="99999999-0000-0000-0000-000000000000"): assert current_trace().uid != step_h.uid assert current_trace() is step_h def test_disabled_opens_nothing(monkeypatch, _fake_client, tmp_path): from app.core import config from app.services.image_capture.context import generation_context monkeypatch.setattr(config.settings, "opik_trace_v2_enabled", False) monkeypatch.setenv("PROJECTS_DIR", str(tmp_path)) with generation_context("p", "e", stage="still_recipe", still_id="x"): assert current_trace() is None assert _fake_client == [] ``` - [x] **Step 2: 실패를 확인한다** Run: `cd backend && ./.venv/bin/pytest tests/services/image_capture/test_shot_trace_scope.py -v` Expected: FAIL — `assert None is not None` - [x] **Step 3: `generation_context` 에 trace scope 를 겹친다** `backend/app/services/image_capture/context.py` 의 `token = _gen_ctx.set(ctx)` 부터 끝까지를 바꾼다. ```python token = _gen_ctx.set(ctx) try: with _shot_trace_scope(ctx): yield queue finally: try: queue.flush() except Exception: # pragma: no cover - flush 자체가 non-fatal raise 안 함 logger.warning("generation_context: flush failed (non-fatal)", exc_info=True) _gen_ctx.reset(token) ``` 같은 파일 아래(모듈 수준)에 헬퍼를 둔다. ```python @contextmanager def _shot_trace_scope(ctx: GenerationContext) -> Iterator[None]: """이 capture scope 를 Opik trace 하나로 연다. ★새 경계를 만들지 않는다 — 이 scope 가 이미 샷 경계다. still_id·scene_index·shot_index 를 이미 들고 있고 worker 전파기도 있다. 설정이 꺼져 있으면 아무것도 안 한다(바이트 동일). """ from app.modules.llm.opik_trace import build_axis_tags, open_trace if ctx.still_id: name = f"still:{str(ctx.still_id)[:8]} · {ctx.stage}" elif ctx.entity_id: name = f"entity:{str(ctx.entity_id)[:8]} · {ctx.stage}" else: name = f"stage:{ctx.stage}" meta = { k: v for k, v in ( ("project_id", ctx.project_id), ("episode_id", ctx.episode_id), ("still_id", ctx.still_id), ("scene_index", ctx.scene_index), ("shot_index", ctx.shot_index), ("entity_id", ctx.entity_id), ("step", ctx.stage), ) if v is not None } # 주행 묶음은 부모(스텝 trace)에서 물려받는다. 부모가 없으면 없는 채로 # 둔다 — 없는 값을 지어내면 묶음이 거짓말을 한다. from app.modules.llm.opik_trace import current_trace parent = current_trace() thread_id = parent.thread_id if parent is not None else None with open_trace(name=name, tags=build_axis_tags(step=ctx.stage), metadata=meta, thread_id=thread_id): yield ``` 파일 상단 import 에 `Iterator` 가 이미 있는지 확인한다(83줄 `-> Iterator["CaptureQueue"]` 에서 쓰므로 있다). - [x] **Step 4: 시험이 통과하는지 본다** Run: `cd backend && ./.venv/bin/pytest tests/services/image_capture/ -v 2>&1 | tail -20` Expected: 새 시험 4건 PASS, 기존 image_capture 시험 실패 증가 0 - [x] **Step 5: commit** ```bash git add backend/app/services/image_capture/context.py \ backend/tests/services/image_capture/test_shot_trace_scope.py git commit -m "feat(opik): capture scope 에 trace 를 겹친다 generation_context 는 still_id·scene_index·shot_index 를 들고 있고 worker 전파기까지 있어 그릇으로 맞다. 그 위에 trace scope 를 겹친다. ★이것만으로는 프로덕션 본류를 못 덮는다 — still_recipe_service.py 에는 generation_context 가 0건이다(Task 7-B). 둘이 한 몸으로 나가야 한다. thread_id 는 부모(스텝 trace)에서 물려받는다 — 부모가 없으면 없는 채로 둔다. 없는 값을 지어내면 묶음이 거짓말을 한다." ``` --- ### Task 7-B: ★프로덕션 샷 루프에 capture scope 를 새로 넣는다 **Files:** - Modify: `backend/app/services/image_capture/context.py` (`capture` 인자 + `capture_enabled` 칸) - Modify: `backend/app/services/image_capture/sink.py:29-40` (`_enqueue` 가 그 칸을 본다) - Modify: `backend/app/services/still_recipe_service.py:2272` (샷 루프) - Test: `backend/tests/unit/test_still_recipe_shot_scope.py` (신규) - Test: `backend/tests/services/image_capture/test_capture_disabled_scope.py` (신규) **Interfaces:** - Consumes: `generation_context` (`image_capture/context.py:83`), Task 7 의 trace 겹치기 - Produces: - `GenerationContext.capture_enabled: bool = True` (frozen dataclass 새 칸) - `generation_context(..., capture: bool = True)` - 샷마다 `current_context().still_id` 와 `current_trace()` 가 그 샷의 것 > ★**scope 를 여는 것은 capture 를 켜는 것이다 — 2026-08-23 Codex 재리뷰로 교정.** > > `sink.py` 의 문서화 문자열이 명시한다: > 「★default-capture-off(중복 0 의 핵심): `generation_context` scope 가 열리지 > 않은 호출(=최종물 경로 등)은 ctx 가 None → 저장하지 않는다. 최종물은 기존 > `_register_image_assets` 가 `image_asset` 을 만들므로 sink 가 또 만들면 > 중복이 된다.」 > > 그냥 scope 를 열면 롤·수리·변환 성공분마다 `asset_type=generated`, > `is_intermediate=True` 행과 spool 파일이 **새로 생긴다**(`queue.py:91-179`). > 첫 판의 「큐가 비어 있을 것」은 **틀렸다.** > > **이 일은 기록 체계화지 자산 늘리기가 아니다.** 사용자가 요구한 것은 > 「어느 샷 것인지 알 수 있게」이지 「중간물을 전부 자산으로 만들라」가 아니다. > 그래서 **capture 를 끈 scope** 를 쓴다 — 신원과 trace 만 얻고 자산 동작은 > 바이트 동일로 둔다. > > (롤을 자산으로 등록하는 것은 그 자체로 값이 있을 수 있지만 **별도 판**이다. > Task 11 이 그것에 기대고 있었으므로 함께 좁힌다.) **배경 (실측):** 이것이 Codex 리뷰의 제일 큰 지적이었다. - `still_recipe_service.py` 에 `generation_context` **0건** - `scene_image_service.py:404` 가 `run_still_recipe_generation` 을 scope 없이 부름 - 샷 루프 `for i, s in enumerate(ordered):` (2272줄)에도 없음 - **still-recipe 계열 Opik trace 529건 전수에 `still_id` 없음(100%)** 부수 이득: 이 scope 가 열리면 `ambient_call_meta()` 가 읽어 **DB 호출 기록에도** `still_id`·`scene_index`·`shot_index` 가 실린다(지금은 안 실린다). - [x] **Step 1: 실패하는 시험을 쓴다** ```python # backend/tests/unit/test_still_recipe_shot_scope.py """프로덕션 샷 루프에 capture scope 가 있어야 한다. 2026-08-23 실측: still_recipe_service.py 에 generation_context 가 0건이라 still-recipe 계열 Opik trace 529건 전수에 still_id 가 없었다(100%). """ import ast from pathlib import Path SRC = (Path(__file__).resolve().parents[2] / "app" / "services" / "still_recipe_service.py") def test_shot_loop_opens_capture_scope(): """샷 루프 안에서 generation_context 가 열려야 한다. ★문자열 검사가 아니라 AST 로 본다 — 주석에 이름만 있어도 통과하면 안 된다(scene_image_service.py:638 이 그 예다). """ tree = ast.parse(SRC.read_text(encoding="utf-8")) opens = [ n for n in ast.walk(tree) if isinstance(n, ast.With) for item in n.items if isinstance(item.context_expr, ast.Call) and getattr(item.context_expr.func, "id", "") == "generation_context" ] assert opens, "샷 루프에 generation_context 가 없다" def test_scope_carries_shot_identity(): """still_id·scene_index·shot_index 를 넘겨야 한다 — 안 넘기면 빈 채로 남는다.""" src = SRC.read_text(encoding="utf-8") tree = ast.parse(src) kwsets = [] for n in ast.walk(tree): if isinstance(n, ast.Call) and \ getattr(n.func, "id", "") == "generation_context": kwsets.append({k.arg for k in n.keywords}) assert kwsets, "generation_context 호출이 없다" assert any({"still_id", "scene_index", "shot_index"} <= s for s in kwsets), \ f"샷 신원을 안 넘긴다: {kwsets}" def test_scope_does_not_rename_the_step(): """★stage 는 지금 쓰이는 스텝 이름 그대로여야 한다. resolve_step_name(image_tracer.py:316)이 ambient["stage"] 를 1순위로 쓴다. 여기에 새 이름을 넣으면 이 경로의 llm_call_log.step_name 이 통째로 바뀐다 — 지금 그 이름('scene_image_pipeline')으로 쌓인 것이 3,707행이라 신·구 대조가 끊긴다. 샷 신원은 still_id 가, 세부 단계는 op: 태그가 말한다. stage 로 말하지 않는다. """ tree = ast.parse(SRC.read_text(encoding="utf-8")) stages = [] for n in ast.walk(tree): if isinstance(n, ast.Call) and \ getattr(n.func, "id", "") == "generation_context": for kw in n.keywords: if kw.arg == "stage" and isinstance(kw.value, ast.Constant): stages.append(kw.value.value) assert stages, "stage 를 상수로 안 넘긴다" assert "scene_image_pipeline" in stages, \ f"stage 가 스텝 이름과 다르다: {stages} — step_name 이 갈린다" def test_scope_disables_capture(): """★capture=False 여야 한다 — 안 그러면 중간물 자산이 무더기로 생긴다.""" tree = ast.parse(SRC.read_text(encoding="utf-8")) for n in ast.walk(tree): if isinstance(n, ast.Call) and \ getattr(n.func, "id", "") == "generation_context": caps = [kw.value for kw in n.keywords if kw.arg == "capture"] assert caps, "capture 를 안 넘긴다 (기본 True = 자산이 생긴다)" assert isinstance(caps[0], ast.Constant) and caps[0].value is False ``` - [x] **Step 2: 실패를 확인한다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_still_recipe_shot_scope.py -v` Expected: FAIL — `AssertionError: 샷 루프에 generation_context 가 없다` - [x] **Step 3: capture 를 끌 수 있게 만든다** `backend/app/services/image_capture/context.py` — `GenerationContext` 에 칸을 하나 보태고(맨 끝, 기본 `True` 라 기존 생성자는 안 깨진다) `generation_context` 에 인자를 하나 보탠다. ```python @dataclass(frozen=True) class GenerationContext: ... entity_id: Optional[str] = None #: 이 scope 가 중간물을 자산으로 포착하는가 (2026-08-23). #: ★scope 를 여는 것은 그 자체로 capture 를 켜는 것이다(sink 의 #: default-capture-off 규약). 신원·trace 만 필요하고 자산은 늘리고 #: 싶지 않은 자리를 위해 끌 수 있게 둔다. capture_enabled: bool = True ``` ```python @contextmanager def generation_context( project_id: str, episode_id: Optional[str], stage: str, *, still_id: Optional[str] = None, scene_index: Optional[int] = None, shot_index: Optional[int] = None, entity_id: Optional[str] = None, capture: bool = True, ) -> Iterator["CaptureQueue"]: ``` `ctx = GenerationContext(...)` 에 `capture_enabled=capture` 를 넘긴다. `backend/app/services/image_capture/sink.py` 의 `_enqueue` 에 한 줄: ```python ctx = current_context() if ctx is None: CAPTURE_DIAG["skipped_no_context"] += 1 return if not getattr(ctx, "capture_enabled", True): # 신원·trace 만 얻으려고 연 scope — 자산을 만들지 않는다. CAPTURE_DIAG["skipped_capture_off"] += 1 return ``` `CAPTURE_DIAG` 에 `skipped_capture_off` 키를 보탠다(`queue.py`). - [x] **Step 4: capture-off 시험을 쓴다** ```python # backend/tests/services/image_capture/test_capture_disabled_scope.py """capture 를 끈 scope — 신원·trace 는 얻고 자산은 안 만든다. ★scope 를 여는 것은 그 자체로 capture 를 켜는 것이다(sink 의 default-capture-off 규약). still-recipe 는 자기 손으로 자산을 저장하므로 여기서 또 포착하면 중간물 행이 무더기로 새로 생긴다 — 이 일은 기록 체계화지 자산 늘리기가 아니다. """ from app.services.image_capture.context import ( current_context, generation_context) def test_capture_off_context_still_carries_identity(tmp_path, monkeypatch): monkeypatch.setenv("PROJECTS_DIR", str(tmp_path)) with generation_context("p", "e", stage="still_recipe", still_id="abc", scene_index=1, shot_index=2, capture=False): ctx = current_context() assert ctx.still_id == "abc" assert ctx.scene_index == 1 assert ctx.capture_enabled is False def test_capture_off_queue_stays_empty(tmp_path, monkeypatch): monkeypatch.setenv("PROJECTS_DIR", str(tmp_path)) from app.services.image_capture.sink import capture_generated_image with generation_context("p", "e", stage="still_recipe", still_id="abc", capture=False) as q: capture_generated_image(b"\x89PNG-fake", role="roll", disposition="diagnostic") assert q._items == [] def test_capture_on_is_unchanged(tmp_path, monkeypatch): """기존 자리는 그대로 포착한다 — 바이트 동일.""" monkeypatch.setenv("PROJECTS_DIR", str(tmp_path)) from app.services.image_capture.sink import capture_generated_image with generation_context("p", "e", stage="background_render", still_id="abc") as q: capture_generated_image(b"\x89PNG-fake", role="bg", disposition="accepted") assert len(q._items) == 1 ``` - [x] **Step 5: 샷 루프 본문을 scope 로 감싼다 — capture 는 끈다** `still_recipe_service.py:2272` 의 `for i, s in enumerate(ordered):` 안. **여는 자리**는 Codex 가 짚은 대로 `still_id` 확정 직후가 아니라, **완료 완전-skip / no-record skip(`2283-2305`)과 `jit_snapshot_before`(`2322-2323`)가 끝난 직후**다 — 안 도는 샷에 trace 를 열 이유가 없고, JIT 스냅숏은 scope 와 무관하다. ```python for i, s in enumerate(ordered): si, shi = int(s["scene_index"]), int(s["shot_index"]) tag = tag_of(si, shi) still_id = s["id"] if execution_scope is not None and still_id not in execution_scope: ... # 기존 skip — 그대로, scope 밖 continue ... # 완료 완전-skip / no-record skip (2283-2305) — scope 밖 ... # jit_snapshot_before (2322-2323) — scope 밖 # ── 샷 경계 (2026-08-23) ─────────────────────────────────────── # ★이 자리에 capture scope 가 없었다. 그래서 still-recipe 계열 # Opik trace 529건 전수에 still_id 가 없었다(100%) — 「이 기록이 # 어느 샷 것이냐」를 한 건도 못 물었다. # ★capture=False 다. scope 를 여는 것은 그 자체로 중간물 포착을 # 켜는 것이라(sink 의 default-capture-off 규약), 그냥 열면 롤· # 수리·변환 성공분마다 is_intermediate 자산 행과 spool 파일이 # 새로 생긴다. 이 일은 기록 체계화지 자산 늘리기가 아니다. from app.services.image_capture.context import generation_context with generation_context( project_id, episode_id, # ★stage 는 반드시 **지금 쓰이는 스텝 이름 그대로** 다. # resolve_step_name(image_tracer.py:316)이 ambient["stage"] 를 # **1순위**로 쓰므로, 여기에 "still_recipe" 를 넣으면 이 경로의 # llm_call_log.step_name 이 통째로 바뀐다 — 지금 그 이름으로 # 쌓인 것이 3,707행이라 신·구 대조가 끊긴다. # 샷 신원은 still_id 가, 세부 단계는 op: 태그가 말한다. stage="scene_image_pipeline", still_id=still_id, scene_index=si, shot_index=shi, capture=False, ): <기존 샷 본문 전체를 여기로 들여쓴다 — 한 줄도 바꾸지 않는다> ``` ★**주의 셋** 1. **`continue`·`break` 가 scope 안에 있어도 된다** — `with` 는 정상 종료로 처리된다 2. **`return` 이 본문에 있으면** scope 가 닫히고 나간다 — 정상이다 3. **첫 유료 가능 호출(confined 판정 `2598`·bg plate 판정 `2697`)이 scope 안**이어야 그 호출들도 샷 신원을 얻는다 — 여는 자리를 더 뒤로 미루지 않는다 - [x] **Step 6: 시험이 통과하는지 본다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_still_recipe_shot_scope.py tests/services/image_capture/ -v 2>&1 | tail -20` Expected: 새 시험 5건 PASS, 기존 image_capture 시험 실패 증가 0 - [x] **Step 7: 자산이 안 늘었는지 실측한다 (동작 변화 점검)** ```bash cd backend && ./.venv/bin/pytest tests/ -q -k "still_recipe or capture" 2>&1 | tail -8 ``` Expected: 실패 증가 0. 그리고 새 주행 첫 샷 몇 개에서: ```bash PGPASSWORD=theroad_dev_2026 psql -h localhost -U theroad -d theroad -c " SELECT stage, asset_type, is_intermediate, count(*) n FROM image_asset WHERE created_at > '<이 주행 시작 시각>' GROUP BY 1,2,3 ORDER BY n DESC LIMIT 10;" ``` Expected: `stage='still_recipe'` 이면서 `is_intermediate=true` 인 행이 **0건**. 있으면 `capture=False` 가 안 먹은 것이다. - [x] **Step 8: 실측 — 관문 0·5·15 의 전제** 새 주행 첫 샷 뒤: ```bash python3 -c " import json,urllib.request BASE='http://192.168.133.87:5173/api'; PID='<프로젝트 id>' r=urllib.request.Request(f'{BASE}/v1/private/traces?project_id={PID}&page=1&size=50', headers={'Comet-Workspace':'default'}) d=json.load(urllib.request.urlopen(r,timeout=30)) n=sum(1 for t in d['content'] if (t.get('metadata') or {}).get('still_id')) print(f'still_id 있는 trace: {n}/{len(d[\"content\"])}') " ``` Expected: **0 이 아니어야 한다.** 지금은 529/529 가 0 이다. - [x] **Step 9: commit** ```bash git add backend/app/services/still_recipe_service.py \ backend/tests/unit/test_still_recipe_shot_scope.py git commit -m "feat(still): 샷 루프에 capture scope 를 넣는다 — 없던 경계다 ★설계가 「샷 경계가 이미 있다」고 전제했는데 틀렸다. generation_context 는 14곳에서 열리지만 still_recipe_service.py 에는 0건이고, 샷을 찍는 본류 (scene_image_service.py:404 → run_still_recipe_generation → 2272줄 루프)에 capture scope 가 아예 없었다. 실측이 그대로 말한다 — still-recipe 계열 Opik trace 529건 전수에 still_id 가 없다(100%). 「이 기록이 어느 샷 것이냐」를 한 건도 못 물었다. 부수 이득: ambient_call_meta 가 이 scope 를 읽어 DB 호출 기록에도 still_id·scene_index·shot_index 가 실린다." ``` --- ### Task 7-C: ★병렬 worker 에 trace 를 전파한다 **Files:** - Modify: `backend/app/modules/llm/opik_trace.py` (`bind_current_trace` 추가) - Modify: `backend/app/modules/pipeline/multiroll_select.py:1385` - Test: `backend/tests/unit/test_opik_trace_worker_propagation.py` (신규) **Interfaces:** - Consumes: `current_trace`, `bind_trace`, `reset_trace` (Task 3) - Produces: `bind_current_trace(fn) -> Callable` — 감싼 함수가 worker thread 에서도 같은 부모를 본다 **배경:** `multiroll_select.py:1372-1385` 는 병렬 롤 worker 에 **budget 과 generation_context 둘만** 명시 전파한다. ```python bound = bind_current_budget(bind_current_generation_context(gen_fn)) ``` trace 핸들은 **세 번째 `ContextVar`** 다. 이 줄에 안 실으면 worker 호출은 부모가 `None` 이라 **다시 낱개 trace 가 된다** — 계층이 병렬 구간에서만 조용히 무너진다. - [x] **Step 1: 실패하는 시험을 쓴다 — ★진짜 thread 를 만든다** ```python # backend/tests/unit/test_opik_trace_worker_propagation.py """병렬 worker 도 같은 부모 trace 를 봐야 한다. ★ContextVar 는 thread 를 안 넘는다. Task 3 의 bind_trace 시험은 같은 thread 안에서만 봤다 — 그것으로는 이 결함을 못 잡는다. 여기서는 진짜 ThreadPoolExecutor 를 돌린다. """ from concurrent.futures import ThreadPoolExecutor import pytest from app.modules.llm.opik_trace import ( TraceHandle, bind_current_trace, bind_trace, current_trace, reset_trace) def test_contextvar_does_not_cross_threads_by_itself(): """전제를 못박는다 — 이것이 참이라 전파기가 필요하다.""" token = bind_trace(TraceHandle(uid="0190-parent", name="still:x")) try: with ThreadPoolExecutor(max_workers=1) as pool: assert pool.submit(current_trace).result() is None finally: reset_trace(token) def test_bind_current_trace_carries_parent_into_worker(): token = bind_trace(TraceHandle(uid="0190-parent", name="still:x")) try: bound = bind_current_trace(lambda: current_trace()) with ThreadPoolExecutor(max_workers=2) as pool: uids = [f.result().uid for f in [pool.submit(bound) for _ in range(4)]] assert uids == ["0190-parent"] * 4 finally: reset_trace(token) def test_worker_restores_after_run(): """worker 안에서 세운 값이 그 thread 에 남으면 다음 작업이 물려받는다.""" token = bind_trace(TraceHandle(uid="0190-a", name="a")) try: bound = bind_current_trace(lambda: current_trace().uid) with ThreadPoolExecutor(max_workers=1) as pool: assert pool.submit(bound).result() == "0190-a" # 같은 thread 를 재사용하는데, 전파기 없이 부르면 비어 있어야 한다 assert pool.submit(current_trace).result() is None finally: reset_trace(token) def test_no_parent_is_not_an_error(): bound = bind_current_trace(lambda: current_trace()) with ThreadPoolExecutor(max_workers=1) as pool: assert pool.submit(bound).result() is None def test_wrapper_is_transparent_to_args_and_exceptions(): token = bind_trace(TraceHandle(uid="0190-a", name="a")) try: def _fn(a, b, *, c): if a == "boom": raise ValueError("터짐") return (a, b, c, current_trace().uid) bound = bind_current_trace(_fn) with ThreadPoolExecutor(max_workers=1) as pool: assert pool.submit(bound, 1, 2, c=3).result() == (1, 2, 3, "0190-a") with pytest.raises(ValueError): pool.submit(bound, "boom", 2, c=3).result() finally: reset_trace(token) ``` - [x] **Step 2: 실패를 확인한다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_opik_trace_worker_propagation.py -v` Expected: FAIL — `ImportError: cannot import name 'bind_current_trace'` - [x] **Step 3: 전파기를 만든다** `backend/app/modules/llm/opik_trace.py` 에 넣는다. `bind_current_generation_context`(`image_capture/context.py:60`)와 **같은 모양**으로 맞춘다 — 옆에 나란히 쓰이므로 다르면 헷갈린다. ```python def bind_current_trace(fn): """지금 trace 를 캡처해 worker thread 안에서 다시 세우는 wrapper. `ContextVar` 는 thread 를 넘지 않는다. 병렬 롤은 `multiroll_select.py:1385` 에서 budget·generation_context 를 명시 전파하는데, trace 핸들은 **세 번째 ContextVar** 라 같이 실어야 한다. 안 실으면 worker 호출이 부모 없이 떨어져 **계층이 병렬 구간에서만 조용히 무너진다.** ★worker 가 끝나면 반드시 되돌린다 — thread pool 은 thread 를 재사용해서, 안 되돌리면 다음 작업이 남의 부모를 물려받는다. """ import functools captured = current_trace() @functools.wraps(fn) def _wrapped(*args, **kwargs): token = _trace_ctx.set(captured) try: return fn(*args, **kwargs) finally: _trace_ctx.reset(token) return _wrapped ``` - [x] **Step 4: 시험이 통과하는지 본다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_opik_trace_worker_propagation.py -v` Expected: PASS (5건) - [x] **Step 5: 병렬 롤 자리에 겹친다** `multiroll_select.py:1385` 를 바꾼다. ```python from app.core.image_call_budget import bind_current_budget from app.modules.llm.opik_trace import bind_current_trace from app.services.image_capture.context import ( bind_current_generation_context, ) # ★셋 다 ContextVar 다 — 하나라도 빠뜨리면 그 축만 조용히 무너진다. bound = bind_current_budget( bind_current_generation_context(bind_current_trace(gen_fn))) ``` - [x] **Step 6: 다른 병렬 자리도 훑는다** 같은 결함이 다른 데 있는지 본다. 있으면 같은 방식으로 겹친다. ```bash grep -rn "bind_current_generation_context\|ThreadPoolExecutor" backend/app --include="*.py" | grep -v test ``` 각 자리에서 `bind_current_trace` 가 함께 있는지 확인한다. **없는 자리를 목록으로 남긴다** — 조용히 빠뜨리지 않기 위해서다. #### 실측 결과 (2026-08-24, AST 로 셌다) `ThreadPoolExecutor` 자리 **35곳**. 그중 전파 현황: | 전파기 | 있는 자리 | |---|---| | `bind_current_trace` | **1** (`multiroll_select.py:1389` — 이 태스크에서 넣은 것) | | `bind_current_generation_context` | 1 (같은 자리) | | `bind_current_budget` | 10 | **즉 34곳은 부모 trace 를 못 본다.** 그중 19곳이 `StepRunner._execute` 안이라 **스텝 trace 가 열린 채** 도는데, worker 는 `current_trace()` 가 `None` 이라 litellm 이 낱개 trace 를 만든다 — 계층이 **텍스트 스텝의 병렬 구간에서만** 안 생긴다. ★같은 자리에서 `set_opik_context`(thread-local)도 thread 를 안 넘는다. 실측 「thread_id 76% None」의 한 갈래일 수 있다 — 확인은 별건. 빠진 34곳(파일:줄 · 함수): ``` app/core/steps/analysis_steps_legacy.py:293 _execute app/core/steps/background_master_plan_step.py:223 _execute app/core/steps/background_prompt_step.py:586 _execute app/core/steps/background_render_step.py:1110 _execute app/core/steps/beat_shot_steps.py:140 _execute app/core/steps/detail_steps.py:1297 _run_shots_with_retry app/core/steps/detail_steps.py:1362 _run_shots_with_retry app/core/steps/detail_steps.py:3926 _execute app/core/steps/entity_steps.py:966 _execute app/core/steps/floor_plan_prompt_step.py:323 _execute app/core/steps/floor_plan_render_step.py:432 _execute app/core/steps/location_consistency_step.py:146 _execute app/core/steps/scene_camera_flow_step.py:172 _execute app/core/steps/scene_consistency_step.py:392 _execute app/core/steps/scene_steps.py:206 _execute app/core/steps/shot_cinematography_step.py:92 _execute app/core/steps/shot_essence_extraction_step.py:125 _execute app/core/steps/shot_selection_step.py:149 _execute app/core/steps/shot_validator_step.py:223 _execute app/modules/llm/grok_image_client.py:208 generate_image app/modules/pipeline/background_chain_planning.py:683 run_background_chain_planning app/modules/pipeline/background_chain_render.py:697 run_background_chain_render app/modules/pipeline/background_chain_render.py:1191 _run_planner_driven_render app/modules/pipeline/entity_extractor_v2_legacy.py:528 extract_entities_multiturn app/modules/pipeline/entity_extractor_v3.py:485 extract_entities app/modules/pipeline/scene_extractor_v2.py:843 extract_scenes_multiturn app/modules/pipeline/scene_summarizer.py:112 summarize_scenes app/modules/pipeline/scene_validator.py:189 validate_visible_entities app/services/export_service.py:775 _build_episode_html ← LLM 아님 app/services/reference_phase1_service.py:186 run app/services/reference_phase2_service.py:193 run app/services/reference_phase3_service.py:188 run app/services/scene_generation_coordinator.py:2024 _generate_scene_in_loop app/services/scene_image_service.py:736 generate_images ``` **이 태스크에서는 안 고친다** — 계획이 짚은 자리는 병렬 롤 하나이고, 34곳은 자리마다 부모의 뜻(스텝이냐 샷이냐)과 검증이 다르다. 단계 3 뒤 별건으로 다룬다. ★「계층이 생겼다」를 말할 때 **이 구간은 빼고 말한다.** - [x] **Step 7: commit** ```bash git add backend/app/modules/llm/opik_trace.py \ backend/app/modules/pipeline/multiroll_select.py \ backend/tests/unit/test_opik_trace_worker_propagation.py git commit -m "fix(opik): 병렬 worker 에 trace 를 전파한다 multiroll_select.py:1385 는 budget 과 generation_context 둘만 worker 에 명시 전파한다. trace 핸들은 세 번째 ContextVar 라 같이 실어야 한다 — 안 실으면 병렬 롤의 호출이 부모 없이 떨어져 계층이 병렬 구간에서만 조용히 무너진다. 시험은 진짜 ThreadPoolExecutor 를 돌린다. 같은 thread 안에서만 보는 시험은 이 결함을 못 잡는다. thread 재사용 때 남의 부모를 물려받지 않는 것도 함께 못박았다." ``` --- ### Task 8: 직접 호출도 부모 밑 span 으로 — `ImageTracer.log` **Files:** - Modify: `backend/app/modules/llm/image_tracer.py:124-200` - Test: `backend/tests/unit/test_image_tracer_span_attach.py` (신규) **Interfaces:** - Consumes: `current_trace` (Task 3), `build_axis_tags` (Task 2) - Produces: 부모가 열려 있으면 `ImageTracer.log` 가 trace 를 새로 만들지 않고 부모에 span 을 붙인다 - [x] **Step 1: 실패하는 시험을 쓴다** ```python # backend/tests/unit/test_image_tracer_span_attach.py """직접 호출(record_provider_call)도 부모 trace 밑 span 이 된다.""" import pytest class _FakeSpan: def __init__(self, calls, **kw): self.calls = calls calls.append(("span", kw)) def end(self, **kw): self.calls.append(("span_end", kw)) class _FakeTrace: def __init__(self, calls, **kw): self.calls = calls calls.append(("trace", kw)) def span(self, **kw): return _FakeSpan(self.calls, **kw) def end(self, **kw): self.calls.append(("trace_end", kw)) @pytest.fixture def tracer(monkeypatch): calls = [] class _Client: def trace(self, **kw): return _FakeTrace(calls, **kw) def span(self, **kw): return _FakeSpan(calls, **kw) from app.modules.llm.image_tracer import ImageTracer t = ImageTracer.__new__(ImageTracer) t._client = _Client() t._explicit_context = None return t, calls def test_v1_makes_its_own_trace(monkeypatch, tracer): from app.core import config monkeypatch.setattr(config.settings, "opik_trace_v2_enabled", False) t, calls = tracer t.log(step="still_recipe_roll", model="x-ai/grok-imagine-image-2.0", prompt="p", duration_ms=120) kinds = [k for k, _ in calls] assert kinds[0] == "trace" assert calls[0][1]["name"] == "still_recipe_roll/x-ai/grok-imagine-image-2.0" def test_v2_attaches_span_to_open_parent(monkeypatch, tracer): from app.core import config from app.modules.llm import opik_trace monkeypatch.setattr(config.settings, "opik_trace_v2_enabled", True) t, calls = tracer handle = opik_trace.TraceHandle(uid="0190-parent", name="still:S1sh3", thread_id="ep_x") token = opik_trace.bind_trace(handle) try: t.log(step="still_recipe_roll", model="x-ai/grok-imagine-image-2.0", prompt="p", duration_ms=120, provider="openrouter", params={"role": "still_recipe_roll"}) finally: opik_trace.reset_trace(token) kinds = [k for k, _ in calls] assert "trace" not in kinds, "부모가 있는데 trace 를 또 만들었다" span_kw = dict(calls[0][1]) assert span_kw["trace_id"] == "0190-parent" assert span_kw["name"] == "still_recipe_roll · x-ai/grok-imagine-image-2.0" def test_v2_without_parent_still_records(monkeypatch, tracer): """부모가 없어도 기록은 남아야 한다 — 기록이 사라지는 것이 최악이다.""" from app.core import config monkeypatch.setattr(config.settings, "opik_trace_v2_enabled", True) t, calls = tracer t.log(step="pdf_validation", model="gpt-5.6-sol", prompt="p", duration_ms=10) assert [k for k, _ in calls][0] == "trace" def test_v2_tags_are_all_prefixed(monkeypatch, tracer): from app.core import config from app.modules.llm import opik_trace monkeypatch.setattr(config.settings, "opik_trace_v2_enabled", True) t, calls = tracer handle = opik_trace.TraceHandle(uid="0190-p", name="n") token = opik_trace.bind_trace(handle) try: t.log(step="s", model="m", prompt="p", duration_ms=1, provider="openrouter", status="error", error="boom") finally: opik_trace.reset_trace(token) tags = calls[0][1]["tags"] assert all(":" in x for x in tags), tags assert "status:error" in tags ``` - [x] **Step 2: 실패를 확인한다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_image_tracer_span_attach.py -v` Expected: `test_v2_*` FAIL - [x] **Step 3: `ImageTracer.log` 에 v2 분기를 넣는다** `backend/app/modules/llm/image_tracer.py` 의 `log` 안, `trace = self._client.trace(` **앞**에 분기를 넣는다. v1 코드는 그대로 둔다. ```python # ── v2 (2026-08-23): 부모가 있으면 span 만 붙인다 ────────── from app.core.config import settings if getattr(settings, "opik_trace_v2_enabled", False): from app.modules.llm.opik_trace import ( build_axis_tags, current_trace) op = (params or {}).get("role") or step axis = build_axis_tags( step=step, op=op, model=model, provider=provider or _provider_of(model), status=None if status == "success" else status, ) parent = current_trace() if parent is not None: s = self._client.span( trace_id=parent.uid, name=f"{op} · {model}", type="llm", model=model, provider=provider or _provider_of(model), input=trace_input, output=trace_output, metadata=span_metadata, tags=axis, ) s.end() return # 부모가 없으면 이름 있는 trace 를 홀로 세운다 — 기록이 # 사라지는 것보다 낫다. tags = axis trace = self._client.trace( ``` ★`span_metadata` 는 지금 `trace = ...` 뒤에서 만들어진다. **`trace_input`/`trace_output` 조립 뒤, `trace = ...` 앞으로 옮긴다** — v2 분기가 그것을 쓴다. - [x] **Step 4: 시험이 통과하는지 본다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_image_tracer_span_attach.py -v` Expected: PASS (4건) - [x] **Step 5: 회귀 — 기존 기록 시험을 태운다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_all_provider_calls_are_traced.py -v` Expected: PASS (기록 표식이 그대로 있어야 한다) - [x] **Step 6: commit** ```bash git add backend/app/modules/llm/image_tracer.py \ backend/tests/unit/test_image_tracer_span_attach.py git commit -m "feat(opik): 직접 호출도 부모 trace 밑 span 으로 record_provider_call 경로가 지금은 호출마다 trace 를 하나씩 만들어 trace:span 이 1:1 이다(계층 없음). 부모가 열려 있으면 span 만 붙인다. 부모가 없으면 지금처럼 홀로 선 trace 를 만든다 — 기록이 사라지는 것이 최악이다. 태그는 전부 접두사 축으로." ``` --- ### Task 9: 단계 2 실물 확인 (관문 0~5, 12, 15) **Files:** - Create: `backend/opik_trace_v2_smoke.py` **Interfaces:** - Consumes: Task 2~8 전부 - Produces: 없음 (확인 전용) **이것은 시험이 아니라 실물 확인이다.** 단위 시험은 가짜 클라이언트를 쓰므로 서버가 실제로 받아 주는지를 못 본다. - [x] **Step 1: 확인 스크립트를 쓴다** ```python # backend/opik_trace_v2_smoke.py """Opik trace v2 실물 확인 — 모델 호출 0건, 자체 호스팅에만 쓴다. `theroad-trace-v2-smoke` 프로젝트에 쓴다. 프로덕션 감사 데이터는 안 건드린다. """ import json import os import time import urllib.request os.environ["OPIK_PROJECT_NAME"] = "theroad-trace-v2-smoke" os.environ["OPIK_TRACE_V2_ENABLED"] = "true" from app.modules.llm.opik_trace import ( # noqa: E402 build_axis_tags, current_trace, episode_thread_id, open_trace) BASE = "http://192.168.133.87:5173/api" HDR = {"Comet-Workspace": "default"} def get(path): return json.load(urllib.request.urlopen( urllib.request.Request(f"{BASE}{path}", headers=HDR), timeout=30)) thread = episode_thread_id(project_name="스모크", episode_title="1회", episode_id="deadbeef-0000-0000-0000-000000000000") print("thread_id =", thread) with open_trace(name="step:scene_image_pipeline", tags=build_axis_tags(step="scene_image_pipeline"), metadata={"project_id": "p", "episode_id": "e"}, thread_id=thread) as step_h: assert step_h is not None, "trace 가 안 열렸다" for i in range(2): with open_trace(name=f"still:smoke{i} · still_recipe", tags=build_axis_tags(step="still_recipe"), metadata={"still_id": f"smoke{i}"}, thread_id=thread) as shot_h: from app.modules.llm.image_tracer import get_image_tracer for op in ("roll", "judge", "fix"): get_image_tracer().log( step="still_recipe", model="probe-model", prompt="확인용", duration_ms=1, params={"role": op}, provider="probe", extra_metadata={"still_id": f"smoke{i}"}) print(f" 샷 {i}: trace={shot_h.uid[:8]} 부모={step_h.uid[:8]}") time.sleep(4) projs = get("/v1/private/projects?page=1&size=100") pid = next(p["id"] for p in projs["content"] if p["name"] == "theroad-trace-v2-smoke") tr = get(f"/v1/private/traces?project_id={pid}&page=1&size=50") print(f"\ntrace {tr['total']}건") bad_name, bad_tag, threads = 0, [], set() for t in tr["content"]: print(f" · {t['name']:34s} span={t['span_count']} thread={t['thread_id']}") if t["name"] == "chat.completion": bad_name += 1 bad_tag += [x for x in (t.get("tags") or []) if ":" not in x] threads.add(t["thread_id"]) print(f"\n관문 1 chat.completion 0건 : {'OK' if bad_name == 0 else f'FAIL {bad_name}'}") print(f"관문 2 thread 1개 : {'OK' if len(threads) == 1 else f'FAIL {threads}'}") print(f"관문 3 계층(span>1) : " f"{'OK' if any(t['span_count'] > 1 for t in tr['content']) else 'FAIL'}") print(f"관문 4 접두사 없는 trace 태그 0 : {'OK' if not bad_tag else f'FAIL {bad_tag}'}") ``` - [x] **Step 2: 돌린다** Run: `cd backend && ./.venv/bin/python opik_trace_v2_smoke.py` Expected: 관문 1~4 전부 `OK`. trace 3건(스텝 1 + 샷 2), 샷 trace 마다 span 3 - [x] **Step 3: 관문 5 — 샷 계보가 trace 하나로 떨어지나** Run: 위 출력에서 `still:smoke0 · still_recipe` 의 `span_count` 가 3인지 확인 Expected: 3 (roll·judge·fix) - [x] **Step 4: 관문 7 준비 — 감사 도구는 아직 trace 를 읽는다** 이 시점에서 `tools/opik_prompt_audit` 은 **깨진다**(본문이 span 으로 옮겨 갔으므로). Task 14 에서 고친다. 지금은 확인만 한다. Run: `cd /Users/manta/Documents/Projects/TheRoad-I1 && ./backend/.venv/bin/python -m tools.opik_prompt_audit.main --help 2>&1 | head -5` Expected: 도구가 뜨는 것만 확인 (실행은 Task 14 뒤) - [x] **Step 5: commit** ```bash git add backend/opik_trace_v2_smoke.py git commit -m "test(opik): trace v2 실물 확인 스크립트 (관문 1~5) 단위 시험은 가짜 클라이언트를 써서 서버가 실제로 받는지를 못 본다. 자체 호스팅 별도 프로젝트(theroad-trace-v2-smoke)에 쓰고 되읽어 확인한다. 모델 호출 0건 — 요금 없음." ``` --- ## 단계 2 재리뷰 반영 (2026-08-24, Codex BLOCK 3건) 세 건 다 `file:line` 을 열어 확인했고 **전부 받았다**. 셋 다 「기록이 깨진다」에 해당한다. ### BLOCK 1 — OFF 인데 payload 가 달라졌다 샷 루프의 `generation_context` 가 설정과 무관하게 열려 있었다. scope 가 열리기만 해도 `ambient_call_meta`(`image_tracer.py`)가 `still_id`· `scene_index`·`shot_index`·`stage` 를 읽어 **v1** trace metadata 와 DB `llm_call_log.metadata` 에 싣는다(`gpt_image_primitive.py` → `record_provider_call` → `ImageTracer.log`). v2 분기는 그보다 **뒤**라서 꺼 놓고도 Opik 으로 나가는 것이 달라졌다 — Global Constraints 의 「OFF 경로는 바이트 동일」 위반이고, 「되돌리기는 한 줄」이라는 안전판이 깨진다. → `_shot_capture_scope`(`still_recipe_service.py`)로 감쌌다. **OFF 면 scope 자체를 안 연다.** 시험은 결함이 드러나던 자리에서 잰다 — `ambient_call_meta()` 의 출력에 `still_id`·`scene_index`·`shot_index`·`stage` 가 하나도 없어야 한다. 루프가 `generation_context` 를 **직접** 열면 붉어지는 AST 시험도 함께 넣었다(플래그를 지나치는 자리를 다시 만들지 않기 위해). ### BLOCK 2 — litellm span 태그가 축으로 안 갈렸다 설계 ⑤ 가 「litellm span 의 이름은 우리가 못 정한다 → 무엇인지는 태그로 가른다」고 못박았는데, `_build_opik_metadata` 가 맨 이름(`[step]`)으로 시작하고 호출자가 실은 태그를 그대로 덧붙였다. `entity_extractor_v3.py:446` 은 **엔티티 이름(한글 고유명사)** 을 태그로 싣는다 — 축도 안 갈리고 카디널리티도 터진다. → v2 에서 호출 단계를 `op:` 축으로 바꾸고, 이미 축화된 태그만 살리고, 맨 이름은 버린다(설계 ⑥ 「맨 이름 — 접두사를 붙여 축을 밝힌다」). exact-tags 시험 4건. OFF 는 맨 이름 그대로임을 시험으로 못박았다. **★`kind:` 는 이번에도 안 배선한다.** 지적은 사실이다 — 실사용 0 이다. 안 하는 이유: - 계획의 어느 태스크도 `kind` 를 채우라고 하지 않았다(Task 8 은 `op` 만). - `op` 에서 `kind` 를 **유도**하려면 `still_recipe_judge` 같은 이름을 글자로 갈라야 하는데, 그것은 이 저장소가 금지한 방식이다(글자·substring 으로 의미를 판단하지 않는다). 제대로 하려면 **호출자가 명시**해야 하고, 그것은 호출 사이트 전수 배선이라 별건이다. - `build_axis_tags` 에 축은 이미 있다 — 나중에 값만 실으면 된다. ### BLOCK 3 — 부모 없는 trace 가 주행 묶음 밖으로 떨어졌다 v2 에서 `StepRunner` 는 `session_id` 를 지우고 `thread_id` 를 넣는데, image step 은 그 dict 를 그대로 `ImageTracer` 의 명시 context 로 넘긴다 (`image_steps.py`). `ImageTracer.log` 는 `session_id` 만 읽어서, 부모가 없을 때 만드는 fallback trace 의 thread 가 `None` 이 됐다. → `ctx.get("thread_id") or ctx.get("session_id")`. OFF 에서는 `thread_id` 가 없어 동작이 같다. 시험은 v2·v1 양쪽을 다 잰다. ### 2차 재리뷰 BLOCK — 정규화 **뒤**에 맨 태그가 다시 들어갔다 (같은 날) `_build_opik_metadata` 가 축을 판 **뒤**, `_add_fallback_tag`(`llm_client.py`)가 안전 sanitize·GPT fallback 태그를 **접두사 없이** 덧붙였다(진입점 6곳). 내 exact-tags 시험은 `_build_opik_metadata` 만 재서 **최초 호출**만 봤고, 정작 훑을 값이 큰 fallback span 을 못 봤다. → 셋 다 고쳤다. 1. `_add_fallback_tag` 가 v2 에서 `status:sanitized` · `status:gpt_fallback` 으로 붙인다. OFF 는 맨 이름 그대로. 2. 축 판별을 **허용 접두사 whitelist**(`is_axis_tag`)로 좁혔다. `":" in tag` 로 보면 `http://192.168.0.9:5173/x` · `붉은 벽:2층` 같은 것이 축 태그로 둔갑해 고카디널리티 맨 태그가 그대로 통과한다. 3. 시험을 **router 로 나가는 최종 payload** 에서 잰다 — `_completion` 을 가로채 Tier 1→2→3 을 실제로 태우고 세 번의 태그를 v2·OFF 양쪽에서 못박았다. ★**교훈**: 「조립하는 함수」를 시험하고 「나가는 것」을 시험했다고 말했다. 그 사이에 태그를 하나 더 붙이는 자리가 있었다. ### 후속 항목 (이번 판 범위 밖) - [ ] **`kind:` 축에 값 싣기** — 호출자나 구조화된 registry 가 명시한다 (`gen`·`judge`·`fix`·`transform`·`analysis`). `op` 를 글자로 갈라 유도하지 않는다. 관문: 「새 주행 구간 span 에 `kind:` 가 실린 것이 0건이 아니다」. - [ ] **병렬 worker 34곳 trace 전파** — Task 7-C 아래 목록. ### 비차단으로 확인된 것 (Codex) 공백 무시 diff 는 scope 22줄뿐이고 기준/현재 AST 에서 scope 내부가 완전히 동일(`continue` 19 · `return` 4 · `raise` 3 동일). `StepRunner` 분리 본문도 기준 AST 와 동일이라 지역 이름 누락 없음. `capture=False` 는 `sink.py` 에서 spool·queue 앞에 반환하므로 새 중간물 자산 경로가 막힌다. --- # 단계 3 — uid 사슬·영향 요약·읽기 도구 ### Task 10: 참조에 자산 id 를 실어 보낸다 (끊김 A) **Files:** - Modify: `backend/app/modules/pipeline/multiroll_select.py:1340-1360`, `1513-1532`, `753`, `1087` - Modify: `backend/app/services/still_recipe_service.py:2819-2880`, `3385` 부근 - Test: `backend/tests/pipeline/test_ref_asset_id_record.py` (신규) **Interfaces:** - Consumes: 없음 - Produces: `records.json` 의 `refs[i]` 에 `asset_id` 칸 (없으면 `None`) **설계 근거:** `labeled_refs` 는 코드 전체에서 **243곳**에 나오고 **3곳에서 제자리 치환**된다(`outdoor_site_layout_step.py:357,464` · `visual_continuity_anchor_step.py:490`). 튜플을 넓히면 2-튜플로 푸는 15곳이 전부 깨진다. **씬 경로에 이미 있는 관례를 따른다** — `scene_generation_coordinator.py:89-114` 의 `ref_role_metadata[i]`, index 정렬 병렬 목록. - [x] **Step 1: 실패하는 시험을 쓴다** ```python # backend/tests/pipeline/test_ref_asset_id_record.py """records.json 의 refs 가 자산 id 를 나른다. 지금은 bytes 참조가 `` 로만 남아 어느 자산인지 모른다. 완주 판 실측: refs 449건 중 248건(55%)이 자산 불명, 대부분이 CHARACTER REFERENCE(222건). """ from app.modules.pipeline.multiroll_select import build_ref_records def test_bytes_ref_keeps_placeholder_and_gains_asset_id(): labeled = [("CHARACTER REFERENCE — 김선영", b"x" * 100)] metas = [{"asset_id": "aaaa-1111", "pipeline_role": "character_ref"}] out = build_ref_records(labeled, metas) assert out == [{ "label": "CHARACTER REFERENCE — 김선영", "path": "", # ★기존 칸은 그대로 — 읽는 데가 있다 "asset_id": "aaaa-1111", "role": "character_ref", }] def test_path_ref_keeps_path(): labeled = [("LOCATION PHOTOGRAPH", "/a/b/plate.png")] out = build_ref_records(labeled, [{"asset_id": None}]) assert out[0]["path"] == "/a/b/plate.png" assert out[0]["asset_id"] is None def test_missing_metadata_is_tolerated(): """병렬 목록이 없거나 짧아도 기록은 나와야 한다.""" labeled = [("A", b"1"), ("B", b"2")] assert [r["asset_id"] for r in build_ref_records(labeled, None)] == [None, None] assert [r["asset_id"] for r in build_ref_records(labeled, [{"asset_id": "x"}])] \ == ["x", None] def test_extra_metadata_is_ignored_not_crashing(): labeled = [("A", b"1")] metas = [{"asset_id": "x"}, {"asset_id": "y"}] assert len(build_ref_records(labeled, metas)) == 1 def test_index_alignment_is_positional_not_label_based(): """같은 라벨이 둘이어도 index 로 갈린다.""" labeled = [("SAME", b"1"), ("SAME", b"2")] metas = [{"asset_id": "first"}, {"asset_id": "second"}] out = build_ref_records(labeled, metas) assert [r["asset_id"] for r in out] == ["first", "second"] ``` - [x] **Step 2: 실패를 확인한다** Run: `cd backend && ./.venv/bin/pytest tests/pipeline/test_ref_asset_id_record.py -v` Expected: FAIL — `ImportError: cannot import name 'build_ref_records'` - [x] **Step 3: 조립 함수를 만든다** `backend/app/modules/pipeline/multiroll_select.py` 에 넣는다(모듈 수준). ```python def build_ref_records( labeled_refs: Sequence[Tuple[str, Any]], ref_role_metadata: Optional[Sequence[Optional[Dict[str, Any]]]] = None, ) -> List[Dict[str, Any]]: """records.json 에 적을 참조 목록 — 자산 id 를 함께 나른다. ★`labeled_refs` 튜플을 넓히지 않는다. 그것은 코드 전체 243곳에 나오고 3곳에서 제자리 치환된다(`outdoor_site_layout_step.py:357,464` · `visual_continuity_anchor_step.py:490`) — 넓히면 2-튜플로 푸는 자리가 전부 깨진다. 씬 경로에 이미 있는 관례를 따른다: `ref_role_metadata[i]` 는 `labeled_refs[i]` 와 **index 로 정렬된** 병렬 목록이다 (`scene_generation_coordinator.collect_actual_attached_refs` 와 동형). 라벨로 맞추지 않는 이유 — 같은 라벨이 둘일 수 있다. ★`path` 칸은 그대로 둔다. 지금 그것을 읽는 것들이 있고, 빼면 그 자리가 조용히 빈다. """ metas = list(ref_role_metadata or []) out: List[Dict[str, Any]] = [] for i, (label, src) in enumerate(labeled_refs): meta = metas[i] if i < len(metas) and isinstance(metas[i], dict) else {} out.append({ "label": str(label), # bytes 참조(엔티티)는 내용 대신 표식만 — record 가 수십 MB 로 # 오염되던 E2E 실측 결함 fix "path": (str(src) if isinstance(src, (str, Path)) else f""), "asset_id": meta.get("asset_id"), "role": meta.get("pipeline_role"), }) return out ``` `Sequence`·`Dict`·`Optional`·`List`·`Tuple`·`Any` 가 이 파일에 import 돼 있는지 확인하고, 없으면 `typing` 에서 보탠다. - [x] **Step 4: 시험이 통과하는지 본다** Run: `cd backend && ./.venv/bin/pytest tests/pipeline/test_ref_asset_id_record.py -v` Expected: PASS (5건) - [x] **Step 5: 기록 자리 둘을 이 함수로 바꾼다** `multiroll_select.py:1519-1530` 의 `"refs": [...]` 를: ```python "refs": build_ref_records(labeled_refs, ref_role_metadata), ``` 그리고 `multiroll_select.py:1344-1356` 의 `roll_refs` 를: ```python record["roll_refs"] = { lab: build_ref_records(roll_refs[lab], roll_ref_metadata.get(lab)) for lab in sorted(roll_refs) } ``` 두 함수(`select_best_roll` 계열, 753·1087줄 시그니처)에 인자를 보탠다. **기본값 `None` 이라 기존 호출자는 안 깨진다.** ```python ref_role_metadata: Optional[Sequence[Optional[Dict[str, Any]]]] = None, roll_ref_metadata: Optional[Dict[str, Sequence[Optional[Dict[str, Any]]]]] = None, ``` `roll_ref_metadata` 는 `.get(lab)` 을 쓰므로 `None` 이면 `{}` 로 바꿔 둔다: ```python roll_ref_metadata = dict(roll_ref_metadata or {}) ``` - [x] **Step 6: 접합 키를 만든다 — 라벨로 맞추면 안 된다** ★**라벨은 두 곳이 서로 다르다.** 실물로 확인했다: | 자리 | 라벨 | |---|---| | `still_recipe_service.py:2870` `_attach("character_ref", name, ...)` | `"김선영"` (이름 그대로) | | `still_recipe.py:1030` `refs.append((char_label.format(name=name), src))` | `"CHARACTER REFERENCE — 김선영: the exact person…"` | 라벨로 맞추면 인물·소품 `asset_id` 가 **전부 `None`** 이 된다. 게다가 `build_still_refs` 는 `handled_by` 갈래에서 char_refs 의 **부분집합만** 싣기도 해(`still_recipe.py:1073-1080`) 순서로 맞추는 것도 못 믿는다. **접합은 「참조 원본의 신원」으로 한다.** 그 값은 두 곳이 같은 객체를 주고받으므로 어긋날 수 없다. `backend/app/modules/pipeline/still_recipe.py` 에 넣는다(양쪽이 같은 함수를 쓰게 — 두 벌로 만들면 갈린다): ```python def ref_source_key(src: Any) -> str: """참조 원본의 신원 — 라벨·순서에 기대지 않는 접합 키. 라벨로 맞출 수 없다: `_attach` 는 이름(`김선영`)을 쓰고 `build_still_refs` 는 서식을 입힌 문안(`CHARACTER REFERENCE — 김선영: …`)을 쓴다. 순서로도 못 맞춘다: `handled_by` 갈래는 char_refs 의 부분집합만 싣는다. 원본 자체는 두 자리가 같은 객체를 주고받으므로 신원이 어긋날 수 없다. ★`id()` 를 쓰지 않는다 — 중간에 바이트를 갈아 끼우는 자리가 있다 (`outdoor_site_layout_step.py:357`·`visual_continuity_anchor_step.py:490`). """ import hashlib from pathlib import Path as _P if isinstance(src, (str, _P)): return f"path:{src}" if isinstance(src, (bytes, bytearray)): return f"sha256:{hashlib.sha256(bytes(src)).hexdigest()}" return f"repr:{src!r}" ``` - [x] **Step 7: ★신원 지도를 `_attach` 와 **분리**한다** > **2026-08-23 Codex 재리뷰로 교정.** 첫 판은 `_attach` 가 불릴 때 지도를 > 채우게 했는데, 그러면 **지도가 늦거나 빈다.** > > | 자리 | 왜 안 되나 | > |---|---| > | `prev_still` (`3037`) | `src` 도 `file_path` 도 **안 넘긴다** — 「파일 참조는 이미 전부 file_path 를 넘긴다」는 내 말이 틀렸다 | > | A/B conti (`4144-4148`) | **승자일 때만** `_attach` 한다. `_run_branch` 는 `4120` 에서 **먼저** 도므로, 신원을 계산하는 시점에 지도에 없다 | > | bgfirst seed 계열 | 같은 「승자 확정 후 부착」 부류 | > > `_attach` 는 **최종 계보(승자만)** 를 위한 것이고, 우리에게 필요한 것은 > **「발송 가능한 참조 전부」의 신원**이다. **둘은 다른 목록이다** — 하나로 > 겹쳐 쓰면 loser 가 최종 계보에 붙거나 roll_refs 가 빈다. **분리한다.** 브랜치가 돌기 **전에** 지도를 채운다. `still_recipe_service.py:2819` 부근(`attached_refs` 옆)에 둔다. ```python #: 원본 신원 → asset_id. **발송 가능한 참조 전부**를 담는다. #: ★attached_refs(최종 계보=승자만)와 **다른 목록**이다. 겹쳐 쓰면 #: loser 가 최종 계보에 붙거나(거짓 edge) roll_refs 가 빈다. #: ★같은 신원에 다른 asset_id 가 오면 마지막 값으로 덮지 않고 #: None 으로 못박는다(fail-closed) — 거짓 계보보다 빈 칸이 낫다. asset_by_src: Dict[str, Optional[str]] = {} _role_by_aid: Dict[str, str] = {} def _register_ref(src: Any, asset_id: Optional[str], role: str = "") -> None: """참조 원본의 신원을 지도에 넣는다. 승패와 무관하게 **미리**.""" if src is None or not asset_id: return from app.modules.pipeline.still_recipe import ref_source_key key = ref_source_key(src) prev = asset_by_src.get(key, _MISSING) if prev is _MISSING: asset_by_src[key] = asset_id elif prev != asset_id: asset_by_src[key] = None # 모호 — 잇지 않는다 if role: _role_by_aid[asset_id] = role ``` `_MISSING` 은 모듈 수준에 둔다(값이 `None` 인 것과 「없는 것」을 가르기 위해). ```python _MISSING = object() ``` **등록 자리** — 참조가 정해지는 곳마다 한 줄씩. 승패를 기다리지 않는다. ★역할 수를 **AST 로** 세었다. **13개**다. > 첫 판에 9개, 그다음 「15개」로 적었는데 **둘 다 틀렸다.** grep 으로 세면서 > `_attach` 호출의 첫 인자가 아닌 문자열(`"subject"`)을 역할로 읽었다. > 문자열 검색으로 세지 말고 AST 로 셀 것. ```bash # 역할 전수 — 구현할 때 이것으로 다시 센다 cd backend && ./.venv/bin/python -c " import ast,pathlib t=ast.parse(pathlib.Path('app/services/still_recipe_service.py').read_text()) r={} for n in ast.walk(t): if isinstance(n,ast.Call) and getattr(n.func,'id','')=='_attach' \\ and n.args and isinstance(n.args[0],ast.Constant): aid=n.args[2] if len(n.args)>2 else None r.setdefault(n.args[0].value,[]).append( (n.lineno,'None' if isinstance(aid,ast.Constant) and aid.value is None else 'asset')) print(len(r)); [print(' ',k,v) for k,v in sorted(r.items())]" ``` | 역할 | 등록 | |---|---| | `character_ref` (`2870`) | `_register_ref(scene_ref_image_map[key], scene_ref_asset_id_map.get(key), "character_ref")` | | `prop_ref` (`2897`) | `_register_ref(scene_ref_image_map[eid], scene_ref_asset_id_map.get(eid), "prop_ref")` | | `location_plate` (`3020`) | `_register_ref(plate, _plate_aid, "location_plate")` | | `conti_light` (`3076`·`4041`·**`4145`**·`4248`) | `_register_ref(conti, conti_entry.get("asset_id"), "conti_light")` — ★`4145` 는 **승자일 때만** 부착하는데 `_run_branch` 는 `4120` 에서 먼저 돈다. 미리 등록해야 한다 | | `prev_still` (`3037` · **`4071`**) | 두 자리 **다** 필요하다 — 아래 별항 | | `lane_storyboard_sketch` (`2984`) | `_register_ref(lane_sketch_path, lane_entry.get("asset_id"), "lane_storyboard_sketch")` | | `lane_canon_master` (`4062`) | `_register_ref(, , "lane_canon_master")` | | `structure_seed_look` (`2996`·`4091`) | `_register_ref(structure_seed_path, structure_seed_asset_id, "structure_seed_look")` | | `location_seed_bg` (`3005`·`4084`) | `_register_ref(seed_bg_path, seed_bg_asset_id, "location_seed_bg")` | | `bgfirst_bg` (`4047`) | `_register_ref(, , "bgfirst_bg")` — ★**만든 직후·`_run_branch` 전에** | | `bgfirst_group_bg` (`4055`) | `_register_ref(<그룹 배경 경로>, , "bgfirst_group_bg")` — 〃 | **자산이 원래 없는 역할 둘** — 등록하지 않는다(지어내지 않는다): | 역할 | 왜 | |---|---| | `confined_fp` (`3545`) | `_attach(..., asset_id=None)` — **명시적 unresolved** | | `era_ref` (`3585`) | `_attach(..., asset_id=None)` — **명시적 unresolved** | ★**`lane_seed` 는 `_attach` 가 아예 없다.** `build_still_refs` 가 `seed_label` 로 내보내는데 자산 계보가 어디에도 없다. 역시 `asset_id=None` 으로 둔다. ### ★bgfirst 갈래는 **plate 도 prev 도** 미리 등록해야 한다 bgfirst 는 `_authority_kind` 에 따라 `plate` 또는 `prev` 를 `refs_b` 에 실어 `3978` 의 `_run_branch` 로 **먼저** 보내는데, 두 경우 다 `_attach` 는 **승자 확정 뒤**에야 돈다. 그래서 그 갈래의 `asset_id` 가 통째로 빈다. **① plate** (2026-08-24 4차 리뷰) `_attach_plate()`(`3014-3023`)는 `3025-3027` 에서 **`not bgfirst_used`** 일 때만 불린다. bgfirst 는 `_authority_aid = None` 으로 시작해(`3671`) plate 경로를 `refs_b` 에 넣고(`3914-3923`), `_attach_plate` 는 `4052` 에서야 돈다. → **plate 자산 id 해석을 helper 로 뺀다.** `_attach_plate` 도 같은 helper 를 써야 두 계산이 갈리지 않는다. ```python def _resolve_plate_aid() -> Optional[str]: """plate 의 자산 id — map_conti 일치 우선, 없으면 fallback. ★`_attach_plate` 와 **같은 함수**를 쓴다. 두 벌로 만들면 갈린다. """ _map_entry = map_conti.get(tag) or {} return ( _map_entry.get("asset_id") if _map_entry.get("plate_path") == str(plate) else None ) or _plate_asset_id(plate) ``` `_attach_plate` 를 그 helper 로 고쳐 쓰고, bgfirst 쪽은 **`_run_branch` 전에**: ```python if _authority_kind == "plate" and _authority_path is not None: _authority_aid = _resolve_plate_aid() _register_ref(_authority_path, _authority_aid, "location_plate") ``` **② prev** — 자리가 하나 더 있다 `prev_still` 의 `3037` 은 **`not bgfirst_used` 조건 안**(`3030-3031`)이다. bgfirst 갈래의 prev 자산 id 는 **`3724-3729`** 에서 따로 정해지고, `3905-3912` 의 `refs_b=refs` 가 **`3978` 의 `_run_branch` 로 먼저 나간다.** → `_authority_kind == "prev"` 직후, **`_run_branch` 전에** 등록한다. ```python # bgfirst 갈래 prev — 3037 은 not bgfirst_used 안이라 여기 안 닿는다. # refs_b 가 3978 의 _run_branch 로 먼저 나가므로 그 전에 등록한다. if _authority_kind == "prev": _register_ref(_authority_path, _authority_aid, "prev_still") ``` ★두 등록은 **`_authority_*` 가 정해진 뒤, `build_ab_branch_refs` (`3914`)보다 앞**이어야 한다. 그 사이 어디든 좋지만 `3978` 뒤면 늦는다. ### ★역할은 `_register_ref` 에서만 정한다 첫 판은 `_attach` 안에서 `_role_by_aid[asset_id] = role` 을 채우게 했는데, 그러면 **늦은 배선**이라 앞의 결함이 되살아난다. 그리고 `asset_id → role` **전역 지도는 역할을 등록 순서에 종속시킨다.** 실제로 같은 자산이 자리에 따라 역할이 다른 경우가 있다 — lane entry 가 `2457` 에서는 sketch 인데 chain 에서는 `2630-2637` 의 conti 로 나간다. → **한 지도에 둘을 함께 담는다**: `source key → (asset_id, role)`. ```python #: 원본 신원 → (asset_id, role). **발송 가능한 참조 전부**를 담는다. #: ★asset_id→role 전역 지도를 따로 두지 않는다 — 같은 자산이 자리에 #: 따라 역할이 다르면(lane sketch ↔ chain conti) 등록 순서에 #: 종속돼 역할이 뒤바뀐다. ref_registry: Dict[str, Optional[Tuple[str, str]]] = {} def _register_ref(src: Any, asset_id: Optional[str], role: str) -> None: """참조 원본의 신원과 역할을 지도에 넣는다. 승패와 무관하게 **미리**.""" if src is None or not asset_id: return from app.modules.pipeline.still_recipe import ref_source_key key = ref_source_key(src) prev = ref_registry.get(key, _MISSING) if prev is _MISSING: ref_registry[key] = (asset_id, role) elif prev != (asset_id, role): ref_registry[key] = None # 모호 — 잇지 않는다 ``` `_meta_for` 도 이 지도를 읽는다. ```python def _meta_for(seq): out = [] for _lab, _src in (seq or []): hit = ref_registry.get(ref_source_key(_src)) out.append({"asset_id": hit[0], "pipeline_role": hit[1]} if hit else None) return out ``` ★**`_attach` 는 손대지 않는다.** 최종 계보 계약(승자만·거짓 edge 금지)이 그대로 살아 있어야 한다. - [x] **Step 7-B: 등록 자리를 하나도 안 빠뜨렸는지 시험으로 잡는다** ```python def test_every_attach_role_has_a_registration(): """★_attach 하는 역할은 신원 등록도 있어야 한다 — 자산이 있는 것만. 지도가 비면 그 참조의 asset_id 가 조용히 None 이 된다 — 고치려던 결함이 그대로 남는데 아무도 모른다. ★역할은 **AST 로** 센다. 문자열 검색으로 세다가 두 번 틀렸다(9 → 15 → 실제 13). _attach 첫 인자가 아닌 문자열을 역할로 읽었다. """ import ast import re from pathlib import Path src = (Path(__file__).resolve().parents[2] / "app" / "services" / "still_recipe_service.py").read_text(encoding="utf-8") tree = ast.parse(src) with_asset, without_asset = set(), set() for n in ast.walk(tree): if not (isinstance(n, ast.Call) and getattr(n.func, "id", "") == "_attach" and n.args and isinstance(n.args[0], ast.Constant)): continue role = n.args[0].value aid = n.args[2] if len(n.args) > 2 else None if isinstance(aid, ast.Constant) and aid.value is None: without_asset.add(role) else: with_asset.add(role) reg_roles = set(re.findall(r'_register_ref\([^)]*"([a-z0-9_]+)"', src)) # 자산이 원래 없는 역할 — asset_id=None 으로 명시 부착한다. 등록 대상 아님. assert without_asset <= {"confined_fp", "era_ref"}, \ f"자산 없는 역할이 늘었다: {sorted(without_asset)}" missing = with_asset - reg_roles assert not missing, f"신원 등록이 없는 역할: {sorted(missing)}" total = with_asset | without_asset assert len(total) == 13, ( f"_attach 역할이 {len(total)}개 — 계획이 센 것은 13개다. " f"늘었으면 등록 목록에 보태고, 줄었으면 왜 사라졌는지 볼 것: " f"{sorted(total)}") def test_bgfirst_authority_refs_are_registered_before_branch(): """★bgfirst 는 plate·prev 를 _run_branch **전에** 등록해야 한다. _attach_plate 는 not bgfirst_used 일 때만(3025-3027), 그리고 bgfirst 는 승자 확정 뒤 4052 에서야 부른다. 그런데 refs_b 는 3914 에서 만들어져 3978 의 _run_branch 로 먼저 나간다 — 그 사이에 등록이 없으면 그 갈래의 asset_id 가 통째로 빈다. ★**줄 번호 비교로는 증명이 안 된다**(2026-08-24 5차 리뷰). `_attach_plate` **함수 본문 안**의 등록도 소스 줄로는 `build_ab_branch_refs` 보다 앞이다 — 그런데 그 함수는 승자 확정 뒤에야 **불린다**. 정의 위치가 아니라 **실행 경로**를 봐야 한다. 그래서 AST 로 「`_authority_kind == "plate"` 를 보는 분기 **안에** `_register_ref(..., "location_plate")` 가 있는가」를 검사한다. """ import ast from pathlib import Path src = (Path(__file__).resolve().parents[2] / "app" / "services" / "still_recipe_service.py").read_text(encoding="utf-8") tree = ast.parse(src) def _registers(node, role: str) -> bool: """이 노드 **아래**에서 그 역할로 _register_ref 를 부르는가.""" for n in ast.walk(node): if (isinstance(n, ast.Call) and getattr(n.func, "id", "") == "_register_ref" and any(isinstance(a, ast.Constant) and a.value == role for a in n.args)): return True return False def _authority_guards(kind: str): """`_authority_kind == ""` 를 조건으로 쓰는 if 블록들.""" out = [] for n in ast.walk(tree): if not isinstance(n, ast.If): continue dump = ast.dump(n.test) if "_authority_kind" in dump and f"'{kind}'" in dump: out.append(n) return out plate_guards = _authority_guards("plate") prev_guards = _authority_guards("prev") assert plate_guards, '_authority_kind == "plate" 분기를 못 찾았다' assert prev_guards, '_authority_kind == "prev" 분기를 못 찾았다' assert any(_registers(g, "location_plate") for g in plate_guards), ( "bgfirst plate 갈래가 _run_branch 전에 location_plate 를 등록하지 " "않는다 — 그 후보의 asset_id 가 통째로 빈다") assert any(_registers(g, "prev_still") for g in prev_guards), ( "bgfirst prev 갈래가 prev_still 을 등록하지 않는다") def test_authority_registration_precedes_branch_refs_build(): """★등록이 refs 조립보다 **실행 순서로** 앞인지, **역할마다 따로** 본다. ★한 목록으로 합쳐 `any(...)` 로 보면 안 된다(2026-08-24 6차 리뷰): prev 가 앞에 있거나 `_attach_plate` **본문 안**의 등록이 앞에 있으면, plate guard 의 direct 등록이 branch 뒤로 밀려도 통과한다. 그래서 **guard 블록 안의 호출만** 뽑아 역할별로 각각 본다. """ import ast from pathlib import Path src = (Path(__file__).resolve().parents[2] / "app" / "services" / "still_recipe_service.py").read_text(encoding="utf-8") tree = ast.parse(src) def _guard_reg_lines(kind: str, role: str): """`_authority_kind == ""` 분기 **안**의 `_register_ref(role)` 줄.""" out = [] for n in ast.walk(tree): if not isinstance(n, ast.If): continue dump = ast.dump(n.test) if "_authority_kind" not in dump or f"'{kind}'" not in dump: continue for c in ast.walk(n): if (isinstance(c, ast.Call) and getattr(c.func, "id", "") == "_register_ref" and any(isinstance(a, ast.Constant) and a.value == role for a in c.args)): out.append(c.lineno) return out build_lines = [ n.lineno for n in ast.walk(tree) if isinstance(n, ast.Call) and getattr(n.func, "id", "") == "build_ab_branch_refs" ] assert build_lines, "build_ab_branch_refs 호출을 못 찾았다" first_build = min(build_lines) plate_lines = _guard_reg_lines("plate", "location_plate") prev_lines = _guard_reg_lines("prev", "prev_still") assert plate_lines, ( 'plate guard 안에 location_plate 등록이 없다 — _attach_plate 본문의 ' '등록은 승자 확정 뒤에야 불리므로 이 갈래를 못 덮는다') assert prev_lines, 'prev guard 안에 prev_still 등록이 없다' # ★역할마다 **따로** 본다 — 합치면 하나가 뒤로 밀려도 통과한다 assert min(plate_lines) < first_build, ( f"plate guard 등록({min(plate_lines)}줄)이 " f"build_ab_branch_refs({first_build}줄) 뒤다") assert min(prev_lines) < first_build, ( f"prev guard 등록({min(prev_lines)}줄)이 " f"build_ab_branch_refs({first_build}줄) 뒤다") def test_plate_aid_resolution_is_shared(): """★plate 자산 id 해석은 한 함수여야 한다 — 두 벌이면 갈린다.""" import re from pathlib import Path src = (Path(__file__).resolve().parents[2] / "app" / "services" / "still_recipe_service.py").read_text(encoding="utf-8") assert "_resolve_plate_aid" in src, "plate aid helper 가 없다" # _attach_plate 도 그 helper 를 써야 한다 m = re.search(r"def _attach_plate\(\).*?(?=\n def |\n if )", src, re.S) assert m and "_resolve_plate_aid" in m.group(0), \ "_attach_plate 가 자기 계산을 따로 한다 — 두 값이 갈린다" def test_prev_still_is_registered_in_both_places(): """★prev_still 은 자리가 둘이다 — 3037(non-bgfirst)과 4071(bgfirst). 3037 은 `not bgfirst_used` 조건 안이라, bgfirst 갈래는 그 등록에 안 닿는다. 그쪽 refs_b 가 3978 의 _run_branch 로 먼저 나가므로 미리 등록해야 한다. """ import re from pathlib import Path src = (Path(__file__).resolve().parents[2] / "app" / "services" / "still_recipe_service.py").read_text(encoding="utf-8") sites = re.findall(r'_register_ref\([^)]*"prev_still"', src) assert len(sites) >= 2, \ f"prev_still 등록이 {len(sites)}곳 — bgfirst 갈래가 빠졌다" def test_lane_seed_has_no_asset_lineage_and_we_admit_it(): """★`lane_seed` 는 `_attach` 가 아예 없다 — 자산 계보가 어디에도 없다. `build_still_refs` 가 `seed_label` 로 내보내는 참조인데 자산 id 가 없다. **없는 것을 지어내지 않는다** — `asset_id=None` 으로 두고, 그 사실이 조용히 잊히지 않게 여기서 못박는다. 나중에 lane_seed 를 자산으로 등록하게 되면 이 시험이 깨진다. 그때 등록 목록에 보태고 이 시험을 지우면 된다. """ import re from pathlib import Path src = (Path(__file__).resolve().parents[2] / "app" / "services" / "still_recipe_service.py").read_text(encoding="utf-8") assert not re.search(r'_attach\(\s*\n?\s*"lane_seed"', src), \ "lane_seed 가 이제 _attach 된다 — 등록 목록에 보태고 이 시험을 지울 것" ``` - [x] **Step 7-C: 모호 신원이 fail-closed 인지 못박는다** ```python def test_ambiguous_source_is_not_linked(): """같은 바이트가 서로 다른 자산으로 등록되면 잇지 않는다. 마지막 값으로 덮으면 거짓 계보가 생긴다 — 빈 칸이 낫다. """ from app.modules.pipeline.still_recipe import ref_source_key _MISSING = object() m = {} def _reg(src, aid): k = ref_source_key(src) prev = m.get(k, _MISSING) if prev is _MISSING: m[k] = aid elif prev != aid: m[k] = None _reg(b"same", "asset-1") assert m[ref_source_key(b"same")] == "asset-1" _reg(b"same", "asset-2") assert m[ref_source_key(b"same")] is None # 모호 → 안 잇는다 _reg(b"same", "asset-1") assert m[ref_source_key(b"same")] is None # 한 번 모호면 계속 모호 ``` - [x] **Step 8: ★신원을 `_run_branch` 안에서 한 번에 계산한다** > 2026-08-23 Codex 지적 반영. 첫 판은 `build_still_refs` 결과 하나에만 신원을 > 붙였는데, 실제로 모델에 나가는 목록은 **훨씬 다양하다**: > > - `_run_branch(branch_tag, branch_refs, …)` 호출 **8곳** > (`3595` · `3958` · `3978` · `4120` · `4156` · `4199` · `4203` · `4258`) > - `roll_refs=` 를 따로 만드는 곳 **4곳** > (`3598` confined · `3964`·`3993` bgfirst · `4130` 4택1) > > 표준 한 벌만 덮으면 이 갈래들은 **위치가 어긋난 `asset_id`** 를 달거나 > `roll_refs` 가 전부 빈다. **호출 자리 8곳에 배선하지 않는다** — 하나만 빠뜨려도 그 갈래가 조용히 빈다. `_run_branch`(`still_recipe_service.py:3422`)는 `branch_refs` 와 `**variant_kw`(그 안에 `roll_refs`)를 **전부 받는 목**이다. 거기서 한 번 계산한다. 이 저장소가 이미 쓰는 원리다 — `ambient_call_meta` 가 인자 대신 주변 scope 를 읽는 이유와 같다(빠뜨릴 자리를 없앤다). ```python def _run_branch(branch_tag: str, branch_refs, rec_key: str, out_stem: Path, rc: Optional[int] = None, jf: Any = None, jt: Any = None, cf: Any = None, ef: Any = None, **variant_kw): # ── 참조 신원 (2026-08-23) ───────────────────────────── # ★여기서 한 번만 계산한다. 호출 자리는 8곳이고 roll_refs 를 # 따로 만드는 곳이 4곳이라, 그 자리마다 배선하면 하나를 # 빠뜨려도 그 갈래만 조용히 빈다. 이 함수가 branch_refs 와 # variant_kw 를 전부 받는 목이다. from app.modules.pipeline.still_recipe import ref_source_key def _meta_for(seq): out = [] for _lab, _src in (seq or []): _aid = asset_by_src.get(ref_source_key(_src)) out.append({"asset_id": _aid, "pipeline_role": _role_by_aid.get(_aid)} if _aid else None) return out variant_kw["ref_role_metadata"] = _meta_for(branch_refs) _rr = variant_kw.get("roll_refs") if _rr: variant_kw["roll_ref_metadata"] = { lab: _meta_for(seq) for lab, seq in _rr.items()} return run_branch_select( branch_tag=branch_tag, branch_refs=branch_refs, ... ``` `_role_by_aid` 는 `_attach` 옆(`still_recipe_service.py:2819` 부근)에서 만든다. ```python #: asset_id → role. 신원 지도와 짝이다. _role_by_aid: Dict[str, str] = {} ``` `_attach` 안, `asset_by_src[...] = asset_id` 옆에 한 줄 보탠다. ```python _role_by_aid[asset_id] = role ``` - [x] **Step 9: `run_branch_select` 가 두 인자를 통과시킨다** `still_recipe.py:565` 의 `run_branch_select` 에 두 키워드를 보태고 그대로 아래로 넘긴다. **기본값 `None` 이라 기존 호출자는 안 깨진다.** ```python ref_role_metadata: Any = None, roll_ref_metadata: Any = None, ``` 그리고 `variant_kwargs`(`still_recipe.py:646` 부근)에 실어 `multiroll_select` 까지 내린다. ```python if ref_role_metadata is not None: variant_kwargs["ref_role_metadata"] = ref_role_metadata if roll_ref_metadata is not None: variant_kwargs["roll_ref_metadata"] = roll_ref_metadata ``` - [x] **Step 10: 갈래 전수를 시험으로 못박는다** ★8곳 중 하나라도 빠지면 그 갈래만 조용히 빈다. 기계로 센다. ```python # backend/tests/pipeline/test_ref_asset_id_record.py 에 이어 붙인다 def test_every_branch_gets_ref_metadata(): """_run_branch 를 지나는 모든 갈래가 신원을 받는다. ★호출 자리 8곳에 배선하지 않고 _run_branch 안에서 한 번 계산하기로 한 설계를 못박는 시험이다. 누가 나중에 호출 자리로 옮기면 여기서 깨진다. """ import ast from pathlib import Path src = (Path(__file__).resolve().parents[2] / "app" / "services" / "still_recipe_service.py").read_text(encoding="utf-8") tree = ast.parse(src) inner = [n for n in ast.walk(tree) if isinstance(n, ast.FunctionDef) and n.name == "_run_branch"] assert len(inner) == 1, "_run_branch 가 하나여야 한다" body = ast.dump(inner[0]) assert "ref_role_metadata" in body, \ "_run_branch 안에서 ref_role_metadata 를 안 만든다" assert "roll_ref_metadata" in body, \ "_run_branch 안에서 roll_ref_metadata 를 안 만든다" def test_roll_ref_sites_are_covered_by_the_funnel(): """roll_refs 를 만드는 자리가 늘어도 _run_branch 를 지나면 덮인다.""" import re from pathlib import Path src = (Path(__file__).resolve().parents[2] / "app" / "services" / "still_recipe_service.py").read_text(encoding="utf-8") sites = len(re.findall(r"\broll_refs=", src)) assert sites >= 4, f"roll_refs 자리가 {sites}개 — 계획이 센 것은 4개였다" # 전부 _run_branch 인자로 들어가는지: roll_refs= 가 _run_branch( 호출 # 블록 안에만 있어야 한다 assert "run_branch_select(" in src ``` - [x] **Step 11: 접합이 실제로 붙는지 시험으로 못박는다** ★이것이 이 태스크에서 제일 미덥지 않은 자리다. 시험으로 고정한다. ```python # backend/tests/pipeline/test_ref_asset_id_record.py 에 이어 붙인다 from app.modules.pipeline.still_recipe import build_still_refs, ref_source_key def test_source_key_is_stable_for_same_bytes(): a, b = b"same-image-bytes", b"same-image-bytes" assert ref_source_key(a) == ref_source_key(b) assert ref_source_key(b"other") != ref_source_key(a) def test_source_key_handles_paths(): from pathlib import Path assert ref_source_key(Path("/a/b.png")) == ref_source_key("/a/b.png") def test_join_survives_label_reformatting(): """★라벨이 두 곳에서 다르다는 것이 이 접합의 존재 이유다. _attach 는 이름('김선영'), build_still_refs 는 서식 문안 ('CHARACTER REFERENCE — 김선영: …'). 원본 신원으로 맞추면 붙는다. """ src = b"char-image-bytes" asset_by_src = {ref_source_key(src): "char-asset-1"} labeled = build_still_refs( bg_only=False, plate=None, conti=None, prev_sel=None, char_refs=[("김선영", src)], prop_refs=[], ) # build_still_refs 가 만든 라벨은 이름과 다르다 assert any("CHARACTER REFERENCE" in lab for lab, _ in labeled) assert not any(lab == "김선영" for lab, _ in labeled) # 그래도 원본 신원으로는 붙는다 hits = [asset_by_src.get(ref_source_key(s)) for _, s in labeled] assert "char-asset-1" in hits ``` Run: `cd backend && ./.venv/bin/pytest tests/pipeline/test_ref_asset_id_record.py -v` Expected: PASS (8건) - [x] **Step 12: 실물로 잰다 — 관문 9** 가장 최근 완주 판으로 사후 계산해 몇 %가 이어지는지 본다. ```bash cd /Users/manta/Documents/Projects/TheRoad-I1 python3 -c " import json,re,collections f='projects/5bddbdfc-2681-42a6-9837-43f35f60049d/images/f5372927-bbec-405d-ad2c-100d587f5373/scene/recipe/records.json' d=json.load(open(f)) c=collections.Counter() for k in [x for x in d if re.fullmatch(r'S\d+sh\d+',x)]: for r in (d[k].get('refs') or []): kind='bytes' if str(r.get('path','')).startswith('&1 | tail -5 grep -n "ref_role_metadata\|asset_id" app/modules/pipeline/multiroll_select.py | \ awk -F: '$1 > 660 && $1 < 700' ``` Expected: 시험 PASS. `compute_input_fingerprint`(671줄) 범위에 새 이름이 **한 줄도 없어야** 한다. - [x] **Step 14: commit** ```bash git add backend/app/modules/pipeline/multiroll_select.py \ backend/app/modules/pipeline/still_recipe.py \ backend/app/services/still_recipe_service.py \ backend/tests/pipeline/test_ref_asset_id_record.py git commit -m "feat(still): records.json 의 refs 가 자산 id 를 나른다 (끊김 A) 완주 판 실측: refs 449건 중 248건(55%)이 로 자산 불명이고 대부분이 CHARACTER REFERENCE(222건) — '이 샷에 어느 인물 참조가 들어갔나'를 못 물었다. 그런데 asset id 는 still_recipe_service.py:2869 아래에서 이미 손에 있었다. labeled_refs 튜플은 안 넓힌다 — 243곳에 나오고 3곳에서 제자리 치환된다. 씬 경로에 이미 있는 관례(ref_role_metadata, index 정렬 병렬 목록)를 따른다. path 칸은 그대로 둔다 — 읽는 데가 있다. ★접합은 라벨이 아니라 참조 원본 신원(ref_source_key)으로 한다. 라벨은 두 곳이 다르고(_attach 는 이름 '김선영', build_still_refs 는 서식 문안 'CHARACTER REFERENCE — 김선영: …'), 순서는 handled_by 갈래에서 부분집합만 실려 어긋난다. 지문에는 한 줄도 안 닿는다." ``` --- ### Task 11: 변환 산출의 직접 입력을 **밝혀 적는다** (끊김 B — 좁힘) > ★**2026-08-23 Codex 재리뷰로 범위를 좁혔다.** > > 첫 판은 `cine_source_sel` 을 **롤의 `image_asset.id`** 로 잇겠다고 했다. > 그런데 **롤은 자산이 아니다** — 파일로만 존재한다. Task 7-B 가 capture 를 > 끄기로 한 이상(그 이유는 거기 적었다) 앞으로도 자산이 되지 않는다. > > 자산이 없는데 자산 id 로 이을 수는 없다. **없는 것을 지어내지 않는다.** > > 그래서 이 태스크는 **「무엇이 직접 입력이었는지를 정확히 적는 것」**으로 > 좁힌다. 자산으로 잇는 것은 롤을 등록할지 정한 뒤의 **별도 판**이다. **Files:** - Modify: `backend/app/services/still_recipe_service.py:4555-4600` - Test: `backend/tests/unit/test_cine_source_asset_link.py` (신규) **Interfaces:** - Consumes: `fix_stage_won` (`still_recipe_service.py` 기존) - Produces: `unresolved_attached_refs` 의 `cine_source_sel` 항목에 `stage`·`unresolved_reason` 칸 **무엇이 달라지나**: 지금은 `{"role": "cine_source_sel", "file": "S55sh8_sel.png", "sha256": "7c866c…"}` 뿐이라, 읽는 사람이 **그 `_sel` 이 원본 롤인지 수리본인지 모른다.** 그 한 칸이 「최종 샷에 무엇이 영향을 미쳤나」의 핵심이다. - [x] **Step 1: 실패하는 시험을 쓴다** ```python # backend/tests/unit/test_cine_source_asset_link.py """변환의 직접 입력이 원본 롤인지 수리본인지 밝혀 적는다. ★파일 이름으로는 못 가른다: _critique_and_fix 는 원본이든 수리본이든 승자를 언제나 canonical _sel 에 복사하므로 cine 의 source_file 은 항상 _sel.png 다 (cine_transform.py:164). fix_stage_won 이 승패를 말한다. ★자산 id 로 잇지는 않는다 — 롤은 자산이 아니다(Task 7-B 가 capture 를 껐다). 없는 것을 지어내지 않는다. """ from app.services.still_recipe_service import describe_cine_source def test_original_roll_won(): assert describe_cine_source( source_file="S12sh3_sel.png", selected="B", fix_won=False, ) == {"stage": "roll", "roll_label": "B", "unresolved_reason": "roll_not_registered_as_asset"} def test_fix_won(): """★첫 판의 결함이 여기다 — _sel.png 라고 롤로 이으면 거짓 계보다.""" assert describe_cine_source( source_file="S12sh3_sel.png", selected="B", fix_won=True, ) == {"stage": "fix", "roll_label": None, "unresolved_reason": "fix_output_not_registered_as_asset"} def test_missing_selected_is_not_guessed(): d = describe_cine_source( source_file="S12sh3_sel.png", selected=None, fix_won=False) assert d["stage"] == "roll" assert d["roll_label"] is None def test_non_sel_source_is_out_of_scope(): """cine 이 sel 이 아닌 것을 봤다면 이 함수의 전제 밖이다.""" assert describe_cine_source( source_file="S12sh3_cine.png", selected="B", fix_won=False) == {} def test_empty_source_is_tolerated(): assert describe_cine_source(source_file="", selected="B", fix_won=False) == {} ``` - [x] **Step 2: 실패를 확인한다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_cine_source_asset_link.py -v` Expected: FAIL — `ImportError: cannot import name 'describe_cine_source'` - [x] **Step 3: 함수를 만든다** `backend/app/services/still_recipe_service.py` 모듈 수준에 넣는다. ```python def describe_cine_source( *, source_file: str, selected: Optional[str], fix_won: bool, ) -> Dict[str, Any]: """변환의 **직접 입력이 무엇이었는지** 밝힌다. 실측: 최종 씬 자산 510개 중 input_image_ids 가 있는 것은 6개(1.2%). 변환이 적용되면 직접 입력이 `_sel.png` 파일이라 unresolved 로 빠진다. ★**파일 이름으로는 못 가른다.** `_critique_and_fix` (`multiroll_select.py:800-815`, `991-1013`)는 원본이든 수리본이든 승자를 언제나 canonical `_sel` 에 복사하므로 `source_file` 은 `cine_transform.py:164` 에서 **항상 `_sel.png`** 다. 승패는 `fix_stage_won(record)` 이 말한다. ★**자산 id 로 잇지 않는다.** 롤도 수리 산출도 `image_asset` 이 아니다 — 파일로만 존재한다. 없는 것을 지어내면 계보 선이 엉뚱한 자산을 가리키고, 그것은 과다 기재보다 나쁘다. 대신 **왜 못 이었는지**를 적는다. """ if not source_file or not source_file.endswith("_sel.png"): return {} if fix_won: return {"stage": "fix", "roll_label": None, "unresolved_reason": "fix_output_not_registered_as_asset"} return {"stage": "roll", "roll_label": str(selected) if selected else None, "unresolved_reason": "roll_not_registered_as_asset"} ``` - [x] **Step 4: 시험이 통과하는지 본다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_cine_source_asset_link.py -v` Expected: PASS (5건) - [x] **Step 5: `cine` 갈래에서 부른다** `still_recipe_service.py:4555` 부근에서 `_direct_unresolved` 의 `cine_source_sel` 항목에 칸을 보탠다. **기존 `file`·`sha256` 은 그대로 둔다** — 대조하는 데가 있다. ```python _cine_desc = describe_cine_source( source_file=(cine_rec or {}).get("source_file", ""), selected=record.get("selected"), # ★승패는 파일 이름이 아니라 이것이 말한다 — _sel 은 원본이 # 이겼을 때도 수리가 이겼을 때도 같은 이름이다. fix_won=fix_stage_won(record), ) if _cine_desc: for _u in _direct_unresolved: if _u.get("role") == "cine_source_sel": _u.update(_cine_desc) ``` - [x] **Step 6: 실측 — 관문 10** 새 주행 첫 샷 몇 개에서: ```bash PGPASSWORD=theroad_dev_2026 psql -h localhost -U theroad -d theroad -t -A -c " SELECT pipeline_metadata_json FROM image_asset WHERE asset_type='scene' ORDER BY created_at DESC LIMIT 5;" | \ python3 -c " import sys,json for line in sys.stdin: if not line.strip(): continue for u in json.loads(line).get('unresolved_attached_refs',[]): if u.get('role')=='cine_source_sel': print('stage:', u.get('stage'), '| roll:', u.get('roll_label'), '| why:', u.get('unresolved_reason')) " ``` Expected: `stage` 가 `roll` 또는 `fix` 로 채워져 있고 `unresolved_reason` 이 있다. **거짓 `asset_id` 는 없다.** - [x] **Step 7: commit** ```bash git add backend/app/services/still_recipe_service.py \ backend/tests/unit/test_cine_source_asset_link.py git commit -m "feat(still): 변환의 직접 입력을 밝혀 적는다 (끊김 B — 좁힘) 첫 판은 cine_source_sel 을 롤의 image_asset.id 로 잇겠다고 했다. 그런데 롤은 자산이 아니다 — 파일로만 존재한다. Task 7-B 가 capture 를 끄기로 한 이상 앞으로도 자산이 안 된다. 없는 것을 지어내지 않는다. 대신 「무엇이 직접 입력이었는지」를 적는다: stage(roll|fix) · roll_label · unresolved_reason. 지금은 file+sha256 뿐이라 그 _sel 이 원본인지 수리본인지 읽는 사람이 모른다 — 그 한 칸이 「최종 샷에 무엇이 영향을 미쳤나」의 핵심이다. 승패는 파일 이름이 아니라 fix_stage_won 이 말한다. _critique_and_fix 가 승자를 언제나 _sel 에 복사하므로 source_file 은 항상 _sel.png 다 (cine_transform.py:164). 롤을 자산으로 등록하는 것은 별도 판으로 남긴다." ``` --- ### Task 12: `shot_run_uid` — 세 곳에 같은 값 **Files:** - Modify: `backend/app/services/image_capture/context.py` (`_shot_trace_scope`) - Modify: `backend/app/modules/pipeline/multiroll_select.py` (record 에 기록) - Modify: `backend/app/services/still_recipe_service.py` (자산 metadata 에 기록) - Test: `backend/tests/unit/test_shot_run_uid.py` (신규) **Interfaces:** - Consumes: `current_trace` (Task 3) - Produces: `current_shot_uid() -> Optional[str]` in `opik_trace.py` - [x] **Step 1: 실패하는 시험을 쓴다** ```python # backend/tests/unit/test_shot_run_uid.py """shot_run_uid — Opik·records.json·image_asset 세 곳에 같은 값.""" import pytest from app.modules.llm.opik_trace import current_shot_uid def test_none_outside_scope(): assert current_shot_uid() is None def test_matches_open_trace_uid(monkeypatch): from app.core import config from app.modules.llm import opik_trace monkeypatch.setattr(config.settings, "opik_trace_v2_enabled", True) handle = opik_trace.TraceHandle(uid="0190-shot-1", name="still:x") token = opik_trace.bind_trace(handle) try: assert current_shot_uid() == "0190-shot-1" finally: opik_trace.reset_trace(token) def test_disabled_gives_none(monkeypatch): from app.core import config from app.modules.llm import opik_trace monkeypatch.setattr(config.settings, "opik_trace_v2_enabled", False) token = opik_trace.bind_trace( opik_trace.TraceHandle(uid="0190-x", name="n")) try: assert current_shot_uid() is None finally: opik_trace.reset_trace(token) ``` - [x] **Step 2: 실패를 확인한다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_shot_run_uid.py -v` Expected: FAIL — `ImportError: cannot import name 'current_shot_uid'` - [x] **Step 3: 함수를 만든다** `backend/app/modules/llm/opik_trace.py` 에 넣는다. ```python def current_shot_uid() -> Optional[str]: """지금 열려 있는 trace 의 uid — 세 곳에 같이 적을 값. Opik trace id 를 그대로 쓴다. 별도 uid 를 하나 더 만들면 둘이 어긋날 자리가 생긴다 — 같은 것을 두 이름으로 부르지 않는다. """ from app.core.config import settings if not getattr(settings, "opik_trace_v2_enabled", False): return None handle = current_trace() return handle.uid if handle is not None else None ``` - [x] **Step 4: 시험이 통과하는지 본다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_shot_run_uid.py -v` Expected: PASS (3건) - [x] **Step 5: ★나가는 문이 여럿이다 — 마무리 한 곳을 만든다** > 2026-08-23 Codex 지적. 첫 판은 uid 를 `record["prompt"]` 옆(1340줄) **한 > 곳에만** 적었는데, 그보다 **먼저 나가는 문**이 있다. `multiroll_select.py` 의 샷 함수가 나가는 자리: | 줄 | 갈래 | 첫 판이면 | |---|---|---| | ~1305 | 지문 일치 완결 — 「전부 skip」 | **uid 가 안 적힌다** (1340 을 못 지난다) | | ~1313 | 그 뒤 갈래 | 〃 | | ~1336 | 〃 | 〃 | | ~1546 | critique 끔 | uid 는 적히지만 **영향 요약이 빈다** | | ~1567 | 정상 끝 | 정상 | **결과**: 재개하면 Opik·`image_asset` 의 uid 와 `records.json` 의 uid 가 **갈린다.** 「세 곳에 같은 값」이 깨진다. > ★★**돈 — 2026-08-23 Codex 재리뷰.** 방문마다 바뀌는 값을 record 에 적으면 > **JIT 재생성 제동이 거짓 발동한다.** > > `_jit_tag_snapshot`(`still_recipe_service.py:167-224`)은 샷 전후의 record > 묶음을 직렬화해 비교하고, **달라지면 「돈이 나갔다」**로 읽는다 > (`:2322-2323` · `:4382-4395`). 지금 걸러 내는 것은 `reused` 와 > `*_this_run` 뿐이다(`_transient = {"reused"}`). > > `shot_run_uid` 는 **방문마다 바뀐다.** 지문이 맞아 아무것도 안 만든 > 재사용 방문도 스냅숏이 달라져 `_records_changed=True` 가 되고, > **완주 판의 래치를 거짓으로 올린다.** 지문 함수는 안 건드려도 **재개 > 계약이 깨진다.** > > → 두 칸을 `_transient` 에 넣는다. 이것을 안 하면 이 태스크는 **돈 문제**다. > ★**두 키를 다 지우면 일한 방문을 못 가려낸다** (2026-08-23 3차 리뷰). > > 복구 갈래가 있다: `_sel` 만 없고 롤 파일과 record 는 남은 모양 > (`multiroll_select.py:1295-1300` — 「sel 존재 + 선정 기록 없음 → sel 폐기, > 판정만 재수행」). 이때 롤 생성은 건너뛰지만 **판정은 유료로 다시 돈다** > (`:1406-1476`). 판정 결과가 전과 같으면 `:1513-1535` 가 **같은 값으로 > 덮어** record 가 안 움직인다 → JIT 가 「지출 없음」으로 읽는다. > > ★이것은 **내 변경이 만든 구멍이 아니다** — 지금도 그렇다. 다만 > `shot_run_produced` 를 transient 로 만들면 **닫을 기회를 버리는** 것이다. > > **더 깨끗하게 닫는다**: 방문마다 바뀌는 값 대신 **만든 방문에서만 오르는 > 계수**를 쓴다. 재사용 방문은 안 오르고(거짓 지출 없음), 유료 재판정 방문은 > 오른다(그 방문이 잡힌다). Opik 이 꺼져 있어도 돈다 — 기록 체계와 > **무관한 로컬 신호**다. 먼저 `_jit_tag_snapshot` 의 `_transient` 를 넓힌다 (`still_recipe_service.py:209` 부근). ```python # · reused: 캐시 적중 여부 # · *_this_run: 이번 실행에서 했는지 표식 # · shot_run_uid / shot_run_produced (2026-08-23): **방문마다 바뀌는** # 기록 신원. 재사용 방문도 값이 달라져 「지출」로 읽히면 완주 판 # 래치가 거짓 발동한다 — 기록을 붙이려다 재개를 깨는 자리다. # ★유료 구간 진입은 shot_run_spend_attempt_count 가 말한다(만든 방문에서만 # 오르는 계수라 재사용에는 안 움직이고, 판정만 다시 돈 유료 복구 # 갈래에는 움직인다). 그 키는 여기 안 넣는다. _transient = {"reused", "shot_run_uid", "shot_run_produced"} ``` 그다음 마무리 헬퍼를 만든다. ```python def _stamp_shot_run(record: Dict[str, Any], *, produced: bool, spend_attempt: bool) -> None: """이 방문의 uid 를 record 에 적고 영향 요약을 trace 에 싣는다. ★**모든 반환이 이 함수를 지나야 한다.** 한 곳에만 적으면 먼저 나가는 문(지문 일치 완결·critique 끔)에서 uid 가 빠져 세 곳이 갈린다. produced: 이번 방문이 **최종 산출의 저자인가**. False(산출 재사용)면 record·Opik 에만 남기고 `image_asset` 의 생성 계보는 **건드리지 않는다** — 그 자산을 만든 것은 이전 주행이다. spend_attempt: 이번 방문이 **유료 호출을 할 수 있는 구간에 들어갔나**. ★「했나」가 아니다 — 콜백이 바깥으로 안 나가고 끝나는 경로도 센다 (후보 하나뿐인 판정 `multiroll_gemini.py:1276-1278`, 참조 결손 로컬 검증 실패). **과금 횟수로 읽지 말 것.** ★`produced` 와도 다르다 — 소급 critique 는 유료 구간인데 수리가 지면 저자는 이전 방문이다(1336 갈래). """ # ★`produced` 는 **언제나 현재 값**으로 덮는다. # uid 가 없을 때(v2 OFF·scope 누락) 이 줄을 건너뛰면, record 에 남아 # 있던 옛 produced=True 와 옛 uid 를 자산 metadata 가 읽어 **새 산출을 # 옛 trace 의 저자로** 기록한다. record["shot_run_produced"] = bool(produced) try: from app.modules.llm.opik_trace import ( build_influence_summary, current_shot_uid, update_trace_output) uid = current_shot_uid() if uid: record["shot_run_uid"] = uid else: # 이번 방문에 trace 가 없으면 옛 uid 를 **지운다** — 남겨 두면 # 자산 계보가 엉뚱한 trace 를 가리킨다. record.pop("shot_run_uid", None) summary = build_influence_summary(record) summary["produced"] = bool(produced) update_trace_output(summary) except Exception as exc: # noqa: BLE001 — 기록은 본 작업을 안 막는다 logger.debug("_stamp_shot_run 실패 (non-fatal): %s", exc) ``` ★`spent` 는 **여기서 안 센다.** 반환 직전에 세면 「성공한 지출」만 잡힌다 — 아래 별항. - [x] **Step 5-B: ★★유료 구간 진입은 호출 「전」에 못박는다 (돈)** > **2026-08-24 4차 리뷰.** 표식을 반환 직전에 올리면 **실패한 시도를 못 잡는다.** > > 과금 가능 호출은 `multiroll_select.py:802`(critique) · `:1369-1403`(생성) · > `:1433-1472`(판정)에서 **예외가 날 수 있다.** 예외는 > `still_recipe_service.py:4372-4380` 으로 빠져 `continue` 하므로 **JIT 전후 > 비교(`:4382-4395`) 자체를 건너뛴다.** 그 시도는 기록에 안 남고 다음 > resume 이 같은 자리를 다시 탄다. > > ★이 저장소가 이미 쓰는 규율이다 — 「선정 결정을 sel 물질화 **이전에** > durable persist」(`multiroll_select.py:1533-1535`). **돈을 쓸 수 있는 구간에 들어가기 전에 적는다.** 샷 함수 안(`record` 와 `_persist` 가 보이는 자리)에 둔다. ```python _spend_marked = False def _mark_spend_attempt_once() -> None: """이번 방문이 유료 호출을 **하기 직전에** 한 번 못박는다. ★반환 직전이 아니라 **호출 전**이다. 과금 후 실패하면 예외가 still_recipe_service.py:4372-4380 으로 빠져 JIT 전후 비교를 통째로 건너뛴다 — 그 시도가 기록에 안 남고 다음 resume 이 같은 자리를 다시 탄다. 방문당 한 번만 올린다(호출이 여럿이어도 「이 방문이 유료 구간에 들어갔다」는 사실 하나면 스냅숏이 움직인다). """ nonlocal _spend_marked if _spend_marked: return _spend_marked = True try: _n = int(record.get("shot_run_spend_attempt_count") or 0) except (TypeError, ValueError): _n = 0 record["shot_run_spend_attempt_count"] = _n + 1 _persist() # ★durable — 여기서 죽어도 시도가 남는다 ``` **유료 호출은 전수 10곳**이다(AST 로 셌다): | 줄 | 호출 | 어디 | |---|---|---| | `802` `806` `907` `962` `970` `976` | `critique_fn` · `composition_critique_fn` · `regen_gen_fn` · `fix_gen_fn` · `fix_rejudge_fn`×2 | **`_critique_and_fix` 안** (모듈 수준 함수, `743-1013`) | | `1400` | `gen_fn` | 롤 생성 | | `1433` `1445` | `judge_fn` (flip 정순·역순) | 판정 | | `1472` | `judge_fn` (비-flip) | 판정 | `_critique_and_fix` 는 **모듈 수준 함수**라 샷 함수의 closure 를 못 쓴다 → **부르기 전에** 찍으면 그 안 여섯이 한꺼번에 덮인다. **부르는 자리 넷** — 전부 **호출 직전**이다. ★아래는 **「유료 호출을 할 수 있는 구간」의 입구**다 — 그 구간이 실제로 바깥으로 나가는지는 콜백이 정한다(위 별항). | 자리 | 무엇 직전 | 덮는 호출 | |---|---|---| | `~1315` | 소급 `_critique_and_fix(...)` 직전 | `802`·`806`·`907`·`962`·`970`·`976` | | `~1369` | 롤 생성 블록 직전 (`missing` 계산 뒤, 생성 전) | `1400` (+병렬 `bound`) | | **`~1406`** | **`# ── 판정·선정 ──` 주석 바로 뒤** = `if judge_flip:` **앞** | `1433`·`1445`·`1472` | | `~1547` | 정상 `_critique_and_fix(...)` 직전 | 위 여섯 | ★**판정은 `1433` 이 아니라 `1406`** 이다. `1433` 에 찍으면 **비-flip 갈래 (`1472`)가 안 덮인다.** 블록 진입점에서 찍어야 세 호출이 다 걸린다. ★`~1369` 는 **`missing` 이 비어 있으면 부르지 않는다** — 생성이 없으면 지출도 없다. ★`_persist` 는 `1188` 에 정의돼 네 자리 모두에서 부를 수 있다. `~1315` 는 재개 갈래라 record 가 이미 디스크에서 온 것이고(지문 일치가 증명된 상태), `~1369` 이후는 `1358` 의 phase 0 persist 뒤다. ### ★★이 계수가 하는 일과 **안 하는 일** — 넘겨짚지 말 것 **한다**: **유료 구간에 들어갔다는 사실**을 durable 하게 남긴다. 성공 방문의 JIT 전후 비교가 그것을 읽어 「이 방문이 일했다」로 센다. 그리고 지금도 있던 구멍 하나를 닫는다 — 판정만 다시 도는 복구 갈래에서 판정 결과가 전과 같아도 그 방문이 잡힌다. **★「과금 횟수」가 아니다 — 이름과 뜻을 좁혀 둔다** (2026-08-24 6차 리뷰) 표식은 **콜백을 부르기 직전**에 오른다. 그래서 콜백이 **바깥으로 안 나가고 끝나는 경로**도 센다. 실제로 그런 자리가 있다: | 경로 | 근거 | |---|---| | 후보가 하나뿐인 판정 | `multiroll_gemini.py:1276-1278` — 코드가 스스로 적어 놨다: 「비교할 상대가 없다 — **유료 호출 없이** 그 후보가 그대로 간다」 | | 참조 결손 로컬 검증 실패 | `gpt_image_primitive.py` 의 `mode/refs` 검사가 provider 호출 **전에** `ValueError` 를 올린다 | 그래서 이 값은 **「유료 호출을 할 수 있는 구간에 몇 번 들어갔나」**다. **과금 횟수도, 정확한 실패 지출 수도 아니다.** ⑭ 의 상한 배선에서도 이것을 정확한 지출 수로 **그대로 쓰면 안 된다** — 그때는 provider seam 이나 Opik 호출 기록으로 따로 세야 한다. ★**치우침의 방향은 안전한 쪽이다.** 과다 계상은 「일했다」로 읽혀 래치 쪽으로 기운다 — 사람 확인을 부를 뿐 돈이 새지 않는다. 반대(과소 계상)가 나쁘다. 그리고 순수 재사용 방문(`1305`)은 표식 전에 반환하므로 **안 센다** — 방문마다 바뀌던 uid 와 달리 훨씬 좁다. **안 한다**: **실패한 지출을 재생성 상한에 넣지 못한다.** 기전을 정확히 적는다(2026-08-24 5차 리뷰, 소스로 확인): 1. 완료 샷 진입에서 `jit_snapshot_before` 를 잡는다(`still_recipe_service.py:2322-2323`) 2. `_mark_spend_attempt_once` 가 계수를 올리고 persist 한다 3. 유료 호출이 실패하면 `except` 가 `jit_failed_tags` 만 보태고 `continue` 한다 (`:4372-4380`) — 전후 비교(`:4393-4395`)와 `jit_regen_count` 증가 (`:4451`·`:4458`)가 **실행되지 않는다** 4. 다음 resume 은 이미 저장된 계수를 **새 기준값**으로 잡는다 — 이전 실패의 변화량이 기준에 흡수된다 5. `jit_regen_count` 는 함수 호출마다 `0` 으로 초기화되고(`:2240-2242`) 성공 경로에서만 오른다. 래치 검사도 그 지역 값만 본다(`:2306-2318`·`:4747-4759`) ★**이것은 내 변경이 만든 구멍이 아니다.** 지금도 실패한 유료 방문은 `jit_regen_count` 를 안 올린다. 계수는 **감사 흔적을 새로 만들 뿐** 그 구멍을 닫지도, 넓히지도 않는다. ★**지금 있는 완화**: 실패한 샷은 `jit_failed_tags` 로 모여 `StillJitVerifyIncomplete` 를 올리고(`:4767`) **스텝을 완료로 못 닫게** 한다. 그래서 실패한 지출이 **조용히 지나가지는 않는다** — 다만 재생성 상한을 소비하지 않을 뿐이다. **왜 이번에 안 고치나**: 실패 지출을 상한에 넣는 것은 **래치가 언제 터지는지를 바꾸는 일**이다. 래치는 실제 사고 때문에 생긴 돈 가드이고, 거짓으로 터지면 주행 전체가 선다. 기록 체계화를 하면서 곁다리로 손댈 자리가 아니다 — **따로, 실측 근거를 갖추고** 해야 한다. → ⑭ 범위 밖에 항목으로 남긴다. ```python missing = [ lab for lab in labels if not _roll_path(out_stem, lab).exists() ] if missing: _mark_spend_attempt_once() # ★생성 전에 못박는다 if parallel_rolls and len(missing) > 1: ... ``` - [x] **Step 5-C: 실패한 지출이 잡히는지 시험으로 못박는다** ```python # backend/tests/unit/test_jit_snapshot_transient.py 에 이어 붙인다 def test_spend_is_marked_before_the_call_not_after(): """★★과금 후 실패해도 지출이 남아야 한다. 예외는 still_recipe_service.py:4372-4380 으로 빠져 JIT 전후 비교를 통째로 건너뛴다. 반환 직전에 세면 「성공한 지출」만 잡히고, 다음 resume 이 같은 돈을 다시 쓴다. 이 시험은 **소스 위치**를 본다 — 계수 증가가 _stamp_shot_run 안이 아니라 별도 함수(_mark_spend_attempt_once)에 있고, 그 함수가 _persist 를 부르는가. """ import ast from pathlib import Path src = (Path(__file__).resolve().parents[2] / "app" / "modules" / "pipeline" / "multiroll_select.py").read_text(encoding="utf-8") tree = ast.parse(src) marker = next( (n for n in ast.walk(tree) if isinstance(n, ast.FunctionDef) and n.name == "_mark_spend_attempt_once"), None) assert marker is not None, "_mark_spend_attempt_once 가 없다 — 지출을 반환 직전에 센다" body = ast.dump(marker) assert "shot_run_spend_attempt_count" in body assert "_persist" in body, "durable 하게 안 적는다 — 죽으면 지출이 사라진다" stamp = next(n for n in ast.walk(tree) if isinstance(n, ast.FunctionDef) and n.name == "_stamp_shot_run") assert "shot_run_spend_attempt_count" not in ast.dump(stamp), \ "_stamp_shot_run 이 아직 계수를 올린다 — 실패한 지출을 놓친다" def test_spend_marker_is_called_before_generation(): """생성·판정·critique **앞**에서 불러야 한다.""" import re from pathlib import Path src = (Path(__file__).resolve().parents[2] / "app" / "modules" / "pipeline" / "multiroll_select.py").read_text(encoding="utf-8") calls = [m.start() for m in re.finditer(r"_mark_spend_attempt_once\(\)", src)] # 정의 1 + 호출 4 assert len(calls) >= 5, f"_mark_spend_attempt_once 호출이 {len(calls) - 1}곳" def test_stale_uid_is_cleared_when_no_trace(): """★v2 OFF·scope 누락이면 옛 uid 를 지운다 — 옛 trace 저자로 안 남긴다.""" rec = {"shot_run_uid": "0190-old", "shot_run_produced": True} # _stamp_shot_run 이 하는 일과 같은 모양 rec["shot_run_produced"] = False # 항상 현재 값 uid = None if uid: rec["shot_run_uid"] = uid else: rec.pop("shot_run_uid", None) assert "shot_run_uid" not in rec assert rec["shot_run_produced"] is False # 자산 metadata 가 읽는 값 assert (rec.get("shot_run_uid") if rec.get("shot_run_produced") else None) \ is None ``` ★`shot_run_spend_attempt_count` 는 **`_transient` 에 안 넣는다.** 그것이 유료 구간 진입을 보수적으로 세는 신호다. 반대로 `shot_run_uid`·`shot_run_produced` 는 방문마다 바뀌므로 넣는다. - [x] **Step 6: 모든 반환 앞에서 부른다** 각 반환 자리 **직전**에 넣는다. `produced` 는 그 갈래가 실제로 산출을 만들었는지로 정한다. `produced` 는 **이번 방문이 최종 산출의 저자인가**로 정한다. Codex 가 소스로 확인해 준 값이다. | 줄 | 갈래 | `produced`(저자) | `spend_attempt`(유료 구간 진입) | |---|---|---|---| | 1305 | `sel` 있음 + `selected` + critique 완료 → 전부 skip | `False` | `False` | | 1313 | 〃 + `critique_enabled=False` — 표식만 남긴다 | `False` | `False` | | 1336 | `sel` 있음 + critique 미실행 → **소급 critique/fix** | **`fix_stage_won(record)`** | **`True`** | | 1546 | critique 끔 (롤·판정을 이번에 돌았다) | `True` | `True` | | 1567 | 정상 끝 | `True` | `True` | ★**1336 이 두 신호가 갈리는 자리다.** `_critique_and_fix` 는 **유료**인데, 수리가 없었거나 졌으면 최종 `_sel` 의 저자는 **이전 방문**이다. 하나로 쓰면 둘 중 하나가 틀린다 — `True` 면 계보를 가로채고, `False` 면 지출을 놓친다. ★1546·1567 에 닿으려면 `sel` 이 없어야 하고, 그 경로는 역순 판정 (`judge_rev_raw = judge_fn(...)`, `1445` 부근)을 **항상** 부른다. 그래서 `spend_attempt=True` 가 과다 계상이 아니다 — 실측으로 확인했다. ★1305·1313 은 호출이 **한 건도 없다**(반환 전까지 `judge_fn`·`critique_fn`· `gen_fn` 을 안 부른다). `spend_attempt=False` 가 맞다. ```python # ~1305 완결 — 호출 0 _stamp_shot_run(record, produced=False, spend_attempt=False) return sel, record # 완결 — 전부 skip # ~1313 완결 + critique 끔 — 표식만, 호출 0 _stamp_shot_run(record, produced=False, spend_attempt=False) _persist() return sel, record # ~1336 소급 critique/fix — **유료**지만 저자는 승패가 가른다 _stamp_shot_run(record, produced=fix_stage_won(record), spend_attempt=True) _persist() return sel, record # ~1546 critique 끔 — 롤·판정을 이번에 돌았다 _stamp_shot_run(record, produced=True, spend_attempt=True) _persist() return sel, record # ~1567 정상 끝 _stamp_shot_run(record, produced=True, spend_attempt=True) _persist() return sel, record ``` ★`_persist()` **앞**에서 부른다 — 그래야 uid 가 파일에 함께 저장된다. ★`fix_stage_won` 은 **`app.modules.pipeline.still_recipe`** 에 있다(2026-08-24 실행 중 확인 — 계획이 `still_recipe_service` 라고 잘못 적었다). `multiroll_select` 에서는 지역 import 로 가져온다. - [x] **Step 7: 반환을 하나도 안 빠뜨렸는지 기계로 확인한다** 사람 눈으로 세면 빠뜨린다. AST 로 센다. ```python # backend/tests/unit/test_shot_run_uid.py 에 이어 붙인다 def _outer_nodes(fn): """바깥 함수 본문의 노드만 — **중첩 함수·lambda 는 건너뛴다**. ★`ast.walk` 를 그냥 쓰면 안쪽 헬퍼(`_crit_refs`·`_roll_prompt`· `_gen_refs`·`_crit_prompt`)의 `return` 까지 「stamp 없는 반환」으로 잡혀 시험이 구현을 막는다. 그 반환들은 이 계약의 대상이 아니다. """ import ast out = [] def _rec(node): for child in ast.iter_child_nodes(node): if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda)): continue # 중첩 정의는 통째로 건너뛴다 out.append(child) _rec(child) _rec(fn) return out def test_every_outer_return_is_stamped(): """★바깥 반환이 전부 _stamp_shot_run 을 지나야 한다. 한 곳만 빠져도 그 갈래에서 세 곳의 uid 가 갈린다. 사람 눈으로 세면 빠뜨리므로 기계로 센다. 소스 기준 바깥 반환은 다섯이다(1305·1313·1336·1546·1567). """ import ast from pathlib import Path src = (Path(__file__).resolve().parents[2] / "app" / "modules" / "pipeline" / "multiroll_select.py").read_text(encoding="utf-8") tree = ast.parse(src) target = next( n for n in ast.walk(tree) if isinstance(n, ast.FunctionDef) and any(isinstance(x, ast.Return) and isinstance(x.value, ast.Tuple) and len(x.value.elts) == 2 for x in _outer_nodes(n)) ) nodes = _outer_nodes(target) returns = [n for n in nodes if isinstance(n, ast.Return) and n.value is not None] stamps = [n.lineno for n in nodes if isinstance(n, ast.Call) and getattr(n.func, "id", "") == "_stamp_shot_run"] plain = [r.lineno for r in returns if not any(r.lineno - 6 <= s < r.lineno for s in stamps)] assert not plain, f"stamp 없이 나가는 바깥 반환: {plain}줄" assert len(returns) == 5, f"바깥 반환이 {len(returns)}개 — 5개여야 한다" def test_nested_helper_returns_are_ignored(): """중첩 헬퍼의 반환은 이 계약 밖이다 — 시험이 구현을 막으면 안 된다.""" import ast fn = ast.parse( "def outer():\n" " def inner():\n" " return 1\n" " _stamp_shot_run(r, produced=True)\n" " return 2\n" ).body[0] rets = [n for n in _outer_nodes(fn) if isinstance(n, ast.Return)] assert len(rets) == 1, "중첩 함수의 return 이 새어 들어왔다" ``` - [x] **Step 8: 자산 metadata 에 적는다 — ★만든 방문만** `still_recipe_service.py` 의 `save_single_scene_asset({...})` 의 `pipeline_metadata` 조립 자리에 넣는다. ```python # ★이번 방문이 실제로 만든 자산에만 uid 를 찍는다. # 산출을 재사용한 방문(지문 일치)의 uid 로 덮으면 # 「이 자산을 만든 것」이 거짓이 된다 — 만든 것은 # 이전 주행이다. "shot_run_uid": ( record.get("shot_run_uid") if record.get("shot_run_produced") else None), ``` - [x] **Step 9: 재개 갈래를 시험으로 못박는다** ```python def test_reused_visit_does_not_claim_authorship(): """산출 재사용 방문의 uid 로 자산 계보를 덮지 않는다.""" rec = {"shot_run_uid": "0190-visit-2", "shot_run_produced": False} stamped = (rec.get("shot_run_uid") if rec.get("shot_run_produced") else None) assert stamped is None rec2 = {"shot_run_uid": "0190-visit-3", "shot_run_produced": True} assert (rec2.get("shot_run_uid") if rec2.get("shot_run_produced") else None) == "0190-visit-3" ``` - [x] **Step 10: ★★JIT 래치가 거짓 발동 안 하는지 못박는다 (돈)** 이것을 빠뜨리면 이 태스크는 **돈 문제**가 된다. 방문마다 바뀌는 uid 가 「지출」로 읽혀 완주 판의 재생성 제동이 거짓으로 걸린다. ```python # backend/tests/unit/test_jit_snapshot_transient.py (신규) """기록 신원이 「돈이 나갔다」로 읽히면 안 된다. _jit_tag_snapshot(still_recipe_service.py:167)은 샷 전후 record 묶음을 직렬화해 비교하고, 달라지면 지출로 읽어 재생성 제동을 건다(:4382-4395). shot_run_uid 는 방문마다 바뀌므로 transient 로 걸러야 한다 — 안 그러면 아무것도 안 만든 재사용 방문이 완주 판의 래치를 거짓으로 올린다. """ from app.services.still_recipe_service import _jit_tag_snapshot class _Rec: def __init__(self, data): self.data = data def _snap(uid, produced): return _jit_tag_snapshot( _Rec({"S1sh1": { "prompt": "P", "selected": "B", "input_fingerprint": "fp1", "shot_run_uid": uid, "shot_run_produced": produced, }}), "S1sh1", {"S1sh1"}, ) def test_visit_uid_alone_does_not_change_snapshot(): """★두 방문이 uid 만 다르면 스냅숏은 같아야 한다.""" assert _snap("0190-visit-1", False) == _snap("0190-visit-2", False) def test_produced_flag_alone_does_not_change_snapshot(): assert _snap("0190-a", False) == _snap("0190-a", True) def test_real_spend_still_changes_snapshot(): """★거르기가 너무 넓으면 일한 방문을 못 가려낸다 — 그것이 더 나쁘다.""" a = _jit_tag_snapshot( _Rec({"S1sh1": {"prompt": "P", "selected": "B", "shot_run_uid": "0190-a"}}), "S1sh1", {"S1sh1"}) b = _jit_tag_snapshot( _Rec({"S1sh1": {"prompt": "P", "selected": "A", # 선정이 바뀌었다 "shot_run_uid": "0190-a"}}), "S1sh1", {"S1sh1"}) assert a != b def test_critique_record_still_counts_as_spend(): a = _jit_tag_snapshot( _Rec({"S1sh1": {"prompt": "P", "shot_run_uid": "0190-a"}}), "S1sh1", {"S1sh1"}) b = _jit_tag_snapshot( _Rec({"S1sh1": {"prompt": "P", "shot_run_uid": "0190-b", "critique": {"issues": [{"issue_ko": "x"}]}}}), "S1sh1", {"S1sh1"}) assert a != b def test_retroactive_critique_counts_as_spend_even_when_fix_loses(): """★★1336 갈래 — 소급 critique 는 유료인데 수리가 지면 저자는 이전 방문. produced 와 spent 를 하나로 쓰면 둘 중 하나가 틀린다: True 면 자산 계보를 가로채고, False 면 지출을 놓친다. """ same = {"prompt": "P", "selected": "B", "critique": {"issues": []}} a = _jit_tag_snapshot( _Rec({"S1sh1": {**same, "shot_run_spend_attempt_count": 1, "shot_run_produced": False}}), "S1sh1", {"S1sh1"}) b = _jit_tag_snapshot( _Rec({"S1sh1": {**same, "shot_run_spend_attempt_count": 2, "shot_run_produced": False}}), "S1sh1", {"S1sh1"}) assert a != b, "수리가 진 소급 critique 의 지출이 안 잡힌다" def test_spent_count_is_a_spend_signal(): """★★유료 복구 갈래 — 판정만 다시 돌아 결과가 같아도 지출은 잡혀야 한다. multiroll_select.py:1295-1300 의 「sel 존재 + 선정 기록 없음 → sel 폐기, 판정만 재수행」 갈래는 롤 생성은 건너뛰지만 판정을 유료로 다시 돈다 (:1406-1476). 판정 결과가 전과 같으면 :1513-1535 가 같은 값으로 덮어 record 가 안 움직인다 — 계수가 없으면 JIT 가 지출을 놓친다. ★이 구멍은 지금도 있다. 계수가 그것을 닫는다. """ same = {"prompt": "P", "selected": "B", "verdicts": [{"label": "B"}]} a = _jit_tag_snapshot( _Rec({"S1sh1": {**same, "shot_run_uid": "0190-a", "shot_run_spend_attempt_count": 1}}), "S1sh1", {"S1sh1"}) b = _jit_tag_snapshot( _Rec({"S1sh1": {**same, "shot_run_uid": "0190-b", "shot_run_spend_attempt_count": 2}}), "S1sh1", {"S1sh1"}) assert a != b, "유료 재판정이 지출로 안 잡힌다" def test_spent_count_unchanged_on_reuse(): """재사용 방문은 계수가 안 오르므로 스냅숏도 같다.""" same = {"prompt": "P", "selected": "B", "shot_run_spend_attempt_count": 3} a = _jit_tag_snapshot( _Rec({"S1sh1": {**same, "shot_run_uid": "0190-a"}}), "S1sh1", {"S1sh1"}) b = _jit_tag_snapshot( _Rec({"S1sh1": {**same, "shot_run_uid": "0190-b"}}), "S1sh1", {"S1sh1"}) assert a == b def test_spent_count_is_not_transient(): """계수를 실수로 _transient 에 넣으면 이 시험이 잡는다.""" from app.services.still_recipe_service import _jit_tag_snapshot as f import inspect src = inspect.getsource(f) assert "shot_run_spend_attempt_count" not in src.split("_transient")[1][:200], \ "표식이 transient 에 들어갔다 — 유료 구간 진입을 놓친다" ``` Run: `cd backend && ./.venv/bin/pytest tests/unit/test_jit_snapshot_transient.py -v` Expected: PASS (7건) ★뒤 네 시험이 중요하다 — **거르기가 너무 넓으면 일한 방문을 못 가려낸다.** 좁게 거르는 것보다 그쪽이 나쁘다. - [x] **Step 10: 지문 회귀를 못박는다** ```bash cd backend && ./.venv/bin/pytest tests/ -q -k "fingerprint" 2>&1 | tail -5 awk 'NR>=671 && NR<=700' app/modules/pipeline/multiroll_select.py | \ grep -c "shot_run_uid\|shot_run_produced" ``` Expected: 시험 PASS · 마지막 명령이 **0** (지문 함수 안에 없다) - [x] **Step 11: commit** ```bash git add backend/app/modules/llm/opik_trace.py \ backend/app/modules/pipeline/multiroll_select.py \ backend/app/services/still_recipe_service.py \ backend/tests/unit/test_shot_run_uid.py git commit -m "feat(opik): shot_run_uid — Opik·records.json·image_asset 을 잇는다 Opik trace id 를 그대로 쓴다. 별도 uid 를 하나 더 만들면 둘이 어긋날 자리가 생긴다 — 같은 것을 두 이름으로 부르지 않는다. ★나가는 문이 여럿이다. uid 를 문안 옆 한 곳에만 적으면 지문 일치 완결 반환(~1305)이 그보다 먼저 나가 uid 가 안 적히고, 재개하면 Opik·image_asset 과 records.json 의 uid 가 갈린다. 모든 반환이 지나는 _stamp_shot_run 한 곳에서 적고, 빠뜨린 반환이 없는지 AST 로 센다. ★산출을 재사용한 방문의 uid 로 자산 계보를 덮지 않는다 — 그 자산을 만든 것은 이전 주행이다. shot_run_produced 로 가른다. 지문에는 안 들어간다(표시 칸일 뿐)." ``` --- ### Task 13: 샷 trace 에 영향 요약을 싣는다 **Files:** - Modify: `backend/app/modules/pipeline/multiroll_select.py` (record 확정 뒤) - Modify: `backend/app/modules/llm/opik_trace.py` (`update_trace_output`) - Test: `backend/tests/unit/test_shot_influence_summary.py` (신규) **Interfaces:** - Consumes: `current_trace`, `finish_trace` (Task 3) - Produces: - `build_influence_summary(record: Dict[str, Any]) -> Dict[str, Any]` - `update_trace_output(output: Dict[str, Any]) -> None` - [x] **Step 1: 실패하는 시험을 쓴다** ```python # backend/tests/unit/test_shot_influence_summary.py """샷 trace 의 영향 요약 — 전문이 아니라 요약이다.""" from app.modules.llm.opik_trace import build_influence_summary def _record(): return { "shot_run_uid": "0190-abc", "input_fingerprint": "e6ad7c49d0c20615", "ref_mode": "그룹 배경+엔티티 (2택1: 무콘티 승)", "share_plan": {"ref_plan": "background"}, "selected": "B", "ranking": ["B", "A"], "totals": {"A": 9, "B": 14}, "verdicts": [{"label": "B", "score": 7, "verdict_ko": "가" * 400}, {"label": "A", "score": 5, "verdict_ko": "나" * 400}], "refs": [ {"label": "LOCATION PHOTOGRAPH", "path": "/a/b.png", "asset_id": "loc-1", "role": "location_plate"}, {"label": "CHARACTER REFERENCE — 김선영", "path": "", "asset_id": "char-1", "role": "character_ref"}, ], "critique": {"issues": [{"issue_ko": "비니가 생략됨"}]}, "fix_skipped": True, "fix_skip_reason": "no_critical_issue", "prompt": "P" * 5000, } def test_summary_keeps_decision_axes(): s = build_influence_summary(_record()) assert s["selected"] == "B" assert s["ranking"] == ["B", "A"] assert s["totals"] == {"A": 9, "B": 14} assert s["input_fingerprint"] == "e6ad7c49d0c20615" assert s["ref_mode"].startswith("그룹 배경+엔티티") assert s["fix_applied"] is False assert s["fix_skip_reason"] == "no_critical_issue" def test_summary_lists_refs_with_asset_ids(): s = build_influence_summary(_record()) assert s["refs"] == [ {"label": "LOCATION PHOTOGRAPH", "asset_id": "loc-1", "role": "location_plate"}, {"label": "CHARACTER REFERENCE — 김선영", "asset_id": "char-1", "role": "character_ref"}, ] def test_summary_never_carries_full_prompt(): """전문은 span 에 이미 있다 — 두 벌로 두면 어느 쪽이 진짜인지 갈린다.""" s = build_influence_summary(_record()) assert "prompt" not in s blob = repr(s) assert "PPPPPPPPPP" not in blob def test_verdicts_are_scores_only(): s = build_influence_summary(_record()) assert s["verdicts"] == [{"label": "B", "score": 7}, {"label": "A", "score": 5}] def test_missing_fields_are_tolerated(): s = build_influence_summary({}) assert s["selected"] is None assert s["refs"] == [] assert s["fix_applied"] is False ``` - [x] **Step 2: 실패를 확인한다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_shot_influence_summary.py -v` Expected: FAIL — `ImportError: cannot import name 'build_influence_summary'` - [x] **Step 3: 요약 함수를 만든다** `backend/app/modules/llm/opik_trace.py` 에 넣는다. ```python def build_influence_summary(record: Dict[str, Any]) -> Dict[str, Any]: """샷 trace 의 `output` 에 실을 **영향 요약**. ★프롬프트 전문·판정 전문은 **안 싣는다.** 그것은 span 에 이미 있다 — 두 벌로 저장하면 어느 쪽이 진짜인지 갈린다. 여기에는 「무엇이 들어갔고 무엇이 골랐나」만 남긴다. """ refs = [] for r in (record.get("refs") or []): if not isinstance(r, dict): continue refs.append({"label": r.get("label"), "asset_id": r.get("asset_id"), "role": r.get("role")}) verdicts = [] for v in (record.get("verdicts") or []): if isinstance(v, dict): verdicts.append({"label": v.get("label"), "score": v.get("score")}) critique = record.get("critique") or {} issue_count = len(critique.get("issues") or []) if isinstance( critique, dict) else 0 return { "shot_run_uid": record.get("shot_run_uid"), "input_fingerprint": record.get("input_fingerprint"), "ref_mode": record.get("ref_mode"), "share_plan": record.get("share_plan"), "refs": refs, "selected": record.get("selected"), "ranking": record.get("ranking"), "totals": record.get("totals"), "verdicts": verdicts, "issue_count": issue_count, "fix_applied": bool(record.get("fix_rejudge") or record.get("repair_mode")) and not bool( record.get("fix_skipped")), "fix_skip_reason": record.get("fix_skip_reason"), "cine_applied": bool((record.get("cine") or {}).get("applied")) if isinstance(record.get("cine"), dict) else False, "needs_reshoot": bool(record.get("needs_reshoot")), } def update_trace_output(output: Dict[str, Any]) -> None: """지금 열려 있는 trace 의 output 을 채운다. 실패는 삼킨다.""" handle = current_trace() if handle is None: return try: live = _LIVE.get(handle.uid) if live is not None: live.update(output=output) except Exception as exc: # noqa: BLE001 logger.debug("update_trace_output 실패 (non-fatal): %s", exc) ``` - [x] **Step 4: 시험이 통과하는지 본다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_shot_influence_summary.py -v` Expected: PASS (5건) - [x] **Step 5: 부르는 자리는 Task 12 의 `_stamp_shot_run` 하나다** > 첫 판은 「마지막 `_persist()` 뒤 한 곳」이라고 썼는데, 그러면 > **critique 를 끄고 나가는 반환(~1546)에서 요약이 빈 채로** trace 가 닫힌다. > Task 12 Step 5 에서 만든 `_stamp_shot_run` 이 uid 와 요약을 **함께** 적고, > **모든 반환이 그것을 지난다.** 여기서는 새로 부르지 않는다. `_stamp_shot_run` 안의 이 두 줄이 그 자리다: ```python summary = build_influence_summary(record) summary["produced"] = bool(produced) update_trace_output(summary) ``` ★**요약 함수만 이 태스크의 몫**이다. 배선은 Task 12 가 이미 했다 — 두 곳에서 부르면 trace output 이 두 번 덮이고 어느 쪽이 마지막인지 갈린다. - [x] **Step 6: commit** ```bash git add backend/app/modules/llm/opik_trace.py \ backend/app/modules/pipeline/multiroll_select.py \ backend/tests/unit/test_shot_influence_summary.py git commit -m "feat(opik): 샷 trace 에 영향 요약을 싣는다 무엇이 들어갔고(refs+asset_id·ref_mode·share_plan) 무엇이 골랐고 (selected·ranking·totals·verdicts 점수) 무엇을 고쳤나(issue_count· fix_applied·fix_skip_reason·cine_applied)를 trace output 에 남긴다. 프롬프트 전문·판정 전문은 안 싣는다 — span 에 이미 있다. 두 벌로 저장하면 어느 쪽이 진짜인지 갈린다." ``` --- ### Task 14: 읽기 도구 `tools/shot_influence.py` **Files:** - Create: `tools/shot_influence.py` - Test: `backend/tests/unit/test_shot_influence_tool.py` (신규) **Interfaces:** - Consumes: `build_influence_summary` (Task 13) - Produces: - `merge_shot_influence(*, record, asset_row, opik_trace_row, opik_spans) -> Dict[str, Any]` - `render_text(merged: Dict[str, Any]) -> str` - [x] **Step 1: 실패하는 시험을 쓴다** ```python # backend/tests/unit/test_shot_influence_tool.py """세 자료원을 합쳐 한 화면으로 — 조회는 밖에서, 합치기는 순수 함수로.""" import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[3])) from tools.shot_influence import merge_shot_influence, render_text # noqa: E402 def _fixture(): return dict( record={ "shot_run_uid": "0190-abc", "selected": "B", "ranking": ["B", "A"], "totals": {"A": 9, "B": 14}, "input_fingerprint": "e6ad7c49", "ref_mode": "그룹 배경+엔티티", "refs": [{"label": "CHARACTER REFERENCE — 김선영", "asset_id": "char-1", "role": "character_ref"}], "verdicts": [{"label": "B", "score": 7}], "fix_skipped": True, "fix_skip_reason": "no_critical_issue", }, asset_row={"id": "asset-9", "file_path": "/p/S12sh3.png", "generation_call_id": "call-3", "shot_run_uid": "0190-abc"}, opik_trace_row={"id": "0190-abc", "name": "still:0190abc · still_recipe", "thread_id": "마지막 임무_1회_f5372927"}, opik_spans=[{"name": "roll · grok-imagine", "usage": {"total_tokens": 0}}, {"name": "judge · grok-4.6", "usage": {"total_tokens": 1500}}], ) def test_merge_keys_off_the_uid(): m = merge_shot_influence(**_fixture()) assert m["shot_run_uid"] == "0190-abc" assert m["sources"] == ["records.json", "image_asset", "opik"] def test_merge_flags_uid_mismatch(): """세 곳의 uid 가 다르면 조용히 넘어가면 안 된다.""" f = _fixture() f["asset_row"]["shot_run_uid"] = "0190-DIFFERENT" m = merge_shot_influence(**f) assert m["uid_mismatch"] == ["image_asset"] def test_merge_survives_missing_sources(): f = _fixture() f["opik_trace_row"] = None f["opik_spans"] = [] m = merge_shot_influence(**f) assert "opik" not in m["sources"] assert m["calls"] == [] def test_render_lists_refs_and_calls(): out = render_text(merge_shot_influence(**_fixture())) assert "CHARACTER REFERENCE — 김선영" in out assert "char-1" in out assert "judge · grok-4.6" in out assert "선정: B" in out def test_render_marks_unresolved_refs(): f = _fixture() f["record"]["refs"].append({"label": "PROP REFERENCE", "asset_id": None}) out = render_text(merge_shot_influence(**f)) assert "자산 불명" in out ``` - [x] **Step 2: 실패를 확인한다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_shot_influence_tool.py -v` Expected: FAIL — `ModuleNotFoundError: No module named 'tools.shot_influence'` - [x] **Step 3: 도구를 만든다** ```python # tools/shot_influence.py """최종 샷 한 장에 무엇이 영향을 미쳤나 — 세 자료원을 합쳐 한 화면으로. ./backend/.venv/bin/python -m tools.shot_influence 자료원 셋: · records.json — 문안·참조·판정·선정·수리·지문 · image_asset — 최종 파일과 계보 · Opik — 실제 호출들(프롬프트 전문·토큰) 합치기는 순수 함수(`merge_shot_influence`)로 두고 조회는 밖에서 한다 — 시험이 망을 안 타게 하기 위해서다. """ from __future__ import annotations import json from typing import Any, Dict, List, Optional def merge_shot_influence( *, record: Optional[Dict[str, Any]], asset_row: Optional[Dict[str, Any]], opik_trace_row: Optional[Dict[str, Any]], opik_spans: Optional[List[Dict[str, Any]]], ) -> Dict[str, Any]: """세 자료원을 uid 로 합친다. 없는 자료원은 조용히 빼고 표시한다.""" record = record or {} uid = (record.get("shot_run_uid") or (asset_row or {}).get("shot_run_uid") or (opik_trace_row or {}).get("id")) sources: List[str] = [] mismatch: List[str] = [] if record: sources.append("records.json") if asset_row: sources.append("image_asset") if asset_row.get("shot_run_uid") and asset_row["shot_run_uid"] != uid: mismatch.append("image_asset") if opik_trace_row: sources.append("opik") if opik_trace_row.get("id") and opik_trace_row["id"] != uid: mismatch.append("opik") return { "shot_run_uid": uid, "sources": sources, "uid_mismatch": mismatch, "refs": list(record.get("refs") or []), "ref_mode": record.get("ref_mode"), "selected": record.get("selected"), "ranking": record.get("ranking"), "totals": record.get("totals"), "verdicts": list(record.get("verdicts") or []), "fix_skipped": bool(record.get("fix_skipped")), "fix_skip_reason": record.get("fix_skip_reason"), "input_fingerprint": record.get("input_fingerprint"), "asset": asset_row or {}, "thread_id": (opik_trace_row or {}).get("thread_id"), "calls": [ {"name": s.get("name"), "tokens": (s.get("usage") or {}).get("total_tokens")} for s in (opik_spans or []) ], } def render_text(merged: Dict[str, Any]) -> str: """사람이 읽는 한 화면.""" L: List[str] = [] L.append(f"샷 uid : {merged.get('shot_run_uid')}") L.append(f"자료원 : {', '.join(merged.get('sources') or []) or '없음'}") if merged.get("uid_mismatch"): L.append(f"★uid 어긋남: {', '.join(merged['uid_mismatch'])}") L.append(f"주행 : {merged.get('thread_id')}") L.append(f"지문 : {merged.get('input_fingerprint')}") L.append("") L.append(f"── 무엇이 들어갔나 (참조 {len(merged.get('refs') or [])}건, " f"방식={merged.get('ref_mode')}) ──") for r in (merged.get("refs") or []): aid = r.get("asset_id") or "자산 불명" L.append(f" · {r.get('label')}") L.append(f" asset_id={aid} role={r.get('role')}") L.append("") L.append("── 무엇이 골랐나 ──") L.append(f" 선정: {merged.get('selected')} 순위: {merged.get('ranking')}" f" 점수: {merged.get('totals')}") for v in (merged.get("verdicts") or []): L.append(f" {v.get('label')}: {v.get('score')}") L.append("") L.append("── 무엇을 고쳤나 ──") if merged.get("fix_skipped"): L.append(f" 수리 건너뜀 — {merged.get('fix_skip_reason')}") else: L.append(" 수리 적용") L.append("") L.append(f"── 실제 호출 {len(merged.get('calls') or [])}건 ──") for c in (merged.get("calls") or []): tok = c.get("tokens") L.append(f" · {c.get('name')}" + (f" 토큰 {tok}" if tok else "")) L.append("") L.append(f"최종 파일: {(merged.get('asset') or {}).get('file_path')}") return "\n".join(L) ``` - [x] **Step 4: 시험이 통과하는지 본다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_shot_influence_tool.py -v` Expected: PASS (5건) - [x] **Step 5: 조회 부분을 붙인다** 같은 파일 아래에 `main()` 을 붙인다. DB 는 `PGPASSWORD`/`DATABASE_URL` 환경을 그대로 쓰고, Opik 은 `settings.opik_url_override` 를 쓴다. `records.json` 은 `still_id` 로 프로젝트 디렉토리를 찾아 읽는다. ```python def _fetch_opik(uid: str) -> tuple: import urllib.parse import urllib.request from app.core.config import settings base = settings.opik_url_override.rstrip("/") hdr = {"Comet-Workspace": settings.opik_workspace} def _get(path: str) -> Dict[str, Any]: req = urllib.request.Request(f"{base}{path}", headers=hdr) return json.load(urllib.request.urlopen(req, timeout=30)) projs = _get("/v1/private/projects?page=1&size=100") pid = next((p["id"] for p in projs.get("content", []) if p["name"] == settings.opik_project_name), None) if not pid: return None, [] tr = _get(f"/v1/private/traces?project_id={pid}&page=1&size=200") row = next((t for t in tr.get("content", []) if t.get("id") == uid), None) if row is None: return None, [] sp = _get(f"/v1/private/spans?project_id={pid}&trace_id={uid}" f"&page=1&size=200") return row, sp.get("content", []) def main(argv: Optional[List[str]] = None) -> int: import argparse ap = argparse.ArgumentParser(description="최종 샷의 영향 요인") ap.add_argument("key", help="still_id 또는 shot_run_uid") ns = ap.parse_args(argv) from app.core.database import SessionLocal from app.models.project import ImageAsset db = SessionLocal() try: # ★계약이 「still_id 또는 shot_run_uid」다 — 둘 다 찾아야 한다. # 첫 판은 still_id 로만 걸어, uid 를 주면 asset·records 를 못 찾고 # Opik 만 보여 「세 자료원 합치기」가 아니었다. q = db.query(ImageAsset).filter(ImageAsset.asset_type == "scene") row = q.filter(ImageAsset.still_id == ns.key).order_by( ImageAsset.created_at.desc()).first() if row is None: # uid 로 준 경우 — metadata 안을 본다. 자산 수가 많지 않아 # 훑어도 되지만, 프로젝트를 좁힐 수 있으면 좁힌다. for cand in q.order_by(ImageAsset.created_at.desc()).limit(5000): try: if json.loads(cand.pipeline_metadata_json or "{}").get( "shot_run_uid") == ns.key: row = cand break except (TypeError, ValueError): continue asset_row = None record = None if row is not None: meta = json.loads(row.pipeline_metadata_json or "{}") asset_row = {"id": row.id, "file_path": row.file_path, "generation_call_id": row.generation_call_id, "still_id": row.still_id, "shot_run_uid": meta.get("shot_run_uid")} from pathlib import Path rj = Path(row.file_path).parent / "recipe" / "records.json" if rj.exists(): allrec = json.loads(rj.read_text(encoding="utf-8")) want = meta.get("shot_run_uid") or ns.key record = next( (v for v in allrec.values() if isinstance(v, dict) and v.get("shot_run_uid") == want), None) uid = (record or {}).get("shot_run_uid") or ( asset_row or {}).get("shot_run_uid") or ns.key trace_row, spans = _fetch_opik(uid) if row is None and trace_row is None: print(f"'{ns.key}' 로 아무것도 못 찾았다 — still_id 인지 " f"shot_run_uid 인지 확인할 것") return 1 print(render_text(merge_shot_influence( record=record, asset_row=asset_row, opik_trace_row=trace_row, opik_spans=spans))) return 0 finally: db.close() if __name__ == "__main__": raise SystemExit(main()) ``` - [x] **Step 6: 실물로 한 번 돌린다 — 관문 11** Run: `cd /Users/manta/Documents/Projects/TheRoad-I1 && ./backend/.venv/bin/python -m tools.shot_influence <새 주행의 still_id>` Expected: 참조·선정·판정·수리·지문·호출이 한 화면에 - [x] **Step 7: commit** ```bash git add tools/shot_influence.py backend/tests/unit/test_shot_influence_tool.py git commit -m "feat(tools): shot_influence — 최종 샷의 영향 요인을 한 화면으로 records.json(문안·참조·판정·선정·수리·지문) + image_asset(파일·계보) + Opik(실제 호출·토큰)을 shot_run_uid 로 합친다. 합치기는 순수 함수로 두고 조회는 밖에 뒀다 — 시험이 망을 안 타게. 세 곳의 uid 가 어긋나면 조용히 넘어가지 않고 표시한다. 갤러리 「영향」 절은 범위 밖(사슬을 이은 뒤 별도 판)." ``` --- # 단계 4 — 감사 도구 이관 ### Task 15: 감사 도구를 span 기반으로 옮긴다 **Files:** - Modify: `tools/opik_prompt_audit/audit/fetch.py:59-100` - Modify: `tools/opik_prompt_audit/cache_baseline.py:40-60` - Test: `backend/tests/unit/test_opik_audit_span_source.py` (신규) **Interfaces:** - Consumes: Task 4~8 의 span 모양 - Produces: - `step_of(row: Dict[str, Any]) -> str` — `step:` 태그 우선, 없으면 기존 경로 - `is_test_origin(row: Dict[str, Any]) -> bool` **배경:** 지금 도구는 **trace 의 `input`** 과 `metadata.trace_name` 을 읽는다. v2 를 켜면 본문이 **span 으로 옮겨 간다.** 이득: span 에는 `usage`(토큰)와 `total_cost` 가 실린다 — 지금 trace 에는 없다. - [x] **Step 1: 실패하는 시험을 쓴다** ```python # backend/tests/unit/test_opik_audit_span_source.py """감사 도구의 자료원이 span 으로 옮겨 간다.""" import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[3])) from tools.opik_prompt_audit.audit.fetch import ( # noqa: E402 is_test_origin, step_of) def test_step_from_axis_tag(): assert step_of({"tags": ["step:scene_detail", "model:gpt-5.6-sol"]}) \ == "scene_detail" def test_step_falls_back_to_legacy_trace_name(): """지난 자료(v1)도 계속 읽혀야 한다 — 대조가 끊기면 안 된다.""" row = {"metadata": {"trace_name": "마지막 임무 > 1회 > scene_detail"}} assert step_of(row) == "scene_detail" def test_call_step_reaches_parent_even_when_span_name_is_set(): """★첫 판의 결함이 여기다. litellm span 의 name 은 `{model}_{obj}_{created}` 라 **절대 비지 않는다**. step_of 를 먼저 부르면 그 이름으로 떨어져 부모를 영영 안 본다. """ from tools.opik_prompt_audit.audit.fetch import step_of_call span = {"trace_id": "T1", "tags": ["model:gpt-5.6-sol"], "name": "gpt-5.6-sol_chat.completion_1755000000"} index = {"T1": {"id": "T1", "metadata": {"trace_name": "P > 1회 > scene_detail"}}} assert step_of_call(span, index) == "scene_detail" def test_call_step_prefers_own_axis_tag(): from tools.opik_prompt_audit.audit.fetch import step_of_call span = {"trace_id": "T1", "tags": ["step:shot_director"], "name": "x_y_1"} index = {"T1": {"id": "T1", "tags": ["step:scene_detail"]}} assert step_of_call(span, index) == "shot_director" def test_call_step_falls_back_to_span_name_last(): from tools.opik_prompt_audit.audit.fetch import step_of_call span = {"trace_id": "T9", "tags": [], "name": "still_recipe_roll/gpt-image-2"} assert step_of_call(span, {}) == "still_recipe_roll/gpt-image-2" def test_step_falls_back_to_name(): assert step_of({"name": "still_recipe_roll/gpt-image-2"}) \ == "still_recipe_roll/gpt-image-2" def test_axis_tag_wins_over_legacy(): row = {"tags": ["step:shot_director"], "metadata": {"trace_name": "a > b > scene_detail"}} assert step_of(row) == "shot_director" def test_test_origin_no_identity_zero_duration(): """갈래 ① — scope 밖에서 부른 시험. 신원 없고 시간 0.""" assert is_test_origin({"metadata": {"step": "s", "duration_ms": 0}}) is True assert is_test_origin({"metadata": { "step": "s", "duration_ms": 120}}) is False assert is_test_origin({"metadata": {}}) is False def test_test_origin_fake_identity(): """★갈래 ② — 가짜 신원을 달고 나간 시험. test_i2i_capture.py:34 는 generation_context("p-i2i", "e-i2i", "i2i_edit") 안에서 부르므로 project_id="p-i2i" 가 실려 갈래 ① 을 통과한다. """ assert is_test_origin({"metadata": { "step": "i2i_edit", "duration_ms": 0, "project_id": "p-i2i"}}) is True assert is_test_origin({"metadata": { "step": "i2i_edit", "duration_ms": 340, "project_id": "p-i2i"}}) is True def test_real_project_id_is_never_filtered(): """진짜 UUID 신원은 무슨 일이 있어도 안 거른다 — 실기록을 지우면 최악이다.""" assert is_test_origin({"metadata": { "step": "still_recipe", "duration_ms": 0, "project_id": "5bddbdfc-2681-42a6-9837-43f35f60049d"}}) is False def test_test_span_is_filtered_via_its_parent(): """★★span 만 보면 시험 기록을 못 거른다 — 부모를 함께 봐야 한다. ImageTracer 는 trace metadata 에만 step 을 넣고(image_tracer.py:171-175) span metadata 에는 안 넣는다(:185-195). is_test_origin 은 "step" 을 요구하므로 span 은 무조건 통과한다. 부모 trace 만 지워지고 자식 span 이 새 감사 SOT 에 그대로 들어간다 — 관문 「시험 기록 0건」이 거짓 통과한다. """ from tools.opik_prompt_audit.audit.fetch import select_call_rows test_span = {"trace_id": "T-test", "metadata": {"duration_ms": 0, "ref_count": 0}, "input": {"prompt": "x"}} real_span = {"trace_id": "T-real", "metadata": {"duration_ms": 340, "project_id": "5bddbdfc-2681-42a6-9837-43f35f60049d"}} index = { "T-test": {"id": "T-test", "metadata": {"step": "i2i_edit", "duration_ms": 0}}, "T-real": {"id": "T-real", "metadata": {"step": "still_recipe", "duration_ms": 340, "project_id": "5bddbdfc-2681-42a6-9837-43f35f60049d"}}, } out = select_call_rows([test_span, real_span], index) assert out == [real_span], "시험 span 이 살아남았다" def test_orphan_span_is_kept(): """부모를 못 찾아도 버리지 않는다 — 못 찾은 것과 시험인 것은 다르다.""" from tools.opik_prompt_audit.audit.fetch import select_call_rows sp = {"trace_id": "T-unknown", "metadata": {"duration_ms": 120}} assert select_call_rows([sp], {}) == [sp] def test_trace_index_widens_the_window(): """★부모 창을 안 넓히면 경계 직전 시작한 긴 trace 의 span 이 부모를 잃는다. 부모를 잃으면 시험 판별도 legacy 스텝 fallback 도 함께 깨진다. """ import inspect from tools.opik_prompt_audit.audit import fetch src = inspect.getsource(fetch.fetch_trace_index) assert "pad_hours" in src, "부모 조회 창을 안 넓힌다" def test_caveat_is_reported(): """★「걸러진 수」가 「오염 총량」으로 읽히면 안 된다.""" from tools.opik_prompt_audit.audit.fetch import test_filter_caveat assert "완전하지 않다" in test_filter_caveat() ``` - [x] **Step 2: 실패를 확인한다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_opik_audit_span_source.py -v` Expected: FAIL — `ImportError: cannot import name 'is_test_origin'` - [x] **Step 3: `step_of` 를 넓히고 `is_test_origin` 을 만든다** `tools/opik_prompt_audit/audit/fetch.py:59` 의 `step_of` 를 바꾼다. ```python def step_of(row: Dict[str, Any]) -> str: """trace/span 의 스텝 이름. 1순위 = `step:` 축 태그(v2). 2순위 = litellm 메타의 `trace_name` 마지막 조각(v1 — 지난 자료를 계속 읽기 위해). 3순위 = 이름. """ for tag in (row.get("tags") or []): if isinstance(tag, str) and tag.startswith("step:"): return tag[len("step:"):] md = row.get("metadata") or {} tn = md.get("trace_name") if isinstance(md, dict) else None if isinstance(tn, str) and ">" in tn: return tn.split(">")[-1].strip() if isinstance(tn, str) and tn: return tn return str(row.get("name") or "?") #: 시험이 쓰는 가짜 신원 — 실제 프로젝트 UUID 가 아니다. #: 늘려야 할 일이 생기면 여기에 보탠다(코드가 아니라 목록이라 안전하다). _TEST_PROJECT_IDS = frozenset({"p-i2i", "p", "proj-1", "p1", "test"}) def is_test_origin(row: Dict[str, Any]) -> bool: """시험이 남긴 기록인가 — 감사 셈에서 뺀다. 2026-08-23 이전에는 돈 가드가 집 안 주소를 통과시켜 시험이 프로덕션 프로젝트에 썼다. 지우지 않고 여기서 거른다 — 지우는 것은 되돌릴 수 없다. ★**이 그물은 완전하지 않다.** 두 갈래로 잡는다: 1. 신원 없이 시간 0 — `record_provider_call` 을 scope 밖에서 부른 시험 2. **가짜 신원** — `test_i2i_capture.py:34` 는 `generation_context("p-i2i", "e-i2i", "i2i_edit")` 안에서 부르므로 `project_id="p-i2i"` 가 실려 1번 그물을 **통과한다** 그래도 못 잡는 것이 남는다(litellm 경로를 타는 시험 등). 그래서 리포트에 **「걸러진 수」와 함께 「이 그물이 못 잡는 모양」**을 같이 적는다 — 「0 건」이 「없다」로 읽히면 안 된다. """ md = row.get("metadata") or {} if not isinstance(md, dict): return False pid = md.get("project_id") # ② 가짜 신원 — 실제 프로젝트 id 는 UUID 다 if isinstance(pid, str) and pid in _TEST_PROJECT_IDS: return True # ① 신원 없이 시간 0 (이미지 기록 모양일 때만) if "step" not in md or "duration_ms" not in md: return False return md.get("duration_ms") == 0 and not pid def test_filter_caveat() -> str: """리포트에 함께 실을 한 줄 — 이 그물의 한계를 밝힌다.""" return ("★시험 기록 필터는 완전하지 않다 — 가짜 신원 목록에 없는 값이나 " "litellm 경로를 탄 시험 기록은 통과한다. 「걸러진 수」를 " "「오염 총량」으로 읽지 말 것.") ``` - [x] **Step 4: 자료원을 span 으로 바꾼다 — ★기존 계약을 그대로 지킨다** > **2026-08-23 Codex 재리뷰로 교정.** 첫 판의 `fetch_call_rows(base, project_id)` 는 > **실제 인터페이스와 달랐다.** 지금 것은 > `fetch_traces(base_url, workspace, project_name, since, until)` > (`fetch.py:13-56`) — `project_id` 가 아니라 **`project_name`** 을 쓰고 > **기간(since/until)** 으로 자른다. 첫 판대로면 기간 창이 사라져 **전 기간을 > 센다.** > > 그리고 **소비처를 안 갈아 끼웠다.** 실제로 부르는 곳은 셋이다: > `main.py:44` · `cache_baseline.py:200` · `metrics.py:33`(각 행에 `step_of`). > 「`cache_baseline` 도 자동으로 따라온다」는 **틀렸다** — 그 파일이 스스로 > `fetch_traces` 를 부른다. `tools/opik_prompt_audit/audit/fetch.py` 에 **기존 계약 그대로** 두 함수를 보탠다. ```python def fetch_spans( base_url: str, workspace: str, project_name: str, since: str, until: str = "9999", ) -> List[Dict[str, Any]]: """감사 대상 = **호출 하나당 한 행**. 그것은 언제나 span 이다. ★span 과 trace 를 그냥 합치면 **지난 자료를 두 번 센다.** `ImageTracer.log`(`image_tracer.py:176-198`)는 호출 하나마다 trace 를 만들고 그 밑에 span 도 만든다. litellm 도 `trace_id` 가 없으면 trace 를 만들고 span 은 **항상** 만든다(`opik_payload_builder/api.py:81-99`). 즉 v1 호출 한 건이 **trace 1 + span 1** 로 남아 있다. 두 배가 된 숫자로는 「팩이 얼마나 커졌나」를 못 묻는다 — 이 도구의 존재 이유가 사라진다. 그래서 **세는 것은 언제나 span** 이고 trace 는 묶음·이름을 얻는 데만 쓴다(`fetch_trace_index`). ★인자·기간 계약은 `fetch_traces` 와 **같게** 둔다. 다르면 소비처가 조용히 전 기간을 센다. """ return _fetch_paged("spans", base_url, workspace, project_name, since, until) def fetch_trace_index( base_url: str, workspace: str, project_name: str, since: str, until: str = "9999", ) -> Dict[str, Dict[str, Any]]: """`trace_id` → trace. **세는 데는 안 쓴다** — 묶음·이름 용도.""" return {t["id"]: t for t in _fetch_paged( "traces", base_url, workspace, project_name, since, until) if t.get("id")} ``` `_fetch_paged` 는 지금 `fetch_traces` 의 페이지 순회 본문을 그대로 옮겨 종류만 인자로 받는 함수다. **`fetch_traces` 는 그것을 부르게 바꾼다** — 페이지·기간 논리를 두 벌로 만들면 갈린다. ```python def _fetch_paged(kind: str, base_url: str, workspace: str, project_name: str, since: str, until: str ) -> List[Dict[str, Any]]: """`traces`/`spans` 공통 페이지 순회. 기간 자르기 규칙은 하나뿐이다. ★**여기서 거르지 않는다.** 시험 기록 판별은 span 과 부모 trace 를 **함께 봐야** 하므로(아래 `select_call_rows`) 이 함수는 raw 로 준다. """ rows: List[Dict[str, Any]] = [] page = 1 while True: q = urllib.parse.urlencode({ "project_name": project_name, "size": PAGE_SIZE, "page": page, "sorting": json.dumps( [{"field": "start_time", "direction": "DESC"}]), }) req = urllib.request.Request( f"{base_url}/v1/private/{kind}?{q}", headers={"Comet-Workspace": workspace}) with urllib.request.urlopen(req, timeout=60) as r: d = json.loads(r.read()) content = d.get("content", []) if not content: break stop = False for t in content: st = t.get("start_time", "") if st < since: stop = True break if st < until: rows.append(t) if stop or page * PAGE_SIZE >= int(d.get("total", 0)): break page += 1 logger.info("%s %d건 수집 (since=%s)", kind, len(rows), since) return rows def fetch_traces(base_url, workspace, project_name, since, until="9999"): """v1 호환 — 지금 소비처가 이것을 부른다.""" return _fetch_paged("traces", base_url, workspace, project_name, since, until) ``` - [x] **Step 4-B: ★시험 기록은 span 과 부모를 **함께** 봐야 걸러진다** > **2026-08-23 3차 리뷰.** 첫 판은 `_fetch_paged` 안에서 trace 와 span 에 > **같은 필터**를 걸었다. 그러면 **바로 그 시험 span 이 살아남는다.** > > `ImageTracer.log`(`image_tracer.py:171-195`)를 보면: > > ```python > trace_metadata = {"model": …, "duration_ms": …, "step": step} # step 있다 > span_metadata = {"duration_ms": …, **(params or {})} # step 없다 > ``` > > `is_test_origin` 은 `"step" in md` 를 요구한다 → **span 은 무조건 통과**한다. > 부모 trace 만 지워지고 자식 span 은 새 감사 SOT 에 그대로 들어간다. > `test_i2i_capture.py:47-56` 의 `prompt="x"` 가 정확히 이 경로다. > > 관문 「시험 기록 0건」이 거짓 통과한다. ```python def select_call_rows(spans: List[Dict[str, Any]], trace_index: Dict[str, Dict[str, Any]] ) -> List[Dict[str, Any]]: """감사 대상 span 만 고른다 — **자기 또는 부모가 시험이면 뺀다**. span metadata 에는 `step` 이 없어서(`image_tracer.py:185`) span 만 보면 시험 판별이 안 된다. 부모 trace 에는 있다. **둘을 함께 본다.** """ out = [] for sp in spans: if is_test_origin(sp): continue parent = trace_index.get(str(sp.get("trace_id"))) if parent is not None and is_test_origin(parent): continue out.append(sp) return out def fetch_spans(base_url, workspace, project_name, since, until="9999"): """감사 대상 = 호출 하나당 한 행. 그것은 언제나 span 이다. (raw)""" return _fetch_paged("spans", base_url, workspace, project_name, since, until) def fetch_trace_index(base_url, workspace, project_name, since, until="9999", *, pad_hours: int = 24 ) -> Dict[str, Dict[str, Any]]: """`trace_id` → trace. **거르지 않는다** — 부모 조회용이다. ★창을 `pad_hours` 만큼 **앞으로 넓힌다.** 창 경계 직전에 시작한 긴 trace 의 창 안 span 이 부모를 잃으면, 시험 판별도 legacy 스텝 fallback 도 함께 깨진다. 부모를 못 찾는 것이 여기서는 가장 나쁘다. """ import datetime as _dt try: _s = _dt.datetime.fromisoformat(since.replace("Z", "+00:00")) wide = (_s - _dt.timedelta(hours=pad_hours)).isoformat() except ValueError: wide = since # 형식이 다르면 그대로 — 좁히지는 않는다 return {t["id"]: t for t in _fetch_paged( "traces", base_url, workspace, project_name, wide, until) if t.get("id")} ``` - [x] **Step 4-B: `step_of_call` 의 순서를 바로잡는다** > 첫 판의 `step_of_call` 은 **부모 fallback 에 절대 도달하지 못한다.** > `step_of` 의 마지막 갈래가 `return str(row.get("name") or "?")` > (`fetch.py:59-67`)인데, litellm span 의 `name` 은 > `f"{model}_{obj_type}_{created}"` 라 **절대 비지 않는다.** `step:` 태그만 따로 뽑고, 없으면 부모 trace 를 보고, 그래도 없으면 이름으로 떨어진다. ```python def _step_tag(row: Dict[str, Any]) -> Optional[str]: """`step:` 축 태그만 — 없으면 None. 이름으로 안 떨어진다.""" for tag in (row.get("tags") or []): if isinstance(tag, str) and tag.startswith("step:"): return tag[len("step:"):] return None def step_of_call(span: Dict[str, Any], trace_index: Dict[str, Dict[str, Any]]) -> str: """호출(span) 하나의 스텝 이름. ① 자기 `step:` 태그 (v2) ② 부모 trace 의 스텝 (v1 — `{step}/{model}` 이름이나 `metadata.trace_name`) ③ 그래도 없으면 span 이름 ★②를 ③보다 먼저 본다. `step_of` 를 먼저 부르면 span 이름으로 떨어져 부모를 영영 안 본다 — litellm span 이름은 비지 않기 때문이다. """ tag = _step_tag(span) if tag: return tag parent = trace_index.get(str(span.get("trace_id"))) if parent is not None: ptag = _step_tag(parent) if ptag: return ptag pname = step_of(parent) if pname and pname != "?": return pname return step_of(span) ``` - [x] **Step 4-C: ★소비처 셋을 실제로 갈아 끼운다** 이것을 빠뜨리면 위 코드는 **아무도 안 부르는 죽은 코드**다. | 파일·줄 | 지금 | 바꿀 것 | |---|---|---| | `main.py:38-47` | `fetch_traces(...)` | `idx = fetch_trace_index(...)` → `rows = select_call_rows(fetch_spans(...), idx)` | | `audit/metrics.py:16,33` | `step_of(t)` | `step_of_call(span, trace_index)` — 함수가 `trace_index` 를 받게 시그니처를 넓힌다 | | `cache_baseline.py:198-200` | `fetch_traces(...)` | 같은 셋(`fetch_trace_index` + `fetch_spans` + `select_call_rows`). 캐시 항목(`usage`)은 span 에 있으므로 오히려 정확해진다 | | `cache_baseline.py:40-60` `step_name` | `step_of(trace)` | `step_of_call(span, trace_index)` | | **`main.py:63-68`** | `{"trace_count": len(traces)}` | `{"call_count": len(rows)}` — 아래 별항 | | **`report.py:45-54`** | 「trace N건」·「trace=Opik 기록 1건」 | 「호출 N건」·「호출=span 1건」 | | **`cache_baseline.py:62-66`·`201-210`** | `trace_count`·「한 trace=한 발송」 | `call_count`·「한 span=한 발송」 | ★`cache_baseline.py` 는 **자기가 `fetch_traces` 를 부른다** — `step_of` 만 고쳐서는 안 따라온다. ### ★라벨을 안 바꾸면 결과를 오독시킨다 세는 대상이 trace 에서 span 으로 바뀌는데 리포트가 여전히 「trace N건」이라고 쓰면, **같은 이름의 값이 다른 것을 가리키게 된다.** 전후 비교가 통째로 틀린다. 이름과 문구를 **함께** 바꾼다. 구 JSON 을 읽는 것이 있으면 `trace_count` 를 별칭으로만 병기하고 새 값은 `call_count` 로 쓴다. ### ★토큰·비용은 이번 범위 밖이다 — 주장을 좁힌다 스펙 ⑩ 이 「span 에 `usage`·`total_cost` 가 있어 **글자 대신 토큰으로 말할 수 있게 된다**」고 썼는데, **그 배선이 이 계획에 없다.** `metrics.per_step` (`metrics.py:22-67`)은 `messages_of` 로 **글자 수만** 세고 `report.py:47-68` 도 문자량만 낸다(`cache_baseline` 만 `usage` 를 읽는다). → 이번 판은 **글자 감사 그대로** 두고, 「토큰으로 말한다」는 주장을 뺀다. span 으로 옮기면 `usage` 가 **손 닿는 곳에 온다**는 것까지만 적는다. 토큰·비용 집계는 `metrics`·`report` 를 함께 고치는 **별도 판**이다. ★산출물 설명이 실제 결과보다 강하면 그것도 결과 오독이다. - [x] **Step 4-D: 리포트에 그물의 한계를 싣는다** `main.py` 의 리포트 머리말에 `test_filter_caveat()` 한 줄을 넣는다. 「걸러진 수」가 「오염 총량」으로 읽히면 안 된다. - [x] **Step 5: 시험이 통과하는지 본다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_opik_audit_span_source.py -v` Expected: PASS (8건) - [x] **Step 6: 관문 7 — 이전 리포트와 스텝 이름이 맞는지 본다** Run: `cd /Users/manta/Documents/Projects/TheRoad-I1 && ./backend/.venv/bin/python -m tools.opik_prompt_audit.main` Expected: 리포트가 나오고, 스텝 이름 목록이 `artifact/20260817_opik_audit_full/report.html` 의 것과 겹친다. 시험 기록(`step` 이 `?` 이거나 프롬프트가 `xxxx`)이 **0건** - [x] **Step 7: commit** ```bash git add tools/opik_prompt_audit/ backend/tests/unit/test_opik_audit_span_source.py git commit -m "fix(audit): 감사 도구 자료원을 span 으로 + 시험 기록 걸러내기 v2 를 켜면 프롬프트 본문이 trace 에서 span 으로 옮겨 간다. span 과 trace 를 둘 다 읽어 지난 자료(v1)와의 대조를 끊지 않는다. 이득: span 에는 usage(토큰)와 total_cost 가 실려 있다 — 지금 trace 에는 없어서 못 보던 값이다. 글자 수 대신 토큰으로 말할 수 있다. 이미 섞인 시험 기록 약 2,600건은 지우지 않고 여기서 거른다(is_test_origin). 지우는 것은 되돌릴 수 없다." ``` --- # 단계 5 — 곁다리 수리 ### Task 16: i2i 경로가 DB 에 한 행도 안 남는 것 **Files:** - Modify: `backend/app/modules/gemini_i2i_editor.py:186-260` (원인에 따라) - Test: `backend/tests/unit/test_i2i_db_logging.py` (신규) **Interfaces:** - Consumes: 없음 - Produces: 없음 (결함 수리) > ★★**2026-08-24 실행 중 전제가 틀린 것으로 판명됐다 — 고칠 결함이 아니었다.** > > 「Opik 에는 i2i 가 있는데 DB 에 없다」를 프로덕션 결함으로 읽었는데, 실제로 > 재 보니 **Opik 의 i2i trace 84건이 전부 시험 기록**이었다: > `project_id="p-i2i"` 21건(가짜 신원) · `project_id=None`·`duration_ms=0`· > `model="m"` 63건(scope 밖 호출). 프로덕션에서 그 경로가 안 돌았고 DB 0건이 > **정상**이다. > > ★이것은 **단계 1 이 고친 바로 그 문제의 결과**다 — 시험이 프로덕션 Opik 에 > 쓰고 있었고, 그 기록을 보고 프로덕션 결함이라고 판단했다. **오염된 자료로는 > 진단도 오염된다.** > > 실제로 한 것: ①`record_provider_call` 의 DB 기록 실패를 `logger.debug` → > `logger.warning` 으로 올렸다(조용히 삼키는 것이 이 조사를 부른 원인이다. > Opik 쪽은 debug 유지 — 자체 호스팅이 죽으면 호출마다 warning 이 쏟아진다) > ②이 판정을 시험으로 못박았다(`test_i2i_db_logging.py`) ③헬퍼가 멀쩡함을 > 확인했다(시그니처 12개가 `log_llm_call` 과 정확히 일치). > > ★**「원인을 확인하기 전에 고치지 않는다」가 실제로 값을 했다.** 계획대로 > 곧장 고쳤다면 없는 결함을 고치느라 호출 경로를 건드렸을 것이다. **배경 (당시 실측 — 위 정정 참조):** Opik 에는 `i2i_edit/...` trace 가 있는데 `llm_call_log` 에는 **`step_name` 이나 `operation_type` 에 i2i 가 든 행이 0건**이다. - [x] **Step 1: 원인을 먼저 잰다 — ★프로덕션에 쓰지 않고** > 2026-08-23 Codex 지적. 첫 판은 `record_provider_call` 을 **일반 backend > 환경에서 그냥 불렀다.** 그 함수는 `to_db=True` 가 기본이라 > (`image_tracer.py:374-408`) 부르는 순간 **프로덕션 `llm_call_log` 에 진짜 > 행이 남고 Opik 에도 쓴다.** 유료 호출은 없지만 **기록이 오염된다** — > 이 문서가 고치려는 바로 그 문제(⑨)를 진단이 반복하는 꼴이었다. **시험 DB + 시험 Opik 프로젝트**에서만 잰다. conftest 가 이미 그 환경을 만들어 주므로 pytest 로 재는 것이 가장 안전하다. ```bash cd backend && ./.venv/bin/pytest tests/unit/ -q -s -k "i2i_db_logging" 2>&1 | tail -20 ``` 시험이 아직 없으니, 재는 것부터 시험으로 쓴다(Step 2). 손으로 확인해야 하면 **반드시** 환경을 갈아입힌다: ```bash cd backend && \ DATABASE_URL="postgresql://theroad:theroad_dev_2026@localhost:5432/theroad_test" \ OPIK_PROJECT_NAME="theroad-scene-lab-test" \ ./.venv/bin/python -c " from app.core.config import settings assert 'theroad_test' in settings.database_url, settings.database_url assert settings.opik_project_name.endswith('-test'), settings.opik_project_name from app.modules.llm.image_tracer import record_provider_call cid = record_provider_call( step='i2i_edit_probe', model='probe-model', prompt='확인', status='success', duration_ms=1, meta={'project_id':'p','episode_id':'e'}, operation='i2i_edit', output_text='[probe]') print('call_id =', cid) " PGPASSWORD=theroad_dev_2026 psql -h localhost -U theroad -d theroad_test -c \ "SELECT id, step_name, operation_type FROM llm_call_log WHERE step_name='i2i_edit_probe';" ``` ★`assert` 두 줄이 핵심이다 — 환경이 안 갈렸으면 **한 줄도 쓰기 전에 멈춘다.** - `call_id` 가 나오고 DB 행도 있으면 → `record_provider_call` 은 멀쩡하다. 원인은 **호출 자리**(`gemini_i2i_editor` 가 그 경로를 안 타거나 예외로 빠짐) - `call_id` 가 `None` 이면 → `log_llm_call` 자체가 실패한다. 로그를 켜서 예외를 본다 **측정 결과에 따라 Step 4 의 수리 내용이 갈린다. 원인을 확인하기 전에 고치지 않는다.** - [x] **Step 2: 헬퍼가 멀쩡한지 못박는 시험을 쓴다** 이 시험은 원인과 무관하게 지금 쓸 수 있다 — `record_provider_call` 이 DB 에 무엇을 넘기는지를 본다. ```python # backend/tests/unit/test_i2i_db_logging.py """i2i 직접 호출도 DB(llm_call_log)에 남아야 한다. 실측(2026-08-23): Opik 에는 i2i_edit trace 가 있는데 llm_call_log 에는 step_name/operation_type 에 i2i 가 든 행이 0건이었다. 한쪽에만 남으면 '이 호출이 무엇을 만들었나'를 DB 로 못 묻는다. """ def test_record_provider_call_passes_step_and_operation(monkeypatch): seen = {} def _fake_log(**kw): seen.update(kw) return "call-1" import app.modules.llm.llm_logger as llm_logger monkeypatch.setattr(llm_logger, "log_llm_call", _fake_log) from app.modules.llm.image_tracer import record_provider_call cid = record_provider_call( step="i2i_edit", model="gemini-3.1-flash-image-preview", prompt="p", status="success", duration_ms=12, meta={"project_id": "p1", "episode_id": "e1"}, operation="i2i_edit", output_text="[image generated]") assert cid == "call-1" assert seen["step_name"] == "i2i_edit" assert seen["operation_type"] == "i2i_edit" assert seen["project_id"] == "p1" def test_db_failure_does_not_lose_opik_record(monkeypatch): """DB 가 터져도 Opik 기록은 남아야 한다 — 둘이 함께 죽으면 안 된다.""" import app.modules.llm.llm_logger as llm_logger def _boom(**kw): raise RuntimeError("DB 죽음") monkeypatch.setattr(llm_logger, "log_llm_call", _boom) logged = [] from app.modules.llm import image_tracer class _T: def log(self, **kw): logged.append(kw) monkeypatch.setattr(image_tracer, "get_image_tracer", lambda: _T()) cid = image_tracer.record_provider_call( step="i2i_edit", model="m", prompt="p", status="success", duration_ms=1, meta={}, operation="i2i_edit") assert cid is None assert len(logged) == 1 ``` - [x] **Step 3: 실패를 확인하고, 원인에 맞는 시험을 하나 더 쓴다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_i2i_db_logging.py -v` 두 시험이 통과하면 **헬퍼는 멀쩡하다** — 원인은 호출 자리다. 그러면 Step 1 의 측정을 근거로 실제 원인을 재현하는 시험을 **하나 더** 쓴다. 통과하지 않으면 헬퍼가 원인이다. ★원인을 확인하기 전에 시험을 짐작해서 쓰지 않는다 — 엉뚱한 것을 굳힌다. - [x] **Step 4: 원인을 고친다** Step 1·3 의 결과에 따라: - **호출 자리가 문제면** — `gemini_i2i_editor.py` 의 성공 경로(241줄)와 `_fail`(187줄)이 실제로 도는지 확인한다. 특히 **함수 안 local import** 가 조용히 실패하는 자리를 본다(이 저장소에 그 결함 유형이 있었다 — `try` 안이라 조용히 failed 로 남는다) - **`log_llm_call` 이 문제면** — `image_tracer.py:395` 의 `logger.debug` 를 `logger.warning` 으로 올려 다음에는 보이게 한다. 조용히 삼키는 것이 이 결함을 넉 달 숨겼다 - [x] **Step 5: 곁다리 — DB `step_name` 빈 행 327건도 같이 본다** spec ⑪-3. 같은 성격(DB 기록 결손)이라 함께 잰다. ```bash PGPASSWORD=theroad_dev_2026 psql -h localhost -U theroad -d theroad -c " SELECT operation_type, model_name, count(*) n FROM llm_call_log WHERE (step_name IS NULL OR step_name = '') AND created_at > '2026-08-14' GROUP BY 1,2 ORDER BY n DESC LIMIT 10;" ``` Task 6·7 의 trace scope 배선으로 `resolve_step_name` 이 `stage` 를 찾게 되면 이 빈 행이 자연히 메워진다. **메워지면 손대지 않는다.** ★**2026-08-24 실측**: 빈 행 340건 중 **327건이 `still_recipe_roll`** (gemini 220 + grok 107)이다. 예측한 그 부류가 맞다 — Task 7-B 의 샷 scope 가 `stage` 를 주면 메워진다. **단 v2 가 켜져야 한다**(OFF 면 scope 를 안 연다 — 그것이 「OFF 는 바이트 동일」계약이다). 나머지 13건은 `operation_type` 까지 비어 있어 호출자가 불명이다 — 소수라 지금 파지 않고 켠 뒤 다시 센다. → 켜기 뒤 확인 항목. 안 메워지면 어느 호출자가 스텝을 안 넘기는지 위 출력의 `operation_type` 으로 짚어 그 자리만 고친다. - [x] **Step 6: 시험이 통과하는지 본다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_i2i_db_logging.py -v` Expected: PASS - [x] **Step 7: commit** ```bash git add backend/app/modules/gemini_i2i_editor.py backend/app/modules/llm/image_tracer.py \ backend/tests/unit/test_i2i_db_logging.py git commit -m "fix(logging): i2i 직접 호출이 DB 에 안 남던 것 Opik 에는 i2i_edit trace 가 있는데 llm_call_log 에는 i2i 가 든 행이 0건이었다. 한쪽에만 남으면 '이 호출이 무엇을 만들었나'를 DB 로 못 묻는다. DB 기록 실패를 logger.debug 로 삼키던 것을 warning 으로 올렸다 — 조용히 삼키는 것이 이 결함을 넉 달 숨겼다." ``` --- ### Task 17: 죽은 키 `trace_name` 을 걷어낸다 > ★★**2026-08-24 보류 — 켜기 직후로 미룬다.** `trace_name` 은 v2 에서 이미 > `pop` 되므로 **v1(설정 OFF) 경로에서만** 만들어진다. 지금 그 줄을 지우면 > **OFF payload 가 바뀌어**, 아직 못 잰 주행 관문(0~11) 전에 「되돌리기는 한 > 줄」이라는 안전판이 먼저 사라진다. 이 태스크 자신이 「단계 2~4 가 나가고 > **관문이 통과한 뒤**」라는 조건을 달아 뒀다 — 그 관문은 주행이 있어야 잰다. > → 켜기 절차(⑭)에서 관문을 통과한 **직후**에 한다. (Codex 도 이 판단에 > 이견 없음) **Files:** - Modify: `backend/app/core/step_runner.py:1620` - Test: `backend/tests/unit/test_opik_no_dead_keys.py` (신규) **★이 태스크는 단계 2~4 가 전부 나가고 관문이 통과한 뒤에 한다.** 감사 도구가 v1 자료의 `metadata.trace_name` 을 아직 읽는다(Task 15 의 2순위 경로). - [ ] **Step 1: 시험을 쓴다** ```python # backend/tests/unit/test_opik_no_dead_keys.py """죽은 키는 내보내지 않는다.""" def test_trace_name_is_not_built_anymore(): """litellm 은 trace_name 을 안 읽는다 — trace 이름은 항상 response_obj["object"] 다. 만들어 실으면 trace metadata 만 더럽힌다.""" from pathlib import Path src = (Path(__file__).resolve().parents[2] / "app" / "core" / "step_runner.py").read_text(encoding="utf-8") assert 'meta["trace_name"]' not in src def test_audit_tool_still_reads_legacy_trace_name(): """지난 자료를 계속 읽을 수 있어야 한다 — 대조가 끊기면 안 된다.""" import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[3])) from tools.opik_prompt_audit.audit.fetch import step_of assert step_of({"metadata": {"trace_name": "a > b > scene_detail"}}) \ == "scene_detail" ``` - [ ] **Step 2: 실패를 확인한다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_opik_no_dead_keys.py -v` Expected: 첫 시험 FAIL - [ ] **Step 3: `step_runner.py:1620` 의 줄을 지운다** ```python ctx = self.opik_context if ctx.get("run_tag"): meta["session_id"] = ctx["run_tag"] # v1 경로 — v2 에서 thread_id 로 ``` `meta["trace_name"] = ...` 한 줄만 지운다. `session_id` 는 v1(설정 OFF) 경로가 아직 쓰므로 **남긴다.** - [ ] **Step 4: 시험이 통과하는지 본다** Run: `cd backend && ./.venv/bin/pytest tests/unit/test_opik_no_dead_keys.py -v` Expected: PASS (2건) - [ ] **Step 5: commit** ```bash git add backend/app/core/step_runner.py backend/tests/unit/test_opik_no_dead_keys.py git commit -m "chore(opik): 죽은 키 trace_name 제거 litellm 은 trace_name 을 안 읽는다 — trace 이름은 항상 response_obj['object'] 로 못박혀 있다. 만들어 실으면 trace metadata 만 더럽힌다. 감사 도구는 지난 자료(v1)의 metadata.trace_name 을 계속 읽는다 — 대조가 끊기지 않게." ``` --- ## 켜기 — 마지막 관문 전 태스크가 끝나고 새 주행을 시작하기 **직전에** 한다. - [ ] `backend/.env` 에 `OPIK_TRACE_V2_ENABLED=true` 를 넣는다 - [ ] 백엔드를 재기동한다 (★`backend` cwd + `.venv` — 기동 시각을 `ps` 로 확인) - [ ] `curl -s http://127.0.0.1:8000/api/v1/health` → `status:ok` - [ ] 새 주행 첫 10~15샷에서 관문 0~11 을 전부 잰다 - [ ] 하나라도 깨지면 `.env` 줄을 지우고 재기동한다 (되돌리기는 한 줄이다) ## 관문 표 (spec ⑬ 과 같다) | 관문 | 무엇 | 통과 기준 | 태스크 | |---|---|---|---| | 0 | 기록이 있나 | 새 trace 가 하나라도 생긴다 | 9 | | 1 | `chat.completion` 이 없나 | 새 주행 구간 trace 이름에 0건 | 9 | | 2 | 주행이 묶이나 | 재개 1회 끼워도 thread_id 1개 | 5·9 | | 3 | 계층이 서나 | 샷 trace 하나에 span 여럿 | 7·8·9 | | 4 | 축이 갈리나 | **trace** 태그가 전부 `접두사:` 꼴 | 2·5·9 | | 5 | 샷 계보 | `still:` trace 하나로 그 샷 호출이 다 보인다 | 7·9 | | 6 | 시험 격리 | 시험 한 바퀴 뒤 프로덕션 trace 증가 0 | 1 | | 7 | 감사 도구 | span 기반 리포트의 스텝 이름이 이전과 맞는다 | 15 | | 8 | uid 가 셋에 다 있나 | 한 샷의 `shot_run_uid` 가 세 곳 같은 값 | 12·14 | | 9 | 참조가 자산으로 | 인물·소품 참조에 `asset_id` 없는 것 0건 | 10 | | 10 | 최종 자산 계보 | `cine_source_sel` 에 `stage`(roll\|fix)·`unresolved_reason` 이 실린다. ★**자산 id 로 잇지 않는다** — 롤은 `image_asset` 이 아니다(Task 11 이 범위를 좁혔다). 거짓 `asset_id` 0건도 함께 본다 | 11 | | 11 | 읽기 도구 | `shot_influence.py` 가 **`still_id` 로도 `uid` 로도** 낸다 | 14 | | 12 | 병렬 계층 | 병렬 롤이 도는 샷에서도 span 이 부모 밑에 있다 | 7-C | | 13 | 재개 uid | 완결 건너뛰기 샷의 `records.json` uid = Opik uid, `image_asset` 계보는 **안 덮임** | 12 | | 14 | 감사 셈 | 같은 v1 호출이 두 번 세어지지 않는다(총건수 대조) | 15 | | 15 | 샷 신원 | still-recipe trace 에 `still_id` 가 실린다(지금 529/529 가 없음) | 7-B |