"""Immutable repository snapshots and dependency-aware knowledge claims.

The database has one writer: a :class:`KnowledgeStore` instance bound to the
thread that created it. SQLite serializes that writer with ``BEGIN IMMEDIATE``
transactions and WAL lets independent readers continue while it commits.
"""

from __future__ import annotations

import fnmatch
import hashlib
import json
import os
import re
import shutil
import signal
import sqlite3
import stat
import subprocess
import tempfile
import threading
import uuid
import zlib
from contextlib import contextmanager
from dataclasses import dataclass
from datetime import datetime, timezone
from enum import Enum
from pathlib import Path, PurePosixPath, PureWindowsPath
from typing import Callable, Iterator, Mapping, Sequence

from .ignore import ALWAYS_SKIP, IgnoreStack


SCHEMA_VERSION = 1
MANIFEST_VERSION = 1
DEFAULT_ANALYZER_VERSION = "byte-repository-intelligence/1"
DATABASE_NAME = "repository-intelligence.sqlite3"
MAX_QUERYABLE_BYTES = 2 * 1024 * 1024
MAX_SNAPSHOT_FILE_BYTES = 64 * 1024 * 1024
MAX_SNAPSHOT_TOTAL_BYTES = 256 * 1024 * 1024
SNAPSHOT_READ_CHUNK_BYTES = 1024 * 1024
MAX_SNAPSHOT_FILES = 100_000
MAX_SNAPSHOT_TREE_ENTRIES = 200_000
MAX_GIT_PATH_OUTPUT_BYTES = 32 * 1024 * 1024
MAX_IGNORE_FILE_BYTES = 1024 * 1024
MAX_KNOWLEDGE_STORE_BYTES = 2 * 1024 * 1024 * 1024
SNAPSHOT_INDEX_ESTIMATE_BYTES = 256
TYPESCRIPT_WORKER_TIMEOUT_SECONDS = 5
MAX_TYPESCRIPT_WORKER_OUTPUT_BYTES = 4 * 1024 * 1024
MAX_TYPESCRIPT_COMPILER_INPUT_BYTES = 64 * 1024 * 1024
MAX_TYPESCRIPT_COMPILER_INPUT_FILES = 10_000
MAX_TYPESCRIPT_QUERY_IDENTIFIERS = 64
MAX_TYPESCRIPT_IDENTIFIER_CHARS = 256
TYPESCRIPT_WORKER_MAX_OLD_SPACE_MIB = 384
TYPESCRIPT_ADAPTER_VERSION = "3"
# Build output names are only traversal hints for the structural graph. They
# cannot be unconditional snapshot exclusions: repositories sometimes track
# authored JS/TS under these directories and explicitly include it from a
# tsconfig. Snapshot file/byte/tree budgets remain the hard resource boundary.
_SNAPSHOT_ALWAYS_SKIP = ALWAYS_SKIP.difference({"build", "dist"})
_SQLITE_MANAGED_SUFFIXES = ("", "-wal", "-shm", "-journal")
_SQLITE_WAL_HEADER_BYTES = 32
_SQLITE_WAL_FRAME_HEADER_BYTES = 24
_SQLITE_WAL_INDEX_REGION_BYTES = 32 * 1024
_SQLITE_WAL_INDEX_FRAMES_PER_REGION = 4096
_MAX_WORKSPACE_PATTERNS = 512
_MAX_WORKSPACE_PATTERN_CHARS = 2048
_MAX_WORKSPACE_BRACE_EXPANSIONS = 512

_TYPESCRIPT_COMPILER_SUFFIXES = frozenset(
    {".cjs", ".cts", ".js", ".json", ".jsonc", ".jsx", ".mjs", ".mts", ".ts", ".tsx"}
)
_TYPESCRIPT_COMPILER_LOCK_NAMES = frozenset(
    {"bun.lock", "bun.lockb", "pnpm-lock.yaml", "yarn.lock"}
)

_CONFIG_NAMES = frozenset(
    {
        ".editorconfig",
        ".npmrc",
        "biome.json",
        "biome.jsonc",
        "cargo.toml",
        "composer.json",
        "deno.json",
        "deno.jsonc",
        "go.mod",
        "go.work",
        "package.json",
        "pnpm-workspace.yaml",
        "pyproject.toml",
        "setup.cfg",
        "tsconfig.json",
    }
)
_LOCK_NAMES = frozenset(
    {
        "bun.lock",
        "bun.lockb",
        "cargo.lock",
        "composer.lock",
        "deno.lock",
        "go.sum",
        "package-lock.json",
        "pnpm-lock.yaml",
        "poetry.lock",
        "uv.lock",
        "yarn.lock",
    }
)


class ClaimStatus(str, Enum):
    SUPPORTED = "supported"
    REFUTED = "refuted"
    UNRESOLVED = "unresolved"


class Applicability(str, Enum):
    CURRENT = "current"
    NEEDS_RECHECK = "needs_recheck"
    UNKNOWN = "unknown"


class SnapshotLimitError(RuntimeError):
    """Raised before persistence when workspace content exceeds fixed snapshot limits."""


class KnowledgeStoreLimitError(SnapshotLimitError):
    """Raised when a SQLite write would exceed the fixed knowledge-store budget."""


class _WorkerOutputLimit(RuntimeError):
    """Raised after terminating a semantic worker that exceeded its output budget."""


class DependencyType(str, Enum):
    EXACT_BLOB = "exact_blob"
    QUERY_SET = "query_set"
    NEGATIVE_QUERY = "negative_query"


@dataclass(frozen=True)
class SnapshotFile:
    path: str
    digest: str
    size: int
    kind: str
    queryable: bool


@dataclass(frozen=True)
class SnapshotManifest:
    analyzer_version: str
    files: tuple[SnapshotFile, ...]
    config_hashes: tuple[tuple[str, str], ...]
    lock_hashes: tuple[tuple[str, str], ...]

    def as_dict(self) -> dict[str, object]:
        return {
            "manifest_version": MANIFEST_VERSION,
            "analyzer_version": self.analyzer_version,
            "files": [
                {
                    "path": entry.path,
                    "digest": entry.digest,
                    "size": entry.size,
                    "kind": entry.kind,
                    "queryable": entry.queryable,
                }
                for entry in self.files
            ],
            "config_hashes": [list(entry) for entry in self.config_hashes],
            "lock_hashes": [list(entry) for entry in self.lock_hashes],
        }


@dataclass(frozen=True)
class Snapshot:
    id: str
    root: str
    created_at: str
    manifest: SnapshotManifest

    @property
    def files(self) -> tuple[SnapshotFile, ...]:
        return self.manifest.files


@dataclass(frozen=True)
class EvidenceInput:
    kind: str
    payload: Mapping[str, object]


@dataclass(frozen=True)
class Evidence:
    id: str
    claim_id: str
    snapshot_id: str
    kind: str
    payload: Mapping[str, object]
    digest: str
    created_at: str


@dataclass(frozen=True)
class ExactBlobDependency:
    path: str


@dataclass(frozen=True)
class QuerySetDependency:
    query_kind: str
    parameters: Mapping[str, object]
    domain: tuple[str, ...] = ("**/*",)


@dataclass(frozen=True)
class NegativeQueryDependency:
    query_kind: str
    parameters: Mapping[str, object]
    domain: tuple[str, ...] = ("**/*",)


@dataclass(frozen=True)
class PrecomputedQueryDependency:
    query_kind: str
    parameters: Mapping[str, object]
    evaluation: QueryEvaluation
    domain: tuple[str, ...] = ("**/*",)
    negative: bool = False


DependencyInput = (
    ExactBlobDependency
    | QuerySetDependency
    | NegativeQueryDependency
    | PrecomputedQueryDependency
)


@dataclass(frozen=True)
class Dependency:
    id: str
    created_snapshot_id: str
    type: DependencyType
    path: str | None
    expected_digest: str | None
    query_kind: str | None
    parameters: Mapping[str, object]
    domain: tuple[str, ...]
    domain_fingerprint: str | None
    result_fingerprint: str | None
    results: tuple[str, ...]
    complete: bool


@dataclass(frozen=True)
class Claim:
    id: str
    statement: str
    status: ClaimStatus
    applicability: Applicability
    created_snapshot_id: str
    current_snapshot_id: str
    scope: Mapping[str, object]
    created_at: str
    updated_at: str
    evidence: tuple[Evidence, ...]
    dependencies: tuple[Dependency, ...]


@dataclass(frozen=True)
class QueryEvaluation:
    domain_fingerprint: str
    result_fingerprint: str
    results: tuple[str, ...]
    complete: bool


@dataclass(frozen=True)
class DependencyCheck:
    dependency_id: str
    type: DependencyType
    state: str
    reason: str


@dataclass(frozen=True)
class RecheckResult:
    check_id: str
    claim_id: str
    target_snapshot_id: str
    status: ClaimStatus
    applicability: Applicability
    dependencies: tuple[DependencyCheck, ...]


@dataclass(frozen=True)
class KnowledgeEvent:
    cursor: int
    kind: str
    claim_id: str | None
    snapshot_id: str | None
    payload: Mapping[str, object]
    created_at: str


@dataclass(frozen=True)
class _CapturedSnapshot:
    id: str
    manifest: SnapshotManifest
    blobs: Mapping[str, bytes]


QueryEvaluator = Callable[["KnowledgeStore", str, Mapping[str, object], tuple[str, ...]], QueryEvaluation]


class KnowledgeStore:
    """Transactional owner of snapshots, claims, evidence, and dependencies."""

    def __init__(
        self,
        root: str | os.PathLike[str],
        *,
        analyzer_version: str = DEFAULT_ANALYZER_VERSION,
        query_evaluators: Mapping[str, QueryEvaluator] | None = None,
    ) -> None:
        self.root = Path(root).expanduser().resolve()
        if not self.root.is_dir():
            raise NotADirectoryError(self.root)
        self.analyzer_version = str(analyzer_version)
        self._owner_thread = threading.get_ident()
        self._closed = False
        self._snapshot_cache: dict[str, Snapshot] = {}
        self._snapshot_entry_cache: dict[str, dict[str, SnapshotFile]] = {}
        self._evaluators: dict[str, QueryEvaluator] = {
            "byte.typescript.consumers": _evaluate_typescript_consumers,
            "text_contains": _evaluate_text_contains,
        }
        if query_evaluators:
            self._evaluators.update(query_evaluators)
        self.path = _validated_store_database(
            self.root, create_store_directory=True, require_database=False
        )
        if os.name != "nt":
            _create_private_store_database(self.path)
            _validated_store_database(
                self.root, create_store_directory=False, require_database=True
            )
        existing_managed_paths = _existing_managed_store_paths(self.path)
        self._connection = sqlite3.connect(self.path, isolation_level=None)
        try:
            _validated_store_database(
                self.root, create_store_directory=False, require_database=True
            )
            self._connection.row_factory = sqlite3.Row
            self._connection.execute("PRAGMA foreign_keys = ON")
            self._connection.execute("PRAGMA busy_timeout = 5000")
            _enforce_store_budget(self._connection, self.path)
            journal_mode = self._connection.execute("PRAGMA journal_mode = WAL").fetchone()[0]
            if str(journal_mode).lower() != "wal":
                raise RuntimeError("repository intelligence store requires SQLite WAL mode")
            _enforce_store_budget(self._connection, self.path)
            _tighten_new_managed_store_files(self.path, existing_managed_paths)
            _validated_store_database(
                self.root, create_store_directory=False, require_database=True
            )
            self._connection.execute("PRAGMA synchronous = NORMAL")
            self._initialize_schema()
            _tighten_new_managed_store_files(self.path, existing_managed_paths)
            _validated_store_database(
                self.root, create_store_directory=False, require_database=True
            )
        except BaseException:
            self._connection.close()
            self._closed = True
            raise

    def __enter__(self) -> "KnowledgeStore":
        self._assert_owner()
        return self

    def __exit__(self, exc_type: object, exc: object, traceback: object) -> None:
        self.close()

    def close(self) -> None:
        if threading.get_ident() != self._owner_thread:
            raise RuntimeError("KnowledgeStore may only be used by its owning thread")
        if self._closed:
            return
        self._connection.close()
        self._closed = True

    @property
    def journal_mode(self) -> str:
        self._assert_owner()
        return str(self._connection.execute("PRAGMA journal_mode").fetchone()[0]).lower()

    def capture_snapshot(self) -> Snapshot:
        """Capture visible workspace content and persist an immutable manifest."""
        self._assert_owner()
        captured = _capture_workspace(self.root, self.analyzer_version)
        now = _now()
        manifest_json = _canonical_json(captured.manifest.as_dict())
        with self._write_transaction():
            existing = self._connection.execute(
                "SELECT manifest_json FROM snapshots WHERE id = ?", (captured.id,)
            ).fetchone()
            if existing is not None and existing["manifest_json"] != manifest_json:
                raise RuntimeError("snapshot digest collision")
            if existing is None:
                added_bytes = (
                    _new_snapshot_blob_bytes(self._connection, captured)
                    + len(manifest_json.encode("utf-8"))
                    + len(captured.manifest.files) * SNAPSHOT_INDEX_ESTIMATE_BYTES
                )
                if _managed_store_bytes(self.path) + added_bytes > MAX_KNOWLEDGE_STORE_BYTES:
                    raise _store_budget_error()
            self._connection.execute(
                """
                INSERT OR IGNORE INTO snapshots(id, root, analyzer_version, manifest_json, created_at)
                VALUES (?, ?, ?, ?, ?)
                """,
                (captured.id, str(self.root), self.analyzer_version, manifest_json, now),
            )
            for entry in captured.manifest.files:
                blob = captured.blobs[entry.digest]
                blob_cursor = self._connection.execute(
                    "INSERT OR IGNORE INTO blobs(digest, encoding, content) VALUES (?, 'zlib', ?)",
                    (entry.digest, sqlite3.Binary(zlib.compress(blob))),
                )
                if blob_cursor.rowcount == 0:
                    stored = self._connection.execute(
                        "SELECT encoding, content FROM blobs WHERE digest = ?", (entry.digest,)
                    ).fetchone()
                    if stored is None or _decode_snapshot_blob(entry, stored) != blob:
                        raise RuntimeError(f"snapshot blob digest collision: {entry.path}")
                self._connection.execute(
                    """
                    INSERT OR IGNORE INTO snapshot_files(
                        snapshot_id, path, digest, size, kind, queryable, blob_stored
                    ) VALUES (?, ?, ?, ?, ?, ?, ?)
                    """,
                    (
                        captured.id,
                        entry.path,
                        entry.digest,
                        entry.size,
                        entry.kind,
                        int(entry.queryable),
                        1,
                    ),
                )
        return self.get_snapshot(captured.id)

    def get_snapshot(self, snapshot_id: str) -> Snapshot:
        self._assert_owner()
        cached = self._snapshot_cache.get(snapshot_id)
        if cached is not None:
            return cached
        row = self._connection.execute(
            "SELECT root, manifest_json, created_at FROM snapshots WHERE id = ?", (snapshot_id,)
        ).fetchone()
        if row is None:
            raise KeyError(f"unknown snapshot: {snapshot_id}")
        manifest = _manifest_from_json(row["manifest_json"])
        if _sha256(_canonical_json(manifest.as_dict()).encode("utf-8")) != snapshot_id:
            raise RuntimeError(f"snapshot manifest integrity check failed: {snapshot_id}")
        entries = {item.path: item for item in manifest.files}
        if len(entries) != len(manifest.files):
            raise RuntimeError(f"snapshot manifest contains duplicate paths: {snapshot_id}")
        snapshot = Snapshot(
            id=snapshot_id,
            root=row["root"],
            created_at=row["created_at"],
            manifest=manifest,
        )
        self._snapshot_cache[snapshot_id] = snapshot
        self._snapshot_entry_cache[snapshot_id] = entries
        return snapshot

    def read_snapshot_file(self, snapshot_id: str, path: str) -> bytes | None:
        """Read bytes from an immutable snapshot, or None when the path is absent."""
        self._assert_owner()
        rel = _normalize_relative_path(path)
        self.get_snapshot(snapshot_id)
        entry = self._snapshot_entry_cache[snapshot_id].get(rel)
        if entry is None:
            return None
        row = self._connection.execute(
            """
            SELECT sf.digest, sf.size, sf.kind, sf.queryable, sf.blob_stored,
                   b.encoding, b.content
            FROM snapshot_files AS sf
            LEFT JOIN blobs AS b ON b.digest = sf.digest
            WHERE sf.snapshot_id = ? AND sf.path = ?
            """,
            (snapshot_id, rel),
        ).fetchone()
        if row is None:
            raise RuntimeError(f"snapshot file metadata is unavailable: {rel}")
        return _verified_snapshot_blob(entry, row)

    def materialize_snapshot(self, snapshot_id: str, destination: str | os.PathLike[str]) -> int:
        """Reconstruct a snapshot into a new or empty directory without changing the store."""
        self._assert_owner()
        snapshot = self.get_snapshot(snapshot_id)
        destination_path = Path(destination).expanduser()
        if destination_path.is_symlink():
            raise ValueError("snapshot destination must be a directory")
        if destination_path.exists():
            if not destination_path.is_dir():
                raise ValueError("snapshot destination must be a directory")
            if any(destination_path.iterdir()):
                raise ValueError("snapshot destination must be empty")
        else:
            destination_path.mkdir(parents=True)
        destination_root = destination_path.resolve()
        rows = self._connection.execute(
            """
            SELECT sf.path, sf.digest, sf.size, sf.kind, sf.queryable, sf.blob_stored,
                   b.encoding, b.content
            FROM snapshot_files AS sf
            LEFT JOIN blobs AS b ON b.digest = sf.digest
            WHERE sf.snapshot_id = ?
            ORDER BY sf.path
            """,
            (snapshot_id,),
        ).fetchall()
        expected_entries = self._snapshot_entry_cache[snapshot.id]
        if len(rows) != len(expected_entries):
            raise RuntimeError(f"snapshot file index integrity check failed: {snapshot_id}")
        regular: list[tuple[Path, bytes]] = []
        links: list[tuple[Path, str]] = []
        seen: set[str] = set()
        for row in rows:
            entry = expected_entries.get(row["path"])
            if entry is None or row["path"] in seen:
                raise RuntimeError(f"snapshot file index contains an unexpected path: {row['path']}")
            seen.add(row["path"])
            target_path = _materialization_path(destination_root, row["path"])
            content = _verified_snapshot_blob(entry, row)
            if row["kind"] == "file":
                regular.append((target_path, content))
            elif row["kind"] == "symlink":
                try:
                    link_target = content.decode("utf-8", errors="surrogateescape")
                except UnicodeError as error:
                    raise ValueError(f"invalid symlink target for {row['path']}") from error
                links.append((target_path, _safe_symlink_target(destination_root, target_path, link_target)))
            else:
                raise RuntimeError(f"unsupported snapshot entry kind: {row['kind']}")
        for target_path, content in regular:
            target_path.parent.mkdir(parents=True, exist_ok=True)
            target_path.write_bytes(content)
        for target_path, link_target in links:
            target_path.parent.mkdir(parents=True, exist_ok=True)
            resolved_target = (target_path.parent / Path(link_target)).resolve(strict=False)
            os.symlink(link_target, target_path, target_is_directory=resolved_target.is_dir())
        _materialize_workspace_package_links(destination_root, expected_entries)
        return len(rows)

    def create_claim(
        self,
        snapshot_id: str,
        statement: str,
        *,
        status: ClaimStatus | str = ClaimStatus.UNRESOLVED,
        scope: Mapping[str, object] | None = None,
        evidence: Sequence[EvidenceInput] = (),
        dependencies: Sequence[DependencyInput] = (),
    ) -> Claim:
        """Record a claim and freeze its dependency baselines at ``snapshot_id``."""
        self._assert_owner()
        self.get_snapshot(snapshot_id)
        normalized_status = ClaimStatus(status)
        claim_id = uuid.uuid4().hex
        now = _now()
        prepared = [self._prepare_dependency(snapshot_id, item) for item in dependencies]
        applicability = (
            Applicability.CURRENT if prepared and all(item.complete for item in prepared) else Applicability.UNKNOWN
        )
        scope_json = _canonical_json(dict(scope or {}))
        with self._write_transaction():
            self._connection.execute(
                """
                INSERT INTO claims(
                    id, statement, status, applicability, created_snapshot_id,
                    current_snapshot_id, scope_json, created_at, updated_at
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    claim_id,
                    str(statement),
                    normalized_status.value,
                    applicability.value,
                    snapshot_id,
                    snapshot_id,
                    scope_json,
                    now,
                    now,
                ),
            )
            for item in evidence:
                self._insert_evidence(claim_id, snapshot_id, item, now)
            for item in prepared:
                self._insert_dependency(claim_id, snapshot_id, item)
            self._insert_event(
                "claim.created",
                {"status": normalized_status.value, "applicability": applicability.value},
                claim_id=claim_id,
                snapshot_id=snapshot_id,
                created_at=now,
            )
        return self.get_claim(claim_id)

    def add_evidence(self, claim_id: str, snapshot_id: str, evidence: EvidenceInput) -> Evidence:
        """Append immutable evidence without replacing earlier observations."""
        self._assert_owner()
        self.get_claim(claim_id)
        self.get_snapshot(snapshot_id)
        evidence_id = uuid.uuid4().hex
        now = _now()
        with self._write_transaction():
            self._insert_evidence(claim_id, snapshot_id, evidence, now, evidence_id=evidence_id)
        return next(item for item in self.get_claim(claim_id).evidence if item.id == evidence_id)

    def set_claim_status(self, claim_id: str, status: ClaimStatus | str) -> Claim:
        """Set the interpreted outcome without changing evidence or applicability."""
        self._assert_owner()
        normalized_status = ClaimStatus(status)
        now = _now()
        with self._write_transaction():
            updated = self._connection.execute(
                "UPDATE claims SET status = ?, updated_at = ? WHERE id = ?",
                (normalized_status.value, now, claim_id),
            ).rowcount
            if not updated:
                raise KeyError(f"unknown claim: {claim_id}")
        return self.get_claim(claim_id)

    def list_claims(self, limit: int = 100) -> tuple[Claim, ...]:
        """List claims by most recent update, with deterministic tie-breaking."""
        self._assert_owner()
        cap = _bounded_limit(limit)
        rows = self._connection.execute(
            "SELECT id FROM claims ORDER BY updated_at DESC, id LIMIT ?", (cap,)
        ).fetchall()
        return tuple(self.get_claim(row["id"]) for row in rows)

    def iter_snapshot_claims(self, snapshot_id: str, *, page_size: int = 100) -> Iterator[Claim]:
        """Visit every claim recorded against a snapshot, independent of its latest check.

        The reader pins candidate membership while callers recheck claims through
        the writer. Keyset pages cannot skip claims when their update times move,
        and records added during this iteration belong to the next analysis.
        """
        self._assert_owner()
        self.get_snapshot(snapshot_id)
        cap = max(1, _bounded_limit(page_size))
        reader = sqlite3.connect(self.path, isolation_level=None)
        reader.row_factory = sqlite3.Row
        try:
            reader.execute("PRAGMA query_only = ON")
            reader.execute("BEGIN")
            after_id = ""
            while True:
                rows = reader.execute(
                    """
                    SELECT c.id FROM claims AS c
                    WHERE c.id > ? AND (
                        c.created_snapshot_id = ?
                        OR EXISTS (
                            SELECT 1 FROM dependencies AS d
                            WHERE d.claim_id = c.id AND d.created_snapshot_id = ?
                        )
                        OR EXISTS (
                            SELECT 1 FROM claim_checks AS checks
                            WHERE checks.claim_id = c.id AND checks.target_snapshot_id = ?
                        )
                        OR EXISTS (
                            SELECT 1 FROM evidence AS e
                            WHERE e.claim_id = c.id AND e.snapshot_id = ?
                        )
                    )
                    ORDER BY c.id LIMIT ?
                    """,
                    (after_id, snapshot_id, snapshot_id, snapshot_id, snapshot_id, cap),
                ).fetchall()
                if not rows:
                    break
                after_id = rows[-1]["id"]
                for row in rows:
                    yield self.get_claim(row["id"])
        finally:
            reader.close()

    def find_claims(self, question: str, limit: int = 10) -> tuple[Claim, ...]:
        """Find claims with deterministic lexical overlap against ``question``."""
        self._assert_owner()
        cap = _bounded_limit(limit)
        query_text = " ".join(str(question).casefold().split())
        query_terms = _lexical_terms(query_text)
        if not query_terms or cap == 0:
            return ()
        rows = self._connection.execute("SELECT id, statement, scope_json FROM claims").fetchall()
        ranked: list[tuple[float, str, str]] = []
        for row in rows:
            statement_text = " ".join(str(row["statement"]).casefold().split())
            scope = json.loads(row["scope_json"])
            scope_question = scope.get("question", "") if isinstance(scope, dict) else ""
            searchable_text = " ".join(
                (statement_text, " ".join(str(scope_question).casefold().split()))
            ).strip()
            searchable_terms = _lexical_terms(searchable_text)
            overlap = len(query_terms & searchable_terms)
            if overlap == 0:
                continue
            score = overlap / len(query_terms) + overlap / max(1, len(searchable_terms))
            if query_text in searchable_text:
                score += 2.0
            ranked.append((-score, statement_text, row["id"]))
        ranked.sort()
        return tuple(self.get_claim(item[2]) for item in ranked[:cap])

    def append_event(
        self,
        kind: str,
        payload: Mapping[str, object],
        *,
        claim_id: str | None = None,
        snapshot_id: str | None = None,
    ) -> KnowledgeEvent:
        """Append a durable event and return its monotonic cursor."""
        self._assert_owner()
        event_kind = str(kind).strip()
        if not event_kind:
            raise ValueError("event kind must not be empty")
        if claim_id is not None:
            self.get_claim(claim_id)
        if snapshot_id is not None:
            self.get_snapshot(snapshot_id)
        now = _now()
        with self._write_transaction():
            cursor = self._insert_event(
                event_kind, dict(payload), claim_id=claim_id, snapshot_id=snapshot_id, created_at=now
            )
        return KnowledgeEvent(cursor, event_kind, claim_id, snapshot_id, dict(payload), now)

    def read_events(
        self,
        after_cursor: int = 0,
        limit: int = 100,
        *,
        claim_id: str | None = None,
    ) -> tuple[KnowledgeEvent, ...]:
        """Read durable events after a cursor in ascending delivery order."""
        self._assert_owner()
        cap = _bounded_limit(limit)
        if claim_id is None:
            rows = self._connection.execute(
                "SELECT * FROM events WHERE cursor > ? ORDER BY cursor LIMIT ?",
                (max(0, int(after_cursor)), cap),
            ).fetchall()
        else:
            rows = self._connection.execute(
                """
                SELECT * FROM events
                WHERE cursor > ? AND claim_id = ?
                ORDER BY cursor LIMIT ?
                """,
                (max(0, int(after_cursor)), claim_id, cap),
            ).fetchall()
        return tuple(_event_from_row(row) for row in rows)

    def get_claim(self, claim_id: str) -> Claim:
        self._assert_owner()
        row = self._connection.execute("SELECT * FROM claims WHERE id = ?", (claim_id,)).fetchone()
        if row is None:
            raise KeyError(f"unknown claim: {claim_id}")
        evidence_rows = self._connection.execute(
            "SELECT * FROM evidence WHERE claim_id = ? ORDER BY created_at, id", (claim_id,)
        ).fetchall()
        dependency_rows = self._connection.execute(
            "SELECT * FROM dependencies WHERE claim_id = ? ORDER BY id", (claim_id,)
        ).fetchall()
        return Claim(
            id=row["id"],
            statement=row["statement"],
            status=ClaimStatus(row["status"]),
            applicability=Applicability(row["applicability"]),
            created_snapshot_id=row["created_snapshot_id"],
            current_snapshot_id=row["current_snapshot_id"],
            scope=json.loads(row["scope_json"]),
            created_at=row["created_at"],
            updated_at=row["updated_at"],
            evidence=tuple(_evidence_from_row(item) for item in evidence_rows),
            dependencies=tuple(_dependency_from_row(item) for item in dependency_rows),
        )

    def recheck_claim(self, claim_id: str, target_snapshot_id: str) -> RecheckResult:
        """Compare dependency baselines to a target snapshot.

        A changed dependency revokes applicability. It never changes the claim's
        supported/refuted/unresolved outcome; interpreting new evidence belongs to
        the claim evaluator that follows this check.
        """
        self._assert_owner()
        claim = self.get_claim(claim_id)
        self.get_snapshot(target_snapshot_id)
        checks = tuple(
            self._check_dependency(item, target_snapshot_id) for item in claim.dependencies
        )
        if any(item.state == "changed" for item in checks):
            applicability = Applicability.NEEDS_RECHECK
        elif not checks or any(item.state == "unknown" for item in checks):
            applicability = Applicability.UNKNOWN
        else:
            applicability = Applicability.CURRENT
        check_id = uuid.uuid4().hex
        now = _now()
        detail_json = _canonical_json(
            [
                {
                    "dependency_id": item.dependency_id,
                    "type": item.type.value,
                    "state": item.state,
                    "reason": item.reason,
                }
                for item in checks
            ]
        )
        with self._write_transaction():
            self._connection.execute(
                """
                INSERT INTO claim_checks(
                    id, claim_id, target_snapshot_id, applicability, details_json, created_at
                ) VALUES (?, ?, ?, ?, ?, ?)
                """,
                (check_id, claim_id, target_snapshot_id, applicability.value, detail_json, now),
            )
            self._connection.execute(
                """
                UPDATE claims
                SET applicability = ?, current_snapshot_id = ?, updated_at = ?
                WHERE id = ?
                """,
                (applicability.value, target_snapshot_id, now, claim_id),
            )
            if applicability is not claim.applicability:
                self._insert_event(
                    "claim.applicability_changed",
                    {
                        "previous": claim.applicability.value,
                        "current": applicability.value,
                        "previous_snapshot_id": claim.current_snapshot_id,
                        "target_snapshot_id": target_snapshot_id,
                    },
                    claim_id=claim_id,
                    snapshot_id=target_snapshot_id,
                    created_at=now,
                )
        return RecheckResult(
            check_id=check_id,
            claim_id=claim_id,
            target_snapshot_id=target_snapshot_id,
            status=claim.status,
            applicability=applicability,
            dependencies=checks,
        )

    def evaluate_query(
        self,
        snapshot_id: str,
        query_kind: str,
        parameters: Mapping[str, object],
        domain: Sequence[str] = ("**/*",),
    ) -> QueryEvaluation:
        """Evaluate a registered deterministic query against captured content."""
        self._assert_owner()
        self.get_snapshot(snapshot_id)
        evaluator = self._evaluators.get(query_kind)
        if evaluator is None:
            return QueryEvaluation("", "", (), False)
        patterns = _normalize_domain(domain)
        return evaluator(self, snapshot_id, dict(parameters), patterns)

    def _prepare_dependency(self, snapshot_id: str, item: DependencyInput) -> Dependency:
        dependency_id = uuid.uuid4().hex
        if isinstance(item, ExactBlobDependency):
            path = _normalize_relative_path(item.path)
            row = self._connection.execute(
                "SELECT digest FROM snapshot_files WHERE snapshot_id = ? AND path = ?",
                (snapshot_id, path),
            ).fetchone()
            return Dependency(
                dependency_id,
                snapshot_id,
                DependencyType.EXACT_BLOB,
                path,
                row["digest"] if row is not None else None,
                None,
                {},
                (),
                None,
                None,
                (),
                True,
            )
        if isinstance(item, (QuerySetDependency, NegativeQueryDependency, PrecomputedQueryDependency)):
            dependency_type = (
                DependencyType.NEGATIVE_QUERY
                if isinstance(item, NegativeQueryDependency)
                or isinstance(item, PrecomputedQueryDependency) and item.negative
                else DependencyType.QUERY_SET
            )
            evaluation = (
                item.evaluation
                if isinstance(item, PrecomputedQueryDependency)
                else self.evaluate_query(snapshot_id, item.query_kind, item.parameters, item.domain)
            )
            complete = evaluation.complete
            if dependency_type is DependencyType.NEGATIVE_QUERY and evaluation.results:
                complete = False
            return Dependency(
                dependency_id,
                snapshot_id,
                dependency_type,
                None,
                None,
                item.query_kind,
                dict(item.parameters),
                _normalize_domain(item.domain),
                evaluation.domain_fingerprint,
                evaluation.result_fingerprint,
                evaluation.results,
                complete,
            )
        raise TypeError(f"unsupported dependency: {type(item).__name__}")

    def _check_dependency(self, item: Dependency, snapshot_id: str) -> DependencyCheck:
        if item.type is DependencyType.EXACT_BLOB:
            row = self._connection.execute(
                "SELECT digest FROM snapshot_files WHERE snapshot_id = ? AND path = ?",
                (snapshot_id, item.path),
            ).fetchone()
            digest = row["digest"] if row is not None else None
            if digest == item.expected_digest:
                return DependencyCheck(item.id, item.type, "unchanged", "path content is unchanged")
            return DependencyCheck(item.id, item.type, "changed", "path content or membership changed")
        if not item.complete:
            return DependencyCheck(item.id, item.type, "unknown", "dependency baseline was incomplete")
        if not item.query_kind:
            return DependencyCheck(item.id, item.type, "unknown", "query identity is missing")
        baseline_analyzer = self.get_snapshot(item.created_snapshot_id).manifest.analyzer_version
        target_analyzer = self.get_snapshot(snapshot_id).manifest.analyzer_version
        if baseline_analyzer != target_analyzer:
            return DependencyCheck(item.id, item.type, "changed", "query analyzer version changed")
        evaluation = self.evaluate_query(snapshot_id, item.query_kind, item.parameters, item.domain)
        if not evaluation.complete:
            return DependencyCheck(item.id, item.type, "unknown", "query could not be evaluated completely")
        if item.type is DependencyType.NEGATIVE_QUERY and evaluation.results:
            return DependencyCheck(item.id, item.type, "changed", "negative query now has results")
        if evaluation.result_fingerprint != item.result_fingerprint:
            return DependencyCheck(item.id, item.type, "changed", "query result changed")
        if evaluation.domain_fingerprint != item.domain_fingerprint:
            return DependencyCheck(
                item.id,
                item.type,
                "unchanged",
                "query domain changed; complete re-evaluation returned the same result",
            )
        return DependencyCheck(item.id, item.type, "unchanged", "query domain and result are unchanged")

    def _insert_evidence(
        self,
        claim_id: str,
        snapshot_id: str,
        evidence: EvidenceInput,
        created_at: str,
        *,
        evidence_id: str | None = None,
    ) -> None:
        payload_json = _canonical_json(dict(evidence.payload))
        digest = _sha256(payload_json.encode("utf-8"))
        self._connection.execute(
            """
            INSERT INTO evidence(id, claim_id, snapshot_id, kind, payload_json, digest, created_at)
            VALUES (?, ?, ?, ?, ?, ?, ?)
            """,
            (
                evidence_id or uuid.uuid4().hex,
                claim_id,
                snapshot_id,
                str(evidence.kind),
                payload_json,
                digest,
                created_at,
            ),
        )

    def _insert_dependency(self, claim_id: str, snapshot_id: str, item: Dependency) -> None:
        self._connection.execute(
            """
            INSERT INTO dependencies(
                id, claim_id, created_snapshot_id, type, path, expected_digest,
                query_kind, parameters_json, domain_json, domain_fingerprint,
                result_fingerprint, results_json, complete
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                item.id,
                claim_id,
                snapshot_id,
                item.type.value,
                item.path,
                item.expected_digest,
                item.query_kind,
                _canonical_json(dict(item.parameters)),
                _canonical_json(list(item.domain)),
                item.domain_fingerprint,
                item.result_fingerprint,
                _canonical_json(list(item.results)),
                int(item.complete),
            ),
        )

    def _insert_event(
        self,
        kind: str,
        payload: Mapping[str, object],
        *,
        claim_id: str | None,
        snapshot_id: str | None,
        created_at: str,
    ) -> int:
        return _insert_event_row(
            self._connection,
            kind,
            payload,
            claim_id=claim_id,
            snapshot_id=snapshot_id,
            created_at=created_at,
        )

    @contextmanager
    def _write_transaction(self) -> Iterator[None]:
        self._assert_owner()
        with _store_write_transaction(self._connection, self.path):
            yield

    def _assert_owner(self) -> None:
        if threading.get_ident() != self._owner_thread:
            raise RuntimeError("KnowledgeStore may only be used by its owning thread")
        if self._closed:
            raise RuntimeError("KnowledgeStore is closed")

    def _initialize_schema(self) -> None:
        with self._write_transaction():
            _execute_sql_script(
                self._connection,
                """
                CREATE TABLE IF NOT EXISTS schema_info(
                    version INTEGER NOT NULL
                );
                CREATE TABLE IF NOT EXISTS snapshots(
                    id TEXT PRIMARY KEY,
                    root TEXT NOT NULL,
                    analyzer_version TEXT NOT NULL,
                    manifest_json TEXT NOT NULL,
                    created_at TEXT NOT NULL
                );
                CREATE TABLE IF NOT EXISTS blobs(
                    digest TEXT PRIMARY KEY,
                    encoding TEXT NOT NULL,
                    content BLOB NOT NULL
                );
                CREATE TABLE IF NOT EXISTS snapshot_files(
                    snapshot_id TEXT NOT NULL REFERENCES snapshots(id),
                    path TEXT NOT NULL,
                    digest TEXT NOT NULL,
                    size INTEGER NOT NULL,
                    kind TEXT NOT NULL,
                    queryable INTEGER NOT NULL,
                    blob_stored INTEGER NOT NULL,
                    PRIMARY KEY(snapshot_id, path)
                );
                CREATE TABLE IF NOT EXISTS claims(
                    id TEXT PRIMARY KEY,
                    statement TEXT NOT NULL,
                    status TEXT NOT NULL CHECK(status IN ('supported', 'refuted', 'unresolved')),
                    applicability TEXT NOT NULL CHECK(applicability IN ('current', 'needs_recheck', 'unknown')),
                    created_snapshot_id TEXT NOT NULL REFERENCES snapshots(id),
                    current_snapshot_id TEXT NOT NULL REFERENCES snapshots(id),
                    scope_json TEXT NOT NULL,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL
                );
                CREATE TABLE IF NOT EXISTS evidence(
                    id TEXT PRIMARY KEY,
                    claim_id TEXT NOT NULL REFERENCES claims(id),
                    snapshot_id TEXT NOT NULL REFERENCES snapshots(id),
                    kind TEXT NOT NULL,
                    payload_json TEXT NOT NULL,
                    digest TEXT NOT NULL,
                    created_at TEXT NOT NULL
                );
                CREATE TABLE IF NOT EXISTS dependencies(
                    id TEXT PRIMARY KEY,
                    claim_id TEXT NOT NULL REFERENCES claims(id),
                    created_snapshot_id TEXT NOT NULL REFERENCES snapshots(id),
                    type TEXT NOT NULL CHECK(type IN ('exact_blob', 'query_set', 'negative_query')),
                    path TEXT,
                    expected_digest TEXT,
                    query_kind TEXT,
                    parameters_json TEXT NOT NULL,
                    domain_json TEXT NOT NULL,
                    domain_fingerprint TEXT,
                    result_fingerprint TEXT,
                    results_json TEXT NOT NULL,
                    complete INTEGER NOT NULL
                );
                CREATE TABLE IF NOT EXISTS claim_checks(
                    id TEXT PRIMARY KEY,
                    claim_id TEXT NOT NULL REFERENCES claims(id),
                    target_snapshot_id TEXT NOT NULL REFERENCES snapshots(id),
                    applicability TEXT NOT NULL,
                    details_json TEXT NOT NULL,
                    created_at TEXT NOT NULL
                );
                CREATE TABLE IF NOT EXISTS events(
                    cursor INTEGER PRIMARY KEY AUTOINCREMENT,
                    kind TEXT NOT NULL,
                    claim_id TEXT REFERENCES claims(id),
                    snapshot_id TEXT REFERENCES snapshots(id),
                    payload_json TEXT NOT NULL,
                    created_at TEXT NOT NULL
                );
                CREATE INDEX IF NOT EXISTS evidence_claim_idx ON evidence(claim_id);
                CREATE INDEX IF NOT EXISTS evidence_snapshot_idx ON evidence(snapshot_id, claim_id);
                CREATE INDEX IF NOT EXISTS dependencies_claim_idx ON dependencies(claim_id);
                CREATE INDEX IF NOT EXISTS dependencies_snapshot_idx ON dependencies(created_snapshot_id, claim_id);
                CREATE INDEX IF NOT EXISTS checks_claim_idx ON claim_checks(claim_id, created_at);
                CREATE INDEX IF NOT EXISTS checks_snapshot_idx ON claim_checks(target_snapshot_id, claim_id);
                CREATE INDEX IF NOT EXISTS events_claim_idx ON events(claim_id, cursor);
                """,
            )
            self._connection.execute(
                """
                INSERT INTO schema_info(version)
                SELECT ? WHERE NOT EXISTS (SELECT 1 FROM schema_info)
                """,
                (SCHEMA_VERSION,),
            )
        row = self._connection.execute("SELECT version FROM schema_info").fetchone()
        if row is None or int(row["version"]) != SCHEMA_VERSION:
            found = "missing" if row is None else row["version"]
            raise RuntimeError(
                f"unsupported repository intelligence schema {found}; expected {SCHEMA_VERSION}"
            )


def _validated_store_database(
    root: Path,
    *,
    create_store_directory: bool,
    require_database: bool,
) -> Path:
    try:
        workspace = root.resolve(strict=True)
    except (OSError, RuntimeError) as error:
        raise RuntimeError(f"repository intelligence workspace is unavailable: {root}") from error
    if not workspace.is_dir():
        raise NotADirectoryError(workspace)

    store_directory = workspace / ".zyte"
    try:
        directory_stat = store_directory.lstat()
    except FileNotFoundError:
        if create_store_directory:
            try:
                store_directory.mkdir(mode=0o700)
            except FileExistsError:
                pass
            directory_stat = store_directory.lstat()
        else:
            database = store_directory / DATABASE_NAME
            if require_database:
                raise FileNotFoundError(database)
            return database
    except OSError as error:
        raise RuntimeError(f"repository intelligence state directory is unsafe: {store_directory}") from error

    if stat.S_ISLNK(directory_stat.st_mode) or not stat.S_ISDIR(directory_stat.st_mode):
        raise RuntimeError("repository intelligence state path .zyte must be a real directory")
    _validate_private_posix_path(
        store_directory, directory_stat, "repository intelligence state directory"
    )
    try:
        resolved_store = store_directory.resolve(strict=True)
    except (OSError, RuntimeError) as error:
        raise RuntimeError("repository intelligence state directory could not be resolved safely") from error
    if resolved_store != store_directory or resolved_store.parent != workspace:
        raise RuntimeError("repository intelligence state directory escapes the workspace")

    database = store_directory / DATABASE_NAME
    database_present = False
    for suffix in _SQLITE_MANAGED_SUFFIXES:
        managed = Path(f"{database}{suffix}")
        try:
            managed_stat = managed.lstat()
        except FileNotFoundError:
            continue
        except OSError as error:
            raise RuntimeError(f"managed SQLite path is unsafe: {managed}") from error
        if stat.S_ISLNK(managed_stat.st_mode) or not stat.S_ISREG(managed_stat.st_mode):
            raise RuntimeError(f"managed SQLite path must be a regular file: {managed}")
        if managed_stat.st_nlink > 1:
            raise RuntimeError(f"managed SQLite path must not be hard-linked: {managed}")
        _validate_private_posix_path(managed, managed_stat, "managed SQLite path")
        try:
            resolved_managed = managed.resolve(strict=True)
        except (OSError, RuntimeError) as error:
            raise RuntimeError(f"managed SQLite path could not be resolved safely: {managed}") from error
        if resolved_managed != managed or resolved_managed.parent != store_directory:
            raise RuntimeError(f"managed SQLite path escapes the state directory: {managed}")
        if not suffix:
            database_present = True
    if require_database and not database_present:
        raise FileNotFoundError(database)
    return database


def _materialize_workspace_package_links(
    destination_root: Path, entries: Mapping[str, SnapshotFile]
) -> None:
    """Recreate npm and pnpm workspace links from captured configuration only."""
    rules = _workspace_package_rules(destination_root, entries)
    if not rules:
        return
    package_names: dict[str, Path] = {}
    for relative_manifest in sorted(entries):
        if relative_manifest == "package.json" or not relative_manifest.endswith("/package.json"):
            continue
        package_directory = relative_manifest[: -len("/package.json")]
        if not _workspace_package_selected(package_directory, rules):
            continue
        manifest_path = _materialization_path(destination_root, relative_manifest)
        try:
            package = json.loads(manifest_path.read_text(encoding="utf-8"))
        except (OSError, UnicodeError, json.JSONDecodeError):
            continue
        name = package.get("name") if isinstance(package, dict) else None
        if not isinstance(name, str) or not re.fullmatch(
            r"(?:@[A-Za-z0-9][A-Za-z0-9._-]*/)?[A-Za-z0-9][A-Za-z0-9._-]*", name
        ):
            continue
        target = _materialization_path(destination_root, package_directory)
        previous = package_names.get(name)
        if previous is not None and previous != target:
            raise RuntimeError(f"snapshot contains duplicate workspace package name: {name}")
        package_names[name] = target
    for name, target in package_names.items():
        parts = name.split("/", 1)
        link = destination_root / "node_modules" / Path(*parts)
        if link.exists() or link.is_symlink():
            raise RuntimeError(f"snapshot workspace link target already exists: {name}")
        link.parent.mkdir(parents=True, exist_ok=True)
        relative_target = os.path.relpath(target, link.parent)
        try:
            os.symlink(relative_target, link, target_is_directory=True)
        except OSError as error:
            raise RuntimeError(f"workspace package link could not be materialized: {name}") from error


def _workspace_package_rules(
    destination_root: Path, entries: Mapping[str, SnapshotFile]
) -> tuple[tuple[bool, tuple[str, ...]], ...]:
    raw_patterns: list[str] = []
    root_manifest = destination_root / "package.json"
    if "package.json" in entries and root_manifest.is_file():
        try:
            metadata = json.loads(root_manifest.read_text(encoding="utf-8"))
        except (OSError, UnicodeError, json.JSONDecodeError):
            metadata = None
        raw_workspaces = metadata.get("workspaces") if isinstance(metadata, dict) else None
        if isinstance(raw_workspaces, dict):
            raw_workspaces = raw_workspaces.get("packages")
        if isinstance(raw_workspaces, list):
            raw_patterns.extend(item for item in raw_workspaces if isinstance(item, str))

    pnpm_manifest = destination_root / "pnpm-workspace.yaml"
    if "pnpm-workspace.yaml" in entries and pnpm_manifest.is_file():
        try:
            pnpm_text = pnpm_manifest.read_text(encoding="utf-8")
        except (OSError, UnicodeError):
            pnpm_text = ""
        raw_patterns.extend(_parse_pnpm_workspace_patterns(pnpm_text))

    if len(raw_patterns) > _MAX_WORKSPACE_PATTERNS:
        return ()
    rules: list[tuple[bool, tuple[str, ...]]] = []
    for raw_pattern in raw_patterns:
        prepared = _prepare_workspace_pattern(raw_pattern)
        if prepared is None:
            continue
        negated, pattern = prepared
        expansions = _expand_workspace_braces(pattern)
        if expansions is None or len(rules) + len(expansions) > _MAX_WORKSPACE_BRACE_EXPANSIONS:
            return ()
        for expansion in expansions:
            parts = _safe_workspace_pattern_parts(expansion)
            if parts is not None:
                rules.append((negated, parts))
    return tuple(rules)


def _prepare_workspace_pattern(raw_pattern: str) -> tuple[bool, str] | None:
    pattern = raw_pattern.strip()
    if not pattern or "\0" in pattern or len(pattern) > _MAX_WORKSPACE_PATTERN_CHARS:
        return None
    negated = pattern.startswith("!")
    if negated:
        pattern = pattern[1:].strip()
    raw_windows = PureWindowsPath(pattern)
    normalized = pattern.replace("\\", "/")
    if (
        not normalized
        or PurePosixPath(normalized).is_absolute()
        or raw_windows.is_absolute()
        or raw_windows.drive
    ):
        return None
    while normalized.startswith("./"):
        normalized = normalized[2:]
    normalized = normalized.rstrip("/")
    if not normalized:
        return None
    return negated, normalized


def _safe_workspace_pattern_parts(pattern: str) -> tuple[str, ...] | None:
    raw_windows = PureWindowsPath(pattern)
    if (
        not pattern
        or len(pattern) > _MAX_WORKSPACE_PATTERN_CHARS
        or PurePosixPath(pattern).is_absolute()
        or raw_windows.is_absolute()
        or raw_windows.drive
    ):
        return None
    parts = tuple(pattern.split("/"))
    if not parts or any(part in ("", ".", "..") or "\0" in part for part in parts):
        return None
    return parts


def _expand_workspace_braces(pattern: str) -> tuple[str, ...] | None:
    expanded: list[str] = []

    def visit(value: str) -> bool:
        opening = value.find("{")
        if opening < 0:
            expanded.append(value)
            return len(expanded) <= _MAX_WORKSPACE_BRACE_EXPANSIONS
        depth = 0
        closing = -1
        for index in range(opening, len(value)):
            if value[index] == "{":
                depth += 1
            elif value[index] == "}":
                depth -= 1
                if depth == 0:
                    closing = index
                    break
        if closing < 0:
            expanded.append(value)
            return len(expanded) <= _MAX_WORKSPACE_BRACE_EXPANSIONS
        alternatives = _split_workspace_brace_alternatives(value[opening + 1 : closing])
        if len(alternatives) < 2:
            expanded.append(value)
            return len(expanded) <= _MAX_WORKSPACE_BRACE_EXPANSIONS
        prefix = value[:opening]
        suffix = value[closing + 1 :]
        return all(visit(f"{prefix}{alternative}{suffix}") for alternative in alternatives)

    if not visit(pattern):
        return None
    return tuple(expanded)


def _split_workspace_brace_alternatives(value: str) -> tuple[str, ...]:
    alternatives: list[str] = []
    start = 0
    depth = 0
    for index, character in enumerate(value):
        if character == "{":
            depth += 1
        elif character == "}":
            depth = max(0, depth - 1)
        elif character == "," and depth == 0:
            alternatives.append(value[start:index])
            start = index + 1
    alternatives.append(value[start:])
    return tuple(alternatives)


def _parse_pnpm_workspace_patterns(text: str) -> tuple[str, ...]:
    lines = text.splitlines()
    for index, raw_line in enumerate(lines):
        if "\t" in raw_line:
            continue
        content = _strip_workspace_yaml_comment(raw_line).rstrip()
        match = re.fullmatch(r"(?P<indent> *)packages\s*:\s*(?P<value>.*)", content)
        if match is None or match.group("indent"):
            continue
        inline = match.group("value").strip()
        if inline:
            return _parse_workspace_yaml_flow_sequence(inline)
        patterns: list[str] = []
        for child_line in lines[index + 1 :]:
            if "\t" in child_line:
                continue
            child = _strip_workspace_yaml_comment(child_line).rstrip()
            if not child.strip():
                continue
            indentation = len(child) - len(child.lstrip(" "))
            if indentation == 0:
                break
            item = child.lstrip(" ")
            if not item.startswith("-"):
                continue
            scalar = _parse_workspace_yaml_scalar(item[1:].strip())
            if scalar is not None:
                patterns.append(scalar)
            if len(patterns) > _MAX_WORKSPACE_PATTERNS:
                return ()
        return tuple(patterns)
    return ()


def _strip_workspace_yaml_comment(value: str) -> str:
    quote: str | None = None
    escaped = False
    for index, character in enumerate(value):
        if quote == '"':
            if escaped:
                escaped = False
            elif character == "\\":
                escaped = True
            elif character == quote:
                quote = None
        elif quote == "'":
            if character == quote:
                quote = None
        elif character in ('"', "'"):
            quote = character
        elif character == "#" and (index == 0 or value[index - 1].isspace()):
            return value[:index]
    return value


def _parse_workspace_yaml_flow_sequence(value: str) -> tuple[str, ...]:
    if not value.startswith("[") or not value.endswith("]"):
        return ()
    items: list[str] = []
    start = 1
    quote: str | None = None
    escaped = False
    brace_depth = 0
    for index in range(1, len(value) - 1):
        character = value[index]
        if quote == '"':
            if escaped:
                escaped = False
            elif character == "\\":
                escaped = True
            elif character == quote:
                quote = None
        elif quote == "'":
            if character == quote:
                quote = None
        elif character in ('"', "'"):
            quote = character
        elif character == "{":
            brace_depth += 1
        elif character == "}":
            brace_depth = max(0, brace_depth - 1)
        elif character == "," and brace_depth == 0:
            scalar = _parse_workspace_yaml_scalar(value[start:index].strip())
            if scalar is not None:
                items.append(scalar)
            start = index + 1
    scalar = _parse_workspace_yaml_scalar(value[start:-1].strip())
    if scalar is not None:
        items.append(scalar)
    return tuple(items)


def _parse_workspace_yaml_scalar(value: str) -> str | None:
    if not value:
        return None
    if value.startswith("'"):
        if len(value) < 2 or not value.endswith("'"):
            return None
        return value[1:-1].replace("''", "'")
    if value.startswith('"'):
        if len(value) < 2 or not value.endswith('"'):
            return None
        try:
            decoded = json.loads(value)
        except json.JSONDecodeError:
            return None
        return decoded if isinstance(decoded, str) else None
    return value


def _workspace_package_selected(
    package_directory: str, rules: Sequence[tuple[bool, tuple[str, ...]]]
) -> bool:
    path_parts = tuple(package_directory.split("/"))
    selected = False
    for negated, pattern_parts in rules:
        if _workspace_path_matches(path_parts, pattern_parts):
            selected = not negated
    return selected


def _workspace_path_matches(path_parts: tuple[str, ...], pattern_parts: tuple[str, ...]) -> bool:
    states = {0}

    def include_zero_length_globstars(indexes: set[int]) -> set[int]:
        expanded = set(indexes)
        frontier = list(indexes)
        while frontier:
            index = frontier.pop()
            if index < len(pattern_parts) and pattern_parts[index] == "**":
                following = index + 1
                if following not in expanded:
                    expanded.add(following)
                    frontier.append(following)
        return expanded

    for path_part in path_parts:
        following_states: set[int] = set()
        for pattern_index in include_zero_length_globstars(states):
            if pattern_index == len(pattern_parts):
                continue
            pattern = pattern_parts[pattern_index]
            if pattern == "**":
                following_states.add(pattern_index)
            elif fnmatch.fnmatchcase(path_part, pattern):
                following_states.add(pattern_index + 1)
        states = following_states
        if not states:
            return False
    return len(pattern_parts) in include_zero_length_globstars(states)


def _validate_private_posix_path(path: Path, item: os.stat_result, description: str) -> None:
    if os.name == "nt":
        return
    get_effective_user = getattr(os, "geteuid", None)
    if get_effective_user is None:
        raise RuntimeError(
            f"{description} ownership cannot be verified on this platform: {path}"
        )
    if item.st_uid != get_effective_user():
        raise RuntimeError(f"{description} must be owned by the current user: {path}")
    if stat.S_IMODE(item.st_mode) & 0o077:
        raise RuntimeError(
            f"{description} permissions must deny group and other access: {path}"
        )


def _create_private_store_database(database: Path) -> None:
    """Create the primary database without exposing it through the process umask."""
    if os.name == "nt":
        return
    flags = os.O_CREAT | os.O_EXCL | os.O_WRONLY
    flags |= getattr(os, "O_CLOEXEC", 0)
    flags |= getattr(os, "O_NOFOLLOW", 0)
    try:
        descriptor = os.open(database, flags, 0o600)
    except FileExistsError:
        return
    except OSError as error:
        raise RuntimeError(
            f"repository intelligence database could not be created safely: {database}"
        ) from error
    try:
        os.fchmod(descriptor, 0o600)
        item = os.fstat(descriptor)
        if not stat.S_ISREG(item.st_mode) or item.st_nlink != 1:
            raise RuntimeError(
                f"repository intelligence database must be a private regular file: {database}"
            )
        _validate_private_posix_path(database, item, "managed SQLite path")
    finally:
        os.close(descriptor)


def _existing_managed_store_paths(database: Path) -> frozenset[Path]:
    return frozenset(
        managed
        for suffix in _SQLITE_MANAGED_SUFFIXES
        if (managed := Path(f"{database}{suffix}")).exists()
    )


def _tighten_new_managed_store_files(
    database: Path, existing_managed_paths: frozenset[Path]
) -> None:
    """Restrict SQLite-created sidecars without repairing untrusted pre-existing state."""
    if os.name == "nt":
        return
    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
    for suffix in _SQLITE_MANAGED_SUFFIXES:
        managed = Path(f"{database}{suffix}")
        if managed in existing_managed_paths:
            continue
        try:
            descriptor = os.open(managed, flags)
        except FileNotFoundError:
            continue
        except OSError as error:
            raise RuntimeError(f"new managed SQLite path is unsafe: {managed}") from error
        try:
            item = os.fstat(descriptor)
            if not stat.S_ISREG(item.st_mode) or item.st_nlink != 1:
                raise RuntimeError(
                    f"new managed SQLite path must be a regular file: {managed}"
                )
            get_effective_user = getattr(os, "geteuid", None)
            if get_effective_user is None or item.st_uid != get_effective_user():
                raise RuntimeError(
                    f"new managed SQLite path must be owned by the current user: {managed}"
                )
            os.fchmod(descriptor, 0o600)
            _validate_private_posix_path(
                managed, os.fstat(descriptor), "new managed SQLite path"
            )
        finally:
            os.close(descriptor)


def _managed_store_bytes(database: Path) -> int:
    total = 0
    for suffix in _SQLITE_MANAGED_SUFFIXES:
        managed = Path(f"{database}{suffix}")
        try:
            item = managed.lstat()
        except FileNotFoundError:
            continue
        if stat.S_ISREG(item.st_mode):
            total += item.st_size
    return total


def _store_budget_error() -> KnowledgeStoreLimitError:
    return KnowledgeStoreLimitError(
        "repository intelligence state exceeds the fixed "
        f"{MAX_KNOWLEDGE_STORE_BYTES}-byte store budget"
    )


def _insert_event_row(
    connection: sqlite3.Connection,
    kind: str,
    payload: Mapping[str, object],
    *,
    claim_id: str | None,
    snapshot_id: str | None,
    created_at: str,
) -> int:
    """Insert an event on the caller's transaction and return its cursor."""
    event_kind = str(kind).strip()
    if not event_kind:
        raise ValueError("event kind must not be empty")
    cursor = connection.execute(
        """
        INSERT INTO events(kind, claim_id, snapshot_id, payload_json, created_at)
        VALUES (?, ?, ?, ?, ?)
        """,
        (event_kind, claim_id, snapshot_id, _canonical_json(dict(payload)), created_at),
    ).lastrowid
    if cursor is None:
        raise RuntimeError("SQLite did not return an event cursor")
    return int(cursor)


def _connection_database(connection: sqlite3.Connection) -> Path:
    rows = connection.execute("PRAGMA database_list").fetchall()
    for row in rows:
        if str(row[1]) == "main" and str(row[2]):
            return Path(str(row[2])).resolve()
    raise RuntimeError("SQLite connection has no file-backed main database")


def _managed_file_bytes(path: Path) -> int:
    try:
        item = path.lstat()
    except FileNotFoundError:
        return 0
    return item.st_size if stat.S_ISREG(item.st_mode) else 0


def _enforce_store_budget(connection: sqlite3.Connection, database: Path) -> None:
    """Apply an atomic SQLite page cap and reject an already oversized store."""
    managed_bytes = _managed_store_bytes(database)
    database_file_bytes = _managed_file_bytes(database)
    sidecar_bytes = managed_bytes - database_file_bytes
    page_size = int(connection.execute("PRAGMA page_size").fetchone()[0])
    page_count = int(connection.execute("PRAGMA page_count").fetchone()[0])
    if page_size <= 0 or page_count < 0:
        raise RuntimeError("SQLite returned invalid page accounting")
    database_bytes = page_size * page_count
    accounted_bytes = max(database_file_bytes, database_bytes) + sidecar_bytes
    if accounted_bytes > MAX_KNOWLEDGE_STORE_BYTES:
        raise _store_budget_error()
    maximum_pages = (MAX_KNOWLEDGE_STORE_BYTES - sidecar_bytes) // page_size
    if maximum_pages < max(1, page_count):
        raise _store_budget_error()
    configured = int(
        connection.execute(f"PRAGMA max_page_count = {maximum_pages}").fetchone()[0]
    )
    if configured > maximum_pages:
        raise _store_budget_error()


def _enforce_wal_commit_budget(connection: sqlite3.Connection, database: Path) -> None:
    """Reserve a conservative upper bound for the WAL commit and its index.

    With cache spill disabled, each dirty database page has at most one final
    frame to flush at commit. Reserving every allocated page also covers pages
    that the transaction did not actually touch.
    """
    if str(connection.execute("PRAGMA journal_mode").fetchone()[0]).lower() != "wal":
        raise RuntimeError("repository intelligence store requires SQLite WAL mode")
    if int(connection.execute("PRAGMA cache_spill").fetchone()[0]) != 0:
        raise RuntimeError("repository intelligence writes require disabled SQLite cache spill")
    if int(connection.execute("PRAGMA wal_autocheckpoint").fetchone()[0]) != 0:
        raise RuntimeError("repository intelligence writes require disabled WAL auto-checkpoint")

    page_size = int(connection.execute("PRAGMA page_size").fetchone()[0])
    page_count = int(connection.execute("PRAGMA page_count").fetchone()[0])
    frame_bytes = page_size + _SQLITE_WAL_FRAME_HEADER_BYTES
    database_file_bytes = _managed_file_bytes(database)
    database_growth = max(0, page_count * page_size - database_file_bytes)
    wal_path = Path(f"{database}-wal")
    shm_path = Path(f"{database}-shm")
    wal_bytes = _managed_file_bytes(wal_path)
    shm_bytes = _managed_file_bytes(shm_path)
    existing_frames = max(
        0,
        (
            max(0, wal_bytes - _SQLITE_WAL_HEADER_BYTES)
            + frame_bytes
            - 1
        )
        // frame_bytes,
    )
    wal_growth = page_count * frame_bytes
    if wal_bytes < _SQLITE_WAL_HEADER_BYTES:
        wal_growth += _SQLITE_WAL_HEADER_BYTES - wal_bytes
    total_frames = existing_frames + page_count
    index_regions = 1 + (
        total_frames + _SQLITE_WAL_INDEX_FRAMES_PER_REGION - 1
    ) // _SQLITE_WAL_INDEX_FRAMES_PER_REGION
    required_shm_bytes = index_regions * _SQLITE_WAL_INDEX_REGION_BYTES
    shm_growth = max(0, required_shm_bytes - shm_bytes)
    # A checkpoint may copy every logical page into the database while the WAL
    # and its index still exist (for example while a reader pins old frames).
    # Reserve both peaks so a successful commit cannot become oversized later.
    if (
        _managed_store_bytes(database)
        + database_growth
        + wal_growth
        + shm_growth
        > MAX_KNOWLEDGE_STORE_BYTES
    ):
        raise _store_budget_error()


@contextmanager
def _store_write_transaction(
    connection: sqlite3.Connection, database: Path | None = None
) -> Iterator[None]:
    """Serialize and budget every write to the repository-intelligence database."""
    if connection.in_transaction:
        raise RuntimeError("nested repository intelligence write transaction")
    database_path = (
        database.resolve() if database is not None else _connection_database(connection)
    )
    connection.execute("PRAGMA cache_spill = OFF")
    connection.execute("PRAGMA wal_autocheckpoint = 0")
    connection.execute("BEGIN IMMEDIATE")
    try:
        _enforce_store_budget(connection, database_path)
        changes_before = connection.total_changes
        schema_before = int(connection.execute("PRAGMA schema_version").fetchone()[0])
        yield
        _enforce_store_budget(connection, database_path)
        schema_after = int(connection.execute("PRAGMA schema_version").fetchone()[0])
        if connection.total_changes != changes_before or schema_after != schema_before:
            _enforce_wal_commit_budget(connection, database_path)
        connection.execute("COMMIT")
    except BaseException as error:
        if connection.in_transaction:
            connection.execute("ROLLBACK")
        if (
            isinstance(error, sqlite3.Error)
            and getattr(error, "sqlite_errorcode", None) == sqlite3.SQLITE_FULL
        ):
            raise _store_budget_error() from error
        raise


def _execute_sql_script(connection: sqlite3.Connection, script: str) -> None:
    """Execute complete statements without ``executescript``'s implicit commit."""
    if not connection.in_transaction:
        raise RuntimeError("schema statements require a write transaction")
    pending = ""
    for line in script.splitlines(keepends=True):
        pending += line
        if sqlite3.complete_statement(pending):
            statement = pending.strip()
            pending = ""
            if statement:
                connection.execute(statement)
    if pending.strip():
        raise RuntimeError("incomplete SQLite schema statement")


def _new_snapshot_blob_bytes(
    connection: sqlite3.Connection, captured: _CapturedSnapshot
) -> int:
    missing = set(captured.blobs)
    digests = sorted(missing)
    for offset in range(0, len(digests), 500):
        chunk = digests[offset : offset + 500]
        placeholders = ",".join("?" for _item in chunk)
        rows = connection.execute(
            f"SELECT digest FROM blobs WHERE digest IN ({placeholders})", chunk
        ).fetchall()
        missing.difference_update(str(row["digest"]) for row in rows)
    return sum(len(captured.blobs[digest]) for digest in missing)


def _capture_workspace(root: Path, analyzer_version: str) -> _CapturedSnapshot:
    if not root.is_dir():
        raise NotADirectoryError(root)
    for _attempt in range(3):
        paths = _visible_paths(root)
        entries: list[SnapshotFile] = []
        blobs: dict[str, bytes] = {}
        signatures: dict[str, tuple[int, int, int, int, int]] = {}
        captured_bytes = 0
        stable = True
        for relative in paths:
            path = root.joinpath(*relative.split("/"))
            try:
                captured = _read_entry(path, MAX_SNAPSHOT_TOTAL_BYTES - captured_bytes)
            except SnapshotLimitError:
                raise
            except RuntimeError:
                stable = False
                break
            if captured is None:
                stable = False
                break
            data, kind, signature = captured
            digest = _sha256(data)
            queryable = kind == "file" and len(data) <= MAX_QUERYABLE_BYTES and _is_utf8_text(data)
            entries.append(SnapshotFile(relative, digest, len(data), kind, queryable))
            blobs.setdefault(digest, data)
            signatures[relative] = signature
            captured_bytes += len(data)
        if not stable or paths != _visible_paths(root):
            continue
        if any(
            _entry_signature(root.joinpath(*relative.split("/"))) != signature
            for relative, signature in signatures.items()
        ):
            continue
        entries.sort(key=lambda item: item.path)
        configs = tuple((item.path, item.digest) for item in entries if _is_config(item.path))
        locks = tuple((item.path, item.digest) for item in entries if _is_lock(item.path))
        manifest = SnapshotManifest(analyzer_version, tuple(entries), configs, locks)
        snapshot_id = _sha256(_canonical_json(manifest.as_dict()).encode("utf-8"))
        return _CapturedSnapshot(snapshot_id, manifest, blobs)
    raise RuntimeError("workspace did not stabilize during snapshot capture")


def _visible_paths(root: Path) -> tuple[str, ...]:
    git_paths = _git_visible_paths(root)
    if git_paths is not None:
        return git_paths
    return _walk_visible_paths(root)


def _trusted_git_executable(root: Path) -> str | None:
    configured = os.environ.get("BYTE_GIT_EXECUTABLE")
    if configured:
        unresolved = Path(configured).expanduser()
        if not unresolved.is_absolute():
            return None
        candidate = _verified_executable(unresolved, root)
        return str(candidate) if candidate is not None else None

    for raw_directory in os.environ.get("PATH", "").split(os.pathsep):
        if not raw_directory:
            continue
        unresolved_directory = Path(raw_directory.strip('"'))
        if not unresolved_directory.is_absolute():
            continue
        try:
            directory = unresolved_directory.resolve(strict=True)
        except (OSError, RuntimeError):
            continue
        if (
            not directory.is_dir()
            or _path_is_within(root, unresolved_directory)
            or _path_is_within(root, directory)
        ):
            continue
        for name in _executable_names("git"):
            candidate = _verified_executable(directory / name, root)
            if candidate is not None:
                return str(candidate)
    return None


def _executable_names(name: str) -> tuple[str, ...]:
    if os.name != "nt":
        return (name,)
    return tuple(f"{name}{suffix}" for suffix in (".exe", ".com"))


def _verified_executable(unresolved: Path, root: Path) -> Path | None:
    if not unresolved.is_absolute() or _path_is_within(root, unresolved):
        return None
    try:
        candidate = unresolved.resolve(strict=True)
        workspace = root.resolve(strict=True)
    except (OSError, RuntimeError):
        return None
    if _path_is_within(workspace, candidate) or not candidate.is_file():
        return None
    if os.name == "nt" and candidate.suffix.casefold() not in {".exe", ".com"}:
        return None
    if not os.access(candidate, os.X_OK):
        return None
    return candidate


def _path_is_within(root: Path, candidate: Path) -> bool:
    try:
        candidate.absolute().relative_to(root.absolute())
    except ValueError:
        return False
    return True


def _git_visible_paths(root: Path) -> tuple[str, ...] | None:
    git = _trusted_git_executable(root)
    if git is None:
        return None
    output = _run_git_ls_files(
        [git, "-C", str(root), "ls-files", "-z", "--cached", "--others", "--exclude-standard"]
    )
    if output is None:
        return None
    found: set[str] = set()
    for raw in output.split(b"\0"):
        if not raw:
            continue
        relative = raw.decode("utf-8", errors="surrogateescape").replace("\\", "/")
        try:
            normalized = _normalize_relative_path(relative)
        except ValueError:
            continue
        if any(part in _SNAPSHOT_ALWAYS_SKIP for part in normalized.split("/")):
            continue
        if _safe_workspace_entry(root, normalized) and normalized not in found:
            if len(found) >= MAX_SNAPSHOT_FILES:
                raise SnapshotLimitError(
                    f"snapshot file count exceeds the fixed {MAX_SNAPSHOT_FILES}-entry limit"
                )
            found.add(normalized)
    return tuple(sorted(found))


def _run_git_ls_files(command: Sequence[str]) -> bytes | None:
    output = bytearray()
    lock = threading.Lock()
    exceeded = threading.Event()
    failed = threading.Event()
    try:
        options: dict[str, object] = {
            "stdout": subprocess.PIPE,
            "stderr": subprocess.DEVNULL,
        }
        if os.name == "nt":
            options["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
        else:
            options["start_new_session"] = True
        process = subprocess.Popen(list(command), **options)
    except OSError:
        return None

    def terminate() -> None:
        try:
            if os.name != "nt":
                os.killpg(process.pid, signal.SIGKILL)
            elif process.poll() is None:
                process.kill()
        except OSError:
            try:
                process.kill()
            except OSError:
                pass

    def drain() -> None:
        assert process.stdout is not None
        try:
            read = getattr(process.stdout, "read1", process.stdout.read)
            while chunk := read(64 * 1024):
                with lock:
                    if len(output) + len(chunk) > MAX_GIT_PATH_OUTPUT_BYTES:
                        exceeded.set()
                        terminate()
                        return
                    output.extend(chunk)
        except (OSError, ValueError):
            failed.set()
            terminate()

    reader = threading.Thread(target=drain, name="byte-git-paths", daemon=True)
    reader.start()
    try:
        returncode = process.wait(timeout=15)
    except subprocess.TimeoutExpired:
        terminate()
        try:
            process.wait(timeout=5)
        except subprocess.TimeoutExpired:
            process.kill()
            process.wait(timeout=5)
        return None
    finally:
        reader.join(timeout=5)
        if process.stdout is not None:
            try:
                process.stdout.close()
            except (OSError, ValueError):
                pass
        reader.join(timeout=1)

    if reader.is_alive() or failed.is_set():
        terminate()
        return None
    if exceeded.is_set():
        raise SnapshotLimitError(
            f"Git path output exceeds the fixed {MAX_GIT_PATH_OUTPUT_BYTES}-byte limit"
        )
    if returncode != 0:
        return None
    with lock:
        return bytes(output)


def _walk_visible_paths(root: Path) -> tuple[str, ...]:
    stack = IgnoreStack()
    found: list[str] = []
    tree_entries = 0

    def visit(directory: Path, relative: str) -> None:
        nonlocal tree_entries
        pushed = False
        ignore_path = directory / ".gitignore"
        try:
            ignore_rules = _read_ignore_rules(ignore_path)
            if ignore_rules is not None:
                stack.push(relative, ignore_rules)
                pushed = True
        except OSError:
            pass
        try:
            entries = []
            with os.scandir(directory) as iterator:
                for entry in iterator:
                    tree_entries += 1
                    if tree_entries > MAX_SNAPSHOT_TREE_ENTRIES:
                        raise SnapshotLimitError(
                            "snapshot traversal exceeds the fixed "
                            f"{MAX_SNAPSHOT_TREE_ENTRIES}-entry limit"
                        )
                    entries.append(entry)
            entries.sort(key=lambda item: item.name)
        except OSError:
            entries = []
        directories: list[tuple[Path, str]] = []
        for entry in entries:
            child = f"{relative}/{entry.name}" if relative else entry.name
            try:
                mode = entry.stat(follow_symlinks=False).st_mode
            except OSError:
                continue
            if stat.S_ISDIR(mode):
                if entry.name in _SNAPSHOT_ALWAYS_SKIP or stack.ignored(child, True):
                    continue
                directories.append((Path(entry.path), child))
            elif (stat.S_ISREG(mode) or stat.S_ISLNK(mode)) and not stack.ignored(child, False):
                if len(found) >= MAX_SNAPSHOT_FILES:
                    raise SnapshotLimitError(
                        f"snapshot file count exceeds the fixed {MAX_SNAPSHOT_FILES}-entry limit"
                    )
                found.append(child)
        for child_path, child in directories:
            visit(child_path, child)
        if pushed:
            stack.pop()

    visit(root, "")
    return tuple(sorted(found))


def _read_ignore_rules(path: Path) -> str | None:
    try:
        before = path.lstat()
    except FileNotFoundError:
        return None
    if not stat.S_ISREG(before.st_mode):
        return None
    if before.st_size > MAX_IGNORE_FILE_BYTES:
        raise SnapshotLimitError(
            f"ignore file exceeds the fixed {MAX_IGNORE_FILE_BYTES}-byte limit: {path}"
        )
    flags = os.O_RDONLY
    flags |= getattr(os, "O_BINARY", 0)
    flags |= getattr(os, "O_CLOEXEC", 0)
    flags |= getattr(os, "O_NOFOLLOW", 0)
    descriptor = os.open(path, flags)
    try:
        opened_before = os.fstat(descriptor)
        if not stat.S_ISREG(opened_before.st_mode) or _stat_identity(before) != _stat_identity(
            opened_before
        ):
            raise RuntimeError(f"ignore file changed during snapshot discovery: {path}")
        data = os.read(descriptor, MAX_IGNORE_FILE_BYTES + 1)
        if len(data) > MAX_IGNORE_FILE_BYTES:
            raise SnapshotLimitError(
                f"ignore file exceeds the fixed {MAX_IGNORE_FILE_BYTES}-byte limit: {path}"
            )
        opened_after = os.fstat(descriptor)
        if _stat_signature(opened_before) != _stat_signature(opened_after):
            raise RuntimeError(f"ignore file changed during snapshot discovery: {path}")
        return data.decode("utf-8", errors="replace")
    finally:
        os.close(descriptor)


def _safe_workspace_entry(root: Path, relative: str) -> bool:
    path = root.joinpath(*relative.split("/"))
    try:
        mode = path.lstat().st_mode
    except OSError:
        return False
    return stat.S_ISREG(mode) or stat.S_ISLNK(mode)


def _entry_signature(path: Path) -> tuple[int, int, int, int, int] | None:
    try:
        current = path.lstat()
    except (OSError, RuntimeError):
        return None
    return _stat_signature(current)


def _read_entry(
    path: Path, remaining_snapshot_bytes: int
) -> tuple[bytes, str, tuple[int, int, int, int, int]] | None:
    for _attempt in range(2):
        try:
            before = path.lstat()
            _check_snapshot_size(path, before.st_size, remaining_snapshot_bytes)
            if stat.S_ISLNK(before.st_mode):
                data = os.readlink(path).encode("utf-8", errors="surrogateescape")
                kind = "symlink"
            else:
                if not stat.S_ISREG(before.st_mode):
                    return None
                data = _read_regular_snapshot_file(path, before, remaining_snapshot_bytes)
                kind = "file"
            _check_snapshot_size(path, len(data), remaining_snapshot_bytes)
            after = path.lstat()
        except OSError:
            return None
        before_signature = _stat_signature(before)
        after_signature = _stat_signature(after)
        if before_signature == after_signature:
            return data, kind, after_signature
    raise RuntimeError(f"workspace file changed during snapshot capture: {path}")


def _read_regular_snapshot_file(
    path: Path, expected: os.stat_result, remaining_snapshot_bytes: int
) -> bytes:
    limit = min(MAX_SNAPSHOT_FILE_BYTES, remaining_snapshot_bytes)
    flags = os.O_RDONLY
    flags |= getattr(os, "O_BINARY", 0)
    flags |= getattr(os, "O_CLOEXEC", 0)
    flags |= getattr(os, "O_NOFOLLOW", 0)
    flags |= getattr(os, "O_NONBLOCK", 0)
    descriptor = os.open(path, flags)
    try:
        opened_before = os.fstat(descriptor)
        if not stat.S_ISREG(opened_before.st_mode):
            raise OSError(f"snapshot entry is not a regular file: {path}")
        if _stat_identity(expected) != _stat_identity(opened_before):
            raise RuntimeError(f"workspace file changed during snapshot capture: {path}")
        chunks: list[bytes] = []
        observed = 0
        while observed <= limit:
            chunk = os.read(
                descriptor,
                min(SNAPSHOT_READ_CHUNK_BYTES, limit + 1 - observed),
            )
            if not chunk:
                break
            chunks.append(chunk)
            observed += len(chunk)
        if observed > limit:
            _raise_snapshot_limit(path, observed, remaining_snapshot_bytes)
        opened_after = os.fstat(descriptor)
        if _stat_signature(opened_before) != _stat_signature(opened_after):
            raise RuntimeError(f"workspace file changed during snapshot capture: {path}")
        return b"".join(chunks)
    finally:
        os.close(descriptor)


def _check_snapshot_size(path: Path, size: int, remaining_snapshot_bytes: int) -> None:
    if size > MAX_SNAPSHOT_FILE_BYTES or size > remaining_snapshot_bytes:
        _raise_snapshot_limit(path, size, remaining_snapshot_bytes)


def _raise_snapshot_limit(path: Path, size: int, remaining_snapshot_bytes: int) -> None:
    if size > MAX_SNAPSHOT_FILE_BYTES:
        raise SnapshotLimitError(
            f"snapshot file exceeds the fixed {MAX_SNAPSHOT_FILE_BYTES}-byte limit: {path}"
        )
    raise SnapshotLimitError(
        "snapshot content exceeds the fixed "
        f"{MAX_SNAPSHOT_TOTAL_BYTES}-byte aggregate limit while reading: {path}"
    )


def _stat_identity(value: os.stat_result) -> tuple[int, int, int]:
    return (stat.S_IFMT(value.st_mode), value.st_dev, value.st_ino)


def _stat_signature(value: os.stat_result) -> tuple[int, int, int, int, int]:
    return (*_stat_identity(value), value.st_size, value.st_mtime_ns)


def query_input_fingerprint(snapshot: Snapshot, domain: Sequence[str]) -> str:
    """Fingerprint source-domain content plus resolution-affecting configuration."""
    patterns = _normalize_domain(domain)
    material = [
        f"source\0{item.path}\0{item.digest}"
        for item in snapshot.files
        if _path_matches_domain(item.path, patterns)
    ]
    material.extend(f"config\0{path}\0{digest}" for path, digest in snapshot.manifest.config_hashes)
    material.extend(f"lock\0{path}\0{digest}" for path, digest in snapshot.manifest.lock_hashes)
    return _fingerprint(sorted(material))


def _evaluate_typescript_consumers(
    store: KnowledgeStore,
    snapshot_id: str,
    parameters: Mapping[str, object],
    domain: tuple[str, ...],
) -> QueryEvaluation:
    identifiers = parameters.get("identifiers")
    baseline_input = parameters.get("baseline_input_fingerprint")
    baseline_domain = parameters.get("baseline_domain_fingerprint")
    baseline_result = parameters.get("baseline_result_fingerprint")
    baseline_results = parameters.get("baseline_results", [])
    baseline_complete = parameters.get("baseline_complete") is True
    adapter_version = parameters.get("adapter_version")
    if (
        not isinstance(identifiers, list)
        or not identifiers
        or any(not isinstance(item, str) or not item for item in identifiers)
        or not isinstance(baseline_input, str)
        or not isinstance(baseline_domain, str)
        or not isinstance(baseline_result, str)
        or not isinstance(baseline_results, list)
        or any(not isinstance(item, str) for item in baseline_results)
        or not isinstance(adapter_version, str)
    ):
        return QueryEvaluation("", "", (), False)
    if adapter_version != TYPESCRIPT_ADAPTER_VERSION:
        return QueryEvaluation("", "", (), False)
    snapshot = store.get_snapshot(snapshot_id)
    current_input = query_input_fingerprint(snapshot, domain)
    if current_input == baseline_input:
        return QueryEvaluation(
            baseline_domain,
            baseline_result,
            tuple(baseline_results),
            baseline_complete,
        )
    result = query_typescript_boundary(store, snapshot_id, identifiers, max_results=1_000)
    if result.get("available") is not True or result.get("adapterVersion") != adapter_version:
        return QueryEvaluation("", "", (), False)
    result_domain = result["domain"]
    consumers = result["consumers"]
    return QueryEvaluation(
        _fingerprint([result_domain["membershipFingerprint"], result_domain["resolutionFingerprint"]]),
        result["resultFingerprint"],
        tuple(sorted(
            "\0".join(str(consumer[key]) for key in ("path", "line", "column", "kind", "name"))
            for consumer in consumers
        )),
        result["complete"] is True and result["truncated"] is not True,
    )


def query_typescript_boundary(
    store: KnowledgeStore,
    snapshot_id: str,
    identifiers: Sequence[str],
    *,
    max_results: int = 250,
) -> dict[str, object]:
    """Run the packaged compiler query only against materialized snapshot bytes."""
    snapshot = store.get_snapshot(snapshot_id)
    if (
        not isinstance(identifiers, (list, tuple))
        or not identifiers
        or len(identifiers) > MAX_TYPESCRIPT_QUERY_IDENTIFIERS
        or any(
            not isinstance(item, str)
            or len(item) > MAX_TYPESCRIPT_IDENTIFIER_CHARS
            or re.fullmatch(r"[A-Za-z_$][A-Za-z0-9_$]*", item) is None
            for item in identifiers
        )
    ):
        raise ValueError(
            "identifiers must be a bounded nonempty list of TypeScript identifier names"
        )
    if not isinstance(max_results, int) or isinstance(max_results, bool) or max_results < 1:
        raise ValueError("max_results must be a positive integer")
    names = sorted(set(identifiers))
    cap = min(max_results, 1_000)

    def unavailable(reason: str) -> dict[str, object]:
        return {
            "snapshot_id": snapshot.id,
            "available": False,
            "complete": False,
            "unknowns": [reason],
        }

    worker = _find_typescript_worker(store.root)
    node = _trusted_node_executable(store.root)
    if worker is None or node is None:
        return unavailable("The packaged TypeScript semantic worker or Node.js is unavailable.")
    compiler_files, compiler_bytes = _typescript_compiler_input_usage(snapshot)
    if (
        compiler_files > MAX_TYPESCRIPT_COMPILER_INPUT_FILES
        or compiler_bytes > MAX_TYPESCRIPT_COMPILER_INPUT_BYTES
    ):
        return unavailable(
            "The immutable snapshot exceeds the fixed TypeScript compiler input budget "
            f"({MAX_TYPESCRIPT_COMPILER_INPUT_FILES} files or "
            f"{MAX_TYPESCRIPT_COMPILER_INPUT_BYTES} bytes)."
        )
    try:
        with tempfile.TemporaryDirectory(prefix="byte-ts-snapshot-") as temporary:
            snapshot_root = Path(temporary) / "snapshot"
            store.materialize_snapshot(snapshot_id, snapshot_root)
            request = _canonical_json(
                {"root": str(snapshot_root), "identifiers": names, "maxResults": cap}
            )
            environment = {
                key: value
                for key, value in os.environ.items()
                if key.upper() in {"PATH", "SYSTEMROOT", "TEMP", "TMP", "TMPDIR", "WINDIR"}
            }
            process = _run_typescript_worker(
                [
                    node,
                    f"--max-old-space-size={TYPESCRIPT_WORKER_MAX_OLD_SPACE_MIB}",
                    str(worker),
                ],
                input=f"{request}\n",
                env=environment,
                cwd=snapshot_root,
                timeout=TYPESCRIPT_WORKER_TIMEOUT_SECONDS,
                output_limit=MAX_TYPESCRIPT_WORKER_OUTPUT_BYTES,
            )
    except subprocess.TimeoutExpired:
        return unavailable("The TypeScript semantic worker exceeded its execution deadline.")
    except _WorkerOutputLimit:
        return unavailable("The TypeScript semantic worker response exceeded the output limit.")
    except (OSError, RuntimeError, ValueError, subprocess.SubprocessError):
        return unavailable("The immutable snapshot could not be analyzed by the TypeScript semantic worker.")
    if process.returncode != 0:
        if _typescript_worker_out_of_memory(process):
            return unavailable(
                "The TypeScript semantic worker reached its fixed memory ceiling."
            )
        return unavailable("The TypeScript semantic worker exited unsuccessfully.")
    try:
        lines = [item for item in process.stdout.splitlines() if item.strip()]
        if len(lines) != 1:
            return unavailable("The TypeScript semantic worker returned an invalid response envelope.")
        envelope = json.loads(lines[0])
        result = envelope.get("result") if isinstance(envelope, dict) and envelope.get("ok") is True else None
    except json.JSONDecodeError:
        return unavailable("The TypeScript semantic worker returned invalid JSON.")
    if not _valid_typescript_boundary(result, names, snapshot, cap):
        return unavailable("The TypeScript semantic worker returned a malformed or mismatched query set.")
    return {**result, "snapshot_id": snapshot.id, "available": True}


def _typescript_compiler_input_usage(snapshot: Snapshot) -> tuple[int, int]:
    files = 0
    total_bytes = 0
    for entry in snapshot.files:
        if entry.kind != "file":
            continue
        path = PurePosixPath(entry.path)
        if (
            path.suffix.casefold() not in _TYPESCRIPT_COMPILER_SUFFIXES
            and path.name.casefold() not in _TYPESCRIPT_COMPILER_LOCK_NAMES
        ):
            continue
        files += 1
        total_bytes += entry.size
    return files, total_bytes


def _typescript_worker_out_of_memory(
    process: subprocess.CompletedProcess[str],
) -> bool:
    stderr = process.stderr.casefold() if isinstance(process.stderr, str) else ""
    return any(
        marker in stderr
        for marker in (
            "heap out of memory",
            "reached heap limit",
            "allocation failed - javascript heap",
        )
    )


class _BoundedWorkerOutput:
    def __init__(self, limit: int) -> None:
        self.limit = limit
        self.lock = threading.Lock()
        self.exceeded = threading.Event()
        self.failed = threading.Event()
        self.buffers = {"stdout": bytearray(), "stderr": bytearray()}
        self.retained = 0
        self.observed = 0
        self.error = ""

    def add(self, stream_name: str, chunk: bytes) -> bool:
        with self.lock:
            remaining = max(0, self.limit - self.retained)
            if remaining:
                kept = chunk[:remaining]
                self.buffers[stream_name].extend(kept)
                self.retained += len(kept)
            self.observed += len(chunk)
            if self.observed > self.limit and not self.exceeded.is_set():
                self.exceeded.set()
                return True
        return False

    def fail(self, error: BaseException) -> bool:
        with self.lock:
            if self.failed.is_set():
                return False
            self.error = str(error)
            self.failed.set()
            return True

    def text(self) -> tuple[str, str]:
        with self.lock:
            stdout = bytes(self.buffers["stdout"])
            stderr = bytes(self.buffers["stderr"])
        return (
            stdout.decode("utf-8", errors="replace"),
            stderr.decode("utf-8", errors="replace"),
        )


def _run_typescript_worker(
    command: Sequence[str],
    *,
    input: str,
    cwd: Path,
    env: Mapping[str, str],
    timeout: float,
    output_limit: int,
) -> subprocess.CompletedProcess[str]:
    """Run the semantic worker while retaining at most its fixed combined byte cap."""
    if output_limit <= 0:
        raise ValueError("semantic worker output limit must be positive")
    output = _BoundedWorkerOutput(output_limit)
    options: dict[str, object] = {
        "cwd": cwd,
        "env": dict(env),
        "stdin": subprocess.PIPE,
        "stdout": subprocess.PIPE,
        "stderr": subprocess.PIPE,
    }
    if os.name == "nt":
        options["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
    else:
        options["start_new_session"] = True
    process = subprocess.Popen(list(command), **options)

    def terminate() -> None:
        try:
            if os.name != "nt":
                os.killpg(process.pid, signal.SIGKILL)
            elif process.poll() is None:
                process.kill()
        except (OSError, ProcessLookupError):
            try:
                process.kill()
            except OSError:
                pass

    def drain(stream_name: str, stream) -> None:
        try:
            read = getattr(stream, "read1", stream.read)
            while chunk := read(64 * 1024):
                if output.add(stream_name, chunk):
                    terminate()
        except (OSError, ValueError) as error:
            if output.fail(error):
                terminate()

    def write_request(stream) -> None:
        try:
            stream.write(input.encode("utf-8"))
            stream.close()
        except BrokenPipeError:
            pass
        except (OSError, ValueError) as error:
            if output.fail(error):
                terminate()

    assert process.stdin is not None
    assert process.stdout is not None
    assert process.stderr is not None
    workers = [
        threading.Thread(
            target=drain,
            args=("stdout", process.stdout),
            name="byte-typescript-stdout",
            daemon=True,
        ),
        threading.Thread(
            target=drain,
            args=("stderr", process.stderr),
            name="byte-typescript-stderr",
            daemon=True,
        ),
        threading.Thread(
            target=write_request,
            args=(process.stdin,),
            name="byte-typescript-stdin",
            daemon=True,
        ),
    ]
    for worker in workers:
        worker.start()
    try:
        returncode = process.wait(timeout=timeout)
    except subprocess.TimeoutExpired:
        terminate()
        try:
            process.wait(timeout=5)
        except subprocess.TimeoutExpired:
            process.kill()
            process.wait(timeout=5)
        stdout, stderr = output.text()
        raise subprocess.TimeoutExpired(command, timeout, output=stdout, stderr=stderr)
    except BaseException:
        terminate()
        try:
            process.wait(timeout=5)
        except BaseException:
            pass
        raise
    finally:
        for worker in workers:
            worker.join(timeout=5)
        for stream in (process.stdin, process.stdout, process.stderr):
            try:
                stream.close()
            except (OSError, ValueError):
                pass
        for worker in workers:
            worker.join(timeout=1)

    if any(worker.is_alive() for worker in workers):
        terminate()
        raise RuntimeError("semantic worker pipe handler did not terminate")
    if output.exceeded.is_set():
        raise _WorkerOutputLimit(
            f"semantic worker combined stdout and stderr exceeded {output_limit} bytes"
        )
    if output.failed.is_set():
        raise RuntimeError(f"semantic worker output could not be observed safely: {output.error}")
    stdout, stderr = output.text()
    return subprocess.CompletedProcess(list(command), returncode, stdout, stderr)


def _valid_typescript_boundary(
    result: object, identifiers: list[str], snapshot: Snapshot, max_results: int
) -> bool:
    if (
        not isinstance(result, dict)
        or result.get("adapter") != "byte.typescript"
        or result.get("adapterVersion") != TYPESCRIPT_ADAPTER_VERSION
    ):
        return False
    query = result.get("query")
    result_domain = result.get("domain")
    consumers = result.get("consumers")
    definitions = result.get("definitions")
    if (
        not isinstance(query, dict)
        or query.get("operation") != "consumers"
        or query.get("identifiers") != identifiers
        or not isinstance(result_domain, dict)
        or not isinstance(consumers, list)
        or len(consumers) > max_results
        or not isinstance(definitions, list)
        or not isinstance(result.get("complete"), bool)
        or not isinstance(result.get("truncated"), bool)
        or result["complete"] and result["truncated"]
        or not isinstance(result.get("unknowns"), list)
        or any(not isinstance(item, str) for item in result["unknowns"])
    ):
        return False
    fingerprints = (
        result.get("queryFingerprint"),
        result.get("resultFingerprint"),
        result_domain.get("membershipFingerprint"),
        result_domain.get("resolutionFingerprint"),
    )
    if any(not isinstance(item, str) or not item for item in fingerprints):
        return False
    paths = {item.path for item in snapshot.files}
    configs = result_domain.get("configFiles")
    discovered_files = result_domain.get("discoveredFiles")
    domain_files = result_domain.get("files")
    expected_applicable = bool(configs) or (
        isinstance(discovered_files, int)
        and not isinstance(discovered_files, bool)
        and discovered_files > 0
    )
    if (
        not isinstance(configs, list)
        or any(not isinstance(item, str) or item not in paths for item in configs)
        or not isinstance(result_domain.get("applicable"), bool)
        or result_domain["applicable"] is not expected_applicable
        or not isinstance(discovered_files, int)
        or isinstance(discovered_files, bool)
        or discovered_files < 0
        or discovered_files > len(paths)
        or not isinstance(result_domain.get("discoveryComplete"), bool)
        or not isinstance(domain_files, int)
        or isinstance(domain_files, bool)
        or domain_files < 0
        or domain_files > len(paths)
        or result_domain.get("mode") not in ("configured", "inferred")
        or (result_domain["mode"] == "configured") is not bool(configs)
    ):
        return False
    for location in [*definitions, *consumers]:
        if not isinstance(location, dict):
            return False
        if not isinstance(location.get("path"), str):
            return False
        try:
            relative = _normalize_relative_path(location["path"])
        except (KeyError, ValueError, TypeError):
            return False
        if relative not in paths or location.get("name") not in identifiers:
            return False
        if not isinstance(location.get("kind"), str) or not location["kind"]:
            return False
        if any(
            not isinstance(location.get(key), int) or isinstance(location[key], bool) or location[key] < 1
            for key in ("line", "column")
        ):
            return False
    if not all(
        item["kind"] in {"call", "construct", "import", "type", "write", "reference"}
        and ("container" not in item or isinstance(item["container"], str))
        for item in consumers
    ):
        return False
    try:
        expected_query_fingerprint = _typescript_sha256(
            (
                "byte.typescript",
                TYPESCRIPT_ADAPTER_VERSION,
                "consumers",
                json.dumps(identifiers, ensure_ascii=False, separators=(",", ":")),
                result_domain["membershipFingerprint"],
                result_domain["resolutionFingerprint"],
            )
        )
        result_parts = [
            _typescript_location_json(item, consumer=False) for item in definitions
        ]
        result_parts.extend(
            _typescript_location_json(item, consumer=True) for item in consumers
        )
        expected_result_fingerprint = _typescript_sha256(result_parts)
    except (UnicodeError, ValueError, TypeError):
        return False
    return (
        result["queryFingerprint"] == expected_query_fingerprint
        and result["resultFingerprint"] == expected_result_fingerprint
    )


def _typescript_location_json(location: Mapping[str, object], *, consumer: bool) -> str:
    """Serialize one result exactly as the TypeScript worker's JSON.stringify call."""
    ordered: dict[str, object] = {
        "path": location["path"],
        "line": location["line"],
        "column": location["column"],
        "name": location["name"],
        "kind": location["kind"],
    }
    if consumer and "container" in location:
        ordered["container"] = location["container"]
    return json.dumps(
        ordered,
        ensure_ascii=False,
        separators=(",", ":"),
        allow_nan=False,
    )


def _typescript_sha256(parts: Sequence[str]) -> str:
    """Mirror TypeScript's length-prefixed SHA-256 over JavaScript strings.

    JavaScript ``String.length`` counts UTF-16 code units, not Unicode code
    points. The payload itself is then encoded as UTF-8 by Node's Hash.update.
    """
    digest = hashlib.sha256()
    for part in parts:
        if not isinstance(part, str):
            raise TypeError("TypeScript fingerprint parts must be strings")
        utf16_units = len(part.encode("utf-16-le")) // 2
        digest.update(str(utf16_units).encode("ascii"))
        digest.update(b":")
        digest.update(part.encode("utf-8"))
    return f"sha256:{digest.hexdigest()}"


def _find_typescript_worker(workspace_root: Path | None = None) -> Path | None:
    configured = os.environ.get("BYTE_TYPESCRIPT_SEMANTIC_WORKER")
    configured_path = Path(configured).expanduser() if configured else None
    candidates = [configured_path] if configured_path is not None and configured_path.is_absolute() else []
    for parent in Path(__file__).resolve().parents:
        candidates.extend(
            (
                parent / "bundle" / "typescript-semantic-worker.js",
                parent / "dist" / "bundle" / "typescript-semantic-worker.js",
                parent
                / "packages"
                / "coding-agent"
                / "dist"
                / "bundle"
                / "typescript-semantic-worker.js",
            )
        )
    for candidate in candidates:
        if not candidate.is_file():
            continue
        resolved = candidate.resolve()
        if workspace_root is not None:
            root = workspace_root.resolve()
            if resolved == root or root in resolved.parents:
                continue
        return resolved
    return None


def _trusted_node_executable(workspace_root: Path) -> str | None:
    """Resolve Node without accepting a repository-local executable shim."""
    configured = os.environ.get("BYTE_NODE_EXECUTABLE")
    if configured:
        unresolved = Path(configured).expanduser()
        if not unresolved.is_absolute():
            return None
    else:
        found = shutil.which("node")
        if found is None:
            return None
        unresolved = Path(found)
    try:
        candidate = unresolved.resolve(strict=True)
        root = workspace_root.resolve(strict=True)
    except OSError:
        return None
    if not candidate.is_file() or candidate == root or root in candidate.parents:
        return None
    if os.name != "nt" and not os.access(candidate, os.X_OK):
        return None
    return str(candidate)


def _evaluate_text_contains(
    store: KnowledgeStore,
    snapshot_id: str,
    parameters: Mapping[str, object],
    domain: tuple[str, ...],
) -> QueryEvaluation:
    needle = parameters.get("text")
    case_sensitive = parameters.get("case_sensitive", True)
    if not isinstance(needle, str) or not needle or not isinstance(case_sensitive, bool):
        return QueryEvaluation("", "", (), False)
    rows = store._connection.execute(
        """
        SELECT path, digest, queryable, blob_stored
        FROM snapshot_files
        WHERE snapshot_id = ?
        ORDER BY path
        """,
        (snapshot_id,),
    ).fetchall()
    domain_rows = [row for row in rows if _path_matches_domain(row["path"], domain)]
    domain_fingerprint = _fingerprint([row["path"] for row in domain_rows])
    results: list[str] = []
    result_material: list[str] = []
    complete = True
    search = needle if case_sensitive else needle.casefold()
    for row in domain_rows:
        if not row["queryable"] or not row["blob_stored"]:
            complete = False
            continue
        raw = store.read_snapshot_file(snapshot_id, row["path"])
        if raw is None:
            complete = False
            continue
        try:
            text = raw.decode("utf-8")
        except UnicodeDecodeError:
            complete = False
            continue
        haystack = text if case_sensitive else text.casefold()
        if search in haystack:
            results.append(row["path"])
            result_material.append(f"{row['path']}\0{row['digest']}")
    return QueryEvaluation(
        domain_fingerprint=domain_fingerprint,
        result_fingerprint=_fingerprint(result_material),
        results=tuple(results),
        complete=complete,
    )


def _manifest_from_json(payload: str) -> SnapshotManifest:
    data = json.loads(payload)
    if data.get("manifest_version") != MANIFEST_VERSION:
        raise RuntimeError(f"unsupported snapshot manifest: {data.get('manifest_version')}")
    files = tuple(
        SnapshotFile(
            path=item["path"],
            digest=item["digest"],
            size=int(item["size"]),
            kind=item["kind"],
            queryable=bool(item["queryable"]),
        )
        for item in data["files"]
    )
    return SnapshotManifest(
        analyzer_version=data["analyzer_version"],
        files=files,
        config_hashes=tuple((str(item[0]), str(item[1])) for item in data["config_hashes"]),
        lock_hashes=tuple((str(item[0]), str(item[1])) for item in data["lock_hashes"]),
    )


def _verified_snapshot_blob(entry: SnapshotFile, row: Mapping[str, object]) -> bytes:
    """Decode one stored blob only when its index metadata and content still match the manifest."""
    if (
        row["digest"] != entry.digest
        or row["size"] != entry.size
        or row["kind"] != entry.kind
        or bool(row["queryable"]) != entry.queryable
        or not bool(row["blob_stored"])
        or row["content"] is None
    ):
        raise RuntimeError(f"snapshot file integrity check failed: {entry.path}")
    return _decode_snapshot_blob(entry, row)


def _decode_snapshot_blob(entry: SnapshotFile, row: Mapping[str, object]) -> bytes:
    if row["content"] is None:
        raise RuntimeError(f"snapshot blob is unavailable: {entry.path}")
    if row["encoding"] != "zlib":
        raise RuntimeError(f"unsupported blob encoding: {row['encoding']}")
    compressed = bytes(row["content"])
    try:
        decoder = zlib.decompressobj()
        content = decoder.decompress(compressed, entry.size + 1)
        if decoder.unconsumed_tail or len(content) > entry.size:
            raise RuntimeError(f"snapshot blob exceeds its declared size: {entry.path}")
        content += decoder.flush()
    except zlib.error as error:
        raise RuntimeError(f"snapshot blob could not be decompressed: {entry.path}") from error
    if (
        not decoder.eof
        or decoder.unused_data
        or len(content) != entry.size
        or _sha256(content) != entry.digest
    ):
        raise RuntimeError(f"snapshot blob integrity check failed: {entry.path}")
    return content


def _evidence_from_row(row: sqlite3.Row) -> Evidence:
    return Evidence(
        id=row["id"],
        claim_id=row["claim_id"],
        snapshot_id=row["snapshot_id"],
        kind=row["kind"],
        payload=json.loads(row["payload_json"]),
        digest=row["digest"],
        created_at=row["created_at"],
    )


def _dependency_from_row(row: sqlite3.Row) -> Dependency:
    return Dependency(
        id=row["id"],
        created_snapshot_id=row["created_snapshot_id"],
        type=DependencyType(row["type"]),
        path=row["path"],
        expected_digest=row["expected_digest"],
        query_kind=row["query_kind"],
        parameters=json.loads(row["parameters_json"]),
        domain=tuple(json.loads(row["domain_json"])),
        domain_fingerprint=row["domain_fingerprint"],
        result_fingerprint=row["result_fingerprint"],
        results=tuple(json.loads(row["results_json"])),
        complete=bool(row["complete"]),
    )


def _event_from_row(row: sqlite3.Row) -> KnowledgeEvent:
    return KnowledgeEvent(
        cursor=int(row["cursor"]),
        kind=row["kind"],
        claim_id=row["claim_id"],
        snapshot_id=row["snapshot_id"],
        payload=json.loads(row["payload_json"]),
        created_at=row["created_at"],
    )


def _materialization_path(destination: Path, relative: str) -> Path:
    normalized = _normalize_relative_path(relative)
    target = destination.joinpath(*normalized.split("/"))
    try:
        target.resolve(strict=False).relative_to(destination)
    except ValueError as error:
        raise ValueError(f"snapshot path escapes destination: {relative!r}") from error
    return target


def _safe_symlink_target(destination: Path, link_path: Path, target: str) -> str:
    if "\0" in target:
        raise ValueError(f"symlink target contains a null byte: {target!r}")
    normalized = target.replace("\\", "/")
    posix_target = PurePosixPath(normalized)
    windows_target = PureWindowsPath(target)
    if posix_target.is_absolute() or windows_target.is_absolute() or windows_target.drive:
        raise ValueError(f"snapshot symlink target must be relative: {target!r}")
    candidate = link_path.parent.joinpath(*posix_target.parts).resolve(strict=False)
    try:
        candidate.relative_to(destination)
    except ValueError as error:
        raise ValueError(f"snapshot symlink target escapes destination: {target!r}") from error
    return normalized


def _normalize_relative_path(path: str) -> str:
    raw = str(path)
    normalized = raw.replace("\\", "/")
    windows_path = PureWindowsPath(raw)
    if PurePosixPath(normalized).is_absolute() or windows_path.is_absolute() or windows_path.drive:
        raise ValueError(f"path must be a normalized workspace-relative path: {path!r}")
    parts = normalized.split("/") if normalized else []
    if not parts or any(part in ("", ".", "..") for part in parts):
        raise ValueError(f"path must be a normalized workspace-relative path: {path!r}")
    return "/".join(parts)


def _bounded_limit(limit: int) -> int:
    return max(0, min(1000, int(limit)))


def _lexical_terms(text: str) -> set[str]:
    return {match.group(0).casefold() for match in re.finditer(r"[A-Za-z0-9_]+", text)}


def _normalize_domain(domain: Sequence[str]) -> tuple[str, ...]:
    patterns = tuple(dict.fromkeys(str(item).replace("\\", "/") for item in domain if str(item)))
    if not patterns:
        raise ValueError("query domain must contain at least one path pattern")
    return patterns


def _path_matches_domain(path: str, domain: Sequence[str]) -> bool:
    return any(pattern == "**/*" or fnmatch.fnmatchcase(path, pattern) for pattern in domain)


def _is_utf8_text(data: bytes) -> bool:
    if b"\0" in data[:8192]:
        return False
    try:
        data.decode("utf-8")
    except UnicodeDecodeError:
        return False
    return True


def _is_config(path: str) -> bool:
    name = path.rsplit("/", 1)[-1].lower()
    return name in _CONFIG_NAMES or name.startswith("tsconfig.") and name.endswith(".json")


def _is_lock(path: str) -> bool:
    name = path.rsplit("/", 1)[-1].lower()
    return name in _LOCK_NAMES or name.startswith("requirements") and name.endswith(".txt")


def _fingerprint(items: Sequence[str]) -> str:
    return _sha256(_canonical_json(list(items)).encode("utf-8"))


def _canonical_json(value: object) -> str:
    return json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":"), allow_nan=False)


def _sha256(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def _now() -> str:
    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")