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

from __future__ import annotations

import json
import os
import re
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable

from .models import (
    AgentStep,
    AttemptDetail,
    DataUsage,
    ExecutionView,
    LifecycleEvent,
    PlannerTurn,
    RunSummary,
    TaskDetail,
    TaskNode,
)

VISUALIZATION_ROOT = Path(__file__).resolve().parents[2]
DEFAULT_AGENT_DATA_ROOT = VISUALIZATION_ROOT.parent / "script_build_host" / ".local" / "agent-data"
BUILD_ID_PATTERN = re.compile(r"Script build (\d+)", re.IGNORECASE)
TASK_KIND_PREFIX = "script-build://task-kinds/"
CONTRACT_PREFIX = "script-build://task-contracts/sha256/"
PLANNER_TOOLS = {
    "inspect_script_plan",
    "plan_script_tasks",
    "dispatch_script_tasks",
    "validate_attempt",
    "decide_script_task",
    "read_input_snapshot",
}
ACTION_LABELS = {
    "inspect_script_plan": "查看整棵任务树",
    "plan_script_tasks": "规划新 Task",
    "dispatch_script_tasks": "派发 Task 执行",
    "validate_attempt": "读取独立验收",
    "decide_script_task": "决定下一步",
    "read_input_snapshot": "读取冻结输入",
}


class RunNotFound(FileNotFoundError):
    pass


def agent_data_root() -> Path:
    configured = os.getenv("SCRIPT_BUILD_AGENT_DATA_ROOT")
    return Path(configured).expanduser().resolve() if configured else DEFAULT_AGENT_DATA_ROOT.resolve()


def _read_json(path: Path) -> dict[str, Any]:
    with path.open("r", encoding="utf-8") as handle:
        value = json.load(handle)
    return value if isinstance(value, dict) else {}


def _message_files(root: Path, trace_id: str) -> list[Path]:
    return sorted((root / "traces" / trace_id / "messages").glob("*.json"))


def _messages(root: Path, trace_id: str) -> list[dict[str, Any]]:
    return [_read_json(path) for path in _message_files(root, trace_id)]


def _build_id_for_root(root: Path, root_trace_id: str) -> int | None:
    meta_path = root / "traces" / root_trace_id / "meta.json"
    if meta_path.exists():
        context = _read_json(meta_path).get("context")
        if isinstance(context, dict):
            build_id = context.get("script_build_id")
            if isinstance(build_id, int) and not isinstance(build_id, bool):
                return build_id

    # Legacy traces may only expose the Build ID through GoalTree or early messages.
    goal_path = root / "traces" / root_trace_id / "goal.json"
    if goal_path.exists():
        match = BUILD_ID_PATTERN.search(str(_read_json(goal_path).get("mission", "")))
        if match:
            return int(match.group(1))
    for message in _messages(root, root_trace_id)[:6]:
        content = message.get("content")
        if isinstance(content, str):
            try:
                payload = json.loads(content)
            except json.JSONDecodeError:
                continue
            if isinstance(payload, dict) and isinstance(payload.get("script_build_id"), int):
                return payload["script_build_id"]
    return None


def _roots_by_build() -> dict[int, str]:
    root = agent_data_root()
    values: dict[int, str] = {}
    for ledger in sorted((root / "task-ledger").glob("*/orchestration/ledger.json")):
        root_trace_id = ledger.parents[1].name
        build_id = _build_id_for_root(root, root_trace_id)
        if build_id is not None:
            values[build_id] = root_trace_id
    return values


def _load_run(script_build_id: int) -> tuple[Path, str, dict[str, Any], list[dict[str, Any]]]:
    root = agent_data_root()
    root_trace_id = _roots_by_build().get(script_build_id)
    if not root_trace_id:
        raise RunNotFound(f"Persisted script build {script_build_id} was not found")
    ledger_path = root / "task-ledger" / root_trace_id / "orchestration" / "ledger.json"
    return root, root_trace_id, _read_json(ledger_path), _messages(root, root_trace_id)


def _task_kind(task: dict[str, Any]) -> str:
    if task.get("parent_task_id") is None:
        return "root"
    specs = task.get("specs") or []
    refs = (specs[-1].get("context_refs") or []) if specs else []
    for ref in refs:
        if isinstance(ref, str) and ref.startswith(TASK_KIND_PREFIX):
            return ref.removeprefix(TASK_KIND_PREFIX)
    return "task"


def _phase(display_path: str, task_kind: str) -> str:
    if task_kind == "root":
        return "mission"
    if display_path.startswith("0.1"):
        return "phase-one"
    if display_path.startswith("0.2"):
        return "phase-two"
    return "delivery"


def _task_view(task: dict[str, Any]) -> TaskNode:
    specs = task.get("specs") or [{}]
    spec = specs[-1]
    kind = _task_kind(task)
    path = str(task.get("display_path", "?"))
    return TaskNode(
        task_id=str(task.get("task_id", "")),
        display_path=path,
        parent_task_id=task.get("parent_task_id"),
        child_task_ids=list(task.get("child_task_ids") or []),
        task_kind=kind,
        phase=_phase(path, kind),
        objective=str(spec.get("objective") or "未记录目标"),
        status=str(task.get("status") or "unknown"),
        blocked_reason=task.get("blocked_reason"),
        attempt_count=len(task.get("attempt_ids") or []),
        validation_count=len(task.get("validation_ids") or []),
        decision_count=len(task.get("decision_ids") or []),
        current_spec_version=int(task.get("current_spec_version") or 1),
        acceptance_criteria=list(spec.get("acceptance_criteria") or []),
        context_refs=list(spec.get("context_refs") or []),
        created_at=str(task.get("created_at") or ""),
        updated_at=str(task.get("updated_at") or ""),
    )


def _parse_object(value: Any) -> Any:
    if not isinstance(value, str):
        return value
    try:
        return json.loads(value)
    except json.JSONDecodeError:
        return value


def _bounded_text(value: Any, limit: int = 3200) -> str:
    if value is None:
        return ""
    text = value if isinstance(value, str) else json.dumps(value, ensure_ascii=False, indent=2)
    normalized = text.strip()
    return normalized if len(normalized) <= limit else f"{normalized[:limit]}\n…（已截断）"


def _ids_from(value: Any, known_ids: set[str]) -> set[str]:
    found: set[str] = set()
    if isinstance(value, dict):
        for key, item in value.items():
            if key in {"task_id", "parent_task_id"} and isinstance(item, str) and item in known_ids:
                found.add(item)
            elif key in {"task_ids", "child_task_ids"} and isinstance(item, list):
                found.update(entry for entry in item if isinstance(entry, str) and entry in known_ids)
            found.update(_ids_from(item, known_ids))
    elif isinstance(value, list):
        for item in value:
            found.update(_ids_from(item, known_ids))
    return found


def _result_status(result: Any) -> str:
    if isinstance(result, dict):
        encoded = json.dumps(result, ensure_ascii=False).lower()
        if any(token in encoded for token in ('"error"', "protocol_violation", "failed")) and '"error": null' not in encoded:
            return "failed"
        return "succeeded"
    return "unknown"


def _planner_turns(
    messages: list[dict[str, Any]], task_views: list[TaskNode]
) -> tuple[list[PlannerTurn], str | None]:
    known_ids = {task.task_id for task in task_views}
    root_ids = {task.task_id for task in task_views if task.parent_task_id is None}
    tool_results = {
        message.get("tool_call_id"): message
        for message in messages
        if message.get("role") == "tool" and message.get("tool_call_id")
    }
    visible_ids = set(root_ids)
    current_phase = "mission"
    turns: list[PlannerTurn] = []
    current_focus: str | None = None
    for message in messages:
        if message.get("role") == "user":
            payload = _parse_object(message.get("content"))
            if isinstance(payload, dict) and payload.get("phase"):
                current_phase = f"phase-{payload['phase']}"
            continue
        if message.get("role") != "assistant" or not isinstance(message.get("content"), dict):
            continue
        content = message["content"]
        for call in content.get("tool_calls") or []:
            function = call.get("function") or {}
            tool_name = str(function.get("name") or "")
            if tool_name not in PLANNER_TOOLS:
                continue
            arguments = _parse_object(function.get("arguments") or "{}")
            arguments = arguments if isinstance(arguments, dict) else {"raw": arguments}
            result_message = tool_results.get(call.get("id"), {})
            result_content = result_message.get("content") or {}
            raw_result = result_content.get("result") if isinstance(result_content, dict) else result_content
            result = _parse_object(raw_result)
            affected = _ids_from(arguments, known_ids) | _ids_from(result, known_ids)
            newly_visible = _ids_from(result, known_ids)
            visible_ids.update(newly_visible)
            explicit_focus = arguments.get("task_id") or arguments.get("parent_task_id")
            if not explicit_focus and isinstance(arguments.get("task_ids"), list) and arguments["task_ids"]:
                explicit_focus = arguments["task_ids"][0]
            if explicit_focus in known_ids:
                current_focus = explicit_focus
            elif affected:
                current_focus = sorted(affected)[0]
            order = len(turns) + 1
            visible_snapshot = sorted(visible_ids)
            turns.append(
                PlannerTurn(
                    turn_id=f"planner-{message.get('sequence', order)}-{len(turns)}",
                    order=order,
                    message_sequence=int(message.get("sequence") or order),
                    phase=current_phase,
                    action=tool_name,
                    action_label=ACTION_LABELS[tool_name],
                    summary=_bounded_text(content.get("text"), 520) or ACTION_LABELS[tool_name],
                    observable_reasoning=_bounded_text(content.get("text"), 2200),
                    result_summary=_bounded_text(result, 900),
                    status=_result_status(result),
                    focus_task_id=current_focus,
                    affected_task_ids=sorted(affected),
                    visible_task_ids=visible_snapshot,
                    tool_name=tool_name,
                    tool_arguments=arguments,
                    tokens=int(message.get("tokens") or 0),
                    cost=float(message.get("cost") or 0),
                    created_at=str(message.get("created_at") or ""),
                )
            )
    if turns:
        turns[-1].visible_task_ids = sorted(known_ids)
    return turns, current_focus


def _checkpoint(tasks: list[TaskNode]) -> str:
    kinds = {task.task_kind for task in tasks}
    if "candidate-portfolio" in kinds:
        if "compose" in kinds:
            return "Phase 2 · Compose 子任务回流"
        return "Phase 2 · Candidate Portfolio"
    if any(task.task_kind == "direction" and task.status == "completed" for task in tasks):
        return "Phase 1 · Direction 已验收"
    return "Phase 1 · 方向与证据"


def _current_gap(tasks: list[TaskNode]) -> str:
    candidates = [task for task in tasks if task.status in {"needs_replan", "failed", "blocked"}]
    candidates.sort(key=lambda item: (item.display_path.count("."), item.updated_at), reverse=True)
    if candidates:
        task = candidates[0]
        reason = task.blocked_reason or task.status
        return f"{task.display_path} {task.task_kind} 等待 Planner 处理：{reason}"
    waiting = [task for task in tasks if task.status == "waiting_children"]
    if waiting:
        task = sorted(waiting, key=lambda item: item.display_path.count("."), reverse=True)[0]
        return f"{task.display_path} 正在等待子 Task 结果返回"
    return "当前持久化快照没有未处理缺口"


def _run_summary(
    script_build_id: int,
    root_trace_id: str,
    ledger: dict[str, Any],
    tasks: list[TaskNode],
    turns: list[PlannerTurn],
    focus: str | None,
) -> RunSummary:
    root_id = ledger.get("root_task_id")
    root_task = next((task for task in tasks if task.task_id == root_id), tasks[0])
    return RunSummary(
        script_build_id=script_build_id,
        title=root_task.objective,
        status=root_task.status,
        checkpoint=_checkpoint(tasks),
        root_trace_id=root_trace_id,
        revision=int(ledger.get("revision") or 0),
        task_count=len(tasks),
        attempt_count=len(ledger.get("attempts") or {}),
        planner_turn_count=len(turns),
        current_focus_task_id=focus,
        current_gap=_current_gap(tasks),
        created_at=root_task.created_at,
        updated_at=max((task.updated_at for task in tasks), default=root_task.updated_at),
    )


def execution_view(script_build_id: int) -> ExecutionView:
    _, root_trace_id, ledger, messages = _load_run(script_build_id)
    tasks = [_task_view(item) for item in (ledger.get("tasks") or {}).values()]
    tasks.sort(key=lambda item: [int(part) for part in item.display_path.split(".")])
    turns, focus = _planner_turns(messages, tasks)
    run = _run_summary(script_build_id, root_trace_id, ledger, tasks, turns, focus)
    warnings = [
        "树的生长顺序来自 Planner 工具结果；Task 状态来自当前 ledger，不伪造历史状态回放。",
        "Reasoning 仅展示模型主动输出的说明、工具参数与验收理由，不等同于隐藏思维链。",
    ]
    return ExecutionView(
        schema_version="script-build-execution-view/v1",
        data_mode="persisted-real-run",
        generated_at=datetime.now(timezone.utc).isoformat(),
        run=run,
        planner_turns=turns,
        tasks=tasks,
        snapshot_basis="Planner 工具调用顺序 + 当前 TaskLedger 持久化快照",
        warnings=warnings,
    )


def list_runs() -> list[RunSummary]:
    values: list[RunSummary] = []
    for build_id in sorted(_roots_by_build(), reverse=True):
        try:
            values.append(execution_view(build_id).run)
        except (OSError, ValueError, KeyError, TypeError, json.JSONDecodeError):
            continue
    return values


def _contract(root: Path, root_trace_id: str, task: TaskNode) -> dict[str, Any] | None:
    for ref in task.context_refs:
        if ref.startswith(CONTRACT_PREFIX):
            digest = ref.removeprefix(CONTRACT_PREFIX)
            path = root / "script-task-contracts" / root_trace_id / "sha256" / digest
            if path.exists():
                return _read_json(path)
    return None


def _step_title(tool_name: str) -> str:
    if tool_name.startswith("read_"):
        return f"读取数据 · {tool_name}"
    if tool_name.startswith(("create_", "batch_create_", "append_")):
        return f"写入草稿 · {tool_name}"
    if tool_name.startswith(("update_", "batch_update_")):
        return f"修改草稿 · {tool_name}"
    if tool_name == "submit_attempt":
        return "提交 Attempt"
    return f"调用工具 · {tool_name}"


def _agent_steps(root: Path, trace_id: str | None) -> tuple[str, list[AgentStep]]:
    if not trace_id:
        return "", []
    prompt = ""
    steps: list[AgentStep] = []
    for message in _messages(root, trace_id):
        role = str(message.get("role") or "unknown")
        content = message.get("content")
        if role == "system":
            prompt = _bounded_text(content, 9000)
            continue
        if role == "user":
            steps.append(
                AgentStep(
                    sequence=int(message.get("sequence") or 0), role=role, step_kind="instruction",
                    title="冻结的 TaskSpec", text=_bounded_text(content), tool_name=None, tool_arguments=None,
                    tokens=int(message.get("tokens") or 0), cost=float(message.get("cost") or 0),
                    duration_ms=message.get("duration_ms"), created_at=str(message.get("created_at") or ""),
                )
            )
            continue
        if role == "assistant" and isinstance(content, dict):
            text = _bounded_text(content.get("text"), 2200)
            if text:
                steps.append(
                    AgentStep(
                        sequence=int(message.get("sequence") or 0), role=role, step_kind="explanation",
                        title="Worker 的可观察说明", text=text, tool_name=None, tool_arguments=None,
                        tokens=int(message.get("tokens") or 0), cost=float(message.get("cost") or 0),
                        duration_ms=message.get("duration_ms"), created_at=str(message.get("created_at") or ""),
                    )
                )
            for call in content.get("tool_calls") or []:
                function = call.get("function") or {}
                name = str(function.get("name") or "unknown_tool")
                args = _parse_object(function.get("arguments") or "{}")
                steps.append(
                    AgentStep(
                        sequence=int(message.get("sequence") or 0), role=role, step_kind="tool_call",
                        title=_step_title(name), text="", tool_name=name,
                        tool_arguments=args if isinstance(args, dict) else {"raw": args},
                        tokens=int(message.get("tokens") or 0), cost=float(message.get("cost") or 0),
                        duration_ms=message.get("duration_ms"), created_at=str(message.get("created_at") or ""),
                    )
                )
        elif role == "tool":
            tool_name = content.get("tool_name") if isinstance(content, dict) else None
            result = content.get("result") if isinstance(content, dict) else content
            steps.append(
                AgentStep(
                    sequence=int(message.get("sequence") or 0), role=role, step_kind="tool_result",
                    title=f"工具返回 · {tool_name or 'unknown'}", text=_bounded_text(_parse_object(result)),
                    tool_name=tool_name, tool_arguments=None, tokens=0, cost=0,
                    duration_ms=message.get("duration_ms"), created_at=str(message.get("created_at") or ""),
                )
            )
    return prompt, steps


def _values_for_task(values: dict[str, Any], task_id: str) -> list[dict[str, Any]]:
    return [value for value in values.values() if value.get("task_id") == task_id]


def _first_for_attempt(values: Iterable[dict[str, Any]], attempt_id: str) -> dict[str, Any] | None:
    return next((value for value in values if value.get("attempt_id") == attempt_id), None)


def task_detail(script_build_id: int, task_id: str) -> TaskDetail:
    root, root_trace_id, ledger, _ = _load_run(script_build_id)
    raw_task = (ledger.get("tasks") or {}).get(task_id)
    if not raw_task:
        raise RunNotFound(f"Task {task_id} was not found in build {script_build_id}")
    task = _task_view(raw_task)
    contract = _contract(root, root_trace_id, task)
    validations = _values_for_task(ledger.get("validations") or {}, task_id)
    decisions = _values_for_task(ledger.get("decisions") or {}, task_id)
    raw_attempts = _values_for_task(ledger.get("attempts") or {}, task_id)
    attempt_details: list[AttemptDetail] = []
    reads: list[dict[str, Any]] = []
    outputs: list[dict[str, Any]] = []
    for attempt in sorted(raw_attempts, key=lambda item: item.get("created_at") or ""):
        prompt, steps = _agent_steps(root, attempt.get("worker_trace_id"))
        for step in steps:
            if step.step_kind == "tool_call" and step.tool_name and step.tool_name.startswith(("read_", "get_", "search_", "query_", "inspect_")):
                reads.append({"attempt_id": attempt.get("attempt_id"), "tool_name": step.tool_name, "arguments": step.tool_arguments, "proof": "worker_trace_tool_call"})
        submission = attempt.get("submission")
        if isinstance(submission, dict):
            for artifact in submission.get("artifact_refs") or []:
                outputs.append({"attempt_id": attempt.get("attempt_id"), "artifact": artifact, "proof": "attempt_submission"})
        validation = _first_for_attempt(validations, str(attempt.get("attempt_id")))
        decision = _first_for_attempt(decisions, str(attempt.get("attempt_id")))
        stats = attempt.get("execution_stats") or {}
        attempt_details.append(
            AttemptDetail(
                attempt_id=str(attempt.get("attempt_id") or ""), task_id=task_id,
                status=str(attempt.get("status") or "unknown"), worker_preset=attempt.get("worker_preset"),
                worker_trace_id=attempt.get("worker_trace_id"), execution_mode=attempt.get("execution_mode"),
                started_at=attempt.get("started_at"), completed_at=attempt.get("completed_at"),
                total_tokens=int(stats.get("total_tokens") or 0), total_cost=float(stats.get("total_cost") or 0),
                error=attempt.get("error"), submission=submission, validation=validation, decision=decision,
                prompt=prompt, steps=steps,
            )
        )
    lifecycle = [
        LifecycleEvent(event_id=f"task-{task_id}", kind="task", title="Planner 创建 Task", status="created", summary=task.objective, created_at=task.created_at)
    ]
    for attempt in raw_attempts:
        lifecycle.append(LifecycleEvent(event_id=str(attempt.get("attempt_id")), kind="attempt", title="Worker 执行 Attempt", status=str(attempt.get("status")), summary=str(attempt.get("worker_preset") or "worker"), created_at=str(attempt.get("created_at") or ""), attempt_id=attempt.get("attempt_id")))
    for validation in validations:
        lifecycle.append(LifecycleEvent(event_id=str(validation.get("validation_id")), kind="validation", title="Validator 独立验收", status=str(validation.get("verdict") or validation.get("status")), summary=str(validation.get("recommendation") or validation.get("summary") or ""), created_at=str(validation.get("created_at") or ""), attempt_id=validation.get("attempt_id")))
    for decision in decisions:
        lifecycle.append(LifecycleEvent(event_id=str(decision.get("decision_id")), kind="decision", title="Planner 决定下一步", status=str(decision.get("action")), summary=str(decision.get("reason") or ""), created_at=str(decision.get("created_at") or ""), attempt_id=decision.get("attempt_id")))
    lifecycle.sort(key=lambda item: item.created_at)
    inherited: list[dict[str, Any]] = [{"ref": ref, "proof": "task_spec_context_ref", "usage": "在冻结输入闭包中"} for ref in task.context_refs]
    if contract:
        for ref in contract.get("input_decision_refs") or []:
            inherited.append({"ref": ref, "proof": "task_contract_input_decision_ref", "usage": "显式接受的上游结果"})
        if contract.get("base_artifact_ref"):
            inherited.append({"ref": contract["base_artifact_ref"], "proof": "task_contract_base_artifact_ref", "usage": "本 Attempt 的基础产物"})
    adoption = [{"decision_id": value.get("decision_id"), "action": value.get("action"), "reason": value.get("reason"), "proof": "planner_decision"} for value in decisions]
    return TaskDetail(
        task=task, contract=contract, lifecycle=lifecycle, attempts=attempt_details,
        data_usage=DataUsage(inherited_inputs=inherited, confirmed_reads=reads, produced_outputs=outputs, adoption=adoption),
    )