主机：汇总任务链路与真实模型用量诊断

Observation view 增加 checkpoint、恢复分类、最后 Task/Attempt/Validation、当前与已恢复错误，以及按唯一 Trace 聚合的 Planner、Worker、Validator provider 用量。

script_build_host/src/script_build_host/application/observation_views.py

@@ -2,8 +2,11 @@
 
 from __future__ import annotations
 
+from datetime import datetime
 from typing import Any
 
+from script_build_host.application.mission_factory import normalize_topic_summary
+
 
 def mission_snapshot_view(ledger: Any) -> dict[str, Any]:
     if not hasattr(ledger, "revision"):
@@ -37,24 +40,14 @@ def mission_snapshot_view(ledger: Any) -> dict[str, Any]:
 
 def task_detail_view(task: Any) -> dict[str, Any]:
     current = task.current_spec
+    specs = getattr(task, "specs", (current,))
     return {
         "task_id": task.task_id,
         "parent_task_id": task.parent_task_id,
         "display_path": task.display_path,
         "status": _enum(task.status),
-        "current_spec": {
-            "version": current.version,
-            "objective": _text(current.objective, 2_000),
-            "acceptance_criteria": [
-                {
-                    "criterion_id": _text(value.criterion_id, 128),
-                    "description": _text(value.description, 500),
-                    "hard": bool(value.hard),
-                }
-                for value in current.acceptance_criteria
-            ],
-            "context_refs": [_safe_ref(value) for value in current.context_refs],
-        },
+        "current_spec": _spec_view(current),
+        "specs": [_spec_view(value) for value in specs],
         "child_task_ids": list(task.child_task_ids),
         "attempt_ids": list(task.attempt_ids),
         "validation_ids": list(task.validation_ids),
@@ -74,6 +67,7 @@ def _attempt_view(value: Any) -> dict[str, Any]:
         "spec_version": value.spec_version,
         "worker_trace_id": value.worker_trace_id,
         "worker_preset": value.worker_preset,
+        "execution_mode": getattr(value, "execution_mode", None),
         "status": _enum(value.status),
         "operation_id": value.operation_id,
         "snapshot_id": value.snapshot_id,
@@ -92,12 +86,17 @@ def _attempt_view(value: Any) -> dict[str, Any]:
             else None
         ),
         "error": _optional_text(value.error, 500),
+        "execution_stats": _execution_stats_view(getattr(value, "execution_stats", None)),
+        "started_at": _timestamp(getattr(value, "started_at", None)),
+        "completed_at": _timestamp(getattr(value, "completed_at", None)),
+        "duration_ms": _stage_duration_ms(value),
         "created_at": str(value.created_at),
         "updated_at": str(value.updated_at),
     }
 
 
 def _validation_view(value: Any) -> dict[str, Any]:
+    plan = getattr(value, "validation_plan", None)
     return {
         "validation_id": value.validation_id,
         "task_id": value.task_id,
@@ -106,6 +105,14 @@ def _validation_view(value: Any) -> dict[str, Any]:
         "snapshot_id": value.snapshot_id,
         "validator_trace_id": value.validator_trace_id,
         "validator_preset": value.validator_preset,
+        "validation_plan": (
+            {
+                "mode": _enum(getattr(plan, "mode", None)),
+                "preflight_rule_ids": list(getattr(plan, "preflight_rule_ids", ())),
+            }
+            if plan is not None
+            else None
+        ),
         "status": _enum(value.status),
         "verdict": _enum(value.verdict),
         "criterion_results": [
@@ -120,6 +127,10 @@ def _validation_view(value: Any) -> dict[str, Any]:
         "summary": _text(value.summary, 500),
         "recommendation": _text(value.recommendation, 256),
         "error": _optional_text(value.error, 500),
+        "execution_stats": _execution_stats_view(getattr(value, "execution_stats", None)),
+        "started_at": _timestamp(getattr(value, "started_at", None)),
+        "completed_at": _timestamp(getattr(value, "completed_at", None)),
+        "duration_ms": _stage_duration_ms(value),
         "created_at": str(value.created_at),
         "updated_at": str(value.updated_at),
     }
@@ -151,11 +162,152 @@ def _operation_view(value: Any) -> dict[str, Any]:
         "validation_ids": list(value.validation_ids),
         "deadline_at": str(value.deadline_at) if value.deadline_at else None,
         "error": _optional_text(value.error, 500),
+        "started_at": _timestamp(getattr(value, "started_at", None)),
+        "completed_at": _timestamp(getattr(value, "completed_at", None)),
+        "duration_ms": _stage_duration_ms(value),
         "created_at": str(value.created_at),
         "updated_at": str(value.updated_at),
     }
 
 
+def task_contract_summary_view(
+    task_id: str,
+    frozen: Any,
+    *,
+    spec_version: int | None = None,
+    spec_created_at: Any = None,
+) -> dict[str, Any]:
+    """Expose the frozen execution contract without storage paths or model configuration."""
+
+    payload = frozen.contract.to_payload()
+    allowed = (
+        "schema_version",
+        "task_kind",
+        "scope_ref",
+        "intent_class",
+        "objective",
+        "input_decision_refs",
+        "base_artifact_ref",
+        "write_scope",
+        "gap_ref",
+        "output_schema",
+        "criteria",
+        "goal_ids",
+        "supersedes_decision_ids",
+        "candidate_closure_decision_refs",
+        "adopted_decision_ids",
+        "held_or_rejected_decision_ids",
+        "compose_order",
+        "comparison_decision_refs",
+    )
+    return {
+        "task_id": task_id,
+        "spec_version": spec_version,
+        "spec_created_at": _timestamp(spec_created_at),
+        "contract_ref": frozen.uri,
+        "contract_digest": frozen.digest,
+        **{key: payload[key] for key in allowed if key in payload},
+    }
+
+
+def input_summary_view(snapshot: Any) -> dict[str, Any]:
+    """Return business-facing frozen input facts; prompts and model manifests stay private."""
+
+    topic_row = snapshot.topic.get("topic") if isinstance(snapshot.topic, dict) else None
+    topic_row = topic_row if isinstance(topic_row, dict) else {}
+    topic = normalize_topic_summary(
+        topic_row.get("result") or topic_row.get("topic_direction") or snapshot.topic_id,
+        max_length=500,
+    )
+    point_types: dict[str, int] = {}
+    for item in snapshot.persona_points:
+        if not isinstance(item, dict):
+            continue
+        label = str(item.get("point_type") or item.get("点类型") or "未分类")[:80]
+        point_types[label] = point_types.get(label, 0) + 1
+    strategies = []
+    for item in snapshot.strategies:
+        if not isinstance(item, dict):
+            continue
+        strategies.append(
+            {
+                "id": item.get("id") or item.get("strategy_id"),
+                "name": _optional_text(item.get("name"), 160),
+                "description": _optional_text(item.get("description"), 300),
+                "mode": _optional_text(item.get("mode"), 40),
+            }
+        )
+    return {
+        "script_build_id": int(snapshot.script_build_id),
+        "input_snapshot_id": str(snapshot.snapshot_id),
+        "input_digest": _text(snapshot.canonical_sha256, 80),
+        "source": {
+            "execution_id": int(snapshot.execution_id),
+            "topic_build_id": int(snapshot.topic_build_id),
+            "topic_id": int(snapshot.topic_id),
+        },
+        "topic": topic,
+        "persona": {
+            "account_name": _optional_text(snapshot.account.get("account_name"), 160),
+            "resolved_account_name": _optional_text(
+                snapshot.account.get("resolved_account_name"), 160
+            ),
+            "point_count": len(snapshot.persona_points),
+            "point_types": point_types,
+            "section_pattern_count": len(snapshot.section_patterns),
+        },
+        "strategies": strategies,
+    }
+
+
+def _spec_view(value: Any) -> dict[str, Any]:
+    return {
+        "version": value.version,
+        "objective": _text(value.objective, 2_000),
+        "acceptance_criteria": [
+            {
+                "criterion_id": _text(item.criterion_id, 128),
+                "description": _text(item.description, 500),
+                "hard": bool(item.hard),
+            }
+            for item in value.acceptance_criteria
+        ],
+        "context_refs": [_safe_ref(item) for item in value.context_refs],
+        "created_at": _timestamp(getattr(value, "created_at", None)),
+    }
+
+
+def _execution_stats_view(value: Any) -> dict[str, Any] | None:
+    if value is None:
+        return None
+    return {
+        "primary_model": _optional_text(getattr(value, "primary_model", None), 200),
+        "total_tokens": getattr(value, "total_tokens", None),
+        "total_cost": getattr(value, "total_cost", None),
+        "failure_code": _enum(getattr(value, "failure_code", None)),
+    }
+
+
+def _stage_duration_ms(value: Any) -> int | None:
+    duration = getattr(value, "duration_ms", None)
+    if duration is not None:
+        return int(duration)
+    started = getattr(value, "started_at", None)
+    completed = getattr(value, "completed_at", None)
+    if not started or not completed:
+        return None
+    try:
+        start = datetime.fromisoformat(str(started).replace("Z", "+00:00"))
+        end = datetime.fromisoformat(str(completed).replace("Z", "+00:00"))
+    except ValueError:
+        return None
+    return max(0, int((end - start).total_seconds() * 1_000))
+
+
+def _timestamp(value: Any) -> str | None:
+    return str(value) if value is not None else None
+
+
 def _artifact_ref(value: Any) -> dict[str, Any]:
     return {
         "uri": _safe_ref(value.uri),
@@ -184,4 +336,9 @@ def _enum(value: Any) -> str | None:
     return str(getattr(value, "value", value))
 
 
-__all__ = ["mission_snapshot_view", "task_detail_view"]
+__all__ = [
+    "input_summary_view",
+    "mission_snapshot_view",
+    "task_contract_summary_view",
+    "task_detail_view",
+]