"""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 을 읽기 시작했다" # ── ★예약 키를 우리 값으로 덮지 않는다 (2026-08-24 E2E 실측) ────────────── # # `project_name` 은 litellm 이 **Opik 프로젝트 이름**으로 읽는 예약 키다 # (`litellm/integrations/opik/opik.py:47-50`). 거기에 파이프라인 프로젝트 # 이름을 실었더니 프로젝트마다 Opik 프로젝트가 새로 생기고 호출이 통째로 # 그리로 빠졌다 — 스텝 trace 는 `settings.opik_project_name` 쪽에 남아 # 부모와 자식이 갈라졌고 계층이 영영 안 섰다. # # ★이전 시험은 `m["project_name"] == "마지막 임무"` 를 **통과**시켰다. # 키가 있는지만 봤고 그 키가 litellm 에서 무슨 뜻인지는 안 봤다. # 그래서 여기서는 **나가는 payload** 를 잰다. def test_v2_does_not_hijack_litellm_project_name(monkeypatch): """StepRunner→router 전 경로를 태우고 나가는 payload 를 본다.""" from app.core import config from app.core.step_runner import StepRunner monkeypatch.setattr(config.settings, "opik_trace_v2_enabled", True) r = StepRunner.__new__(StepRunner) r.project_id, r.episode_id, r.step_id = "p1", "e1", "scene_detail" r.opik_context = {"project_name": "마지막 임무", "episode_title": "1회", "run_tag": "RT-1"} set_opik_context(r.build_opik_metadata()) md = _build_opik_metadata("scene_detail")["opik"] assert "project_name" not in md, ( "litellm 예약 키를 덮었다 — Opik 프로젝트가 갈라진다") # 이름 자체는 잃지 않는다(태그가 아니라 metadata 로 보존한다는 원래 의도). assert md["pipeline_project_name"] == "마지막 임무" def test_v2_only_fills_the_three_honored_keys(monkeypatch): """litellm 예약 키 넷 중 우리가 채우는 것은 셋뿐이다.""" 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="step:scene_detail") token = opik_trace.bind_trace(handle) try: set_opik_context({"thread_id": "ep-abc", "tags": ["step:scene_detail"], "pipeline_project_name": "마지막 임무"}) md = _build_opik_metadata("scene_detail")["opik"] finally: opik_trace.reset_trace(token) reserved = {"project_name", "current_span_data", "tags", "thread_id"} filled = reserved & set(md) assert filled == {"current_span_data", "tags", "thread_id"}, filled 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 # ── litellm span 의 태그도 축으로 갈려야 한다 (Codex BLOCK 2, 2026-08-24) ── # # 설계 ⑤ 가 못박았다: litellm span 의 **이름은 우리가 못 정한다** # (`{model}_{obj_type}_{created}`). 그래서 「무엇인지는 태그(op:·kind:)와 # metadata 로 가른다」. 그런데 태그가 맨 이름이면 그 유일한 식별 근거가 # 안 걸린다. 설계 ⑥ 도 「맨 이름 — 접두사를 붙여 축을 밝힌다」고 적었다. def test_v2_call_step_becomes_op_axis(monkeypatch): from app.core import config monkeypatch.setattr(config.settings, "opik_trace_v2_enabled", True) md = _build_opik_metadata("entity_t2i")["opik"] assert md["tags"] == ["op:entity_t2i"] def test_v2_keeps_already_axed_tags_from_step_runner(monkeypatch): """StepRunner 가 실은 축 태그는 그대로 살린다.""" from app.core import config monkeypatch.setattr(config.settings, "opik_trace_v2_enabled", True) set_opik_context({"tags": ["step:scene_image_pipeline", "status:retry"]}) md = _build_opik_metadata("still_recipe_judge")["opik"] assert md["tags"] == ["op:still_recipe_judge", "step:scene_image_pipeline", "status:retry"] def test_v2_drops_bare_names(monkeypatch): """★맨 이름은 버린다 — entity_extractor_v3.py:446 이 엔티티 이름을 싣는다. 한글 고유명사가 태그로 나가면 축도 안 갈리고 카디널리티도 터진다. """ from app.core import config monkeypatch.setattr(config.settings, "opik_trace_v2_enabled", True) md = _build_opik_metadata( "entity_t2i", {"tags": ["entity_t2i", "김보라의 낡은 가방"]})["opik"] assert md["tags"] == ["op:entity_t2i"] assert all(":" in t for t in md["tags"]) def test_v1_tags_are_untouched(monkeypatch): """설정 OFF 는 맨 이름 그대로 — 바이트 동일.""" from app.core import config monkeypatch.setattr(config.settings, "opik_trace_v2_enabled", False) md = _build_opik_metadata( "entity_t2i", {"tags": ["entity_t2i", "김보라의 낡은 가방"]})["opik"] assert md["tags"] == ["entity_t2i", "entity_t2i", "김보라의 낡은 가방"] def test_axis_whitelist_not_just_a_colon(monkeypatch): """★':' 이 있다고 축 태그가 아니다 — URL·동적 이름이 둔갑한다. 2026-08-24 Codex 재리뷰. 허용 축(step/op/kind/model/provider/status) 접두사만 살린다. """ from app.core import config monkeypatch.setattr(config.settings, "opik_trace_v2_enabled", True) md = _build_opik_metadata( "scene_detail", {"tags": ["http://192.168.0.9:5173/x", "step:a", "붉은 벽:2층"]})["opik"] assert md["tags"] == ["op:scene_detail", "step:a"] # ── ★router 로 나가는 **최종** payload 를 잰다 (Codex 재리뷰 2026-08-24) ── # # `_build_opik_metadata` 만 재면 **최초 호출**만 본다. 안전 sanitize(Tier 2)와 # GPT fallback(Tier 3)은 그 뒤 `_add_fallback_tag` 로 태그를 하나 더 붙이는데, # 거기서 접두사 없이 붙으면 앞에서 판 축이 다시 섞인다. def _drive_three_tiers(monkeypatch, *, opik_metadata=None): """Tier 1→2→3 을 전부 태우고 router 로 나간 태그를 순서대로 돌려준다.""" from app.modules.llm import llm_client seen = [] def _fake_completion(binding, model, kwargs): seen.append(list(kwargs["metadata"]["opik"]["tags"])) raise RuntimeError("safety filter blocked the prompt") monkeypatch.setattr(llm_client, "_get_router_binding", lambda: object()) monkeypatch.setattr(llm_client, "_completion", _fake_completion) with pytest.raises(Exception): llm_client.call_structured( step="scene_detail", system_prompt="s", user_prompt="u", response_schema={"type": "object"}, schema_name="t", opik_metadata=opik_metadata, ) return seen def test_v2_fallback_tags_carry_the_status_axis(monkeypatch): from app.core import config monkeypatch.setattr(config.settings, "opik_trace_v2_enabled", True) tiers = _drive_three_tiers(monkeypatch) assert tiers == [ ["op:scene_detail"], ["op:scene_detail", "status:sanitized"], ["op:scene_detail", "status:gpt_fallback"], ] for tags in tiers: assert all(":" in t for t in tags), tags def test_v1_fallback_tags_are_untouched(monkeypatch): """설정 OFF 면 sanitized / gpt_fallback 바이트 그대로.""" from app.core import config monkeypatch.setattr(config.settings, "opik_trace_v2_enabled", False) tiers = _drive_three_tiers(monkeypatch) assert tiers == [ ["scene_detail"], ["scene_detail", "sanitized"], ["scene_detail", "gpt_fallback"], ]