script-build-and-framework-tmp/visualization/backend/tests/test_source_lineage.py

from __future__ import annotations

from copy import deepcopy

from fastapi.testclient import TestClient
import pytest

from app import main
from app.card_details.common import context_input_fields
from app.module_audit.log_context import locate_log_module
from app.source_lineage.comparison import _normalize_business, _source_module_ids
from app.source_lineage.projector import _all_bindings
from app.source_lineage.resolver import SourceResolver


client = TestClient(main.app)


@pytest.fixture(autouse=True)
def clear_source_inspector_caches():
    main._inspector_cache.clear()
    main._build_view_cache.clear()
    yield
    main._inspector_cache.clear()
    main._build_view_cache.clear()


def _module(payload: dict, module_id: str) -> dict:
    return next(item for item in payload["modules"] if item["id"] == module_id)


def _row_bindings(module: dict, row_index: int = 0) -> list[dict]:
    ids = set(module["rows"][row_index]["bindingIds"])
    return [binding for binding in module["bindings"] if binding["id"] in ids]


def _event_detail(summary: dict, *, input_content=None, output_content=None):
    return {
        **summary,
        "input": {"contentType": "json", "content": input_content, "truncated": False},
        "output": {"contentType": "json", "content": output_content, "truncated": False},
    }


def test_inspector_view_planning_fields_have_exact_independent_bindings(
    monkeypatch, current_bundle
):
    bundle = deepcopy(current_bundle)
    event = {
        "id": 60,
        "event_seq": 1,
        "event_type": "tool_call",
        "event_name": "think_and_plan",
        "agent_role": "main",
        "agent_depth": 0,
        "status": "completed",
        "started_at": "2026-07-13T09:59:00",
        "ended_at": "2026-07-13T09:59:01",
    }
    bundle["events"].insert(0, event)
    detail = _event_detail(
        event,
        input_content={
            "thought_summary": "先确定承重结构",
            "thought": "完整思考原文",
            "plan": "1. 写入方向\n2. 开始本轮",
            "action": "保存创作目标",
        },
        output_content={"ok": True},
    )
    calls = []
    monkeypatch.setattr(main.provider, "load_bundle", lambda _build_id: deepcopy(bundle))
    monkeypatch.setattr(
        main.provider,
        "load_event_details",
        lambda _build_id, ids: calls.append(list(ids)) or {60: deepcopy(detail)},
    )
    monkeypatch.setattr(main.provider, "load_event_detail", lambda _build_id, _event_id: deepcopy(detail))
    main._inspector_cache.clear()

    response = client.get("/api/script-builds/88001/inspector-view/event:60")
    assert response.status_code == 200
    payload = response.json()
    assert payload["schemaVersion"] == "inspector-source-v2"
    assert payload["businessProjection"]["businessSections"][0]["content"] == "先确定承重结构"
    assert calls == [[60]]
    paths = {
        module["id"]: _row_bindings(module)[0]["selector"].get("path")
        for module in payload["modules"]
    }
    assert paths == {
        "summary": "/input/content/thought_summary",
        "thought": "/input/content/thought",
        "plan": "/input/content/plan",
        "action": "/input/content/action",
    }
    assert list(payload["sources"]) == ["event:60"]


def test_optional_objective_and_creative_blocks_use_stable_semantic_ids():
    objective = _normalize_business({
        "blocks": [
            {"type": "items", "title": "根据以下信息做出的决策", "items": [{"label": "账号", "value": "A"}]},
            {"type": "summary", "title": "当前保存的创作方向", "value": "方向原文"},
        ]
    }, card_kind="objective")
    assert [item["_semanticKey"] for item in objective] == [
        "block:inputs",
        "block:direction",
    ]
    assert _source_module_ids("objective", "block:inputs", 1) == ["inputs"]
    assert _source_module_ids("objective", "block:direction", 2) == ["direction"]

    creative = _normalize_business({
        "blocks": [
            {"type": "items", "title": "可确认的上游信息", "items": [{"label": "案例", "value": "A"}]},
            {"type": "creative-process", "title": "创作处理", "steps": [{"action": "改写"}]},
            {"type": "summary", "title": "产出的候选表", "value": "候选 A"},
            {"type": "notice", "title": "记录说明", "value": "过程只保存了一部分"},
        ]
    }, card_kind="creative")
    assert [item["_semanticKey"] for item in creative] == [
        "block:available",
        "block:process",
        "block:candidate",
        "block:record-note",
    ]
    assert [
        _source_module_ids("creative", item["_semanticKey"], index)
        for index, item in enumerate(creative, 1)
        ] == [["available"], ["process"], ["candidate"], ["record-note"]]


def test_context_input_fields_classifies_business_refs_as_database_records():
    context = {
        "inputs": [
            {"label": "多路决策", "summary": "采用方案 1", "detailRef": "multipath-decision:21"},
            {"label": "数据取舍", "summary": "采用两条证据", "detailRef": "data-decision:11"},
            {"label": "领域事实", "summary": "事实 A", "detailRef": "domain-info:31"},
            {"label": "开启本轮", "summary": "目标 A", "detailRef": "event:60", "relation": "direct-read"},
        ]
    }
    event = {"id": 60, "event_name": "begin_round"}
    fields = context_input_fields(context, "test", [event])
    assert [item["source"]["kind"] for item in fields] == [
        "database", "database", "database", "runtime-event"
    ]
    assert [item["source"]["label"] for item in fields[:3]] == [
        "script_build_multipath_decision",
        "script_build_data_decision",
        "script_build_domain_info",
    ]
    assert [item["source"]["fieldPath"] for item in fields] == [
        "decision", "decision", "content", "input.content.goal"
    ]


def test_implementation_task_sections_bind_to_exact_task_spans(
    monkeypatch, current_bundle
):
    bundle = deepcopy(current_bundle)
    task = """交付说明：输出一份完整候选表
任务目标：形成三段式文章结构
method: 先搭建段落骨架
目标：让读者理解因果关系
必用素材：柔性排产案例
前情提要：本轮已核实领域事实
重要约束：不改动事实数字
禁止事项：不得伪造引用"""
    event = next(item for item in bundle["events"] if item["id"] == 50)
    event["inputData"] = {"task": task}
    detail = _event_detail(event, input_content={"task": task}, output_content={"summary": "完成"})
    monkeypatch.setattr(main.provider, "load_bundle", lambda _build_id: deepcopy(bundle))
    monkeypatch.setattr(
        main.provider,
        "load_event_details",
        lambda _build_id, ids: {50: deepcopy(detail)} if 50 in ids else {},
    )
    main._inspector_cache.clear()

    response = client.get(
        "/api/script-builds/88001/inspector-view/round:1:branch:1:task"
    )
    assert response.status_code == 200
    payload = response.json()
    expected_modules = {
        "task-notes-1": "deliverable",
        "task-scope-2": "scope",
        "task-method-3": "method",
        "task-goal-4": "goal",
        "task-materials-5": "constraints",
        "task-context-6": "constraints",
        "task-constraints-7": "constraints",
        "task-avoid-8": "avoid",
    }
    for module_id, source_module_id in expected_modules.items():
        module = _module(payload, module_id)
        bindings = _row_bindings(module)
        assert bindings, module_id
        assert all(binding["resolution"] == "resolved" for binding in bindings)
        assert all(
            binding["selector"]["kind"] == "text-span"
            for binding in bindings
            if binding["sourceId"] == "event:50"
        )
        if source_module_id != "scope":
            event_binding = next(binding for binding in bindings if binding["sourceId"] == "event:50")
            assert event_binding["selector"]["kind"] == "text-span"
            assert event_binding["selector"]["path"] == "/input/content/task"
            section = next(
                item
                for item in payload["businessProjection"]["businessSections"]
                if item["id"] == module_id
            )
            assert event_binding["selector"]["exactText"] == section["content"]


def test_inspector_view_query_binds_each_condition_and_case(monkeypatch, current_bundle):
    bundle = deepcopy(current_bundle)
    event = next(item for item in bundle["events"] if item["id"] == 53)
    detail = _event_detail(
        event,
        input_content={"keyword": "因果结构", "top_k": 2},
        output_content={
            "count": 2,
            "results": [
                {"post_id": "A", "account": "账号A", "data": "Case A"},
                {"post_id": "B", "account": "账号B", "data": "Case B"},
            ],
        },
    )
    monkeypatch.setattr(main.provider, "load_bundle", lambda _build_id: deepcopy(bundle))
    monkeypatch.setattr(main.provider, "load_event_details", lambda _build_id, _ids: {53: deepcopy(detail)})
    monkeypatch.setattr(main.provider, "load_event_detail", lambda _build_id, _event_id: deepcopy(detail))
    main._inspector_cache.clear()

    payload = client.get("/api/script-builds/88001/inspector-view/event:53").json()
    purpose = _module(payload, "conditions")
    result = _module(payload, "results")
    status = _module(payload, "outcome")
    assert [_row_bindings(purpose, index)[0]["selector"]["path"] for index in range(2)] == [
        "/input/content/keyword",
        "/input/content/top_k",
    ]
    assert [_row_bindings(result, index)[0]["selector"]["path"] for index in range(2)] == [
        "/output/content/results/0",
        "/output/content/results/1",
    ]
    assert len({binding["id"] for binding in result["bindings"]}) == 2
    assert all(
        binding["selector"].get("path") == "/output/content"
        and binding["transform"]["operation"] == "retrieval-outcome-v1"
        for binding in _row_bindings(status)
    )
    assert result["bindings"][0]["evidence"] == {
        "availability": "returned-to-agent",
        "adoption": "not-recorded",
        "confidence": "deterministic",
    }
    assert result["bindings"][0]["transform"]["operation"] == "retrieval-result-item-v1"
    assert all(binding["resolution"] == "resolved" for binding in status["bindings"])
    assert not any(":error:" in binding["id"] for binding in status["bindings"])
    status_binding_ids = {binding["id"] for binding in status["bindings"]}
    assert all(
        set(row["bindingIds"]).issubset(status_binding_ids)
        for row in status["rows"]
    )


@pytest.mark.parametrize(
    ("event_name", "output_content", "expected_paths"),
    [
        (
            "get_domain_info",
            {"domain_info": [{"id": 1, "content": "事实 A"}, {"id": 2, "content": "事实 B"}]},
            [
                "/output/content/domain_info/0",
                "/output/content/domain_info/1",
            ],
        ),
        (
            "get_script_snapshot",
            {"script_build": {"status": "running"}, "paragraphs": []},
            ["/output/content"],
        ),
    ],
)
def test_direct_tool_result_paths_follow_the_shared_business_contract(
    monkeypatch, current_bundle, event_name, output_content, expected_paths
):
    bundle = deepcopy(current_bundle)
    event = {
        "id": 60,
        "event_seq": 60,
        "event_type": "tool_call",
        "event_name": event_name,
        "agent_role": "main",
        "agent_depth": 0,
        "status": "completed",
        "ended_at": "2026-07-13T10:00:01",
    }
    bundle["events"].append(event)
    detail = _event_detail(event, input_content={}, output_content=output_content)
    monkeypatch.setattr(main.provider, "load_bundle", lambda _build_id: deepcopy(bundle))
    monkeypatch.setattr(
        main.provider,
        "load_event_details",
        lambda _build_id, ids: {60: deepcopy(detail)} if 60 in ids else {},
    )
    main._inspector_cache.clear()

    payload = client.get("/api/script-builds/88001/inspector-view/event:60").json()
    results = _module(payload, "results")
    paths = [
        _row_bindings(results, index)[0]["selector"].get("path")
        for index in range(len(results["rows"]))
    ]
    assert paths == expected_paths


def test_generic_event_with_saved_body_never_falls_back_to_false_missing(
    monkeypatch, current_bundle
):
    bundle = deepcopy(current_bundle)
    event = {
        "id": 60,
        "event_seq": 60,
        "event_type": "tool_call",
        "event_name": "save_script_direction",
        "agent_role": "main",
        "agent_depth": 0,
        "status": "completed",
    }
    bundle["events"].insert(0, event)
    detail = _event_detail(
        event,
        input_content={"direction": "完整创作方向"},
        output_content="创作脚本方向已保存",
    )
    monkeypatch.setattr(main.provider, "load_bundle", lambda _build_id: deepcopy(bundle))
    monkeypatch.setattr(
        main.provider,
        "load_event_details",
        lambda _build_id, ids: {60: deepcopy(detail)} if 60 in ids else {},
    )
    main._inspector_cache.clear()

    payload = client.get("/api/script-builds/88001/inspector-view/event:60").json()
    assert payload["sources"]["event:60"]["resultState"] == "present"
    bindings = [
        binding
        for module in payload["modules"]
        for binding in module["bindings"]
    ]
    assert bindings
    assert all(binding["sourceId"] == "event:60" for binding in bindings)
    assert all(binding["resolution"] == "resolved" for binding in bindings)


def test_data_decision_and_final_result_bind_to_real_database_fields(
    monkeypatch, current_bundle
):
    bundle = deepcopy(current_bundle)
    bundle["record"].update({"status": "success", "summary": "构建摘要"})
    monkeypatch.setattr(main.provider, "load_bundle", lambda _build_id: deepcopy(bundle))
    monkeypatch.setattr(main.provider, "load_event_details", lambda _build_id, _ids: {})
    main._inspector_cache.clear()

    decision = client.get(
        "/api/script-builds/88001/inspector-view/data-decision:11"
    ).json()
    source_module = _module(decision, "block-1")
    for index, row in enumerate(source_module["rows"]):
        row_bindings = _row_bindings(source_module, index)
        assert len(row_bindings) == 1
        assert row_bindings[0]["selector"]["path"] == f"/sources/{index}"
        assert row_bindings[0]["evidence"]["adoption"] == "explicitly-adopted"
        assert decision["sources"][row_bindings[0]["sourceId"]]["locator"]["table"] == "script_build_data_decision"
    assert len(_row_bindings(_module(decision, "block-2"))) == 1

    final = client.get(
        "/api/script-builds/88001/inspector-view/run:final-result"
    ).json()
    conclusion = _module(final, "conclusion")
    summary = _module(final, "summary")
    assert final["businessProjection"]["businessSections"][0]["content"] == "构建已完成。"
    assert _row_bindings(conclusion)[0]["selector"]["path"] == "/status"
    assert _row_bindings(conclusion)[0]["transform"]["kind"] == "parsed"
    assert _row_bindings(summary)[0]["selector"]["path"] == "/summary"
    stats = _module(final, "statistics")
    assert _row_bindings(stats)[0]["selector"]["kind"] == "members"
    assert _row_bindings(stats)[0]["selector"]["members"]
    assert _row_bindings(stats)[0]["selector"]["reducer"] == "Round / Branch 记录计数"


def test_every_visible_module_has_a_resolved_or_explicit_unresolved_binding(
    monkeypatch, current_bundle
):
    bundle = deepcopy(current_bundle)
    monkeypatch.setattr(main.provider, "load_bundle", lambda _build_id: deepcopy(bundle))
    monkeypatch.setattr(main.provider, "load_event_details", lambda _build_id, _ids: {})
    main._inspector_cache.clear()

    payload = client.get(
        "/api/script-builds/88001/inspector-view/round:1:goal"
    ).json()
    assert payload["modules"]
    for module in payload["modules"]:
        assert module["rows"]
        for row in module["rows"]:
            assert row["businessSelector"]
            bindings = _row_bindings(module, module["rows"].index(row))
            assert bindings
            for binding in bindings:
                assert binding["resolution"] in {"resolved", "partial", "missing", "unsafe"}
                if binding["resolution"] == "missing":
                    assert binding["selector"]["kind"] == "unresolved"
                    assert binding["selector"]["reason"]
                if binding["resolution"] == "unsafe":
                    assert binding["evidence"]["confidence"] == "unconfirmed"
                assert binding["sourceId"] in payload["sources"]


def test_business_projection_is_the_same_projection_as_readable_inspector(
    monkeypatch, current_bundle
):
    bundle = deepcopy(current_bundle)
    monkeypatch.setattr(main.provider, "load_bundle", lambda _build_id: deepcopy(bundle))
    monkeypatch.setattr(main.provider, "load_event_details", lambda _build_id, _ids: {})
    main._inspector_cache.clear()

    for detail_ref in (
        "run:objective",
        "round:1:goal",
        "round:1:plan",
        "data-decision:11",
        "round:1:branch:1:task",
        "round:1:result",
        "run:final-result",
    ):
        old = client.get(f"/api/script-builds/88001/activities/{detail_ref}")
        source = client.get(f"/api/script-builds/88001/inspector-view/{detail_ref}")
        assert old.status_code == 200, detail_ref
        assert source.status_code == 200, detail_ref
        expected = {
            key: value
            for key, value in old.json().items()
            if key not in {"technical", "changes"}
        }
        assert source.json()["businessProjection"] == expected, detail_ref


def test_zero_domain_facts_are_a_complete_empty_query_not_missing(
    monkeypatch, current_bundle
):
    bundle = deepcopy(current_bundle)
    bundle["domainInfo"] = []
    monkeypatch.setattr(main.provider, "load_bundle", lambda _build_id: deepcopy(bundle))
    monkeypatch.setattr(main.provider, "load_event_details", lambda _build_id, _ids: {})
    main._inspector_cache.clear()

    payload = client.get(
        "/api/script-builds/88001/inspector-view/round:1:branch:2:output"
    ).json()
    assert payload["businessProjection"]["businessSections"][0]["content"] == (
        "已查询完成，本方案没有新增领域事实"
    )
    output = _module(payload, "output")
    binding = _row_bindings(output)[0]
    source = payload["sources"][binding["sourceId"]]
    assert binding["resolution"] == "resolved"
    assert source["completeness"] == "complete"
    assert source["resultState"] == "empty"
    assert source["locator"] == {
        "table": "script_build_domain_info",
        "filters": {"branch_id": 2},
        "resultCount": 0,
    }
    assert source["rawRecord"] == []

    round_result = client.get(
        "/api/script-builds/88001/inspector-view/round:1:result"
    ).json()
    domain = _module(round_result, "facts")
    domain_source = round_result["sources"][_row_bindings(domain)[0]["sourceId"]]
    assert domain_source["resultState"] == "empty"
    assert domain_source["rawRecord"]["value"] == []


def test_completeness_scan_includes_item_level_bindings():
    modules = [
        {
            "bindings": [{"id": "module", "resolution": "resolved"}],
            "business": {
                "items": [
                    {
                        "bindings": [
                            {
                                "id": "child",
                                "resolution": "missing",
                                "selector": {"kind": "unresolved", "reason": "missing"},
                            }
                        ]
                    }
                ]
            },
        }
    ]
    assert {item["resolution"] for item in _all_bindings(modules)} == {
        "resolved",
        "missing",
    }


def test_direct_tool_group_has_one_item_binding_per_call(monkeypatch, current_bundle):
    bundle = deepcopy(current_bundle)
    summary = next(item for item in bundle["events"] if item["id"] == 51)
    detail = _event_detail(summary, input_content={}, output_content={"paragraphs": [1, 2]})
    monkeypatch.setattr(main.provider, "load_bundle", lambda _build_id: deepcopy(bundle))
    monkeypatch.setattr(main.provider, "load_event_details", lambda _build_id, _ids: {51: deepcopy(detail)})
    monkeypatch.setattr(main.provider, "load_event_detail", lambda _build_id, _event_id: deepcopy(detail))
    main._inspector_cache.clear()

    payload = client.get(
        "/api/script-builds/88001/inspector-view/retrieval-direct:51"
    ).json()
    activities = _module(payload, "activities")
    assert len(activities["rows"]) == 1
    activity_binding = _row_bindings(activities)[0]
    assert activity_binding["selector"]["kind"] == "whole-record"
    assert activity_binding["transform"]["operation"] == "activity-summary-v1"


def test_whole_event_and_parsed_text_fallback_are_resolved():
    event = _event_detail(
        {
            "id": 91,
            "event_name": "script_evaluator",
            "status": "completed",
            "duration_ms": 20,
        },
        input_content={"task": "评审两个方案"},
        output_content={"summary": "原始 Markdown **结论**"},
    )
    resolver = SourceResolver(bundle={}, event_details={91: event})
    whole = resolver.binding_for_event(91, binding_id="whole", path="")
    assert whole["selector"] == {"kind": "whole-record"}
    assert whole["resolution"] == "resolved"

    parsed = resolver.text_span_binding(
        91,
        binding_id="parsed",
        path="output.content.summary",
        exact_text="清理后的结论",
        role="output",
    )
    assert parsed["selector"] == {
        "kind": "json-pointer",
        "path": "/output/content/summary",
    }
    assert parsed["transform"]["kind"] == "parsed"
    assert parsed["resolution"] == "resolved"


def test_round_goal_and_plan_bind_their_real_save_events(monkeypatch, current_bundle):
    bundle = deepcopy(current_bundle)
    begin = {
        "id": 60,
        "event_seq": 60,
        "event_type": "tool_call",
        "event_name": "begin_round",
        "round_index": 1,
        "status": "completed",
    }
    revision = {
        "id": 61,
        "event_seq": 61,
        "event_type": "tool_call",
        "event_name": "record_multipath_plan",
        "round_index": 1,
        "status": "completed",
    }
    bundle["events"].extend([begin, revision])
    details = {
        60: _event_detail(begin, input_content={"goal": bundle["rounds"][0]["goal"]}, output_content={"round_index": 1}),
        61: _event_detail(revision, input_content={"paths": bundle["rounds"][0]["multipath_plan"], "mode": "分工", "note": "保存规划"}, output_content={"saved": True}),
    }
    monkeypatch.setattr(main.provider, "load_bundle", lambda _build_id: deepcopy(bundle))
    monkeypatch.setattr(main.provider, "load_event_details", lambda _build_id, ids: {event_id: deepcopy(details[event_id]) for event_id in ids if event_id in details})
    main._inspector_cache.clear()

    goal = client.get("/api/script-builds/88001/inspector-view/round:1:goal").json()
    goal_block = _module(goal, "goal")
    entered = next(binding for binding in _row_bindings(goal_block) if binding["sourceId"] == "event:60")
    assert entered["selector"]["path"] == "/input/content/goal"
    assert entered["evidence"]["availability"] == "produced-by-run"

    plan = client.get("/api/script-builds/88001/inspector-view/round:1:plan").json()
    formation = _module(plan, "basis")
    revisions = next(binding for binding in _row_bindings(formation) if binding["sourceId"] == "event:61")
    assert revisions["selector"]["path"] == "/input/content"
    assert revisions["evidence"]["availability"] == "produced-by-run"


def test_round_result_review_uses_the_real_event_record(monkeypatch, current_bundle):
    bundle = deepcopy(current_bundle)
    summary = next(item for item in bundle["events"] if item["id"] == 41)
    detail = _event_detail(
        summary,
        input_content={"task": "评审第 1 轮主脚本"},
        output_content={"summary": summary["agentOutputData"]["summary"]},
    )
    monkeypatch.setattr(main.provider, "load_bundle", lambda _build_id: deepcopy(bundle))
    monkeypatch.setattr(main.provider, "load_event_details", lambda _build_id, ids: {41: deepcopy(detail)} if 41 in ids else {})
    main._inspector_cache.clear()

    payload = client.get("/api/script-builds/88001/inspector-view/round:1:result").json()
    review_bindings = [binding for binding in _all_bindings(payload["modules"]) if binding["sourceId"] == "event:41"]
    assert review_bindings
    assert all(binding["selector"]["kind"] == "whole-record" for binding in review_bindings)
    assert all(binding["resolution"] == "resolved" for binding in review_bindings)


def test_log_window_loader_ignores_events_without_msg_id(monkeypatch):
    captured = []
    monkeypatch.setattr(
        main.audit_repository,
        "load_event_log_windows",
        lambda build_id, anchors: captured.extend(anchors) or {},
    )
    result = main._safe_load_log_windows(443, {9: {"msg_id": None, "event_seq": 77}})
    assert captured == []
    assert result == {}


def test_log_context_requires_the_exact_event_msg_id():
    content = """[FOLD:第一段]
[MSG_ANCHOR:msg_id=message-a:seq=7]
内容 A
[FOLD:第二段]
[TOOL_ANCHOR:msg_id=message-b:seq=7:tool=test]
内容 B
"""
    matched = locate_log_module(content, {"msg_id": "message-b", "event_seq": 7})
    assert matched is not None
    assert matched["msgId"] == "message-b"
    assert "内容 B" in matched["content"]
    assert "内容 A" not in matched["content"]
    assert locate_log_module(content, {"msg_id": None, "event_seq": 7}) is None
    assert locate_log_module(content, {"msg_id": "message-c", "event_seq": 7}) is None


def test_missing_review_and_decision_cards_have_resolved_empty_sources(
    monkeypatch, current_bundle
):
    bundle = deepcopy(current_bundle)
    bundle["events"] = [
        event
        for event in bundle["events"]
        if event.get("event_name") != "script_multipath_evaluator"
    ]
    bundle["multipathDecisions"] = []
    monkeypatch.setattr(main.provider, "load_bundle", lambda _build_id: deepcopy(bundle))
    monkeypatch.setattr(main.provider, "load_event_details", lambda _build_id, _ids: {})
    main._inspector_cache.clear()
    main._build_view_cache.clear()

    execution = client.get("/api/script-builds/88001/execution-view").json()
    batch = execution["rounds"][0]["convergenceBatches"][0]
    refs = [
        batch["missingReview"]["detailRef"],
        batch["missingDecision"]["detailRef"],
    ]
    for detail_ref in refs:
        response = client.get(
            f"/api/script-builds/88001/inspector-view/{detail_ref}"
        )
        assert response.status_code == 200
        payload = response.json()
        assert payload["businessProjection"]["businessSections"][0]["id"] == "status"
        bindings = _row_bindings(_module(payload, "status"))
        assert bindings
        assert all(binding["resolution"] == "resolved" for binding in bindings)
        assert any(
            source["resultState"] == "empty"
            for source in payload["sources"].values()
        )