"""从完整 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()