patch_layout.py

"""Place MiniMax H3 continuation guides on the target timeline.

Stock ComfyUI builds keyframe conditioning rows at one of two time
coordinates and rejects everything else:

    if pixel_index == 0:
        cond_t = float(text_len)
    elif frame_count is not None and pixel_index == frame_count - 1:
        cond_t = float(text_len) + sum(_video_t_spans(latent_t)) - FRAME_RESCALE
    else:
        raise ValueError("only first/last keyframe anchors are supported")

Both branches are the same expression. Each video token spans
FRAME_RESCALE * FRAME_PER_TOKEN[k % 5] and covers FRAME_PER_TOKEN[k % 5]
pixel frames, so the cumulative time at pixel frame p is exactly
FRAME_RESCALE * p, for every p. Substituting p = frame_count - 1
reproduces the second branch identically:

    text_len + FRAME_RESCALE * (frame_count - 1)
      == text_len + FRAME_RESCALE * frame_count - FRAME_RESCALE
      == text_len + sum(_video_t_spans(latent_t)) - FRAME_RESCALE

So the general position is:

    cond_t = text_len + FRAME_RESCALE * pixel_index

We do NOT rewrite the source of PackedLayout.__init__. Instead every
keyframe is handed to stock code with resolved_frame_index = 0, which is
always legal, and the real index rides along under MC_KEY. After the
stock constructor returns we rewrite the time column of each cond
segment's rows in position_ids. RoPE is built at forward time from
position_ids, so this lands before anything reads it.

That keeps the patch surface to one attribute we can verify rather than a
copy of a 90-line constructor that would rot on the next ComfyUI change.

ComfyUI PR #15439 adds native arbitrary-position video/audio guides, merges
guide and Ref2VA payloads, aligns guides to the target origin after reference
blocks, and drops the ``frame_count`` constructor argument. On that API this
module detects the native implementation and leaves ``PackedLayout`` entirely
core-owned. Everything below is retained only as a guarded fallback for older
ComfyUI builds.
"""

import inspect
import logging
import sys

import torch

import comfy.ldm.minimax.model as mm

MC_KEY = "motion_context_index"
MC_AUDIO_KEY = "motion_context_audio_end_frame"

# Marker set on our wrapper so a second copy of this file, vendored into
# another pack, can recognise it and stand down instead of wrapping it.
# Shared ABI across every pack that vendors this patch, exactly like
# MC_KEY and MC_AUDIO_KEY: rename it in all of them or in none.
PATCH_MARKER = "_h3_motion_context_layout_patch"
NATIVE_GUIDES_MARKER = "_h3_motion_context_native_guides"
_LOG = logging.getLogger("h3_motion_context")

_orig_init = None
_applied = False

# SolAttn's H3 Morton hook wraps PackedLayout only to record the final video
# span. It calls the constructor it captured, does not mutate the layout, and
# keeps the layout object alive so later in-place position fixes remain visible.
# That makes it safe to compose with this patch in either load order. Keep this
# recognition deliberately narrow: unknown wrappers still fail closed.
SOLATTN_LAYOUT_MODULE_SUFFIX = "._morton_h3"
# The CUDA PR117 development checkout currently installs its H3 observer from
# the pack root rather than ``_morton_h3`` and assigns the checkout path to the
# nested wrapper's ``__module__``.  ComfyUI therefore reports the literal path
# below as PackedLayout.__init__'s owner.  Keep this identity exact: it is safe
# to compose only after the same ``original_init`` closure check used for the
# upstream Kijai wrapper succeeds.
SOLATTN_CUDA_PR117_LAYOUT_MODULE = "ComfyUI-SolAttn-CUDA-PR117"
# Folder names are not an ABI.  ComfyUI loads package roots under their full
# install path, and PR test helpers are commonly copied to names such as
# ``sol_attn_minimax_v2``.  Recognise those copies by the upstream observer's
# read-only span-registration fingerprint instead of growing a folder-name
# allowlist forever.  A future upstream marker is accepted as the clean ABI.
SOLATTN_LAYOUT_OBSERVER_MARKER = "_sol_attn_h3_layout_observer"

# The Kijai observer's fingerprint is a nested ``__init__`` in a module
# that owns two very specific globals: a callable ``_video_span`` that reads
# the target video span off the layout, and a dict ``_SPANS`` that records it
# by ``position_ids`` identity.  The wrapper must reference BOTH those globals
# from its body -- that is what proves this specific module's observer is what
# is installed, not some unrelated wrapper that happens to live alongside them.
SOLATTN_LAYOUT_SIGNATURE_GLOBALS = frozenset({"_video_span", "_SPANS"})
# The wrapper must also touch both PackedLayout attributes used by the
# observer. Two access patterns appear in the wild and both are equally
# observe-only:
#   * ``self.segments`` / ``self.position_ids`` -- attribute lookup, name lands
#     in ``code.co_names`` (Kijai's upstream ``_morton_h3`` does this).
#   * ``getattr(self, "segments", [])`` -- defensive lookup, name lands in
#     ``code.co_consts`` as a string literal (comfy-kitchen PR #117's
#     ``sol_attn_minimax_v2`` does this).
# Requiring both attributes via either access route accepts both styles while
# ensuring that an arbitrary ``__init__`` in the same module does not match.
SOLATTN_LAYOUT_ATTRIBUTE_NAMES = frozenset({"position_ids", "segments"})


REF_SEGMENT_KINDS = ("ref_img", "ref_audio")


def _target_origin(layout):
    """The coordinate the target clip starts at, read off the built layout.

    Stock lays reference blocks out from a cursor that starts at text_len,
    and the target rows take the cursor's final value as their origin.
    Keyframe coordinates are computed from text_len directly and never
    compensated, so without this term any reference slides the anchors
    backwards relative to the clip they are anchoring.

    Earlier versions recomputed that cursor with a local copy of stock's
    per-kind advance arithmetic. Reading it back out of the layout instead
    means there is nothing to keep in sync: if upstream changes how a
    reference kind advances the cursor, the number here changes with it.
    The target video segment is always last and always has at least one
    latent step, and _video_grid puts its first row exactly on the cursor.
    """
    a, b, kind = layout.segments[-1]
    if kind != "video" or b <= a:
        raise RuntimeError(
            "h3_motion_context: expected the target video rows to be the "
            "last layout segment, found %r spanning %d rows. Upstream "
            "layout change; refusing to rewrite positions." % (kind, b - a))
    return float(layout.position_ids[a, 0])


def _expected_ref_segments(blk):
    """The segment kinds one reference block emits, in emission order.

    Mirrors the branches of the stock constructor:
      image         one ref_img
      audio         one ref_audio, or nothing at all when the window is
                    empty (stock skips the segment but still advances)
      video         the block's audio rows pack immediately before its
      video_audio   video rows, so ref_audio then ref_img
    """
    kind = blk.get("kind")
    if kind == "image":
        return ("ref_img",)
    if kind == "audio":
        return ("ref_audio",) if int(blk.get("ref_audio_t", 0)) > 0 else ()
    if kind in ("video", "video_audio"):
        if int(blk.get("ref_audio_t", 0)) > 0:
            return ("ref_audio", "ref_img")
        return ("ref_img",)
    raise RuntimeError(
        "h3_motion_context: unknown reference kind %r; cannot tell which "
        "layout rows belong to it." % (kind,))


def _ref_segment_map(layout, refs):
    """Which rows each reference block actually produced.

    Returns {block_index: {segment_kind: (start, stop)}}.

    This is the whole point of the multi-reference support. The rows of
    one reference block could be found by working out the coordinate span
    it ought to occupy and selecting everything inside it, but that means
    duplicating stock's cursor arithmetic and then hoping nothing else
    shares the range. Stock keyframe rows genuinely do land inside it,
    which is why the coordinate approach needed an explicit exclusion for
    them, and a second reference of the same kind would need another.

    The layout already publishes a segment table, and reference blocks
    emit their segments in list order, so the mapping is exact. Nothing
    is inferred from coordinates and nothing has to be excluded.
    """
    ref_segs = [(a, b, k) for a, b, k in layout.segments
                if k in REF_SEGMENT_KINDS]
    want = [(i, k) for i, blk in enumerate(refs or [])
            for k in _expected_ref_segments(blk)]
    if len(want) != len(ref_segs):
        raise RuntimeError(
            "h3_motion_context: %d reference blocks should have produced %d "
            "layout segments, the layout has %d. Upstream layout change; "
            "refusing to move rows." % (len(refs or []), len(want),
                                        len(ref_segs)))
    out = {}
    for (i, kind), (a, b, got) in zip(want, ref_segs):
        if got != kind:
            raise RuntimeError(
                "h3_motion_context: reference block %d (%r) should have "
                "emitted a %s segment, the layout has %s. Upstream layout "
                "change; refusing to move rows."
                % (i, refs[i].get("kind"), kind, got))
        out.setdefault(i, {})[kind] = (a, b)
    return out


def _cond_t(text_len, latent_t, frame_count, p):
    """Time coordinate for a keyframe anchored at pixel frame p.

    The endpoints reuse stock's exact expressions rather than the general
    formula. They are mathematically identical, but stock accumulates
    latent_t float additions where the general form does one multiply, and
    those differ in the last bits (about 7e-15). Matching stock bit for bit
    means an existing first/last graph builds byte-identical positions
    after this patch is applied, and lets the self-test stay strict.
    """
    if p == 0:
        return float(text_len)
    if frame_count is not None and p == frame_count - 1:
        return float(text_len) + sum(mm._video_t_spans(latent_t)) - mm.FRAME_RESCALE
    return float(text_len) + mm.FRAME_RESCALE * float(p)


def _fixup(layout, text_len, latent_t, frame_count, keyframes, refs=None):
    """Rewrite cond-row time coordinates to the general position formula.

    `refs` is accepted but no longer read for arithmetic: the compensation
    a reference block owes the anchors is now taken from where the target
    actually landed, not recomputed from the block list.
    """
    offset = _target_origin(layout) - float(text_len)
    if offset and any(kf.get(MC_KEY) is None for kf in keyframes):
        # keyframes without MC_KEY are left exactly as stock built them,
        # which means they do NOT get the reference compensation. Mixing
        # them with MC keyframes under a reference would slide the stock
        # anchors relative to ours and to the target. Nothing produces
        # this today; refuse loudly in case something ever does.
        raise RuntimeError(
            "h3_motion_context: stock and motion-context keyframes mixed in "
            "one graph alongside a ref; their coordinates would disagree. "
            "Give every keyframe a %s entry or remove the refs." % MC_KEY)
    cond_spans = [(a, b) for a, b, kind in layout.segments if kind == "cond"]
    if len(cond_spans) != len(keyframes):
        raise RuntimeError(
            "h3_motion_context: expected %d cond segments, layout has %d. "
            "Refusing to rewrite positions."
            % (len(keyframes), len(cond_spans)))
    for (a, b), kf in zip(cond_spans, keyframes):
        p = kf.get(MC_KEY)
        if p is None:
            continue
        layout.position_ids[a:b, 0] = _cond_t(text_len, latent_t, frame_count, p) + offset


def _fixup_audio(layout, text_len, refs):
    """Move the marked audio ref's rows onto the target timeline.

    References and keyframes carry identical row machinery; what makes the
    model read a reference as "a separate clip to imitate" rather than
    "this clip, continued" is that its coordinates sit in a span before
    the target. That distinction decided continuation vs reproduction for
    video, and seam analysis showed the audio reference producing
    phase-unlocked imitation. So: keep the audio on the reference path for
    construction and payload (rows built, latents filled, all stock code
    untouched) and TRANSLATE its time coordinates so the window END lands
    at target frame MC_AUDIO_KEY, the same instant the pinned video ends.

    Translation, not per-row assignment: new = old + shift preserves
    whatever intra-block structure stock built. Stock lays an audio
    reference out channel-major, the same rt coordinates once per stereo
    channel, and a uniform shift keeps that intact without this code
    having to know about it.

    The block keeps its place in the cursor, so the coordinates it vacates
    are left empty. An audio window longer than the video window therefore
    spills backwards into empty space rather than onto the text rows, so
    the collision that made `before` mode fail for video does not arise.

    Other reference blocks are untouched. A Ref2VA graph can carry its own
    image, video and audio references and this moves only the one block
    the node marked, wherever in the list it sits.
    """
    marked = [i for i, r in enumerate(refs or [])
              if r.get(MC_AUDIO_KEY) is not None]
    if len(marked) != 1:
        raise RuntimeError(
            "h3_motion_context: audio timeline placement needs exactly one "
            "reference marked with %s; the layout has %d references and %d "
            "marked. If this appeared during startup, check for more than "
            "one H3 Motion Context folder in custom_nodes."
            % (MC_AUDIO_KEY, len(refs or []), len(marked)))
    idx = marked[0]
    blk = refs[idx]
    if blk.get("kind") != "audio":
        raise RuntimeError(
            "h3_motion_context: %s set on a %r ref; only audio refs can be "
            "moved onto the timeline." % (MC_AUDIO_KEY, blk.get("kind")))
    rt = int(blk.get("ref_audio_t", 0))
    if rt <= 0:
        return

    seg = _ref_segment_map(layout, refs).get(idx, {}).get("ref_audio")
    if seg is None:
        raise RuntimeError(
            "h3_motion_context: the marked audio reference produced no "
            "ref_audio segment. Upstream layout change; refusing to move "
            "rows.")
    a, b = seg
    if b - a != 2 * rt:
        # stock emits exactly rt rows per stereo channel. An exact count
        # rather than a tolerance: if this ever changes, the intra-block
        # structure a translation is preserving has changed with it.
        raise RuntimeError(
            "h3_motion_context: the marked audio reference has %d rows for "
            "%d latent steps, expected %d (stereo, channel-major). Upstream "
            "layout change; refusing to move rows." % (b - a, rt, 2 * rt))

    target_origin = _target_origin(layout)
    slot_start = float(layout.position_ids[a, 0])
    end_frame = float(blk[MC_AUDIO_KEY])
    # window end at target time FRAME_RESCALE * end_frame, width rt steps
    desired_start = target_origin + mm.FRAME_RESCALE * end_frame - float(rt)
    layout.position_ids[a:b, 0] = (layout.position_ids[a:b, 0]
                                   + (desired_start - slot_start))


def _patched_init(self, text_len, latent_t, latent_h, latent_w, audio_t,
                  keyframes=None, refs=None, frame_count=None):
    _orig_init(self, text_len, latent_t, latent_h, latent_w, audio_t,
               keyframes=keyframes, refs=refs, frame_count=frame_count)
    has_mc_kf = bool(keyframes) and any(
        kf.get(MC_KEY) is not None for kf in keyframes)
    has_mc_audio = bool(refs) and any(
        r.get(MC_AUDIO_KEY) is not None for r in refs)
    if has_mc_kf:
        _fixup(self, text_len, latent_t, frame_count, keyframes, refs)
    if has_mc_audio:
        _fixup_audio(self, text_len, refs)
    # neither marked: stock graph, leave it exactly as built


def _self_test():
    """Prove the rewrite reproduces stock positions before committing.

    Builds the two anchors stock code already supports, once the stock way
    and once through our mechanism, and requires the position tensors to
    match exactly. Then exercises the parts stock has no equivalent of:
    interior anchors, reference compensation, the audio move, and the
    same audio move inside a multi-reference Ref2VA layout. If ComfyUI
    changes the position maths or the segment table underneath us this
    fails and the patch is not applied.
    """
    text_len, latent_t, lh, lw, audio_t = 7, 7, 22, 38, 16
    frame_count = sum(mm.FRAME_PER_TOKEN[k % 5] for k in range(latent_t))

    def build(keyframes=None, refs=None, fix=False, move=False):
        lay = mm.PackedLayout.__new__(mm.PackedLayout)
        _orig_init(lay, text_len, latent_t, lh, lw, audio_t,
                   keyframes=keyframes, refs=refs, frame_count=frame_count)
        if fix:
            _fixup(lay, text_len, latent_t, frame_count, keyframes, refs)
        if move:
            _fixup_audio(lay, text_len, refs)
        return lay

    def cond_ts(lay):
        return [float(lay.position_ids[a, 0])
                for a, _, k in lay.segments if k == "cond"]

    # 1. the two anchors stock supports must come out bit-identical
    stock_kf = [{"resolved_frame_index": 0},
                {"resolved_frame_index": frame_count - 1}]
    ours_kf = [{"resolved_frame_index": 0, MC_KEY: 0},
               {"resolved_frame_index": 0, MC_KEY: frame_count - 1}]
    a = build(keyframes=stock_kf)
    b = build(keyframes=ours_kf, fix=True)
    if a.position_ids.shape != b.position_ids.shape:
        raise RuntimeError("position_ids shape mismatch in self-test")
    if not torch.equal(a.position_ids, b.position_ids):
        bad = (a.position_ids != b.position_ids).any(dim=1).nonzero().flatten()
        raise RuntimeError("position mismatch at rows %s" % bad[:8].tolist())

    # 2. a consecutive run lands on strictly increasing coordinates inside
    # the span the two endpoints define
    run = [{"resolved_frame_index": 0, MC_KEY: i} for i in range(4)]
    c = build(keyframes=run, fix=True)
    ts = cond_ts(c)
    if len(ts) != len(run):
        raise RuntimeError("expected %d cond segments, got %d" % (len(run), len(ts)))
    if any(ts[i] >= ts[i + 1] for i in range(len(ts) - 1)):
        raise RuntimeError("consecutive anchors not strictly increasing: %s" % ts)
    t_last = float(text_len) + mm.FRAME_RESCALE * (frame_count - 1)
    if not (ts[0] == float(text_len) and ts[-1] < t_last):
        raise RuntimeError("run %s escapes the [%.4f, %.4f] span"
                           % (ts, float(text_len), t_last))

    # 3. adding a reference must not move the anchors relative to the
    # target. Stock cond rows cannot be the reference here: stock computes
    # them from text_len and never compensates, which is the very bug the
    # compensation exists to fix. The ground truth is the target rows
    # themselves, so the anchor-to-end gap must be identical with and
    # without the reference.
    ref = [{"kind": "audio", "ref_audio_t": 8}]
    d = build(keyframes=run, refs=ref, fix=True)
    ts_ref = cond_ts(d)
    if len(ts_ref) != len(ts):
        raise RuntimeError("cond segment count changed when a ref was added")
    tol = 1e-3
    gap = float(c.position_ids[:, 0].max()) - ts[0]
    gap_ref = float(d.position_ids[:, 0].max()) - ts_ref[0]
    if abs(gap - gap_ref) > tol:
        raise RuntimeError(
            "ref compensation off by %.6f: anchor-to-target gap %.6f without "
            "ref, %.6f with. The target origin read back from the layout no "
            "longer matches its cursor arithmetic." % (gap_ref - gap, gap, gap_ref))
    shifts = [y - x for x, y in zip(ts, ts_ref)]
    if any(abs(sh - shifts[0]) > tol for sh in shifts):
        raise RuntimeError("ref shifted anchors unevenly: %s" % shifts)

    # 4. the audio move: exactly the marked block's rows shift, all by one
    # uniform amount, every other row bit-identical
    end_frame, rt = 4, 8
    ref_mc = [{"kind": "audio", "ref_audio_t": rt, MC_AUDIO_KEY: end_frame}]
    e = build(keyframes=run, refs=ref_mc, fix=True, move=True)
    _check_move(d, e, ref_mc, 0, "single-ref")

    # 5. the same move inside a Ref2VA layout: image, video and audio
    # references of the graph's own must come through untouched. The
    # marked block sits in the MIDDLE of the list, not at the end, because
    # nothing about locating it by segment depends on its position.
    r_lh, r_lw, r_vt = 8, 12, 3
    others = [
        {"kind": "image", "latent_h": r_lh, "latent_w": r_lw},
        {"kind": "video_audio", "latent_h": r_lh, "latent_w": r_lw,
         "latent_t": r_vt, "ref_audio_t": 5},
        {"kind": "audio", "ref_audio_t": 3},
    ]
    marked = {"kind": "audio", "ref_audio_t": rt, MC_AUDIO_KEY: end_frame}
    plain = {"kind": "audio", "ref_audio_t": rt}
    multi_plain = others[:2] + [plain] + others[2:]
    multi_marked = others[:2] + [marked] + others[2:]
    f = build(keyframes=run, refs=multi_plain, fix=True)
    g = build(keyframes=run, refs=multi_marked, fix=True, move=True)
    _check_move(f, g, multi_marked, 2, "multi-ref")

    # 6. the segment map must agree with how the layout is actually laid
    # out. Not by recomputing the cursor, which is the duplication this
    # rewrite exists to remove, but by checking the structural properties
    # the move depends on: reference blocks appear in list order, their
    # rows sit before the target, and no block's rows overlap another's.
    smap = _ref_segment_map(f, multi_plain)
    prev_hi = float(text_len) - 1e-9
    origin = _target_origin(f)
    for i in range(len(multi_plain)):
        spans = smap.get(i)
        if not spans:
            continue
        rows = [r for a0, b0 in spans.values() for r in range(a0, b0)]
        lo = min(float(f.position_ids[r, 0]) for r in rows)
        hi = max(float(f.position_ids[r, 0]) for r in rows)
        if lo < prev_hi - 1e-9:
            raise RuntimeError(
                "reference block %d starts at %.6f, before block %d ended "
                "at %.6f. Reference blocks are not laid out in list order."
                % (i, lo, i - 1, prev_hi))
        if hi >= origin - 1e-9:
            raise RuntimeError(
                "reference block %d reaches %.6f, at or past the target "
                "origin %.6f. Reference rows should sit before the target."
                % (i, hi, origin))
        prev_hi = hi


def _check_move(before, after, refs, idx, label):
    """Only the marked block's rows moved, uniformly, on the time axis."""
    if after.position_ids.shape != before.position_ids.shape:
        raise RuntimeError("%s: audio move changed the layout shape" % label)
    if not torch.equal(before.position_ids[:, 1:], after.position_ids[:, 1:]):
        raise RuntimeError(
            "%s: audio move touched a non-time coordinate column" % label)
    a, b = _ref_segment_map(before, refs)[idx]["ref_audio"]
    expect_moved = set(range(a, b))
    tb, ta = before.position_ids[:, 0], after.position_ids[:, 0]
    moved = set(i for i in range(len(tb)) if float(tb[i]) != float(ta[i]))
    if not moved:
        raise RuntimeError("%s: audio move moved no rows" % label)
    if moved != expect_moved:
        raise RuntimeError(
            "%s: audio move touched the wrong rows: %d moved, %d expected, "
            "e.g. %s" % (label, len(moved), len(expect_moved),
                         sorted(moved ^ expect_moved)[:8]))
    deltas = [float(ta[i]) - float(tb[i]) for i in sorted(moved)]
    if any(abs(dd - deltas[0]) > 1e-9 for dd in deltas):
        raise RuntimeError("%s: audio rows shifted non-uniformly: %s"
                           % (label, deltas[:4]))
    # The size of the shift is deliberately NOT asserted. It depends on
    # how far the reference cursor advanced, which is stock's business,
    # and pinning it here would put a copy of that arithmetic back in.
    # What must hold is where the window ENDS: on the target timeline,
    # FRAME_RESCALE * end_frame past the target origin.
    blk = refs[idx]
    rt = int(blk["ref_audio_t"])
    want_end = (_target_origin(after)
                + mm.FRAME_RESCALE * float(blk[MC_AUDIO_KEY]))
    got_end = float(after.position_ids[a, 0]) + float(rt)
    if abs(got_end - want_end) > 1e-9:
        raise RuntimeError(
            "%s: audio window ends at %.6f, should end at %.6f"
            % (label, got_end, want_end))


setattr(_patched_init, PATCH_MARKER, True)


def _is_supported_solattn_layout_module(init):
    """Whether ``init`` comes from one of the audited SolAttn observers."""
    module = str(getattr(init, "__module__", "") or "")
    if module.endswith(SOLATTN_LAYOUT_MODULE_SUFFIX):
        return True
    normalized = module.replace("\\", "/").rstrip("/")
    tail = normalized.rsplit("/", 1)[-1]
    return (tail == SOLATTN_CUDA_PR117_LAYOUT_MODULE
            or tail.endswith("." + SOLATTN_CUDA_PR117_LAYOUT_MODULE))


def _is_structural_solattn_layout_observer(init):
    """Recognise a Kijai H3 layout observer independently of its folder.

    The observer's fingerprint is a nested ``__init__`` in a module that owns
    two very specific globals -- a callable ``_video_span`` that reads the
    target span off the layout, and a dict ``_SPANS`` that records it by
    ``position_ids`` identity -- and the wrapper's body references both.  It
    must also touch both PackedLayout attributes. Each touch can be a direct
    attribute lookup (``self.segments``, in which case ``segments`` is in
    ``co_names``) or a defensive getattr with a string literal
    (``getattr(self, "segments", [])``, in which case ``"segments"`` is in
    ``co_consts``).  Both patterns appear in the wild -- Kijai's upstream
    ``_morton_h3`` uses direct access, comfy-kitchen PR #117's
    ``sol_attn_minimax_v2`` uses getattr -- and both are equally observe-only.
    Requiring the complete attribute pair keeps the match fail-closed without
    imposing a particular access style.
    """
    if bool(getattr(init, SOLATTN_LAYOUT_OBSERVER_MARKER, False)):
        return True
    code = getattr(init, "__code__", None)
    namespace = getattr(init, "__globals__", None)
    if code is None or code.co_name != "__init__" or not isinstance(
            namespace, dict):
        return False
    # Signature globals: two module-level objects with matching types.  These
    # names are unique to the Kijai observer design; any module that
    # ships both a callable ``_video_span`` and a dict ``_SPANS`` is one of
    # the audited observer variants.
    if not (callable(namespace.get("_video_span"))
            and isinstance(namespace.get("_SPANS"), dict)):
        return False
    names = set(code.co_names)
    # The wrapper must actually reference the signature globals from its body,
    # not merely inherit them from the module.  This rules out unrelated
    # ``__init__`` methods that happen to live in the same module.
    if not SOLATTN_LAYOUT_SIGNATURE_GLOBALS.issubset(names):
        return False
    # And it must touch both PackedLayout attributes we care about.
    # Attribute access puts the name in ``co_names``; ``getattr(self, "name",
    # default)`` puts the name in ``co_consts`` as a string literal.  Accept
    # either route -- a real observer references both under one of the two
    # patterns.
    consts = {c for c in getattr(code, "co_consts", ()) if isinstance(c, str)}
    if not SOLATTN_LAYOUT_ATTRIBUTE_NAMES.issubset(names | consts):
        return False
    return True


def _is_supported_solattn_layout_observer(init):
    return bool(
        _is_supported_solattn_layout_module(init)
        or _is_structural_solattn_layout_observer(init)
    )


def _solattn_wrapped_init(init):
    """Return the constructor captured by Kijai's H3 Morton wrapper.

    The upstream wrapper is a nested ``__init__`` in ``_patch_packed_layout``
    and captures the previous constructor in a closure named
    ``original_init``. Checking that exact closure together with either an
    audited module identity or the read-only span-registration fingerprint is
    intentionally stricter than accepting an arbitrary renamed wrapper.
    """
    if not _is_supported_solattn_layout_observer(init):
        return None
    closure = getattr(init, "__closure__", None) or ()
    freevars = getattr(getattr(init, "__code__", None), "co_freevars", ())
    for name, cell in zip(freevars, closure):
        if name != "original_init":
            continue
        try:
            original = cell.cell_contents
        except ValueError:
            return None
        return original if callable(original) and original is not init else None
    wrapped = getattr(init, "__wrapped__", None)
    if bool(getattr(init, SOLATTN_LAYOUT_OBSERVER_MARKER, False)):
        return wrapped if callable(wrapped) and wrapped is not init else None
    return None


def _unwrap_solattn_layout_observers(init):
    """Return the constructor beneath every audited observer in ``init``.

    Multiple renamed copies can be installed in one process.  They are all
    read-only, so unwrapping the full chain is safe; cycles and excessive
    wrapper depth are treated as unknown rather than guessed through.
    """
    current = init
    wrappers = []
    seen = set()
    for _depth in range(16):
        if id(current) in seen:
            return None, tuple(wrappers)
        seen.add(id(current))
        original = _solattn_wrapped_init(current)
        if original is None:
            return current, tuple(wrappers)
        wrappers.append(current)
        current = original
    return None, tuple(wrappers)


def _replace_solattn_wrapped_init(init, replacement):
    """Replace the constructor captured by the innermost audited observer."""
    current = init
    seen = set()
    for _depth in range(16):
        if id(current) in seen or not _is_supported_solattn_layout_observer(
                current):
            return False
        seen.add(id(current))
        closure = getattr(current, "__closure__", None) or ()
        freevars = getattr(
            getattr(current, "__code__", None), "co_freevars", ())
        for name, cell in zip(freevars, closure):
            if name != "original_init":
                continue
            try:
                original = cell.cell_contents
            except ValueError:
                return False
            if _solattn_wrapped_init(original) is not None:
                current = original
                break
            cell.cell_contents = replacement
            return True
        else:
            return False
    return False


def _uses_native_guide_api(init):
    """Whether a constructor exposes PR #15439's frame_count-free API."""
    original, _wrappers = _unwrap_solattn_layout_observers(init)
    if original is None:
        return False
    if getattr(original, NATIVE_GUIDES_MARKER, False):
        return True
    try:
        parameters = inspect.signature(original).parameters
    except (TypeError, ValueError):
        return False
    return "frame_count" not in parameters


def native_guides_available():
    """Whether ComfyUI provides the merged, core-owned H3 guide API.

    SolAttn may wrap the core constructor as a read-only observer, and an older
    installed copy of this pack may already have wrapped the native API. The
    narrow unwrapping/marker checks in ``_uses_native_guide_api`` recognise
    both cases without changing either implementation.
    """
    cls = getattr(mm, "PackedLayout", None)
    init = getattr(cls, "__init__", None) if cls is not None else None
    if init is None:
        return False
    original, _wrappers = _unwrap_solattn_layout_observers(init)
    if original is None:
        return False
    if (getattr(original, NATIVE_GUIDES_MARKER, False)
            or getattr(original, "__name__", "") == "_patched_init_native"):
        return True
    # A frame_count-free signature is meaningful only on core's own
    # constructor. Arbitrary third-party wrappers often accept *args/**kwargs;
    # treating those as the native API would silently bypass collision checks.
    if hasattr(original, "__wrapped__"):
        return False
    home = str(getattr(cls, "__module__", "") or "")
    where = str(getattr(original, "__module__", "") or "")
    if home and where != home:
        return False
    return _uses_native_guide_api(original)


def compatibility_report():
    """Inspect guide-layout compatibility without installing any patch.

    Preflight and authoring UIs call this before model execution.  It must
    remain read-only: no constructor replacement, self-test, or global owner
    claim is performed here.
    """
    cls = getattr(mm, "PackedLayout", None)
    init = getattr(cls, "__init__", None) if cls is not None else None
    if init is None:
        return {
            "ok": False,
            "mode": "unavailable",
            "owner": "",
            "message": "MiniMax H3 PackedLayout is unavailable.",
            "action": "Install or update a ComfyUI build with MiniMax H3 support.",
        }
    owner = str(getattr(init, "__module__", "") or "")
    if native_guides_available():
        return {
            "ok": True,
            "mode": "native",
            "owner": owner,
            "message": "ComfyUI's native H3 guide API is available.",
            "action": "",
        }
    who = _already_patched()
    if who == "foreign":
        return {
            "ok": False,
            "mode": "conflict",
            "owner": owner,
            "message": (
                "Another extension owns MiniMax H3 PackedLayout with an "
                "unsupported wrapper."),
            "action": (
                "Disable the conflicting H3 layout extension or update "
                "ComfyUI to the native H3 guide API, then restart ComfyUI."),
        }
    if who == "solattn":
        return {
            "ok": True,
            "mode": "legacy_solattn_compatible",
            "owner": owner,
            "message": (
                "The supported CUDA SolAttn layout observer can compose with "
                "the legacy H3 guide bridge."),
            "action": "",
        }
    if who in ("same", "other"):
        return {
            "ok": True,
            "mode": "legacy_active",
            "owner": owner,
            "message": "A compatible legacy H3 guide bridge is already active.",
            "action": "",
        }
    return {
        "ok": True,
        "mode": "legacy_available",
        "owner": owner,
        "message": "The legacy H3 guide bridge can be installed at runtime.",
        "action": "Update ComfyUI to use the native H3 guide API when practical.",
    }


def _already_patched():
    """Has another copy of this file already wrapped the constructor?

    Returns None, "same", "other", "solattn", or "foreign".

    Two copies of this patch in one ComfyUI is normal enough: several
    packs vendor it, and forks of this repo carry their own. Whichever
    loads second would otherwise capture the first's wrapper as its
    original and wrap a wrapper. That is worse than it sounds, because
    each copy self-tests through whatever is already installed, so a copy
    with newer tests gets checked against older behaviour and refuses
    over a limitation that no longer exists.

    Three checks, in decreasing confidence.

    The marker is set by copies new enough to set it. That is a matching
    version and we stand down quietly.

    A wrapper merely NAMED like ours is an older copy of this code, or a
    fork. We stand down and say so, because whichever one loaded first is
    the one deciding what the patch supports.

    SolAttn's narrowly identified Morton wrapper is safe to compose: it only
    observes the layout and records its video span. If it captured a copy of
    this patch, we stand down for that copy; if it captured stock, we install
    outside it so both features remain active.

    Anything else sitting where the stock constructor should be is a
    DIFFERENT pack patching the same thing. Several H3 packs lift the
    same first/last restriction independently, and they cannot both own
    the constructor. Detected by comparing where the function was defined
    against where the class was: stock's __init__ comes from the same
    module as PackedLayout itself, a wrapper comes from somewhere else.
    functools.wraps copies __module__ across, so __wrapped__ is checked
    too. A wrapper that hides both is indistinguishable from stock and
    nothing can be done about that.
    """
    cls = getattr(mm, "PackedLayout", None)
    init = getattr(cls, "__init__", None)
    if init is None:
        return None
    if getattr(init, PATCH_MARKER, False):
        return "same"
    if getattr(init, "__name__", "") in (
            "_patched_init", "_patched_init_native"):
        return "other"
    solattn_original, solattn_wrappers = _unwrap_solattn_layout_observers(init)
    if solattn_wrappers:
        if solattn_original is None:
            return "foreign"
        if getattr(solattn_original, PATCH_MARKER, False):
            return "same"
        if getattr(solattn_original, "__name__", "") in (
                "_patched_init", "_patched_init_native"):
            return "other"
        home = getattr(cls, "__module__", None)
        where = getattr(solattn_original, "__module__", None)
        if home and where and where == home and not hasattr(
                solattn_original, "__wrapped__"):
            return "solattn"
        return "foreign"
    if hasattr(init, "__wrapped__"):
        return "foreign"
    home = getattr(cls, "__module__", None)
    where = getattr(init, "__module__", None)
    if home and where and where != home:
        return "foreign"
    return None


def apply_patch():
    global _orig_init, _applied
    if _applied:
        return True
    current_init = getattr(getattr(mm, "PackedLayout", None), "__init__", None)
    if native_guides_available():
        _LOG.info(
            "h3_motion_context: native H3 guides detected; PackedLayout "
            "remains core-owned")
        return True
    who = _already_patched()
    if who == "foreign":
        _LOG.warning(
            "h3_motion_context: another pack has already patched H3's "
            "layout (PackedLayout.__init__ now comes from %r). Several "
            "ComfyUI packs lift the same first/last keyframe restriction "
            "independently and they cannot both own it, so this one is "
            "refusing rather than wrapping a wrapper and producing joins "
            "neither pack intended. Disable one of them and restart.",
            getattr(getattr(getattr(mm, "PackedLayout", None), "__init__",
                            None), "__module__", "?"))
        return False
    if who and who != "solattn":
        # report success: the patch IS active, just not ours, and the
        # calling pack's nodes check is_applied() before they will run
        _applied = True
        if who == "same":
            _LOG.info(
                "h3_motion_context: interior keyframe anchors already "
                "enabled by another pack, standing down")
        else:
            _LOG.warning(
                "h3_motion_context: the H3 layout patch is already installed "
                "by a DIFFERENT copy of this code (another version, or a "
                "fork). Standing down; that copy decides what the patch "
                "supports, so features added since it may be unavailable. "
                "If you have more than one H3 Motion Context folder in "
                "custom_nodes, keep one and remove the rest. Renaming a "
                "folder does not stop ComfyUI loading it.")
        return True
    if who == "solattn":
        _LOG.info(
            "h3_motion_context: composing with SolAttn's H3 Morton layout "
            "observer; interior anchors and Morton ordering remain enabled")
    if not hasattr(mm, "PackedLayout") or not hasattr(mm, "FRAME_RESCALE"):
        _LOG.warning("h3_motion_context: MiniMax H3 model module missing expected "
                     "attributes, patch not applied")
        return False
    _orig_init = mm.PackedLayout.__init__
    try:
        _self_test()
    except Exception as exc:
        _orig_init = None
        _LOG.warning("h3_motion_context: self-test failed (%s), patch not "
                     "applied. Continuation guide placement unavailable.", exc)
        _LOG.warning(
            "h3_motion_context: if you have more than one H3 Motion Context "
            "folder in custom_nodes (a fork, a backup, a manual clone "
            "alongside a Manager install), that is the usual cause: each "
            "copy self-tests against whichever one loaded first. Keep one "
            "and remove the rest. Renaming a folder does not stop ComfyUI "
            "loading it. Otherwise this is an upstream ComfyUI change and "
            "the message above says what moved.")
        return False
    mm.PackedLayout.__init__ = _patched_init
    _applied = True
    _LOG.info("h3_motion_context: legacy interior keyframe anchors enabled")
    return True


def claim_patch_ownership():
    """Prefer this copy over an older compatible copy of the same patch.

    The operation is explicit because it changes process-global ownership.
    Only wrappers carrying this patch family's marker, or its exact historical
    function names, are replaceable. SolAttn's narrowly recognised observer is
    retained around the new owner; unknown wrappers fail closed.

    Returns ``(ok, detail)`` for the visible Patch Priority pass-through node.
    """
    global _orig_init, _applied
    cls = getattr(mm, "PackedLayout", None)
    current = getattr(cls, "__init__", None) if cls is not None else None
    if current is None or not hasattr(mm, "FRAME_RESCALE"):
        return False, "MiniMax H3 PackedLayout is unavailable"
    if native_guides_available():
        return True, "native guides are core-owned; no patch required"
    if current is _patched_init:
        _applied = True
        return True, "layout owned by this pack"

    who = _already_patched()
    if who in (None, "solattn"):
        # No sibling owns the patch yet. Normal activation already has all
        # compatibility checks needed for stock or SolAttn-first load order.
        _applied = False
        if not apply_patch():
            return False, "layout activation failed its compatibility test"
        return True, ("layout activated by this pack"
                      if who is None else
                      "layout activated while retaining SolAttn")
    if who == "foreign":
        return False, (
            "layout owner is unknown; only another H3 Motion Context copy "
            "can be safely replaced")

    family_wrapper, solattn_wrappers = _unwrap_solattn_layout_observers(
        current)
    if family_wrapper is None:
        return False, "SolAttn observer chain is cyclic or unexpectedly deep"
    owner_module = sys.modules.get(str(getattr(
        family_wrapper, "__module__", "")))
    original = getattr(owner_module, "_orig_init", None)
    if not callable(original) or original is family_wrapper:
        return False, (
            "the existing H3 Motion Context layout wrapper does not expose "
            "its captured constructor; restart with only one copy enabled")

    stock, _original_observers = _unwrap_solattn_layout_observers(original)
    if stock is None:
        return False, "captured SolAttn observer chain is invalid"
    home = str(getattr(cls, "__module__", "") or "")
    where = str(getattr(stock, "__module__", "") or "")
    if (hasattr(stock, "__wrapped__") or (home and where != home)
            or getattr(stock, PATCH_MARKER, False)):
        return False, (
            "the existing layout wrapper captured another unknown wrapper; "
            "refusing to discard it")

    previous_original = _orig_init
    _orig_init = original
    try:
        _self_test()
    except Exception as exc:
        _orig_init = previous_original
        return False, "replacement layout self-test failed: %s" % exc

    replacement = _patched_init
    if solattn_wrappers:
        if not _replace_solattn_wrapped_init(current, replacement):
            _orig_init = previous_original
            return False, "could not preserve SolAttn's layout observer"
    else:
        cls.__init__ = replacement
    _applied = True
    _LOG.info(
        "h3_motion_context: this pack claimed H3 layout ownership from "
        "compatible module %s", getattr(family_wrapper, "__module__", "?"))
    return True, "layout ownership claimed from a compatible older copy"


def is_applied():
    return _applied


def native_guides_active():
    """Backward-compatible alias for native guide capability detection."""
    return native_guides_available()