v1.0 alpha: keys-Auto Receipts Studio (iPhone / may add Autonomous Lamp Skill)

.env.example

@@ -0,0 +1,28 @@
+# All local. No cloud keys. Copy to .env
+
+# Gemma 4 12B Unified omni (vision + embed). Lamp: set URLs to the GPU box.
+# Does not fit on the Lamp (6 GB). GPU util cap 0.85.
+RECEIPT_LLM_BACKEND=gemma
+RECEIPT_LLM_BASE_URL=http://127.0.0.1:8080/v1
+RECEIPT_LLM_MODEL=google/gemma-4-12B-it
+RECEIPT_LLM_ACCEPTS_IMAGES=true
+RECEIPT_LLM_MAX_TOKENS=8192
+
+RECEIPT_EMBED_BACKEND=omni
+RECEIPT_EMBED_BASE_URL=http://127.0.0.1:8080/v1
+RECEIPT_EMBED_MODEL=google/gemma-4-12B-it
+RECEIPT_EMBED_DIM=3840
+
+# Optional OCR assist (vision extract does not need it)
+RECEIPT_OCR_BACKEND=none
+
+# Lamp HAL camera
+RECEIPT_CAMERA_URL=http://127.0.0.1:5001
+RECEIPT_SNAPSHOT_WIDTH=1280
+RECEIPT_SNAPSHOT_QUALITY=85
+
+# UI / inbox
+RECEIPT_UI_HOST=127.0.0.1
+RECEIPT_UI_PORT=7860
+RECEIPT_UI_SHARE_LAN=false
+RECEIPT_IDLE_SECONDS=30

.gitignore

@@ -0,0 +1,30 @@
+.venv/
+venv/
+__pycache__/
+*.py[cod]
+*.egg-info/
+.pytest_cache/
+.mypy_cache/
+.ruff_cache/
+.env
+*.db
+*.db-wal
+*.db-shm
+data/*
+!data/.gitkeep
+inbox/*
+!inbox/.gitkeep
+processing/*
+!processing/.gitkeep
+processed/*
+!processed/.gitkeep
+failed/*
+!failed/.gitkeep
+exports/*
+!exports/.gitkeep
+.DS_Store
+Thumbs.db
+dist/
+build/
+*.log
+.grok/sessions/

.grok/skills/keys-receipt-scanner/SKILL.md

@@ -0,0 +1,14 @@
+---
+name: keys-receipt-scanner
+description: Scan receipts/docs via Lamp camera or inbox, Gemma 4 12B Unified omni extract+embed, sqlite-vec store. Use when the user mentions receipts, invoices, Lamp scan, Autonomous OS skill, expense capture, or /keys-receipt-scanner.
+---
+
+# keys-receipt-scanner (Grok)
+
+Work in `keys-automatic-receipt-doc-scanner`. Architecture is in `AGENTS.md`.
+
+- Lamp 6 GB: skill + camera only. Gemma 4 12B Unified stays on the GPU box.
+- Default LLM/embed: Gemma 4 12B Unified, dim 3840, `embed_backend=omni`.
+- Qwen3.8-27B ADay777 is the vision fallback (`llm_backend=nvidia`). Lightning is text-only — never send images.
+- Do not download weights. Do not raise GPU util above 0.85.
+- Tests: `pytest` with mocked HTTP, no live models.

AGENTS.md

@@ -0,0 +1,95 @@
+# keys-automatic-receipt-doc-scanner
+
+Local receipt/document scanner for **Autonomous OS + Autonomous Lamp**.
+Lamp camera captures; a GPU box runs the model. Nothing cloud.
+
+## Hardware split (do not blur this)
+
+| Where | Fits | Does not fit |
+|---|---|---|
+| **Lamp** (ARM64, **6 GB RAM**) | This skill (`SKILL.md` + `scripts/`), HAL snapshot, SQLite | Gemma 4 12B Unified, Qwen3.8-27B, Nemotron 3.5 Lightning |
+| **GPU box** (DGX Spark / Omen) | Gemma 4 12B Unified (omni: vision + embed) or Qwen3.8-27B VLM + Embed-1B | — |
+
+Gemma 4 12B Unified is ~12B dense (`hidden_size` **3840**, `Gemma4UnifiedForConditionalGeneration`). Weights alone exceed the Lamp's 6 GB even in NVFP4. Treat the Lamp as eyes/hands; treat the GPU box as the brain.
+
+## Default brain: Gemma 4 12B Unified (omni)
+
+Encoder-free VLM. One OpenAI-compatible server does:
+
+1. **Vision extract** — `POST /v1/chat/completions` with `image_url` data URI (receipt/doc JPEG)
+2. **Embed** — `POST /v1/embeddings` against the **same** server (mean-pool / convert-embed). Dim **3840**.
+
+Never mix 3840 (Gemma) and 2048 (Nemotron-3-Embed-1B) in one sqlite-vec index.
+
+## Fallback brains (same skill, env only)
+
+- **Qwen3.8-27B ADay777** VLM at `:8078` (`qwen38-nvfp4`) for extract; Nemotron-3-Embed-1B 2048-d for embed
+- **Lightning** is text-only. Never send images to it.
+
+Fleet GB10: `--gpu-memory-utilization` **0.85** hard cap.
+
+## Pipeline
+
+```
+voice / phone / drop → Lamp camera or inbox/
+        → optional OCR assist
+        → Gemma4 (or Qwen) vision JSON extract + category
+        → omni embed (or Nemotron-3-Embed-1B)
+        → sqlite-vec vendor/SKU/category match
+        → review / speak summary
+```
+
+## Rules
+
+- No cloud APIs. Backends behind `OCRBackend` / `LLMBackend` / `EmbedBackend`.
+- Lightning: `accepts_images=False`. Never attach image parts.
+- Python 3.12, typed, pytest. No notebooks. Don't vendor weights.
+- Idle-batch inbox 30s. Dedup sha256.
+
+## Extract JSON
+
+```json
+{
+  "doc_kind": "receipt|invoice|document",
+  "category": "groceries|dining|transport|household|health|entertainment|utilities|office|travel|other",
+  "vendor": "string|null",
+  "date": "YYYY-MM-DD|null",
+  "tax": "number|null",
+  "total": "number|null",
+  "currency": "string|null",
+  "line_items": [
+    {"description": "string", "qty": "number|null", "unit_price": "number|null",
+     "amount": "number|null", "sku": "string|null"}
+  ]
+}
+```
+
+Money stored as integer cents.
+
+## Match (cosine similarity = 1 - sqlite-vec distance)
+
+| | Auto | Review | Unmatched |
+|---|---|---|---|
+| SKU / line | ≥ 0.88 | 0.72–0.88 | < 0.72 |
+| Vendor | ≥ 0.82 | 0.65–0.82 | < 0.65 |
+
+Exact catalog SKU wins first.
+
+## Layout
+
+```
+app/          config, schemas, media, camera, extract, embed, db, match,
+              pipeline, watcher, cli, ui
+backends/     base, openai_compat, gemma, nvidia, ollama, apple, cpu
+skills/keys-receipt-scanner/   Autonomous OS built-in skill (Lamp)
+inbox/ processing/ processed/ failed/ exports/
+```
+
+## Autonomous OS skill
+
+`skills/keys-receipt-scanner/` is a **built-in skill** in Autonomous OS format:
+
+- `SKILL.md` + `skill.json` (`capabilities: ["vision"]`)
+- Installs on any body that declares vision (Lamp, Reachy Mini — not Intern)
+- Acts via HAL `GET :5001/camera/snapshot` then `python -m app.cli scan --image PATH`
+- Does not load 12B weights on the robot

LICENSE

@@ -0,0 +1,17 @@
+Apache License
+Version 2.0, January 2004
+http://www.apache.org/licenses/
+
+Copyright 2026 keys
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.

README.md

@@ -0,0 +1,226 @@
+# keys-Auto Receipts Studio (iPhone / may add Autonomous Lamp Skill)
+
+**v1.0 alpha**
+
+iPhone (or desktop) photo → local **Gemma 4 12B-it** vision extract → SQLite. Optional [Autonomous OS](https://github.com/autonomous-ai/autonomous-os) Lamp skill (camera on the robot, 12B on the GPU box — the Lamp’s 6 GB cannot hold 12B).
+
+GitHub: `drowzeys/keys-Auto-Receipts-Studio`  
+Hugging Face: `drowzeys/keys-Auto-Receipts-Studio`
+
+## One-shot (Linux GPU box — NVIDIA + vLLM)
+
+This is the Spark/Omen path. It installs the app, fetches **google/gemma-4-12B-it** if missing, starts vLLM at **`--gpu-memory-utilization 0.15`** (FP8, max-model-len 8192, never above 0.85), and opens the UI.
+
+```bash
+git clone https://github.com/drowzeys/keys-Auto-Receipts-Studio.git
+cd keys-Auto-Receipts-Studio
+bash oneshot.sh
+```
+
+When it prints READY:
+
+| | |
+|---|---|
+| Review | http://127.0.0.1:7860 |
+| iPhone (same Wi‑Fi, **Safari**) | http://<this-pc-lan-ip>:7860/phone |
+
+`hf auth login` once if the Gemma weights are not already at `~/models-gemma4-12b-it`.
+
+Desktop icon after that: `bash scripts/install-launcher.sh`
+
+## Windows (PC)
+
+12B does **not** run on a typical Windows GPU from this script. Run Gemma on a Linux NVIDIA box (`oneshot.sh` there), then this PC is the UI + iPhone hotspot.
+
+1. Install [Python 3.12](https://www.python.org/downloads/) — check **Add python.exe to PATH**.
+2. Double-click `oneshot.bat` (or `scripts\start-ui.bat`).
+3. Edit `.env`:
+
+```
+RECEIPT_LLM_BASE_URL=http://<spark-lan-ip>:8080/v1
+RECEIPT_EMBED_BASE_URL=http://<spark-lan-ip>:8080/v1
+RECEIPT_LLM_MODEL=google/gemma-4-12B-it
+RECEIPT_EMBED_MODEL=google/gemma-4-12B-it
+RECEIPT_EMBED_DIM=3840
+```
+
+4. Phone: `http://<this-windows-lan-ip>:7860/phone`
+
+If this PC **is** an NVIDIA box with `vllm` on PATH, set the same `.env` to `127.0.0.1:8080` and run `bash scripts/serve-gemma.sh` from Git Bash, or install WSL2 and use the Linux one-shot.
+
+## macOS
+
+Same split: Gemma on the Linux GPU box; Mac is UI + iPhone.
+
+```bash
+git clone https://github.com/drowzeys/keys-Auto-Receipts-Studio.git
+cd keys-Auto-Receipts-Studio
+python3 -m venv .venv
+.venv/bin/pip install -e ".[dev]"
+cp .env.example .env
+# point RECEIPT_LLM_BASE_URL at the Spark, then:
+bash scripts/install-launcher.sh
+```
+
+Double-click **Desktop → Receipt Studio.command** (first time: right-click → **Open**).
+
+Apple Silicon will not load 12B next to this app as vLLM-NVIDIA. Use the Spark.
+
+## Linux without NVIDIA
+
+Same as Mac: `bash oneshot.sh` will install the UI and skip/fail vLLM if `vllm` is missing. Point `.env` at a machine that already serves Gemma.
+
+---
+
+Built-in **Autonomous OS** skill for **Autonomous Lamp**: hold up a receipt; Lamp snapshots; the GPU box runs Gemma 4 12B-it. Also a Mac / Windows / Linux inbox (phone upload + Gradio).
+
+## Does Gemma 4 12B Unified fit on the Lamp?
+
+**No.** Lamp is 8-core ARM64 with **6 GB RAM**. Gemma 4 12B Unified is a 12B dense omni model (`Gemma4UnifiedForConditionalGeneration`, hidden size **3840**). Weights alone will not boot beside Autonomous OS + HAL.
+
+| Piece | Fits on Lamp? | Fits on GPU box? |
+|---|---|---|
+| `skills/keys-receipt-scanner/` (this skill) | yes — built-in skill format | yes |
+| HAL `GET /camera/snapshot` | yes | n/a |
+| SQLite + sqlite-vec + HTTP client | yes | yes |
+| **Gemma 4 12B Unified weights** | **no** | yes (vLLM, util **0.15**) |
+| Qwen3.8-27B ADay777 VLM | no | yes |
+| Nemotron-3-Embed-1B | no | yes |
+
+The skill is **built-in to Autonomous OS** (markdown + `skill.json` `capabilities: ["vision"]`). The brain is remote. Intern has no camera, so this skill will not install there.
+
+## Built-in skill (Lamp)
+
+```
+skills/keys-receipt-scanner/
+  SKILL.md          # agent instructions + HAL camera contract
+  skill.json        # {"capabilities": ["vision"]}
+  scripts/scan.py
+```
+
+Install onto a robot (no reboot):
+
+```bash
+make push-skill SKILL=./skills/keys-receipt-scanner TARGET=pi@lamp-xxxx.local
+```
+
+On the Lamp, point the brain at the GPU box:
+
+```bash
+export RECEIPT_LLM_BASE_URL=http://<spark-or-omen>:8080/v1
+export RECEIPT_EMBED_BASE_URL=http://<spark-or-omen>:8080/v1
+export RECEIPT_LLM_MODEL=google/gemma-4-12B-it
+export RECEIPT_EMBED_MODEL=google/gemma-4-12B-it
+export RECEIPT_EMBED_DIM=3840
+export RECEIPT_EMBED_BACKEND=omni
+```
+
+Say **“scan this receipt”** while holding paper to the camera.
+
+To ship it in a fork of [autonomous-os](https://github.com/autonomous-ai/autonomous-os): copy `skills/keys-receipt-scanner/` into `skills/`, run their `python skills/skill-creator/scripts/quick_validate.py`, `make skills-catalog`, open the PR.
+
+## GPU box — Gemma 4 12B Unified (omni)
+
+One server for vision extract **and** embeddings. Do not raise util above **0.85**.
+
+```bash
+bash scripts/serve-gemma.sh
+# util 0.15, FP8, max-model-len 8192  (or: bash oneshot.sh)
+```
+
+If `/v1/embeddings` 404s on a generate-only runner, either:
+
+- serve a pooling convert on another port and set `RECEIPT_EMBED_BASE_URL`, or
+- set `RECEIPT_EMBED_BACKEND=nvidia`, `RECEIPT_EMBED_DIM=2048`, and run Nemotron-3-Embed-1B. **Never mix 3840 and 2048 in one DB.**
+
+### Fallback: Qwen3.8-27B ADay777 (vision only)
+
+```bash
+export RECEIPT_LLM_BACKEND=nvidia
+export RECEIPT_LLM_BASE_URL=http://127.0.0.1:8078/v1
+export RECEIPT_LLM_MODEL=qwen38-nvfp4
+# thinking off is sent automatically
+```
+
+Lightning is **text-only**. This skill will not attach images to it.
+
+## One-click (Windows / macOS / Linux)
+
+Does **not** start Gemma. Point `.env` at the GPU box, then double-click:
+
+| OS | One-click |
+|---|---|
+| **Linux** | `bash scripts/install-launcher.sh` once → Desktop **Receipt Studio** |
+| **macOS** | `bash scripts/install-launcher.sh` once → Desktop **Receipt Studio.command** (first time: right-click → Open) |
+| **Windows** | Copy `scripts/start-ui.bat` to the Desktop (or double-click it in the repo). First run creates `.venv`. |
+
+Same entry from a terminal:
+
+```bash
+# Linux / macOS
+./scripts/start-ui.sh
+
+# Windows
+scripts\start-ui.bat
+```
+
+That starts the UI on the LAN if needed and opens **Review** in the browser. Phone page: `http://<this-machine-lan-ip>:7860/phone`.
+
+On a **CUDA box** (this Spark), the same click also starts Gemma 4 12B if `:8080` is down:
+
+- `--gpu-memory-utilization **0.15**` (~18.3 GiB of 121.7 GiB; never above **0.85**)
+- **FP8** — BF16 weights are ~23GB and cannot fit in that pool
+- `--max-model-len **8192**` (receipt photo is ~280 vision tokens + JSON)
+
+Context at util 0.15 (after ~12.5GB FP8 weights):
+
+| Estimate | Tokens |
+|---|---|
+| Conservative (all 48 layers full attn) | **~12k** |
+| Hybrid (8 full + 40× sliding-1024) | **~65k** |
+| Model native window | 262,144 (not at 0.15) |
+
+A receipt scan uses ~1–2k tokens. Raise `RECEIPT_VLLM_MAX_MODEL_LEN` only after a boot log shows `GPU KV cache size` large enough.
+
+If Gemma is already running, the launcher leaves it alone. To apply the 15GB cap, stop the current `vllm` process, then click again.
+
+```bash
+# apply 15GB cap (stops the current unconstrained serve)
+pkill -x vllm   # only if you intend to restart it
+./scripts/start-ui.sh
+```
+
+On a Mac/Windows laptop with no GPU, set in `.env`:
+
+```
+RECEIPT_LLM_BASE_URL=http://<spark-lan-ip>:8080/v1
+RECEIPT_EMBED_BASE_URL=http://<spark-lan-ip>:8080/v1
+```
+
+## Desktop / phone (same pipeline)
+
+```bash
+python -m venv .venv && source .venv/bin/activate
+pip install -e ".[dev]"
+cp .env.example .env
+python -m app.cli ui
+```
+
+- Inbox watcher: files idle **30s** in `inbox/` then process
+- Phone on LAN: `RECEIPT_UI_SHARE_LAN=true` → `http://<lan-ip>:7860/phone`
+- Syncthing: phone camera/share folder → `inbox/`
+
+```bash
+python -m app.cli scan --image path/to/receipt.jpg
+python -m app.cli query --category groceries
+pytest
+```
+
+## What you still run yourself
+
+- Serve Gemma 4 12B Unified (or Qwen) on the GPU box
+- Pair Lamp on Wi-Fi via the Autonomous app
+- `make push-skill` (or Skill Store / PR) for the built-in skill
+- Optional: Syncthing on iOS/Android
+
+No model weights in this repo. No PyInstaller in this release.

app/__init__.py

@@ -0,0 +1,3 @@
+"""keys-automatic-receipt-doc-scanner"""
+
+__version__ = "1.0.0a1"

app/__main__.py

@@ -0,0 +1,4 @@
+from app.cli import main
+
+if __name__ == "__main__":
+    main()

app/camera.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import httpx
+
+from app.config import Settings
+
+
+class CameraError(RuntimeError):
+    pass
+
+
+def snapshot(settings: Settings, *, client: httpx.Client | None = None) -> Path:
+    """HAL snapshot used by Autonomous Lamp. Returns the saved JPEG path."""
+    url = (
+        f"{settings.camera_url.rstrip('/')}/camera/snapshot"
+        f"?save=true&width={settings.snapshot_width}&quality={settings.snapshot_quality}"
+    )
+    own = client is None
+    http = client or httpx.Client(timeout=30.0)
+    try:
+        response = http.get(url)
+        response.raise_for_status()
+        payload = response.json()
+    except httpx.HTTPError as exc:
+        raise CameraError(f"Lamp camera snapshot failed: {exc}") from exc
+    finally:
+        if own:
+            http.close()
+    path = payload.get("path") if isinstance(payload, dict) else None
+    if not path:
+        raise CameraError(f"snapshot JSON missing path: {payload!r}")
+    return Path(path)

app/cli.py

@@ -0,0 +1,124 @@
+from __future__ import annotations
+
+import argparse
+import json
+import shutil
+import sys
+from pathlib import Path
+
+from app.camera import snapshot
+from app.config import load_settings
+from app.db import get_receipt, list_line_items, list_receipts, open_db
+from app.pipeline import process_file
+from app.schemas import ProcessResult
+
+
+def _print_result(result: ProcessResult) -> None:
+    payload = result.model_dump(mode="json")
+    json.dump(payload, sys.stdout, indent=2)
+    sys.stdout.write("\n")
+
+
+def cmd_scan(args: argparse.Namespace) -> int:
+    settings = load_settings()
+    if args.image:
+        image = Path(args.image)
+    else:
+        image = snapshot(settings)
+    if args.inbox:
+        dest = settings.inbox_dir / image.name
+        shutil.copy2(image, dest)
+        image = dest
+    result = process_file(image, settings)
+    _print_result(result)
+    return 0 if result.status.value not in {"failed"} else 1
+
+
+def cmd_query(args: argparse.Namespace) -> int:
+    settings = load_settings()
+    con = open_db(settings)
+    try:
+        rows = list_receipts(con, status=args.status, limit=args.limit)
+        out = []
+        for row in rows:
+            item = dict(row)
+            if args.category and item.get("category") != args.category:
+                continue
+            if args.vendor and (item.get("vendor") or "").lower().find(args.vendor.lower()) < 0:
+                continue
+            item["line_items"] = [dict(x) for x in list_line_items(con, int(row["id"]))]
+            out.append(item)
+        json.dump(out, sys.stdout, indent=2, default=str)
+        sys.stdout.write("\n")
+    finally:
+        con.close()
+    return 0
+
+
+def cmd_show(args: argparse.Namespace) -> int:
+    settings = load_settings()
+    con = open_db(settings)
+    try:
+        row = get_receipt(con, args.id)
+        if row is None:
+            print(f"not found: {args.id}", file=sys.stderr)
+            return 1
+        payload = dict(row)
+        payload["line_items"] = [dict(x) for x in list_line_items(con, args.id)]
+        json.dump(payload, sys.stdout, indent=2, default=str)
+        sys.stdout.write("\n")
+    finally:
+        con.close()
+    return 0
+
+
+def cmd_snapshot(_args: argparse.Namespace) -> int:
+    settings = load_settings()
+    path = snapshot(settings)
+    print(path)
+    return 0
+
+
+def cmd_ui(_args: argparse.Namespace) -> int:
+    from app.ui import main
+
+    main()
+    return 0
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(prog="keys-scan")
+    sub = parser.add_subparsers(dest="cmd", required=True)
+
+    scan = sub.add_parser("scan", help="process one image (Lamp camera if omitted)")
+    scan.add_argument("--image", help="path to jpeg/png/pdf; omit to snapshot Lamp camera")
+    scan.add_argument("--inbox", action="store_true", help="copy into inbox/ first")
+    scan.set_defaults(func=cmd_scan)
+
+    query = sub.add_parser("query", help="list stored receipts")
+    query.add_argument("--status")
+    query.add_argument("--category")
+    query.add_argument("--vendor")
+    query.add_argument("--limit", type=int, default=50)
+    query.set_defaults(func=cmd_query)
+
+    show = sub.add_parser("show", help="show one receipt")
+    show.add_argument("id", type=int)
+    show.set_defaults(func=cmd_show)
+
+    snap = sub.add_parser("snapshot", help="HAL camera snapshot only")
+    snap.set_defaults(func=cmd_snapshot)
+
+    ui = sub.add_parser("ui", help="Gradio review UI")
+    ui.set_defaults(func=cmd_ui)
+    return parser
+
+
+def main(argv: list[str] | None = None) -> None:
+    parser = build_parser()
+    args = parser.parse_args(argv)
+    raise SystemExit(args.func(args))
+
+
+if __name__ == "__main__":
+    main()

app/config.py

@@ -0,0 +1,119 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+ROOT = Path(__file__).resolve().parent.parent
+
+CATEGORIES = (
+    "groceries",
+    "dining",
+    "transport",
+    "household",
+    "health",
+    "entertainment",
+    "utilities",
+    "office",
+    "travel",
+    "other",
+)
+
+DOC_KINDS = ("receipt", "invoice", "document")
+
+ACCEPTED_SUFFIXES = {
+    ".jpg",
+    ".jpeg",
+    ".png",
+    ".webp",
+    ".heic",
+    ".heif",
+    ".tif",
+    ".tiff",
+    ".pdf",
+    ".txt",
+}
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(
+        env_prefix="RECEIPT_",
+        env_file=str(ROOT / ".env"),
+        env_file_encoding="utf-8",
+        extra="ignore",
+    )
+
+    root_dir: Path = ROOT
+    data_dir: Path = ROOT / "data"
+    inbox_dir: Path = ROOT / "inbox"
+    processing_dir: Path = ROOT / "processing"
+    processed_dir: Path = ROOT / "processed"
+    failed_dir: Path = ROOT / "failed"
+    exports_dir: Path = ROOT / "exports"
+
+    idle_seconds: float = 30.0
+
+    # Gemma 4 12B Unified is the default omni brain (vision extract + embed).
+    # It does not fit on the Lamp (6 GB). Point these URLs at the GPU box.
+    llm_backend: str = "gemma"
+    llm_base_url: str = "http://127.0.0.1:8080/v1"
+    llm_model: str = "google/gemma-4-12B-it"
+    llm_api_key: str = "local"
+    llm_accepts_images: bool = True
+    llm_max_tokens: int = 8192
+    llm_timeout_s: float = 180.0
+
+    embed_backend: str = "omni"
+    embed_base_url: str = ""
+    embed_model: str = "google/gemma-4-12B-it"
+    embed_dim: int = 3840
+    embed_api_key: str = "local"
+    embed_timeout_s: float = 120.0
+    embed_prefix: bool = True
+
+    ocr_backend: str = "none"
+    ocr_base_url: str = "http://127.0.0.1:11434/v1"
+    ocr_model: str = ""
+    ocr_api_key: str = "local"
+
+    camera_url: str = "http://127.0.0.1:5001"
+    snapshot_width: int = 1280
+    snapshot_quality: int = 85
+
+    sku_auto: float = 0.88
+    sku_review: float = 0.72
+    vendor_auto: float = 0.82
+    vendor_review: float = 0.65
+
+    ui_host: str = "127.0.0.1"
+    ui_port: int = 7860
+    ui_share_lan: bool = False
+
+    jpeg_max_edge: int = 2048
+
+    def model_post_init(self, _context: object) -> None:
+        if not self.embed_base_url:
+            object.__setattr__(self, "embed_base_url", self.llm_base_url)
+        if not self.embed_model:
+            object.__setattr__(self, "embed_model", self.llm_model)
+
+    @property
+    def db_path(self) -> Path:
+        return self.data_dir / "receipts.db"
+
+    def ensure_dirs(self) -> None:
+        for path in (
+            self.data_dir,
+            self.inbox_dir,
+            self.processing_dir,
+            self.processed_dir,
+            self.failed_dir,
+            self.exports_dir,
+        ):
+            path.mkdir(parents=True, exist_ok=True)
+
+
+def load_settings(**overrides: object) -> Settings:
+    settings = Settings(**overrides)
+    settings.ensure_dirs()
+    return settings

app/db.py

@@ -0,0 +1,412 @@
+from __future__ import annotations
+
+import json
+import sqlite3
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Iterator
+
+from app.config import Settings
+from app.schemas import MatchHit, ReceiptExtract, ReceiptStatus, to_cents
+
+try:
+    import sqlite_vec
+    from sqlite_vec import serialize_float32
+except ImportError:  # pragma: no cover
+    sqlite_vec = None
+    serialize_float32 = None  # type: ignore[assignment]
+
+
+class VecLoadError(RuntimeError):
+    pass
+
+
+class EmbedIndexError(RuntimeError):
+    pass
+
+
+def _utc_now() -> str:
+    return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
+
+
+def connect(path: Path) -> sqlite3.Connection:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    con = sqlite3.connect(str(path))
+    con.row_factory = sqlite3.Row
+    con.execute("PRAGMA foreign_keys = ON")
+    if sqlite_vec is None:
+        raise VecLoadError("sqlite-vec is not installed")
+    try:
+        con.enable_load_extension(True)
+        sqlite_vec.load(con)
+        con.enable_load_extension(False)
+    except Exception as exc:
+        con.close()
+        raise VecLoadError(f"sqlite-vec load failed: {exc}") from exc
+    return con
+
+
+def _create_vec_table(con: sqlite3.Connection, name: str, pk: str, dim: int) -> None:
+    ddl = (
+        f"CREATE VIRTUAL TABLE IF NOT EXISTS {name} USING vec0("
+        f"{pk} INTEGER PRIMARY KEY, embedding float[{dim}] distance_metric=cosine)"
+    )
+    try:
+        con.execute(ddl)
+    except sqlite3.OperationalError:
+        con.execute(
+            f"CREATE VIRTUAL TABLE IF NOT EXISTS {name} USING vec0("
+            f"{pk} INTEGER PRIMARY KEY, embedding float[{dim}])"
+        )
+
+
+def init_schema(con: sqlite3.Connection, settings: Settings) -> None:
+    con.executescript(
+        """
+        CREATE TABLE IF NOT EXISTS meta (
+            key TEXT PRIMARY KEY,
+            value TEXT NOT NULL
+        );
+        CREATE TABLE IF NOT EXISTS receipts (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            source_path TEXT NOT NULL,
+            sha256 TEXT NOT NULL UNIQUE,
+            status TEXT NOT NULL,
+            doc_kind TEXT,
+            category TEXT,
+            vendor TEXT,
+            receipt_date TEXT,
+            tax_cents INTEGER,
+            total_cents INTEGER,
+            currency TEXT,
+            ocr_text TEXT,
+            extract_json TEXT,
+            error TEXT,
+            created_at TEXT NOT NULL,
+            updated_at TEXT NOT NULL
+        );
+        CREATE INDEX IF NOT EXISTS idx_receipts_category ON receipts(category);
+        CREATE INDEX IF NOT EXISTS idx_receipts_vendor ON receipts(vendor);
+        CREATE INDEX IF NOT EXISTS idx_receipts_date ON receipts(receipt_date);
+        CREATE TABLE IF NOT EXISTS line_items (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            receipt_id INTEGER NOT NULL REFERENCES receipts(id) ON DELETE CASCADE,
+            description TEXT NOT NULL,
+            qty REAL,
+            unit_price_cents INTEGER,
+            amount_cents INTEGER,
+            sku TEXT,
+            match_catalog_id INTEGER,
+            match_score REAL,
+            match_status TEXT
+        );
+        CREATE TABLE IF NOT EXISTS catalog (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            sku TEXT,
+            vendor TEXT,
+            description TEXT NOT NULL,
+            size TEXT,
+            unit_price_cents INTEGER,
+            metadata_json TEXT
+        );
+        """
+    )
+    _create_vec_table(con, "receipt_vec", "receipt_id", settings.embed_dim)
+    _create_vec_table(con, "catalog_vec", "catalog_id", settings.embed_dim)
+    _check_or_set_meta(con, settings)
+    con.commit()
+
+
+def _meta(con: sqlite3.Connection, key: str) -> str | None:
+    row = con.execute("SELECT value FROM meta WHERE key = ?", (key,)).fetchone()
+    return None if row is None else str(row["value"])
+
+
+def _check_or_set_meta(con: sqlite3.Connection, settings: Settings) -> None:
+    stored_model = _meta(con, "embed_model")
+    stored_dim = _meta(con, "embed_dim")
+    if stored_model is None:
+        con.execute(
+            "INSERT INTO meta(key, value) VALUES ('embed_model', ?), ('embed_dim', ?)",
+            (settings.embed_model, str(settings.embed_dim)),
+        )
+        return
+    if stored_model != settings.embed_model or stored_dim != str(settings.embed_dim):
+        raise EmbedIndexError(
+            f"index is {stored_model} dim={stored_dim}; config is "
+            f"{settings.embed_model} dim={settings.embed_dim}. Never mix embedding models."
+        )
+
+
+def open_db(settings: Settings) -> sqlite3.Connection:
+    con = connect(settings.db_path)
+    init_schema(con, settings)
+    return con
+
+
+def get_by_sha(con: sqlite3.Connection, sha256: str) -> sqlite3.Row | None:
+    return con.execute("SELECT * FROM receipts WHERE sha256 = ?", (sha256,)).fetchone()
+
+
+def insert_receipt(
+    con: sqlite3.Connection,
+    *,
+    source_path: str,
+    sha256: str,
+    status: ReceiptStatus,
+    extract: ReceiptExtract | None = None,
+    ocr_text: str | None = None,
+    error: str | None = None,
+) -> int:
+    now = _utc_now()
+    cur = con.execute(
+        """
+        INSERT INTO receipts (
+            source_path, sha256, status, doc_kind, category, vendor, receipt_date,
+            tax_cents, total_cents, currency, ocr_text, extract_json, error,
+            created_at, updated_at
+        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+        """,
+        (
+            source_path,
+            sha256,
+            status.value,
+            None if extract is None else extract.doc_kind,
+            None if extract is None else extract.category,
+            None if extract is None else extract.vendor,
+            None if extract is None else (extract.date.isoformat() if extract.date else None),
+            None if extract is None else to_cents(extract.tax),
+            None if extract is None else to_cents(extract.total),
+            None if extract is None else extract.currency,
+            ocr_text,
+            None if extract is None else extract.model_dump_json(),
+            error,
+            now,
+            now,
+        ),
+    )
+    receipt_id = int(cur.lastrowid)
+    if extract is not None:
+        for item in extract.line_items:
+            con.execute(
+                """
+                INSERT INTO line_items (
+                    receipt_id, description, qty, unit_price_cents, amount_cents, sku
+                ) VALUES (?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    receipt_id,
+                    item.description,
+                    item.qty,
+                    to_cents(item.unit_price),
+                    to_cents(item.amount),
+                    item.sku,
+                ),
+            )
+    con.commit()
+    return receipt_id
+
+
+def update_receipt_status(
+    con: sqlite3.Connection,
+    receipt_id: int,
+    status: ReceiptStatus,
+    *,
+    error: str | None = None,
+) -> None:
+    con.execute(
+        "UPDATE receipts SET status = ?, error = ?, updated_at = ? WHERE id = ?",
+        (status.value, error, _utc_now(), receipt_id),
+    )
+    con.commit()
+
+
+def update_extract(
+    con: sqlite3.Connection,
+    receipt_id: int,
+    extract: ReceiptExtract,
+    status: ReceiptStatus,
+) -> None:
+    con.execute("DELETE FROM line_items WHERE receipt_id = ?", (receipt_id,))
+    con.execute(
+        """
+        UPDATE receipts SET
+            status = ?, doc_kind = ?, category = ?, vendor = ?, receipt_date = ?,
+            tax_cents = ?, total_cents = ?, currency = ?, extract_json = ?,
+            error = NULL, updated_at = ?
+        WHERE id = ?
+        """,
+        (
+            status.value,
+            extract.doc_kind,
+            extract.category,
+            extract.vendor,
+            extract.date.isoformat() if extract.date else None,
+            to_cents(extract.tax),
+            to_cents(extract.total),
+            extract.currency,
+            extract.model_dump_json(),
+            _utc_now(),
+            receipt_id,
+        ),
+    )
+    for item in extract.line_items:
+        con.execute(
+            """
+            INSERT INTO line_items (
+                receipt_id, description, qty, unit_price_cents, amount_cents, sku
+            ) VALUES (?, ?, ?, ?, ?, ?)
+            """,
+            (
+                receipt_id,
+                item.description,
+                item.qty,
+                to_cents(item.unit_price),
+                to_cents(item.amount),
+                item.sku,
+            ),
+        )
+    con.commit()
+
+
+def delete_receipt(con: sqlite3.Connection, receipt_id: int, *, unlink_file: bool = True) -> bool:
+    row = con.execute(
+        "SELECT source_path FROM receipts WHERE id = ?", (receipt_id,)
+    ).fetchone()
+    if row is None:
+        return False
+    try:
+        con.execute("DELETE FROM receipt_vec WHERE receipt_id = ?", (receipt_id,))
+    except sqlite3.Error:
+        pass
+    con.execute("DELETE FROM receipts WHERE id = ?", (receipt_id,))
+    con.commit()
+    if unlink_file:
+        path = Path(row["source_path"] or "")
+        if path.is_file():
+            try:
+                path.unlink()
+            except OSError:
+                pass
+    return True
+
+
+def set_line_match(
+    con: sqlite3.Connection,
+    line_id: int,
+    hit: MatchHit,
+) -> None:
+    con.execute(
+        """
+        UPDATE line_items SET match_catalog_id = ?, match_score = ?, match_status = ?
+        WHERE id = ?
+        """,
+        (hit.catalog_id, hit.similarity, hit.band.value, line_id),
+    )
+    con.commit()
+
+
+def list_receipts(con: sqlite3.Connection, *, status: str | None = None, limit: int = 50) -> list[sqlite3.Row]:
+    if status:
+        return list(
+            con.execute(
+                "SELECT * FROM receipts WHERE status = ? ORDER BY id DESC LIMIT ?",
+                (status, limit),
+            )
+        )
+    return list(con.execute("SELECT * FROM receipts ORDER BY id DESC LIMIT ?", (limit,)))
+
+
+def get_receipt(con: sqlite3.Connection, receipt_id: int) -> sqlite3.Row | None:
+    return con.execute("SELECT * FROM receipts WHERE id = ?", (receipt_id,)).fetchone()
+
+
+def list_line_items(con: sqlite3.Connection, receipt_id: int) -> list[sqlite3.Row]:
+    return list(
+        con.execute("SELECT * FROM line_items WHERE receipt_id = ? ORDER BY id", (receipt_id,))
+    )
+
+
+def add_catalog_item(
+    con: sqlite3.Connection,
+    *,
+    description: str,
+    sku: str | None = None,
+    vendor: str | None = None,
+    size: str | None = None,
+    unit_price_cents: int | None = None,
+    metadata: dict[str, Any] | None = None,
+) -> int:
+    cur = con.execute(
+        """
+        INSERT INTO catalog (sku, vendor, description, size, unit_price_cents, metadata_json)
+        VALUES (?, ?, ?, ?, ?, ?)
+        """,
+        (
+            sku,
+            vendor,
+            description,
+            size,
+            unit_price_cents,
+            None if metadata is None else json.dumps(metadata),
+        ),
+    )
+    con.commit()
+    return int(cur.lastrowid)
+
+
+def list_catalog(con: sqlite3.Connection, limit: int = 200) -> list[sqlite3.Row]:
+    return list(con.execute("SELECT * FROM catalog ORDER BY id DESC LIMIT ?", (limit,)))
+
+
+def find_catalog_by_sku(con: sqlite3.Connection, sku: str) -> sqlite3.Row | None:
+    return con.execute(
+        "SELECT * FROM catalog WHERE sku = ? COLLATE NOCASE LIMIT 1", (sku,)
+    ).fetchone()
+
+
+def upsert_vector(con: sqlite3.Connection, table: str, pk_col: str, pk: int, vec: list[float]) -> None:
+    if serialize_float32 is None:
+        raise VecLoadError("sqlite-vec missing")
+    blob = serialize_float32(vec)
+    con.execute(f"DELETE FROM {table} WHERE {pk_col} = ?", (pk,))
+    con.execute(
+        f"INSERT INTO {table}({pk_col}, embedding) VALUES (?, ?)",
+        (pk, blob),
+    )
+    con.commit()
+
+
+def knn(
+    con: sqlite3.Connection,
+    table: str,
+    pk_col: str,
+    query: list[float],
+    *,
+    k: int = 5,
+) -> list[tuple[int, float]]:
+    if serialize_float32 is None:
+        raise VecLoadError("sqlite-vec missing")
+    blob = serialize_float32(query)
+    rows = con.execute(
+        f"""
+        SELECT {pk_col} AS id, distance
+        FROM {table}
+        WHERE embedding MATCH ?
+          AND k = ?
+        """,
+        (blob, k),
+    ).fetchall()
+    return [(int(row["id"]), float(row["distance"])) for row in rows]
+
+
+def receipt_to_extract(row: sqlite3.Row) -> ReceiptExtract | None:
+    raw = row["extract_json"]
+    if not raw:
+        return None
+    return ReceiptExtract.model_validate_json(raw)
+
+
+def iter_rows(rows: list[sqlite3.Row]) -> Iterator[dict[str, Any]]:
+    for row in rows:
+        yield dict(row)

app/embed.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+from app.config import Settings
+from app.schemas import LineItem, ReceiptExtract
+from backends.base import EmbedBackend, InputType
+from backends.openai_compat import apply_embed_prefix
+
+
+def format_embed_input(text: str, input_type: InputType, *, enabled: bool = True) -> str:
+    return apply_embed_prefix(text, input_type, enabled=enabled)
+
+
+def line_query_text(extract: ReceiptExtract, item: LineItem) -> str:
+    parts = [
+        extract.vendor or "",
+        item.sku or "",
+        item.description,
+        f"qty {item.qty}" if item.qty is not None else "",
+        f"amount {item.amount}" if item.amount is not None else "",
+    ]
+    return " | ".join(p for p in parts if p)
+
+
+def catalog_passage_text(
+    *,
+    vendor: str | None,
+    sku: str | None,
+    description: str,
+    size: str | None = None,
+) -> str:
+    parts = [vendor or "", sku or "", description, size or ""]
+    return " | ".join(p for p in parts if p)
+
+
+def vendor_query_text(vendor: str) -> str:
+    return vendor.strip()
+
+
+def embed_texts(
+    backend: EmbedBackend,
+    texts: list[str],
+    *,
+    input_type: InputType,
+    settings: Settings | None = None,
+) -> list[list[float]]:
+    del settings
+    return backend.embed(texts, input_type=input_type)

app/extract.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+from app.config import CATEGORIES, DOC_KINDS, Settings
+from app.schemas import ReceiptExtract, parse_extract_json
+from backends.base import LLMBackend
+
+SYSTEM = (
+    "You extract structured data from a photo of a receipt, invoice, or paper document. "
+    "Reply with a single JSON object only — no markdown, no commentary, no trailing text."
+)
+
+_SCHEMA_HINT = f"""
+Required keys:
+  doc_kind: one of {list(DOC_KINDS)}
+  category: one of {list(CATEGORIES)}
+  vendor: string or null
+  date: YYYY-MM-DD or null
+  tax: number or null
+  total: number or null
+  currency: string or null (ISO 4217 if known)
+  line_items: array of objects with description (string), qty (number|null),
+              unit_price (number|null), amount (number|null), sku (string|null)
+
+Rules:
+- Money as numbers, not strings. Unknown fields must be null.
+- Do not invent SKUs, vendors, or totals. If unreadable, use null.
+- Prefer the printed total over summing line items when they disagree.
+- category is the spend bucket (groceries, dining, …), not the store name.
+""".strip()
+
+
+def build_user_prompt(*, ocr_text: str | None, hint: str | None = None) -> str:
+    parts = [_SCHEMA_HINT]
+    if ocr_text:
+        parts.append("OCR assist (may be noisy):\n" + ocr_text.strip()[:8000])
+    if hint:
+        parts.append(hint)
+    parts.append("Extract the JSON now.")
+    return "\n\n".join(parts)
+
+
+def extract_receipt(
+    llm: LLMBackend,
+    *,
+    settings: Settings,
+    image_jpeg: bytes | None,
+    ocr_text: str | None,
+) -> ReceiptExtract:
+    del settings
+    if image_jpeg and not llm.accepts_images:
+        image_jpeg = None
+    if image_jpeg is None and not (ocr_text and ocr_text.strip()):
+        raise ValueError("need an image (vision LLM) or OCR/text to extract")
+    user = build_user_prompt(ocr_text=ocr_text)
+    raw = llm.complete_json(system=SYSTEM, user=user, image_jpeg=image_jpeg)
+    try:
+        return parse_extract_json(raw)
+    except (ValueError, Exception) as first:
+        retry = build_user_prompt(
+            ocr_text=ocr_text,
+            hint=f"Previous output failed validation: {first}. Return corrected JSON only.",
+        )
+        raw2 = llm.complete_json(system=SYSTEM, user=retry, image_jpeg=image_jpeg)
+        return parse_extract_json(raw2)

app/launch.py

@@ -0,0 +1,154 @@
+"""One-click start: optional Gemma vLLM (≤15GB), then LAN UI + browser."""
+
+from __future__ import annotations
+
+import os
+import shutil
+import socket
+import subprocess
+import sys
+import time
+import urllib.error
+import urllib.request
+import webbrowser
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+HOST = "127.0.0.1"
+PORT = int(os.environ.get("RECEIPT_UI_PORT", "7860"))
+VLLM_PORT = int(os.environ.get("RECEIPT_VLLM_PORT", "8080"))
+START_VLLM = os.environ.get("RECEIPT_START_VLLM", "1").lower() not in {"0", "false", "no"}
+MAX_GB = os.environ.get("RECEIPT_VLLM_MAX_GB", "15")  # unused if UTIL is set in serve-gemma.sh
+
+
+def _lan_ip() -> str:
+    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    try:
+        sock.connect(("192.0.2.1", 1))
+        return sock.getsockname()[0]
+    except OSError:
+        return "127.0.0.1"
+    finally:
+        sock.close()
+
+
+def _port_up(port: int) -> bool:
+    sock = socket.socket()
+    sock.settimeout(0.4)
+    try:
+        sock.connect((HOST, port))
+        return True
+    except OSError:
+        return False
+    finally:
+        sock.close()
+
+
+def _vllm_ready() -> bool:
+    try:
+        with urllib.request.urlopen(
+            f"http://127.0.0.1:{VLLM_PORT}/v1/models", timeout=2
+        ) as response:
+            return response.status == 200
+    except (urllib.error.URLError, TimeoutError, OSError):
+        return False
+
+
+def _can_serve_gemma() -> bool:
+    if shutil.which("vllm") is None:
+        return False
+    model = Path(os.environ.get("RECEIPT_GEMMA_PATH", str(Path.home() / "models-gemma4-12b-it")))
+    return (model / "config.json").is_file()
+
+
+def _spawn(cmd: list[str], log: Path, *, bash: bool = False) -> None:
+    log.parent.mkdir(parents=True, exist_ok=True)
+    creation = 0
+    if sys.platform == "win32":
+        creation = getattr(subprocess, "CREATE_NEW_CONSOLE", 0)
+    argv = cmd
+    if bash:
+        argv = ["bash", *cmd]
+    with log.open("a", encoding="utf-8") as handle:
+        subprocess.Popen(
+            argv,
+            cwd=str(ROOT),
+            env=os.environ.copy(),
+            stdout=handle,
+            stderr=handle,
+            creationflags=creation,
+            start_new_session=sys.platform != "win32",
+        )
+
+
+def _ensure_vllm() -> None:
+    if not START_VLLM:
+        return
+    if _vllm_ready():
+        print(f"Gemma already up on :{VLLM_PORT} (not restarted; 15GB cap applies on a fresh serve).")
+        return
+    if not _can_serve_gemma():
+        print("No local vLLM/Gemma — skip serve. Point .env at the GPU box.")
+        return
+    script = ROOT / "scripts" / "serve-gemma.sh"
+    if not script.is_file():
+        print(f"missing {script}", file=sys.stderr)
+        return
+    os.environ.setdefault("RECEIPT_VLLM_MAX_GB", MAX_GB)
+    print("Starting Gemma 4 12B vLLM at gpu_memory_utilization=0.15 (FP8, max-model-len 8192)…")
+    _spawn([str(script)], ROOT / "data" / "vllm-gemma.log", bash=True)
+    for _ in range(120):
+        if _vllm_ready():
+            print("Gemma ready.")
+            return
+        time.sleep(5)
+    print(
+        f"vLLM still starting. Watch {ROOT / 'data' / 'vllm-gemma.log'}",
+        file=sys.stderr,
+    )
+
+
+def _spawn_ui() -> None:
+    env = os.environ.copy()
+    env["RECEIPT_UI_SHARE_LAN"] = "true"
+    env.setdefault("RECEIPT_IDLE_SECONDS", "5")
+    log = ROOT / "data" / "ui.log"
+    log.parent.mkdir(parents=True, exist_ok=True)
+    creation = 0
+    if sys.platform == "win32":
+        creation = getattr(subprocess, "CREATE_NEW_CONSOLE", 0)
+    with log.open("a", encoding="utf-8") as handle:
+        subprocess.Popen(
+            [sys.executable, "-m", "app.cli", "ui"],
+            cwd=str(ROOT),
+            env=env,
+            stdout=handle,
+            stderr=handle,
+            creationflags=creation,
+            start_new_session=sys.platform != "win32",
+        )
+
+
+def main() -> None:
+    os.chdir(ROOT)
+    _ensure_vllm()
+    if not _port_up(PORT):
+        print("Starting Receipt Studio UI…")
+        _spawn_ui()
+        for _ in range(40):
+            if _port_up(PORT):
+                break
+            time.sleep(0.25)
+        else:
+            print(f"UI did not bind :{PORT}. See {ROOT / 'data' / 'ui.log'}", file=sys.stderr)
+            raise SystemExit(1)
+    lan = _lan_ip()
+    review = f"http://127.0.0.1:{PORT}"
+    phone = f"http://{lan}:{PORT}/phone"
+    print(f"Review: {review}")
+    print(f"Phone:  {phone}")
+    webbrowser.open(review)
+
+
+if __name__ == "__main__":
+    main()

app/match.py

@@ -0,0 +1,147 @@
+from __future__ import annotations
+
+import sqlite3
+
+from app.config import Settings
+from app.db import find_catalog_by_sku, knn
+from app.embed import catalog_passage_text, embed_texts, line_query_text, vendor_query_text
+from app.schemas import LineItem, MatchBand, MatchHit, ReceiptExtract
+from backends.base import EmbedBackend
+
+
+def distance_to_similarity(distance: float) -> float:
+    return 1.0 - distance
+
+
+def band_for(similarity: float, *, auto: float, review: float) -> MatchBand:
+    if similarity >= auto:
+        return MatchBand.auto
+    if similarity >= review:
+        return MatchBand.review
+    return MatchBand.unmatched
+
+
+def unmatched(reason: str) -> MatchHit:
+    return MatchHit(similarity=0.0, band=MatchBand.unmatched, reason=reason)
+
+
+def match_line_item(
+    con: sqlite3.Connection,
+    settings: Settings,
+    embed: EmbedBackend,
+    extract: ReceiptExtract,
+    item: LineItem,
+    *,
+    k: int = 5,
+) -> MatchHit:
+    if item.sku:
+        row = find_catalog_by_sku(con, item.sku)
+        if row is not None:
+            return MatchHit(
+                catalog_id=int(row["id"]),
+                sku=row["sku"],
+                vendor=row["vendor"],
+                description=row["description"],
+                similarity=1.0,
+                band=MatchBand.exact,
+                reason="exact sku",
+            )
+    catalog_count = con.execute("SELECT COUNT(*) AS n FROM catalog").fetchone()["n"]
+    if catalog_count == 0:
+        return unmatched("empty catalog")
+    query_vec = embed_texts(
+        embed, [line_query_text(extract, item)], input_type="query", settings=settings
+    )[0]
+    hits = knn(con, "catalog_vec", "catalog_id", query_vec, k=k)
+    if not hits:
+        return unmatched("no vectors")
+    catalog_id, distance = hits[0]
+    similarity = distance_to_similarity(distance)
+    row = con.execute("SELECT * FROM catalog WHERE id = ?", (catalog_id,)).fetchone()
+    return MatchHit(
+        catalog_id=catalog_id,
+        sku=None if row is None else row["sku"],
+        vendor=None if row is None else row["vendor"],
+        description=None if row is None else row["description"],
+        similarity=similarity,
+        band=band_for(similarity, auto=settings.sku_auto, review=settings.sku_review),
+        reason="knn",
+    )
+
+
+def match_vendor(
+    con: sqlite3.Connection,
+    settings: Settings,
+    embed: EmbedBackend,
+    vendor: str,
+    *,
+    k: int = 5,
+) -> MatchHit:
+    if not vendor.strip():
+        return unmatched("no vendor")
+    exact = con.execute(
+        "SELECT * FROM catalog WHERE vendor = ? COLLATE NOCASE LIMIT 1", (vendor,)
+    ).fetchone()
+    if exact is not None:
+        return MatchHit(
+            catalog_id=int(exact["id"]),
+            sku=exact["sku"],
+            vendor=exact["vendor"],
+            description=exact["description"],
+            similarity=1.0,
+            band=MatchBand.exact,
+            reason="exact vendor",
+        )
+    query_vec = embed_texts(
+        embed, [vendor_query_text(vendor)], input_type="query", settings=settings
+    )[0]
+    hits = knn(con, "catalog_vec", "catalog_id", query_vec, k=k)
+    if not hits:
+        return unmatched("no vectors")
+    catalog_id, distance = hits[0]
+    similarity = distance_to_similarity(distance)
+    row = con.execute("SELECT * FROM catalog WHERE id = ?", (catalog_id,)).fetchone()
+    return MatchHit(
+        catalog_id=catalog_id,
+        sku=None if row is None else row["sku"],
+        vendor=None if row is None else row["vendor"],
+        description=None if row is None else row["description"],
+        similarity=similarity,
+        band=band_for(
+            similarity, auto=settings.vendor_auto, review=settings.vendor_review
+        ),
+        reason="vendor knn",
+    )
+
+
+def match_receipt(
+    con: sqlite3.Connection,
+    settings: Settings,
+    embed: EmbedBackend,
+    extract: ReceiptExtract,
+) -> list[MatchHit]:
+    hits = [match_line_item(con, settings, embed, extract, item) for item in extract.line_items]
+    if extract.vendor:
+        hits.append(match_vendor(con, settings, embed, extract.vendor))
+    return hits
+
+
+def embed_catalog_row(
+    con: sqlite3.Connection,
+    settings: Settings,
+    embed: EmbedBackend,
+    catalog_id: int,
+) -> None:
+    from app.db import upsert_vector
+
+    row = con.execute("SELECT * FROM catalog WHERE id = ?", (catalog_id,)).fetchone()
+    if row is None:
+        return
+    text = catalog_passage_text(
+        vendor=row["vendor"],
+        sku=row["sku"],
+        description=row["description"],
+        size=row["size"],
+    )
+    vec = embed_texts(embed, [text], input_type="passage", settings=settings)[0]
+    upsert_vector(con, "catalog_vec", "catalog_id", catalog_id, vec)

app/media.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+import io
+from pathlib import Path
+
+from PIL import Image
+
+from app.config import Settings
+
+try:
+    from pillow_heif import register_heif_opener
+
+    register_heif_opener()
+except ImportError:
+    pass
+
+
+def _load_heif() -> None:
+    try:
+        from pillow_heif import register_heif_opener
+
+        register_heif_opener()
+    except ImportError:
+        return
+
+
+def pdf_first_page_jpeg(path: Path, max_edge: int) -> bytes:
+    import pypdfium2 as pdfium
+
+    pdf = pdfium.PdfDocument(str(path))
+    try:
+        page = pdf[0]
+        bitmap = page.render(scale=150 / 72)
+        image = bitmap.to_pil().convert("RGB")
+    finally:
+        pdf.close()
+    return _pil_to_jpeg(image, max_edge)
+
+
+def _pil_to_jpeg(image: Image.Image, max_edge: int) -> bytes:
+    rgb = image.convert("RGB")
+    rgb.thumbnail((max_edge, max_edge), Image.Resampling.LANCZOS)
+    buf = io.BytesIO()
+    rgb.save(buf, format="JPEG", quality=85, optimize=True)
+    return buf.getvalue()
+
+
+def to_jpeg_bytes(path: Path, settings: Settings) -> bytes | None:
+    suffix = path.suffix.lower()
+    if suffix == ".txt":
+        return None
+    if suffix == ".pdf":
+        return pdf_first_page_jpeg(path, settings.jpeg_max_edge)
+    if suffix in {".heic", ".heif"}:
+        _load_heif()
+    with Image.open(path) as image:
+        return _pil_to_jpeg(image, settings.jpeg_max_edge)

app/ocr.py

@@ -0,0 +1,16 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from backends.base import OCRBackend, OCRResult
+
+
+def maybe_ocr(backend: OCRBackend | None, path: Path) -> OCRResult | None:
+    if backend is None:
+        sidecar = path.with_suffix(".txt")
+        if sidecar.is_file():
+            return OCRResult(text=sidecar.read_text(encoding="utf-8", errors="replace"), engine="sidecar")
+        if path.suffix.lower() == ".txt":
+            return OCRResult(text=path.read_text(encoding="utf-8", errors="replace"), engine="plaintext")
+        return None
+    return backend.ocr(path)

app/pipeline.py

@@ -0,0 +1,171 @@
+from __future__ import annotations
+
+import hashlib
+import shutil
+import threading
+from pathlib import Path
+
+from app.config import Settings
+from app.db import (
+    get_by_sha,
+    insert_receipt,
+    open_db,
+    set_line_match,
+    update_receipt_status,
+    upsert_vector,
+)
+from app.embed import embed_texts
+from app.extract import extract_receipt
+from app.match import match_receipt
+from app.media import to_jpeg_bytes
+from app.ocr import maybe_ocr
+from app.schemas import ProcessResult, ReceiptStatus
+from backends import build_embed, build_llm, build_ocr
+
+_LOCK = threading.Lock()
+
+
+def sha256_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _safe_move(src: Path, dest_dir: Path) -> Path:
+    dest_dir.mkdir(parents=True, exist_ok=True)
+    dest = dest_dir / src.name
+    stem, suffix = dest.stem, dest.suffix
+    n = 1
+    while dest.exists():
+        dest = dest_dir / f"{stem}-{n}{suffix}"
+        n += 1
+    shutil.move(str(src), str(dest))
+    return dest
+
+
+def process_file(path: Path, settings: Settings) -> ProcessResult:
+    with _LOCK:
+        return _process_file_locked(path, settings)
+
+
+def _process_file_locked(path: Path, settings: Settings) -> ProcessResult:
+    settings.ensure_dirs()
+    src = Path(path)
+    digest = sha256_file(src)
+    con = open_db(settings)
+    try:
+        existing = get_by_sha(con, digest)
+        if existing is not None:
+            if src.parent.resolve() == settings.inbox_dir.resolve():
+                _safe_move(src, settings.processed_dir)
+            return ProcessResult(
+                receipt_id=int(existing["id"]),
+                status=ReceiptStatus.duplicate,
+                source_path=str(src),
+                error="duplicate sha256",
+            )
+        working = src
+        if src.parent.resolve() == settings.inbox_dir.resolve():
+            working = _safe_move(src, settings.processing_dir)
+
+        ocr_backend = build_ocr(settings)
+        llm = build_llm(settings)
+        embed = build_embed(settings)
+        ocr = maybe_ocr(ocr_backend, working)
+        ocr_text = None if ocr is None else ocr.text
+        try:
+            image = to_jpeg_bytes(working, settings)
+        except Exception as exc:
+            failed = _safe_move(working, settings.failed_dir)
+            rid = insert_receipt(
+                con,
+                source_path=str(failed),
+                sha256=digest,
+                status=ReceiptStatus.failed,
+                ocr_text=ocr_text,
+                error=str(exc),
+            )
+            return ProcessResult(
+                receipt_id=rid,
+                status=ReceiptStatus.failed,
+                source_path=str(failed),
+                error=str(exc),
+            )
+
+        if image is None and not ocr_text:
+            rid = insert_receipt(
+                con,
+                source_path=str(working),
+                sha256=digest,
+                status=ReceiptStatus.needs_ocr,
+            )
+            return ProcessResult(
+                receipt_id=rid,
+                status=ReceiptStatus.needs_ocr,
+                source_path=str(working),
+                error="no image/text for extract",
+            )
+
+        try:
+            extract = extract_receipt(
+                llm, settings=settings, image_jpeg=image, ocr_text=ocr_text
+            )
+        except Exception as exc:
+            final = _safe_move(working, settings.processed_dir)
+            rid = insert_receipt(
+                con,
+                source_path=str(final),
+                sha256=digest,
+                status=ReceiptStatus.needs_extract,
+                ocr_text=ocr_text,
+                error=str(exc),
+            )
+            return ProcessResult(
+                receipt_id=rid,
+                status=ReceiptStatus.needs_extract,
+                source_path=str(final),
+                error=str(exc),
+            )
+
+        final = _safe_move(working, settings.processed_dir)
+        rid = insert_receipt(
+            con,
+            source_path=str(final),
+            sha256=digest,
+            status=ReceiptStatus.needs_review,
+            extract=extract,
+            ocr_text=ocr_text,
+        )
+        try:
+            if extract.vendor or extract.line_items:
+                blob = " | ".join(
+                    [
+                        extract.doc_kind,
+                        extract.category,
+                        extract.vendor or "",
+                        extract.date.isoformat() if extract.date else "",
+                        *(item.description for item in extract.line_items[:12]),
+                    ]
+                )
+                vec = embed_texts(embed, [blob], input_type="passage", settings=settings)[0]
+                upsert_vector(con, "receipt_vec", "receipt_id", rid, vec)
+            matches = match_receipt(con, settings, embed, extract)
+            line_rows = con.execute(
+                "SELECT id FROM line_items WHERE receipt_id = ? ORDER BY id", (rid,)
+            ).fetchall()
+            for row, hit in zip(line_rows, matches, strict=False):
+                if hit.reason != "vendor knn":
+                    set_line_match(con, int(row["id"]), hit)
+        except Exception:
+            matches = []
+        return ProcessResult(
+            receipt_id=rid,
+            status=ReceiptStatus.needs_review,
+            source_path=str(final),
+            extract=extract,
+            matches=matches,
+        )
+    finally:
+        con.close()

app/schemas.py

@@ -0,0 +1,135 @@
+from __future__ import annotations
+
+import json
+import re
+from datetime import date as Date
+from datetime import datetime as DateTime
+from decimal import Decimal, ROUND_HALF_UP
+from enum import StrEnum
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+from app.config import CATEGORIES, DOC_KINDS
+
+_FENCE_RE = re.compile(r"```(?:json)?\s*(.*?)\s*```", re.DOTALL | re.IGNORECASE)
+
+
+class ReceiptStatus(StrEnum):
+    queued = "queued"
+    processing = "processing"
+    needs_ocr = "needs_ocr"
+    needs_extract = "needs_extract"
+    needs_review = "needs_review"
+    confirmed = "confirmed"
+    failed = "failed"
+    duplicate = "duplicate"
+
+
+class MatchBand(StrEnum):
+    exact = "exact"
+    auto = "auto"
+    review = "review"
+    unmatched = "unmatched"
+    confirmed = "confirmed"
+
+
+class LineItem(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    description: str
+    qty: float | None = None
+    unit_price: Decimal | None = None
+    amount: Decimal | None = None
+    sku: str | None = None
+
+
+class ReceiptExtract(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    doc_kind: str = "receipt"
+    category: str = "other"
+    vendor: str | None = None
+    date: Date | None = None
+    tax: Decimal | None = None
+    total: Decimal | None = None
+    currency: str | None = None
+    line_items: list[LineItem] = Field(default_factory=list)
+
+    @field_validator("doc_kind")
+    @classmethod
+    def _doc_kind(cls, value: str) -> str:
+        kind = (value or "receipt").strip().lower()
+        return kind if kind in DOC_KINDS else "document"
+
+    @field_validator("category")
+    @classmethod
+    def _category(cls, value: str) -> str:
+        cat = (value or "other").strip().lower()
+        return cat if cat in CATEGORIES else "other"
+
+    @field_validator("date", mode="before")
+    @classmethod
+    def _date(cls, value: object) -> object:
+        if value in (None, "", "null"):
+            return None
+        if isinstance(value, Date):
+            return value
+        text = str(value).strip()[:10]
+        return DateTime.strptime(text, "%Y-%m-%d").date()
+
+
+class MatchHit(BaseModel):
+    catalog_id: int | None = None
+    sku: str | None = None
+    vendor: str | None = None
+    description: str | None = None
+    similarity: float
+    band: MatchBand
+    reason: str
+
+
+class ProcessResult(BaseModel):
+    receipt_id: int | None = None
+    status: ReceiptStatus
+    source_path: str
+    extract: ReceiptExtract | None = None
+    matches: list[MatchHit] = Field(default_factory=list)
+    error: str | None = None
+
+
+def to_cents(value: Decimal | float | int | None) -> int | None:
+    if value is None:
+        return None
+    quantized = (Decimal(str(value)) * 100).quantize(Decimal("1"), rounding=ROUND_HALF_UP)
+    return int(quantized)
+
+
+def cents_to_decimal(cents: int | None) -> Decimal | None:
+    if cents is None:
+        return None
+    return (Decimal(cents) / Decimal("100")).quantize(Decimal("0.01"))
+
+
+def extract_json_object(text: str) -> str:
+    stripped = text.strip()
+    fenced = _FENCE_RE.search(stripped)
+    if fenced:
+        stripped = fenced.group(1).strip()
+    start = stripped.find("{")
+    end = stripped.rfind("}")
+    if start < 0 or end <= start:
+        raise ValueError("no JSON object in model output")
+    return stripped[start : end + 1]
+
+
+def parse_extract_json(text: str) -> ReceiptExtract:
+    payload = json.loads(extract_json_object(text))
+    if not isinstance(payload, dict):
+        raise ValueError("extract JSON must be an object")
+    if "line_items" not in payload or payload["line_items"] is None:
+        payload["line_items"] = []
+    return ReceiptExtract.model_validate(payload)
+
+
+EXTRACT_JSON_SCHEMA: dict[str, Any] = ReceiptExtract.model_json_schema()

app/ui.py

@@ -0,0 +1,634 @@
+from __future__ import annotations
+
+import json
+import shutil
+import socket
+import threading
+import uuid
+from pathlib import Path
+
+import gradio as gr
+from fastapi import FastAPI, File, UploadFile
+from fastapi.responses import HTMLResponse, JSONResponse
+
+from app.config import CATEGORIES, DOC_KINDS, Settings, load_settings
+from app.db import (
+    add_catalog_item,
+    delete_receipt,
+    get_receipt,
+    list_catalog,
+    list_line_items,
+    list_receipts,
+    open_db,
+    update_extract,
+    update_receipt_status,
+)
+from app.pipeline import process_file
+from app.schemas import ReceiptExtract, ReceiptStatus
+from app.watcher import start_inbox_watcher
+from backends import build_embed, build_llm
+
+CSS = """
+.gradio-container {max-width: 1200px !important;}
+"""
+
+PHONE_HTML = """<!doctype html>
+<html lang="en"><head>
+<meta charset="utf-8"/>
+<meta name="viewport" content="width=device-width,initial-scale=1,viewport-fit=cover"/>
+<meta name="apple-mobile-web-app-capable" content="yes"/>
+<title>Scan a receipt</title>
+<style>
+ :root { color-scheme: dark; }
+ * { -webkit-tap-highlight-color: transparent; }
+ body{font-family:-apple-system,system-ui,sans-serif;background:#111814;color:#e8f0e8;margin:0;padding:24px}
+ .card{max-width:480px;margin:0 auto;background:#1b2420;border-radius:16px;padding:24px}
+ h1{font-size:1.5rem;margin:0 0 8px}
+ p{line-height:1.4;color:#c5d4c8}
+ /* iOS ignores taps on display:none file inputs. Cover the whole button with an opacity-0 input. */
+ .hit{position:relative;display:block;box-sizing:border-box;width:100%;margin:14px 0 0;min-height:56px;
+      border-radius:12px;font:inherit;font-weight:600;text-align:center;line-height:56px;
+      overflow:hidden;touch-action:manipulation}
+ .cam{background:#c4a35a;color:#1b1408}
+ .lib{background:#2b3a32;color:#e8f0e8;border:1px solid #4d6556}
+ .hit input[type=file]{
+   position:absolute;left:0;top:0;width:100%;height:100%;margin:0;padding:0;
+   opacity:0;font-size:24px;cursor:pointer;z-index:2}
+ .ok{color:#9fdfb2;margin-top:12px;word-break:break-word}
+ .err{color:#ff8a80;margin-top:12px;word-break:break-word}
+</style></head>
+<body><div class="card">
+<h1>Scan a receipt</h1>
+<p>Tap <b>Take photo</b> for the camera, or <b>Choose file</b> for Photos / Files. Use Safari.</p>
+<div class="hit cam"><span>Take photo</span>
+  <input id="cam" type="file" accept="image/*" capture="environment" tabindex="0"/>
+</div>
+<div class="hit lib"><span>Choose file</span>
+  <input id="lib" type="file" accept="image/*,image/heic,image/heif,application/pdf,.heic,.heif,.pdf,.jpg,.jpeg,.png" tabindex="0"/>
+</div>
+<div id="out"></div>
+<script>
+const out = document.getElementById('out');
+function add(cls, msg) {
+  const d = document.createElement('div');
+  d.className = cls;
+  d.textContent = msg;
+  out.prepend(d);
+}
+async function waitDone(jobId) {
+  add('ok', 'Uploaded. Reading the receipt on the Spark…');
+  const deadline = Date.now() + 180000;
+  while (Date.now() < deadline) {
+    await new Promise(r => setTimeout(r, 2000));
+    const r = await fetch('/api/jobs/' + jobId);
+    const j = await r.json();
+    if (j.state === 'processing') continue;
+    if (j.state === 'done') {
+      const bits = [];
+      if (j.vendor) bits.push(j.vendor);
+      if (j.total) bits.push('$' + j.total);
+      if (j.category) bits.push(j.category);
+      if (j.receipt_id != null) bits.push('#' + j.receipt_id);
+      const line = bits.join(' · ') || j.status || 'done';
+      if (j.status === 'failed' || j.status === 'needs_extract') {
+        add('err', 'Finished with issues: ' + line + (j.error ? ' — ' + j.error : ''));
+      } else {
+        add('ok', 'Completed: ' + line + '. Confirm it on the desktop Review tab.');
+      }
+      return;
+    }
+    add('err', 'Failed: ' + (j.error || j.state));
+    return;
+  }
+  add('err', 'Still processing after 3 minutes — check Review on the desktop.');
+}
+async function upload(file) {
+  if (!file) { add('err', 'No file selected'); return; }
+  add('ok', 'Uploading ' + (file.name || 'photo') + '…');
+  const body = new FormData();
+  const name = file.name && file.name !== 'image.jpg' ? file.name : ('iphone-' + Date.now() + '.jpg');
+  body.append('file', file, name);
+  try {
+    const r = await fetch('/api/inbox', { method: 'POST', body });
+    const text = await r.text();
+    let j;
+    try { j = JSON.parse(text); } catch (e) { throw new Error('Server returned ' + r.status + ': ' + text.slice(0, 180)); }
+    if (!r.ok || !j.ok) throw new Error(text.slice(0, 180));
+    await waitDone(j.job_id);
+  } catch (err) {
+    add('err', String(err));
+  }
+}
+function bind(id) {
+  const el = document.getElementById(id);
+  el.addEventListener('change', () => {
+    const files = el.files;
+    if (!files || !files.length) { add('err', 'Picker closed with no file'); return; }
+    for (const f of files) upload(f);
+    el.value = '';
+  });
+}
+bind('cam');
+bind('lib');
+</script></div></body></html>
+"""
+
+
+def _lan_ip() -> str:
+    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    try:
+        sock.connect(("192.0.2.1", 1))
+        return sock.getsockname()[0]
+    except OSError:
+        return "127.0.0.1"
+    finally:
+        sock.close()
+
+
+def _as_text(raw: object) -> str:
+    if raw is None:
+        return ""
+    if isinstance(raw, list):
+        return "\n".join(str(part) for part in raw)
+    return str(raw)
+
+
+def pretty_json(raw: object) -> str:
+    text = _as_text(raw).strip() or "{}"
+    return json.dumps(json.loads(text), indent=2)
+
+
+def parse_rid(value: object) -> int | None:
+    if value is None or value == "":
+        return None
+    try:
+        return int(float(str(value)))
+    except (TypeError, ValueError):
+        return None
+
+
+def _table_rows(data: object) -> list[list[object]]:
+    if data is None:
+        return []
+    if hasattr(data, "values"):
+        return [list(row) for row in data.values]
+    return [list(row) for row in data]
+
+
+def _image_preview(path: str) -> str | None:
+    if not path:
+        return None
+    suffix = Path(path).suffix.lower()
+    if suffix in {".jpg", ".jpeg", ".png", ".webp", ".tif", ".tiff"}:
+        return path
+    return None
+
+
+def _save_upload(settings: Settings, name: str, data: bytes) -> Path:
+    safe = Path(name).name or "upload.bin"
+    dest = settings.inbox_dir / safe
+    n = 1
+    while dest.exists():
+        dest = settings.inbox_dir / f"{dest.stem}-{n}{dest.suffix}"
+        n += 1
+    dest.write_bytes(data)
+    return dest
+
+
+_JOBS: dict[str, dict[str, object]] = {}
+_JOBS_LOCK = threading.Lock()
+
+
+def build_app(settings: Settings) -> FastAPI:
+    settings.ensure_dirs()
+    api = FastAPI(title="keys-automatic-receipt-doc-scanner")
+
+    @api.post("/api/inbox")
+    async def api_inbox(file: UploadFile = File(...)) -> JSONResponse:
+        data = await file.read()
+        dest = _save_upload(settings, file.filename or "iphone.jpg", data)
+        job_id = uuid.uuid4().hex
+        with _JOBS_LOCK:
+            _JOBS[job_id] = {"state": "processing"}
+
+        def _run() -> None:
+            try:
+                result = process_file(dest, settings)
+                extract = result.extract
+                payload: dict[str, object] = {
+                    "state": "done",
+                    "status": result.status.value,
+                    "receipt_id": result.receipt_id,
+                    "vendor": extract.vendor if extract else None,
+                    "total": str(extract.total) if extract and extract.total is not None else None,
+                    "category": extract.category if extract else None,
+                    "error": result.error,
+                }
+            except Exception as exc:
+                payload = {"state": "error", "error": str(exc)}
+            with _JOBS_LOCK:
+                _JOBS[job_id] = payload
+
+        threading.Thread(target=_run, daemon=True, name="inbox-scan").start()
+        return JSONResponse(
+            {
+                "ok": True,
+                "job_id": job_id,
+                "path": str(dest),
+                "bytes": len(data),
+                "processing": True,
+            }
+        )
+
+    @api.get("/api/jobs/{job_id}")
+    async def job_status(job_id: str) -> JSONResponse:
+        with _JOBS_LOCK:
+            job = _JOBS.get(job_id)
+        if job is None:
+            return JSONResponse({"state": "error", "error": "unknown job"}, status_code=404)
+        return JSONResponse(job)
+
+    @api.get("/phone", response_class=HTMLResponse)
+    async def phone() -> str:
+        return PHONE_HTML
+
+    def health() -> str:
+        llm = build_llm(settings)
+        embed = build_embed(settings)
+        return (
+            f"LLM {settings.llm_backend}/{settings.llm_model} @ {settings.llm_base_url} "
+            f"vision={llm.accepts_images} health={llm.health()}\n"
+            f"Embed {settings.embed_backend}/{settings.embed_model} dim={settings.embed_dim} "
+            f"health={embed.health()}\n"
+            f"Camera {settings.camera_url}  idle={settings.idle_seconds}s"
+        )
+
+    def inbox_list() -> str:
+        files = sorted(p.name for p in settings.inbox_dir.iterdir() if p.is_file())
+        return "\n".join(files) or "(empty inbox)"
+
+    def ingest(files: list) -> str:
+        if not files:
+            return "no files"
+        names = []
+        for item in files:
+            src = Path(item if isinstance(item, str) else item.name)
+            dest = settings.inbox_dir / src.name
+            shutil.copy2(src, dest)
+            names.append(dest.name)
+        return "queued: " + ", ".join(names)
+
+    def review_table() -> list[list[str]]:
+        con = open_db(settings)
+        try:
+            rows = list_receipts(con, limit=40)
+            return [
+                [
+                    str(r["id"]),
+                    r["status"],
+                    r["doc_kind"] or "",
+                    r["category"] or "",
+                    r["vendor"] or "",
+                    r["receipt_date"] or "",
+                    "" if r["total_cents"] is None else f"{r['total_cents']/100:.2f}",
+                ]
+                for r in rows
+            ]
+        finally:
+            con.close()
+
+    def _empty_load() -> tuple:
+        return (
+            None,
+            "",
+            "receipt",
+            "other",
+            "",
+            "",
+            "",
+            "",
+            "",
+            "{}",
+            [],
+            "not found",
+        )
+
+    def load_one(receipt_id: object) -> tuple:
+        rid = parse_rid(receipt_id)
+        if rid is None:
+            return _empty_load()
+        con = open_db(settings)
+        try:
+            row = get_receipt(con, rid)
+            if row is None:
+                return _empty_load()
+            try:
+                payload = pretty_json(row["extract_json"] or "{}")
+            except json.JSONDecodeError:
+                payload = row["extract_json"] or "{}"
+            extract = None
+            try:
+                extract = ReceiptExtract.model_validate_json(payload)
+            except Exception:
+                extract = None
+            lines = [
+                [
+                    item.description,
+                    "" if item.qty is None else item.qty,
+                    "" if item.unit_price is None else str(item.unit_price),
+                    "" if item.amount is None else str(item.amount),
+                    item.sku or "",
+                ]
+                for item in (extract.line_items if extract else [])
+            ]
+            if not lines:
+                lines = [
+                    [
+                        r["description"],
+                        r["qty"] if r["qty"] is not None else "",
+                        "" if r["unit_price_cents"] is None else f"{r['unit_price_cents']/100:.2f}",
+                        "" if r["amount_cents"] is None else f"{r['amount_cents']/100:.2f}",
+                        r["sku"] or "",
+                    ]
+                    for r in list_line_items(con, rid)
+                ]
+            return (
+                _image_preview(row["source_path"] or ""),
+                str(rid),
+                (extract.doc_kind if extract else row["doc_kind"]) or "receipt",
+                (extract.category if extract else row["category"]) or "other",
+                (extract.vendor if extract else row["vendor"]) or "",
+                (
+                    extract.date.isoformat()
+                    if extract and extract.date
+                    else (row["receipt_date"] or "")
+                ),
+                "" if extract is None or extract.tax is None else str(extract.tax),
+                "" if extract is None or extract.total is None else str(extract.total),
+                (extract.currency if extract else row["currency"]) or "",
+                payload,
+                lines,
+                f"loaded #{rid}",
+            )
+        finally:
+            con.close()
+
+    def load_from_table(data: object, evt: gr.SelectData) -> tuple:
+        rows = _table_rows(data)
+        index = evt.index
+        row_i = index[0] if isinstance(index, (list, tuple)) else index
+        if row_i is None or row_i < 0 or row_i >= len(rows) or not rows[row_i]:
+            return _empty_load()
+        return load_one(rows[row_i][0])
+
+    def save_extract(receipt_id: object, raw: object) -> tuple[str, list[list[str]]]:
+        rid = parse_rid(receipt_id)
+        if rid is None:
+            return "pick a receipt (click a row or enter id)", review_table()
+        try:
+            extract = ReceiptExtract.model_validate_json(pretty_json(raw))
+        except Exception as exc:
+            return f"invalid JSON: {exc}", review_table()
+        con = open_db(settings)
+        try:
+            update_extract(con, rid, extract, ReceiptStatus.needs_review)
+        finally:
+            con.close()
+        return f"saved JSON for #{rid}", review_table()
+
+    def save_fields(
+        receipt_id: object,
+        doc_kind: str,
+        category: str,
+        vendor: str,
+        receipt_date: str,
+        tax: str,
+        total: str,
+        currency: str,
+        lines: object,
+    ) -> tuple[str, str, list[list[str]]]:
+        rid = parse_rid(receipt_id)
+        if rid is None:
+            return "pick a receipt first", "{}", review_table()
+        def _num(val: object) -> str | None:
+            text = str(val).strip()
+            return None if text in {"", "None", "null"} else text
+
+        items = []
+        for row in _table_rows(lines):
+            if not row or not str(row[0]).strip():
+                continue
+            items.append(
+                {
+                    "description": str(row[0]).strip(),
+                    "qty": _num(row[1] if len(row) > 1 else None),
+                    "unit_price": _num(row[2] if len(row) > 2 else None),
+                    "amount": _num(row[3] if len(row) > 3 else None),
+                    "sku": (str(row[4]).strip() or None) if len(row) > 4 else None,
+                }
+            )
+        payload = {
+            "doc_kind": doc_kind or "receipt",
+            "category": category or "other",
+            "vendor": vendor.strip() or None,
+            "date": receipt_date.strip() or None,
+            "tax": tax.strip() or None,
+            "total": total.strip() or None,
+            "currency": currency.strip() or None,
+            "line_items": items,
+        }
+        raw = json.dumps(payload, indent=2)
+        try:
+            extract = ReceiptExtract.model_validate(payload)
+        except Exception as exc:
+            return f"invalid fields: {exc}", raw, review_table()
+        con = open_db(settings)
+        try:
+            update_extract(con, rid, extract, ReceiptStatus.needs_review)
+        finally:
+            con.close()
+        return f"saved receipt #{rid}", pretty_json(extract.model_dump_json()), review_table()
+
+    def confirm(receipt_id: object) -> tuple[str, list[list[str]]]:
+        rid = parse_rid(receipt_id)
+        if rid is None:
+            return "pick a receipt first", review_table()
+        con = open_db(settings)
+        try:
+            update_receipt_status(con, rid, ReceiptStatus.confirmed)
+        finally:
+            con.close()
+        return f"confirmed #{rid}", review_table()
+
+    def delete_one(receipt_id: object) -> tuple:
+        rid = parse_rid(receipt_id)
+        empty = _empty_load()
+        if rid is None:
+            return (*empty[:-1], "pick a receipt first", review_table())
+        con = open_db(settings)
+        try:
+            found = delete_receipt(con, rid)
+        finally:
+            con.close()
+        if not found:
+            return (*empty[:-1], f"not found #{rid}", review_table())
+        return (*empty[:-1], f"deleted #{rid}", review_table())
+
+    def catalog_table() -> list[list[str]]:
+        con = open_db(settings)
+        try:
+            return [
+                [str(r["id"]), r["sku"] or "", r["vendor"] or "", r["description"]]
+                for r in list_catalog(con)
+            ]
+        finally:
+            con.close()
+
+    def add_sku(sku: str, vendor: str, description: str) -> str:
+        if not description.strip():
+            return "description required"
+        con = open_db(settings)
+        try:
+            add_catalog_item(con, sku=sku or None, vendor=vendor or None, description=description)
+        finally:
+            con.close()
+        return "added"
+
+    def process_now(path: str) -> str:
+        if not path:
+            return "no path"
+        result = process_file(Path(path), settings)
+        return result.model_dump_json(indent=2)
+
+    with gr.Blocks(title="Receipt Studio", theme=gr.themes.Soft(primary_hue="amber"), css=CSS) as demo:
+        gr.Markdown(
+            "# Receipt Studio\n"
+            "Lamp camera · phone upload · Gemma 4 12B Unified on the GPU box (not on the Lamp)."
+        )
+        with gr.Tab("Inbox"):
+            gr.Markdown(
+                f"Drop files here or on your phone: `http://{_lan_ip()}:{settings.ui_port}/phone` "
+                f"(bind `0.0.0.0` / `RECEIPT_UI_SHARE_LAN=true`). Syncthing can also land in `inbox/`."
+            )
+            files = gr.File(label="Photos / PDFs", file_count="multiple", type="filepath")
+            ingest_btn = gr.Button("Queue in inbox")
+            ingest_out = gr.Textbox(label="Queued")
+            listing = gr.Textbox(label="Inbox", lines=8)
+            refresh = gr.Button("Refresh inbox")
+            ingest_btn.click(ingest, inputs=[files], outputs=[ingest_out]).then(
+                inbox_list, outputs=[listing]
+            )
+            refresh.click(inbox_list, outputs=[listing])
+            demo.load(inbox_list, outputs=[listing])
+        with gr.Tab("Review"):
+            gr.Markdown(
+                "Click a row to load. Edit **Kind / Category / Vendor / Date / Tax / Total** "
+                "(or the JSON), then **Save fields + lines**. "
+                "**Delete** removes a bad scan from the database and disk."
+            )
+            table = gr.Dataframe(
+                headers=["id", "status", "kind", "category", "vendor", "date", "total"],
+                datatype=["str"] * 7,
+                interactive=False,
+                wrap=True,
+            )
+            refresh_r = gr.Button("Refresh queue")
+            with gr.Row():
+                img = gr.Image(label="Scan", type="filepath")
+                with gr.Column():
+                    rid = gr.Textbox(label="Receipt id")
+                    kind = gr.Dropdown(choices=list(DOC_KINDS), label="Kind", value="receipt")
+                    category = gr.Dropdown(choices=list(CATEGORIES), label="Category", value="other")
+                    vendor = gr.Textbox(label="Vendor")
+                    receipt_date = gr.Textbox(label="Date (YYYY-MM-DD)")
+                    tax = gr.Textbox(label="Tax")
+                    total = gr.Textbox(label="Total")
+                    currency = gr.Textbox(label="Currency")
+            raw = gr.Textbox(
+                label="Extract JSON (editable)",
+                lines=18,
+                max_lines=40,
+                interactive=True,
+            )
+            lines = gr.Dataframe(
+                headers=["description", "qty", "unit_price", "amount", "sku"],
+                datatype=["str", "str", "str", "str", "str"],
+                label="Line items (editable)",
+                interactive=True,
+                wrap=True,
+            )
+            with gr.Row():
+                load_btn = gr.Button("Load id")
+                save_fields_btn = gr.Button("Save fields + lines")
+                save_json_btn = gr.Button("Save JSON")
+                ok_btn = gr.Button("Confirm")
+                del_btn = gr.Button("Delete", variant="stop")
+            msg = gr.Textbox(label="Status")
+            load_outputs = [
+                img,
+                rid,
+                kind,
+                category,
+                vendor,
+                receipt_date,
+                tax,
+                total,
+                currency,
+                raw,
+                lines,
+                msg,
+            ]
+            refresh_r.click(review_table, outputs=[table])
+            table.select(load_from_table, inputs=[table], outputs=load_outputs)
+            load_btn.click(load_one, inputs=[rid], outputs=load_outputs)
+            save_fields_btn.click(
+                save_fields,
+                inputs=[rid, kind, category, vendor, receipt_date, tax, total, currency, lines],
+                outputs=[msg, raw, table],
+            )
+            save_json_btn.click(save_extract, inputs=[rid, raw], outputs=[msg, table])
+            ok_btn.click(confirm, inputs=[rid], outputs=[msg, table])
+            del_btn.click(delete_one, inputs=[rid], outputs=load_outputs + [table])
+            demo.load(review_table, outputs=[table])
+        with gr.Tab("Catalog"):
+            cat = gr.Dataframe(headers=["id", "sku", "vendor", "description"])
+            sku = gr.Textbox(label="SKU")
+            vendor = gr.Textbox(label="Vendor")
+            desc = gr.Textbox(label="Description")
+            add_btn = gr.Button("Add SKU")
+            add_msg = gr.Textbox()
+            add_btn.click(add_sku, inputs=[sku, vendor, desc], outputs=[add_msg]).then(
+                catalog_table, outputs=[cat]
+            )
+            demo.load(catalog_table, outputs=[cat])
+        with gr.Tab("Settings"):
+            gr.Markdown(
+                "Gemma 4 12B Unified does **not** load on the Lamp (6 GB). "
+                "This UI talks to the GPU box URLs in `.env`."
+            )
+            gr.Textbox(value=health, label="Backends", every=15)
+            path = gr.Textbox(label="Process this path now")
+            run = gr.Button("Process")
+            run_out = gr.Textbox(lines=16)
+            run.click(process_now, inputs=[path], outputs=[run_out])
+
+    return gr.mount_gradio_app(api, demo, path="/")
+
+
+def main() -> None:
+    settings = load_settings()
+    if settings.ui_share_lan:
+        settings.ui_host = "0.0.0.0"
+    start_inbox_watcher(settings)
+    import uvicorn
+
+    uvicorn.run(
+        build_app(settings),
+        host=settings.ui_host,
+        port=settings.ui_port,
+        log_level="info",
+    )
+
+
+if __name__ == "__main__":
+    main()

app/watcher.py

@@ -0,0 +1,133 @@
+from __future__ import annotations
+
+import threading
+import time
+from collections.abc import Callable
+from dataclasses import dataclass
+from pathlib import Path
+
+from watchdog.events import FileSystemEvent, FileSystemEventHandler
+from watchdog.observers import Observer
+
+from app.config import ACCEPTED_SUFFIXES, Settings
+from app.pipeline import process_file
+
+OnBatch = Callable[[list[Path]], None]
+
+
+def is_ignored(path: Path) -> bool:
+    name = path.name
+    if name.startswith("."):
+        return True
+    if name.endswith(".tmp") or name.endswith(".part"):
+        return True
+    if ".syncthing." in name or name.startswith(".syncthing"):
+        return True
+    return path.suffix.lower() not in ACCEPTED_SUFFIXES
+
+
+@dataclass
+class _State:
+    sig: tuple[int, float]
+    last_change: float
+
+
+class IdleBatchWatcher:
+    """Settle files until size+mtime are unchanged for idle_seconds, then batch."""
+
+    def __init__(
+        self,
+        inbox: Path,
+        *,
+        idle_seconds: float = 30.0,
+        on_batch: OnBatch | None = None,
+    ) -> None:
+        self.inbox = Path(inbox)
+        self.idle_seconds = idle_seconds
+        self.on_batch = on_batch
+        self._state: dict[Path, _State] = {}
+        self._lock = threading.Lock()
+        self._running = False
+        self._observer: Observer | None = None
+        self._thread: threading.Thread | None = None
+
+    def note(self, path: Path, now: float) -> None:
+        if not path.is_file() or is_ignored(path):
+            return
+        stat = path.stat()
+        sig = (stat.st_size, stat.st_mtime)
+        with self._lock:
+            prev = self._state.get(path)
+            if prev is None or prev.sig != sig:
+                self._state[path] = _State(sig=sig, last_change=now)
+
+    def tick(self, now: float | None = None) -> list[Path]:
+        clock = time.monotonic() if now is None else now
+        self.inbox.mkdir(parents=True, exist_ok=True)
+        for path in self.inbox.iterdir():
+            self.note(path, clock)
+        ready: list[Path] = []
+        with self._lock:
+            for path, state in list(self._state.items()):
+                if not path.is_file():
+                    self._state.pop(path, None)
+                    continue
+                if clock - state.last_change >= self.idle_seconds:
+                    ready.append(path)
+                    self._state.pop(path, None)
+        if ready and self.on_batch is not None:
+            self.on_batch(ready)
+        return ready
+
+    def start(self) -> None:
+        if self._running:
+            return
+        self._running = True
+        handler = _Handler(self)
+        observer = Observer()
+        observer.schedule(handler, str(self.inbox), recursive=False)
+        observer.start()
+        self._observer = observer
+        self._thread = threading.Thread(target=self._loop, daemon=True)
+        self._thread.start()
+
+    def stop(self) -> None:
+        self._running = False
+        if self._observer is not None:
+            self._observer.stop()
+            self._observer.join(timeout=2)
+            self._observer = None
+
+    def _loop(self) -> None:
+        while self._running:
+            self.tick()
+            time.sleep(min(0.25, max(0.05, self.idle_seconds / 4)))
+
+
+class _Handler(FileSystemEventHandler):
+    def __init__(self, watcher: IdleBatchWatcher) -> None:
+        self.watcher = watcher
+
+    def on_any_event(self, event: FileSystemEvent) -> None:
+        if event.is_directory:
+            return
+        path = Path(str(event.src_path))
+        self.watcher.note(path, time.monotonic())
+
+
+def process_batch(paths: list[Path], settings: Settings) -> None:
+    for path in paths:
+        try:
+            process_file(path, settings)
+        except Exception:
+            continue
+
+
+def start_inbox_watcher(settings: Settings) -> IdleBatchWatcher:
+    watcher = IdleBatchWatcher(
+        settings.inbox_dir,
+        idle_seconds=settings.idle_seconds,
+        on_batch=lambda paths: process_batch(paths, settings),
+    )
+    watcher.start()
+    return watcher

backends/__init__.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+from app.config import Settings
+from backends.apple import AppleEmbed, AppleLLM, AppleOCR
+from backends.base import EmbedBackend, LLMBackend, OCRBackend
+from backends.cpu import CpuOCR
+from backends.gemma import GemmaEmbed, GemmaLLM
+from backends.nvidia import NvidiaEmbed, NvidiaLLM, NvidiaOCR
+from backends.ollama import OllamaEmbed, OllamaLLM, OllamaOCR
+
+__all__ = [
+    "EmbedBackend",
+    "LLMBackend",
+    "OCRBackend",
+    "build_embed",
+    "build_llm",
+    "build_ocr",
+]
+
+
+def build_llm(settings: Settings, *, client=None) -> LLMBackend:
+    name = settings.llm_backend.lower().strip()
+    if name in {"gemma", "gemma4", "unified"}:
+        return GemmaLLM(settings, client=client)
+    if name in {"nvidia", "vllm", "qwen", "qwen38"}:
+        return NvidiaLLM(settings, client=client)
+    if name in {"ollama", "lightning"}:
+        return OllamaLLM(settings, client=client)
+    if name == "apple":
+        return AppleLLM()
+    raise ValueError(f"unknown llm_backend: {settings.llm_backend}")
+
+
+def build_embed(settings: Settings, *, client=None) -> EmbedBackend:
+    name = settings.embed_backend.lower().strip()
+    if name in {"omni", "gemma", "gemma4"}:
+        return GemmaEmbed(settings, client=client)
+    if name in {"nvidia", "nemotron", "vllm"}:
+        return NvidiaEmbed(settings, client=client)
+    if name in {"openai", "ollama"}:
+        return OllamaEmbed(settings, client=client)
+    if name == "apple":
+        return AppleEmbed()
+    raise ValueError(f"unknown embed_backend: {settings.embed_backend}")
+
+
+def build_ocr(settings: Settings, *, client=None) -> OCRBackend | None:
+    name = settings.ocr_backend.lower().strip()
+    if name in {"", "none", "off"}:
+        return None
+    if name == "ollama":
+        return OllamaOCR(settings, client=client)
+    if name == "nvidia":
+        return NvidiaOCR()
+    if name == "apple":
+        return AppleOCR()
+    if name == "cpu":
+        return CpuOCR()
+    raise ValueError(f"unknown ocr_backend: {settings.ocr_backend}")

backends/apple.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from backends.base import InputType, OCRResult
+
+
+class _Stub:
+    name = "apple"
+
+    def health(self) -> bool:
+        return False
+
+
+class AppleLLM(_Stub):
+    accepts_images = True
+
+    def complete_json(self, *, system: str, user: str, image_jpeg: bytes | None = None) -> str:
+        raise NotImplementedError("Apple Vision / Foundation Models backend is Phase 2.")
+
+
+class AppleEmbed(_Stub):
+    dim = 0
+
+    def embed(self, texts: list[str], *, input_type: InputType) -> list[list[float]]:
+        raise NotImplementedError("Apple embed backend is Phase 2.")
+
+
+class AppleOCR(_Stub):
+    def ocr(self, path: Path) -> OCRResult:
+        raise NotImplementedError(f"Apple Vision OCR is Phase 2 (path={path}).")

backends/base.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Literal, Protocol, runtime_checkable
+
+InputType = Literal["query", "passage"]
+
+
+class OCRResult:
+    __slots__ = ("text", "engine")
+
+    def __init__(self, text: str, engine: str) -> None:
+        self.text = text
+        self.engine = engine
+
+
+@runtime_checkable
+class LLMBackend(Protocol):
+    accepts_images: bool
+    name: str
+
+    def complete_json(
+        self,
+        *,
+        system: str,
+        user: str,
+        image_jpeg: bytes | None = None,
+    ) -> str: ...
+
+    def health(self) -> bool: ...
+
+
+@runtime_checkable
+class EmbedBackend(Protocol):
+    name: str
+    dim: int
+
+    def embed(self, texts: list[str], *, input_type: InputType) -> list[list[float]]: ...
+
+    def health(self) -> bool: ...
+
+
+@runtime_checkable
+class OCRBackend(Protocol):
+    name: str
+
+    def ocr(self, path: Path) -> OCRResult: ...

backends/cpu.py

@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from backends.base import OCRResult
+
+
+class CpuOCR:
+    name = "cpu"
+
+    def ocr(self, path: Path) -> OCRResult:
+        raise NotImplementedError(
+            f"RapidOCR CPU fallback is Phase 2 and must not auto-download weights (path={path})."
+        )

backends/gemma.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+import httpx
+
+from app.config import Settings
+from backends.openai_compat import OpenAICompatEmbed, OpenAICompatLLM
+
+
+class GemmaLLM(OpenAICompatLLM):
+    """Gemma 4 12B Unified on vLLM — vision extract. Runs on the GPU box, not the Lamp."""
+
+    def __init__(self, settings: Settings, *, client: httpx.Client | None = None) -> None:
+        super().__init__(
+            settings,
+            name="gemma4-unified",
+            accepts_images=settings.llm_accepts_images,
+            extra_body={},
+            client=client,
+        )
+
+
+class GemmaEmbed(OpenAICompatEmbed):
+    """Same Gemma 4 12B Unified server, /v1/embeddings (omni). Dim 3840."""
+
+    def __init__(self, settings: Settings, *, client: httpx.Client | None = None) -> None:
+        super().__init__(settings, name="gemma4-omni-embed", client=client)

backends/nvidia.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import httpx
+
+from app.config import Settings
+from backends.base import OCRResult
+from backends.openai_compat import OpenAICompatEmbed, OpenAICompatLLM
+
+
+class NvidiaLLM(OpenAICompatLLM):
+    """vLLM Qwen3.8-27B ADay777 (or other NVIDIA VLM). Thinking off."""
+
+    def __init__(self, settings: Settings, *, client: httpx.Client | None = None) -> None:
+        super().__init__(
+            settings,
+            name="nvidia-vllm",
+            accepts_images=settings.llm_accepts_images,
+            extra_body={"chat_template_kwargs": {"enable_thinking": False}},
+            client=client,
+        )
+
+
+class NvidiaEmbed(OpenAICompatEmbed):
+    def __init__(self, settings: Settings, *, client: httpx.Client | None = None) -> None:
+        super().__init__(settings, name="nemotron-embed", client=client)
+
+
+class NvidiaOCR:
+    name = "nvidia-ocr"
+
+    def ocr(self, path: Path) -> OCRResult:
+        raise NotImplementedError(
+            "Nemotron OCR v2 is not in this skill. Use Gemma/Qwen vision extract "
+            f"(path={path})."
+        )

backends/ollama.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+import base64
+from pathlib import Path
+
+import httpx
+
+from app.config import Settings
+from backends.base import OCRResult
+from backends.openai_compat import OpenAICompatEmbed, OpenAICompatLLM
+
+
+def _is_lightning(model: str) -> bool:
+    return "lightning" in model.lower()
+
+
+class OllamaLLM(OpenAICompatLLM):
+    def __init__(self, settings: Settings, *, client: httpx.Client | None = None) -> None:
+        accepts = settings.llm_accepts_images and not _is_lightning(settings.llm_model)
+        super().__init__(
+            settings,
+            name="ollama",
+            accepts_images=accepts,
+            extra_body={},
+            client=client,
+        )
+
+
+class OllamaEmbed(OpenAICompatEmbed):
+    def __init__(self, settings: Settings, *, client: httpx.Client | None = None) -> None:
+        super().__init__(settings, name="ollama-embed", client=client)
+
+
+class OllamaOCR:
+    name = "ollama-ocr"
+
+    def __init__(self, settings: Settings, *, client: httpx.Client | None = None) -> None:
+        self.settings = settings
+        self._client = client
+
+    def ocr(self, path: Path) -> OCRResult:
+        if not self.settings.ocr_model:
+            raise NotImplementedError("set RECEIPT_OCR_MODEL for Ollama vision OCR")
+        jpeg = path.read_bytes()
+        b64 = base64.b64encode(jpeg).decode("ascii")
+        own = self._client is None
+        http = self._client or httpx.Client(
+            base_url=self.settings.ocr_base_url.rstrip("/"),
+            timeout=self.settings.llm_timeout_s,
+            headers={"Authorization": f"Bearer {self.settings.ocr_api_key}"},
+        )
+        try:
+            response = http.post(
+                "/chat/completions",
+                json={
+                    "model": self.settings.ocr_model,
+                    "messages": [
+                        {
+                            "role": "user",
+                            "content": [
+                                {
+                                    "type": "image_url",
+                                    "image_url": {
+                                        "url": f"data:image/jpeg;base64,{b64}"
+                                    },
+                                },
+                                {
+                                    "type": "text",
+                                    "text": "Transcribe this document verbatim.",
+                                },
+                            ],
+                        }
+                    ],
+                    "temperature": 0,
+                    "max_tokens": 4096,
+                },
+            )
+            response.raise_for_status()
+            text = response.json()["choices"][0]["message"]["content"]
+        finally:
+            if own:
+                http.close()
+        return OCRResult(text=text or "", engine=self.name)

backends/openai_compat.py

@@ -0,0 +1,208 @@
+from __future__ import annotations
+
+import base64
+import math
+from typing import Any
+
+import httpx
+
+from app.config import Settings
+from backends.base import InputType
+
+
+class EmbedDimensionError(ValueError):
+    pass
+
+
+class OpenAICompatError(RuntimeError):
+    pass
+
+
+def _normalize_base(url: str) -> str:
+    return url.rstrip("/")
+
+
+class OpenAICompatClient:
+    def __init__(
+        self,
+        *,
+        base_url: str,
+        api_key: str,
+        timeout_s: float,
+        client: httpx.Client | None = None,
+    ) -> None:
+        self.base_url = _normalize_base(base_url)
+        self.api_key = api_key
+        self._owns = client is None
+        self._client = client or httpx.Client(
+            base_url=self.base_url,
+            timeout=timeout_s,
+            headers={"Authorization": f"Bearer {api_key}"},
+        )
+
+    def close(self) -> None:
+        if self._owns:
+            self._client.close()
+
+    def health(self) -> bool:
+        try:
+            response = self._client.get("/models")
+            return response.status_code < 500
+        except httpx.HTTPError:
+            return False
+
+    def chat_completions(self, body: dict[str, Any]) -> dict[str, Any]:
+        response = self._client.post("/chat/completions", json=body)
+        try:
+            response.raise_for_status()
+        except httpx.HTTPStatusError as exc:
+            raise OpenAICompatError(
+                f"chat/completions {exc.response.status_code}: {exc.response.text[:500]}"
+            ) from exc
+        return response.json()
+
+    def embeddings(self, body: dict[str, Any]) -> dict[str, Any]:
+        response = self._client.post("/embeddings", json=body)
+        if response.status_code == 404:
+            # vLLM pooling runner
+            response = self._client.post("/pooling", json={**body, "task": "embed"})
+        try:
+            response.raise_for_status()
+        except httpx.HTTPStatusError as exc:
+            raise OpenAICompatError(
+                f"embeddings {exc.response.status_code}: {exc.response.text[:500]}"
+            ) from exc
+        return response.json()
+
+
+class OpenAICompatLLM:
+    def __init__(
+        self,
+        settings: Settings,
+        *,
+        name: str,
+        accepts_images: bool,
+        extra_body: dict[str, Any] | None = None,
+        client: httpx.Client | None = None,
+    ) -> None:
+        self.name = name
+        self.accepts_images = accepts_images
+        self.model = settings.llm_model
+        self.max_tokens = settings.llm_max_tokens
+        self.extra_body = extra_body or {}
+        self._http = OpenAICompatClient(
+            base_url=settings.llm_base_url,
+            api_key=settings.llm_api_key,
+            timeout_s=settings.llm_timeout_s,
+            client=client,
+        )
+
+    def health(self) -> bool:
+        return self._http.health()
+
+    def complete_json(
+        self,
+        *,
+        system: str,
+        user: str,
+        image_jpeg: bytes | None = None,
+    ) -> str:
+        if self.accepts_images and image_jpeg:
+            b64 = base64.b64encode(image_jpeg).decode("ascii")
+            user_content: Any = [
+                {
+                    "type": "image_url",
+                    "image_url": {"url": f"data:image/jpeg;base64,{b64}"},
+                },
+                {"type": "text", "text": user},
+            ]
+        else:
+            user_content = user
+        body: dict[str, Any] = {
+            "model": self.model,
+            "messages": [
+                {"role": "system", "content": system},
+                {"role": "user", "content": user_content},
+            ],
+            "temperature": 0,
+            "max_tokens": self.max_tokens,
+            "response_format": {"type": "json_object"},
+        }
+        body.update(self.extra_body)
+        payload = self._http.chat_completions(body)
+        try:
+            return str(payload["choices"][0]["message"]["content"] or "")
+        except (KeyError, IndexError, TypeError) as exc:
+            raise OpenAICompatError(f"unexpected chat response: {payload!r}") from exc
+
+
+def apply_embed_prefix(text: str, input_type: InputType, *, enabled: bool) -> str:
+    if not enabled:
+        return text
+    prefix = "query: " if input_type == "query" else "passage: "
+    stripped = text.lstrip()
+    if stripped.startswith("query:") or stripped.startswith("passage:"):
+        return text
+    return prefix + text
+
+
+def l2_normalize(vec: list[float]) -> list[float]:
+    norm = math.sqrt(sum(x * x for x in vec)) or 1.0
+    return [x / norm for x in vec]
+
+
+def parse_embedding_payload(payload: dict[str, Any]) -> list[list[float]]:
+    if "data" in payload:
+        rows = sorted(payload["data"], key=lambda row: row.get("index", 0))
+        return [list(map(float, row["embedding"])) for row in rows]
+    if "embeddings" in payload:
+        embeddings = payload["embeddings"]
+        if isinstance(embeddings, dict) and "float" in embeddings:
+            embeddings = embeddings["float"]
+        return [list(map(float, row)) for row in embeddings]
+    raise OpenAICompatError(f"unexpected embed response keys: {list(payload)}")
+
+
+class OpenAICompatEmbed:
+    def __init__(
+        self,
+        settings: Settings,
+        *,
+        name: str,
+        client: httpx.Client | None = None,
+    ) -> None:
+        self.name = name
+        self.dim = settings.embed_dim
+        self.model = settings.embed_model
+        self.prefix = settings.embed_prefix
+        self._http = OpenAICompatClient(
+            base_url=settings.embed_base_url or settings.llm_base_url,
+            api_key=settings.embed_api_key or settings.llm_api_key,
+            timeout_s=settings.embed_timeout_s,
+            client=client,
+        )
+
+    def health(self) -> bool:
+        return self._http.health()
+
+    def embed(self, texts: list[str], *, input_type: InputType) -> list[list[float]]:
+        if not texts:
+            return []
+        prefixed = [
+            apply_embed_prefix(text, input_type, enabled=self.prefix) for text in texts
+        ]
+        body = {
+            "model": self.model,
+            "input": prefixed,
+            "encoding_format": "float",
+            "input_type": input_type,
+        }
+        payload = self._http.embeddings(body)
+        vectors = [l2_normalize(vec) for vec in parse_embedding_payload(payload)]
+        for vec in vectors:
+            if len(vec) != self.dim:
+                raise EmbedDimensionError(
+                    f"embed dim {len(vec)} != configured {self.dim}. "
+                    "Never mix Gemma-3840 and Nemotron-2048 in one index."
+                )
+        return vectors

oneshot.bat

@@ -0,0 +1,11 @@
+@echo off
+REM Windows: Gemma 4 12B-it must already be served (this PC or a Spark).
+REM This script installs the app venv and opens the UI.
+cd /d "%~dp0"
+py -3.12 -m venv .venv 2>nul || python -m venv .venv
+.venv\Scripts\python.exe -m pip install -q -U pip
+.venv\Scripts\python.exe -m pip install -q -e ".[dev]"
+if not exist .env copy .env.example .env
+echo If Gemma is on another box, set RECEIPT_LLM_BASE_URL in .env to http://SPARK:8080/v1
+.venv\Scripts\python.exe -m app.launch
+if errorlevel 1 pause

oneshot.sh

@@ -0,0 +1,112 @@
+#!/usr/bin/env bash
+# =============================================================================
+# ONE-SHOT: keys-Auto Receipts Studio (iPhone / may add Autonomous Lamp Skill)
+#   1. Python 3.12 venv + app
+#   2. Gemma 4 12B-it weights (skip if already on disk)
+#   3. vLLM serve  --gpu-memory-utilization 0.15  FP8  max-model-len 8192
+#   4. Gradio UI on the LAN + print phone URL
+# Idempotent. Re-run anytime. Never raises GPU util above 0.85.
+# =============================================================================
+set -euo pipefail
+ROOT="$(cd "$(dirname "$0")" && pwd)"
+cd "$ROOT"
+
+MODEL_ID="${RECEIPT_HF_MODEL:-google/gemma-4-12B-it}"
+MODEL_DIR="${RECEIPT_GEMMA_PATH:-$HOME/models-gemma4-12b-it}"
+PORT_LLM="${RECEIPT_VLLM_PORT:-8080}"
+PORT_UI="${RECEIPT_UI_PORT:-7860}"
+UTIL="${RECEIPT_GPU_MEMORY_UTILIZATION:-0.15}"
+
+say(){ printf '\n\033[1;36m==> %s\033[0m\n' "$*"; }
+die(){ printf '\n\033[1;31mFAILED: %s\033[0m\n' "$*" >&2; exit 1; }
+
+python3 -c 'import sys; assert sys.version_info >= (3,12), sys.version' \
+  || die "Python 3.12+ required"
+
+if python3 -c "u=float('$UTIL'); assert u<=0.85" 2>/dev/null; then :; else
+  die "gpu_memory_utilization $UTIL > 0.85 hard cap"
+fi
+
+say "1/5 venv + install"
+if [[ ! -x .venv/bin/python ]]; then
+  python3 -m venv .venv
+fi
+.venv/bin/pip install -q -U pip
+.venv/bin/pip install -q -e ".[dev]"
+[[ -f .env ]] || cp .env.example .env
+
+say "2/5 Gemma 4 12B-it weights → $MODEL_DIR"
+if [[ -f "$MODEL_DIR/config.json" ]] && ls "$MODEL_DIR"/*.safetensors >/dev/null 2>&1; then
+  echo "  present"
+else
+  command -v hf >/dev/null || .venv/bin/pip install -q huggingface_hub
+  mkdir -p "$MODEL_DIR"
+  hf download "$MODEL_ID" --local-dir "$MODEL_DIR" \
+    || python3 - "$MODEL_ID" "$MODEL_DIR" <<'PY' || die "weight download failed (hf auth login)"
+import sys
+from huggingface_hub import snapshot_download
+snapshot_download(sys.argv[1], local_dir=sys.argv[2])
+print("  downloaded")
+PY
+fi
+
+say "3/5 vLLM Gemma (util=$UTIL FP8, :$PORT_LLM)"
+if curl -sf -m3 "http://127.0.0.1:$PORT_LLM/v1/models" >/dev/null 2>&1; then
+  echo "  already serving"
+else
+  command -v vllm >/dev/null || die "vllm not on PATH (pip install vllm, or use this Spark's install)"
+  mkdir -p data
+  nohup bash "$ROOT/scripts/serve-gemma.sh" >> data/vllm-gemma.log 2>&1 &
+  echo "  pid $!  log data/vllm-gemma.log"
+fi
+
+say "4/5 wait until Gemma answers /v1/models (first boot compiles kernels)"
+ok=0
+for i in $(seq 1 120); do
+  if curl -sf -m3 "http://127.0.0.1:$PORT_LLM/v1/models" >/dev/null 2>&1; then
+    echo "  healthy ($i)"
+    ok=1
+    break
+  fi
+  sleep 5
+done
+[[ "$ok" = 1 ]] || die "vLLM not healthy — tail data/vllm-gemma.log"
+
+say "5/5 UI on LAN :$PORT_UI"
+export RECEIPT_UI_SHARE_LAN=true
+export RECEIPT_LLM_BASE_URL="http://127.0.0.1:${PORT_LLM}/v1"
+export RECEIPT_EMBED_BASE_URL="http://127.0.0.1:${PORT_LLM}/v1"
+export RECEIPT_LLM_MODEL="$MODEL_ID"
+export RECEIPT_EMBED_MODEL="$MODEL_ID"
+export RECEIPT_EMBED_DIM=3840
+export RECEIPT_EMBED_BACKEND=omni
+if curl -sf -m2 "http://127.0.0.1:$PORT_UI/phone" >/dev/null 2>&1; then
+  echo "  UI already up"
+else
+  nohup .venv/bin/python -m app.cli ui >> data/ui.log 2>&1 &
+  echo "  pid $!"
+  for i in $(seq 1 40); do
+    curl -sf -m2 "http://127.0.0.1:$PORT_UI/phone" >/dev/null 2>&1 && break
+    sleep 0.25
+  done
+fi
+
+LAN="$(python3 - <<'PY'
+import socket
+s=socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+try:
+    s.connect(("192.0.2.1",1)); print(s.getsockname()[0])
+except OSError:
+    print("127.0.0.1")
+finally:
+    s.close()
+PY
+)"
+
+printf '\n\033[1;32m✅ READY\033[0m  keys-Auto Receipts Studio\n'
+printf '   Review (this machine):  http://127.0.0.1:%s\n' "$PORT_UI"
+printf '   iPhone Safari:          http://%s:%s/phone\n' "$LAN" "$PORT_UI"
+printf '   Gemma /v1:              http://127.0.0.1:%s/v1  model %s  util=%s\n' "$PORT_LLM" "$MODEL_ID" "$UTIL"
+printf '   Desktop launcher:       bash scripts/install-launcher.sh\n'
+printf '\n   Hold a receipt up → Take photo on the phone page.\n'
+printf '   Lamp skill (optional):  skills/keys-receipt-scanner/  — 12B does not fit in 6GB RAM.\n'

pyproject.toml

@@ -0,0 +1,40 @@
+[build-system]
+requires = ["setuptools>=69"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "keys-automatic-receipt-doc-scanner"
+version = "1.0.0a1"
+description = "Lamp camera + Qwen3.8 vision extract + sqlite-vec receipt/doc scanner"
+readme = "README.md"
+requires-python = ">=3.12"
+license = { text = "Apache-2.0" }
+authors = [{ name = "keys" }]
+dependencies = [
+  "watchdog>=4.0",
+  "httpx>=0.27",
+  "pydantic>=2.8",
+  "pydantic-settings>=2.4",
+  "gradio>=4.44",
+  "sqlite-vec>=0.1.6",
+  "pillow>=10.4",
+  "pillow-heif>=0.18",
+  "pypdfium2>=4.30",
+  "uvicorn>=0.30",
+  "python-multipart>=0.0.9",
+]
+
+[project.optional-dependencies]
+dev = ["pytest>=8.3"]
+
+[project.scripts]
+keys-scan = "app.cli:main"
+receipt-studio = "app.ui:main"
+
+[tool.setuptools.packages.find]
+include = ["app*", "backends*"]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+pythonpath = ["."]
+addopts = "-q"

requirements-dev.txt

@@ -0,0 +1,2 @@
+-r requirements.txt
+pytest>=8.3

requirements.txt

@@ -0,0 +1,11 @@
+watchdog>=4.0
+httpx>=0.27
+pydantic>=2.8
+pydantic-settings>=2.4
+gradio>=4.44
+sqlite-vec>=0.1.6
+pillow>=10.4
+pillow-heif>=0.18
+pypdfium2>=4.30
+uvicorn>=0.30
+python-multipart>=0.0.9

scripts/Receipt-Studio.desktop

@@ -0,0 +1,12 @@
+[Desktop Entry]
+Version=1.0
+Type=Application
+Name=Receipt Studio
+Comment=Open the receipt scanner Review UI (phone upload on LAN)
+Exec=bash -c '"%k"'
+# %k is unreliable across desktops; the installer copies a wrapper that cds to the repo.
+TryExec=
+Icon=applications-office
+Terminal=false
+Categories=Office;Utility;
+StartupNotify=true

scripts/install-launcher.sh

@@ -0,0 +1,33 @@
+#!/usr/bin/env bash
+# Linux: one-click icon on the Desktop. macOS: aliases start-ui.command.
+set -euo pipefail
+ROOT="$(cd "$(dirname "$0")/.." && pwd)"
+chmod +x "$ROOT/scripts/start-ui.sh" "$ROOT/scripts/start-ui.command"
+
+if [[ "$(uname -s)" == "Darwin" ]]; then
+  DEST="$HOME/Desktop/Receipt Studio.command"
+  ln -sf "$ROOT/scripts/start-ui.command" "$DEST"
+  chmod +x "$DEST"
+  echo "Mac: double-click Desktop/Receipt Studio.command (first time: right-click → Open)"
+  exit 0
+fi
+
+DESKTOP="${XDG_DESKTOP_DIR:-$HOME/Desktop}"
+mkdir -p "$DESKTOP" "$HOME/.local/share/applications"
+APP="$DESKTOP/Receipt Studio.desktop"
+cat > "$APP" <<EOF
+[Desktop Entry]
+Version=1.0
+Type=Application
+Name=Receipt Studio
+Comment=Open the receipt scanner Review UI
+Exec=$ROOT/scripts/start-ui.sh
+Path=$ROOT
+Icon=applications-office
+Terminal=false
+Categories=Office;Utility;
+StartupNotify=true
+EOF
+chmod +x "$APP"
+cp "$APP" "$HOME/.local/share/applications/receipt-studio.desktop"
+echo "Linux: double-click Desktop/Receipt Studio.desktop (Mark as trusted / Allow Launching if asked)"

scripts/serve-gemma.sh

@@ -0,0 +1,39 @@
+#!/usr/bin/env bash
+# Gemma 4 12B Unified. GPU util 0.15 (~18.3GiB of 121.7GiB). Never above 0.85.
+# BF16 weights ~23GB cannot fit; FP8 ~12.5GB + KV in the rest.
+set -euo pipefail
+MODEL="${RECEIPT_GEMMA_PATH:-$HOME/models-gemma4-12b-it}"
+NAME="${RECEIPT_LLM_MODEL:-google/gemma-4-12B-it}"
+HOST="${RECEIPT_VLLM_HOST:-0.0.0.0}"
+PORT="${RECEIPT_VLLM_PORT:-8080}"
+# User-set 0.15. Fleet hard cap 0.85.
+UTIL="${RECEIPT_GPU_MEMORY_UTILIZATION:-0.15}"
+MAX_LEN="${RECEIPT_VLLM_MAX_MODEL_LEN:-8192}"
+
+if [[ ! -f "$MODEL/config.json" ]]; then
+  echo "Gemma checkpoint not found: $MODEL" >&2
+  exit 1
+fi
+
+python3 - "$UTIL" <<'PY'
+import sys
+util = float(sys.argv[1])
+if util > 0.85:
+    raise SystemExit(f"gpu_memory_utilization {util} > 0.85 hard cap")
+print(f"util={util:.4f}  pool~{util*121.69:.1f}GiB of 121.7GiB")
+print("context: max-model-len default 8192 (receipts). KV estimate at 0.15:")
+print("  conservative (48-layer full attn fp16): ~12k tokens")
+print("  hybrid (8 full + 40 sliding-1024):     ~65k tokens")
+print("  model native max_position_embeddings:  262144 (not reachable at 0.15)")
+PY
+
+exec vllm serve "$MODEL" \
+  --served-model-name "$NAME" \
+  --host "$HOST" \
+  --port "$PORT" \
+  --gpu-memory-utilization "$UTIL" \
+  --max-model-len "$MAX_LEN" \
+  --max-num-seqs 2 \
+  --max-num-batched-tokens 2048 \
+  --quantization fp8 \
+  --enforce-eager

scripts/start-ui.bat

@@ -0,0 +1,10 @@
+@echo off
+REM Windows: double-click this file.
+cd /d "%~dp0\.."
+if not exist ".venv\Scripts\python.exe" (
+  echo Creating .venv (one time)...
+  py -3.12 -m venv .venv || python -m venv .venv
+  .venv\Scripts\python.exe -m pip install -e ".[dev]"
+)
+.venv\Scripts\python.exe -m app.launch
+if errorlevel 1 pause

scripts/start-ui.command

@@ -0,0 +1,4 @@
+#!/bin/bash
+# macOS: double-click this file (first time: right-click → Open).
+cd "$(dirname "$0")"
+exec ./start-ui.sh

scripts/start-ui.sh

@@ -0,0 +1,11 @@
+#!/usr/bin/env bash
+# Linux / macOS: double-click start-ui.command on Mac, or run this script.
+set -euo pipefail
+ROOT="$(cd "$(dirname "$0")/.." && pwd)"
+cd "$ROOT"
+if [[ ! -x .venv/bin/python ]]; then
+  echo "Creating .venv (one time)…"
+  python3 -m venv .venv
+  .venv/bin/pip install -e ".[dev]"
+fi
+exec .venv/bin/python -m app.launch

skills/keys-receipt-scanner/SKILL.md

@@ -0,0 +1,82 @@
+---
+name: keys-receipt-scanner
+description: Scan receipts and paper documents with the Lamp camera, extract totals/line items, categorize, and store them. Use when the user says scan this, scan the receipt, snap this invoice, log this expense, what did I just buy, hold this up, look at this receipt, capture this document, add this to expenses, or shows paper to the camera.
+---
+
+# keys-receipt-scanner
+
+Built-in Autonomous OS skill. The **Lamp camera** is the eye. The **GPU box** runs Gemma 4 12B Unified (omni vision + embed). Do **not** load 12B weights on the robot — Lamp has 6 GB RAM.
+
+## When to use
+
+- User holds up a receipt, invoice, statement, or letter
+- "Scan this", "log this expense", "what did this cost", "save this document"
+- Phone/Syncthing drop is handled by the same `scan` CLI on the GPU box; on Lamp, still snapshot then scan
+
+Do **not** use for "what do you see" about the room (that is `camera`) or privacy toggles (`camera` disable/enable).
+
+## Capture (Lamp HAL)
+
+Reuse `[vision-image] <path>` if this turn already has one. Otherwise:
+
+```bash
+curl -s "http://127.0.0.1:5001/camera/snapshot?save=true&width=1280&quality=85"
+```
+
+Read `path` from the JSON. Receipts need **1280** px, not 768 — small print.
+
+If they say it is on the desk: curl `POST http://127.0.0.1:5001/servo/aim` with `{"direction":"down"}` **before** snapshot (`[HW:…]` would move after the photo).
+
+Then:
+
+```
+[HW:/emotion:{"emotion":"curious","intensity":0.6}]
+```
+
+## Extract + store
+
+From the skill checkout / install prefix (repo root on the GPU box, or `/opt/keys-receipt-scanner` on Lamp if you copied the package):
+
+```bash
+python -m app.cli scan --image "$SNAP_PATH"
+```
+
+The CLI POSTs the JPEG to Gemma 4 12B Unified (`RECEIPT_LLM_BASE_URL`, default `http://127.0.0.1:8080/v1`) with `image_url`, categorizes, embeds on the same omni server (dim 3840), writes SQLite.
+
+On the Lamp, set `RECEIPT_LLM_BASE_URL` / `RECEIPT_EMBED_BASE_URL` to the GPU box (LAN). Never `api.x.ai`. Never Lightning with an image.
+
+## Speak
+
+After JSON comes back, say in the user's language:
+
+- kind + category + vendor
+- date and total (with currency)
+- 1–2 notable line items
+- match band if SKU auto/review
+
+Then:
+
+```
+[HW:/emotion:{"emotion":"acknowledge","intensity":0.7}]
+```
+
+If `status` is `needs_extract` / `failed`, say you could not read it and ask them to hold it flatter / closer. Do not invent totals.
+
+## Query
+
+```bash
+python -m app.cli query --category groceries --limit 10
+python -m app.cli show <id>
+```
+
+## Privacy
+
+Only snapshot when they asked to scan paper. Camera off stays a `camera` skill concern. `/camera/snapshot` auto-enables for the frame.
+
+## Fit
+
+| Piece | Runs on |
+|---|---|
+| This SKILL.md + snapshot curl | Lamp (Autonomous OS) |
+| `app.cli scan` HTTP client + SQLite | Lamp **or** GPU box |
+| Gemma 4 12B Unified weights | GPU box only |

skills/keys-receipt-scanner/references/hardware.md

@@ -0,0 +1,7 @@
+# Hardware split
+
+Lamp: 6 GB RAM. This skill + HAL snapshot only.
+
+Gemma 4 12B Unified (`hidden_size` 3840) and Qwen3.8-27B run on the GPU box.
+
+Camera: `GET http://127.0.0.1:5001/camera/snapshot?save=true&width=1280&quality=85`

skills/keys-receipt-scanner/scripts/scan.py

@@ -0,0 +1,19 @@
+#!/usr/bin/env python3
+"""Lamp/agent entry: python skills/keys-receipt-scanner/scripts/scan.py --image PATH"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[3]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from app.cli import main
+
+if __name__ == "__main__":
+    argv = sys.argv[1:]
+    if not argv or argv[0] not in {"scan", "query", "show", "snapshot", "ui"}:
+        argv = ["scan", *argv]
+    main(argv)

skills/keys-receipt-scanner/skill.json

@@ -0,0 +1,6 @@
+{
+  "name": "keys-receipt-scanner",
+  "capabilities": ["vision"],
+  "category": "Productivity",
+  "tags": ["receipt", "document", "camera", "expense", "ocr", "scanner"]
+}

tests/conftest.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+import io
+from pathlib import Path
+
+import pytest
+from PIL import Image
+
+from app.config import Settings, load_settings
+
+
+def tiny_jpeg() -> bytes:
+    image = Image.new("RGB", (16, 16), (240, 240, 240))
+    buf = io.BytesIO()
+    image.save(buf, format="JPEG")
+    return buf.getvalue()
+
+
+@pytest.fixture
+def settings(tmp_path: Path) -> Settings:
+    return load_settings(
+        root_dir=tmp_path,
+        data_dir=tmp_path / "data",
+        inbox_dir=tmp_path / "inbox",
+        processing_dir=tmp_path / "processing",
+        processed_dir=tmp_path / "processed",
+        failed_dir=tmp_path / "failed",
+        exports_dir=tmp_path / "exports",
+        idle_seconds=0.05,
+        llm_backend="gemma",
+        llm_base_url="http://llm.test/v1",
+        llm_model="google/gemma-4-12B-it",
+        embed_backend="omni",
+        embed_base_url="http://llm.test/v1",
+        embed_model="google/gemma-4-12B-it",
+        embed_dim=8,
+        embed_prefix=True,
+    )

tests/test_camera.py

@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import httpx
+
+from app.camera import snapshot
+from app.config import Settings
+
+
+def test_snapshot_reads_path(settings: Settings, tmp_path: Path) -> None:
+    saved = tmp_path / "snap.jpg"
+    saved.write_bytes(b"x")
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert "width=1280" in str(request.url)
+        return httpx.Response(200, json={"path": str(saved)})
+
+    client = httpx.Client(transport=httpx.MockTransport(handler), base_url="http://hal.test")
+    settings = settings.model_copy(update={"camera_url": "http://hal.test"})
+    assert snapshot(settings, client=client) == saved

tests/test_db.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+import pytest
+
+from app.config import Settings
+from app.db import (
+    EmbedIndexError,
+    VecLoadError,
+    connect,
+    delete_receipt,
+    init_schema,
+    insert_receipt,
+)
+from app.schemas import ReceiptExtract, ReceiptStatus
+
+
+def test_schema_and_meta(settings: Settings) -> None:
+    try:
+        con = connect(settings.db_path)
+        init_schema(con, settings)
+    except VecLoadError:
+        pytest.skip("sqlite-vec not loadable")
+    tables = {
+        row[0]
+        for row in con.execute("SELECT name FROM sqlite_master WHERE type IN ('table','view')")
+    }
+    assert "receipts" in tables
+    assert "catalog" in tables
+    rid = insert_receipt(
+        con,
+        source_path="x.jpg",
+        sha256="abc",
+        status=ReceiptStatus.needs_review,
+        extract=ReceiptExtract(vendor="A", category="dining"),
+    )
+    assert rid == 1
+    other = settings.model_copy(update={"embed_dim": 3840, "embed_model": "other"})
+    with pytest.raises(EmbedIndexError):
+        init_schema(con, other)
+    assert delete_receipt(con, rid, unlink_file=False) is True
+    assert con.execute("SELECT COUNT(*) AS n FROM receipts").fetchone()["n"] == 0
+    assert delete_receipt(con, rid, unlink_file=False) is False
+    con.close()

tests/test_embed_prefix.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+import httpx
+
+from app.config import Settings
+from backends.gemma import GemmaEmbed
+from backends.openai_compat import apply_embed_prefix, EmbedDimensionError
+import pytest
+
+
+def test_prefix_query_and_passage() -> None:
+    assert apply_embed_prefix("milk", "query", enabled=True) == "query: milk"
+    assert apply_embed_prefix("milk 2%", "passage", enabled=True) == "passage: milk 2%"
+    assert apply_embed_prefix("query: already", "query", enabled=True) == "query: already"
+
+
+def test_embed_request_body_has_prefix_and_input_type(settings: Settings) -> None:
+    recorded: list[tuple[str, dict]] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        recorded.append((request.url.path, request.read().decode()))
+        return httpx.Response(
+            200,
+            json={
+                "data": [
+                    {"index": 0, "embedding": [0.5, 0.5, 0.5, 0.5, 0.5, 0.5, 0.5, 0.5]}
+                ]
+            },
+        )
+
+    client = httpx.Client(transport=httpx.MockTransport(handler), base_url="http://llm.test/v1")
+    embed = GemmaEmbed(settings, client=client)
+    vecs = embed.embed(["milk"], input_type="query")
+    assert len(vecs[0]) == 8
+    path, body = recorded[0]
+    assert path.endswith("/embeddings")
+    assert "query: milk" in body
+    assert '"input_type": "query"' in body or '"input_type":"query"' in body
+
+
+def test_wrong_dim_rejected(settings: Settings) -> None:
+    def handler(_request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json={"data": [{"index": 0, "embedding": [1.0, 0.0]}]})
+
+    client = httpx.Client(transport=httpx.MockTransport(handler), base_url="http://llm.test/v1")
+    embed = GemmaEmbed(settings, client=client)
+    with pytest.raises(EmbedDimensionError):
+        embed.embed(["x"], input_type="passage")

tests/test_extract_vision.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+import json
+
+import httpx
+
+from app.config import Settings
+from app.extract import extract_receipt
+from backends.gemma import GemmaLLM
+from backends.ollama import OllamaLLM
+from tests.conftest import tiny_jpeg
+
+EXTRACT = {
+    "doc_kind": "receipt",
+    "category": "dining",
+    "vendor": "Cafe",
+    "date": "2026-08-21",
+    "tax": 0.5,
+    "total": 8.0,
+    "currency": "USD",
+    "line_items": [],
+}
+
+
+def _chat_handler(sink: list[dict]):
+    def handler(request: httpx.Request) -> httpx.Response:
+        payload = json.loads(request.content)
+        sink.append(payload)
+        return httpx.Response(
+            200,
+            json={"choices": [{"message": {"content": json.dumps(EXTRACT)}}]},
+        )
+
+    return handler
+
+
+def test_gemma_sends_image_url(settings: Settings) -> None:
+    sink: list[dict] = []
+    client = httpx.Client(
+        transport=httpx.MockTransport(_chat_handler(sink)),
+        base_url="http://llm.test/v1",
+    )
+    llm = GemmaLLM(settings, client=client)
+    extract = extract_receipt(
+        llm, settings=settings, image_jpeg=tiny_jpeg(), ocr_text=None
+    )
+    assert extract.vendor == "Cafe"
+    content = sink[0]["messages"][1]["content"]
+    assert isinstance(content, list)
+    kinds = {part["type"] for part in content}
+    assert "image_url" in kinds
+    url = next(part["image_url"]["url"] for part in content if part["type"] == "image_url")
+    assert url.startswith("data:image/jpeg;base64,")
+
+
+def test_lightning_never_sends_image(settings: Settings) -> None:
+    settings = settings.model_copy(
+        update={
+            "llm_backend": "ollama",
+            "llm_model": "nemotron-3.5-lightning",
+            "llm_accepts_images": False,
+        }
+    )
+    sink: list[dict] = []
+    client = httpx.Client(
+        transport=httpx.MockTransport(_chat_handler(sink)),
+        base_url="http://llm.test/v1",
+    )
+    llm = OllamaLLM(settings, client=client)
+    assert llm.accepts_images is False
+    extract = extract_receipt(
+        llm, settings=settings, image_jpeg=tiny_jpeg(), ocr_text="Cafe 8.00"
+    )
+    assert extract.total is not None
+    content = sink[0]["messages"][1]["content"]
+    assert isinstance(content, str)
+    dumped = json.dumps(sink[0])
+    assert "image_url" not in dumped
+    assert "data:image" not in dumped