# Search five approved media types in one local index > Check candidate paths against approved originals and use local receipts to recover interrupted indexing or search calls. By BIGCHANGE.AI Editorial Published: 2026-10-27T14:48:38.399Z Updated: 2026-10-27T14:48:38.399Z Canonical: https://bigchange.ai/blog/gemini-five-media-local-search-chroma-langchain-guide ![A landscape print beside an open laptop displaying a similar image, with a portable drive connected at right.](https://bigchange.ai/api/media/file/gemini-five-media-source-check-hero-v1.png) AI-generated conceptual illustration by BIG CHANGE. For a small collection, finding a clip, diagram or page can be harder than storing it. Index approved TXT, PNG, WAV, MP4 and PDF files together. A typed question returns candidate paths and hashes. You still open the files to decide whether the results answer it. ## The big change - **What changed:** One checked local collection holds the approved files' search vectors and source receipts. - **Why it matters:** The local receipts tie each candidate to the source bytes and make an interrupted indexing or query call inspectable. - **What to watch:** Retrieval quality, account charges and permission to send each file remain the collection owner's decisions. The project uses a hosted embedding call, local Chroma storage and a LangChain vector query. BIG CHANGE's program supplies the file validation, source binding and recovery rules. Its 25 offline tests passed with fake provider and store adapters. We did not install the pinned native packages, call a provider, index real media or measure retrieval accuracy. The documentation and package versions were checked on October 27, 2026. ## 1. Choose the collection and data route This is for a developer with a small, permissioned collection and authority over the account that will be billed. Use one private project folder. Put no more than 50 approved files directly in `data/`. Use UTF-8 text and one-page PDFs. Each file can be 1 through 8,000,000 bytes, inclusive. Text has an 8,000-character cap. Limit WAV to 180 seconds and MP4 to 120 seconds, with one video stream. Video audio is omitted. PDF content above 8,192 tokens may be truncated, so make the source unit short enough to inspect as a whole. These project and preflight bounds are not a spending limit. Review the [embedding manual](https://ai.google.dev/gemini-api/docs/embeddings), [current pricing and data-use table](https://ai.google.dev/gemini-api/docs/pricing) and [project rate limits](https://ai.google.dev/gemini-api/docs/rate-limits) before using real files. Confirm which files you may send, the project's tier and acceptable data terms. Keep the key in the process environment; do not put it in the project. Protect `data/` and `local-index/` under the same retention policy: the latter holds paths, titles, hashes, vectors and receipts. If an MP4 check crashes, inspect the operating system's temporary area too, because the check briefly copies validated video bytes there. **Result to aim for:** an explicit list of owned files and a permitted account route. If either is uncertain, keep the project at the offline `check` stage. ## 2. Build the project and run offline checks Create this tree on a local filesystem that supports same-directory hard links. Copy the four complete reader files from the end of this article into the named paths. The fenced blocks are the file contents, not commands to paste into a terminal. Start with an empty `local-index/`; keep code, input files and run receipts in their own locations. Run only one command process against this project at a time, so its single pending marker remains unambiguous. ```text five-media-search/ search.py test_search.py requirements.txt manifest.template.json manifest.json # your edited copy of the template data/ # your approved files local-index/ # created by the program; start empty ``` Use Python 3.11 or newer, below 4. From the project folder, create the virtual environment and use its interpreter for every later command. The following PowerShell commands avoid changing the machine's script execution policy: ```powershell py -3.11 -m venv .venv .\.venv\Scripts\python.exe --version .\.venv\Scripts\python.exe -m pip install -r requirements.txt ``` On macOS or Linux, use: ```sh python3 -m venv .venv ./.venv/bin/python --version ./.venv/bin/python -m pip install -r requirements.txt ``` If a new shell opens, return to the project directory and use the same interpreter path again. Confirm its printed version and record installed package versions. MP4 preflight also needs `ffprobe` from an FFmpeg distribution on `PATH`; check `ffprobe -version` if you include video. These are setup instructions, not a claim that BIG CHANGE installed the dependencies. Copy `manifest.template.json` to `manifest.json`, then edit it. Replace *each entire angle-bracket placeholder*, including its brackets, with a real value. `path` is one filename directly under `data/`, such as `approved-note.txt`, not a full path. Quote a command argument if your own path or question contains spaces. Give every file a stable opaque ID, a checked title, its actual source revision and a private reference to the permission decision. Add one entry per file. Do not copy the illustrative placeholder as a real receipt. The script rejects repeated IDs, missing files, symlinks, traversal and other file types. Run the local validation before setting a key: ```powershell .\.venv\Scripts\python.exe search.py check .\.venv\Scripts\python.exe -B -m unittest test_search.py ``` ```sh ./.venv/bin/python search.py check ./.venv/bin/python -B -m unittest test_search.py ``` The first command prints the current files' hashes without starting the index or making a paid request. Compare every title, filename, revision, permission reference and hash with the approved originals. The second command exercises synthetic adapter cases, including interrupted writes. It cannot prove native installation, a real model response or ranking quality. Repair an input mismatch, then run `check` again. Keep your manifest and source files still during an index or query run. ## 3. Index the approved files Set `GEMINI_API_KEY` in the process environment. Run the platform's interpreter-bound command: ```powershell .\.venv\Scripts\python.exe search.py index ``` ```sh ./.venv/bin/python search.py index ``` For a new input, the program records a pending run ID before the request. A returned vector is saved before the store write; then a store readback and final receipt complete the run. The terminal prints `indexed `, `unchanged ` or `metadata updated `, not a hash. For a newly indexed file, open its final `local-index/index-receipts/.json` and compare `metadata.id`, `metadata.path` and `metadata.sha256` with `check` and the approved original. For a metadata-only update, inspect the corresponding `local-index/metadata-receipts/` entry. An unchanged item need not make a new receipt. Text `title` edits re-embed; path, revision or permission-reference edits update local metadata when the embedding input stays the same. A file removed from the manifest stops appearing in current query results; its stored vector remains until an owner-approved retention action. **If the run stops:** read `local-index/index-pending.json`, its `runId`, `index-vectors/.json`, `index-receipts/` and the account usage record. A hidden `.stage` file is incomplete; inspect it, then use final receipts to choose recovery. With a saved vector and unchanged source binding, rerun `index` to finish storage or receipt work without a new embedding. A published `started` marker without a saved vector is an uncertain call. Review account usage and decide whether to abandon it before running: ```powershell .\.venv\Scripts\python.exe search.py resolve-index --run-id --ack-reviewed-account-usage ``` ```sh ./.venv/bin/python search.py resolve-index --run-id --ack-reviewed-account-usage ``` Substitute the whole `` token with the value you just read. If a saved vector belongs to a source or manifest that changed, review the mismatch and add `--abandon-saved-vector` to the same resolution command; do not reuse that vector against changed input. Resolution archives the pending record under `retired/` and records the decision. It does not undo a provider charge. An unstarted pending run can resume without a prior request. ## 4. Query, verify and recover Ask a short question you are permitted to send to the provider. For example, replace the quoted text with one of your real inspection questions: ```powershell .\.venv\Scripts\python.exe search.py query "" --count 5 ``` ```sh ./.venv/bin/python search.py query "" --count 5 ``` Replace the *whole* quoted placeholder, keeping the quote marks. The query path verifies current file bytes and the indexed collection, then uses LangChain to rank candidates. The result lists current IDs, titles, paths and SHA-256 hashes. Open each returned file and compare its bytes, path and content with the request. A high rank alone says only that the file is a candidate. The cache is tied to the question, count and actual corpus state; another count or changed corpus may need another embedding call. If a query stops, inspect `local-index/query-pending.json`, `query-vectors/.json` and `queries/`. With a saved vector and unchanged corpus, rerun the **identical** question and count. If `started` is present but no saved vector exists, review account usage before abandoning the uncertain call: ```powershell .\.venv\Scripts\python.exe search.py resolve-query --run-id --ack-reviewed-account-usage ``` ```sh ./.venv/bin/python search.py resolve-query --run-id --ack-reviewed-account-usage ``` If the corpus changed after a query vector was saved, inspect the binding, add `--abandon-saved-vector` to the resolution command, finish indexing the current corpus and ask again. Preserve the retired receipts. Do not blindly replay a call whose billing outcome is uncertain. For a useful acceptance check, index at least one approved example of each required type: TXT, PNG, WAV, MP4 and PDF. The collection owner should name a relevant original for a real question about each type. Run each question once, write down the known file's rank and inspect misses in the originals. If audio or video misses a detail, confirm the whole indexed file contains it; this program does not locate timestamped clips. If a long text or PDF misses a passage, consider an approved smaller source unit. Any changed bytes need a new index run before the result is interpretable. The result is a candidate locator for one operator. It has no answer generator, login system, shared server or automatic refresh. If you need multiuser access, design its authorization separately. For a local downloadable encoder prototype, see [our EmbeddingGemma 2 guide](https://bigchange.ai/blog/embeddinggemma-2-local-multimodal-search-guide). This project makes a different data-route and persistence choice. ## Complete reader files Create these files exactly as shown. `manifest.template.json` stays a template; copy and edit it for your own `manifest.json`. Keep source data and generated receipts outside these four code files. ### search.py ```python """Local five-media search with source-bound vectors and receipt-preserving recovery.""" from __future__ import annotations import argparse import hashlib import io import json import math import os from pathlib import Path import re import shutil import subprocess import sys import tempfile import uuid import wave HERE = Path(__file__).resolve().parent DATA = HERE / "data" STORE = HERE / "local-index" MANIFEST = HERE / "manifest.json" MODEL = "gemini-embedding-2" DIM = 768 FORMAT = "owned-source-v2-search-result-v1" COLLECTION = "owned_multimodal_v2" MIMES = {".txt": "text/plain", ".png": "image/png", ".wav": "audio/wav", ".mp4": "video/mp4", ".pdf": "application/pdf"} MAX_BYTES = 8_000_000 # Local bound, not the provider token limit. MAX_ITEMS = 50 def digest(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def canonical(obj: object) -> bytes: return (json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + "\n").encode() def read_json(path: Path): return json.loads(path.read_text(encoding="utf-8")) def write_stage(path: Path, payload: bytes) -> Path: """Fsync unique same-directory bytes before any final name becomes visible.""" fd, name = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".stage", dir=path.parent) stage = Path(name) try: with os.fdopen(fd, "wb") as f: f.write(payload) f.flush() os.fsync(f.fileno()) if stage.read_bytes() != payload: raise OSError("Staged bytes did not match the intended receipt") return stage except BaseException: stage.unlink(missing_ok=True) raise def atomic_new(path: Path, obj: object) -> None: """Publish only a complete receipt; same-volume hard link refuses overwrite.""" path.parent.mkdir(parents=True, exist_ok=True) stage = write_stage(path, canonical(obj)) try: os.link(stage, path) finally: stage.unlink(missing_ok=True) def record_once(path: Path, obj: object) -> None: if path.exists(): if read_json(path) != obj: raise RuntimeError(f"Conflicting receipt {path.name}; inspect it") else: atomic_new(path, obj) def retire(path: Path, run_id: str) -> None: archive = STORE / "retired" / f"{run_id}.json" archive.parent.mkdir(parents=True, exist_ok=True) if archive.exists(): raise RuntimeError("Run archive already exists; inspect it") os.replace(path, archive) # Preserve exact pending bytes. def source_unchanged(row: dict) -> None: if digest(MANIFEST.read_bytes()) != row["_manifestSha256"]: raise RuntimeError("Approved manifest changed; recheck before continuing") path = DATA / row["path"] if path.is_symlink() or not path.is_file() or path.resolve().parent != DATA.resolve(): raise RuntimeError("Approved source path changed or disappeared") if path.read_bytes() != row["_content"]: raise RuntimeError("Approved source bytes changed; preserve pending and recheck") def parse_manifest() -> list[dict]: manifest_bytes = MANIFEST.read_bytes() raw = json.loads(manifest_bytes) if not isinstance(raw, dict) or raw.get("model") != MODEL or raw.get("dimensions") != DIM: raise ValueError("Manifest model/dimensions differ from this index contract") items = raw.get("items") if not isinstance(items, list) or not 1 <= len(items) <= MAX_ITEMS: raise ValueError("Supply 1 to 50 permissioned items") ids, paths, rows = set(), set(), [] for item in items: if not isinstance(item, dict) or set(item) != { "id", "path", "title", "sourceRevision", "permissionReceipt"}: raise ValueError("Item fields do not match the printed contract") item_id, rel = item["id"], item["path"] if (not isinstance(item_id, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,60}", item_id) or item_id in ids or not isinstance(rel, str) or rel in paths): raise ValueError("Duplicate or unsafe ID/path") if any(not isinstance(item[k], str) or not item[k].strip() or "<" in item[k] for k in ("title", "sourceRevision", "permissionReceipt")): raise ValueError("Real title, revision and permission receipt required") p = Path(rel) if p.is_absolute() or ".." in p.parts or len(p.parts) != 1 or p.suffix.lower() not in MIMES: raise ValueError("Use one supported file directly under data/") file = DATA / p if file.is_symlink() or not file.is_file() or file.resolve().parent != DATA.resolve(): raise ValueError("Missing, linked or escaping source file") content = file.read_bytes() # This immutable buffer is the embedding input. if not content or len(content) > MAX_BYTES: raise ValueError("Empty or over local byte cap") mime = MIMES[p.suffix.lower()] if mime == "text/plain": body = content.decode("utf-8") if not body.strip() or len(body) > 8000: raise ValueError("Text must be nonempty and at most 8000 characters") elif mime == "image/png" and not content.startswith(b"\x89PNG\r\n\x1a\n"): raise ValueError("PNG signature mismatch") elif mime == "audio/wav": with wave.open(io.BytesIO(content), "rb") as audio: if audio.getframerate() <= 0 or audio.getnframes() / audio.getframerate() > 180: raise ValueError("WAV exceeds 180 seconds") elif mime == "video/mp4": if not shutil.which("ffprobe"): raise ValueError("ffprobe is required to check MP4 duration") with tempfile.NamedTemporaryFile(prefix="approved-media-", suffix=".mp4", delete=False) as tmp: tmp.write(content) probe_path = Path(tmp.name) try: probe = subprocess.run(["ffprobe", "-v", "error", "-show_entries", "format=duration:stream=codec_name,codec_type", "-of", "json", str(probe_path)], capture_output=True, text=True, timeout=20, check=True) finally: probe_path.unlink(missing_ok=True) details = json.loads(probe.stdout) duration = float(details["format"]["duration"]) if not math.isfinite(duration) or not 0 < duration <= 120: raise ValueError("MP4 duration is outside 0–120 seconds") codecs = [s["codec_name"] for s in details.get("streams", []) if s.get("codec_type") == "video"] if len(codecs) != 1 or codecs[0] not in {"h264", "hevc", "av1", "vp9"}: raise ValueError("MP4 must have one supported video codec") elif mime == "application/pdf": if not content.startswith(b"%PDF-"): raise ValueError("PDF signature mismatch") from pypdf import PdfReader if len(PdfReader(io.BytesIO(content)).pages) != 1: raise ValueError("This kit accepts one PDF page to reduce truncation risk") ids.add(item_id) paths.add(rel) rows.append({**item, "mime": mime, "sha256": digest(content), "bytes": len(content), "_content": content, "_manifestSha256": digest(manifest_bytes)}) return rows def input_key(row: dict) -> str: return digest(canonical({"sha256": row["sha256"], "mime": row["mime"], "textTitle": row["title"] if row["mime"] == "text/plain" else None, "model": MODEL, "dimensions": DIM, "format": FORMAT})) def metadata(row: dict) -> dict: return {"id": row["id"], "path": row["path"], "title": row["title"], "sourceRevision": row["sourceRevision"], "permissionSha256": digest(row["permissionReceipt"].encode()), "sha256": row["sha256"], "mime": row["mime"], "inputKey": input_key(row)} def vector_from_response(response: object) -> list[float]: embeddings = getattr(response, "embeddings", None) if not embeddings or len(embeddings) != 1: raise ValueError("Expected exactly one embedding") values = getattr(embeddings[0], "values", None) if not isinstance(values, (list, tuple)) or len(values) != DIM: raise ValueError("Embedding dimensions differ from collection") vector = [float(x) for x in values] if not all(math.isfinite(x) for x in vector): raise ValueError("Nonfinite embedding") return vector def developer_client(): key = os.environ.get("GEMINI_API_KEY", "") if not key.strip(): raise RuntimeError("Set an approved GEMINI_API_KEY before a paid request") if os.environ.get("GOOGLE_API_KEY") or os.environ.get("GOOGLE_GENAI_USE_VERTEXAI"): raise RuntimeError("Conflicting Google API/backend environment setting; clear it") from google import genai return genai.Client(api_key=key, vertexai=False) def embed(client: object, data: bytes, mime: str, title: str = "") -> list[float]: from google.genai import types if mime == "text/plain": contents = f"title: {title} | text: {data.decode('utf-8')}" elif mime == "query": contents = f"task: search result | query: {data.decode('utf-8')}" else: contents = [types.Part.from_bytes(data=data, mime_type=mime)] response = client.models.embed_content( model=MODEL, contents=contents, config=types.EmbedContentConfig(output_dimensionality=DIM)) return vector_from_response(response) def get_collection(): import chromadb STORE.mkdir(exist_ok=True) client = chromadb.PersistentClient(path=str(STORE / "chroma")) return client, client.get_or_create_collection(name=COLLECTION, embedding_function=None) def get_item(collection, item_id: str): found = collection.get(ids=[item_id], include=["metadatas", "embeddings"]) if not found["ids"]: return None vector = [float(x) for x in found["embeddings"][0]] if len(vector) != DIM or not all(math.isfinite(x) for x in vector): raise RuntimeError("Stored vector invalid") return found["metadatas"][0], vector def matches(found: dict, expected: dict) -> bool: return all(found.get(k) == v for k, v in expected.items()) def saved_vector(path: Path, pending: dict) -> list[float]: record = read_json(path) vector = record.get("vector") if (record.get("binding") != pending or not isinstance(vector, list) or len(vector) != DIM or not all(isinstance(x, (float, int)) and math.isfinite(x) for x in vector)): raise RuntimeError("Saved vector/binding invalid") return vector def mark_started(path: Path, pending: dict) -> dict: started = {**pending, "started": True} if read_json(path) == started: return started if read_json(path) != pending: raise RuntimeError("Pending marker changed; inspect before a paid call") stage = write_stage(path, canonical(started)) try: if read_json(path) != pending: raise RuntimeError("Pending marker changed during staging") os.replace(stage, path) finally: stage.unlink(missing_ok=True) if read_json(path) != started: raise RuntimeError("Started marker readback failed; inspect before a paid call") return started def index(rows: list[dict]) -> None: _, collection = get_collection() path = STORE / "index-pending.json" pending = read_json(path) if path.exists() else None by_id = {r["id"]: r for r in rows} if pending and (pending.get("kind") != "index" or pending.get("id") not in by_id or pending.get("inputKey") != input_key(by_id[pending["id"]])): raise RuntimeError("Pending index source/contract changed; resolve saved run") ordered = sorted(rows, key=lambda r: r["id"] != pending["id"]) if pending else rows for row in ordered: if pending and pending["id"] != row["id"]: raise RuntimeError("Pending index run must finish first") current = get_item(collection, row["id"]) expected = metadata(row) if not pending and current and current[0].get("inputKey") == input_key(row): source_unchanged(row) if not matches(current[0], expected): collection.upsert(ids=[row["id"]], embeddings=[current[1]], documents=[row["title"]], metadatas=[expected]) fresh = get_item(collection, row["id"]) if (not fresh or not matches(fresh[0], expected) or digest(canonical(fresh[1])) != digest(canonical(current[1]))): raise RuntimeError("Metadata readback failed") source_unchanged(row) record_once(STORE / "metadata-receipts" / f"{row['id']}-{digest(canonical(expected))}.json", {"id": row["id"], "metadata": expected, "status": "updated-without-provider-call"}) print(f"metadata updated {row['id']}") else: print(f"unchanged {row['id']}") continue binding = {"kind": "index", "id": row["id"], "inputKey": input_key(row), "manifestSha256": row["_manifestSha256"]} if pending: if {k: pending.get(k) for k in binding} != binding: raise RuntimeError("Pending index binding changed; resolve saved run") else: source_unchanged(row) pending = {**binding, "runId": uuid.uuid4().hex} atomic_new(path, pending) vector_path = STORE / "index-vectors" / f"{pending['runId']}.json" if vector_path.exists(): vector = saved_vector(vector_path, pending) elif pending.get("started"): raise RuntimeError("Uncertain paid index call; inspect usage and resolve-index") else: client = developer_client() # No journaled paid call before route validation. source_unchanged(row) pending = mark_started(path, pending) vector = embed(client, row["_content"], row["mime"], row["title"]) atomic_new(vector_path, {"binding": pending, "vector": vector}) source_unchanged(row) collection.upsert(ids=[row["id"]], embeddings=[vector], documents=[row["title"]], metadatas=[expected]) fresh = get_item(collection, row["id"]) if (not fresh or not matches(fresh[0], expected) or any(not math.isclose(a, b, rel_tol=1e-5, abs_tol=1e-6) for a, b in zip(fresh[1], vector))): raise RuntimeError("Collection readback did not match approved source") source_unchanged(row) record_once(STORE / "index-receipts" / f"{pending['runId']}.json", {"binding": pending, "metadata": expected, "vectorSha256": digest(canonical(vector)), "status": "indexed"}) retire(path, pending["runId"]) pending = None print(f"indexed {row['id']}") def corpus_binding(rows: list[dict], collection) -> str: if (STORE / "index-pending.json").exists(): raise RuntimeError("Resolve pending index call before query") items = [] for row in sorted(rows, key=lambda r: r["id"]): source_unchanged(row) current = get_item(collection, row["id"]) if not current or not matches(current[0], metadata(row)): raise RuntimeError("Index incomplete or stale; run index before query") items.append({"id": row["id"], "metadata": metadata(row), "storedVectorSha256": digest(canonical(current[1]))}) return digest(canonical({"model": MODEL, "dimensions": DIM, "format": FORMAT, "items": items})) def search_by_vector(vector: list[float], collection, k: int): from langchain_chroma import Chroma from langchain_core.documents import Document import chromadb lc = Chroma(client=chromadb.PersistentClient(path=str(STORE / "chroma")), collection_name=COLLECTION, embedding_function=None) found = lc.similarity_search_by_vector(embedding=vector, k=k) if not all(isinstance(doc, Document) for doc in found): raise RuntimeError("Unexpected LangChain search result") return found def query(rows: list[dict], text: str, count: int) -> dict: if not text.strip() or len(text) > 1000 or not 1 <= count <= 10: raise ValueError("Use a nonempty query of at most 1000 characters and 1–10 results") _, collection = get_collection() corpus = corpus_binding(rows, collection) # Always before cache read. binding = {"querySha256": digest(text.encode()), "count": count, "model": MODEL, "dimensions": DIM, "format": FORMAT, "corpusSha256": corpus} result_path = STORE / "queries" / f"{digest(canonical(binding))}.json" path = STORE / "query-pending.json" pending = read_json(path) if path.exists() else None if pending and (pending.get("kind") != "query" or pending.get("binding") != binding): raise RuntimeError("Different pending query; resolve saved run first") if result_path.exists() and not pending: result = read_json(result_path) if result.get("binding") != binding: raise RuntimeError("Cached query binding differs") current = {r["id"]: r for r in rows} hits = result.get("results") if (not isinstance(hits, list) or len(hits) > count or any(not isinstance(hit, dict) or hit.get("id") not in current or hit.get("sha256") != current[hit["id"]]["sha256"] or hit.get("path") != current[hit["id"]]["path"] or hit.get("title") != current[hit["id"]]["title"] for hit in hits)): raise RuntimeError("Cached query results differ from current approved sources") if corpus_binding(rows, collection) != corpus: raise RuntimeError("Corpus changed during cache read") print(json.dumps(result, indent=2)) return result if not pending: pending = {"kind": "query", "binding": binding, "runId": uuid.uuid4().hex} atomic_new(path, pending) vector_path = STORE / "query-vectors" / f"{pending['runId']}.json" if vector_path.exists(): vector = saved_vector(vector_path, pending) elif pending.get("started"): raise RuntimeError("Uncertain paid query call; inspect usage and resolve-query") else: client = developer_client() pending = mark_started(path, pending) vector = embed(client, text.encode(), "query") atomic_new(vector_path, {"binding": pending, "vector": vector}) if corpus_binding(rows, collection) != corpus: raise RuntimeError("Corpus changed during query; preserve pending") if result_path.exists(): result = read_json(result_path) if result.get("binding") != binding: raise RuntimeError("Saved result binding differs") else: current = {r["id"]: r for r in rows} ranked = search_by_vector(vector, collection, collection.count()) results = [] for doc in ranked: md = doc.metadata row = current.get(md.get("id")) if row and md.get("sha256") == row["sha256"] and md.get("inputKey") == input_key(row): results.append({"id": row["id"], "title": row["title"], "path": row["path"], "sha256": row["sha256"]}) if len(results) >= count: break result = {"binding": binding, "results": results, "note": "Vector proximity locates candidates; inspect the originals."} atomic_new(result_path, result) if corpus_binding(rows, collection) != corpus: raise RuntimeError("Corpus changed before query retirement; preserve pending") retire(path, pending["runId"]) print(json.dumps(result, indent=2)) return result def resolve(kind: str, run_id: str, abandon_saved_vector: bool = False) -> None: path = STORE / f"{kind}-pending.json" if not path.exists(): raise RuntimeError("No pending run to resolve") pending = read_json(path) if pending.get("kind") != kind or pending.get("runId") != run_id: raise RuntimeError("Run ID/kind does not match pending receipt") vector_path = STORE / f"{kind}-vectors" / f"{run_id}.json" if vector_path.exists() and not abandon_saved_vector: raise RuntimeError("Saved vector exists; resume or explicitly archive it") record_once(STORE / "abandonments" / f"{run_id}.json", {"kind": kind, "runId": run_id, "pendingSha256": digest(path.read_bytes()), "savedVectorSha256": digest(vector_path.read_bytes()) if vector_path.exists() else None, "decision": "operator-reviewed-abandonment"}) retire(path, run_id) print(f"Archived {kind} run {run_id}; a new request may incur another charge") def main() -> None: parser = argparse.ArgumentParser() sub = parser.add_subparsers(dest="command", required=True) sub.add_parser("check", help="Validate local files; no provider or index") sub.add_parser("index", help="Embed changed inputs or resume saved vectors") q = sub.add_parser("query", help="Search the current, fully indexed corpus") q.add_argument("text") q.add_argument("--count", type=int, default=5) for name in ("resolve-index", "resolve-query"): p = sub.add_parser(name, help="Archive a reviewed pending call without deleting its receipt") p.add_argument("--run-id", required=True) p.add_argument("--ack-reviewed-account-usage", action="store_true", required=True) p.add_argument("--abandon-saved-vector", action="store_true") args = parser.parse_args() if args.command.startswith("resolve-"): resolve(args.command.split("-")[1], args.run_id, args.abandon_saved_vector) return rows = parse_manifest() if args.command == "check": print(json.dumps({"model": MODEL, "dimensions": DIM, "items": [{k: v for k, v in r.items() if not k.startswith("_")} for r in rows]}, indent=2)) elif args.command == "index": index(rows) else: query(rows, args.text, args.count) if __name__ == "__main__": try: main() except Exception as exc: print(f"HOLD: {exc}", file=sys.stderr) raise SystemExit(1) ``` ### test\_search.py ```python """Offline lifecycle regressions with fake provider/store; no install or network.""" import contextlib import io import json from pathlib import Path import sys import tempfile import types import unittest import wave from unittest.mock import patch import search as s TEXT_NAME = "note.txt" RENAMED_NAME = "renamed.txt" SECOND_NAME = "second.txt" PICTURE = "picture.png" SOUND = "sound.wav" CLIP = "clip.mp4" PAGE = "page.pdf" class Collection: def __init__(self): self.items = {} self.fail_upsert = False def get(self, ids, include): found = [x for x in ids if x in self.items] return {"ids": found, "metadatas": [self.items[x][0] for x in found], "embeddings": [self.items[x][1] for x in found]} def upsert(self, ids, embeddings, documents, metadatas): if self.fail_upsert: self.fail_upsert = False raise RuntimeError("simulated store interruption") for i, vector, doc, md in zip(ids, embeddings, documents, metadatas): self.items[i] = (md, vector, doc) def count(self): return len(self.items) class Lifecycle(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) root = Path(self.temp.name) (root / "data").mkdir() (root / "local-index").mkdir() self.patches = [patch.object(s, "DATA", root / "data"), patch.object(s, "STORE", root / "local-index"), patch.object(s, "MANIFEST", root / "manifest.json")] for p in self.patches: p.start() self.addCleanup(p.stop) self.items = [{"id": "note1", "path": TEXT_NAME, "title": "Checked title", "sourceRevision": "rev1", "permissionReceipt": "local-approval"}] (s.DATA / TEXT_NAME).write_bytes(b"first approved bytes") self.write_manifest() self.collection = Collection() p = patch.object(s, "get_collection", return_value=(None, self.collection)) p.start(); self.addCleanup(p.stop) p = patch.object(s, "developer_client", return_value=object()) p.start(); self.addCleanup(p.stop) self.sent = [] def fake_embed(client, data, mime, title=""): self.sent.append((bytes(data), mime, title)) return [float(len(self.sent)) / 10] * s.DIM p = patch.object(s, "embed", side_effect=fake_embed) p.start(); self.addCleanup(p.stop) def fake_search(vector, collection, k): return [types.SimpleNamespace(metadata=x[0]) for _, x in sorted(collection.items.items())][:k] p = patch.object(s, "search_by_vector", side_effect=fake_search) p.start(); self.addCleanup(p.stop) def write_manifest(self): s.MANIFEST.write_text(json.dumps({"model": s.MODEL, "dimensions": s.DIM, "items": self.items}), encoding="utf-8") def run_index(self): with contextlib.redirect_stdout(io.StringIO()): s.index(s.parse_manifest()) def run_query(self, text="real question", count=5): with contextlib.redirect_stdout(io.StringIO()): return s.query(s.parse_manifest(), text, count) def test_check_manifest_and_cli_no_provider(self): with patch.object(sys, "argv", ["search.py", "check"]), contextlib.redirect_stdout(io.StringIO()) as out: s.main() record = json.loads(out.getvalue()) self.assertEqual(record["items"][0]["sha256"], s.digest(b"first approved bytes")) self.assertNotIn("_content", out.getvalue()) self.assertEqual(self.sent, []) def test_reject_traversal_duplicate_missing_permission(self): self.items[0]["path"] = "../other.txt"; self.write_manifest() with self.assertRaises(ValueError): s.parse_manifest() self.items[0]["path"] = TEXT_NAME self.items.append(dict(self.items[0])); self.write_manifest() with self.assertRaises(ValueError): s.parse_manifest() self.items.pop(); self.items[0]["permissionReceipt"] = ""; self.write_manifest() with self.assertRaises(ValueError): s.parse_manifest() def test_changed_bytes_between_check_and_send_hold(self): rows = s.parse_manifest() (s.DATA / TEXT_NAME).write_bytes(b"changed after check") with self.assertRaisesRegex(RuntimeError, "source bytes changed"): with contextlib.redirect_stdout(io.StringIO()): s.index(rows) self.assertEqual(self.sent, []) self.assertFalse((s.STORE / "index-pending.json").exists()) def test_changed_manifest_between_check_and_send_hold(self): rows = s.parse_manifest() self.items[0]["title"] = "new title"; self.write_manifest() with self.assertRaisesRegex(RuntimeError, "manifest changed"): with contextlib.redirect_stdout(io.StringIO()): s.index(rows) self.assertEqual(self.sent, []) def test_after_api_source_mutation_preserves_vector_and_requires_archive(self): def mutate(client, data, mime, title=""): self.sent.append((bytes(data), mime, title)) (s.DATA / TEXT_NAME).write_bytes(b"later bytes") return [0.1] * s.DIM with patch.object(s, "embed", side_effect=mutate): with self.assertRaisesRegex(RuntimeError, "source bytes changed"): self.run_index() pending = s.read_json(s.STORE / "index-pending.json") self.assertEqual(self.sent[0][0], b"first approved bytes") self.assertTrue((s.STORE / "index-vectors" / f"{pending['runId']}.json").exists()) with self.assertRaisesRegex(RuntimeError, "Pending index source"): self.run_index() with self.assertRaisesRegex(RuntimeError, "Saved vector exists"): s.resolve("index", pending["runId"]) with contextlib.redirect_stdout(io.StringIO()): s.resolve("index", pending["runId"], abandon_saved_vector=True) self.assertTrue((s.STORE / "retired" / f"{pending['runId']}.json").exists()) self.run_index() self.assertEqual(self.collection.items["note1"][0]["sha256"], s.digest(b"later bytes")) def test_corpus_count_and_source_bound_cache(self): self.run_index() five = self.run_query(count=5) self.assertEqual(len(five["results"]), 1) calls = len(self.sent) self.assertEqual(self.run_query(count=5), five) self.assertEqual(len(self.sent), calls) # Exact current cache hit. one = self.run_query(count=1) self.assertEqual(one["binding"]["count"], 1) self.assertEqual(len(self.sent), calls + 1) (s.DATA / TEXT_NAME).write_bytes(b"new current bytes") with self.assertRaisesRegex(RuntimeError, "Index incomplete or stale"): self.run_query(count=5) self.run_index() new = self.run_query(count=5) self.assertNotEqual(new["binding"]["corpusSha256"], five["binding"]["corpusSha256"]) self.assertEqual(new["results"][0]["sha256"], s.digest(b"new current bytes")) def test_stored_vector_change_invalidates_cache_and_tampered_hit_holds(self): self.run_index() first = self.run_query() md, vector, doc = self.collection.items["note1"] self.collection.items["note1"] = (md, [0.9] * s.DIM, doc) second = self.run_query() self.assertNotEqual(first["binding"]["corpusSha256"], second["binding"]["corpusSha256"]) cache = next(p for p in (s.STORE / "queries").glob("*.json") if s.read_json(p)["binding"] == second["binding"]) bad = s.read_json(cache) bad["results"][0]["sha256"] = "wrong" cache.write_text(json.dumps(bad), encoding="utf-8") with self.assertRaisesRegex(RuntimeError, "Cached query results differ"): self.run_query() def test_text_title_reembeds_but_provenance_and_path_refresh_without_api(self): self.run_index(); calls = len(self.sent) self.items[0].update(path=RENAMED_NAME, sourceRevision="rev2", permissionReceipt="second-approval") (s.DATA / TEXT_NAME).rename(s.DATA / RENAMED_NAME) self.write_manifest(); self.run_index() self.assertEqual(len(self.sent), calls) md = self.collection.items["note1"][0] self.assertEqual((md["path"], md["sourceRevision"]), (RENAMED_NAME, "rev2")) self.items[0]["title"] = "Corrected title"; self.write_manifest(); self.run_index() self.assertEqual(len(self.sent), calls + 1) self.assertEqual(self.sent[-1][2], "Corrected title") self.assertEqual(self.collection.items["note1"][0]["title"], "Corrected title") def test_embedding_format_or_model_change_requires_new_vector(self): self.run_index(); calls = len(self.sent) with patch.object(s, "FORMAT", "new-format"): self.run_index() self.assertEqual(len(self.sent), calls + 1) with patch.object(s, "MODEL", "new-model"): self.write_manifest() self.run_index() self.assertEqual(len(self.sent), calls + 2) def test_removed_and_readded_id_does_not_return_retired_row(self): (s.DATA / SECOND_NAME).write_bytes(b"second owned bytes") second = {**self.items[0], "id": "note2", "path": SECOND_NAME, "title": "Second"} self.items.append(second); self.write_manifest(); self.run_index() both = self.run_query() self.assertEqual({x["id"] for x in both["results"]}, {"note1", "note2"}) self.items.remove(second); self.write_manifest() only = self.run_query() self.assertEqual([x["id"] for x in only["results"]], ["note1"]) calls = len(self.sent) self.items.append(second); self.write_manifest(); self.run_index() self.assertEqual(len(self.sent), calls) again = self.run_query() self.assertEqual({x["id"] for x in again["results"]}, {"note1", "note2"}) def test_new_manifest_item_holds_query_until_indexed(self): self.run_index() (s.DATA / SECOND_NAME).write_bytes(b"other approved bytes") self.items.append({**self.items[0], "id": "note2", "path": SECOND_NAME}) self.write_manifest() with self.assertRaisesRegex(RuntimeError, "Index incomplete or stale"): self.run_query() self.run_index() self.assertEqual(len(self.run_query()["results"]), 2) def test_store_interruption_resumes_saved_vector_without_paid_replay(self): self.collection.fail_upsert = True with self.assertRaisesRegex(RuntimeError, "store interruption"): self.run_index() calls = len(self.sent) self.run_index() self.assertEqual(len(self.sent), calls) self.assertFalse((s.STORE / "index-pending.json").exists()) self.assertEqual(len(list((s.STORE / "index-receipts").glob("*.json"))), 1) def test_index_receipt_save_and_retirement_resumable(self): real = s.atomic_new def fail_receipt(path, obj): if path.parent.name == "index-receipts": raise RuntimeError("receipt-save interruption") return real(path, obj) with patch.object(s, "atomic_new", side_effect=fail_receipt): with self.assertRaisesRegex(RuntimeError, "receipt-save interruption"): self.run_index() calls = len(self.sent) self.run_index() self.assertEqual(len(self.sent), calls) self.assertEqual(len(list((s.STORE / "index-receipts").glob("*.json"))), 1) (s.DATA / TEXT_NAME).write_bytes(b"new approved version") real_retire = s.retire first = [True] def fail_retire(path, run_id): if first[0]: first[0] = False; raise RuntimeError("index retirement interruption") return real_retire(path, run_id) with patch.object(s, "retire", side_effect=fail_retire): with self.assertRaisesRegex(RuntimeError, "index retirement interruption"): self.run_index() calls = len(self.sent) self.run_index() self.assertEqual(len(self.sent), calls) self.assertFalse((s.STORE / "index-pending.json").exists()) def test_uncertain_index_call_hold_then_receipt_preserving_archive(self): real = s.atomic_new def fail_vector(path, obj): if path.parent.name == "index-vectors": raise RuntimeError("vector-save interruption") return real(path, obj) with patch.object(s, "atomic_new", side_effect=fail_vector): with self.assertRaisesRegex(RuntimeError, "vector-save interruption"): self.run_index() pending = s.read_json(s.STORE / "index-pending.json") with self.assertRaisesRegex(RuntimeError, "Uncertain paid index"): self.run_index() with contextlib.redirect_stdout(io.StringIO()): s.resolve("index", pending["runId"]) self.assertTrue((s.STORE / "retired" / f"{pending['runId']}.json").exists()) self.assertTrue((s.STORE / "abandonments" / f"{pending['runId']}.json").exists()) def test_query_result_save_and_retirement_resumable(self): self.run_index() real = s.atomic_new def fail_result(path, obj): if path.parent.name == "queries": raise RuntimeError("result-save interruption") return real(path, obj) with patch.object(s, "atomic_new", side_effect=fail_result): with self.assertRaisesRegex(RuntimeError, "result-save interruption"): self.run_query() calls = len(self.sent) self.run_query() self.assertEqual(len(self.sent), calls) self.assertFalse((s.STORE / "query-pending.json").exists()) real_retire = s.retire first = [True] def fail_retire(path, run_id): if first[0]: first[0] = False; raise RuntimeError("retirement interruption") return real_retire(path, run_id) with patch.object(s, "retire", side_effect=fail_retire): with self.assertRaisesRegex(RuntimeError, "retirement interruption"): self.run_query("another question") calls = len(self.sent) self.run_query("another question") self.assertEqual(len(self.sent), calls) def test_query_uncertain_call_can_be_archived_without_deleting_receipt(self): self.run_index() real = s.atomic_new def fail_vector(path, obj): if path.parent.name == "query-vectors": raise RuntimeError("query-vector interruption") return real(path, obj) with patch.object(s, "atomic_new", side_effect=fail_vector): with self.assertRaisesRegex(RuntimeError, "query-vector interruption"): self.run_query() pending = s.read_json(s.STORE / "query-pending.json") with self.assertRaisesRegex(RuntimeError, "Uncertain paid query"): self.run_query() with contextlib.redirect_stdout(io.StringIO()): s.resolve("query", pending["runId"]) self.assertTrue((s.STORE / "retired" / f"{pending['runId']}.json").exists()) def test_all_five_media_preflight_and_query_path(self): (s.DATA / PICTURE).write_bytes(b"\x89PNG\r\n\x1a\n" + b"small image") with wave.open(str(s.DATA / SOUND), "wb") as f: f.setnchannels(1); f.setsampwidth(2); f.setframerate(8000); f.writeframes(b"\0\0" * 80) (s.DATA / CLIP).write_bytes(b"fake MP4 preflight only") (s.DATA / PAGE).write_bytes(b"%PDF-fake preflight only") for name in (PICTURE, SOUND, CLIP, PAGE): self.items.append({"id": name.split(".")[0], "path": name, "title": name, "sourceRevision": "rev1", "permissionReceipt": "local-approval"}) self.write_manifest() pdf = types.ModuleType("pypdf") pdf.PdfReader = lambda path: types.SimpleNamespace(pages=[object()]) probe = types.SimpleNamespace(stdout=json.dumps({"format": {"duration": "1.0"}, "streams": [{"codec_type": "video", "codec_name": "h264"}]})) with patch.dict(sys.modules, {"pypdf": pdf}), patch.object(s.shutil, "which", return_value="ffprobe"), \ patch.object(s.subprocess, "run", return_value=probe): rows = s.parse_manifest() with contextlib.redirect_stdout(io.StringIO()): s.index(rows) with contextlib.redirect_stdout(io.StringIO()): result = s.query(rows, "five media", 5) self.assertEqual(len(rows), 5) self.assertEqual(len(result["results"]), 5) self.assertEqual({x[1] for x in self.sent[:5]}, set(s.MIMES.values())) calls = len(self.sent) next(x for x in self.items if x["id"] == "picture")["title"] = "Corrected image title" self.write_manifest() with patch.dict(sys.modules, {"pypdf": pdf}), patch.object(s.shutil, "which", return_value="ffprobe"), \ patch.object(s.subprocess, "run", return_value=probe): self.run_index() self.assertEqual(len(self.sent), calls) # Binary title is retrieval metadata only. class Route(unittest.TestCase): def test_client_explicit_backend_and_conflict_rejection(self): args = [] fake = types.ModuleType("google") fake.genai = types.SimpleNamespace(Client=lambda **kw: args.append(kw) or object()) with patch.dict(sys.modules, {"google": fake}), \ patch.dict(s.os.environ, {"GEMINI_API_KEY": "private-test-key"}, clear=True): s.developer_client() self.assertEqual(args, [{"api_key": "private-test-key", "vertexai": False}]) s.os.environ["GOOGLE_API_KEY"] = "other-secret" with self.assertRaisesRegex(RuntimeError, "Conflicting Google") as exc: s.developer_client() self.assertNotIn("other-secret", str(exc.exception)) del s.os.environ["GOOGLE_API_KEY"] s.os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "true" with self.assertRaisesRegex(RuntimeError, "Conflicting Google"): s.developer_client() def test_embed_uses_expected_sdk_contract_without_provider(self): seen = [] class Part: @staticmethod def from_bytes(**kwargs): seen.append(("part", kwargs)) return ("part", kwargs) class Config: def __init__(self, **kwargs): seen.append(("config", kwargs)) self.kwargs = kwargs class Models: def embed_content(self, **kwargs): seen.append(("embed", kwargs)) return types.SimpleNamespace(embeddings=[types.SimpleNamespace(values=[0.1]*s.DIM)]) sdk = types.ModuleType("google.genai") sdk.types = types.SimpleNamespace(Part=Part, EmbedContentConfig=Config) google = types.ModuleType("google") google.genai = sdk with patch.dict(sys.modules, {"google": google, "google.genai": sdk}): vector = s.embed(types.SimpleNamespace(models=Models()), b"image-bytes", "image/png") self.assertEqual(len(vector), s.DIM) self.assertEqual(seen[0], ("part", {"data": b"image-bytes", "mime_type": "image/png"})) self.assertEqual(seen[1], ("config", {"output_dimensionality": s.DIM})) self.assertEqual(seen[2][1]["model"], s.MODEL) class StagedFailure(unittest.TestCase): """Interrupt actual staged writes and marker publication, then use normal reruns.""" setUp = Lifecycle.setUp write_manifest = Lifecycle.write_manifest run_index = Lifecycle.run_index run_query = Lifecycle.run_query def test_partial_query_result_stage_recovers_without_paid_replay(self): self.run_index() real = s.write_stage def partial(path, payload): if path.parent.name == "queries": path.parent.mkdir(parents=True, exist_ok=True) stage = path.parent / ".partial-test.stage" with stage.open("xb") as f: f.write(payload[:11]); f.flush(); s.os.fsync(f.fileno()) raise OSError("simulated partial stage write") return real(path, payload) with patch.object(s, "write_stage", side_effect=partial): with self.assertRaisesRegex(OSError, "partial stage"): self.run_query() self.assertEqual(list((s.STORE / "queries").glob("*.json")), []) calls = len(self.sent) result = self.run_query() self.assertEqual(len(self.sent), calls) self.assertEqual(len(result["results"]), 1) def test_partial_atomic_new_write_never_exposes_final(self): path = s.STORE / "index-receipts" / "receipt.json" real_fdopen = s.os.fdopen class Interrupted: def __init__(self, file): self.file = file def __enter__(self): return self def __exit__(self, *args): self.file.close() def write(self, data): self.file.write(data[:7]); self.file.flush(); s.os.fsync(self.file.fileno()) raise OSError("disk-full after partial bytes") def flush(self): self.file.flush() def fileno(self): return self.file.fileno() with patch.object(s.os, "fdopen", side_effect=lambda fd, mode: Interrupted(real_fdopen(fd, mode))): with self.assertRaisesRegex(OSError, "disk-full"): s.atomic_new(path, {"binding": "approved"}) self.assertFalse(path.exists()) self.assertEqual(list(path.parent.glob("*.stage")), []) s.atomic_new(path, {"binding": "approved"}) self.assertEqual(s.read_json(path), {"binding": "approved"}) with self.assertRaises(FileExistsError): s.atomic_new(path, {"binding": "other"}) self.assertEqual(s.read_json(path), {"binding": "approved"}) def test_partial_index_pending_stage_does_not_start_or_poison_run(self): real_fdopen = s.os.fdopen class Partial: def __init__(self, f): self.f = f def __enter__(self): return self def __exit__(self, *args): self.f.close() def write(self, data): self.f.write(data[:5]); self.f.flush() raise OSError("partial pending stage") def flush(self): self.f.flush() def fileno(self): return self.f.fileno() with patch.object(s.os, "fdopen", side_effect=lambda fd, mode: Partial(real_fdopen(fd, mode))): with self.assertRaisesRegex(OSError, "partial pending"): self.run_index() self.assertFalse((s.STORE / "index-pending.json").exists()) self.assertEqual(self.sent, []) self.run_index() self.assertEqual(len(self.sent), 1) def test_partial_vector_stage_leaves_uncertain_call_for_review(self): real = s.write_stage def partial_vector(path, payload): if path.parent.name == "index-vectors": stage = path.parent / ".vector-partial.stage" stage.parent.mkdir(parents=True, exist_ok=True) stage.write_bytes(payload[:13]) raise OSError("partial vector stage") return real(path, payload) with patch.object(s, "write_stage", side_effect=partial_vector): with self.assertRaisesRegex(OSError, "partial vector"): self.run_index() pending = s.read_json(s.STORE / "index-pending.json") self.assertTrue(pending["started"]) self.assertFalse((s.STORE / "index-vectors" / f"{pending['runId']}.json").exists()) with self.assertRaisesRegex(RuntimeError, "Uncertain paid index"): self.run_index() self.assertEqual(len(self.sent), 1) def test_started_marker_write_or_pre_replace_failure_reruns_without_call(self): real = s.write_stage def fail_stage(path, payload): if path.name == "index-pending.json" and b'"started":true' in payload: path.parent.mkdir(parents=True, exist_ok=True) stage = path.parent / ".marker-partial.stage" stage.write_bytes(payload[:9]) raise OSError("marker staged write interruption") return real(path, payload) with patch.object(s, "write_stage", side_effect=fail_stage): with self.assertRaisesRegex(OSError, "marker staged"): self.run_index() self.assertNotIn("started", s.read_json(s.STORE / "index-pending.json")) self.assertEqual(self.sent, []) real_replace = s.os.replace def fail_replace(src, dst): if str(dst).endswith("index-pending.json"): raise OSError("marker replace interruption") return real_replace(src, dst) with patch.object(s.os, "replace", side_effect=fail_replace): with self.assertRaisesRegex(OSError, "marker replace"): self.run_index() self.assertNotIn("started", s.read_json(s.STORE / "index-pending.json")) self.assertEqual(self.sent, []) self.run_index() self.assertEqual(len(self.sent), 1) def test_after_marker_replace_failure_holds_uncertain_call(self): real = s.os.replace def replace_then_fail(src, dst): real(src, dst) if str(dst).endswith("index-pending.json"): raise OSError("post-replace interruption") with patch.object(s.os, "replace", side_effect=replace_then_fail): with self.assertRaisesRegex(OSError, "post-replace"): self.run_index() self.assertTrue(s.read_json(s.STORE / "index-pending.json")["started"]) with self.assertRaisesRegex(RuntimeError, "Uncertain paid index"): self.run_index() self.assertEqual(self.sent, []) if __name__ == "__main__": unittest.main() ``` ### requirements.txt ```text google-genai==2.33.0 langchain-chroma==1.1.0 chromadb==1.5.9 pypdf==6.21.2 ``` ### manifest.template.json ```json { "model": "gemini-embedding-2", "dimensions": 768, "items": [ { "id": "", "path": "", "title": "", "sourceRevision": "", "permissionReceipt": "" } ] } ``` ## Sources - [Google embedding manual](https://ai.google.dev/gemini-api/docs/embeddings) — Checked October 27, 2026. Supports media limits; local validation and recovery are this guide's implementation. - [Versioned Google Python SDK](https://pypi.org/project/google-genai/2.33.0/) — Documents client construction. No live account response was tested. - [Gemini API pricing](https://ai.google.dev/gemini-api/docs/pricing) — Check current tier charges and data terms against the reader's own project. - [Gemini API rate limits](https://ai.google.dev/gemini-api/docs/rate-limits) — Project quotas must be checked before a paid run. - [LangChain Chroma integration](https://docs.langchain.com/oss/python/integrations/vectorstores/chroma) — Documents the vector-query integration; local source checks are the program's own controls. - [Pinned LangChain Chroma implementation](https://github.com/langchain-ai/langchain/blob/langchain-chroma%3D%3D1.1.0/libs/partners/chroma/langchain_chroma/vectorstores.py) — Shows the selected version's constructor and query method. - [Pinned LangChain Chroma package constraints](https://github.com/langchain-ai/langchain/blob/langchain-chroma%3D%3D1.1.0/libs/partners/chroma/pyproject.toml) — Supports the interpreter and dependency range used for the reader setup. - [Chroma collection management](https://docs.trychroma.com/docs/collections/manage-collections) — Supports local collection setup. - [Chroma adding data](https://docs.trychroma.com/docs/collections/add-data) — Supports supplied-vector insertion. - [Chroma updating data](https://docs.trychroma.com/docs/collections/update-data) — Supports the vector update route. - [pypdf 6.21.2](https://pypi.org/project/pypdf/6.21.2/) — Supports the PDF parser used for local page-count preflight. - [FFmpeg ffprobe manual](https://ffmpeg.org/ffprobe.html) — Supports the video inspection fields used during preflight. - [BIG CHANGE local encoder prototype](https://bigchange.ai/blog/embeddinggemma-2-local-multimodal-search-guide) — Related local prototype using a different encoder and workflow.