#!/usr/bin/env python3 """Attach a legacy zero-context take, then output only its mixed-size chapter.""" import copy import importlib.util import json from pathlib import Path import tempfile from unittest.mock import patch ROOT = Path(__file__).resolve().parents[1] spec = importlib.util.spec_from_file_location( "attributed_chapter_helpers", ROOT / "tests/_checkpoint_local_output_unit_test.py") h = importlib.util.module_from_spec(spec) spec.loader.exec_module(h) chain = h.chain def main(): with tempfile.TemporaryDirectory() as directory: root = Path(directory) h.h.folder_paths.output_directory = directory run = root / "h3_chains" / "attributed_chapter" metadata, lineage = [], [] for scene in range(1, 10): compatibility = { "width":1344 if scene < 8 else 960, "height":768 if scene < 8 else 544, "fps":24, "audio_mode":"generated_audio", "context_length":39, "audio_context_length":39, "continuation_mode":"audio_feathered_av", "generation_fingerprint":f"catalog-{scene}", } item, _ = h.h.write_revision( run, scene, str(scene) * 32, scene, active=True, predecessor=metadata[-1] if metadata else None, run_name=run.name, compatibility=compatibility, context_length=5 if scene == 9 else 0, audio_context_length=0) metadata.append(item) lineage.append({"scene":scene, "revision":str(scene) * 32}) previous, _ = h.h.write_revision( run, 9, "a" * 32, 90, run_name=run.name, predecessor=metadata[7], compatibility=metadata[8]["compatibility"], context_length=0, audio_context_length=0) candidate, _ = h.h.write_revision( run, 10, "b" * 32, 100, run_name=run.name, predecessor=previous, compatibility={**metadata[8]["compatibility"], "generation_fingerprint":"older-catalog"}, context_length=0, audio_context_length=39, generated_continuity="on", recovery_archive=True) # Legacy AV saves retain a 39f audio default although the zero-video # early return consumed neither predecessor video nor audio. candidate["segment"].update({ "resolved_context_length":0, "resolved_audio_context_length":39, "continuation_mode":"audio_feathered_av", "raw_frames":141, "delivered_frames":141}) (root / candidate["segment"]["revision_metadata"]).write_text(json.dumps(candidate)) (run / "plan.json").write_text(json.dumps({ "run_name":run.name, "shots":[{"id":f"scene_{i}"} for i in range(1, 13)]})) # Modern saves share their immutable recovery snapshot when attached. # The alias must resolve this original snapshot, not demand a copy # under its new lineage id or silently use today's Run-level Plan. archived_plan = root / candidate["archives"]["plan"] archived_plan.write_text((run / "plan.json").read_text()) (run / "plan.json").write_text('{"shots":[]}') (run / "editorial.json").write_text(json.dumps({ "scene_order":[{"scene":i, "scene_id":f"scene_{i}"} for i in range(1, 13)], "chapters":[{"id":"one", "start_scene_id":"scene_1"}, {"id":"two", "start_scene_id":"scene_8"}], })) manager = chain.CheckpointGraphManager(directory) graph = manager.graph(run.name) before = h.snapshot(run) parent_branch = next(b for b in graph["branches"] if b["leaf_revision"] == "9" * 32) assert {"scene":10, "revision":"b" * 32} in parent_branch["attribution_slot"]["candidates"] attached = manager.attribute(run.name, 9, "9" * 32, 10, "b" * 32) assert attached["created"] after = h.snapshot(run) assert len(after) == len(before) + 1, sorted(set(after) - set(before)) assert all(after[path] == value for path, value in before.items()), \ "attachment must not rewrite media, original metadata or active pointers" alias = json.loads((run / "checkpoints" / f"clip_0010.{attached['revision']}.json").read_text()) assert alias["segment"]["segment"] == candidate["segment"]["segment"] assert alias["segment"]["checkpoint"] == candidate["segment"]["checkpoint"] selection = {"run_name":run.name, "output_mode":"workflow_local", "output_scope":"chapter", "scope_start_scene":8, "scope_end_scene":12, "lineage":[*lineage, {"scene":10, "revision":attached["revision"]}]} # Calling the real node, not only the graph's candidate listing, is the # regression: chapter 1 must not participate in the geometry check. with patch.object(chain, "_atomic_json", side_effect=AssertionError("output wrote project files")): result = chain.MiniMaxH3ChainCheckpointManager().passthrough(json.dumps(selection))[0] assert [s["index"] for s in result["segments"]] == [8, 9, 10] assert result["segments"][-1]["revision"] == attached["revision"] assert result["segments"][-1]["delivered_frames"] == 141 assert result["compatibility"]["width"] == 960 assert result["archives"] == candidate["archives"] assert h.snapshot(run) == after # Ordinary saves still cannot claim another take's archive. An alias # can share only the origin explicitly recorded by attribution. def archives(value): return chain._checkpoint_run_archives({"run_name":run.name}, value) invalid = copy.deepcopy(alias) invalid.pop("adoption") h.rejected(lambda: archives(invalid), "different revision") invalid = copy.deepcopy(alias) invalid["adoption"]["source_revision"] = "c" * 32 h.rejected(lambda: archives(invalid), "origin is inconsistent") invalid = copy.deepcopy(alias) invalid["adoption"]["source_scene"] = 9 h.rejected(lambda: archives(invalid), "origin is inconsistent") invalid = copy.deepcopy(alias) invalid["archives"]["workflow"] = invalid["archives"]["workflow"].replace("b" * 32, "c" * 32) h.rejected(lambda: archives(invalid), "outside revision") h.rejected(lambda: chain._checkpoint_run_archives({"run_name":"another_run"}, alias), "outside revision") workflow_archive = root / alias["archives"]["workflow"] workflow_bytes = workflow_archive.read_bytes() workflow_archive.unlink() h.rejected(lambda: archives(alias), "workflow is missing") workflow_archive.write_bytes(workflow_bytes) # Deleting an inactive original sidecar must not strand surviving # aliases: they already carry the origin and share its artifacts. candidate_path = root / candidate["segment"]["revision_metadata"] candidate_bytes = candidate_path.read_bytes() candidate_path.unlink() assert archives(alias) == candidate["archives"] candidate_path.write_bytes(candidate_bytes) assert h.snapshot(run) == after h.rejected(lambda: chain._checkpoint_selection_manifest({ **selection, "output_scope":"project"}), "compatibility") # The unattached original and a genuinely different canvas remain # invalid; fixing scope must not remove lineage or geometry checks. h.rejected(lambda: chain._checkpoint_selection_manifest({ **selection, "lineage":[*lineage, {"scene":10, "revision":"b" * 32}]}), "depends on") alias_path = run / "checkpoints" / f"clip_0010.{attached['revision']}.json" alias["compatibility"]["width"] = 1024 alias_path.write_text(json.dumps(alias)) h.rejected(lambda: chain._checkpoint_selection_manifest(selection), "compatible takes within this chapter") print("Attributed chapter output: 960x544 context-free candidate after 5f take, mixed earlier chapter, immutable media and compatibility guards pass") if __name__ == "__main__": main()