feat(真实E2E): 增加分段恢复 检查点 与详细诊断报告

运行入口支持 script-build-id、resume、start-at 和 stop-after，并等待 durable partial/success 状态以消除竞态。报告增加恢复分类、Root/Operation、最后 Task/Attempt/Validation、模型与工具用量、阶段耗时及 deterministic preflight 命中情况，继续原子写盘。

script_build_host/src/script_build_host/internal_e2e.py

@@ -6,7 +6,7 @@ import argparse
 import asyncio
 import json
 import os
-from dataclasses import asdict, dataclass
+from dataclasses import asdict, dataclass, replace
 from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any, TypedDict, cast
@@ -16,8 +16,10 @@ import httpx
 from sqlalchemy import text
 
 from script_build_host.application.mission_service import (
+    PHASE_ONE_CAPABILITY_BOUNDARY,
     PHASE_TWO_CANDIDATE_PORTFOLIO_READY,
 )
+from script_build_host.domain.records import BuildStatus
 from script_build_host.infrastructure.config import ScriptBuildSettings
 from script_build_host.internal_runtime import create_internal_runtime
 from script_build_host.production import compose_production_host
@@ -42,6 +44,9 @@ class RealE2EReport:
     detail_status: str
     paragraph_count: int
     element_count: int
+    checkpoint: str = "finalized"
+    recovery_classification: str = ""
+    diagnostics: dict[str, Any] | None = None
 
 
 class _Source(TypedDict):
@@ -60,6 +65,10 @@ async def run_real_e2e(
     account_name: str | None = None,
     timeout_seconds: float = 3_600,
     poll_seconds: float = 2.0,
+    script_build_id: int | None = None,
+    resume: bool = False,
+    start_at: str = "auto",
+    stop_after: str | None = None,
 ) -> RealE2EReport:
     """Run real inputs and Qwen through Phase1-3, final publication, and readback."""
 
@@ -67,7 +76,8 @@ async def run_real_e2e(
         raise RuntimeError("the real internal E2E gate is allowed only in test environment")
     if settings.enabled_phase != 3:
         raise RuntimeError("SCRIPT_BUILD_ENABLED_PHASE=3 is required for the real E2E gate")
-    if settings.role_model is None:
+    model = settings.role_model
+    if model is None:
         raise RuntimeError("SCRIPT_BUILD_ROLE_MODEL is required for the real E2E gate")
     supplied = (execution_id, topic_build_id, topic_id)
     if any(value is not None for value in supplied) and not all(
@@ -75,20 +85,40 @@ async def run_real_e2e(
     ):
         raise ValueError("execution_id, topic_build_id, and topic_id must be supplied together")
 
+    if start_at not in {"auto", "phase-two", "phase-three", "finalize"}:
+        raise ValueError("start_at is invalid")
+    if stop_after not in {None, "phase-one", "structure", "portfolio", "root"}:
+        raise ValueError("stop_after is invalid")
+    if start_at != "auto" and script_build_id is None:
+        raise ValueError("a segmented start requires script_build_id")
+    if resume and script_build_id is None:
+        raise ValueError("resume requires script_build_id")
+    if stop_after == "phase-one" and script_build_id is None:
+        settings = settings.model_copy(update={"enabled_phase": 1})
+
     started_at = datetime.now(UTC)
     host = compose_production_host(settings, create_internal_runtime(settings))
     report_root = settings.agent_data_root / "e2e-reports"
     report_root.mkdir(parents=True, exist_ok=True)
     run_id = started_at.strftime("%Y%m%dT%H%M%SZ") + "-" + uuid4().hex[:8]
     report_path = report_root / f"{run_id}.json"
-    script_build_id: int | None = None
+    active_build_id: int | None = script_build_id
+    diagnostics: dict[str, Any] = {}
+    source: _Source
     await host.validate_startup()
     try:
-        if all(value is not None for value in supplied):
+        if script_build_id is not None and not all(value is not None for value in supplied):
+            source = {
+                "execution_id": execution_id or 0,
+                "topic_build_id": topic_build_id or 0,
+                "topic_id": topic_id or 0,
+                "account_name": account_name or "",
+            }
+        elif all(value is not None for value in supplied):
             assert execution_id is not None
             assert topic_build_id is not None
             assert topic_id is not None
-            source: _Source = {
+            source = {
                 "execution_id": execution_id,
                 "topic_build_id": topic_build_id,
                 "topic_id": topic_id,
@@ -108,36 +138,108 @@ async def run_real_e2e(
             base_url="http://script-build-internal",
             timeout=120,
         ) as client:
-            start = await client.post(
-                "/api/pattern/script_builds",
-                headers={"Idempotency-Key": f"real-e2e-start-{run_id}"},
-                json={
-                    "execution_id": source["execution_id"],
-                    "topic_build_id": source["topic_build_id"],
-                    "topic_id": source["topic_id"],
-                    "agent_type": "AigcAgent",
-                },
-            )
-            start_payload = _successful_json(start, "start")
-            script_build_id = int(start_payload["script_build_id"])
-            root_trace_id = str(start_payload["root_trace_id"])
-            print(f"已启动 script_build_id={script_build_id}", flush=True)
+            recovery_classification = ""
+            if active_build_id is None:
+                start = await client.post(
+                    "/api/pattern/script_builds",
+                    headers={"Idempotency-Key": f"real-e2e-start-{run_id}"},
+                    json={
+                        "execution_id": source["execution_id"],
+                        "topic_build_id": source["topic_build_id"],
+                        "topic_id": source["topic_id"],
+                        "agent_type": "AigcAgent",
+                    },
+                )
+                start_payload = _successful_json(start, "start")
+                active_build_id = int(start_payload["script_build_id"])
+                root_trace_id = str(start_payload["root_trace_id"])
+                print(f"已启动 script_build_id={active_build_id}", flush=True)
+            else:
+                binding = await host.composition.mission_service.bindings.get_by_build(
+                    active_build_id
+                )
+                root_trace_id = binding.root_trace_id
+                if resume:
+                    response = await client.post(
+                        f"/api/pattern/script_builds/{active_build_id}/resume",
+                        headers={"Idempotency-Key": f"real-e2e-resume-{run_id}"},
+                    )
+                    recovery_classification = str(
+                        _successful_json(response, "resume").get("classification") or ""
+                    )
 
-            phase_two_root = await _wait_for_root(
-                client,
-                script_build_id,
-                expected_status="blocked",
-                expected_reason=PHASE_TWO_CANDIDATE_PORTFOLIO_READY,
-                timeout_seconds=timeout_seconds,
-                poll_seconds=poll_seconds,
-            )
-            print("Phase 1/2 已到 CandidatePortfolio 边界", flush=True)
+            assert active_build_id is not None
+            script_build_id = active_build_id
 
-            advance = await client.post(
-                f"/api/pattern/script_builds/{script_build_id}/phase-three/advance",
-                headers={"Idempotency-Key": f"real-e2e-phase-three-{run_id}"},
-            )
-            _successful_json(advance, "phase-three advance")
+            if start_at == "phase-two":
+                response = await client.post(
+                    f"/api/pattern/script_builds/{script_build_id}/phase-two/advance",
+                    headers={"Idempotency-Key": f"real-e2e-phase-two-{run_id}"},
+                )
+                _successful_json(response, "phase-two advance")
+            if stop_after in {"phase-one", "structure"}:
+                checkpoint = await _wait_for_checkpoint(
+                    client,
+                    script_build_id,
+                    checkpoint=stop_after,
+                    timeout_seconds=timeout_seconds,
+                    poll_seconds=poll_seconds,
+                )
+                await host.composition.mission_service.pause_at_debug_checkpoint(
+                    script_build_id, checkpoint_code=checkpoint
+                )
+                report = _checkpoint_report(
+                    started_at=started_at,
+                    settings=settings,
+                    source=source,
+                    script_build_id=script_build_id,
+                    root_trace_id=root_trace_id,
+                    checkpoint=checkpoint,
+                    recovery_classification=recovery_classification,
+                )
+                report = await _attach_diagnostics(host, report)
+                _atomic_write_report(report_path, {"success": True, **asdict(report)})
+                return report
+
+            if start_at in {"auto", "phase-two"}:
+                phase_two_root = await _wait_for_root(
+                    client,
+                    script_build_id,
+                    expected_status="blocked",
+                    expected_reason=PHASE_TWO_CANDIDATE_PORTFOLIO_READY,
+                    timeout_seconds=timeout_seconds,
+                    poll_seconds=poll_seconds,
+                )
+                print("Phase 1/2 已到 CandidatePortfolio 边界", flush=True)
+                await _wait_for_build_status(
+                    host.composition.mission_service.legacy_state,
+                    script_build_id,
+                    expected=BuildStatus.PARTIAL,
+                    timeout_seconds=timeout_seconds,
+                    poll_seconds=poll_seconds,
+                )
+                if stop_after == "portfolio":
+                    report = _checkpoint_report(
+                        started_at=started_at,
+                        settings=settings,
+                        source=source,
+                        script_build_id=script_build_id,
+                        root_trace_id=root_trace_id,
+                        checkpoint="portfolio",
+                        recovery_classification=recovery_classification,
+                    )
+                    report = await _attach_diagnostics(host, report)
+                    _atomic_write_report(report_path, {"success": True, **asdict(report)})
+                    return report
+            else:
+                phase_two_root = {"blocked_reason": PHASE_TWO_CANDIDATE_PORTFOLIO_READY}
+
+            if start_at != "finalize":
+                advance = await client.post(
+                    f"/api/pattern/script_builds/{script_build_id}/phase-three/advance",
+                    headers={"Idempotency-Key": f"real-e2e-phase-three-{run_id}"},
+                )
+                _successful_json(advance, "phase-three advance")
             completed_root = await _wait_for_root(
                 client,
                 script_build_id,
@@ -147,14 +249,38 @@ async def run_real_e2e(
                 poll_seconds=poll_seconds,
             )
             print("Phase 3 Root Worker/Validator/ACCEPT 已完成", flush=True)
-
-            finalize = await client.post(
-                f"/api/pattern/script_builds/{script_build_id}/finalize",
-                headers={"Idempotency-Key": f"real-e2e-finalize-{run_id}"},
+            settled_status = await _wait_for_build_status(
+                host.composition.mission_service.legacy_state,
+                script_build_id,
+                expected=(BuildStatus.PARTIAL, BuildStatus.SUCCESS),
+                timeout_seconds=timeout_seconds,
+                poll_seconds=poll_seconds,
             )
-            final_payload = _successful_json(finalize, "finalize")
-            if not final_payload.get("committed"):
-                raise RuntimeError("final publication did not report a committed transaction")
+            if stop_after == "root":
+                report = _checkpoint_report(
+                    started_at=started_at,
+                    settings=settings,
+                    source=source,
+                    script_build_id=script_build_id,
+                    root_trace_id=root_trace_id,
+                    checkpoint="root",
+                    recovery_classification=recovery_classification,
+                )
+                report = await _attach_diagnostics(host, report)
+                _atomic_write_report(report_path, {"success": True, **asdict(report)})
+                return report
+
+            final_payload: dict[str, Any] = {}
+            if settled_status is not BuildStatus.SUCCESS:
+                finalize = await client.post(
+                    f"/api/pattern/script_builds/{script_build_id}/finalize",
+                    headers={"Idempotency-Key": f"real-e2e-finalize-{run_id}"},
+                )
+                final_payload = _successful_json(finalize, "finalize")
+                if not final_payload.get("committed"):
+                    raise RuntimeError(
+                        "final publication did not report a committed transaction"
+                    )
 
             publication_response = await client.get(
                 f"/api/pattern/script_builds/{script_build_id}/publication",
@@ -178,7 +304,7 @@ async def run_real_e2e(
             report = RealE2EReport(
                 started_at=started_at.isoformat(),
                 completed_at=datetime.now(UTC).isoformat(),
-                model=settings.role_model,
+                model=model,
                 execution_id=int(source["execution_id"]),
                 topic_build_id=int(source["topic_build_id"]),
                 topic_id=int(source["topic_id"]),
@@ -188,24 +314,36 @@ async def run_real_e2e(
                 phase_two_boundary=str(phase_two_root.get("blocked_reason") or ""),
                 root_status=str(completed_root["status"]),
                 publication_state=str(publication_value.get("state") or ""),
-                publication_digest=str(final_payload.get("legacy_projection_digest") or ""),
+                publication_digest=str(
+                    final_payload.get("legacy_projection_digest")
+                    or publication_value.get("expected_sha256")
+                    or ""
+                ),
                 detail_status=str(build["status"]),
                 paragraph_count=len(paragraphs),
                 element_count=len(elements),
+                recovery_classification=recovery_classification,
             )
+            report = await _attach_diagnostics(host, report)
             _atomic_write_report(report_path, {"success": True, **asdict(report)})
             print(f"E2E 报告: {report_path}", flush=True)
             return report
     except BaseException as exc:
+        if active_build_id is not None:
+            try:
+                diagnostics = await _collect_diagnostics(host, active_build_id)
+            except Exception as diagnostic_error:
+                diagnostics = {"diagnostic_error": str(diagnostic_error)[:500]}
         _atomic_write_report(
             report_path,
             {
                 "success": False,
                 "started_at": started_at.isoformat(),
                 "failed_at": datetime.now(UTC).isoformat(),
-                "script_build_id": script_build_id,
+                "script_build_id": active_build_id,
                 "error_type": type(exc).__name__,
                 "error": str(exc)[:2_000],
+                **diagnostics,
             },
         )
         print(f"失败报告: {report_path}", flush=True)
@@ -331,6 +469,209 @@ async def _wait_for_root(
     )
 
 
+async def _wait_for_build_status(
+    legacy_state: Any,
+    script_build_id: int,
+    *,
+    expected: BuildStatus | tuple[BuildStatus, ...],
+    timeout_seconds: float,
+    poll_seconds: float,
+) -> BuildStatus:
+    expected_statuses = (expected,) if isinstance(expected, BuildStatus) else expected
+    expected_values = ",".join(item.value for item in expected_statuses)
+    deadline = asyncio.get_running_loop().time() + timeout_seconds
+    last: BuildStatus | None = None
+    while asyncio.get_running_loop().time() < deadline:
+        last = await legacy_state.get_status(script_build_id)
+        if last in expected_statuses:
+            return last
+        if last in {BuildStatus.FAILED, BuildStatus.STOPPING, BuildStatus.STOPPED}:
+            raise RuntimeError(
+                f"build entered {last.value} while waiting for {expected_values}"
+            )
+        await asyncio.sleep(poll_seconds)
+    raise TimeoutError(
+        f"timed out waiting for build status={expected_values}; "
+        f"last={last.value if last is not None else None}"
+    )
+
+
+async def _wait_for_checkpoint(
+    client: httpx.AsyncClient,
+    script_build_id: int,
+    *,
+    checkpoint: str,
+    timeout_seconds: float,
+    poll_seconds: float,
+) -> str:
+    deadline = asyncio.get_running_loop().time() + timeout_seconds
+    while asyncio.get_running_loop().time() < deadline:
+        payload = _successful_json(
+            await client.get(f"/api/pattern/script_builds/{script_build_id}/mission"),
+            "checkpoint poll",
+        )
+        tasks = payload.get("tasks", [])
+        if checkpoint == "phase-one":
+            root_id = payload.get("root_task_id")
+            root = next(
+                (
+                    item
+                    for item in tasks
+                    if isinstance(item, dict) and item.get("task_id") == root_id
+                ),
+                {},
+            )
+            if (
+                root.get("status") == "blocked"
+                and root.get("blocked_reason") == PHASE_ONE_CAPABILITY_BOUNDARY
+            ):
+                return PHASE_ONE_CAPABILITY_BOUNDARY
+        else:
+            for task in tasks:
+                refs = (
+                    task.get("current_spec", {}).get("context_refs", [])
+                    if isinstance(task, dict)
+                    else []
+                )
+                if (
+                    task.get("status") == "completed"
+                    and "script-build://task-kinds/structure" in refs
+                ):
+                    return "STRUCTURE_ACCEPTED"
+        await asyncio.sleep(poll_seconds)
+    raise TimeoutError(f"timed out waiting for debug checkpoint {checkpoint}")
+
+
+def _checkpoint_report(
+    *,
+    started_at: datetime,
+    settings: ScriptBuildSettings,
+    source: _Source,
+    script_build_id: int,
+    root_trace_id: str,
+    checkpoint: str,
+    recovery_classification: str,
+) -> RealE2EReport:
+    return RealE2EReport(
+        started_at=started_at.isoformat(),
+        completed_at=datetime.now(UTC).isoformat(),
+        model=settings.role_model or "",
+        execution_id=int(source["execution_id"]),
+        topic_build_id=int(source["topic_build_id"]),
+        topic_id=int(source["topic_id"]),
+        account_name=str(source["account_name"]),
+        script_build_id=script_build_id,
+        root_trace_id=root_trace_id,
+        phase_two_boundary=(
+            PHASE_TWO_CANDIDATE_PORTFOLIO_READY if checkpoint == "portfolio" else ""
+        ),
+        root_status="completed" if checkpoint == "root" else "blocked",
+        publication_state="",
+        publication_digest="",
+        detail_status="partial",
+        paragraph_count=0,
+        element_count=0,
+        checkpoint=checkpoint,
+        recovery_classification=recovery_classification,
+    )
+
+
+async def _collect_diagnostics(host: Any, script_build_id: int) -> dict[str, Any]:
+    binding = await host.composition.mission_service.bindings.get_by_build(script_build_id)
+    ledger = await host.composition.coordinator.task_store.load(binding.root_trace_id)
+    ledger_value = ledger.to_dict()
+    root = ledger.tasks[ledger.root_task_id]
+    active = [
+        ledger_value["operations"][item.operation_id]
+        for item in ledger.operations.values()
+        if item.status.value in {"pending", "running", "stop_requested"}
+    ]
+    attempts = list(ledger.attempts.values())
+    validations = list(ledger.validations.values())
+    task_seconds: dict[str, float] = {}
+    preset_usage: dict[str, dict[str, int | float]] = {}
+    domain_errors: list[dict[str, str]] = []
+    for item in [*attempts, *validations]:
+        task = ledger.tasks[item.task_id]
+        kind = next(
+            (
+                ref.rsplit("/", 1)[-1]
+                for ref in task.current_spec.context_refs
+                if "/task-kinds/" in ref
+            ),
+            "root",
+        )
+        if item.duration_ms is not None:
+            task_seconds[kind] = task_seconds.get(kind, 0.0) + item.duration_ms / 1000
+        stats = item.execution_stats
+        if stats is not None:
+            key = getattr(item, "worker_preset", None) or getattr(
+                item, "validator_preset", "unknown"
+            )
+            usage = preset_usage.setdefault(
+                str(key),
+                {"calls": 0, "tokens": 0, "timeouts": 0, "failures": 0, "seconds": 0.0},
+            )
+            usage["calls"] += 1
+            usage["tokens"] += int(stats.total_tokens or 0)
+            usage["seconds"] += (item.duration_ms or 0) / 1000
+            failure_code = str(stats.failure_code or "")
+            if failure_code:
+                usage["failures"] += 1
+                if "timeout" in failure_code.lower():
+                    usage["timeouts"] += 1
+                domain_errors.append({"code": failure_code, "message": str(item.error or "")})
+
+    trace_ids = {binding.root_trace_id}
+    trace_ids.update(item.worker_trace_id for item in attempts)
+    trace_ids.update(item.validator_trace_id for item in validations)
+    tool_calls: dict[str, int] = {}
+    for trace_id in trace_ids:
+        for message in await host.composition.mission_service.runner.trace_store.get_trace_messages(
+            trace_id
+        ):
+            content = message.content
+            if not isinstance(content, dict):
+                continue
+            for call in content.get("tool_calls") or []:
+                name = str(call.get("function", {}).get("name") or "unknown")
+                tool_calls[name] = tool_calls.get(name, 0) + 1
+    last_task = max(ledger.tasks.values(), key=lambda item: item.updated_at)
+    last_attempt = max(attempts, key=lambda item: item.updated_at) if attempts else None
+    last_validation = max(validations, key=lambda item: item.updated_at) if validations else None
+    return {
+        "checkpoint": root.blocked_reason or root.status.value,
+        "root": {"status": root.status.value, "blocked_reason": root.blocked_reason},
+        "active_operations": active,
+        "last_task": ledger_value["tasks"][last_task.task_id],
+        "last_attempt": (
+            ledger_value["attempts"][last_attempt.attempt_id] if last_attempt else None
+        ),
+        "last_validation": (
+            ledger_value["validations"][last_validation.validation_id] if last_validation else None
+        ),
+        "last_domain_error": domain_errors[-1] if domain_errors else None,
+        "phase_timing_seconds": task_seconds,
+        "preset_usage": preset_usage,
+        "tool_calls": tool_calls,
+        "paragraph_tool_calls": {
+            "single": tool_calls.get("create_script_paragraph", 0),
+            "batch": tool_calls.get("create_script_paragraphs", 0),
+        },
+        "deterministic_preflight_rejection": any(
+            "Deterministic preflight rejected" in item.summary for item in validations
+        ),
+    }
+
+
+async def _attach_diagnostics(host: Any, report: RealE2EReport) -> RealE2EReport:
+    try:
+        diagnostics = await _collect_diagnostics(host, report.script_build_id)
+    except Exception as exc:
+        diagnostics = {"diagnostic_error": str(exc)[:500]}
+    return replace(report, diagnostics=diagnostics)
+
+
 def _successful_json(response: httpx.Response, operation: str) -> dict[str, Any]:
     try:
         payload = response.json()
@@ -365,6 +706,12 @@ def main() -> None:
     parser.add_argument("--account-name")
     parser.add_argument("--timeout-seconds", type=float, default=3_600)
     parser.add_argument("--poll-seconds", type=float, default=2.0)
+    parser.add_argument("--script-build-id", type=int)
+    parser.add_argument("--resume", action="store_true")
+    parser.add_argument(
+        "--start-at", choices=("auto", "phase-two", "phase-three", "finalize"), default="auto"
+    )
+    parser.add_argument("--stop-after", choices=("phase-one", "structure", "portfolio", "root"))
     args = parser.parse_args()
     report = asyncio.run(
         run_real_e2e(
@@ -375,6 +722,10 @@ def main() -> None:
             account_name=args.account_name,
             timeout_seconds=args.timeout_seconds,
             poll_seconds=args.poll_seconds,
+            script_build_id=args.script_build_id,
+            resume=args.resume,
+            start_at=args.start_at,
+            stop_after=args.stop_after,
         )
     )
     print(json.dumps(asdict(report), ensure_ascii=False, indent=2, sort_keys=True))