Give spoken words a memory with an instant attached

An audio file used to produce no memory at all: _ingest_file returned an empty
batch for any suffix it did not know, with no log and no reason. It now raises
UnsupportedFile when you name the file and records a SkippedFile when it was a
directory walk, so a refusal is visible either way.

Speech is measured, not promised. Over 210.5 seconds of real Video-MME audio,
faster-whisper-tiny on CPU int8 transcribed at 51x realtime and reached word
error rate 0.242 against community subtitles nobody here wrote (a caption track
drops words a transcript keeps, so that is an upper bound). 66 speech atoms now
carry 550 word-level timestamps, 0 words out of order and 0 starting before
their own segment. The snapshot is 78.2 MB, under the 500 MB cap, and the report
is bench/reports/ASR.md.

Two library facts were measured rather than assumed. faster-whisper's own audio
loader is broken against PyAV 19 (av.open got an unexpected keyword
'metadata_errors') and it declares av with no version bound, so asr.decode_audio
decodes with PyAV directly and hands the model a float32 array instead of
pinning an old av with no cp314 wheel. And av.Error does not exist on PyAV 19 -
the real base is av.error.FFmpegError - so that is the only thing caught and my
own ValueErrors keep their messages.

Files over the 500-line rule were split rather than excused: ingestion.py (748)
into ingestion_common / ingestion_media / ingestion_repository / ingestion, and
cli.py (544) into cli_senses plus cli_common, following the cli_transactive
register pattern. MediaIngest declares _source_record abstract, so a half-wired
host fails at construction instead of returning None.

Verification: 207 tests pass; ruff clean on the 13 touched files; mypy clean on
them, and repo-wide mypy fell from 29 errors to 24 (5 fixed, none added);
pyright reports 0 errors on the touched files.

CHECKLIST.md

@@ -153,10 +153,35 @@
   - 单屏内：MRR 0.908（recall@10 是 1.000 但 chance 也是 1.000，所以那一行只有 MRR 有意义）
   - 本轮没测双语：ScreenSpot-v2 的指令全是英文，报告里写明了
   - 验收：无文字图像被图像查询召回有 recall 数字；`pytest tests/test_visual_tokens.py` 绿
-- [ ] **p1-2** 音频摄入
+- [x] **p1-2** 音频摄入
-  - [ ] p1-2.1 `Modality` 加 audio/transcript；未知扩展名不再静默跳过
+  - [x] p1-2.1 `Modality` 加 audio；未知扩展名不再静默跳过
-  - [ ] p1-2.2 ASR 接入，转写 atom 带词级时间戳
+  - [x] p1-2.2 ASR 接入，转写 atom 带词级时间戳
-  - 验收：`pytest tests/test_audio_ingest.py` 绿；ASR 跑通或明确 `DEFERRED`
+  - 验收：`pytest backend/tests/test_asr.py` 绿（8 个测试）；ASR **实跑**，不是 DEFERRED
+  - 实测证据（`bench/reports/ASR.md`，commit 03ca357）：3 段 Video-MME 真视频 210.5 秒音频，
+    对着第三方字幕 WER **0.242**，CPU int8 **51× 实时**，产出 66 个 speech atom、550 个词级时间戳，
+    0 个词时序倒错、0 个词早于自己所属片段。模型 `Systran/faster-whisper-tiny`（78.2MB，
+    model.bin sha256 dcb76c6586fc06cb…），远低于 500MB 上限。字幕不是逐字稿，所以 WER 是上界。
+  - 偏差与根因：
+    - 验收路径 `tests/test_audio_ingest.py` 不存在，实际是 `backend/tests/test_asr.py`。
+    - `Modality` 只加 `audio`，没加 `transcript`：一份转写就是带时间戳的 audio atom，
+      两个成员是同一件事的两个名字。
+    - faster-whisper 自带的音频加载器在 PyAV 19 上是坏的（`av.open(..., metadata_errors=)` →
+      `TypeError`），而它声明 `av` 时没写版本界。所以 `asr.decode_audio()` 自己用 PyAV 解码成
+      float32/16kHz 单声道再喂给 `model.transcribe()`，不降级 pin 一个没有 cp314 wheel 的旧 av。
+    - PyAV 19 的异常面是量出来的不是猜的：`av.Error` 不存在（第一版会死在 except 里），
+      真实基类是 `av.error.FFmpegError`；只 catch 它，我自己的 `ValueError` 消息才不会被吞。
+    - 未知扩展名：显式给路径 → 抛 `UnsupportedFile`（“no reader for .xyz”）；目录扫描 →
+      记进 `IngestResponse.skipped` 带原因，继续摄入能读的。以前是返回空 batch，无声无息。
+    - 视频现在除了帧 atom 还会产出 speech atom（同一个 mp4：18 帧 + 31 段语音）。
+    - 顺手把 p2-4 提前做了：`ingestion.py` 748 行超了 500 规则，拆成
+      `ingestion_common.py`(99) / `ingestion_media.py`(ABC mixin, 285) / `ingestion_repository.py`(346) /
+      `ingestion.py`(193)；`cli.py` 544 行也超了，按已有的 `cli_transactive.register` 模式拆出
+      `cli_senses.py`(84) 和 `cli_common.py`(json_dump)，`cli.py` 现在 479 行。
+    - 拆完顺手修掉搬过来的旧类型债：`MediaIngest._source_record` 改成 `@abstractmethod`
+      （半接好的 host 现在在构造时就失败，不会静默返回 None）、PDF 页循环与 bbox 减法、
+      `page.get_text` 的 union 收窄；`pyproject.toml` 里失效的 `fitz.*` mypy override 换成 `faster_whisper.*`。
+    - 全仓 mypy 从 29 个错降到 **24**（本任务净修 5 个，没新增）；改动文件 pyright **0 错**、
+      ruff 干净、207 个测试绿。
 - [ ] **p1-3** adapter 加载
   - [ ] p1-3.1 按卡带/场景解析 6 个 adapter，不再只有 `runtime.py:317`
   - [ ] p1-3.2 加载结果可观测：日志 + `GET /api/runtime/adapters`

backend/src/ely_eye/asr.py

@@ -0,0 +1,129 @@
+"""Speech to text with word-level timings, so a spoken memory can be found and pointed at."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+
+from .config import Settings, get_settings
+
+DOWNLOAD_HINT = (
+    "fetch a faster-whisper CTranslate2 snapshot and set ELY_EYE_ASR_MODEL_DIR "
+    "to the directory holding it"
+)
+DEFERRED = "deferred: no speech model snapshot"
+MODEL_GRAPH = "model.bin"
+SAMPLE_RATE = 16_000
+
+
+@dataclass(frozen=True)
+class SpokenWord:
+    word: str
+    start: float
+    end: float
+    confidence: float
+
+
+@dataclass(frozen=True)
+class SpeechSegment:
+    text: str
+    start: float
+    end: float
+    words: tuple[SpokenWord, ...]
+    confidence: float
+
+
+def decode_audio(path: Path) -> np.ndarray:
+    """Decode any container PyAV reads into mono 16 kHz floats, the shape the model wants.
+
+    The decode lives here rather than inside faster-whisper because its own decoder calls
+    av.open(metadata_errors=...), a keyword PyAV 19 no longer accepts. PyAV 19 raises its
+    failures as OSError and ValueError subclasses, so FFmpegError is the only thing caught
+    here and a caller's own error keeps its own message.
+    """
+    import av
+    from av.error import FFmpegError
+
+    frames: list[np.ndarray] = []
+    try:
+        with av.open(str(path)) as container:
+            audio_streams = container.streams.audio
+            if not audio_streams:
+                raise ValueError(f"{path}: the file holds no audio track")
+            resampler = av.AudioResampler(format="flt", layout="mono", rate=SAMPLE_RATE)
+            for chunk in container.decode(audio_streams[0]):
+                if not isinstance(chunk, av.AudioFrame):
+                    continue
+                for resampled in resampler.resample(chunk):
+                    frames.append(np.asarray(resampled.to_ndarray()).reshape(-1))
+    except FFmpegError as error:
+        raise ValueError(f"{path}: the audio cannot be decoded ({error})") from error
+    if not frames:
+        raise ValueError(f"{path}: the audio track is empty")
+    return np.concatenate(frames).astype(np.float32)
+
+
+def _seconds(value: Any) -> float:
+    return float(value)
+
+
+def _confidence(words: list[Any]) -> float:
+    scores = [float(word.probability) for word in words if getattr(word, "probability", None) is not None]
+    if not scores:
+        return 0.0
+    return min(sum(scores) / len(scores), 1.0)
+
+
+def _segment(segment: Any) -> SpeechSegment:
+    words = list(getattr(segment, "words", None) or [])
+    return SpeechSegment(
+        text=segment.text.strip(),
+        start=_seconds(segment.start),
+        end=_seconds(segment.end),
+        words=tuple(
+            SpokenWord(
+                word=spoken.word.strip(),
+                start=_seconds(spoken.start),
+                end=_seconds(spoken.end),
+                confidence=float(getattr(spoken, "probability", 0.0) or 0.0),
+            )
+            for spoken in words
+        ),
+        confidence=_confidence(words),
+    )
+
+
+class SpeechTranscriber:
+    """Whisper on CTranslate2: 89 seconds of speech in 1.5 seconds of CPU, int8."""
+
+    def __init__(self, settings: Settings | None = None) -> None:
+        self.settings = settings or get_settings()
+        self._model: Any = None
+
+    def model_dir(self) -> Path | None:
+        return self.settings.asr_model_dir
+
+    def available(self) -> bool:
+        directory = self.model_dir()
+        return directory is not None and (directory / MODEL_GRAPH).exists()
+
+    def transcribe(self, path: Path) -> list[SpeechSegment]:
+        """Words with the instant each was spoken; the timings are the point of the feature."""
+        directory = self.model_dir()
+        if directory is None:
+            raise SystemExit(f"no speech model snapshot is configured; {DOWNLOAD_HINT}")
+        if not (directory / MODEL_GRAPH).exists():
+            raise SystemExit(f"{directory} holds no {MODEL_GRAPH}; {DOWNLOAD_HINT}")
+        model = self._load(directory)
+        segments, _info = model.transcribe(decode_audio(path), word_timestamps=True)
+        return [_segment(segment) for segment in segments]
+
+    def _load(self, directory: Path) -> Any:
+        if self._model is None:
+            from faster_whisper import WhisperModel
+
+            self._model = WhisperModel(str(directory), device="cpu", compute_type="int8")
+        return self._model

backend/src/ely_eye/cli.py

@@ -16,11 +16,12 @@ from .benchmarks import (
 )
 from .cartridge import CartridgeService
 from .cartridge_assets import CartridgeAssetService
+from .cli_common import json_dump
+from .cli_senses import register as register_sense_commands
+from .cli_senses import senses_report
 from .cli_transactive import register as register_transactive_commands
-from .config import Settings, get_settings
+from .config import get_settings
 from .db import Database
-from .describe import DEFERRED, VisionDescriber
-from .embeddings import EmbeddingService
 from .ingestion import IngestionService
 from .memory import MemoryService
 from .profiles import runtime_profile_report
@@ -32,18 +33,12 @@ from .retrieval import RetrievalService
 from .runtime import QwenRuntime
 from .schemas import MemoryOrigin, SearchFilters, TimePrecision, TrainingAdapterKind
 from .training import TrainingService
-from .visual_index import VisualIndex
 
 app = typer.Typer(no_args_is_help=True)
 
 
-def json_dump(value: object) -> str:
-    import json
-
-    return json.dumps(value, ensure_ascii=False, indent=2)
-
-
 register_transactive_commands(app, lambda: get_settings())
+register_sense_commands(app, lambda: get_settings())
 
 
 @app.command()
@@ -51,29 +46,6 @@ def serve(host: str = "127.0.0.1", port: int = 8765) -> None:
     uvicorn.run(create_app(), host=host, port=port)
 
 
-def _senses(settings: Settings, db: Database) -> dict[str, dict[str, object]]:
-    """What this machine can see, and what it has to pretend about.
-
-    A missing snapshot is reported, not hidden: an unrun capability is a plan, and the status
-    output is where an operator learns which claims about this install are real today.
-    """
-    visual = VisualIndex(settings, db)
-    describer = VisionDescriber(settings)
-    index_present = visual.index_path.exists() and visual.map_path.exists()
-    return {
-        "picture_search": {
-            "model_dir": str(visual.model_dir() or ""),
-            "model_ready": visual.available(),
-            "index": "present" if index_present else "absent",
-        },
-        "picture_description": {
-            "model_dir": str(describer.model_dir() or ""),
-            "model_ready": describer.available(),
-            "note": "" if describer.available() else DEFERRED,
-        },
-    }
-
-
 @app.command()
 def status() -> None:
     settings = get_settings()
@@ -91,7 +63,7 @@ def status() -> None:
                 ),
                 "cartridges": db.cartridge_count(),
                 "data_home": str(settings.home),
-                "senses": _senses(settings, db),
+                "senses": senses_report(settings, db),
             }
         )
     )
@@ -144,21 +116,6 @@ def search(
     typer.echo(result.model_dump_json(indent=2))
 
 
-@app.command()
-def describe_image(path: Path) -> None:
-    """Say what a picture shows; names the setup step when no vision model is installed."""
-    settings = get_settings()
-    typer.echo(json_dump({"image": str(path), "caption": VisionDescriber(settings).describe(path)}))
-
-
-@app.command()
-def index_visual() -> None:
-    """Encode every image atom so a picture or a sentence can find it."""
-    settings = get_settings()
-    database = Database(settings)
-    typer.echo(json_dump(VisualIndex(settings, database).build()))
-
-
 @app.command()
 def remember(
     text: str,
@@ -213,13 +170,6 @@ def forget(atom_id: str = "", source_id: str = "", reason: str = "") -> None:
     typer.echo(report.model_dump_json(indent=2))
 
 
-@app.command("build-dense-index")
-def build_dense_index() -> None:
-    index = EmbeddingService(get_settings()).build_index()
-    dimensions = int(index.vectors.shape[1]) if index.vectors.size else 0
-    typer.echo(json_dump({"atoms": len(index.atom_ids), "dimensions": dimensions}))
-
-
 @app.command("list-cartridges")
 def list_cartridges() -> None:
     for manifest in CartridgeService(get_settings()).list_manifests():

backend/src/ely_eye/cli_common.py

@@ -0,0 +1,9 @@
+"""Shared plumbing for the CLI faces, so every command answers in JSON a pipe can read."""
+
+from __future__ import annotations
+
+import json
+
+
+def json_dump(value: object) -> str:
+    return json.dumps(value, ensure_ascii=False, indent=2)

backend/src/ely_eye/cli_senses.py

@@ -0,0 +1,84 @@
+"""The senses: what this machine can see and hear, and the commands that exercise them.
+
+Each command here depends on a model snapshot the operator supplies. The report says which
+snapshots are present so a claim about this install can be checked instead of believed.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import asdict
+from pathlib import Path
+from typing import Any
+
+import typer
+
+from .asr import DEFERRED as SPEECH_DEFERRED
+from .asr import SpeechTranscriber
+from .config import Settings
+from .db import Database
+from .describe import DEFERRED, VisionDescriber
+from .embeddings import EmbeddingService
+from .visual_index import VisualIndex
+
+
+def senses_report(settings: Settings, db: Database) -> dict[str, dict[str, object]]:
+    """What this machine can see, and what it has to pretend about.
+
+    A missing snapshot is reported, not hidden: an unrun capability is a plan, and the status
+    output is where an operator learns which claims about this install are real today.
+    """
+    visual = VisualIndex(settings, db)
+    describer = VisionDescriber(settings)
+    transcriber = SpeechTranscriber(settings)
+    index_present = visual.index_path.exists() and visual.map_path.exists()
+    return {
+        "picture_search": {
+            "model_dir": str(visual.model_dir() or ""),
+            "model_ready": visual.available(),
+            "index": "present" if index_present else "absent",
+        },
+        "picture_description": {
+            "model_dir": str(describer.model_dir() or ""),
+            "model_ready": describer.available(),
+            "note": "" if describer.available() else DEFERRED,
+        },
+        "speech": {
+            "model_dir": str(transcriber.model_dir() or ""),
+            "model_ready": transcriber.available(),
+            "note": "" if transcriber.available() else SPEECH_DEFERRED,
+        },
+    }
+
+
+def register(app: typer.Typer, settings: Callable[[], Settings]) -> None:
+    """Add the media commands. The settings thunk lets one patch in a test redirect every command."""
+
+    def json_dump(value: Any) -> str:
+        import json
+
+        return json.dumps(value, ensure_ascii=False, indent=2)
+
+    @app.command()
+    def transcribe(path: Path) -> None:
+        """Words with the instant each was spoken; the snapshot stays the operator's to supply."""
+        segments = SpeechTranscriber(settings()).transcribe(path)
+        typer.echo(json_dump([asdict(segment) for segment in segments]))
+
+    @app.command("describe-image")
+    def describe_image(path: Path) -> None:
+        """Say what a picture shows; names the setup step when no vision model is installed."""
+        typer.echo(json_dump({"image": str(path), "caption": VisionDescriber(settings()).describe(path)}))
+
+    @app.command("index-visual")
+    def index_visual() -> None:
+        """Encode every image atom so a picture or a sentence can find it."""
+        database = Database(settings())
+        typer.echo(json_dump(VisualIndex(settings(), database).build()))
+
+    @app.command("build-dense-index")
+    def build_dense_index() -> None:
+        """Encode every atom into the dense index the text retriever falls back to."""
+        index = EmbeddingService(settings()).build_index()
+        dimensions = int(index.vectors.shape[1]) if index.vectors.size else 0
+        typer.echo(json_dump({"atoms": len(index.atom_ids), "dimensions": dimensions}))

backend/src/ely_eye/config.py

@@ -17,10 +17,11 @@ class Settings(BaseSettings):
     model_dir: Path | None = None
     kv_cache_dtype: str = "auto"
     planner_model_id: str = "Qwen/Qwen3.5-0.8B"
-    embedding_model_id: str = "Qwen/Qwen3-VL-Embedding-8B"
+    embedding_model_id: str = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
     clip_model_dir: Path | None = None
     vlm_model_dir: Path | None = None
     vlm_max_new_tokens: int = 96
+    asr_model_dir: Path | None = None
     adapter_dir: Path | None = None
     runtime_backend: Literal["sglang", "transformers"] = "transformers"
     sglang_base_url: str = "http://127.0.0.1:8000/v1"

backend/src/ely_eye/ingestion.py

@@ -1,28 +1,31 @@
+"""Ingest anything the store can hold, and say out loud what it refused to read."""
+
 from __future__ import annotations
 
-import ast
-import json
 import os
-import re
-import subprocess
-import tempfile
-from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
-import cv2
+from .asr import SpeechTranscriber
-import pymupdf as fitz
-from PIL import Image
-
 from .config import Settings, get_settings
 from .db import Database
-from .describe import DEFERRED, VisionDescriber
+from .describe import VisionDescriber
-from .ocr import read_text_from_image
+from .ingestion_common import (
+    SUPPORTED_AUDIO_EXTENSIONS,
+    SUPPORTED_CODE_EXTENSIONS,
+    SUPPORTED_IMAGE_EXTENSIONS,
+    SUPPORTED_TEXT_EXTENSIONS,
+    SUPPORTED_VIDEO_EXTENSIONS,
+    IngestBatch,
+    UnsupportedFile,
+)
+from .ingestion_media import MediaIngest
+from .ingestion_repository import RepositoryIngest, extract_symbols
 from .schemas import (
     EvidenceAtom,
     IngestResponse,
-    LayoutBox,
     Modality,
+    SkippedFile,
     SourceRecord,
     TrustScores,
 )
@@ -30,62 +33,12 @@ from .storage import (
     ObjectStore,
     guess_mime,
     safe_source_id,
-    sha256_bytes,
     sha256_file,
     token_equivalent,
-    visual_token_equivalent,
 )
 
-SUPPORTED_TEXT_EXTENSIONS = {
-    ".txt",
-    ".md",
-    ".json",
-    ".jsonl",
-    ".yaml",
-    ".yml",
-    ".toml",
-    ".csv",
-    ".html",
-    ".css",
-    ".xml",
-}
 
-SUPPORTED_CODE_EXTENSIONS = {
+class IngestionService(MediaIngest, RepositoryIngest):
-    ".py",
-    ".ts",
-    ".tsx",
-    ".js",
-    ".jsx",
-    ".rs",
-    ".go",
-    ".java",
-    ".cpp",
-    ".c",
-    ".h",
-    ".hpp",
-    ".cs",
-    ".swift",
-    ".kt",
-    ".sql",
-}
-
-SUPPORTED_IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".webp", ".bmp", ".tif", ".tiff"}
-SUPPORTED_VIDEO_EXTENSIONS = {".mp4", ".mov", ".mkv", ".avi", ".webm", ".m4v"}
-REPOSITORY_INDEX_LIMIT = 5000
-REPOSITORY_COMMIT_LIMIT = 50
-
-
-@dataclass
-class IngestBatch:
-    sources: list[SourceRecord]
-    atoms: list[EvidenceAtom]
-
-    @property
-    def token_equivalent(self) -> int:
-        return sum(atom.token_equivalent for atom in self.atoms)
-
-
-class IngestionService:
     def __init__(
         self,
         settings: Settings | None = None,
@@ -96,6 +49,7 @@ class IngestionService:
         self.db = database or Database(self.settings)
         self.objects = object_store or ObjectStore(self.settings)
         self.describer = VisionDescriber(self.settings)
+        self.transcriber = SpeechTranscriber(self.settings)
 
     def ingest_path(self, path: Path, cartridge_name: str | None = None) -> IngestResponse:
         path = path.expanduser().resolve()
@@ -103,8 +57,16 @@ class IngestionService:
             raise FileNotFoundError(path)
         files = self._discover_files(path)
         batch = IngestBatch(sources=[], atoms=[])
+        skipped: list[SkippedFile] = []
+        explicit = path.is_file()
         for file_path in files:
-            file_batch = self._ingest_file(file_path)
+            try:
+                file_batch = self._ingest_file(file_path)
+            except UnsupportedFile as error:
+                if explicit:
+                    raise
+                skipped.append(SkippedFile(path=str(file_path), reason=str(error)))
+                continue
             batch.sources.extend(file_batch.sources)
             batch.atoms.extend(file_batch.atoms)
         if path.is_dir():
@@ -132,6 +94,7 @@ class IngestionService:
             cartridge_id=cartridge_id,
             token_equivalent=batch.token_equivalent,
             sources=batch.sources,
+            skipped=skipped,
         )
 
     def _discover_files(self, path: Path) -> list[Path]:
@@ -163,11 +126,13 @@ class IngestionService:
             return self._ingest_image(path)
         if suffix in SUPPORTED_VIDEO_EXTENSIONS:
             return self._ingest_video(path)
+        if suffix in SUPPORTED_AUDIO_EXTENSIONS:
+            return self._ingest_audio(path)
         if suffix in SUPPORTED_CODE_EXTENSIONS:
             return self._ingest_code(path)
         if suffix in SUPPORTED_TEXT_EXTENSIONS:
             return self._ingest_text(path, Modality.text)
-        return IngestBatch(sources=[], atoms=[])
+        raise UnsupportedFile(f"no reader for {suffix or 'a name with no extension'}")
 
     def _source_record(self, path: Path, kind: str, metadata: dict[str, Any] | None = None) -> SourceRecord:
         digest = sha256_file(path)
@@ -207,475 +172,3 @@ class IngestionService:
         batch.atoms[0].metadata["line_count"] = source_text.count("\n") + 1
         return batch
 
-    def _ingest_image(self, path: Path) -> IngestBatch:
-        source = self._source_record(path, self._image_kind(path), metadata=image_metadata(path))
-        image_ref = self.objects.put_file(path, "images")
-        ocr = read_text_from_image(path)
-        caption = ""
-        caption_note = ""
-        if not ocr.text.strip():
-            # Words already on the picture are free; only a wordless one is worth a model call.
-            try:
-                caption = self.describer.describe_or_none(path) or ""
-            except (OSError, RuntimeError, ValueError) as error:
-                caption, caption_note = "", f"failed: {error}"
-            else:
-                caption_note = "vlm" if caption else DEFERRED
-        text = caption or ocr.text
-        width, height = image_size(path)
-        atom = EvidenceAtom(
-            atom_id=f"{source.source_id}:image",
-            modality=Modality.ui_screenshot if source.kind == "ui_screenshot" else Modality.image,
-            source_id=source.source_id,
-            source=str(path),
-            text=text,
-            image_ref=image_ref,
-            layout=LayoutBox(x=0, y=0, w=width, h=height),
-            trust=TrustScores(parser=1.0, ocr=ocr.confidence, model=0.6 if caption else 0.0),
-            token_equivalent=token_equivalent(text) + visual_token_equivalent(width, height),
-            metadata={
-                "ocr_regions": ocr.regions,
-                "caption": caption_note,
-                **image_metadata(path),
-            },
-        )
-        return IngestBatch(sources=[source], atoms=[atom])
-
-    def _ingest_pdf(self, path: Path) -> IngestBatch:
-        source = self._source_record(path, "pdf")
-        atoms: list[EvidenceAtom] = []
-        with fitz.open(path) as document:
-            source.metadata["page_count"] = document.page_count
-            for page_index, page in enumerate(document, start=1):
-                rect = page.rect
-                text = page.get_text("text").strip()
-                pixmap = page.get_pixmap(matrix=fitz.Matrix(1.5, 1.5), alpha=False)
-                image_bytes = pixmap.tobytes("png")
-                image_ref = self.objects.put_bytes(image_bytes, ".png", "pdf_pages")
-                if not text:
-                    with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as temp_file:
-                        temp_file.write(image_bytes)
-                        temp_path = Path(temp_file.name)
-                    try:
-                        ocr = read_text_from_image(temp_path)
-                        text = ocr.text
-                        ocr_confidence = ocr.confidence
-                    finally:
-                        temp_path.unlink(missing_ok=True)
-                else:
-                    ocr_confidence = 0.0
-                atoms.append(
-                    EvidenceAtom(
-                        atom_id=f"{source.source_id}:page:{page_index:05d}",
-                        modality=Modality.pdf_page,
-                        source_id=source.source_id,
-                        source=f"{path}#page={page_index}",
-                        text=text,
-                        image_ref=image_ref,
-                        layout=LayoutBox(x=0, y=0, w=float(rect.width), h=float(rect.height), page=page_index),
-                        trust=TrustScores(parser=1.0, ocr=ocr_confidence, model=0.0),
-                        token_equivalent=token_equivalent(text) + max(1, int(rect.width * rect.height) // 50_000),
-                        metadata={"page": page_index, "width": rect.width, "height": rect.height},
-                    )
-                )
-                atoms.extend(self._pdf_region_atoms(path, source.source_id, page_index, page))
-        return IngestBatch(sources=[source], atoms=atoms)
-
-    def _pdf_region_atoms(
-        self, path: Path, source_id: str, page_index: int, page: fitz.Page
-    ) -> list[EvidenceAtom]:
-        regions: list[EvidenceAtom] = []
-        blocks = page.get_text("blocks")
-        for block_index, block in enumerate(blocks):
-            if len(block) < 5:
-                continue
-            x0, y0, x1, y1, text = block[:5]
-            text = str(text).strip()
-            if not text:
-                continue
-            regions.append(
-                EvidenceAtom(
-                    atom_id=f"{source_id}:page:{page_index:05d}:region:{block_index:04d}",
-                    modality=Modality.pdf_region,
-                    source_id=source_id,
-                    source=f"{path}#page={page_index}&region={block_index}",
-                    text=text,
-                    layout=LayoutBox(x=float(x0), y=float(y0), w=float(x1 - x0), h=float(y1 - y0), page=page_index),
-                    trust=TrustScores(parser=1.0, ocr=0.0, model=0.0),
-                    token_equivalent=token_equivalent(text),
-                    metadata={"page": page_index, "block": block_index},
-                )
-            )
-        return regions
-
-    def _ingest_video(self, path: Path) -> IngestBatch:
-        source = self._source_record(path, "video")
-        capture = cv2.VideoCapture(str(path))
-        if not capture.isOpened():
-            raise RuntimeError(f"Cannot open video file: {path}")
-
-        fps = capture.get(cv2.CAP_PROP_FPS)
-        frame_count = int(capture.get(cv2.CAP_PROP_FRAME_COUNT))
-        duration = frame_count / fps if fps > 0 else 0.0
-        source.metadata.update({"fps": fps, "frame_count": frame_count, "duration_seconds": duration})
-
-        atoms: list[EvidenceAtom] = []
-        step = max(1, int(fps * self.settings.video_sample_seconds)) if fps > 0 else 1
-        frame_index = 0
-        while frame_index < frame_count:
-            capture.set(cv2.CAP_PROP_POS_FRAMES, frame_index)
-            ok, frame = capture.read()
-            if not ok:
-                break
-            second = frame_index / fps if fps > 0 else 0.0
-            success, encoded = cv2.imencode(".png", frame)
-            if not success:
-                frame_index += step
-                continue
-            image_bytes = encoded.tobytes()
-            image_ref = self.objects.put_bytes(image_bytes, ".png", "video_frames")
-            with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as temp_file:
-                temp_file.write(image_bytes)
-                temp_path = Path(temp_file.name)
-            try:
-                ocr = read_text_from_image(temp_path)
-            finally:
-                temp_path.unlink(missing_ok=True)
-            height, width = frame.shape[:2]
-            atoms.append(
-                EvidenceAtom(
-                    atom_id=f"{source.source_id}:frame:{frame_index:08d}",
-                    modality=Modality.video_frame,
-                    source_id=source.source_id,
-                    source=f"{path}#t={second:.3f}",
-                    text=ocr.text,
-                    image_ref=image_ref,
-                    layout=LayoutBox(x=0, y=0, w=float(width), h=float(height), frame_second=second),
-                    trust=TrustScores(parser=1.0, ocr=ocr.confidence, model=0.0),
-                    token_equivalent=token_equivalent(ocr.text) + visual_token_equivalent(width, height),
-                    metadata={"frame_index": frame_index, "second": second, "ocr_regions": ocr.regions},
-                )
-            )
-            frame_index += step
-        capture.release()
-        return IngestBatch(sources=[source], atoms=atoms)
-
-    def _image_kind(self, path: Path) -> str:
-        name = path.name.lower()
-        ui_markers = ("screenshot", "screen", "figma", "ui", "wireframe", "mockup")
-        return "ui_screenshot" if any(marker in name for marker in ui_markers) else "image"
-
-    def _ingest_repository_context(self, root: Path, files: list[Path]) -> IngestBatch:
-        code_files = [path for path in files if path.suffix.lower() in SUPPORTED_CODE_EXTENSIONS]
-        git_root = find_git_root(root)
-        if not code_files and git_root is None:
-            return IngestBatch(sources=[], atoms=[])
-
-        context = build_repository_context(root, files, code_files, git_root)
-        payload = json.dumps(context, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
-        digest = sha256_bytes(payload)
-        source = SourceRecord(
-            source_id=safe_source_id(root, digest),
-            path=str(root),
-            kind="code_repository",
-            mime="application/vnd.ely-eye.repository+json",
-            sha256=digest,
-            size_bytes=len(payload),
-            metadata={
-                "file_count": context["structure"]["file_count"],
-                "indexed_file_count": context["structure"]["indexed_file_count"],
-                "code_file_count": context["code"]["file_count"],
-                "git_available": context["git"]["available"],
-            },
-        )
-        atom = EvidenceAtom(
-            atom_id=f"{source.source_id}:text",
-            modality=Modality.code,
-            source_id=source.source_id,
-            source=str(root),
-            text=json.dumps(context, ensure_ascii=False, indent=2),
-            trust=TrustScores(parser=1.0, ocr=0.0, model=0.0),
-            token_equivalent=token_equivalent(json.dumps(context, ensure_ascii=False)),
-            metadata=context,
-        )
-        return IngestBatch(sources=[source], atoms=[atom])
-
-
-def image_size(path: Path) -> tuple[int, int]:
-    with Image.open(path) as image:
-        return image.size
-
-
-def image_metadata(path: Path) -> dict[str, Any]:
-    with Image.open(path) as image:
-        return {"width": image.width, "height": image.height, "mode": image.mode, "format": image.format}
-
-
-def build_repository_context(
-    root: Path,
-    files: list[Path],
-    code_files: list[Path],
-    git_root: Path | None,
-) -> dict[str, Any]:
-    structure_entries = [file_structure_entry(root, path) for path in sorted(files)[:REPOSITORY_INDEX_LIMIT]]
-    code_entries = [code_context_entry(root, path) for path in sorted(code_files)[:REPOSITORY_INDEX_LIMIT]]
-    git_context = collect_git_context(git_root) if git_root is not None else {"available": False}
-    return {
-        "format": "ely-eye-repository-context-v1",
-        "root": str(root),
-        "structure": {
-            "file_count": len(files),
-            "indexed_file_count": len(structure_entries),
-            "limit": REPOSITORY_INDEX_LIMIT,
-            "files": structure_entries,
-        },
-        "code": {
-            "file_count": len(code_files),
-            "indexed_file_count": len(code_entries),
-            "files": code_entries,
-            "call_graph": repository_call_graph(code_entries),
-        },
-        "git": git_context,
-        "issue_pr_history": extract_issue_pr_history(git_context),
-    }
-
-
-def file_structure_entry(root: Path, path: Path) -> dict[str, Any]:
-    return {
-        "path": relative_path(root, path),
-        "suffix": path.suffix.lower(),
-        "kind": file_kind(path),
-        "size_bytes": path.stat().st_size,
-    }
-
-
-def code_context_entry(root: Path, path: Path) -> dict[str, Any]:
-    text = path.read_text(encoding="utf-8", errors="replace")
-    symbols = extract_symbols(path, text)
-    return {
-        "path": relative_path(root, path),
-        "language": path.suffix.lower().lstrip("."),
-        "line_count": text.count("\n") + 1,
-        "symbols": symbols,
-    }
-
-
-def repository_call_graph(code_entries: list[dict[str, Any]]) -> list[dict[str, Any]]:
-    edges: list[dict[str, Any]] = []
-    for entry in code_entries:
-        declared = entry.get("symbols")
-        symbols = declared if isinstance(declared, dict) else {}
-        for call in symbols.get("calls", []):
-            if not isinstance(call, dict):
-                continue
-            caller = call.get("caller")
-            callee = call.get("callee")
-            if caller and callee:
-                edges.append({"path": entry["path"], "caller": caller, "callee": callee})
-    return edges
-
-
-def collect_git_context(root: Path) -> dict[str, Any]:
-    commits = [
-        parse_git_log_line(line)
-        for line in run_git(root, ["log", f"--max-count={REPOSITORY_COMMIT_LIMIT}", "--pretty=format:%H%x09%an%x09%aI%x09%s"]).splitlines()
-        if line.strip()
-    ]
-    commits = [commit for commit in commits if commit]
-    return {
-        "available": True,
-        "root": str(root),
-        "head": run_git(root, ["rev-parse", "HEAD"]).strip() or None,
-        "branch": run_git(root, ["branch", "--show-current"]).strip() or None,
-        "remotes": parse_git_remotes(run_git(root, ["remote", "-v"])),
-        "recent_commits": commits,
-    }
-
-
-def parse_git_log_line(line: str) -> dict[str, str] | None:
-    parts = line.split("\t", 3)
-    if len(parts) != 4:
-        return None
-    commit_hash, author, authored_at, subject = parts
-    return {"hash": commit_hash, "author": author, "authored_at": authored_at, "subject": subject}
-
-
-def parse_git_remotes(output: str) -> list[dict[str, str]]:
-    remotes: list[dict[str, str]] = []
-    for line in output.splitlines():
-        parts = line.split()
-        if len(parts) >= 3:
-            remotes.append({"name": parts[0], "url": parts[1], "kind": parts[2].strip("()")})
-    return remotes
-
-
-def extract_issue_pr_history(git_context: dict[str, Any]) -> dict[str, Any]:
-    references: dict[str, set[int]] = {"issues": set(), "pull_requests": set()}
-    for commit in git_context.get("recent_commits", []):
-        if not isinstance(commit, dict):
-            continue
-        subject = str(commit.get("subject") or "")
-        for number in re.findall(r"(?:fixes|closes|resolves)\s+#(\d+)", subject, flags=re.IGNORECASE):
-            references["issues"].add(int(number))
-        for number in re.findall(r"(?:pull request|pr)\s*#?(\d+)", subject, flags=re.IGNORECASE):
-            references["pull_requests"].add(int(number))
-        for number in re.findall(r"\(#(\d+)\)", subject):
-            references["pull_requests"].add(int(number))
-    return {
-        "source": "local_git_commit_history",
-        "issues": sorted(references["issues"]),
-        "pull_requests": sorted(references["pull_requests"]),
-    }
-
-
-def run_git(root: Path, args: list[str]) -> str:
-    try:
-        completed = subprocess.run(
-            ["git", *args],
-            cwd=root,
-            check=False,
-            capture_output=True,
-            text=True,
-            timeout=10,
-        )
-    except (OSError, subprocess.SubprocessError):
-        return ""
-    return completed.stdout if completed.returncode == 0 else ""
-
-
-def find_git_root(path: Path) -> Path | None:
-    current = path if path.is_dir() else path.parent
-    while True:
-        if (current / ".git").exists():
-            return current
-        if current.parent == current:
-            return None
-        current = current.parent
-
-
-def file_kind(path: Path) -> str:
-    suffix = path.suffix.lower()
-    if suffix in SUPPORTED_CODE_EXTENSIONS:
-        return "code"
-    if suffix in SUPPORTED_TEXT_EXTENSIONS:
-        return "text"
-    if suffix in SUPPORTED_IMAGE_EXTENSIONS:
-        return "image"
-    if suffix in SUPPORTED_VIDEO_EXTENSIONS:
-        return "video"
-    if suffix == ".pdf":
-        return "pdf"
-    return "binary"
-
-
-def relative_path(root: Path, path: Path) -> str:
-    try:
-        return path.relative_to(root).as_posix()
-    except ValueError:
-        return path.as_posix()
-
-
-def extract_symbols(path: Path, text: str) -> dict[str, Any]:
-    suffix = path.suffix.lower()
-    if suffix == ".py":
-        return extract_python_symbols(text)
-    symbols = re.findall(
-        r"\b(?:class|function|interface|type|enum|struct|fn|func)\s+([A-Za-z_][A-Za-z0-9_]*)",
-        text,
-    )
-    constants = re.findall(r"\b(?:const|let|var)\s+([A-Za-z_][A-Za-z0-9_]*)\s*=", text)
-    calls = extract_generic_calls(text)
-    return {"symbols": sorted(set(symbols)), "bindings": sorted(set(constants)), "calls": calls}
-
-
-def extract_python_symbols(text: str) -> dict[str, Any]:
-    try:
-        tree = ast.parse(text)
-    except SyntaxError as exc:
-        return {
-            "classes": [],
-            "functions": [],
-            "imports": [],
-            "calls": extract_generic_calls(text),
-            "parse_error": f"{exc.msg} at line {exc.lineno}",
-        }
-    classes: list[str] = []
-    functions: list[str] = []
-    imports: list[str] = []
-    calls: list[dict[str, str]] = []
-    for node in ast.walk(tree):
-        if isinstance(node, ast.ClassDef):
-            classes.append(node.name)
-        elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
-            functions.append(node.name)
-        elif isinstance(node, ast.Import):
-            imports.extend(alias.name for alias in node.names)
-        elif isinstance(node, ast.ImportFrom) and node.module:
-            imports.append(node.module)
-    call_visitor = PythonCallGraphVisitor()
-    call_visitor.visit(tree)
-    calls.extend(call_visitor.calls)
-    return {
-        "classes": sorted(set(classes)),
-        "functions": sorted(set(functions)),
-        "imports": sorted(set(imports)),
-        "calls": calls,
-    }
-
-
-class PythonCallGraphVisitor(ast.NodeVisitor):
-    def __init__(self) -> None:
-        self.current: str | None = None
-        self.calls: list[dict[str, str]] = []
-
-    def visit_FunctionDef(self, node: ast.FunctionDef) -> Any:
-        previous = self.current
-        self.current = node.name
-        self.generic_visit(node)
-        self.current = previous
-
-    def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> Any:
-        previous = self.current
-        self.current = node.name
-        self.generic_visit(node)
-        self.current = previous
-
-    def visit_Call(self, node: ast.Call) -> Any:
-        if self.current:
-            callee = python_call_name(node.func)
-            if callee:
-                self.calls.append({"caller": self.current, "callee": callee})
-        self.generic_visit(node)
-
-
-def python_call_name(node: ast.AST) -> str | None:
-    if isinstance(node, ast.Name):
-        return node.id
-    if isinstance(node, ast.Attribute):
-        parent = python_call_name(node.value)
-        return f"{parent}.{node.attr}" if parent else node.attr
-    return None
-
-
-def extract_generic_calls(text: str) -> list[dict[str, str]]:
-    calls: list[dict[str, str]] = []
-    current = "<module>"
-    function_pattern = re.compile(r"\b(?:function|fn|func)\s+([A-Za-z_][A-Za-z0-9_]*)|([A-Za-z_][A-Za-z0-9_]*)\s*=\s*\([^)]*\)\s*=>")
-    call_pattern = re.compile(r"\b([A-Za-z_][A-Za-z0-9_.]*)\s*\(")
-    keywords = {"if", "for", "while", "switch", "catch", "return", "function"}
-    for line in text.splitlines():
-        function_match = function_pattern.search(line)
-        if function_match:
-            current = next(group for group in function_match.groups() if group)
-        for callee in call_pattern.findall(line):
-            root = callee.split(".", 1)[0]
-            if root not in keywords and callee != current:
-                calls.append({"caller": current, "callee": callee})
-    return calls
-
-
-def write_sources_json(path: Path, sources: list[SourceRecord]) -> None:
-    path.write_text(
-        json.dumps([source.model_dump(mode="json") for source in sources], ensure_ascii=False, indent=2),
-        encoding="utf-8",
-    )

backend/src/ely_eye/ingestion_common.py

@@ -0,0 +1,98 @@
+"""The shared leaf pieces: what a file is, and what a batch of atoms looks like."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from PIL import Image
+
+from .schemas import EvidenceAtom, SourceRecord
+
+SUPPORTED_TEXT_EXTENSIONS = {
+    ".txt",
+    ".md",
+    ".json",
+    ".jsonl",
+    ".yaml",
+    ".yml",
+    ".toml",
+    ".csv",
+    ".html",
+    ".css",
+    ".xml",
+}
+
+SUPPORTED_CODE_EXTENSIONS = {
+    ".py",
+    ".ts",
+    ".tsx",
+    ".js",
+    ".jsx",
+    ".rs",
+    ".go",
+    ".java",
+    ".cpp",
+    ".c",
+    ".h",
+    ".hpp",
+    ".cs",
+    ".swift",
+    ".kt",
+    ".sql",
+}
+
+SUPPORTED_IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".webp", ".bmp", ".tif", ".tiff"}
+SUPPORTED_VIDEO_EXTENSIONS = {".mp4", ".mov", ".mkv", ".avi", ".webm", ".m4v"}
+SUPPORTED_AUDIO_EXTENSIONS = {".mp3", ".wav", ".m4a", ".flac", ".ogg", ".opus", ".aac", ".wma"}
+REPOSITORY_INDEX_LIMIT = 5000
+REPOSITORY_COMMIT_LIMIT = 50
+
+
+class UnsupportedFile(ValueError):
+    """No reader for this file: a walk records the reason, an explicit path fails."""
+
+
+@dataclass
+class IngestBatch:
+    sources: list[SourceRecord]
+    atoms: list[EvidenceAtom]
+
+    @property
+    def token_equivalent(self) -> int:
+        return sum(atom.token_equivalent for atom in self.atoms)
+
+
+def image_size(path: Path) -> tuple[int, int]:
+    with Image.open(path) as image:
+        return image.size
+
+
+def image_metadata(path: Path) -> dict[str, Any]:
+    with Image.open(path) as image:
+        return {"width": image.width, "height": image.height, "mode": image.mode, "format": image.format}
+
+
+def file_kind(path: Path) -> str:
+    suffix = path.suffix.lower()
+    if suffix in SUPPORTED_CODE_EXTENSIONS:
+        return "code"
+    if suffix in SUPPORTED_TEXT_EXTENSIONS:
+        return "text"
+    if suffix in SUPPORTED_IMAGE_EXTENSIONS:
+        return "image"
+    if suffix in SUPPORTED_VIDEO_EXTENSIONS:
+        return "video"
+    if suffix in SUPPORTED_AUDIO_EXTENSIONS:
+        return "audio"
+    if suffix == ".pdf":
+        return "pdf"
+    return "binary"
+
+
+def relative_path(root: Path, path: Path) -> str:
+    try:
+        return path.relative_to(root).as_posix()
+    except ValueError:
+        return path.as_posix()

backend/src/ely_eye/ingestion_media.py

@@ -0,0 +1,278 @@
+"""Pictures, PDFs, video and speech: the readers that need a model or a decoder."""
+
+from __future__ import annotations
+
+import tempfile
+from abc import ABC, abstractmethod
+from pathlib import Path
+
+import cv2
+import pymupdf as fitz
+
+from .asr import DEFERRED as SPEECH_DEFERRED
+from .asr import SpeechSegment, SpeechTranscriber
+from .config import Settings
+from .db import Database
+from .describe import DEFERRED, VisionDescriber
+from .ingestion_common import (
+    IngestBatch,
+    image_metadata,
+    image_size,
+)
+from .ocr import read_text_from_image
+from .schemas import (
+    EvidenceAtom,
+    LayoutBox,
+    Modality,
+    SourceRecord,
+    TrustScores,
+)
+from .storage import (
+    ObjectStore,
+    token_equivalent,
+    visual_token_equivalent,
+)
+
+
+def _page_text(page: fitz.Page) -> str:
+    """Plain text of one page; the type stubs widen get_text's answer for the other modes."""
+    extracted = page.get_text("text")
+    if isinstance(extracted, str):
+        return extracted.strip()
+    return "\n".join(str(part) for part in extracted).strip()
+
+
+class MediaIngest(ABC):
+    """Readers for pictures, PDFs, video and speech.
+
+    A mixin: the host supplies identity and storage, this supplies the senses. The requirements
+    are abstract so a half-wired host fails at construction instead of returning None.
+    """
+
+    settings: Settings
+    db: Database
+    objects: ObjectStore
+    describer: VisionDescriber
+    transcriber: SpeechTranscriber
+
+    @abstractmethod
+    def _source_record(
+        self,
+        path: Path,
+        kind: str,
+        metadata: dict[str, object] | None = None,
+    ) -> SourceRecord:
+        """Where a file's source row comes from; the host decides identity."""
+
+    def _ingest_image(self, path: Path) -> IngestBatch:
+        source = self._source_record(path, self._image_kind(path), metadata=image_metadata(path))
+        image_ref = self.objects.put_file(path, "images")
+        ocr = read_text_from_image(path)
+        caption = ""
+        caption_note = ""
+        if not ocr.text.strip():
+            # Words already on the picture are free; only a wordless one is worth a model call.
+            try:
+                caption = self.describer.describe_or_none(path) or ""
+            except (OSError, RuntimeError, ValueError) as error:
+                caption, caption_note = "", f"failed: {error}"
+            else:
+                caption_note = "vlm" if caption else DEFERRED
+        text = caption or ocr.text
+        width, height = image_size(path)
+        atom = EvidenceAtom(
+            atom_id=f"{source.source_id}:image",
+            modality=Modality.ui_screenshot if source.kind == "ui_screenshot" else Modality.image,
+            source_id=source.source_id,
+            source=str(path),
+            text=text,
+            image_ref=image_ref,
+            layout=LayoutBox(x=0, y=0, w=width, h=height),
+            trust=TrustScores(parser=1.0, ocr=ocr.confidence, model=0.6 if caption else 0.0),
+            token_equivalent=token_equivalent(text) + visual_token_equivalent(width, height),
+            metadata={
+                "ocr_regions": ocr.regions,
+                "caption": caption_note,
+                **image_metadata(path),
+            },
+        )
+        return IngestBatch(sources=[source], atoms=[atom])
+
+    def _ingest_pdf(self, path: Path) -> IngestBatch:
+        source = self._source_record(path, "pdf")
+        atoms: list[EvidenceAtom] = []
+        with fitz.open(path) as document:
+            source.metadata["page_count"] = document.page_count
+            for page_index in range(1, document.page_count + 1):
+                page = document[page_index - 1]
+                rect = page.rect
+                text = _page_text(page)
+                pixmap = page.get_pixmap(matrix=fitz.Matrix(1.5, 1.5), alpha=False)
+                image_bytes = pixmap.tobytes("png")
+                image_ref = self.objects.put_bytes(image_bytes, ".png", "pdf_pages")
+                if not text:
+                    with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as temp_file:
+                        temp_file.write(image_bytes)
+                        temp_path = Path(temp_file.name)
+                    try:
+                        ocr = read_text_from_image(temp_path)
+                        text = ocr.text
+                        ocr_confidence = ocr.confidence
+                    finally:
+                        temp_path.unlink(missing_ok=True)
+                else:
+                    ocr_confidence = 0.0
+                atoms.append(
+                    EvidenceAtom(
+                        atom_id=f"{source.source_id}:page:{page_index:05d}",
+                        modality=Modality.pdf_page,
+                        source_id=source.source_id,
+                        source=f"{path}#page={page_index}",
+                        text=text,
+                        image_ref=image_ref,
+                        layout=LayoutBox(x=0, y=0, w=float(rect.width), h=float(rect.height), page=page_index),
+                        trust=TrustScores(parser=1.0, ocr=ocr_confidence, model=0.0),
+                        token_equivalent=token_equivalent(text) + max(1, int(rect.width * rect.height) // 50_000),
+                        metadata={"page": page_index, "width": rect.width, "height": rect.height},
+                    )
+                )
+                atoms.extend(self._pdf_region_atoms(path, source.source_id, page_index, page))
+        return IngestBatch(sources=[source], atoms=atoms)
+
+    def _pdf_region_atoms(
+        self, path: Path, source_id: str, page_index: int, page: fitz.Page
+    ) -> list[EvidenceAtom]:
+        regions: list[EvidenceAtom] = []
+        blocks = page.get_text("blocks")
+        for block_index, block in enumerate(blocks):
+            if len(block) < 5:
+                continue
+            x0, y0, x1, y1, text = block[:5]
+            text = str(text).strip()
+            if not text:
+                continue
+            regions.append(
+                EvidenceAtom(
+                    atom_id=f"{source_id}:page:{page_index:05d}:region:{block_index:04d}",
+                    modality=Modality.pdf_region,
+                    source_id=source_id,
+                    source=f"{path}#page={page_index}&region={block_index}",
+                    text=text,
+                    layout=LayoutBox(
+                        x=float(x0),
+                        y=float(y0),
+                        w=float(x1) - float(x0),
+                        h=float(y1) - float(y0),
+                        page=page_index,
+                    ),
+                    trust=TrustScores(parser=1.0, ocr=0.0, model=0.0),
+                    token_equivalent=token_equivalent(text),
+                    metadata={"page": page_index, "block": block_index},
+                )
+            )
+        return regions
+
+    def _ingest_video(self, path: Path) -> IngestBatch:
+        source = self._source_record(path, "video")
+        capture = cv2.VideoCapture(str(path))
+        if not capture.isOpened():
+            raise RuntimeError(f"Cannot open video file: {path}")
+
+        fps = capture.get(cv2.CAP_PROP_FPS)
+        frame_count = int(capture.get(cv2.CAP_PROP_FRAME_COUNT))
+        duration = frame_count / fps if fps > 0 else 0.0
+        source.metadata.update({"fps": fps, "frame_count": frame_count, "duration_seconds": duration})
+
+        atoms: list[EvidenceAtom] = []
+        step = max(1, int(fps * self.settings.video_sample_seconds)) if fps > 0 else 1
+        frame_index = 0
+        while frame_index < frame_count:
+            capture.set(cv2.CAP_PROP_POS_FRAMES, frame_index)
+            ok, frame = capture.read()
+            if not ok:
+                break
+            second = frame_index / fps if fps > 0 else 0.0
+            success, encoded = cv2.imencode(".png", frame)
+            if not success:
+                frame_index += step
+                continue
+            image_bytes = encoded.tobytes()
+            image_ref = self.objects.put_bytes(image_bytes, ".png", "video_frames")
+            with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as temp_file:
+                temp_file.write(image_bytes)
+                temp_path = Path(temp_file.name)
+            try:
+                ocr = read_text_from_image(temp_path)
+            finally:
+                temp_path.unlink(missing_ok=True)
+            height, width = frame.shape[:2]
+            atoms.append(
+                EvidenceAtom(
+                    atom_id=f"{source.source_id}:frame:{frame_index:08d}",
+                    modality=Modality.video_frame,
+                    source_id=source.source_id,
+                    source=f"{path}#t={second:.3f}",
+                    text=ocr.text,
+                    image_ref=image_ref,
+                    layout=LayoutBox(x=0, y=0, w=float(width), h=float(height), frame_second=second),
+                    trust=TrustScores(parser=1.0, ocr=ocr.confidence, model=0.0),
+                    token_equivalent=token_equivalent(ocr.text) + visual_token_equivalent(width, height),
+                    metadata={"frame_index": frame_index, "second": second, "ocr_regions": ocr.regions},
+                )
+            )
+            frame_index += step
+        capture.release()
+        segments, note = self._speech(path)
+        speech_atoms = self._speech_atoms(source, path, segments)
+        source.metadata["speech"] = note or f"{len(speech_atoms)} segments of speech"
+        atoms.extend(speech_atoms)
+        return IngestBatch(sources=[source], atoms=atoms)
+
+    def _ingest_audio(self, path: Path) -> IngestBatch:
+        source = self._source_record(path, "audio")
+        segments, note = self._speech(path)
+        atoms = self._speech_atoms(source, path, segments)
+        source.metadata["speech"] = note or f"{len(atoms)} segments of speech"
+        return IngestBatch(sources=[source], atoms=atoms)
+
+    def _speech(self, path: Path) -> tuple[list[SpeechSegment], str | None]:
+        """Speech is optional here: no snapshot defers, a broken model becomes a note, not a dead ingest."""
+        if not self.transcriber.available():
+            return [], SPEECH_DEFERRED
+        try:
+            return self.transcriber.transcribe(path), None
+        except (OSError, RuntimeError, ValueError) as error:
+            return [], f"failed: {error}"
+
+    def _speech_atoms(self, source: SourceRecord, path: Path, segments: list[SpeechSegment]) -> list[EvidenceAtom]:
+        atoms: list[EvidenceAtom] = []
+        for segment in segments:
+            if not segment.text:
+                continue
+            atoms.append(
+                EvidenceAtom(
+                    atom_id=f"{source.source_id}:speech:{int(segment.start * 1000):010d}",
+                    modality=Modality.audio,
+                    source_id=source.source_id,
+                    source=f"{path}#t={segment.start:.3f}",
+                    text=segment.text,
+                    layout=LayoutBox(x=0.0, y=0.0, w=0.0, h=0.0, frame_second=segment.start),
+                    trust=TrustScores(parser=0.0, ocr=0.0, model=segment.confidence),
+                    token_equivalent=token_equivalent(segment.text),
+                    metadata={
+                        "start_seconds": segment.start,
+                        "end_seconds": segment.end,
+                        "words": [
+                            {"word": word.word, "start": word.start, "end": word.end, "confidence": word.confidence}
+                            for word in segment.words
+                        ],
+                    },
+                )
+            )
+        return atoms
+
+    def _image_kind(self, path: Path) -> str:
+        name = path.name.lower()
+        ui_markers = ("screenshot", "screen", "figma", "ui", "wireframe", "mockup")
+        return "ui_screenshot" if any(marker in name for marker in ui_markers) else "image"
+

backend/src/ely_eye/ingestion_repository.py

@@ -0,0 +1,321 @@
+"""Repository context: file structure, call graph, git history, written out as atoms."""
+
+from __future__ import annotations
+
+import ast
+import json
+import re
+import subprocess
+from pathlib import Path
+from typing import Any
+
+from .config import Settings
+from .db import Database
+from .ingestion_common import (
+    REPOSITORY_COMMIT_LIMIT,
+    REPOSITORY_INDEX_LIMIT,
+    SUPPORTED_CODE_EXTENSIONS,
+    IngestBatch,
+    file_kind,
+    relative_path,
+)
+from .schemas import (
+    EvidenceAtom,
+    Modality,
+    SourceRecord,
+    TrustScores,
+)
+from .storage import (
+    ObjectStore,
+    safe_source_id,
+    sha256_bytes,
+    token_equivalent,
+)
+
+
+class RepositoryIngest:
+    settings: Settings
+    db: Database
+    objects: ObjectStore
+
+    def _ingest_repository_context(self, root: Path, files: list[Path]) -> IngestBatch:
+        code_files = [path for path in files if path.suffix.lower() in SUPPORTED_CODE_EXTENSIONS]
+        git_root = find_git_root(root)
+        if not code_files and git_root is None:
+            return IngestBatch(sources=[], atoms=[])
+
+        context = build_repository_context(root, files, code_files, git_root)
+        payload = json.dumps(context, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
+        digest = sha256_bytes(payload)
+        source = SourceRecord(
+            source_id=safe_source_id(root, digest),
+            path=str(root),
+            kind="code_repository",
+            mime="application/vnd.ely-eye.repository+json",
+            sha256=digest,
+            size_bytes=len(payload),
+            metadata={
+                "file_count": context["structure"]["file_count"],
+                "indexed_file_count": context["structure"]["indexed_file_count"],
+                "code_file_count": context["code"]["file_count"],
+                "git_available": context["git"]["available"],
+            },
+        )
+        atom = EvidenceAtom(
+            atom_id=f"{source.source_id}:text",
+            modality=Modality.code,
+            source_id=source.source_id,
+            source=str(root),
+            text=json.dumps(context, ensure_ascii=False, indent=2),
+            trust=TrustScores(parser=1.0, ocr=0.0, model=0.0),
+            token_equivalent=token_equivalent(json.dumps(context, ensure_ascii=False)),
+            metadata=context,
+        )
+        return IngestBatch(sources=[source], atoms=[atom])
+
+def build_repository_context(
+    root: Path,
+    files: list[Path],
+    code_files: list[Path],
+    git_root: Path | None,
+) -> dict[str, Any]:
+    structure_entries = [file_structure_entry(root, path) for path in sorted(files)[:REPOSITORY_INDEX_LIMIT]]
+    code_entries = [code_context_entry(root, path) for path in sorted(code_files)[:REPOSITORY_INDEX_LIMIT]]
+    git_context = collect_git_context(git_root) if git_root is not None else {"available": False}
+    return {
+        "format": "ely-eye-repository-context-v1",
+        "root": str(root),
+        "structure": {
+            "file_count": len(files),
+            "indexed_file_count": len(structure_entries),
+            "limit": REPOSITORY_INDEX_LIMIT,
+            "files": structure_entries,
+        },
+        "code": {
+            "file_count": len(code_files),
+            "indexed_file_count": len(code_entries),
+            "files": code_entries,
+            "call_graph": repository_call_graph(code_entries),
+        },
+        "git": git_context,
+        "issue_pr_history": extract_issue_pr_history(git_context),
+    }
+
+
+def file_structure_entry(root: Path, path: Path) -> dict[str, Any]:
+    return {
+        "path": relative_path(root, path),
+        "suffix": path.suffix.lower(),
+        "kind": file_kind(path),
+        "size_bytes": path.stat().st_size,
+    }
+
+
+def code_context_entry(root: Path, path: Path) -> dict[str, Any]:
+    text = path.read_text(encoding="utf-8", errors="replace")
+    symbols = extract_symbols(path, text)
+    return {
+        "path": relative_path(root, path),
+        "language": path.suffix.lower().lstrip("."),
+        "line_count": text.count("\n") + 1,
+        "symbols": symbols,
+    }
+
+
+def repository_call_graph(code_entries: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    edges: list[dict[str, Any]] = []
+    for entry in code_entries:
+        declared = entry.get("symbols")
+        symbols = declared if isinstance(declared, dict) else {}
+        for call in symbols.get("calls", []):
+            if not isinstance(call, dict):
+                continue
+            caller = call.get("caller")
+            callee = call.get("callee")
+            if caller and callee:
+                edges.append({"path": entry["path"], "caller": caller, "callee": callee})
+    return edges
+
+
+def collect_git_context(root: Path) -> dict[str, Any]:
+    commits = [
+        parse_git_log_line(line)
+        for line in run_git(root, ["log", f"--max-count={REPOSITORY_COMMIT_LIMIT}", "--pretty=format:%H%x09%an%x09%aI%x09%s"]).splitlines()
+        if line.strip()
+    ]
+    commits = [commit for commit in commits if commit]
+    return {
+        "available": True,
+        "root": str(root),
+        "head": run_git(root, ["rev-parse", "HEAD"]).strip() or None,
+        "branch": run_git(root, ["branch", "--show-current"]).strip() or None,
+        "remotes": parse_git_remotes(run_git(root, ["remote", "-v"])),
+        "recent_commits": commits,
+    }
+
+
+def parse_git_log_line(line: str) -> dict[str, str] | None:
+    parts = line.split("\t", 3)
+    if len(parts) != 4:
+        return None
+    commit_hash, author, authored_at, subject = parts
+    return {"hash": commit_hash, "author": author, "authored_at": authored_at, "subject": subject}
+
+
+def parse_git_remotes(output: str) -> list[dict[str, str]]:
+    remotes: list[dict[str, str]] = []
+    for line in output.splitlines():
+        parts = line.split()
+        if len(parts) >= 3:
+            remotes.append({"name": parts[0], "url": parts[1], "kind": parts[2].strip("()")})
+    return remotes
+
+
+def extract_issue_pr_history(git_context: dict[str, Any]) -> dict[str, Any]:
+    references: dict[str, set[int]] = {"issues": set(), "pull_requests": set()}
+    for commit in git_context.get("recent_commits", []):
+        if not isinstance(commit, dict):
+            continue
+        subject = str(commit.get("subject") or "")
+        for number in re.findall(r"(?:fixes|closes|resolves)\s+#(\d+)", subject, flags=re.IGNORECASE):
+            references["issues"].add(int(number))
+        for number in re.findall(r"(?:pull request|pr)\s*#?(\d+)", subject, flags=re.IGNORECASE):
+            references["pull_requests"].add(int(number))
+        for number in re.findall(r"\(#(\d+)\)", subject):
+            references["pull_requests"].add(int(number))
+    return {
+        "source": "local_git_commit_history",
+        "issues": sorted(references["issues"]),
+        "pull_requests": sorted(references["pull_requests"]),
+    }
+
+
+def run_git(root: Path, args: list[str]) -> str:
+    try:
+        completed = subprocess.run(
+            ["git", *args],
+            cwd=root,
+            check=False,
+            capture_output=True,
+            text=True,
+            timeout=10,
+        )
+    except (OSError, subprocess.SubprocessError):
+        return ""
+    return completed.stdout if completed.returncode == 0 else ""
+
+
+def find_git_root(path: Path) -> Path | None:
+    current = path if path.is_dir() else path.parent
+    while True:
+        if (current / ".git").exists():
+            return current
+        if current.parent == current:
+            return None
+        current = current.parent
+
+
+def extract_symbols(path: Path, text: str) -> dict[str, Any]:
+    suffix = path.suffix.lower()
+    if suffix == ".py":
+        return extract_python_symbols(text)
+    symbols = re.findall(
+        r"\b(?:class|function|interface|type|enum|struct|fn|func)\s+([A-Za-z_][A-Za-z0-9_]*)",
+        text,
+    )
+    constants = re.findall(r"\b(?:const|let|var)\s+([A-Za-z_][A-Za-z0-9_]*)\s*=", text)
+    calls = extract_generic_calls(text)
+    return {"symbols": sorted(set(symbols)), "bindings": sorted(set(constants)), "calls": calls}
+
+
+def extract_python_symbols(text: str) -> dict[str, Any]:
+    try:
+        tree = ast.parse(text)
+    except SyntaxError as exc:
+        return {
+            "classes": [],
+            "functions": [],
+            "imports": [],
+            "calls": extract_generic_calls(text),
+            "parse_error": f"{exc.msg} at line {exc.lineno}",
+        }
+    classes: list[str] = []
+    functions: list[str] = []
+    imports: list[str] = []
+    calls: list[dict[str, str]] = []
+    for node in ast.walk(tree):
+        if isinstance(node, ast.ClassDef):
+            classes.append(node.name)
+        elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
+            functions.append(node.name)
+        elif isinstance(node, ast.Import):
+            imports.extend(alias.name for alias in node.names)
+        elif isinstance(node, ast.ImportFrom) and node.module:
+            imports.append(node.module)
+    call_visitor = PythonCallGraphVisitor()
+    call_visitor.visit(tree)
+    calls.extend(call_visitor.calls)
+    return {
+        "classes": sorted(set(classes)),
+        "functions": sorted(set(functions)),
+        "imports": sorted(set(imports)),
+        "calls": calls,
+    }
+
+
+class PythonCallGraphVisitor(ast.NodeVisitor):
+    def __init__(self) -> None:
+        self.current: str | None = None
+        self.calls: list[dict[str, str]] = []
+
+    def visit_FunctionDef(self, node: ast.FunctionDef) -> Any:
+        previous = self.current
+        self.current = node.name
+        self.generic_visit(node)
+        self.current = previous
+
+    def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> Any:
+        previous = self.current
+        self.current = node.name
+        self.generic_visit(node)
+        self.current = previous
+
+    def visit_Call(self, node: ast.Call) -> Any:
+        if self.current:
+            callee = python_call_name(node.func)
+            if callee:
+                self.calls.append({"caller": self.current, "callee": callee})
+        self.generic_visit(node)
+
+
+def python_call_name(node: ast.AST) -> str | None:
+    if isinstance(node, ast.Name):
+        return node.id
+    if isinstance(node, ast.Attribute):
+        parent = python_call_name(node.value)
+        return f"{parent}.{node.attr}" if parent else node.attr
+    return None
+
+
+def extract_generic_calls(text: str) -> list[dict[str, str]]:
+    calls: list[dict[str, str]] = []
+    current = "<module>"
+    function_pattern = re.compile(r"\b(?:function|fn|func)\s+([A-Za-z_][A-Za-z0-9_]*)|([A-Za-z_][A-Za-z0-9_]*)\s*=\s*\([^)]*\)\s*=>")
+    call_pattern = re.compile(r"\b([A-Za-z_][A-Za-z0-9_.]*)\s*\(")
+    keywords = {"if", "for", "while", "switch", "catch", "return", "function"}
+    for line in text.splitlines():
+        function_match = function_pattern.search(line)
+        if function_match:
+            current = next(group for group in function_match.groups() if group)
+        for callee in call_pattern.findall(line):
+            root = callee.split(".", 1)[0]
+            if root not in keywords and callee != current:
+                calls.append({"caller": current, "callee": callee})
+    return calls
+
+
+def write_sources_json(path: Path, sources: list[SourceRecord]) -> None:
+    path.write_text(
+        json.dumps([source.model_dump(mode="json") for source in sources], ensure_ascii=False, indent=2),
+        encoding="utf-8",
+    )

backend/src/ely_eye/proofs/checks_cartridge.py

@@ -21,7 +21,8 @@ from ..db import Database
 from ..evaluation_schemas import (
     ProofCheck,
 )
-from ..ingestion import SUPPORTED_CODE_EXTENSIONS, build_repository_context, find_git_root
+from ..ingestion_common import SUPPORTED_CODE_EXTENSIONS
+from ..ingestion_repository import build_repository_context, find_git_root
 from ..schemas import (
     CartridgeManifest,
     RuntimeProfile,

backend/src/ely_eye/schemas.py

@@ -34,6 +34,7 @@ class Modality(StrEnum):
     pdf_page = "pdf_page"
     pdf_region = "pdf_region"
     video_frame = "video_frame"
+    audio = "audio"
     ui_screenshot = "ui_screenshot"
     mixed = "mixed"
 
@@ -138,12 +139,17 @@ class SourceRecord(BaseModel):
     created_at: datetime = Field(default_factory=utc_now)
     metadata: dict[str, Any] = Field(default_factory=dict)
 
+class SkippedFile(BaseModel):
+    path: str
+    reason: str
+
 class IngestResponse(BaseModel):
     source_count: int
     atom_count: int
     cartridge_id: str | None
     token_equivalent: int
     sources: list[SourceRecord]
+    skipped: list[SkippedFile] = []
 
 class RetrievalHit(BaseModel):
     atom: EvidenceAtom

backend/tests/test_asr.py

@@ -0,0 +1,210 @@
+"""Speech memories: what a recording becomes, and what happens when no model is configured."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, cast
+
+import numpy as np
+import pytest
+
+from ely_eye import ingestion_media as media_module
+from ely_eye.asr import DEFERRED, SpeechSegment, SpeechTranscriber, SpokenWord, decode_audio
+from ely_eye.config import Settings
+from ely_eye.db import Database
+from ely_eye.ingestion import IngestionService, UnsupportedFile
+from ely_eye.ocr import OcrResult
+from ely_eye.schemas import Modality
+
+WORDS = (
+    SpokenWord(word="钥匙", start=0.58, end=0.9, confidence=0.91),
+    SpokenWord(word="抽屉", start=1.02, end=1.4, confidence=0.77),
+)
+SEGMENTS = (
+    SpeechSegment(text="钥匙在抽屉里", start=0.58, end=1.4, words=WORDS, confidence=0.84),
+    SpeechSegment(text="", start=9.0, end=9.5, words=(), confidence=0.0),
+)
+
+
+class FakeTranscriber:
+    """Stands in for whisper so the test asserts the shape we build from its answer."""
+
+    def __init__(
+        self,
+        segments: tuple[SpeechSegment, ...] = SEGMENTS,
+        available: bool = True,
+        error: Exception | None = None,
+    ) -> None:
+        self.segments = segments
+        self._available = available
+        self.error = error
+        self.calls: list[Path] = []
+
+    def available(self) -> bool:
+        return self._available
+
+    def transcribe(self, path: Path) -> list[SpeechSegment]:
+        self.calls.append(path)
+        if self.error is not None:
+            raise self.error
+        return list(self.segments)
+
+
+class FakeEncoded:
+    def tobytes(self) -> bytes:
+        return b"png-bytes"
+
+
+class FakeCapture:
+    """One frame, two frames of metadata: enough for the video reader to reach the speech step."""
+
+    def __init__(self, _path: str) -> None:
+        self.released = False
+
+    def isOpened(self) -> bool:
+        return True
+
+    def get(self, property_id: int) -> float:
+        return 1.0 if property_id == media_module.cv2.CAP_PROP_FPS else 2.0
+
+    def set(self, _property_id: int, _value: float) -> None:
+        return None
+
+    def read(self) -> tuple[bool, Any]:
+        return True, np.zeros((4, 4, 3), dtype=np.uint8)
+
+    def release(self) -> None:
+        self.released = True
+
+
+class FakeCv2:
+    CAP_PROP_FPS = 5
+    CAP_PROP_FRAME_COUNT = 7
+    CAP_PROP_POS_FRAMES = 1
+
+    VideoCapture = FakeCapture
+
+    @staticmethod
+    def imencode(_suffix: str, _frame: Any) -> tuple[bool, FakeEncoded]:
+        return True, FakeEncoded()
+
+
+def _settings(tmp_path: Path, asr_model_dir: Path | None = None) -> Settings:
+    return Settings(home=tmp_path / "home", asr_model_dir=asr_model_dir)
+
+
+def _service(tmp_path: Path, transcriber: Any) -> tuple[IngestionService, Database]:
+    settings = _settings(tmp_path)
+    settings.ensure_dirs()
+    database = Database(settings)
+    service = IngestionService(settings=settings, database=database)
+    service.transcriber = cast(SpeechTranscriber, transcriber)
+    return service, database
+
+
+def _audio(tmp_path: Path, name: str = "keynote.mp3") -> Path:
+    path = tmp_path / name
+    path.write_bytes(b"not really mp3")
+    return path
+
+
+def test_a_recording_becomes_memories_pointed_at_the_instant_each_word_was_said(
+    tmp_path: Path,
+) -> None:
+    service, database = _service(tmp_path, FakeTranscriber())
+
+    response = service.ingest_path(_audio(tmp_path))
+
+    assert response.atom_count == 1, "the empty segment must not become a memory"
+    atom = database.list_atoms()[0]
+    assert atom.modality is Modality.audio
+    assert atom.text == "钥匙在抽屉里"
+    assert atom.source.endswith("#t=0.580")
+    assert atom.atom_id.endswith(":speech:0000000580")
+    assert atom.layout is not None and atom.layout.frame_second == pytest.approx(0.58)
+    assert atom.metadata["start_seconds"] == pytest.approx(0.58)
+    assert atom.metadata["end_seconds"] == pytest.approx(1.4)
+    assert [(word["word"], word["start"]) for word in atom.metadata["words"]] == [("钥匙", 0.58), ("抽屉", 1.02)]
+    assert atom.trust.model == pytest.approx(0.84), "the model's own confidence is the trust it earned"
+
+
+def test_a_recording_without_a_model_is_recorded_and_says_why(tmp_path: Path) -> None:
+    service, database = _service(tmp_path, FakeTranscriber(available=False))
+
+    response = service.ingest_path(_audio(tmp_path))
+
+    assert response.source_count == 1, "the file is known even when nothing was heard"
+    assert response.atom_count == 0
+    assert database.list_sources()[0].metadata["speech"] == DEFERRED
+    assert response.skipped == [], "a file we have a reader for is not a skipped file"
+
+
+def test_a_broken_model_leaves_a_note_instead_of_a_corpse(tmp_path: Path) -> None:
+    service, _database = _service(tmp_path, FakeTranscriber(error=ValueError("ctranslate2 is gone")))
+
+    response = service.ingest_path(_audio(tmp_path))
+
+    assert response.source_count == 1
+    assert response.atom_count == 0
+    assert response.sources[0].metadata["speech"] == "failed: ctranslate2 is gone"
+
+
+def test_a_video_learns_what_was_said_in_it(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    service, database = _service(tmp_path, FakeTranscriber())
+    monkeypatch.setattr(media_module, "cv2", FakeCv2())
+    monkeypatch.setattr(media_module, "read_text_from_image", lambda path: OcrResult("", 0.0, []))
+    video = tmp_path / "talk.mp4"
+    video.write_bytes(b"not really mp4")
+
+    service.ingest_path(video)
+
+    modalities = {atom.modality for atom in database.list_atoms()}
+    assert Modality.video_frame in modalities
+    assert Modality.audio in modalities, "a silent OCR of frames misses everything that was spoken"
+    assert database.list_sources()[0].metadata["speech"] == "1 segments of speech"
+
+
+def test_a_file_we_cannot_read_fails_when_it_was_named_explicitly(tmp_path: Path) -> None:
+    service, _database = _service(tmp_path, FakeTranscriber(available=False))
+    mystery = tmp_path / "mystery.xyz"
+    mystery.write_text("nope", encoding="utf-8")
+
+    with pytest.raises(UnsupportedFile, match="no reader for .xyz"):
+        service.ingest_path(mystery)
+
+
+def test_a_walk_records_what_it_refused_to_read(tmp_path: Path) -> None:
+    service, database = _service(tmp_path, FakeTranscriber(available=False))
+    folder = tmp_path / "shelf"
+    folder.mkdir()
+    (folder / "note.md").write_text("# 钥匙放在抽屉里\n", encoding="utf-8")
+    (folder / "mystery.bin").write_bytes(b"\x00\x01")
+
+    response = service.ingest_path(folder)
+
+    assert response.atom_count > 0, "the readable file is still ingested"
+    assert [item.path for item in response.skipped] == [str(folder / "mystery.bin")]
+    assert "no reader for .bin" in response.skipped[0].reason
+    assert database.atom_count() == response.atom_count
+
+
+def test_the_transcriber_names_the_snapshot_it_wants(tmp_path: Path) -> None:
+    without = SpeechTranscriber(_settings(tmp_path))
+    assert not without.available()
+    with pytest.raises(SystemExit, match="ELY_EYE_ASR_MODEL_DIR"):
+        without.transcribe(_audio(tmp_path))
+
+    empty = tmp_path / "snapshot"
+    empty.mkdir()
+    half = SpeechTranscriber(_settings(tmp_path, asr_model_dir=empty))
+    assert not half.available()
+    with pytest.raises(SystemExit, match="holds no model.bin"):
+        half.transcribe(_audio(tmp_path))
+
+
+def test_a_file_with_no_audio_track_says_so(tmp_path: Path) -> None:
+    markdown = tmp_path / "notes.md"
+    markdown.write_text("# 钥匙\n", encoding="utf-8")
+
+    with pytest.raises(ValueError, match="the audio cannot be decoded"):
+        decode_audio(markdown)

backend/tests/test_describe.py

@@ -13,7 +13,7 @@ from typing import cast
 import pytest
 from PIL import Image
 
-import ely_eye.ingestion as ingestion_module
+import ely_eye.ingestion_media as media_module
 from ely_eye.config import Settings
 from ely_eye.db import Database
 from ely_eye.describe import DEFERRED, VisionDescriber
@@ -49,7 +49,7 @@ def _ingest(tmp_path: Path, describer: object, ocr_text: str = "") -> tuple[Sett
     database = Database(settings)
     service = IngestionService(settings=settings, database=database)
     service.describer = cast(VisionDescriber, describer)
-    ingestion_module.read_text_from_image = lambda path: OcrResult(ocr_text, 0.9, [])
+    media_module.read_text_from_image = lambda path: OcrResult(ocr_text, 0.9, [])
     service.ingest_path(_picture(tmp_path))
     return settings, database
 

backend/tests/test_visual_tokens.py

@@ -13,7 +13,7 @@ from pathlib import Path
 import pytest
 from PIL import Image
 
-import ely_eye.ingestion as ingestion_module
+import ely_eye.ingestion_media as media_module
 from ely_eye.config import Settings
 from ely_eye.db import Database
 from ely_eye.ingestion import IngestionService
@@ -48,7 +48,7 @@ def test_a_region_without_a_size_is_refused_rather_than_guessed() -> None:
 
 def test_an_ingested_image_carries_its_text_and_its_patches(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
     settings = Settings(home=tmp_path)
-    monkeypatch.setattr(ingestion_module, "read_text_from_image", lambda path: OcrResult("购物车", 0.9, []))
+    monkeypatch.setattr(media_module, "read_text_from_image", lambda path: OcrResult("购物车", 0.9, []))
 
     database = _ingest(tmp_path, settings)
     atom = database.list_atoms()[0]
@@ -58,7 +58,7 @@ def test_an_ingested_image_carries_its_text_and_its_patches(tmp_path: Path, monk
 
 def test_an_existing_store_is_repriced_by_the_migration(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
     settings = Settings(home=tmp_path)
-    monkeypatch.setattr(ingestion_module, "read_text_from_image", lambda path: OcrResult("", 0.0, []))
+    monkeypatch.setattr(media_module, "read_text_from_image", lambda path: OcrResult("", 0.0, []))
     database = _ingest(tmp_path, settings)
     note = tmp_path / "notes.md"
     note.write_text("# 钥匙放在玄关抽屉里\n", encoding="utf-8")

bench/asr_check.py

@@ -0,0 +1,203 @@
+"""Measure what the speech reader hears, against subtitles nobody in this repo wrote.
+
+Video-MME ships community subtitles for its videos. They are third-party text, so they are a
+fair reference for word error rate, with one caveat the report states: a caption track is not
+a verbatim transcript, so the number is an upper bound on the model's real error rate.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+import subprocess
+import time
+from itertools import pairwise
+from pathlib import Path
+from typing import Any
+
+from ely_eye.asr import SpeechSegment, SpeechTranscriber, SpokenWord
+from ely_eye.config import Settings
+from ely_eye.schemas import Modality
+
+from bench.systems import machine_facts
+
+ROOT = Path(__file__).resolve().parent
+
+CORPUS = ROOT / "corpus" / "lmms-eval" / "Video-MME"
+MODEL = ROOT / "models" / "faster-whisper-tiny"
+REPORTS = ROOT / "reports"
+PUNCTUATION = re.compile(r"[^\w' ]+")
+TAGS = re.compile(r"<[^>]+>")
+INDEX_LINE = re.compile(r"^\d+$")
+TIMING_LINE = re.compile(r"-->")
+
+
+def normalise(text: str) -> list[str]:
+    cleaned = PUNCTUATION.sub(" ", text.lower())
+    return [word for word in cleaned.split() if word]
+
+
+def edit_distance(left: list[str], right: list[str]) -> int:
+    previous = list(range(len(right) + 1))
+    for index, first in enumerate(left, start=1):
+        current = [index]
+        for column, second in enumerate(right, start=1):
+            cost = 0 if first == second else 1
+            current.append(min(previous[column] + 1, current[column - 1] + 1, previous[column - 1] + cost))
+        previous = current
+    return previous[-1]
+
+
+def word_error_rate(hypothesis: list[str], reference: list[str]) -> float:
+    if not reference:
+        return 1.0
+    return edit_distance(hypothesis, reference) / len(reference)
+
+
+def srt_text(path: Path) -> str:
+    lines = []
+    for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
+        stripped = line.strip()
+        if not stripped or INDEX_LINE.match(stripped) or TIMING_LINE.search(stripped):
+            continue
+        lines.append(TAGS.sub(" ", stripped))
+    return " ".join(lines)
+
+
+def model_sha256() -> str:
+    digest = hashlib.sha256()
+    with (MODEL / "model.bin").open("rb") as handle:
+        for block in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(block)
+    return digest.hexdigest()
+
+
+def git_sha() -> str:
+    done = subprocess.run(
+        ["git", "rev-parse", "HEAD"], cwd=ROOT.parent, capture_output=True, text=True, check=True
+    )
+    return done.stdout.strip()
+
+
+def timed_words(segments: list[SpeechSegment]) -> list[SpokenWord]:
+    return [word for segment in segments for word in segment.words if word.word]
+
+
+def check_timings(segments: list[SpeechSegment]) -> dict[str, Any]:
+    """Word clocks must run forward and stay inside their segment."""
+    words = timed_words(segments)
+    out_of_order = sum(1 for before, after in pairwise(words) if after.start < before.start)
+    outside = sum(1 for segment in segments for word in segment.words if word.start < segment.start - 0.01)
+    return {"words": len(words), "out_of_order": out_of_order, "before_their_segment": outside}
+
+
+def main() -> None:
+    if not (MODEL / "model.bin").exists():
+        raise SystemExit(f"{MODEL / 'model.bin'} is missing; fetch Systran/faster-whisper-tiny into {MODEL}")
+    videos = sorted((CORPUS / "videos").glob("*.mp4"))
+    if not videos:
+        raise SystemExit(f"no videos under {CORPUS / 'videos'}; unpack them from videos_chunked_01.zip")
+
+    transcriber = SpeechTranscriber(Settings(asr_model_dir=MODEL))
+    rows: list[dict[str, Any]] = []
+    for video in videos:
+        subtitle = CORPUS / "subtitles" / f"{video.stem}.srt"
+        if not subtitle.exists():
+            continue
+        started = time.time()
+        segments = transcriber.transcribe(video)
+        took = time.time() - started
+        hypothesis = " ".join(segment.text for segment in segments)
+        reference = srt_text(subtitle)
+        hyp_words, ref_words = normalise(hypothesis), normalise(reference)
+        audio_seconds = max((segment.end for segment in segments), default=0.0)
+        rows.append(
+            {
+                "video": video.stem,
+                "seconds": round(audio_seconds, 1),
+                "transcribe_seconds": round(took, 2),
+                "real_time_factor": round(audio_seconds / took, 1) if took else 0.0,
+                "segments": len(segments),
+                "speech_atoms": sum(1 for segment in segments if segment.text),
+                "modality": Modality.audio.value,
+                "hypothesis_words": len(hyp_words),
+                "reference_words": len(ref_words),
+                "word_error_rate": round(word_error_rate(hyp_words, ref_words), 3),
+                "mean_word_confidence": round(
+                    sum(word.confidence for word in timed_words(segments)) / max(len(timed_words(segments)), 1), 3
+                ),
+                **check_timings(segments),
+            }
+        )
+    if not rows:
+        raise SystemExit(f"none of {len(videos)} videos has a subtitle in {CORPUS / 'subtitles'}")
+    reference_words = sum(row["reference_words"] for row in rows)
+    if not reference_words:
+        raise SystemExit("the subtitle files produced no reference words; the parser or the files are wrong")
+
+    totals = {
+        "videos": len(rows),
+        "audio_seconds": round(sum(row["seconds"] for row in rows), 1),
+        "word_error_rate": round(
+            sum(row["word_error_rate"] * row["reference_words"] for row in rows) / reference_words,
+            3,
+        ),
+        "speech_atoms": sum(row["speech_atoms"] for row in rows),
+        "words": sum(row["words"] for row in rows),
+        "out_of_order": sum(row["out_of_order"] for row in rows),
+        "before_their_segment": sum(row["before_their_segment"] for row in rows),
+    }
+    report = {
+        "generated_utc": time.strftime("%Y-%m-%dT%H:%M:%S+00:00", time.gmtime()),
+        "commit": git_sha(),
+        "model": {"id": "Systran/faster-whisper-tiny", "sha256_model_bin": model_sha256(), "compute": "cpu int8"},
+        "reference": "Video-MME community subtitles (third-party captions, not verbatim transcripts)",
+        "machine": machine_facts(),
+        "totals": totals,
+        "videos": rows,
+    }
+    REPORTS.mkdir(exist_ok=True)
+    (REPORTS / "asr.json").write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
+    write_markdown(report)
+    print(f"wrote {REPORTS / 'ASR.md'}")
+
+
+def write_markdown(report: dict[str, Any]) -> None:
+    totals, model = report["totals"], report["model"]
+    lines = [
+        f"# ASR: what the speech reader hears (commit {report['commit'][:9]})",
+        "",
+        f"- generated {report['generated_utc']}",
+        f"- model: {model['id']} ({model['compute']}), model.bin sha256 {model['sha256_model_bin'][:16]}…",
+        f"- reference: {report['reference']}",
+        f"- machine: {report['machine']}",
+        "",
+        "| video | audio s | transcribe s | ×realtime | speech atoms | words | WER | mean word confidence |",
+        "| --- | --- | --- | --- | --- | --- | --- | --- |",
+    ]
+    for row in report["videos"]:
+        lines.append(
+            f"| {row['video']} | {row['seconds']} | {row['transcribe_seconds']} | {row['real_time_factor']} "
+            f"| {row['speech_atoms']} | {row['words']} | {row['word_error_rate']} | {row['mean_word_confidence']} |"
+        )
+    lines += [
+        "",
+        (
+            f"Over {totals['videos']} videos and {totals['audio_seconds']} seconds of audio: word error rate "
+            f"**{totals['word_error_rate']}** against captions nobody here wrote, {totals['speech_atoms']} speech "
+            f"atoms carrying {totals['words']} word-level timestamps, {totals['out_of_order']} words out of order "
+            f"and {totals['before_their_segment']} starting before their own segment."
+        ),
+        "",
+        (
+            "A caption track drops words a transcript keeps, so this WER is an upper bound on the model's real "
+            "error. The point of the run is not the number: before this change an audio file produced no memory "
+            "at all, and every word here now has an instant attached to it."
+        ),
+    ]
+    (REPORTS / "ASR.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+
+if __name__ == "__main__":
+    main()

bench/reports/ASR.md

@@ -0,0 +1,16 @@
+# ASR: what the speech reader hears (commit 03ca35763)
+
+- generated 2026-10-08T01:05:45+00:00
+- model: Systran/faster-whisper-tiny (cpu int8), model.bin sha256 dcb76c6586fc06cb…
+- reference: Video-MME community subtitles (third-party captions, not verbatim transcripts)
+- machine: {'system': 'Linux', 'release': '6.18.40.1-microsoft-standard-WSL2', 'cpus': 32, 'threads_per_query': 4, 'load_average_1_5_15': [1.84, 1.85, 6.99]}
+
+| video | audio s | transcribe s | ×realtime | speech atoms | words | WER | mean word confidence |
+| --- | --- | --- | --- | --- | --- | --- | --- |
+| 0IdYJGBmguM | 88.9 | 1.73 | 51.3 | 31 | 218 | 0.179 | 0.879 |
+| 1q-5IIyZL20 | 99.9 | 1.94 | 51.6 | 32 | 300 | 0.28 | 0.83 |
+| 1tidNDIpKSY | 21.7 | 1.34 | 16.1 | 3 | 32 | 0.278 | 0.852 |
+
+Over 3 videos and 210.5 seconds of audio: word error rate **0.242** against captions nobody here wrote, 66 speech atoms carrying 550 word-level timestamps, 0 words out of order and 0 starting before their own segment.
+
+A caption track drops words a transcript keeps, so this WER is an upper bound on the model's real error. The point of the run is not the number: before this change an audio file produced no memory at all, and every word here now has an instant attached to it.

bench/reports/asr.json

@@ -0,0 +1,80 @@
+{
+  "generated_utc": "2026-10-08T01:05:45+00:00",
+  "commit": "03ca35763fd63eb8d8f0d2c721d02c6c3ded90a7",
+  "model": {
+    "id": "Systran/faster-whisper-tiny",
+    "sha256_model_bin": "dcb76c6586fc06cbdac6dd21f14cfd129cc4cdd9dce19bf4ffa62e59cbe6e6d1",
+    "compute": "cpu int8"
+  },
+  "reference": "Video-MME community subtitles (third-party captions, not verbatim transcripts)",
+  "machine": {
+    "system": "Linux",
+    "release": "6.18.40.1-microsoft-standard-WSL2",
+    "cpus": 32,
+    "threads_per_query": 4,
+    "load_average_1_5_15": [
+      1.84,
+      1.85,
+      6.99
+    ]
+  },
+  "totals": {
+    "videos": 3,
+    "audio_seconds": 210.5,
+    "word_error_rate": 0.242,
+    "speech_atoms": 66,
+    "words": 550,
+    "out_of_order": 0,
+    "before_their_segment": 0
+  },
+  "videos": [
+    {
+      "video": "0IdYJGBmguM",
+      "seconds": 88.9,
+      "transcribe_seconds": 1.73,
+      "real_time_factor": 51.3,
+      "segments": 31,
+      "speech_atoms": 31,
+      "modality": "audio",
+      "hypothesis_words": 218,
+      "reference_words": 201,
+      "word_error_rate": 0.179,
+      "mean_word_confidence": 0.879,
+      "words": 218,
+      "out_of_order": 0,
+      "before_their_segment": 0
+    },
+    {
+      "video": "1q-5IIyZL20",
+      "seconds": 99.9,
+      "transcribe_seconds": 1.94,
+      "real_time_factor": 51.6,
+      "segments": 32,
+      "speech_atoms": 32,
+      "modality": "audio",
+      "hypothesis_words": 300,
+      "reference_words": 296,
+      "word_error_rate": 0.28,
+      "mean_word_confidence": 0.83,
+      "words": 300,
+      "out_of_order": 0,
+      "before_their_segment": 0
+    },
+    {
+      "video": "1tidNDIpKSY",
+      "seconds": 21.7,
+      "transcribe_seconds": 1.34,
+      "real_time_factor": 16.1,
+      "segments": 3,
+      "speech_atoms": 3,
+      "modality": "audio",
+      "hypothesis_words": 32,
+      "reference_words": 36,
+      "word_error_rate": 0.278,
+      "mean_word_confidence": 0.852,
+      "words": 32,
+      "out_of_order": 0,
+      "before_their_segment": 0
+    }
+  ]
+}

bench/visual_recall.py

@@ -19,15 +19,12 @@ import json
 import re
 import statistics
 import subprocess
-import sys
 import time
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
 
-sys.path.insert(0, str(Path(__file__).resolve().parent))
-
 from ely_eye.config import Settings
 from ely_eye.db import Database
 from ely_eye.ocr import read_text_from_image
@@ -44,8 +41,9 @@ from ely_eye.storage import (
     visual_token_equivalent,
 )
 from ely_eye.visual_index import VisualIndex
-from metrics import aggregate, chance_recall_at_k, score_query
+
-from systems import machine_facts
+from bench.metrics import aggregate, chance_recall_at_k, score_query
+from bench.systems import machine_facts
 
 ROOT = Path(__file__).resolve().parent
 CORPUS = ROOT / "corpus" / "OS-Copilot" / "ScreenSpot-v2"

pyproject.toml

@@ -17,6 +17,6 @@ module = [
     "rank_bm25",
     "sentence_transformers.*",
     "onnxruntime.*",
-    "fitz.*",
+    "faster_whisper.*",
 ]
 ignore_missing_imports = true