diff --git a/TESTING.md b/TESTING.md index e33db33..199525a 100644 --- a/TESTING.md +++ b/TESTING.md @@ -1747,7 +1747,9 @@ Review queue schema (#249, `database/review_schema.py`, `tests/test_review_schema.py`): the group `media_review_metadata` (one row: schema version, `review_index_revision`), `media_review_items` (one ambiguous resolver result per `(byte_id, resolver_version, evidence_hash)`, citing an -annotation byte record `ON DELETE RESTRICT`, with 0..16 candidates, `limited`, +annotation byte record `ON DELETE RESTRICT`, with the file's recorded +`spelled_path` (the locator is case-folded on Windows, a listing target is +not), 0..16 candidates, `limited`, `state` pending/selected/unresolved and a revision) and `media_review_decisions` (one row per item decision revision; `selected` exactly when a `candidate_hash` is present; `pair_id`/`image_id` navigation is set to NULL, @@ -1786,7 +1788,47 @@ unchanged; the job's `scope` gains `review_items` (new items) and `review_errors`, and the log ends with `? N images added to Needs review` (and `! N images could not be added to Needs review` on failures, e.g. capture capacity). The monitor records after its fallback link, best effort: a failure -is logged and the link stays. A new item moves `review_index_revision`. +is logged and the link stays. A new item moves `review_index_revision`. A +rescan does not copy or hash an ambiguous file whose current size and +`mtime_ns` equal the verified fingerprint of bytes an item already cites (a +stat compare); the file still stays in the default "new" scope, because it is +not a known image. + +Needs review decisions and queue reads (#249, `py/api/review.py`, +`database/review.py`, `database/review_decisions.py`, +`tests/test_api_review.py`): `GET /prompt_manager/review/items?limit=1..40&cursor=` +answers `{success, library_epoch, total, items, next_cursor}` with pending items +oldest first; `total` is the exact pending count. The cursor is an HMAC keyset +on `(created_at, id)` bound to the storage epoch and `review_index_revision`: a +forged one is 400, one from before a restore is 409 `storage-changed`, one from +before any queue change (a decision, a new item) is 409 `review-conflict`. +Targets are formatted from `spelled_path` against the current roots (the +roots' real paths are the only file access); a file under no current root is +`unavailable` with a null target. `GET .../items/{id}` adds `candidates` +(`hash`, `sampler_id`, `state` resolved/unresolved, `reason`, `label` " / " from the saved prompt's Library title else the +first line, and the role texts), `dependency_fingerprint` (SHA-256 of the +file's exact legacy rows, positive hashes and pair links with their negative +hashes, as the ingester reads them) and `sha256`; a detail over 8 MiB, or a +limited item, has no candidates. `POST .../items/{id}/decision +{expected_epoch, expected_revision, capture_token, sha256, dependency_fingerprint, +choice}` with a candidate hash (Apply) requires a capture of the item's file +whose bytes are the item's, re-hashes the current file (`file-changed` +otherwise, also for a fresh capture of new bytes), re-reads the capture and +requires the same evidence digest (`review-conflict` otherwise), then in one +`_pair_transaction` checks the epoch, the item revision and state +(`review-conflict` with the current item for a repeat), the dependency +fingerprint and the public ingester's compatible-only rules through +`_ingest_resolved_pair_in` (either refusal is 409 `association-conflict` +with zero writes, explained inline; Apply never reparents a link), and writes +the link with `kind:"user-selection"` provenance (decision id, reviewed digest, +candidate hash, resolver version, role provenance), the decision row, the +item's new state and revision and `review_index_revision`. `"unresolved"` +records the decision only and accepts `capture_token: null`. A successful +decision releases its capture. No decision changes run or usage counts or +calls an execution writer; a Tag save never changes a review item. Errors carry +fixed sentences: `review-failed` / `decision-failed` 500s never echo exception +text or paths. Capture store (#249, `utils/media_capture.py`, `tests/test_media_capture.py`): one explicit action captures one rooted media file into an owner-only diff --git a/database/operations.py b/database/operations.py index bce3446..803f467 100644 --- a/database/operations.py +++ b/database/operations.py @@ -14,6 +14,7 @@ from .library import LibraryOperations, titled from .image_favorites import ImageFavoriteOperations from .annotations import AnnotationOperations from .review import ReviewOperations +from .review_decisions import ReviewDecisionOperations from .saved_views import SavedViewOperations from .library_commands import LibraryCommandOperations from .vocabulary import VocabularyOperations @@ -343,6 +344,7 @@ class PromptDatabase( ImageFavoriteOperations, AnnotationOperations, ReviewOperations, + ReviewDecisionOperations, SavedViewOperations, ): """Database operations class for managing prompts.""" diff --git a/database/pair_ingestion.py b/database/pair_ingestion.py index cad3641..c23184c 100644 --- a/database/pair_ingestion.py +++ b/database/pair_ingestion.py @@ -19,118 +19,144 @@ def ingest_resolved_pair( state = resolution.get("state", "unresolved") if state != "resolved": return {"status": state, "pairs_added": 0, "links_added": 0} + with db._pair_transaction() as conn: + return _ingest_resolved_pair_in(conn, db, image_path, metadata, resolution) + + +def _conflict() -> Dict[str, Any]: + return {"status": "conflict", "pairs_added": 0, "links_added": 0} + + +def _ingest_resolved_pair_in( + conn: Any, + db: Any, + image_path: str, + metadata: Dict[str, Any], + resolution: Dict[str, Any], + *, + provenance: Dict[str, Any] | None = None, +) -> Dict[str, Any]: + """``ingest_resolved_pair`` inside the caller's open ``_pair_transaction``. + + The compatible-only rules are the public ingester's: an image with more + than one exact row, another positive, or a prior pair with another + negative is a ``conflict`` and nothing is written. The caller owns the + transaction, so its own later failure rolls the link back too. + ``provenance`` replaces the link's default resolver provenance (Needs + review records the user's selection there). + """ positive, negative = resolution["positive"], resolution["negative"] positive_text = positive.get("storage_text", positive["text"]) negative_text = negative.get("storage_text", negative["text"]) normalized = normalize_image_path(image_path) - provenance = {"resolver_version": 1, "positive": positive, "negative": negative} + if provenance is None: + provenance = { + "resolver_version": 1, + "positive": positive, + "negative": negative, + } encoded = json.dumps( provenance, sort_keys=True, separators=(",", ":"), allow_nan=False ) - with db._pair_transaction() as conn: - images = conn.execute( - "SELECT gi.*,p.hash AS positive_hash FROM generated_images gi " - "JOIN prompts p ON p.id=gi.prompt_id " - "WHERE gi.file_path=? OR gi.image_path=?", - (normalized, image_path), + images = conn.execute( + "SELECT gi.*,p.hash AS positive_hash FROM generated_images gi " + "JOIN prompts p ON p.id=gi.prompt_id " + "WHERE gi.file_path=? OR gi.image_path=?", + (normalized, image_path), + ).fetchall() + if len(images) > 1 or ( + images and images[0]["positive_hash"] != generate_prompt_hash(positive_text) + ): + return _conflict() + existing = images[0] if images else None + if existing: + prior = conn.execute( + "SELECT n.hash AS negative_hash FROM image_prompt_pairs ip " + "JOIN prompt_pairs pp ON pp.id=ip.pair_id " + "LEFT JOIN prompts n ON n.id=pp.negative_prompt_id WHERE ip.image_id=?", + (existing["id"],), ).fetchall() - if len(images) > 1 or ( - images and images[0]["positive_hash"] != generate_prompt_hash(positive_text) - ): - return {"status": "conflict", "pairs_added": 0, "links_added": 0} - existing = images[0] if images else None - if existing: - prior = conn.execute( - "SELECT n.hash AS negative_hash FROM image_prompt_pairs ip " - "JOIN prompt_pairs pp ON pp.id=ip.pair_id " - "LEFT JOIN prompts n ON n.id=pp.negative_prompt_id WHERE ip.image_id=?", - (existing["id"],), - ).fetchall() - expected = ( - generate_prompt_hash(negative_text) if negative_text.strip() else None - ) - if any(row["negative_hash"] != expected for row in prior): - return {"status": "conflict", "pairs_added": 0, "links_added": 0} - prior_pair = conn.execute( - "SELECT pp.id FROM prompt_pairs pp " - "JOIN prompts p ON p.id=pp.positive_prompt_id " - "LEFT JOIN prompts n ON n.id=pp.negative_prompt_id " - "WHERE p.hash=? AND n.hash IS ?", - ( - generate_prompt_hash(positive_text), - generate_prompt_hash(negative_text) if negative_text.strip() else None, - ), - ).fetchone() - pair_id = db._save_pair(conn, positive_text, negative_text) - pair = conn.execute( - "SELECT * FROM prompt_pairs WHERE id=?", (pair_id,) - ).fetchone() - pairs_added = int(prior_pair is None) - file_size, width, height, image_format = _coerce_file_info(metadata) - snapshots = { - column: json.dumps(metadata[key], sort_keys=True, allow_nan=False) - for column, key in ( - ("workflow_data", "workflow"), - ("prompt_metadata", "prompt"), - ("parameters", "parameters"), - ) - if key in metadata - } - info = { - "file_size": file_size, - "width": width, - "height": height, - "format": image_format, - } - snapshots.update( - {key: value for key, value in info.items() if value is not None} + expected = ( + generate_prompt_hash(negative_text) if negative_text.strip() else None ) - changed = False - if existing: - image_id = existing["id"] - updates = { - key: value for key, value in snapshots.items() if existing[key] != value - } - if updates: - conn.execute( - "UPDATE generated_images SET " - + ",".join(f"{key}=?" for key in updates) - + " WHERE id=?", - [*updates.values(), image_id], - ) - changed = True - else: - columns = ["prompt_id", "image_path", "filename", "file_path", *snapshots] - values = [ - pair["positive_prompt_id"], - image_path, - os.path.basename(image_path), - normalized, - *snapshots.values(), - ] - image_id = conn.execute( - "INSERT INTO generated_images(" - + ",".join(columns) - + ") VALUES(" - + ",".join("?" for _ in columns) - + ")", - values, - ).lastrowid - linked = conn.execute( - "SELECT 1 FROM image_prompt_pairs WHERE pair_id=? " - "AND image_id=? AND provenance=?", - (pair_id, image_id, encoded), - ).fetchone() - db._link_image_to_pair(conn, pair_id, image_id, provenance) - status = ( - "added" - if existing is None - else "enriched" if changed or not linked else "unchanged" + if any(row["negative_hash"] != expected for row in prior): + return _conflict() + prior_pair = conn.execute( + "SELECT pp.id FROM prompt_pairs pp " + "JOIN prompts p ON p.id=pp.positive_prompt_id " + "LEFT JOIN prompts n ON n.id=pp.negative_prompt_id " + "WHERE p.hash=? AND n.hash IS ?", + ( + generate_prompt_hash(positive_text), + generate_prompt_hash(negative_text) if negative_text.strip() else None, + ), + ).fetchone() + pair_id = db._save_pair(conn, positive_text, negative_text) + pair = conn.execute("SELECT * FROM prompt_pairs WHERE id=?", (pair_id,)).fetchone() + pairs_added = int(prior_pair is None) + file_size, width, height, image_format = _coerce_file_info(metadata) + snapshots = { + column: json.dumps(metadata[key], sort_keys=True, allow_nan=False) + for column, key in ( + ("workflow_data", "workflow"), + ("prompt_metadata", "prompt"), + ("parameters", "parameters"), ) - return { - "status": status, - "pairs_added": pairs_added, - "links_added": int(not linked), - "pair_id": pair_id, - "image_id": image_id, + if key in metadata + } + info = { + "file_size": file_size, + "width": width, + "height": height, + "format": image_format, + } + snapshots.update({key: value for key, value in info.items() if value is not None}) + changed = False + if existing: + image_id = existing["id"] + updates = { + key: value for key, value in snapshots.items() if existing[key] != value } + if updates: + conn.execute( + "UPDATE generated_images SET " + + ",".join(f"{key}=?" for key in updates) + + " WHERE id=?", + [*updates.values(), image_id], + ) + changed = True + else: + columns = ["prompt_id", "image_path", "filename", "file_path", *snapshots] + values = [ + pair["positive_prompt_id"], + image_path, + os.path.basename(image_path), + normalized, + *snapshots.values(), + ] + image_id = conn.execute( + "INSERT INTO generated_images(" + + ",".join(columns) + + ") VALUES(" + + ",".join("?" for _ in columns) + + ")", + values, + ).lastrowid + linked = conn.execute( + "SELECT 1 FROM image_prompt_pairs WHERE pair_id=? " + "AND image_id=? AND provenance=?", + (pair_id, image_id, encoded), + ).fetchone() + db._link_image_to_pair(conn, pair_id, image_id, provenance) + status = ( + "added" + if existing is None + else "enriched" if changed or not linked else "unchanged" + ) + return { + "status": status, + "pairs_added": pairs_added, + "links_added": int(not linked), + "pair_id": pair_id, + "image_id": image_id, + } diff --git a/database/review.py b/database/review.py index 73428cf..160fdfd 100644 --- a/database/review.py +++ b/database/review.py @@ -12,20 +12,29 @@ nothing here reads the filesystem. Writes run in one ``_pair_transaction`` fenced by the storage epoch the capture was taken under. """ +import hashlib import json +import os +import re import sqlite3 from typing import Iterator +try: + from ..utils.hashing import generate_prompt_hash +except ImportError: + from utils.hashing import generate_prompt_hash + from .annotation_schema import ( BYTES_TABLE, FILES_TABLE, METADATA_TABLE as ANNOTATION_METADATA_TABLE, ) from .annotations import _new_id, _validate_locator, _validate_target -from .models import utc_now_iso +from .models import normalize_image_path, utc_now_iso from .review_schema import ( ITEMS_TABLE, MAX_CANDIDATES, + MAX_SPELLED_PATH_BYTES, METADATA_TABLE, canonical_evidence, evidence_digest, @@ -33,9 +42,9 @@ from .review_schema import ( # Only this module's table constants are interpolated into the SQL below. _INSERT_ITEM = ( - f"INSERT OR IGNORE INTO {ITEMS_TABLE}(id,byte_id,resolver_version," # nosec B608 - "evidence_hash,evidence,candidate_count,limited,state,revision,created_at," - "updated_at) VALUES(?,?,?,?,?,?,?,'pending',0,?,?)" + f"INSERT OR IGNORE INTO {ITEMS_TABLE}(id,byte_id,spelled_path," # nosec B608 + "resolver_version,evidence_hash,evidence,candidate_count,limited,state," + "revision,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,'pending',0,?,?)" ) _BUMP_REVIEW_INDEX = ( f"UPDATE {METADATA_TABLE} SET " # nosec B608 @@ -57,6 +66,73 @@ _ITEMS_OF_LOCATOR = ( f"JOIN {ITEMS_TABLE} i ON i.byte_id=b.id " "WHERE f.locator=? ORDER BY i.created_at,i.id" ) +_QUEUED = ( + f"SELECT 1 FROM {FILES_TABLE} f " # nosec B608 + f"JOIN {BYTES_TABLE} b ON b.file_id=f.id " + f"JOIN {ITEMS_TABLE} i ON i.byte_id=b.id " + "WHERE f.locator=? AND b.size=? AND b.mtime_ns=? AND i.resolver_version=? " + "LIMIT 1" +) +_EPOCH = "SELECT storage_epoch FROM execution_history_metadata WHERE singleton_id=1" +_PENDING_COUNT = ( + f"SELECT COUNT(*) FROM {ITEMS_TABLE} WHERE state='pending'" # nosec B608 +) +_QUEUE_COLUMNS = ( + "id", + "state", + "revision", + "created_at", + "candidate_count", + "limited", + "spelled_path", + "locator", +) +_QUEUE_SELECT = ( + "SELECT i.id,i.state,i.revision,i.created_at,i.candidate_count," # nosec B608 + f"i.limited,i.spelled_path,f.locator FROM {ITEMS_TABLE} i " + f"JOIN {BYTES_TABLE} b ON b.id=i.byte_id " + f"JOIN {FILES_TABLE} f ON f.id=b.file_id " +) +_QUEUE_FIRST = _QUEUE_SELECT + ( + "WHERE i.state='pending' ORDER BY i.created_at,i.id LIMIT ?" +) +_QUEUE_AFTER = _QUEUE_SELECT + ( + "WHERE i.state='pending' AND (i.created_at,i.id)>(?,?) " + "ORDER BY i.created_at,i.id LIMIT ?" +) +_DETAIL_COLUMNS = _QUEUE_COLUMNS + ( + "byte_id", + "resolver_version", + "evidence_hash", + "evidence", + "updated_at", + "sha256", +) +_DETAIL = ( + "SELECT i.id,i.state,i.revision,i.created_at,i.candidate_count," # nosec B608 + "i.limited,i.spelled_path,f.locator,i.byte_id,i.resolver_version," + "i.evidence_hash,i.evidence,i.updated_at,b.sha256 " + f"FROM {ITEMS_TABLE} i JOIN {BYTES_TABLE} b ON b.id=i.byte_id " + f"JOIN {FILES_TABLE} f ON f.id=b.file_id WHERE i.id=?" +) +# The legacy rows and pair links the ingester's compatibility rules read. +_DEPENDENT_IMAGES = ( + "SELECT gi.id,p.hash 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" +) +_DEPENDENT_LINKS = ( + "SELECT DISTINCT ip.pair_id,n.hash FROM image_prompt_pairs ip " + "JOIN prompt_pairs pp ON pp.id=ip.pair_id " + "LEFT JOIN prompts n ON n.id=pp.negative_prompt_id " + "WHERE ip.image_id=? ORDER BY ip.pair_id" +) +_TITLES = ( + "SELECT p.hash,s.title FROM prompts p " + "JOIN prompt_library_state s ON s.prompt_id=p.id " + "WHERE p.hash IN ({marks}) AND s.title IS NOT NULL ORDER BY p.id" +) +_ITEM_ID = re.compile(r"[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}") _ITEM_COLUMNS = ( "id", "byte_id", @@ -73,6 +149,34 @@ _ITEM_COLUMNS = ( ) +def _validate_spelling(spelled_path: object, locator: str) -> None: + if ( + not isinstance(spelled_path, str) + or not spelled_path + or len(spelled_path.encode("utf-8", "surrogatepass")) > MAX_SPELLED_PATH_BYTES + or os.path.normcase(spelled_path) != locator + ): + raise ValueError("spelled_path must be the file's spelling of its locator") + + +def dependency_fingerprint(conn: sqlite3.Connection, image_path: str) -> 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. + """ + images = conn.execute( + _DEPENDENT_IMAGES, (normalize_image_path(image_path), image_path) + ).fetchall() + rows = [] + for image_id, positive_hash in images: + 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}) + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + class ReviewOperations: """PromptDatabase mixin: insert and read Needs review items.""" @@ -80,6 +184,7 @@ class ReviewOperations: self, *, locator: str, + spelled_path: str, sha256: str, size: int, mtime_ns: int, @@ -92,11 +197,14 @@ class ReviewOperations: """Store one ambiguous result for these exact bytes; True when new. The byte record is created (or its verified fingerprint refreshed) - like an observe. Raises ``LibraryCommandError("storage-changed")`` when - the storage epoch is no longer ``epoch`` (a restore since the capture) - and ``ValueError`` for invalid facts or evidence. + like an observe. ``spelled_path`` is the file's own spelling (it must + fold to ``locator``); targets are formatted from it. Raises + ``LibraryCommandError("storage-changed")`` when the storage epoch is + no longer ``epoch`` (a restore since the capture) and ``ValueError`` + for invalid facts or evidence. """ _validate_target(locator, sha256, size, mtime_ns) + _validate_spelling(spelled_path, locator) if type(resolver_version) is not int or resolver_version < 1: raise ValueError("resolver_version must be a positive integer") if ( @@ -122,6 +230,7 @@ class ReviewOperations: ( _new_id(), byte_id, + spelled_path, resolver_version, evidence_hash, text, @@ -155,3 +264,91 @@ class ReviewOperations: item["evidence"] = json.loads(item["evidence"]) item["limited"] = bool(item["limited"]) yield item + + # -- queue reads (#249 E3) ------------------------------------------------- + + def review_fingerprint_queued( + self, locator: str, size: int, mtime_ns: int, resolver_version: int + ) -> bool: + """True when an item cites a byte record verified at this exact stat. + + A rescan uses it to skip the copy and hash of a file it already + queued: the same ``(size, mtime_ns)`` the bytes were verified with. + """ + row = ( + self.model.get_connection() + .execute(_QUEUED, (locator, size, mtime_ns, resolver_version)) + .fetchone() + ) + return row is not None + + def review_queue_page(self, *, limit: int, after: tuple | None = None) -> dict: + """One page of pending items, oldest first, from one snapshot. + + ``after`` is the ``(created_at, id)`` keyset of the last item already + shown. Returns ``{epoch, revision, total, items, more}``; ``total`` is + the exact pending count and ``items`` carry their ``spelled_path`` and + ``locator`` (server-side only). + """ + if type(limit) is not int or limit < 1: + raise ValueError("limit must be a positive integer") + with self._annotation_read() as conn: + epoch = conn.execute(_EPOCH).fetchone() + revision = conn.execute(_REVIEW_INDEX).fetchone() + if epoch is None or revision is None: + raise sqlite3.DatabaseError("Missing review queue metadata") + total = conn.execute(_PENDING_COUNT).fetchone()[0] + if after is None: + rows = conn.execute(_QUEUE_FIRST, (limit + 1,)).fetchall() + else: + rows = conn.execute(_QUEUE_AFTER, (*after, limit + 1)).fetchall() + items = [_queue_row(row, _QUEUE_COLUMNS) for row in rows[:limit]] + return { + "epoch": epoch[0], + "revision": revision[0], + "total": total, + "items": items, + "more": len(rows) > limit, + } + + def review_item(self, item_id: str) -> dict | None: + """One item with its parsed evidence and the pending total, or None.""" + if not isinstance(item_id, str) or _ITEM_ID.fullmatch(item_id) is None: + return None + with self._annotation_read() as conn: + epoch = conn.execute(_EPOCH).fetchone() + if epoch is None: + raise sqlite3.DatabaseError("Missing execution history metadata") + row = conn.execute(_DETAIL, (item_id,)).fetchone() + if row is None: + return None + item = _queue_row(row, _DETAIL_COLUMNS) + item["evidence"] = json.loads(item["evidence"]) + item["epoch"] = epoch[0] + item["total"] = conn.execute(_PENDING_COUNT).fetchone()[0] + return item + + def review_dependency_fingerprint(self, image_path: str) -> str: + with self._annotation_read() as conn: + return dependency_fingerprint(conn, image_path) + + def stored_prompt_titles(self, texts) -> dict: + """Library titles of saved prompts by text (dedup hash), when set.""" + hashes = {generate_prompt_hash(text): text for text in texts if text.strip()} + if not hashes: + return {} + marks = ",".join("?" * len(hashes)) + found: dict[str, tuple] = {} + conn = self.model.get_connection() + for prompt_hash, title in conn.execute( + _TITLES.format(marks=marks), list(hashes) + ): + text = hashes[prompt_hash] + found[text] = found.get(text, ()) + (title,) + return found + + +def _queue_row(row, columns) -> dict: + item = dict(zip(columns, tuple(row))) + item["limited"] = bool(item["limited"]) + return item diff --git a/database/review_decisions.py b/database/review_decisions.py new file mode 100644 index 0000000..b305927 --- /dev/null +++ b/database/review_decisions.py @@ -0,0 +1,252 @@ +"""Needs review decisions (#249 E3): explicit, auditable and compatible-only. + +A decision is one row of ``media_review_decisions`` for one item revision. +``selected`` links the chosen candidate's pair to the file through the public +ingester's compatible-only rules (``_ingest_resolved_pair_in``) in the same +``_pair_transaction`` as the decision row, the item's new state and revision +and the review index revision; any conflict or later failure rolls all of it +back. ``unresolved`` records the decision and links nothing. No decision +counts a run, touches usage or calls an execution writer. + +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 sqlite3 + +from .annotation_schema import BYTES_TABLE +from .annotations import _new_id +from .library_commands import LibraryCommandError +from .models import utc_now_iso +from .review import ( + _BUMP_REVIEW_INDEX, + _DETAIL, + _DETAIL_COLUMNS, + _PENDING_COUNT, + _queue_row, + dependency_fingerprint, +) +from .review_schema import DECISIONS_TABLE, ITEMS_TABLE, canonical_evidence + +DECISION_KIND = "user-selection" +_INSERT_DECISION = ( + f"INSERT INTO {DECISIONS_TABLE}(id,item_id,decision_revision," # nosec B608 + "created_at,outcome,candidate_hash,evidence,pair_id,image_id) " + "VALUES(?,?,?,?,?,?,?,?,?)" +) +_ADVANCE_ITEM = ( + f"UPDATE {ITEMS_TABLE} SET state=?,revision=revision+1," # nosec B608 + "updated_at=? WHERE id=? AND revision=?" +) +_ITEM_BYTES = ( + f"SELECT b.sha256 FROM {ITEMS_TABLE} i " # nosec B608 + f"JOIN {BYTES_TABLE} b ON b.id=i.byte_id WHERE i.id=?" +) + + +class ReviewCommandError(LibraryCommandError): + """A decision lost a race or would rewrite an association. + + ``code`` is ``review-conflict`` (the item was decided or changed since it + was read), ``association-conflict`` (the choice is not compatible with + the file's saved links, or those links changed), ``file-changed`` (the + item is not about these bytes) or ``review-missing``. ``item`` is the + current item read inside the transaction, ``total`` the pending count and + ``fingerprint`` the current dependency fingerprint, when known; + ``changed`` marks an association conflict caused by links that changed + since the reviewer read them (rather than an incompatible choice). + """ + + def __init__( + self, code, epoch, *, item=None, total=None, fingerprint=None, changed=False + ): + super().__init__(code, epoch) + self.item = item + self.total = total + self.fingerprint = fingerprint + self.changed = changed + + +class _Rollback(Exception): + """Raised inside the transaction so the context manager rolls back.""" + + def __init__(self, error: ReviewCommandError): + super().__init__(error.code) + self.error = error + + +class ReviewDecisionOperations: + """PromptDatabase mixin: record one decision on one Needs review item.""" + + def decide_review_item( + self, + *, + item_id: str, + expected_epoch: str, + expected_revision: int, + sha256: str, + outcome: str, + candidate: dict | None = None, + image_path: str | None = None, + metadata: dict | None = None, + expected_fingerprint: str | 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. + Raises ``LibraryCommandError("storage-changed")`` after a restore and + :class:`ReviewCommandError` for every refused decision, with zero + writes either way. + """ + if outcome not in ("selected", "unresolved"): + raise ValueError("outcome must be selected or unresolved") + if outcome == "selected" and ( + not isinstance(candidate, dict) or not image_path + ): + raise ValueError("A selected decision needs a candidate and a file") + try: + with self._pair_transaction() as conn: + return self._decide_in( + conn, + item_id=item_id, + expected_epoch=expected_epoch, + expected_revision=expected_revision, + sha256=sha256, + outcome=outcome, + candidate=candidate, + image_path=image_path, + metadata=metadata or {}, + expected_fingerprint=expected_fingerprint, + ) + except _Rollback as rollback: + raise rollback.error from None + + def _decide_in(self, conn: sqlite3.Connection, **command) -> dict: + epoch = self._library_epoch(conn, command["expected_epoch"]) + item_id = command["item_id"] + current = _current_item(conn, item_id) + if current is None: + raise _Rollback(ReviewCommandError("review-missing", epoch)) + total = conn.execute(_PENDING_COUNT).fetchone()[0] + if ( + current["state"] != "pending" + or current["revision"] != command["expected_revision"] + ): + raise _Rollback( + ReviewCommandError("review-conflict", epoch, item=current, total=total) + ) + if conn.execute(_ITEM_BYTES, (item_id,)).fetchone()[0] != command["sha256"]: + raise _Rollback(ReviewCommandError("file-changed", epoch)) + decision_id = _new_id() + pair = None + candidate = command["candidate"] if command["outcome"] == "selected" else None + if candidate is not None: + pair = self._apply_candidate( + conn, epoch, current, decision_id, total, **command + ) + evidence = { + "kind": DECISION_KIND, + "decision_id": decision_id, + "outcome": command["outcome"], + "reviewed_digest": command["sha256"], + "item_evidence_hash": current["evidence_hash"], + "resolver_version": current["resolver_version"], + "candidate": candidate, + "dependency_fingerprint": command["expected_fingerprint"], + "pair": pair, + } + now = utc_now_iso() + self._insert_review_decision( + conn, + ( + decision_id, + item_id, + current["revision"] + 1, + now, + command["outcome"], + candidate["hash"] if candidate is not None else None, + canonical_evidence(evidence), + pair["id"] if pair else None, + pair["image_id"] if pair else None, + ), + ) + conn.execute( + _ADVANCE_ITEM, (command["outcome"], now, item_id, current["revision"]) + ) + conn.execute(_BUMP_REVIEW_INDEX) + return { + "item": _current_item(conn, item_id), + "total": conn.execute(_PENDING_COUNT).fetchone()[0], + "pair": pair, + "decision_id": decision_id, + } + + def _apply_candidate( + self, conn, epoch, current, decision_id, total, **command + ) -> dict: + """Link the candidate's pair, compatible-only; raise on any conflict.""" + # Imported here: pair_ingestion imports operations, which composes this. + from .pair_ingestion import _ingest_resolved_pair_in + + image_path = command["image_path"] + fingerprint = dependency_fingerprint(conn, image_path) + if fingerprint != command["expected_fingerprint"]: + raise _Rollback( + ReviewCommandError( + "association-conflict", + epoch, + item=current, + total=total, + fingerprint=fingerprint, + changed=True, + ) + ) + candidate = command["candidate"] + positive, negative = candidate["positive"], candidate["negative"] + provenance = { + "kind": DECISION_KIND, + "decision_id": decision_id, + "reviewed_digest": command["sha256"], + "candidate_hash": candidate["hash"], + "resolver_version": current["resolver_version"], + "positive": positive, + "negative": negative, + } + result = _ingest_resolved_pair_in( + conn, + self, + image_path, + command["metadata"], + {"state": "resolved", "positive": positive, "negative": negative}, + provenance=provenance, + ) + if result["status"] == "conflict": + raise _Rollback( + ReviewCommandError( + "association-conflict", + epoch, + item=current, + total=total, + fingerprint=fingerprint, + ) + ) + return { + "id": result["pair_id"], + "image_id": result["image_id"], + "status": result["status"], + } + + @staticmethod + def _insert_review_decision(conn: sqlite3.Connection, values: tuple) -> None: + conn.execute(_INSERT_DECISION, values) + + +def _current_item(conn: sqlite3.Connection, item_id: str) -> dict | None: + row = conn.execute(_DETAIL, (item_id,)).fetchone() + if row is None: + return None + item = _queue_row(row, _DETAIL_COLUMNS) + item.pop("evidence") + return item diff --git a/database/review_schema.py b/database/review_schema.py index e311448..f610744 100644 --- a/database/review_schema.py +++ b/database/review_schema.py @@ -7,8 +7,10 @@ annotation group, because every item cites one annotation byte record. It owns - ``media_review_metadata``: one row, the group's schema version and the review index revision (moved by every queue change); - ``media_review_items``: one ambiguous resolver result for one byte record, - keyed by ``(byte_id, resolver_version, evidence_hash)``, with its canonical - evidence, candidate count, ``limited`` flag, state and revision; + keyed by ``(byte_id, resolver_version, evidence_hash)``, with the file's + spelling when it was recorded (the locator is case-folded on Windows, a + target is not), its canonical evidence, candidate count, ``limited`` flag, + state and revision; - ``media_review_decisions``: the auditable decision history of an item, one row per decision revision, with optional navigation to the pair and image the decision linked (nulled, never cascaded, when those are deleted). @@ -40,6 +42,9 @@ DECISIONS_TABLE = "media_review_decisions" TABLES = (METADATA_TABLE, ITEMS_TABLE, DECISIONS_TABLE) MAX_EVIDENCE_BYTES = 8 * 1024 * 1024 MAX_CANDIDATES = 16 +# The file's own spelling when it was recorded (bytes of UTF-8 text). Locators +# are case-folded on Windows, so a target is formatted from this instead. +MAX_SPELLED_PATH_BYTES = 16384 ITEM_STATES = ("pending", "selected", "unresolved") DECISION_OUTCOMES = ("selected", "unresolved") _TIMESTAMP_BYTES = 64 @@ -57,6 +62,8 @@ _TABLE_DDL = { ITEMS_TABLE: ( f"CREATE TABLE {ITEMS_TABLE} (id TEXT PRIMARY KEY CHECK(length(id)=36), " f"byte_id TEXT NOT NULL REFERENCES {BYTES_TABLE}(id) ON DELETE RESTRICT, " + "spelled_path TEXT NOT NULL CHECK(length(CAST(spelled_path AS BLOB)) " + f"BETWEEN 1 AND {MAX_SPELLED_PATH_BYTES}), " "resolver_version INTEGER NOT NULL, evidence_hash TEXT NOT NULL " f"CHECK({_HEX64.format(c='evidence_hash')}), " f"evidence TEXT NOT NULL {_EVIDENCE_CHECK}, " @@ -109,6 +116,7 @@ _COLUMNS = { ITEMS_TABLE: ( ("id", "TEXT", 0, None, 1), ("byte_id", "TEXT", 1, None, 0), + ("spelled_path", "TEXT", 1, None, 0), ("resolver_version", "INTEGER", 1, None, 0), ("evidence_hash", "TEXT", 1, None, 0), ("evidence", "TEXT", 1, None, 0), @@ -243,6 +251,8 @@ _ROW_PREDICATES = { ( _uuid("id"), _uuid("byte_id"), + "typeof(spelled_path) = 'text' AND length(CAST(spelled_path AS BLOB)) " + f"BETWEEN 1 AND {MAX_SPELLED_PATH_BYTES}", "typeof(resolver_version) = 'integer'", _hex64("evidence_hash"), _evidence("evidence"), diff --git a/py/api/__init__.py b/py/api/__init__.py index b45b3d5..3eaed5e 100644 --- a/py/api/__init__.py +++ b/py/api/__init__.py @@ -7,6 +7,7 @@ Split into domain-specific mixins for maintainability: - LoggingRoutesMixin: log management endpoints - AutotagRoutesMixin: auto-tagging model management and tagging - JobRoutesMixin: the session job registry behind Activity (py/api/jobs.py) +- ReviewRoutesMixin: the Needs review queue, item detail and decisions """ import asyncio @@ -27,6 +28,7 @@ from .pairs import PairRoutesMixin from .library import LibraryRoutesMixin from .vocabulary import VocabularyRoutesMixin from .annotations import AnnotationRoutesMixin +from .review import ReviewRoutesMixin from .saved_views import SavedViewRoutesMixin from .history import HistoryRoutesMixin from .images import ImageRoutesMixin @@ -254,6 +256,7 @@ class PromptManagerAPI( LibraryRoutesMixin, VocabularyRoutesMixin, AnnotationRoutesMixin, + ReviewRoutesMixin, SavedViewRoutesMixin, HistoryRoutesMixin, PromptRoutesMixin, @@ -448,6 +451,7 @@ class PromptManagerAPI( self._register_library_routes(routes) self._register_vocabulary_routes(routes) self._register_annotation_routes(routes) + self._register_review_routes(routes) self._register_saved_view_routes(routes) self._register_history_routes(routes) self._register_prompt_routes(routes) diff --git a/py/api/review.py b/py/api/review.py new file mode 100644 index 0000000..5e39635 --- /dev/null +++ b/py/api/review.py @@ -0,0 +1,579 @@ +"""Needs review routes (#249 E3): the pending queue, item detail, decisions. + +``GET /prompt_manager/review/items`` lists pending items oldest first behind +an HMAC keyset cursor bound to the storage epoch and the review index +revision (any decision, new item or restore makes an old cursor stale); +``total`` is the exact pending count. ``GET .../items/{id}`` adds the +candidates and the dependency fingerprint. ``POST .../items/{id}/decision`` +records one explicit decision: Apply links the chosen sampler's pair only +when the capture is the item's bytes, the current file still hashes to them, +the evidence they yield is unchanged and the ingester's compatible-only rules +allow it; Leave unresolved links nothing. + +Targets are formatted from the recorded spelling against the current roots +(no file access beyond the roots' real paths); a file under no current root +is ``unavailable``. Responses carry no server path, no locator and no +exception text. +""" + +import base64 +import binascii +import hashlib +import hmac +import json +import os +import re + +from aiohttp import web + +from .annotations import _HEX64, _TOKEN, _UUID, _capture_error, _error +from .prompts import bad_request + +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 + +ITEMS_URL = "/prompt_manager/review/items" +DEFAULT_LIMIT = 40 +MAX_LIMIT = 40 +MAX_CURSOR = 4096 +MAX_BODY_BYTES = 64 * 1024 +# The serialized detail stays under this; past it the detail is ``limited``. +MAX_DETAIL_BYTES = 8 * 1024 * 1024 +UNRESOLVED = "unresolved" +_DECISION_FIELDS = frozenset( + { + "expected_epoch", + "expected_revision", + "capture_token", + "sha256", + "dependency_fingerprint", + "choice", + } +) +_CURSOR_SECRET = os.urandom(32) +_LIMIT = re.compile(r"[0-9]{1,3}") +_NO_STORE = {"Cache-Control": "no-store"} + +_STATUS = { + "review-missing": 404, + "review-conflict": 409, + "association-conflict": 409, + "file-changed": 409, + "storage-changed": 409, + "source-changed": 409, + "capture-expired": 409, +} +_MESSAGES = { + "review-missing": "This review item no longer exists. Reload the queue.", + "review-conflict": ( + "This item was decided or changed since it was opened. Reload the queue." + ), + "association-conflict": ( + "This image is already linked to another prompt or negative, so this " + "sampler path cannot be attached safely. The existing links were " + "preserved." + ), + "dependencies-changed": ( + "The image's saved links changed since this item was opened. Reopen it " + "to see the current association." + ), + "file-changed": ( + "The file changed since it was reviewed. A metadata scan adds its new " + "bytes as a new item." + ), + "storage-changed": "Prompt storage was restored; reload the review queue.", + "source-changed": ( + "The source changed. Refresh the image collection before reviewing this " + "file." + ), + "capture-expired": "The preview expired. Open the item again.", + "other-file": "The preview belongs to another file. Open the item again.", + "evidence-changed": ( + "This file's metadata no longer gives the evidence that was reviewed. " + "Reload the queue." + ), +} + + +_DEFAULT_MESSAGE = "The decision could not be saved. Reload the queue." + + +def _coded(code: str, message_key: str | None = None, **extra) -> web.Response: + message = _MESSAGES.get(message_key or code, _DEFAULT_MESSAGE) + return _error(_STATUS.get(code, 409), code, message, **extra) + + +def _failed() -> web.Response: + return _error(500, "review-failed", "The review queue could not be read.") + + +def format_review_target(spelled_path, locator, real_roots, *, pathmod=os.path): + """``(root_index, relative)`` of the first root that lists the file, or None. + + ``spelled_path`` is the file's recorded spelling, ``locator`` its folded + identity and ``real_roots`` ``(index, realpath)`` pairs. The relative + path keeps the recorded spelling (a Windows locator is case-folded, a + listing target is not) and must fold back to the locator. + """ + sep = pathmod.sep + for index, real_root in real_roots: + head = spelled_path[: len(real_root)] + if pathmod.normcase(head) != pathmod.normcase(real_root): + continue + rest = spelled_path[len(real_root) :] + if not real_root.endswith(sep): + if not rest.startswith(sep): + continue + rest = rest[1:] + parts = rest.split(sep) + if any( + part in ("", ".", "..") or "/" in part or "\\" in part or ":" in part + for part in parts + ): + continue + if pathmod.normcase(pathmod.join(real_root, *parts)) != locator: + continue + return index, "/".join(parts) + return None + + +def _encode_cursor(epoch: str, revision: int, key: tuple) -> str: + payload = json.dumps( + {"v": 1, "e": epoch, "r": revision, "k": list(key)}, + separators=(",", ":"), + ensure_ascii=True, + ).encode() + signature = hmac.new(_CURSOR_SECRET, payload, hashlib.sha256).digest() + return base64.urlsafe_b64encode(signature + payload).decode().rstrip("=") + + +def _decode_cursor(token: str) -> tuple: + """``(epoch, revision, (created_at, id))`` of a cursor; ValueError if forged.""" + try: + if len(token) > MAX_CURSOR: + raise ValueError() + raw = base64.b64decode( + token + "=" * (-len(token) % 4), altchars=b"-_", validate=True + ) + signature, payload = raw[:32], raw[32:] + expected = hmac.new(_CURSOR_SECRET, payload, hashlib.sha256).digest() + if not hmac.compare_digest(signature, expected): + raise ValueError() + data = json.loads(payload) + key = data["k"] + if ( + data["v"] != 1 + or not isinstance(data["e"], str) + or type(data["r"]) is not int + or not isinstance(key, list) + or len(key) != 2 + or not all(isinstance(part, str) for part in key) + ): + raise ValueError() + return data["e"], data["r"], tuple(key) + except (ValueError, KeyError, TypeError, UnicodeError, binascii.Error): + raise ValueError("Invalid review cursor; reload the queue") from None + + +def _limit(value): + if value is None: + return DEFAULT_LIMIT + if _LIMIT.fullmatch(value) is None or not 1 <= int(value) <= MAX_LIMIT: + return None + return int(value) + + +def _side_text(side) -> str | None: + if isinstance(side, dict) and isinstance(side.get("text"), str): + return side["text"] + return None + + +def _workable(candidate: dict) -> bool: + return ( + candidate.get("state") == "resolved" + and _side_text(candidate.get("positive")) is not None + and _side_text(candidate.get("negative")) is not None + ) + + +def _side_label(titles: dict, side, empty: str) -> str: + """The Library title of the side's saved prompt, else its first line.""" + text = _side_text(side) + if text is None: + return empty + stored = side.get("storage_text") + stored = stored if isinstance(stored, str) else text + return display_title(titles.get(stored, ()), text) or empty + + +class ReviewRoutesMixin: + """The Needs review queue over E1's schema and E2's recorded items.""" + + def _register_review_routes(self, routes: web.RouteTableDef) -> None: + routes.get(ITEMS_URL)(self.list_review_items) + routes.get(ITEMS_URL + "/{item_id}")(self.get_review_item) + routes.post(ITEMS_URL + "/{item_id}/decision")(self.decide_review_item) + + # -- shared ---------------------------------------------------------------- + + def _review_roots(self): + """``(roots, real_roots)``: the current gallery roots and their realpaths.""" + roots = list(self._get_all_output_dirs()) + return roots, [(i, os.path.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).""" + found = format_review_target(row["spelled_path"], row["locator"], real_roots) + target = None + if found is not None: + index, relative = found + target = { + "root_index": index, + "root_scope": self._gallery_root_scope(roots[index], index), + "path": relative, + } + return { + "id": row["id"], + "state": row["state"], + "revision": row["revision"], + "created_at": row["created_at"], + "filename": os.path.basename(row["spelled_path"]), + "target": target, + "unavailable": target is None, + "candidate_count": row["candidate_count"], + "limited": row["limited"], + } + + @staticmethod + def _review_image_path(row: dict, roots, target) -> str: + """The path the scan and the monitor link this file under.""" + if target is None: + return row["spelled_path"] + root = roots[target["root_index"]] + return os.path.join(str(root), *target["path"].split("/")) + + # -- queue ----------------------------------------------------------------- + + async def list_review_items(self, request: web.Request) -> web.Response: + """GET ``?limit=1..40&cursor=``: one page of pending items.""" + limit = _limit(request.query.get("limit")) + if limit is None: + return bad_request(f"limit must be an integer from 1 to {MAX_LIMIT}") + bound = None + cursor = request.query.get("cursor") + if cursor: + try: + bound = _decode_cursor(cursor) + 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, + ) + roots, real_roots = self._review_roots() + except Exception: + self.logger.exception("Review queue read failed") + return _failed() + if bound is not None and bound[0] != page["epoch"]: + return _coded("storage-changed", library_epoch=page["epoch"]) + if bound is not None and bound[1] != page["revision"]: + return _coded("review-conflict", total=page["total"]) + items = page["items"] + next_cursor = None + if page["more"] and items: + last = items[-1] + next_cursor = _encode_cursor( + page["epoch"], page["revision"], (last["created_at"], last["id"]) + ) + return web.json_response( + { + "success": True, + "library_epoch": page["epoch"], + "total": page["total"], + "items": [self._review_view(row, roots, real_roots) for row in items], + "next_cursor": next_cursor, + }, + headers=_NO_STORE, + ) + + # -- 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"] + ) + 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() + detail = { + "success": True, + "library_epoch": item["epoch"], + **view, + "candidates": candidates, + "dependency_fingerprint": fingerprint, + "sha256": item["sha256"], + "total": item["total"], + } + if len(json.dumps(detail).encode("utf-8")) > MAX_DETAIL_BYTES: + detail = {**detail, "candidates": [], "limited": True} + return web.json_response(detail, headers=_NO_STORE) + + def _review_candidates(self, item: dict) -> list: + """The detail's candidates: role texts, a label and selectability.""" + if item["limited"]: + return [] + candidates = [ + c for c in item["evidence"].get("candidates", []) if isinstance(c, dict) + ] + texts = set() + for candidate in candidates: + for role in ("positive", "negative"): + side = candidate.get(role) + if isinstance(side, dict): + for key in ("text", "storage_text"): + if isinstance(side.get(key), str): + texts.add(side[key]) + titles = self.db.stored_prompt_titles(texts) + rows = [] + for candidate in candidates: + positive = _side_text(candidate.get("positive")) + negative = _side_text(candidate.get("negative")) + label = ( + _side_label(titles, candidate.get("positive"), "No positive") + + " / " + + _side_label(titles, candidate.get("negative"), "No negative") + ) + rows.append( + { + "hash": candidate.get("hash"), + "sampler_id": candidate.get("sampler_id"), + "state": "resolved" if _workable(candidate) else "unresolved", + "reason": candidate.get("reason") or "", + "label": label, + "positive": {"text": positive} if positive is not None else None, + "negative": {"text": negative} if negative is not None else None, + } + ) + return rows + + # -- decision -------------------------------------------------------------- + + async def decide_review_item(self, request: web.Request) -> web.Response: + """POST one explicit decision on one pending item.""" + data, error = await self._read_json_body(request, max_bytes=MAX_BODY_BYTES) + if error is not None: + return error + guards, error = _decision_guards(data) + if error is not None: + return error + try: + item = await self._run_in_executor( + self.db.review_item, request.match_info["item_id"] + ) + except Exception: + self.logger.exception("Review item read failed") + return _failed() + if item is None: + return _coded("review-missing") + choice = guards["choice"] + candidate = None + if choice != UNRESOLVED: + candidate = next( + ( + c + for c in item["evidence"].get("candidates", []) + if isinstance(c, dict) and c.get("hash") == choice + ), + None, + ) + if candidate is None or item["limited"]: + return bad_request("choice is not one of this item's candidates") + if not _workable(candidate): + 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 + ) + 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) + + async def _verified_review_capture(self, item, guards, roots, *, need_metadata): + """``((capture, 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 + bytes; for Apply, the capture's metadata must still yield exactly the + reviewed evidence. + """ + store = self._captures + capture = store.get(guards["capture_token"]) if store is not None else None + if capture is None: + return None, _coded("capture-expired") + if capture.locator != item["locator"]: + return None, _coded("capture-expired", "other-file") + if guards["sha256"] != capture.sha256: + return None, bad_request("sha256 does not match the preview") + if capture.sha256 != item["sha256"]: + return None, _coded("file-changed") + 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 + ) + except CaptureError as exc: + return None, _capture_error(exc) + 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): + 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 + + async def _record_decision(self, item, guards, candidate, verified, roots): + capture, image_path, metadata = verified or (None, None, None) + try: + result = await self._run_in_executor( + self.db.decide_review_item, + item_id=item["id"], + expected_epoch=guards["expected_epoch"], + expected_revision=guards["expected_revision"], + sha256=item["sha256"], + outcome="selected" if candidate is not None else UNRESOLVED, + candidate=candidate, + image_path=image_path, + metadata=metadata, + expected_fingerprint=guards["dependency_fingerprint"], + ) + except ReviewCommandError as exc: + return self._decision_refused(exc, roots) + except LibraryCommandError as exc: + return _coded(exc.code, library_epoch=exc.epoch) + except Exception: + self.logger.exception("Review decision failed") + return _error(500, "decision-failed", "The decision could not be saved.") + 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), + "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: + 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) + message = None + if exc.fingerprint is not None: + extra["dependency_fingerprint"] = exc.fingerprint + if exc.code == "association-conflict" and exc.changed: + message = "dependencies-changed" + return _coded(exc.code, message, **extra) + + +def _decision_guards(data: dict): + """The validated decision fields, or ``(None, error)``.""" + if set(data) != _DECISION_FIELDS: + return None, bad_request( + "Request must contain exactly " + ", ".join(sorted(_DECISION_FIELDS)) + ) + epoch, revision = data["expected_epoch"], data["expected_revision"] + if not isinstance(epoch, str) or _UUID.fullmatch(epoch) is None: + return None, bad_request("expected_epoch must be a canonical UUID") + if type(revision) is not int or revision < 0: + return None, bad_request("expected_revision must be a non-negative integer") + token = data["capture_token"] + if token is not None and ( + not isinstance(token, str) or _TOKEN.fullmatch(token) is None + ): + return None, bad_request("Invalid capture token") + for name in ("sha256", "dependency_fingerprint"): + value = data[name] + if not isinstance(value, str) or _HEX64.fullmatch(value) is None: + return None, bad_request( + f"{name} must be 64 lowercase hexadecimal characters" + ) + choice = data["choice"] + if not isinstance(choice, str) or ( + choice != UNRESOLVED and _HEX64.fullmatch(choice) is None + ): + return None, bad_request("choice must be a candidate hash or unresolved") + 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.""" + facts = store.verify(root, relative) + captured = captured_review(store, capture) if need_metadata else None + return facts, captured + + +def _same_evidence(resolution: dict, item: dict) -> bool: + """True when the captured bytes still yield the item's exact evidence.""" + 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: + return False diff --git a/tests/test_api_core.py b/tests/test_api_core.py index 14f7c46..7e64377 100644 --- a/tests/test_api_core.py +++ b/tests/test_api_core.py @@ -462,6 +462,19 @@ class TestUiRoutes(CoreRoutesTestCase): ): self.assertIn(entry, table) + async def test_review_routes_are_registered(self): + table = { + (route.method, route.resource.canonical) + for route in self.app.router.routes() + if route.resource is not None + } + for entry in ( + ("GET", "/prompt_manager/review/items"), + ("GET", "/prompt_manager/review/items/{item_id}"), + ("POST", "/prompt_manager/review/items/{item_id}/decision"), + ): + self.assertIn(entry, table) + class TestStaticRoutes(CoreRoutesTestCase): diff --git a/tests/test_api_review.py b/tests/test_api_review.py new file mode 100644 index 0000000..b482b73 --- /dev/null +++ b/tests/test_api_review.py @@ -0,0 +1,754 @@ +"""Needs review decisions and queue reads (#249 E3). + +The queue lists pending items oldest first behind an HMAC cursor bound to the +storage epoch and the review index revision; ``total`` is the exact pending +count. A decision is explicit: Apply links the chosen sampler's pair only +under the ingester's compatible rules and only for the current, re-hashed +bytes of the reviewed file, in one transaction with its decision row; Leave +unresolved records the decision and links nothing. Responses never carry a +server path. +""" + +import ntpath +import os +import posixpath +import sqlite3 +from unittest.mock import patch + +try: + from tests.test_review import ( + ReviewTestCase, + ambiguous_graph, + digest, + write_png, + ) +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 utils.hashing import generate_prompt_hash + +try: # guarded so a missing module fails assertions, not test loading + from py.api import review as review_api +except ImportError: + review_api = None + +ITEMS_URL = "/prompt_manager/review/items" +# Every table a decision may touch; compared whole before and after. +WRITE_TABLES = ( + "prompts", + "generated_images", + "prompt_pairs", + "image_prompt_pairs", + "media_review_items", + "media_review_decisions", + "media_review_metadata", +) + + +class ReviewRoutesTestCase(ReviewTestCase): + def _require_routes(self): + self.assertIsNotNone(review_api, "py.api.review is not implemented") + + def _ambiguous(self, name="two.png", positives=("cat", "dog"), color=(255, 0, 0)): + path = self.output_dir / name + path.parent.mkdir(parents=True, exist_ok=True) + write_png(path, prompt=ambiguous_graph(*positives), color=color) + return path + + def _record(self, path): + self._roots() + outcome = self.api.record_review_file(str(path)) + self.assertIn(outcome, ("recorded", "limited"), outcome) + return outcome + + async def _json(self, method, url, **kwargs): + resp = await self.client.request(method, url, **kwargs) + data = await resp.json() + self.responses.append(data) + return resp.status, data + + async def asyncSetUp(self): + await super().asyncSetUp() + self.responses = [] + + async def _list(self, query=""): + status, data = await self._json("GET", ITEMS_URL + query) + self.assertEqual(status, 200, data) + return data + + async def _detail(self, item_id): + status, data = await self._json("GET", f"{ITEMS_URL}/{item_id}") + self.assertEqual(status, 200, data) + return data + + async def _observe(self, target): + status, data = await self._json( + "POST", "/prompt_manager/image-annotations/observe", json={"target": target} + ) + self.assertEqual(status, 200, data) + return data + + async def _open(self): + """The first pending item, its detail and a fresh capture of its file.""" + listed = await self._list() + item = listed["items"][0] + detail = await self._detail(item["id"]) + observed = await self._observe(item["target"]) + return item, detail, observed + + @staticmethod + def _candidate(detail, positive): + return next( + c for c in detail["candidates"] if c["positive"]["text"] == positive + ) + + def _body(self, detail, observed, choice): + return { + "expected_epoch": observed["library_epoch"], + "expected_revision": detail["revision"], + "capture_token": observed["capture"]["token"], + "sha256": observed["capture"]["sha256"], + "dependency_fingerprint": detail["dependency_fingerprint"], + "choice": choice, + } + + async def _decide(self, item_id, body): + return await self._json("POST", f"{ITEMS_URL}/{item_id}/decision", json=body) + + def _snapshot(self): + conn = self.api.db.model.get_connection() + return { + table: [tuple(row) for row in conn.execute(f"SELECT * FROM {table}")] + for table in WRITE_TABLES + } + + def _decisions(self): + conn = self.api.db.model.get_connection() + return [ + dict(row) + for row in conn.execute( + "SELECT * FROM media_review_decisions ORDER BY decision_revision" + ) + ] + + def _no_paths(self): + for data in self.responses: + text = canonical_evidence({"r": data}) + self.assertNotIn(self.tmpdir, text) + self.assertNotIn(self.tmpdir.replace(os.sep, "/"), text) + self.assertNotIn(os.path.normcase(self.tmpdir), text) + + +class TestQueueReads(ReviewRoutesTestCase): + async def test_total_is_exact_pending_count(self): + self._require_routes() + for index in range(3): + self._record(self._ambiguous(f"f{index}.png", color=(index, 0, 0))) + page = await self._list("?limit=1") + self.assertEqual(page["total"], 3) + self.assertEqual(len(page["items"]), 1) + seen = [page["items"][0]["id"]] + while page["next_cursor"]: + page = await self._list(f"?limit=1&cursor={page['next_cursor']}") + self.assertEqual(page["total"], 3) + seen.extend(item["id"] for item in page["items"]) + rows = self._items() + self.assertEqual(seen, [row["id"] for row in rows]) + item = (await self._list())["items"][0] + self.assertEqual( + set(item), + { + "id", + "state", + "revision", + "created_at", + "filename", + "target", + "unavailable", + "candidate_count", + "limited", + }, + ) + self.assertEqual( + (item["state"], item["revision"], item["filename"]), + ("pending", 0, "f0.png"), + ) + self.assertEqual(item["candidate_count"], 2) + self.assertIs(item["unavailable"], False) + self.assertEqual(item["target"]["path"], "f0.png") + self.assertEqual(item["target"]["root_index"], 0) + conn = self.api.db.model.get_connection() + with conn: + conn.execute( + "UPDATE media_review_items SET state='selected', revision=1 " + "WHERE id=?", + (seen[0],), + ) + conn.execute( + "UPDATE media_review_items SET state='unresolved', revision=1 " + "WHERE id=?", + (seen[1],), + ) + page = await self._list() + self.assertEqual(page["total"], 1) + self.assertEqual([i["id"] for i in page["items"]], [seen[2]]) + self.assertIsNone(page["next_cursor"]) + self._no_paths() + + async def test_empty_queue_and_limit_validation(self): + self._require_routes() + page = await self._list() + self.assertEqual( + (page["success"], page["total"], page["items"], page["next_cursor"]), + (True, 0, [], None), + ) + self.assertEqual(page["library_epoch"], self.api.db.get_execution_epoch()) + for query in ("?limit=0", "?limit=41", "?limit=abc", "?limit=-1", "?limit=1.5"): + status, data = await self._json("GET", ITEMS_URL + query) + self.assertEqual(status, 400, query) + self.assertFalse(data["success"]) + status, data = await self._json("GET", ITEMS_URL + "?cursor=" + "x" * 5000) + self.assertEqual(status, 400) + + async def test_cursor_invalidated_by_decision_and_restore(self): + self._require_routes() + for index in range(3): + self._record(self._ambiguous(f"f{index}.png", color=(index, 0, 0))) + first = await self._list("?limit=1") + cursor = first["next_cursor"] + self.assertTrue(cursor) + # Tampered: one character changed, or a cursor of another shape. + broken = ("A" if cursor[5] != "A" else "B").join((cursor[:5], cursor[6:])) + status, data = await self._json("GET", f"{ITEMS_URL}?cursor={broken}") + self.assertEqual(status, 400, data) + # A decision moves the review index revision: the old cursor is stale. + item, detail, observed = await self._open() + status, data = await self._decide( + item["id"], {**self._body(detail, observed, "unresolved")} + ) + self.assertEqual(status, 200, data) + status, data = await self._json("GET", f"{ITEMS_URL}?limit=1&cursor={cursor}") + self.assertEqual(status, 409, data) + self.assertEqual(data["code"], "review-conflict") + self.assertEqual(data["total"], 2) + fresh = (await self._list("?limit=1"))["next_cursor"] + conn = self.api.db.model.get_connection() + with conn: # a restore replaces the storage epoch + conn.execute( + "UPDATE execution_history_metadata SET " + "storage_epoch='00000000-0000-4000-8000-000000000000'" + ) + status, data = await self._json("GET", f"{ITEMS_URL}?limit=1&cursor={fresh}") + self.assertEqual(status, 409, data) + self.assertEqual(data["code"], "storage-changed") + self.assertEqual(data["library_epoch"], "00000000-0000-4000-8000-000000000000") + self._no_paths() + + async def test_detail_size_cap_and_limited_item_has_no_candidates(self): + self._require_routes() + self._roots() + limited = self.output_dir / "limited.png" + write_png( + limited, + parameters=( + "cat\nNegative prompt: blur\nNegative prompt: noise\n" + "Steps: 20, Sampler: Euler" + ), + ) + self.assertEqual(self.api.record_review_file(str(limited)), "limited") + self._record(self._ambiguous("two.png")) + items = (await self._list())["items"] + self.assertEqual([i["limited"] for i in items], [True, False]) + detail = await self._detail(items[0]["id"]) + self.assertEqual((detail["limited"], detail["candidates"]), (True, [])) + self.assertEqual(detail["candidate_count"], 0) + detail = await self._detail(items[1]["id"]) + self.assertEqual(len(detail["candidates"]), 2) + candidate = self._candidate(detail, "cat") + self.assertEqual( + set(candidate), + {"hash", "sampler_id", "state", "reason", "label", "positive", "negative"}, + ) + self.assertEqual(candidate["positive"], {"text": "cat"}) + self.assertEqual(candidate["negative"], {"text": "bad"}) + self.assertEqual(candidate["state"], "resolved") + self.assertEqual(candidate["label"], "cat / bad") + self.assertEqual(detail["sha256"], digest(self.output_dir / "two.png")) + self.assertRegex(detail["dependency_fingerprint"], r"^[0-9a-f]{64}$") + with patch.object(review_api, "MAX_DETAIL_BYTES", 200): + capped = await self._detail(items[1]["id"]) + self.assertEqual((capped["limited"], capped["candidates"]), (True, [])) + self.assertEqual(capped["id"], items[1]["id"]) + for missing in ("00000000-0000-4000-8000-000000000000", "not-a-uuid"): + status, data = await self._json("GET", f"{ITEMS_URL}/{missing}") + self.assertEqual(status, 404, data) + self.assertFalse(data["success"]) + self._no_paths() + + async def test_label_uses_library_title_of_a_saved_prompt(self): + self._require_routes() + prompt_id = self.api.db.save_prompt( + "cat", prompt_hash=generate_prompt_hash("cat") + ) + conn = self.api.db.model.get_connection() + with conn: + conn.execute( + "INSERT INTO prompt_library_state(prompt_id,title) VALUES(?,?)", + (prompt_id, "Dunes"), + ) + self._record(self._ambiguous("two.png")) + item = (await self._list())["items"][0] + detail = await self._detail(item["id"]) + self.assertEqual(self._candidate(detail, "cat")["label"], "Dunes / bad") + self.assertEqual(self._candidate(detail, "dog")["label"], "dog / bad") + + async def test_unavailable_when_no_current_root_lists_the_file(self): + self._require_routes() + self._record(self._ambiguous("two.png")) + other = self.output_dir / "elsewhere" + other.mkdir() + from py.config import GalleryConfig + + GalleryConfig.MONITORING_DIRECTORIES = [self._root(other)] + item = (await self._list())["items"][0] + self.assertEqual( + (item["unavailable"], item["target"], item["filename"]), + (True, None, "two.png"), + ) + detail = await self._detail(item["id"]) + self.assertIs(detail["unavailable"], True) + self._no_paths() + + def test_target_formatting_keeps_the_recorded_spelling_on_windows(self): + self._require_routes() + spelled = "C:\\Out\\Sub\\Image_0001.PNG" + locator = ntpath.normcase(spelled) + found = review_api.format_review_target( + spelled, locator, [(0, "D:\\Other"), (1, "C:\\Out")], pathmod=ntpath + ) + self.assertEqual(found, (1, "Sub/Image_0001.PNG")) + self.assertIsNone( + review_api.format_review_target( + spelled, locator, [(0, "C:\\Outer")], pathmod=ntpath + ) + ) + # A recorded spelling that no longer folds to the locator is refused. + self.assertIsNone( + review_api.format_review_target( + spelled, "c:\\out\\sub\\other.png", [(0, "C:\\Out")], pathmod=ntpath + ) + ) + self.assertEqual( + review_api.format_review_target( + "/out/a/b.png", "/out/a/b.png", [(0, "/out")], pathmod=posixpath + ), + (0, "a/b.png"), + ) + + +class TestDecisions(ReviewRoutesTestCase): + async def test_apply_candidate_links_pair_and_records_decision_atomically(self): + self._require_routes() + path = self._ambiguous("two.png") + self._record(path) + revision = self._review_revision() + item, detail, observed = await self._open() + cat = self._candidate(detail, "cat") + before = self._snapshot() + # A failure after the link rolls the link back with everything else. + with patch.object( + type(self.api.db), + "_insert_review_decision", + side_effect=sqlite3.OperationalError("disk I/O error at /secret/path"), + ): + status, data = await self._decide( + item["id"], self._body(detail, observed, cat["hash"]) + ) + self.assertEqual(status, 500, data) + self.assertEqual(data["code"], "decision-failed") + self.assertNotIn("secret", data["error"]) + self.assertEqual(self._snapshot(), before) + + # A failed decision keeps its capture for the retry. + status, data = await self._decide( + item["id"], self._body(detail, observed, cat["hash"]) + ) + self.assertEqual(status, 200, data) + self.assertTrue(data["success"]) + self.assertEqual(data["total"], 0) + self.assertEqual( + (data["item"]["id"], data["item"]["state"], data["item"]["revision"]), + (item["id"], "selected", 1), + ) + conn = self.api.db.model.get_connection() + image = conn.execute( + "SELECT gi.id, p.text FROM generated_images gi " + "JOIN prompts p ON p.id=gi.prompt_id WHERE gi.image_path=?", + (str(path),), + ).fetchone() + self.assertEqual(image[1], "cat") + pair = conn.execute( + "SELECT pp.id, p.text, n.text FROM prompt_pairs pp " + "JOIN prompts p ON p.id=pp.positive_prompt_id " + "JOIN prompts n ON n.id=pp.negative_prompt_id" + ).fetchall() + self.assertEqual([tuple(row)[1:] for row in pair], [("cat", "bad")]) + self.assertEqual(data["pair"]["id"], pair[0][0]) + self.assertEqual(data["pair"]["image_id"], image[0]) + links = self.api.db.get_image_prompt_pairs(image[0]) + self.assertEqual(len(links), 1) + provenance = links[0]["provenance"] + self.assertEqual(provenance["kind"], "user-selection") + self.assertEqual(provenance["candidate_hash"], cat["hash"]) + self.assertEqual(provenance["reviewed_digest"], digest(path)) + self.assertEqual(provenance["resolver_version"], 1) + self.assertEqual(provenance["positive"]["text"], "cat") + (decision,) = self._decisions() + self.assertEqual(provenance["decision_id"], decision["id"]) + self.assertEqual( + ( + decision["item_id"], + decision["decision_revision"], + decision["outcome"], + decision["candidate_hash"], + decision["pair_id"], + decision["image_id"], + ), + (item["id"], 1, "selected", cat["hash"], pair[0][0], image[0]), + ) + self.assertEqual(self._review_revision(), revision + 1) + self.assertEqual((await self._list())["total"], 0) + self.assertEqual(self._capture_files(), []) + self._no_paths() + + async def test_apply_conflicting_legacy_owner_returns_association_conflict_with_zero_writes( # noqa: E501 + self, + ): + self._require_routes() + path = self._ambiguous("two.png") + # The monitor's fallback link: the file is linked to "cat" already. + prompt_id = self.api.db.save_prompt( + "cat", prompt_hash=generate_prompt_hash("cat") + ) + image_id = self.api.db.link_image_to_prompt(prompt_id, str(path)) + self._record(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, 409, data) + self.assertEqual(data["code"], "association-conflict") + self.assertIn("linked", data["error"]) + self.assertEqual(self._snapshot(), before) + # The compatible choice reuses the existing image row. + 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(data["pair"]["image_id"], image_id) + conn = self.api.db.model.get_connection() + self.assertEqual( + conn.execute("SELECT COUNT(*) FROM generated_images").fetchone()[0], 1 + ) + self._no_paths() + + async def test_changed_dependencies_are_an_association_conflict(self): + self._require_routes() + path = self._ambiguous("two.png") + self._record(path) + item, detail, observed = await self._open() + # Another writer links the file after the detail was read. + prompt_id = self.api.db.save_prompt( + "cat", prompt_hash=generate_prompt_hash("cat") + ) + self.api.db.link_image_to_prompt(prompt_id, str(path)) + before = self._snapshot() + cat = self._candidate(detail, "cat") + status, data = await self._decide( + item["id"], self._body(detail, observed, cat["hash"]) + ) + self.assertEqual(status, 409, data) + self.assertEqual(data["code"], "association-conflict") + self.assertNotEqual( + data["dependency_fingerprint"], detail["dependency_fingerprint"] + ) + self.assertEqual(self._snapshot(), before) + + async def test_leave_unresolved_records_decision_and_leaves_pending_list(self): + self._require_routes() + path = self._ambiguous("two.png") + self._record(path) + self._record(self._ambiguous("other.png", color=(0, 0, 9))) + item, detail, _observed = await self._open() + body = { + "expected_epoch": self.api.db.get_execution_epoch(), + "expected_revision": 0, + "capture_token": None, + "sha256": detail["sha256"], + "dependency_fingerprint": detail["dependency_fingerprint"], + "choice": "unresolved", + } + status, data = await self._decide(item["id"], body) + self.assertEqual(status, 200, data) + self.assertEqual(data["item"]["state"], "unresolved") + self.assertEqual(data["total"], 1) + self.assertNotIn("pair", data) + (decision,) = self._decisions() + self.assertEqual( + (decision["outcome"], decision["candidate_hash"], decision["pair_id"]), + ("unresolved", None, None), + ) + conn = self.api.db.model.get_connection() + for table in ("prompt_pairs", "image_prompt_pairs", "generated_images"): + self.assertEqual( + conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0], 0, table + ) + listed = await self._list() + self.assertEqual(listed["total"], 1) + self.assertNotIn(item["id"], [i["id"] for i in listed["items"]]) + # The item still answers its detail, now decided. + self.assertEqual((await self._detail(item["id"]))["state"], "unresolved") + self._no_paths() + + async def test_changed_bytes_or_evidence_reject_apply(self): + self._require_routes() + path = self._ambiguous("two.png") + self._record(path) + item, detail, observed = await self._open() + cat = self._candidate(detail, "cat") + before = self._snapshot() + + # The file is rewritten after the capture: the re-hash refuses it. + write_png(path, prompt=ambiguous_graph("cat", "dog"), color=(0, 255, 0)) + status, data = await self._decide( + item["id"], self._body(detail, observed, cat["hash"]) + ) + self.assertEqual((status, data["code"]), (409, "file-changed"), data) + self.assertEqual(self._snapshot(), before) + + # A fresh capture of the new bytes is not the reviewed item's bytes. + observed = await self._observe(item["target"]) + status, data = await self._decide( + item["id"], self._body(detail, observed, cat["hash"]) + ) + self.assertEqual((status, data["code"]), (409, "file-changed"), data) + self.assertEqual( + self._snapshot()["media_review_items"], before["media_review_items"] + ) + + # Back to the reviewed bytes, but the evidence they yield has changed. + write_png(path, prompt=ambiguous_graph("cat", "dog")) + self.assertEqual(digest(path), detail["sha256"]) + observed = await self._observe(item["target"]) + changed = { + "state": "ambiguous", + "reason": "multiple samplers", + "candidates": [], + } + with patch("utils.review_capture.resolve_prompt_pair", return_value=changed): + status, data = await self._decide( + item["id"], self._body(detail, observed, cat["hash"]) + ) + self.assertEqual((status, data["code"]), (409, "review-conflict"), data) + + # A choice that is not one of the item's candidates is refused. + status, data = await self._decide( + item["id"], self._body(detail, observed, "f" * 64) + ) + self.assertEqual(status, 400, data) + # A restore since the capture. + status, data = await self._decide( + item["id"], + { + **self._body(detail, observed, cat["hash"]), + "expected_epoch": "00000000-0000-4000-8000-000000000000", + }, + ) + self.assertEqual((status, data["code"]), (409, "storage-changed"), data) + # A stale revision. + status, data = await self._decide( + item["id"], + {**self._body(detail, observed, cat["hash"]), "expected_revision": 3}, + ) + self.assertEqual((status, data["code"]), (409, "review-conflict"), data) + # Apply needs a capture; Leave unresolved does not. + status, data = await self._decide( + item["id"], + {**self._body(detail, observed, cat["hash"]), "capture_token": None}, + ) + self.assertEqual(status, 400, data) + # A capture of another file. + other = self._ambiguous("other.png", color=(1, 2, 3)) + self._record(other) + target = {**item["target"], "path": "other.png"} + foreign = await self._observe(target) + status, data = await self._decide( + item["id"], + { + **self._body(detail, observed, cat["hash"]), + "capture_token": foreign["capture"]["token"], + "sha256": foreign["capture"]["sha256"], + }, + ) + self.assertEqual((status, data["code"]), (409, "capture-expired"), data) + self.assertEqual(self._decisions(), []) + self.assertEqual(self._items()[0]["state"], "pending") + self._no_paths() + + async def test_decision_body_validation(self): + self._require_routes() + self._record(self._ambiguous("two.png")) + item, detail, observed = await self._open() + good = self._body(detail, observed, "unresolved") + cases = [ + {**good, "extra": 1}, + {k: v for k, v in good.items() if k != "choice"}, + {**good, "expected_epoch": "nope"}, + {**good, "expected_revision": -1}, + {**good, "expected_revision": True}, + {**good, "capture_token": "bad token!"}, + {**good, "sha256": "A" * 64}, + {**good, "dependency_fingerprint": "x"}, + {**good, "choice": "maybe"}, + {**good, "choice": 7}, + {**good, "sha256": "0" * 64}, + ] + for body in cases: + with self.subTest(body=body): + status, data = await self._decide(item["id"], body) + self.assertEqual(status, 400, data) + self.assertFalse(data["success"]) + status, data = await self._json( + "POST", f"{ITEMS_URL}/{item['id']}/decision", data=b"[1" + ) + self.assertEqual(status, 400, data) + status, data = await self._decide("00000000-0000-4000-8000-000000000000", good) + self.assertEqual(status, 404, data) + self.assertEqual(self._decisions(), []) + + async def test_duplicate_apply_after_commit_is_review_conflict_with_current_state( + self, + ): + self._require_routes() + self._record(self._ambiguous("two.png")) + 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) + after = self._snapshot() + observed = await self._observe(item["target"]) + status, data = await self._decide( + item["id"], self._body(detail, observed, cat["hash"]) + ) + self.assertEqual(status, 409, data) + self.assertEqual(data["code"], "review-conflict") + self.assertEqual( + (data["item"]["state"], data["item"]["revision"]), ("selected", 1) + ) + self.assertEqual(data["total"], 0) + self.assertEqual(self._snapshot(), after) + self.assertEqual(len(self._decisions()), 1) + + async def test_tag_save_never_changes_review_item(self): + self._require_routes() + self._record(self._ambiguous("two.png")) + item, _detail, observed = await self._open() + before = self._snapshot() + status, data = await self._json( + "PUT", + "/prompt_manager/image-annotations/tags", + json={ + "target": item["target"], + "capture_token": observed["capture"]["token"], + "sha256": observed["capture"]["sha256"], + "expected_epoch": observed["library_epoch"], + "expected_tag_revision": observed["tag_revision"], + "tags": ["sunset"], + }, + ) + self.assertEqual(status, 200, data) + after = self._snapshot() + for table in ( + "media_review_items", + "media_review_decisions", + "media_review_metadata", + "generated_images", + "image_prompt_pairs", + ): + self.assertEqual(after[table], before[table], table) + self.assertEqual((await self._list())["total"], 1) + + async def test_run_and_usage_counts_unchanged(self): + self._require_routes() + path = self._ambiguous("two.png") + prompt_id = self.api.db.save_prompt( + "cat", prompt_hash=generate_prompt_hash("cat") + ) + conn = self.api.db.model.get_connection() + with conn: + conn.execute( + "UPDATE prompts SET run_count=5, last_used_at='2026-01-01T00:00:00Z' " + "WHERE id=?", + (prompt_id,), + ) + self._record(path) + executions = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' " + "AND name LIKE 'execution_%'" + ).fetchall() + counts = { + row[0]: conn.execute(f"SELECT COUNT(*) FROM {row[0]}").fetchone()[0] + for row in executions + } + item, detail, observed = await self._open() + status, data = await self._decide( + item["id"], + self._body(detail, observed, self._candidate(detail, "cat")["hash"]), + ) + self.assertEqual(status, 200, data) + rows = conn.execute( + "SELECT text, run_count, last_used_at FROM prompts" + ).fetchall() + self.assertIn(("cat", 5, "2026-01-01T00:00:00Z"), [tuple(r) for r in rows]) + self.assertEqual({tuple(r)[1] for r in rows if r[0] != "cat"}, {0}) + self.assertEqual( + { + name: conn.execute(f"SELECT COUNT(*) FROM {name}").fetchone()[0] + for name in counts + }, + counts, + ) + (pair,) = self.api.db.list_prompt_pairs()["pairs"] + self.assertEqual(pair["run_count"], 0) + + async def test_no_absolute_paths_in_responses(self): + self._require_routes() + self._record(self._ambiguous("two.png")) + item, detail, observed = await self._open() + await self._json("GET", f"{ITEMS_URL}?cursor=bad") + await self._decide(item["id"], {**self._body(detail, observed, "f" * 64)}) + with patch.object( + type(self.api.db), + "review_queue_page", + side_effect=sqlite3.OperationalError(f"no such table at {self.tmpdir}"), + ): + status, data = await self._json("GET", ITEMS_URL) + self.assertEqual((status, data["code"]), (500, "review-failed")) + with patch.object( + type(self.api.db), + "review_item", + side_effect=sqlite3.OperationalError(f"locked {self.tmpdir}"), + ): + status, data = await self._json("GET", f"{ITEMS_URL}/{item['id']}") + self.assertEqual((status, data["code"]), (500, "review-failed")) + status, data = await self._decide( + item["id"], self._body(detail, observed, "unresolved") + ) + self.assertEqual(status, 200, data) + self._no_paths() diff --git a/tests/test_pair_ingestion.py b/tests/test_pair_ingestion.py index 4cfdfa9..fab7ef9 100644 --- a/tests/test_pair_ingestion.py +++ b/tests/test_pair_ingestion.py @@ -160,3 +160,69 @@ class TestPairIngestion(unittest.TestCase): info = self.db.get_image_prompt_info(self.path) self.assertEqual(info["pair_context"]["positive_text"], "cat, lora_trigger") self.assertEqual(info["pair_context"]["text_source"], "execution") + + def test_public_ingest_resolved_pair_contract_unchanged(self): + """The public ingester keeps its result shape and its own transaction; + the extracted ``_ingest_resolved_pair_in`` runs in the caller's.""" + from database import pair_ingestion + + self.assertTrue( + hasattr(pair_ingestion, "_ingest_resolved_pair_in"), + "_ingest_resolved_pair_in is not extracted", + ) + for state in ("ambiguous", "unresolved", "conflict"): + self.assertEqual( + ingest_resolved_pair(self.db, self.path, {}, {"state": state}), + {"status": state, "pairs_added": 0, "links_added": 0}, + ) + first = ingest_resolved_pair(self.db, self.path, {}, resolved()) + self.assertEqual( + set(first), + {"status", "pairs_added", "links_added", "pair_id", "image_id"}, + ) + self.assertEqual(first["status"], "added") + (link,) = self.db.get_image_prompt_pairs(first["image_id"]) + self.assertEqual( + link["provenance"], + { + "resolver_version": 1, + "positive": resolved()["positive"], + "negative": resolved()["negative"], + }, + ) + conn = self.db.model.get_connection() + self.assertFalse(conn.in_transaction) + # Inside the caller's transaction, with the caller's provenance; an + # exception afterwards rolls the link back with the caller's writes. + provenance = {"kind": "user-selection", "resolver_version": 1} + with self.assertRaises(RuntimeError): + with self.db._pair_transaction() as held: + inner = pair_ingestion._ingest_resolved_pair_in( + held, + self.db, + self.path + "2", + {}, + resolved("dog"), + provenance=provenance, + ) + self.assertEqual(inner["status"], "added") + raise RuntimeError("caller failed") + self.assertIsNone(self.db.get_image_prompt_info(self.path + "2")) + with self.db._pair_transaction() as held: + inner = pair_ingestion._ingest_resolved_pair_in( + held, + self.db, + self.path + "2", + {}, + resolved("dog"), + provenance=provenance, + ) + (link,) = self.db.get_image_prompt_pairs(inner["image_id"]) + self.assertEqual(link["provenance"], provenance) + with self.db._pair_transaction() as held: + conflict = pair_ingestion._ingest_resolved_pair_in( + held, self.db, self.path, {}, resolved("bird") + ) + self.assertEqual( + conflict, {"status": "conflict", "pairs_added": 0, "links_added": 0} + ) diff --git a/tests/test_review.py b/tests/test_review.py index a9ebdd1..a169eba 100644 --- a/tests/test_review.py +++ b/tests/test_review.py @@ -275,6 +275,46 @@ class TestScanRecording(ReviewTestCase): self.assertEqual(self._review_revision(), revision) self.assertEqual(self._capture_files(), []) + async def test_queued_fingerprint_skips_the_copy_and_hash(self): + """A rescan of a queued file whose size and mtime match its item's + verified fingerprint neither copies nor hashes it, and the file stays + in the default "new" scope (it is still read like any unknown file).""" + self._require_module() + self._roots() + path = self.output_dir / "two.png" + write_png(path, prompt=ambiguous_graph("cat", "dog")) + await self._scan("new") + captures = [] + original = CaptureStore.capture + + def spy(store, root, relative, **kwargs): + captures.append(relative) + return original(store, root, relative, **kwargs) + + reads = [] + extract = self.api._extract_comfyui_metadata + + def read_spy(file_path): + reads.append(os.path.basename(file_path)) + return extract(file_path) + + self.api._extract_comfyui_metadata = read_spy + with patch.object(CaptureStore, "capture", spy): + second = await self._scan("new") + self.assertEqual(captures, []) + self.assertEqual(reads, ["two.png"]) + self.assertEqual(second["scope"]["review_items"], 0) + # A touched file (same bytes, new mtime) is captured again: a + # duplicate of its item, whose fingerprint is refreshed. + stat = path.stat() + os.utime(path, ns=(stat.st_atime_ns, stat.st_mtime_ns + 10**9)) + await self._scan("new") + self.assertEqual(captures, ["two.png"]) + items = self._items() + self.assertEqual(len(items), 1) + self.assertEqual(items[0]["mtime_ns"], path.stat().st_mtime_ns) + self.assertEqual(self._capture_files(), []) + async def test_repeat_scan_keeps_decisions(self): self._require_module() self._roots() @@ -601,6 +641,7 @@ class TestReviewEdges(ReviewTestCase): epoch = self.api.db.get_execution_epoch() good = { "locator": "/out/a.png", + "spelled_path": "/out/a.png", "sha256": "a" * 64, "size": 1, "mtime_ns": 2, @@ -619,6 +660,9 @@ class TestReviewEdges(ReviewTestCase): {"limited": True, "candidate_count": 2}, {"sha256": "A" * 64}, {"evidence": ["not", "an", "object"]}, + {"spelled_path": ""}, + {"spelled_path": None}, + {"spelled_path": "/out/b.png"}, ): with self.assertRaises(ValueError, msg=change): self.api.db.record_review_item(**{**good, **change}) diff --git a/tests/test_review_schema.py b/tests/test_review_schema.py index 674ced9..dfd6140 100644 --- a/tests/test_review_schema.py +++ b/tests/test_review_schema.py @@ -53,7 +53,9 @@ DDL = { ITEMS: ( f"CREATE TABLE {ITEMS}(id TEXT PRIMARY KEY CHECK(length(id)=36), " "byte_id TEXT NOT NULL REFERENCES media_annotation_bytes(id) " - "ON DELETE RESTRICT, resolver_version INTEGER NOT NULL, " + "ON DELETE RESTRICT, spelled_path TEXT NOT NULL CHECK(length(CAST(" + "spelled_path AS BLOB)) BETWEEN 1 AND 16384), " + "resolver_version INTEGER NOT NULL, " "evidence_hash TEXT NOT NULL CHECK(length(evidence_hash)=64 AND " "evidence_hash NOT GLOB '*[^0-9a-f]*'), evidence TEXT NOT NULL " f"CHECK(length(CAST(evidence AS BLOB)) <= {MAX_EVIDENCE_BYTES}), " @@ -175,6 +177,7 @@ def insert_item(conn, *, item_id=ITEM_ID, evidence=EVIDENCE, candidates=2, **ext values = { "id": item_id, "byte_id": BYTE_ID, + "spelled_path": "/out/img.png", "resolver_version": 1, "evidence_hash": hashlib.sha256(evidence.encode("utf-8")).hexdigest(), "evidence": evidence, @@ -531,6 +534,9 @@ class ReviewSchemaTests(unittest.TestCase): "long created_at": ((f"UPDATE {ITEMS} SET created_at=?", ("x" * 65,)),), "blob updated_at": (f"UPDATE {ITEMS} SET updated_at=X'00'",), "text resolver version": (f"UPDATE {ITEMS} SET resolver_version='one'",), + "blob spelled path": ( + f"UPDATE {ITEMS} SET spelled_path=CAST(spelled_path AS BLOB)", + ), "evidence hash mismatch": ( (f"UPDATE {ITEMS} SET evidence_hash=?", (CANDIDATE,)), ), @@ -549,6 +555,7 @@ class ReviewSchemaTests(unittest.TestCase): ), "bad state": (f"UPDATE {ITEMS} SET state='done'",), "limited two": (f"UPDATE {ITEMS} SET limited=2",), + "empty spelled path": (f"UPDATE {ITEMS} SET spelled_path=''",), "negative revision": (f"UPDATE {ITEMS} SET revision=-1",), "negative candidates": (f"UPDATE {ITEMS} SET candidate_count=-1",), "bad outcome": (f"UPDATE {DECISIONS} SET outcome='maybe'",), diff --git a/utils/review_capture.py b/utils/review_capture.py index 0ab9c34..1af9bd7 100644 --- a/utils/review_capture.py +++ b/utils/review_capture.py @@ -22,6 +22,7 @@ from __future__ import annotations import hashlib import json import logging +import os from pathlib import Path from typing import Any @@ -53,11 +54,16 @@ def record_ambiguous_file(db, store, root, root_index: int, relative: str) -> st Returns ``recorded`` (new item with candidates), ``limited`` (new item without candidates), ``duplicate`` (these bytes already carry this - evidence) or ``skipped`` (the file changed, vanished, is not ambiguous in - its captured bytes, or storage was restored meanwhile). + evidence, or the file's current stat is the verified fingerprint of + bytes an item cites, so it is not even copied) or ``skipped`` (the file + changed, vanished, is not ambiguous in its captured bytes, or storage was + restored meanwhile). ``CaptureCapacity`` and database errors propagate to the caller, which logs and counts them. """ + spelled_path = _queued_spelling(db, root, relative) + if spelled_path is None: + return "duplicate" epoch = db.get_execution_epoch() try: capture = store.capture( @@ -69,15 +75,17 @@ def record_ambiguous_file(db, store, root, root_index: int, relative: str) -> st logger.debug("Review capture skipped (%s): %s", error.code, error) return "skipped" try: - resolution = _captured_resolution(store, capture) + captured = captured_review(store, capture) finally: store.release(capture.token) + resolution = captured[1] if captured is not None else None if resolution is None or resolution.get("state") != "ambiguous": return "skipped" evidence, count, limited = review_evidence(resolution) try: inserted = db.record_review_item( locator=capture.locator, + spelled_path=spelled_path, sha256=capture.sha256, size=capture.size, mtime_ns=capture.mtime_ns, @@ -94,18 +102,45 @@ def record_ambiguous_file(db, store, root, root_index: int, relative: str) -> st return "limited" if limited else "recorded" -def _captured_resolution(store, capture) -> dict | None: - """The ingestion resolver's result for the capture's own metadata.""" +def _queued_spelling(db, root, relative: str) -> str | None: + """The file's spelling, or None when its current stat is already queued. + + The spelling is the capture's own candidate path (the real root joined + with the listed relative path), so it folds to the capture's locator. A + file whose size and ``mtime_ns`` equal the verified fingerprint of bytes + an item already cites is not copied or hashed again (maintainer ruling: + the rescan stays cheap and the file stays in the default scope). A stat + failure leaves the decision to the capture. + """ + spelled = os.path.join(os.path.realpath(os.fspath(root)), *relative.split("/")) + try: + facts = os.lstat(spelled) + except OSError: + return spelled + if db.review_fingerprint_queued( + os.path.normcase(spelled), facts.st_size, facts.st_mtime_ns, RESOLVER_VERSION + ): + return None + return spelled + + +def captured_review(store, capture) -> tuple[dict, dict] | None: + """``(metadata, resolution)`` of the capture's own bytes, else None. + + ``metadata`` is what the scan hands the ingester: prompt and workflow + JSON-decoded and ``file_info`` (size, dimensions, format) from the + capture. None when the capture is gone or carries no prompt metadata. + """ if store.acquire(capture.token) is None: return None try: try: - metadata = read_inspection_file(Path(capture.path)).metadata + read = read_inspection_file(Path(capture.path)) except MetadataReadError: return None finally: store.release_use(capture.token) - canonical = dict(metadata) + canonical = dict(read.metadata) for key in ("prompt", "workflow"): if isinstance(canonical.get(key), str): try: @@ -115,7 +150,13 @@ def _captured_resolution(store, capture) -> dict | None: pass if not canonical.get("prompt") and not canonical.get("parameters"): return None - return resolve_prompt_pair(canonical) + info = {"size": capture.size} + if read.width is not None and read.height is not None: + info["dimensions"] = [read.width, read.height] + if read.format is not None: + info["format"] = read.format + canonical["file_info"] = info + return canonical, resolve_prompt_pair(canonical) def review_evidence(resolution: dict) -> tuple[dict, int, bool]: