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

from __future__ import annotations

import re
from collections import defaultdict
from typing import Any

from .business_detail_text import branch_task_text, business_text
from .decision_projection import AgentDecisionProjector
from .runtime_event_index import RuntimeEventIndex, event_succeeded, integer
from .runtime_tool_catalog import (
    RETRIEVAL_AGENT_LABELS,
    TOOL_CATALOG,
    business_tool_label,
    tools_in_category,
)


_CREATIVE_ACTION_TOOLS = tools_in_category("creative-action")
_DECISION_INPUT_TOOLS = tools_in_category("decision-input")


class CreativeEventProjector:
    """Project only safely owned implementer ``think_and_plan`` records."""

    def __init__(self, decisions: AgentDecisionProjector | None = None):
        self._decisions = decisions or AgentDecisionProjector()

    def project(
        self,
        index: RuntimeEventIndex,
        *,
        valid_rounds: set[int],
        valid_branches: set[int],
    ) -> dict[str, Any]:
        implementers: dict[int, dict[str, Any]] = {}
        for event in index.ordered:
            if (
                str(event.get("event_type") or "") == "agent_invoke"
                and str(event.get("event_name") or "") == "script_implementer"
                and (branch_id := integer(event.get("branch_id"))) in valid_branches
            ):
                implementers[branch_id] = event

        events_by_branch: dict[int, list[dict[str, Any]]] = defaultdict(list)
        for event in index.ordered:
            if (
                str(event.get("event_type") or "") != "tool_call"
                or str(event.get("event_name") or "") != "think_and_plan"
                or not event_succeeded(event)
            ):
                continue
            branch_id = integer(event.get("branch_id"))
            round_index = integer(event.get("round_index"))
            implementer = implementers.get(branch_id or -1)
            if (
                branch_id not in valid_branches
                or round_index not in valid_rounds
                or implementer is None
                or integer(event.get("scope_event_id"))
                not in {integer(implementer.get("id")), integer(implementer.get("scope_event_id"))}
                or str(event.get("agent_role") or "") not in {"script_implementer", "implementer"}
            ):
                continue
            events_by_branch[branch_id].append(event)

        projected: dict[int, dict[str, Any]] = {}
        for branch_id, records in events_by_branch.items():
            implementer = implementers[branch_id]
            latest = records[-1]
            task = _task(implementer)
            uncertainty = _uncertainties(records)
            steps = _steps(records)
            reasoning = _thought(latest)
            runtime_actions = _runtime_actions(index, implementer)
            available_upstream = _available_upstream(index, implementer)
            decision = self._decisions.creative(
                {
                    "id": f"creative:{integer(latest.get('id')) or 0}",
                    "detailRef": f"creative:{integer(latest.get('id')) or 0}",
                    "promptRef": f"prompt:event:{integer(implementer.get('id')) or 0}",
                    "safeLink": True,
                    "event": latest,
                    "task": task,
                    "steps": steps,
                    "reasoning": reasoning,
                    "uncertainties": uncertainty,
                    "inputs": available_upstream,
                    "completeness": "partial",
                    "notices": [
                        {
                            "code": "creative-record-partial",
                            "message": "当前运行只保存了通用思考与计划，没有稳定的完整创作四元组。",
                        }
                    ],
                }
            )
            if decision is None:
                continue
            decision["body"] = {
                **(decision.get("body") or {}),
                "availableUpstream": available_upstream,
                "runtimeActions": runtime_actions,
            }
            blocks = list((decision.get("detail") or {}).get("blocks") or [])
            insert_at = next(
                (
                    index_ + 1
                    for index_, block in enumerate(blocks)
                    if isinstance(block, dict)
                    and block.get("type") == "creative-process"
                ),
                0,
            )
            additions = []
            if available_upstream:
                additions.append(
                    {
                        "type": "items",
                        "title": "可确认的上游信息",
                        "items": [
                            {
                                "label": item["label"],
                                "value": item.get("summary"),
                                "note": "可确认在上游存在或已返回，但无法确认是否被本次创作采用。",
                                "detailRef": item.get("detailRef"),
                            }
                            for item in available_upstream
                        ],
                        "visualRole": "evidence",
                        "presentation": "list",
                        "collapsible": True,
                    }
                )
            if runtime_actions:
                additions.append(
                    {
                        "type": "items",
                        "title": "实际脚本改动",
                        "items": runtime_actions,
                        "visualRole": "section",
                        "presentation": "list",
                        "collapsible": True,
                    }
                )
            blocks[insert_at:insert_at] = additions
            decision["detail"] = {
                **(decision.get("detail") or {}),
                "blocks": blocks,
            }
            ref_set = {
                *(f"event:{integer(item.get('id')) or 0}" for item in records),
                *(str(item.get("detailRef")) for item in available_upstream if item.get("detailRef")),
                *(str(item.get("detailRef")) for item in runtime_actions if item.get("detailRef")),
            }
            decision["eventRefs"] = [
                ref
                for item in index.ordered
                if (ref := f"event:{integer(item.get('id')) or 0}") in ref_set
            ]
            decision["technicalRefs"] = decision["eventRefs"]
            projected[branch_id] = decision
        return {"creativeByBranch": projected, "unassigned": []}


def _task(event: dict[str, Any]) -> str | None:
    value = event.get("inputData")
    if isinstance(value, dict) and isinstance(value.get("task"), str):
        text = value["task"].strip()
        if match := re.search(r"(?:^|\s)目标\s*[：:]\s*(.+)$", text, re.S):
            text = match.group(1).strip()
        return branch_task_text(text) or None
    return None


def _steps(events: list[dict[str, Any]]) -> list[dict[str, Any]]:
    values: list[dict[str, Any]] = []
    for fallback_index, event in enumerate(events, start=1):
        payload = event.get("inputData")
        if not isinstance(payload, dict):
            continue
        summary = _creative_action_text(payload.get("thought_summary"))
        action = _creative_action_text(payload.get("action")) or summary
        plan = _creative_action_text(payload.get("plan"))
        reasoning = _creative_action_text(payload.get("thought"))
        if not any((action, plan, summary, reasoning)):
            continue
        values.append(
            {
                "stepIndex": integer(payload.get("thought_number")) or fallback_index,
                "action": action or f"步骤 {fallback_index}",
                "summary": summary,
                "plan": plan,
                "reasoning": reasoning,
                "eventRef": f"event:{integer(event.get('id')) or 0}",
            }
        )
    return values


def _creative_action_text(value: Any) -> str | None:
    if not isinstance(value, str) or not value.strip():
        return None
    text = value.strip()
    exact = text.lower()
    if exact in RETRIEVAL_AGENT_LABELS:
        return f"委派{RETRIEVAL_AGENT_LABELS[exact]}取数"
    if exact in TOOL_CATALOG:
        definition = TOOL_CATALOG[exact]
        if definition.category == "decision-input":
            return f"读取{definition.business_label}"
        return definition.business_label
    if re.fullmatch(r"[a-z][a-z0-9_]{2,}", exact):
        return None
    replacements = {
        **{
            name: f"委派{label}取数"
            for name, label in RETRIEVAL_AGENT_LABELS.items()
        },
        **{
            name: (
                f"读取{definition.business_label}"
                if definition.category == "decision-input"
                else definition.business_label
            )
            for name, definition in TOOL_CATALOG.items()
        },
    }
    for name in sorted(replacements, key=len, reverse=True):
        text = re.sub(
            rf"(?<![A-Za-z0-9_]){re.escape(name)}(?![A-Za-z0-9_])",
            replacements[name],
            text,
            flags=re.I,
        )
    text = business_text(text)
    text = re.sub(r"调用\s*(读取|记录|新增|更新|补充|批量)", r"\1", text)
    text = re.sub(r"委派\s*委派", "委派", text)
    return text.strip() or None


def _thought(event: dict[str, Any]) -> str | None:
    payload = event.get("inputData")
    if not isinstance(payload, dict):
        return None
    value = payload.get("thought_summary") or payload.get("thought")
    return value.strip() if isinstance(value, str) and value.strip() else None


def _uncertainties(events: list[dict[str, Any]]) -> list[str]:
    values: list[str] = []
    for event in events:
        payload = event.get("inputData")
        if not isinstance(payload, dict):
            continue
        text = "\n".join(
            str(payload.get(key) or "") for key in ("thought", "thought_summary", "plan")
        )
        for sentence in re.split(r"[。！？\n]+", text):
            if any(token in sentence for token in ("缺少直接证据", "未验证", "常识", "推导", "无法确认")):
                cleaned = sentence.strip()
                if cleaned and cleaned not in values:
                    values.append(cleaned)
    return values[:3]


def _runtime_actions(
    index: RuntimeEventIndex, implementer: dict[str, Any]
) -> list[dict[str, Any]]:
    scope_id = integer(implementer.get("id"))
    if scope_id is None:
        return []
    values: list[dict[str, Any]] = []
    for event in index.events_in_scope(scope_id):
        name = str(event.get("event_name") or "")
        if str(event.get("event_type") or "") != "tool_call" or name not in _CREATIVE_ACTION_TOOLS:
            continue
        event_id = integer(event.get("id")) or 0
        status = str(event.get("status") or "unknown")
        values.append(
            {
                "label": business_tool_label(name),
                "value": "已完成" if event_succeeded(event) else "未完成",
                "note": f"运行状态：{status}",
                "detailRef": f"event:{event_id}",
            }
        )
    return values


def _available_upstream(
    index: RuntimeEventIndex, implementer: dict[str, Any]
) -> list[dict[str, Any]]:
    implementer_id = integer(implementer.get("id"))
    branch_id = integer(implementer.get("branch_id"))
    round_index = integer(implementer.get("round_index"))
    if implementer_id is None:
        return []
    values: list[dict[str, Any]] = []
    for event in index.events_in_scope(implementer_id):
        name = str(event.get("event_name") or "")
        if str(event.get("event_type") or "") != "tool_call" or name not in _DECISION_INPUT_TOOLS:
            continue
        if not event_succeeded(event):
            continue
        values.append(
            {
                "label": business_tool_label(name),
                "summary": "实现 Agent 在创作过程中直接读取",
                "observedRelation": "read-by-actor",
                "decisionUse": "not-recorded",
                "detailRef": f"event:{integer(event.get('id')) or 0}",
            }
        )
    for event in index.ordered:
        name = str(event.get("event_name") or "")
        if (
            str(event.get("event_type") or "") != "agent_invoke"
            or name not in RETRIEVAL_AGENT_LABELS
            or integer(event.get("parent_event_id")) != implementer_id
            or integer(event.get("branch_id")) != branch_id
            or integer(event.get("round_index")) != round_index
            or not event_succeeded(event)
        ):
            continue
        values.append(
            {
                "label": RETRIEVAL_AGENT_LABELS[name],
                "summary": "取数结果已返回实现 Agent",
                "observedRelation": "returned-to-actor",
                "decisionUse": "not-recorded",
                "detailRef": f"event:{integer(event.get('id')) or 0}",
            }
        )
    unique: dict[tuple[Any, Any], dict[str, Any]] = {}
    for item in values:
        unique[(item.get("label"), item.get("detailRef"))] = item
    return list(unique.values())