from __future__ import annotations import json from datetime import datetime, timezone from typing import Any, Literal from .models import ( DataRef, InputSummary, JourneyEdge, JourneyStep, JourneyView, ProcessView, StepEvidence, ) from .narrator import narrate_contract, task_label def build_journey(script_build_id: int, source: dict[str, Any]) -> JourneyView: snapshot = source["snapshot"] tasks = {str(item["task_id"]): item for item in snapshot.get("tasks") or []} attempts = {str(item["attempt_id"]): item for item in snapshot.get("attempts") or []} validations = {str(item["validation_id"]): item for item in snapshot.get("validations") or []} decisions = {str(item["decision_id"]): item for item in snapshot.get("decisions") or []} contracts = source.get("contracts") or {} messages = source.get("messages") or {} root_trace_id = str(snapshot.get("root_trace_id") or "") planner_calls = _planner_calls(messages.get(root_trace_id) or []) operations = snapshot.get("operations") or [] parallel = _parallel_attempts(operations) event_sequences = _event_sequences(source.get("events") or []) steps: list[JourneyStep] = [] plan_by_task: dict[str, str] = {} execute_by_attempt: dict[str, str] = {} validation_by_id: dict[str, str] = {} for task_id, task in tasks.items(): if task.get("parent_task_id") is None or task_id not in contracts: continue contract = contracts[task_id] narration = narrate_contract(contract) call = planner_calls.get(task_id) observable = _plain((call or {}).get("assistant_text")) step_id = f"plan:{task_id}" plan_by_task[task_id] = step_id steps.append( JourneyStep( step_id=step_id, step_type="plan", title=f"Planner 规划·{narration.output}", status=str(task.get("status") or "planned"), task_id=task_id, reasoning=observable or narration.reasoning, reasoning_source="observable" if observable else "contract", process=ProcessView( label=narration.action, actor="Planner", tool_name="plan_script_tasks", trace_id=root_trace_id, details={"task_kind": contract.get("task_kind")}, ), input_data=_contract_inputs(contract), output_data=[ DataRef(label="TaskContract", kind="contract", ref=contract.get("contract_ref")) ], contract=narration, created_at=str(task.get("created_at") or ""), evidence=StepEvidence( contract=contract, tool_calls=[call["tool_call"]] if call and call.get("tool_call") else [], ), ) ) for attempt_id, attempt in attempts.items(): task_id = str(attempt.get("task_id") or "") task = tasks.get(task_id) or {} contract = contracts.get(task_id) worker_trace_id = str(attempt.get("worker_trace_id") or "") worker_messages = messages.get(worker_trace_id) or [] observable = _assistant_text(worker_messages) narration = narrate_contract(contract) if contract else None tools = _tool_calls(worker_messages) position, total, group_id = parallel.get(attempt_id, (None, None, None)) stats = attempt.get("execution_stats") or {} step_id = f"execute:{attempt_id}" execute_by_attempt[attempt_id] = step_id steps.append( JourneyStep( step_id=step_id, step_type="execute", title=f"Worker 执行·{_task_name(task, contract)}", status=str(attempt.get("status") or "unknown"), task_id=task_id, attempt_id=attempt_id, reasoning=observable or ( f"执行任务合同：{contract.get('objective')}" if contract else "执行已冻结的任务合同" ), reasoning_source="observable" if observable else "contract", process=ProcessView( label=_worker_process_label(tools), actor="Worker", trace_id=worker_trace_id or None, details={"execution_mode": attempt.get("execution_mode")}, ), input_data=_contract_inputs(contract or {}), output_data=_attempt_outputs(attempt), contract=narration, parallel_group_id=group_id, parallel_position=position, parallel_total=total, model=stats.get("primary_model"), tokens=stats.get("total_tokens"), cost=stats.get("total_cost"), duration_ms=attempt.get("duration_ms"), created_at=str(attempt.get("started_at") or attempt.get("created_at") or ""), evidence=StepEvidence( contract=contract, tool_calls=tools, artifact_refs=_artifact_refs(attempt), ), ) ) for validation_id, validation in validations.items(): attempt_id = str(validation.get("attempt_id") or "") attempt = attempts.get(attempt_id) or {} task_id = str(validation.get("task_id") or attempt.get("task_id") or "") plan = validation.get("validation_plan") or {} stats = validation.get("execution_stats") or {} reasoning = _validation_reason(validation) task_name = _task_name(tasks.get(task_id) or {}, contracts.get(task_id)) step_id = f"validate:{validation_id}" validation_by_id[validation_id] = step_id steps.append( JourneyStep( step_id=step_id, step_type="validate", title=f"Validator 验收·{task_name}", status=str(validation.get("verdict") or validation.get("status") or "unknown"), task_id=task_id, attempt_id=attempt_id or None, validation_id=validation_id, reasoning=reasoning, reasoning_source="validation", process=ProcessView( label=( "确定性规则预检" if plan.get("mode") == "deterministic" else "独立 Validator 语义质检" ), actor="Validator", trace_id=validation.get("validator_trace_id"), details={ "mode": plan.get("mode"), "preflight_rule_ids": plan.get("preflight_rule_ids") or [], }, ), input_data=_attempt_outputs(attempt), output_data=[ DataRef( label=f"Validation·{validation.get('verdict') or validation.get('status')}", kind="validation", ref=validation_id, detail=_plain(validation.get("summary")), ) ], model=stats.get("primary_model"), tokens=stats.get("total_tokens"), cost=stats.get("total_cost"), duration_ms=validation.get("duration_ms"), created_at=str(validation.get("started_at") or validation.get("created_at") or ""), evidence=StepEvidence(validation=validation, artifact_refs=_artifact_refs(attempt)), ) ) for decision_id, decision in decisions.items(): task_id = str(decision.get("task_id") or "") action = str(decision.get("action") or "unknown").upper() validation_id = str(decision.get("validation_id") or "") attempt_id = str(decision.get("attempt_id") or "") steps.append( JourneyStep( step_id=f"decide:{decision_id}", step_type="decide", title=f"Planner 决定·{_action_label(action)}", status=action.lower(), task_id=task_id, attempt_id=attempt_id or None, validation_id=validation_id or None, decision_id=decision_id, reasoning=_plain(decision.get("reason")) or "Planner 根据验收结果选择下一步", reasoning_source="decision", process=ProcessView( label=_action_process(action), actor="Planner", tool_name="decide_script_task", trace_id=root_trace_id, details={ "from_status": decision.get("from_status"), "to_status": decision.get("to_status"), }, ), input_data=[ DataRef(label="Validation", kind="validation", ref=validation_id or None) ], output_data=[ DataRef(label=_action_label(action), kind="decision", ref=decision_id) ], created_at=str(decision.get("created_at") or ""), evidence=StepEvidence(decision=decision), ) ) steps.sort(key=lambda step: _step_sort_key(step, event_sequences)) steps = [step.model_copy(update={"sequence": index}) for index, step in enumerate(steps, 1)] edges = _edges(steps, tasks, plan_by_task, execute_by_attempt, validation_by_id) root_task = next((item for item in tasks.values() if item.get("parent_task_id") is None), {}) raw_input = source["input_summary"] return JourneyView( schema_version="script-build-journey/v1", generated_at=datetime.now(timezone.utc).isoformat(), script_build_id=script_build_id, root_trace_id=root_trace_id, status=str(root_task.get("status") or "unknown"), objective=_plain(snapshot.get("root_objective")) or "完成脚本创作与交付", input_summary=InputSummary( topic=_plain(raw_input.get("topic")) or "未记录选题", account_name=(raw_input.get("persona") or {}).get("resolved_account_name") or (raw_input.get("persona") or {}).get("account_name"), persona_point_count=int((raw_input.get("persona") or {}).get("point_count") or 0), strategy_names=[ str(item["name"]) for item in raw_input.get("strategies") or [] if isinstance(item, dict) and item.get("name") ], snapshot_id=str(raw_input.get("input_snapshot_id") or ""), digest=str(raw_input.get("input_digest") or ""), ), steps=steps, edges=edges, warnings=_warnings(steps), ) def _planner_calls(messages: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: results = { str(message.get("tool_call_id")): _parse_result(message.get("content")) for message in messages if message.get("role") == "tool" and message.get("tool_call_id") } output: dict[str, dict[str, Any]] = {} last_text = "" for message in sorted(messages, key=lambda value: int(value.get("sequence") or 0)): if message.get("role") != "assistant": continue content = message.get("content") if not isinstance(content, dict): continue if _plain(content.get("text")): last_text = _plain(content.get("text")) for call in content.get("tool_calls") or []: function = call.get("function") or {} if function.get("name") != "plan_script_tasks": continue result = results.get(str(call.get("id"))) or {} for task_id in result.get("task_ids") or []: output[str(task_id)] = { "assistant_text": last_text, "tool_call": { "name": "plan_script_tasks", "arguments": _parse_json(function.get("arguments")), }, } return output def _tool_calls(messages: list[dict[str, Any]]) -> list[dict[str, Any]]: output: list[dict[str, Any]] = [] for message in messages: content = message.get("content") if message.get("role") != "assistant" or not isinstance(content, dict): continue for call in content.get("tool_calls") or []: function = call.get("function") or {} output.append( { "name": str(function.get("name") or "unknown"), "arguments": _parse_json(function.get("arguments")), } ) return output def _assistant_text(messages: list[dict[str, Any]]) -> str: values = [] for message in messages: content = message.get("content") if message.get("role") == "assistant" and isinstance(content, dict): text = _plain(content.get("text")) if text: values.append(text) return " ".join(values[:3])[:1_200] def _contract_inputs(contract: dict[str, Any]) -> list[DataRef]: values = [] for item in contract.get("input_decision_refs") or []: if not isinstance(item, dict): continue artifact = item.get("artifact_ref") or {} label = f"已接受的{task_label(str(item.get('expected_task_kind') or '上游成果'))}" values.append( DataRef( label=label, kind=str(artifact.get("kind") or "artifact"), ref=artifact.get("uri"), detail=str(item.get("decision_id") or ""), ) ) if not values: values.append(DataRef(label="冻结创作输入", kind="input-snapshot")) return values def _attempt_outputs(attempt: dict[str, Any]) -> list[DataRef]: submission = attempt.get("submission") or {} values = [] for item in submission.get("artifact_refs") or []: if isinstance(item, dict): values.append( DataRef( label=str(item.get("summary") or item.get("kind") or "Artifact"), kind=str(item.get("kind") or "artifact"), ref=item.get("uri"), ) ) return values or [DataRef(label="待提交的任务成果", kind="artifact")] def _artifact_refs(attempt: dict[str, Any]) -> list[dict[str, Any]]: submission = attempt.get("submission") or {} return [item for item in submission.get("artifact_refs") or [] if isinstance(item, dict)] def _validation_reason(validation: dict[str, Any]) -> str: summary = _plain(validation.get("summary")) failures = [ f"{item.get('criterion_id')}：{_plain(item.get('reason'))}" for item in validation.get("criterion_results") or [] if isinstance(item, dict) and str(item.get("verdict")) not in {"passed", "pass"} ] return "；".join(failures) or summary or "Validator 已逐项检查验收标准" def _parallel_attempts(operations: list[dict[str, Any]]) -> dict[str, tuple[int, int, str]]: output: dict[str, tuple[int, int, str]] = {} for operation in operations: attempt_ids = [str(value) for value in operation.get("attempt_ids") or []] if len(attempt_ids) < 2: continue group_id = str(operation.get("operation_id") or "") for index, attempt_id in enumerate(attempt_ids, start=1): output[attempt_id] = (index, len(attempt_ids), group_id) return output def _loop_target( action: str, task_id: str, attempt_id: str, tasks: dict[str, dict[str, Any]], plan_by_task: dict[str, str], execute_by_attempt: dict[str, str], ) -> str | None: if action == "RETRY": return execute_by_attempt.get(attempt_id) if action == "REVISE": return plan_by_task.get(task_id) if action == "SPLIT": children = tasks.get(task_id, {}).get("child_task_ids") or [] return next( (plan_by_task.get(str(value)) for value in children if plan_by_task.get(str(value))), None, ) return None def _edges( steps: list[JourneyStep], tasks: dict[str, dict[str, Any]], plan_by_task: dict[str, str], execute_by_attempt: dict[str, str], validation_by_id: dict[str, str], ) -> list[JourneyEdge]: edges: list[JourneyEdge] = [] for left, right in zip(steps, steps[1:]): edges.append( JourneyEdge( edge_id=f"sequence:{left.step_id}:{right.step_id}", source=left.step_id, target=right.step_id, relation="sequence", label="然后", ) ) for step in steps: if step.step_type == "execute" and step.attempt_id: source = plan_by_task.get(step.task_id) if source: edges.append(_edge(source, step.step_id, "creates", "派发执行")) elif step.step_type == "validate" and step.attempt_id: source = execute_by_attempt.get(step.attempt_id) if source: edges.append(_edge(source, step.step_id, "validates", "提交验收")) elif step.step_type == "decide" and step.validation_id: source = validation_by_id.get(step.validation_id) if source: edges.append(_edge(source, step.step_id, "decides", "验收结论")) target = _loop_target( step.status.upper(), step.task_id, step.attempt_id or "", tasks, plan_by_task, execute_by_attempt, ) if target: edges.append(_edge(step.step_id, target, "loop", _loop_label(step.status.upper()))) return edges def _edge( source: str, target: str, relation: Literal["creates", "validates", "decides", "loop"], label: str, ) -> JourneyEdge: return JourneyEdge( edge_id=f"{relation}:{source}:{target}", source=source, target=target, relation=relation, label=label, ) def _warnings(steps: list[JourneyStep]) -> list[str]: warnings = [] if not steps: warnings.append("当前运行还没有可展示的创作步骤。") if any(step.reasoning_source == "contract" for step in steps): warnings.append("部分“思考”由冻结任务合同转述，不是隐藏思维链。") return warnings def _worker_process_label(tools: list[dict[str, Any]]) -> str: names = [str(item.get("name")) for item in tools if item.get("name")] if not names: return "根据任务合同生成候选成果" shown = "、".join(names[:3]) suffix = f"等 {len(names)} 次工具调用" if len(names) > 3 else "" return f"执行 {shown}{suffix}" def _task_name(task: dict[str, Any], contract: dict[str, Any] | None) -> str: if contract: return task_label(str(contract.get("task_kind") or "task")) return _plain((task.get("current_spec") or {}).get("objective"))[:60] or "任务" def _action_label(action: str) -> str: return { "ACCEPT": "接受成果", "RETRY": "重新执行", "REVISE": "修订任务", "SPLIT": "拆分子任务", "BLOCK": "暂停在边界", "CANCEL": "取消任务", }.get(action, action) def _action_process(action: str) -> str: return { "ACCEPT": "接受当前成果，继续规划下一步", "RETRY": "保留任务合同，让 Worker 再执行一次", "REVISE": "修订任务合同后重新执行", "SPLIT": "把当前问题拆成多个子任务", "BLOCK": "保存当前成果，等待下一阶段", }.get(action, "更新任务状态") def _loop_label(action: str) -> str: return {"RETRY": "重试", "REVISE": "修订后返工", "SPLIT": "拆分后继续"}.get(action, "返工") def _event_sequences(events: list[dict[str, Any]]) -> dict[str, int]: result: dict[str, int] = {} for event in events: sequence = int(event.get("sequence") or 0) payload = event.get("payload") or {} ids = list(payload.get("task_ids") or []) ids.extend( payload.get(key) for key in ("task_id", "attempt_id", "validation_id", "decision_id") if payload.get(key) ) for entity_id in ids: result.setdefault(str(entity_id), sequence) return result def _step_sort_key(step: JourneyStep, event_sequences: dict[str, int]) -> tuple[int, str, int, str]: priority = {"plan": 0, "execute": 1, "validate": 2, "decide": 3} entity_id = step.decision_id or step.validation_id or step.attempt_id or step.task_id event_sequence = event_sequences.get(entity_id, 2**31) return event_sequence, step.created_at, priority[step.step_type], step.step_id def _parse_result(content: Any) -> dict[str, Any]: if isinstance(content, dict): return _parse_json(content.get("result", content)) return _parse_json(content) def _parse_json(value: Any) -> Any: if not isinstance(value, str): return value try: return json.loads(value) except json.JSONDecodeError: return value def _plain(value: Any) -> str: return " ".join(str(value or "").split())