"""발송 장부 — **호출 경계**에서 시도를 센다 (2026-09-20 Codex (y)).

## 왜 필요한가

지금까지 「돈이 나갔나」의 근거는 두 가지뿐이었다.
  · record 묶음이 움직였나 → 그것은 **기록 변경**이지 발송이 아니다
  · 기존 로그를 사후에 세기 → ★**「로그 행 없음 = 전송 0」이 아니다**
    (기록 실패를 삼키는 자리가 있고, 재시도 `continue` 가 실패 로그보다
    **앞**이다)

그래서 **보내기 직전에** 적는다.

## 이 시험이 잠그는 것

    · `generate_image` **부른 횟수**가 아니라 **보낸 횟수**를 센다
      (재시도 루프가 함수 **안**에 있다)
    · **결과 불명**을 실패와 가른다 — 돈이 나갔을 수 있다
    · **관측 단위를 스스로 밝힌다** — 합산해서 「전송 총수」라 하지 않는다
    · 팬아웃(다른 스레드)의 발송도 잡힌다
    · 장부는 **제동이 아니다** — 없으면 아무 일도 안 한다
"""
from __future__ import annotations

from concurrent.futures import ThreadPoolExecutor
from typing import Any, List

import pytest

from app.core.send_ledger import (
    GRAIN_RAW,
    GRAIN_SLOT,
    STATE_FAILED,
    STATE_OK,
    SendLedger,
    bind_current_ledger,
    install_ledger,
    record_send,
)


@pytest.fixture(autouse=True)
def _clean():
    prev = install_ledger(None)
    yield
    install_ledger(prev)


def test_no_ledger_means_no_op():
    """★장부가 없으면 **아무 일도 안 한다** — 관측이 본 작업을 안 막는다."""
    assert record_send(kind="image", granularity=GRAIN_RAW, source="t") is None


def test_each_send_is_counted_not_each_call():
    """★★**보낸 횟수**를 센다 — 재시도 루프가 함수 **안**에 있다.

    함수 입구에서 세면 429 로 세 번 보내도 1 이 된다.
    """
    led = SendLedger()
    install_ledger(led)

    def _one_logical_call_with_retries():
        for _ in range(3):                    # 429 → 재시도 사다리
            record_send(kind="image", granularity=GRAIN_RAW,
                        source="grok.urlopen", model="grok-x")

    _one_logical_call_with_retries()
    assert led.counts()["image/request_attempt"]["sent"] == 3, (
        "부른 횟수(1)를 보낸 횟수로 셌다")


def test_unknown_is_not_failure():
    """★★**결과 불명**은 실패와 다르다 — 돈이 나갔을 수 있다."""
    led = SendLedger()
    install_ledger(led)
    a = record_send(kind="image", granularity=GRAIN_RAW, source="t")
    a.failed("http_400")
    b = record_send(kind="image", granularity=GRAIN_RAW, source="t")
    b.unknown("http_502")

    c = led.counts()["image/request_attempt"]
    assert c["sent"] == 2 and c["failed"] == 1 and c["unknown"] == 1
    assert "결과불명 1" in led.summary()


def test_grains_are_never_summed():
    """★★**관측 단위를 합산하지 않는다**.

    `slot_entry` 는 그 **아래에** 자체 재시도가 있어 물리 전송이 아니다.
    두 단위를 더해 「전송 총수」라고 부르면 그 자리에서 틀린다.
    """
    led = SendLedger()
    install_ledger(led)
    record_send(kind="image", granularity=GRAIN_RAW, source="a")
    record_send(kind="llm", granularity=GRAIN_SLOT, source="b")

    keys = set(led.counts())
    assert keys == {"image/request_attempt", "llm/slot_entry"}
    assert "total" not in led.summary()


def test_a_fanned_out_send_is_still_caught():
    """★롤 생성은 팬아웃이다 — 안 실으면 그 안의 발송이 통째로 안 잡힌다."""
    led = SendLedger()
    install_ledger(led)

    def _worker():
        record_send(kind="image", granularity=GRAIN_RAW, source="roll")

    with ThreadPoolExecutor(max_workers=2) as pool:
        for _ in range(2):
            pool.submit(bind_current_ledger(_worker))
    assert led.counts()["image/request_attempt"]["sent"] == 2


def test_an_unbound_thread_does_not_silently_count_elsewhere():
    """★싣지 않은 스레드는 **안 잡힌다** — 그 사실이 드러나야 한다.

    조용히 부모 장부에 섞이면 「어느 방문의 지출인가」가 뒤섞인다.
    """
    led = SendLedger()
    install_ledger(led)

    def _worker():
        record_send(kind="image", granularity=GRAIN_RAW, source="roll")

    with ThreadPoolExecutor(max_workers=1) as pool:
        pool.submit(_worker).result()       # ★안 싣고 보낸다
    assert led.counts() == {}, "싣지 않은 스레드가 부모 장부에 섞였다"


def test_the_attempt_carries_what_it_is_for():
    """시도마다 **무엇을 위한 발송인가**를 이어 둔다(attempt id 로 감사)."""
    led = SendLedger()
    install_ledger(led)
    a = record_send(kind="image", granularity=GRAIN_RAW, source="s",
                    model="grok-x", work="S1sh1", visit="U1")
    assert a.attempt_id and len(a.attempt_id) == 32
    assert (a.kind, a.model, a.work, a.visit) == ("image", "grok-x",
                                                  "S1sh1", "U1")


# ── 실제 경계에 달렸는가 ───────────────────────────────────────────

def test_the_raw_boundary_is_inside_the_retry_loop():
    """★★**재시도 루프 안**이어야 한다 — 밖이면 「부른 횟수」가 된다."""
    import inspect

    from app.modules.llm import grok_image_client as mod

    src = inspect.getsource(mod)
    i = src.find("for attempt in range(1, max_retries + 2):")
    j = src.find('source="grok_image_client.urlopen"', i)
    k = src.find("urllib.request.urlopen(", j)
    assert 0 < i < j < k, "발송 표가 루프 밖이거나 요청 뒤에 있다"
    assert "_record_send(" in src[j - 400:k], "그 자리에 장부가 없다"


def test_the_vlm_boundary_declares_it_is_not_physical():
    """★Router 진입은 **그 아래에 재시도가 있다** — 물리 전송이 아니다."""
    import inspect

    from app.modules.llm import llm_client as mod

    src = inspect.getsource(mod)
    i = src.find("_record_send(")
    assert i > 0, "VLM 경계에 장부가 없다"
    assert "_GRAIN_SLOT" in src[i:i + 400], (
        "요청 시도 단위로 적고 있다 — Router 재시도가 안 보이는 자리다")
    assert 'kind="llm"' in src[i:i + 400], (
        "★`vlm` 으로 적으면 「이미지 관찰 횟수」로 오독된다 — 이 경로는 "
        "**텍스트 호출도 지난다**")


def test_the_fanout_carries_the_ledger():
    """★롤 생성 팬아웃에 장부를 **싣는다** — 안 실으면 전송이 안 잡힌다."""
    import inspect

    from app.modules.pipeline import multiroll_select as mod

    src = inspect.getsource(mod.run_multiroll_select)
    i = src.find("bound = bind_current_budget(")
    assert i > 0
    seg = src[i:i + 300]
    assert "bind_current_ledger(" in seg, (
        "팬아웃에 장부를 안 실었다 — 그 안의 발송이 통째로 안 잡힌다")


def test_the_ledger_is_a_report_not_a_brake():
    """★★장부는 **보고값**이다 — 기존 제동에 끼어들지 않는다.

    못 재는 경로를 0 으로 치고 제동을 푸는 것이 이 판의 병이었다.
    장부 수가 상한·래치 판단에 들어가면 같은 병을 다시 낸다.
    """
    import inspect

    from app.services import still_recipe_service as svc

    src = inspect.getsource(svc)
    # 상한·래치를 정하는 이름들 곁에 장부가 없어야 한다
    for guard in ("jit_regen_count", "jit_regen_limit",
                  "_jit_write_latch", "reserve_current_call"):
        for line in src.splitlines():
            if guard in line and "send_ledger" in line:
                raise AssertionError(
                    f"장부가 제동 판단에 끼어들었다: {line.strip()}")
    # 그리고 **보고에는** 나와야 한다
    assert "still_recipe 발송 장부" in src


def test_the_ledger_is_not_merged_with_the_audit_log():
    """★`llm_call_log`·Opik 과 **합산하지 않는다** — 같은 발송을 두 번 센다."""
    import inspect

    from app.core import send_ledger as mod

    src = inspect.getsource(mod)
    assert "합산" in src, "중복 계상 금지를 적어 두지 않았다"
    # ★**설명글이 아니라 코드**를 본다 — 주석에 이름이 나온 것을 「읽는다」로
    #  읽으면 시험이 제 말에 걸린다(실제로 그랬다).
    import ast

    tree = ast.parse(src)
    names = {
        n.module for n in ast.walk(tree) if isinstance(n, ast.ImportFrom)
    } | {
        a.name for n in ast.walk(tree) if isinstance(n, ast.Import)
        for a in n.names
    }
    for banned in ("llm_call_log", "opik", "image_tracer"):
        assert not any(banned in str(m or "") for m in names), (
            f"장부가 감사 경로({banned})를 import 한다 — 합산의 문이 된다")
    calls = {
        n.func.id for n in ast.walk(tree)
        if isinstance(n, ast.Call) and isinstance(n.func, ast.Name)
    }
    assert "log_llm_call" not in calls


# ── 판정 팬아웃이 장부를 잃지 않는가 (Codex BLOCK 1) ───────────────

def test_the_cross_model_judge_fanout_keeps_the_ledger():
    """★★**선정·재판정의 실제 자리**가 장부를 잃으면 안 된다.

    롤 팬아웃에만 실으면 **별도 판정 pool 로는 전파되지 않는다** —
    두 슬롯의 `_completion` 이 통째로 안 잡혀 「관측 0」이 된다.
    ★★**이 시험이 잠근 범위**(Codex NON-BLOCK): 실제 스레드는 돌리지만
     `_judge_cross_model_order` 도 `_completion` 도 **부르지 않는다**.
     프로덕션과 **같은 감싸기 순서**를 복제해 `record_send` 를 직접 부른다.
     즉 **「감싸기 조합이 전달되는가」**까지다 — 「프로덕션 교차 판정 →
     `_completion` → router 대역」의 **끝점 시험은 아직 아니다**.
     배선 자체는 아래 소스 검사가 따로 본다.
    """
    from app.core.image_call_budget import bind_current_budget
    from app.core.send_ledger import bind_current_ledger
    from app.modules.llm.opik_trace import bind_current_trace
    from app.services.image_capture.context import (
        bind_current_generation_context,
    )

    led = SendLedger()
    install_ledger(led)

    def _slot(model: str) -> str:
        record_send(kind="llm", granularity=GRAIN_SLOT,
                    source="llm_client._completion", model=model)
        return model

    # 프로덕션이 쓰는 **그 감싸기 순서** 그대로
    bound = bind_current_budget(
        bind_current_ledger(
            bind_current_generation_context(bind_current_trace(_slot))))
    with ThreadPoolExecutor(max_workers=2) as pool:
        futs = [pool.submit(bound, m) for m in ("gemini", "gpt")]
        [f.result() for f in futs]

    assert led.counts()["llm/slot_entry"]["sent"] == 2, (
        "판정 pool 의 두 슬롯이 부모 장부에 안 남았다")


@pytest.mark.parametrize("fn_name,pool_hint", [
    ("_judge_cross_model_order", "max_workers=2"),
    ("_judge_gq", "max_workers=len(models)"),
    # ★관찰 pool 도 **같은 걷기에서 켤 수 있다** — 목록에서 빠져 있었다
    #  (2026-09-20 Codex NON-BLOCK).
    ("make_gg46_critique_fn", "observer_models"),
])
def test_every_judge_pool_carries_the_ledger(fn_name, pool_hint):
    """★**켤 수 있는 판정 pool 마다** 실려 있어야 한다."""
    import inspect

    from app.modules.pipeline import multiroll_gemini as mod

    src = inspect.getsource(getattr(mod, fn_name))
    i = src.find("bind_current_budget(")
    assert i > 0, f"{fn_name} 에 팬아웃 감싸기가 없다"
    assert "bind_current_ledger(" in src[i:i + 300], (
        f"{fn_name} pool 이 장부를 안 나른다 — 그 안의 판정이 「관측 0」")


def test_the_dual_vlm_pool_carries_the_ledger():
    import inspect

    from app.modules.llm import dual_vlm as mod

    src = inspect.getsource(mod.ask_both)
    i = src.find("bind_current_budget(")
    assert i > 0
    assert "bind_current_ledger(" in src[i:i + 300]


# ── 중도 종료 (Codex BLOCK 2) ──────────────────────────────────────

def test_an_aborted_walk_restores_and_still_reports():
    """★★**중단돼도** 원래 장부로 돌아가고 **관측한 수를 보고한다**.

    ★범위: `ledger_scope` **helper** 시험이다 — 실제 걷기의 래치 경로를
     태운 것이 아니다(Codex NON-BLOCK). 걷기에 씌워졌는지는 아래 소스
     검사가 따로 본다.

    정상 흐름에서만 복원·보고하면 ①앞 샷들의 발송 보고가 사라지고
    ②끝난 걷기의 장부가 **스레드에 남아** 다음 작업이 거기 들어간다.
    """
    from app.core.send_ledger import get_current_ledger, ledger_scope

    outer = SendLedger()
    install_ledger(outer)
    seen: List[Any] = []

    with pytest.raises(RuntimeError, match="상한"):
        with ledger_scope(SendLedger(), report=seen.append):
            record_send(kind="image", granularity=GRAIN_RAW, source="s")
            raise RuntimeError("걷기 도중 상한")

    assert get_current_ledger() is outer, "중단 뒤 옛 장부로 안 돌아왔다"
    assert seen and seen[0].counts()["image/request_attempt"]["sent"] == 1, (
        "중단까지 관측한 수를 보고하지 않았다")
    assert outer.attempts == [], "중단된 걷기의 발송이 바깥 장부에 섞였다"


def test_a_failing_report_does_not_mask_the_original_error():
    """★보고 실패가 **원래 중단 예외를 덮지 않는다**."""
    from app.core.send_ledger import ledger_scope

    def _boom(_led):
        raise ValueError("보고가 터졌다")

    with pytest.raises(RuntimeError, match="진짜 원인"):
        with ledger_scope(SendLedger(), report=_boom):
            record_send(kind="image", granularity=GRAIN_RAW, source="s")
            raise RuntimeError("진짜 원인")


def test_the_walk_installs_the_ledger_outside_its_body():
    """★설치가 **본문 바깥**이어야 예외 경로까지 덮인다."""
    import inspect

    from app.services import still_recipe_service as svc

    # ★`functools.wraps` 라 `getsource` 는 **본문**을 돌려준다 — 그래서
    #  껍질은 데코레이터 쪽에서 본다(이름을 바꾸면 본문을 읽는 기존 시험
    #  여덟이 엉뚱한 소스를 본다).
    assert "@_with_send_ledger" in inspect.getsource(svc), (
        "걷기에 장부 범위를 안 씌운다")
    deco = inspect.getsource(svc._with_send_ledger)
    assert "ledger_scope(" in deco
    assert "functools.wraps" in deco, (
        "본문 소스를 읽는 기존 시험들이 껍질을 보게 된다")


# ── 발송 맥락 — **소유자가 정한다** (Codex (δ)) ────────────────────

def test_the_owner_sets_the_context_not_the_boundary():
    """★★관측이 **스스로 캐지 않는다** — 소유자가 정해 넘긴 것을 쓴다.

    `record_send` 가 로그·DB·문안을 뒤져 「이게 어느 샷이지」를 추측하면
    **틀려도 아무도 모른다**.
    """
    from app.core.send_ledger import send_context

    led = SendLedger()
    install_ledger(led)
    with send_context(work="S1sh1", visit="U1"):
        record_send(kind="image", granularity=GRAIN_RAW, source="s")
    a = led.attempts[0]
    assert (a.work, a.visit) == ("S1sh1", "U1")


def test_an_unknown_visit_stays_unknown():
    """★★모르면 **미상으로 남긴다** — 숫자나 `still_id` 로 안 메운다.

    trace 가 꺼지면 방문 신원이 없다. 그때 아무 값이나 넣으면 「누가
    만들었나」가 **거짓으로** 남는다.
    """
    from app.core.send_ledger import send_context

    led = SendLedger()
    install_ledger(led)
    with send_context(work="S1sh1", visit=""):
        record_send(kind="image", granularity=GRAIN_RAW, source="s")
    assert led.attempts[0].visit == "", "미상을 무언가로 메웠다"


def test_a_shared_job_overrides_only_the_work():
    """★★공유 배경의 발송은 **그 샷 것이 아니다**.

    `work` 만 그룹 키로 덮고 **방문 신원은 잇는다**. 그리고 나온 뒤에는
    **되돌아온다** — 안 그러면 그 뒤 발송이 전부 그룹 것으로 적힌다.
    """
    from app.core.send_ledger import send_context

    led = SendLedger()
    install_ledger(led)
    with send_context(work="S1sh1", visit="U1"):
        record_send(kind="image", granularity=GRAIN_RAW, source="본체")
        with send_context(work="groupbg::어떤장소", visit=None):
            record_send(kind="image", granularity=GRAIN_RAW, source="공유")
        record_send(kind="image", granularity=GRAIN_RAW, source="본체2")

    works = [(a.source, a.work, a.visit) for a in led.attempts]
    assert works == [
        ("본체", "S1sh1", "U1"),
        ("공유", "groupbg::어떤장소", "U1"),   # ★방문은 이어진다
        ("본체2", "S1sh1", "U1"),              # ★되돌아왔다
    ], works


def test_an_explicit_argument_wins_over_the_context():
    """호출부가 **명시로** 준 값이 있으면 그것이 우선이다."""
    from app.core.send_ledger import send_context

    led = SendLedger()
    install_ledger(led)
    with send_context(work="S1sh1", visit="U1"):
        record_send(kind="image", granularity=GRAIN_RAW, source="s",
                    work="S9sh9")
    assert led.attempts[0].work == "S9sh9"


def test_the_fanout_carries_the_context_too():
    """★장부만 싣고 맥락을 빠뜨리면 팬아웃 발송이 **「어느 샷인지 모름」**."""
    from app.core.send_ledger import send_context

    led = SendLedger()
    install_ledger(led)

    def _worker():
        record_send(kind="image", granularity=GRAIN_RAW, source="roll")

    with send_context(work="S1sh1", visit="U1"):
        with ThreadPoolExecutor(max_workers=1) as pool:
            pool.submit(bind_current_ledger(_worker)).result()
    assert (led.attempts[0].work, led.attempts[0].visit) == ("S1sh1", "U1")


def test_the_boundary_never_digs_for_context():
    """★★경계가 **스스로 캐는 코드**를 두지 않는다.

    `record_send` 안에서 Opik·DB·문안을 뒤지면 그 추측이 틀려도 모른다.
    """
    import ast
    import inspect

    from app.core import send_ledger as mod

    tree = ast.parse(inspect.getsource(mod))
    names = {n.module for n in ast.walk(tree)
             if isinstance(n, ast.ImportFrom)}
    for banned in ("opik", "models", "db", "session"):
        assert not any(banned in str(m or "") for m in names), (
            f"장부가 {banned} 를 뒤진다 — 맥락은 소유자가 준다")


def test_the_walk_installs_the_context_at_the_shot_scope():
    """★소유자(샷 루프)가 **canonical tag** 로 설치한다."""
    import inspect

    from app.services import still_recipe_service as svc

    src = inspect.getsource(svc.run_still_recipe_generation)
    i = src.find("_send_context(work=tag")
    assert i > 0, "샷 루프가 맥락을 안 씌운다"
    # ★`still_id` 가 아니라 **tag** 여야 본체/자식/공유가 갈린다
    assert "_send_context(work=still_id" not in src
    # 공유 배경은 work 만 덮는다
    assert 'work=f"groupbg::{group_key}", visit=None' in src, (
        "공유 배경 발송이 마침 처음 닿은 샷의 지출로 적힌다")


# ── ② OpenAI 슬롯 경계 (2026-09-20) ────────────────────────────────

def _slot_client(targets):
    """`FailoverOpenAIClient` 에 **가짜 슬롯**을 물린다.

    `targets` = [(슬롯 이름, 그 슬롯에서 부를 함수), …] — 순서대로 쓴다.
    """
    from app.core import openai_keys as ok

    seq = list(targets)
    used: list = []

    class _Leaf:
        def __init__(self, fn):
            self._fn = fn

        def __call__(self, *a, **kw):
            return self._fn(*a, **kw)

    class _Ns:
        def __init__(self, fn):
            self._fn = fn

        def __getattr__(self, name):
            return _Leaf(self._fn)

    class _Root:
        def __init__(self, fn):
            self._fn = fn

        def __getattr__(self, name):
            return _Ns(self._fn)

    cli = ok.FailoverOpenAIClient()

    def _raw():
        slot, fn = seq[min(len(used), len(seq) - 1)]
        used.append(slot)
        return slot, _Root(fn)

    cli._raw = _raw                                    # type: ignore
    return cli, used


def test_the_openai_slot_boundary_records_each_entry(monkeypatch):
    """★★키 슬롯을 옮기면 **두 번 나간다** — 두 번 적혀야 한다."""
    from app.core import openai_keys as ok

    calls: list = []

    def _boom(*a, **kw):
        calls.append("primary")
        raise RuntimeError("키가 죽었다")

    def _fine(*a, **kw):
        calls.append("secondary")
        return "OK"

    cli, used = _slot_client([("p", _boom), ("s", _fine)])
    monkeypatch.setattr(ok, "slot_count", lambda: 2)
    monkeypatch.setattr(ok, "failover_on", lambda *a, **k: True)
    monkeypatch.setattr(
        "app.core.research_call_budget.reserve_current_research_call",
        lambda **k: None)

    led = SendLedger()
    install_ledger(led)
    out = cli.images.edit(model="gpt-image-2.5", prompt="x")

    assert out == "OK" and calls == ["primary", "secondary"]
    assert len(led.attempts) == 2, (
        f"슬롯 진입을 다 안 적었다: {led.summary()}")
    assert [a.state for a in led.attempts] == [STATE_FAILED, STATE_OK]
    assert all(a.kind == "image" for a in led.attempts), (
        "`images.*` 인데 이미지로 안 적혔다")
    assert all(a.granularity == GRAIN_SLOT for a in led.attempts), (
        "SDK 안 재시도는 안 보인다 — `slot_entry` 로 적어야 한다")
    assert all(a.model == "gpt-image-2.5" for a in led.attempts)


def test_the_openai_slot_boundary_tells_image_from_text(monkeypatch):
    """★종류는 **호출 경로**로 가른다 — 문안을 훑어 짐작하지 않는다."""
    from app.core import openai_keys as ok

    cli, _ = _slot_client([("p", lambda *a, **kw: "OK")])
    monkeypatch.setattr(ok, "slot_count", lambda: 1)
    monkeypatch.setattr(
        "app.core.research_call_budget.reserve_current_research_call",
        lambda **k: None)

    led = SendLedger()
    install_ledger(led)
    cli.images.edit(model="gpt-image-2.5", prompt="x")
    cli.responses.create(model="gpt-6", input="x")

    assert [a.kind for a in led.attempts] == ["image", "llm"]


def test_the_openai_slot_boundary_does_not_dig_for_context(monkeypatch):
    """★★경계는 **소유자가 씌운 맥락**을 받아 쓴다 — 스스로 안 캔다."""
    from app.core import openai_keys as ok
    from app.core.send_ledger import send_context

    cli, _ = _slot_client([("p", lambda *a, **kw: "OK")])
    monkeypatch.setattr(ok, "slot_count", lambda: 1)
    monkeypatch.setattr(
        "app.core.research_call_budget.reserve_current_research_call",
        lambda **k: None)

    led = SendLedger()
    install_ledger(led)
    with send_context(work="S1sh1::cine", visit="U7"):
        cli.images.edit(model="m", prompt="x")

    a = led.attempts[0]
    assert (a.work, a.visit) == ("S1sh1::cine", "U7")