video-image-production-build/run_segment.py

"""从完整 SegmentPackage 运行或恢复一个正式 Segment。"""

from __future__ import annotations

import argparse
import hashlib
import json
from pathlib import Path

from dotenv import load_dotenv
from langchain_core.language_models import BaseChatModel

from production_build_agents.agents.segment.run import run_segment_production
from production_build_agents.contracts.production_execution_models import (
    SegmentPackage,
)
from production_build_agents.contracts.production_models import (
    SegmentValidationReport,
)
from production_build_agents.production.runtime import (
    PRODUCTION_PROTOCOL_VERSION,
    validate_production_run_directory_boundary,
)
from production_build_agents.run.layout import run_directory
from production_build_agents.run.records import VersionConflictError
from production_build_agents.run.registry import register_thread
from production_build_agents.tools.registry import ToolRegistry


PROJECT_ROOT = Path(__file__).resolve().parent


def segment_request_sha256(package_bytes: bytes) -> str:
    """以完整 Package 的精确字节作为 standalone 请求身份。"""

    return hashlib.sha256(package_bytes).hexdigest()


def run_production_segment(
    package_path: Path,
    *,
    output_root: Path,
    thread_id: str | None = None,
    executor_model: BaseChatModel | None = None,
    validator_model: BaseChatModel | None = None,
    tool_registry: ToolRegistry | None = None,
    registry_dir: Path | None = None,
) -> SegmentValidationReport:
    """校验完整 Package 并复用 Graph 的 Executor/Validator 阶段服务。"""

    load_dotenv(PROJECT_ROOT / ".env", override=False)
    source = package_path.resolve()
    if not source.is_file():
        raise FileNotFoundError(f"SegmentPackage 不存在：{source}")
    source_bytes = source.read_bytes()
    package = SegmentPackage.model_validate_json(source_bytes)
    request_sha256 = segment_request_sha256(source_bytes)
    active_thread_id = thread_id or package.run_id
    if active_thread_id != package.run_id:
        raise ValueError(
            "standalone thread_id 必须与 SegmentPackage.run_id 一致"
        )
    run_dir = run_directory(
        output_root.resolve(),
        active_thread_id,
    ).resolve()
    validate_production_run_directory_boundary(run_dir)

    def prepare_locked_run(locked_run_dir: Path) -> None:
        validate_production_run_directory_boundary(locked_run_dir)
        if source.read_bytes() != source_bytes:
            raise VersionConflictError(
                "SegmentPackage 输入文件在封印前已变化"
            )
        register_thread(
            registry_dir=registry_dir or PROJECT_ROOT / ".run_registry",
            thread_id=active_thread_id,
            protocol_version=PRODUCTION_PROTOCOL_VERSION,
            input_path=source,
            input_sha256=request_sha256,
            run_dir=locked_run_dir,
        )

    return run_segment_production(
        package,
        run_dir=run_dir,
        executor_model=executor_model,
        validator_model=validator_model,
        tool_registry=tool_registry,
        locked_preflight=prepare_locked_run,
    )


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="运行一个正式 SegmentPackage，直到终态或明确失败"
    )
    parser.add_argument("package", type=Path, help="完整 SegmentPackage JSON")
    parser.add_argument(
        "--output-dir",
        type=Path,
        default=Path("demo_output"),
    )
    parser.add_argument("--thread-id", default=None)
    return parser


def main() -> None:
    args = _build_parser().parse_args()
    report = run_production_segment(
        args.package,
        output_root=args.output_dir,
        thread_id=args.thread_id,
    )
    print(
        json.dumps(
            {
                "run_id": report.run_id,
                "segment_id": report.segment_id,
                "verdict": report.verdict,
            },
            ensure_ascii=False,
            indent=2,
        )
    )
    if report.verdict != "PASS":
        raise SystemExit(1)


if __name__ == "__main__":
    main()