From 1692492eab1712fd1d4e623b3f67064355a373a7 Mon Sep 17 00:00:00 2001 From: Jason Hoku Date: Tue, 3 Nov 2026 23:56:28 -0800 Subject: [PATCH] feat(manifest): comparisons storage + save/clear routes Pairwise-review votes live in manifest["comparisons"]. Locked read-modify-write helper, disk-wins preservation in every manifest writer, two new routes, tests. Co-Authored-By: Claude Fable 5.1 --- __init__.py | 79 ++++++++++++ image_generation.py | 4 + manifest_utils.py | 126 ++++++++++++++++++ remote_vae.py | 4 + tests/conftest.py | 2 + tests/test_comparisons_manifest.py | 197 +++++++++++++++++++++++++++++ 6 files changed, 412 insertions(+) create mode 100644 tests/test_comparisons_manifest.py diff --git a/__init__.py b/__init__.py index 2a6478c..1ac5a8e 100644 --- a/__init__.py +++ b/__init__.py @@ -409,6 +409,81 @@ async def save_changes(request): traceback.print_exc() return web.Response(status=500, text=str(e)) +# ============================================================================= +# API: PAIRWISE REVIEW VOTES (manifest "comparisons") +# ============================================================================= + +def _comparisons_manifest_path(session_name): + """Resolve a sanitized session name to its manifest path; returns (path, error_response).""" + if session_name: + session_name = re.sub(r'[^\w\-]', '', str(session_name)) + if not session_name: + return None, web.Response(status=400, text="Missing session_name") + base_dir = os.path.join(folder_paths.get_output_directory(), "benchmarks", session_name) + if not _is_path_within(base_dir, _get_benchmarks_base()): + return None, web.Response(status=403, text="Forbidden: path outside benchmarks directory") + manifest_path = os.path.join(base_dir, "manifest.json") + if not os.path.exists(manifest_path): + return None, web.Response(status=404, text=f"Session '{session_name}' not found") + return manifest_path, None + + +@server.PromptServer.instance.routes.post("/config_tester/save_comparisons") +async def save_comparisons(request): + """ + Append / remove blind-compare votes in manifest["comparisons"]. + Body: {session_name, append: [{a,b,winner,ts,mode,blind?}], remove: ["ts:a:b", ...]} + Only the comparisons list is touched; items/meta are written back as read. + """ + try: + from .manifest_utils import update_manifest_comparisons + data = await request.json() + append = data.get("append") or [] + remove = data.get("remove") or [] + if not isinstance(append, list) or not isinstance(remove, list): + return web.Response(status=400, text="append and remove must be lists") + if (not append and not remove) or len(append) > 5000 or len(remove) > 5000: + return web.Response(status=400, text="Nothing to save or too many votes") + manifest_path, err = _comparisons_manifest_path(data.get("session_name")) + if err: + return err + try: + result = await asyncio.to_thread( + update_manifest_comparisons, manifest_path, append, + [str(k) for k in remove]) + except FileNotFoundError: + return web.Response(status=404, text="Session manifest not found") + except ValueError: # includes json.JSONDecodeError (manifest mid-write): client retries + return web.Response(status=409, text="Manifest unreadable, retry") + print(f"[ConfigTester] ⚖️ Votes: +{result['added']} -{result['removed']} = {result['total']} in {data.get('session_name')}") + return web.json_response(result) + except Exception as e: + print(f"[ConfigTester] Error saving comparisons: {e}") + return web.Response(status=500, text=str(e)) + + +@server.PromptServer.instance.routes.post("/config_tester/clear_comparisons") +async def clear_comparisons(request): + """Delete every vote of a session (body: {session_name}).""" + try: + from .manifest_utils import update_manifest_comparisons + data = await request.json() + manifest_path, err = _comparisons_manifest_path(data.get("session_name")) + if err: + return err + try: + result = await asyncio.to_thread(update_manifest_comparisons, manifest_path, (), (), True) + except FileNotFoundError: + return web.Response(status=404, text="Session manifest not found") + except ValueError: + return web.Response(status=409, text="Manifest unreadable, retry") + print(f"[ConfigTester] ⚖️ Votes cleared ({result['removed']}) in {data.get('session_name')}") + return web.json_response(result) + except Exception as e: + print(f"[ConfigTester] Error clearing comparisons: {e}") + return web.Response(status=500, text=str(e)) + + # ============================================================================= # API: SAVE MANIFEST (Legacy - Full Save) # ============================================================================= @@ -478,6 +553,10 @@ async def save_manifest(request): if "meta" in server_manifest: manifest_data["meta"] = server_manifest["meta"] + # Preserve pairwise-review votes (disk wins; the dashboard copy can be stale) + if "comparisons" in server_manifest: + manifest_data["comparisons"] = server_manifest["comparisons"] + # Save merged manifest os.makedirs(base_dir, exist_ok=True) with open(manifest_path, "w") as f: diff --git a/image_generation.py b/image_generation.py index f7ea9f5..d8e8a50 100644 --- a/image_generation.py +++ b/image_generation.py @@ -1084,6 +1084,10 @@ def flush_batch_with_vae(pending_batch, vae, img_dir, existing_data, session_nam try: with open(manifest_path, "r") as f: disk_manifest = json.load(f) + + # Pairwise-review votes live on disk (disk wins), never drop them on this save + from .manifest_utils import preserve_disk_comparisons + preserve_disk_comparisons(disk_manifest, existing_data) # Create a lookup map for items currently in memory memory_items_map = { diff --git a/manifest_utils.py b/manifest_utils.py index afe8a99..912499f 100644 --- a/manifest_utils.py +++ b/manifest_utils.py @@ -6,12 +6,135 @@ Handles saving and merging manifest.json with user modifications import json import os import threading +import time # Thread-safe lock for manifest file writes. # Required when multiple distributed workers submit results concurrently. _manifest_lock = threading.Lock() +# --------------------------------------------------------------------------- +# Pairwise review votes (manifest top-level "comparisons" list) +# --------------------------------------------------------------------------- +_WINNERS = ("a", "b", "tie") + + +def comparison_key(c): + """Dedupe key of a vote: "ts:a:b" (string).""" + return f"{c.get('ts')}:{c.get('a')}:{c.get('b')}" + + +def validate_comparison(c): + """Return a cleaned vote dict, or None when the entry is not a valid vote.""" + if not isinstance(c, dict): + return None + a, b = c.get("a"), c.get("b") + if type(a) is not int or type(b) is not int or a == b: + return None + if c.get("winner") not in _WINNERS: + return None + ts = c.get("ts") + if type(ts) not in (int, float) or ts != ts or ts < 0: + return None + mode = c.get("mode") + mode = mode[:32] if isinstance(mode, str) and mode else "matched" + out = {"a": a, "b": b, "winner": c["winner"], "ts": ts, "mode": mode} + if isinstance(c.get("blind"), bool): + out["blind"] = c["blind"] + return out + + +def merge_comparison_lists(existing, append, remove_keys): + """ + Merge votes: keep existing order, drop entries whose key is in remove_keys, + append valid new entries whose key is not present yet. + Invalid existing entries are kept untouched. Returns (merged, added, removed). + """ + remove = set(remove_keys or ()) + merged = [] + seen = set() + removed = 0 + for c in (existing or []): + if isinstance(c, dict) and validate_comparison(c) is not None: + k = comparison_key(c) + if k in remove: + removed += 1 + continue + seen.add(k) + merged.append(c) + added = 0 + for c in (append or []): + v = validate_comparison(c) + if v is None: + continue + k = comparison_key(v) + if k in seen or k in remove: + continue + seen.add(k) + merged.append(v) + added += 1 + return merged, added, removed + + +def preserve_disk_comparisons(disk_manifest, target_manifest): + """ + Copy the on-disk votes onto target_manifest (disk wins). Only the compare + routes create or remove votes, so a stale in-memory copy must never + resurrect undone/cleared votes or drop new ones. + """ + try: + if isinstance(disk_manifest, dict) and "comparisons" in disk_manifest and isinstance(target_manifest, dict): + target_manifest["comparisons"] = disk_manifest["comparisons"] + except Exception: + pass + + +def _atomic_write_json(path, data): + tmp = path + ".tmp" + with open(tmp, "w") as f: + json.dump(data, f, indent=4) + last = None + for _ in range(5): + try: + os.replace(tmp, path) + return + except PermissionError as e: # Windows: a reader has the file open + last = e + time.sleep(0.04) + # Fall back to a plain in-place write (same discipline as save_manifest) + with open(path, "w") as f: + json.dump(data, f, indent=4) + try: + os.remove(tmp) + except OSError: + pass + + +def update_manifest_comparisons(manifest_path, append=(), remove_keys=(), clear=False): + """ + Locked read-modify-write of ONLY the "comparisons" list. + Raises FileNotFoundError when the manifest is missing and ValueError/JSON + errors when it is unreadable (nothing is written in that case). + """ + with _manifest_lock: + with open(manifest_path, "r") as f: + data = json.load(f) + if not isinstance(data, dict): + raise ValueError("manifest is not a JSON object") + if clear: + old = data.get("comparisons") + removed = len(old) if isinstance(old, list) else 0 + data["comparisons"], added = [], 0 + else: + existing = data.get("comparisons") + if not isinstance(existing, list): + existing = [] + data["comparisons"], added, removed = merge_comparison_lists(existing, append, remove_keys) + _atomic_write_json(manifest_path, data) + return {"comparisons": data["comparisons"], "added": added, + "removed": removed, "total": len(data["comparisons"])} + + def merge_manifest_user_changes(manifest_path, existing_data): """ Reload manifest and merge user changes (favorites, rejected, notes) to preserve them. @@ -24,6 +147,9 @@ def merge_manifest_user_changes(manifest_path, existing_data): try: with open(manifest_path, "r") as f: current_manifest = json.load(f) + + # Votes are owned by the compare routes: the disk copy always wins + preserve_disk_comparisons(current_manifest, existing_data) # Create lookup dict of current items by ID from DISK current_items_dict = { diff --git a/remote_vae.py b/remote_vae.py index 3cf7204..21ff6c0 100644 --- a/remote_vae.py +++ b/remote_vae.py @@ -231,6 +231,10 @@ class RemoteVAEDecodeWorker: try: with open(self.manifest_path, "r") as f: disk_manifest = json.load(f) + + # Pairwise-review votes live on disk (disk wins), never drop them on this save + from .manifest_utils import preserve_disk_comparisons + preserve_disk_comparisons(disk_manifest, self.existing_data) # Create a lookup map for items currently in memory # This allows us to update self.existing_data in-place diff --git a/tests/conftest.py b/tests/conftest.py index 62950d5..da962b6 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -106,6 +106,8 @@ _GEN_ORCH_DEPS = { "manifest_utils": { "load_existing_manifest": lambda *a, **kw: {}, "save_manifest": lambda *a, **kw: None, + "preserve_disk_comparisons": lambda *a, **kw: None, + "update_manifest_comparisons": lambda *a, **kw: {"comparisons": [], "added": 0, "removed": 0, "total": 0}, }, "model_loader": { "load_checkpoint": lambda *a, **kw: None, diff --git a/tests/test_comparisons_manifest.py b/tests/test_comparisons_manifest.py new file mode 100644 index 0000000..2cda795 --- /dev/null +++ b/tests/test_comparisons_manifest.py @@ -0,0 +1,197 @@ +"""Pairwise-review votes stored in manifest["comparisons"]. + +Loads the real manifest_utils.py (tests/conftest.py registers a stub under the +package name) via importlib, like test_pinned_clip.py does. +""" +import copy +import importlib.util +import json +import os +import threading + +import pytest + +_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +_spec = importlib.util.spec_from_file_location("manifest_utils_real", os.path.join(_ROOT, "manifest_utils.py")) +mu = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(mu) + + +def V(a, b, winner="a", ts=1000, **kw): + d = {"a": a, "b": b, "winner": winner, "ts": ts} + d.update(kw) + return d + + +def _write(path, data): + with open(path, "w") as f: + json.dump(data, f, indent=4) + + +def _read(path): + with open(path) as f: + return json.load(f) + + +def _manifest(): + return {"session_name": "s", "meta": {"model": "m.safetensors"}, + "items": [{"id": 3, "favorited": True, "note": "n", "file": "/view?x=3"}, + {"id": 2, "rejected": True, "file": "/view?x=2"}, + {"id": 1, "file": "/view?x=1"}]} + + +# ---- validate_comparison ---------------------------------------------------- +def test_validate_rejects_bad_entries(): + assert mu.validate_comparison(V(1, 1)) is None + assert mu.validate_comparison(V(True, 2)) is None + assert mu.validate_comparison(V("1", 2)) is None + assert mu.validate_comparison(V(1.0, 2)) is None + assert mu.validate_comparison(V(1, 2, winner="x")) is None + assert mu.validate_comparison(V(1, 2, ts=-1)) is None + assert mu.validate_comparison(V(1, 2, ts="5")) is None + assert mu.validate_comparison("nope") is None + + +def test_validate_cleans_fields(): + v = mu.validate_comparison(V(1, 2, "tie", mode="x" * 100, blind=False, junk=1)) + assert v == {"a": 1, "b": 2, "winner": "tie", "ts": 1000, "mode": "x" * 32, "blind": False} + assert mu.validate_comparison(V(1, 2))["mode"] == "matched" + assert "blind" not in mu.validate_comparison(V(1, 2, blind="yes")) + assert mu.validate_comparison(V(180000000000000, 2))["a"] == 180000000000000 + + +# ---- merge_comparison_lists ------------------------------------------------- +def test_merge_append_dedupe_remove(): + m, added, removed = mu.merge_comparison_lists([], [V(1, 2), V(1, 2), V(3, 4, ts=5)], []) + assert (len(m), added, removed) == (2, 2, 0) + m2, added, removed = mu.merge_comparison_lists(m, [V(1, 2)], []) + assert (len(m2), added) == (2, 0) + m3, added, removed = mu.merge_comparison_lists(m, [], ["1000:1:2"]) + assert (len(m3), removed) == (1, 1) and m3[0]["a"] == 3 + m4, added, removed = mu.merge_comparison_lists(m, [], ["nope"]) + assert (m4, removed) == (m, 0) + + +def test_merge_preserves_invalid_existing(): + junk = {"weird": True} + m, added, removed = mu.merge_comparison_lists([junk, V(1, 2)], [V(5, 6)], ["1000:1:2"]) + assert m[0] == junk and len(m) == 2 and (added, removed) == (1, 1) + + +# ---- update_manifest_comparisons ------------------------------------------- +def test_update_roundtrip_leaves_items_meta_alone(tmp_path): + p = str(tmp_path / "manifest.json") + orig = _manifest() + _write(p, orig) + r = mu.update_manifest_comparisons(p, append=[V(1, 2), V(2, 3, ts=7)]) + assert r["added"] == 2 and r["total"] == 2 + d = _read(p) + assert d["items"] == orig["items"] and d["meta"] == orig["meta"] and len(d["comparisons"]) == 2 + mu.update_manifest_comparisons(p, remove_keys=["1000:1:2"]) + d = _read(p) + assert d["items"] == orig["items"] and len(d["comparisons"]) == 1 + r = mu.update_manifest_comparisons(p, clear=True) + d = _read(p) + assert d["comparisons"] == [] and d["items"] == orig["items"] and r["removed"] == 1 + assert not os.path.exists(p + ".tmp") + + +def test_update_missing_file_raises(tmp_path): + with pytest.raises(FileNotFoundError): + mu.update_manifest_comparisons(str(tmp_path / "nope.json"), append=[V(1, 2)]) + + +def test_update_corrupt_json_raises_and_keeps_file(tmp_path): + p = str(tmp_path / "manifest.json") + with open(p, "w") as f: + f.write('{"items": [ {"id": 1}, ') + before = open(p).read() + with pytest.raises(ValueError): # JSONDecodeError subclasses ValueError + mu.update_manifest_comparisons(p, append=[V(1, 2)]) + assert open(p).read() == before + + +def test_update_creates_comparisons_when_absent(tmp_path): + p = str(tmp_path / "manifest.json") + _write(p, _manifest()) + assert "comparisons" not in _read(p) + mu.update_manifest_comparisons(p, append=[V(1, 2)]) + assert len(_read(p)["comparisons"]) == 1 + + +# ---- regression: other writers must not drop / resurrect votes ------------- +def test_stale_in_memory_manifest_saved_after_vote_keeps_vote(tmp_path): + p = str(tmp_path / "manifest.json") + _write(p, _manifest()) + stale = copy.deepcopy(_manifest()) # the generator's start-of-run copy: no votes + mu.update_manifest_comparisons(p, append=[V(1, 2), V(2, 3, ts=7)]) + stale["items"].insert(0, {"id": 4, "file": "/view?x=4"}) + mu.save_manifest(p, stale) + d = _read(p) + assert [i["id"] for i in d["items"]] == [4, 3, 2, 1] + assert len(d["comparisons"]) == 2 + + +def test_stale_copy_with_old_votes_does_not_resurrect_after_clear(tmp_path): + p = str(tmp_path / "manifest.json") + m = _manifest() + m["comparisons"] = [V(1, 2)] + _write(p, m) + stale = copy.deepcopy(m) + mu.update_manifest_comparisons(p, clear=True) + mu.save_manifest(p, stale) + assert _read(p)["comparisons"] == [] + + +def test_stale_copy_does_not_resurrect_undone_vote(tmp_path): + p = str(tmp_path / "manifest.json") + m = _manifest() + m["comparisons"] = [V(1, 2), V(2, 3, ts=7)] + _write(p, m) + stale = copy.deepcopy(m) + mu.update_manifest_comparisons(p, remove_keys=["7:2:3"]) + mu.save_manifest(p, stale) + assert [c["ts"] for c in _read(p)["comparisons"]] == [1000] + + +def test_preserve_disk_comparisons_noop_without_field(): + t = {"comparisons": [V(1, 2)]} + mu.preserve_disk_comparisons({"items": []}, t) + assert t["comparisons"] == [V(1, 2)] + mu.preserve_disk_comparisons({"comparisons": []}, t) + assert t["comparisons"] == [] + + +# ---- concurrency ------------------------------------------------------------ +def test_concurrent_saves_and_votes_lose_nothing(tmp_path): + p = str(tmp_path / "manifest.json") + base = _manifest() + _write(p, base) + state = copy.deepcopy(base) # the generator's long-lived in-memory dict + errors = [] + N = 50 + + def gen(): + try: + for i in range(N): + state["items"].insert(0, {"id": 1000 + i, "file": f"/view?x={i}"}) + mu.save_manifest(p, state) + except Exception as e: # pragma: no cover + errors.append(e) + + def vote(): + try: + for i in range(N): + mu.update_manifest_comparisons(p, append=[V(1, 2, ts=i + 1)]) + except Exception as e: # pragma: no cover + errors.append(e) + + ts = [threading.Thread(target=gen), threading.Thread(target=vote)] + for t in ts: + t.start() + for t in ts: + t.join() + assert not errors + d = _read(p) + assert len(d["items"]) == 3 + N + assert len(d["comparisons"]) == N