#!/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") 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 project_path(value: object, label: str) -> Path: if not isinstance(value, str) or not value: raise RuntimeError(f"missing project path: {label}") 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"project path escapes root: {label}") 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 valid_png(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")) revision = project.get("artwork_revision") or {} version = revision.get("version") if revision.get("status") != f"intro_{version}_prepared" or version not in {"v2", "v3", "v4", "v5", "v6", "v7"}: raise RuntimeError("active intro revision is absent or unsupported") promotion_path = resolve_active(project, "promotion_receipt") prompts_path = resolve_active(project, "image_prompts") output = resolve_active(project, "intro_master") receipt = resolve_active(project, "artwork_generation_receipt") left = resolve_active(project, "left_panel_master") right = resolve_active(project, "right_panel_master") prior_visual_path = project_path(revision.get("prior_visual_receipt"), "prior_visual_receipt") prior_orientation_path = project_path(revision.get("prior_orientation_receipt"), "prior_orientation_receipt") prior_intro = project_path(revision.get("prior_intro_preserved"), "prior_intro_preserved") promotion = json.loads(promotion_path.read_text(encoding="utf-8")) prompts = json.loads(prompts_path.read_text(encoding="utf-8")) prior_visual = json.loads(prior_visual_path.read_text(encoding="utf-8")) prior_orientation = json.loads(prior_orientation_path.read_text(encoding="utf-8")) canon_hash = promotion.get("canon_sha256") if promotion.get("status") != "completed" or promotion.get("verified") is not True: raise RuntimeError("promotion is not verified") prompt_revision = prompts.get("revision") or {} if prompts.get("source_canon_sha256") != canon_hash or prompts.get("source_candidate_sha256") != promotion.get("candidate_sha256"): raise RuntimeError("revision prompt is stale") if prompts.get("status") != f"revision_intro_{version}_prepared" or prompt_revision.get("asset") != "intro" or prompt_revision.get("version") != version: raise RuntimeError("prompt revision contract does not match active revision") if prior_visual.get("status") != "failed" or prior_visual.get("verified") is not False or not prior_visual.get("blockers"): raise RuntimeError("prior visual failure does not authorize revision") if prior_orientation.get("status") != "completed" or prior_orientation.get("verified") is not True: raise RuntimeError("prior orientation evidence is absent") expected_panels = revision.get("accepted_panel_hashes") or {} current_panels = {"left_panel": sha256(left), "right_panel": sha256(right)} if expected_panels != current_panels: raise RuntimeError("accepted panel drift detected") prior_hashes = prior_visual.get("artifact_sha256") or {} if prior_hashes.get("left_panel") != current_panels["left_panel"] or prior_hashes.get("right_panel") != current_panels["right_panel"]: raise RuntimeError("prior visual receipt does not match accepted panels") if prior_hashes.get("intro") != sha256(prior_intro) or prior_orientation.get("artifact_sha256") != prior_hashes: raise RuntimeError("prior intro or orientation evidence drift detected") if output.exists() or receipt.exists(): raise RuntimeError("revision output/receipt already exists and requires audit") if not GENERATOR.is_file(): raise RuntimeError("configured image generator is missing") output.parent.mkdir(parents=True, exist_ok=True) command = [ sys.executable, str(GENERATOR), "--prompt", prompts["prompts"]["intro"], "--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 valid_png(output): raise RuntimeError(f"intro {version} generation failed: exit={completed.returncode}; stderr={completed.stderr[-1000:]}") result = { "status": "completed_pending_visual_and_orientation_review", "verified": False, "revision_asset": "intro", "revision_version": version, "source_canon_sha256": canon_hash, "prompt_manifest": str(prompts_path), "prompt_manifest_sha256": sha256(prompts_path), "preserved_assets": { "left_panel": {"path": str(left), "sha256": sha256(left)}, "right_panel": {"path": str(right), "sha256": sha256(right)}, "prior_intro": {"path": str(prior_intro), "sha256": sha256(prior_intro)}, }, "prior_evidence": { "visual": {"path": str(prior_visual_path), "sha256": sha256(prior_visual_path)}, "orientation": {"path": str(prior_orientation_path), "sha256": sha256(prior_orientation_path)}, }, "generated": [{"kind": "intro", "path": str(output), "sha256": sha256(output), "bytes": output.stat().st_size, "png_signature_verified": True}], "visual_review_required": True, "orientation_gate_required": True, "completed_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), } atomic_json(receipt, result) prompts["status"] = f"revision_intro_{version}_generated_pending_review" prompts["revision"]["output_path"] = str(output) prompts["revision"]["output_sha256"] = sha256(output) prompts["revision"]["generated_at"] = result["completed_at"] atomic_json(prompts_path, prompts) print(json.dumps({"status": result["status"], "version": version, "generated_count": 1, "path": str(output), "sha256": sha256(output), "receipt": str(receipt)}, 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)