script-build-and-framework-tmp/visualization/backend/app/journey.py

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