from __future__ import annotations import re from typing import Any from .business_detail_text import ( assessment_text, batches_markdown, branch_task_sections, branch_task_text, branches_markdown, business_text, decisions_markdown, event_business_sections, source_markdown, clean_report, ) from .sanitizer import sanitize from .retrieval_detail_projection import ( is_retrieval_detail_event, project_retrieval_event, ) def project_round_detail( script_build_id: int, round_: dict[str, Any], bundle: dict[str, Any] | None = None, ) -> dict[str, Any]: raw_round = next( ( item for item in (bundle or {}).get("rounds", []) if int(item.get("round_index") or 0) == int(round_.get("roundIndex") or 0) ), None, ) sections = [ _section( "goal", "本轮目标", (raw_round or {}).get("goal") or (round_.get("summary", {}).get("goal") or {}).get("headline"), ), _section("plan", "当前保存的规划", _plan_text(round_.get("plan") or {})), _section("branches", "候选方案", branches_markdown(round_.get("branches") or [])), _section( "decisions", "候选评审与主 Agent 决策", _convergence_markdown(round_.get("convergenceBatches") or []), ), _section( "evaluation", "主脚本整体评审", _node_value(round_.get("overallEvaluation")) or "未找到明确整轮评审结论", ), _section("result", "本轮产出", _node_value(round_.get("result"))), ] result = round_.get("result") or {} changes = None if result.get("hasChanges"): changes = { "summary": _node_value(result), "sections": [ item for item in [ _section( "content", "内容方案", _round_content_changes(round_.get("branches") or []), ), _section( "domain", "领域信息", _round_domain_changes(round_.get("branches") or []), ), ] if item ], "artifactRef": "artifact:base:current", "exactness": "current-only", } return _payload( sections, { "scriptBuildId": script_build_id, "round": raw_round, "multipathDecisions": [ item for item in (bundle or {}).get("multipathDecisions", []) if int(item.get("round_index") or 0) == int(round_.get("roundIndex") or 0) ], }, changes, ) def project_activity_detail( script_build_id: int, activity_id: str, view: dict[str, Any], bundle: dict[str, Any], event_detail: dict[str, Any] | None = None, ) -> dict[str, Any]: if activity_id == "run:final-result": return _final_result_detail(script_build_id, view, bundle) missing_convergence = _find_missing_convergence(view, activity_id) if missing_convergence is not None: node, batch = missing_convergence return _payload( [ _section( "status", str(node.get("title") or "记录状态"), _node_value(node), "summary", ) ], { "scriptBuildId": script_build_id, "roundIndex": batch.get("roundIndex"), "branchIds": batch.get("branchIds") or [], "association": batch.get("association"), "unscopedReviewRefs": node.get("unscopedReviewRefs") or [], }, ) decision = _find_decision_projection(view, activity_id) if decision is not None: return _decision_detail_payload( script_build_id, decision, bundle, activity_id, event_detail ) if activity_id.startswith("domain-info:"): record_id = _suffix_int(activity_id) row = _find_by_id(bundle.get("domainInfo") or [], record_id) if not row: raise KeyError(activity_id) return _payload( [ _section("fact", "已核实领域事实", row.get("content")), _section("verification", "核实说明", row.get("note")), _section("source", "信息出处", _source_value(row.get("source"))), _section( "use", "用途与存疑状态", "已作为领域事实独立保存；是否进入具体候选脚本文案未记录。", ), ], {"scriptBuildId": script_build_id, "table": "script_build_domain_info", "record": row}, { "summary": "这条领域事实已经独立写入领域信息库。", "sections": [], "exactness": "exact", }, ) if activity_id.startswith("retrieval-direct:"): group = _find_retrieval_item(view, "directToolGroups", activity_id) if not group: raise KeyError(activity_id) calls = group.get("calls") or [] payload = _payload( [ _section("purpose", "读取方式", "实现 Agent 直接调用只读工具，补齐创作所需的上下文。"), _section("sources", "读取了什么", _direct_sources_markdown(group)), _section("result", "得到什么", _direct_result_markdown(calls), "summary"), ], { "scriptBuildId": script_build_id, "retrievalGroup": group, "events": _events_for_ids(bundle, [item.get("eventId") for item in calls]), }, ) payload.update( { "detailKind": "retrieval-group", "retrievalKind": "direct-tools", "activities": [ { "id": item.get("id") or item.get("detailRef"), "kind": "tool", "label": item.get("businessLabel") or "读取相关信息", "status": item.get("status") or "unknown", "summary": item.get("resultSummary") or "", "detailRef": item.get("detailRef"), } for item in calls ], } ) return payload if activity_id.startswith("retrieval-agent:"): run = _find_retrieval_item(view, "agentRuns", activity_id) if not run: raise KeyError(activity_id) event = event_detail or _event_by_id(bundle, run.get("eventId")) screening = run.get("screening") or {} full_task = _event_nested_value(event, "input", "task") raw_screening_text = _event_nested_value(event, "output", "summary") if not raw_screening_text: raw_screening_text = screening.get("preview") screening_text = clean_report(raw_screening_text) payload = _payload( [ _section("task", "完整取数任务", full_task or run.get("taskPreview") or "取数目标未记录"), _section("queries", "查询过程", _agent_query_markdown(run)), _section("hits", "原始命中", _agent_hit_markdown(run)), _section( "screening", "初筛整理", screening_text or _screening_fallback(screening.get("state")), "report", ), ], { "scriptBuildId": script_build_id, "agentRun": run, "agentEvent": event, "queryEvents": _events_for_ids(bundle, [item.get("eventId") for item in run.get("attempts") or []]), }, ) activities = [ { "id": item.get("id") or item.get("detailRef"), "kind": "query", "label": item.get("queryLabel") or "查询内容详见详情", "status": item.get("status") or "unknown", "resultCount": item.get("resultCount"), "detailRef": item.get("detailRef"), } for item in run.get("attempts") or [] ] payload.update( { "detailKind": "retrieval-agent", "retrievalKind": "agent", "activities": activities, } ) if not activities: if script_build_id <= 433: payload["missingActivityNotice"] = "该历史运行没有结构化查询明细。" elif run.get("agentType") == "retrieve_data_pattern_relation": payload["missingActivityNotice"] = ( "本次运行未保存可下钻的结构化查询记录。" ) return payload round_index, branch_id, role = _activity_parts(activity_id) round_ = _find_round(view, round_index) if round_index is not None else None if round_ is None: raise KeyError(activity_id) if branch_id is None: if role is None: return project_round_detail(script_build_id, round_, bundle) node = _round_node(round_, role) if not node: raise KeyError(activity_id) raw_round = next( ( item for item in bundle.get("rounds") or [] if int(item.get("round_index") or 0) == round_index ), None, ) if role == "result": return _round_result_detail(script_build_id, round_, raw_round, bundle) return _node_detail(script_build_id, node, raw_round) branch = next( ( item for item in round_.get("branches") or [] if int(item.get("branchId") or 0) == branch_id ), None, ) raw_branch = next( ( item for item in bundle.get("branches") or [] if int(item.get("branch_id") or 0) == branch_id ), None, ) if not branch: raise KeyError(activity_id) if role is None: return _branch_detail(script_build_id, branch, raw_branch, bundle) if role == "retrieval": return _retrieval_stage_detail(script_build_id, branch, raw_branch, bundle) node = _branch_node(branch, role) if not node: raise KeyError(activity_id) if role == "output": return _branch_output_detail(script_build_id, branch, raw_branch, bundle) if role == "task": return _branch_task_detail( script_build_id, branch, raw_branch, bundle, round_index=round_index, branch_id=branch_id, ) return _node_detail(script_build_id, node, raw_branch) def project_event_detail( script_build_id: int, event: dict[str, Any], *, run_status: str | None = None, ) -> dict[str, Any]: if is_retrieval_detail_event(event): payload = project_retrieval_event(event, run_status=run_status) payload["technical"] = sanitize( { "source": { "scriptBuildId": script_build_id, "table": "script_build_event / script_build_event_body", "detailRef": f"event:{event.get('id')}", }, **_event_technical_groups(event), }, max_text=None, ) return sanitize(payload, max_text=None) if ( event.get("event_name") == "think_and_plan" and event.get("agent_role") == "main" and int(event.get("agent_depth") or 0) == 0 ): raw_input = event.get("input") content = raw_input.get("content") if isinstance(raw_input, dict) else None data = content if isinstance(content, dict) else event.get("inputData") data = data if isinstance(data, dict) else {} return _payload( [ _section( "summary", "规划概要", data.get("thought_summary"), "summary", visual_role="key-result", ), _section( "thought", "完整思考", data.get("thought"), "report", visual_role="section", ), _section( "plan", "执行计划", data.get("plan"), visual_role="section", ), _section( "action", "下一步", data.get("action"), visual_role="reason", ), ], { "scriptBuildId": script_build_id, "table": "script_build_event / script_build_event_body", "event": event, }, ) return _payload( event_business_sections(event), {"scriptBuildId": script_build_id, "table": "script_build_event / script_build_event_body", "event": event}, ) def project_artifact_detail( script_build_id: int, snapshot_ref: str, artifact: dict[str, Any] ) -> dict[str, Any]: paragraphs, elements, links = _normalized_artifact_records(artifact) exactness = "current-only" if "base:current" in snapshot_ref else _artifact_exactness(artifact) context = artifact.get("artifactContext") if isinstance(artifact.get("artifactContext"), dict) else {} context = { "type": context.get("type") or ("final" if "base:current" in snapshot_ref else "candidate"), **context, "exactness": exactness, } candidate = context.get("type") == "candidate" scale_title = "候选脚本规模" if candidate else "主脚本规模" content_summary = "候选脚本内容" if candidate else "当前主脚本内容" payload = _payload( [ _section("summary", scale_title, f"{len(paragraphs)} 个段落、{len(elements)} 个元素、{len(links)} 条关联"), ], { "scriptBuildId": script_build_id, "snapshotRef": snapshot_ref, "sourceOrigin": artifact.get("sourceOrigin") or "database", "exactness": exactness, "artifactContext": context, "recordCounts": { "paragraphs": len(paragraphs), "elements": len(elements), "links": len(links), }, }, { "summary": content_summary, "sections": [ _section( "counts", "内容规模", f"{len(paragraphs)} 个段落、{len(elements)} 个元素、{len(links)} 条关联", ), ], "artifactRef": snapshot_ref, "exactness": exactness, }, ) payload["scriptTable"] = sanitize( _script_table_projection(snapshot_ref, artifact, exactness), max_text=None, ) payload["artifactContext"] = sanitize(context, max_text=None) return payload _ELEMENT_REFERENCE = re.compile(r"\[([^\]:]+):(\d+)\]") _SCRIPT_SECTIONS = ( ("theme", "主题"), ("form", "形式"), ("function", "作用"), ("feeling", "感受"), ) def _script_table_projection( snapshot_ref: str, artifact: dict[str, Any], exactness: str ) -> dict[str, Any]: snapshot = artifact.get("snapshot") if isinstance(artifact.get("snapshot"), dict) else artifact snapshot = snapshot if isinstance(snapshot, dict) else {} paragraphs, elements, links = _normalized_artifact_records(artifact) element_names: dict[int, str] = {} for element in elements: element_id = _as_int(element.get("effectiveId") or element.get("id")) row_id = _as_int(element.get("rowId") or element.get("id")) name = str(element.get("name") or "").strip() if element_id is not None: element_names[element_id] = name if row_id is not None: element_names[row_id] = name paragraph_rows = _ordered_paragraphs( [_script_paragraph(row, element_names) for row in paragraphs] ) paragraph_lookup: dict[int, dict[str, Any]] = {} for paragraph in paragraph_rows: for key in ("id", "rowId"): value = _as_int(paragraph.get(key)) if value is not None: paragraph_lookup[value] = paragraph linked_paragraphs: dict[int, list[dict[str, Any]]] = {} for link in links: element_id = _as_int(link.get("element_id") or link.get("elementId")) paragraph_id = _as_int(link.get("paragraph_id") or link.get("paragraphId")) paragraph = paragraph_lookup.get(paragraph_id) if paragraph_id is not None else None if element_id is None or paragraph is None: continue linked_paragraphs.setdefault(element_id, []).append( { "id": paragraph.get("id"), "index": paragraph.get("index"), "name": paragraph.get("name"), } ) element_rows = [] for element in elements: element_id = element.get("effectiveId") or element.get("id") row_id = element.get("rowId") or element.get("id") link_keys = { value for value in (_as_int(row_id), _as_int(element_id)) if value is not None } linked = [ paragraph for key in link_keys for paragraph in linked_paragraphs.get(key, []) ] element_rows.append( { "id": element_id, "rowId": row_id, "name": element.get("name") or "未命名元素", "primaryDimension": element.get("dimension_primary"), "secondaryDimension": element.get("dimension_secondary"), "paragraphs": _unique_records(linked, "id"), } ) element_rows.sort(key=lambda row: (str(row.get("name") or ""), _sort_number(row.get("id")))) return { "snapshotRef": snapshot_ref, "exactness": exactness, "sourceOrigin": artifact.get("sourceOrigin") or snapshot.get("sourceOrigin") or "database", "counts": { "paragraphs": len(paragraph_rows), "elements": len(element_rows), "links": len(links), }, "maxDepth": ( max((int(row.get("depth") or 0) for row in paragraph_rows), default=0) + 1 if paragraph_rows else 0 ), "paragraphs": paragraph_rows, "elements": element_rows, } def _normalized_artifact_records( artifact: dict[str, Any], ) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]: """Normalize current overlays and frozen branch snapshots to one table shape. Current overlays store a flat ``paragraphs`` list plus top-level ``paragraphElements``. Frozen ``content_snapshot`` rows store second-level paragraphs under ``sub_paragraphs`` and links under each paragraph's ``linked_elements``. The visualizer must preserve both without pretending that a truncated top-level list is the complete candidate script. """ snapshot = artifact.get("snapshot") if isinstance(artifact.get("snapshot"), dict) else artifact snapshot = snapshot if isinstance(snapshot, dict) else {} paragraphs: list[dict[str, Any]] = [] seen_paragraphs: set[tuple[str, str]] = set() def visit(row: Any, parent_id: Any = None) -> None: if not isinstance(row, dict): return current = dict(row) children = current.pop("sub_paragraphs", None) or current.pop("subParagraphs", None) or [] if parent_id is not None and current.get("parent_id") is None and current.get("parentId") is None: current["parent_id"] = parent_id row_id = current.get("rowId") or current.get("id") key = (str(row_id), str(current.get("baseRefId") or current.get("base_ref_id") or "")) if key not in seen_paragraphs: seen_paragraphs.add(key) paragraphs.append(current) effective_parent = current.get("effectiveId") or current.get("id") or row_id for child in children if isinstance(children, list) else []: visit(child, effective_parent) for paragraph in snapshot.get("paragraphs") or []: visit(paragraph) elements = [item for item in snapshot.get("elements") or [] if isinstance(item, dict)] links: list[dict[str, Any]] = [] seen_links: set[tuple[str, str]] = set() def append_link(link: Any, paragraph_id: Any = None) -> None: if not isinstance(link, dict): return current = dict(link) if paragraph_id is not None and current.get("paragraph_id") is None and current.get("paragraphId") is None: current["paragraph_id"] = paragraph_id element_id = current.get("element_id") or current.get("elementId") linked_paragraph = current.get("paragraph_id") or current.get("paragraphId") if element_id is None or linked_paragraph is None: return key = (str(linked_paragraph), str(element_id)) if key in seen_links: return seen_links.add(key) links.append(current) for link in snapshot.get("paragraphElements") or snapshot.get("paragraph_elements") or []: append_link(link) for paragraph in paragraphs: paragraph_id = paragraph.get("effectiveId") or paragraph.get("id") or paragraph.get("rowId") nested_links = paragraph.get("linked_elements") or paragraph.get("linkedElements") or [] if isinstance(nested_links, list) and nested_links: for link in nested_links: append_link(link, paragraph_id) else: for element_id in paragraph.get("linked_element_ids") or paragraph.get("linkedElementIds") or []: append_link({"element_id": element_id}, paragraph_id) return paragraphs, elements, links def _script_paragraph(row: dict[str, Any], element_names: dict[int, str]) -> dict[str, Any]: full_description = str(row.get("full_description") or "") return { "id": row.get("effectiveId") or row.get("id"), "rowId": row.get("rowId") or row.get("id"), "parentId": row.get("parent_id"), "depth": 0, "index": row.get("paragraph_index"), "level": row.get("level"), "name": row.get("name") or "未命名段落", "contentRange": _content_range(row.get("content_range")), "sections": { key: { "label": label, "description": row.get(key), "descriptionParts": _description_parts( str(row.get(key) or ""), element_names ), "dimensions": _dimensions(row.get(f"{key}_elements")), } for key, label in _SCRIPT_SECTIONS }, "description": row.get("description"), "fullDescription": full_description, "fullDescriptionParts": _description_parts(full_description, element_names), } def _ordered_paragraphs(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: """Keep every child next to its real parent without guessing from level labels.""" by_reference: dict[int, dict[str, Any]] = {} for row in rows: for value in (row.get("rowId"), row.get("id")): reference = _as_int(value) if reference is not None: by_reference[reference] = row children: dict[int, list[dict[str, Any]]] = {} roots: list[dict[str, Any]] = [] for row in rows: parent = by_reference.get(_as_int(row.get("parentId"))) if parent is None or parent is row: row["parentId"] = None roots.append(row) continue row["parentId"] = parent.get("id") children.setdefault(id(parent), []).append(row) ordered: list[dict[str, Any]] = [] visited: set[int] = set() def visit(row: dict[str, Any], depth: int, ancestry: set[int]) -> None: marker = id(row) if marker in visited: return if marker in ancestry: row["parentId"] = None depth = 0 visited.add(marker) row["depth"] = depth ordered.append(row) next_ancestry = {*ancestry, marker} for child in sorted(children.get(marker, []), key=_paragraph_sort_key): visit(child, depth + 1, next_ancestry) for root in sorted(roots, key=_paragraph_sort_key): visit(root, 0, set()) for row in sorted(rows, key=_paragraph_sort_key): if id(row) not in visited: row["parentId"] = None visit(row, 0, set()) return ordered def _paragraph_sort_key(row: dict[str, Any]) -> tuple[Any, ...]: return (_sort_number(row.get("index")), _sort_number(row.get("id"))) def _content_range(value: Any) -> list[str]: if not isinstance(value, dict): return [_display_value(value)] if value not in (None, "") else [] tags: list[str] = [] for key, item in value.items(): if key == "标题": tags.append("标题") elif key == "图片" and isinstance(item, list): if not item: continue values = [str(entry) for entry in item] tags.append(f"图{values[0]}" if len(values) == 1 else f"图{values[0]}-{values[-1]}") elif key == "视频": tags.append(_display_value(item)) elif key == "正文": body = "\n".join(str(entry) for entry in item) if isinstance(item, list) else _display_value(item) tags.append(f"正文: {body}") else: rendered = ",".join(str(entry) for entry in item) if isinstance(item, list) else _display_value(item) tags.append(f"{key}: {rendered}") return tags def _dimensions(value: Any) -> list[dict[str, Any]]: rows = value if isinstance(value, list) else [value] if isinstance(value, dict) else [] return [ { "type": row.get("维度类型") or row.get("dimension_type"), "name": row.get("维度") or row.get("dimension"), "value": row.get("原子点") or row.get("value"), **({"id": row.get("id")} if row.get("id") is not None else {}), } for row in rows if isinstance(row, dict) ] def _description_parts(value: str, element_names: dict[int, str]) -> list[dict[str, Any]]: parts: list[dict[str, Any]] = [] cursor = 0 for match in _ELEMENT_REFERENCE.finditer(value): if match.start() > cursor: parts.append({"type": "text", "text": value[cursor : match.start()]}) element_id = int(match.group(2)) name = element_names.get(element_id) parts.append( { "type": "element", "id": element_id, "name": name, "resolved": name is not None, "text": match.group(0), } ) cursor = match.end() if name: duplicate = re.match(r"\s*" + re.escape(name), value[cursor:]) if duplicate: cursor += duplicate.end() if cursor < len(value): parts.append({"type": "text", "text": value[cursor:]}) return parts def _display_value(value: Any) -> str: if value is True: return "包含" if value is False: return "不包含" if value is None: return "" return str(value) def _as_int(value: Any) -> int | None: try: return int(value) if value is not None else None except (TypeError, ValueError): return None def _sort_number(value: Any) -> tuple[int, str]: number = _as_int(value) return (0, str(number).zfill(12)) if number is not None else (1, str(value or "")) def _unique_records(rows: list[dict[str, Any]], key: str) -> list[dict[str, Any]]: seen: set[Any] = set() result = [] for row in rows: value = row.get(key) if value in seen: continue seen.add(value) result.append(row) return result def _branch_detail(script_build_id: int, branch: dict[str, Any], raw: Any, bundle: dict[str, Any]) -> dict[str, Any]: facts = _branch_facts(branch, bundle) return _payload( [ _section("task", "实现任务", branch_task_text(branch.get("summary", {}).get("task"))), _section("data", "数据取舍", decisions_markdown([item.get("decision") for item in branch.get("dataDecisions") or []])), _section("output", "候选产出", branch.get("summary", {}).get("output")), ], {"scriptBuildId": script_build_id, "branch": raw}, { "summary": branch.get("summary", {}).get("output"), "sections": [_section("facts", "领域事实", _domain_facts_markdown(facts))] if facts else [], "artifactRef": branch.get("output", {}).get("snapshotRef"), "exactness": _output_exactness(branch.get("output") or {}), } if branch.get("output", {}).get("node", {}).get("hasChanges") else None, ) def _branch_task_detail( script_build_id: int, branch: dict[str, Any], raw: Any, bundle: dict[str, Any], *, round_index: int, branch_id: int, ) -> dict[str, Any]: task_event = next( ( event for event in reversed(bundle.get("events") or []) if str(event.get("event_type") or "") == "agent_invoke" and str(event.get("event_name") or "") == "script_implementer" and int(event.get("round_index") or 0) == round_index and int(event.get("branch_id") or 0) == branch_id ), None, ) event_input = (task_event or {}).get("inputData") event_task = event_input.get("task") if isinstance(event_input, dict) else None raw_task = (raw or {}).get("impl_task") if isinstance(raw, dict) else None complete_task = branch_task_text( event_task or raw_task or branch.get("summary", {}).get("task") ) task_sections = branch_task_sections( event_task or raw_task or branch.get("summary", {}).get("task") ) visual_roles = { "goal": ("summary", "key-result"), "materials": ("report", "evidence"), "avoid": ("report", "notice"), "context": ("report", "evidence"), } return _payload( [ _section( f"task-{section['kind']}-{index}", section["title"], section["content"], visual_roles.get(section["kind"], ("default", "section"))[0], visual_role=visual_roles.get(section["kind"], ("default", "section"))[1], presentation="document" if section["kind"] in {"materials", "constraints", "avoid", "context"} else "prose", ) for index, section in enumerate(task_sections, 1) ] or [_section("task-notes-1", "任务说明", complete_task)], { "scriptBuildId": script_build_id, "branch": raw, "taskEvent": task_event, "taskSource": ( "script_build_event.inputData.task" if event_task else "script_build_branch.impl_task" if raw_task else "execution-view summary" ), }, ) def _branch_output_detail(script_build_id: int, branch: dict[str, Any], raw: Any, bundle: dict[str, Any]) -> dict[str, Any]: output = branch.get("output") or {} facts = _branch_facts(branch, bundle) changes = { "summary": output.get("summary"), "sections": [_section("facts", "已核实领域事实", _domain_facts_markdown(facts))] if facts else [], "artifactRef": output.get("snapshotRef"), "exactness": _output_exactness(output), } return _payload( [ _section("output", "形成了什么", output.get("summary")), _section("assessment", "实现者说明", assessment_text((raw or {}).get("self_assessment")), "report"), ], {"scriptBuildId": script_build_id, "branch": raw, "output": output}, changes if output.get("node", {}).get("hasChanges") else None, ) def _retrieval_stage_detail(script_build_id: int, branch: dict[str, Any], raw: Any, bundle: dict[str, Any]) -> dict[str, Any]: stage = branch.get("retrievalStage") or {} direct = stage.get("directToolGroups") or [] agents = stage.get("agentRuns") or [] return _payload( [ _section( "purpose", "取数目的", branch_task_text(branch.get("summary", {}).get("task")) or "为当前实现任务补齐可用的上游信息。", ), _section("mode", "执行方式", _retrieval_mode_text(stage.get("observedMode"))), _section("waves", "执行波次", _retrieval_waves_markdown(stage.get("waves") or [])), _section("direct", "工具取数", _direct_groups_markdown(direct)), _section("agents", "Agent 取数", _agent_runs_markdown(agents)), _section("result", "阶段结果与失败", _retrieval_stage_result(stage)), ], { "scriptBuildId": script_build_id, "branch": raw, "retrievalStage": stage, "note": "取数和初筛来自结构化运行事件；正式数据取舍来自独立业务表，不在本详情内反推。", }, ) def _node_detail(script_build_id: int, node: dict[str, Any], technical: Any) -> dict[str, Any]: fields = [] primary = node.get("card", {}).get("primary") if primary: fields.append(_section("primary", primary.get("label") or node.get("title"), primary.get("value"))) for line in node.get("card", {}).get("secondary") or []: fields.append(_section(str(line.get("key")), line.get("label") or "补充说明", line.get("value"))) return _payload(fields, {"scriptBuildId": script_build_id, "record": technical}) def _round_result_detail( script_build_id: int, round_: dict[str, Any], raw_round: dict[str, Any] | None, bundle: dict[str, Any], ) -> dict[str, Any]: result = round_.get("result") or {} branches = round_.get("branches") or [] round_index = int(round_.get("roundIndex") or 0) facts = [ item for item in bundle.get("domainInfo") or [] if int(item.get("round_index") or 0) == round_index ] evaluation = round_.get("overallEvaluation") or {} next_step = next( ( line.get("value") for line in (result.get("card") or {}).get("secondary") or [] if line.get("key") == "next" ), None, ) sections = [ _section( "goal", "本轮目标完成情况", "\n".join( part for part in ( business_text((raw_round or {}).get("goal")), f"收口状态：{result.get('status') or round_.get('state') or '未记录'}", ) if part ), ), _section("outcomes", "方案处置", _branch_outcomes_markdown(branches)), _section("facts", "领域事实", _domain_facts_markdown(facts) or "本轮没有新增领域事实记录。"), _section( "evaluation", "整体评审", _node_value(evaluation) or "未找到明确的本轮整体评审结论。", ), _section( "changes", "主脚本变化", _round_content_changes(branches), ), _section("next", "下一步", next_step or "下一步未记录。"), ] changes = None if result.get("hasChanges"): changes = { "summary": _node_value(result), "sections": [ _section("content", "内容方案", _round_content_changes(branches)), _section("domain", "领域信息", f"本轮独立保存 {len(facts)} 条领域事实。"), ], "artifactRef": "artifact:base:current", "exactness": "current-only", } return _payload( sections, { "scriptBuildId": script_build_id, "round": raw_round, "resultProjection": result, "note": "本轮主脚本历史快照无法完整还原，产物入口指向当前主脚本。", }, changes, ) def _final_result_detail( script_build_id: int, view: dict[str, Any], bundle: dict[str, Any] ) -> dict[str, Any]: final = view.get("finalResult") or {} record = bundle.get("record") or {} branches = final.get("branchCounts") or {} artifact = final.get("artifact") or {} status = str(final.get("status") or record.get("status") or "unknown") conclusion = { "success": "构建已完成。", "completed": "构建已完成。", "partial": "构建部分完成。", "failed": "构建失败。", "stopped": "构建已停止。", "running": "构建仍在进行。", }.get(status.lower(), f"构建状态：{status}。") summary_or_error = final.get("summary") if final.get("errorMessage"): summary_or_error = f"失败原因：{final['errorMessage']}" sections = [ _section("conclusion", "构建结论", conclusion, "summary"), _section("summary", "构建摘要或失败原因", summary_or_error or "未另外记录构建摘要。"), _section( "statistics", "轮次与方案统计", ( f"共 {int(final.get('roundCount') or 0)} 轮，" f"{int(branches.get('total') or 0)} 个方案；" f"采用 {int(branches.get('merged') or 0)}，" f"暂存 {int(branches.get('parked') or 0)}，" f"未采用 {int(branches.get('discarded') or 0)}，" f"待处理 {int(branches.get('open') or 0)}。" ), ), _section( "artifact", "最终主脚本规模", ( f"{int(artifact.get('paragraphCount') or 0)} 个段落、" f"{int(artifact.get('elementCount') or 0)} 个元素、" f"{int(artifact.get('linkCount') or 0)} 条关联。" ), ), _section( "version", "版本说明", "当前只能确认数据库保存的最新主脚本，不将它伪装成某一轮结束时的历史快照。", ), ] return _payload( sections, { "scriptBuildId": script_build_id, "table": "script_build_record", "record": record, "finalResultProjection": final, }, { "summary": "打开当前主脚本。", "sections": [], "artifactRef": artifact.get("snapshotRef") or "artifact:base:current", "exactness": "current-only", }, ) def _find_decision_projection( view: dict[str, Any], detail_ref: str ) -> dict[str, Any] | None: candidates: list[dict[str, Any]] = [] objective = view.get("objective") or {} if isinstance(objective.get("decision"), dict): candidates.append(objective["decision"]) for round_ in view.get("rounds") or []: for node in ( round_.get("goal"), (round_.get("plan") or {}).get("node"), round_.get("overallEvaluation"), ): if isinstance(node, dict) and isinstance(node.get("decision"), dict): candidates.append(node["decision"]) for branch in round_.get("branches") or []: for item in branch.get("dataDecisions") or []: projection = item.get("decisionProjection") if isinstance(projection, dict): candidates.append(projection) output_node = (branch.get("output") or {}).get("node") or {} if isinstance(output_node.get("decision"), dict): candidates.append(output_node["decision"]) for batch in round_.get("convergenceBatches") or []: review = batch.get("review") or {} decision = batch.get("decision") or {} if isinstance(review.get("decisionProjection"), dict): candidates.append(review["decisionProjection"]) if isinstance(decision.get("decisionProjection"), dict): candidates.append(decision["decisionProjection"]) return next( ( item for item in candidates if detail_ref in {item.get("id"), item.get("detailRef")} ), None, ) def _find_missing_convergence( view: dict[str, Any], detail_ref: str ) -> tuple[dict[str, Any], dict[str, Any]] | None: for round_ in view.get("rounds") or []: for batch in round_.get("convergenceBatches") or []: for key in ("missingReview", "missingDecision"): node = batch.get(key) if isinstance(node, dict) and detail_ref in { node.get("id"), node.get("detailRef"), }: return node, batch return None def _decision_detail_payload( script_build_id: int, decision: dict[str, Any], bundle: dict[str, Any], activity_id: str, event_detail: dict[str, Any] | None = None, ) -> dict[str, Any]: detail = decision.get("detail") if isinstance(decision.get("detail"), dict) else {} blocks = [ block for block in detail.get("blocks") or [] if isinstance(block, dict) ] for notice in decision.get("notices") or []: if not isinstance(notice, dict) or not notice.get("message"): continue blocks.append( { "type": "notice", "level": "warning" if "missing" in str(notice.get("code")) else "info", "title": "记录说明", "value": notice["message"], "visualRole": "notice", "presentation": "prose", } ) event_refs = [ ref for ref in [ *(decision.get("eventRefs") or []), *(decision.get("technicalRefs") or []), ] if isinstance(ref, str) and ref.startswith("event:") ] event_refs = list(dict.fromkeys(event_refs)) technical: dict[str, Any] = { "source": { "scriptBuildId": script_build_id, "detailRef": activity_id, "technicalRefs": event_refs, }, "association": { "completeness": decision.get("completeness"), "notices": decision.get("notices") or [], }, "rawDecisionProjection": decision, "rawBusinessRecord": _decision_raw_record(bundle, activity_id), } if event_refs: technical["runtimeEvents"] = _events_for_ids( bundle, [_suffix_int(ref) for ref in event_refs] ) if event_detail: technical.update(_event_technical_groups(event_detail)) payload: dict[str, Any] = { "detailKind": "agent-decision", "decisionType": decision.get("decisionType"), "actor": decision.get("actor"), "authority": decision.get("authority"), "question": decision.get("question"), "promptRef": decision.get("promptRef"), "blocks": sanitize(blocks, max_text=None), "technical": sanitize(technical, max_text=None), } artifact_ref = decision.get("artifactRef") if artifact_ref: payload["changes"] = { "summary": (decision.get("body") or {}).get("output") or "已形成可查看的候选产出。", "sections": [], "artifactRef": artifact_ref, "exactness": "historical-snapshot", } return payload def _event_technical_groups(event: dict[str, Any]) -> dict[str, Any]: """Split a lazily loaded event body into the Inspector's technical groups. The execution bundle only carries bounded previews. Once a user opens an event-backed decision, the full (sanitized) body belongs here rather than being silently replaced by those previews. """ metadata = { key: value for key, value in event.items() if key not in {"input", "output", "duration_ms"} } groups: dict[str, Any] = { "input": event.get("input"), "output": event.get("output"), "rawEvent": metadata, } if event.get("duration_ms") is not None: groups["cost"] = {"durationMs": event.get("duration_ms")} return groups def _decision_raw_record(bundle: dict[str, Any], activity_id: str) -> Any: if activity_id.startswith("data-decision:"): return _find_by_id(bundle.get("dataDecisions") or [], _suffix_int(activity_id)) if activity_id.startswith("multipath-decision:"): return _find_by_id( bundle.get("multipathDecisions") or [], _suffix_int(activity_id) ) if activity_id.startswith("event:"): return _event_by_id(bundle, _suffix_int(activity_id)) if activity_id.startswith("creative:"): event_id = _suffix_int(activity_id) event = _event_by_id(bundle, event_id) branch_id = int((event or {}).get("branch_id") or 0) branch = next( ( item for item in bundle.get("branches") or [] if int(item.get("branch_id") or 0) == branch_id ), None, ) return { "anchorEvent": event, "candidateSnapshot": (branch or {}).get("candidate_snapshot"), } if activity_id == "run:objective": return bundle.get("record") round_index, _, role = _activity_parts(activity_id) if role in {"goal", "plan"}: return next( ( item for item in bundle.get("rounds") or [] if int(item.get("round_index") or 0) == int(round_index or 0) ), None, ) return None def _payload(sections: list[dict[str, Any] | None], technical: Any, changes: dict[str, Any] | None = None) -> dict[str, Any]: payload: dict[str, Any] = { "businessSections": [item for item in sections if item], "technical": sanitize(technical, max_text=None), } if changes: changes["sections"] = [item for item in changes.get("sections", []) if item] payload["changes"] = sanitize(changes, max_text=None) return payload def _section( section_id: str, title: str, content: Any, variant: str = "default", *, visual_role: str | None = None, presentation: str = "prose", collapsible: bool = False, default_open: bool = False, ) -> dict[str, Any] | None: if content is None or content == "" or content == []: return None if not isinstance(content, str): return None content = business_text(content) if not content: return None return { "id": section_id, "title": title, "content": sanitize(content, max_text=None), "variant": variant, "visualRole": visual_role or ("key-result" if variant == "summary" else "section"), "presentation": presentation, "collapsible": collapsible, "defaultOpen": default_open, } def _items_section( section_id: str, title: str, items: list[dict[str, Any] | None], *, fallback: Any = None, variant: str = "default", visual_role: str | None = None, collapsible: bool = False, default_open: bool = False, ) -> dict[str, Any] | None: values = [item for item in items if item] if not values and fallback in (None, "", []): return None content = "\n".join( f"{item.get('label')}：{item.get('value')}" for item in values ) or str(fallback) section = _section( section_id, title, content, variant, visual_role=visual_role, presentation="list", collapsible=collapsible, default_open=default_open, ) if section is not None and values: section["items"] = sanitize(values, max_text=None) return section def _detail_item(label: Any, value: Any) -> dict[str, Any] | None: if value in (None, "", []): return None return { "label": sanitize(business_text(str(label)), max_text=None), "value": sanitize(business_text(str(value)), max_text=None), } def _activity_parts(activity_id: str) -> tuple[int | None, int | None, str | None]: parts = activity_id.split(":") if len(parts) < 2 or parts[0] != "round": return None, None, None try: round_index = int(parts[1]) except ValueError: return None, None, None if len(parts) >= 4 and parts[2] == "branch": try: branch_id = int(parts[3]) except ValueError: return round_index, None, None return round_index, branch_id, parts[4] if len(parts) > 4 else None return round_index, None, parts[2] if len(parts) > 2 else None def _round_node(round_: dict[str, Any], role: str) -> dict[str, Any] | None: if role == "goal": return round_.get("goal") if role == "plan": return (round_.get("plan") or {}).get("node") if role == "result": return round_.get("result") if role == "evaluation": return round_.get("overallEvaluation") return None def _branch_node(branch: dict[str, Any], role: str) -> dict[str, Any] | None: if role == "task": return branch.get("task") if role == "output": return branch.get("output", {}).get("node") return None def _convergence_markdown(batches: list[dict[str, Any]]) -> str: if not batches: return "未找到候选评审或主 Agent 多路决策记录。" rows: list[str] = [] for index, batch in enumerate(batches, 1): review = batch.get("review") or {} decision = batch.get("decision") or {} branch_ids = batch.get("branchIds") or [] rows.append( f"### 第 {index} 批 · " + ("、".join(f"方案 {value}" for value in branch_ids) or "方案范围未记录") ) rows.append( f"**多方案评审：** {review.get('recommendation') or review.get('comparison') or review.get('conclusion') or '评审结论未记录'}" ) rows.append( f"**主 Agent 决策：** {business_text(decision.get('decision')) or '决策尚未记录'}" ) return "\n\n".join(rows) def _find_retrieval_item(view: dict[str, Any], collection: str, item_id: str) -> dict[str, Any] | None: for round_ in view.get("rounds") or []: for branch in round_.get("branches") or []: for item in (branch.get("retrievalStage") or {}).get(collection) or []: if item.get("id") == item_id: return item return None def _event_by_id(bundle: dict[str, Any], event_id: Any) -> dict[str, Any] | None: return next( (item for item in bundle.get("events") or [] if int(item.get("id") or 0) == int(event_id or 0)), None, ) def _events_for_ids(bundle: dict[str, Any], event_ids: list[Any]) -> list[dict[str, Any]]: wanted = {int(value) for value in event_ids if value is not None} return [item for item in bundle.get("events") or [] if int(item.get("id") or 0) in wanted] def _event_nested_value( event: dict[str, Any] | None, side: str, key: str ) -> Any: if not isinstance(event, dict): return None aliases = ( ("inputData", "input_data") if side == "input" else ("agentOutputData", "outputData", "output_data") ) for alias in aliases: value = event.get(alias) if isinstance(value, dict) and value.get(key) not in (None, ""): return value.get(key) wrapped = event.get(side) if isinstance(wrapped, dict): content = wrapped.get("content") if isinstance(content, dict): return content.get(key) if key == "summary" and isinstance(content, str): return content return None def _direct_sources_markdown(group: dict[str, Any]) -> str: rows = group.get("sources") or [] return "\n".join( f"- {item.get('businessLabel') or '相关信息'}：{int(item.get('callCount') or 0)} 次" for item in rows ) or "未记录具体读取项" def _direct_result_markdown(calls: list[dict[str, Any]]) -> str: success = sum(1 for item in calls if item.get("status") == "success") failure = sum(1 for item in calls if item.get("status") == "failure") running = sum(1 for item in calls if item.get("status") == "running") interrupted = sum(1 for item in calls if item.get("status") == "interrupted") return f"共 {len(calls)} 次读取：{success} 次完成，{failure} 次失败，{running} 次进行中，{interrupted} 次因构建中断。" def _agent_query_markdown(run: dict[str, Any]) -> str: summary = run.get("querySummary") or {} total = int(summary.get("total") or 0) if total == 0 and not (run.get("attempts") or []): return "Agent 已完成，没有发起查询。" header = ( f"共 {total} 次查询：" f"{int(summary.get('hit') or 0)} 次命中，" f"{int(summary.get('empty') or 0)} 次无结果，" f"{int(summary.get('failure') or 0)} 次失败，" f"{int(summary.get('interrupted') or 0)} 次因构建中断。" ) rows = [] for index, attempt in enumerate(run.get("attempts") or [], 1): query = attempt.get("queryLabel") or attempt.get("queryPreview") or attempt.get("businessLabel") or "查询" label = {"hit": "命中", "empty": "无结果", "failure": "失败", "running": "进行中", "interrupted": "构建中断", "unknown": "结果数不明"}.get(attempt.get("status"), "结果数不明") count = attempt.get("resultCount") suffix = f"· {count} 条" if isinstance(count, int) and count > 0 else f"· {label}" rows.append(f"{index:02d}. {query} {suffix}") return "\n".join([header, *rows]) def _agent_hit_markdown(run: dict[str, Any]) -> str: attempts = run.get("attempts") or [] hits = [item for item in attempts if item.get("status") == "hit"] empty = [item for item in attempts if item.get("status") == "empty"] failures = [ item for item in attempts if item.get("status") in {"failure", "interrupted"} ] rows = [ f"- {item.get('queryLabel') or '查询'}：{int(item.get('resultCount') or 0)} 条" for item in hits ] rows.append(f"成功命中 {len(hits)} 次，成功为空 {len(empty)} 次，技术失败或中断 {len(failures)} 次。") return "\n".join(rows) def _screening_fallback(state: Any) -> str: return {"running": "Agent 仍在整理查询结果。", "missing": "未找到 Agent 的初筛整理记录。"}.get(str(state), "初筛整理未记录。") def _retrieval_mode_text(mode: Any) -> str: return { "single": "本次只有一个取数操作。", "sequential": "各项取数依次执行。", "parallel": "各项取数在同一时间段并行执行。", "mixed": "本次同时存在并行取数和后续取数。", "unknown": "运行时间记录不足，无法确认执行先后。", }.get(str(mode), "执行方式未记录。") def _retrieval_waves_markdown(waves: list[dict[str, Any]]) -> str: if not waves: return "运行时间或并发信息不足，无法安全还原执行波次。" rows = [] for index, wave in enumerate(waves, 1): operations = wave.get("operations") or wave.get("operationIds") or [] rows.append( f"- 第 {index} 波：{len(operations)} 个取数操作" ) return "\n".join(rows) def _retrieval_stage_result(stage: dict[str, Any]) -> str: direct_calls = [ call for group in stage.get("directToolGroups") or [] for call in group.get("calls") or [] ] attempts = [ attempt for run in stage.get("agentRuns") or [] for attempt in run.get("attempts") or [] ] failures = sum( 1 for item in [*direct_calls, *attempts] if item.get("status") in {"failure", "interrupted"} ) hits = sum(1 for item in attempts if item.get("status") == "hit") empty = sum(1 for item in attempts if item.get("status") == "empty") return ( f"阶段状态：{stage.get('status') or '未记录'}。" f"工具读取 {len(direct_calls)} 次，Agent 查询 {len(attempts)} 次；" f"命中 {hits} 次，成功为空 {empty} 次，失败或中断 {failures} 次。" ) def _direct_groups_markdown(groups: list[dict[str, Any]]) -> str: if not groups: return "本次未记录实现 Agent 直接读取工具。" return "\n".join( f"- 第 {index} 组：{'、'.join(item.get('businessLabel') or '相关信息' for item in group.get('sources') or [])}（{int(group.get('callCount') or 0)} 次）" for index, group in enumerate(groups, 1) ) def _agent_runs_markdown(runs: list[dict[str, Any]]) -> str: if not runs: return "本次未记录委派取数 Agent。" return "\n".join( f"- {item.get('businessLabel') or '取数 Agent'}：{int((item.get('querySummary') or {}).get('total') or 0)} 次查询，初筛{'已记录' if (item.get('screening') or {}).get('state') == 'available' else '未完整记录'}" for item in runs ) def _find_round(view: dict[str, Any], round_index: int | None) -> dict[str, Any] | None: return next((item for item in view.get("rounds") or [] if int(item.get("roundIndex") or 0) == int(round_index or 0)), None) def _find_by_id(rows: list[dict[str, Any]], record_id: int) -> dict[str, Any] | None: return next((item for item in rows if int(item.get("id") or 0) == record_id), None) def _suffix_int(value: str) -> int: try: return int(value.rsplit(":", 1)[1]) except (IndexError, ValueError) as exc: raise KeyError(value) from exc def _node_value(node: dict[str, Any] | None) -> str | None: return ((node or {}).get("card") or {}).get("primary", {}).get("value") def _plan_text(plan: dict[str, Any]) -> str: paths = plan.get("paths") or [] parts = [f"方式：{plan.get('mode') or '未记录'}", f"当前保存 {len(paths)} 个方案"] if plan.get("note"): parts.append(f"说明：{plan['note']}") revisions = int(plan.get("runtimeRevisionCount") or 0) if revisions > 1: parts.append(f"运行事件中找到 {revisions} 次规划记录；历史版本只在技术详情中查看。") return "\n".join(parts) def _round_content_changes(branches: list[dict[str, Any]]) -> str: merged = [item for item in branches if item.get("pathType") != "领域信息" and item.get("status") == "merged"] return f"确认采用 {len(merged)} 个内容方案；无法据此还原本轮结束时的完整主脚本快照。" def _branch_outcomes_markdown(branches: list[dict[str, Any]]) -> str: if not branches: return "本轮没有可展示的方案处置记录。" rows = [] for item in branches: outcome = item.get("outcome") or {} line = ( f"- 方案 {item.get('branchId')}：" f"{outcome.get('label') or item.get('status') or '状态未记录'}" ) if outcome.get("reasoning"): line += f"。{business_text(outcome.get('reasoning'))}" rows.append(line) return "\n".join(rows) def _round_domain_changes(branches: list[dict[str, Any]]) -> str: count = sum(int(item.get("output", {}).get("factCount") or 0) for item in branches if item.get("pathType") == "领域信息") return f"新增 {count} 条独立保存的领域事实。" def _branch_facts(branch: dict[str, Any], bundle: dict[str, Any]) -> list[dict[str, Any]]: refs = set(branch.get("output", {}).get("factRefs") or []) return [item for item in bundle.get("domainInfo") or [] if f"domain-info:{item.get('id')}" in refs] def _output_exactness(output: dict[str, Any]) -> str: if output.get("type") == "domain-info": return "exact" historical = output.get("accuracy", {}).get("historical") return "historical-snapshot" if historical == "exact" else "current-only" if output.get("snapshotRef") else "unknown" def _artifact_exactness(artifact: dict[str, Any]) -> str: return "historical-snapshot" if artifact.get("historicalAccuracy") == "exact" else "current-only" def _domain_facts_markdown(rows: list[dict[str, Any]]) -> str: values = [business_text(row.get("content")) for row in rows] return "\n".join(f"- {value}" for value in values if value) def _source_value(value: Any) -> str: if isinstance(value, str): return value if isinstance(value, list): return source_markdown(value) if isinstance(value, dict): return source_markdown([value]) or business_text(value.get("url") or value.get("name")) return ""