ComfyUI-MiniMaxH3-Context-Loop/ComfyUI-MAINodes/tests/test_concept_lab_contracts.py

#!/usr/bin/env python3
"""Unit test for concept_lab's contracts, space and verb layer.

  python tests/test_concept_lab_contracts.py
      Synthetic, no GPU, no models, no renders, and no torch: the point of
      the T1 contracts is that they are readable and writable by things
      that are not ComfyUI, and check 1 enforces exactly that in a
      subprocess.

What it checks:

  1. THE EPISTEMIC BOUNDARY. concept_lab.types and concept_lab.space import
     in a subprocess with no torch and no comfy loaded. If this ever fails,
     the contracts have grown a dependency on the runtime they exist to
     outlive.
  2. DETERMINISTIC IDS. The same inputs give the same capture id in a fresh
     process; changing any identity field moves it; changing a
     non-identity field (a note, a timestamp) does NOT, because otherwise
     every bank on disk goes stale the next time someone adds a field.
  3. ROUNDTRIP. Every record type survives to_json/from_json with its
     identity intact, and an unknown field from a newer schema is dropped
     rather than reinterpreted.
  4. REFUSALS. Bad splits, bad arms, bad stores, a sketch tap with no seed,
     and a TapSpec that names both explicit steps and quantiles are all
     refused at construction with a message that names the field.
  5. THE COLLISION CHECK. Writing the same record id with different content
     raises rather than overwriting. This is the capture bank's whole
     safety story: same id, different bytes means the id does not describe
     the condition.
  6. CONFOUNDS. A corpus where every wave clip also smiles reports the
     smile label as `perfect`, and one where they vary independently
     reports `free`. This is the check that runs before GPU time.
  7. EXPORT / IMPORT. A space exports to a bundle, a second space imports
     it, the records land under imported/<origin>/ with origin stamped,
     re-importing is idempotent, and a private pack is held back from the
     export with its reason named.
  8. THE H3 BACKEND'S ANCHORS. The segment vocabulary matches what
     PackedLayout actually builds in the installed ComfyUI, checked by
     reading the source rather than trusting this file. Skipped with a
     printed note if ComfyUI is not on this machine.
  9. UNBUILT VERBS REFUSE. capture_run/factorize/compile raise
     NotYetImplemented naming their task, instead of returning an empty
     result that would get recorded as a measurement.
 10. THE CLI SURFACE. Every command either prints something or exits
     non-zero, and no option shadows a subcommand's dispatch key. Minted
     from a real bug: --group collided with the subparser's own dest, so
     `corpus add --group alice` matched no branch, printed nothing, exited
     0, and dropped the row. A thin surface is exactly where a silent
     no-op hides, so it gets driven rather than trusted.
 11. DEFINITIONS WANT VERSION CONTROL. An index root outside a git work
     tree is reported and warned about by status and the corpus verbs.
     Corpora and plans are definitions, not results: no tensor
     reconstructs which contrasts somebody chose to shoot.
 12. DOCTOR HASHES. Tensor refs are verified byte-for-byte, not merely
     found; a flipped byte reads as hash_mismatch, and a ref that never
     carried a sha256 is a warning rather than a pass.
 13. THE TWO GUARDS THAT MUST FIRE. Split containment (same content hash
     or same source_id in two splits) refuses at plan time, and
     arms_check refuses a control that is not shape-matched, naming the
     key. A key neither row recorded comes back as unverifiable.
"""
import json
import os
import shutil
import subprocess
import sys
import tempfile

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from concept_lab import types as T                                # noqa: E402
from concept_lab import api                                       # noqa: E402
from concept_lab.space import Space, SpaceError                   # noqa: E402

PACK = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
COMFY = os.path.dirname(os.path.dirname(PACK))

_fails = []


def ck(cond, label, detail=""):
    if not cond:
        _fails.append(label)
    print(f"  [{'PASS' if cond else 'FAIL'}] {label}"
          + (f"  {detail}" if detail else ""))


def _stack(**kw):
    base = dict(checkpoint="minimax-h3.safetensors",
                checkpoint_hash="sha256:abc", vae="h3-vae", backend="h3",
                quantization="fp8", comfy_commit="7fe8a613")
    base.update(kw)
    return T.ModelStackFingerprint(**base)


def _capture(**kw):
    src = T.ConditioningSource(kind="image", ref="/tmp/a.png",
                               content_hash="sha256:aaa")
    var = T.ConditionVariant(name="actual_ref", arm="R", sources=(src,))
    tap = T.TapSpec(blocks=(0, 10), step_quantiles=(0.15, 0.85),
                    segments=("video",), store="stats")
    base = dict(stack=_stack(), variant=var, tap=tap, seed=42,
                sampler={"sampler": "euler", "steps": 18, "shift": 12.0})
    base.update(kw)
    return T.FunctionalCaptureManifest(**base)


# ---------------------------------------------------------------- 1
def check_boundary():
    print("\n[1] the epistemic boundary: contracts import with no torch, no comfy")
    prog = (
        "import sys, json;"
        "sys.path.insert(0, %r);"
        "import concept_lab.types as T, concept_lab.space as S;"
        "loaded=[m for m in ('torch','comfy','folder_paths','numpy') "
        "        if m in sys.modules];"
        "print(json.dumps({'loaded': loaded, "
        "                  'proto': T.PROTOCOL_VERSION, "
        "                  'dirs': list(S.DIRS)}))" % PACK)
    r = subprocess.run([sys.executable, "-c", prog], capture_output=True,
                       text=True)
    ok = r.returncode == 0
    ck(ok, "subprocess import succeeded", r.stderr.strip()[:200])
    if not ok:
        return
    got = json.loads(r.stdout.strip().splitlines()[-1])
    ck(got["loaded"] == [], "no torch/comfy/numpy in the process",
       f"loaded={got['loaded']}")
    ck(got["proto"] == T.PROTOCOL_VERSION, "protocol version agrees")


# ---------------------------------------------------------------- 2
def check_ids():
    print("\n[2] deterministic ids, and which fields are allowed to move them")
    a, b = _capture(), _capture()
    ck(a.capture_id == b.capture_id, "same inputs give the same capture id",
       a.capture_id)

    ck(_capture(seed=43).capture_id != a.capture_id, "seed moves the id")
    ck(_capture(stack=_stack(quantization="bf16")).capture_id != a.capture_id,
       "a model-stack change moves the id")
    tap2 = T.TapSpec(blocks=(0, 11), step_quantiles=(0.15, 0.85),
                     segments=("video",), store="stats")
    ck(_capture(tap=tap2).capture_id != a.capture_id, "a tap change moves the id")
    ck(_capture(replicate=1).capture_id != a.capture_id,
       "a replicate index moves the id")

    # The process launch: bit-identical within one interpreter, only close
    # across two, so which launch measured it is a condition (DEC-CL-0015).
    proc = T.process_fingerprint()
    p1 = _capture(process=dict(proc, launch_id="aaaa"))
    p2 = _capture(process=dict(proc, launch_id="bbbb"))
    ck(p1.capture_id != p2.capture_id, "a different process launch moves the id",
       f"{p1.capture_id} vs {p2.capture_id}")
    ck(_capture(process=dict(proc, launch_id="aaaa", pid=1)).capture_id
       == p1.capture_id, "the pid does NOT move the id")
    ck(_capture(process=dict(proc, launch_id="aaaa", host="other")).capture_id
       == p1.capture_id, "the host does NOT move the id")
    ck(proc["launch_id"] == T.process_fingerprint()["launch_id"],
       "the launch id is minted once per interpreter, not once per call",
       proc["launch_id"])

    d = T.FunctionalDeltaManifest(a_capture_id="a", b_capture_id="b")
    ck(d.same_process() is None,
       "a delta with no launch ids answers None, not False")
    ck(T.FunctionalDeltaManifest(a_capture_id="a", b_capture_id="b",
                                 a_process_launch_id="x",
                                 b_process_launch_id="x").same_process()
       is True, "two arms from one launch compare same_process True")
    ck(T.FunctionalDeltaManifest(a_capture_id="a", b_capture_id="b",
                                 a_process_launch_id="x",
                                 b_process_launch_id="y").same_process()
       is False, "arms from two launches compare same_process False")

    # The one that protects every bank on disk:
    ck(_capture(notes="scribbled later").capture_id == a.capture_id,
       "a note does NOT move the id")
    ck(_capture(created="1999-01-01T00:00:00+00:00").capture_id == a.capture_id,
       "a timestamp does NOT move the id")
    ck(_capture(status=T.Status.SCOPED).capture_id == a.capture_id,
       "a status change does NOT move the id")

    # float normalisation: the same condition named two ways is one id
    s1 = _capture(sampler={"shift": 12.0})
    s2 = _capture(sampler={"shift": 12.000000000000002})
    ck(s1.capture_id == s2.capture_id,
       "float noise below 12 significant digits does not fork the id")

    ck(T.digest({"a": 1}) == T.digest({"a": 1}), "digest is stable")
    ck(T.canonical_json({"b": 1, "a": 2}) == '{"a":2,"b":1}',
       "canonical json sorts keys and drops whitespace")


# ---------------------------------------------------------------- 3
def check_roundtrip():
    print("\n[3] roundtrip, and forward compatibility")
    for rec in (_stack(), _capture(),
                T.TapSpec(blocks=(1,), segments=("video",)),
                T.ConceptCorpusRow(
                    source=T.ConditioningSource(kind="video", ref="c.mp4"),
                    labels={"motion": "wave"}, split="validate"),
                T.ConceptPackManifest(name="wave"),
                T.ConceptPlan(name="p", positive={"wave.core": 0.8})):
        back = type(rec).from_json(rec.to_json())
        ck(back.identity() == rec.identity(),
           f"{type(rec).__name__} roundtrips with identity intact")

    d = _capture().to_dict()
    d["a_field_from_the_future"] = 7
    back = T.FunctionalCaptureManifest.from_dict(d)
    ck(back.capture_id == _capture().capture_id,
       "an unknown field from a newer schema is dropped, not reinterpreted")

    ck(type(T.load_record(_stack().to_dict())) is T.ModelStackFingerprint,
       "load_record rebuilds from the _type stamp")


# ---------------------------------------------------------------- 4
def check_refusals():
    print("\n[4] refusals name their field")
    def refuses(fn, label):
        try:
            fn()
        except T.ContractError as e:
            ck(True, label, str(e)[:70])
            return
        except Exception as e:
            ck(False, label, f"wrong exception {type(e).__name__}")
            return
        ck(False, label, "did not refuse")

    refuses(lambda: T.ConceptCorpusRow(
        source=T.ConditioningSource(kind="image"), split="trainset"),
        "an unknown split is refused")
    refuses(lambda: T.ConditionVariant(name="x", arm="maybe"),
            "an unknown arm is refused")
    refuses(lambda: T.TapSpec(store="everything"),
            "an unknown store is refused")
    refuses(lambda: T.TapSpec(store="sketch"),
            "a sketch tap with no dim/seed is refused (unreproducible)")
    refuses(lambda: T.TapSpec(steps=(1,), step_quantiles=(0.5,)),
            "a tap selected two ways at once is refused")
    refuses(lambda: T.ConditioningSource(kind="hologram"),
            "an unknown source kind is refused")
    refuses(lambda: T.ConceptPackManifest(name="x", visibility="public"),
            "an unknown visibility is refused")
    refuses(lambda: T.FunctionalDeltaManifest(a_capture_id="a",
                                              b_capture_id="b",
                                              alignment="rows_probably"),
            "an unknown alignment is refused")


# ---------------------------------------------------------------- 5
def check_collision(root):
    print("\n[5] same id, different bytes is refused")
    s = Space(os.path.join(root, "s5"), os.path.join(root, "t5"))
    s.init(label="five")
    cap = _capture()
    s.put("captures", cap.capture_id, cap)
    s.put("captures", cap.capture_id, cap)
    ck(True, "writing identical content twice is idempotent")
    try:
        s.put("captures", cap.capture_id, _capture(notes="different bytes"))
        ck(False, "same id with different content is refused")
    except SpaceError as e:
        ck("DIFFERENT content" in str(e),
           "same id with different content is refused", str(e)[:60])
        # The refusal has to say which of the two stories it is: a defect
        # within a launch, or a capture that forgot its process fingerprint.
        ck("DEC-CL-0015" in str(e),
           "the refusal points at the cross-launch case too",
           str(e)[str(e).find("Same id"):][:120])


# ---------------------------------------------------------------- 6
def check_confounds():
    print("\n[6] confounds: the check that runs before GPU time")
    def row(subj, motion, face, split="extract"):
        return T.ConceptCorpusRow(
            source=T.ConditioningSource(kind="video",
                                        ref=f"{subj}_{motion}_{face}.mp4"),
            labels={"subject": subj, "motion": motion, "face": face},
            split=split, matched_group=subj)

    # every wave also smiles: motion and face are in bijection
    bad = T.ConceptCorpus(name="bad", rows=tuple(
        r.to_dict() for r in (row("alice", "wave", "smile"),
                              row("bob", "wave", "smile"),
                              row("alice", "neutral", "flat"),
                              row("bob", "neutral", "flat"))))
    rep = bad.confounds("motion")
    ck(rep["labels"]["face"]["verdict"] == "perfect",
       "a face label that tracks the motion exactly reads 'perfect'",
       json.dumps(rep["labels"]["face"]))
    ck(rep["labels"]["subject"]["verdict"] == "free",
       "a subject label that varies within each motion reads 'free'")

    good = T.ConceptCorpus(name="good", rows=tuple(
        r.to_dict() for r in (row("alice", "wave", "smile"),
                              row("bob", "wave", "flat"),
                              row("alice", "neutral", "flat"),
                              row("bob", "neutral", "smile"))))
    ck(good.confounds("motion")["labels"]["face"]["verdict"] == "free",
       "breaking the correlation clears the confound")

    ck("error" in bad.confounds("nosuchlabel"),
       "a target label nobody carries reports an error, not an empty verdict")


# ---------------------------------------------------------------- 7
def check_export_import(root):
    print("\n[7] export and import keep provenance straight")
    a_idx, a_ten = os.path.join(root, "A"), os.path.join(root, "At")
    b_idx, b_ten = os.path.join(root, "B"), os.path.join(root, "Bt")
    A = Space(a_idx, a_ten); A.init(label="mine")
    B = Space(b_idx, b_ten); B.init(label="theirs")
    ck(A.space_id != B.space_id, "two spaces get different ids")

    api.corpus_new("wave", index_root=a_idx, tensor_root=a_ten)
    api.corpus_add("wave", "video", "/nonexistent/clip.mp4",
                   labels={"motion": "wave"}, split="extract",
                   index_root=a_idx, tensor_root=a_ten)
    shareable = T.ConceptPackManifest(name="open", visibility="shareable")
    private = T.ConceptPackManifest(name="likeness", visibility="private")
    A.put("packs", shareable.pack_id, shareable)
    A.put("packs", private.pack_id, private)

    bundle = os.path.join(root, "out")
    res = api.space_export(bundle, index_root=a_idx, tensor_root=a_ten)
    kinds = {(i["kind"], i["id"]) for i in res["included"]}
    ck(("corpora", "wave") in kinds, "the corpus is in the bundle")
    ck(("packs", shareable.pack_id) in kinds, "a shareable pack is exported")
    held = [e for e in res["excluded"] if e["id"] == private.pack_id]
    ck(len(held) == 1 and "private" in held[0]["reason"],
       "a private pack is held back WITH its reason", held[0]["reason"]
       if held else "")

    got = api.space_import(res["bundle"], index_root=b_idx, tensor_root=b_ten)
    ck(got["origin_space_id"] == A.space_id, "the bundle names its origin")
    landed = B.list("corpora")
    ck(len(landed) == 1 and landed[0]["origin"] == A.space_id
       and not landed[0]["local"],
       "imported records are tagged with the origin space and are not local")
    ck(os.path.isdir(os.path.join(b_idx, "imported", A.space_id)),
       "imported records live in their own namespace")
    stamped = json.load(open(landed[0]["path"]))
    ck(stamped.get("_imported_from") == A.space_id,
       "the record itself carries where it came from")

    again = api.space_import(res["bundle"], index_root=b_idx,
                             tensor_root=b_ten)
    ck(not again["refused"], "re-importing the same bundle is idempotent")

    try:
        api.space_import(res["bundle"], index_root=a_idx, tensor_root=a_ten)
        ck(False, "importing your own export back into itself is refused")
    except SpaceError as e:
        ck("THIS space" in str(e),
           "importing your own export back into itself is refused",
           str(e)[:60])

    res2 = api.space_export(os.path.join(root, "out2"), include_private=True,
                            index_root=a_idx, tensor_root=a_ten)
    ids = {i["id"] for i in res2["included"]}
    ck(private.pack_id in ids, "include_private does include it, when asked")


# ---------------------------------------------------------------- 8
def check_h3_anchors():
    print("\n[8] the H3 backend's anchors against the installed ComfyUI")
    from concept_lab.backends import get_backend
    from concept_lab.backends.base import BackendError
    be = get_backend("h3")

    src = os.path.join(COMFY, "comfy", "ldm", "minimax", "model.py")
    if not os.path.isfile(src):
        print(f"  [SKIP] no ComfyUI at {src}")
    else:
        text = open(src, encoding="utf-8").read()
        missing = [s for s in be.segment_vocabulary()
                   if f'"{s}"' not in text]
        ck(not missing,
           "every declared segment name appears in PackedLayout's source",
           f"missing={missing}")
        ck('patches_replace' in text and '"double_block"' in text,
           "the block-replace seam is still present upstream")
        ck("num_layers=50" in text and "hidden_size=5376" in text,
           "the 50-block / 5376-width defaults still hold",
           "if this fails the capability report is stale, not the code")

    try:
        be.validate_tap(T.TapSpec(blocks=(0,), segments=("nonsense",)))
        ck(False, "a tap naming an unknown segment is refused")
    except BackendError as e:
        ck("nonsense" in str(e), "a tap naming an unknown segment is refused",
           str(e)[:60])
    be.validate_tap(T.TapSpec(blocks=(0, 49), segments=("video", "ref_img")))
    ck(True, "a legal tap validates")

    arms = be.plan_arms(T.ConditioningSource(kind="video", ref="c.mp4"))
    ck("frames" in arms["arm_N"]["must_match"],
       "a video control must match frame count, not only dimensions")
    ck(len(arms["arm_N"]["strata_to_try"]) >= 3,
       "several control strata are offered, so R-N is not one arbitrary choice")


# ---------------------------------------------------------------- 9
def check_unbuilt_refuse(root):
    print("\n[9] unbuilt verbs refuse instead of returning nothing")
    for name in ("capture_run", "delta_build", "factorize", "pack_compile"):
        try:
            getattr(api, name)()
            ck(False, f"{name} refuses")
        except api.NotYetImplemented as e:
            ck(bool(e.task), f"{name} refuses and names its task",
               f"blocked on {e.task}")

    idx, ten = os.path.join(root, "P"), os.path.join(root, "Pt")
    api.space_init(index_root=idx, tensor_root=ten, label="plan")
    api.corpus_new("w", index_root=idx, tensor_root=ten)
    for subj in ("alice", "bob"):
        for m in ("wave", "neutral"):
            api.corpus_add("w", "video", f"/none/{subj}_{m}.mp4",
                           labels={"subject": subj, "motion": m},
                           matched_group=subj, index_root=idx, tensor_root=ten)
    plan = api.capture_plan("w", index_root=idx, tensor_root=ten)
    ck(plan["forwards"] == 4 * 3, "plan costs rows x arms forwards",
       f"forwards={plan['forwards']}")
    ck(plan["tap_points"] > 0, "plan reports what the capture would write",
       f"tap_points={plan['tap_points']}")
    ck(any("VALIDATE" in w for w in plan["warnings"]),
       "the plan carries the corpus's protocol gaps")
    ck("no measurement" in plan["note"] or "Nothing here is a measurement"
       in plan["note"], "the plan says plainly that it measured nothing")

    # Forwards are the unit everyone reaches for and the wrong one on H3.
    ck(plan["cost"]["cost_model"]["step_cost_exponent_range"] == [1.5, 1.7],
       "the plan prices sequence, not forwards",
       json.dumps(plan["cost"]["cost_model"]["step_cost_exponent_range"]))
    ck(plan["process_launch_id"] == T.process_fingerprint()["launch_id"],
       "the plan stamps the launch it was planned in")
    ck(any("ONE process launch" in n for n in plan["protocol_notes"]),
       "the plan states the one-launch constraint on the arms")

    p2 = api.capture_plan("w", arms=("R", "0"), index_root=idx,
                          tensor_root=ten)
    ck(any("shape-matched" in w for w in p2["warnings"]),
       "dropping the control arm earns a warning about the confound")


# --------------------------------------------------------------- 10
def check_cli(root):
    """Every CLI command produces output or refuses. Nothing exits 0 silently.

    Minted from a real bug: `--group` was declared on the corpus subparser
    while the subcommand itself parsed into dest "group", so argparse wrote
    the option's value over the dispatch key and `corpus add --group alice`
    matched no branch, printed nothing, and exited 0. Rows vanished without
    a single error. A thin surface is exactly where that hides, so the
    surface gets driven here rather than trusted.
    """
    print("\n[10] the CLI surface: no command exits 0 in silence")
    from concept_lab import cli

    idx, ten = os.path.join(root, "C"), os.path.join(root, "Ct")
    base = ["--index-root", idx, "--tensor-root", ten]
    calls = []
    orig = cli._out
    cli._out = lambda o: calls.append(o)
    try:
        cmds = [
            (["space", "init", "--label", "cli"], 0),
            (["space", "status"], 0),
            (["space", "doctor"], 0),
            (["space", "list", "corpora"], 0),
            (["corpus", "new", "w"], 0),
            (["corpus", "add", "w", "--kind", "video", "--ref", "/none/a.mp4",
              "--label", "motion=wave", "--label", "face=smile",
              "--split", "extract", "--group", "alice"], 0),
            (["corpus", "add", "w", "--kind", "video", "--ref", "/none/b.mp4",
              "--label", "motion=neutral", "--label", "face=flat",
              "--split", "validate", "--group", "alice"], 0),
            (["corpus", "status", "w"], 0),
            (["corpus", "confounds", "w", "--target", "motion"], 0),
            (["capture", "plan", "w"], 0),
            (["pack", "list"], 0),
        ]
        for argv, want_rc in cmds:
            before = len(calls)
            rc = cli.main(base + argv)
            label = " ".join(argv[:3])
            ck(rc == want_rc and len(calls) > before,
               f"`{label}` returned {want_rc} AND printed something",
               f"rc={rc} outputs={len(calls) - before}")

        # the flag whose collision caused the bug actually reaches the corpus
        st = [c for c in calls if c.get("matched_groups") is not None][-1]
        ck(st["matched_groups"].get("alice") == 2,
           "--group reaches the corpus and both rows land in the group",
           json.dumps(st["matched_groups"]))
        ck(st["n_rows"] == 2, "both CLI-added rows persisted",
           f"n_rows={st['n_rows']}")

        rc = cli.main(base + ["capture", "run"])
        ck(rc == 3, "an unbuilt verb exits 3, not 0", f"rc={rc}")
    finally:
        cli._out = orig

    # Structural guard for the whole class, not just the one instance: no
    # option anywhere may write the namespace key a subparser dispatches on.
    import argparse

    dispatch_keys, collisions = set(), []

    def walk(parser, path):
        for act in parser._actions:
            if isinstance(act, argparse._SubParsersAction):
                dispatch_keys.add(act.dest)
                for name, sp in act.choices.items():
                    walk(sp, path + [name])
            elif act.dest in dispatch_keys and act.option_strings:
                collisions.append((" ".join(path), act.option_strings,
                                   act.dest))

    walk(cli.build_parser(), [])
    ck(not collisions,
       "no option shadows a subcommand dispatch key",
       f"collisions={collisions}" if collisions else
       f"dispatch keys guarded: {sorted(dispatch_keys)}")


# --------------------------------------------------------------- 11
def check_version_control(root):
    """Definitions are not results, and only one of them is reconstructible.

    Captures and deltas can be re-measured from a corpus plus a stack.
    Corpora, tap specs and plans cannot be re-derived from anything: they
    are the record of which contrasts somebody chose. The ancestor program
    lost exactly those, under a gitignored output dir, behind four findings.
    """
    print("\n[11] definitions want version control, and the space says so")
    idx, ten = os.path.join(root, "V"), os.path.join(root, "Vt")
    api.space_init(index_root=idx, tensor_root=ten, label="vc")

    st = api.space_status(index_root=idx, tensor_root=ten)
    ck(st["index_root_tracked"] is False,
       "an index root outside a git work tree reports untracked")
    ck(any("DEC-CL-0016" in w for w in st["warnings"]),
       "status warns, citing the decision", st["warnings"][0][:70]
       if st["warnings"] else "")

    made = api.corpus_new("defs", index_root=idx, tensor_root=ten)
    ck(any("H3_CONCEPT_INDEX_DIR" in w for w in made["warnings"]),
       "corpus new carries the warning with the fix in it")
    added = api.corpus_add("defs", "video", "/none/x.mp4",
                           labels={"motion": "wave"},
                           index_root=idx, tensor_root=ten)
    ck(any("DEC-CL-0016" in w for w in added["warnings"]),
       "corpus add carries it too")

    os.makedirs(os.path.join(idx, ".git"), exist_ok=True)
    st2 = api.space_status(index_root=idx, tensor_root=ten)
    ck(st2["index_root_tracked"] is True,
       "the same root inside a git work tree reports tracked")
    ck(not st2["warnings"], "and stops warning", f"warnings={st2['warnings']}")
    clean = api.corpus_new("defs2", index_root=idx, tensor_root=ten)
    ck(not clean["warnings"], "the corpus verbs stop too")


# --------------------------------------------------------------- 12
def check_doctor_hashes(root):
    """Presence is not integrity. A truncated tensor is present."""
    print("\n[12] doctor hashes the tensors it can, and says which it cannot")
    idx, ten = os.path.join(root, "D"), os.path.join(root, "Dt")
    s = Space(idx, ten)
    s.init(label="doc")
    tdir = s.tensor_dir()
    os.makedirs(tdir, exist_ok=True)

    rel = "video_b0_s0.bin"
    p = os.path.join(tdir, rel)
    with open(p, "wb") as fh:
        fh.write(b"pretend this is a block state" * 8)
    good = T.TensorRef(key="video/b0", path=rel, sha256=T.file_sha256(p),
                       segment="video", block=0)
    cap = _capture(tensors=(good,))
    s.put("captures", cap.capture_id, cap)
    d = s.doctor()
    ck(d["ok"] and not d["problems"], "a tensor whose bytes match is clean",
       json.dumps(d["problems"])[:80])

    rel2 = "video_b1_s0.bin"
    p2 = os.path.join(tdir, rel2)
    with open(p2, "wb") as fh:
        fh.write(b"unhashed")
    cap2 = _capture(seed=99, tensors=(T.TensorRef(key="video/b1", path=rel2),))
    s.put("captures", cap2.capture_id, cap2)
    d = s.doctor()
    ck(d["ok"], "an unhashed ref does not fail the space")
    ck(any(w["kind"] == "unhashed_tensor" and w["detail"] == p2
           for w in d["warnings"]),
       "an unhashed ref is a warning naming the file",
       json.dumps(d["warnings"])[:90])

    with open(p, "r+b") as fh:            # one byte, the classic corruption
        fh.seek(3)
        fh.write(b"X")
    d = s.doctor()
    bad = [x for x in d["problems"] if x["kind"] == "hash_mismatch"]
    ck(not d["ok"] and len(bad) == 1 and bad[0]["detail"] == p,
       "one flipped byte is reported as hash_mismatch naming the path",
       bad[0]["why"][:80] if bad else json.dumps(d["problems"])[:80])


# --------------------------------------------------------------- 13
def check_containment_and_arms(root):
    """Two guards that have to FIRE, not merely exist.

    A held-out score over a row the extraction set already contains is an
    extraction score, and a control that is not shape-matched turns R-N
    from semantic content into content plus layout.
    """
    print("\n[13] split containment refuses, and the control arm is asserted")
    idx, ten = os.path.join(root, "K"), os.path.join(root, "Kt")
    api.space_init(index_root=idx, tensor_root=ten, label="contain")
    r = {"index_root": idx, "tensor_root": ten}

    asset = os.path.join(root, "shared_clip.mp4")
    with open(asset, "wb") as fh:
        fh.write(b"same bytes in two splits")

    api.corpus_new("leaky", **r)
    api.corpus_add("leaky", "video", asset, labels={"motion": "wave"},
                   split="extract", **r)
    api.corpus_add("leaky", "video", asset, labels={"motion": "wave"},
                   split="validate", **r)
    h = T.file_sha256(asset)
    st = api.corpus_status("leaky", **r)
    ck(st["containment"]["content_hash"]
       and st["containment"]["content_hash"][0]["value"] == h,
       "corpus status lists the colliding content hash",
       json.dumps(st["containment"]["content_hash"])[:90])
    ck(any(w.startswith("CONTAINMENT") for w in st["warnings"]),
       "and warns about it in words")
    try:
        api.capture_plan("leaky", **r)
        ck(False, "capture_plan refuses a corpus that leaks a content hash")
    except T.ContractError as e:
        ck(h in str(e) and "extract" in str(e),
           "capture_plan refuses a corpus that leaks a content hash",
           str(e)[:80])
    forced = api.capture_plan("leaky", allow_containment=True, **r)
    ck(len(forced["containment_overridden"]) == 1,
       "an override is allowed and lands in the plan as a record")

    api.corpus_new("lineage", **r)
    api.corpus_add("lineage", "video", "/none/take7_a.mp4",
                   source_id="take7", split="extract", **r)
    api.corpus_add("lineage", "video", "/none/take7_b.mp4",
                   source_id="take7", split="validate", **r)
    try:
        api.capture_plan("lineage", **r)
        ck(False, "different bytes from one take still leak, and refuse")
    except T.ContractError as e:
        ck("take7" in str(e),
           "different bytes from one take still leak, and refuse",
           str(e)[:80])

    api.corpus_new("clean", **r)
    pre = {"width": 1024, "height": 576, "frames": 49, "fps": 24}
    rr = api.corpus_add("clean", "video", "/none/alice_wave.mp4",
                        source_id="take1", split="extract", preprocess=pre,
                        **r)
    nn = api.corpus_add("clean", "video", "/none/control.mp4",
                        source_id="take2", split="extract", preprocess=pre,
                        **r)
    bad = api.corpus_add("clean", "video", "/none/control_wide.mp4",
                         source_id="take3", split="extract",
                         preprocess=dict(pre, width=1280), **r)
    api.corpus_add("clean", "video", "/none/bob_neutral.mp4",
                   source_id="take4", split="validate", **r)
    plan = api.capture_plan("clean", **r)
    ck(plan["forwards"] == 4 * 3, "a clean corpus plans without refusing",
       f"forwards={plan['forwards']}")

    ok = api.arms_check("clean", rr["row_id"], nn["row_id"], **r)
    ck(ok["ok"] and {c["key"] for c in ok["checked"]} >=
       {"width", "height", "frames", "fps", "modality"},
       "a shape-matched control passes on every recorded key",
       json.dumps([c["key"] for c in ok["checked"]]))
    ck("reference_count" in ok["unverifiable"],
       "a key neither row recorded is unverifiable, not a pass",
       json.dumps(ok["unverifiable"]))
    try:
        api.arms_check("clean", rr["row_id"], bad["row_id"], **r)
        ck(False, "a control with the wrong width is refused")
    except T.ContractError as e:
        ck("width" in str(e) and "1280" in str(e),
           "a control with the wrong width is refused, naming the key",
           str(e)[:80])


def main():
    root = tempfile.mkdtemp(prefix="concept_lab_test_")
    try:
        print("=== concept_lab contracts (synthetic, no GPU, no torch) ===")
        check_boundary()
        check_ids()
        check_roundtrip()
        check_refusals()
        check_collision(root)
        check_confounds()
        check_export_import(root)
        check_h3_anchors()
        check_unbuilt_refuse(root)
        check_cli(root)
        check_version_control(root)
        check_doctor_hashes(root)
        check_containment_and_arms(root)
    finally:
        shutil.rmtree(root, ignore_errors=True)

    print(f"\n=== {'ALL PASS' if not _fails else str(len(_fails)) + ' FAILED'} ===")
    for f in _fails:
        print(f"  FAILED: {f}")
    return 0 if not _fails else 1


if __name__ == "__main__":
    sys.exit(main())