"""Read-only Bookmark Canvas package ingestion and a persistent SQL query API.

Only the derived database is written. JSON/Canvas remains the source of truth.
Absent section files are retained; a supplied canvas replaces the previous entry.
"""

import hashlib
import json
import math
import re
import sqlite3
from contextlib import contextmanager
from datetime import datetime, timezone
from pathlib import Path, PurePosixPath
from urllib.parse import urlsplit

from package_paths import (
    PROTOCOL_DIRS,
    dir_matches_section_type,
    is_package_dir,
    resolve_protocol_reference,
)


# The trigram tokenizer indexes three-character sequences; shorter patterns
# cannot use it and keep the LIKE fallback.
TRIGRAM_MINIMUM = 3

# URL shape is a retrieval signal, not a claim about the page: a documentation
# or changelog page is usually a better citation than a vendor homepage, while
# a login or search-result URL is almost never the requested source.
URL_KIND_RANK = {"docs": 0, "changelog": 0, "repo": 1, "other": 2, "homepage": 3, "login": 4,
                 "search": 4, "local_file": 5}


def url_kind(url):
    """Classify a bookmark URL for preflight ordering; never fetch it."""
    try:
        parts = urlsplit(url)
        host = (parts.hostname or "").lower()
        path = (parts.path or "").lower()
        scheme = (parts.scheme or "").lower()
    except (TypeError, ValueError):
        return "other"
    if scheme not in ("http", "https"):
        return "local_file"
    if any(token in path for token in ("changelog", "release-notes", "release_notes", "whats-new", "whatsnew")):
        return "changelog"
    if host in ("github.com", "gitlab.com", "bitbucket.org", "codeberg.org"):
        return "repo"
    if host.startswith("login.") or any(token in path for token in ("/login", "/signin", "/auth/", "/oauth")):
        return "login"
    if path.rstrip("/") in ("", "/") or path in ("", "/"):
        return "homepage"
    if "search" in path and parts.query:
        return "search"
    if (host.startswith(("docs.", "developer.", "developers.", "documentation.", "help.")) or ".docs." in host
            or any(token in path for token in ("/docs/", "/doc/", "/document/", "/documentation/", "/api/", "/reference/"))):
        return "docs"
    return "other"



# A question is reduced to literal terms: ASCII runs whole, CJK runs expanded
# into bigrams because a phrase the user typed rarely appears verbatim.
PLAN_RUN = re.compile(r"[0-9A-Za-z][0-9A-Za-z+._-]*|[㐀-䶿一-鿿]+")
QUESTION_URL = re.compile(r"https?://[^\s<>()\"']+")
PLAN_MAX_TERMS = 12


def question_urls(text):
    """The URLs a question already contains; the one shared URL extractor."""
    return QUESTION_URL.findall(text) if isinstance(text, str) else []


def title_has_term(title, term):
    """Whether a title contains the term as a token, not merely a substring.

    ASCII terms use word boundaries so ``quantum`` does not match
    ``QuantumNous``. CJK has no whitespace boundary; a literal occurrence is the
    most this layer can honestly claim.
    """
    if not title or not term:
        return False
    if term.isascii():
        return re.search(r"(?<![0-9A-Za-z])" + re.escape(term) + r"(?![0-9A-Za-z])",
                         title, re.I) is not None
    return term in title

# Added-at is deliberately literal: a tempId_/syncId_ date segment is a
# generation clue, a dateAdded epoch is the bookmark's own add time, and an
# edge id ends with the millisecond it was created. Nothing here guesses a
# missing time, and each source keeps its own precision and confidence.
ID_DATE = re.compile(r"^(?:temp|sync)Id_(\d{8})_")
EPOCH_MS = (10 ** 12, 10 ** 13)  # 2001-09-09 .. 2286-11-20
ADDED_AT_SOURCES = {"dateAdded": ("millisecond", "high"), "edge_epoch": ("millisecond", "medium"),
                    "id_date_segment": ("day", "low")}


def _added_at_from_id(item_id):
    """The YYYYMMDD segment of a tempId_/syncId_ id as an ISO instant, or None."""
    match = ID_DATE.match(item_id if isinstance(item_id, str) else "")
    if match is None:
        return None
    try:
        moment = datetime.strptime(match.group(1), "%Y%m%d")
    except ValueError:
        return None
    return moment.replace(tzinfo=timezone.utc).isoformat()


def _added_at_from_epoch(value):
    """An epoch-millisecond number as an ISO instant, or None when implausible.

    The range check is what keeps a stray counter, such as the middle number of
    an edge id, from being read as a date.
    """
    if type(value) is not int or not EPOCH_MS[0] <= value < EPOCH_MS[1]:
        return None
    return datetime.fromtimestamp(value / 1000, timezone.utc).isoformat()


def _added_at_from_edge_id(edge_id):
    """The trailing millisecond of an ``edge - 9 -1782572278545`` id, or None."""
    tail = (edge_id if isinstance(edge_id, str) else "").rsplit("-", 1)[-1]
    return _added_at_from_epoch(int(tail)) if tail.strip().isdigit() else None


def _added_at_fields(item_id, raw):
    """(added_at, source, confidence) for one item; dateAdded wins over an id segment.

    dateAdded is when the bookmark was added; an id date segment is only a
    generation or import clue, so it is reported with low confidence and never
    overrides a real add time.
    """
    if isinstance(raw, dict):
        epoch = _added_at_from_epoch(raw.get("dateAdded"))
        if epoch is not None:
            return epoch, "dateAdded", ADDED_AT_SOURCES["dateAdded"][1]
    segment = _added_at_from_id(item_id)
    if segment is not None:
        return segment, "id_date_segment", ADDED_AT_SOURCES["id_date_segment"][1]
    return None, None, None


def _dump(value):
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))


def _now():
    return datetime.now(timezone.utc).isoformat()


def _inside(path, directory):
    try:
        path.relative_to(directory)
        return True
    except ValueError:
        return False


def _unique_object(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError("Duplicate JSON key: " + key)
        result[key] = value
    return result


def _reject_constant(value):
    raise ValueError("Non-finite JSON value: " + value)


def _string(value, label, required=False):
    if not isinstance(value, str) or "\x00" in value or (required and not value):
        raise ValueError("Expected %s string" % label)
    return value


def _reference(reference, package_files=None):
    """Resolve the protocol directory suffix, not the referring file's parent.

    The vault root is external to an exported/moved package. Retaining the full
    protocol suffix accommodates its vault prefix without reading outside it.
    """
    return resolve_protocol_reference(reference, package_files=package_files)


SCHEMA = """
CREATE TABLE IF NOT EXISTS sources (
 source_id TEXT PRIMARY KEY, package_path TEXT NOT NULL,
 created_at TEXT NOT NULL, last_sync TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS files (
 source_id TEXT NOT NULL, file_path TEXT NOT NULL, kind TEXT NOT NULL,
 sha256 TEXT NOT NULL, raw_json TEXT NOT NULL, revision INTEGER NOT NULL DEFAULT 1,
 PRIMARY KEY(source_id,file_path),
 FOREIGN KEY(source_id) REFERENCES sources(source_id)
);
CREATE TABLE IF NOT EXISTS sections (
 source_id TEXT NOT NULL, section_id TEXT NOT NULL, file_path TEXT NOT NULL,
 section_type TEXT NOT NULL, label TEXT NOT NULL, slot TEXT NOT NULL,
 title TEXT NOT NULL, description_md TEXT NOT NULL, is_anchor INTEGER NOT NULL,
 inherit_path TEXT, canonical_id TEXT, raw_json TEXT NOT NULL,
 revision INTEGER NOT NULL DEFAULT 1, PRIMARY KEY(source_id,section_id),
 FOREIGN KEY(source_id) REFERENCES sources(source_id)
);
CREATE INDEX IF NOT EXISTS sections_file ON sections(source_id,file_path);
CREATE TABLE IF NOT EXISTS items (
 pk INTEGER PRIMARY KEY, source_id TEXT NOT NULL, section_id TEXT NOT NULL,
 item_id TEXT NOT NULL, parent_id TEXT, item_type TEXT NOT NULL,
 position INTEGER NOT NULL, preorder INTEGER NOT NULL, title TEXT NOT NULL,
 url TEXT NOT NULL, note TEXT NOT NULL, note_color TEXT,
 tags_json TEXT NOT NULL, tag_text TEXT NOT NULL, path_json TEXT NOT NULL,
 folder_path TEXT NOT NULL, ancestor_ids_json TEXT NOT NULL,
 raw_json TEXT NOT NULL, metadata_json TEXT NOT NULL,
 added_at TEXT, added_at_source TEXT, added_at_confidence TEXT,
 revision INTEGER NOT NULL DEFAULT 1,
 UNIQUE(source_id,section_id,item_id),
 FOREIGN KEY(source_id,section_id) REFERENCES sections(source_id,section_id)
);
CREATE INDEX IF NOT EXISTS items_parent ON items(source_id,section_id,parent_id);
CREATE INDEX IF NOT EXISTS items_url ON items(source_id,url);
CREATE TABLE IF NOT EXISTS item_tags (
 item_pk INTEGER NOT NULL REFERENCES items(pk) ON DELETE CASCADE,
 position INTEGER NOT NULL, color TEXT NOT NULL, text TEXT NOT NULL,
 raw_json TEXT NOT NULL, PRIMARY KEY(item_pk,color,text)
);
CREATE VIRTUAL TABLE IF NOT EXISTS items_fts USING fts5(
 title,url,note,tag_text,folder_path,content='items',content_rowid='pk'
);
CREATE TRIGGER IF NOT EXISTS items_fts_ai AFTER INSERT ON items BEGIN
 INSERT INTO items_fts(rowid,title,url,note,tag_text,folder_path)
 VALUES(new.pk,new.title,new.url,new.note,new.tag_text,new.folder_path);
END;
CREATE TRIGGER IF NOT EXISTS items_fts_ad AFTER DELETE ON items BEGIN
 INSERT INTO items_fts(items_fts,rowid,title,url,note,tag_text,folder_path)
 VALUES('delete',old.pk,old.title,old.url,old.note,old.tag_text,old.folder_path);
END;
CREATE TRIGGER IF NOT EXISTS items_fts_au AFTER UPDATE ON items BEGIN
 INSERT INTO items_fts(items_fts,rowid,title,url,note,tag_text,folder_path)
 VALUES('delete',old.pk,old.title,old.url,old.note,old.tag_text,old.folder_path);
 INSERT INTO items_fts(rowid,title,url,note,tag_text,folder_path)
 VALUES(new.pk,new.title,new.url,new.note,new.tag_text,new.folder_path);
END;
CREATE TABLE IF NOT EXISTS nodes (
 source_id TEXT NOT NULL, canvas_path TEXT NOT NULL, node_id TEXT NOT NULL,
 node_type TEXT NOT NULL, section_id TEXT, file_path TEXT, label TEXT NOT NULL,
 text TEXT NOT NULL, x REAL NOT NULL, y REAL NOT NULL, width REAL NOT NULL,
 height REAL NOT NULL, position INTEGER NOT NULL, raw_json TEXT NOT NULL,
 revision INTEGER NOT NULL DEFAULT 1, PRIMARY KEY(source_id,canvas_path,node_id)
);
CREATE TABLE IF NOT EXISTS edges (
 source_id TEXT NOT NULL, canvas_path TEXT NOT NULL, edge_id TEXT NOT NULL,
 from_node TEXT NOT NULL, to_node TEXT NOT NULL,
 from_end TEXT NOT NULL, to_end TEXT NOT NULL, label TEXT NOT NULL,
 raw_json TEXT NOT NULL, revision INTEGER NOT NULL DEFAULT 1,
 PRIMARY KEY(source_id,canvas_path,edge_id)
);
CREATE TABLE IF NOT EXISTS memberships (
 source_id TEXT NOT NULL, canvas_path TEXT NOT NULL,
 group_id TEXT NOT NULL, node_id TEXT NOT NULL,
 PRIMARY KEY(source_id,canvas_path,group_id,node_id)
);
PRAGMA user_version=2;
"""


class BookmarkIndex:
    def __init__(self, db_path):
        self.db_path = None if str(db_path) == ":memory:" else Path(db_path).expanduser().resolve()
        if self.db_path is not None:
            # Refuse before connecting, so even an accidentally chosen package
            # path does not acquire an empty database, journal or WAL file.
            for parent in self.db_path.parents:
                if is_package_dir(parent):
                    raise ValueError("Database must be outside imported package directories")
            self.db_path.parent.mkdir(parents=True, exist_ok=True)
        self.connection = sqlite3.connect(str(self.db_path) if self.db_path else ":memory:")
        self.connection.row_factory = sqlite3.Row
        self.connection.execute("PRAGMA foreign_keys=ON")
        self.connection.execute("PRAGMA busy_timeout=5000")
        version = self.connection.execute("PRAGMA user_version").fetchone()[0]
        if version not in (0, 1, 2):
            self.close()
            raise ValueError("Unsupported bookmark index schema version: %s" % version)
        if version == 1:
            # Add the added-at columns and fill them from stored raw_json before
            # the schema script runs: CREATE TABLE IF NOT EXISTS cannot add a
            # column, and the script would advance the version past this repair.
            try:
                self._upgrade_items_added_at()
            except Exception:
                self.close()
                raise
        try:
            self.connection.executescript(SCHEMA)
        except Exception:
            self.close()
            raise
        self._fts_mode = self._prepare_search_index()
        self._window_functions = self._supports_window_functions()

    def _upgrade_items_added_at(self):
        """v1 → v2: derive added-at for existing rows without touching the package.

        The index is derived data, but its package may be gone or unreadable, so
        the backfill reads the stored raw_json and item_id rather than re-importing.
        """
        with self.transaction():
            for column in ("added_at", "added_at_source", "added_at_confidence"):
                self.connection.execute("ALTER TABLE items ADD COLUMN %s TEXT" % column)
            for row in self.connection.execute("SELECT pk,item_id,raw_json FROM items").fetchall():
                try:
                    raw = json.loads(row["raw_json"])
                except ValueError:
                    raw = {}
                added_at, source, confidence = _added_at_fields(row["item_id"], raw)
                self.connection.execute(
                    "UPDATE items SET added_at=?,added_at_source=?,added_at_confidence=? WHERE pk=?",
                    (added_at, source, confidence, row["pk"]))
            self.connection.execute("PRAGMA user_version=2")

    def _supports_window_functions(self):
        """SQLite 3.25+ can return the pre-LIMIT count alongside the page."""
        try:
            self.connection.execute("SELECT row_number() OVER () FROM (SELECT 1)").fetchone()
        except sqlite3.DatabaseError:
            return False
        return True

    def _prepare_search_index(self):
        """Keep items_fts able to answer literal substring searches.

        The default FTS5 tokenizer treats a whole run of CJK characters as one
        token, so ``MATCH '推理'`` cannot find ``推理加速``. The trigram tokenizer
        does substring matching, which is the same contract as the LIKE fallback
        used for short terms; when it is unavailable the index stays usable and
        every term falls back to LIKE.
        """
        row = self.connection.execute("SELECT sql FROM sqlite_master WHERE name='items_fts'").fetchone()
        existing = (row["sql"] or "") if row is not None else ""
        if row is not None and "trigram" in existing:
            return "trigram"
        if row is not None and not self._trigram_available():
            return "plain"
        with self.transaction():
            self.connection.execute("DROP TABLE IF EXISTS items_fts")
            self.connection.execute("CREATE VIRTUAL TABLE items_fts USING fts5("
                                    "title,url,note,tag_text,folder_path,content='items',content_rowid='pk',"
                                    "tokenize='trigram')")
            self.connection.execute("INSERT INTO items_fts(items_fts) VALUES('rebuild')")
        return "trigram"

    def _trigram_available(self):
        try:
            self.connection.execute("CREATE VIRTUAL TABLE temp.__trigram_probe USING fts5(body, tokenize='trigram')")
        except sqlite3.DatabaseError:
            return False
        finally:
            try:
                self.connection.execute("DROP TABLE IF EXISTS temp.__trigram_probe")
            except sqlite3.DatabaseError:
                pass
        return True

    def close(self):
        self.connection.close()

    def __enter__(self):
        return self

    def __exit__(self, *_):
        self.close()

    def _source(self, source_id):
        row = self.connection.execute("SELECT * FROM sources WHERE source_id=?", (source_id,)).fetchone()
        if row is None:
            raise ValueError("Unknown source_id: " + str(source_id))
        return dict(row)

    def _input_path(self, source_id, fallback):
        """Use the live input for live sources and the managed copy for snapshots."""
        try:
            row = self.connection.execute(
                "SELECT record_json FROM source_records WHERE source_id=?", (source_id,)).fetchone()
        except sqlite3.OperationalError:
            # Direct BookmarkIndex.sync() callers do not have the optional
            # source_records table; the package path fallback is exact for them.
            row = None
        if row is not None:
            try:
                record = json.loads(row[0])
                if record.get("mode") == "snapshot":
                    return Path(fallback)
                value = record.get("input_path")
                if isinstance(value, str) and value:
                    path = Path(value).expanduser()
                    if path.is_dir():
                        return path
            except (TypeError, ValueError, KeyError):
                pass
        return Path(fallback)

    def _upsert(self, table, keys, values):
        """Internal table/column names only; user values are always parameters."""
        where = " AND ".join(key + "=?" for key in keys)
        row = self.connection.execute("SELECT * FROM %s WHERE %s" % (table, where), tuple(keys.values())).fetchone()
        if row is not None and all(row[key] == value for key, value in values.items()):
            return "unchanged"
        if row is None:
            fields = dict(keys, **values)
            self.connection.execute("INSERT INTO %s (%s) VALUES (%s)" % (
                table, ",".join(fields), ",".join("?" for _ in fields)), tuple(fields.values()))
            return "inserted"
        self.connection.execute("UPDATE %s SET %s,revision=revision+1 WHERE %s" % (
            table, ",".join(key + "=?" for key in values), where), tuple(values.values()) + tuple(keys.values()))
        return "updated"

    @staticmethod
    def _section(file_path, obj, package_files=None):
        if obj.get("format") != "bookmark-canvas-section":
            raise ValueError("Not a Bookmark Canvas section: " + file_path)
        section_type = obj.get("sectionType")
        if section_type not in ("permanent", "temporary"):
            raise ValueError("Unknown sectionType in " + file_path)
        if not dir_matches_section_type(file_path, section_type):
            raise ValueError("sectionType does not match protocol directory: " + file_path)
        anchor = obj.get("fileRole") == "copy-anchor" or obj.get("anchorOnly") is True
        slot = _string(obj.get("slot", ""), "slot")
        if section_type == "permanent":
            if not slot:
                raise ValueError("Permanent section needs slot: " + file_path)
            if anchor:
                if "tree" in obj:
                    raise ValueError("Copy anchors must not duplicate a tree")
                copy_id = _string(obj.get("copyId"), "copyId", True)
                section_id = obj.get("id", "permanent-section-copy-" + copy_id)
            else:
                section_id = obj.get("id", "permanent-section" if slot == "A" else "permanent-section-" + slot)
            label = slot
        else:
            if anchor:
                raise ValueError("Only permanent sections can be copy anchors")
            section_id = obj.get("id")
            label = obj.get("label", "unknown")
        _string(section_id, "section id", True)
        _string(label, "section label", True)
        return section_id, {
            "file_path": file_path, "section_type": section_type, "label": label,
            "slot": slot, "title": _string(obj.get("title", ""), "section title"),
            "description_md": _string(obj.get("descriptionMd", ""), "descriptionMd"),
            "is_anchor": int(anchor), "inherit_path": _reference(obj.get("inheritFrom"), package_files=package_files) if anchor else None,
            "raw_json": _dump(obj),
        }

    @staticmethod
    def _items(section_id, section, obj):
        if section["is_anchor"]:
            return []
        permanent = section["section_type"] == "permanent"
        roots = obj.get("tree") if permanent else obj.get("items")
        if permanent and isinstance(roots, dict):
            roots = [roots]
        if not isinstance(roots, list):
            raise ValueError("Section needs a complete tree/items value: " + section_id)
        metadata = obj.get("identityMap", []) if permanent else []
        if isinstance(metadata, dict):
            meta = {}
            for key, value in metadata.items():
                if not isinstance(value, dict):
                    raise ValueError("identityMap entries must be objects")
                if value.get("syncId", key) != key:
                    raise ValueError("identityMap key/syncId mismatch")
                meta[key] = value
        elif isinstance(metadata, list):
            meta = {}
            for value in metadata:
                if not isinstance(value, dict):
                    raise ValueError("identityMap entries must be objects")
                key = _string(value.get("syncId"), "identityMap syncId", True)
                if key in meta:
                    raise ValueError("Duplicate identityMap syncId: " + key)
                meta[key] = value
        else:
            raise ValueError("identityMap must be an array or keyed object")
        result, seen = [], set()

        def visit(nodes, parent_id=None, titles=(), ancestor_ids=()):
            for position, node in enumerate(nodes):
                if not isinstance(node, dict):
                    raise ValueError("Items must be JSON objects")
                item_id = _string(node.get("id"), "item id", True)
                if item_id in seen:
                    raise ValueError("Duplicate item id in section: " + item_id)
                seen.add(item_id)
                if permanent and "parentId" in node and node["parentId"] != parent_id:
                    raise ValueError("parentId disagrees with tree position: " + item_id)
                if not permanent and node.get("sectionId", section_id) != section_id:
                    raise ValueError("Temporary item sectionId mismatch: " + item_id)
                title = _string(node.get("title", ""), "item title")
                url = _string(node.get("url", ""), "item URL")
                item_type = ("bookmark" if url else "folder") if permanent else node.get("type")
                if item_type not in ("bookmark", "folder") or (item_type == "bookmark" and not url):
                    raise ValueError("Invalid bookmark/folder type: " + item_id)
                children = node.get("children", [])
                if not isinstance(children, list) or (children and item_type == "bookmark"):
                    raise ValueError("Invalid item children: " + item_id)
                item_meta = meta.get(item_id, {}) if permanent else node
                tags = item_meta.get("tags", [])
                if not isinstance(tags, list):
                    raise ValueError("tags must be an array")
                normalized, tag_keys = [], set()
                for tag in tags:
                    if not isinstance(tag, dict):
                        raise ValueError("tags must contain color/text objects")
                    key = (_string(tag.get("color"), "tag color", True), _string(tag.get("text", ""), "tag text"))
                    if key not in tag_keys:
                        normalized.append(tag)
                        tag_keys.add(key)
                note = _string(item_meta.get("note", ""), "note").strip()
                note_color = item_meta.get("noteColor", "orange") if note else None
                if note_color is not None:
                    _string(note_color, "noteColor")
                added = _added_at_fields(item_id, node)
                result.append((item_id, {
                    "parent_id": parent_id, "item_type": item_type, "position": position,
                    "preorder": len(result), "title": title, "url": url, "note": note,
                    "note_color": note_color, "tags_json": _dump(normalized),
                    "tag_text": "\n".join(tag["text"] for tag in normalized),
                    "path_json": _dump(list(titles)), "folder_path": " / ".join(titles),
                    "ancestor_ids_json": _dump(list(ancestor_ids)),
                    # Full child arrays also survive in files/sections.raw_json.
                    "raw_json": _dump({key: value for key, value in node.items() if key != "children"}),
                    "metadata_json": _dump(item_meta if permanent else {key: node[key] for key in ("tags", "note", "noteColor") if key in node}),
                    "added_at": added[0], "added_at_source": added[1], "added_at_confidence": added[2],
                }))
                visit(children, item_id, titles + ((title,) if title else ()), ancestor_ids + (item_id,))

        visit(roots)
        return result

    @staticmethod
    def _canvas(obj, package_files=None):
        if not isinstance(obj.get("nodes"), list) or not isinstance(obj.get("edges"), list):
            raise ValueError("Canvas requires nodes[] and edges[]")
        nodes, edges, seen = [], [], set()
        for position, node in enumerate(obj["nodes"]):
            if not isinstance(node, dict):
                raise ValueError("Canvas nodes must be objects")
            node_id = _string(node.get("id"), "node id", True)
            if node_id in seen:
                raise ValueError("Duplicate canvas node: " + node_id)
            seen.add(node_id)
            node_type = node.get("type")
            if node_type not in ("file", "text", "group"):
                raise ValueError("Unsupported package canvas node type: " + str(node_type))
            geometry = {}
            for key in ("x", "y", "width", "height"):
                value = node.get(key)
                if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value):
                    raise ValueError("Invalid canvas geometry: " + key)
                if key in ("width", "height") and value <= 0:
                    raise ValueError("Canvas dimensions must be positive")
                geometry[key] = value
            nodes.append((node_id, dict(geometry, node_type=node_type,
                file_path=_reference(node.get("file"), package_files=package_files) if node_type == "file" else None,
                label=_string(node.get("label", ""), "node label"),
                text=_string(node.get("text", ""), "node text"), position=position, raw_json=_dump(node))))
        edge_ids = set()
        for edge in obj["edges"]:
            if not isinstance(edge, dict):
                raise ValueError("Edges must be objects")
            edge_id = _string(edge.get("id"), "edge id", True)
            if edge_id in edge_ids:
                raise ValueError("Duplicate edge id")
            edge_ids.add(edge_id)
            if edge.get("fromNode") not in seen or edge.get("toNode") not in seen:
                raise ValueError("Canvas edge endpoint does not exist: " + edge_id)
            from_end, to_end = edge.get("fromEnd", "none"), edge.get("toEnd", "arrow")
            if from_end not in ("none", "arrow") or to_end not in ("none", "arrow"):
                raise ValueError("Unknown canvas edge end")
            edges.append((edge_id, {"from_node": edge["fromNode"], "to_node": edge["toNode"],
                "from_end": from_end, "to_end": to_end, "label": _string(edge.get("label", ""), "edge label"),
                "raw_json": _dump(edge)}))
        return nodes, edges

    @contextmanager
    def read_snapshot(self):
        self.connection.execute("SAVEPOINT bookmark_read")
        try:
            yield
        finally:
            self.connection.execute("RELEASE SAVEPOINT bookmark_read")

    @contextmanager
    def transaction(self):
        """Serialize writers; allow an importer to publish metadata atomically."""
        nested = self.connection.in_transaction
        if nested:
            self.connection.execute("SAVEPOINT bookmark_sync")
        else:
            self.connection.execute("BEGIN IMMEDIATE")
        try:
            yield
            if nested:
                self.connection.execute("RELEASE SAVEPOINT bookmark_sync")
            else:
                self.connection.commit()
        except BaseException:
            if nested:
                self.connection.execute("ROLLBACK TO SAVEPOINT bookmark_sync")
                self.connection.execute("RELEASE SAVEPOINT bookmark_sync")
            else:
                self.connection.rollback()
            raise

    def sync(self, package_path, source_id=None, completeness="partial"):
        with self.transaction():
            return self._sync(package_path, source_id, completeness)

    def _sync(self, package_path, source_id, completeness):
        if completeness not in ("partial", "complete"):
            raise ValueError("completeness must be partial or complete")
        package = Path(package_path).expanduser().resolve()
        if not package.is_dir():
            raise ValueError("Package directory does not exist: " + str(package))
        if self.db_path is not None and _inside(self.db_path, package):
            raise ValueError("Database must be outside the imported package")
        if source_id is None:
            existing = self.connection.execute("SELECT source_id FROM sources WHERE package_path=?", (str(package),)).fetchall()
            if len(existing) > 1:
                raise ValueError("Multiple sources use this path; provide source_id")
            source_id = existing[0][0] if existing else "source-" + hashlib.sha256(str(package).encode()).hexdigest()[:12]
        _string(source_id, "source_id", True)
        paths = list(package.glob("*.canvas"))
        if len(paths) > 1:
            raise ValueError("A package must have at most one .canvas entry; use separate sources for separate canvases")
        canvas_path = paths[0].relative_to(package).as_posix() if paths else None
        for directory in PROTOCOL_DIRS:
            paths.extend((package / directory).rglob("*.json"))
        paths = sorted(paths)
        if not paths:
            raise ValueError("No protocol JSON or .canvas files in package")
        old_files = {row["file_path"]: dict(row) for row in self.connection.execute(
            "SELECT * FROM files WHERE source_id=?", (source_id,))}
        indexed_section_files = {row[0] for row in self.connection.execute(
            "SELECT file_path FROM sections WHERE source_id=?", (source_id,))}
        package_section_files = {p.relative_to(package).as_posix() for p in paths if p.suffix == ".json"} | indexed_section_files
        changed, observed, section_ids = [], set(), set()
        # Parse and validate every changed file before starting the mutation.
        for path in paths:
            if not _inside(path.resolve(), package):
                raise ValueError("Package contains an external symlink: " + str(path))
            relative = path.relative_to(package).as_posix()
            observed.add(relative)
            raw = path.read_bytes()
            digest = hashlib.sha256(raw).hexdigest()
            if (relative in old_files and old_files[relative]["sha256"] == digest
                    and (path.suffix == ".canvas" or relative in indexed_section_files)):
                continue
            try:
                obj = json.loads(raw.decode("utf-8-sig"), object_pairs_hook=_unique_object, parse_constant=_reject_constant)
                if not isinstance(obj, dict):
                    raise ValueError("File must contain one JSON object")
                if path.suffix == ".canvas":
                    parsed = self._canvas(obj, package_files=package_section_files)
                else:
                    section_id, section = self._section(relative, obj, package_files=package_section_files)
                    if section_id in section_ids:
                        raise ValueError("Duplicate section ID in package: " + section_id)
                    section_ids.add(section_id)
                    parsed = (section_id, section, self._items(section_id, section, obj))
            except (ValueError, TypeError, RecursionError) as exc:
                raise ValueError("Invalid package file %s: %s" % (relative, exc)) from exc
            changed.append((relative, digest, obj, parsed))
        summary = {"source_id": source_id, "package_path": str(package),
            "changed_files": len(changed), "skipped_files": len(observed) - len(changed),
            "items_inserted": 0, "items_updated": 0, "items_deleted": 0,
            "retained_missing_files": sorted(set(old_files) - observed), "warnings": []}
        summary["replaced_canvas_files"] = [relative for relative, record in old_files.items()
            if canvas_path is not None and record["kind"] == "canvas" and relative != canvas_path]
        timestamp = _now()
        with self.transaction():
            self.connection.execute("""INSERT INTO sources VALUES(?,?,?,?)
                ON CONFLICT(source_id) DO UPDATE SET package_path=excluded.package_path,last_sync=excluded.last_sync""",
                (source_id, str(package), timestamp, timestamp))
            # One logical source has one canvas entry. A supplied entry replaces
            # its prior path, while a partial export without one retains the layout.
            for previous in summary["replaced_canvas_files"]:
                for table in ("memberships", "edges", "nodes"):
                    self.connection.execute("DELETE FROM %s WHERE source_id=? AND canvas_path=?" % table,
                                            (source_id, previous))
                self.connection.execute("DELETE FROM files WHERE source_id=? AND file_path=?", (source_id, previous))
            for relative, digest, obj, parsed in changed:
                if relative.endswith(".canvas"):
                    continue
                section_id, values, items = parsed
                # A renamed file may retain its stable section ID. A second
                # currently observed file with that ID is ambiguous, however.
                prior = self.connection.execute("SELECT file_path FROM sections WHERE source_id=? AND section_id=?", (source_id, section_id)).fetchone()
                if prior and prior[0] != relative and prior[0] in observed:
                    raise ValueError("Two present files share section ID: " + section_id)
                self._upsert("sections", {"source_id": source_id, "section_id": section_id}, values)
                present_items = set()
                for item_id, item in items:
                    present_items.add(item_id)
                    action = self._upsert("items", {"source_id": source_id, "section_id": section_id, "item_id": item_id}, item)
                    if action != "unchanged":
                        summary["items_" + action] += 1
                        pk = self.connection.execute("SELECT pk FROM items WHERE source_id=? AND section_id=? AND item_id=?", (source_id, section_id, item_id)).fetchone()[0]
                        self.connection.execute("DELETE FROM item_tags WHERE item_pk=?", (pk,))
                        for position, tag in enumerate(json.loads(item["tags_json"])):
                            self.connection.execute("INSERT INTO item_tags VALUES(?,?,?,?,?)", (pk, position, tag["color"], tag["text"], _dump(tag)))
                for row in self.connection.execute("SELECT pk,item_id FROM items WHERE source_id=? AND section_id=?", (source_id, section_id)).fetchall():
                    if row["item_id"] not in present_items:
                        self.connection.execute("DELETE FROM items WHERE pk=?", (row["pk"],))
                        summary["items_deleted"] += 1
            # A complete directory is a mirror. Resolve renames by section ID
            # first so moving a file preserves item row IDs and revisions.
            summary["removed_files"] = []
            if completeness == "complete":
                for row in self.connection.execute("SELECT section_id,file_path FROM sections WHERE source_id=?", (source_id,)).fetchall():
                    if row["file_path"] not in observed:
                        summary["items_deleted"] += self.connection.execute(
                            "DELETE FROM items WHERE source_id=? AND section_id=?", (source_id, row["section_id"])).rowcount
                        self.connection.execute("DELETE FROM sections WHERE source_id=? AND section_id=?", (source_id, row["section_id"]))
                for previous in set(old_files) - observed:
                    for table in ("memberships", "edges", "nodes"):
                        self.connection.execute("DELETE FROM %s WHERE source_id=? AND canvas_path=?" % table, (source_id, previous))
                    self.connection.execute("DELETE FROM files WHERE source_id=? AND file_path=?", (source_id, previous))
                    summary["removed_files"].append(previous)
                summary["removed_files"].sort()
            sections = {row["section_id"]: dict(row) for row in self.connection.execute("SELECT * FROM sections WHERE source_id=?", (source_id,))}
            by_file = {}
            for sid, section in sections.items():
                if section["file_path"] in by_file:
                    raise ValueError("A section file changed identity; use a new source or preserve its ID")
                by_file[section["file_path"]] = sid

            def canonical(sid, visiting=()):
                if sid in visiting:
                    raise ValueError("Copy-anchor inheritance cycle")
                section = sections[sid]
                if not section["is_anchor"]:
                    return sid
                target = by_file.get(section["inherit_path"])
                if target is None:
                    # A retained anchor still identifies its previously resolved
                    # tree when a partial export renames that stable section ID.
                    prior = section["canonical_id"]
                    if section["file_path"] not in observed and prior in sections:
                        summary["warnings"].append("Retained copy-anchor uses prior file reference: " + sid)
                        target = prior
                    else:
                        return None
                if sections[target]["section_type"] != "permanent":
                    raise ValueError("Copy anchor must inherit a permanent section")
                return canonical(target, visiting + (sid,))

            for sid, section in sections.items():
                resolved = canonical(sid)
                if resolved != section["canonical_id"]:
                    self.connection.execute("UPDATE sections SET canonical_id=? WHERE source_id=? AND section_id=?", (resolved, source_id, sid))
                if resolved is None:
                    summary["warnings"].append("Unresolved copy-anchor target: " + sid)
            for relative, digest, obj, parsed in changed:
                if relative.endswith(".canvas"):
                    nodes, edges = parsed
                    for table, id_column, rows in (("nodes", "node_id", nodes), ("edges", "edge_id", edges)):
                        retained = set()
                        for identifier, values in rows:
                            retained.add(identifier)
                            if table == "nodes":
                                values = dict(values, section_id=by_file.get(values["file_path"]))
                            self._upsert(table, {"source_id": source_id, "canvas_path": relative, id_column: identifier}, values)
                        old = self.connection.execute("SELECT %s FROM %s WHERE source_id=? AND canvas_path=?" % (id_column, table), (source_id, relative)).fetchall()
                        for row in old:
                            if row[0] not in retained:
                                self.connection.execute("DELETE FROM %s WHERE source_id=? AND canvas_path=? AND %s=?" % (table, id_column), (source_id, relative, row[0]))
                    self._memberships(source_id, relative)
                self._upsert("files", {"source_id": source_id, "file_path": relative}, {
                    "kind": "canvas" if relative.endswith(".canvas") else "section", "sha256": digest, "raw_json": _dump(obj)})
            # A stable section ID moving to a new path supersedes its old file
            # hash. Keeping that hash would skip parsing if it later moved back.
            # Missing sections themselves remain retained for partial exports.
            self.connection.execute("""DELETE FROM files WHERE source_id=? AND kind='section'
                AND NOT EXISTS (SELECT 1 FROM sections s
                    WHERE s.source_id=files.source_id AND s.file_path=files.file_path)""", (source_id,))
            # A partial package can supply a previously unresolved referenced
            # section without changing any canvas bytes.
            for node in self.connection.execute("SELECT * FROM nodes WHERE source_id=? AND node_type='file'", (source_id,)).fetchall():
                resolved = by_file.get(node["file_path"])
                if resolved is None and node["canvas_path"] not in observed and node["section_id"] in sections:
                    # Keep known card identity for an omitted (older) layout; a
                    # currently supplied canvas always resolves its actual paths.
                    resolved = node["section_id"]
                    summary["warnings"].append("Retained canvas uses prior file reference: " + node["file_path"])
                if resolved != node["section_id"]:
                    self.connection.execute("UPDATE nodes SET section_id=? WHERE source_id=? AND canvas_path=? AND node_id=?", (resolved, source_id, node["canvas_path"], node["node_id"]))
                if resolved is None:
                    summary["warnings"].append("Unresolved canvas file: " + node["file_path"])
            if self.connection.execute("PRAGMA foreign_key_check").fetchone():
                raise ValueError("Derived database failed relationship validation")
        summary["counts"] = self.status(source_id)["counts"]
        summary["retained_missing_files"] = sorted(row[0] for row in self.connection.execute(
            "SELECT file_path FROM files WHERE source_id=?", (source_id,)) if row[0] not in observed)
        return summary

    def _memberships(self, source_id, canvas_path):
        nodes = self.connection.execute("SELECT * FROM nodes WHERE source_id=? AND canvas_path=?", (source_id, canvas_path)).fetchall()
        expected = set()
        for group in nodes:
            if group["node_type"] != "group":
                continue
            for node in nodes:
                if group["node_id"] == node["node_id"]:
                    continue
                if (group["x"] <= node["x"] and group["y"] <= node["y"]
                        and node["x"] + node["width"] <= group["x"] + group["width"]
                        and node["y"] + node["height"] <= group["y"] + group["height"]):
                    # Equal rectangles do not define reciprocal group nesting.
                    if node["node_type"] == "group" and all(node[k] == group[k] for k in ("x", "y", "width", "height")):
                        continue
                    expected.add((group["node_id"], node["node_id"]))
        old = {(r[0], r[1]) for r in self.connection.execute("SELECT group_id,node_id FROM memberships WHERE source_id=? AND canvas_path=?", (source_id, canvas_path))}
        for pair in old - expected:
            self.connection.execute("DELETE FROM memberships WHERE source_id=? AND canvas_path=? AND group_id=? AND node_id=?", (source_id, canvas_path) + pair)
        for pair in expected - old:
            self.connection.execute("INSERT INTO memberships VALUES(?,?,?,?)", (source_id, canvas_path) + pair)

    def refresh(self, source_id):
        return self.sync(self._source(source_id)["package_path"], source_id=source_id)

    def _sections(self, source_id, selector=None):
        rows = self.connection.execute("SELECT * FROM sections WHERE source_id=? ORDER BY file_path,section_id", (source_id,)).fetchall()
        if selector is not None:
            rows = [row for row in rows if selector in (row["section_id"], row["label"], row["slot"], row["file_path"])]
            if not rows:
                raise ValueError("Unknown section: " + str(selector))
        return rows

    def _group(self, source_id, group_id):
        rows = self.connection.execute("SELECT * FROM nodes WHERE source_id=? AND node_type='group' AND node_id=?", (source_id, group_id)).fetchall()
        if not rows:
            rows = self.connection.execute("SELECT * FROM nodes WHERE source_id=? AND node_type='group' AND label=?", (source_id, group_id)).fetchall()
        if len(rows) != 1:
            raise ValueError("Unknown or ambiguous group; use a unique group ID: " + str(group_id))
        return rows[0]

    def _scope(self, source_id, section=None, group_id=None, folder_id=None, tags=None, tag_colors=None,
               item_types=None):
        self._source(source_id)
        selected_types = list(item_types) if item_types else ["bookmark"]
        clauses = ["i.source_id=?", "i.item_type IN (%s)" % ",".join("?" for _ in selected_types)]
        args = [source_id, *selected_types]
        sections = self._sections(source_id, section) if section is not None or group_id is not None else None
        if group_id is not None:
            group = self._group(source_id, group_id)
            members = {row[0] for row in self.connection.execute("""SELECT n.section_id FROM memberships m
              JOIN nodes n ON n.source_id=m.source_id AND n.canvas_path=m.canvas_path AND n.node_id=m.node_id
              WHERE m.source_id=? AND m.canvas_path=? AND m.group_id=?""",
                (source_id, group["canvas_path"], group["node_id"]))}
            # Intersect the requested cards before following copy anchors to
            # the shared tree: a copy's group does not contain its primary card.
            sections = [row for row in sections if row["section_id"] in members]
        if sections is not None:
            ids = {row["canonical_id"] for row in sections if row["canonical_id"]}
            clauses.append("i.section_id IN (%s)" % (",".join("?" for _ in ids) or "NULL"))
            args.extend(sorted(ids))
        if folder_id is not None:
            folders = self.connection.execute("SELECT section_id FROM items WHERE source_id=? AND item_id=? AND item_type='folder'", (source_id, folder_id)).fetchall()
            if not folders:
                raise ValueError("Unknown folder ID: " + str(folder_id))
            # Use relational ancestry rather than requiring SQLite's optional
            # JSON1 extension or inferring ancestry from folder-name strings.
            clauses.append("""EXISTS (WITH RECURSIVE subtree(section_id,item_id) AS (
                SELECT section_id,item_id FROM items
                WHERE source_id=? AND item_id=? AND item_type='folder'
                UNION ALL
                SELECT child.section_id,child.item_id FROM items child JOIN subtree p
                ON child.section_id=p.section_id AND child.parent_id=p.item_id
                WHERE child.source_id=?
            ) SELECT 1 FROM subtree d WHERE d.section_id=i.section_id AND d.item_id=i.item_id)""")
            args.extend((source_id, folder_id, source_id))
        for tag in tags or []:
            _string(tag, "tag", True)
            clauses.append("EXISTS (SELECT 1 FROM item_tags t WHERE t.item_pk=i.pk AND t.text=?)")
            args.append(tag)
        for color in tag_colors or []:
            _string(color, "tag color", True)
            clauses.append("EXISTS (SELECT 1 FROM item_tags t WHERE t.item_pk=i.pk AND t.color=?)")
            args.append(color)
        return clauses, args

    @staticmethod
    def plan_terms(question):
        """Question text into literal search terms; no semantics, no guessing.

        ASCII runs of two or more characters are kept whole. A CJK run of four
        or fewer characters is kept whole as well; a longer run contributes its
        bigrams instead, so 本地部署 still reaches a card that only says 部署.
        A single ASCII character is dropped, because it would match almost
        everything; a single CJK character is kept, because it is often the
        whole entity. The order is the order of appearance and the list is
        capped, because a bounded OR is cheaper than an unbounded one and a
        fixed order is testable.
        """
        if not isinstance(question, str):
            raise ValueError("question must be a string")
        if not question.strip():
            raise ValueError("question must not be empty")
        # A URL in the question is an instruction to read it, not a search term:
        # its scheme and host would otherwise match half the package.
        terms = []
        for run in PLAN_RUN.findall(QUESTION_URL.sub(" ", question)):
            if run[0].isascii():
                candidates = [run] if len(run) >= 2 else []
            else:
                candidates = [run] if len(run) <= 4 else []
                candidates.extend(run[index:index + 2] for index in range(len(run) - 1))
            for candidate in candidates:
                if candidate not in terms:
                    terms.append(candidate)
        return terms[:PLAN_MAX_TERMS]

    def plan_scope(self, source_id, question, max_cards=3, max_urls=8, homepage_limit=0):
        """Rank the cards a question touches and the bookmarks inside them.

        Local only: literal matching over indexed metadata, no fetch and no
        reasoning. ``cards`` is ranked by term weight scaled by the card's own
        density, not by raw hits: a term contributes its length divided by the
        number of cards it matches (so a specific name outranks a common word
        that appears everywhere), and the sum is multiplied by hits/bookmarks
        (so a 1,053-bookmark tree cannot win on its size alone). One card per
        canonical tree is kept, so a copy anchor cannot occupy two of the
        limited slots, and each card reports the terms it matched, its density
        and its weight. ``term_cards`` counts the canonical trees each term
        points at, which is how ``plan_research`` tells a name (one tree) from
        a common word. ``urls`` holds the matched bookmarks of the selected
        cards, deduplicated by URL, with the terms each URL's own rows matched
        and the source-wide instance count, which is not the selected scope.
        """
        source = self._source(source_id)
        package = self._input_path(source_id, source["package_path"])
        # Partial imports keep cards whose file disappeared. They stay indexed,
        # but ranking them above current cards hides the fact that their text is
        # no longer in the package. Their URLs remain useful as a supplement.
        missing_files = {row[0] for row in self.connection.execute(
            "SELECT file_path FROM files WHERE source_id=?", (source_id,))
            if not (package / row[0]).is_file()}
        if type(max_cards) is not int or not 1 <= max_cards <= 3:
            raise ValueError("max_cards must be an integer from 1 to 3")
        if type(max_urls) is not int or not 1 <= max_urls <= 8:
            raise ValueError("max_urls must be an integer from 1 to 8")
        if type(homepage_limit) is not int or not 0 <= homepage_limit <= 8:
            raise ValueError("homepage_limit must be an integer from 0 to 8")
        terms = self.plan_terms(question)
        result = {"source_id": source_id, "terms": terms, "term_cards": {}, "cards": [], "urls": [],
                  "truncated": False, "errors": []}
        if not terms:
            result["errors"].append("The question has no searchable term: a single ASCII character is not "
                                    "searched. Ask with a longer name, a phrase or a URL")
            return result
        clauses = [self._match(term) for term in terms]
        where = " OR ".join(clause for clause, _ in clauses)
        parameters = [value for _, values in clauses for value in values]
        # One grouped count per term answers both questions at once: how many
        # cards a term touches (its weight denominator) and which cards matched
        # it. No per-card loop is needed.
        per_term = {}
        for term, (clause, values) in zip(terms, clauses):
            per_term[term] = {row["section_id"]: row["hits"] for row in self.connection.execute(
                "SELECT i.section_id AS section_id, count(*) AS hits FROM items i"
                " WHERE i.source_id=? AND i.item_type='bookmark' AND " + clause + " GROUP BY i.section_id",
                (source_id, *values))}
        hits = {row["section_id"]: row["hits"] for row in self.connection.execute(
            "SELECT i.section_id AS section_id, count(*) AS hits FROM items i"
            " WHERE i.source_id=? AND i.item_type='bookmark' AND (" + where + ")"
            " GROUP BY i.section_id", (source_id, *parameters))}
        if not hits:
            result["errors"].append("No bookmark in this source matches the question terms")
            return result
        sections = {row["section_id"]: row for row in self._sections(source_id)}
        scores = {}
        for term, matched in per_term.items():
            if not matched:
                continue
            weight = float(len(term)) / len(matched)
            for section_id in matched:
                scores[section_id] = scores.get(section_id, 0.0) + weight
        # A title occurrence is a stronger signal than a note/URL hit. Only
        # longer terms need it, and the check is token-aware for ASCII so
        # "quantum" does not count inside "QuantumNous".
        # A term that reaches exactly one canonical tree names that tree; a term
        # that reaches many is a common word, whatever it looks like.
        for term, matched in per_term.items():
            result["term_cards"][term] = len({sections[value]["canonical_id"] or value
                                              for value in matched}) if matched else 0
        # A title occurrence is a stronger signal than a note/URL hit. Only
        # terms that reach a few trees need the extra scan; a common word that
        # appears across the whole package cannot be a name.
        title_hits = {}
        for term in terms:
            if len(term) < 4 or result["term_cards"].get(term, 0) > 3:
                continue
            escaped = term.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
            counts = {}
            for row in self.connection.execute(
                    "SELECT section_id, title FROM items WHERE source_id=? AND item_type='bookmark'"
                    " AND title LIKE ? ESCAPE '\\'", (source_id, "%" + escaped + "%")):
                if title_has_term(row["title"], term):
                    counts[row["section_id"]] = counts.get(row["section_id"], 0) + 1
            title_hits[term] = counts
        bookmarks = {row["section_id"]: row["count"] for row in self.connection.execute(
            "SELECT section_id, count(*) AS count FROM items WHERE source_id=? AND item_type='bookmark'"
            " GROUP BY section_id", (source_id,))}
        newest = {row["section_id"]: (row["newest"], row["newest_high"]) for row in self.connection.execute(
            "SELECT section_id, max(added_at) AS newest,"
            " max(CASE WHEN added_at_confidence='high' THEN added_at END) AS newest_high"
            " FROM items WHERE source_id=? AND item_type='bookmark'"
            " AND added_at IS NOT NULL GROUP BY section_id", (source_id,))}
        # Size must not win by itself: the term weight is scaled by the card's own
        # hit density, so an 82-of-107 topic card outranks an 18-of-1,053 tree.
        density = {value: (hits[value] / bookmarks[value]) if bookmarks.get(value) else 0.0
                   for value in scores}
        # Two passes keep mixed sort directions readable: the first fixes the
        # ascending tie-breaks, the second orders by score, hits and freshness,
        # and equal rows keep the order the first pass gave them.
        ordered = sorted(hits, key=lambda value: (sections[value]["file_path"], value))
        ordered.sort(key=lambda value: (scores[value] * density[value], hits[value],
                                        (newest.get(value) or (None, None))[0] or ""), reverse=True)
        trees, ranked = set(), []
        for section_id in ordered:
            row = sections[section_id]
            tree = row["canonical_id"] or section_id
            if tree in trees:
                continue
            trees.add(tree)
            ranked.append({"section_id": section_id, "label": row["label"], "title": row["title"] or "",
                "file_path": row["file_path"],
                "hits": hits[section_id], "score": round(scores[section_id] * density[section_id], 6),
                "density": round(density[section_id], 6),
                "newest_added_at": (newest.get(section_id) or (None, None))[0],
                "newest_high_added_at": (newest.get(section_id) or (None, None))[1],
                "bookmarks": bookmarks.get(section_id, 0),
                "matched_terms": [term for term in terms if section_id in per_term[term]],
                "term_hits": {term: per_term[term][section_id] for term in terms
                              if section_id in per_term[term]},
                "term_title_hits": {term: title_hits.get(term, {}).get(section_id, 0) for term in terms
                                    if title_hits.get(term, {}).get(section_id)},
                "stale_missing_file": row["file_path"] in missing_files,
                "anchor_of": row["canonical_id"] if row["is_anchor"] else None})
        eligible = [card for card in ranked if not card["stale_missing_file"]]
        stale = [card for card in ranked if card["stale_missing_file"]]
        # A stale card never takes a top slot while a current card exists; its
        # URLs still join the candidate pool below, so a deleted card's precise
        # entry points are not lost.
        result["cards"] = (eligible or ranked)[:max_cards]
        result["stale_missing_files"] = sorted(missing_files)
        selected = [card["section_id"] for card in result["cards"]]
        stale_sections = [card["section_id"] for card in stale[:max_cards]
                          if card["section_id"] not in selected]
        selected_for_urls = selected + stale_sections
        placeholders = ",".join("?" for _ in selected_for_urls)
        rows = [dict(row) for row in self.connection.execute(
            "SELECT i.url AS url, count(*) AS instances, min(i.title) AS title,"
            " group_concat(DISTINCT i.section_id) AS sections FROM items i"
            " WHERE i.source_id=? AND i.item_type='bookmark' AND i.url<>''"
            " AND i.section_id IN (" + placeholders + ") AND (" + where + ")"
            " GROUP BY i.url", (source_id, *selected_for_urls, *parameters))]
        # A URL is as relevant as the most specific term its own rows matched, so
        # a bookmark that matched a name comes before incidental hits on a common
        # word, wherever inside the selection it lives. Card score, raw instance
        # counts and the URL itself only break ties.
        url_weight, url_terms = {}, {}
        for term, (clause, values) in zip(terms, clauses):
            if not per_term[term]:
                continue
            weight = float(len(term)) / len(per_term[term])
            for url, in self.connection.execute(
                    "SELECT DISTINCT i.url FROM items i WHERE i.source_id=? AND i.item_type='bookmark'"
                    " AND i.url<>'' AND i.section_id IN (" + placeholders + ") AND " + clause,
                    (source_id, *selected_for_urls, *values)):
                url_weight[url] = max(url_weight.get(url, 0.0), weight)
                url_terms.setdefault(url, []).append(term)
        card_score = {card["section_id"]: card["score"] for card in result["cards"] + stale}
        rows.sort(key=lambda row: (-url_weight.get(row["url"], 0.0), URL_KIND_RANK[url_kind(row["url"])],
            -max(card_score.get(value, 0.0) for value in row["sections"].split(",")),
            -row["instances"], row["url"]))
        if homepage_limit:
            kept, homepage_counts = [], {}
            for row in rows:
                if url_kind(row["url"]) == "homepage":
                    sections = row["sections"].split(",")
                    if any(homepage_counts.get(section, 0) >= homepage_limit for section in sections):
                        continue
                    for section in sections:
                        homepage_counts[section] = homepage_counts.get(section, 0) + 1
                kept.append(row)
            rows = kept
        result["truncated"] = len(rows) > max_urls
        stale_ids = set(stale_sections) | {card["section_id"] for card in result["cards"]
                                           if card["stale_missing_file"]}
        for row in rows[:max_urls]:
            row["section_ids"] = sorted(row.pop("sections").split(","))
            row["matched_terms"] = url_terms.get(row["url"], [])
            row["url_kind"] = url_kind(row["url"])
            row["stale"] = bool(row["section_ids"]) and all(value in stale_ids for value in row["section_ids"])
            row["source_instances"] = self.connection.execute(
                "SELECT count(*) FROM items WHERE source_id=? AND item_type='bookmark' AND url=?",
                (source_id, row["url"])).fetchone()[0]
            result["urls"].append(row)
        # Defensive invariant: the importer refuses a bookmark without a URL, so
        # a matching card always yields at least one URL.
        if not result["urls"]:
            result["errors"].append("Matched cards contain no bookmark matching the question terms")
        return result

    def _match(self, target):
        """Literal substring matching: every whitespace-separated term must occur.

        Terms of three or more characters use the trigram full-text index, which
        is substring-based and case-insensitive. Shorter terms (and every term
        when trigram is unavailable) keep the LIKE fallback, because a two
        character CJK term cannot be answered by a trigram index.
        """
        terms = target.split() or [target]
        clauses, parameters = [], []
        for term in terms:
            if self._fts_mode == "trigram" and len(term) >= TRIGRAM_MINIMUM:
                clauses.append("i.pk IN (SELECT rowid FROM items_fts WHERE items_fts MATCH ?)")
                parameters.append('"' + term.replace('"', '""') + '"')
                continue
            # Escape all SQL wildcard characters; FTS query syntax is not exposed.
            escaped = term.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
            fields = ("title", "url", "note", "tag_text", "folder_path")
            clauses.append("(" + " OR ".join("i.%s LIKE ? ESCAPE '\\'" % field for field in fields) + ")")
            parameters.extend(["%" + escaped + "%"] * len(fields))
        return "(" + " AND ".join(clauses) + ")", parameters

    def _section_anchors(self, source_id):
        """Map a canonical section to the copy-anchor sections that share its tree."""
        anchors = {}
        for row in self.connection.execute(
                "SELECT section_id, canonical_id, is_anchor FROM sections WHERE source_id=?", (source_id,)):
            if row["is_anchor"] and row["canonical_id"] and row["canonical_id"] != row["section_id"]:
                anchors.setdefault(row["canonical_id"], []).append(row["section_id"])
        return anchors

    def _item_result(self, row, anchors=None, section_cache=None):
        keys = row.keys()
        result = {key: row[key] for key in ("source_id", "section_id", "item_id", "parent_id", "item_type", "title",
                                            "url", "note", "note_color", "position")
                  + (("added_at", "added_at_source", "added_at_confidence") if "added_at" in keys else ())}
        result["added_at_precision"] = ADDED_AT_SOURCES.get(result.get("added_at_source"), (None,))[0]
        result.update(source=row["source_id"], section=row["section_id"], item=row["item_id"],
            path=json.loads(row["path_json"]), folder_path=row["folder_path"], tags=json.loads(row["tags_json"]),
            raw_json=json.loads(row["raw_json"]), metadata=json.loads(row["metadata_json"]))
        # An untitled parentless folder is the tree container a canvas package
        # synthesizes, not a folder the user created. Mark it so counts and
        # listings can tell them apart.
        if row["item_type"] == "folder" and not row["title"] and not row["parent_id"]:
            result["synthetic"] = True
        if anchors:
            # Which copy cards display this bookmark; the query just answers one
            # of them, so "which cards contain it" would otherwise be incomplete.
            shown = anchors.get(row["section_id"])
            if shown:
                result["shown_in_anchors"] = sorted(shown)
        key = row["source_id"], row["section_id"]
        cache = {} if section_cache is None else section_cache
        if key not in cache:
            cache[key] = self.connection.execute("SELECT label,file_path FROM sections WHERE source_id=? AND section_id=?", key).fetchone()
        section = cache[key]
        result["section_label"], result["file_path"] = section[0], section[1]
        return result

    def search(self, source_id, targets=None, section=None, group_id=None, folder_id=None, tags=None,
               limit=20, offset=0, tag_colors=None, item_types=None, count_only=False, compact=False):
        """Literal structured search. ``limit=0`` or ``count_only`` returns counts only."""
        if isinstance(limit, bool) or not isinstance(limit, int) or not 0 <= limit <= 1000:
            raise ValueError("limit must be an integer from 0 to 1000; 0 returns counts only")
        if isinstance(offset, bool) or not isinstance(offset, int) or offset < 0:
            raise ValueError("offset must be a non-negative integer")
        if type(compact) is not bool:
            raise ValueError("compact must be a boolean")
        if isinstance(targets, str) or isinstance(tags, str) or isinstance(tag_colors, str):
            raise ValueError("targets, tags and tag_colors must be lists of strings")
        targets = list(dict.fromkeys(targets or []))
        if len(targets) > 100:
            raise ValueError("At most 100 targets per request")
        for target in targets:
            _string(target, "target", True)
        if item_types is not None:
            if (not isinstance(item_types, list) or not 1 <= len(item_types) <= 2
                    or any(value not in ("bookmark", "folder") for value in item_types)
                    or len(set(item_types)) != len(item_types)):
                raise ValueError("item_types must select bookmark and/or folder without duplicates")
        clauses, args = self._scope(source_id, section, group_id, folder_id, tags, tag_colors, item_types)
        page_size = 0 if count_only else limit
        anchors_by_section = self._section_anchors(source_id)
        section_cache = {}

        def query(chosen, page=True):
            predicates, parameters = list(clauses), list(args)
            if chosen:
                matches = [self._match(target) for target in chosen]
                predicates.append("(" + " OR ".join(match[0] for match in matches) + ")")
                parameters.extend(value for _, values in matches for value in values)
            where = " AND ".join(predicates)
            rows, total = [], None
            if page and page_size:
                if self._window_functions:
                    # One statement returns the page and its pre-LIMIT count, so a
                    # batch of targets does not pay a second scan per target.
                    rows = self.connection.execute(
                        "SELECT i.*, count(*) OVER () AS computed_total FROM items i WHERE " + where +
                        " ORDER BY i.section_id,i.preorder,i.item_id LIMIT ? OFFSET ?",
                        parameters + [page_size, offset]).fetchall()
                    total = rows[0]["computed_total"] if rows else self.connection.execute(
                        "SELECT count(*) FROM items i WHERE " + where, parameters).fetchone()[0]
                else:
                    total = self.connection.execute("SELECT count(*) FROM items i WHERE " + where, parameters).fetchone()[0]
                    rows = self.connection.execute(
                        "SELECT i.* FROM items i WHERE " + where + " ORDER BY i.section_id,i.preorder,i.item_id LIMIT ? OFFSET ?",
                        parameters + [page_size, offset]).fetchall()
            else:
                total = self.connection.execute("SELECT count(*) FROM items i WHERE " + where, parameters).fetchone()[0]
            unique = self.connection.execute(
                "SELECT count(DISTINCT i.url) FROM items i WHERE " + where + " AND i.url<>''", parameters).fetchone()[0]
            return {"total": total, "unique_urls": unique, "limit": limit, "offset": offset,
                    "results": [self._item_result(row, anchors_by_section, section_cache) for row in rows]}

        result = query(targets)
        section_rows = self._sections(source_id, section) if section is not None else []
        anchors = {row["section_id"]: row["canonical_id"] for row in section_rows if row["is_anchor"]}
        result.update(source_id=source_id,
            query_scope={"section": section, "group_id": group_id, "folder_id": folder_id,
                         "tags": list(tags or []), "tag_colors": list(tag_colors or []),
                         "item_types": list(item_types) if item_types else ["bookmark"]},
            match_mode="literal_substring", match_index="fts_trigram+like" if self._fts_mode == "trigram" else "like",
            searched_fields=["title", "url", "note", "tags.text", "folder_path"])
        result["counts"] = {"instances": result["total"], "unique_urls": result["unique_urls"],
            "note": "instances counts rows of query_scope.item_types, including synthetic folders when selected; unique_urls counts distinct non-empty "
                    "URLs. Per-section totals must not be summed when a copy anchor shares another card's tree."}
        if anchors:
            result["anchor_of"] = anchors
            result["anchor_note"] = ("The requested card is a copy anchor of another card; this result is the shared "
                                     "tree, so it already appears under that card's own section.")
        canvas_matches = self._canvas_matches(source_id, targets)
        if canvas_matches:
            result["canvas_matches"] = canvas_matches
            result["canvas_match_note"] = ("Canvas text cards, group labels and edge labels matching a target. They are "
                                           "not bookmark items, so they are excluded from item totals and result pages; "
                                           "use get_context for the surrounding layout.")
        shown = offset + len(result["results"]) if result["results"] else offset
        result["next_offset"] = None if page_size == 0 else (shown if shown < result["total"] else None)
        # A single-target union is identical to its per-target page.
        result["targets"] = [dict(target=target, **({key: result[key] for key in
            ("total", "unique_urls", "limit", "offset", "results")} if len(targets) == 1 else query([target])))
            for target in targets]
        if compact:
            from bookmark_output import compact_search
            return compact_search(result)
        return result

    def _canvas_matches(self, source_id, targets):
        """Find canvas text nodes, group labels and edge labels for these targets.

        These are not bookmark items, so they never enter ``results`` or
        ``total``. They are reported separately because a user asking about
        "the card labelled 123" otherwise gets unrelated short-substring
        bookmark hits and no sign that 123 is a group or edge label.
        """
        if not targets:
            return []
        patterns = []
        for target in targets:
            escaped = target.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
            patterns.append("%" + escaped + "%")
        # One query per canvas table for the whole batch, not two per target.
        clause = " OR ".join("label LIKE ? ESCAPE '\\' OR text LIKE ? ESCAPE '\\'" for _ in patterns)
        found = []
        for row in self.connection.execute(
                "SELECT canvas_path,node_id,node_type,label,text FROM nodes WHERE source_id=? AND (" + clause + ") "
                "ORDER BY canvas_path,position", (source_id, *[value for pattern in patterns for value in (pattern, pattern)])):
            found.append({"kind": "text" if row["node_type"] == "text" else "group",
                          "canvas_path": row["canvas_path"], "node_id": row["node_id"],
                          "label": row["label"], "text": row["text"], "node_type": row["node_type"]})
        clause = " OR ".join("label LIKE ? ESCAPE '\\'" for _ in patterns)
        for row in self.connection.execute(
                "SELECT canvas_path,edge_id,from_node,to_node,label FROM edges WHERE source_id=? AND (" + clause + ") "
                "ORDER BY canvas_path,edge_id", (source_id, *patterns)):
            found.append({"kind": "edge", "canvas_path": row["canvas_path"], "edge_id": row["edge_id"],
                          "label": row["label"], "from_node": row["from_node"], "to_node": row["to_node"]})
        # Attribute each canvas hit to every target it answers.
        matches = []
        for row in found:
            haystack = ((row.get("label") or "") + "\n" + (row.get("text") or "")).casefold()
            for target in targets:
                needle = target.casefold()
                if needle and needle in haystack:
                    matches.append({"target": target, **row})
        return matches

    def context(self, source_id, section=None, group_id=None, item_id=None):
        source = self._source(source_id)
        sections = self._sections(source_id, section)
        nodes = [dict(row) for row in self.connection.execute("SELECT * FROM nodes WHERE source_id=? ORDER BY canvas_path,position", (source_id,))]
        all_nodes = list(nodes)
        memberships = [dict(row) for row in self.connection.execute("SELECT * FROM memberships WHERE source_id=? ORDER BY canvas_path,group_id,node_id", (source_id,))]
        edges = [dict(row) for row in self.connection.execute("SELECT * FROM edges WHERE source_id=? ORDER BY canvas_path,edge_id", (source_id,))]
        item_rows = []
        if item_id is not None:
            canonical_ids = {row["canonical_id"] for row in sections}
            item_rows = [row for row in self.connection.execute("SELECT * FROM items WHERE source_id=? AND item_id=?", (source_id, item_id)) if row["section_id"] in canonical_ids]
            if not item_rows:
                raise ValueError("Unknown item in requested scope: " + str(item_id))
            ids = {row["section_id"] for row in item_rows}
            sections = [row for row in sections if row["canonical_id"] in ids]
        if group_id is not None:
            group = self._group(source_id, group_id)
            member_ids = {m["node_id"] for m in memberships if m["canvas_path"] == group["canvas_path"] and m["group_id"] == group["node_id"]}
            selected = [node for node in nodes if node["canvas_path"] == group["canvas_path"] and node["node_id"] in member_ids | {group["node_id"]}]
            section_ids = {node["section_id"] for node in selected if node["section_id"]}
            sections = [row for row in sections if row["section_id"] in section_ids]
            if item_id is not None:
                canonical_ids = {row["canonical_id"] for row in sections}
                item_rows = [row for row in item_rows if row["section_id"] in canonical_ids]
                if not item_rows:
                    raise ValueError("Item is outside the requested group")
            if section is not None or item_id is not None:
                section_ids = {row["section_id"] for row in sections}
                selected = [node for node in selected
                    if node["section_id"] in section_ids or node["node_id"] == group["node_id"]]
            nodes = selected
        elif section is not None or item_id is not None:
            section_ids = {row["section_id"] for row in sections}
            selected_ids = {(node["canvas_path"], node["node_id"]) for node in nodes if node["section_id"] in section_ids}
            selected_ids.update((m["canvas_path"], m["group_id"]) for m in memberships if (m["canvas_path"], m["node_id"]) in selected_ids)
            nodes = [node for node in nodes if (node["canvas_path"], node["node_id"]) in selected_ids]
        if section is not None or group_id is not None or item_id is not None:
            selected_ids = {(node["canvas_path"], node["node_id"]) for node in nodes}
            edges = [edge for edge in edges if (edge["canvas_path"], edge["from_node"]) in selected_ids or (edge["canvas_path"], edge["to_node"]) in selected_ids]
            memberships = [m for m in memberships if (m["canvas_path"], m["group_id"]) in selected_ids or (m["canvas_path"], m["node_id"]) in selected_ids]
        for edge in edges:
            edge["direction"] = ("both" if edge["from_end"] == edge["to_end"] == "arrow" else
                "forward" if edge["to_end"] == "arrow" else "reverse" if edge["from_end"] == "arrow" else "none")
            edge["raw_json"] = json.loads(edge["raw_json"])
            edge["added_at"] = _added_at_from_edge_id(edge["edge_id"])
            edge["added_at_source"] = "edge_epoch" if edge["added_at"] else None
            edge["added_at_confidence"] = ADDED_AT_SOURCES["edge_epoch"][1] if edge["added_at"] else None
        # Chronological inside each canvas; an edge whose id carries no time
        # stays in stable id order at the end instead of being invented a date.
        edges.sort(key=lambda edge: (edge["canvas_path"], edge["added_at"] is None,
                                     edge["added_at"] or "", edge["edge_id"]))
        for node in nodes:
            node["raw_json"] = json.loads(node["raw_json"])
        selected_node_ids = {(node["canvas_path"], node["node_id"]) for node in nodes}
        endpoint_ids = {(edge["canvas_path"], edge[end]) for edge in edges for end in ("from_node", "to_node")}
        endpoint_ids.update((member["canvas_path"], member[end])
            for member in memberships for end in ("group_id", "node_id"))
        related_nodes = [node for node in all_nodes
            if (node["canvas_path"], node["node_id"]) in endpoint_ids - selected_node_ids]
        for node in related_nodes:
            node["raw_json"] = json.loads(node["raw_json"])
        section_results = []
        for row in sections:
            entry = dict(row)
            # Context is bounded to section headers. The entire source object,
            # including all unknown fields, remains in the database raw_json.
            entry["raw_json"] = {key: value for key, value in json.loads(entry["raw_json"]).items()
                if key not in ("tree", "items", "identityMap")}
            entry["bookmark_count"] = self.connection.execute("SELECT count(*) FROM items WHERE source_id=? AND section_id=? AND item_type='bookmark'", (source_id, entry["canonical_id"])).fetchone()[0]
            # A copy anchor reports the shared tree's count, so summing section
            # counts double counts. Say so explicitly instead of leaving a
            # self-consistent but inflated total.
            entry["anchor_of"] = entry["canonical_id"] if entry["is_anchor"] else None
            entry["counted_in_totals"] = not bool(entry["is_anchor"])
            section_results.append(entry)
        items = [self._item_result(row) for row in item_rows]
        for item, row in zip(items, item_rows):
            item["ancestors"] = []
            for ancestor in json.loads(row["ancestor_ids_json"]):
                ancestor_row = self.connection.execute("SELECT * FROM items WHERE source_id=? AND section_id=? AND item_id=?", (source_id, row["section_id"], ancestor)).fetchone()
                if ancestor_row:
                    item["ancestors"].append(self._item_result(ancestor_row))
        return {"source_id": source_id, "source": source, "sections": section_results,
            "counts": self._scope_counts(source_id, section_results),
            "items": items, "nodes": nodes, "groups": [node for node in nodes if node["node_type"] == "group"],
            "edges": edges, "related_nodes": related_nodes, "memberships": memberships}

    def _scope_counts(self, source_id, section_results):
        """Distinct totals for the returned sections, so copy anchors are not summed twice."""
        counted = [row for row in section_results if row["counted_in_totals"]]
        instances, urls = 0, set()
        for row in counted:
            instances += row["bookmark_count"]
            urls.update(value for value in self.connection.execute(
                "SELECT DISTINCT url FROM items WHERE source_id=? AND section_id=? AND item_type='bookmark' AND url<>''",
                (source_id, row["canonical_id"])))
        return {"sections": len(section_results), "counted_sections": len(counted),
                "instances": instances, "unique_urls": len(urls),
                "note": "Section totals, not item or folder subtree totals. Sum of counted_sections only; copy anchors share another card's tree and are excluded."}

    def inventory(self, source_ids):
        """Freeze the complete indexed input, independently of query previews.

        Exact URLs identify research rows; bookmarks remain distinct instances.
        Copy anchors add appearances of a canonical tree, never extra instances.
        One SQLite read transaction keeps files, items and canvas links coherent
        when another client synchronizes the index at the same time.
        """
        if not isinstance(source_ids, list) or not source_ids:
            raise ValueError("source_ids must contain indexed package IDs")
        identifiers = sorted(set(_string(value, "source_id", True) for value in source_ids))
        from archive import SourceArchive
        result = {"schema_version": 1, "index_state": "last_synchronized",
                  "source_ids": identifiers, "sources": [], "files": [],
                  "sections": [], "folders": [], "nodes": [], "edges": [],
                  "memberships": [], "entries": []}
        by_url = {}
        section_cache = {}
        self.connection.execute("SAVEPOINT inventory_snapshot")
        try:
            for source_id in identifiers:
                context = self.context(source_id)
                files = [dict(row) for row in self.connection.execute(
                    "SELECT source_id,file_path,kind,sha256,revision FROM files "
                    "WHERE source_id=? ORDER BY file_path", (source_id,))]
                version_files = [{key: row[key] for key in ("file_path", "kind", "sha256")} for row in files]
                result["sources"].append({**context["source"], "input_version":
                    hashlib.sha256(_dump(version_files).encode("utf-8")).hexdigest()})
                result["files"].extend(files)
                for collection in ("sections", "nodes", "edges", "memberships"):
                    result[collection].extend(context[collection])
                appearances = {}
                for section in context["sections"]:
                    if not section["canonical_id"]:
                        continue
                    nodes = [node for node in context["nodes"] if node["section_id"] == section["section_id"]]
                    node_ids = {(node["canvas_path"], node["node_id"]) for node in nodes}
                    appearance = {key: section[key] for key in
                        ("section_id", "file_path", "label", "title", "description_md", "is_anchor", "canonical_id")}
                    appearance["nodes"] = [{"canvas_path": node["canvas_path"], "node_id": node["node_id"]} for node in nodes]
                    appearance["memberships"] = [member for member in context["memberships"]
                        if (member["canvas_path"], member["node_id"]) in node_ids]
                    appearance["edges"] = [edge for edge in context["edges"]
                        if any((edge["canvas_path"], edge[end]) in node_ids for end in ("from_node", "to_node"))]
                    appearances.setdefault(section["canonical_id"], []).append(appearance)
                rows = self.connection.execute("SELECT * FROM items WHERE source_id=? "
                    "ORDER BY section_id,preorder,item_id", (source_id,)).fetchall()
                for row in rows:
                    item = self._item_result(row, section_cache=section_cache)
                    item.update(ancestor_ids=json.loads(row["ancestor_ids_json"]),
                                revision=row["revision"], preorder=row["preorder"])
                    if row["item_type"] == "folder":
                        # Descendants are preserved as individual rows; avoid
                        # duplicating an entire nested tree at each ancestor.
                        item["raw_json"] = {key: value for key, value in item["raw_json"].items() if key != "children"}
                        result["folders"].append(item)
                        continue
                    original = row["url"]
                    if original not in by_url:
                        try:
                            retrieval = SourceArchive._key(original)
                        except (ValueError, UnicodeError):
                            retrieval = None
                        try:
                            scheme = urlsplit(original).scheme
                        except ValueError:
                            scheme = ""
                        by_url[original] = {"id": "u-" + hashlib.sha256(original.encode("utf-8")).hexdigest()[:32],
                            "original_url": original, "retrieval_url": retrieval,
                            "url_kind": "web" if retrieval else "local" if scheme == "file" else "unsupported",
                            "instances": []}
                    identity = [source_id, row["section_id"], row["item_id"]]
                    item.update(instance_id="b-" + hashlib.sha256(_dump(identity).encode("utf-8")).hexdigest()[:32],
                                canonical_section_id=row["section_id"],
                                appearances=appearances.get(row["section_id"], []))
                    by_url[original]["instances"].append(item)
            result["entries"] = sorted(by_url.values(), key=lambda row: row["id"])
            version = [{key: row[key] for key in ("source_id", "file_path", "kind", "sha256")}
                       for row in result["files"]]
            result["input_version"] = hashlib.sha256(_dump(version).encode("utf-8")).hexdigest()
            result["counts"] = {"unique_urls": len(result["entries"]),
                "bookmark_instances": sum(len(row["instances"]) for row in result["entries"]),
                **{key: len(result[key]) for key in ("sources", "files", "sections", "folders", "nodes", "edges", "memberships")}}
            return result
        finally:
            self.connection.execute("RELEASE SAVEPOINT inventory_snapshot")

    def input_version(self, source_ids):
        """Fingerprint the indexed input without rebuilding the full inventory."""
        records = []
        self.connection.execute("SAVEPOINT input_version_snapshot")
        try:
            for source_id in sorted(set(source_ids)):
                self._source(source_id)
                records.extend(dict(row) for row in self.connection.execute(
                    "SELECT source_id,file_path,kind,sha256 FROM files WHERE source_id=? ORDER BY file_path", (source_id,)))
            return hashlib.sha256(_dump(records).encode("utf-8")).hexdigest()
        finally:
            self.connection.execute("RELEASE SAVEPOINT input_version_snapshot")

    def status(self, source_id=None):
        if source_id is None:
            ids = [row[0] for row in self.connection.execute("SELECT source_id FROM sources ORDER BY source_id")]
            return {"database": str(self.db_path) if self.db_path else ":memory:", "schema_version": 1,
                "sources": [self.status(identifier) for identifier in ids]}
        result = self._source(source_id)
        counts = dict(self.connection.execute("""SELECT count(*) AS items,
            coalesce(sum(item_type='bookmark'),0) AS bookmarks,
            coalesce(sum(item_type='folder'),0) AS folders,
            coalesce(sum(item_type='folder' AND parent_id IS NULL AND title=''),0) AS tree_roots,
            count(DISTINCT CASE WHEN item_type='bookmark' THEN url END) AS unique_urls
            FROM items WHERE source_id=?""", (source_id,)).fetchone())
        # "folders" is the raw row count, which includes the synthetic tree
        # container; user_folders is what a person sees in the canvas.
        counts["user_folders"] = counts["folders"] - counts["tree_roots"]
        for table in ("sections", "files", "nodes", "edges", "memberships"):
            counts[table] = self.connection.execute("SELECT count(*) FROM " + table + " WHERE source_id=?", (source_id,)).fetchone()[0]
        result["counts"] = counts
        result.update(bookmarks=counts["bookmarks"], unique_urls=counts["unique_urls"])
        package = Path(result["package_path"])
        result["package_exists"] = package.is_dir()
        result["retained_missing_files"] = [row[0] for row in self.connection.execute("SELECT file_path FROM files WHERE source_id=? ORDER BY file_path", (source_id,)) if not (package / row[0]).is_file()]
        if result["retained_missing_files"]:
            # partial imports keep removed cards on purpose; say so at the top
            # level so the stale rows are not mistaken for current content.
            result["retained_missing_note"] = ("%d imported file(s) are no longer in the package and were kept because "
                "completeness=partial. Re-import with completeness=complete to reconcile the full mirror, or ignore "
                "them deliberately." % len(result["retained_missing_files"]))
        result["sections"] = [{"section_id": row["section_id"], "label": row["label"], "is_anchor": bool(row["is_anchor"]), "canonical_id": row["canonical_id"]} for row in self._sections(source_id)]
        return result