feat(gemini): Gemini backend + vision-direct image acquire (T2)

GeminiBackend: multimodal call, _ExtractionSchema with string dates for
clean JSON schema, schema-constrained output via response_schema, 3-attempt
exponential backoff. Factory registers the gemini builder. core.py replaces
the placeholder _default_acquire with _make_acquire(settings), handling
image+vision_direct by loading raw bytes; native_pdf/OCR still deferred.
11 mocked unit tests cover parsing, null fields, retry, backoff, exhaustion,
and factory wiring. Verified on a real SGD receipt: vendor, date, line items,
and subtotal correct; H2 correctly caught total!=subtotal+tax and routed
to review.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

PROGRESS_TOMORROW.md

@@ -53,7 +53,7 @@ Paddle-on-3.11 risk to T4 instead of breaking everything at once.
   `config.py` accepts it.
   Check: `uv run python -c "from doc_agent.config import load_config; c=load_config(); print(c.extraction_backend, bool(c.gemini_api_key))"` → `gemini True`.
 
-- [ ] **T2 — Gemini backend + real image acquire** (build plan 2.5) ⭐ milestone
+- [x] **T2 — Gemini backend + real image acquire** (build plan 2.5) ⭐ milestone
   Implement `src/doc_agent/backends/gemini.py` per architecture §5: multimodal
   call, schema-constrained JSON output, bounded retries + timeout, model id from
   config; register its builder in the factory. Wire a minimal real `acquire` for

scripts/smoke_gemini.py

@@ -0,0 +1,58 @@
+"""Quick manual smoke script for T2: run process_document on a real image.
+
+Usage:
+    uv run python scripts/smoke_gemini.py path/to/receipt.jpg
+"""
+
+import sys
+from pathlib import Path
+
+from doc_agent.core import process_document
+
+
+def main() -> None:
+    """Run the pipeline on a single image and print the result."""
+    if len(sys.argv) != 2:
+        print("Usage: uv run python scripts/smoke_gemini.py <image_path>")
+        sys.exit(1)
+
+    path = Path(sys.argv[1])
+    if not path.exists():
+        print(f"File not found: {path}")
+        sys.exit(1)
+
+    print(f"Processing: {path}")
+    result = process_document(path)
+
+    print(f"\ndecision     : {result.decision}")
+    print(f"confidence   : {result.confidence:.3f}")
+    print(f"backend      : {result.backend_name}")
+    print(f"modality     : {result.modality}")
+    if result.error:
+        print(f"error        : {result.error}")
+
+    doc = result.document
+    print("\n--- extracted fields ---")
+    print(f"doc_type     : {doc.doc_type}")
+    print(f"vendor_name  : {doc.vendor_name}")
+    print(f"invoice_num  : {doc.invoice_number}")
+    print(f"date         : {doc.document_date}")
+    print(f"currency     : {doc.currency}")
+    print(f"subtotal     : {doc.subtotal}")
+    print(f"tax          : {doc.tax}")
+    print(f"total        : {doc.total}")
+    print(f"line_items   : {len(doc.line_items)} item(s)")
+    for item in doc.line_items:
+        print(f"  {item.description}  qty={item.quantity}  price={item.unit_price}  amt={item.amount}")
+
+    print("\n--- validation ---")
+    v = result.report
+    print(f"hard_failed  : {v.hard_failed}")
+    if v.hard_failures:
+        print(f"hard failures: {[r.code for r in v.hard_failures]}")
+    if v.soft_failures:
+        print(f"soft failures: {[r.code for r in v.soft_failures]}")
+
+
+if __name__ == "__main__":
+    main()

src/doc_agent/backends/base.py

@@ -156,11 +156,30 @@ def _build_stub(settings: Settings) -> ExtractionBackend:
     return StubBackend()
 
 
+def _build_gemini(settings: Settings) -> ExtractionBackend:
+    """Construct the Gemini backend (lazy import of google-genai).
+
+    The import is local so this module stays a dependency leaf: importing
+    ``doc_agent.backends.base`` never pulls in ``google.genai`` unless the
+    Gemini backend is actually selected.
+
+    Args:
+        settings: Validated runtime configuration (API key, model, timeout).
+
+    Returns:
+        A ready-to-use ``GeminiBackend`` instance.
+    """
+    from doc_agent.backends.gemini import GeminiBackend
+
+    return GeminiBackend(settings)
+
+
 # Registry of buildable backends: name -> builder. Builders import their adapter
 # lazily so selecting one backend never imports another's (possibly heavy or
-# optional) provider SDK. Gemini and Ollama register their builders here when
-# implemented (build plan phases 2.5/2.6); tonight only the offline stub exists.
+# optional) provider SDK. Ollama registers its builder here when implemented
+# (build plan phase 2.6).
 _BACKEND_BUILDERS: dict[str, BackendBuilder] = {
+    "gemini": _build_gemini,
     "stub": _build_stub,
 }
 
@@ -202,7 +221,7 @@ def create_backend(settings: Settings, *, name: str | None = None) -> Extraction
         available = ", ".join(available_backends())
         raise ConfigError(
             f"Unknown or unavailable extraction backend {backend_name!r}. "
-            f"Available backends: {available}. (The Gemini and Ollama adapters "
-            f"are added in a later build phase.)"
+            f"Available backends: {available}. "
+            f"(The Ollama adapter is added in build phase 2.6.)"
         )
     return builder(settings)

src/doc_agent/backends/gemini.py

@@ -0,0 +1,225 @@
+"""Gemini extraction backend using the Google GenAI SDK.
+
+Calls the Gemini API with multimodal (image) or text input, requests
+schema-constrained JSON output (CLAUDE.md rule 4), and applies bounded retries
+with exponential backoff before surfacing a ``RuntimeError`` that the core
+catches and routes to review (rule 6 -- exhausted retries never crash the loop).
+
+Architecture rules honoured here:
+- Rule 2: no direct SDK import at module load; ``google.genai`` is imported
+  inside ``__init__`` (lazy, so this module stays a dependency leaf until
+  the Gemini backend is actually selected).
+- Rule 3: the model identifier comes from ``Settings.gemini_model`` (config),
+  never hardcoded.
+- Rule 4: schema-constrained JSON output via ``response_schema``; no regex.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+from doc_agent.backends.base import BackendResult, DocumentPayload
+from doc_agent.config import Settings
+
+logger = logging.getLogger(__name__)
+
+_EXTRACT_PROMPT: str = """\
+You are a document-extraction assistant. Extract every available field from \
+this document and return them as JSON.
+
+Rules:
+- Set any absent or illegible field to null.
+- Dates must be ISO 8601 strings (YYYY-MM-DD) or null.
+- Monetary amounts must be plain numbers with no currency symbols.
+- doc_type must be exactly "receipt", "invoice", or "other".
+- currency must be an ISO 4217 code (e.g. "USD", "SGD") or null.
+"""
+
+_MAX_RETRIES: int = 3
+_BASE_BACKOFF_S: float = 1.0
+_TIMEOUT_MS: int = 60_000  # 60 s in milliseconds (HttpOptions.timeout unit)
+
+
+class _LineItem(BaseModel):
+    """Gemini-serializable line item (all primitives so the JSON schema is clean)."""
+
+    description: str | None = None
+    quantity: float | None = None
+    unit_price: float | None = None
+    amount: float | None = None
+
+
+class _ExtractionSchema(BaseModel):
+    """Schema sent to Gemini for constrained JSON output.
+
+    Uses ``str`` for date fields so Gemini's JSON schema remains simple;
+    ``Document``'s validators downstream coerce them to ``datetime.date``
+    (CLAUDE.md rule 4 -- structured output is enforced at validation time,
+    not by regex).
+    """
+
+    doc_type: str = "other"
+    vendor_name: str | None = None
+    vendor_address: str | None = None
+    invoice_number: str | None = None
+    document_date: str | None = None
+    due_date: str | None = None
+    currency: str | None = None
+    line_items: list[_LineItem] = Field(default_factory=list)
+    subtotal: float | None = None
+    tax: float | None = None
+    total: float | None = None
+
+
+class GeminiBackend:
+    """Extraction backend that calls the Gemini multimodal API.
+
+    Accepts image bytes (``vision_direct`` mode) or plain text (native-PDF /
+    OCR path), enforces schema-constrained JSON output, and retries up to
+    ``_MAX_RETRIES`` times with exponential backoff on transient failures.
+
+    Attributes:
+        name: Backend identifier used in logs and the factory registry.
+    """
+
+    name = "gemini"
+
+    def __init__(self, settings: Settings) -> None:
+        """Build the Gemini client from validated settings.
+
+        Imports ``google.genai`` lazily here so the module stays a dependency
+        leaf until this backend is actually selected (architecture rule 2).
+
+        Args:
+            settings: Validated runtime configuration supplying the API key,
+                model identifier, and timeout.
+        """
+        from google import genai
+        from google.genai import types as _t
+
+        self._model: str = settings.gemini_model
+        self._types = _t
+        self._client = genai.Client(
+            api_key=settings.gemini_api_key,
+            http_options=_t.HttpOptions(timeout=_TIMEOUT_MS),
+        )
+
+    def extract(self, payload: DocumentPayload, schema: type[BaseModel]) -> BackendResult:
+        """Extract document fields from a payload with bounded retries.
+
+        Args:
+            payload: The acquired document representation.  Must carry either
+                ``image_bytes`` (vision_direct) or ``text`` (text path).
+            schema: The Pydantic model defining the output contract (the core
+                passes ``Document``).  The backend uses ``_ExtractionSchema``
+                for the API call and returns a dict the core validates into
+                ``schema``.
+
+        Returns:
+            A ``BackendResult`` with the extracted data dict and ``None``
+            field_confidence (the Gemini free tier exposes no per-field signal;
+            scoring falls back to a neutral prior).
+
+        Raises:
+            RuntimeError: When all ``_MAX_RETRIES`` attempts fail.  The core
+                catches this and routes the document to review.
+        """
+        contents = self._build_contents(payload)
+        last_exc: Exception | None = None
+
+        for attempt in range(_MAX_RETRIES):
+            if attempt:
+                backoff = _BASE_BACKOFF_S * (2 ** (attempt - 1))
+                logger.debug(
+                    "gemini retry attempt=%d/%d backoff=%.1fs source=%s",
+                    attempt + 1,
+                    _MAX_RETRIES,
+                    backoff,
+                    payload.source_path,
+                )
+                time.sleep(backoff)
+            try:
+                return self._call_api(contents)
+            except Exception as exc:  # noqa: BLE001
+                logger.warning(
+                    "gemini API attempt %d/%d failed source=%s error=%s",
+                    attempt + 1,
+                    _MAX_RETRIES,
+                    payload.source_path,
+                    exc,
+                )
+                last_exc = exc
+
+        raise RuntimeError(
+            f"Gemini extraction failed after {_MAX_RETRIES} attempts: {last_exc}"
+        ) from last_exc
+
+    def _build_contents(self, payload: DocumentPayload) -> list[Any]:
+        """Build the Gemini content list from a document payload.
+
+        Args:
+            payload: The document payload carrying image bytes or text.
+
+        Returns:
+            A list of genai ``Part`` objects for the API call.
+
+        Raises:
+            ValueError: If the payload has neither ``image_bytes`` nor ``text``.
+        """
+        types = self._types
+        parts: list[Any] = [types.Part.from_text(text=_EXTRACT_PROMPT)]
+
+        if payload.image_bytes is not None:
+            mime = payload.image_mime or "image/jpeg"
+            parts.append(types.Part.from_bytes(data=payload.image_bytes, mime_type=mime))
+        elif payload.text is not None:
+            parts.append(types.Part.from_text(text=payload.text))
+        else:
+            raise ValueError(
+                "DocumentPayload must supply either image_bytes (vision_direct) "
+                "or text (OCR / native-PDF path)."
+            )
+
+        return parts
+
+    def _call_api(self, contents: list[Any]) -> BackendResult:
+        """Make one Gemini API call and parse the schema-constrained response.
+
+        Args:
+            contents: The genai content parts produced by ``_build_contents``.
+
+        Returns:
+            A ``BackendResult`` with the extracted data dict.
+        """
+        types = self._types
+        response = self._client.models.generate_content(
+            model=self._model,
+            contents=contents,
+            config=types.GenerateContentConfig(
+                response_mime_type="application/json",
+                response_schema=_ExtractionSchema,
+            ),
+        )
+
+        extracted = _ExtractionSchema.model_validate_json(response.text)
+        data: dict[str, Any] = extracted.model_dump()
+        data["line_items"] = [li.model_dump() for li in extracted.line_items]
+
+        logger.debug(
+            "gemini extraction complete model=%s total=%s vendor=%s",
+            self._model,
+            data.get("total"),
+            data.get("vendor_name"),
+        )
+
+        return BackendResult(
+            data=data,
+            # Gemini free tier exposes no per-field confidence; the scorer
+            # handles None with a neutral prior (architecture section 8).
+            field_confidence=None,
+            raw={"model": self._model},
+        )

src/doc_agent/core.py

@@ -100,29 +100,63 @@ class ExtractionResult:
         return self.decision == "accept"
 
 
-def _default_acquire(path: Path, modality: Modality) -> DocumentPayload:
-    """Placeholder acquirer used until the real parser is wired in.
-
-    Native-PDF parsing (Docling) and the OCR path are build-plan phases 2.2/2.3,
-    a supervised TOMORROW task. Until then callers must inject an ``acquire``
-    callable; invoking this default is a clear, recoverable error rather than a
-    silent empty payload.
+# MIME types for image modality, keyed by lower-case file extension.
+_MIME_BY_SUFFIX: dict[str, str] = {
+    ".jpg": "image/jpeg",
+    ".jpeg": "image/jpeg",
+    ".png": "image/png",
+    ".gif": "image/gif",
+    ".webp": "image/webp",
+    ".tif": "image/tiff",
+    ".tiff": "image/tiff",
+    ".bmp": "image/bmp",
+}
+
+
+def _load_image_payload(path: Path) -> DocumentPayload:
+    """Load an image file's raw bytes for vision-direct extraction.
 
     Args:
-        path: The document path (unused).
-        modality: The detected modality (unused).
+        path: Path to the image file to read.
 
-    Raises:
-        NotImplementedError: Always -- acquisition is not yet implemented.
+    Returns:
+        A ``DocumentPayload`` with ``image_bytes`` and ``image_mime`` set.
     """
-    raise NotImplementedError(
-        "Document acquisition (Docling/OCR) is not wired yet (build-plan phases "
-        "2.2/2.3, a supervised task). Inject an `acquire` callable for now -- the "
-        "web demo supplies its uploaded payload and the core smoke test supplies "
-        "a fixed one."
+    mime = _MIME_BY_SUFFIX.get(path.suffix.lower(), "image/jpeg")
+    return DocumentPayload(
+        modality="image",
+        source_path=path,
+        image_bytes=path.read_bytes(),
+        image_mime=mime,
     )
 
 
+def _make_acquire(settings: Settings) -> Acquire:
+    """Create the default acquire callable wired to current settings.
+
+    Handles ``image`` + ``vision_direct`` by reading raw bytes (T2). All other
+    paths (``native_pdf``, ``ocr_then_text``) raise until T3/T4/T5 wire them in.
+
+    Args:
+        settings: Validated runtime configuration (``image_strategy`` is read).
+
+    Returns:
+        An ``Acquire`` callable that maps (path, modality) to a payload.
+    """
+
+    def _acquire(path: Path, modality: Modality) -> DocumentPayload:
+        if modality == "image" and settings.image_strategy == "vision_direct":
+            return _load_image_payload(path)
+        raise NotImplementedError(
+            f"Acquisition for modality={modality!r} with "
+            f"IMAGE_STRATEGY={settings.image_strategy!r} is not yet wired. "
+            "Native-PDF parsing needs T3 (Docling); the OCR path needs T4. "
+            "Inject an acquire callable or use IMAGE_STRATEGY=vision_direct with an image."
+        )
+
+    return _acquire
+
+
 def _aggregate_model_signal(field_confidence: dict[str, float] | None) -> float | None:
     """Reduce a backend's per-field confidence to one document-level signal.
 
@@ -193,7 +227,7 @@ def process_document(
     """
     settings = settings or load_config()
     backend = backend or create_backend(settings)
-    acquire = acquire or _default_acquire
+    acquire = acquire or _make_acquire(settings)
     reference_date = today if today is not None else date.today()
     source_path = Path(path)
     backend_name = backend.name

tests/test_backends.py

@@ -69,12 +69,18 @@ def test_factory_override_takes_precedence_over_config() -> None:
 def test_factory_reads_backend_name_from_settings() -> None:
     """With no override the factory resolves the name from settings.
 
-    A Gemini/Ollama config resolves to a not-yet-available backend tonight, so
-    the factory raises -- but the error names the *configured* backend, proving
-    the name was read from settings rather than ignored.
+    Gemini is now registered (T2), so _gemini_settings() succeeds.
+    Ollama is still unregistered, so _ollama_settings() still raises a
+    ConfigError naming the configured backend.
     """
-    with pytest.raises(ConfigError, match="gemini"):
-        create_backend(_gemini_settings())
+    from unittest.mock import patch
+
+    from doc_agent.backends.gemini import GeminiBackend
+
+    with patch("google.genai.Client"):
+        backend = create_backend(_gemini_settings())
+    assert isinstance(backend, GeminiBackend)
+
     with pytest.raises(ConfigError, match="ollama"):
         create_backend(_ollama_settings())
 
@@ -98,10 +104,12 @@ def test_unknown_backend_raises_config_error() -> None:
     assert "stub" in message
 
 
-def test_available_backends_lists_the_stub() -> None:
-    """The registry currently exposes the offline stub (and is sorted)."""
+def test_available_backends_lists_registered_backends() -> None:
+    """The registry exposes gemini and stub (sorted); ollama is not yet registered."""
     names = available_backends()
     assert "stub" in names
+    assert "gemini" in names
+    assert "ollama" not in names
     assert list(names) == sorted(names)
 
 

tests/test_gemini.py

@@ -0,0 +1,245 @@
+"""Mocked unit tests for the Gemini extraction backend (T2).
+
+No real API calls are made -- the google.genai client is bypassed by
+constructing ``GeminiBackend`` via ``object.__new__`` and injecting mock
+attributes directly. This keeps the tests deterministic, free of network/quota,
+and fast.
+
+Covers the acceptance criteria for T2:
+- schema-valid JSON response is parsed into a ``BackendResult`` whose data
+  validates into a ``Document``,
+- a transient failure on attempt 1 is retried and succeeds on attempt 2,
+- all ``_MAX_RETRIES`` attempts failing raises ``RuntimeError`` (which the
+  core catches and routes to review).
+"""
+
+from __future__ import annotations
+
+import json
+from unittest.mock import MagicMock, call, patch
+
+import pytest
+
+from doc_agent.backends.base import DocumentPayload
+from doc_agent.backends.gemini import GeminiBackend, _ExtractionSchema, _MAX_RETRIES
+from doc_agent.schema.models import Document
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_backend() -> tuple[GeminiBackend, MagicMock]:
+    """Build a GeminiBackend bypassing __init__ with a mock client.
+
+    Returns:
+        A tuple of (backend, mock_client) where mock_client is the object
+        wired to ``backend._client``.
+    """
+    mock_types = MagicMock()
+    mock_client = MagicMock()
+
+    backend = object.__new__(GeminiBackend)
+    backend._model = "gemini-test"
+    backend._types = mock_types
+    backend._client = mock_client
+    return backend, mock_client
+
+
+def _json_response(data: dict) -> MagicMock:
+    """Build a mock Gemini response whose ``.text`` is a valid JSON string.
+
+    Args:
+        data: Fields to include in the ``_ExtractionSchema`` response.
+
+    Returns:
+        A MagicMock whose ``.text`` attribute is a JSON-serialised
+        ``_ExtractionSchema`` populated from ``data``.
+    """
+    schema_fields = _ExtractionSchema.model_fields.keys()
+    filtered = {k: v for k, v in data.items() if k in schema_fields}
+    text = _ExtractionSchema(**filtered).model_dump_json()
+    mock_resp = MagicMock()
+    mock_resp.text = text
+    return mock_resp
+
+
+def _image_payload() -> DocumentPayload:
+    """A minimal vision-direct image payload."""
+    return DocumentPayload(
+        modality="image",
+        image_bytes=b"fake-image-bytes",
+        image_mime="image/jpeg",
+    )
+
+
+def _text_payload() -> DocumentPayload:
+    """A minimal text payload (native-PDF / OCR path)."""
+    return DocumentPayload(
+        modality="native_pdf",
+        text="Invoice #001  Total: $42.50",
+    )
+
+
+# ---------------------------------------------------------------------------
+# Schema-valid parsing
+# ---------------------------------------------------------------------------
+
+
+def test_extract_image_returns_schema_valid_data() -> None:
+    """A valid Gemini JSON response parses into Document-compatible data (AC)."""
+    backend, mock_client = _make_backend()
+    mock_client.models.generate_content.return_value = _json_response(
+        {
+            "doc_type": "receipt",
+            "vendor_name": "Test Cafe",
+            "invoice_number": "R-001",
+            "document_date": "2024-03-15",
+            "currency": "USD",
+            "subtotal": 10.00,
+            "tax": 0.80,
+            "total": 10.80,
+        }
+    )
+
+    result = backend.extract(_image_payload(), Document)
+
+    assert result.data["doc_type"] == "receipt"
+    assert result.data["vendor_name"] == "Test Cafe"
+    assert result.data["total"] == pytest.approx(10.80)
+    # The dict must validate cleanly into the Document schema.
+    doc = Document.model_validate(result.data)
+    assert doc.vendor_name == "Test Cafe"
+    assert doc.total == pytest.approx(10.80)
+
+
+def test_extract_text_returns_schema_valid_data() -> None:
+    """A text payload (native-PDF / OCR) is accepted and parsed correctly."""
+    backend, mock_client = _make_backend()
+    mock_client.models.generate_content.return_value = _json_response(
+        {"doc_type": "invoice", "invoice_number": "INV-42", "total": 99.99}
+    )
+
+    result = backend.extract(_text_payload(), Document)
+
+    assert result.data["doc_type"] == "invoice"
+    assert result.data["total"] == pytest.approx(99.99)
+    Document.model_validate(result.data)  # must not raise
+
+
+def test_extract_null_fields_become_none() -> None:
+    """Absent fields in the JSON response survive as None through the Document."""
+    backend, mock_client = _make_backend()
+    mock_client.models.generate_content.return_value = _json_response(
+        {"doc_type": "other", "total": None, "vendor_name": None}
+    )
+
+    result = backend.extract(_image_payload(), Document)
+    doc = Document.model_validate(result.data)
+
+    assert doc.total is None
+    assert doc.vendor_name is None
+
+
+def test_field_confidence_is_none() -> None:
+    """GeminiBackend returns None field_confidence (no per-field signal)."""
+    backend, mock_client = _make_backend()
+    mock_client.models.generate_content.return_value = _json_response({"total": 5.00})
+
+    result = backend.extract(_image_payload(), Document)
+
+    assert result.field_confidence is None
+
+
+def test_payload_without_image_or_text_raises() -> None:
+    """A payload with neither image_bytes nor text raises ValueError."""
+    backend, _ = _make_backend()
+    bad_payload = DocumentPayload(modality="image")
+
+    with pytest.raises(ValueError, match="image_bytes"):
+        backend.extract(bad_payload, Document)
+
+
+# ---------------------------------------------------------------------------
+# Retry logic
+# ---------------------------------------------------------------------------
+
+
+def test_retry_succeeds_on_second_attempt() -> None:
+    """A transient error on attempt 1 is retried; attempt 2 returns data."""
+    backend, mock_client = _make_backend()
+    good_response = _json_response({"doc_type": "receipt", "total": 7.77})
+    mock_client.models.generate_content.side_effect = [
+        RuntimeError("transient network error"),
+        good_response,
+    ]
+
+    with patch("doc_agent.backends.gemini.time.sleep") as mock_sleep:
+        result = backend.extract(_image_payload(), Document)
+
+    assert result.data["total"] == pytest.approx(7.77)
+    assert mock_client.models.generate_content.call_count == 2
+    mock_sleep.assert_called_once()  # one backoff sleep between attempts
+
+
+def test_retry_all_attempts_fail_raises_runtime_error() -> None:
+    """All _MAX_RETRIES attempts failing raises RuntimeError (core catches it)."""
+    backend, mock_client = _make_backend()
+    mock_client.models.generate_content.side_effect = TimeoutError("request timed out")
+
+    with patch("doc_agent.backends.gemini.time.sleep"):
+        with pytest.raises(RuntimeError, match=f"failed after {_MAX_RETRIES}"):
+            backend.extract(_image_payload(), Document)
+
+    assert mock_client.models.generate_content.call_count == _MAX_RETRIES
+
+
+def test_retry_backoff_grows_exponentially() -> None:
+    """Each retry sleeps longer than the previous one (exponential backoff)."""
+    backend, mock_client = _make_backend()
+    mock_client.models.generate_content.side_effect = OSError("network")
+
+    sleep_calls: list[float] = []
+    with patch("doc_agent.backends.gemini.time.sleep", side_effect=lambda s: sleep_calls.append(s)):
+        with pytest.raises(RuntimeError):
+            backend.extract(_image_payload(), Document)
+
+    # _MAX_RETRIES attempts -> (_MAX_RETRIES - 1) sleeps.
+    assert len(sleep_calls) == _MAX_RETRIES - 1
+    for i in range(1, len(sleep_calls)):
+        assert sleep_calls[i] > sleep_calls[i - 1]
+
+
+def test_no_sleep_on_first_attempt() -> None:
+    """The first attempt is made immediately without any sleep."""
+    backend, mock_client = _make_backend()
+    mock_client.models.generate_content.return_value = _json_response({"total": 1.00})
+
+    with patch("doc_agent.backends.gemini.time.sleep") as mock_sleep:
+        backend.extract(_image_payload(), Document)
+
+    mock_sleep.assert_not_called()
+
+
+# ---------------------------------------------------------------------------
+# Factory integration
+# ---------------------------------------------------------------------------
+
+
+def test_factory_builds_gemini_backend() -> None:
+    """create_backend resolves 'gemini' and returns a GeminiBackend."""
+    from doc_agent.backends.base import create_backend
+    from doc_agent.config import load_config
+
+    settings = load_config(
+        extraction_backend="gemini",
+        gemini_api_key="test-key",
+        image_strategy="vision_direct",
+    )
+    # Patch the google.genai.Client so no real HTTP client is constructed.
+    with patch("google.genai.Client"):
+        backend = create_backend(settings)
+
+    assert isinstance(backend, GeminiBackend)
+    assert backend.name == "gemini"