fix(review): announce items that leave the queue, and seed only queueable files
#284 now queues a file only when its final outcome is ambiguous and unlinked, and it removes the pending items of files that are now linked. That removal is a queue change, so it calls notify_review_change after its commit, like recording and decisions: an open Needs review page re-reads its queue, and its count follows. The review browser fixture wrote graphs that the scan's fallback links, and it produced the conflict item through the monitor's fallback link. Neither is queued any more. Its files now carry ComfyUI's Infinity stamp, which the legacy parser cannot read, so they stay unlinked with the same two candidates. The conflict item is queued first and its file is linked to another prompt afterwards. "monitor" additions go through record_review_file from a worker thread.
This commit is contained in:
@@ -433,6 +433,8 @@ class ReviewOperations:
|
||||
removed += conn.execute(_SETTLE_ITEM, (item_id,)).rowcount
|
||||
if removed:
|
||||
conn.execute(_BUMP_REVIEW_INDEX)
|
||||
if removed:
|
||||
notify_review_change(self.model.db_path)
|
||||
return removed
|
||||
|
||||
def review_queue_page(self, *, limit: int, after: tuple | None = None) -> dict:
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
// Needs review (#249 PR F, plan Tasks F1 and F2, Mock inventory M4): the SAVED VIEWS row and its count, the Metadata
|
||||
// review page over the real review routes, and Activity's Review metadata. Items are recorded by the real recorder
|
||||
// (tests/browser/workspace_review_fixture.py), as the live monitor records a new output; the conflict item is a real
|
||||
// monitor fallback link. An open page re-reads its queue from the review change stream (F2).
|
||||
// (tests/browser/workspace_review_fixture.py) through the monitor's entry point; the conflict item's file was linked
|
||||
// to another prompt after it was queued. An open page re-reads its queue from the review change stream (F2).
|
||||
import { startCoverage, saveCoverage } from '../../tools/vendor/browser-coverage.mjs';
|
||||
import { test, expect } from './fixtures.mjs';
|
||||
|
||||
@@ -103,7 +103,7 @@ test('apply candidate links the pair and the item leaves the queue and the count
|
||||
});
|
||||
|
||||
test('conflict leaves links unchanged and explains', async ({ page, request }) => {
|
||||
// A real monitor fallback: the new output is linked to the most recent prompt, then queued.
|
||||
// Queued first, then linked to another prompt (before a scan or monitor pass could settle it).
|
||||
await seed(request, 0, { conflict: true });
|
||||
const before = await links(request);
|
||||
expect(before).toEqual([{ filename: 'conflict.png', prompt: 'Misty forest path', pair_id: null, positive: null,
|
||||
|
||||
@@ -150,7 +150,7 @@ def register_activity_commands(app, routes, api):
|
||||
scope = {
|
||||
"files": str(data.get("files", "new")),
|
||||
"output_folders": count(data.get("output_folders"), 2),
|
||||
# A real scan's count of files it added to Needs review.
|
||||
# A real scan's count of its files in the Needs review queue.
|
||||
**(
|
||||
{"review_items": count(data["review_items"])}
|
||||
if "review_items" in data
|
||||
|
||||
@@ -1,18 +1,19 @@
|
||||
"""Owned, deterministic Needs review items for the workspace page (#249 PR F).
|
||||
|
||||
Each file is a PNG whose workflow has two samplers with their own positive
|
||||
encoders and one shared negative, so the resolver finds it ``ambiguous``; it
|
||||
is recorded through the real recorder (``record_review_file``), exactly as the
|
||||
live monitor records a new output. No path or SQL comes from a request.
|
||||
encoders and one shared negative, so the resolver finds it ``ambiguous``. Its
|
||||
loader node carries ComfyUI's ``Infinity`` stamp, which the legacy pages'
|
||||
parser (the scan's fallback) cannot read, so nothing links the file and it is
|
||||
a Needs review case (#284: only ambiguous, unlinked files are queued). It is
|
||||
recorded through the producers' entry point (``record_review_file``). No path
|
||||
or SQL comes from a request.
|
||||
|
||||
``POST /__test__/review/reset`` also takes the special items F2 needs, queued
|
||||
after the plain ones in this order:
|
||||
|
||||
- ``conflict``: a new output handled by the real monitor pipeline
|
||||
(``ImageGenerationHandler.process_new_image``). Its positive encoders are
|
||||
PromptManager nodes whose texts are not saved, so the monitor falls back to
|
||||
the most recent prompt ("Misty forest path"), links the file to it and
|
||||
records the item; applying a sampler path then conflicts with that link.
|
||||
- ``conflict``: an item whose file was linked to another prompt ("Misty
|
||||
forest path") after it was queued (the next scan or monitor pass would
|
||||
remove the item); applying a sampler path then conflicts with that link.
|
||||
- ``limited``: an A1111 ``parameters`` block with two negative lines, which
|
||||
the recorder stores ``limited`` (no candidates).
|
||||
- ``unavailable``: a file recorded under the ``temp`` directory while it was a
|
||||
@@ -25,8 +26,10 @@ its first negative, the one the concept's panel shows), so the panel's text
|
||||
wraps as the concept's does.
|
||||
|
||||
``POST /__test__/review/add {kind}`` adds one file while a page is open:
|
||||
``monitor`` runs the monitor pipeline on a new ambiguous output (recorded),
|
||||
``file`` only writes one (a metadata scan records it). ``GET
|
||||
``monitor`` records a new ambiguous, unlinked output from a worker thread
|
||||
through ``record_review_file`` (the monitor's own entry point; the monitor
|
||||
never queues an output it linked itself), ``file`` only writes one (a
|
||||
metadata scan records it). ``GET
|
||||
/__test__/review/links`` answers each review file's prompt and pair links, and
|
||||
``GET /__test__/review/streams`` the open change streams and captures.
|
||||
"""
|
||||
@@ -50,7 +53,7 @@ if TYPE_CHECKING:
|
||||
|
||||
POSITIVES = ("Golden dunes at dusk", "Harbor lights at night")
|
||||
NEGATIVE = "Soft negative, blur"
|
||||
# The monitor's fallback: the most recent saved prompt.
|
||||
# The prompt the conflict item's file is linked to after it was queued.
|
||||
FALLBACK = "Misty forest path"
|
||||
SPECIAL = ("conflict", "limited", "unavailable")
|
||||
# The concept's first three images (concept app.js `images`) and first prompt.
|
||||
@@ -91,7 +94,13 @@ def _graph(
|
||||
negative: str = NEGATIVE,
|
||||
) -> dict:
|
||||
graph = {
|
||||
"1": {"class_type": "CheckpointLoaderSimple", "inputs": {"ckpt_name": "m"}},
|
||||
# ``Infinity`` (as ComfyUI writes it): the legacy parser cannot read the
|
||||
# graph, so the scan's fallback links nothing and the file is queued.
|
||||
"1": {
|
||||
"class_type": "CheckpointLoaderSimple",
|
||||
"inputs": {"ckpt_name": "m"},
|
||||
"is_changed": [float("inf")],
|
||||
},
|
||||
"2": {"class_type": "CLIPTextEncode", "inputs": {"text": negative}},
|
||||
}
|
||||
for index, text in enumerate(positives):
|
||||
@@ -138,36 +147,42 @@ def _stamp(path: Path, age: float) -> None:
|
||||
|
||||
|
||||
class _Monitor:
|
||||
"""The real monitor pipeline over this server's database (one per server)."""
|
||||
"""The monitor's recorder over this server's database (one per server)."""
|
||||
|
||||
def __init__(self, api: PromptManagerAPI) -> None:
|
||||
from utils.image_monitor import ImageGenerationHandler
|
||||
from utils.prompt_tracker import PromptTracker
|
||||
|
||||
self.handler = ImageGenerationHandler(api.db, PromptTracker(api.db))
|
||||
self.handler.review_recorder = api.record_review_file
|
||||
self.api = api
|
||||
self.lock = threading.Lock()
|
||||
|
||||
def new_output(self, path: Path) -> None:
|
||||
"""What the monitor does once a new output has settled."""
|
||||
"""The monitor's Needs review pass for a new output, from a worker thread."""
|
||||
outcome = []
|
||||
with self.lock:
|
||||
self.handler.process_new_image(str(path))
|
||||
worker = threading.Thread(
|
||||
target=lambda: outcome.append(self.api.record_review_file(str(path)))
|
||||
)
|
||||
worker.start()
|
||||
worker.join()
|
||||
if outcome != ["recorded"]:
|
||||
raise RuntimeError(f"{path.name} was {outcome}, not recorded")
|
||||
|
||||
|
||||
def _monitored_output(api: PromptManagerAPI, monitor: _Monitor, path: Path) -> None:
|
||||
_png(path, (150, 60, 40))
|
||||
monitor.new_output(path)
|
||||
|
||||
|
||||
def _linked_after_queueing(api: PromptManagerAPI, path: Path) -> None:
|
||||
from utils.hashing import generate_prompt_hash
|
||||
|
||||
if api.db.get_prompt_by_hash(generate_prompt_hash(FALLBACK)) is None:
|
||||
api.db.save_prompt(FALLBACK, prompt_hash=generate_prompt_hash(FALLBACK))
|
||||
_png(path, (150, 60, 40), graph=_graph("PromptManager"))
|
||||
before = len(api.db.review_items_for(_locator(path)))
|
||||
monitor.new_output(path)
|
||||
if len(api.db.review_items_for(_locator(path))) != before + 1:
|
||||
raise RuntimeError(f"the monitor did not record {path.name}")
|
||||
|
||||
|
||||
def _locator(path: Path) -> str:
|
||||
return os.path.normcase(os.path.realpath(path))
|
||||
_png(path, (150, 60, 40))
|
||||
_record(api, path, "recorded")
|
||||
prompt = api.db.get_prompt_by_hash(generate_prompt_hash(FALLBACK))
|
||||
prompt_id = (
|
||||
prompt["id"]
|
||||
if prompt is not None
|
||||
else api.db.save_prompt(FALLBACK, prompt_hash=generate_prompt_hash(FALLBACK))
|
||||
)
|
||||
api.db.link_image_to_prompt(prompt_id, str(path))
|
||||
|
||||
|
||||
def _unavailable(api: PromptManagerAPI, root: Path, path: Path) -> None:
|
||||
@@ -220,7 +235,7 @@ def seed_review(
|
||||
relative = f"review/{kind}.png"
|
||||
path = (root / "temp" if kind == "unavailable" else output) / relative
|
||||
if kind == "conflict":
|
||||
_monitored_output(api, monitor, path)
|
||||
_linked_after_queueing(api, path)
|
||||
elif kind == "limited":
|
||||
_png(
|
||||
path,
|
||||
|
||||
@@ -15,8 +15,10 @@ import threading
|
||||
|
||||
try:
|
||||
from tests.test_api_review import ReviewRoutesTestCase
|
||||
from tests.test_review import unlinkable_graph, write_png
|
||||
except ImportError: # discovered with tests/ as the top-level directory
|
||||
from test_api_review import ReviewRoutesTestCase
|
||||
from test_review import unlinkable_graph, write_png
|
||||
|
||||
try: # guarded so a missing module fails assertions, not test loading
|
||||
from database import review_changes
|
||||
@@ -76,6 +78,13 @@ class ReviewEventsTestCase(ReviewRoutesTestCase):
|
||||
self.assertEqual(name, "ready")
|
||||
return data
|
||||
|
||||
def _queueable(self, name="two.png", color=(255, 0, 0)):
|
||||
"""A file the producers queue: ambiguous, and nothing links it."""
|
||||
path = self.output_dir / name
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
write_png(path, prompt=unlinkable_graph("cat", "dog"), color=color)
|
||||
return path
|
||||
|
||||
async def _record_in_thread(self, path):
|
||||
"""Record as the monitor does: from a worker thread, not the loop."""
|
||||
self._roots()
|
||||
@@ -92,7 +101,7 @@ class TestReviewEvents(ReviewEventsTestCase):
|
||||
async def test_monitor_recording_sends_one_change(self):
|
||||
self._require_events()
|
||||
stream = await self._stream()
|
||||
self.assertEqual(await self._record_in_thread(self._ambiguous()), "recorded")
|
||||
self.assertEqual(await self._record_in_thread(self._queueable()), "recorded")
|
||||
self.assertEqual(await self._event(stream), "change")
|
||||
|
||||
async def test_duplicate_recording_and_refused_decision_send_nothing(self):
|
||||
@@ -111,7 +120,7 @@ class TestReviewEvents(ReviewEventsTestCase):
|
||||
calls.append(db_path)
|
||||
original(db_path)
|
||||
|
||||
path = self._ambiguous("two.png")
|
||||
path = self._queueable("two.png")
|
||||
self._record(path)
|
||||
with (
|
||||
patch.object(review_db, "notify_review_change", spy),
|
||||
@@ -182,12 +191,25 @@ class TestReviewEvents(ReviewEventsTestCase):
|
||||
async def test_scan_recording_sends_a_change(self):
|
||||
self._require_events()
|
||||
self._roots()
|
||||
self._ambiguous("sub/two.png")
|
||||
self._queueable("sub/two.png")
|
||||
stream = await self._stream()
|
||||
job = await self._scan("all")
|
||||
self.assertEqual(job["scope"]["review_items"], 1)
|
||||
self.assertEqual(await self._event(stream), "change")
|
||||
|
||||
async def test_a_linked_file_leaving_the_queue_sends_a_change(self):
|
||||
"""The queue heals (#284 maintainer test): a pending item whose file is
|
||||
linked now leaves the queue, and the open page hears about it."""
|
||||
self._require_events()
|
||||
self._roots()
|
||||
path = self._ambiguous("two.png")
|
||||
self._record(path) # as the pre-fix scan queued fallback-linked files
|
||||
stream = await self._stream()
|
||||
job = await self._scan("new")
|
||||
self.assertEqual(job["scope"]["review_settled"], 1)
|
||||
self.assertEqual(self._items(), [])
|
||||
self.assertEqual(await self._event(stream), "change")
|
||||
|
||||
async def test_ready_carries_the_version_a_list_read_answers(self):
|
||||
# F3 (no double read per visit): the page skips the re-read on
|
||||
# ``ready`` when its first read already answered this version.
|
||||
|
||||
Reference in New Issue
Block a user