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() )