from __future__ import annotations import json import re from datetime import datetime, timezone from typing import Any, Iterable from .evaluation_report_parser import EvaluationReportParser from .main_decision_text import parse_goal_text, parse_objective_text from .runtime_event_index import ( EventBounds, RuntimeEventIndex, detail_ref, event_input, event_output, event_sequence, event_summary, event_text_output, integer, ) from .runtime_tool_catalog import business_tool_label class MainAgentDecisionProjector: """Build honest decision contexts for the first three main-Agent stages. Database rows are passed in by the caller and remain the current business truth. Runtime events only describe how that persisted result was reached. """ def project( self, index: RuntimeEventIndex, *, script_direction: str | None, rounds: Iterable[dict[str, Any]], multipath_decisions: Iterable[dict[str, Any]] = (), ) -> dict[str, Any]: round_rows = sorted( (dict(row) for row in rounds), key=lambda row: integer(row.get("round_index")) or 0, ) decisions = [dict(item) for item in multipath_decisions] planning_analysis = PlanningAnalysisProjector().project(index) objective = ObjectiveDecisionProjector().project( index, script_direction=script_direction ) goals: dict[int, dict[str, Any]] = {} plans: dict[int, dict[str, Any]] = {} for position, row in enumerate(round_rows): round_index = integer(row.get("round_index")) if round_index is None: continue goals[round_index] = RoundGoalDecisionProjector().project( index, round_row=row, objective_context=objective, previous_round=(round_rows[position - 1] if position else None), multipath_decisions=decisions, ) plans[round_index] = ImplementationPlanDecisionProjector().project( index, round_row=row, goal_context=goals[round_index] ) projection = { "planningAnalysis": planning_analysis, "objective": objective, "roundGoalsByRound": goals, "implementationPlansByRound": plans, } assigned_technical_refs = { ref for context in [planning_analysis, objective, *goals.values(), *plans.values()] for ref in context.get("technicalRefs") or [] } projection["unassigned"] = [ event_summary(event, "failed-main-decision-call-unassigned") for name in ("save_script_direction", "begin_round", "record_multipath_plan") for event in index.anchors(name, successful=False) if detail_ref(event) not in assigned_technical_refs ] return projection class PlanningAnalysisProjector: """Project the last real main-Agent planning note before direction is saved.""" def project(self, index: RuntimeEventIndex) -> dict[str, Any]: direction_saves = index.anchors("save_script_direction", successful=True) first_save_seq = event_sequence(direction_saves[0]) if direction_saves else None candidates = index.anchors( "think_and_plan", successful=True, bounds=EventBounds(before_seq=first_save_seq), ) event = candidates[-1] if candidates else None raw_input = event_input(event) if event else None data = raw_input if isinstance(raw_input, dict) else {} event_ref = detail_ref(event) if event else None return { "stage": "planning-analysis", "summary": _detail_text(data.get("thought_summary")), "thought": _detail_text(data.get("thought")), "plan": _detail_text(data.get("plan")), "action": _detail_text(data.get("action")), "eventRef": event_ref, "technicalRefs": [event_ref] if event_ref else [], "completeness": "complete" if event_ref and data else "missing", } class ObjectiveDecisionProjector: def project( self, index: RuntimeEventIndex, *, script_direction: str | None ) -> dict[str, Any]: successful = index.anchors("save_script_direction", successful=True) failed = index.anchors("save_script_direction", successful=False) current_anchor = successful[-1] if successful else None previous_seq = event_sequence(successful[-2]) if len(successful) > 1 else None current_seq = event_sequence(current_anchor) if current_anchor else None scope_id = integer(current_anchor.get("scope_event_id")) if current_anchor else None bounds = EventBounds(after_seq=previous_seq, before_seq=current_seq) reads = index.main_direct_reads( scope_id=scope_id, bounds=bounds, ) failed_reads = index.main_direct_reads( scope_id=scope_id, bounds=bounds, successful=False ) parsed = parse_script_direction(script_direction) revisions = [_direction_revision(event, script_direction) for event in successful] return { "stage": "objective", "inputs": [_decision_read(event) for event in reads], "decisionItems": parsed["goals"], "explicitReasoning": parsed["valueReasoning"], "constraints": parsed["constraints"], "revisionRefs": [ref for event in successful if (ref := detail_ref(event))], "completeness": _completeness(bool(script_direction), bool(current_anchor)), "revisions": revisions, "currentRevisionRef": _current_revision_ref(revisions), "technicalRefs": [ ref for event in [*failed_reads, *failed] if (ref := detail_ref(event)) ], "currentSavedValue": script_direction, "sourceDocument": parsed["raw"], "valueJudgements": parsed["valueJudgements"], } class RoundGoalDecisionProjector: def project( self, index: RuntimeEventIndex, *, round_row: dict[str, Any], objective_context: dict[str, Any], previous_round: dict[str, Any] | None, multipath_decisions: Iterable[dict[str, Any]], ) -> dict[str, Any]: round_index = integer(round_row.get("round_index")) goal = _text(round_row.get("goal")) begin_events = index.successful_begin_rounds().get(round_index or -1, []) current_anchor = begin_events[-1] if begin_events else None current_seq = event_sequence(current_anchor) if current_anchor else None inputs: list[dict[str, Any]] = [] constraints: list[str] = [] late_decision_refs: list[str] = [] if round_index == 1: if objective_context.get("decisionItems"): inputs.append( { "label": "当前保存的创作目标", "summary": _detail_text( "\n\n".join( str(item) for item in objective_context["decisionItems"] ) ), "actor": "main", "relation": "standing-constraint", "detailRef": objective_context.get("currentRevisionRef"), } ) lower_seq = _last_objective_save_seq(index) else: previous_index = integer((previous_round or {}).get("round_index")) lower_seq = None reports = index.agent_returns( "script_evaluator", round_index=previous_index, before_seq=current_seq, ) if reports: evaluator, relation = reports[-1] parsed_report = EvaluationReportParser().parse( event_text_output(evaluator) or "", kind="overall" ) inputs.append( { "label": "上一轮整体评审", "summary": _parsed_report_summary(parsed_report), "actor": "overall-evaluator", "relation": relation, "detailRef": detail_ref(evaluator), } ) constraints.extend(_parsed_report_guidance(parsed_report)) lower_seq = event_sequence(evaluator) previous_decisions = sorted( [ item for item in multipath_decisions if integer(item.get("round_index")) == previous_index ], key=lambda item: (str(item.get("created_at") or ""), integer(item.get("id")) or 0), ) for decision in previous_decisions: decision_ref = ( f"multipath-decision:{decision.get('id')}" if decision.get("id") is not None else None ) if not _record_precedes_event(decision, current_anchor): if decision_ref: late_decision_refs.append(decision_ref) continue inputs.append( { "label": "上一轮主 Agent 多路决策", "summary": _detail_text(decision.get("decision")), "actor": "main", "relation": "persisted-record", "detailRef": decision_ref, } ) if current_anchor is not None: scope_id = integer(current_anchor.get("scope_event_id")) read_bounds = EventBounds(after_seq=lower_seq, before_seq=current_seq) reads = index.main_direct_reads( scope_id=scope_id, bounds=read_bounds, ) inputs.extend(_decision_read(event) for event in reads) failed_reads = index.main_direct_reads( scope_id=scope_id, bounds=read_bounds, successful=False ) else: failed_reads = [] failed = [ event for event in index.anchors("begin_round", successful=False) if _explicit_output_round_hint(event) == round_index ] return { "stage": "round-goal", "inputs": _dedupe_inputs(inputs), "decisionItems": [goal] if goal else [], "sourceDocument": goal, "explicitReasoning": _labelled_reasoning(goal), "constraints": _unique(constraints), "revisionRefs": [ref for event in begin_events if (ref := detail_ref(event))], "completeness": _completeness(bool(goal), bool(current_anchor)), "technicalRefs": [ ref for event in [*failed_reads, *failed] if (ref := detail_ref(event)) ] + late_decision_refs, "currentSavedValue": goal, "roundIndex": round_index, } class ImplementationPlanDecisionProjector: def project( self, index: RuntimeEventIndex, *, round_row: dict[str, Any], goal_context: dict[str, Any], ) -> dict[str, Any]: round_index = integer(round_row.get("round_index")) paths = round_row.get("multipath_plan") paths = paths if isinstance(paths, list) else [] mode = _text(round_row.get("race_or_divide")) note = _text(round_row.get("plan_note")) successful: list[dict[str, Any]] = [] failed: list[dict[str, Any]] = [] for event in index.anchors("record_multipath_plan", successful=None): resolved_round = _round_hint(event) if resolved_round != round_index: continue (successful if _anchor_success(event) else failed).append(event) current_anchor = successful[-1] if successful else None begin_events = index.successful_begin_rounds().get(round_index or -1, []) lower_seq = event_sequence(begin_events[-1]) if begin_events else None current_seq = event_sequence(current_anchor) if current_anchor else None scope_id = integer(current_anchor.get("scope_event_id")) if current_anchor else None inputs: list[dict[str, Any]] = [] goal_items = goal_context.get("decisionItems") or [] if goal_items: inputs.append( { "label": "本轮目标", "summary": _detail_text("\n".join(str(item) for item in goal_items)), "actor": "main", "relation": "standing-constraint", "detailRef": (goal_context.get("revisionRefs") or [None])[-1], } ) if current_anchor is not None: read_bounds = EventBounds(after_seq=lower_seq, before_seq=current_seq) inputs.extend( _decision_read(event) for event in index.main_direct_reads( scope_id=scope_id, bounds=read_bounds, ) ) failed_reads = index.main_direct_reads( scope_id=scope_id, bounds=read_bounds, successful=False ) else: failed_reads = [] revisions = [_plan_revision(event, paths, mode, note) for event in successful] display_mode = "单路" if len(paths) == 1 else mode return { "stage": "implementation-plan", "inputs": _dedupe_inputs(inputs), "decisionItems": [_path_summary(path, index + 1) for index, path in enumerate(paths)], "explicitReasoning": note, "constraints": [], "revisionRefs": [ref for event in successful if (ref := detail_ref(event))], "completeness": _completeness(bool(paths), bool(current_anchor)), "revisions": revisions, "currentRevisionRef": _current_revision_ref(revisions), "technicalRefs": [ ref for event in [*failed_reads, *failed] if (ref := detail_ref(event)) ], "roundIndex": round_index, "displayMode": display_mode, "rawMode": mode, "pathCount": len(paths), "routeItems": [_path_item(path, index + 1) for index, path in enumerate(paths)], } def parse_script_direction(value: str | None) -> dict[str, Any]: parsed = parse_objective_text(value) goals = list(parsed.get("targets") or []) if not goals and parsed.get("raw"): goals = [str(parsed.get("headline"))] constraints = _unique( [ *(str(item) for item in parsed.get("domainDimensions") or []), *(str(item) for item in parsed.get("constraints") or []), ] ) value_judgements = [ {"label": label, "value": content} for label, content in ( ("选题价值主张", parsed.get("topicValue")), ("账号价值主张", parsed.get("accountValue")), ) if content ] reasoning = "\n".join( f"{item['label']}:{item['value']}" for item in value_judgements ) or None return { "goals": _unique(goals), "constraints": _unique(constraints), "valueJudgements": value_judgements, "valueReasoning": reasoning, "raw": str(parsed.get("raw") or ""), } def parse_goal_items(value: str | None) -> list[str]: parsed = parse_goal_text(value) if not parsed.get("raw"): return [] return [str(parsed["raw"])] def _decision_read(event: dict[str, Any]) -> dict[str, Any]: return { "label": business_tool_label(str(event.get("event_name") or "")), "summary": _read_result_summary(event), "actor": "main", "relation": "direct-read", "detailRef": detail_ref(event), } def _read_result_summary(event: dict[str, Any]) -> str: name = str(event.get("event_name") or "") output = event_output(event) if isinstance(output, dict): if name == "get_topic_detail": topic = output.get("topic") if isinstance(output.get("topic"), dict) else {} points = output.get("points") if isinstance(output.get("points"), list) else [] point_name = next( ( _text(point.get("point_result")) for point in points if isinstance(point, dict) and _text(point.get("point_result")) ), None, ) topic_result = _text(topic.get("result")) summary = ":".join(value for value in (point_name, topic_result) if value) if summary: return _detail_text(summary) or "已读取选题" if name == "get_account_script_section_patterns": summary = _detail_text(output.get("分段规律摘要")) if summary: return summary if name == "get_account_script_persona_points": persona = _persona_summary(output) if persona: return persona for key in ("count", "returned_count", "result_count", "total"): count = integer(output.get(key)) if count is not None: return f"返回 {count} 条记录" for key in ("summary", "message"): value = _detail_text(output.get(key)) if value: return value raw_preview = str(event.get("output_preview") or "") if name == "get_topic_detail": topic_result = _json_string_field(raw_preview, "result") point_name = _json_string_field(raw_preview, "point_result") summary = ":".join(value for value in (point_name, topic_result) if value) if summary: return _detail_text(summary) or "已读取选题" if name == "get_account_script_section_patterns": summary = _json_string_field(raw_preview, "分段规律摘要") if summary: return _detail_text(summary) or "已读取账号段落模式" if name == "get_account_script_persona_points": count_match = re.search(r'"returned_count"\s*:\s*(\d+)', raw_preview) columns = _json_string_fields(raw_preview, "列") if count_match: suffix = f",主要涉及{'、'.join(columns[:3])}" if columns else "" return f"返回 {count_match.group(1)} 个账号表达特征{suffix}" return "已读取" def _persona_summary(output: dict[str, Any]) -> str | None: count = integer(output.get("returned_count")) records = output.get("records") if isinstance(output.get("records"), list) else [] columns = _unique( str(item.get("列")) for item in records if isinstance(item, dict) and item.get("列") ) if count is None: count = len(records) if records else None if count is None: return None suffix = f",主要涉及{'、'.join(columns[:3])}" if columns else "" return f"返回 {count} 个账号表达特征{suffix}" def _json_string_field(value: str, key: str) -> str | None: values = _json_string_fields(value, key) return values[0] if values else None def _json_string_fields(value: str, key: str) -> list[str]: results: list[str] = [] pattern = re.compile( rf'"{re.escape(key)}"\s*:\s*"((?:\\.|[^"\\])*)"' ) for match in pattern.finditer(value): try: decoded = json.loads(f'"{match.group(1)}"') except json.JSONDecodeError: decoded = match.group(1) text = _text(decoded) if text and text not in results: results.append(text) return results def _direction_revision( event: dict[str, Any], current_direction: str | None ) -> dict[str, Any]: payload = event_input(event) saved = payload.get("script_direction") if isinstance(payload, dict) else None return { "eventId": integer(event.get("id")), "eventSeq": event_sequence(event), "savedValue": saved, "current": _normalized_compare(saved) == _normalized_compare(current_direction), "detailRef": detail_ref(event), } def _plan_revision( event: dict[str, Any], current_paths: list[Any], current_mode: str | None, current_note: str | None ) -> dict[str, Any]: payload = event_input(event) payload = payload if isinstance(payload, dict) else {} paths = payload.get("paths") if isinstance(payload.get("paths"), list) else [] mode = _normalized_plan_mode(payload.get("mode")) normalized_current_mode = _normalized_plan_mode(current_mode) note = _text(payload.get("note")) current = ( _canonical_json(paths) == _canonical_json(current_paths) and (mode or normalized_current_mode) == normalized_current_mode and (note if note is not None else current_note) == current_note ) return { "eventId": integer(event.get("id")), "eventSeq": event_sequence(event), "paths": paths, "mode": mode, "note": note, "current": current, "detailRef": detail_ref(event), } def _normalized_plan_mode(value: Any) -> str | None: mode = _text(value) aliases = {"race": "竞争", "赛马": "竞争", "divide": "分工"} return aliases.get((mode or "").lower(), mode) def _path_item(path: Any, fallback_index: int) -> dict[str, Any]: data = path if isinstance(path, dict) else {} return { "pathIndex": integer(data.get("path_index")) or fallback_index, "pathType": _text(data.get("path_type")), "action": _text(data.get("action")), "target": _text(data.get("target")), "method": _text(data.get("method")), "emphasis": _text(data.get("emphasis")), } def _path_summary(path: Any, fallback_index: int) -> str: item = _path_item(path, fallback_index) parts = [part for part in (item["action"], item["target"]) if part] headline = ":".join(parts) if parts else f"方案 {item['pathIndex']}" return f"方案 {item['pathIndex']} · {headline}" def _parsed_report_summary(report: dict[str, Any]) -> str | None: if report.get("conclusion"): return _detail_text(report["conclusion"]) if report.get("recommendation"): return _detail_text(report["recommendation"]) problems = report.get("problems") or [] if problems and isinstance(problems[0], dict): return _detail_text(problems[0].get("summary")) return None def _parsed_report_guidance(report: dict[str, Any]) -> list[str]: values: list[str] = [] for item in report.get("problems") or []: if isinstance(item, dict) and (summary := _text(item.get("summary"))): values.append(f"问题:{summary}") if next_goal := _text(report.get("nextGoal")): values.append(f"下一步:{next_goal}") return _unique(values) def _record_precedes_event( record: dict[str, Any], event: dict[str, Any] | None ) -> bool: if event is None: return False record_time = _date_value(record.get("created_at")) event_time = _date_value(event.get("started_at") or event.get("ended_at")) return record_time is not None and event_time is not None and record_time <= event_time def _date_value(value: Any) -> datetime | None: if isinstance(value, datetime): parsed = value elif isinstance(value, str) and value.strip(): try: parsed = datetime.fromisoformat(value.strip().replace("Z", "+00:00")) except ValueError: return None else: return None if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=timezone.utc) return parsed.astimezone(timezone.utc) def _labelled_reasoning(text: str | None) -> str | None: value = _normalize_markdown(text) match = re.search(r"(?im)^\s*(?:\u7406\u7531|\u76ee\u6807\u6765\u6e90)\s*[\uff1a:]\s*(.+)$", value) return _clean_line(match.group(1)) if match else None def _last_objective_save_seq(index: RuntimeEventIndex) -> int | None: saves = index.anchors("save_script_direction", successful=True) return event_sequence(saves[-1]) if saves else None def _round_hint(event: dict[str, Any]) -> int | None: output = event_output(event) if isinstance(output, dict): value = integer(output.get("round_index")) if value is not None: return value return integer(event.get("round_index")) def _explicit_output_round_hint(event: dict[str, Any]) -> int | None: output = event_output(event) return integer(output.get("round_index")) if isinstance(output, dict) else None def _anchor_success(event: dict[str, Any]) -> bool: # Imported lazily to keep the public API above concise and avoid duplicating # success/error interpretation. from .runtime_event_index import event_succeeded return event_succeeded(event) def _current_revision_ref(revisions: list[dict[str, Any]]) -> str | None: for revision in reversed(revisions): if revision.get("current"): return revision.get("detailRef") return None def _normalize_markdown(value: Any) -> str: text = str(value or "").strip() if "\\n" in text: text = text.replace("\\n", "\n") return text.replace("\r\n", "\n").replace("\r", "\n") def _clean_line(value: Any) -> str: text = str(value or "") text = re.sub(r"^\s*(?:[-*+]\s+|\d+[.、]\s+|[\u2460-\u2473]\s*)", "", text) text = re.sub(r"[`*#]", "", text) return re.sub(r"\s+", " ", text).strip(" ::") def _plain_text(value: Any) -> str: lines = [_clean_line(line) for line in _normalize_markdown(value).splitlines()] return " ".join(line for line in lines if line) def _detail_text(value: Any) -> str: text = _normalize_markdown(value) text = re.sub(r"(?m)^\s{0,3}#{1,6}\s*", "", text) text = re.sub(r"\*\*([^*]+)\*\*", r"\1", text) text = re.sub(r"`([^`]*)`", r"\1", text) return text.strip() def _text(value: Any) -> str | None: text = str(value).strip() if value is not None else "" return text or None def _canonical_json(value: Any) -> str: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) def _normalized_compare(value: Any) -> str: return re.sub(r"\s+", "", str(value or "")) def _completeness(has_fact: bool, has_runtime_anchor: bool) -> str: if has_fact and has_runtime_anchor: return "complete" if has_fact or has_runtime_anchor: return "partial" return "missing" def _unique(values: Iterable[str]) -> list[str]: result: list[str] = [] for value in values: if value and value not in result: result.append(value) return result def _dedupe_inputs(values: Iterable[dict[str, Any]]) -> list[dict[str, Any]]: result: list[dict[str, Any]] = [] seen: set[tuple[Any, ...]] = set() for value in values: key = (value.get("label"), value.get("detailRef"), value.get("relation")) if key in seen: continue seen.add(key) result.append(value) return result