#!/usr/bin/env python3 from __future__ import annotations import datetime import hashlib import json import os import subprocess import sys import tempfile from pathlib import Path ROOT = Path(__file__).resolve().parents[1] PROJECT_MANIFEST = ROOT / "script/project-manifest.json" GENERATOR = Path("/home/hermes/.hermes/skills/content-creation/tao-anh/scripts/generate_image.py") AUTHORITY_KEYS = ["story_brief", "identity_registry", "reveal_ledger", "outline"] TARGET_KEYS = { "right_panel": "right_panel_master", "left_panel": "left_panel_master", "intro": "intro_master", } def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(8 * 1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def resolve_active(manifest: dict, key: str) -> Path: value = (manifest.get("active_paths") or {}).get(key) if not isinstance(value, str) or not value: raise RuntimeError(f"active_paths.{key} is missing") path = Path(value) if not path.is_absolute(): path = ROOT / path try: path.resolve().relative_to(ROOT.resolve()) except ValueError as exc: raise RuntimeError(f"active path escapes project: {key}") from exc return path def atomic_json(path: Path, value: dict) -> None: path.parent.mkdir(parents=True, exist_ok=True) data = (json.dumps(value, ensure_ascii=False, indent=2) + "\n").encode("utf-8") fd, name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) try: with os.fdopen(fd, "wb") as handle: handle.write(data) handle.flush() os.fsync(handle.fileno()) os.replace(name, path) finally: if os.path.exists(name): os.unlink(name) def png_valid(path: Path) -> bool: return path.exists() and path.stat().st_size > 1024 and path.read_bytes()[:8] == b"\x89PNG\r\n\x1a\n" def main() -> int: project = json.loads(PROJECT_MANIFEST.read_text(encoding="utf-8")) promotion_path = resolve_active(project, "promotion_receipt") prompts_path = resolve_active(project, "image_prompts") reconciliation_path = resolve_active(project, "image_prompt_reconciliation_receipt") receipt_path = resolve_active(project, "artwork_generation_receipt") targets = {name: resolve_active(project, key) for name, key in TARGET_KEYS.items()} authority = {key: resolve_active(project, key) for key in AUTHORITY_KEYS} promotion = json.loads(promotion_path.read_text(encoding="utf-8")) prompts = json.loads(prompts_path.read_text(encoding="utf-8")) reconciliation = json.loads(reconciliation_path.read_text(encoding="utf-8")) if promotion.get("status") != "completed" or promotion.get("verified") is not True: raise RuntimeError("promotion is absent or not verified") canon_hash = promotion.get("canon_sha256") current = {key: sha256(path) for key, path in authority.items()} if prompts.get("authority_hashes") != current: raise RuntimeError("image prompt authority drift detected") if prompts.get("source_candidate_sha256") != promotion.get("candidate_sha256") or prompts.get("source_canon_sha256") != canon_hash: raise RuntimeError("image prompts are stale for current promotion") if reconciliation.get("verified") is not True or reconciliation.get("canon_sha256") != canon_hash: raise RuntimeError("image prompt reconciliation is absent or stale") if reconciliation.get("prompt_manifest_sha256") != sha256(prompts_path): raise RuntimeError("image prompt manifest changed after reconciliation") if prompts.get("image_api_called") is not False: raise RuntimeError("image prompt manifest already marked as called") if not GENERATOR.is_file(): raise RuntimeError("configured image generator is missing") existing = [str(path) for path in [receipt_path, *targets.values()] if path.exists()] if existing: raise RuntimeError(f"artwork outputs already exist and require explicit audit/invalidation: {existing}") generated: list[dict] = [] for key in ("right_panel", "left_panel", "intro"): output = targets[key] output.parent.mkdir(parents=True, exist_ok=True) command = [ sys.executable, str(GENERATOR), "--prompt", prompts["prompts"][key], "--output", str(output), "--n", "1", "--detail", "high", "--format", "png", "--timeout", "600", ] completed = subprocess.run(command, text=True, capture_output=True, timeout=660) if completed.returncode != 0 or not png_valid(output): raise RuntimeError( f"{key} generation failed: exit={completed.returncode}; stderr={completed.stderr[-1000:]}" ) generated.append({ "kind": key, "path": str(output), "sha256": sha256(output), "bytes": output.stat().st_size, "png_signature_verified": True, }) atomic_json(receipt_path, { "status": "running", "verified": False, "source_canon_sha256": canon_hash, "authority_hashes": current, "prompt_manifest": str(prompts_path), "prompt_manifest_sha256": sha256(prompts_path), "generated": generated, "visual_review_required": True, "orientation_gate_required": True, "updated_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), }) prompts["status"] = "generated_pending_visual_and_orientation_review" prompts["image_api_called"] = True prompts["generated"] = generated prompts["generated_at"] = datetime.datetime.now(datetime.timezone.utc).isoformat() atomic_json(prompts_path, prompts) receipt = { "status": "completed_pending_visual_and_orientation_review", "verified": False, "source_canon_sha256": canon_hash, "authority_hashes": current, "prompt_manifest": str(prompts_path), "prompt_manifest_sha256": sha256(prompts_path), "generated": generated, "visual_review_required": True, "orientation_gate_required": True, "completed_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), "hash_method": "sha256 streaming 8 MiB", } atomic_json(receipt_path, receipt) print(json.dumps({ "status": receipt["status"], "verified": False, "generated_count": len(generated), "receipt": str(receipt_path), }, ensure_ascii=False)) return 0 if __name__ == "__main__": try: sys.exit(main()) except Exception as exc: failure = { "status": "failed", "verified": False, "error": f"{type(exc).__name__}: {exc}", "failed_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), } try: project = json.loads(PROJECT_MANIFEST.read_text(encoding="utf-8")) atomic_json(resolve_active(project, "artwork_generation_receipt"), failure) except Exception: pass print(json.dumps(failure, ensure_ascii=False), file=sys.stderr) sys.exit(1)