diff --git a/database/pair_ingestion.py b/database/pair_ingestion.py index c23184c..ff1262e 100644 --- a/database/pair_ingestion.py +++ b/database/pair_ingestion.py @@ -9,8 +9,10 @@ from .operations import _coerce_file_info try: from ..utils.hashing import generate_prompt_hash + from ..utils.prompt_pair_resolver import RESOLVER_VERSION except ImportError: from utils.hashing import generate_prompt_hash + from utils.prompt_pair_resolver import RESOLVER_VERSION def ingest_resolved_pair( @@ -51,7 +53,7 @@ def _ingest_resolved_pair_in( normalized = normalize_image_path(image_path) if provenance is None: provenance = { - "resolver_version": 1, + "resolver_version": RESOLVER_VERSION, "positive": positive, "negative": negative, } diff --git a/database/review.py b/database/review.py index 160fdfd..db7df29 100644 --- a/database/review.py +++ b/database/review.py @@ -117,7 +117,7 @@ _DETAIL = ( ) # The legacy rows and pair links the ingester's compatibility rules read. _DEPENDENT_IMAGES = ( - "SELECT gi.id,p.hash FROM generated_images gi " + "SELECT gi.id,p.hash,gi.image_path FROM generated_images gi " "JOIN prompts p ON p.id=gi.prompt_id " "WHERE gi.file_path=? OR gi.image_path=? ORDER BY gi.id" ) @@ -127,6 +127,12 @@ _DEPENDENT_LINKS = ( "LEFT JOIN prompts n ON n.id=pp.negative_prompt_id " "WHERE ip.image_id=? ORDER BY ip.pair_id" ) +# Rows whose spelling ends with the file's name: the candidates for another +# spelling of the same physical file (the link summaries' filename fallback). +_SAME_NAME_ROWS = ( + "SELECT id,image_path FROM generated_images " + "WHERE image_path LIKE ? ESCAPE '\\' ORDER BY id" +) _TITLES = ( "SELECT p.hash,s.title FROM prompts p " "JOIN prompt_library_state s ON s.prompt_id=p.id " @@ -159,18 +165,47 @@ def _validate_spelling(spelled_path: object, locator: str) -> None: raise ValueError("spelled_path must be the file's spelling of its locator") -def dependency_fingerprint(conn: sqlite3.Connection, image_path: str) -> str: +def dependent_images(conn: sqlite3.Connection, spellings) -> list: + """``(id, image_path, positive_hash)`` of every legacy row of the file. + + ``spellings`` are the paths the file is linked under (the Apply path and + every alias of the same physical file); each is matched as the ingester + matches one path. Rows are distinct and ordered by id. + """ + found = {} + for spelled in dict.fromkeys(spellings): + for row in conn.execute( + _DEPENDENT_IMAGES, (normalize_image_path(spelled), spelled) + ): + found[row[0]] = (row[0], row[2], row[1]) + return [found[key] for key in sorted(found)] + + +def same_name_rows(conn: sqlite3.Connection, name: str) -> list: + """``(id, image_path)`` of rows whose spelling may end with ``name``.""" + # Imported here: operations composes this module's mixin. + from .operations import escape_like + + return [ + (row[0], row[1]) + for row in conn.execute(_SAME_NAME_ROWS, (f"%{escape_like(name)}",)) + ] + + +def dependency_fingerprint( + conn: sqlite3.Connection, image_path: str, aliases=() +) -> str: """SHA-256 of what Apply's compatibility depends on for ``image_path``. - The exact legacy rows of the file with their positive hashes, and each - row's pair links with the negative hash, read as the ingester reads them. - Unrelated prompt changes (tags, titles) do not move it. + The exact legacy rows of the file under ``image_path`` and every alias + spelling of it, with their positive hashes, and each row's pair links + with the negative hash, read as the ingester reads them. Unrelated prompt + changes (tags, titles) do not move it. """ - images = conn.execute( - _DEPENDENT_IMAGES, (normalize_image_path(image_path), image_path) - ).fetchall() rows = [] - for image_id, positive_hash in images: + for image_id, _spelled, positive_hash in dependent_images( + conn, (image_path, *aliases) + ): links = conn.execute(_DEPENDENT_LINKS, (image_id,)).fetchall() rows.append([image_id, positive_hash, [list(link) for link in links]]) text = canonical_evidence({"v": 1, "images": rows}) @@ -328,9 +363,14 @@ class ReviewOperations: item["total"] = conn.execute(_PENDING_COUNT).fetchone()[0] return item - def review_dependency_fingerprint(self, image_path: str) -> str: + def review_dependency_fingerprint(self, image_path: str, aliases=()) -> str: with self._annotation_read() as conn: - return dependency_fingerprint(conn, image_path) + return dependency_fingerprint(conn, image_path, aliases) + + def review_same_name_rows(self, name: str) -> list: + """``(id, image_path)`` of legacy rows that may spell a file ``name``.""" + with self._annotation_read() as conn: + return same_name_rows(conn, name) def stored_prompt_titles(self, texts) -> dict: """Library titles of saved prompts by text (dedup hash), when set.""" diff --git a/database/review_decisions.py b/database/review_decisions.py index b305927..3b5b1ff 100644 --- a/database/review_decisions.py +++ b/database/review_decisions.py @@ -12,6 +12,7 @@ Callers verify the file first (the capture must be the item's bytes and the current file must still hash to them); nothing here reads the filesystem. """ +import os import sqlite3 from .annotation_schema import BYTES_TABLE @@ -25,6 +26,8 @@ from .review import ( _PENDING_COUNT, _queue_row, dependency_fingerprint, + dependent_images, + same_name_rows, ) from .review_schema import DECISIONS_TABLE, ITEMS_TABLE, canonical_evidence @@ -65,6 +68,8 @@ class ReviewCommandError(LibraryCommandError): self.total = total self.fingerprint = fingerprint self.changed = changed + # The public view of ``item``, set by the API when it formats one. + self.view = None class _Rollback(Exception): @@ -90,12 +95,18 @@ class ReviewDecisionOperations: image_path: str | None = None, metadata: dict | None = None, expected_fingerprint: str | None = None, + aliases: tuple = (), + known_rows: frozenset | None = None, ) -> dict: """Record the decision; return ``{item, total, pair}``. ``selected`` needs the evidence ``candidate`` (with its ``hash``), the ``image_path`` the scan and the monitor link under, the captured ``metadata`` and the ``expected_fingerprint`` the reviewer saw. + ``aliases`` are the other spellings legacy rows of the same physical + file use (a symlinked, junctioned or mapped root) and ``known_rows`` + the ids of every same-name row the caller classified; a same-name row + it did not see is a changed association. Raises ``LibraryCommandError("storage-changed")`` after a restore and :class:`ReviewCommandError` for every refused decision, with zero writes either way. @@ -119,6 +130,8 @@ class ReviewDecisionOperations: image_path=image_path, metadata=metadata or {}, expected_fingerprint=expected_fingerprint, + aliases=tuple(aliases), + known_rows=known_rows, ) except _Rollback as rollback: raise rollback.error from None @@ -191,8 +204,14 @@ class ReviewDecisionOperations: from .pair_ingestion import _ingest_resolved_pair_in image_path = command["image_path"] - fingerprint = dependency_fingerprint(conn, image_path) - if fingerprint != command["expected_fingerprint"]: + aliases = command["aliases"] + fingerprint = dependency_fingerprint(conn, image_path, aliases) + known = command["known_rows"] + unseen = known is not None and any( + row_id not in known + for row_id, _path in same_name_rows(conn, os.path.basename(image_path)) + ) + if unseen or fingerprint != command["expected_fingerprint"]: raise _Rollback( ReviewCommandError( "association-conflict", @@ -203,6 +222,22 @@ class ReviewDecisionOperations: changed=True, ) ) + rows = dependent_images(conn, (image_path, *aliases)) + if len(rows) > 1: + # One file linked under several spellings: no single row to enrich. + raise _Rollback( + ReviewCommandError( + "association-conflict", + epoch, + item=current, + total=total, + fingerprint=fingerprint, + ) + ) + if rows: + # Enrich the one row of this file under the spelling it was + # linked with, so the ingester's compatibility rules see it. + image_path = rows[0][1] candidate = command["candidate"] positive, negative = candidate["positive"], candidate["negative"] provenance = { diff --git a/py/api/review.py b/py/api/review.py index 5e39635..5980dd6 100644 --- a/py/api/review.py +++ b/py/api/review.py @@ -33,14 +33,12 @@ try: from ...database.library import display_title from ...database.library_commands import LibraryCommandError from ...database.review_decisions import ReviewCommandError - from ...database.review_schema import canonical_evidence, evidence_digest from ...utils.media_capture import CaptureError from ...utils.review_capture import captured_review, review_evidence except ImportError: from database.library import display_title from database.library_commands import LibraryCommandError from database.review_decisions import ReviewCommandError - from database.review_schema import canonical_evidence, evidence_digest from utils.media_capture import CaptureError from utils.review_capture import captured_review, review_evidence @@ -63,6 +61,8 @@ _DECISION_FIELDS = frozenset( } ) _CURSOR_SECRET = os.urandom(32) +# The real path of a spelling; module-level so a test can fake a junction. +_realpath = os.path.realpath _LIMIT = re.compile(r"[0-9]{1,3}") _NO_STORE = {"Cache-Control": "no-store"} @@ -230,9 +230,12 @@ class ReviewRoutesMixin: # -- shared ---------------------------------------------------------------- def _review_roots(self): - """``(roots, real_roots)``: the current gallery roots and their realpaths.""" + """``(roots, real_roots)``: the current gallery roots and their realpaths. + + Blocking (it lists the roots and resolves them): call it on a worker. + """ roots = list(self._get_all_output_dirs()) - return roots, [(i, os.path.realpath(str(root))) for i, root in enumerate(roots)] + return roots, [(i, _realpath(str(root))) for i, root in enumerate(roots)] def _review_view(self, row: dict, roots, real_roots) -> dict: """The public ReviewItem of one stored row (no path, no locator).""" @@ -280,12 +283,9 @@ class ReviewRoutesMixin: except ValueError as exc: return bad_request(str(exc)) try: - page = await self._run_in_executor( - self.db.review_queue_page, - limit=limit, - after=bound[2] if bound else None, + page, views = await self._run_in_executor( + self._review_page, limit, bound[2] if bound else None ) - roots, real_roots = self._review_roots() except Exception: self.logger.exception("Review queue read failed") return _failed() @@ -305,32 +305,34 @@ class ReviewRoutesMixin: "success": True, "library_epoch": page["epoch"], "total": page["total"], - "items": [self._review_view(row, roots, real_roots) for row in items], + "items": views, "next_cursor": next_cursor, }, headers=_NO_STORE, ) + def _review_page(self, limit: int, after): + """Worker thread: one queue page and the public views of its items.""" + page = self.db.review_queue_page(limit=limit, after=after) + roots, real_roots = self._review_roots() + return page, [ + self._review_view(row, roots, real_roots) for row in page["items"] + ] + # -- detail ---------------------------------------------------------------- async def get_review_item(self, request: web.Request) -> web.Response: """GET one item with its candidates and dependency fingerprint.""" try: - item = await self._run_in_executor( - self.db.review_item, request.match_info["item_id"] + read = await self._run_in_executor( + self._review_detail, request.match_info["item_id"] ) - if item is None: - return _coded("review-missing") - roots, real_roots = self._review_roots() - view = self._review_view(item, roots, real_roots) - image_path = self._review_image_path(item, roots, view["target"]) - fingerprint = await self._run_in_executor( - self.db.review_dependency_fingerprint, image_path - ) - candidates = await self._run_in_executor(self._review_candidates, item) except Exception: self.logger.exception("Review item read failed") return _failed() + if read is None: + return _coded("review-missing") + item, view, fingerprint, candidates = read detail = { "success": True, "library_epoch": item["epoch"], @@ -344,6 +346,18 @@ class ReviewRoutesMixin: detail = {**detail, "candidates": [], "limited": True} return web.json_response(detail, headers=_NO_STORE) + def _review_detail(self, item_id: str): + """Worker thread: ``(item, view, fingerprint, candidates)`` or None.""" + item = self.db.review_item(item_id) + if item is None: + return None + roots, real_roots = self._review_roots() + view = self._review_view(item, roots, real_roots) + image_path = self._review_image_path(item, roots, view["target"]) + aliases, _known = physical_spellings(self.db, image_path, item["locator"]) + fingerprint = self.db.review_dependency_fingerprint(image_path, aliases) + return item, view, fingerprint, self._review_candidates(item) + def _review_candidates(self, item: dict) -> list: """The detail's candidates: role texts, a label and selectability.""" if item["limited"]: @@ -418,20 +432,19 @@ class ReviewRoutesMixin: return bad_request("This sampler path cannot be applied") if guards["capture_token"] is None: return bad_request("Apply needs a capture of the file") - roots = list(self._get_all_output_dirs()) verified = None if guards["capture_token"] is not None: verified, error = await self._verified_review_capture( - item, guards, roots, need_metadata=candidate is not None + item, guards, need_metadata=candidate is not None ) if error is not None: return error elif guards["sha256"] != item["sha256"]: return bad_request("sha256 does not match the item") - return await self._record_decision(item, guards, candidate, verified, roots) + return await self._record_decision(item, guards, candidate, verified) - async def _verified_review_capture(self, item, guards, roots, *, need_metadata): - """``((capture, image_path, metadata), None)`` or ``(None, error)``. + async def _verified_review_capture(self, item, guards, *, need_metadata): + """``((capture, facts, image_path, metadata), None)`` or ``(None, error)``. The capture must be the item's file and bytes, under the expected epoch and a current root; the current file must re-hash to the same @@ -451,52 +464,74 @@ class ReviewRoutesMixin: if guards["expected_epoch"] != capture.epoch: current = await self._run_in_executor(self.db.get_execution_epoch) return None, _coded("storage-changed", library_epoch=current) - root, error = self._scoped_root(capture.root_index, capture.scope) - if error is not None: - return None, error - found = format_review_target( - item["spelled_path"], - item["locator"], - [(capture.root_index, os.path.realpath(str(root)))], - ) - if found is None: - return None, _coded("source-changed") - relative = found[1] try: - facts, captured = await self._run_in_executor( - _verify_and_read, store, capture, root, relative, need_metadata + root, error = await self._run_in_executor( + self._scoped_root, capture.root_index, capture.scope + ) + if error is not None: + return None, error + read = await self._run_in_executor( + _verify_and_read, store, capture, item, root, need_metadata ) except CaptureError as exc: return None, _capture_error(exc) + if read is None: + return None, _coded("source-changed") + relative, facts, captured = read if facts.locator != item["locator"]: return None, _coded("capture-expired", "other-file") if facts.sha256 != item["sha256"]: return None, _coded("file-changed") metadata = None if need_metadata: - if captured is None or not _same_evidence(captured[1], item): + if captured is None or not _candidate_still_offered( + captured[1], guards["choice"] + ): return None, _coded("review-conflict", "evidence-changed") metadata = captured[0] image_path = os.path.join(str(root), *relative.split("/")) - return (capture, image_path, metadata), None + return (capture, facts, image_path, metadata), None - async def _record_decision(self, item, guards, candidate, verified, roots): - capture, image_path, metadata = verified or (None, None, None) + def _decide_sync(self, item, guards, candidate, verified): + """Worker thread: resolve the file's spellings, decide, build the view. + + Returns ``(result, view)``; a refusal raises with ``exc.view`` set. + """ + _capture, facts, image_path, metadata = verified or (None, None, None, None) + aliases, known = (), None + if candidate is not None: + aliases, known = physical_spellings(self.db, image_path, item["locator"]) try: - result = await self._run_in_executor( - self.db.decide_review_item, + result = self.db.decide_review_item( item_id=item["id"], expected_epoch=guards["expected_epoch"], expected_revision=guards["expected_revision"], - sha256=item["sha256"], + # The bytes the re-hash proved, or the body's for Leave unresolved. + sha256=facts.sha256 if facts is not None else guards["sha256"], outcome="selected" if candidate is not None else UNRESOLVED, candidate=candidate, image_path=image_path, metadata=metadata, expected_fingerprint=guards["dependency_fingerprint"], + aliases=aliases, + known_rows=known, ) except ReviewCommandError as exc: - return self._decision_refused(exc, roots) + if exc.item is not None: + roots, real_roots = self._review_roots() + exc.view = self._review_view(exc.item, roots, real_roots) + raise + roots, real_roots = self._review_roots() + return result, self._review_view(result["item"], roots, real_roots) + + async def _record_decision(self, item, guards, candidate, verified): + capture = verified[0] if verified else None + try: + result, view = await self._run_in_executor( + self._decide_sync, item, guards, candidate, verified + ) + except ReviewCommandError as exc: + return self._decision_refused(exc) except LibraryCommandError as exc: return _coded(exc.code, library_epoch=exc.epoch) except Exception: @@ -505,24 +540,22 @@ class ReviewRoutesMixin: if capture is not None: # The decision used the capture; the page returns to Images. self._captures.release(capture.token) - real_roots = [(i, os.path.realpath(str(r))) for i, r in enumerate(roots)] body = { "success": True, "library_epoch": guards["expected_epoch"], - "item": self._review_view(result["item"], roots, real_roots), + "item": view, "total": result["total"], } if result["pair"] is not None: body["pair"] = result["pair"] return web.json_response(body, headers=_NO_STORE) - def _decision_refused(self, exc: ReviewCommandError, roots) -> web.Response: + def _decision_refused(self, exc: ReviewCommandError) -> web.Response: extra = {"library_epoch": exc.epoch} if exc.total is not None: extra["total"] = exc.total - if exc.item is not None: - real_roots = [(i, os.path.realpath(str(r))) for i, r in enumerate(roots)] - extra["item"] = self._review_view(exc.item, roots, real_roots) + if exc.view is not None: + extra["item"] = exc.view message = None if exc.fingerprint is not None: extra["dependency_fingerprint"] = exc.fingerprint @@ -561,19 +594,70 @@ def _decision_guards(data: dict): return {**data}, None -def _verify_and_read(store, capture, root, relative: str, need_metadata: bool): - """Worker thread: re-hash the current file, then read the capture.""" +def _verify_and_read(store, capture, item, root, need_metadata: bool): + """Worker thread: ``(relative, facts, captured)``, or None off the root. + + Formats the item's target under the capture's root (resolving the root's + real path), re-hashes the current file, then reads the capture. + """ + found = format_review_target( + item["spelled_path"], + item["locator"], + [(capture.root_index, _realpath(str(root)))], + ) + if found is None: + return None + relative = found[1] facts = store.verify(root, relative) captured = captured_review(store, capture) if need_metadata else None - return facts, captured + return relative, facts, captured -def _same_evidence(resolution: dict, item: dict) -> bool: - """True when the captured bytes still yield the item's exact evidence.""" +def _same_file(spelled: str, image_path: str, locator: str) -> bool: + """True when ``spelled`` reaches the physical file at ``locator``.""" + try: + if os.path.normcase(_realpath(spelled)) == locator: + return True + return os.path.samefile(spelled, image_path) + except (OSError, ValueError): + return False + + +def physical_spellings(db, image_path: str, locator: str): + """``(aliases, known_rows)`` for the legacy rows of one physical file. + + The roots are real paths, while the monitor links a file under the + directory it watches, so a symlinked or junctioned root, a ``subst`` or a + mapped drive spells the same file otherwise. Every row whose spelling + ends with the file's name (the link summaries' fallback lookup) is a + candidate; ``aliases`` are the candidates that resolve to ``locator`` + (``normcase(realpath)``, as the capture forms it) or are the same file, + and ``known_rows`` every candidate id, so a decision can refuse a + same-name row it did not classify. Blocking: call it on a worker. + """ + rows = db.review_same_name_rows(os.path.basename(image_path)) + aliases = tuple( + spelled + for _row_id, spelled in rows + if isinstance(spelled, str) + and spelled != image_path + and _same_file(spelled, image_path, locator) + ) + return tuple(dict.fromkeys(aliases)), frozenset(row_id for row_id, _ in rows) + + +def _candidate_still_offered(resolution: dict, choice: str) -> bool: + """True when the captured bytes still offer the chosen candidate. + + Only what the decision depends on is re-checked: the bytes are still + ``ambiguous`` and the chosen candidate (its hash covers its sampler, + sides and provenance) is still among their candidates. A resolver change + that rewords the top-level reason or reorders candidates leaves a + pending item applicable. + """ if not isinstance(resolution, dict) or resolution.get("state") != "ambiguous": return False - evidence, _count, _limited = review_evidence(resolution) - try: - return evidence_digest(canonical_evidence(evidence)) == item["evidence_hash"] - except ValueError: + evidence, _count, limited = review_evidence(resolution) + if limited: return False + return any(c.get("hash") == choice for c in evidence.get("candidates", ())) diff --git a/tests/test_api_review.py b/tests/test_api_review.py index b482b73..be67bce 100644 --- a/tests/test_api_review.py +++ b/tests/test_api_review.py @@ -9,10 +9,12 @@ unresolved records the decision and links nothing. Responses never carry a server path. """ +import json import ntpath import os import posixpath import sqlite3 +from pathlib import Path from unittest.mock import patch try: @@ -25,7 +27,10 @@ try: except ImportError: # discovered with tests/ as the top-level directory from test_review import ReviewTestCase, ambiguous_graph, digest, write_png -from database.review_schema import canonical_evidence +from database.review_schema import canonical_evidence, evidence_digest +from py.config import GalleryConfig +from utils import review_capture as review_capture_module +from utils.media_capture import CaptureChanged, FileFacts from utils.hashing import generate_prompt_hash try: # guarded so a missing module fails assertions, not test loading @@ -752,3 +757,302 @@ class TestDecisions(ReviewRoutesTestCase): ) self.assertEqual(status, 200, data) self._no_paths() + + +class TestPhysicalIdentity(ReviewRoutesTestCase): + """Apply sees a legacy row of the same file under any spelling (I1). + + The roots are real paths, but the monitor links a file under the + directory it watches: a symlink, a Windows junction, ``subst`` or a + mapped drive reaches the same file under another spelling. + """ + + def _monitor_link(self, text, spelled): + prompt_id = self.api.db.save_prompt( + text, prompt_hash=generate_prompt_hash(text) + ) + return self.api.db.link_image_to_prompt(prompt_id, str(spelled)) + + def _rows(self): + conn = self.api.db.model.get_connection() + return [ + tuple(row) + for row in conn.execute( + "SELECT gi.image_path, p.text FROM generated_images gi " + "JOIN prompts p ON p.id=gi.prompt_id ORDER BY gi.id" + ) + ] + + async def _conflict_then_compatible(self, alias_path): + image_id = self._monitor_link("cat", alias_path) + item, detail, observed = await self._open() + before = self._snapshot() + dog = self._candidate(detail, "dog") + status, data = await self._decide( + item["id"], self._body(detail, observed, dog["hash"]) + ) + self.assertEqual((status, data.get("code")), (409, "association-conflict")) + self.assertEqual(self._snapshot(), before) + cat = self._candidate(detail, "cat") + status, data = await self._decide( + item["id"], self._body(detail, observed, cat["hash"]) + ) + self.assertEqual(status, 200, data) + # The monitor's row is enriched; no second row for the same file. + self.assertEqual(data["pair"]["image_id"], image_id) + self.assertEqual(self._rows(), [(str(alias_path), "cat")]) + self._no_paths() + + def _symlinked_root(self): + alias = Path(self.tmpdir) / "alias-output" + try: + os.symlink(str(self.output_dir), str(alias), target_is_directory=True) + except (OSError, NotImplementedError, AttributeError) as exc: + self.skipTest(f"symlinks unavailable: {exc}") + return alias + + async def test_symlinked_root_legacy_row_is_an_association_conflict(self): + self._require_routes() + path = self._ambiguous("two.png") + alias = self._symlinked_root() + self._record(path) + await self._conflict_then_compatible(alias / "two.png") + + async def test_junction_or_mapped_drive_legacy_row_is_seen(self): + """A portable fake of a junction: its realpath is the real root.""" + self._require_routes() + path = self._ambiguous("two.png") + self._record(path) + junction = os.path.join(self.tmpdir, "Junction") + real = os.path.realpath + + def fake_realpath(value, *args, **kwargs): + text = os.fspath(value) + if os.path.normcase(text).startswith(os.path.normcase(junction)): + text = str(self.output_dir) + text[len(junction) :] + return real(text, *args, **kwargs) + + with patch.object(review_api, "_realpath", side_effect=fake_realpath): + await self._conflict_then_compatible(os.path.join(junction, "two.png")) + + async def test_two_rows_under_two_spellings_are_an_association_conflict(self): + self._require_routes() + path = self._ambiguous("two.png") + alias = self._symlinked_root() + self._monitor_link("cat", alias / "two.png") + self._monitor_link("cat", path) + self._record(path) + item, detail, observed = await self._open() + before = self._snapshot() + cat = self._candidate(detail, "cat") + status, data = await self._decide( + item["id"], self._body(detail, observed, cat["hash"]) + ) + self.assertEqual((status, data.get("code")), (409, "association-conflict")) + self.assertEqual(self._snapshot(), before) + + async def test_same_name_in_another_folder_is_not_the_file(self): + self._require_routes() + path = self._ambiguous("two.png") + other = self.output_dir / "elsewhere" / "two.png" + other.parent.mkdir(parents=True) + write_png(other, prompt=ambiguous_graph("cat", "dog"), color=(9, 9, 9)) + self._monitor_link("dog", other) + self._record(path) + item, detail, observed = await self._open() + cat = self._candidate(detail, "cat") + status, data = await self._decide( + item["id"], self._body(detail, observed, cat["hash"]) + ) + self.assertEqual(status, 200, data) + self.assertEqual( + sorted(self._rows()), sorted([(str(other), "dog"), (str(path), "cat")]) + ) + + async def test_a_new_alias_row_after_the_check_is_a_changed_conflict(self): + self._require_routes() + path = self._ambiguous("two.png") + alias = self._symlinked_root() + self._record(path) + item, detail, observed = await self._open() + cat = self._candidate(detail, "cat") + # The monitor links the file under its alias after the reviewer opened it. + self._monitor_link("cat", alias / "two.png") + before = self._snapshot() + status, data = await self._decide( + item["id"], self._body(detail, observed, cat["hash"]) + ) + self.assertEqual((status, data.get("code")), (409, "association-conflict")) + self.assertNotEqual( + data["dependency_fingerprint"], detail["dependency_fingerprint"] + ) + self.assertEqual(self._snapshot(), before) + + +class TestApplyRefusals(ReviewRoutesTestCase): + """Every Apply refusal answers a code and writes nothing (I2, M1, M4).""" + + async def _opened(self): + path = self._ambiguous("two.png") + self._record(path) + item, detail, observed = await self._open() + return path, item, detail, observed, self._candidate(detail, "cat") + + async def _refused(self, item, body, status, code=None): + before = self._snapshot() + got, data = await self._decide(item["id"], body) + self.assertEqual(got, status, data) + if code is not None: + self.assertEqual(data["code"], code, data) + self.assertEqual(self._snapshot(), before) + self.assertEqual(self._decisions(), []) + return data + + async def test_resolver_change_keeps_the_chosen_candidate_applicable(self): + self._require_routes() + _path, item, detail, observed, cat = await self._opened() + real = review_capture_module.resolve_prompt_pair + + def reworded(metadata): + resolution = real(metadata) + return { + **resolution, + "reason": "a reworded reason from a newer resolver", + "candidates": list(reversed(resolution["candidates"])), + } + + with ( + patch.object( + review_capture_module, "resolve_prompt_pair", side_effect=reworded + ), + patch.object(review_capture_module, "RESOLVER_VERSION", 2), + ): + status, data = await self._decide( + item["id"], self._body(detail, observed, cat["hash"]) + ) + self.assertEqual(status, 200, data) + self.assertEqual(data["item"]["state"], "selected") + + async def test_chosen_candidate_no_longer_offered_is_a_review_conflict(self): + self._require_routes() + _path, item, detail, observed, cat = await self._opened() + real = review_capture_module.resolve_prompt_pair + + def without_cat(metadata): + resolution = real(metadata) + kept = [ + c + for c in resolution["candidates"] + if (c.get("positive") or {}).get("text") != "cat" + ] + return {**resolution, "candidates": kept * 2} + + with patch.object( + review_capture_module, "resolve_prompt_pair", side_effect=without_cat + ): + await self._refused( + item, + self._body(detail, observed, cat["hash"]), + 409, + "review-conflict", + ) + + async def test_decision_passes_the_verified_digest(self): + self._require_routes() + path, item, detail, observed, cat = await self._opened() + real = type(self.api.db).decide_review_item + seen = [] + + def spy(db, **kwargs): + seen.append(kwargs["sha256"]) + return real(db, **kwargs) + + with patch.object(type(self.api.db), "decide_review_item", spy): + status, data = await self._decide( + item["id"], self._body(detail, observed, cat["hash"]) + ) + self.assertEqual(status, 200, data) + self.assertEqual(seen, [digest(path)]) + + async def test_expired_capture(self): + self._require_routes() + _path, item, detail, observed, cat = await self._opened() + self.api._captures.release(observed["capture"]["token"]) + await self._refused( + item, self._body(detail, observed, cat["hash"]), 409, "capture-expired" + ) + + async def test_a_candidate_that_cannot_be_applied(self): + self._require_routes() + _path, item, detail, observed, cat = await self._opened() + conn = self.api.db.model.get_connection() + (text,) = conn.execute( + "SELECT evidence FROM media_review_items WHERE id=?", (item["id"],) + ).fetchone() + evidence = json.loads(text) + for candidate in evidence["candidates"]: + if candidate["hash"] == cat["hash"]: + candidate["state"] = "unresolved" + text = canonical_evidence(evidence) + with conn: + conn.execute( + "UPDATE media_review_items SET evidence=?, evidence_hash=? WHERE id=?", + (text, evidence_digest(text), item["id"]), + ) + data = await self._refused(item, self._body(detail, observed, cat["hash"]), 400) + self.assertIn("cannot be applied", data["error"]) + + async def test_leave_unresolved_with_another_digest(self): + self._require_routes() + _path, item, detail, observed, _cat = await self._opened() + body = { + **self._body(detail, observed, "unresolved"), + "capture_token": None, + "sha256": "0" * 64, + } + data = await self._refused(item, body, 400) + self.assertIn("does not match the item", data["error"]) + + async def test_root_no_longer_configured_is_source_changed(self): + self._require_routes() + _path, item, detail, observed, cat = await self._opened() + # Another root takes index 0. + elsewhere = self.output_dir / "elsewhere" + elsewhere.mkdir() + GalleryConfig.MONITORING_DIRECTORIES = [self._root(elsewhere)] + await self._refused( + item, self._body(detail, observed, cat["hash"]), 409, "source-changed" + ) + + async def test_file_no_longer_under_its_root_is_source_changed(self): + self._require_routes() + _path, item, detail, observed, cat = await self._opened() + moved = os.path.join(self.tmpdir, "moved-root") + real = os.path.realpath + + def other_root(value, *args, **kwargs): + if os.path.normcase(os.fspath(value)) == os.path.normcase( + str(self.output_dir) + ): + return moved + return real(value, *args, **kwargs) + + with patch.object(review_api, "_realpath", side_effect=other_root): + await self._refused( + item, self._body(detail, observed, cat["hash"]), 409, "source-changed" + ) + + async def test_verify_failures(self): + self._require_routes() + _path, item, detail, observed, cat = await self._opened() + store = self.api._captures + body = self._body(detail, observed, cat["hash"]) + with patch.object(store, "verify", side_effect=CaptureChanged("gone")): + await self._refused(item, body, 409, "file-changed") + other = FileFacts(item_locator := self._items()[0]["locator"], "1" * 64, 1, 1) + with patch.object(store, "verify", return_value=other): + await self._refused(item, body, 409, "file-changed") + foreign = FileFacts(item_locator + "x", detail["sha256"], 1, 1) + with patch.object(store, "verify", return_value=foreign): + data = await self._refused(item, body, 409, "capture-expired") + self.assertIn("another file", data["error"]) diff --git a/utils/prompt_pair_resolver.py b/utils/prompt_pair_resolver.py index f64fee9..dfcdf7a 100644 --- a/utils/prompt_pair_resolver.py +++ b/utils/prompt_pair_resolver.py @@ -39,6 +39,12 @@ from .pair_execution import ExecutionEvidenceError, read_pair_execution _LOGGER = logging.getLogger("prompt_manager.prompt_pair_resolver") +# The ingestion resolver's version, recorded in pair-link provenance and on +# Needs review items. Bump it when a change would resolve the same metadata +# otherwise; a pending item stays applicable while its chosen candidate is +# still offered by the captured bytes. +RESOLVER_VERSION = 1 + @dataclass(frozen=True) class OutputAdapter: diff --git a/utils/review_capture.py b/utils/review_capture.py index 1af9bd7..d136a2c 100644 --- a/utils/review_capture.py +++ b/utils/review_capture.py @@ -31,18 +31,16 @@ try: from ..database.review_schema import MAX_CANDIDATES, canonical_evidence from .media_capture import CaptureCapacity, CaptureError from .prompt_inspector import MetadataReadError, read_inspection_file - from .prompt_pair_resolver import resolve_prompt_pair + from .prompt_pair_resolver import RESOLVER_VERSION, resolve_prompt_pair except ImportError: # imported with the repo root on sys.path from database.library_commands import LibraryCommandError from database.review_schema import MAX_CANDIDATES, canonical_evidence from utils.media_capture import CaptureCapacity, CaptureError from utils.prompt_inspector import MetadataReadError, read_inspection_file - from utils.prompt_pair_resolver import resolve_prompt_pair + from utils.prompt_pair_resolver import RESOLVER_VERSION, resolve_prompt_pair logger = logging.getLogger("prompt_manager.review_capture") -# The ingestion resolver's provenance version (``database.pair_ingestion``). -RESOLVER_VERSION = 1 # Capture scope label: review captures are never handed to a client. REVIEW_SCOPE = "needs-review" # Nested ambiguous stages are flattened at most this deep.