rookiestar28-ComfyUI-Sigmax/tests/test_node_registration.py

"""M4-06 namespaced registration and schema-discovery contracts."""

from __future__ import annotations

import ast
import json
from dataclasses import FrozenInstanceError, replace
from pathlib import Path
from types import SimpleNamespace
from typing import Any, cast

import pytest
from comfyui_sigmax.adapters import (
    ComfyApiLifecycle,
    ComfyApiProbe,
    ComfyNodeSchemaForm,
)
from comfyui_sigmax.adapters.registration import (
    NODE_REGISTRATION_SCHEMA_ID,
    NODE_REGISTRATION_SCHEMA_VERSION,
    SIGMAX_NODE_MODULE,
    NodeDefinitionKind,
    NodeRegistration,
    NodeRegistrationError,
    NodeRegistrationReason,
    NodeRegistry,
    builtin_node_registry,
    discover_legacy_registration,
    discover_v3_registration,
    registration_from_node_definition_v2,
    require_registration_compatible,
)
from comfyui_sigmax.core import ScheduleContractError
from comfyui_sigmax.profiles import HostCapabilityLifecycle

ROOT = Path(__file__).resolve().parents[1]


class LegacyNode:
    RETURN_TYPES = ("SIGMAS",)
    RETURN_NAMES = ("sigmas",)
    OUTPUT_IS_LIST = (False,)
    CATEGORY = "Sigmax/testing"
    DESCRIPTION = "Legacy test node"
    OUTPUT_NODE = False
    DEPRECATED = False
    EXPERIMENTAL = False

    @classmethod
    def INPUT_TYPES(cls) -> dict[str, object]:
        return {
            "required": {
                "model": ("MODEL",),
                "mode": (["simple", "normal"],),
            },
            "optional": {
                "strength": ("FLOAT", {"default": 1.0}),
            },
        }


def _v3_info(
    *,
    node_id: str = "Sigmax.V3Fixture",
    experimental: bool = True,
) -> dict[str, object]:
    return {
        "input": {
            "required": {
                "model": ["MODEL"],
                "sigmas": ["SIGMAS"],
            }
        },
        "output": ["SIGMAS"],
        "output_is_list": [False],
        "output_name": ["sigmas"],
        "name": node_id,
        "display_name": "Sigmax V3 Fixture",
        "description": "V3 test node",
        "python_module": "custom_nodes.any_directory",
        "category": "Sigmax/testing",
        "output_node": False,
        "deprecated": False,
        "experimental": experimental,
    }


class V3Node:
    @classmethod
    def GET_SCHEMA(cls) -> object:
        return SimpleNamespace(
            node_id="Sigmax.V3Fixture",
            display_name="Sigmax V3 Fixture",
            description="V3 test node",
            category="Sigmax/testing",
            is_output_node=False,
            is_deprecated=False,
            is_experimental=True,
        )

    @classmethod
    def GET_NODE_INFO_V1(cls) -> dict[str, object]:
        return _v3_info()


def _v2_payload(
    *,
    node_id: str = "Sigmax.V2Fixture",
    experimental: bool = False,
) -> dict[str, object]:
    return {
        "inputs": {
            "sampler_name": {
                "name": "sampler_name",
                "type": "COMBO",
                "options": ["euler", "dpmpp_2m"],
                "isOptional": False,
            }
        },
        "outputs": [
            {
                "index": 0,
                "name": "sampler",
                "type": "SAMPLER",
                "is_list": False,
            }
        ],
        "name": node_id,
        "display_name": "Sigmax V2 Fixture",
        "description": "V2 test node",
        "category": "Sigmax/testing",
        "output_node": False,
        "python_module": "comfyui_sigmax.nodes",
        "deprecated": False,
        "experimental": experimental,
    }


def _probe(*, stable: bool = False) -> ComfyApiProbe:
    return ComfyApiProbe(
        module_name="comfy_api.v0_0_2" if not stable else "comfy_api.v0_0_3",
        api_version="0.0.2" if not stable else "0.0.3",
        is_numbered=True,
        lifecycle=ComfyApiLifecycle.EXPERIMENTAL if not stable else ComfyApiLifecycle.LANDED,
        public_symbols=("ComfyExtension", "io", "ui"),
    )


def _legacy() -> NodeRegistration:
    return discover_legacy_registration(
        node_id="Sigmax.LegacyFixture",
        display_name="Sigmax Legacy Fixture",
        node_class=LegacyNode,
    )


def _v3() -> NodeRegistration:
    return discover_v3_registration(V3Node)


def _v2() -> NodeRegistration:
    return registration_from_node_definition_v2(
        node_class=type("V2Node", (), {}),
        payload=_v2_payload(),
    )


def test_legacy_discovery_uses_explicit_namespaced_identity() -> None:
    registration = _legacy()

    assert registration.schema_id == NODE_REGISTRATION_SCHEMA_ID
    assert registration.schema_version == NODE_REGISTRATION_SCHEMA_VERSION == "1"
    assert registration.node_id == "Sigmax.LegacyFixture"
    assert registration.display_name == "Sigmax Legacy Fixture"
    assert registration.definition_kind is NodeDefinitionKind.LEGACY_V1
    assert registration.definition.schema_form is ComfyNodeSchemaForm.OBJECT_INFO_V1
    assert registration.definition.node_id == registration.node_id
    assert registration.python_module == SIGMAX_NODE_MODULE
    assert registration.lifecycle is HostCapabilityLifecycle.LANDED


def test_v3_discovery_requires_public_schema_and_compatibility_projection() -> None:
    registration = _v3()

    assert registration.node_id == "Sigmax.V3Fixture"
    assert registration.definition_kind is NodeDefinitionKind.COMFY_V3
    assert registration.definition.schema_form is ComfyNodeSchemaForm.OBJECT_INFO_V1
    assert registration.lifecycle is HostCapabilityLifecycle.EXPERIMENTAL
    assert tuple(item.name for item in registration.definition.inputs) == ("model", "sigmas")


def test_node_definition_v2_discovery_preserves_wire_form_and_lifecycle() -> None:
    registration = _v2()

    assert registration.node_id == "Sigmax.V2Fixture"
    assert registration.definition_kind is NodeDefinitionKind.NODE_DEFINITION_V2
    assert registration.definition.schema_form is ComfyNodeSchemaForm.NODE_DEFINITION_V2
    assert registration.lifecycle is HostCapabilityLifecycle.LANDED


def test_registry_is_copy_on_write_canonical_and_idempotent() -> None:
    empty = NodeRegistry.empty()
    first = empty.register(_v3()).register(_legacy())
    repeated = first.register(_legacy())

    assert empty.entries == ()
    assert tuple(item.node_id for item in first.entries) == (
        "Sigmax.LegacyFixture",
        "Sigmax.V3Fixture",
    )
    assert repeated is first


def test_register_many_accepts_mixed_public_definition_forms() -> None:
    registry = NodeRegistry.empty().register_many((_v3(), _v2(), _legacy()))

    assert tuple(item.definition_kind for item in registry.entries) == (
        NodeDefinitionKind.LEGACY_V1,
        NodeDefinitionKind.NODE_DEFINITION_V2,
        NodeDefinitionKind.COMFY_V3,
    )


def test_collision_never_overwrites_unrelated_class_or_schema() -> None:
    registry = NodeRegistry.empty().register(_legacy())
    different_class = discover_legacy_registration(
        node_id="Sigmax.LegacyFixture",
        display_name="Sigmax Legacy Fixture",
        node_class=type(
            "OtherLegacy",
            (),
            {
                "INPUT_TYPES": classmethod(lambda cls: {"required": {}}),
                "RETURN_TYPES": (),
            },
        ),
    )

    with pytest.raises(NodeRegistrationError) as captured:
        registry.register(different_class)

    assert captured.value.reason is NodeRegistrationReason.NODE_ID_COLLISION
    assert registry.class_mappings()["Sigmax.LegacyFixture"] is LegacyNode


def test_mapping_and_display_projections_are_fresh_and_exact() -> None:
    registry = NodeRegistry.empty().register_many((_legacy(), _v3()))

    classes = registry.class_mappings()
    displays = registry.display_name_mappings()
    classes.clear()
    displays.clear()

    assert registry.class_mappings() == {
        "Sigmax.LegacyFixture": LegacyNode,
        "Sigmax.V3Fixture": V3Node,
    }
    assert registry.display_name_mappings() == {
        "Sigmax.LegacyFixture": "Sigmax Legacy Fixture",
        "Sigmax.V3Fixture": "Sigmax V3 Fixture",
    }


def test_object_info_projection_round_trips_through_m4_adapter_shape() -> None:
    registry = NodeRegistry.empty().register_many((_legacy(), _v3(), _v2()))
    wire = registry.object_info_projection()

    assert tuple(wire) == (
        "Sigmax.LegacyFixture",
        "Sigmax.V2Fixture",
        "Sigmax.V3Fixture",
    )
    assert wire["Sigmax.LegacyFixture"]["input"]["required"]["mode"][0] == [
        "simple",
        "normal",
    ]
    assert wire["Sigmax.V3Fixture"]["experimental"] is True
    assert wire["Sigmax.V2Fixture"]["experimental"] is False


def test_node_definition_v2_projection_is_documented_and_deterministic() -> None:
    registry = NodeRegistry.empty().register_many((_legacy(), _v3(), _v2()))
    wire = registry.node_definition_v2_projection()

    assert tuple(wire) == (
        "Sigmax.LegacyFixture",
        "Sigmax.V2Fixture",
        "Sigmax.V3Fixture",
    )
    assert wire["Sigmax.LegacyFixture"]["inputs"]["strength"]["isOptional"] is True
    assert wire["Sigmax.V2Fixture"] == _v2_payload()
    assert wire["Sigmax.V3Fixture"]["experimental"] is True


def test_projection_mutation_does_not_change_registry() -> None:
    registry = NodeRegistry.empty().register(_legacy())
    first = registry.object_info_projection()
    first["Sigmax.LegacyFixture"]["input"]["required"].clear()

    second = registry.object_info_projection()
    assert tuple(second["Sigmax.LegacyFixture"]["input"]["required"]) == ("mode", "model")


def test_node_id_never_depends_on_installation_directory_or_module_name() -> None:
    original_module = LegacyNode.__module__
    try:
        LegacyNode.__module__ = "custom_nodes.renamed_directory.nodes"
        registration = _legacy()
    finally:
        LegacyNode.__module__ = original_module

    assert registration.node_id == "Sigmax.LegacyFixture"
    assert registration.python_module == "comfyui_sigmax.nodes"


@pytest.mark.parametrize(
    "node_id",
    [
        "LegacyFixture",
        "sigmax.LegacyFixture",
        "Sigmax/LegacyFixture",
        "Sigmax.legacy_fixture",
        "Sigmax.",
        "Sigmax." + ("A" * 65),
    ],
)
def test_non_namespaced_or_unstable_node_ids_reject(node_id: str) -> None:
    with pytest.raises(NodeRegistrationError) as captured:
        discover_legacy_registration(
            node_id=node_id,
            display_name="Fixture",
            node_class=LegacyNode,
        )

    assert captured.value.reason is NodeRegistrationReason.INVALID_NODE_ID


def test_current_experimental_api_allows_discovery_but_rejects_v3_activation() -> None:
    registration = _v3()

    with pytest.raises(NodeRegistrationError) as captured:
        require_registration_compatible(registration, _probe())

    assert captured.value.reason is NodeRegistrationReason.V3_API_NOT_STABLE
    assert "stable numbered Comfy API" in captured.value.action


def test_stable_api_allows_v3_and_legacy_does_not_require_v3() -> None:
    assert require_registration_compatible(_v3(), _probe(stable=True)) == _v3()
    assert require_registration_compatible(_legacy(), _probe()) == _legacy()


def test_v3_schema_and_object_info_identity_disagreement_rejects() -> None:
    class Mismatch(V3Node):
        @classmethod
        def GET_NODE_INFO_V1(cls) -> dict[str, object]:
            return _v3_info(node_id="Sigmax.Other")

    with pytest.raises(NodeRegistrationError) as captured:
        discover_v3_registration(Mismatch)

    assert captured.value.reason is NodeRegistrationReason.SCHEMA_ID_MISMATCH


def test_v3_lifecycle_disagreement_rejects() -> None:
    class Mismatch(V3Node):
        @classmethod
        def GET_NODE_INFO_V1(cls) -> dict[str, object]:
            return _v3_info(experimental=False)

    with pytest.raises(NodeRegistrationError) as captured:
        discover_v3_registration(Mismatch)

    assert captured.value.reason is NodeRegistrationReason.SCHEMA_LIFECYCLE_MISMATCH


@pytest.mark.parametrize(
    "node_class",
    [
        object(),
        type("MissingInputTypes", (), {"RETURN_TYPES": ()}),
        type(
            "MissingReturns",
            (),
            {"INPUT_TYPES": classmethod(lambda cls: {"required": {}})},
        ),
        type(
            "BrokenInputs",
            (),
            {
                "INPUT_TYPES": classmethod(lambda cls: []),
                "RETURN_TYPES": (),
            },
        ),
    ],
)
def test_legacy_discovery_fails_actionably_for_invalid_public_class(
    node_class: object,
) -> None:
    with pytest.raises(NodeRegistrationError) as captured:
        discover_legacy_registration(
            node_id="Sigmax.Broken",
            display_name="Broken",
            node_class=cast(Any, node_class),
        )

    assert captured.value.reason in {
        NodeRegistrationReason.INVALID_NODE_CLASS,
        NodeRegistrationReason.SCHEMA_MALFORMED,
    }
    assert captured.value.action


@pytest.mark.parametrize(
    "node_class",
    [
        object(),
        type("MissingInfo", (), {"GET_SCHEMA": classmethod(lambda cls: object())}),
        type(
            "BrokenInfo",
            (),
            {
                "GET_SCHEMA": classmethod(
                    lambda cls: SimpleNamespace(
                        node_id="Sigmax.Broken",
                        display_name="Broken",
                        description="",
                        category="Sigmax/testing",
                        is_output_node=False,
                        is_deprecated=False,
                        is_experimental=False,
                    )
                ),
                "GET_NODE_INFO_V1": classmethod(lambda cls: []),
            },
        ),
    ],
)
def test_v3_discovery_fails_actionably_for_invalid_public_class(
    node_class: object,
) -> None:
    with pytest.raises(NodeRegistrationError) as captured:
        discover_v3_registration(cast(Any, node_class))

    assert captured.value.reason in {
        NodeRegistrationReason.INVALID_NODE_CLASS,
        NodeRegistrationReason.SCHEMA_MALFORMED,
    }
    assert captured.value.action


def test_registration_contracts_are_immutable() -> None:
    registration = _legacy()
    registry = NodeRegistry.empty().register(registration)

    with pytest.raises(FrozenInstanceError):
        registration.node_id = "Sigmax.Changed"  # type: ignore[misc]
    with pytest.raises(FrozenInstanceError):
        registry.entries = ()  # type: ignore[misc]


def test_builtin_registry_and_package_mappings_expose_only_validated_product_nodes() -> None:
    import comfyui_sigmax
    from comfyui_sigmax.nodes import (
        AdvancedFlowMatchScheduler,
        AnimaSigmaScheduler,
        AuraFlowSigmaScheduler,
        CheckpointEvidenceInspector,
        Flux1SchnellSigmaScheduler,
        HunyuanImage21SigmaScheduler,
        Krea2ConditioningRebalance,
        Krea2SigmaScheduler,
        LTXSigmaScheduler,
        Lumina2SigmaScheduler,
        MiniMaxH3SigmaScheduler,
        ModelAwareSigmaScheduler,
        ProfileInspector,
        QwenImageSigmaScheduler,
        RawWorkflowOutput,
        ScheduleComparison,
        ScheduleConcatenate,
        ScheduleInspector,
        ScheduleResample,
        ScheduleSlice,
        SD3SigmaScheduler,
        TurboWorkflowOutput,
        WanSigmaScheduler,
        ZImageSigmaScheduler,
    )

    registry = builtin_node_registry()
    assert tuple(item.node_id for item in registry.entries) == (
        "Sigmax.AdvancedFlowMatchScheduler",
        "Sigmax.AnimaSigmaScheduler",
        "Sigmax.AuraFlowSigmaScheduler",
        "Sigmax.CheckpointEvidenceInspector",
        "Sigmax.Flux1SchnellSigmaScheduler",
        "Sigmax.HunyuanImage21SigmaScheduler",
        "Sigmax.Krea2ConditioningRebalance",
        "Sigmax.Krea2SigmaScheduler",
        "Sigmax.LTXSigmaScheduler",
        "Sigmax.Lumina2SigmaScheduler",
        "Sigmax.MiniMaxH3SigmaScheduler",
        "Sigmax.ModelAwareSigmaScheduler",
        "Sigmax.ProfileInspector",
        "Sigmax.QwenImageSigmaScheduler",
        "Sigmax.RawWorkflowOutput",
        "Sigmax.SD3SigmaScheduler",
        "Sigmax.ScheduleComparison",
        "Sigmax.ScheduleConcatenate",
        "Sigmax.ScheduleInspector",
        "Sigmax.ScheduleResample",
        "Sigmax.ScheduleSlice",
        "Sigmax.TurboWorkflowOutput",
        "Sigmax.WanSigmaScheduler",
        "Sigmax.ZImageSigmaScheduler",
    )
    assert {
        "Sigmax.AdvancedFlowMatchScheduler": AdvancedFlowMatchScheduler,
        "Sigmax.AnimaSigmaScheduler": AnimaSigmaScheduler,
        "Sigmax.AuraFlowSigmaScheduler": AuraFlowSigmaScheduler,
        "Sigmax.CheckpointEvidenceInspector": CheckpointEvidenceInspector,
        "Sigmax.Flux1SchnellSigmaScheduler": Flux1SchnellSigmaScheduler,
        "Sigmax.HunyuanImage21SigmaScheduler": HunyuanImage21SigmaScheduler,
        "Sigmax.Krea2ConditioningRebalance": Krea2ConditioningRebalance,
        "Sigmax.Krea2SigmaScheduler": Krea2SigmaScheduler,
        "Sigmax.LTXSigmaScheduler": LTXSigmaScheduler,
        "Sigmax.Lumina2SigmaScheduler": Lumina2SigmaScheduler,
        "Sigmax.MiniMaxH3SigmaScheduler": MiniMaxH3SigmaScheduler,
        "Sigmax.QwenImageSigmaScheduler": QwenImageSigmaScheduler,
        "Sigmax.ModelAwareSigmaScheduler": ModelAwareSigmaScheduler,
        "Sigmax.SD3SigmaScheduler": SD3SigmaScheduler,
        "Sigmax.ProfileInspector": ProfileInspector,
        "Sigmax.RawWorkflowOutput": RawWorkflowOutput,
        "Sigmax.ScheduleComparison": ScheduleComparison,
        "Sigmax.ScheduleConcatenate": ScheduleConcatenate,
        "Sigmax.ScheduleInspector": ScheduleInspector,
        "Sigmax.ScheduleResample": ScheduleResample,
        "Sigmax.ScheduleSlice": ScheduleSlice,
        "Sigmax.TurboWorkflowOutput": TurboWorkflowOutput,
        "Sigmax.WanSigmaScheduler": WanSigmaScheduler,
        "Sigmax.ZImageSigmaScheduler": ZImageSigmaScheduler,
    } == comfyui_sigmax.NODE_CLASS_MAPPINGS
    assert comfyui_sigmax.NODE_DISPLAY_NAME_MAPPINGS == {
        "Sigmax.AdvancedFlowMatchScheduler": "Advanced FlowMatch Scheduler",
        "Sigmax.AnimaSigmaScheduler": "Anima Sigma Scheduler",
        "Sigmax.AuraFlowSigmaScheduler": "AuraFlow Sigma Scheduler",
        "Sigmax.CheckpointEvidenceInspector": "Checkpoint Evidence Inspector",
        "Sigmax.Flux1SchnellSigmaScheduler": "FLUX.1-schnell Sigma Scheduler",
        "Sigmax.HunyuanImage21SigmaScheduler": "HunyuanImage 2.1 Sigma Scheduler",
        "Sigmax.Krea2ConditioningRebalance": "Krea 2 Conditioning Rebalance",
        "Sigmax.Krea2SigmaScheduler": "Krea 2 Sigma Scheduler",
        "Sigmax.LTXSigmaScheduler": "LTX Sigma Scheduler",
        "Sigmax.Lumina2SigmaScheduler": "Lumina-Image 2.0 Sigma Scheduler",
        "Sigmax.MiniMaxH3SigmaScheduler": "MiniMax H3 Sigma Scheduler",
        "Sigmax.QwenImageSigmaScheduler": "Qwen Image Sigma Scheduler",
        "Sigmax.ModelAwareSigmaScheduler": "Model-Aware Sigma Scheduler",
        "Sigmax.SD3SigmaScheduler": "Stable Diffusion 3 Sigma Scheduler",
        "Sigmax.ProfileInspector": "Profile Inspector",
        "Sigmax.RawWorkflowOutput": "RAW Workflow Output",
        "Sigmax.ScheduleComparison": "Schedule Comparison",
        "Sigmax.ScheduleConcatenate": "Schedule Concatenate",
        "Sigmax.ScheduleInspector": "Schedule Inspector",
        "Sigmax.ScheduleResample": "Schedule Resample",
        "Sigmax.ScheduleSlice": "Schedule Slice",
        "Sigmax.TurboWorkflowOutput": "Turbo Workflow Output",
        "Sigmax.WanSigmaScheduler": "Wan Sigma Scheduler",
        "Sigmax.ZImageSigmaScheduler": "Z-Image Sigma Scheduler",
    }
    assert comfyui_sigmax.NODE_CLASS_MAPPINGS is not builtin_node_registry().class_mappings()


def test_bootstrap_uses_catalog_without_inert_v3_entrypoint() -> None:
    package_source = (ROOT / "comfyui_sigmax" / "__init__.py").read_text(encoding="utf-8")

    assert "builtin_node_registry" in package_source
    assert "comfy_entrypoint" not in package_source


def test_registration_source_never_imports_or_mutates_comfy_sampler_registry() -> None:
    path = ROOT / "comfyui_sigmax" / "adapters" / "registration.py"
    source = path.read_text(encoding="utf-8")
    tree = ast.parse(source, filename=path.name)
    imports = {
        alias.name
        for node in ast.walk(tree)
        if isinstance(node, ast.Import)
        for alias in node.names
    } | {
        node.module
        for node in ast.walk(tree)
        if isinstance(node, ast.ImportFrom) and node.module is not None
    }

    assert not any(name == "comfy" or name.startswith("comfy.") for name in imports)
    assert "comfy.samplers" not in source
    assert "add_scheduler" not in source


def test_wrong_argument_types_fail_closed() -> None:
    with pytest.raises(ScheduleContractError):
        NodeRegistry(entries=cast(Any, []))
    with pytest.raises(ScheduleContractError):
        NodeRegistry.empty().register(cast(Any, object()))
    with pytest.raises(ScheduleContractError):
        NodeRegistry.empty().register_many(cast(Any, object()))
    with pytest.raises(ScheduleContractError):
        require_registration_compatible(cast(Any, object()), _probe())
    with pytest.raises(ScheduleContractError):
        require_registration_compatible(_legacy(), cast(Any, object()))


def test_registration_reason_and_action_do_not_echo_private_payload() -> None:
    class PrivateNode:
        RETURN_TYPES = ()

        @classmethod
        def INPUT_TYPES(cls) -> object:
            return {"required": {"C:\\private\\model.ckpt": [object()]}}

    with pytest.raises(NodeRegistrationError) as captured:
        discover_legacy_registration(
            node_id="Sigmax.Private",
            display_name="Private",
            node_class=PrivateNode,
        )

    assert "C:\\private" not in str(captured.value)
    assert "model.ckpt" not in str(captured.value)


@pytest.mark.parametrize(
    "arguments",
    [
        {"reason": cast(Any, "bad"), "action": "fix"},
        {"reason": NodeRegistrationReason.SCHEMA_MALFORMED, "action": ""},
        {
            "reason": NodeRegistrationReason.SCHEMA_MALFORMED,
            "action": cast(Any, None),
        },
    ],
)
def test_registration_error_contract_rejects_invalid_construction(
    arguments: dict[str, object],
) -> None:
    with pytest.raises(ScheduleContractError):
        NodeRegistrationError(**cast(Any, arguments))


@pytest.mark.parametrize(
    "changes",
    [
        {"schema_id": "other"},
        {"schema_version": "2"},
        {"node_id": "Other"},
        {"display_name": cast(Any, None)},
        {"display_name": ""},
        {"display_name": "x" * 513},
        {"display_name": "line\nbreak"},
        {"description": cast(Any, None)},
        {"category": "testing"},
        {"python_module": "custom_nodes.folder"},
        {"output_node": cast(Any, 0)},
        {"node_class": cast(Any, object())},
        {"definition_kind": cast(Any, "legacy_v1")},
        {"definition": cast(Any, object())},
        {"definition": replace(_legacy().definition, node_id="Sigmax.Other")},
        {"lifecycle": cast(Any, "landed")},
        {"lifecycle": HostCapabilityLifecycle.EXPERIMENTAL},
        {"source_payload_json": cast(Any, b"{}")},
        {"source_payload_json": "[]"},
        {"source_payload_json": '{ "name": "not-canonical" }'},
    ],
)
def test_registration_contract_rejects_invalid_values(
    changes: dict[str, object],
) -> None:
    with pytest.raises((ScheduleContractError, NodeRegistrationError)):
        replace(_legacy(), **cast(Any, changes))


def test_deprecated_lifecycle_is_explicitly_unsupported() -> None:
    class DeprecatedLegacy(LegacyNode):
        DEPRECATED = True

    registration = discover_legacy_registration(
        node_id="Sigmax.Deprecated",
        display_name="Deprecated",
        node_class=DeprecatedLegacy,
    )

    assert registration.lifecycle is HostCapabilityLifecycle.UNSUPPORTED


def test_registry_contract_rejects_duplicates_noncanonical_and_invalid_items() -> None:
    legacy = _legacy()
    v3 = _v3()

    with pytest.raises(ScheduleContractError):
        NodeRegistry(entries=(legacy, legacy))
    with pytest.raises(ScheduleContractError):
        NodeRegistry(entries=(v3, legacy))
    with pytest.raises(ScheduleContractError):
        NodeRegistry(entries=cast(Any, (object(),)))
    with pytest.raises(ScheduleContractError):
        NodeRegistry.empty().register_many((cast(Any, object()),))


def test_legacy_discovery_catches_input_execution_and_wire_encoding_failures() -> None:
    class Raises:
        RETURN_TYPES = ()

        @classmethod
        def INPUT_TYPES(cls) -> object:
            raise RuntimeError("private payload")

    with pytest.raises(NodeRegistrationError) as captured:
        discover_legacy_registration(
            node_id="Sigmax.Raises",
            display_name="Raises",
            node_class=Raises,
        )
    assert captured.value.reason is NodeRegistrationReason.SCHEMA_MALFORMED
    assert "private payload" not in str(captured.value)

    class NonFinite(LegacyNode):
        @classmethod
        def INPUT_TYPES(cls) -> dict[str, object]:
            return {"required": {"value": ("FLOAT", {"default": float("nan")})}}

    with pytest.raises(NodeRegistrationError):
        discover_legacy_registration(
            node_id="Sigmax.NonFinite",
            display_name="Non Finite",
            node_class=NonFinite,
        )

    class Oversized(LegacyNode):
        @classmethod
        def INPUT_TYPES(cls) -> dict[str, object]:
            return {"required": {"value": ("STRING", {"default": "x" * 1_048_577})}}

    with pytest.raises(NodeRegistrationError):
        discover_legacy_registration(
            node_id="Sigmax.Oversized",
            display_name="Oversized",
            node_class=Oversized,
        )


def test_legacy_discovery_rejects_invalid_output_fields_and_output_node() -> None:
    for node_class in (
        type(
            "StringReturns",
            (),
            {
                "INPUT_TYPES": classmethod(lambda cls: {"required": {}}),
                "RETURN_TYPES": "SIGMAS",
            },
        ),
        type(
            "BrokenNames",
            (),
            {
                "INPUT_TYPES": classmethod(lambda cls: {"required": {}}),
                "RETURN_TYPES": ("SIGMAS",),
                "RETURN_NAMES": object(),
            },
        ),
        type(
            "BrokenLists",
            (),
            {
                "INPUT_TYPES": classmethod(lambda cls: {"required": {}}),
                "RETURN_TYPES": ("SIGMAS",),
                "OUTPUT_IS_LIST": object(),
            },
        ),
        type(
            "BrokenOutputNode",
            (),
            {
                "INPUT_TYPES": classmethod(lambda cls: {"required": {}}),
                "RETURN_TYPES": (),
                "OUTPUT_NODE": "false",
            },
        ),
    ):
        with pytest.raises((NodeRegistrationError, TypeError)):
            discover_legacy_registration(
                node_id="Sigmax.BrokenOutput",
                display_name="Broken Output",
                node_class=node_class,
            )


@pytest.mark.parametrize(
    "schema_changes",
    [
        {"is_experimental": "false"},
        {"is_deprecated": "false"},
        {"is_deprecated": True},
    ],
)
def test_v3_schema_lifecycle_validation_covers_all_paths(
    schema_changes: dict[str, object],
) -> None:
    class Lifecycle(V3Node):
        @classmethod
        def GET_SCHEMA(cls) -> object:
            values: dict[str, object] = {
                "node_id": "Sigmax.V3Fixture",
                "display_name": "Sigmax V3 Fixture",
                "description": "V3 test node",
                "category": "Sigmax/testing",
                "is_output_node": False,
                "is_deprecated": False,
                "is_experimental": True,
            }
            values.update(schema_changes)
            return SimpleNamespace(**values)

    with pytest.raises(NodeRegistrationError) as captured:
        discover_v3_registration(Lifecycle)

    assert captured.value.reason is NodeRegistrationReason.SCHEMA_LIFECYCLE_MISMATCH


def test_v3_schema_defaults_display_and_metadata_when_omitted() -> None:
    class MinimalV3:
        @classmethod
        def GET_SCHEMA(cls) -> object:
            return SimpleNamespace(node_id="Sigmax.MinimalV3")

        @classmethod
        def GET_NODE_INFO_V1(cls) -> dict[str, object]:
            return {
                "input": {},
                "output": [],
                "name": "Sigmax.MinimalV3",
                "display_name": "Sigmax.MinimalV3",
                "description": "",
                "category": "Sigmax",
                "output_node": False,
            }

    registration = discover_v3_registration(MinimalV3)

    assert registration.display_name == "Sigmax.MinimalV3"
    assert registration.description == ""
    assert registration.category == "Sigmax"
    assert registration.output_node is False


def test_v3_public_method_execution_failure_is_actionable() -> None:
    class Raises:
        @classmethod
        def GET_SCHEMA(cls) -> object:
            raise ValueError("private schema failure")

        @classmethod
        def GET_NODE_INFO_V1(cls) -> dict[str, object]:
            return {}

    with pytest.raises(NodeRegistrationError) as captured:
        discover_v3_registration(Raises)

    assert captured.value.reason is NodeRegistrationReason.SCHEMA_MALFORMED
    assert "private schema failure" not in str(captured.value)


def test_v2_registration_rejects_invalid_class_module_and_payload() -> None:
    with pytest.raises(NodeRegistrationError):
        registration_from_node_definition_v2(
            node_class=cast(Any, object()),
            payload=_v2_payload(),
        )
    wrong_module = _v2_payload()
    wrong_module["python_module"] = "custom_nodes.folder"
    with pytest.raises(NodeRegistrationError):
        registration_from_node_definition_v2(
            node_class=type("Node", (), {}),
            payload=wrong_module,
        )
    with pytest.raises(NodeRegistrationError):
        registration_from_node_definition_v2(
            node_class=type("Node", (), {}),
            payload=[],
        )


def test_v2_object_info_projection_handles_noncombo_and_hidden_exclusion() -> None:
    payload = _v2_payload(node_id="Sigmax.V2Values")
    payload["inputs"] = {
        "model": {"name": "model", "type": "MODEL", "isOptional": False},
        "mask": {"name": "mask", "type": "MASK", "isOptional": True},
    }
    registration = registration_from_node_definition_v2(
        node_class=type("V2Values", (), {}),
        payload=payload,
    )
    registry = NodeRegistry.empty().register(registration)

    object_info = registry.object_info_projection()["Sigmax.V2Values"]
    assert object_info["input"] == {
        "required": {"model": ["MODEL"]},
        "optional": {"mask": ["MASK"]},
    }

    hidden_definition = replace(
        _legacy().definition,
        inputs=(
            *_legacy().definition.inputs,
            cast(
                Any,
                type(_legacy().definition.inputs[0])(
                    name="prompt",
                    type_name="PROMPT",
                    section="hidden",
                    optional=False,
                ),
            ),
        ),
    )
    hidden_registration = replace(
        _legacy(),
        definition=hidden_definition,
        source_payload_json=json.dumps(
            {
                **json.loads(_legacy().source_payload_json),
                "input": {
                    **json.loads(_legacy().source_payload_json)["input"],
                    "hidden": {"prompt": ["PROMPT"]},
                },
            },
            sort_keys=True,
            separators=(",", ":"),
        ),
    )
    v2_wire = (
        NodeRegistry.empty()
        .register(hidden_registration)
        .node_definition_v2_projection()["Sigmax.LegacyFixture"]
    )
    assert "prompt" not in v2_wire["inputs"]