script-build-and-framework-tmp/visualization/backend/app/module_audit/service.py

from __future__ import annotations

import json
from hashlib import sha1
from dataclasses import dataclass
from threading import RLock
from time import monotonic
from typing import Any, Iterable

from ..execution_builder import ExecutionViewBuilder
from ..inspector_projection import (
    project_activity_detail,
    project_event_detail,
    project_round_detail,
)
from ..providers import LocalDatabaseProvider
from ..sanitizer import sanitize
from .log_context import locate_log_module
from .repository import AuditRepository


class ModuleNotFound(LookupError):
    pass


@dataclass(frozen=True)
class _RunSnapshot:
    bundle: dict[str, Any]
    view: dict[str, Any]
    modules: list[dict[str, Any]]
    expires_at: float


class ModuleAuditService:
    def __init__(
        self,
        provider: LocalDatabaseProvider | None = None,
        builder: ExecutionViewBuilder | None = None,
        repository: AuditRepository | None = None,
    ):
        self.provider = provider or LocalDatabaseProvider()
        self.builder = builder or ExecutionViewBuilder()
        self.repository = repository or AuditRepository()
        self._snapshot_cache: dict[int, _RunSnapshot] = {}
        self._detail_cache: dict[tuple[int, str], tuple[float, dict[str, Any]]] = {}
        self._event_cache: dict[tuple[int, int], tuple[float, dict[str, Any] | None]] = {}
        self._log_cache: dict[int, tuple[float, str | None]] = {}
        self._snapshot_lock = RLock()

    def _snapshot(self, script_build_id: int) -> _RunSnapshot:
        now = monotonic()
        with self._snapshot_lock:
            cached = self._snapshot_cache.get(script_build_id)
            if cached is not None and cached.expires_at > now:
                return cached

        bundle = self.provider.load_bundle(script_build_id)
        view = self.builder.from_bundle(bundle)
        modules = _collect_modules(view)
        status = str((view.get("header") or {}).get("status") or "").lower()
        ttl = 300.0 if status in {"completed", "success", "failed", "interrupted", "cancelled"} else 3.0
        snapshot = _RunSnapshot(
            bundle=bundle,
            view=view,
            modules=modules,
            expires_at=monotonic() + ttl,
        )
        with self._snapshot_lock:
            self._snapshot_cache[script_build_id] = snapshot
            if len(self._snapshot_cache) > 12:
                stale_ids = sorted(
                    self._snapshot_cache,
                    key=lambda item: self._snapshot_cache[item].expires_at,
                )[:-12]
                for stale_id in stale_ids:
                    self._snapshot_cache.pop(stale_id, None)
        return snapshot

    def index(self, script_build_id: int) -> dict[str, Any]:
        snapshot = self._snapshot(script_build_id)
        view = snapshot.view
        modules = snapshot.modules
        return sanitize(
            {
                "schemaVersion": "1",
                "run": {
                    "id": script_build_id,
                    "status": (view.get("header") or {}).get("status"),
                    "capturedAt": view.get("capturedAt"),
                    "dataShape": view.get("dataShape"),
                },
                "columns": ["模块实际使用", "模块实际读取", "原始记录与上下文"],
                "modules": [
                    {
                        "id": item["id"],
                        "title": item["title"],
                        "group": item["group"],
                        "path": item["path"],
                        "kind": item["kind"],
                        "role": item.get("role"),
                        "roundIndex": item.get("roundIndex"),
                        "branchId": item.get("branchId"),
                        "detailRef": item.get("detailRef"),
                        "displayVariant": item.get("displayVariant", "expanded"),
                        "cardCharacters": _characters(_card_use(item)),
                        "completeness": "available",
                    }
                    for item in modules
                ],
                "summary": {
                    "moduleCount": len(modules),
                    "groupCount": len({item["group"] for item in modules}),
                    "cardCharacters": sum(_characters(_card_use(item)) for item in modules),
                },
            },
            max_text=None,
        )

    def detail(self, script_build_id: int, module_id: str) -> dict[str, Any]:
        snapshot = self._snapshot(script_build_id)
        cache_key = (script_build_id, module_id)
        now = monotonic()
        with self._snapshot_lock:
            cached_detail = self._detail_cache.get(cache_key)
            if cached_detail is not None and cached_detail[0] > now:
                return cached_detail[1]
        bundle = snapshot.bundle
        view = snapshot.view
        module = next((item for item in snapshot.modules if item["id"] == module_id), None)
        if module is None:
            raise ModuleNotFound(module_id)

        inspector, inspector_notice = self._inspector(script_build_id, module, view, bundle)
        card = _card_use(module)
        reads = self._reads(script_build_id, module, inspector, bundle)
        reads = _annotate_usage_areas(reads, inspector)
        use_by_consumer = {
            "card": _strings(card),
            "inspector": _strings(inspector),
        }
        use_snippets = sorted(
            set(use_by_consumer["card"] + use_by_consumer["inspector"]),
            key=len,
            reverse=True,
        )
        contexts = self._contexts(
            script_build_id,
            reads,
            use_snippets,
            expires_at=snapshot.expires_at,
        )
        reads, contexts = _annotate_lineage(module["id"], reads, contexts)
        completeness = _completeness(inspector_notice, reads, contexts)
        payload = sanitize(
            {
                "schemaVersion": "1",
                "id": module["id"],
                "title": module["title"],
                "group": module["group"],
                "path": module["path"],
                "actualUse": {
                    "card": card,
                    "inspector": inspector,
                    "inspectorNotice": inspector_notice,
                    "characters": {
                        "card": _characters(card),
                        "inspectorBusiness": _characters((inspector or {}).get("businessSections")),
                        "inspectorDecision": _characters((inspector or {}).get("blocks")),
                        "inspectorChanges": _characters((inspector or {}).get("changes")),
                        "inspectorTechnical": _characters((inspector or {}).get("technical")),
                    },
                },
                "actualReads": [
                    _public_read(item, use_by_consumer) for item in reads
                ],
                "contexts": contexts,
                "completeness": completeness,
            },
            max_text=None,
        )
        with self._snapshot_lock:
            self._detail_cache[cache_key] = (snapshot.expires_at, payload)
            if len(self._detail_cache) > 512:
                stale_keys = sorted(
                    self._detail_cache,
                    key=lambda item: self._detail_cache[item][0],
                )[:-512]
                for stale_key in stale_keys:
                    self._detail_cache.pop(stale_key, None)
        return payload

    def _inspector(
        self,
        script_build_id: int,
        module: dict[str, Any],
        view: dict[str, Any],
        bundle: dict[str, Any],
    ) -> tuple[dict[str, Any] | None, str | None]:
        detail_ref = str(module.get("detailRef") or "")
        if not detail_ref:
            return None, "该模块没有 Inspector 入口。"
        if module["kind"] == "final-result" or detail_ref.startswith("artifact:"):
            return None, "该按钮进入脚本表查看器，不属于 Inspector。"
        try:
            if module["kind"] == "round-summary":
                return project_round_detail(
                    script_build_id,
                    module["payload"],
                    bundle,
                ), None
            if detail_ref.startswith("event:"):
                event_id = int(detail_ref.rsplit(":", 1)[1])
                event_detail = self.provider.load_event_detail(script_build_id, event_id)
                try:
                    return project_activity_detail(
                        script_build_id,
                        detail_ref,
                        view,
                        bundle,
                        event_detail=event_detail,
                    ), None
                except KeyError:
                    return project_event_detail(
                        script_build_id,
                        event_detail,
                        run_status=(view.get("header") or {}).get("status"),
                    ), None
            return project_activity_detail(
                script_build_id,
                detail_ref,
                view,
                bundle,
            ), None
        except (KeyError, ValueError, LookupError) as exc:
            return None, f"Inspector 记录不可用：{type(exc).__name__}"

    def _reads(
        self,
        script_build_id: int,
        module: dict[str, Any],
        inspector: dict[str, Any] | None,
        bundle: dict[str, Any],
    ) -> list[dict[str, Any]]:
        reads = _explicit_reads(module, bundle)
        technical = (inspector or {}).get("technical")
        if isinstance(technical, dict):
            for key, value in technical.items():
                if key in {"scriptBuildId", "table"} or value in (None, "", [], {}):
                    continue
                source = _source_from_technical(key, value)
                reads.append(
                    {
                        "consumer": "inspector",
                        "label": _technical_label(key),
                        "disposition": "selected",
                        "relation": "exact-record" if isinstance(value, (dict, list)) else "projected",
                        "source": source,
                        "fieldPath": f"technical.{key}",
                        "value": value,
                        "characters": _characters(value),
                        "context": value,
                    }
                )

        event_ids = _event_ids(module, reads)
        known = {
            int(item["source"]["eventId"])
            for item in reads
            if isinstance(item.get("source"), dict)
            and item["source"].get("eventId") is not None
        }
        for event_id in event_ids:
            if event_id in known:
                continue
            event = self.repository.load_event(script_build_id, event_id)
            if event:
                reads.append(_event_read(event, "module", "关联运行事件", "calculation-input"))
        return _dedupe_reads(reads)

    def _contexts(
        self,
        script_build_id: int,
        reads: list[dict[str, Any]],
        use_snippets: list[str],
        *,
        expires_at: float,
    ) -> list[dict[str, Any]]:
        contexts: list[dict[str, Any]] = []
        event_ids: set[int] = set()
        for read in reads:
            source = read.get("source") or {}
            event_id = _integer(source.get("eventId"))
            if event_id is not None:
                event_ids.add(event_id)
                if source.get("sourceKind") == "database":
                    continue
            content = read.get("context")
            if content in (None, "", [], {}):
                continue
            function_result = source.get("sourceKind") == "function"
            contexts.append(
                {
                    "id": _context_id(source, read.get("fieldPath")),
                    "kind": "function-result" if function_result else "database-record",
                    "title": (
                        f"函数处理结果 · {read.get('label') or '数据'}"
                        if function_result
                        else read.get("label") or "原始数据库记录"
                    ),
                    "source": source,
                    "sourceKey": _source_key(source),
                    "relatedSourceKey": (
                        f"database:event:{source.get('eventId')}"
                        if function_result and source.get("eventId") is not None
                        else None
                    ),
                    "content": content,
                    "characters": _characters(content),
                    "readPaths": ["$"] if function_result else [read.get("fieldPath")],
                    "usedFragments": _matching_snippets(content, use_snippets),
                    "match": read.get("relation"),
                    "completeness": "complete",
                }
            )

        log_content = self._load_log(script_build_id, expires_at) if event_ids else None
        for event_id in sorted(event_ids):
            event = self._load_event(script_build_id, event_id, expires_at)
            if not event:
                contexts.append(
                    {
                        "id": f"event:{event_id}:missing",
                        "kind": "event",
                        "title": f"Event {event_id}",
                        "source": {
                            "sourceKind": "database",
                            "table": "script_build_event + script_build_event_body",
                            "eventId": event_id,
                        },
                        "sourceKey": f"database:event:{event_id}",
                        "content": "事件记录不存在。",
                        "characters": 8,
                        "readPaths": [],
                        "usedFragments": [],
                        "match": "missing",
                        "completeness": "missing",
                    }
                )
                continue
            contexts.append(
                {
                    "id": f"event:{event_id}",
                    "kind": "event-record",
                    "title": f"script_build_event #{event_id}",
                    "source": {
                        "sourceKind": "database",
                        "table": "script_build_event + script_build_event_body",
                        "eventId": event_id,
                    },
                    "sourceKey": f"database:event:{event_id}",
                    "content": event,
                    "characters": _characters(event),
                    "readPaths": ["eventBody.input_content", "eventBody.output_content", "inputData", "outputPreview"],
                    "usedFragments": _matching_snippets(event, use_snippets),
                    "match": "exact-record",
                    "completeness": "complete" if event.get("eventBody") else "body-missing",
                }
            )
            log_module = locate_log_module(log_content, event)
            if log_module:
                contexts.append(
                    {
                        "id": f"event:{event_id}:log",
                        "kind": "log-module",
                        "title": f"锚定日志模块 · Event {event_id}",
                        "source": {
                            "sourceKind": "log",
                            "table": "script_build_log",
                            "eventId": event_id,
                            "anchor": log_module["anchor"],
                        },
                        "sourceKey": f"log:event:{event_id}",
                        "relatedSourceKey": f"database:event:{event_id}",
                        "content": log_module["content"],
                        "characters": log_module["characters"],
                        "readPaths": [],
                        "usedFragments": _matching_snippets(log_module["content"], use_snippets),
                        "match": "anchored",
                        "completeness": "complete",
                    }
                )
        return _dedupe_contexts(contexts)

    def _load_event(
        self,
        script_build_id: int,
        event_id: int,
        expires_at: float,
    ) -> dict[str, Any] | None:
        key = (script_build_id, event_id)
        now = monotonic()
        with self._snapshot_lock:
            cached = self._event_cache.get(key)
            if cached is not None and cached[0] > now:
                return cached[1]
            event = self.repository.load_event(script_build_id, event_id)
            self._event_cache[key] = (expires_at, event)
            if len(self._event_cache) > 512:
                stale_keys = sorted(
                    self._event_cache,
                    key=lambda item: self._event_cache[item][0],
                )[:-512]
                for stale_key in stale_keys:
                    self._event_cache.pop(stale_key, None)
            return event

    def _load_log(self, script_build_id: int, expires_at: float) -> str | None:
        now = monotonic()
        with self._snapshot_lock:
            cached = self._log_cache.get(script_build_id)
            if cached is not None and cached[0] > now:
                return cached[1]
            log_content = self.repository.load_log(script_build_id)
            self._log_cache[script_build_id] = (expires_at, log_content)
            if len(self._log_cache) > 12:
                stale_ids = sorted(
                    self._log_cache,
                    key=lambda item: self._log_cache[item][0],
                )[:-12]
                for stale_id in stale_ids:
                    self._log_cache.pop(stale_id, None)
            return log_content


def _collect_modules(view: dict[str, Any]) -> list[dict[str, Any]]:
    modules: list[dict[str, Any]] = []

    def add(
        payload: Any,
        *,
        group: str,
        path: list[str],
        kind: str,
        round_index: int | None = None,
        branch_id: int | None = None,
        display_variant: str = "expanded",
        forced_id: str | None = None,
        fallback_id: str | None = None,
        fallback_title: str | None = None,
        fallback_detail_ref: str | None = None,
    ) -> None:
        if not isinstance(payload, dict):
            return
        module_id = str(forced_id or payload.get("id") or fallback_id or "").strip()
        if not module_id:
            return
        modules.append(
            {
                "id": module_id,
                "title": str(payload.get("title") or fallback_title or module_id),
                "group": group,
                "path": path,
                "kind": kind,
                "role": payload.get("role"),
                "roundIndex": round_index,
                "branchId": branch_id,
                "detailRef": payload.get("detailRef") or fallback_detail_ref,
                "displayVariant": display_variant,
                "payload": payload,
            }
        )

    add(view.get("planning"), group="构建前", path=["构建前"], kind="story")
    add(view.get("objective"), group="构建前", path=["构建前"], kind="story")
    for round_ in view.get("rounds") or []:
        round_index = _integer(round_.get("roundIndex")) or 0
        group = f"第 {round_index} 轮"
        add(round_.get("goal"), group=group, path=[group, "主 Agent"], kind="story", round_index=round_index)
        add((round_.get("plan") or {}).get("node"), group=group, path=[group, "主 Agent"], kind="story", round_index=round_index)
        for branch in round_.get("branches") or []:
            branch_id = _integer(branch.get("branchId")) or 0
            branch_label = f"方案 {branch_id}"
            add(branch.get("task"), group=group, path=[group, branch_label], kind="story", round_index=round_index, branch_id=branch_id)
            stage = branch.get("retrievalStage") or {}
            add(stage, group=group, path=[group, branch_label, "取数"], kind="retrieval-stage", round_index=round_index, branch_id=branch_id, fallback_title="取数阶段")
            for direct in stage.get("directToolGroups") or []:
                add(direct, group=group, path=[group, branch_label, "取数", "工具取数"], kind="retrieval-direct", round_index=round_index, branch_id=branch_id)
                for call in direct.get("calls") or []:
                    add(call, group=group, path=[group, branch_label, "取数", "工具调用"], kind="retrieval-call", round_index=round_index, branch_id=branch_id, fallback_title=call.get("businessLabel"))
            for agent in stage.get("agentRuns") or []:
                add(agent, group=group, path=[group, branch_label, "取数", "Agent 取数"], kind="retrieval-agent", round_index=round_index, branch_id=branch_id, fallback_title=agent.get("businessLabel"))
                for attempt in agent.get("attempts") or []:
                    add(attempt, group=group, path=[group, branch_label, "取数", "单次查询"], kind="query-attempt", round_index=round_index, branch_id=branch_id, fallback_title=attempt.get("queryLabel"))
            for decision in branch.get("dataDecisions") or []:
                add(decision.get("node") or decision, group=group, path=[group, branch_label], kind="story", round_index=round_index, branch_id=branch_id)
            add((branch.get("output") or {}).get("node"), group=group, path=[group, branch_label], kind="story", round_index=round_index, branch_id=branch_id)
        for batch in round_.get("convergenceBatches") or []:
            add(batch.get("review") or batch.get("missingReview"), group=group, path=[group, "收敛"], kind="review", round_index=round_index)
            add((batch.get("decision") or {}).get("node") or batch.get("decision") or batch.get("missingDecision"), group=group, path=[group, "收敛"], kind="decision", round_index=round_index)
        add(round_.get("overallEvaluation"), group=group, path=[group, "整体评审"], kind="story", round_index=round_index)
        add(round_.get("result"), group=group, path=[group, "本轮产出"], kind="story", round_index=round_index)
    add(view.get("finalResult"), group="最终结果", path=["最终结果"], kind="final-result")
    return _unique_modules(modules)


def _card_use(module: dict[str, Any]) -> dict[str, Any]:
    payload = module["payload"]
    card = payload.get("card")
    if not isinstance(card, dict):
        card = ((payload.get("decisionProjection") or {}).get("card") if isinstance(payload.get("decisionProjection"), dict) else None)
    if isinstance(card, dict):
        fields = []
        primary = card.get("primary")
        if isinstance(primary, dict):
            fields.append({"label": primary.get("label"), "value": primary.get("value"), "kind": "primary"})
        for item in card.get("secondary") or []:
            if isinstance(item, dict):
                fields.append({"label": item.get("label"), "value": item.get("value"), "kind": "secondary"})
        return {"title": module["title"], "fields": fields}

    kind = module["kind"]
    if kind == "round-summary":
        summary = payload.get("summary") or {}
        return {
            "title": f"第 {module.get('roundIndex')} 轮摘要",
            "fields": [
                {"label": "本轮目标", "value": (summary.get("goal") or {}).get("headline")},
                {"label": "尝试方案", "value": summary.get("approach")},
                {"label": "主 Agent 决策", "value": summary.get("decision")},
                {"label": "本轮产出", "value": summary.get("output")},
            ],
        }
    if kind == "branch-summary":
        summary = payload.get("summary") or {}
        return {"title": module["title"], "fields": [{"label": "任务", "value": summary.get("task")}, {"label": "产出", "value": summary.get("output")}, {"label": "状态", "value": payload.get("status")}]}
    if kind == "retrieval-stage":
        return {"title": "取数阶段", "fields": [{"label": "模式", "value": payload.get("observedMode")}, {"label": "状态", "value": payload.get("status")}, {"label": "取数模块", "value": f"{len(payload.get('directToolGroups') or [])} 个工具组、{len(payload.get('agentRuns') or [])} 个 Agent"}]}
    if kind == "retrieval-direct":
        sources = "、".join(str(item.get("businessLabel") or "") for item in payload.get("sources") or [] if isinstance(item, dict))
        result = f"{payload.get('failureCount')} 次失败" if payload.get("failureCount") else f"{payload.get('interruptedCount')} 次中断" if payload.get("interruptedCount") else "进行中" if payload.get("runningCount") else "已完成"
        return {"title": sources or "工具取数", "fields": [{"label": "读取", "value": f"{payload.get('callCount') or 0} 次"}, {"label": "结果", "value": result}]}
    if kind == "retrieval-agent":
        query = payload.get("querySummary") or {}
        return {"title": payload.get("businessLabel") or "Agent 取数", "fields": [{"label": "取数目标", "value": payload.get("taskPreview")}, {"label": "查询过程", "value": query}, {"label": "初筛整理", "value": (payload.get("screening") or {}).get("preview")}]}
    if kind in {"query-attempt", "retrieval-call"}:
        return {"title": module["title"], "fields": [{"label": "状态", "value": payload.get("status")}, {"label": "结果数量", "value": payload.get("resultCount")}, {"label": "结果摘要", "value": payload.get("resultSummary")}]}
    if kind == "final-result":
        return {"title": module["title"], "fields": [{"label": "执行结果", "value": payload.get("summary")}, {"label": "状态", "value": payload.get("status")}, {"label": "轮次", "value": payload.get("roundCount")}, {"label": "方案", "value": payload.get("branchCounts")}]}
    return {"title": module["title"], "fields": [{"label": "模块数据", "value": payload}]}


def _explicit_reads(module: dict[str, Any], bundle: dict[str, Any]) -> list[dict[str, Any]]:
    reads: list[dict[str, Any]] = []
    role = str(module.get("role") or "")
    round_index = module.get("roundIndex")
    branch_id = module.get("branchId")
    raw_round = _find(bundle.get("rounds"), "round_index", round_index)
    raw_branch = _find(bundle.get("branches"), "branch_id", branch_id)

    if role == "planning-analysis":
        _append_related_events(reads, module, bundle, "card", "规划事件")
    elif role == "objective":
        record = bundle.get("record") or {}
        reads.append(_record_read("card", "创作目标", "script_build_record", record, "script_direction"))
    elif role == "goal" and raw_round:
        reads.append(_record_read("card", "本轮目标", "script_build_round", raw_round, "goal"))
    elif role == "plan" and raw_round:
        for field, label in (("multipath_plan", "实现路线"), ("race_or_divide", "组织方式"), ("plan_note", "规划理由")):
            if raw_round.get(field) not in (None, "", [], {}):
                reads.append(_record_read("card", label, "script_build_round", raw_round, field))
    elif role == "branch-task" and raw_branch:
        reads.append(_record_read("card", "实现任务", "script_build_branch", raw_branch, "impl_task"))
        task_event = next(
            (
                event
                for event in reversed(bundle.get("events") or [])
                if event.get("event_name") == "script_implementer"
                and _integer(event.get("round_index")) == round_index
                and _integer(event.get("branch_id")) == branch_id
            ),
            None,
        )
        if task_event:
            event_value = (task_event.get("inputData") or {}).get("task") if isinstance(task_event.get("inputData"), dict) else None
            reads.append(_event_summary_read(task_event, "inspector", "实现 Agent 派发任务", "inputData.task", event_value, "selected" if event_value else "missing"))
            reads.append({**_record_read("inspector", "历史回退任务", "script_build_branch", raw_branch, "impl_task"), "disposition": "fallback" if event_value else "selected"})
        else:
            reads.append({**_record_read("inspector", "实现任务", "script_build_branch", raw_branch, "impl_task"), "disposition": "selected"})
    elif role == "data-decision":
        row = _find(bundle.get("dataDecisions"), "id", _suffix_int(module["id"]))
        if row:
            reads.append(_whole_record_read("card", "取数取舍记录", "script_build_data_decision", row))
    elif role in {"domain-output", "candidate-output"} and raw_branch:
        reads.append(_whole_record_read("card", "实现方案记录", "script_build_branch", raw_branch))
        if role == "domain-output":
            for row in bundle.get("domainInfo") or []:
                if _integer(row.get("branch_id")) == branch_id:
                    reads.append(_whole_record_read("card", "领域事实", "script_build_domain_info", row))
    elif role == "multipath-decision":
        row = _find(bundle.get("multipathDecisions"), "id", _suffix_int(module["id"]))
        if row:
            reads.append(_whole_record_read("card", "多方案决定", "script_build_multipath_decision", row))
    elif role in {"round-evaluation", "round-result"} and raw_round:
        reads.append(_whole_record_read("card", "轮次记录", "script_build_round", raw_round))
        _append_related_events(reads, module, bundle, "card", "轮次运行事件")

    kind = module["kind"]
    payload = module["payload"]
    if kind == "round-summary" and raw_round:
        reads.append(_whole_record_read("card", "折叠态轮次记录", "script_build_round", raw_round))
    elif kind == "branch-summary" and raw_branch:
        reads.append(_whole_record_read("card", "折叠态方案记录", "script_build_branch", raw_branch))
    elif kind.startswith("retrieval") or kind == "query-attempt":
        event_ids = _event_ids_from_payload(payload)
        for event_id in event_ids:
            event = next((item for item in bundle.get("events") or [] if _integer(item.get("id")) == event_id), None)
            if event:
                reads.append(_event_summary_read(event, "card", "取数运行事件", "event", event, "calculation-input"))
    elif kind == "review":
        _append_related_events(reads, module, bundle, "card", "评审事件")
    elif kind == "final-result":
        reads.append(_whole_record_read("card", "构建记录", "script_build_record", bundle.get("record") or {}))
        reads.append(_whole_record_read("card", "当前主脚本快照", "script_build_artifact", bundle.get("currentArtifact") or {}))

    if not reads:
        reads.append(
            {
                "consumer": "module",
                "label": "来源未确认",
                "disposition": "unconfirmed",
                "relation": "unconfirmed",
                "source": {"type": "unknown", "sourceKind": "unknown"},
                "fieldPath": None,
                "value": "当前审计规则无法可靠确认这个模块的字段来源。",
                "characters": 24,
                "context": None,
            }
        )
    return reads


def _record_read(consumer: str, label: str, table: str, row: dict[str, Any], field: str) -> dict[str, Any]:
    value = row.get(field)
    return {
        "consumer": consumer,
        "label": label,
        "disposition": "selected",
        "relation": (
            "projected"
            if consumer == "card"
            else "exact" if isinstance(value, str) else "structured"
        ),
        "source": {
            "type": "business-record",
            "sourceKind": "database",
            "table": table,
            "recordId": row.get("id"),
        },
        "fieldPath": field,
        "value": value,
        "characters": _characters(value),
        "context": row,
    }


def _whole_record_read(consumer: str, label: str, table: str, row: dict[str, Any]) -> dict[str, Any]:
    return {
        "consumer": consumer,
        "label": label,
        "disposition": "selected",
        "relation": (
            "combined"
            if table == "script_build_artifact"
            else "projected" if consumer == "card" else "exact-record"
        ),
        "source": {
            "type": "business-record",
            "sourceKind": "database",
            "table": table,
            "recordId": row.get("id"),
        },
        "fieldPath": "$",
        "value": row,
        "characters": _characters(row),
        "context": row,
    }


def _event_summary_read(event: dict[str, Any], consumer: str, label: str, field_path: str, value: Any, disposition: str) -> dict[str, Any]:
    return {
        "consumer": consumer,
        "label": label,
        "disposition": disposition,
        "relation": (
            "calculation-input"
            if disposition == "calculation-input"
            else "exact" if isinstance(value, str) else "structured"
        ),
        "source": {
            "type": "runtime-event",
            "sourceKind": "database",
            "table": "script_build_event + script_build_event_body",
            "eventId": event.get("id"),
            "eventName": event.get("event_name"),
        },
        "fieldPath": field_path,
        "value": value,
        "characters": _characters(value),
        "context": None,
    }


def _event_read(event: dict[str, Any], consumer: str, label: str, disposition: str) -> dict[str, Any]:
    return {
        "consumer": consumer,
        "label": label,
        "disposition": disposition,
        "relation": "exact-record",
        "source": {
            "type": "runtime-event",
            "sourceKind": "database",
            "table": "script_build_event + script_build_event_body",
            "eventId": event.get("id"),
            "eventName": event.get("event_name"),
        },
        "fieldPath": "$",
        "value": event,
        "characters": _characters(event),
        "context": None,
    }


def _append_related_events(reads: list[dict[str, Any]], module: dict[str, Any], bundle: dict[str, Any], consumer: str, label: str) -> None:
    for event_id in _event_ids_from_payload(module.get("payload") or {}):
        event = next((item for item in bundle.get("events") or [] if _integer(item.get("id")) == event_id), None)
        if event:
            reads.append(_event_summary_read(event, consumer, label, "event", event, "calculation-input"))


def _source_from_technical(key: str, value: Any) -> dict[str, Any]:
    event_id = None
    table = None
    if isinstance(value, dict):
        event_id = _integer(value.get("eventId") or value.get("id")) if "event" in key.lower() else None
        table = value.get("table")
    return {
        "type": "inspector-projection",
        "sourceKind": "function",
        "functionName": "Inspector 详情组装",
        "table": table,
        "eventId": event_id,
        "technicalGroup": key,
    }


def _event_ids(module: dict[str, Any], reads: list[dict[str, Any]]) -> list[int]:
    values = _event_ids_from_payload(module.get("payload") or {})
    values.extend(
        _integer((item.get("source") or {}).get("eventId"))
        for item in reads
    )
    return sorted({value for value in values if value is not None})


def _event_ids_from_payload(value: Any) -> list[int]:
    result: list[int] = []
    if isinstance(value, dict):
        for key, nested in value.items():
            if key in {"eventId", "implementerEventId", "eventRef"}:
                parsed = _integer(nested)
                if parsed is None:
                    parsed = _suffix_int(nested)
                if parsed is not None:
                    result.append(parsed)
            elif key == "detailRef" and str(nested or "").startswith("event:"):
                parsed = _suffix_int(nested)
                if parsed is not None:
                    result.append(parsed)
            elif key in {"technicalRefs", "runtimeRevisionRefs"} and isinstance(nested, list):
                result.extend(_suffix_int(item) for item in nested if _suffix_int(item) is not None)
            elif key not in {"card", "summary", "body", "content_snapshot", "candidate_snapshot"}:
                result.extend(_event_ids_from_payload(nested))
    elif isinstance(value, list):
        for item in value:
            result.extend(_event_ids_from_payload(item))
    return sorted({item for item in result if item is not None})


def _public_read(
    value: dict[str, Any], use_by_consumer: dict[str, list[str]]
) -> dict[str, Any]:
    public = {key: item for key, item in value.items() if key != "context"}
    source = value.get("source") or {}
    public["sourceKey"] = _source_key(source)
    disposition = str(value.get("disposition") or "")
    relation = str(value.get("relation") or "")
    public["sourcePriority"] = {
        "selected": "已采用来源",
        "fallback": "备用来源",
        "calculation-input": "计算输入",
        "unconfirmed": "来源未确认",
        "missing": "来源缺失",
    }.get(disposition, "已读取来源")
    public["transformation"] = {
        "exact": "直接摘取",
        "exact-record": "完整记录读取",
        "structured": "结构化字段读取",
        "projected": "展示处理",
        "combined": "多来源重组",
        "calculation-input": "参与计算",
        "unconfirmed": "无法确认",
    }.get(relation, relation or "未标记")
    ratio = _usage_ratio(
        value.get("value"),
        use_by_consumer.get(str(value.get("consumer") or ""), []),
        relation,
        disposition,
    )
    public["usageRatio"] = (
        {
            "calculable": True,
            "value": ratio,
            "label": f"{ratio * 100:.1f}%",
        }
        if ratio is not None
        else {
            "calculable": False,
            "value": None,
            "label": "不可直接计算",
        }
    )
    return public


def _annotate_lineage(
    module_id: str,
    reads: list[dict[str, Any]],
    contexts: list[dict[str, Any]],
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    """Attach field-level pointers without inventing lineage for derived values."""
    contexts_by_source: dict[str, list[dict[str, Any]]] = {}
    for context in contexts:
        contexts_by_source.setdefault(str(context.get("sourceKey") or ""), []).append(context)

    for index, read in enumerate(reads):
        source_key = _source_key(read.get("source") or {})
        identity = "|".join(
            (
                module_id,
                str(read.get("consumer") or "module"),
                source_key,
                str(read.get("fieldPath") or "$"),
                str(read.get("label") or "data"),
                str(index),
            )
        )
        read_id = f"read:{sha1(identity.encode('utf-8')).hexdigest()[:14]}"
        read["readId"] = read_id
        refs: list[dict[str, Any]] = []
        for context in contexts_by_source.get(source_key, []):
            context_path, relation = _lineage_path(read, context)
            refs.append(
                {
                    "contextId": context.get("id"),
                    "sourceKey": source_key,
                    "contextFieldPath": context_path,
                    "relation": relation,
                    "exact": relation in {"direct", "normalized-alias", "whole-record"},
                }
            )
        read["contextRefs"] = refs

    read_ids_by_context: dict[str, list[str]] = {}
    paths_by_context: dict[str, list[str]] = {}
    for read in reads:
        for ref in read.get("contextRefs") or []:
            context_id = str(ref.get("contextId") or "")
            if not context_id:
                continue
            read_ids_by_context.setdefault(context_id, []).append(str(read["readId"]))
            field_path = str(ref.get("contextFieldPath") or "$")
            paths_by_context.setdefault(context_id, []).append(field_path)

    for context in contexts:
        context_id = str(context.get("id") or "")
        context["readIds"] = list(dict.fromkeys(read_ids_by_context.get(context_id, [])))
        mapped_paths = list(dict.fromkeys(paths_by_context.get(context_id, [])))
        if mapped_paths:
            context["readPaths"] = mapped_paths
    return reads, contexts


def _lineage_path(
    read: dict[str, Any], context: dict[str, Any]
) -> tuple[str, str]:
    field_path = str(read.get("fieldPath") or "$")
    content = context.get("content")
    source_kind = str((read.get("source") or {}).get("sourceKind") or "")

    if source_kind == "function":
        return "$", "transformed"
    if field_path in {"$", "event"}:
        return "$", "whole-record"

    candidates = [field_path]
    if field_path.startswith("inputData."):
        suffix = field_path.removeprefix("inputData.")
        candidates.append(f"eventBody.input_content.{suffix}")
    elif field_path == "inputData":
        candidates.append("eventBody.input_content")
    if field_path.startswith("agentOutputData."):
        suffix = field_path.removeprefix("agentOutputData.")
        candidates.append(f"eventBody.output_content.{suffix}")
    elif field_path in {"agentOutputData", "outputPreview", "output_preview"}:
        candidates.append("eventBody.output_content")

    for candidate in candidates:
        if _value_at_path(content, candidate) is not _MISSING:
            relation = "direct" if candidate == field_path else "normalized-alias"
            return candidate, relation
    return "$", "record-only"


_MISSING = object()


def _value_at_path(value: Any, field_path: str) -> Any:
    if not field_path or field_path == "$":
        return value
    current = value
    for part in field_path.removeprefix("$.").split("."):
        if isinstance(current, str):
            text = current.strip()
            if text[:1] in {"{", "["}:
                try:
                    current = json.loads(text)
                except (TypeError, ValueError, json.JSONDecodeError):
                    return _MISSING
        if not isinstance(current, dict) or part not in current:
            return _MISSING
        current = current[part]
    return current


def _annotate_usage_areas(
    reads: list[dict[str, Any]], inspector: dict[str, Any] | None
) -> list[dict[str, Any]]:
    has_business = bool(
        (inspector or {}).get("businessSections") or (inspector or {}).get("blocks")
    )
    has_changes = bool((inspector or {}).get("changes"))
    has_technical = isinstance((inspector or {}).get("technical"), dict)

    for read in reads:
        consumer = read.get("consumer")
        field_path = str(read.get("fieldPath") or "")
        disposition = read.get("disposition")
        areas: list[str] = []
        if disposition == "fallback":
            read["usageAreas"] = areas
            continue
        if consumer == "card":
            areas.append("卡片")
            if has_business:
                areas.append("Inspector 业务详情")
            if has_changes:
                areas.append("Inspector 产出与改动")
        elif consumer == "inspector":
            areas.append(
                "Inspector 技术详情"
                if field_path.startswith("technical.")
                else "Inspector 业务详情"
            )
        elif consumer == "module":
            areas.append("模块处理")
        read["usageAreas"] = areas

    if has_technical:
        technical_refs = {
            _underlying_record_key(read.get("source") or {})
            for read in reads
            if str(read.get("fieldPath") or "").startswith("technical.")
        }
        technical_refs.discard(None)
        for read in reads:
            source = read.get("source") or {}
            if (
                source.get("sourceKind") == "database"
                and _underlying_record_key(source) in technical_refs
                and read.get("disposition") != "fallback"
                and "Inspector 技术详情" not in read["usageAreas"]
            ):
                read["usageAreas"].append("Inspector 技术详情")
    return reads


def _underlying_record_key(source: dict[str, Any]) -> str | None:
    event_id = _integer(source.get("eventId"))
    if event_id is not None:
        return f"event:{event_id}"
    table = source.get("table")
    record_id = source.get("recordId")
    if table and record_id is not None:
        return f"record:{table}:{record_id}"
    return None


def _source_key(source: dict[str, Any]) -> str:
    kind = str(source.get("sourceKind") or "unknown")
    underlying = _underlying_record_key(source)
    if kind == "database" and underlying:
        return f"database:{underlying}"
    if kind == "log":
        return f"log:event:{source.get('eventId') or 'unknown'}"
    if kind == "function":
        group = source.get("technicalGroup") or source.get("functionName") or "detail"
        event = source.get("eventId")
        return f"function:{group}:{event or 'module'}"
    return f"unknown:{source.get('type') or 'source'}"


def _usage_ratio(
    value: Any, snippets: list[str], relation: str, disposition: str
) -> float | None:
    if (
        disposition != "selected"
        or relation != "exact"
        or not isinstance(value, str)
        or not value
    ):
        return None
    normalized_source = value.strip()
    if not normalized_source:
        return None
    candidates = []
    for snippet in snippets:
        normalized = snippet.strip().removesuffix("…").rstrip()
        if len(normalized) >= 8 and normalized in normalized_source:
            candidates.append(normalized)
    if not candidates:
        return None
    return min(1.0, len(max(candidates, key=len)) / len(normalized_source))


def _dedupe_reads(reads: list[dict[str, Any]]) -> list[dict[str, Any]]:
    seen: set[tuple[str, str, str, str]] = set()
    result = []
    for item in reads:
        source = item.get("source") or {}
        key = (str(item.get("consumer")), str(source.get("table")), str(source.get("eventId") or source.get("recordId")), str(item.get("fieldPath")))
        if key in seen:
            continue
        seen.add(key)
        result.append(item)
    return result


def _dedupe_contexts(contexts: list[dict[str, Any]]) -> list[dict[str, Any]]:
    result = []
    seen = set()
    for item in contexts:
        key = item.get("id")
        if key in seen:
            continue
        seen.add(key)
        result.append(item)
    return result


def _unique_modules(modules: list[dict[str, Any]]) -> list[dict[str, Any]]:
    seen = set()
    result = []
    for item in modules:
        if item["id"] in seen:
            continue
        seen.add(item["id"])
        result.append(item)
    return result


def _find(values: Any, key: str, expected: Any) -> dict[str, Any] | None:
    if expected is None:
        return None
    return next((item for item in values or [] if _integer(item.get(key)) == _integer(expected)), None)


def _characters(value: Any) -> int:
    if value is None:
        return 0
    if isinstance(value, str):
        return len(value)
    try:
        return len(json.dumps(value, ensure_ascii=False, sort_keys=True, default=str, separators=(",", ":")))
    except (TypeError, ValueError):
        return len(str(value))


def _strings(value: Any) -> list[str]:
    result: list[str] = []
    if isinstance(value, str):
        if len(value.strip()) >= 8:
            result.append(value.strip())
    elif isinstance(value, dict):
        for nested in value.values():
            result.extend(_strings(nested))
    elif isinstance(value, list):
        for nested in value:
            result.extend(_strings(nested))
    return sorted(set(result), key=len, reverse=True)


def _matching_snippets(value: Any, snippets: list[str]) -> list[str]:
    text = value if isinstance(value, str) else json.dumps(value, ensure_ascii=False, default=str)
    return [snippet for snippet in snippets if snippet in text][:20]


def _technical_label(key: str) -> str:
    return {
        "source": "数据来源与定位",
        "association": "关联方式与完整性",
        "input": "原始输入",
        "output": "原始输出",
        "cost": "耗时与成本",
        "record": "原始业务记录",
        "event": "原始运行事件",
        "taskEvent": "实现任务事件",
        "taskSource": "任务来源判定",
        "rawEvent": "运行事件摘要",
        "branch": "实现方案记录",
        "round": "轮次记录",
    }.get(key, key)


def _context_id(source: dict[str, Any], field_path: Any) -> str:
    return ":".join(str(item) for item in (source.get("table") or source.get("type"), source.get("recordId") or source.get("eventId") or "record", field_path or "$") )


def _completeness(notice: str | None, reads: list[dict[str, Any]], contexts: list[dict[str, Any]]) -> dict[str, Any]:
    missing = [item for item in contexts if item.get("completeness") in {"missing", "body-missing"}]
    unconfirmed = [item for item in reads if item.get("disposition") == "unconfirmed"]
    return {
        "state": "partial" if notice or missing or unconfirmed else "complete",
        "inspectorAvailable": notice is None,
        "sourceCount": len(reads),
        "contextCount": len(contexts),
        "missingContextCount": len(missing),
        "unconfirmedSourceCount": len(unconfirmed),
        "notice": notice,
        "redacted": True,
    }


def _suffix_int(value: Any) -> int | None:
    text = str(value or "")
    try:
        return int(text.rsplit(":", 1)[-1])
    except (TypeError, ValueError):
        return None


def _integer(value: Any) -> int | None:
    try:
        return int(value) if value is not None else None
    except (TypeError, ValueError):
        return None