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