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

from __future__ import annotations

import os
import time
from threading import Lock
from typing import Optional

from fastapi import FastAPI, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware

from .execution_builder import ExecutionViewBuilder
from .execution_view_payload import compact_execution_view
from .inspector_projection import (
    project_activity_detail,
    project_artifact_detail,
    project_event_detail,
    project_round_detail,
)
from .providers import LocalDatabaseProvider
from .repositories import ActivityNotFound, BuildNotFound
from .runtime_bridge import database_host_override, runtime_dir
from .prompt_projection import project_prompt
from .sanitizer import sanitize
from .module_audit import router as module_audit_router
from .module_audit.repository import AuditRepository
from .card_details import CardBusinessDataProjector, CardDataNotFound
from .card_details.common import find_detail
from .source_lineage import InspectorWorkbenchProjector, InspectorViewNotFound


app = FastAPI(
    title="Script Build Runtime Visualization",
    description="Read-only V8 projection of script-build decisions and structured runtime events.",
    version="8.0.0",
)

origins = [
    value.strip()
    for value in os.getenv(
        "VISUALIZATION_CORS_ORIGINS",
        "http://127.0.0.1:3008,http://localhost:3008",
    ).split(",")
    if value.strip()
]
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=False,
    allow_methods=["GET"],
    allow_headers=["Accept", "Content-Type"],
)
app.add_middleware(GZipMiddleware, minimum_size=1_000, compresslevel=5)
app.include_router(module_audit_router)

provider = LocalDatabaseProvider()
builder = ExecutionViewBuilder()
card_data_projector = CardBusinessDataProjector()
inspector_workbench_projector = InspectorWorkbenchProjector()
audit_repository = AuditRepository()
_inspector_cache: dict[tuple[int, str], tuple[float, dict]] = {}
_inspector_cache_lock = Lock()
_build_view_cache: dict[int, tuple[float, dict, dict]] = {}
_build_view_cache_lock = Lock()


def _load_bundle_and_view(script_build_id: int) -> tuple[dict, dict]:
    """Share one read-only run snapshot across Inspector source requests."""
    now = time.monotonic()
    with _build_view_cache_lock:
        cached = _build_view_cache.get(script_build_id)
        if cached and cached[0] > now:
            return cached[1], cached[2]
        # The first parallel Inspector burst for a run must not issue the same
        # ORM bundle query from several worker threads.  Build the immutable
        # snapshot once while holding the short per-cache critical section.
        bundle = provider.load_bundle(script_build_id)
        view = builder.from_bundle(bundle)
        terminal = str((bundle.get("record") or {}).get("status") or "").lower() in {
            "success", "completed", "partial", "failed", "stopped", "cancelled"
        }
        ttl = 300.0 if terminal else 3.0
        _build_view_cache[script_build_id] = (now + ttl, bundle, view)
        if len(_build_view_cache) > 16:
            expired = [key for key, value in _build_view_cache.items() if value[0] <= now]
            for key in expired or list(_build_view_cache)[:4]:
                _build_view_cache.pop(key, None)
        return bundle, view


def execution_view(script_build_id: int) -> dict:
    _bundle, view = _load_bundle_and_view(script_build_id)
    return compact_execution_view(view)


def _safe_load_log_windows(
    script_build_id: int,
    event_details: dict[int, dict],
) -> dict[int, str]:
    try:
        anchors = [
            (
                event_id,
                str(event.get("msg_id") or "").strip(),
                int(event.get("event_seq")) if event.get("event_seq") is not None else None,
            )
            for event_id, event in event_details.items()
            if str(event.get("msg_id") or "").strip()
        ]
        return audit_repository.load_event_log_windows(script_build_id, anchors)
    except Exception:
        # Log context is supplemental; DB/Event lineage must remain usable
        # when the historical log table is absent or temporarily unavailable.
        return {}


@app.get("/api/health")
def health():
    source_ok = False
    source_error = None
    try:
        provider.list_builds(limit=1)
        source_ok = True
    except Exception as exc:  # Health remains inspectable when DB is unavailable.
        source_error = f"{type(exc).__name__}: {exc}"
    return {
        "ok": True,
        "source": {
            "mode": provider.mode,
            "available": source_ok,
            "error": sanitize(source_error),
        },
        "runtimeDir": str(runtime_dir()),
        "readOnly": True,
        "databaseHostOverrideEnabled": bool(database_host_override()),
        "databaseSessionReadOnly": True,
    }


@app.get("/api/capabilities")
def capabilities():
    return {
        "schemaVersion": "8",
        "sourceMode": provider.mode,
        "businessSources": [
            "script_build_record",
            "script_build_round",
            "script_build_branch",
            "script_build_data_decision",
            "script_build_multipath_decision",
            "script_build_domain_info",
            "script_build_paragraph",
            "script_build_element",
            "script_build_paragraph_element",
        ],
        "runtimeSupplement": ["script_build_event", "script_build_event_body"],
        "readOnly": True,
        "databaseHostOverrideEnabled": bool(database_host_override()),
        "databaseSessionReadOnly": True,
    }


@app.get("/api/script-builds")
def list_script_builds(
    limit: int = Query(30, ge=1, le=100), status: Optional[str] = None
):
    try:
        return {
            "items": provider.list_builds(limit=limit, status=status),
            "sourceMode": provider.mode,
            "warnings": [],
        }
    except Exception as exc:
        raise HTTPException(
            status_code=503,
            detail=f"读取构建列表失败：{type(exc).__name__}: {sanitize(str(exc))}",
        ) from exc


@app.get("/api/script-builds/{script_build_id}/execution-view")
def get_execution_view(script_build_id: int):
    try:
        return execution_view(script_build_id)
    except BuildNotFound as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except Exception as exc:
        raise HTTPException(
            status_code=503,
            detail=f"读取构建失败：{type(exc).__name__}: {sanitize(str(exc))}",
        ) from exc


@app.get("/api/script-builds/{script_build_id}/card-data/{detail_ref:path}")
def get_card_data(script_build_id: int, detail_ref: str):
    try:
        bundle, view = _load_bundle_and_view(script_build_id)
        return sanitize(card_data_projector.project(
            detail_ref,
            bundle=bundle,
            view=view,
            load_event=lambda event_id: provider.load_event_detail(
                script_build_id, event_id
            ),
        ))
    except BuildNotFound as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except (ActivityNotFound, CardDataNotFound) as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except Exception as exc:
        raise HTTPException(
            status_code=503,
            detail=f"读取卡片数据流失败：{type(exc).__name__}: {sanitize(str(exc))}",
        ) from exc


@app.get("/api/script-builds/{script_build_id}/inspector-view/{detail_ref:path}")
def get_inspector_view(script_build_id: int, detail_ref: str):
    """Return business, evidence and raw runtime columns in one response."""
    try:
        key = (script_build_id, detail_ref)
        now = time.monotonic()
        with _inspector_cache_lock:
            cached = _inspector_cache.get(key)
            if cached and cached[0] > now:
                return cached[1]

        bundle, view = _load_bundle_and_view(script_build_id)
        record = bundle.get("record") or {}
        event_ids = inspector_workbench_projector.related_event_ids(
            detail_ref,
            bundle=bundle,
            view=view,
        )
        event_details = provider.load_event_details(script_build_id, event_ids)
        # Event bodies have already been redacted and bounded by the
        # repository.  Do not impose a second lower text limit here, which
        # would silently invalidate selectors without updating completeness.
        result = sanitize(
            inspector_workbench_projector.project(
                detail_ref,
                script_build_id=script_build_id,
                bundle=bundle,
                view=view,
                event_details=event_details,
                log_contents=_safe_load_log_windows(script_build_id, event_details),
            ),
            max_text=None,
        )
        terminal = str(record.get("status") or "").lower() in {
            "success", "completed", "partial", "failed", "stopped", "cancelled"
        }
        ttl = 300.0 if terminal else 3.0
        with _inspector_cache_lock:
            _inspector_cache[key] = (now + ttl, result)
            if len(_inspector_cache) > 256:
                expired = [item for item, value in _inspector_cache.items() if value[0] <= now]
                for item in expired or list(_inspector_cache)[:64]:
                    _inspector_cache.pop(item, None)
        return result
    except BuildNotFound as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except (ActivityNotFound, InspectorViewNotFound) as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except Exception as exc:
        raise HTTPException(
            status_code=503,
            detail=f"读取 Inspector 数据来源失败：{type(exc).__name__}: {sanitize(str(exc))}",
        ) from exc


@app.get("/api/script-builds/{script_build_id}/rounds/{round_index}")
def get_round_detail(script_build_id: int, round_index: int):
    try:
        bundle = provider.load_bundle(script_build_id)
        view = builder.from_bundle(bundle)
    except BuildNotFound as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except Exception as exc:
        raise HTTPException(
            status_code=503,
            detail=f"读取轮次失败：{type(exc).__name__}: {sanitize(str(exc))}",
        ) from exc
    round_ = next(
        (
            item
            for item in view.get("rounds", [])
            if int(item.get("roundIndex") or 0) == round_index
        ),
        None,
    )
    if round_ is None:
        raise HTTPException(status_code=404, detail=f"未找到第 {round_index} 轮")
    return project_round_detail(script_build_id, round_, bundle)


@app.get("/api/script-builds/{script_build_id}/activities/{activity_id:path}")
def get_activity_detail(script_build_id: int, activity_id: str):
    if activity_id.startswith("event:"):
        try:
            bundle, view = _load_bundle_and_view(script_build_id)
            event_id = int(activity_id.rsplit(":", 1)[1])
            event_detail = provider.load_event_detail(script_build_id, event_id)
            try:
                return project_activity_detail(
                    script_build_id,
                    activity_id,
                    view,
                    bundle,
                    event_detail=event_detail,
                )
            except KeyError:
                pass
            return project_event_detail(
                script_build_id,
                event_detail,
                run_status=(view.get("header") or {}).get("status"),
            )
        except (ValueError, ActivityNotFound) as exc:
            raise HTTPException(status_code=404, detail=f"未找到活动 {activity_id}") from exc
        except Exception as exc:
            raise HTTPException(
                status_code=503,
                detail=f"读取运行事件失败：{type(exc).__name__}: {sanitize(str(exc))}",
            ) from exc
    try:
        bundle, view = _load_bundle_and_view(script_build_id)
        event_detail = None
        if activity_id.startswith("retrieval-agent:"):
            run = find_detail(view, activity_id)
            event_id = int((run or {}).get("eventId") or 0)
            if event_id:
                event_detail = provider.load_event_detail(script_build_id, event_id)
        return project_activity_detail(
            script_build_id,
            activity_id,
            view,
            bundle,
            event_detail=event_detail,
        )
    except BuildNotFound as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except KeyError as exc:
        raise HTTPException(status_code=404, detail=f"未找到活动 {activity_id}") from exc
    except Exception as exc:
        raise HTTPException(
            status_code=503,
            detail=f"读取活动详情失败：{type(exc).__name__}: {sanitize(str(exc))}",
        ) from exc


@app.get("/api/script-builds/{script_build_id}/decision-prompts/{prompt_ref:path}")
def get_decision_prompt(script_build_id: int, prompt_ref: str):
    try:
        return project_prompt(
            provider.load_prompt_context(script_build_id, prompt_ref)
        )
    except BuildNotFound as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except ActivityNotFound as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except Exception as exc:
        raise HTTPException(
            status_code=503,
            detail=f"读取 Agent 提示词失败：{type(exc).__name__}: {sanitize(str(exc))}",
        ) from exc


@app.get("/api/script-builds/{script_build_id}/artifacts/{snapshot_ref:path}")
def get_artifact(script_build_id: int, snapshot_ref: str):
    if snapshot_ref in {"base:current", "artifact:base:current"}:
        try:
            artifact = provider.load_current_artifact(script_build_id)
        except BuildNotFound as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc
        except Exception as exc:
            raise HTTPException(
                status_code=503,
                detail=f"读取创作表失败：{type(exc).__name__}: {sanitize(str(exc))}",
            ) from exc
        return project_artifact_detail(
            script_build_id,
            "artifact:base:current",
            {
                **artifact,
                "artifactContext": {
                    "type": "final",
                    "versionKind": "current-base",
                    "note": "当前保存的最终主脚本",
                },
            },
        )
    round_ref = snapshot_ref.removeprefix("artifact:").removeprefix("round:")
    if round_ref.isdigit() and "round:" in snapshot_ref:
        round_index = int(round_ref)
        try:
            bundle = provider.load_bundle(script_build_id)
        except BuildNotFound as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc
        except Exception as exc:
            raise HTTPException(
                status_code=503,
                detail=f"读取本轮产出脚本表失败：{type(exc).__name__}: {sanitize(str(exc))}",
            ) from exc
        if not any(int(item.get("round_index") or 0) == round_index for item in bundle.get("rounds", [])):
            raise HTTPException(status_code=404, detail=f"未找到第 {round_index} 轮")
        return project_artifact_detail(
            script_build_id,
            f"artifact:round:{round_index}",
            {
                **(bundle.get("currentArtifact") or {}),
                "artifactContext": {
                    "type": "round",
                    "roundIndex": round_index,
                    "versionKind": "current-base",
                    "note": f"当前数据库未保存第 {round_index} 轮结束时的独立快照；这里展示当前保存的主脚本。",
                },
            },
        )
    try:
        bundle = provider.load_bundle(script_build_id)
    except BuildNotFound as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except Exception as exc:
        raise HTTPException(
            status_code=503,
            detail=f"读取创作表失败：{type(exc).__name__}: {sanitize(str(exc))}",
        ) from exc
    match = snapshot_ref.removeprefix("artifact:").removeprefix("branch:")
    if match.isdigit():
        branch = next(
            (
                item
                for item in bundle.get("branches", [])
                if int(item.get("branch_id") or 0) == int(match)
            ),
            None,
        )
        if branch:
            if str(branch.get("path_type") or "").strip() == "领域信息":
                raise HTTPException(
                    status_code=404,
                    detail="领域信息方案不产出候选脚本表",
                )
            candidate = branch.get("candidate_snapshot") or {}
            if not isinstance(candidate, dict) or not isinstance(candidate.get("snapshot"), dict):
                raise HTTPException(
                    status_code=404,
                    detail="该实现方案没有可用的候选脚本快照",
                )
            return project_artifact_detail(
                script_build_id,
                f"artifact:branch:{match}",
                {
                    **candidate,
                    "artifactContext": {
                        "type": "candidate",
                        "roundIndex": branch.get("round_index"),
                        "branchId": branch.get("branch_id"),
                        "branchStatus": branch.get("status"),
                        "pathType": branch.get("path_type"),
                        "versionKind": candidate.get("snapshotKind"),
                        "historicalAccuracy": candidate.get("historicalAccuracy"),
                        "currentProjectionAccuracy": candidate.get("currentProjectionAccuracy"),
                        "note": candidate.get("note"),
                    },
                },
            )
    raise HTTPException(status_code=404, detail=f"未找到创作表快照 {snapshot_ref}")