from __future__ import annotations import re from collections import Counter, defaultdict from datetime import datetime, timezone from typing import Any from zoneinfo import ZoneInfo from .business_detail_text import branch_task_sections, branch_task_text, business_text from .decision_projection import AgentDecisionProjector from .main_decision_text import ( parse_goal_text, parse_objective_text, plan_mode_label, plan_path_summary, preview as decision_preview, ) from .runtime_event_projection import RuntimeEventProjector _DECISIONS = AgentDecisionProjector() TERMINAL_STATUSES = { "success", "partial", "failed", "stopped", "completed", "cancelled", "cancel", "cancle", } class ExecutionViewBuilder: """Deterministic V8 business narrative projector. The projector consumes plain dictionaries and has no database, HTTP or framework dependencies. Business rows win over runtime events; events may only fill clearly labelled evaluation/activity detail. """ def __init__(self, event_projector: RuntimeEventProjector | None = None): self._event_projector = event_projector or RuntimeEventProjector() def from_bundle(self, bundle: dict[str, Any]) -> dict[str, Any]: record = bundle.get("record") or {} raw_rounds = sorted( bundle.get("rounds") or [], key=lambda item: (_integer(item.get("round_index")) or 0, _integer(item.get("id")) or 0), ) raw_branches = sorted( bundle.get("branches") or [], key=lambda item: ( _integer(item.get("round_index")) or 0, _integer(item.get("branch_id")) or 0, ), ) data_decisions = [ item for item in bundle.get("dataDecisions") or [] if (_integer(item.get("branch_id")) or 0) > 0 ] multipath_decisions = bundle.get("multipathDecisions") or [] domain_info = bundle.get("domainInfo") or [] events = bundle.get("events") or [] valid_rounds = { value for value in (_integer(item.get("round_index")) for item in raw_rounds) if value is not None } valid_branches = { value for value in (_integer(item.get("branch_id")) for item in raw_branches) if value is not None } captured_at = datetime.now(timezone.utc) runtime = self._event_projector.project( events, valid_rounds=valid_rounds, valid_branches=valid_branches, captured_at=captured_at, run_status=str(record.get("status") or "unknown"), script_direction=record.get("script_direction"), rounds=raw_rounds, multipath_decisions=multipath_decisions, ) implementer_tasks = _implementer_tasks_by_branch(events) decisions_by_branch: dict[int, list[dict[str, Any]]] = defaultdict(list) for item in data_decisions: branch_id = _integer(item.get("branch_id")) if branch_id is not None: decisions_by_branch[branch_id].append(item) domain_by_branch: dict[int, list[dict[str, Any]]] = defaultdict(list) domain_by_round: dict[int, list[dict[str, Any]]] = defaultdict(list) for item in domain_info: branch_id = _integer(item.get("branch_id")) round_index = _integer(item.get("round_index")) if branch_id is not None: domain_by_branch[branch_id].append(item) if round_index is not None: domain_by_round[round_index].append(item) multipath_by_round: dict[int, list[dict[str, Any]]] = defaultdict(list) unassigned_business: list[dict[str, Any]] = [] for item in multipath_decisions: round_index = _integer(item.get("round_index")) if round_index in valid_rounds: multipath_by_round[round_index].append(item) else: unassigned_business.append( { "type": "multipath-decision", "id": item.get("id"), "reason": "没有可确认的所属轮次", } ) projected_rounds: list[dict[str, Any]] = [] for index, raw_round in enumerate(raw_rounds): round_index = _integer(raw_round.get("round_index")) or index + 1 branch_rows = [ item for item in raw_branches if _integer(item.get("round_index")) == round_index ] next_round = raw_rounds[index + 1] if index + 1 < len(raw_rounds) else None projected_rounds.append( self._round_story( record, raw_round, branch_rows, decisions_by_branch, domain_by_branch, domain_by_round.get(round_index, []), multipath_by_round.get(round_index, []), runtime, implementer_tasks, next_round, ) ) data_shape = _data_shape( raw_branches, multipath_decisions, domain_info, events ) warnings = _warnings( data_shape=data_shape, ignored_legacy_count=_integer( bundle.get("ignoredLegacyDataDecisionCount") ) or 0, rounds=projected_rounds, runtime_unassigned=runtime.get("unassigned") or [], business_unassigned=unassigned_business, ) artifact = bundle.get("currentArtifact") or {} current_round = projected_rounds[-1]["roundIndex"] if projected_rounds else None view = { "schemaVersion": "8", "capturedAt": captured_at.isoformat(), "dataShape": data_shape, "header": _header(record, current_round), "planning": _planning_analysis( (runtime.get("mainAgentDecisions") or {}).get("planningAnalysis"), ), "objective": _objective( record, (runtime.get("mainAgentDecisions") or {}).get("objective"), ), "rounds": projected_rounds, "finalResult": _final_result(record, projected_rounds, artifact), "unassigned": { "runtimeEvents": runtime.get("unassigned") or [], "businessRecords": unassigned_business, }, "warnings": warnings, } if view["planning"] is None: view.pop("planning") return view def _round_story( self, record: dict[str, Any], raw_round: dict[str, Any], branch_rows: list[dict[str, Any]], decisions_by_branch: dict[int, list[dict[str, Any]]], domain_by_branch: dict[int, list[dict[str, Any]]], round_domain_info: list[dict[str, Any]], multipath_rows: list[dict[str, Any]], runtime: dict[str, Any], implementer_tasks: dict[int, str], next_round: dict[str, Any] | None, ) -> dict[str, Any]: round_index = _integer(raw_round.get("round_index")) or 0 main_decisions = runtime.get("mainAgentDecisions") or {} goal_context = (main_decisions.get("roundGoalsByRound") or {}).get(round_index) plan_context = (main_decisions.get("implementationPlansByRound") or {}).get( round_index ) plan = _plan_projection( raw_round, plan_context, ) branches = [ _branch_story( round_index, row, decisions_by_branch.get(_integer(row.get("branch_id")) or 0, []), domain_by_branch.get(_integer(row.get("branch_id")) or 0, []), runtime.get("retrievalStagesByBranch", {}).get( _integer(row.get("branch_id")) or 0 ), runtime.get("creativeByBranch", {}).get( _integer(row.get("branch_id")) or 0 ), implementer_tasks.get(_integer(row.get("branch_id")) or 0), str(record.get("status") or "unknown"), ) for row in branch_rows ] branch_ids = {branch["branchId"] for branch in branches} decision_batches = [ _decision_batch(round_index, row, branch_ids, branches) for row in sorted(multipath_rows, key=_record_order) ] convergence_batches, unassigned_reviews = _convergence_batches( round_index, runtime.get("multipathReviewsByRound", {}).get(round_index, []), decision_batches, str(record.get("status") or "unknown"), ) if not convergence_batches and branches: convergence_batches = [ _convergence_batch( round_index, None, None, "records-missing", str(record.get("status") or "unknown"), branch_ids=sorted(branch_ids), ) ] _annotate_unscoped_round_reviews( convergence_batches, runtime.get("unassigned") or [], round_index, ) if unassigned_reviews: runtime.setdefault("unassigned", []).extend(unassigned_reviews) overall_evaluation = _round_evaluation( round_index, runtime.get("overallEvaluationsByRound", {}).get(round_index, []), str(record.get("status") or "unknown"), ) state = _round_state(record, raw_round, overall_evaluation) result = _round_result( record, round_index, branches, round_domain_info, state, next_round, ) goal_summary = _goal_summary(raw_round.get("goal")) round_summary = _round_summary( round_index=round_index, goal=goal_summary, plan=plan, branches=branches, batches=decision_batches, domain_info=round_domain_info, run_status=str(record.get("status") or "unknown"), ) goal_parsed = parse_goal_text(raw_round.get("goal")) goal_lines: list[dict[str, Any]] = [] focus_items = goal_parsed.get("focusItems") or [] if focus_items: focus_summary = "；".join(str(item) for item in focus_items[:2]) if len(focus_items) > 2: focus_summary += f"；另有 {len(focus_items) - 2} 项，详情查看" goal_lines.append(_line("focus", "本轮聚焦", focus_summary)) visible_inputs = _context_input_labels(goal_context) if visible_inputs: goal_lines.append(_line("visible-before", "形成前可见", " · ".join(visible_inputs))) goal_decision = _direction_decision( id=f"round:{round_index}:goal", subtype="round-goal", context=goal_context, decision_items=(goal_context or {}).get("decisionItems") or [goal_parsed["headline"]], primary_label="要解决", primary_value=goal_parsed["headline"], secondary=goal_lines, prompt_ref="prompt:main", ) return { "id": f"round:{round_index}", "roundIndex": round_index, "state": state, "summary": round_summary, "goal": _story( id=f"round:{round_index}:goal", role="goal", kind="decision", title="本轮目标", label="要解决", value=goal_parsed["headline"], secondary=goal_lines, detail_ref=f"round:{round_index}:goal", source_notice=None if raw_round.get("goal") else "record-missing", decision=goal_decision, ), "plan": plan, "branches": branches, "convergenceBatches": convergence_batches, "overallEvaluation": overall_evaluation, "result": result, "detailRef": f"round:{round_index}", } def _header(record: dict[str, Any], current_round: int | None) -> dict[str, Any]: agent_config = record.get("agent_config") model = None if isinstance(agent_config, dict): model = agent_config.get("model_name") or agent_config.get("model") return { "id": str(record.get("id") or ""), "title": f"脚本构建 #{record.get('id')}", "status": str(record.get("status") or "unknown"), "summary": _plain(record.get("summary")), "errorMessage": _plain(record.get("error_message")), "model": model, "totalCost": record.get("cost_usd"), "currentRound": current_round, "createdAt": record.get("start_time"), "updatedAt": record.get("end_time"), "direction": { "currentValue": _plain(record.get("script_direction")), "note": "这是数据库当前保存的创作方向。", }, } def _objective( record: dict[str, Any], decision_context: dict[str, Any] | None ) -> dict[str, Any]: raw_direction = record.get("script_direction") direction = _plain(raw_direction) # Section parsing needs the persisted Markdown line breaks. ``direction`` # is only the compact truthy/display fallback used by the surrounding view. parsed = parse_objective_text(raw_direction) secondary: list[dict[str, Any]] = [] input_labels = _context_input_labels(decision_context, relation="direct-read") if input_labels: secondary.append( _line("inputs", "决策前读取", " · ".join(input_labels)) ) structure: list[str] = [] target_count = int(parsed.get("targetCount") or 0) dimension_count = int(parsed.get("domainDimensionCount") or 0) if target_count: structure.append(f"{target_count} 个创作目标") if dimension_count: structure.append(f"{dimension_count} 个领域评估维度") if structure: secondary.append(_line("structure", "目标结构", " · ".join(structure))) decision = _direction_decision( id="run:objective", subtype="objective", context=decision_context, decision_items=(decision_context or {}).get("decisionItems") or ([parsed.get("headline")] if parsed.get("headline") else []), primary_label="核心目标", primary_value=parsed.get("headline"), secondary=secondary, prompt_ref="prompt:main", ) return _story( id="run:objective", role="objective", kind="decision" if direction else "missing", title="创作目标", label="核心目标", value=parsed.get("headline") or "创作方向未记录", secondary=secondary, detail_ref="run:objective", source_notice=None if direction else "record-missing", decision=decision, ) def _planning_analysis(context: dict[str, Any] | None) -> dict[str, Any] | None: if not context or context.get("completeness") != "complete" or not context.get("eventRef"): return None secondary = [ _line("plan", "执行计划", context.get("plan")), _line("next", "下一步", context.get("action")), ] return _story( id="run:planning-analysis", role="planning-analysis", kind="execution", title="创作规划解析", label="规划概要", value=context.get("summary") or "主 Agent 已完成首次创作规划", secondary=secondary, detail_ref=context.get("eventRef"), status="completed", ) def _plan_projection( raw_round: dict[str, Any], decision_context: dict[str, Any] | None ) -> dict[str, Any]: round_index = _integer(raw_round.get("round_index")) or 0 paths = raw_round.get("multipath_plan") if not isinstance(paths, list): paths = [] raw_mode = _plain(raw_round.get("race_or_divide")) mode = plan_mode_label(raw_mode, len(paths)) summary = ( f"{mode or '方式未记录'} · {len(paths)} 个方案" if paths else "多路规划未记录" ) decision = _direction_decision( id=f"round:{round_index}:plan", subtype="implementation-plan", context=decision_context, decision_items=(decision_context or {}).get("decisionItems") or [summary], primary_label="实现方式", primary_value=summary, secondary=_plan_card_lines(paths, raw_round.get("plan_note")), prompt_ref="prompt:main", ) node = _story( id=f"round:{round_index}:plan", role="plan", kind="decision" if paths else "missing", title="实现规划", label="实现方式", value=summary, secondary=_plan_card_lines(paths, raw_round.get("plan_note")), detail_ref=f"round:{round_index}:plan", source_notice=None if paths else "record-missing", decision=decision, ) return { "mode": mode, "rawMode": raw_mode, "paths": paths, "note": _plain(raw_round.get("plan_note")), "currentOnly": True, "runtimeRevisionCount": len((decision_context or {}).get("revisionRefs") or []), "runtimeRevisionRefs": list((decision_context or {}).get("revisionRefs") or []), "node": node, } def _branch_story( round_index: int, row: dict[str, Any], decisions: list[dict[str, Any]], domain_info: list[dict[str, Any]], retrieval_stage: dict[str, Any] | None, creative_decision: dict[str, Any] | None, implementer_task: str | None, run_status: str, ) -> dict[str, Any]: branch_id = _integer(row.get("branch_id")) or 0 path_type = str(row.get("path_type") or "unknown") if path_type not in {"内容", "领域信息"}: path_type = "unknown" task = ( branch_task_text(implementer_task) or branch_task_text(row.get("impl_task")) or business_text(row.get("target")) ) retrieval_stage = retrieval_stage or _missing_retrieval_stage( round_index, branch_id, run_status ) projected_data_decisions = [ _DECISIONS.data_tradeoff( { **item, "decision": business_text(item.get("decision")), "reasoning": business_text(item.get("reasoning")), "id": f"data-decision:{item.get('id')}", "detailRef": f"data-decision:{item.get('id')}", "promptRef": _implementer_prompt_ref(retrieval_stage), } ) for item in decisions ] for item, projection in zip(decisions, projected_data_decisions): full_conclusion = business_text(item.get("decision")) if full_conclusion: projection.setdefault("card", {}).setdefault("primary", {})[ "value" ] = full_conclusion decision_nodes = [ _story( id=f"data-decision:{item.get('id')}", role="data-decision", kind="decision", title="数据取舍", label="取舍结论", value=projection["card"]["primary"]["value"], secondary=projection["card"]["secondary"], detail_ref=f"data-decision:{item.get('id')}", decision=projection, ) for item, projection in zip(decisions, projected_data_decisions) ] output = _branch_output(row, path_type, domain_info) if creative_decision and output.get("type") == "script-artifact": creative_decision = dict(creative_decision) explicit_basis = [ { "label": business_text(item.get("decision")) or "已记录一项数据取舍", "value": business_text(item.get("reasoning")) or None, "detailRef": f"data-decision:{item.get('id')}", } for item in decisions if item.get("decision") ] creative_decision["artifactRef"] = output.get("snapshotRef") creative_decision["body"] = { **(creative_decision.get("body") or {}), "output": output.get("summary"), "explicitBasis": explicit_basis, "candidateSnapshot": { "snapshotRef": output.get("snapshotRef"), "paragraphCount": output.get("paragraphCount"), "elementCount": output.get("elementCount"), "linkCount": output.get("linkCount"), "exactness": _candidate_output_exactness(output), }, } detail = creative_decision.get("detail") if isinstance(detail, dict): blocks = [ block for block in detail.get("blocks") or [] if not ( isinstance(block, dict) and block.get("title") in {"实际产出", "产出的候选表"} ) ] if explicit_basis: blocks.insert( 0, { "type": "items", "title": "明确采用的数据取舍", "items": explicit_basis, "visualRole": "evidence", "presentation": "list", "collapsible": True, }, ) blocks.append( { "type": "summary", "title": "产出的候选表", "value": output.get("summary"), "visualRole": "key-result", "presentation": "prose", } ) creative_decision["detail"] = {**detail, "blocks": blocks} secondary = list((creative_decision.get("card") or {}).get("secondary") or []) if explicit_basis: secondary.append(_line("basis", "采用数据", f"{len(explicit_basis)} 项明确取舍")) secondary.append(_line("output", "产出的候选表", output.get("summary"))) creative_decision["card"] = { **(creative_decision.get("card") or {}), "secondary": secondary[:3], } output["node"] = _story( id=creative_decision["id"], role="candidate-output", kind="decision", title="产生候选创作表", label="创作处理", value=creative_decision["card"]["primary"]["value"], secondary=creative_decision["card"]["secondary"], detail_ref=creative_decision["detailRef"], artifact_ref=output.get("snapshotRef"), has_changes=bool(output.get("snapshotRef")), decision=creative_decision, ) status = str(row.get("status") or "open") disposition_context = _branch_disposition_context(status, run_status) title = _plain(row.get("target")) or f"方案 {branch_id}" return { "id": f"round:{round_index}:branch:{branch_id}", "branchId": branch_id, "pathType": path_type, "status": status, "title": title, "summary": { "task": task or "实现任务未记录", "output": output["summary"], }, "task": _branch_task_story( id=f"round:{round_index}:branch:{branch_id}:task", task=task, detail_ref=f"round:{round_index}:branch:{branch_id}:task", ), "retrievalStage": retrieval_stage, "dataDecisions": [ { "id": f"data-decision:{item.get('id')}", "recordId": item.get("id"), "decision": _plain(item.get("decision")), "reasoning": _plain(item.get("reasoning")), "sources": item.get("sources") or [], "createdAt": item.get("created_at"), "node": node, "decisionProjection": projection, } for item, node, projection in zip( decisions, decision_nodes, projected_data_decisions ) ], "output": output, "outcome": { "status": status, "label": disposition_context["label"], "reasoning": _plain(row.get("decision_reasoning")), "sourceNotice": disposition_context["sourceNotice"], }, "detailRef": f"round:{round_index}:branch:{branch_id}", } def _branch_output( row: dict[str, Any], path_type: str, domain_info: list[dict[str, Any]] ) -> dict[str, Any]: branch_id = _integer(row.get("branch_id")) or 0 round_index = _integer(row.get("round_index")) or 0 if path_type == "领域信息": facts = [ item for item in domain_info if _integer(item.get("branch_id")) == branch_id and _integer(item.get("round_index")) in {None, round_index} ] summary = ( f"新增 {len(facts)} 条已核实领域事实" if facts else "已查询完成，本方案没有新增领域事实" ) node = _story( id=f"round:{round_index}:branch:{branch_id}:output", role="domain-output", kind="result", title="领域事实", label="产出", value=summary, detail_ref=f"round:{round_index}:branch:{branch_id}:output", has_changes=bool(facts), source_notice=None, ) return { "type": "domain-info", "summary": summary, "factCount": len(facts), "factRefs": [f"domain-info:{item.get('id')}" for item in facts], "node": node, } candidate = row.get("candidate_snapshot") or {} snapshot = candidate.get("snapshot") if isinstance(candidate, dict) else None counts = _snapshot_counts(snapshot) if any(counts.values()): summary = ( f"候选脚本包含 {counts['paragraphs']} 个段落、" f"{counts['elements']} 个元素" ) elif row.get("self_assessment"): assessment = business_text(row.get("self_assessment")) first_item = next( (line.strip(" -*") for line in assessment.splitlines() if line.strip()), assessment, ) summary = f"实现 Agent 自评：{first_item}" else: summary = "候选脚本内容未记录" has_output = bool(snapshot) or bool(row.get("self_assessment")) node = _story( id=f"round:{round_index}:branch:{branch_id}:output", role="candidate-output", kind="result" if has_output else "missing", title="候选产出", label="产出", value=summary, detail_ref=f"round:{round_index}:branch:{branch_id}:output", has_changes=bool(snapshot), source_notice=None if has_output else "record-missing", artifact_ref=f"artifact:branch:{branch_id}" if snapshot else None, ) return { "type": "script-artifact" if has_output else "missing", "summary": summary, "snapshotRef": f"artifact:branch:{branch_id}" if snapshot else None, "paragraphCount": counts["paragraphs"], "elementCount": counts["elements"], "linkCount": counts["links"], "accuracy": { "historical": candidate.get("historicalAccuracy") if isinstance(candidate, dict) else "unknown", "currentProjection": candidate.get("currentProjectionAccuracy") if isinstance(candidate, dict) else "unknown", }, "node": node, } def _branch_task_story( *, id: str, task: str | None, detail_ref: str ) -> dict[str, Any]: if not task: return _story( id=id, role="branch-task", kind="missing", title="实现任务", label="实现任务", value="实现任务未记录", detail_ref=detail_ref, source_notice="record-missing", ) sections = branch_task_sections(task) preferred = next( ( item for kind in ("scope", "method", "goal", "notes") for item in sections if item.get("kind") == kind and item.get("content") ), None, ) primary = (preferred or {}).get("content") or task secondary = [ _line(str(item.get("kind") or index), item.get("title") or "任务说明", item.get("content")) for index, item in enumerate(sections, 1) if item is not preferred and item.get("content") ][:3] node = _story( id=id, role="branch-task", kind="execution", title="实现任务", label=(preferred or {}).get("title") or "实现任务", value=primary, secondary=secondary, detail_ref=detail_ref, ) node["taskStructure"] = { "sectionCount": len(sections), "sectionKinds": [item.get("kind") for item in sections], } return node def _candidate_output_exactness(output: dict[str, Any]) -> str: historical = (output.get("accuracy") or {}).get("historical") if historical == "exact": return "historical-snapshot" return "current-only" if output.get("snapshotRef") else "unknown" def _missing_retrieval_stage( round_index: int, branch_id: int, run_status: str ) -> dict[str, Any]: running = str(run_status or "").lower() not in TERMINAL_STATUSES return { "id": f"round:{round_index}:branch:{branch_id}:retrieval", "roundIndex": round_index, "branchId": branch_id, "implementerEventId": None, "observedMode": "unknown", "waves": [], "directToolGroups": [], "agentRuns": [], "status": "running" if running else "missing", "startedAt": None, "endedAt": None, "detailRef": f"round:{round_index}:branch:{branch_id}:retrieval", "sourceNotice": None if running else "record-missing", } def _decision_batch( round_index: int, row: dict[str, Any], valid_branch_ids: set[int], branches: list[dict[str, Any]], ) -> dict[str, Any]: branch_ids = _ids(row.get("branch_ids")) missing = [branch_id for branch_id in branch_ids if branch_id not in valid_branch_ids] record_id = row.get("id") decision = _plain(row.get("decision")) or "决策内容未记录" display_decision = _business_terms(decision) secondary = [ _line( "branches", "比较方案", "、".join(f"方案 {branch_id}" for branch_id in branch_ids) or "涉及方案未记录", ) ] if row.get("reasoning"): secondary.append( _line("reason", "理由", _preview(row.get("reasoning"), 120)) ) branch_by_id = {branch["branchId"]: branch for branch in branches} outcomes = [ { "branchId": branch_id, "status": branch_by_id.get(branch_id, {}).get("outcome", {}).get("status", "unknown"), "label": branch_by_id.get(branch_id, {}).get("outcome", {}).get("label", "状态未记录"), "reasoning": branch_by_id.get(branch_id, {}).get("outcome", {}).get("reasoning"), } for branch_id in branch_ids ] if outcomes: secondary.insert( 1, _line( "outcomes", "各方案结果", "；".join( f"方案 {item['branchId']} · {item['label']}" for item in outcomes ), ), ) projection = _DECISIONS.multipath_tradeoff( { "id": f"multipath-decision:{record_id}", "detailRef": f"multipath-decision:{record_id}", "promptRef": "prompt:main", "branchIds": branch_ids, "scopeItems": [f"方案 {branch_id}" for branch_id in branch_ids], "decision": display_decision, "reasoning": business_text(row.get("reasoning")), "selected": [ f"方案 {item['branchId']}" for item in outcomes if item["status"] == "merged" ], "parked": [ f"方案 {item['branchId']}" for item in outcomes if item["status"] == "parked" ], "rejected": [ f"方案 {item['branchId']}" for item in outcomes if item["status"] == "discarded" ], "completeness": "partial" if missing or not branch_ids else "complete", } ) projection.setdefault("card", {}).setdefault("primary", {})[ "value" ] = display_decision return { "id": f"multipath-decision:{record_id}", "recordId": record_id, "roundIndex": round_index, "branchIds": branch_ids, "decision": decision, "reasoning": _plain(row.get("reasoning")), "createdAt": row.get("created_at"), "completeness": "partial" if missing or not branch_ids else "complete", "missingBranchIds": missing, "branchOutcomes": outcomes, "decisionProjection": projection, "node": _story( id=f"multipath-decision:{record_id}", role="multipath-decision", kind="decision", title="主 Agent 多路决策", label="决定", value=projection["card"]["primary"]["value"], secondary=projection["card"]["secondary"], detail_ref=f"multipath-decision:{record_id}", source_notice="record-missing" if missing else None, decision=projection, ), } def _convergence_batches( round_index: int, reviews: list[dict[str, Any]], decisions: list[dict[str, Any]], run_status: str, ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: ordered_reviews = sorted(reviews, key=_runtime_order) ordered_decisions = sorted(decisions, key=_decision_order) unmatched_reviews = set(range(len(ordered_reviews))) batches: list[dict[str, Any]] = [] for decision in ordered_decisions: decision_set = set(decision.get("branchIds") or []) decision_time = _date_value(decision.get("createdAt")) candidates = [ index for index in unmatched_reviews if set(ordered_reviews[index].get("branchIds") or []) == decision_set and _not_after(ordered_reviews[index].get("endedAt"), decision_time) ] review_index = max( candidates, key=lambda index: _runtime_order(ordered_reviews[index]), default=None, ) review = ordered_reviews[review_index] if review_index is not None else None if review_index is not None: unmatched_reviews.remove(review_index) batches.append( _convergence_batch( round_index, review, decision, "exact-branch-set-and-time" if review else "decision-only", run_status, ) ) for index in sorted(unmatched_reviews, key=lambda value: _runtime_order(ordered_reviews[value])): review = ordered_reviews[index] batches.append( _convergence_batch( round_index, review, None, "review-only", run_status ) ) batches.sort(key=_convergence_order) return batches, [] def _annotate_unscoped_round_reviews( batches: list[dict[str, Any]], unassigned: list[dict[str, Any]], round_index: int, ) -> None: """Keep unscoped reviews visible without fabricating a candidate link.""" reviews = [ item for item in unassigned if str(item.get("name") or "") == "script_multipath_evaluator" and _integer(item.get("roundIndex")) == round_index and str(item.get("association") or "") == "missing-branch-scope" ] if not reviews: return refs = [item.get("detailRef") for item in reviews if item.get("detailRef")] for batch in batches: missing = batch.get("missingReview") if not isinstance(missing, dict): continue missing["card"] = { "primary": { "label": "评审关联", "value": "发现同轮评审，但无法安全关联到具体候选", }, "secondary": [ _line("records", "未关联记录", f"{len(refs)} 条评审记录") ], } missing["sourceNotice"] = "association-incomplete" missing["unscopedReviewRefs"] = refs def _convergence_batch( round_index: int, review: dict[str, Any] | None, decision: dict[str, Any] | None, association: str, run_status: str, branch_ids: list[int] | None = None, ) -> dict[str, Any]: branch_ids = branch_ids or list( dict.fromkeys( (review or {}).get("branchIds") or (decision or {}).get("branchIds") or [] ) ) suffix = ( f"review-{review.get('eventId')}" if review else f"decision-{decision.get('recordId')}" if decision else "records-missing" ) missing_context = _missing_multipath_context(run_status) if review is not None and not review.get("decisionProjection"): review = dict(review) review["decisionProjection"] = _evaluation_decision( review, subtype="multipath-evaluation", detail_ref=review.get("detailRef"), prompt_ref=review.get("promptRef"), ) missing_review = None if review is None: missing_review_id = f"round:{round_index}:{suffix}:review-missing" missing_review = _story( id=missing_review_id, role="multipath-review", kind=missing_context["kind"], title="多方案评审", label="评审记录", value="未找到可与本次决策安全关联的多方案评审记录", detail_ref=missing_review_id, source_notice=missing_context["sourceNotice"], status=missing_context.get("status"), ) missing_decision = None if decision is None: missing_decision_id = f"round:{round_index}:{suffix}:decision-missing" missing_decision = _story( id=missing_decision_id, role="multipath-decision", kind=missing_context["kind"], title="主 Agent 多路决策", label=missing_context["label"], value=missing_context["value"], detail_ref=missing_decision_id, source_notice=missing_context["sourceNotice"], status=missing_context.get("status"), ) if review is not None and decision is not None: _attach_review_comparison(review, decision, association) return { "id": f"round:{round_index}:convergence:{suffix}", "roundIndex": round_index, "branchIds": branch_ids, "review": review, "decision": decision, "missingReview": missing_review, "missingDecision": missing_decision, "association": association, } def _round_evaluation( round_index: int, evaluations: list[dict[str, Any]], run_status: str ) -> dict[str, Any]: if not evaluations: context = _missing_evaluation_context(run_status) status = context.get("status") if not status: status = ( "not-evaluated" if context["sourceNotice"] == "record-missing" else "unknown" ) return _story( id=f"round:{round_index}:evaluation", role="round-evaluation", kind=context["kind"], title="主脚本整体评审", label=context["label"], value=context["value"], detail_ref=f"round:{round_index}:evaluation", source_notice=context["sourceNotice"], status=status, ) latest = evaluations[-1] conclusion = _plain(latest.get("conclusion")) projection = _evaluation_decision( latest, subtype="overall-evaluation", detail_ref=latest.get("detailRef"), prompt_ref=latest.get("promptRef"), ) return _story( id=f"round:{round_index}:evaluation", role="round-evaluation", kind="decision" if conclusion else "missing", title="主脚本整体评审", label="评审结论", value=projection["card"]["primary"]["value"], secondary=projection["card"]["secondary"], detail_ref=latest.get("detailRef"), source_notice="runtime-associated", status=_evaluation_state(conclusion), decision=projection, ) def _evaluation_decision( evaluation: dict[str, Any], *, subtype: str, detail_ref: Any, prompt_ref: Any, ) -> dict[str, Any]: projection = _DECISIONS.evaluation( { "id": evaluation.get("id") or detail_ref, "detailRef": detail_ref, "promptRef": prompt_ref, "subtype": subtype, "subjects": evaluation.get("subjects") or [ f"方案 {value}" for value in evaluation.get("branchIds") or [] ], "criteria": evaluation.get("criteria") or [], "itemConclusions": evaluation.get("itemConclusions") or evaluation.get("branchConclusions") or [], "branchEvaluations": evaluation.get("branchEvaluations") or [], "comparisonRows": evaluation.get("comparisonRows") or [], "achievements": evaluation.get("achievements") or [], "problems": evaluation.get("problems") or [], "conclusion": evaluation.get("conclusion"), "recommendation": evaluation.get("recommendation") or evaluation.get("comparison"), "nextGoal": evaluation.get("nextGoal"), "completeness": "partial" if evaluation.get("notices") else "complete", } ) full_conclusion = business_text( evaluation.get("conclusion") or evaluation.get("recommendation") or evaluation.get("comparison") ) if full_conclusion: projection.setdefault("card", {}).setdefault("primary", {})[ "value" ] = full_conclusion return projection def _attach_review_comparison( review: dict[str, Any], decision: dict[str, Any], association: str ) -> None: projection = decision.get("decisionProjection") if not isinstance(projection, dict): return conclusions = { int(item.get("branchId") or 0): item for item in review.get("branchConclusions") or [] if isinstance(item, dict) } outcomes = { int(item.get("branchId") or 0): item for item in decision.get("branchOutcomes") or [] if isinstance(item, dict) } reasoning = business_text(decision.get("reasoning")) rows: list[dict[str, Any]] = [] for branch_id in decision.get("branchIds") or []: row = { "candidate": f"方案 {branch_id}", "recommendation": (conclusions.get(branch_id) or {}).get("conclusion") or review.get("recommendation") or "评审未给出单独结论", "finalDecision": (outcomes.get(branch_id) or {}).get("label") or "最终状态未记录", } handling = _explicit_recommendation_handling(reasoning, branch_id) if handling: row["handling"] = handling rows.append(row) projection.setdefault("body", {})["reviewComparison"] = rows projection["body"]["reviewAssociation"] = association review_input = { "label": "同批候选方案的评审建议", "summary": business_text( review.get("recommendation") or review.get("comparison") ), "observedRelation": "returned-to-actor", "decisionUse": "not-recorded", "detailRef": review.get("detailRef"), } projection.setdefault("inputs", []).append(review_input) projection.setdefault("notices", []).append( { "code": "same-batch-association", "message": "评审与最终决定按同一批候选方案和发生顺序关联；系统没有保存直接的一对一关系。", } ) if rows: blocks = projection.setdefault("detail", {}).setdefault("blocks", []) blocks.insert( 0, { "type": "items", "title": "参与判断", "items": [ { "label": review_input["label"], "value": review_input["summary"], "note": "可确认评审在最终决定前返回；是否采用应以下方的逐方案对照为准。", "detailRef": review_input["detailRef"], } ], "visualRole": "evidence", "presentation": "list", "collapsible": True, }, ) blocks.insert( 3, { "type": "comparison", "title": "评审建议与最终决定", "rows": rows, "visualRole": "section", "presentation": "comparison", }, ) def _explicit_recommendation_handling(reasoning: str, branch_id: int) -> str | None: if not reasoning: return None for sentence in re.split(r"[。；;\n]+", reasoning): if not re.search(rf"方案\s*{branch_id}\b", sentence): continue if any( token in sentence for token in ( "采纳", "吸收", "保留", "延后", "转入", "下一轮", "转译", "不直接合并", "未采用", "拒绝", "不采纳", ) ): return sentence.strip() return None _DATABASE_TIMEZONE = ZoneInfo("Asia/Shanghai") def _date_value(value: Any) -> datetime | None: if isinstance(value, datetime): parsed = value elif isinstance(value, str) and value: try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: return None else: return None if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=_DATABASE_TIMEZONE) return parsed.astimezone(timezone.utc) def _not_after(value: Any, boundary: datetime | None) -> bool: current = _date_value(value) return boundary is None or current is None or current <= boundary def _record_order(row: dict[str, Any]) -> tuple[str, int]: return (str(row.get("created_at") or ""), _integer(row.get("id")) or 0) def _runtime_order(row: dict[str, Any]) -> tuple[str, int]: return ( str(row.get("startedAt") or row.get("endedAt") or ""), _integer(row.get("eventId")) or 0, ) def _decision_order(row: dict[str, Any]) -> tuple[str, int]: return (str(row.get("createdAt") or ""), _integer(row.get("recordId")) or 0) def _convergence_order(row: dict[str, Any]) -> tuple[str, int]: review = row.get("review") or {} decision = row.get("decision") or {} return ( str(review.get("startedAt") or decision.get("createdAt") or ""), _integer(review.get("eventId") or decision.get("recordId")) or 0, ) def _round_result( record: dict[str, Any], round_index: int, branches: list[dict[str, Any]], domain_info: list[dict[str, Any]], state: str, next_round: dict[str, Any] | None, ) -> dict[str, Any]: merged_content = [ branch for branch in branches if branch["pathType"] != "领域信息" and branch["status"] == "merged" ] script_summary = ( f"确认采用 {len(merged_content)} 个内容方案" if merged_content else "没有可确认的内容方案合入记录" ) domain_summary = ( f"新增 {len(domain_info)} 条领域事实" if domain_info else "本轮没有新增领域事实记录" ) if next_round: next_step = f"数据库已记录第 {next_round.get('round_index')} 轮" elif str(record.get("status") or "").lower() in TERMINAL_STATUSES: next_step = "构建已经结束" else: next_step = "等待本轮收口" primary = f"{script_summary}；{domain_summary}" return _story( id=f"round:{round_index}:result", role="round-result", kind="result", title="本轮产出", label="产出", value=primary, secondary=[_line("next", "下一步", next_step)], detail_ref=f"round:{round_index}:result", artifact_ref=f"artifact:round:{round_index}", has_changes=bool(merged_content or domain_info), status=state, ) def _round_state( record: dict[str, Any], raw_round: dict[str, Any], evaluation: dict[str, Any] | None, ) -> str: if evaluation and evaluation.get("status") not in {None, "unknown"}: return str(evaluation["status"]) run_status = str(record.get("status") or "").lower() if str(raw_round.get("status") or "").lower() == "open" and run_status not in TERMINAL_STATUSES: return "running" return "unknown" def _evaluation_state(conclusion: str | None) -> str: text = str(conclusion or "") if any(token in text for token in ("未通过", "不通过", "回退", "重试")): return "needs-retry" if "部分通过" in text: return "partially-passed" if "通过" in text: return "passed" return "unknown" def _final_result( record: dict[str, Any], rounds: list[dict[str, Any]], artifact: dict[str, Any] ) -> dict[str, Any]: branches = [branch for round_ in rounds for branch in round_["branches"]] statuses = Counter(branch["status"] for branch in branches) counts = _snapshot_counts(artifact) status = str(record.get("status") or "unknown") summary = _plain(record.get("summary")) error = _plain(record.get("error_message")) primary = summary or error or _final_status_text(status) return { "id": "run:final-result", "role": "final-result", "kind": "result", "title": "最终结果", "status": status, "terminal": status.lower() in TERMINAL_STATUSES, "roundCount": len(rounds), "branchCounts": { "total": len(branches), "merged": statuses.get("merged", 0), "parked": statuses.get("parked", 0), "discarded": statuses.get("discarded", 0), "open": statuses.get("open", 0), }, "summary": _plain(record.get("summary")), "errorMessage": _plain(record.get("error_message")), "artifact": { "snapshotRef": "artifact:base:current", "paragraphCount": counts["paragraphs"], "elementCount": counts["elements"], "linkCount": counts["links"], "historicalVersion": "current-only", }, "card": { "primary": {"label": "构建结论", "value": primary}, "secondary": [ _line("rounds", "轮次", f"{len(rounds)} 轮"), _line( "branches", "方案处置", f"采用 {statuses.get('merged', 0)} · 暂存 {statuses.get('parked', 0)} · 未采用 {statuses.get('discarded', 0)}", ), _line( "artifact", "主脚本规模", f"{counts['paragraphs']} 段落 · {counts['elements']} 元素 · {counts['links']} 关联", ), ], }, "artifactRef": "artifact:base:current", "detailRef": "run:final-result", } def _final_status_text(status: str) -> str: return { "success": "构建已完成", "completed": "构建已完成", "partial": "构建部分完成", "failed": "构建失败", "stopped": "构建已停止", "cancelled": "构建已取消", "running": "构建进行中", }.get(str(status or "").lower(), "构建状态已记录") def _story( *, id: str, role: str, kind: str, title: str, label: str, value: Any, secondary: list[dict[str, Any]] | None = None, detail_ref: str | None = None, has_changes: bool = False, source_notice: str | None = None, status: str | None = None, artifact_ref: str | None = None, decision: dict[str, Any] | None = None, ) -> dict[str, Any]: node = { "id": id, "role": role, "kind": kind, "title": title, "card": (decision or {}).get("card") or { # Callers pass a deliberately selected card field. Keep that field # intact; line clamping is a view concern, not a data projection. "primary": {"label": label, "value": _plain(value)}, "secondary": [item for item in (secondary or []) if item.get("value")], }, "hasChanges": bool(has_changes), } if detail_ref: node["detailRef"] = detail_ref if source_notice: node["sourceNotice"] = source_notice if status: node["status"] = status if artifact_ref: node["artifactRef"] = artifact_ref if decision: node["decision"] = decision return node def _line(key: str, label: str, value: Any, tone: str = "normal") -> dict[str, Any]: return {"key": key, "label": label, "value": _plain(value), "tone": tone} def _plan_card_lines(paths: list[Any], note: Any) -> list[dict[str, Any]]: lines: list[dict[str, Any]] = [] if paths: route_values = [plan_path_summary(path, index) for index, path in enumerate(paths[:2], 1)] route_summary = "；".join(route_values) if len(paths) > 2: route_summary += f"；另有 {len(paths) - 2} 路，详情查看" lines.append(_line("routes", "各路任务", route_summary)) reason = _plain(note) if reason: lines.append(_line("reason", "规划理由", reason)) return lines[:2] def _context_input_labels( context: dict[str, Any] | None, *, relation: str | None = None ) -> list[str]: result: list[str] = [] for item in (context or {}).get("inputs") or []: if not isinstance(item, dict): continue if relation and item.get("relation") != relation: continue label = _plain(item.get("label")) if label and label not in result: result.append(label) return result[:3] def _direction_decision( *, id: str, subtype: str, context: dict[str, Any] | None, decision_items: list[Any], primary_label: str, primary_value: Any, secondary: list[dict[str, Any]], prompt_ref: str, ) -> dict[str, Any]: context = context if isinstance(context, dict) else {} decision = _DECISIONS.direction( { "id": id, "detailRef": id, "promptRef": prompt_ref, "subtype": subtype, "decisionItems": [str(item) for item in decision_items if item], "explicitReasoning": context.get("explicitReasoning"), "constraints": context.get("constraints") or [], "inputs": context.get("inputs") or [], "sourceDocument": context.get("sourceDocument"), "implementationPlan": { "mode": context.get("displayMode"), "rawMode": context.get("rawMode"), "routes": context.get("routeItems") or [], "reasoning": context.get("explicitReasoning"), } if subtype == "implementation-plan" else None, "completeness": context.get("completeness"), } ) decision["card"] = { "primary": { "label": primary_label, "value": _plain(primary_value) or "决策结论未记录", }, "secondary": [item for item in secondary if item.get("value")][:3], } revisions = [] for revision in context.get("revisions") or []: if not isinstance(revision, dict): continue paths = revision.get("paths") if isinstance(revision.get("paths"), list) else [] parts = [str(revision.get("mode"))] if revision.get("mode") else [] if paths: parts.append(f"{len(paths)} 路方案") revisions.append( { "current": bool(revision.get("current")), "detailRef": revision.get("detailRef"), "summary": " · ".join(parts) or "已保存一次版本", } ) if revisions: decision["body"]["revisions"] = revisions decision["technicalRefs"] = list(context.get("technicalRefs") or []) return decision def _implementer_prompt_ref(retrieval_stage: dict[str, Any] | None) -> str | None: event_id = _integer((retrieval_stage or {}).get("implementerEventId")) return f"prompt:event:{event_id}" if event_id is not None else None def _snapshot_counts(value: Any) -> dict[str, int]: if isinstance(value, dict) and isinstance(value.get("snapshot"), dict): value = value["snapshot"] if not isinstance(value, dict): return {"paragraphs": 0, "elements": 0, "links": 0} stats = value.get("stats") if isinstance(value.get("stats"), dict) else {} if stats: return { "paragraphs": int(stats.get("paragraph_count") or 0), "elements": int(stats.get("element_count") or 0), "links": int(stats.get("link_count") or 0), } return { "paragraphs": len(value.get("paragraphs") or []), "elements": len(value.get("elements") or []), "links": len(value.get("paragraphElements") or value.get("paragraph_elements") or []), } def _ids(value: Any) -> list[int]: if not isinstance(value, list): return [] result: list[int] = [] for item in value: number = _integer(item) if number is not None and number not in result: result.append(number) return result def _implementer_tasks_by_branch( events: list[dict[str, Any]], ) -> dict[int, str]: """Return complete dispatch tasks when the lightweight event carries them. Historical bundles may only retain previews; callers deliberately keep the branch business record as a labelled fallback in that case. """ result: dict[int, str] = {} for event in events: if ( str(event.get("event_type") or "") != "agent_invoke" or str(event.get("event_name") or "") != "script_implementer" ): continue branch_id = _integer(event.get("branch_id")) if branch_id is None: continue data = event.get("inputData") task = data.get("task") if isinstance(data, dict) else None if not task: wrapped = event.get("input") content = wrapped.get("content") if isinstance(wrapped, dict) else None task = content.get("task") if isinstance(content, dict) else None text = branch_task_text(task) if text: result[branch_id] = text return result def _round_summary( *, round_index: int, goal: dict[str, Any], plan: dict[str, Any], branches: list[dict[str, Any]], batches: list[dict[str, Any]], domain_info: list[dict[str, Any]], run_status: str, ) -> dict[str, Any]: paths = plan.get("paths") or [] branch_ids = sorted( { branch_id for batch in batches for branch_id in batch.get("branchIds") or [] } ) decision_context = _missing_multipath_context(run_status) decision_summary = ( "；".join(_preview(_business_terms(batch.get("decision")), 64) for batch in batches[:2]) if batches else decision_context["value"] ) if len(batches) > 2: decision_summary += f"；另有 {len(batches) - 2} 个决策批次" merged_content_count = sum( 1 for branch in branches if branch.get("pathType") != "领域信息" and branch.get("status") == "merged" ) script_summary = ( f"{merged_content_count} 个内容方案进入主脚本" if merged_content_count else "没有可确认的内容方案合入记录" ) domain_summary = ( f"新增 {len(domain_info)} 条领域事实" if domain_info else "没有新增领域事实记录" ) return { "title": f"第 {round_index} 轮", "goal": goal, "approach": { "mode": plan.get("mode"), "candidateCount": len(branches) or len(paths), "plannedPathCount": len(paths), "contentBranchCount": sum( 1 for branch in branches if branch.get("pathType") == "内容" ), "domainBranchCount": sum( 1 for branch in branches if branch.get("pathType") == "领域信息" ), "summary": _plan_summary(plan), }, "decision": { "batchCount": len(batches), "branchIds": branch_ids, "summary": decision_summary, }, "output": { "mergedContentCount": merged_content_count, "domainFactCount": len(domain_info), "scriptSummary": script_summary, "domainSummary": domain_summary, "summary": f"{script_summary}；{domain_summary}", }, } def _goal_summary(value: Any) -> dict[str, Any]: parsed = parse_goal_text(value) if not parsed.get("raw"): return { "headline": "本轮目标未记录", "itemCount": 0, "previewItems": [], "truncated": False, } items = list(parsed.get("focusItems") or []) headline_source = str(parsed.get("headline") or parsed.get("raw")) headline = _preview(headline_source, 96) preview_items = [_preview(item, 72) for item in items[:3]] return { "headline": headline, "itemCount": len(items), "previewItems": preview_items, "truncated": bool( items or len(headline_source) > len(headline) or len(items) > len(preview_items) ), } def _business_terms(value: Any) -> str: text = _plain(value) or "未记录" text = re.sub(r"(?i)\s*\bbranch\s*#?(\d+)\b\s*", r"方案 \1 ", text) text = re.sub(r"(?i)\s*\bbase\b", "主脚本", text) text = re.sub(r"\s+([，。；！？、：）】,.])", r"\1", text) return text def _missing_multipath_context(run_status: str) -> dict[str, Any]: status = str(run_status or "unknown").lower() if status == "running": return { "kind": "execution", "label": "当前进度", "value": "等待主 Agent 比较候选方案", "sourceNotice": None, "status": "running", } interrupted = { "stopped": "构建停止前尚未形成多路决策", "cancelled": "构建取消前尚未形成多路决策", "cancel": "构建取消前尚未形成多路决策", "cancle": "构建取消前尚未形成多路决策", "failed": "构建失败前尚未形成多路决策", } if status in interrupted: return { "kind": "missing", "label": "流程进度", "value": interrupted[status], "sourceNotice": "process-incomplete", } return { "kind": "missing", "label": "记录状态", "value": "构建已经结束，但未找到主 Agent 多路决策记录", "sourceNotice": "record-missing", } def _missing_evaluation_context(run_status: str) -> dict[str, Any]: status = str(run_status or "unknown").lower() if status == "running": return { "kind": "execution", "label": "当前进度", "value": "等待整体评审", "sourceNotice": None, "status": "running", } interrupted = { "stopped": "构建停止前尚未进行整体评审", "cancelled": "构建取消前尚未进行整体评审", "cancel": "构建取消前尚未进行整体评审", "cancle": "构建取消前尚未进行整体评审", "failed": "构建失败前尚未形成整体评审结论", } if status in interrupted: return { "kind": "missing", "label": "流程进度", "value": interrupted[status], "sourceNotice": "process-incomplete", } return { "kind": "missing", "label": "记录状态", "value": "构建已经结束，但未找到明确的整体评审结论", "sourceNotice": "record-missing", } def _branch_disposition_context(branch_status: str, run_status: str) -> dict[str, Any]: if branch_status != "open": return { "kind": "decision", "label": _status_label(branch_status), "sourceNotice": None, } status = str(run_status or "unknown").lower() if status == "running": return { "kind": "execution", "label": "等待主 Agent 处置", "sourceNotice": None, } interrupted = { "stopped": "构建停止时尚未处置", "cancelled": "构建取消时尚未处置", "cancel": "构建取消时尚未处置", "cancle": "构建取消时尚未处置", "failed": "构建失败时尚未处置", } if status in interrupted: return { "kind": "missing", "label": interrupted[status], "sourceNotice": "process-incomplete", } return { "kind": "missing", "label": "构建结束时仍未处置", "sourceNotice": "record-missing", } def _plan_summary(plan: dict[str, Any]) -> str: paths = plan.get("paths") or [] if not paths: return "多路规划未记录" return f"{plan.get('mode') or '方式未记录'}：{len(paths)} 个候选方案" def _status_label(status: str) -> str: return { "merged": "已采用", "parked": "暂存", "discarded": "未采用", "open": "尚未处置", }.get(status, status or "状态未记录") def _data_shape( branches: list[dict[str, Any]], multipath_decisions: list[dict[str, Any]], domain_info: list[dict[str, Any]], events: list[dict[str, Any]], ) -> str: if ( multipath_decisions or domain_info or events or any(branch.get("path_type") for branch in branches) ): return "current" return "legacy-incomplete" def _warnings( *, data_shape: str, ignored_legacy_count: int, rounds: list[dict[str, Any]], runtime_unassigned: list[dict[str, Any]], business_unassigned: list[dict[str, Any]], ) -> list[dict[str, str]]: warnings: list[dict[str, str]] = [] if data_shape == "legacy-incomplete": warnings.append( { "id": "legacy-shape", "level": "warning", "message": "这是旧结构记录，只展示新数据契约能够直接确认的内容。", } ) if ignored_legacy_count: warnings.append( { "id": "ignored-legacy-data-decisions", "level": "info", "message": f"已忽略 {ignored_legacy_count} 条旧语义主决策记录。", } ) missing_rounds = [ round_["roundIndex"] for round_ in rounds if round_["branches"] and not any( batch.get("decision") for batch in round_.get("convergenceBatches") or [] ) and any( (batch.get("missingDecision") or {}).get("sourceNotice") == "record-missing" for batch in round_.get("convergenceBatches") or [] ) ] if missing_rounds: warnings.append( { "id": "missing-multipath-decisions", "level": "warning", "message": "第 " + "、".join(map(str, missing_rounds)) + " 轮未记录主 Agent 多路决策。", } ) if runtime_unassigned: warnings.append( { "id": "unassigned-runtime-events", "level": "info", "message": f"有 {len(runtime_unassigned)} 条运行记录无法安全归属，已留在技术详情中。", } ) if business_unassigned: warnings.append( { "id": "unassigned-business-records", "level": "warning", "message": f"有 {len(business_unassigned)} 条多路决策缺少可确认的轮次归属。", } ) return warnings def _plain(value: Any) -> str | None: if value is None: return None text = str(value).replace("\r\n", "\n").strip() if not text: return None text = re.sub(r"```[a-zA-Z0-9_-]*", "", text).replace("```", "") text = re.sub(r"^\s{0,3}#{1,6}\s*", "", text, flags=re.M) text = re.sub(r"\*\*([^*]+)\*\*", r"\1", text) text = re.sub(r"(?m)^\s*[-*]\s+", "• ", text) return text.strip() def _preview(value: Any, limit: int = 140) -> str: text = _plain(value) or "未记录" text = re.sub(r"\s+", " ", text) if len(text) <= limit: return text window = text[:limit] cut = max(window.rfind(mark) for mark in "。！？；") if cut >= max(30, limit // 2): return window[: cut + 1] return window.rstrip("，、；：,. ") + "…" def _integer(value: Any) -> int | None: try: return int(value) if value is not None else None except (TypeError, ValueError): return None