Acquisition verify gate (plan 22)

CHANGELOG.md

@@ -16,6 +16,17 @@ All notable changes to this project are documented here. The format is based on
   new `COLOPHON_BOOKS_ROOT` (the host path the library mounts from); unset ⇒ feature off.
   New module `colophon/epub.py` (stdlib-only) plus `hardcover.book_by_isbn` and
   `grimmory.epub_path` helpers.
+- **Acquisition-side `verify`** — a read-only `verify(requested, file)` (new module
+  `colophon/verify.py`; CLI `colophon verify <file> --hcid <id>` / `--title`) answering
+  *is this downloaded file the requested work?* at work granularity, for the
+  identifier-verified acquisition gate (plan 22). Deterministic path: the file's embedded
+  OPF ISBN → Hardcover work, compared to the requested work via the shared `canonical_id`.
+  No resolvable ISBN: the file is identified through the resolver's adjudicator (search by
+  the file's own title/author, fold in its colophon text — the LLM only ever selects a real
+  candidate) and that work is compared to the requested one; below `_LLM_MIN` (0.8)
+  confidence the file is held as `unverifiable`. Reuses `epub` + `hardcover` + `matcher` +
+  `resolver`; no writes, no grimmory DB at runtime. Exits 0/3/4 (match/mismatch/unverifiable).
+  `epub.inspect` now also returns `opf_author` (dc:creator).
 
 ## [0.2.0] — 2026-06-04
 

colophon/cli.py

@@ -17,6 +17,7 @@
 from .grimmory import Grimmory, GrimmoryError
 from .heal import assert_preconditions, heal_book, revert_run, PreconditionError
 from .store import Store
+from .verify import verify
 
 
 def _fmt(snap, keys=("title", "series_name", "series_number", "isbn_13",
@@ -227,6 +228,22 @@ def cmd_maintain(args, g, store):
     return 0 if (res and res["ok"]) else 1
 
 
+def cmd_verify(args, g, store):
+    """Acquisition gate: is the file at <file> the requested work? Read-only — prints a
+    JSON verdict (match/mismatch/unverifiable) and exits 0/3/4 for scripting. The gate's
+    hook shim reads the JSON, not the exit code."""
+    requested = {}
+    if args.hcid:
+        requested["hcid"] = args.hcid
+    if args.title:
+        requested["title"] = args.title
+    if args.author:
+        requested["authors"] = args.author
+    result = verify(requested, args.file)
+    print(json.dumps(result, ensure_ascii=False, indent=2))
+    return {"match": 0, "mismatch": 3, "unverifiable": 4}.get(result["verdict"], 4)
+
+
 def main(argv=None):
     p = argparse.ArgumentParser(prog="colophon", description="autonomous library metadata healer (Phase 0)")
     sub = p.add_subparsers(dest="cmd", required=True)
@@ -266,13 +283,19 @@ def main(argv=None):
     mt.add_argument("--apply", action="store_true", help="actually write (default: dry-run)")
     mt.add_argument("--force", action="store_true", help="re-query cached-unresolvable mis-seeds this run")
     mt.add_argument("--email", action="store_true", help="always email the summary (a daily heartbeat)")
+    ve = sub.add_parser("verify", help="acquisition gate: is a downloaded file the requested work? (read-only)")
+    ve.add_argument("file", help="path to the downloaded book file")
+    ve.add_argument("--hcid", help="requested Hardcover work id (primary anchor)")
+    ve.add_argument("--title", help="requested title (degraded fallback when no --hcid)")
+    ve.add_argument("--author", help="requested author(s), used with --title")
     args = p.parse_args(argv)
 
     g, store = Grimmory(), Store()
     fn = {"precheck": cmd_precheck, "heal": cmd_heal, "log": cmd_log,
           "revert": cmd_revert, "backfill": cmd_backfill, "audit": cmd_audit,
           "resolve": cmd_resolve, "series-audit": cmd_series_audit,
-          "oversight": cmd_oversight, "maintain": cmd_maintain}[args.cmd]
+          "oversight": cmd_oversight, "maintain": cmd_maintain,
+          "verify": cmd_verify}[args.cmd]
     try:
         return fn(args, g, store)
     except (GrimmoryError, PreconditionError) as e:

colophon/epub.py

@@ -53,7 +53,7 @@ def _dc(opf, tag):
 
 
 def inspect(path):
-    """Return {opf_isbns, colophon_text, opf_title, publisher, year} or None.
+    """Return {opf_isbns, colophon_text, opf_title, opf_author, publisher, year} or None.
 
     `colophon_text` is "" when no page looks like a colophon (e.g. indie EPUBs with
     only a cover) — the caller should treat empty as "no usable colophon signal"."""
@@ -98,6 +98,7 @@ def inspect(path):
         "opf_isbns": _isbns(opf),
         "colophon_text": best_text[:_MAX_COLOPHON] if best_hits else "",
         "opf_title": _dc(opf, "title"),
+        "opf_author": _dc(opf, "creator"),
         "publisher": _dc(opf, "publisher"),
         "year": (_dc(opf, "date") or "")[:4] or None,
     }

colophon/grimmory.py

@@ -114,6 +114,20 @@ def refresh(self, book_ids, refresh_covers=True, replace_mode="REPLACE_ALL"):
             raise GrimmoryError(f"refresh task start returned {st}: {resp[:200]}")
         return resp
 
+    def delete_books(self, book_ids):
+        """Hard-delete books by id: DELETE /api/v1/books?ids=<csv>. `ids` is a
+        @RequestParam Set<Long> — a QUERY parameter, NOT a JSON body (a body 500s;
+        confirmed against a live throwaway record). grimmory removes the record AND
+        its files; the response reports any `failedFileDeletions`, for which the
+        caller's own file unlink is the fallback. Used only by the plan-22 gate."""
+        ids = [int(b) for b in book_ids]
+        if not ids:
+            return None
+        st, resp = self._call("DELETE", "/books?ids=" + ",".join(str(i) for i in ids), None)
+        if st not in (200, 204):
+            raise GrimmoryError(f"DELETE /books?ids={ids} returned {st}: {resp[:200]}")
+        return resp
+
 
 # --- read-only snapshot ------------------------------------------------------
 SNAPSHOT_COLS = [
@@ -136,6 +150,16 @@ def _db(sql):
     return r.stdout
 
 
+def book_ids_by_filename(file_name):
+    """book_id(s) whose `book_file.file_name` matches exactly — read-only. Lets the
+    acquisition gate resolve a just-landed grab to its grimmory record(s) before
+    deleting it. Single quotes in the name are SQL-escaped (titles like
+    "Salvation's Child")."""
+    safe = (file_name or "").replace("'", "''")
+    out = _db(f"SELECT book_id FROM book_file WHERE file_name='{safe}';").strip()
+    return [int(x) for x in out.split() if x.strip().isdigit()]
+
+
 def snapshot(book_id):
     """Current metadata for a book as a dict, or None if it doesn't exist."""
     bid = int(book_id)

colophon/verify.py

@@ -0,0 +1,134 @@
+"""Acquisition-side verification — is a grabbed file the requested work?
+
+Read-only, no writes, no grimmory DB. Given a requested identity (a Hardcover
+work id — *primary*; or title/author — *degraded fallback*) and a downloaded book
+file, decide match / mismatch / unverifiable at **work** granularity. Reuses
+`epub.inspect` + `hardcover` + `matcher`; never originates an identifier.
+
+This is the comparator the acquisition gate (plan 22) calls per grab. The gate
+acts on the returned verdict (promote / hold) — never on a process exit code.
+
+Deterministic happy path (no LLM): the file's embedded OPF ISBN resolves to a
+Hardcover book; compare its work to the requested work (id, via the shared
+canonical). A file with no resolvable ISBN is held as `unverifiable` — the
+title/colophon LLM adjudication is a later increment, not a guess.
+"""
+from . import epub, hardcover, matcher
+
+MATCH = "match"
+MISMATCH = "mismatch"
+UNVERIFIABLE = "unverifiable"
+
+# Below this adjudicator confidence the file is held (unverifiable), never matched or
+# rejected on a guess. LLM verdicts are also capped below the deterministic-ISBN 0.97.
+_LLM_MIN = 0.8
+
+
+def _result(verdict, confidence, reason, **evidence):
+    out = {"verdict": verdict, "confidence": round(confidence, 2), "reason": reason}
+    out.update({k: v for k, v in evidence.items() if v is not None})
+    return out
+
+
+def _canon(book):
+    """Work-level key for a Hardcover book dict: its canonical id, else its own id."""
+    return str(book.get("canonical_id") or book.get("hcid"))
+
+
+def _same_work(a, b):
+    """Two Hardcover book dicts are the same work when their ids or canonicals agree
+    (an edition/duplicate record points at the work via `canonical_id`)."""
+    return _canon(a) == _canon(b) or str(a.get("hcid")) == str(b.get("hcid"))
+
+
+def _file_book_from_isbns(isbns):
+    """First OPF ISBN that resolves to a Hardcover book → (isbn, book), else (None, None)."""
+    for isbn in isbns:
+        try:
+            book = hardcover.book_by_isbn(isbn)
+        except Exception:
+            continue
+        if book and book.get("hcid"):
+            return isbn, book
+    return None, None
+
+
+def verify(requested, file_path):
+    """requested: {"hcid": <id>} (primary) or {"title": ..., "authors": ...} (fallback).
+    file_path: the downloaded book file. Returns a verdict dict
+    {"verdict", "confidence", "reason", ...evidence}."""
+    sig = epub.inspect(file_path)
+    if not sig:
+        return _result(UNVERIFIABLE, 0.0,
+                       "no embedded signals (non-EPUB or unreadable file)",
+                       source="no-file-signal")
+
+    isbns = sig.get("opf_isbns") or []
+    req_hcid = str(requested["hcid"]) if requested.get("hcid") else None
+    req_title = requested.get("title")
+
+    isbn, file_book = _file_book_from_isbns(isbns)
+
+    # --- Deterministic path: the file's embedded ISBN resolves to a Hardcover work ---
+    if file_book:
+        file_hcid = str(file_book.get("hcid"))
+        file_title = file_book.get("title")
+        if req_hcid:
+            try:
+                req_book = hardcover.book_by_id(req_hcid)
+            except Exception:
+                req_book = None
+            same = _same_work(file_book, req_book) if req_book else (file_hcid == req_hcid)
+            verdict = MATCH if same else MISMATCH
+            rel = "same work as" if same else "different work from"
+            return _result(verdict, 0.97 if same else 0.9,
+                           f"file ISBN {isbn} -> hc#{file_hcid} {file_title!r}; {rel} requested hc#{req_hcid}",
+                           source="isbn-id", isbn=isbn, file_hcid=file_hcid, file_title=file_title)
+        if req_title:
+            same = matcher._title_match(file_title or "", req_title)
+            verdict = MATCH if same else MISMATCH
+            rel = "~ requested title" if same else "!= requested title"
+            return _result(verdict, 0.80 if same else 0.75,
+                           f"file ISBN {isbn} -> hc#{file_hcid} {file_title!r} {rel} {req_title!r} (no id given)",
+                           source="isbn-title", isbn=isbn, file_hcid=file_hcid, file_title=file_title)
+        return _result(UNVERIFIABLE, 0.0, "no requested identity supplied", source="no-request")
+
+    # --- No resolvable embedded ISBN: identify the file via the resolver's adjudicator
+    # (search Hardcover by the file's own title/author, fold in its colophon text), then
+    # compare that work to the requested one. The LLM only ever selects a real candidate. ---
+    opf_title, opf_author = sig.get("opf_title"), sig.get("opf_author")
+    if not opf_title:
+        return _result(UNVERIFIABLE, 0.0,
+                       "no embedded ISBN and no title in the file to adjudicate on",
+                       source="no-signal")
+
+    from . import resolver  # lazy: keeps verify's import surface light (epub/hardcover/matcher)
+    try:
+        prop = resolver.resolve({"book_id": None, "title": opf_title, "authors": opf_author or ""},
+                                file_signals=sig)
+    except Exception as e:
+        return _result(UNVERIFIABLE, 0.0, f"adjudication failed: {str(e)[:80]}", source="llm-error")
+
+    conf = prop.get("confidence") or 0.0
+    chosen = str(prop["chosen_id"]) if prop.get("chosen_id") else None
+    if prop.get("action") != "propose" or not chosen or conf < _LLM_MIN:
+        return _result(UNVERIFIABLE, round(conf, 2),
+                       f"file not confidently identified ({prop.get('reason') or prop.get('action')})",
+                       source="llm-lowconf", opf_title=opf_title, chosen_id=chosen)
+
+    file_title = prop.get("chosen_title")
+    if req_hcid:
+        try:
+            same = _same_work(hardcover.book_by_id(chosen), hardcover.book_by_id(req_hcid))
+        except Exception:
+            same = chosen == req_hcid
+    elif req_title:
+        same = matcher._title_match(file_title or "", req_title)
+    else:
+        return _result(UNVERIFIABLE, 0.0, "no requested identity supplied", source="no-request")
+
+    verdict = MATCH if same else MISMATCH
+    rel = "same work as" if same else "different work from"
+    return _result(verdict, round(min(conf, 0.9), 2),
+                   f"file adjudicated -> hc#{chosen} {file_title!r}; {rel} requested",
+                   source="llm-adjudicated", file_hcid=chosen, file_title=file_title)

tests/test_epub.py

@@ -79,6 +79,7 @@ def test_colophon_text_found(self):
     def test_metadata_fields(self):
         r = epub.inspect(self._epub(['<dc:identifier>urn:isbn:9781466849358</dc:identifier>']))
         self.assertEqual(r["opf_title"], "Lock In")
+        self.assertEqual(r["opf_author"], "John Scalzi")
         self.assertEqual(r["publisher"], "Tor")
         self.assertEqual(r["year"], "2014")