video-image-production-build/run_pipeline.py

"""顺序运行或恢复 Global Data 与 Production 两个正式 Run。"""

from __future__ import annotations

import argparse
import json
import re
from pathlib import Path
from typing import Any, Mapping

from dotenv import load_dotenv

from production_build_agents.run.layout import (
    global_data_delivery_path,
    run_directory,
)
from production_build_agents.observability.pipeline import PipelineObservation
from run_global_data import run_global_data
from run_production import run_full_production


PROJECT_ROOT = Path(__file__).resolve().parent
_ROUND_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")


def _validate_round_id(round_id: str) -> str:
    value = round_id.strip()
    if not _ROUND_ID.fullmatch(value):
        raise ValueError(
            "round_id 只能包含字母、数字、点、下划线和连字符，"
            "且必须以字母或数字开头"
        )
    return value


def _pipeline_result(
    *,
    round_id: str,
    global_data_thread_id: str,
    production_thread_id: str,
    global_data_result: Mapping[str, Any],
    production_result: Mapping[str, Any] | None,
) -> dict[str, Any]:
    active = (
        production_result
        if production_result is not None
        else global_data_result
    )
    return {
        "round_id": round_id,
        "status": active.get("status"),
        "phase": (
            "PRODUCTION"
            if production_result is not None
            else "GLOBAL_DATA"
        ),
        "global_data_thread_id": global_data_thread_id,
        "production_thread_id": production_thread_id,
        "global_data_result": dict(global_data_result),
        "production_result": (
            dict(production_result)
            if production_result is not None
            else None
        ),
    }


def run_pipeline(
    input_path: Path,
    *,
    output_root: Path,
    round_id: str,
    shared_visual_anchor_path: Path,
    max_plan_revisions: int = 5,
    pause_after_progress_review: bool = False,
) -> dict[str, Any]:
    """顺序运行或恢复两个阶段，并只在 Global Data 完成后进入 Production。

    编排层没有第三套 checkpoint。相同 ``round_id`` 会稳定映射到两个独立
    thread_id；重复调用时，两个正式 runtime 分别执行自己的恢复与终态重验。
    """

    load_dotenv(PROJECT_ROOT / ".env", override=False)
    active_round_id = _validate_round_id(round_id)
    output_root = output_root.resolve()
    global_data_thread_id = f"{active_round_id}-global-data"
    production_thread_id = f"{active_round_id}-production"

    with PipelineObservation(
        project_root=PROJECT_ROOT,
        round_id=active_round_id,
        input_path=input_path,
        output_root=output_root,
        shared_visual_anchor_path=shared_visual_anchor_path,
    ) as pipeline_observation:
        with pipeline_observation.stage(
            "global_data",
            input_value={
                "thread_id": global_data_thread_id,
                "input_path": str(input_path.resolve()),
            },
        ) as global_data_stage:
            global_result = run_global_data(
                input_path,
                output_root=output_root,
                thread_id=global_data_thread_id,
                observation_parent_run_id=(
                    global_data_stage.parent_run_id
                ),
                observation_parent_inst_id=(
                    global_data_stage.parent_inst_id
                ),
                pipeline_round_id=active_round_id,
            )
            global_data_stage.finish(global_result)
        if global_result.get("status") != "COMPLETED":
            pipeline_result = _pipeline_result(
                round_id=active_round_id,
                global_data_thread_id=global_data_thread_id,
                production_thread_id=production_thread_id,
                global_data_result=global_result,
                production_result=None,
            )
            pipeline_observation.finish(pipeline_result)
            return pipeline_result

        if global_result.get("run_id") != global_data_thread_id:
            raise ValueError(
                "Global Data 完成结果 run_id 与 Pipeline 身份不一致"
            )
        expected_delivery = global_data_delivery_path(
            run_directory(output_root, global_data_thread_id)
        ).resolve()
        raw_delivery_path = global_result.get("global_data_delivery_path")
        if not isinstance(raw_delivery_path, str) or not raw_delivery_path:
            raise ValueError("Global Data 完成结果缺少正式 Delivery 路径")
        actual_delivery = Path(raw_delivery_path).resolve()
        if actual_delivery != expected_delivery:
            raise ValueError(
                "Global Data 完成结果指向非当前 Run 的 Delivery："
                f"{actual_delivery}"
            )
        if not actual_delivery.is_file():
            raise FileNotFoundError(
                f"Global Data 正式 Delivery 不存在：{actual_delivery}"
            )

        with pipeline_observation.stage(
            "production",
            input_value={
                "thread_id": production_thread_id,
                "global_data_thread_id": global_data_thread_id,
                "global_data_delivery_path": str(actual_delivery),
            },
        ) as production_stage:
            production_result = run_full_production(
                actual_delivery,
                output_root=output_root,
                thread_id=production_thread_id,
                shared_visual_anchor_path=shared_visual_anchor_path,
                max_plan_revisions=max_plan_revisions,
                pause_after_progress_review=pause_after_progress_review,
                observation_parent_run_id=(
                    production_stage.parent_run_id
                ),
                observation_parent_inst_id=(
                    production_stage.parent_inst_id
                ),
                pipeline_round_id=active_round_id,
                upstream_global_data_thread_id=global_data_thread_id,
            )
            production_stage.finish(production_result)
        pipeline_result = _pipeline_result(
            round_id=active_round_id,
            global_data_thread_id=global_data_thread_id,
            production_thread_id=production_thread_id,
            global_data_result=global_result,
            production_result=production_result,
        )
        pipeline_observation.finish(pipeline_result)
        return pipeline_result


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description=(
            "顺序运行或恢复 Global Data → Production；两个阶段保留独立 Run"
        )
    )
    parser.add_argument("input", type=Path, help="原始 production_final.json")
    parser.add_argument(
        "--output-dir",
        type=Path,
        default=Path("demo_output"),
    )
    parser.add_argument(
        "--round-id",
        required=True,
        help="本轮稳定身份；派生出 <round>-global-data 和 <round>-production",
    )
    parser.add_argument(
        "--shared-visual-anchor",
        type=Path,
        required=True,
        help="Production 外部已接纳的共享视觉锚点图片",
    )
    parser.add_argument(
        "--max-plan-revisions",
        type=int,
        default=5,
        choices=range(0, 6),
        metavar="{0..5}",
        help="Production 允许的 ADAPT/REPLAN 总次数",
    )
    parser.add_argument(
        "--pause-after-progress-review",
        action="store_true",
        help="Production 在下一次 Progress Review 后以 RUNNING 状态暂停",
    )
    return parser


def main() -> None:
    args = _build_parser().parse_args()
    result = run_pipeline(
        args.input,
        output_root=args.output_dir,
        round_id=args.round_id,
        shared_visual_anchor_path=args.shared_visual_anchor,
        max_plan_revisions=args.max_plan_revisions,
        pause_after_progress_review=args.pause_after_progress_review,
    )
    print(json.dumps(result, ensure_ascii=False, indent=2))
    if result.get("status") == "COMPLETED":
        return
    if (
        result.get("phase") == "PRODUCTION"
        and args.pause_after_progress_review
        and result.get("status") == "RUNNING"
    ):
        return
    raise SystemExit(1)


if __name__ == "__main__":
    main()