script-build-and-framework-tmp/visualization/backend/app/source_lineage/comparison.py

from __future__ import annotations

import re
from copy import deepcopy
from typing import Any

from .resolver import SourceResolver, pointer_lookup
from .schema import SourceBinding, unresolved_binding


# These mappings are keyed by the stable business projection shape, never by
# fuzzy title matching.  Values point at the richer source modules built by the
# card-specific projectors.
_MODULE_MAP: dict[str, dict[str, list[str]]] = {
    "objective": {
        "block:direction": ["direction"],
        "block:inputs": ["inputs"],
    },
    "round-goal": {
        "block:goal": ["goal", "why"],
        "block:dependencies": ["dependencies"],
    },
    "implementation-plan": {
        "block:plan": ["mode", "routes", "revisions"],
        "block:basis": ["basis", "revisions"],
    },
    "implementation-task": {
        "section:scope": ["scope"],
        "section:method": ["method"],
        "section:goal": ["goal"],
        "section:constraints": ["constraints"],
        "section:avoid": ["avoid"],
        "section:deliverable": ["deliverable"],
        "section:action": ["action"],
        "section:request": ["request"],
        "section:result": ["result"],
    },
    "creative": {
        "block:basis": ["basis"],
        "block:process": ["process"],
        "block:available": ["available"],
        "block:changes": ["changes"],
        "block:candidate": ["candidate"],
        "block:uncertainty": ["uncertainty"],
        "block:record-note": ["record-note"],
    },
    "data-decision": {
        "block:0": ["sources"],
        "block:1": ["decision", "tradeoff"],
        "block:2": ["reason"],
    },
    "multipath-decision": {
        "block:0": ["strength"],
        "block:1": ["scope"],
        "block:2": ["comparison"],
        "block:3": ["comparison"],
        "block:4": ["reason"],
        "block:5": ["strength"],
    },
    "domain-facts": {
        "section:output": ["facts", "verification", "sources", "usage"],
        "section:assessment": [],
    },
    "candidate-output-fallback": {
        "section:output": ["output"],
        "section:assessment": ["assessment"],
    },
    "round-result": {
        "section:goal": ["completion"],
        "section:outcomes": ["dispositions"],
        "section:facts": ["domain"],
        "section:evaluation": ["review"],
        "section:changes": ["changes"],
        "section:next": ["next"],
    },
    "final-result": {
        "section:conclusion": ["conclusion"],
        "section:summary": ["summary"],
        "section:statistics": ["stats"],
        "section:artifact": ["scale", "artifact"],
        "section:version": ["version"],
    },
    "retrieval-agent": {
        "section:task": ["task"],
        "section:queries": ["queries"],
        "section:hits": ["raw"],
        "section:screening": ["screening"],
        "special:activities": ["activities"],
    },
    "retrieval-direct-group": {
        "section:purpose": ["purpose"],
        "section:sources": ["calls"],
        "section:result": ["results", "failures"],
        "special:activities": ["calls", "results", "failures"],
    },
    "retrieval-event": {
        "special:conditions": ["purpose"],
        "special:outcome": ["status"],
        "special:results": ["result", "results"],
        "special:completeness": ["status"],
    },
}


def business_only_projection(detail: dict[str, Any]) -> dict[str, Any]:
    """Keep exactly what the readable Inspector uses, without duplicating raw JSON."""
    hidden = {"technical", "changes"}
    return deepcopy({key: value for key, value in detail.items() if key not in hidden})


def align_business_projection(
    *,
    detail_ref: str,
    card_kind: str,
    business: dict[str, Any],
    source_modules: list[dict[str, Any]],
    resolver: SourceResolver,
    bundle: dict[str, Any],
) -> list[dict[str, Any]]:
    """Turn the readable Inspector projection into row-aligned source modules."""
    projection = business_only_projection(business)
    normalized = _normalize_business(projection, card_kind=card_kind)
    source_by_id = {str(module.get("id")): module for module in source_modules}
    aligned: list[dict[str, Any]] = []
    for module_index, module in enumerate(normalized, 1):
        semantic_key = str(module.pop("_semanticKey"))
        source_ids = _source_module_ids(card_kind, semantic_key, module_index)
        candidates = [source_by_id[source_id] for source_id in source_ids if source_id in source_by_id]
        if not candidates and card_kind == "implementation-task" and "historical-task" in source_by_id:
            candidates = [source_by_id["historical-task"]]
        rows: list[dict[str, Any]] = []
        module_bindings: list[SourceBinding] = []
        for row_index, row in enumerate(module.get("rows") or [], 1):
            found, business_value = pointer_lookup(
                projection, str(row.get("businessSelector") or "")
            )
            if card_kind == "domain-facts" and semantic_key == "section:assessment":
                branch_id = _branch_id(detail_ref)
                branch = next((item for item in bundle.get("branches") or [] if _integer(item.get("branch_id")) == branch_id), None)
                bindings = [resolver.binding_for_field({
                    "id": f"branch:{branch_id}:assessment",
                    "label": "实现者说明",
                    "value": (branch or {}).get("self_assessment"),
                    "source": {
                        "kind": "database",
                        "label": "script_build_branch",
                        "ref": detail_ref,
                        "fieldPath": "self_assessment",
                    },
                    "relation": "persisted-output",
                    "completeness": "complete" if branch and branch.get("self_assessment") else "partial",
                }, binding_id=f"{row['id']}:source:1", role="output")]
            elif semantic_key in {"special:question", "block:record-note"}:
                bindings = [resolver.binding_for_field({
                    "id": f"{module['id']}:display-text",
                    "label": "业务详情确定性文案",
                    "value": business_value if found else None,
                    "source": {
                        "kind": "calculation",
                        "label": "业务详情确定性说明",
                        "ref": f"calculation:{module['id']}:display-text",
                        "fieldPath": "value",
                    },
                    "relation": "persisted-output",
                    "completeness": "complete",
                }, binding_id=f"{row['id']}:source:1", role="status")]
            elif card_kind == "retrieval-event" and semantic_key in {
                "special:outcome",
                "special:completeness",
            }:
                event_id = _event_id(detail_ref)
                output_key = str(row.get("id") or "").rsplit(":", 1)[-1]
                path = "output.content" if semantic_key == "special:outcome" else ""
                binding = resolver.binding_for_event(
                    event_id or 0,
                    binding_id=f"{row['id']}:source:1",
                    path=path,
                    role="output" if semantic_key == "special:outcome" else "status",
                    availability=(
                        "returned-to-agent"
                        if semantic_key == "special:outcome"
                        else "produced-by-run"
                    ),
                )
                binding["transform"] = {
                    "kind": "parsed",
                    "operation": (
                        "retrieval-outcome-v1"
                        if semantic_key == "special:outcome"
                        else "event-body-completeness-v1"
                    ),
                    "outputKey": output_key,
                }
                binding["evidence"]["confidence"] = "deterministic"
                bindings = [binding]
            elif semantic_key == "special:activities":
                ref = ""
                if found and isinstance(business_value, dict):
                    ref = str(
                        business_value.get("detailRef")
                        or business_value.get("id")
                        or ""
                    )
                event_id = _event_id(ref)
                if event_id is not None:
                    binding = resolver.binding_for_event(
                        event_id,
                        binding_id=f"{row['id']}:source:1",
                        role="process",
                        availability="produced-by-run",
                    )
                    binding["transform"] = {
                        "kind": "parsed",
                        "operation": "activity-summary-v1",
                        "outputKey": str(row.get("id") or "activity"),
                    }
                    binding["evidence"]["confidence"] = "deterministic"
                    bindings = [binding]
                elif found and (
                    business_value == []
                    or str(row.get("businessSelector") or "")
                    == "/missingActivityNotice"
                ):
                    empty_binding = resolver.binding_for_field(
                        {
                            "id": f"{row['id']}:empty",
                            "label": "可见查询运行记录数量",
                            "value": business_value,
                            "source": {
                                "kind": "calculation",
                                "label": "取数 Agent 结构化查询记录确定性检查",
                                "ref": f"calculation:{row['id']}:empty",
                                "fieldPath": "value",
                            },
                            "relation": "persisted-output",
                            "completeness": "complete",
                        },
                        binding_id=f"{row['id']}:source:1",
                        role="status",
                    )
                    empty_source = resolver.sources.get(empty_binding["sourceId"])
                    if empty_source is not None:
                        empty_source["resultState"] = "empty"
                        empty_source.setdefault("locator", {})["resultCount"] = 0
                    bindings = [empty_binding]
                else:
                    bindings = []
            else:
                bindings = _bindings_for_row(
                    module_id=str(module["id"]),
                    row=row,
                    row_index=row_index,
                    candidates=candidates,
                    resolver=resolver,
                    business_value=business_value if found else None,
                    narrow_event_text=card_kind == "implementation-task",
                )
                if card_kind == "implementation-plan" and semantic_key == "block:plan":
                    revision = _implementation_plan_revision_binding(
                        row=row,
                        business_value=business_value if found else None,
                        candidates=candidates,
                        resolver=resolver,
                    )
                    if revision is not None:
                        bindings.append(revision)
            if not bindings:
                binding = unresolved_binding(
                    f"{module['id']}:row:{row_index}:unresolved",
                    reason=f"业务详情「{module['title']}」的这一项没有保存可安全定位的原始记录",
                    code="binding-map-missing",
                )
                resolver.add_unresolved_source(binding)
                bindings = [binding]
            _mark_row_transform(
                bindings,
                business_value if found else None,
                resolver,
                str(module["id"]),
            )
            row = {key: value for key, value in row.items() if not key.startswith("_")}
            row["bindingIds"] = [binding["id"] for binding in bindings]
            rows.append(row)
            module_bindings.extend(bindings)
        module["rows"] = rows
        module["bindings"] = _unique_bindings(module_bindings)
        module["runtimeRefs"] = list(dict.fromkeys(
            binding["sourceId"]
            for binding in module["bindings"]
            if binding["sourceId"].startswith("event:")
        ))
        aligned.append(module)
    return aligned


def _source_module_ids(card_kind: str, semantic_key: str, index: int) -> list[str]:
    explicit = _MODULE_MAP.get(card_kind, {}).get(semantic_key)
    if explicit is not None:
        return explicit
    if semantic_key.startswith("section:"):
        return [semantic_key.split(":", 1)[1]]
    if semantic_key.startswith("block:"):
        try:
            return [f"block-{int(semantic_key.split(':', 1)[1]) + 1}"]
        except ValueError:
            return [f"block-{index}"]
    return [semantic_key.split(":", 1)[-1]]


def _normalize_business(
    projection: dict[str, Any], *, card_kind: str = ""
) -> list[dict[str, Any]]:
    modules: list[dict[str, Any]] = []
    question = projection.get("question")
    if question not in (None, ""):
        modules.append(_module(
            "question",
            "这一步要回答什么",
            "/question",
            "special:question",
            [_row("question:value", None, "/question")],
            default_open=True,
        ))

    for index, section in enumerate(projection.get("businessSections") or []):
        if not isinstance(section, dict):
            continue
        section_id = str(section.get("id") or f"section-{index + 1}")
        path = f"/businessSections/{index}"
        rows = []
        if isinstance(section.get("items"), list):
            for item_index, item in enumerate(section.get("items") or []):
                label = str((item or {}).get("label") or f"第 {item_index + 1} 项") if isinstance(item, dict) else f"第 {item_index + 1} 项"
                item_path = f"{path}/items/{item_index}"
                rows.append(_row(f"{section_id}:item:{item_index + 1}", label, item_path, item_index=item_index))
        elif section.get("content") is not None:
            rows.append(_row(f"{section_id}:content", None, f"{path}/content"))
        semantic_id = _section_semantic_id(section_id)
        modules.append(_module(
            section_id,
            str(section.get("title") or f"业务模块 {index + 1}"),
            path,
            f"section:{semantic_id}",
            rows,
            presentation=str(section.get("presentation") or "text"),
            default_open=not bool(section.get("collapsible")) or bool(section.get("defaultOpen")),
        ))

    for index, block in enumerate(projection.get("blocks") or []):
        if not isinstance(block, dict):
            continue
        path = f"/blocks/{index}"
        semantic_id = _block_semantic_id(card_kind, block, index)
        module_id = semantic_id or f"block-{index + 1}"
        modules.append(_module(
            module_id,
            str(block.get("title") or "记录说明"),
            path,
            f"block:{semantic_id}" if semantic_id else f"block:{index}",
            _block_rows(module_id, path, block),
            presentation=str(block.get("presentation") or block.get("type") or "text"),
            default_open=not bool(block.get("collapsible")) or bool(block.get("defaultOpen")),
        ))

    conditions = projection.get("queryConditions")
    if projection.get("detailKind") == "retrieval-event" and isinstance(conditions, list):
        rows = [
            _row(
                f"conditions:item:{index + 1}",
                str(item.get("label") or item.get("key") or f"条件 {index + 1}"),
                f"/queryConditions/{index}/value",
                item_index=index,
            )
            for index, item in enumerate(conditions)
            if isinstance(item, dict)
        ]
        if not rows:
            rows = [_row("conditions:empty", None, "/queryConditions")]
        modules.append(_module("conditions", "查询条件", "/queryConditions", "special:conditions", rows))

    outcome = projection.get("outcome")
    if isinstance(outcome, dict):
        rows = []
        for key, label in (("message", "执行结果"), ("count", "返回数量"), ("errorKind", "失败类型")):
            if outcome.get(key) is not None:
                rows.append(_row(f"outcome:{key}", label, f"/outcome/{key}"))
        modules.append(_module("outcome", "查询结果", "/outcome", "special:outcome", rows))

    result_items = projection.get("items")
    if isinstance(result_items, list) and result_items:
        rows = [
            _row(
                f"results:item:{index + 1}",
                str((item or {}).get("title") or f"结果 {index + 1}") if isinstance(item, dict) else f"结果 {index + 1}",
                f"/items/{index}",
                item_index=index,
            )
            for index, item in enumerate(result_items)
        ]
        modules.append(_module("results", "返回结果", "/items", "special:results", rows))

    activities = projection.get("activities")
    if isinstance(activities, list):
        rows = [
            _row(
                f"activities:item:{index + 1}",
                str((item or {}).get("label") or f"运行记录 {index + 1}") if isinstance(item, dict) else f"运行记录 {index + 1}",
                f"/activities/{index}",
                item_index=index,
            )
            for index, item in enumerate(activities)
        ]
        if not rows:
            notice_path = "/missingActivityNotice" if projection.get("missingActivityNotice") else "/activities"
            rows = [_row("activities:empty", None, notice_path)]
        modules.append(_module("activities", "运行记录", "/activities", "special:activities", rows, presentation="process"))

    completeness = projection.get("completeness")
    if projection.get("detailKind") == "retrieval-event" and isinstance(completeness, dict):
        rows = []
        if completeness.get("bodyAvailable") is False:
            rows.append(_row("completeness:body", "完整返回", "/completeness/bodyAvailable"))
        if completeness.get("truncated"):
            rows.append(_row(
                "completeness:truncated",
                "截断说明",
                "/completeness/omittedCharacters" if completeness.get("omittedCharacters") is not None else "/completeness/truncated",
            ))
        if rows:
            modules.append(_module("completeness", "记录完整性", "/completeness", "special:completeness", rows, presentation="notice"))
    return modules


def _block_rows(module_id: str, path: str, block: dict[str, Any]) -> list[dict[str, Any]]:
    block_type = str(block.get("type") or "summary")
    if block_type in {"summary", "notice"}:
        return [_row(f"{module_id}:value", None, f"{path}/value")]
    if block_type == "items":
        return [
            _row(
                f"{module_id}:item:{index + 1}",
                str(
                    item.get("label")
                    or item.get("title")
                    or item.get("subject")
                    or item.get("summary")
                    or f"第 {index + 1} 项"
                ) if isinstance(item, dict) else f"第 {index + 1} 项",
                f"{path}/items/{index}",
                item_index=index,
            )
            for index, item in enumerate(block.get("items") or [])
        ]
    if block_type == "implementation-plan":
        rows = []
        if block.get("mode") is not None:
            rows.append(_row(f"{module_id}:mode", "规划方式", f"{path}/mode", binding_slot="mode:0"))
        for route_index, route in enumerate(block.get("routes") or []):
            if not isinstance(route, dict):
                continue
            route_number = route.get("pathIndex") or route_index + 1
            for key, label in (("target", "作用范围"), ("method", "实施方法"), ("action", "具体动作"), ("emphasis", "路线侧重"), ("pathType", "路线类型")):
                if route.get(key) is None:
                    continue
                source_key = {"pathIndex": "path_index", "pathType": "path_type"}.get(key, key)
                rows.append(_row(
                    f"{module_id}:route:{route_index + 1}:{key}",
                    f"路线 {route_number} · {label}",
                    f"{path}/routes/{route_index}/{key}",
                    binding_slot="routes:0",
                    source_suffix=f"/{route_index}/{source_key}",
                ))
        if block.get("reasoning") is not None:
            rows.append(_row(f"{module_id}:reasoning", "为什么这样规划", f"{path}/reasoning", binding_slot="mode:1"))
        return rows
    if block_type == "creative-process":
        rows = []
        for step_index, step in enumerate(block.get("steps") or []):
            if not isinstance(step, dict):
                continue
            number = step.get("stepIndex") or step_index + 1
            for key, label in (("action", "动作"), ("summary", "结果"), ("plan", "执行计划"), ("reasoning", "思考依据")):
                if step.get(key) is not None:
                    row = _row(f"{module_id}:step:{step_index + 1}:{key}", f"步骤 {number} · {label}", f"{path}/steps/{step_index}/{key}", item_index=step_index)
                    row["_fieldKey"] = key
                    rows.append(row)
        return rows
    if block_type == "evaluation-branches":
        rows = []
        for branch_index, branch in enumerate(block.get("branches") or []):
            if not isinstance(branch, dict):
                continue
            subject = str(branch.get("subject") or f"方案 {branch_index + 1}")
            for key, label in (("conclusion", "结论"), ("achievements", "已达成"), ("problems", "需关注"), ("evidence", "依据")):
                if branch.get(key) not in (None, [], ""):
                    rows.append(_row(f"{module_id}:branch:{branch_index + 1}:{key}", f"{subject} · {label}", f"{path}/branches/{branch_index}/{key}", item_index=branch_index))
        return rows
    if block_type == "evaluation-matrix":
        return [
            _row(f"{module_id}:row:{index + 1}", str((row or {}).get("criterion") or f"标准 {index + 1}"), f"{path}/rows/{index}", item_index=index)
            for index, row in enumerate(block.get("rows") or [])
        ]
    if block_type == "comparison":
        return [
            _row(f"{module_id}:row:{index + 1}", str((row or {}).get("candidate") or f"方案 {index + 1}"), f"{path}/rows/{index}", item_index=index)
            for index, row in enumerate(block.get("rows") or [])
        ]
    if block_type == "artifact":
        return [
            _row(f"{module_id}:artifact:{index + 1}", str((ref or {}).get("label") or f"产物 {index + 1}"), f"{path}/refs/{index}", item_index=index)
            for index, ref in enumerate(block.get("refs") or [])
        ]
    return [_row(f"{module_id}:record", None, path)]


def _bindings_for_row(
    *,
    module_id: str,
    row: dict[str, Any],
    row_index: int,
    candidates: list[dict[str, Any]],
    resolver: SourceResolver,
    business_value: Any,
    narrow_event_text: bool = False,
) -> list[SourceBinding]:
    slot = str(row.get("_bindingSlot") or "")
    bindings: list[SourceBinding] = []
    if slot:
        source_id, _, raw_index = slot.partition(":")
        source_module = next((module for module in candidates if str(module.get("id")) == source_id), None)
        source_bindings = list((source_module or {}).get("bindings") or [])
        try:
            binding = source_bindings[int(raw_index or 0)]
        except (IndexError, ValueError):
            binding = None
        if binding is not None:
            cloned = _clone_binding(
                binding,
                binding_id=f"{row['id']}:source:1",
                source_suffix=str(row.get("_sourceSuffix") or ""),
                resolver=resolver,
            )
            bindings = [
                _narrow_event_text_binding(cloned, business_value, resolver, module_id)
                if narrow_event_text
                else cloned
            ]
    if not bindings:
        item_index = row.get("_itemIndex")
        for source_module in candidates:
            items = ((source_module.get("business") or {}).get("items") or [])
            source_bindings = list(source_module.get("bindings") or [])
            if isinstance(item_index, int) and item_index < len(items):
                item = items[item_index] or {}
                item_bindings = list(item.get("bindings") or [])
                field_key = str(row.get("_fieldKey") or "")
                if field_key and item_bindings:
                    event_id = _event_id(item_bindings[0].get("sourceId"))
                    event_paths = {
                        "action": "input.content.action",
                        "summary": "input.content.thought_summary",
                        "plan": "input.content.plan",
                        "reasoning": "input.content.thought",
                    }
                    if event_id and field_key in event_paths:
                        binding = resolver.binding_for_event(
                            event_id,
                            binding_id=f"{row['id']}:source:1",
                            path=event_paths[field_key],
                            role="process",
                        )
                        found, selected = pointer_lookup(
                            (resolver.sources.get(binding["sourceId"]) or {}).get("rawRecord"),
                            str((binding.get("selector") or {}).get("path") or ""),
                        )
                        if not found or selected != business_value:
                            binding["transform"] = {
                                "kind": "parsed",
                                "operation": "creative-step-business-label-v1",
                                "outputKey": field_key,
                            }
                            binding["evidence"]["confidence"] = "deterministic"
                        return [binding]
                matched = _bindings_matching_value(
                    item.get("value"), item_bindings, business_value
                )
                bindings.extend(matched or item_bindings)
            elif (
                isinstance(item_index, int)
                and isinstance(((source_module.get("business") or {}).get("value")), list)
                and item_index < len(source_bindings)
            ):
                # The readable Inspector may split one source list into one row
                # per item while CardData keeps it as one aggregate module.
                # The persisted source order is explicit, so bind the same
                # index instead of repeating every source on every row.
                bindings.append(source_bindings[item_index])
            elif isinstance(item_index, int) and item_index < len(items):
                bindings.extend((items[item_index] or {}).get("bindings") or [])
        if not bindings:
            bindings = [
                binding
                for source_module in candidates
                for binding in source_module.get("bindings") or []
            ]
        physical_bindings = _unique_physical_bindings(bindings)
        usable_bindings = [
            binding
            for binding in physical_bindings
            if binding.get("resolution") != "missing"
        ]
        if usable_bindings:
            physical_bindings = usable_bindings
        bindings = [
            _narrow_event_text_binding(cloned, business_value, resolver, module_id)
            if narrow_event_text
            else cloned
            for index, binding in enumerate(physical_bindings, 1)
            for cloned in [
                _clone_binding(
                binding,
                binding_id=f"{row['id']}:source:{index}",
                source_suffix="",
                resolver=resolver,
                )
            ]
        ]
    return bindings


def _mark_row_transform(
    bindings: list[SourceBinding],
    business_value: Any,
    resolver: SourceResolver,
    module_id: str,
) -> None:
    """Never call a normalized/combined business row a direct projection."""
    mismatched: list[SourceBinding] = []
    for binding in bindings:
        if (binding.get("transform") or {}).get("kind") != "direct":
            continue
        source = resolver.sources.get(str(binding.get("sourceId") or "")) or {}
        selector = binding.get("selector") or {}
        selected: Any = None
        found = False
        if selector.get("kind") == "json-pointer":
            found, selected = pointer_lookup(
                source.get("rawRecord"), str(selector.get("path") or "")
            )
        elif selector.get("kind") == "whole-record":
            found, selected = True, source.get("rawRecord")
        elif selector.get("kind") == "text-span":
            selected = selector.get("exactText")
            found = True
        if found and selected != business_value:
            mismatched.append(binding)
    for binding in mismatched:
        binding["transform"] = (
            {
                "kind": "combined",
                "operation": f"business-row:{module_id}",
            }
            if len(bindings) > 1
            else {
                "kind": "parsed",
                "operation": "business-row-projection-v1",
                "outputKey": module_id,
            }
        )
        binding["evidence"]["confidence"] = "deterministic"


def _section_semantic_id(section_id: str) -> str:
    """Map task section instance ids to the stable business module ids."""
    if not section_id.startswith("task-"):
        return section_id
    match = re.match(r"^task-([a-z-]+)-\d+$", section_id)
    kind = match.group(1) if match else "notes"
    return {
        "scope": "scope",
        "method": "method",
        "goal": "goal",
        "materials": "constraints",
        "context": "constraints",
        "constraints": "constraints",
        "avoid": "avoid",
        "notes": "deliverable",
    }.get(kind, "deliverable")


def _block_semantic_id(
    card_kind: str, block: dict[str, Any], index: int
) -> str | None:
    """Return a stable semantic id for optional/reordered decision blocks."""
    explicit = str(block.get("semanticId") or "").strip()
    if explicit:
        return explicit
    block_type = str(block.get("type") or "").strip()
    title = str(block.get("title") or "").strip()
    by_title = ({
        "当前保存的创作方向": "direction",
        "最终方向": "direction",
        "根据以下信息做出的决策": "inputs",
    } if card_kind == "objective" else {
        "明确依据": "basis",
        "明确采用的数据取舍": "basis",
        "创作处理": "process",
        "可确认的上游信息": "available",
        "实际脚本改动": "changes",
        "实际产出": "candidate",
        "产出的候选表": "candidate",
        "存疑项": "uncertainty",
        "记录说明": "record-note",
    } if card_kind == "creative" else {
        "本轮目标构成": "goal",
        "根据以下信息做出的决策": "dependencies",
    } if card_kind == "round-goal" else {
        "本轮实施路线": "plan",
        "根据以下信息做出的决策": "basis",
    } if card_kind == "implementation-plan" else {})
    if title in by_title:
        return by_title[title]
    if card_kind == "creative" and block_type == "creative-process":
        return "process"
    if card_kind == "creative" and block_type == "notice":
        return "record-note"
    return None


def _narrow_event_text_binding(
    binding: SourceBinding,
    business_value: Any,
    resolver: SourceResolver,
    module_id: str = "",
) -> SourceBinding:
    """Narrow a complete implementation task binding to its exact visible section."""
    if not isinstance(business_value, str) or not business_value.strip():
        return binding
    event_id = _event_id(binding.get("sourceId"))
    selector = binding.get("selector") or {}
    if event_id is None or selector.get("kind") != "json-pointer":
        return binding
    path = str(selector.get("path") or "")
    source = resolver.sources.get(str(binding.get("sourceId") or "")) or {}
    found, raw_text = pointer_lookup(source.get("rawRecord"), path)
    if not found or not isinstance(raw_text, str):
        return binding
    exact_text = business_value if business_value in raw_text else _task_source_section(raw_text, module_id)
    if not exact_text:
        return binding
    narrowed = resolver.text_span_binding(
        event_id,
        binding_id=str(binding.get("id") or "implementation-task:span"),
        path=path.lstrip("/").replace("/", "."),
        exact_text=exact_text,
        role=str(binding.get("role") or "input"),
    )
    if exact_text != business_value:
        narrowed["transform"] = {
            "kind": "parsed",
            "operation": "implementation-task-section-normalization-v1",
            "outputKey": module_id,
        }
        narrowed["evidence"]["confidence"] = "deterministic"
    return narrowed


def _task_source_section(task: str, module_id: str) -> str | None:
    """Return the exact raw task section behind a normalized business block."""
    module_id = _section_semantic_id(module_id)
    line_keys = {
        "scope": ("target", "实现范围"),
        "method": ("method", "实施方法"),
    }
    if module_id in line_keys:
        keys = "|".join(re.escape(key) for key in line_keys[module_id])
        match = re.search(rf"(?im)^(?:{keys})\s*[:：]\s*(.+)$", task)
        return match.group(1).strip() if match else None
    headings = {
        "goal": (r"目标",),
        "constraints": (
            r"取证方向与约束",
            r"参考依据与约束",
            r"严格约束",
            r"约束(?:（[^）]*）)?",
        ),
        "avoid": (r"禁止事项", r"需要避免"),
        "deliverable": (r"预期交付", r"交付"),
    }
    constraint_pattern = "|".join(headings["constraints"])
    if module_id == "deliverable" and not re.search(
        rf"(?im)^(?:{'|'.join(headings['deliverable'])})\s*[:：]", task
    ):
        # Unlabelled task notes are the exact prefix before a later explicit
        # constraint section.  The readable Inspector may normalize terms
        # inside it, so retain the raw prefix as a parsed text span.
        match = re.search(
            rf"(?ims)^(.*?)(?=\n\s*\n(?:{constraint_pattern})\s*[:：]|\Z)",
            task,
        )
        return match.group(1).strip() if match else task.strip()
    names = headings.get(module_id)
    if not names:
        return None
    all_names = [name for values in headings.values() for name in values]
    start_pattern = "|".join(names)
    next_pattern = "|".join(all_names)
    match = re.search(
        rf"(?ims)^(?:{start_pattern})\s*[:：]\s*(.*?)(?=\n\s*\n(?:{next_pattern})\s*[:：]|\Z)",
        task,
    )
    return match.group(1).strip() if match else None


def _implementation_plan_revision_binding(
    *,
    row: dict[str, Any],
    business_value: Any,
    candidates: list[dict[str, Any]],
    resolver: SourceResolver,
) -> SourceBinding | None:
    """Bind the displayed final plan row to the revision that formed it.

    A round may call ``record_multipath_plan`` several times.  The database
    value is authoritative, while the latest revision containing the exact
    displayed value is the formation record.  This prevents round 4 from
    stopping at Events 4301/4369 and omitting the final Event 4434.
    """
    revision = next(
        (module for module in candidates if str(module.get("id")) == "revisions"),
        None,
    )
    event_ids = sorted(
        {
            event_id
            for binding in (revision or {}).get("bindings") or []
            for event_id in [_event_id(binding.get("sourceId"))]
            if event_id is not None
        },
        reverse=True,
    )
    row_id = str(row.get("id") or "")
    path = ""
    if row_id.endswith(":mode"):
        path = "input.content.mode"
    elif row_id.endswith(":reasoning"):
        path = "input.content.note"
    else:
        match = re.search(r":route:(\d+):([A-Za-z]+)$", row_id)
        if match:
            route_index = int(match.group(1)) - 1
            key = {"pathType": "path_type"}.get(match.group(2), match.group(2))
            path = f"input.content.paths.{route_index}.{key}"
    if not path:
        return None
    fallback_event: int | None = None
    for event_id in event_ids:
        event = resolver.event_details.get(event_id) or {}
        found, selected = pointer_lookup(event, "/" + path.replace(".", "/"))
        if not found:
            continue
        fallback_event = fallback_event or event_id
        if selected != business_value:
            continue
        return resolver.binding_for_event(
            event_id,
            binding_id=f"{row_id}:revision",
            path=path,
            role="process",
            availability="produced-by-run",
        )
    # The readable mode can normalize a one-route saved plan to "单路".
    # Retain the latest concrete revision but mark that normalization openly.
    if row_id.endswith(":mode") and fallback_event:
        binding = resolver.binding_for_event(
            fallback_event,
            binding_id=f"{row_id}:revision",
            path=path,
            role="process",
            availability="produced-by-run",
        )
        binding["transform"] = {
            "kind": "parsed",
            "operation": "single-route-mode-normalization-v1",
            "outputKey": "mode",
        }
        binding["evidence"]["confidence"] = "deterministic"
        return binding
    return None


def _bindings_matching_value(
    source_value: Any,
    bindings: list[SourceBinding],
    business_value: Any,
) -> list[SourceBinding]:
    source_leaves = _leaf_texts(source_value)
    wanted = set(_leaf_texts(business_value))
    if not wanted or len(source_leaves) != len(bindings):
        return []
    return [
        binding
        for leaf, binding in zip(source_leaves, bindings)
        if leaf in wanted
    ]


def _leaf_texts(value: Any) -> list[str]:
    if isinstance(value, str):
        return [value.strip()] if value.strip() else []
    if isinstance(value, dict):
        result: list[str] = []
        for child in value.values():
            result.extend(_leaf_texts(child))
        return result
    if isinstance(value, list):
        result = []
        for child in value:
            result.extend(_leaf_texts(child))
        return result
    return [str(value)] if value is not None else []


def _event_id(source_id: Any) -> int | None:
    value = str(source_id or "")
    if not value.startswith("event:"):
        return None
    try:
        return int(value.split(":", 1)[1])
    except ValueError:
        return None


def _clone_binding(
    binding: SourceBinding,
    *,
    binding_id: str,
    source_suffix: str,
    resolver: SourceResolver,
) -> SourceBinding:
    cloned: SourceBinding = deepcopy(binding)
    cloned["id"] = binding_id
    selector = cloned.get("selector") or {}
    if source_suffix and selector.get("kind") == "json-pointer":
        pointer = f"{str(selector.get('path') or '').rstrip('/')}{source_suffix}"
        source = resolver.sources.get(cloned["sourceId"]) or {}
        found, selected = pointer_lookup(source.get("rawRecord"), pointer)
        if found:
            cloned["selector"] = {"kind": "json-pointer", "path": pointer}
            source.setdefault("selectedValues", []).append({
                "bindingId": binding_id,
                "fieldId": binding_id,
                "label": binding_id,
                "path": pointer,
                "value": selected,
            })
        else:
            cloned["selector"] = {
                "kind": "unresolved",
                "reason": f"原始记录中无法定位字段 {pointer}",
                "code": "field-path-mismatch",
            }
            cloned["resolution"] = "partial"
            cloned["evidence"]["confidence"] = "unconfirmed"
    return cloned


def _module(
    module_id: str,
    title: str,
    business_path: str,
    semantic_key: str,
    rows: list[dict[str, Any]],
    *,
    presentation: str = "text",
    default_open: bool = True,
) -> dict[str, Any]:
    return {
        "id": module_id,
        "title": title,
        "businessPath": business_path,
        "presentation": presentation,
        "rows": rows,
        "defaultOpen": default_open,
        "_semanticKey": semantic_key,
    }


def _row(
    row_id: str,
    label: str | None,
    business_selector: str,
    *,
    item_index: int | None = None,
    binding_slot: str | None = None,
    source_suffix: str | None = None,
) -> dict[str, Any]:
    row: dict[str, Any] = {
        "id": row_id,
        "label": label,
        "businessSelector": business_selector,
    }
    if item_index is not None:
        row["_itemIndex"] = item_index
    if binding_slot is not None:
        row["_bindingSlot"] = binding_slot
    if source_suffix is not None:
        row["_sourceSuffix"] = source_suffix
    return row


def _unique_bindings(bindings: list[SourceBinding]) -> list[SourceBinding]:
    result: list[SourceBinding] = []
    seen: set[str] = set()
    for binding in bindings:
        key = str(binding.get("id") or "")
        if not key or key in seen:
            continue
        seen.add(key)
        result.append(binding)
    return result


def _unique_physical_bindings(bindings: list[SourceBinding]) -> list[SourceBinding]:
    result: list[SourceBinding] = []
    seen: set[str] = set()
    for binding in bindings:
        selector = binding.get("selector") or {}
        key = "|".join((
            str(binding.get("sourceId") or ""),
            str(selector.get("kind") or ""),
            str(selector.get("path") or selector.get("exactText") or selector.get("reason") or selector),
            str(binding.get("role") or ""),
        ))
        if not key or key in seen:
            continue
        seen.add(key)
        result.append(binding)
    return result


def _branch_id(value: str) -> int | None:
    import re
    match = re.search(r":branch:(\d+)", value)
    return _integer(match.group(1)) if match else None


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