"""litellm 을 우회하는 provider 직접 호출은 **반드시 Opik 기록을 달고 있어야 한다**. 2026-08-07 사용자 지시 — "모든 호출 opik 에 로그 남겨야해". litellm 경로는 `llm_client` 의 `litellm.callbacks = ["opik"]` 로 자동이다. 문제는 그것을 우회하는 직접 호출이었다: openai SDK 를 그대로 부르거나 (`images.generate/edit`, `responses.create`), **urllib 로 provider endpoint 를 직접 치거나**(`api.openai.com/v1/responses`), Gemini REST 를 직접 친다 (`generateContent`). 그 경로들은 기록이 통째로 없어서, 그 단계가 도는 동안 "작업이 살아 있나"를 볼 근거가 아무것도 없었다 — 실제로 그 때문에 살아 있는 작업을 죽었다고 판단해 상태를 잘못 내린 사고가 났다. 검사 단위는 **함수**다 (파일이 아니라). 한 파일에 기록 있는 함수 하나와 기록 없는 함수 하나가 같이 있으면 파일 단위 검사는 거짓 통과한다 — Codex 리뷰가 짚은 구멍이다. 함수를 감싼 `@traced_call` 데코레이터도 기록으로 인정한다. """ from __future__ import annotations import ast from pathlib import Path from typing import List, Set, Tuple APP = Path(__file__).resolve().parents[2] / "app" # provider 를 직접 치는 속성 호출 (owner.attr 쌍) DIRECT_ATTR_CALLS = { ("images", "generate"), ("images", "edit"), ("responses", "create"), ("responses", "parse"), ("responses", "stream"), ("completions", "create"), ("messages", "create"), # Anthropic SDK ("models", "generate_content"), # google-genai SDK } # provider endpoint 를 가리키는 URL 조각 — 이게 있는 파일의 urlopen 은 # provider 직접 호출로 본다. PROVIDER_HOSTS = ("api.openai.com", "generativelanguage.googleapis.com", "api.anthropic.com", "fal.run", "queue.fal.run", # qwen(dashscope) — 현재 프로덕션 사용처 0 이지만, 실험 # 코드가 프로덕션으로 이관될 때 미계측으로 새는 것을 막는다. "dashscope.aliyuncs.com", "dashscope-intl.aliyuncs.com") # 기록이 붙었다고 인정하는 표식 TRACE_MARKERS = ( "record_provider_call", # 공용 헬퍼 "traced_call", # 공용 데코레이터 "_tracer", # gemini_image_client 의 기존 방식 "get_image_tracer", "_trace_opik", # gemini_text_client 의 자기 헬퍼(안에서 공용을 부른다) ) # 기록이 필요 없는 자리 — 이유를 함께 적는다. EXEMPT_FILES: Set[str] = { # 키 broker: 원본 클라이언트에 위임만 하고 자기가 호출을 만들지 않는다. "core/openai_keys.py", } EXEMPT_FUNCS: Set[Tuple[str, str]] = { # 외부 참조 0 인 legacy (2026-08-07 확인: 모듈 밖에서 import 하는 곳 없음). # 살아나면 여기서 빼고 기록을 붙일 것. ("modules/pipeline/entity_extractor_v2_legacy.py", "_call_openai"), ("modules/pipeline/entity_extractor_v2_legacy.py", "_gpt_review_entity_list"), # fal queue 운반 원시층 (2026-08-25) — 자기는 기록하지 않는다. **한 번의 # 유료 작업이 submit·poll·result 여러 왕복으로 나뉘어** 있어서 여기서 # 남기면 한 작업이 여러 번 기록되고, poll 이 길어질수록 기록만 불어난다. # 기록은 사슬 끝이 정확히 한 번 남긴다 — 성공은 `_finish`, 실패는 # `_fail_log`(둘 다 `log_llm_call` + `get_image_tracer`). # ★면제는 「안 남긴다」는 뜻이 아니라 「여기가 남기는 자리가 아니다」는 # 뜻이다. 실제로 남는지는 자가신고가 아니라 실증으로 못박는다 — # tests/unit/test_cine_provider_swap.py 의 기록 시험 두 건이 # 성공·실패 양쪽에서 각각 1회 남는 것을 확인한다. ("modules/llm/reve_image_client.py", "_request"), } def _rel(path: Path) -> str: return str(path.relative_to(APP)) def _is_direct_provider_call(node: ast.AST, file_has_provider_host: bool) -> bool: if not isinstance(node, ast.Call): return False func = node.func if isinstance(func, ast.Attribute): owner = func.value if isinstance(owner, ast.Attribute) and (owner.attr, func.attr) in DIRECT_ATTR_CALLS: return True if isinstance(owner, ast.Name) and (owner.id, func.attr) in DIRECT_ATTR_CALLS: return True # 별칭: images = client.images 후 images.generate(...) # urllib.request.urlopen — provider host 상수가 같은 파일에 있을 때만 if func.attr == "urlopen" and file_has_provider_host: return True # litellm.completion 을 직접 (Router 우회) if func.attr in ("completion", "acompletion") and isinstance(owner, ast.Name) \ and owner.id == "litellm": return True return False def _own_nodes(fn: ast.AST): """그 함수가 **자기 몸으로** 가진 노드들 — 중첩 함수·lambda 본문은 뺀다. ★`ast.walk` 를 그냥 쓰면 안 된다 (Codex 리뷰 2026-08-07): 안쪽 함수의 기록 표식이 **바깥 함수를 면제해** 버린다. 바깥에 미계측 호출이 있어도 안쪽 어딘가에 표식만 있으면 통과하는 거짓 통과가 생긴다. 데코레이터는 자기 것으로 친다 — `@traced_call` 이 거기 붙는다. """ nested = (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda) stack = [n for n in (list(getattr(fn, "decorator_list", [])) + list(getattr(fn, "body", []))) if not isinstance(n, nested)] while stack: node = stack.pop() yield node for child in ast.iter_child_nodes(node): if isinstance(child, nested): continue # 중첩 함수는 그 자체로 따로 검사된다 stack.append(child) def _untraced_functions() -> List[str]: """provider 를 직접 치는데 기록 표식이 없는 **함수**들.""" bad: List[str] = [] for path in APP.rglob("*.py"): rel = _rel(path) if rel in EXEMPT_FILES: continue try: src = path.read_text(encoding="utf-8") tree = ast.parse(src) except (OSError, SyntaxError): continue has_host = any(h in src for h in PROVIDER_HOSTS) bad.extend(f"{rel}::{name}" for name in _untraced_in_tree( tree, has_host, exempt={n for (f, n) in EXEMPT_FUNCS if f == rel})) return bad def _untraced_in_tree(tree: ast.AST, has_host: bool, exempt: Set[str] | None = None) -> List[str]: """한 AST 에서 미계측 함수 이름들. 합성 표본 시험도 이것을 쓴다.""" skip = exempt or set() out: List[str] = [] for fn in ast.walk(tree): if not isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)): continue if fn.name in skip: continue own = list(_own_nodes(fn)) if not any(_is_direct_provider_call(n, has_host) for n in own): continue if not any(m in ast.dump(n) for n in own for m in TRACE_MARKERS): out.append(fn.name) return out def test_direct_provider_calls_carry_opik_tracing(): """provider 를 직접 치는 함수는 기록을 함께 가져야 한다.""" missing = _untraced_functions() assert not missing, ( "litellm 을 우회하는 provider 직접 호출인데 Opik 기록이 없다 — " f"{missing}. `record_provider_call` 을 부르거나 `@traced_call` 로 " "감쌀 것. 호출을 만들지 않는 자리면 EXEMPT 에 이유와 함께 넣을 것." ) # ── 탐지기 자체를 거는 표본 (Codex 리뷰 권고 2026-08-07) ────────────── # # 위 전수 시험은 **탐지기가 조용히 죽으면 언제나 통과한다** — 아무것도 못 # 찾으면 미계측도 0 이기 때문이다. 그래서 문법마다 작은 소스를 직접 넣어 # "이건 반드시 잡힌다 / 이건 잡히면 안 된다"를 고정한다. production 함수명에 # 묶지 않으므로 rename 에 깨지지 않으면서도 탐지 규칙의 죽음은 잡는다. _POSITIVE = { "images.generate": "def f():\n client.images.generate(prompt='x')\n", "images.edit": "def f():\n client.images.edit(image=i, prompt='x')\n", "별칭 images": "def f():\n images = client.images\n images.generate(prompt='x')\n", "responses.create": "def f():\n client.responses.create(model='m')\n", "responses.parse": "def f():\n client.responses.parse(model='m')\n", "responses.stream": "def f():\n client.responses.stream(model='m')\n", "messages.create": "def f():\n client.messages.create(model='m')\n", "models.generate_content": "def f():\n client.models.generate_content(model='m')\n", "litellm.completion": "def f():\n litellm.completion(model='m')\n", "litellm.acompletion": "def f():\n litellm.acompletion(model='m')\n", "중첩 함수의 표식은 바깥을 면제하지 않는다": ( "def outer():\n" " def inner():\n" " record_provider_call(step='s')\n" " client.responses.create(model='m')\n" ), } _NEGATIVE = { "표식 있는 함수": ( "def f():\n" " client.responses.create(model='m')\n" " record_provider_call(step='s')\n" ), "@traced_call 로 감싼 함수": ( "@traced_call(operation='x')\n" "def f():\n" " client.responses.create(model='m')\n" ), "provider 아닌 호출": "def f():\n db.session.create(x=1)\n", } def test_scanner_catches_each_call_syntax(): """탐지기가 살아 있는지 — 문법마다 반드시 잡히는지 고정한다.""" for label, src in _POSITIVE.items(): found = _untraced_in_tree(ast.parse(src), has_host=False) assert found, f"탐지 규칙이 죽었다 — 「{label}」 을 못 잡는다" def test_scanner_does_not_cry_wolf(): """기록이 있거나 provider 가 아니면 잡지 않는다.""" for label, src in _NEGATIVE.items(): found = _untraced_in_tree(ast.parse(src), has_host=False) assert not found, f"거짓 양성 — 「{label}」 을 잡았다: {found}" def test_scanner_urlopen_needs_provider_host(): """urlopen 은 그 파일에 provider 주소가 있을 때만 직접 호출로 본다.""" src = "def f():\n urllib.request.urlopen(req, timeout=3)\n" assert _untraced_in_tree(ast.parse(src), has_host=True), \ "provider 주소가 있는 파일의 urlopen 을 못 잡는다" assert not _untraced_in_tree(ast.parse(src), has_host=False), \ "무관한 urlopen 을 잡았다 — 거짓 양성" def test_scanner_still_sees_a_real_call_site(): """통합 카나리 하나 — 실제 코드에서도 탐지가 도는지 본다. 문법 표본이 주 방어선이고, 이것은 "실제 트리에서도 같은 규칙이 도는가"만 확인한다. 그래서 자리 하나로 족하다(함수명 결속을 최소로). """ path = APP / "modules" / "llm" / "gpt_image_primitive.py" src = path.read_text(encoding="utf-8") tree = ast.parse(src) has_host = any(h in src for h in PROVIDER_HOSTS) found_any = any( _is_direct_provider_call(n, has_host) for fn in ast.walk(tree) if isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)) for n in _own_nodes(fn)) assert found_any, "실제 파일에서 provider 직접 호출을 하나도 못 찾는다"