Fast_deploy / ComfyUI-MiniMaxH3-Context-Loop /tests /_checkpoint_activation_unit_test.py

Upload folder using huggingface_hub (part 2)

bc50ea0 verified 30 days ago

12.6 kB

#!/usr/bin/env python3
"""Focused mixed-policy checkpoint branch activation regression."""

import asyncio
import importlib.util
import json
import pathlib
import tempfile


ROOT = pathlib.Path(__file__).resolve().parents[1]
HELPERS_PATH = ROOT / "tests" / "_checkpoint_revision_unit_test.py"
spec = importlib.util.spec_from_file_location(
    "h3_checkpoint_activation_helpers", HELPERS_PATH)
helpers = importlib.util.module_from_spec(spec)
spec.loader.exec_module(helpers)


async def check():
    with tempfile.TemporaryDirectory() as temporary:
        helpers.folder_paths.output_directory = temporary
        run = pathlib.Path(temporary) / "h3_chains" / "mixed_policy"
        first = "a" * 32
        second = "b" * 32
        first_meta, _ = helpers.write_revision(
            run, 1, first, 1001, active=True, run_name="mixed_policy")
        helpers.write_revision(
            run, 2, second, 1002, predecessor=first_meta,
            run_name="mixed_policy", compatibility={
                "context_length": 39,
                "audio_context_length": 39,
                "audio_mode": "generated_audio",
            })
        selection = {
            "run_name": "mixed_policy",
            "resume_scene": 3,
            "revisions": [
                {"scene": 1, "revision": first},
                {"scene": 2, "revision": second},
            ],
        }

        strict = await helpers.chain._restore_checkpoint_revisions(
            helpers.JsonRequest(selection))
        assert strict.status == 400
        assert "different Plan compatibility" in json.loads(
            strict.text)["error"]
        assert not (run / "checkpoints" / "clip_0002.json").exists()

        promoted = await helpers.chain._restore_checkpoint_revisions(
            helpers.JsonRequest({**selection, "activate_only": True}))
        assert promoted.status == 200
        assert json.loads(promoted.text)["activate_only"] is True
        assert json.loads((
            run / "checkpoints" / "clip_0002.json").read_text())[
                "segment"]["revision"] == second
        graph = helpers.chain.CheckpointGraphManager(
            temporary).graph("mixed_policy")
        assert [item["revision"] for item in graph["revisions"]
                if item["active"]] == [first, second]
        assert graph["branches"][0]["active"] is True

        # One builder scene can depend on several saved picture revisions,
        # including repeated windows from the same source. Checkpoint Manager
        # must protect every distinct source, not only the nearest block.
        builder_run = pathlib.Path(temporary) / "h3_chains" / "builder_graph"
        builder_one = "8" * 32
        builder_two = "9" * 32
        builder_three = "c" * 32
        builder_four = "d" * 32
        builder_one_meta, _ = helpers.write_revision(
            builder_run, 1, builder_one, 3001, active=True,
            run_name="builder_graph")
        builder_two_meta, _ = helpers.write_revision(
            builder_run, 2, builder_two, 3002, active=True,
            predecessor=builder_one_meta, run_name="builder_graph")
        builder_three_meta, _ = helpers.write_revision(
            builder_run, 3, builder_three, 3003, active=True,
            predecessor=builder_two_meta, run_name="builder_graph")
        builder_four_meta, _ = helpers.write_revision(
            builder_run, 4, builder_four, 3004, active=True,
            predecessor=builder_three_meta, run_name="builder_graph",
            generated_continuity="off")
        builder_four_meta["segment"]["visual_context_blocks"] = [
            {
                "source_scene": 1, "source_revision": builder_one,
                "source_checkpoint_sha256": builder_one_meta["segment"][
                    "checkpoint_sha256"], "frames": 5,
            },
            {
                "source_scene": 2, "source_revision": builder_two,
                "source_checkpoint_sha256": builder_two_meta["segment"][
                    "checkpoint_sha256"], "frames": 12,
            },
            {
                "source_scene": 1, "source_revision": builder_one,
                "source_checkpoint_sha256": builder_one_meta["segment"][
                    "checkpoint_sha256"], "frames": 22,
            },
        ]
        builder_metadata = builder_run / "checkpoints" / (
            "clip_0004.%s.json" % builder_four)
        builder_metadata.write_text(
            json.dumps(builder_four_meta), encoding="utf-8")
        (builder_run / "checkpoints" / "clip_0004.json").write_text(
            json.dumps(builder_four_meta), encoding="utf-8")
        builder_graph = helpers.chain.CheckpointGraphManager(
            temporary).graph("builder_graph")
        builder_record = next(
            item for item in builder_graph["revisions"]
            if item["scene"] == 4 and item["revision"] == builder_four)
        assert builder_record["dependencies"] == [
            {"scene": 1, "revision": builder_one},
            {"scene": 2, "revision": builder_two},
            {"scene": 3, "revision": builder_three},
        ]
        assert not helpers.chain.CheckpointGraphManager(
            temporary).deletion_preview(
                "builder_graph", 1, builder_one)["allowed"]

        # Editorial chapters are independent activation scopes. A Chapter 2
        # branch may retain provenance to an inactive Chapter 1 revision, but
        # promoting it must not rewrite Chapter 1's active pointers.
        scoped_run = pathlib.Path(temporary) / "h3_chains" / "chapter_scopes"
        one = "1" * 32
        two_active = "2" * 32
        two_other = "3" * 32
        three_active = "4" * 32
        four_active = "5" * 32
        three_other = "6" * 32
        four_other = "7" * 32
        one_meta, _ = helpers.write_revision(
            scoped_run, 1, one, 2001, active=True,
            run_name="chapter_scopes")
        two_active_meta, _ = helpers.write_revision(
            scoped_run, 2, two_active, 2002, active=True,
            predecessor=one_meta, run_name="chapter_scopes")
        two_other_meta, _ = helpers.write_revision(
            scoped_run, 2, two_other, 2003, predecessor=one_meta,
            run_name="chapter_scopes")
        three_active_meta, _ = helpers.write_revision(
            scoped_run, 3, three_active, 2004, active=True,
            predecessor=two_active_meta, run_name="chapter_scopes")
        helpers.write_revision(
            scoped_run, 4, four_active, 2005, active=True,
            predecessor=three_active_meta, run_name="chapter_scopes")
        three_other_meta, _ = helpers.write_revision(
            scoped_run, 3, three_other, 2006,
            predecessor=two_other_meta, run_name="chapter_scopes",
            context_length=0, audio_context_length=0,
            generated_continuity="off")
        helpers.write_revision(
            scoped_run, 4, four_other, 2007,
            predecessor=three_other_meta, run_name="chapter_scopes")

        (scoped_run / "plan.json").write_text(json.dumps({
            "plan_hash": "chapter-scope-test",
            "shots": [{"id": "scene_%d" % scene} for scene in range(1, 5)],
        }), encoding="utf-8")
        helpers.chain._save_run_editorial_document({
            "run_name": "chapter_scopes",
            "scene_order": [
                {"scene": scene, "scene_id": "scene_%d" % scene}
                for scene in range(1, 5)
            ],
            "chapters": [
                {"id": "chapter_1", "title": "Chapter 1",
                 "start_scene_id": "scene_1"},
                {"id": "chapter_2", "title": "Chapter 2",
                 "start_scene_id": "scene_3"},
            ],
        })

        chapter_two = await helpers.chain._restore_checkpoint_revisions(
            helpers.JsonRequest({
                "run_name": "chapter_scopes",
                "resume_scene": 5,
                "scope_start_scene": 3,
                "scope_end_scene": 4,
                "activate_only": True,
                "revisions": [
                    {"scene": 3, "revision": three_other},
                    {"scene": 4, "revision": four_other},
                ],
            }))
        assert chapter_two.status == 200
        chapter_two_body = json.loads(chapter_two.text)
        assert chapter_two_body["scope_start_scene"] == 3
        assert chapter_two_body["scope_end_scene"] == 4
        assert chapter_two_body["retired_scope_pointers"] == 0
        pointers = {
            scene: json.loads((scoped_run / "checkpoints" /
                               ("clip_%04d.json" % scene)).read_text())["segment"][
                                   "revision"]
            for scene in range(1, 5)
        }
        assert pointers == {
            1: one, 2: two_active, 3: three_other, 4: four_other}

        manifest = helpers.chain._checkpoint_selection_manifest({
            "run_name": "chapter_scopes",
            "lineage": [
                {"scene": 1, "revision": one},
                {"scene": 2, "revision": two_active},
                {"scene": 3, "revision": three_other},
                {"scene": 4, "revision": four_other},
            ],
            "scope_start_scene": 3,
            "scope_end_scene": 4,
        })
        assert [item["revision"] for item in manifest["segments"]] == [
            one, two_active, three_other, four_other]
        assert manifest["selection_scope"] == {
            "start_scene": 3, "end_scene": 4}
        scoped_graph = helpers.chain.CheckpointGraphManager(
            temporary).graph("chapter_scopes")
        active_branches = [branch for branch in scoped_graph["branches"]
                           if branch["active"]]
        assert [[item["scene"] for item in branch["path"]]
                for branch in active_branches] == [[3, 4], [1, 2]]
        chapter_root = next(
            item for item in scoped_graph["revisions"]
            if item["scene"] == 3 and item["revision"] == three_other)
        assert chapter_root["parent"] is None
        assert chapter_root["dependencies"] == []
        explicit_root = next(
            item for item in scoped_graph["revisions"]
            if item["scene"] == 3 and item["revision"] == three_active)
        assert explicit_root["parent"] is None
        assert explicit_root["dependencies"] == [
            {"scene": 2, "revision": two_active}]
        protected = helpers.chain.CheckpointGraphManager(
            temporary).deletion_preview(
                "chapter_scopes", 2, two_active)
        assert not protected["allowed"]
        assert any(item["revision"] == three_active
                   for item in protected["dependents"])

        chapter_one = await helpers.chain._restore_checkpoint_revisions(
            helpers.JsonRequest({
                "run_name": "chapter_scopes",
                "resume_scene": 3,
                "scope_start_scene": 1,
                "scope_end_scene": 2,
                "activate_only": True,
                "revisions": [
                    {"scene": 1, "revision": one},
                    {"scene": 2, "revision": two_other},
                ],
            }))
        assert chapter_one.status == 200
        pointers = {
            scene: json.loads((scoped_run / "checkpoints" /
                               ("clip_%04d.json" % scene)).read_text())["segment"][
                                   "revision"]
            for scene in range(1, 5)
        }
        assert pointers == {
            1: one, 2: two_other, 3: three_other, 4: four_other}
        independent = helpers.chain.CheckpointGraphManager(
            temporary).deletion_preview(
                "chapter_scopes", 2, two_other)
        assert independent["allowed"] and independent["rollback"]

        incompatible_dependency = await (
            helpers.chain._restore_checkpoint_revisions(
                helpers.JsonRequest({
                    "run_name": "chapter_scopes",
                    "resume_scene": 5,
                    "scope_start_scene": 3,
                    "scope_end_scene": 4,
                    "activate_only": True,
                    "revisions": [
                        {"scene": 3, "revision": three_active},
                        {"scene": 4, "revision": four_active},
                    ],
                })))
        assert incompatible_dependency.status == 400
        assert "explicitly depends on scene 2" in json.loads(
            incompatible_dependency.text)["error"]


if __name__ == "__main__":
    asyncio.run(check())
    print("H3 checkpoint activation: mixed policies and independent chapter scopes pass")

	#!/usr/bin/env python3
	"""Focused mixed-policy checkpoint branch activation regression."""

	import asyncio
	import importlib.util
	import json
	import pathlib
	import tempfile


	ROOT = pathlib.Path(__file__).resolve().parents[1]
	HELPERS_PATH = ROOT / "tests" / "_checkpoint_revision_unit_test.py"
	spec = importlib.util.spec_from_file_location(
	"h3_checkpoint_activation_helpers", HELPERS_PATH)
	helpers = importlib.util.module_from_spec(spec)
	spec.loader.exec_module(helpers)


	async def check():
	with tempfile.TemporaryDirectory() as temporary:
	helpers.folder_paths.output_directory = temporary
	run = pathlib.Path(temporary) / "h3_chains" / "mixed_policy"
	first = "a" * 32
	second = "b" * 32
	first_meta, _ = helpers.write_revision(
	run, 1, first, 1001, active=True, run_name="mixed_policy")
	helpers.write_revision(
	run, 2, second, 1002, predecessor=first_meta,
	run_name="mixed_policy", compatibility={
	"context_length": 39,
	"audio_context_length": 39,
	"audio_mode": "generated_audio",
	})
	selection = {
	"run_name": "mixed_policy",
	"resume_scene": 3,
	"revisions": [
	{"scene": 1, "revision": first},
	{"scene": 2, "revision": second},
	],
	}

	strict = await helpers.chain._restore_checkpoint_revisions(
	helpers.JsonRequest(selection))
	assert strict.status == 400
	assert "different Plan compatibility" in json.loads(
	strict.text)["error"]
	assert not (run / "checkpoints" / "clip_0002.json").exists()

	promoted = await helpers.chain._restore_checkpoint_revisions(
	helpers.JsonRequest({**selection, "activate_only": True}))
	assert promoted.status == 200
	assert json.loads(promoted.text)["activate_only"] is True
	assert json.loads((
	run / "checkpoints" / "clip_0002.json").read_text())[
	"segment"]["revision"] == second
	graph = helpers.chain.CheckpointGraphManager(
	temporary).graph("mixed_policy")
	assert [item["revision"] for item in graph["revisions"]
	if item["active"]] == [first, second]
	assert graph["branches"][0]["active"] is True

	# One builder scene can depend on several saved picture revisions,
	# including repeated windows from the same source. Checkpoint Manager
	# must protect every distinct source, not only the nearest block.
	builder_run = pathlib.Path(temporary) / "h3_chains" / "builder_graph"
	builder_one = "8" * 32
	builder_two = "9" * 32
	builder_three = "c" * 32
	builder_four = "d" * 32
	builder_one_meta, _ = helpers.write_revision(
	builder_run, 1, builder_one, 3001, active=True,
	run_name="builder_graph")
	builder_two_meta, _ = helpers.write_revision(
	builder_run, 2, builder_two, 3002, active=True,
	predecessor=builder_one_meta, run_name="builder_graph")
	builder_three_meta, _ = helpers.write_revision(
	builder_run, 3, builder_three, 3003, active=True,
	predecessor=builder_two_meta, run_name="builder_graph")
	builder_four_meta, _ = helpers.write_revision(
	builder_run, 4, builder_four, 3004, active=True,
	predecessor=builder_three_meta, run_name="builder_graph",
	generated_continuity="off")
	builder_four_meta["segment"]["visual_context_blocks"] = [
	{
	"source_scene": 1, "source_revision": builder_one,
	"source_checkpoint_sha256": builder_one_meta["segment"][
	"checkpoint_sha256"], "frames": 5,
	},
	{
	"source_scene": 2, "source_revision": builder_two,
	"source_checkpoint_sha256": builder_two_meta["segment"][
	"checkpoint_sha256"], "frames": 12,
	},
	{
	"source_scene": 1, "source_revision": builder_one,
	"source_checkpoint_sha256": builder_one_meta["segment"][
	"checkpoint_sha256"], "frames": 22,
	},
	]
	builder_metadata = builder_run / "checkpoints" / (
	"clip_0004.%s.json" % builder_four)
	builder_metadata.write_text(
	json.dumps(builder_four_meta), encoding="utf-8")
	(builder_run / "checkpoints" / "clip_0004.json").write_text(
	json.dumps(builder_four_meta), encoding="utf-8")
	builder_graph = helpers.chain.CheckpointGraphManager(
	temporary).graph("builder_graph")
	builder_record = next(
	item for item in builder_graph["revisions"]
	if item["scene"] == 4 and item["revision"] == builder_four)
	assert builder_record["dependencies"] == [
	{"scene": 1, "revision": builder_one},
	{"scene": 2, "revision": builder_two},
	{"scene": 3, "revision": builder_three},
	]
	assert not helpers.chain.CheckpointGraphManager(
	temporary).deletion_preview(
	"builder_graph", 1, builder_one)["allowed"]

	# Editorial chapters are independent activation scopes. A Chapter 2
	# branch may retain provenance to an inactive Chapter 1 revision, but
	# promoting it must not rewrite Chapter 1's active pointers.
	scoped_run = pathlib.Path(temporary) / "h3_chains" / "chapter_scopes"
	one = "1" * 32
	two_active = "2" * 32
	two_other = "3" * 32
	three_active = "4" * 32
	four_active = "5" * 32
	three_other = "6" * 32
	four_other = "7" * 32
	one_meta, _ = helpers.write_revision(
	scoped_run, 1, one, 2001, active=True,
	run_name="chapter_scopes")
	two_active_meta, _ = helpers.write_revision(
	scoped_run, 2, two_active, 2002, active=True,
	predecessor=one_meta, run_name="chapter_scopes")
	two_other_meta, _ = helpers.write_revision(
	scoped_run, 2, two_other, 2003, predecessor=one_meta,
	run_name="chapter_scopes")
	three_active_meta, _ = helpers.write_revision(
	scoped_run, 3, three_active, 2004, active=True,
	predecessor=two_active_meta, run_name="chapter_scopes")
	helpers.write_revision(
	scoped_run, 4, four_active, 2005, active=True,
	predecessor=three_active_meta, run_name="chapter_scopes")
	three_other_meta, _ = helpers.write_revision(
	scoped_run, 3, three_other, 2006,
	predecessor=two_other_meta, run_name="chapter_scopes",
	context_length=0, audio_context_length=0,
	generated_continuity="off")
	helpers.write_revision(
	scoped_run, 4, four_other, 2007,
	predecessor=three_other_meta, run_name="chapter_scopes")

	(scoped_run / "plan.json").write_text(json.dumps({
	"plan_hash": "chapter-scope-test",
	"shots": [{"id": "scene_%d" % scene} for scene in range(1, 5)],
	}), encoding="utf-8")
	helpers.chain._save_run_editorial_document({
	"run_name": "chapter_scopes",
	"scene_order": [
	{"scene": scene, "scene_id": "scene_%d" % scene}
	for scene in range(1, 5)
	],
	"chapters": [
	{"id": "chapter_1", "title": "Chapter 1",
	"start_scene_id": "scene_1"},
	{"id": "chapter_2", "title": "Chapter 2",
	"start_scene_id": "scene_3"},
	],
	})

	chapter_two = await helpers.chain._restore_checkpoint_revisions(
	helpers.JsonRequest({
	"run_name": "chapter_scopes",
	"resume_scene": 5,
	"scope_start_scene": 3,
	"scope_end_scene": 4,
	"activate_only": True,
	"revisions": [
	{"scene": 3, "revision": three_other},
	{"scene": 4, "revision": four_other},
	],
	}))
	assert chapter_two.status == 200
	chapter_two_body = json.loads(chapter_two.text)
	assert chapter_two_body["scope_start_scene"] == 3
	assert chapter_two_body["scope_end_scene"] == 4
	assert chapter_two_body["retired_scope_pointers"] == 0
	pointers = {
	scene: json.loads((scoped_run / "checkpoints" /
	("clip_%04d.json" % scene)).read_text())["segment"][
	"revision"]
	for scene in range(1, 5)
	}
	assert pointers == {
	1: one, 2: two_active, 3: three_other, 4: four_other}

	manifest = helpers.chain._checkpoint_selection_manifest({
	"run_name": "chapter_scopes",
	"lineage": [
	{"scene": 1, "revision": one},
	{"scene": 2, "revision": two_active},
	{"scene": 3, "revision": three_other},
	{"scene": 4, "revision": four_other},
	],
	"scope_start_scene": 3,
	"scope_end_scene": 4,
	})
	assert [item["revision"] for item in manifest["segments"]] == [
	one, two_active, three_other, four_other]
	assert manifest["selection_scope"] == {
	"start_scene": 3, "end_scene": 4}
	scoped_graph = helpers.chain.CheckpointGraphManager(
	temporary).graph("chapter_scopes")
	active_branches = [branch for branch in scoped_graph["branches"]
	if branch["active"]]
	assert [[item["scene"] for item in branch["path"]]
	for branch in active_branches] == [[3, 4], [1, 2]]
	chapter_root = next(
	item for item in scoped_graph["revisions"]
	if item["scene"] == 3 and item["revision"] == three_other)
	assert chapter_root["parent"] is None
	assert chapter_root["dependencies"] == []
	explicit_root = next(
	item for item in scoped_graph["revisions"]
	if item["scene"] == 3 and item["revision"] == three_active)
	assert explicit_root["parent"] is None
	assert explicit_root["dependencies"] == [
	{"scene": 2, "revision": two_active}]
	protected = helpers.chain.CheckpointGraphManager(
	temporary).deletion_preview(
	"chapter_scopes", 2, two_active)
	assert not protected["allowed"]
	assert any(item["revision"] == three_active
	for item in protected["dependents"])

	chapter_one = await helpers.chain._restore_checkpoint_revisions(
	helpers.JsonRequest({
	"run_name": "chapter_scopes",
	"resume_scene": 3,
	"scope_start_scene": 1,
	"scope_end_scene": 2,
	"activate_only": True,
	"revisions": [
	{"scene": 1, "revision": one},
	{"scene": 2, "revision": two_other},
	],
	}))
	assert chapter_one.status == 200
	pointers = {
	scene: json.loads((scoped_run / "checkpoints" /
	("clip_%04d.json" % scene)).read_text())["segment"][
	"revision"]
	for scene in range(1, 5)
	}
	assert pointers == {
	1: one, 2: two_other, 3: three_other, 4: four_other}
	independent = helpers.chain.CheckpointGraphManager(
	temporary).deletion_preview(
	"chapter_scopes", 2, two_other)
	assert independent["allowed"] and independent["rollback"]

	incompatible_dependency = await (
	helpers.chain._restore_checkpoint_revisions(
	helpers.JsonRequest({
	"run_name": "chapter_scopes",
	"resume_scene": 5,
	"scope_start_scene": 3,
	"scope_end_scene": 4,
	"activate_only": True,
	"revisions": [
	{"scene": 3, "revision": three_active},
	{"scene": 4, "revision": four_active},
	],
	})))
	assert incompatible_dependency.status == 400
	assert "explicitly depends on scene 2" in json.loads(
	incompatible_dependency.text)["error"]


	if __name__ == "__main__":
	asyncio.run(check())
	print("H3 checkpoint activation: mixed policies and independent chapter scopes pass")