script-build-and-framework-tmp/visualization-old/backend/app/decision_projection.py

from __future__ import annotations

import re
from typing import Any, Iterable, Mapping

from .business_detail_text import business_text


_SUCCESS_STATUSES = {"ok", "success", "succeeded", "completed", "complete"}
_DECISION_BASIS_TITLE = "根据以下信息做出的决策"
_RELATIONS = {
    "direct-read": "read-by-actor",
    "read-by-actor": "read-by-actor",
    "returned-report": "returned-to-actor",
    "returned-to-actor": "returned-to-actor",
    "persisted-record": "persisted-before-decision",
    "persisted-before-decision": "persisted-before-decision",
    "standing-constraint": "standing-constraint",
}


class AgentDecisionProjector:
    """Create the V8 decision union from already-associated, plain records.

    The projector deliberately does not scan event timelines or infer causal
    links.  Callers must first establish ownership/association, then pass those
    facts here.  In particular an observed read remains a read unless its input
    explicitly records ``decisionUse=explicit-basis``.
    """

    def direction(self, payload: Mapping[str, Any]) -> dict[str, Any]:
        subtype = _machine(payload.get("subtype")) or "direction"
        actor = _actor(payload.get("actor"), default_role="main", default_label="主 Agent")
        decision_items = _strings(
            payload.get("decisionItems") or payload.get("decisions")
        )
        reasoning = _text(payload.get("reasoning") or payload.get("explicitReasoning"))
        constraints = _strings(payload.get("constraints"))
        inputs = _inputs(payload.get("inputs"))
        primary = decision_items[0] if decision_items else _text(payload.get("decision"))
        secondary = []
        if inputs:
            secondary.append(_card_line("inputs", _DECISION_BASIS_TITLE, _input_labels(inputs)))
        if constraints:
            secondary.append(_card_line("constraints", "关键约束", _join(constraints, 2)))
        if reasoning:
            secondary.append(_card_line("reasoning", "决策理由", reasoning))
        body = {
            "type": "direction",
            "decisionItems": decision_items,
            "reasoning": reasoning,
            "constraints": constraints,
        }
        if subtype == "implementation-plan":
            implementation_plan = _implementation_plan(payload.get("implementationPlan"), reasoning)
            body["implementationPlan"] = implementation_plan
            blocks = [
                _implementation_plan_block(implementation_plan),
                _items_block(_DECISION_BASIS_TITLE, [_input_business_item(item) for item in inputs], visual_role="evidence"),
                _items_block("关键约束", constraints),
            ]
        elif subtype == "objective" and _source_text(payload.get("sourceDocument")):
            source_document = _source_text(payload.get("sourceDocument"))
            body["sourceDocument"] = source_document
            blocks = [
                _source_document_block(
                    "当前保存的创作方向",
                    source_document,
                ),
                _items_block(
                    _DECISION_BASIS_TITLE,
                    [_input_business_item(item) for item in inputs],
                    visual_role="evidence",
                ),
            ]
        elif subtype == "round-goal" and _source_text(payload.get("sourceDocument")):
            source_document = _source_text(payload.get("sourceDocument"))
            body["sourceDocument"] = source_document
            blocks = [
                _source_document_block(
                    "本轮目标构成",
                    source_document,
                ),
                _items_block(
                    _DECISION_BASIS_TITLE,
                    [_input_business_item(item) for item in inputs],
                    visual_role="evidence",
                ),
                _summary_block("目标来源", reasoning, visual_role="reason"),
            ]
        else:
            blocks = [
                _decision_items_block("最终方向", decision_items or ([primary] if primary else []), visual_role="key-result"),
                _items_block(_DECISION_BASIS_TITLE, [_input_business_item(item) for item in inputs], visual_role="evidence"),
                _items_block("关键约束", constraints),
                _summary_block("明确理由", reasoning, visual_role="reason"),
            ]
        return _decision(
            payload,
            decision_type="direction",
            subtype=subtype,
            actor=actor,
            authority="final",
            question=_text(payload.get("question")) or _direction_question(subtype),
            inputs=inputs,
            body=body,
            primary_label=_text(payload.get("primaryLabel")) or "最终方向",
            primary=primary,
            secondary=secondary,
            blocks=blocks,
        )

    def data_tradeoff(self, row: Mapping[str, Any]) -> dict[str, Any]:
        inputs = _inputs(row.get("inputs") or _source_inputs(row.get("sources")))
        decision_text = _text(row.get("decision"))
        reasoning = _text(row.get("reasoning"))
        scope_items = _strings(row.get("scopeItems"))
        body = {
            "type": "tradeoff",
            "scopeItems": scope_items,
            "decision": decision_text,
            "selected": _strings(row.get("selected")),
            "combined": _strings(row.get("combined")),
            "parked": _strings(row.get("parked")),
            "rejected": _strings(row.get("rejected")),
            "reasoning": reasoning,
        }
        secondary = []
        if inputs:
            secondary.append(_card_line("scope", "参与判断", f"{len(inputs)} 类信息"))
        if reasoning:
            secondary.append(_card_line("reasoning", "理由", reasoning))
        return _decision(
            row,
            decision_type="tradeoff",
            subtype="data-tradeoff",
            actor=_actor(
                row.get("actor"),
                default_role="implementer",
                default_label="实现 Agent",
            ),
            authority="implementation-scope",
            question=_text(row.get("question")) or "哪些信息用于当前实现方案？",
            inputs=inputs,
            body=body,
            primary_label="取舍结论",
            primary=decision_text,
            secondary=secondary,
            blocks=[
                _items_block("取舍对象", scope_items),
                _items_block("参与判断", [_input_business_item(item) for item in inputs], visual_role="evidence", collapsible=True),
                _summary_block("明确取舍", decision_text, visual_role="key-result"),
                _summary_block("理由", reasoning, visual_role="reason"),
            ],
        )

    def multipath_tradeoff(self, row: Mapping[str, Any]) -> dict[str, Any]:
        branch_ids = _unique_values(row.get("branchIds") or row.get("branch_ids"))
        inputs = _inputs(row.get("inputs"))
        decision_text = _text(row.get("decision"))
        reasoning = _text(row.get("reasoning"))
        selected = _strings(row.get("selected"))
        rejected = _strings(row.get("rejected"))
        body = {
            "type": "tradeoff",
            "scopeItems": _strings(row.get("scopeItems")),
            "decision": decision_text,
            "selected": selected,
            "combined": _strings(row.get("combined")),
            "parked": _strings(row.get("parked")),
            "rejected": rejected,
            "reasoning": reasoning,
            # Identity is useful to flow wiring but is not rendered as business copy.
            "branchIds": branch_ids,
            "reviewComparison": row.get("reviewComparison"),
        }
        secondary = []
        if branch_ids:
            secondary.append(
                _card_line("scope", "取舍范围", f"比较 {len(branch_ids)} 个候选方案")
            )
        if reasoning:
            secondary.append(_card_line("reasoning", "决策理由", reasoning))
        if row.get("reviewComparison"):
            secondary.append(_card_line("review", "评审对照", "可在详情中查看"))
        return _decision(
            row,
            decision_type="tradeoff",
            subtype="multipath-tradeoff",
            actor=_actor(row.get("actor"), default_role="main", default_label="主 Agent"),
            authority="final",
            question=_text(row.get("question")) or "多个候选方案最终如何处理？",
            inputs=inputs,
            body=body,
            primary_label="最终决定",
            primary=decision_text,
            secondary=secondary,
            blocks=[
                _summary_block(
                    "取舍对象",
                    f"{len(branch_ids)} 个候选方案" if branch_ids else None,
                ),
                _items_block("参与判断", [_input_business_item(item) for item in inputs], visual_role="evidence", collapsible=True),
                _summary_block("最终决定", decision_text, visual_role="key-result"),
                _summary_block("决策理由", reasoning, visual_role="reason"),
                _comparison_block(row.get("reviewComparison")),
            ],
        )

    def evaluation(self, payload: Mapping[str, Any]) -> dict[str, Any]:
        subtype = _machine(payload.get("subtype")) or "overall-evaluation"
        is_multipath = subtype == "multipath-evaluation"
        inputs = _inputs(payload.get("inputs"))
        conclusion = _text(payload.get("conclusion"))
        recommendation = _text(payload.get("recommendation"))
        subjects = _strings(payload.get("subjects"))
        criteria = _strings(payload.get("criteria"))
        item_conclusions = _business_dicts(payload.get("itemConclusions"))
        achievements = _strings(payload.get("achievements"))
        problems = _business_dicts(payload.get("problems"))
        branch_evaluations = _mapping_list(payload.get("branchEvaluations"))
        comparison_rows = _mapping_list(payload.get("comparisonRows"))
        body = {
            "type": "evaluation",
            "subjects": subjects,
            "criteria": criteria,
            "itemConclusions": item_conclusions,
            "achievements": achievements,
            "problems": problems,
            "branchEvaluations": branch_evaluations,
            "comparisonRows": comparison_rows,
            "conclusion": conclusion,
            "recommendation": recommendation,
            "nextGoal": _text(payload.get("nextGoal")),
        }
        secondary = []
        if achievements:
            secondary.append(_card_line("achievement", "已达成", _join(achievements, 2)))
        if problems:
            secondary.append(
                _card_line("problems", "需要关注", _problem_summary(problems), "warning")
            )
        if recommendation:
            secondary.append(_card_line("recommendation", "评审建议", recommendation))
        primary = conclusion or _item_conclusion_summary(item_conclusions) or recommendation
        if is_multipath and branch_evaluations:
            detail_blocks = [
                _evaluation_branches_block(branch_evaluations),
                _evaluation_matrix_block(comparison_rows) or _items_block("评审标准", criteria),
                _summary_block("评审建议", recommendation),
            ]
        else:
            detail_blocks = [
                _items_block("评审对象", subjects),
                _items_block("评审标准", criteria),
                _items_block("逐项结论", item_conclusions),
                _items_block("已达成", achievements),
                _items_block("问题", problems),
                _summary_block("总结论", conclusion, visual_role="key-result"),
                _summary_block("评审建议", recommendation),
            ]
        return _decision(
            payload,
            decision_type="evaluation",
            subtype=subtype,
            actor=_actor(
                payload.get("actor"),
                default_role=("multipath-evaluator" if is_multipath else "overall-evaluator"),
                default_label=("多方案评审 Agent" if is_multipath else "整体评审 Agent"),
            ),
            authority="recommendation",
            question=_text(payload.get("question"))
            or ("哪些候选方案更值得采用？" if is_multipath else "当前主脚本是否达到目标？"),
            inputs=inputs,
            body=body,
            primary_label="评审结论",
            primary=primary,
            secondary=secondary,
            blocks=detail_blocks,
        )

    def creative(self, payload: Mapping[str, Any]) -> dict[str, Any] | None:
        """Project an implementer creative decision only from a safe event link.

        An artifact reference may be attached to a real creative record, but it
        is never sufficient to create that record.  This avoids inferring the
        implementer's plan from the eventual candidate script.
        """

        if payload.get("safeLink") is not True:
            return None
        event = payload.get("event")
        event = event if isinstance(event, Mapping) else payload
        if not _is_successful_think_and_plan(event):
            return None
        raw_input = event.get("inputData") or event.get("input")
        if not isinstance(raw_input, Mapping):
            return None
        thought = _text(
            payload.get("reasoning")
            or raw_input.get("thought")
            or raw_input.get("thought_summary")
        )
        steps = _creative_steps(payload.get("steps"))
        actions = _strings(payload.get("actions"))
        if not steps:
            fallback_action = _text(raw_input.get("action"))
            fallback_plan = _source_text(raw_input.get("plan"))
            if fallback_action or fallback_plan or thought:
                steps = [{
                    "stepIndex": 1,
                    "action": fallback_action or "创作处理",
                    "summary": _text(raw_input.get("thought_summary")),
                    "plan": fallback_plan,
                    "reasoning": _source_text(raw_input.get("thought")),
                }]
        if not actions:
            actions = [str(step["action"]) for step in steps if step.get("action")]
        if not any((thought, actions, steps)):
            return None
        task = _text(payload.get("task"))
        explicit_basis = _strings(payload.get("basis"))
        uncertainties = _strings(payload.get("uncertainties"))
        body = {
            "type": "creative",
            "task": task,
            "basis": explicit_basis,
            "actions": actions,
            "steps": steps,
            "reasoning": thought,
            "output": _text(payload.get("output")),
            "uncertainties": uncertainties,
        }
        secondary = []
        projected_payload = dict(payload)
        projected_payload.setdefault("id", f"creative-event:{event.get('id')}")
        return _decision(
            projected_payload,
            decision_type="creative",
            subtype="creative-handling",
            actor=_actor(
                payload.get("actor"),
                default_role="implementer",
                default_label="实现 Agent",
            ),
            authority="implementation-scope",
            question=_text(payload.get("question")) or "当前实现方案如何形成候选产出？",
            inputs=_inputs(payload.get("inputs")),
            body=body,
            primary_label="创作处理",
            primary=actions[0] if actions else thought,
            secondary=secondary,
            blocks=[
                _items_block("明确依据", explicit_basis, visual_role="evidence", collapsible=True),
                _creative_process_block(steps) or _items_block("创作处理", actions),
                _summary_block("产出的候选表", body["output"], visual_role="key-result"),
                _items_block("存疑项", uncertainties, visual_role="notice"),
            ],
        )


def _decision(
    payload: Mapping[str, Any],
    *,
    decision_type: str,
    subtype: str,
    actor: dict[str, str],
    authority: str,
    question: str,
    inputs: list[dict[str, Any]],
    body: dict[str, Any],
    primary_label: str,
    primary: str | None,
    secondary: list[dict[str, Any]],
    blocks: list[dict[str, Any] | None],
) -> dict[str, Any]:
    completeness = _completeness(payload.get("completeness"), primary)
    notices = _notices(payload.get("notices"))
    if not primary and not any(item.get("code") == "record-missing" for item in notices):
        notices.append({"code": "record-missing", "message": "没有找到结构化决策结论。"})
    decision_id = _machine(payload.get("id") or payload.get("detailRef")) or f"{subtype}:unknown"
    detail_ref = _machine(payload.get("detailRef")) or decision_id
    result = {
        "id": decision_id,
        "semanticKind": "decision",
        "decisionType": decision_type,
        "subtype": subtype,
        "actor": actor,
        "authority": authority,
        "question": question,
        "inputs": inputs,
        "body": body,
        "card": {
            "primary": {
                "label": primary_label,
                "value": primary or "决策结论未记录",
            },
            "secondary": [item for item in secondary if item.get("value")][:3],
        },
        "detailRef": detail_ref,
        "promptRef": _machine(payload.get("promptRef")),
        "artifactRef": _machine(payload.get("artifactRef")),
        "completeness": completeness,
        "notices": notices,
        "detail": {
            "detailKind": "agent-decision",
            "decisionType": decision_type,
            "actor": actor,
            "authority": authority,
            "question": question,
            "blocks": [block for block in blocks if block],
        },
    }
    return result


def _inputs(value: Any) -> list[dict[str, Any]]:
    if not isinstance(value, Iterable) or isinstance(value, (str, bytes, Mapping)):
        return []
    result: list[dict[str, Any]] = []
    for raw in value:
        if not isinstance(raw, Mapping):
            continue
        observed = _RELATIONS.get(
            _text(raw.get("observedRelation") or raw.get("relation")) or "",
            "persisted-before-decision",
        )
        explicit_use = _machine(raw.get("decisionUse"))
        decision_use = explicit_use if explicit_use in {"explicit-basis", "not-recorded"} else "not-recorded"
        if not (raw.get("observedRelation") or raw.get("relation")):
            continue
        label = _text(raw.get("label") or raw.get("name") or raw.get("type"))
        if not label:
            continue
        result.append(
            {
                "label": label,
                "summary": _text(raw.get("summary")),
                "observedRelation": observed,
                "decisionUse": decision_use,
                "detailRef": _text(raw.get("detailRef")),
            }
        )
    return result


def _source_inputs(value: Any) -> list[dict[str, Any]]:
    if not isinstance(value, list):
        return []
    result = []
    for index, item in enumerate(value, 1):
        if isinstance(item, Mapping):
            label = _text(
                item.get("label")
                or item.get("data_type")
                or item.get("source_type")
                or item.get("type")
            )
            summary = _text(
                item.get("summary")
                or item.get("data_content")
                or item.get("title")
                or item.get("name")
            )
            explicit_use = item.get("decisionUse")
        else:
            label, summary, explicit_use = _text(item), None, None
        result.append(
            {
                "label": label or f"信息来源 {index}",
                "summary": summary,
                "observedRelation": "persisted-before-decision",
                "decisionUse": explicit_use or "not-recorded",
            }
        )
    return result


def _actor(value: Any, *, default_role: str, default_label: str) -> dict[str, str]:
    if isinstance(value, Mapping):
        return {
            "role": _machine(value.get("role")) or default_role,
            "label": _text(value.get("label")) or default_label,
        }
    return {"role": default_role, "label": default_label}


def _summary_block(
    title: str,
    value: Any,
    *,
    visual_role: str = "section",
    presentation: str = "prose",
) -> dict[str, Any] | None:
    text = _text(value)
    return {
        "type": "summary",
        "title": title,
        "value": text,
        "visualRole": visual_role,
        "presentation": presentation,
    } if text else None


def _source_document_block(title: str, value: Any) -> dict[str, Any] | None:
    text = _source_text(value)
    return {
        "type": "summary",
        "title": title,
        "value": text,
        "visualRole": "key-result",
        "presentation": "document",
    } if text else None


def _items_block(
    title: str,
    items: Any,
    *,
    visual_role: str = "section",
    presentation: str = "list",
    collapsible: bool = False,
    default_open: bool = False,
) -> dict[str, Any] | None:
    if not isinstance(items, list) or not items:
        return None
    return {
        "type": "items",
        "title": title,
        "items": items,
        "visualRole": visual_role,
        "presentation": presentation,
        "collapsible": collapsible,
        "defaultOpen": default_open,
    }


def _creative_process_block(steps: list[dict[str, Any]]) -> dict[str, Any] | None:
    if not steps:
        return None
    return {
        "type": "creative-process",
        "title": "创作处理",
        "steps": steps,
        "visualRole": "section",
        "presentation": "process",
    }


def _evaluation_branches_block(branches: list[dict[str, Any]]) -> dict[str, Any] | None:
    if not branches:
        return None
    return {
        "type": "evaluation-branches",
        "title": "逐方案评审",
        "branches": branches,
        "visualRole": "section",
        "presentation": "grouped-review",
    }


def _evaluation_matrix_block(rows: list[dict[str, Any]]) -> dict[str, Any] | None:
    if not rows:
        return None
    return {
        "type": "evaluation-matrix",
        "title": "评审标准与方案对比",
        "rows": rows,
        "visualRole": "section",
        "presentation": "matrix",
    }


def _decision_items_block(
    title: str,
    items: list[str],
    *,
    visual_role: str = "section",
) -> dict[str, Any] | None:
    if not items:
        return None
    if len(items) == 1:
        return _summary_block(title, items[0], visual_role=visual_role)
    return _items_block(title, items, visual_role=visual_role)


def _implementation_plan(value: Any, reasoning: str | None) -> dict[str, Any]:
    plan = value if isinstance(value, Mapping) else {}
    routes = plan.get("routes") if isinstance(plan.get("routes"), list) else []
    return {
        "mode": _text(plan.get("mode")) or _text(plan.get("rawMode")),
        "reasoning": _text(plan.get("reasoning")) or reasoning,
        "routes": [
            {
                "pathIndex": route.get("pathIndex") or index,
                "pathType": _business_text(route.get("pathType")),
                "action": _business_text(route.get("action")),
                "target": _business_text(route.get("target")),
                "method": _business_text(route.get("method")),
                "emphasis": _business_text(route.get("emphasis")),
            }
            for index, route in enumerate(routes, 1)
            if isinstance(route, Mapping)
        ],
    }


def _implementation_plan_block(plan: Mapping[str, Any]) -> dict[str, Any] | None:
    routes = plan.get("routes") if isinstance(plan.get("routes"), list) else []
    if not routes and not plan.get("reasoning"):
        return None
    return {
        "type": "implementation-plan",
        "title": "本轮实施路线",
        "visualRole": "section",
        "presentation": "plan",
        "mode": plan.get("mode"),
        "routes": routes,
        "reasoning": plan.get("reasoning"),
    }


def _item_conclusion_summary(items: list[dict[str, Any]]) -> str | None:
    summaries = []
    for item in items[:3]:
        subject = _text(item.get("subject") or item.get("label"))
        conclusion = _text(item.get("conclusion") or item.get("value"))
        if subject and conclusion:
            summaries.append(f"{subject}：{conclusion}")
    if not summaries:
        return None
    suffix = f"；另有 {len(items) - 3} 项" if len(items) > 3 else ""
    return "；".join(summaries) + suffix


def _comparison_block(value: Any) -> dict[str, Any] | None:
    if not isinstance(value, list) or not value:
        return None
    return {
        "type": "comparison",
        "title": "评审建议与最终决定",
        "visualRole": "section",
        "presentation": "comparison",
        "rows": _business_value(value),
    }


def _input_business_item(item: Mapping[str, Any]) -> dict[str, str]:
    result = {"label": str(item.get("label") or "信息")}
    summary = _business_text(item.get("summary"))
    if summary:
        result["value"] = summary
    relation = {
        "read-by-actor": "决策前直接读取",
        "returned-to-actor": "评审返回给决策者",
        "persisted-before-decision": "此前已保存",
        "standing-constraint": "持续约束",
    }.get(str(item.get("observedRelation") or ""))
    use = (
        "本轮必须遵守"
        if item.get("observedRelation") == "standing-constraint"
        else "明确作为决策依据"
        if item.get("decisionUse") == "explicit-basis"
        else "是否采用未记录"
    )
    result["note"] = "；".join(part for part in (relation, use) if part)
    return result


def _business_text(value: Any) -> str | None:
    text = business_text(value)
    return text or None


def _card_line(key: str, label: str, value: Any, tone: str | None = None) -> dict[str, Any]:
    line = {"key": key, "label": label, "value": _preview(value, 160)}
    if tone:
        line["tone"] = tone
    return line


def _notices(value: Any) -> list[dict[str, str]]:
    if not isinstance(value, list):
        return []
    result = []
    for item in value:
        if isinstance(item, Mapping):
            code = _machine(item.get("code"))
            message = _text(item.get("message"))
        else:
            code, message = "notice", _text(item)
        if code and message:
            result.append({"code": code, "message": message})
    return result


def _business_dicts(value: Any) -> list[Any]:
    if not isinstance(value, list):
        return []
    result: list[Any] = []
    for item in value:
        if isinstance(item, Mapping):
            cleaned = {
                key: _business_value(item[key])
                for key in (
                    "label",
                    "subject",
                    "conclusion",
                    "summary",
                    "severity",
                    "recommendation",
                    "achievements",
                    "problems",
                )
                if item.get(key) not in (None, "")
            }
            if cleaned:
                result.append(cleaned)
        elif text := _text(item):
            result.append(text)
    return result


def _mapping_list(value: Any) -> list[dict[str, Any]]:
    if not isinstance(value, list):
        return []
    return [
        _business_value(dict(item))
        for item in value
        if isinstance(item, Mapping)
    ]


def _creative_steps(value: Any) -> list[dict[str, Any]]:
    if not isinstance(value, list):
        return []
    result: list[dict[str, Any]] = []
    for fallback_index, item in enumerate(value, start=1):
        if not isinstance(item, Mapping):
            continue
        action = _text(item.get("action"))
        plan = _source_text(item.get("plan"))
        summary = _text(item.get("summary"))
        reasoning = _source_text(item.get("reasoning"))
        if not any((action, plan, summary, reasoning)):
            continue
        result.append(
            {
                "stepIndex": int(item.get("stepIndex") or fallback_index),
                "action": action or f"步骤 {fallback_index}",
                "summary": summary,
                "plan": plan,
                "reasoning": reasoning,
                "eventRef": _text(item.get("eventRef")),
            }
        )
    return result


def _problem_summary(problems: list[Any]) -> str:
    values = []
    for item in problems[:2]:
        if isinstance(item, Mapping):
            values.append(_text(item.get("summary") or item.get("label")) or "待处理问题")
        else:
            values.append(str(item))
    return _join(values, 2)


def _input_labels(inputs: list[dict[str, Any]]) -> str:
    return _join([str(item["label"]) for item in inputs], 3)


def _direction_question(subtype: str) -> str:
    return {
        "objective": "这次构建要形成什么创作方向？",
        "round-goal": "本轮优先解决什么问题？",
        "implementation-plan": "本轮应如何组织实现方案？",
    }.get(subtype, "这一步确定了什么方向？")


def _is_successful_think_and_plan(event: Mapping[str, Any]) -> bool:
    return (
        str(event.get("event_type") or "") == "tool_call"
        and str(event.get("event_name") or "") == "think_and_plan"
        and str(event.get("agent_role") or "")
        in {"script_implementer", "implementer"}
        and str(event.get("status") or "").lower() in _SUCCESS_STATUSES
    )


def _completeness(value: Any, primary: str | None) -> str:
    explicit = _machine(value)
    if explicit in {"complete", "partial", "missing"}:
        return explicit
    return "complete" if primary else "missing"


def _strings(value: Any) -> list[str]:
    if not isinstance(value, list):
        return []
    return [text for item in value if (text := _text(item))]


def _unique_values(value: Any) -> list[Any]:
    if not isinstance(value, list):
        return []
    result = []
    for item in value:
        if item not in result:
            result.append(item)
    return result


def _join(values: list[str], limit: int) -> str:
    kept = [value for value in values if value][:limit]
    suffix = f" · 另有 {len(values) - limit} 项" if len(values) > limit else ""
    return " · ".join(kept) + suffix


def _preview(value: Any, limit: int) -> str:
    text = _text(value) or ""
    if len(text) <= limit:
        return text
    cut = text[:limit]
    boundary = max(cut.rfind(mark) for mark in "。！？；")
    if boundary >= max(24, limit // 2):
        return cut[: boundary + 1]
    return cut.rstrip() + "…"


def _text(value: Any) -> str | None:
    if value is None:
        return None
    if isinstance(value, (dict, list, tuple, set)):
        return None
    text = str(value).replace("\\n", "\n")
    text = re.sub(r"[`*_#]+", "", text)
    text = re.sub(r"\s+", " ", text).strip(" ：:")
    return _business_copy(text) or None


def _source_text(value: Any) -> str | None:
    """Keep authoritative saved Markdown intact for source-document rendering."""

    if value is None or isinstance(value, (dict, list, tuple, set)):
        return None
    text = str(value).replace("\\n", "\n").strip()
    return text or None


def _business_value(value: Any) -> Any:
    if isinstance(value, str):
        return _business_copy(value)
    if isinstance(value, list):
        return [_business_value(item) for item in value]
    if isinstance(value, Mapping):
        return {key: _business_value(item) for key, item in value.items()}
    return value


def _business_copy(text: str) -> str:
    """Translate machine-facing tokens without deleting ordinary numbers."""

    replacements = (
        (r"\bretrieve_data_decode_case\b|\bretrievedatadecodecase\b", "解构案例取数"),
        (r"\bretrieve_data_knowledge\b|\bretrievedataknowledge\b", "知识取数"),
        (r"\bget_account_script_persona_points\b|\bgetaccountscriptpersonapoints\b", "账号人设"),
        (r"\bget_account_script_section_patterns\b", "账号段落模式"),
        (r"\bget_topic_detail\b", "选题内容"),
        (r"\bget_script_snapshot\b", "当前主脚本"),
        (r"\brecord_data_decision\b", "记录数据取舍"),
        (r"\bthink_and_plan\b", "实现思考与计划"),
    )
    for pattern, label in replacements:
        text = re.sub(pattern, label, text, flags=re.IGNORECASE)
    text = re.sub(
        r"\bpost_?id\s*[:=]?(?:\s*[A-Za-z0-9-]+)?",
        "账号内容样本",
        text,
        flags=re.IGNORECASE,
    )
    text = re.sub(r"\bscript_build_id\s*[:=]\s*\d+\b", "", text, flags=re.IGNORECASE)
    text = re.sub(r"\bround_index\s*[:=]\s*(\d+)\b", r"第 \1 轮", text, flags=re.IGNORECASE)
    text = re.sub(r"\bbranch[_\s-]*(\d+)\b", r"方案 \1", text, flags=re.IGNORECASE)
    text = re.sub(r"\bP(\d+)\b", r"段落 \1", text)
    text = re.sub(r"\bD(\d+)\b", r"评估维度 \1", text)
    text = re.sub(r"\bbase\b", "主脚本", text, flags=re.IGNORECASE)
    return re.sub(r"\s+", " ", text).strip(" ：:")


def _machine(value: Any) -> str | None:
    if value is None or isinstance(value, (dict, list, tuple, set)):
        return None
    text = str(value).strip()
    return text or None