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 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Fable 5.1
parent
198138e51c
commit
1692492eab
+79
@@ -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:
|
||||
|
||||
@@ -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 = {
|
||||
|
||||
@@ -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 = {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user