feat: block-influence recording is on by default; four-group comparison with shuffled-label control, and tools/taste_groups.py to read it

Co-Authored-By: Claude Haiku 5.5 <noreply@anthropic.com>

conftest.py

@@ -168,6 +168,16 @@ def _quarantine_in_tmp(tmp_path, monkeypatch):
     monkeypatch.setattr(config, "QUARANTINE_FILE", tmp_path / "quarantine.json")
 
 
+@pytest.fixture(autouse=True)
+def _block_influence_off_in_tmp(tmp_path, monkeypatch):
+    """Recording is on by default, so a test that rates a clip would also pick up its block-influence
+    capture. Off here; a test that wants it on turns it on itself."""
+    from modules.sampling.block_influence import measure
+    switch = tmp_path / "block_influence.enabled"
+    switch.write_text("0")
+    monkeypatch.setattr(measure, "SWITCH", switch)
+
+
 class TinyLTX:
     """A REAL LTX audio+video model -- ComfyUI's own LTXAVModel, four blocks wide as a thumb.
 

modules/sampling/block_influence/measure.py

@@ -14,10 +14,11 @@ problem = None
 
 
 def enabled() -> bool:
+    # On until someone turns it off: with no switch file yet (a fresh rental) recording runs.
     try:
-        return SWITCH.read_text().strip() == "1"
+        return SWITCH.read_text().strip() != "0"
     except OSError:
-        return False
+        return True
 
 
 def set_enabled(on: bool) -> bool:
@@ -142,6 +143,68 @@ def profile(rows) -> dict:
             "flatness": flat, "n_liked": int(liked.sum()), "n_disliked": int(disliked.sum())}
 
 
+GROUPS = ("liked", "image", "composition", "both")
+
+
+def _group(row):
+    """liked; or a dislike sorted by its axis: image, composition, or both (plain dislike)."""
+    reward = float(row["reward"])
+    if reward > 0:
+        return "liked"
+    if reward == 0:
+        return None
+    return row.get("axis") if row.get("axis") in ("image", "composition") else "both"
+
+
+def groups(rows, shuffles=200, seed=0) -> dict:
+    """Do the four groups point in different directions over the block profile?
+
+    For each dislike group, `d = mean(group) - mean(liked)`; the answer is the cosine between
+    those directions. The control reshuffles the group labels over the same clips (same group
+    sizes) and asks how often that cosine comes out as high. A real difference should sit
+    above the shuffled range; inside it, the ratings are not telling the groups apart yet.
+    """
+    rows = [r for r in rows if "ratio" in r["rows"]]
+    if not rows:
+        return {"used": 0, "counts": {}, "cos": {}, "note": "nothing recorded yet"}
+    n = len(rows[-1]["rows"]["ratio"])
+    rows = [r for r in rows if len(r["rows"]["ratio"]) == n]
+    stack = torch.stack([r["rows"]["ratio"].float() for r in rows])
+    stack = torch.where(torch.isfinite(stack), stack, torch.nanmean(stack, dim=0).nan_to_num(0.0))
+    labels = [_group(r) for r in rows]
+    counts = {g: labels.count(g) for g in GROUPS}
+
+    def directions(labs):
+        means = {}
+        for g in GROUPS:
+            idx = [i for i, x in enumerate(labs) if x == g]
+            if len(idx) >= MIN_PER_GROUP:
+                means[g] = stack[idx].mean(dim=0)
+        if "liked" not in means:
+            return {}
+        return {g: means[g] - means["liked"] for g in ("image", "composition", "both") if g in means}
+
+    def cosines(d):
+        pairs = {}
+        names = sorted(d)
+        for i, a in enumerate(names):
+            for b in names[i + 1:]:
+                pairs[f"{a} vs {b}"] = float(torch.nn.functional.cosine_similarity(d[a], d[b], dim=0))
+        return pairs
+
+    observed = cosines(directions(labels))
+    if not observed:
+        return {"used": len(rows), "counts": counts, "cos": {}, "note": "not enough clips in each group yet"}
+    gen = torch.Generator().manual_seed(seed)
+    null = {k: [] for k in observed}
+    for _ in range(shuffles):
+        perm = torch.randperm(len(labels), generator=gen).tolist()
+        for k, v in cosines(directions([labels[i] for i in perm])).items():
+            null[k].append(v)
+    chance = {k: {"mean": sum(v) / len(v), "p95": sorted(v)[int(0.95 * (len(v) - 1))]} for k, v in null.items() if v}
+    return {"used": len(rows), "counts": counts, "cos": observed, "chance": chance}
+
+
 def _kind(key):
     kind = registry.current().ask("taste_kind", key, KIND)
     if kind is None:

modules/sampling/block_influence/tests/test_block_influence.py

@@ -26,9 +26,11 @@ def _load(tiny, key="fox"):
     return FunPackLoadModifiers.execute(tiny.patcher, settings).result
 
 
-def test_off_by_default_it_stores_nothing_changes_nothing_and_says_so(tiny_h3):
+def test_switched_off_it_stores_nothing_changes_nothing_and_says_so(tiny_h3):
     from core import log
+    from modules.sampling.block_influence import measure
     from modules.system.taste import store
+    measure.set_enabled(False)
     base = tiny_h3.run()
     patched, _ = _load(tiny_h3)
     log.new_run()
@@ -199,6 +201,7 @@ def test_a_bad_key_is_refused_before_the_switch_flips_and_a_placeholder_finds_th
         async def json(self):
             return self._body
 
+    measure.set_enabled(False)
     out = asyncio.run(handlers[("POST", "/b/enabled")](Req({"enabled": True, "key": "no/slash"})))
     assert out.status == 400 and not measure.enabled()
     patched, _ = _load(tiny_h3, key="mystyle")

modules/sampling/block_influence/tests/test_groups.py

@@ -0,0 +1,46 @@
+"""The four-group comparison and the recording default, on synthetic rows (no model needed)."""
+
+import torch
+
+from modules.sampling.block_influence import measure
+
+
+def _row(reward, axis, profile):
+    return {"reward": reward, "axis": axis, "rows": {"ratio": torch.tensor(profile, dtype=torch.float32)}}
+
+
+def test_recording_is_on_when_no_switch_file_exists(tmp_path, monkeypatch):
+    monkeypatch.setattr(measure, "SWITCH", tmp_path / "switch")
+    assert measure.enabled() is True
+    measure.set_enabled(False)
+    assert measure.enabled() is False
+    measure.set_enabled(True)
+    assert measure.enabled() is True
+
+
+def test_groups_sort_dislikes_by_axis():
+    assert measure._group(_row(1.0, None, [0])) == "liked"
+    assert measure._group(_row(-1.0, "image", [0])) == "image"
+    assert measure._group(_row(-1.0, "composition", [0])) == "composition"
+    assert measure._group(_row(-1.0, None, [0])) == "both"
+    assert measure._group(_row(0.0, None, [0])) is None
+
+
+def test_opposite_directions_show_as_opposed_cosines():
+    rows = [_row(1, None, [1, 0, 0]) for _ in range(3)] + [_row(-1, "image", [0, 1, 0]) for _ in range(3)] \
+         + [_row(-1, "composition", [0, 0, 1]) for _ in range(3)] + [_row(-1, None, [0, 0, 0]) for _ in range(3)]
+    out = measure.groups(rows, shuffles=50)
+    assert out["counts"] == {"liked": 3, "image": 3, "composition": 3, "both": 3}
+    # Each dislike is "liked minus a different block": the two directions share the "away from liked"
+    # part (the -1 on block 0), so they agree half way, not fully and not at all.
+    assert abs(out["cos"]["composition vs image"] - 0.5) < 1e-5
+    assert set(out["chance"]) == set(out["cos"])
+
+
+def test_too_few_clips_says_so_instead_of_comparing():
+    out = measure.groups([_row(1, None, [1, 0]), _row(-1, "image", [0, 1])], shuffles=5)
+    assert out["cos"] == {} and "not enough" in out["note"]
+
+
+def test_nothing_recorded_is_said():
+    assert measure.groups([])["note"] == "nothing recorded yet"

tools/taste_groups.py

@@ -0,0 +1,33 @@
+"""Print the four-group comparison for a Taste key's block-influence recordings.
+
+    python tools/taste_groups.py <taste key>
+
+Reads the rows the recording wrote (liked, bad image, bad composition, both bad) and prints how
+the group directions line up, next to what shuffled labels give by chance. Reads only; changes nothing.
+"""
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+
+def main(key: str) -> None:
+    from modules.sampling.block_influence import measure
+    from modules.system.taste import store
+
+    rows = store.load(key, measure.KIND)["rows"]
+    out = measure.groups(rows)
+    print(f"key {key!r}: {out['used']} clips rated")
+    print("counts:", out.get("counts", {}))
+    if out.get("note"):
+        print(out["note"])
+    for pair, cos in out.get("cos", {}).items():
+        c = out["chance"][pair]
+        print(f"{pair}: {cos:+.3f}   (chance: mean {c['mean']:+.3f}, 95th {c['p95']:+.3f})")
+
+
+if __name__ == "__main__":
+    if len(sys.argv) != 2:
+        sys.exit(__doc__)
+    main(sys.argv[1])