feat(真实E2E): 新增旧输入到最终回读的可执行门禁

自动发现完整旧版真实输入，通过 ASGI API 驱动 Phase1、Phase2、Phase3、finalize、publication 与 legacy detail 回读。

持久化脱敏运行报告，并限制启动期 HTTP 容错次数，让持续 500 立即暴露而不是假性卡死。

script_build_host/pyproject.toml

@@ -18,6 +18,7 @@ dependencies = [
 [project.scripts]
 script-build-host = "script_build_host.cli:main"
 script-build-rebuild-decode-index = "script_build_host.infrastructure.decode_index_builder:main"
+script-build-real-e2e = "script_build_host.internal_e2e:main"
 
 [project.optional-dependencies]
 dev = [

script_build_host/src/script_build_host/internal_e2e.py

@@ -0,0 +1,384 @@
+"""Executable real-model E2E gate for the internal Script Build environment."""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import os
+from dataclasses import asdict, dataclass
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any, TypedDict, cast
+from uuid import uuid4
+
+import httpx
+from sqlalchemy import text
+
+from script_build_host.application.mission_service import (
+    PHASE_TWO_CANDIDATE_PORTFOLIO_READY,
+)
+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
+from script_build_host.repositories import LegacySqlAlchemyInputReader
+
+
+@dataclass(frozen=True, slots=True)
+class RealE2EReport:
+    started_at: str
+    completed_at: str
+    model: str
+    execution_id: int
+    topic_build_id: int
+    topic_id: int
+    account_name: str
+    script_build_id: int
+    root_trace_id: str
+    phase_two_boundary: str
+    root_status: str
+    publication_state: str
+    publication_digest: str
+    detail_status: str
+    paragraph_count: int
+    element_count: int
+
+
+class _Source(TypedDict):
+    execution_id: int
+    topic_build_id: int
+    topic_id: int
+    account_name: str
+
+
+async def run_real_e2e(
+    settings: ScriptBuildSettings,
+    *,
+    execution_id: int | None = None,
+    topic_build_id: int | None = None,
+    topic_id: int | None = None,
+    account_name: str | None = None,
+    timeout_seconds: float = 3_600,
+    poll_seconds: float = 2.0,
+) -> RealE2EReport:
+    """Run real inputs and Qwen through Phase1-3, final publication, and readback."""
+
+    if settings.environment.lower() != "test":
+        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:
+        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(
+        value is not None for value in supplied
+    ):
+        raise ValueError("execution_id, topic_build_id, and topic_id must be supplied together")
+
+    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
+    await host.validate_startup()
+    try:
+        if 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 = {
+                "execution_id": execution_id,
+                "topic_build_id": topic_build_id,
+                "topic_id": topic_id,
+                "account_name": account_name or "",
+            }
+        else:
+            source = await _discover_source(host, settings, account_name=account_name)
+        print(
+            "真实输入: "
+            f"execution={source['execution_id']} build={source['topic_build_id']} "
+            f"topic={source['topic_id']} account={source['account_name']}",
+            flush=True,
+        )
+        transport = httpx.ASGITransport(app=host.composition.app, raise_app_exceptions=False)
+        async with httpx.AsyncClient(
+            transport=transport,
+            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)
+
+            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)
+
+            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,
+                expected_status="completed",
+                expected_reason=None,
+                timeout_seconds=timeout_seconds,
+                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}"},
+            )
+            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",
+                params={"type": "final"},
+            )
+            publication = _successful_json(publication_response, "publication readback")
+            detail_response = await client.get(f"/api/pattern/script_builds/{script_build_id}")
+            detail = _successful_json(detail_response, "legacy detail readback")
+            build = detail.get("build")
+            if not isinstance(build, dict) or build.get("status") != "success":
+                raise RuntimeError("legacy detail does not report a successful build")
+            paragraphs = detail.get("paragraphs")
+            elements = detail.get("elements")
+            if not isinstance(paragraphs, list) or not paragraphs:
+                raise RuntimeError("legacy detail contains no published paragraphs")
+            if not isinstance(elements, list):
+                raise RuntimeError("legacy detail elements are not an array")
+            publication_value = publication.get("publication", publication)
+            if not isinstance(publication_value, dict):
+                raise RuntimeError("final publication readback is not an object")
+            report = RealE2EReport(
+                started_at=started_at.isoformat(),
+                completed_at=datetime.now(UTC).isoformat(),
+                model=settings.role_model,
+                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=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 ""),
+                detail_status=str(build["status"]),
+                paragraph_count=len(paragraphs),
+                element_count=len(elements),
+            )
+            _atomic_write_report(report_path, {"success": True, **asdict(report)})
+            print(f"E2E 报告: {report_path}", flush=True)
+            return report
+    except BaseException as exc:
+        _atomic_write_report(
+            report_path,
+            {
+                "success": False,
+                "started_at": started_at.isoformat(),
+                "failed_at": datetime.now(UTC).isoformat(),
+                "script_build_id": script_build_id,
+                "error_type": type(exc).__name__,
+                "error": str(exc)[:2_000],
+            },
+        )
+        print(f"失败报告: {report_path}", flush=True)
+        raise
+    finally:
+        await host.close()
+
+
+async def _discover_source(
+    host: Any,
+    settings: ScriptBuildSettings,
+    *,
+    account_name: str | None,
+) -> _Source:
+    persona_accounts = {path.name for path in settings.persona_root.iterdir() if path.is_dir()}
+    async with host.database.read() as session:
+        rows = (
+            (
+                await session.execute(
+                    text(
+                        "SELECT t.execution_id, t.build_id AS topic_build_id, "
+                        "t.id AS topic_id, b.personal_config, "
+                        "(SELECT COUNT(*) FROM topic_build_point p "
+                        " WHERE p.topic_id=t.id AND p.build_id=t.build_id AND p.is_active=1) "
+                        "AS point_count, "
+                        "(SELECT COUNT(*) FROM topic_build_composition_item i "
+                        " WHERE i.topic_id=t.id AND i.build_id=t.build_id AND i.is_active=1) "
+                        "AS item_count "
+                        "FROM topic_build_topic t "
+                        "JOIN topic_build_record b ON b.id=t.build_id "
+                        "JOIN topic_pattern_execution e ON e.id=t.execution_id "
+                        "WHERE e.status='success' AND b.status='success' "
+                        "AND b.is_deleted=0 AND t.status IN ('mature','success') "
+                        "ORDER BY b.end_time DESC, t.id DESC LIMIT 200"
+                    )
+                )
+            )
+            .mappings()
+            .all()
+        )
+    reader = LegacySqlAlchemyInputReader(host.database.read)
+    candidates: list[tuple[int, int, _Source]] = []
+    for raw in rows:
+        personal = raw.get("personal_config")
+        if isinstance(personal, str):
+            try:
+                personal = json.loads(personal)
+            except json.JSONDecodeError:
+                personal = {}
+        resolved_account = (
+            str(personal.get("account_name") or "") if isinstance(personal, dict) else ""
+        )
+        if account_name and resolved_account != account_name:
+            continue
+        if persona_accounts and resolved_account not in persona_accounts:
+            continue
+        if int(raw["point_count"] or 0) < 1 or int(raw["item_count"] or 0) < 1:
+            continue
+        source: _Source = {
+            "execution_id": int(cast(Any, raw["execution_id"])),
+            "topic_build_id": int(cast(Any, raw["topic_build_id"])),
+            "topic_id": int(cast(Any, raw["topic_id"])),
+            "account_name": resolved_account,
+        }
+        candidates.append((int(raw["item_count"]), int(raw["point_count"]), source))
+    for _, __, source in sorted(candidates, key=lambda item: (item[0], item[1])):
+        try:
+            await reader.read_topic_graph(
+                execution_id=source["execution_id"],
+                topic_build_id=source["topic_build_id"],
+                topic_id=source["topic_id"],
+            )
+        except Exception:
+            continue
+        return source
+    suffix = f" for account {account_name}" if account_name else ""
+    raise RuntimeError(f"no complete real Script Build input source was found{suffix}")
+
+
+async def _wait_for_root(
+    client: httpx.AsyncClient,
+    script_build_id: int,
+    *,
+    expected_status: str,
+    expected_reason: str | None,
+    timeout_seconds: float,
+    poll_seconds: float,
+) -> dict[str, Any]:
+    deadline = asyncio.get_running_loop().time() + timeout_seconds
+    last_state: tuple[str, str | None] | None = None
+    received_snapshot = False
+    transient_failures = 0
+    while asyncio.get_running_loop().time() < deadline:
+        response = await client.get(f"/api/pattern/script_builds/{script_build_id}/mission")
+        if not received_snapshot and response.status_code in {404, 500} and transient_failures < 5:
+            transient_failures += 1
+            await asyncio.sleep(poll_seconds)
+            continue
+        payload = _successful_json(response, "mission poll")
+        received_snapshot = True
+        root_id = payload.get("root_task_id")
+        tasks = payload.get("tasks")
+        if not isinstance(tasks, list):
+            raise RuntimeError("mission snapshot tasks are not an array")
+        root = next(
+            (item for item in tasks if isinstance(item, dict) and item.get("task_id") == root_id),
+            None,
+        )
+        if root is None:
+            raise RuntimeError("mission snapshot does not contain its root task")
+        state = (str(root.get("status")), root.get("blocked_reason"))
+        if state != last_state:
+            print(f"Root 状态: status={state[0]} reason={state[1]}", flush=True)
+            last_state = state
+        if state[0] in {"failed", "cancelled"}:
+            raise RuntimeError(f"root task entered terminal failure state: {state}")
+        if state[0] == expected_status and (expected_reason is None or state[1] == expected_reason):
+            return root
+        await asyncio.sleep(poll_seconds)
+    raise TimeoutError(
+        f"timed out waiting for root status={expected_status} reason={expected_reason}; "
+        f"last={last_state}"
+    )
+
+
+def _successful_json(response: httpx.Response, operation: str) -> dict[str, Any]:
+    try:
+        payload = response.json()
+    except ValueError as exc:
+        raise RuntimeError(f"{operation} returned non-JSON HTTP {response.status_code}") from exc
+    if not response.is_success:
+        code = payload.get("error_code") if isinstance(payload, dict) else None
+        message = payload.get("message") if isinstance(payload, dict) else None
+        raise RuntimeError(
+            f"{operation} failed with HTTP {response.status_code}: {code or message or 'unknown'}"
+        )
+    if not isinstance(payload, dict):
+        raise RuntimeError(f"{operation} response root is not an object")
+    return payload
+
+
+def _atomic_write_report(path: Path, value: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_name(f".{path.name}.{uuid4().hex}.tmp")
+    temporary.write_text(
+        json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
+        encoding="utf-8",
+    )
+    os.replace(temporary, path)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="运行内部真实脚本构建 E2E")
+    parser.add_argument("--execution-id", type=int)
+    parser.add_argument("--topic-build-id", type=int)
+    parser.add_argument("--topic-id", type=int)
+    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)
+    args = parser.parse_args()
+    report = asyncio.run(
+        run_real_e2e(
+            ScriptBuildSettings(),  # type: ignore[call-arg]
+            execution_id=args.execution_id,
+            topic_build_id=args.topic_build_id,
+            topic_id=args.topic_id,
+            account_name=args.account_name,
+            timeout_seconds=args.timeout_seconds,
+            poll_seconds=args.poll_seconds,
+        )
+    )
+    print(json.dumps(asdict(report), ensure_ascii=False, indent=2, sort_keys=True))
+
+
+if __name__ == "__main__":
+    main()

script_build_host/tests/test_internal_e2e.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+import httpx
+import pytest
+
+from script_build_host.internal_e2e import _wait_for_root
+
+
+@pytest.mark.asyncio
+async def test_mission_poll_allows_only_bounded_startup_failures() -> None:
+    calls = 0
+
+    def handler(_request: httpx.Request) -> httpx.Response:
+        nonlocal calls
+        calls += 1
+        if calls <= 5:
+            return httpx.Response(500, json={"error_code": "LEDGER_NOT_READY"})
+        return httpx.Response(
+            200,
+            json={
+                "root_task_id": "root",
+                "tasks": [
+                    {
+                        "task_id": "root",
+                        "status": "blocked",
+                        "blocked_reason": "BOUNDARY",
+                    }
+                ],
+            },
+        )
+
+    async with httpx.AsyncClient(
+        transport=httpx.MockTransport(handler), base_url="http://internal"
+    ) as client:
+        result = await _wait_for_root(
+            client,
+            7,
+            expected_status="blocked",
+            expected_reason="BOUNDARY",
+            timeout_seconds=1,
+            poll_seconds=0,
+        )
+
+    assert result["task_id"] == "root"
+    assert calls == 6
+
+
+@pytest.mark.asyncio
+async def test_mission_poll_surfaces_persistent_server_failure() -> None:
+    def handler(_request: httpx.Request) -> httpx.Response:
+        return httpx.Response(500, json={"error_code": "InternalServerError"})
+
+    async with httpx.AsyncClient(
+        transport=httpx.MockTransport(handler), base_url="http://internal"
+    ) as client:
+        with pytest.raises(RuntimeError, match="mission poll failed with HTTP 500"):
+            await _wait_for_root(
+                client,
+                7,
+                expected_status="blocked",
+                expected_reason="BOUNDARY",
+                timeout_seconds=1,
+                poll_seconds=0,
+            )