from __future__ import annotations import hashlib import json from pathlib import Path import yaml from orchestration_harness_v1 import run_plan from pipeline_runtime_anomaly_lib_v1 import finalize_runtime_profile, runtime_profile_from_steps ROOT = Path(__file__).resolve().parents[1] DIST = ROOT / "dist" RUNTIME_PROFILE = ROOT / "Temp" / "build_bundle_runtime_profile_v1.json" CACHE_PATH = ROOT / "Temp" / "build_bundle_cache_v1.json" def read(path: str) -> str: return (ROOT / path).read_text(encoding="utf-8").rstrip() def load_bundle_contents(paths: list[str]) -> dict[str, str]: contents: dict[str, str] = {} for path in paths: contents[path] = read(path) return contents def compute_content_signature(contents: dict[str, str]) -> str: digest = hashlib.sha256() for path in sorted(contents): digest.update(path.encode("utf-8")) digest.update(b"\0") digest.update(contents[path].encode("utf-8")) digest.update(b"\0") return digest.hexdigest() def flatten_manifest_files(manifest: dict) -> list[str]: ordered: list[str] = ["spec/RetirementAssetPortfolio.yaml", "AGENTS.md"] for step in (manifest.get("load_sequence") or {}).values(): for file_name in step.get("files", []): if "*" not in file_name and file_name not in ordered: ordered.append(file_name) for extra in ( "spec/ownership_map.yaml", "spec/aliases.yaml", "spec/xref_matrix.yaml", "prompts/analysis_prompt.md", "prompts/review_prompt.md", ): if extra not in ordered and (ROOT / extra).exists(): ordered.append(extra) return ordered def bundle_profile(manifest: dict, mode: str) -> dict: if mode == "full": return { "output": "dist/retirement_portfolio_bundle.yaml", "purpose": "분리된 문서를 manifest 순서로 묶은 전체 LLM 입력용 합본", "files": flatten_manifest_files(manifest), } profiles = manifest.get("bundle_profiles") or {} profile = profiles.get(mode) if not isinstance(profile, dict): raise ValueError(f"missing bundle profile: {mode}") return profile def write_bundle( paths: list[str], output: Path, mode: str = "full", purpose_text: str | None = None, *, contents: dict[str, str] | None = None, ) -> None: title_suffix = { "full": "Full", "compact": "Compact", "ultra_compact": "Ultra Compact", }[mode] purpose = purpose_text or "분리된 문서를 manifest 순서로 묶은 LLM 입력용 합본" chunks = [ "meta:", f" title: \"은퇴자산포트폴리오 {title_suffix} Bundle\"", " generated_by: \"tools/build_bundle.py\"", f" purpose: \"{purpose}\"", f" mode: \"{mode}\"", f" file_count: {len(paths)}", "", "bundle_files:", ] for path in paths: chunks.append(f" - \"{path}\"") chunks.append("") chunks.append("bundle_content:") for path in paths: text = contents[path] if contents and path in contents else read(path) indented = "\n".join(" " + line for line in text.splitlines()) key = path.replace("\\", "/").replace("/", "__").replace(".", "_") chunks.append(f" {key}: |") chunks.append(indented if indented else " ") output.write_text("\n".join(chunks).rstrip() + "\n", encoding="utf-8") def _write_bundle_job(mode: str, profile: dict, contents: dict[str, str]) -> dict[str, str]: output = ROOT / profile["output"] paths = [path for path in profile["files"] if "*" not in path] write_bundle(paths, output, mode=mode, purpose_text=profile.get("purpose"), contents=contents) return {"mode": mode, "output": str(output), "status": "OK"} def _validate_bundle(path: Path) -> dict[str, str]: yaml.safe_load(path.read_text(encoding="utf-8")) return {"path": str(path), "status": "OK"} def load_cache() -> dict[str, str]: if not CACHE_PATH.exists(): return {} try: data = json.loads(CACHE_PATH.read_text(encoding="utf-8")) except Exception: return {} return data if isinstance(data, dict) else {} def save_cache(payload: dict[str, str]) -> None: CACHE_PATH.parent.mkdir(parents=True, exist_ok=True) CACHE_PATH.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") def main() -> int: DIST.mkdir(exist_ok=True) # 1) 공통 입력 로드는 순차로 고정한다. manifest = yaml.safe_load((ROOT / "spec" / "RetirementAssetPortfolio.yaml").read_text(encoding="utf-8")) profiles = {mode: bundle_profile(manifest, mode) for mode in ("full", "compact", "ultra_compact")} all_profile_paths = [path for profile in profiles.values() for path in profile.get("files", [])] all_unique_paths = sorted({path for path in all_profile_paths if "*" not in path}) missing = [path for path in all_profile_paths if "*" not in path and not (ROOT / path).exists()] if missing: print("BUNDLE BUILD FAIL") for path in missing: print(f"- missing: {path}") return 1 shared_contents = load_bundle_contents(all_unique_paths) content_signature = compute_content_signature(shared_contents) latest_source_mtime = max((ROOT / path).stat().st_mtime for path in all_unique_paths) if all_unique_paths else 0.0 cache = load_cache() output_paths = [ROOT / profiles[mode]["output"] for mode in ("full", "compact", "ultra_compact")] cache_hit = bool(cache) and all(path.exists() for path in output_paths) and min(path.stat().st_mtime for path in output_paths) >= latest_source_mtime bundle_paths = [ ROOT / profiles["full"]["output"], ROOT / profiles["compact"]["output"], ROOT / profiles["ultra_compact"]["output"], ] if cache_hit: # 소스와 산출물 해시가 같으면 기존 번들을 재사용한다. orchestration_steps = [] for mode, profile in profiles.items(): orchestration_steps.append({ "name": f"build_{mode}", "callable": (lambda m=mode, p=profile: {"mode": m, "output": str(ROOT / p["output"]), "status": "CACHED"}), }) for mode, path in zip(("full", "compact", "ultra_compact"), bundle_paths): orchestration_steps.append({ "name": f"validate_{path.stem}", "depends_on": [f"build_{mode}"], "callable": (lambda p=path: {"path": str(p), "status": "OK_CACHED"}), }) results = run_plan(orchestration_steps, label="build-bundle:cached") else: # 2) 서로 독립인 3개 번들 생성은 병렬, 각 생성물 검증은 해당 생성 직후 수행한다. orchestration_steps = [] for mode, profile in profiles.items(): orchestration_steps.append({ "name": f"build_{mode}", "callable": (lambda m=mode, p=profile, c=shared_contents: _write_bundle_job(m, p, c)), }) for mode, path in zip(("full", "compact", "ultra_compact"), bundle_paths): orchestration_steps.append({ "name": f"validate_{path.stem}", "depends_on": [f"build_{mode}"], "callable": (lambda p=path: _validate_bundle(p)), }) results = run_plan(orchestration_steps, label="build-bundle") save_cache({ "content_signature": content_signature, "latest_source_mtime": latest_source_mtime, "bundle_outputs": [str(p) for p in bundle_paths], }) profile = runtime_profile_from_steps( harness_name="build-bundle", mode="bundle", steps=results, runtime_context={ "harness_name": "build-bundle", "mode": "bundle", }, file_count=len(all_profile_paths), gate_status="OK", ) profile["cache_hit"] = cache_hit analysis = finalize_runtime_profile( profile_path=RUNTIME_PROFILE, payload=profile, ) if analysis.get("status") == "ALERT": print("RUNTIME_ANOMALY:", ";".join(analysis.get("anomaly_reason_codes") or [])) print("BUNDLE BUILD OK" + (" (cached)" if cache_hit else "")) return 0 if __name__ == "__main__": raise SystemExit(main())