Cost "f": price the agent at the model it runs; thinking/vision token breakdown; model switch; near-duplicate gate

The cost column priced gemini-3.5-flash calls at Gemini 2.5 Flash rates
($0.30 / $2.50 per M) while the model is $1.50 / $9.00: it showed about a
quarter of the bill (runs 544-593: $12.98 shown, ~$48.50 real, ~$0.41 per
PDF; AI Studio billed $4.27 on 28 Sep for cycles the Space showed at <$1).

- config: MODEL_PRICES per model (3.5 flash, 3.5 flash-lite, 3 flash preview,
2.5 flash, 2.5 flash-lite), BATCH_DISCOUNT, model_price(), cost_usd(model,
batch), run_cost_usd() pricing text and vision tokens at their own models;
AGENT_PRICE_*_PER_M only as an explicit override of every model.
- extraction: usage_detail() -> (in, out, thinking); Extraction.tokens_thinking;
text_model()/vision_model() read GEMINI_MODEL / GEMINI_VISION_MODEL at call
time (dataclass model defaults were bound at import); extraction_from_response()
split out (accepts dicts, for the Batch API).
- figures: calls go to vision_model(); VisionStats.tokens_thinking + model.
- batch_ingest: PdfResult tokens_thinking, vision_tokens_in/out, text_model,
vision_model, duplicate_of; process_pdf(pre_extracted=...) skips the text
call; summarize() reports the breakdown.
- agent_runs: tokens_thinking, vision_tokens_in/out, duplicate_text, text_model,
vision_model, cost_usd (stored at finish, priced per model).
- Run Control "Gemini cost": text model, vision model (default: same), near-dup
threshold; Logs: Thinking, Near-dup, Model, Cost (stored; older runs re-priced
at gemini-3.5-flash), $/PDF, and a spend line over the listed cycles.
- agent/textdedupe.py: MinHash (128 x word 5-shingles) of the PDF text; a PDF
>= dedupe_text_similarity (default 0.9) to an ingested document is skipped
before any Gemini call (registry 'duplicate_text'); claims are registered at
the gate under a lock (parallel workers) and released when a PDF fails for good.
- test_cost_e2e.py (38 checks).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018Db9kKLweSLwRrQm63Uot3

agent/agdb.py

@@ -157,6 +157,15 @@ _RUNS_EXTRA_COLS = (
     ("heartbeat_at", "timestamptz"),               # last sign of progress (events)
     ("kind", "text DEFAULT 'cycle'"),              # 'cycle' | 'discovery' (28 Sep 2026)
     ("queued", "integer DEFAULT 0"),               # discovery pass: candidates queued
+    # 30 Sep 2026: cost breakdown and per-run pricing (the old derived cost
+    # used Gemini 2.5 Flash prices for a gemini-3.5-flash agent).
+    ("tokens_thinking", "bigint DEFAULT 0"),       # thinking share of tokens_out
+    ("vision_tokens_in", "bigint DEFAULT 0"),      # vision share of tokens_in
+    ("vision_tokens_out", "bigint DEFAULT 0"),     # vision share of tokens_out
+    ("duplicate_text", "integer DEFAULT 0"),       # PDFs skipped as near-duplicates
+    ("text_model", "text"),                        # Gemini model of the text calls
+    ("vision_model", "text"),                      # Gemini model of the vision calls
+    ("cost_usd", "double precision"),              # priced per model at finish
 )
 
 # Integer counters persisted by finish_run, in column order. Metric keys equal
@@ -167,13 +176,16 @@ _RUN_COUNTER_COLS = (
     "figures_found", "figures_mined", "figure_rows", "vision_calls",
     "relevance_rejected", "url_seen_skipped", "download_failed",
     "tokens_in", "tokens_out", "rows_linked", "queued",
+    "tokens_thinking", "vision_tokens_in", "vision_tokens_out", "duplicate_text",
 )
 _METRIC_ALIASES = {"url_seen_skipped": ("url_seen_skipped", "skipped_url")}
 
 
 def ensure_agent_schema(conn) -> None:
+    from agent import textdedupe
     with conn.cursor() as cur:
         cur.execute(DDL)
+        cur.execute(textdedupe.DDL)
         for name, typ in _RUNS_EXTRA_COLS:
             cur.execute(f"ALTER TABLE agent_runs ADD COLUMN IF NOT EXISTS {name} {typ}")
     conn.commit()
@@ -289,14 +301,20 @@ def finish_run(conn, run_id: int, status: str, metrics: dict, report: dict) -> N
         return 0
 
     sources = metrics.get("sources")
+    from agent import config as C
+    spent = int(metrics.get("tokens_in", 0) or 0) + int(metrics.get("tokens_out", 0) or 0)
+    text_model = (metrics.get("text_model") or None) if spent else None
+    vision_model = (metrics.get("vision_model") or None) if int(metrics.get("vision_tokens_in", 0) or 0) else None
+    cost = round(C.run_cost_usd(metrics), 6) if spent else 0.0
     with conn.cursor() as cur:
         cur.execute(
             f"UPDATE agent_runs SET finished_at = now(), status = %s, {sets}, "
-            f"sources = %s::jsonb, report = %s "
+            f"sources = %s::jsonb, report = %s, text_model = %s, vision_model = %s, "
+            f"cost_usd = %s "
             f"WHERE id = %s AND status = 'running'",
             (status, *[_val(c) for c in cols],
              json.dumps(sources) if sources else None,
-             redact(json.dumps(report))[:100_000], run_id),
+             redact(json.dumps(report))[:100_000], text_model, vision_model, cost, run_id),
         )
     conn.commit()
 

agent/config.py

@@ -132,6 +132,18 @@ DEFAULTS = {
     # Thinking tokens are billed as output (8x input); "low" cut a cycle's
     # bill by roughly two thirds. extraction.THINKING is set from this each cycle.
     "gemini_thinking": os.environ.get("GEMINI_THINKING", "low").strip().lower() or "dynamic",
+    # Gemini models (30 Sep 2026). gemini-3.5-flash is $1.50 / $9.00 per M
+    # in/out; gemini-3.5-flash-lite $0.30 / $2.50. The text model is what the
+    # paper's numbers are stamped with (rows carry it in `model`), so it only
+    # changes on purpose. Vision model "" = same as the text model; the one
+    # classify call per PDF is an easy task for the Lite model.
+    "gemini_model": os.environ.get("GEMINI_MODEL", "gemini-3.5-flash").strip() or "gemini-3.5-flash",
+    "gemini_vision_model": os.environ.get("GEMINI_VISION_MODEL", "").strip(),
+    # Near-duplicate gate: a PDF whose text is >= this similar to a document
+    # already ingested (regional datasheet variants, a preprint and its
+    # published version, the same PDF under another URL) is skipped before
+    # any Gemini call. 0 disables.
+    "dedupe_text_similarity": _f("AGENT_DEDUPE_TEXT_SIMILARITY", 0.9),
     "max_figures_per_pdf": _i("AGENT_MAX_FIGURES", 12),
     # Figure citation linking (figure_links.py): a text row that cites
     # "Fig. 3" gets figure_id/figure_ref/score/signals + the PNG embedded in
@@ -172,16 +184,74 @@ DISCOVERY_TIMEOUT_MINUTES = _f("AGENT_DISCOVERY_TIMEOUT_MINUTES", 45.0)
 # query). 0 disables; the hard budget still applies.
 CYCLE_STALL_MINUTES = _f("AGENT_CYCLE_STALL_MINUTES", 15.0)
 
-# Display-only price table for the cost column (USD per million tokens).
-# Gemini 2.5 Flash list price at the time of writing; override via env when
-# it changes. Tokens are what is stored; dollars are derived at read time.
-PRICE_IN_PER_M = _f("AGENT_PRICE_IN_PER_M", 0.30)
-PRICE_OUT_PER_M = _f("AGENT_PRICE_OUT_PER_M", 2.50)
+# Gemini list prices, USD per million tokens (input, output; thinking tokens
+# bill as output). From ai.google.dev/gemini-api/docs/pricing, 30 Sep 2026.
+# Until 30 Sep the cost column used one pair, 0.30 / 2.50 (Gemini 2.5 Flash),
+# while the agent ran gemini-3.5-flash at 1.50 / 9.00: it showed ~1/4 of the
+# real bill. Runs now store their cost (priced per model at the end of the
+# cycle); older runs are re-priced at read time with LEGACY_RUN_MODEL.
+MODEL_PRICES = {
+    "gemini-3.5-flash": (1.50, 9.00),
+    "gemini-3.5-flash-lite": (0.30, 2.50),
+    "gemini-3-flash-preview": (0.50, 3.00),
+    "gemini-2.5-flash": (0.30, 2.50),
+    "gemini-2.5-flash-lite": (0.10, 0.40),
+}
+BATCH_DISCOUNT = 0.5          # Batch API: half the standard price
+# Every run before per-run pricing (tokens were first logged on 22 Sep 2026)
+# used this model.
+LEGACY_RUN_MODEL = "gemini-3.5-flash"
+# Explicit override for a model not in the table (or a price change): both
+# must be set; they then apply to every model.
+_PRICE_IN_ENV = os.environ.get("AGENT_PRICE_IN_PER_M", "").strip()
+_PRICE_OUT_ENV = os.environ.get("AGENT_PRICE_OUT_PER_M", "").strip()
+
+
+def model_price(model: str | None) -> tuple[float, float]:
+    """(input, output) USD per million tokens for `model`."""
+    if _PRICE_IN_ENV and _PRICE_OUT_ENV:
+        try:
+            return float(_PRICE_IN_ENV), float(_PRICE_OUT_ENV)
+        except ValueError:
+            pass
+    m = (model or LEGACY_RUN_MODEL).strip().lower()
+    if m.startswith("models/"):
+        m = m[len("models/"):]
+    if m in MODEL_PRICES:
+        return MODEL_PRICES[m]
+    # dated / suffixed variants ("gemini-3.5-flash-001"): longest known prefix
+    for k in sorted(MODEL_PRICES, key=len, reverse=True):
+        if m.startswith(k):
+            return MODEL_PRICES[k]
+    return MODEL_PRICES[LEGACY_RUN_MODEL]
+
+
+# Kept for callers that show "the" price (the Logs column help).
+PRICE_IN_PER_M, PRICE_OUT_PER_M = model_price(os.environ.get("GEMINI_MODEL", LEGACY_RUN_MODEL))
+
+
+def cost_usd(tokens_in: int, tokens_out: int, model: str | None = None,
+             batch: bool = False) -> float:
+    pin, pout = model_price(model)
+    c = (float(tokens_in or 0) * pin + float(tokens_out or 0) * pout) / 1e6
+    return c * (BATCH_DISCOUNT if batch else 1.0)
 
 
-def cost_usd(tokens_in: int, tokens_out: int) -> float:
-    return (float(tokens_in or 0) * PRICE_IN_PER_M
-            + float(tokens_out or 0) * PRICE_OUT_PER_M) / 1e6
+def run_cost_usd(metrics: dict) -> float:
+    """A cycle's cost from its metrics: text and vision tokens priced at their
+    own models, batch-mode text calls at the batch discount."""
+    tin = int(metrics.get("tokens_in", 0) or 0)
+    tout = int(metrics.get("tokens_out", 0) or 0)
+    vin = int(metrics.get("vision_tokens_in", 0) or 0)
+    vout = int(metrics.get("vision_tokens_out", 0) or 0)
+    bin_ = int(metrics.get("batch_tokens_in", 0) or 0)
+    bout = int(metrics.get("batch_tokens_out", 0) or 0)
+    tm = metrics.get("text_model") or LEGACY_RUN_MODEL
+    vm = metrics.get("vision_model") or tm
+    text_in, text_out = max(0, tin - vin - bin_), max(0, tout - vout - bout)
+    return (cost_usd(text_in, text_out, tm)
+            + cost_usd(bin_, bout, tm, batch=True)
+            + cost_usd(vin, vout, vm))
 
 # Keep-alive: the container requests its own public URL every N minutes so
 # the hosting platform never sees the Space idle (free Spaces sleep after 48 h

agent/orchestrator.py

@@ -88,6 +88,9 @@ def node_plan(state: AgentState) -> AgentState:
         # Cost knob: how much every Gemini call in this cycle may think.
         import extraction as _ex
         _ex.THINKING = str(cfg.get("gemini_thinking", _ex.THINKING) or "dynamic").strip().lower()
+        # Cost knob: which models the text and figure calls go to.
+        _ex.GEMINI_MODEL = str(cfg.get("gemini_model") or _ex.GEMINI_MODEL).strip()
+        _ex.GEMINI_VISION_MODEL = str(cfg.get("gemini_vision_model") or "").strip()
         # Budget hold: Gemini refused the previous cycle for budget reasons
         # (prepaid credits depleted, spending cap). Downloading more would
         # only pile PDFs up on the ephemeral disk, so this cycle downloads
@@ -651,6 +654,10 @@ def node_ingest(state: AgentState) -> AgentState:
     m.setdefault("vision_calls", 0)
     m.setdefault("tokens_in", 0)
     m.setdefault("tokens_out", 0)
+    m.setdefault("tokens_thinking", 0)
+    m.setdefault("vision_tokens_in", 0)
+    m.setdefault("vision_tokens_out", 0)
+    m.setdefault("duplicate_text", 0)
     m.setdefault("rows_linked", 0)
     results = []
     _adopt_leftover_pdfs(state)
@@ -721,6 +728,16 @@ def node_ingest(state: AgentState) -> AgentState:
                                + "; ".join(problems)
                                + " -- run `python pg_migrate.py --apply`",
                                level="warn", node="ingest", run_id=state["run_id"])
+        # Near-duplicate gate (agent/textdedupe.py): skip a PDF whose text
+        # repeats a document already ingested, before any Gemini call.
+        text_gate = None
+        try:
+            sim = float(cfg.get("dedupe_text_similarity", 0.0) or 0.0)
+        except (TypeError, ValueError):
+            sim = 0.0
+        if 0.0 < sim <= 1.0:
+            from agent import textdedupe
+            text_gate = textdedupe.Gate(agdb.connect, sim)
         rows_by_file: dict[str, int] = {}
         workers = max(1, min(int(cfg.get("ingest_workers", C.INGEST_WORKERS_DEFAULT)),
                              C.HARD_MAX_INGEST_WORKERS, len(state["downloads"])))
@@ -736,7 +753,7 @@ def node_ingest(state: AgentState) -> AgentState:
         kept: list[str] = []             # PDFs left on disk for the next cycle
         budget_reason = ""
         for rec, result, exc in _ingest_all(state["downloads"], figure_opts, link_opts,
-                                            workers, stop):
+                                            workers, stop, text_gate=text_gate):
             pdf_path = C.PDF_DIR / rec["filename"]
             if isinstance(exc, _BudgetStop) or (result is not None and getattr(result, "quota", False)):
                 # Gemini refused for budget reasons (monthly spending cap,
@@ -754,6 +771,8 @@ def node_ingest(state: AgentState) -> AgentState:
             if exc is not None:
                 m["errors"] += 1
                 agdb.update_ingest_status(meta_conn, rec["filename"], f"error")
+                if text_gate is not None:
+                    text_gate.release_file(rec["filename"])
                 if rec.get("_cand_id"):
                     agdb.mark_candidate(meta_conn, rec["_cand_id"], "failed", str(exc))
                 tb = getattr(exc, "__traceback_text__", "")
@@ -762,10 +781,23 @@ def node_ingest(state: AgentState) -> AgentState:
                                + (f"\n{tb}" if tb else ""),
                                level="error", node="ingest", run_id=state["run_id"])
                 continue
-            results.append(result)
-            m["tokens_in"] += int(getattr(result, "tokens_in", 0) or 0)
-            m["tokens_out"] += int(getattr(result, "tokens_out", 0) or 0)
+            _add_usage(m, result)
             m["rows_linked"] += int(getattr(result, "rows_linked", 0) or 0)
+            if result.error == "duplicate_text":
+                # Near-duplicate of a document already ingested: nothing was
+                # sent to Gemini. Final, like a DOI-level duplicate.
+                m["duplicate_text"] += 1
+                agdb.update_ingest_status(meta_conn, rec["filename"], "duplicate_text")
+                if rec.get("_cand_id"):
+                    agdb.mark_candidate(meta_conn, rec["_cand_id"], "duplicate",
+                                        f"text repeats {result.duplicate_of}")
+                agdb.log_event(meta_conn, f"{rec['filename']}: skipped before extraction — "
+                                          f"its text repeats an ingested document "
+                                          f"({str(result.duplicate_of)[:12]})",
+                               node="ingest", run_id=state["run_id"])
+                pdf_path.unlink(missing_ok=True)
+                continue
+            results.append(result)
             if result.error:
                 if str(result.error).startswith("gemini_error"):
                     # Transient (rate limit, 5xx, network): keep the PDF for
@@ -777,6 +809,8 @@ def node_ingest(state: AgentState) -> AgentState:
                                    level="warn", node="ingest", run_id=state["run_id"])
                     continue
                 agdb.update_ingest_status(meta_conn, rec["filename"], result.error)
+                if text_gate is not None:
+                    text_gate.release_file(rec["filename"])   # a variant may stand in later
                 if rec.get("_cand_id"):
                     agdb.mark_candidate(meta_conn, rec["_cand_id"], "failed", result.error)
                 agdb.log_event(meta_conn, f"{rec['filename']}: {result.error}",
@@ -864,12 +898,24 @@ def node_ingest(state: AgentState) -> AgentState:
     return state
 
 
+def _add_usage(m: dict, result) -> None:
+    """Fold one PdfResult's Gemini usage into the cycle metrics."""
+    for k in ("tokens_in", "tokens_out", "tokens_thinking",
+              "vision_tokens_in", "vision_tokens_out"):
+        m[k] = int(m.get(k, 0) or 0) + int(getattr(result, k, 0) or 0)
+    if getattr(result, "text_model", ""):
+        m["text_model"] = result.text_model
+    if getattr(result, "vision_model", ""):
+        m["vision_model"] = result.vision_model
+
+
 class _BudgetStop(Exception):
     """Marker: this PDF was not attempted because Gemini already refused
     this cycle for budget reasons (spending cap / quota)."""
 
 
-def _ingest_one(rec: dict, figure_opts, link_opts, stop: Optional[threading.Event] = None):
+def _ingest_one(rec: dict, figure_opts, link_opts, stop: Optional[threading.Event] = None,
+                text_gate=None):
     """Worker: extract one PDF on a private Postgres connection.
     Returns (rec, PdfResult or None, exception or None)."""
     pdf_path = C.PDF_DIR / rec["filename"]
@@ -877,6 +923,12 @@ def _ingest_one(rec: dict, figure_opts, link_opts, stop: Optional[threading.Even
         return rec, None, _BudgetStop()
     pg = None
     try:
+        if text_gate is not None:
+            dup = _near_duplicate(text_gate, pdf_path)
+            if dup:
+                res = batch_ingest._empty_result(pdf_path, time.time(), "duplicate_text")
+                res.duplicate_of = dup
+                return rec, res, None
         pg = pg_mirror.connect_from_env()
         result = batch_ingest.process_pdf(pdf_path, pg, C.GEMINI_API_KEY,
                                           db=pg_mirror,
@@ -902,8 +954,20 @@ def _ingest_one(rec: dict, figure_opts, link_opts, stop: Optional[threading.Even
                 pass
 
 
+def _near_duplicate(text_gate, pdf_path) -> Optional[str]:
+    """sha1 of the ingested document this PDF's text repeats, or None. The
+    gate is an optimisation: any failure means 'not a duplicate'."""
+    try:
+        data = pdf_path.read_bytes()
+        import hashlib
+        return text_gate(data, hashlib.sha1(data).hexdigest(), pdf_path.name)
+    except Exception as exc:
+        log.warning("near-duplicate gate failed for %s: %s", pdf_path.name, exc)
+        return None
+
+
 def _ingest_all(downloads: list, figure_opts, link_opts, workers: int,
-                stop: Optional[threading.Event] = None):
+                stop: Optional[threading.Event] = None, text_gate=None):
     """Yield (rec, result, exc) for every download, in completion order.
     workers == 1 keeps the old one-at-a-time behaviour (and the old
     connection count) — the only difference is the private connection.
@@ -911,11 +975,12 @@ def _ingest_all(downloads: list, figure_opts, link_opts, workers: int,
     started come back as _BudgetStop without a request."""
     if workers <= 1:
         for rec in downloads:
-            yield _ingest_one(rec, figure_opts, link_opts, stop)
+            yield _ingest_one(rec, figure_opts, link_opts, stop, text_gate)
         return
     from concurrent.futures import ThreadPoolExecutor, as_completed
     with ThreadPoolExecutor(max_workers=workers, thread_name_prefix="ingest") as pool:
-        futures = [pool.submit(_ingest_one, rec, figure_opts, link_opts, stop) for rec in downloads]
+        futures = [pool.submit(_ingest_one, rec, figure_opts, link_opts, stop, text_gate)
+                   for rec in downloads]
         for fut in as_completed(futures):
             yield fut.result()
 
@@ -1008,8 +1073,11 @@ def node_report(state: AgentState) -> AgentState:
                        f"{m.get('figures_found', 0)} harvested, "
                        f"{m.get('figure_rows', 0)} estimate rows, "
                        f"{m.get('rows_linked', 0)} rows linked; "
-                       f"tokens {m.get('tokens_in', 0)} in / {m.get('tokens_out', 0)} out "
-                       f"(~${C.cost_usd(m.get('tokens_in', 0), m.get('tokens_out', 0)):.4f})",
+                       f"tokens {m.get('tokens_in', 0)} in / {m.get('tokens_out', 0)} out, "
+                       f"{m.get('tokens_thinking', 0)} of it thinking "
+                       f"(~${C.run_cost_usd(m):.4f} at {m.get('text_model') or C.LEGACY_RUN_MODEL} prices)"
+                       + (f"; {m['duplicate_text']} near-duplicate PDF(s) skipped"
+                          if m.get("duplicate_text") else ""),
                        node="report", run_id=state["run_id"])
     finally:
         conn.close()

agent/textdedupe.py

@@ -0,0 +1,182 @@
+"""Near-duplicate gate: skip a PDF whose text repeats a document already
+ingested, before any Gemini call is made (30 Sep 2026).
+
+Why: the same content arrives under different bytes — a manufacturer's
+regional datasheet variants (…global_version… / …eu_version…), a preprint
+and its published version, the same paper from a repository and a publisher.
+The sha1 check only catches byte-identical files, so each variant was sent
+to Gemini (~$0.10–0.40 each) for rows the dedup grain then threw away.
+
+How: MinHash over word 5-shingles of the PDF's extracted text (PyMuPDF, the
+same text the grounding check uses). 128 slots, stored as 1 KB per document
+in agent_text_fingerprints. The estimated Jaccard similarity of two
+documents is the share of equal slots; at >= the threshold (default 0.9,
+Run Control) the new PDF is skipped. Different datasheets from one template
+share boilerplate but not their numbers, and every changed number breaks
+five shingles, so a family of sheets stays well below 0.9.
+
+A fingerprint is registered when its PDF passes the gate (a claim), so two
+variants in the same cycle — even on parallel workers — cannot both pass.
+If the claimed PDF then fails for good, release() drops the claim.
+Documents ingested before this existed have no fingerprint (their PDFs are
+gone), so the gate only knows documents from its deploy on.
+"""
+from __future__ import annotations
+
+import hashlib
+import logging
+import re
+import threading
+from typing import Callable, Optional
+
+import numpy as np
+
+log = logging.getLogger("agent.textdedupe")
+
+NUM_PERM = 128
+SHINGLE = 5
+MIN_SHINGLES = 40          # below this there is too little text to judge
+_PRIME = np.uint64((1 << 61) - 1)
+_rng = np.random.RandomState(20260930)
+_A = _rng.randint(1, 1 << 31, size=NUM_PERM).astype(np.uint64)
+_B = _rng.randint(0, 1 << 31, size=NUM_PERM).astype(np.uint64)
+_WORD = re.compile(r"[a-z0-9]+(?:[.,][0-9]+)*")
+
+DDL = """
+CREATE TABLE IF NOT EXISTS agent_text_fingerprints (
+    sha1 text PRIMARY KEY,
+    filename text,
+    minhash bytea NOT NULL,
+    n_shingles integer,
+    created_at timestamptz DEFAULT now()
+);
+"""
+
+
+def _words(text: str) -> list[str]:
+    return _WORD.findall((text or "").lower())
+
+
+def fingerprint(text: str) -> Optional[tuple[bytes, int]]:
+    """(minhash bytes, number of distinct shingles), or None for too little text."""
+    w = _words(text)
+    if len(w) < SHINGLE:
+        return None
+    sh = {" ".join(w[i:i + SHINGLE]) for i in range(len(w) - SHINGLE + 1)}
+    if len(sh) < MIN_SHINGLES:
+        return None
+    # one 31-bit hash per shingle, then NUM_PERM universal hashes of it
+    h = np.fromiter((int.from_bytes(hashlib.blake2b(s.encode(), digest_size=4).digest(), "little")
+                     for s in sh), dtype=np.uint64, count=len(sh))
+    perm = (np.outer(h, _A) + _B) % _PRIME          # (n_shingles, NUM_PERM), < 2^61: no overflow
+    return perm.min(axis=0).astype(np.uint64).tobytes(), len(sh)
+
+
+def similarity(a: bytes, b: bytes) -> float:
+    x = np.frombuffer(a, dtype=np.uint64)
+    y = np.frombuffer(b, dtype=np.uint64)
+    if x.shape != y.shape or not len(x):
+        return 0.0
+    return float(np.mean(x == y))
+
+
+def ensure_table(conn) -> None:
+    with conn.cursor() as cur:
+        cur.execute(DDL)
+    conn.commit()
+
+
+class Gate:
+    """Callable for batch_ingest.process_pdf(text_gate=...):
+    gate(pdf_bytes, sha1) -> sha1 of the ingested document this PDF repeats,
+    or None (and the PDF's fingerprint is claimed)."""
+
+    _lock = threading.Lock()          # parallel ingest workers share one process
+
+    def __init__(self, connect: Callable, threshold: float,
+                 text_of: Optional[Callable[[bytes], str]] = None):
+        self.connect = connect
+        self.threshold = float(threshold or 0.0)
+        self.text_of = text_of or _pdf_text
+        self._cache: Optional[list[tuple[str, bytes]]] = None
+        self.claimed: dict[str, str] = {}     # sha1 -> filename, claimed by this gate
+
+    @property
+    def enabled(self) -> bool:
+        return 0.0 < self.threshold <= 1.0
+
+    def _load(self, conn) -> list[tuple[str, bytes]]:
+        if self._cache is None:
+            with conn.cursor() as cur:
+                cur.execute("SELECT sha1, minhash FROM agent_text_fingerprints")
+                self._cache = [(r[0], bytes(r[1])) for r in cur.fetchall()]
+        return self._cache
+
+    def __call__(self, pdf_bytes: bytes, sha1: str, filename: str = "") -> Optional[str]:
+        if not self.enabled:
+            return None
+        fp = fingerprint(self.text_of(pdf_bytes))
+        if fp is None:
+            return None
+        mh, n = fp
+        with self._lock:
+            conn = self.connect()
+            try:
+                best, best_sha = 0.0, None
+                for other_sha, other in self._load(conn):
+                    if other_sha == sha1:
+                        continue              # itself (a kept PDF coming back)
+                    s = similarity(mh, other)
+                    if s > best:
+                        best, best_sha = s, other_sha
+                if best_sha is not None and best >= self.threshold:
+                    log.info("near-duplicate: %s ~ %s (%.2f)", sha1[:12], best_sha[:12], best)
+                    return best_sha
+                with conn.cursor() as cur:
+                    cur.execute(
+                        "INSERT INTO agent_text_fingerprints (sha1, filename, minhash, n_shingles) "
+                        "VALUES (%s, %s, %s, %s) ON CONFLICT (sha1) DO NOTHING",
+                        (sha1, filename or None, mh, n))
+                conn.commit()
+                if self._cache is not None and not any(s == sha1 for s, _ in self._cache):
+                    self._cache.append((sha1, mh))
+                self.claimed[sha1] = filename
+                return None
+            finally:
+                try:
+                    conn.close()
+                except Exception:
+                    pass
+
+    def release(self, sha1: str) -> None:
+        """Drop a claim whose PDF failed for good (so a variant can stand in)."""
+        if not sha1:
+            return
+        with self._lock:
+            conn = self.connect()
+            try:
+                with conn.cursor() as cur:
+                    cur.execute("DELETE FROM agent_text_fingerprints WHERE sha1 = %s", (sha1,))
+                conn.commit()
+                if self._cache is not None:
+                    self._cache = [(s, m) for s, m in self._cache if s != sha1]
+                self.claimed.pop(sha1, None)
+            finally:
+                try:
+                    conn.close()
+                except Exception:
+                    pass
+
+
+    def release_file(self, filename: str) -> None:
+        for sha, fn in list(self.claimed.items()):
+            if fn == filename:
+                self.release(sha)
+
+
+def _pdf_text(pdf_bytes: bytes) -> str:
+    import extraction
+    try:
+        return "\n".join(extraction.pdf_page_texts(pdf_bytes))
+    except Exception:
+        return ""

batch_ingest.py

@@ -157,6 +157,16 @@ class PdfResult:
     # call was made; tokens_out includes thinking tokens, billed as output)
     tokens_in: int = 0
     tokens_out: int = 0
+    # Breakdown for cost analysis (all included in tokens_in / tokens_out):
+    # thinking tokens of every call, and the vision calls' share.
+    tokens_thinking: int = 0
+    vision_tokens_in: int = 0
+    vision_tokens_out: int = 0
+    text_model: str = ""          # model of the text call ("" = no call made)
+    vision_model: str = ""        # model of the vision calls ("" = none made)
+    # near-duplicate gate: sha1 of the already-ingested document this one
+    # repeats (then nothing was sent to Gemini)
+    duplicate_of: Optional[str] = None
     # figure-linking phase (all zero when --link-figures is off)
     rows_linked: int = 0          # text rows that got a figure link (new or backfilled)
     link_figures_found: int = 0   # figures harvested for the link pass (local, no API call)
@@ -526,6 +536,11 @@ def _run_figure_stage(
         result.vision_calls = stage.vision.total
         result.tokens_in += getattr(stage.vision, "tokens_in", 0)
         result.tokens_out += getattr(stage.vision, "tokens_out", 0)
+        result.tokens_thinking += getattr(stage.vision, "tokens_thinking", 0)
+        result.vision_tokens_in += getattr(stage.vision, "tokens_in", 0)
+        result.vision_tokens_out += getattr(stage.vision, "tokens_out", 0)
+        if stage.vision.total:
+            result.vision_model = getattr(stage.vision, "model", "") or extraction.vision_model()
         result.figure_filters = {
             k: v for k, v in dataclasses.asdict(stage.harvest).items() if v
         }
@@ -650,10 +665,15 @@ def process_pdf(
     db: Any = None,
     figure_opts: Optional[FigureOptions] = None,
     link_opts: Optional[LinkOptions] = None,
+    *,
+    pre_extracted: Optional[Extraction] = None,
 ) -> PdfResult:
     # `db` = backend module providing seen_sha1 / record_source /
     # already_inserted / insert_row. Defaults to this module (SQLite);
     # main() passes pg_mirror for --pg. Same logic either way.
+    #
+    # pre_extracted: the text call's result obtained elsewhere (the Gemini
+    # Batch API); no text call is made here then.
     db = db or sys.modules[__name__]
     started = time.time()
     pdf_bytes = pdf_path.read_bytes()
@@ -684,7 +704,8 @@ def process_pdf(
         return result
 
     try:
-        extracted = extract_from_pdf(pdf_bytes, pdf_path.name, api_key)
+        extracted = (pre_extracted if pre_extracted is not None
+                     else extract_from_pdf(pdf_bytes, pdf_path.name, api_key))
     except requests.RequestException as exc:
         # requests quotes the keyed URL in HTTPError/ConnectionError messages.
         err = _empty_result(pdf_path, started,
@@ -693,6 +714,8 @@ def process_pdf(
         return err
     tokens_in = int(getattr(extracted, "tokens_in", 0) or 0)
     tokens_out = int(getattr(extracted, "tokens_out", 0) or 0)
+    tokens_thinking = int(getattr(extracted, "tokens_thinking", 0) or 0)
+    text_model_used = (getattr(extracted, "model", "") or "") if (tokens_in or tokens_out) else ""
 
     if extracted.doc_status == "scanned_no_text":
         # Don't fabricate rows from an image-only PDF (Task 10). Whole-page
@@ -704,6 +727,7 @@ def process_pdf(
         # Billed even though nothing came back: keep the usage on the result.
         empty = _empty_result(pdf_path, started, "empty_extraction")
         empty.tokens_in, empty.tokens_out = tokens_in, tokens_out
+        empty.tokens_thinking, empty.text_model = tokens_thinking, text_model_used
         return empty
 
     # Ground every value against the PDF text (Task 1).
@@ -751,6 +775,8 @@ def process_pdf(
         material_classes=sorted(set(classes)),
         tokens_in=tokens_in,
         tokens_out=tokens_out,
+        tokens_thinking=tokens_thinking,
+        text_model=text_model_used,
     )
     for k, v in link_info.items():
         setattr(result, k, v)
@@ -895,6 +921,9 @@ def summarize(results: list[PdfResult]) -> dict[str, Any]:
         "tokens": {
             "tokens_in": sum(r.tokens_in for r in results),
             "tokens_out": sum(r.tokens_out for r in results),
+            "tokens_thinking": sum(r.tokens_thinking for r in results),
+            "vision_tokens_in": sum(r.vision_tokens_in for r in results),
+            "vision_tokens_out": sum(r.vision_tokens_out for r in results),
         },
         "figures": {
             "figures_found": sum(r.figures_found for r in results),

extraction.py

@@ -64,8 +64,24 @@ log = logging.getLogger("extraction")
 # extraction call. Previews get retired with ~2 weeks notice; stable models
 # don't. Overridable via env so a future swap needs no code change.
 GEMINI_MODEL = os.environ.get("GEMINI_MODEL", "gemini-3.5-flash")
+# Model for the figure (vision) calls: classify + mining. Empty = same as
+# GEMINI_MODEL. Classification is an easy task (one call per PDF picking a
+# figure kind), so a cheaper model here costs little quality. The agent sets
+# both from Run Control each cycle; read them at call time via text_model() /
+# vision_model(), never through a name imported at module load.
+GEMINI_VISION_MODEL = os.environ.get("GEMINI_VISION_MODEL", "").strip()
 PROMPT_VERSION = "2.0"
 
+
+def text_model() -> str:
+    """The model the text extraction call uses right now."""
+    return (GEMINI_MODEL or "gemini-3.5-flash").strip()
+
+
+def vision_model() -> str:
+    """The model the figure classify / mining calls use right now."""
+    return (GEMINI_VISION_MODEL or "").strip() or text_model()
+
 # How much the model may "think" per call. Thinking tokens are billed as
 # output (8x the input price): on 28 Sep 2026 a 25-paper cycle cost $2.02,
 # $1.86 of it output tokens, ~30k per PDF where the extraction JSON itself
@@ -248,13 +264,31 @@ class Material:
 @dataclasses.dataclass
 class Extraction:
     materials: list[Material] = dataclasses.field(default_factory=list)
-    model: str = GEMINI_MODEL
+    model: str = dataclasses.field(default_factory=text_model)
     prompt_version: str = PROMPT_VERSION
     doc_status: str = "ok"  # "ok" | "scanned_no_text" | "empty_extraction"
     # Gemini usage for the text call (0 when no call was made). tokens_out
-    # includes the model's thinking tokens, which are billed as output.
+    # includes the model's thinking tokens, which are billed as output;
+    # tokens_thinking is that thinking share on its own (for cost analysis).
     tokens_in: int = 0
     tokens_out: int = 0
+    tokens_thinking: int = 0
+
+
+def usage_detail(resp: Any) -> tuple[int, int, int]:
+    """(tokens_in, tokens_out, tokens_thinking) from a generateContent
+    response's ``usageMetadata``. tokens_out is billed output (candidates +
+    thinking); tokens_thinking is the thinking part alone. (0, 0, 0) when
+    absent or unparseable; never raises."""
+    try:
+        data = resp.json() if hasattr(resp, "json") else (resp or {})
+        u = (data or {}).get("usageMetadata") or {}
+        tin = int(u.get("promptTokenCount") or 0)
+        think = int(u.get("thoughtsTokenCount") or 0)
+        tout = int(u.get("candidatesTokenCount") or 0) + think
+        return max(0, tin), max(0, tout), max(0, think)
+    except Exception:
+        return 0, 0, 0
 
 
 def usage_from_response(resp: Any) -> tuple[int, int]:
@@ -318,7 +352,7 @@ class PropertyRow:
     status: str = "ok"
     flag_reason: str = ""
     # bookkeeping
-    model: str = GEMINI_MODEL
+    model: str = dataclasses.field(default_factory=text_model)
     prompt_version: str = PROMPT_VERSION
     # provenance kind (figure-mining phase): 'text' (grounded in the PDF
     # text) or 'figure' (read off a plot/table image — an estimate; see
@@ -609,8 +643,8 @@ def _upload_pdf_file(
         uri = info.get("uri", uri)
 
 
-def _parse_extraction_json(resp: requests.Response) -> Optional[dict[str, Any]]:
-    data = resp.json()
+def _parse_extraction_json(resp: Any) -> Optional[dict[str, Any]]:
+    data = resp.json() if hasattr(resp, "json") else (resp or {})
     candidates = data.get("candidates", [])
     if not candidates:
         return None
@@ -697,18 +731,28 @@ def extract_from_pdf(pdf_bytes: bytes, filename: str, api_key: str) -> Extractio
             "responseSchema": EXTRACTION_SCHEMA,
         },
     }
-    url = GEMINI_URL_TEMPLATE.format(model=GEMINI_MODEL, key=api_key)
+    model = text_model()
+    url = GEMINI_URL_TEMPLATE.format(model=model, key=api_key)
     resp = gemini_request(url, payload)
     if resp is None:
-        return Extraction(doc_status="empty_extraction")
-    tokens_in, tokens_out = usage_from_response(resp)
+        return Extraction(doc_status="empty_extraction", model=model)
+    return extraction_from_response(resp, model)
+
+
+def extraction_from_response(resp: Any, model: Optional[str] = None) -> Extraction:
+    """Extraction (with usage) from a generateContent response object or its
+    parsed JSON dict (the Batch API hands back dicts)."""
+    model = model or text_model()
+    tokens_in, tokens_out, tokens_thinking = usage_detail(resp)
     raw = _parse_extraction_json(resp)
     if not raw:
         # The call was made and billed even though nothing usable came back.
-        return Extraction(doc_status="empty_extraction",
-                          tokens_in=tokens_in, tokens_out=tokens_out)
+        return Extraction(doc_status="empty_extraction", model=model,
+                          tokens_in=tokens_in, tokens_out=tokens_out,
+                          tokens_thinking=tokens_thinking)
     out = _extraction_from_json(raw)
-    out.tokens_in, out.tokens_out = tokens_in, tokens_out
+    out.model = model
+    out.tokens_in, out.tokens_out, out.tokens_thinking = tokens_in, tokens_out, tokens_thinking
     return out
 
 

figures.py

@@ -152,7 +152,7 @@ class Figure:
     material_key: str = ""
     mining_status: str = ""     # "" | not_mined | skipped_kind | mined | mining_failed | classify_failed
     n_values: int = 0
-    model: str = GEMINI_MODEL
+    model: str = dataclasses.field(default_factory=extraction.vision_model)
     figure_prompt_version: str = FIGURE_PROMPT_VERSION
     # not persisted: raw PNG bytes (on-disk artifact) + the JPEG copy sent to Gemini
     png_bytes: bytes = dataclasses.field(default=b"", repr=False, compare=False)
@@ -200,6 +200,8 @@ class VisionStats:
     incomplete_calls: int = 0     # classify responses that skipped some indices
     tokens_in: int = 0            # Gemini usageMetadata over all vision calls
     tokens_out: int = 0           # (candidates + thinking tokens)
+    tokens_thinking: int = 0      # the thinking share of tokens_out
+    model: str = ""               # the model the calls went to (for pricing)
 
     @property
     def total(self) -> int:
@@ -212,14 +214,17 @@ class VisionStats:
         self.incomplete_calls += other.incomplete_calls
         self.tokens_in += other.tokens_in
         self.tokens_out += other.tokens_out
+        self.tokens_thinking += other.tokens_thinking
+        self.model = self.model or other.model
 
     def record_usage(self, resp: Any) -> None:
         """Add a response's usageMetadata (no-op for None / unparseable)."""
         if resp is None:
             return
-        tin, tout = extraction.usage_from_response(resp)
+        tin, tout, think = extraction.usage_detail(resp)
         self.tokens_in += tin
         self.tokens_out += tout
+        self.tokens_thinking += think
 
 
 @dataclasses.dataclass
@@ -724,7 +729,7 @@ def classify_figures(
     text_materials: list[str],
     api_key: str,
     *,
-    model: str = GEMINI_MODEL,
+    model: Optional[str] = None,
     stats: Optional[VisionStats] = None,
 ) -> list[Figure]:
     """One Gemini vision call classifies all of a PDF's figures.
@@ -734,6 +739,7 @@ def classify_figures(
     and ``figure_kind='other'`` (nothing raises; the caller counts it).
     """
     vs = stats if stats is not None else VisionStats()
+    model = model or extraction.vision_model()
     if not figures:
         return figures
     for batch in _batches_by_size(figures, CLASSIFY_MAX_BYTES):
@@ -883,7 +889,7 @@ def mine_figure(
     text_materials: list[str],
     api_key: str,
     *,
-    model: str = GEMINI_MODEL,
+    model: Optional[str] = None,
     stats: Optional[VisionStats] = None,
 ) -> Optional[MinedFigure]:
     """One vision call: structured readout of a property_plot / table_image.
@@ -892,6 +898,7 @@ def mine_figure(
     sets ``'mined'`` and ``fig.n_values`` on success.
     """
     vs = stats if stats is not None else VisionStats()
+    model = model or extraction.vision_model()
     mats = ", ".join(f"'{m}'" for m in dict.fromkeys(text_materials) if m) or "(none)"
     prompt = MINING_PROMPT.format(
         caption=fig.caption or "(no caption)", materials=mats,
@@ -1093,7 +1100,7 @@ def figure_properties_to_rows(
             mat.properties.append(prop)
             fig_of_prop[id(prop)] = fig
 
-    ext = Extraction(materials=list(buckets.values()), model=GEMINI_MODEL,
+    ext = Extraction(materials=list(buckets.values()), model=extraction.vision_model(),
                      prompt_version=FIGURE_PROMPT_VERSION)
     rows = to_rows(ext, source_pdf, source_sha1)
     # to_rows iterates materials x properties in order; walk the same order to
@@ -1104,7 +1111,7 @@ def figure_properties_to_rows(
         fig = fig_of_prop[id(prop)]
         row.origin = "figure"
         row.figure_id = fig.figure_id
-        row.model = fig.model or GEMINI_MODEL
+        row.model = fig.model or extraction.vision_model()
         row.prompt_version = fig.figure_prompt_version
         row.source_quote = fig.caption or f"[{fig.label}, no caption]"
         row.page = fig.page
@@ -1137,7 +1144,7 @@ def run_figure_stage(
     out_dir: Path | str = DEFAULT_FIGURES_DIR,
     max_figures: int = DEFAULT_MAX_FIGURES,
     mine: bool = True,
-    model: str = GEMINI_MODEL,
+    model: Optional[str] = None,
     done_figure_ids: Optional[set[str]] = None,
 ) -> FigureStageResult:
     """harvest -> classify -> mine -> rows. Never raises; errors are recorded
@@ -1152,6 +1159,8 @@ def run_figure_stage(
     to overwrite their stored status.
     """
     hs, vs = HarvestStats(), VisionStats()
+    model = model or extraction.vision_model()
+    vs.model = model
     res = FigureStageResult(figures=[], rows=[], harvest=hs, vision=vs)
     done = done_figure_ids or set()
     try:

page_files/Logs.py

@@ -134,7 +134,8 @@ def _history_block() -> pd.DataFrame:
         "candidates, relevance_rejected, skipped_doi, url_seen_skipped, "
         "download_failed, queued, downloaded, pdfs_ingested, rows_inserted, "
         "rows_flagged, duplicates, errors, figures_found, figures_mined, "
-        "figure_rows, vision_calls, rows_linked, tokens_in, tokens_out, "
+        "figure_rows, vision_calls, rows_linked, duplicate_text, tokens_in, tokens_out, "
+        "tokens_thinking, vision_tokens_in, vision_tokens_out, text_model, cost_usd, "
         "heartbeat_at, report "
         "FROM agent_runs ORDER BY id DESC LIMIT 50")
 
@@ -189,13 +190,31 @@ def _history_block() -> pd.DataFrame:
                       "skipped_doi", "url_seen_skipped", "download_failed", "queued", "downloaded",
                       "pdfs_ingested", "rows_inserted", "rows_flagged", "duplicates",
                       "errors", "figures_found", "figures_mined", "figure_rows",
-                      "vision_calls", "rows_linked", "tokens_in", "tokens_out"):
+                      "vision_calls", "rows_linked", "duplicate_text", "tokens_in", "tokens_out",
+                      "tokens_thinking"):
                 if c in disp.columns:
                     disp[c] = _to_int64(disp[c])
-            # Dollars are derived from the stored tokens at the configured price.
+            # Cost: stored per run since 30 Sep 2026 (each model at its own
+            # price); older runs re-priced from their tokens at the model they
+            # ran on (gemini-3.5-flash) — the old derived column used Gemini
+            # 2.5 Flash prices and showed about a quarter of the real bill.
             disp["cost_usd"] = [
-                C.cost_usd(i, o) if (pd.notna(i) and pd.notna(o)) else None
-                for i, o in zip(disp["tokens_in"], disp["tokens_out"])]
+                (float(c) if pd.notna(c) else
+                 (C.cost_usd(i, o, C.LEGACY_RUN_MODEL) if (pd.notna(i) and pd.notna(o)) else None))
+                for c, i, o in zip(disp["cost_usd"], disp["tokens_in"], disp["tokens_out"])]
+            disp["cost_per_pdf"] = [
+                (c / n) if (c is not None and pd.notna(n) and int(n) > 0) else None
+                for c, n in zip(disp["cost_usd"], disp["pdfs_ingested"])]
+            disp["text_model"] = [
+                _s(mm) or (C.LEGACY_RUN_MODEL if (pd.notna(i) and int(i or 0) > 0) else "")
+                for mm, i in zip(disp["text_model"], disp["tokens_in"])]
+            disp = disp.drop(columns=["vision_tokens_in", "vision_tokens_out"])
+            _spent = sum(c for c in disp["cost_usd"] if c is not None)
+            _pdfs = int(pd.to_numeric(disp["pdfs_ingested"], errors="coerce").fillna(0).sum())
+            if _spent > 0:
+                st.caption(f"Gemini spend over these cycles: **${_spent:,.2f}** for {_pdfs:,} "
+                           f"PDFs ingested" + (f" — **${_spent / _pdfs:.3f} per PDF**" if _pdfs else "")
+                           + ". Near-duplicates skipped before extraction cost nothing.")
             disp["trigger"] = disp["trigger"].map(_s)
             disp["status"] = disp["status"].map(_s)
 
@@ -264,10 +283,22 @@ def _history_block() -> pd.DataFrame:
                     "tokens_out": st.column_config.NumberColumn(
                         "Tokens out", format="localized",
                         help="Gemini output tokens incl. thinking (text + vision)"),
+                    "tokens_thinking": st.column_config.NumberColumn(
+                        "Thinking", format="localized",
+                        help="The thinking share of Tokens out (billed as output)"),
+                    "duplicate_text": st.column_config.NumberColumn(
+                        "Near-dup", format="localized",
+                        help="PDFs skipped before extraction: their text repeats a document "
+                             "already ingested"),
+                    "text_model": st.column_config.TextColumn(
+                        "Model", width="small", help="Gemini model of the text extraction calls"),
                     "cost_usd": st.column_config.NumberColumn(
                         "Cost", format="$%.4f",
-                        help=f"Derived: tokens x list price (${C.PRICE_IN_PER_M:g} / "
-                             f"${C.PRICE_OUT_PER_M:g} per M in/out; AGENT_PRICE_*_PER_M)"),
+                        help="Gemini list price per model (text and vision calls each at their "
+                             "own model; batch calls at half price). Runs before 30 Sep are "
+                             "re-priced at gemini-3.5-flash ($1.50 / $9.00 per M in/out)."),
+                    "cost_per_pdf": st.column_config.NumberColumn(
+                        "$/PDF", format="$%.3f", help="Cost / PDFs ingested"),
                 },
                 hide_index=True, width='stretch',
                 height=min(35 * 11 + 3, 35 * (len(disp) + 1) + 3))

page_files/Run_Control.py

@@ -272,9 +272,25 @@ _THINK_OPTS = ["low", "minimal", "off", "dynamic", "high"]
 _think_now = str(cfg.get("gemini_thinking", "low") or "dynamic").lower()
 if _think_now not in _THINK_OPTS:
     _THINK_OPTS.append(_think_now)              # an explicit token budget set by env
+_MODEL_OPTS = list(C.MODEL_PRICES)
+_model_now = str(cfg.get("gemini_model") or "gemini-3.5-flash").strip()
+if _model_now not in _MODEL_OPTS:
+    _MODEL_OPTS.insert(0, _model_now)
+_VMODEL_OPTS = ["(same as text model)"] + list(C.MODEL_PRICES)
+_vmodel_now = str(cfg.get("gemini_vision_model") or "").strip() or "(same as text model)"
+if _vmodel_now not in _VMODEL_OPTS:
+    _VMODEL_OPTS.append(_vmodel_now)
+
+
+def _price_label(m: str) -> str:
+    pin, pout = C.model_price(m)
+    return f"{m}  (${pin:g} / ${pout:g} per M)"
+
+
 with st.container(border=True):
-    T.card_title("Gemini cost", sub="thinking tokens are billed as output — 8× the input price")
-    k1, k2 = st.columns([1, 2], vertical_alignment="bottom")
+    T.card_title("Gemini cost", sub="thinking tokens are billed as output (6× the input price "
+                                    "on gemini-3.5-flash)")
+    k1, k2, k3 = st.columns(3, vertical_alignment="bottom")
     with k1:
         gemini_thinking = st.selectbox("Thinking per call", _THINK_OPTS,
                                        index=_THINK_OPTS.index(_think_now),
@@ -287,9 +303,31 @@ with st.container(border=True):
                                             "If the model rejects the option the call is repeated "
                                             "without it.")
     with k2:
-        st.caption("Rule of thumb from run #543 (25 papers, dynamic): $2.02, of which $1.86 "
-                   "output tokens. Watch the Cost column in Runs & Logs after a change; "
-                   "rows per paper and the flagged share tell you if recall moved.")
+        gemini_model = st.selectbox("Text extraction model", _MODEL_OPTS,
+                                    index=_MODEL_OPTS.index(_model_now),
+                                    format_func=_price_label,
+                                    disabled=not admin, key="rc_model",
+                                    help="Every row is stamped with this model. Change it only "
+                                         "after comparing on the gold set (cost_ab.py) — the "
+                                         "paper's numbers are per model.")
+    with k3:
+        gemini_vision_model = st.selectbox("Figure (vision) model", _VMODEL_OPTS,
+                                           index=_VMODEL_OPTS.index(_vmodel_now),
+                                           format_func=lambda m: m if m.startswith("(")
+                                           else _price_label(m),
+                                           disabled=not admin, key="rc_vmodel",
+                                           help="The one classify call per PDF (and mining, if "
+                                                "on). An easy task: the Lite model is ~5x cheaper.")
+    dedupe_sim = st.slider("Skip near-duplicate PDFs (text similarity ≥)", 0.0, 1.0,
+                           float(cfg.get("dedupe_text_similarity", 0.9) or 0.0), 0.01,
+                           disabled=not admin, key="rc_dedupe",
+                           help="Before any Gemini call, a PDF whose text is this similar to one "
+                                "already ingested (regional datasheet variants, preprint vs "
+                                "published version) is skipped. 0 = off.")
+    st.caption("Real cost of the 28–29 Sep cycles on gemini-3.5-flash with dynamic thinking and "
+               "mining on: ~$0.41 per PDF (36k in / 40k out tokens). Watch Cost and $/PDF in "
+               "Runs & Logs after a change; rows per paper and the flagged share tell you if "
+               "recall moved.")
 
 # --- figure mining ----------------------------------------------------------
 with st.container(border=True):
@@ -349,6 +387,10 @@ with st.container(horizontal=True, vertical_alignment="center", gap="medium"):
                 "discovery_per_query": int(disc_per_query),
                 "figures_enabled": figures_on, "figure_mining": figure_mine,
                 "gemini_thinking": str(gemini_thinking),
+                "gemini_model": str(gemini_model),
+                "gemini_vision_model": ("" if str(gemini_vision_model).startswith("(")
+                                        else str(gemini_vision_model)),
+                "dedupe_text_similarity": float(dedupe_sim),
                 "max_figures_per_pdf": int(max_figs),
                 "link_figures": link_on,
             })

test_cost_e2e.py

@@ -0,0 +1,326 @@
+"""Cost follow-up "f" (30 Sep 2026), scratch Postgres, no network:
+
+* per-model pricing — the old cost column priced a gemini-3.5-flash agent
+  at Gemini 2.5 Flash rates (~1/4 of the bill); runs now store cost_usd
+  priced per model, text and vision separately;
+* thinking / vision token breakdown on PdfResult and agent_runs;
+* the text and vision models follow Run Control at call time;
+* the near-duplicate gate skips a PDF whose text repeats an ingested one
+  before any Gemini call, and releases its claim when a PDF fails for good.
+
+  python test_cost_e2e.py      (same DB_* env as test_e2e.py)
+"""
+import hashlib
+import json
+import os
+import sys
+
+os.environ.setdefault("AGENT_WORK_DIR", "/tmp/agent_cost_test")
+os.environ["GEMINI_API_KEY"] = "test-key-not-used"
+os.environ["AGENT_USE_NTRS"] = "0"
+os.environ["AGENT_USE_S2"] = "0"
+os.environ["AGENT_USE_OPENALEX_TOPICS"] = "0"
+os.environ["AGENT_USE_DATASHEETS"] = "0"
+os.environ["AGENT_SEED_QUERY_GRID"] = "0"
+os.environ.pop("AGENT_PRICE_IN_PER_M", None)
+os.environ.pop("AGENT_PRICE_OUT_PER_M", None)
+
+import fitz  # noqa: E402
+
+import batch_ingest  # noqa: E402
+import extraction  # noqa: E402
+import figures  # noqa: E402
+import pdf_crawler  # noqa: E402
+from agent import agdb, config as C, orchestrator, textdedupe  # noqa: E402
+
+C.GEMINI_API_KEY = "test-key-not-used"
+fails = []
+
+
+def check(name, cond):
+    print(("PASS " if cond else "FAIL ") + name)
+    if not cond:
+        fails.append(name)
+
+
+def close(a, b, tol=1e-6):
+    return abs(float(a) - float(b)) <= tol
+
+
+class FakeResp:
+    def __init__(self, payload):
+        self._p = payload
+        self.status_code = 200
+        self.headers = {}
+
+    def json(self):
+        return self._p
+
+
+# --- pricing ------------------------------------------------------------------
+check("price: gemini-3.5-flash is $1.50 / $9.00", C.model_price("gemini-3.5-flash") == (1.50, 9.00))
+check("price: gemini-3.5-flash-lite is $0.30 / $2.50",
+      C.model_price("gemini-3.5-flash-lite") == (0.30, 2.50))
+check("price: 'models/' prefix and dated suffix resolve",
+      C.model_price("models/gemini-3.5-flash-001") == (1.50, 9.00)
+      and C.model_price("gemini-3.5-flash-lite-preview-06") == (0.30, 2.50))
+check("price: unknown model falls back to the legacy run model",
+      C.model_price("some-future-model") == C.model_price(C.LEGACY_RUN_MODEL))
+# runs 544-593 (29 Sep): 4.25M in / 4.68M out -> $48.5 real, not the $12.98 shown
+check("cost: runs 544-593 re-priced at gemini-3.5-flash = $48.50",
+      close(C.cost_usd(4_250_000, 4_680_000), 48.495))
+check("cost: batch halves the price", close(C.cost_usd(1e6, 1e6, "gemini-3.5-flash", batch=True), 5.25))
+rc = C.run_cost_usd({"tokens_in": 1_000_000, "tokens_out": 200_000,
+                     "vision_tokens_in": 400_000, "vision_tokens_out": 50_000,
+                     "text_model": "gemini-3.5-flash", "vision_model": "gemini-3.5-flash-lite"})
+# text: 600k in, 150k out at 3.5 flash = 0.9 + 1.35; vision 400k/50k at lite = 0.12 + 0.125
+check("run cost: text and vision priced at their own models", close(rc, 0.9 + 1.35 + 0.12 + 0.125))
+
+# --- usage breakdown ------------------------------------------------------------
+resp = {"candidates": [{"content": {"parts": [{"text": "{}"}]}}],
+        "usageMetadata": {"promptTokenCount": 9000, "candidatesTokenCount": 3000,
+                          "thoughtsTokenCount": 25000}}
+check("usage_detail: (in, out incl. thinking, thinking)",
+      extraction.usage_detail(FakeResp(resp)) == (9000, 28000, 25000))
+check("usage_detail: works on a plain dict (Batch API responses)",
+      extraction.usage_detail(resp) == (9000, 28000, 25000))
+check("usage_detail: garbage -> zeros", extraction.usage_detail(None) == (0, 0, 0))
+
+# --- models follow the runtime setting --------------------------------------------
+doc = fitz.open()
+page = doc.new_page()
+page.insert_text((72, 100), "The tensile modulus of PEEK-CF30 was measured as 24.5 GPa. " * 3)
+TINY = doc.tobytes()
+doc.close()
+GOOD = json.dumps({"materials": [{
+    "material_name": "PEEK CF30", "material_abbreviation": "PEEK-CF30",
+    "material_class": "Composite", "matrix": "PEEK", "fiber": "Carbon",
+    "properties": [{"section": "Mechanical", "property_name": "Tensile Modulus",
+                    "value_raw": "24.5", "value_num": 24.5, "unit": "GPa",
+                    "source_quote": "measured as 24.5 GPa", "page": 1}]}]})
+URLS = []
+_orig_req = extraction.gemini_request
+
+
+def fake_req(url, payload, **kw):
+    URLS.append(url)
+    return FakeResp({"candidates": [{"content": {"parts": [{"text": GOOD}]}}],
+                     "usageMetadata": {"promptTokenCount": 100, "candidatesTokenCount": 10,
+                                       "thoughtsTokenCount": 40}})
+
+
+try:
+    extraction.gemini_request = fake_req
+    figures.gemini_request = fake_req
+    extraction.GEMINI_MODEL, extraction.GEMINI_VISION_MODEL = "gemini-3.5-flash-lite", ""
+    ex = extraction.extract_from_pdf(TINY, "t.pdf", "k")
+    check("text call goes to the runtime model", "models/gemini-3.5-flash-lite:" in URLS[-1])
+    check("Extraction stamped with the runtime model and thinking tokens",
+          ex.model == "gemini-3.5-flash-lite" and ex.tokens_thinking == 40 and ex.tokens_out == 50)
+    rows = extraction.to_rows(ex, "t.pdf", "abc")
+    check("rows carry the runtime model", rows and all(r.model == "gemini-3.5-flash-lite" for r in rows))
+    extraction.GEMINI_MODEL, extraction.GEMINI_VISION_MODEL = "gemini-3.5-flash", "gemini-3.5-flash-lite"
+    check("vision_model(): explicit vision model", extraction.vision_model() == "gemini-3.5-flash-lite")
+    vs = figures.VisionStats()
+    vs.record_usage(fake_req("x", {}))
+    check("VisionStats keeps thinking separately", (vs.tokens_out, vs.tokens_thinking) == (50, 40))
+    extraction.GEMINI_VISION_MODEL = ""
+    check("vision_model(): empty = the text model", extraction.vision_model() == "gemini-3.5-flash")
+finally:
+    extraction.gemini_request = _orig_req
+    figures.gemini_request = _orig_req
+    extraction.GEMINI_MODEL, extraction.GEMINI_VISION_MODEL = "gemini-3.5-flash", ""
+
+# figure classification URL: run classify on a stub figure list via the batch helper
+URLS.clear()
+try:
+    figures.gemini_request = fake_req
+    extraction.GEMINI_VISION_MODEL = "gemini-3.5-flash-lite"
+    f1 = figures.Figure(figure_id="f1", source_pdf="t.pdf", source_sha1="abc", page=1,
+                        bbox=(0, 0, 100, 100), caption="Fig. 1 Tensile modulus", image_path="",
+                        image_sha256="", width_px=200, height_px=200, route="raster",
+                        jpeg_bytes=b"\xff\xd8fakejpeg")
+    check("Figure.model defaults to the runtime vision model", f1.model == "gemini-3.5-flash-lite")
+    figures.classify_figures([f1], ["PEEK"], "k", stats=figures.VisionStats())
+    check("figure classify call goes to the vision model",
+          bool(URLS) and "models/gemini-3.5-flash-lite:" in URLS[-1])
+finally:
+    figures.gemini_request = _orig_req
+    extraction.GEMINI_VISION_MODEL = ""
+
+# --- MinHash fingerprints -----------------------------------------------------------
+BASE = (" ".join(f"Hexcel HexPly 8552 epoxy matrix product data sheet section {i}: "
+                 f"tensile strength {1500 + i} MPa at 23 C, compression modulus {60 + i} GPa, "
+                 f"interlaminar shear {90 + i} MPa after cure 180 C for 120 minutes."
+                 for i in range(25)))
+VARIANT = BASE.replace("section 0:", "section 0 (EU version, Hexcel Composites GmbH):") + \
+    " Copyright Hexcel Corporation. EU edition."
+OTHER = BASE.replace("8552", "8551")
+for n in range(25):  # a sibling datasheet: same template, different numbers
+    OTHER = OTHER.replace(f"{1500 + n} MPa", f"{1300 + 3 * n} MPa").replace(f"{60 + n} GPa", f"{50 + 2 * n} GPa")
+fa, fb, fc = textdedupe.fingerprint(BASE), textdedupe.fingerprint(VARIANT), textdedupe.fingerprint(OTHER)
+check("fingerprint: identical text -> similarity 1.0", textdedupe.similarity(fa[0], fa[0]) == 1.0)
+sab = textdedupe.similarity(fa[0], fb[0])
+sac = textdedupe.similarity(fa[0], fc[0])
+print(f"   regional variant ~ {sab:.2f}, sibling datasheet ~ {sac:.2f}")
+check("fingerprint: regional variant >= 0.9", sab >= 0.9)
+check("fingerprint: sibling datasheet (same template, other numbers) < 0.9", sac < 0.9)
+check("fingerprint: too little text -> None", textdedupe.fingerprint("tensile modulus 24 GPa") is None)
+
+
+# --- E2E cycle: models from config, breakdown + cost stored, duplicate skipped ------
+def make_pdf(text: str) -> bytes:
+    d = fitz.open()
+    words = text.split()
+    for start in range(0, len(words), 350):
+        p = d.new_page()
+        chunk = words[start:start + 350]
+        y = 60
+        for i in range(0, len(chunk), 12):
+            p.insert_text((40, y), " ".join(chunk[i:i + 12]), fontsize=8)
+            y += 11
+    data = d.tobytes()
+    d.close()
+    return data
+
+
+PDFS = {"base": make_pdf(BASE), "variant": make_pdf(VARIANT), "other": make_pdf(OTHER)}
+ORDER = ["base", "variant", "other"]
+EXTRACTED = []
+
+
+def fake_search_openalex(query, limit):
+    for k in ORDER:
+        yield pdf_crawler.Candidate(title=f"{k} hexply 8552 datasheet", pdf_url=f"https://example.org/{k}.pdf",
+                                    source="openalex", query=query, doi=f"10.9999/aim.cost.{k}", year="2026",
+                                    abstract="thermoplastic composite tensile modulus carbon fiber datasheet")
+
+
+def fake_download(cand, pdf_dir, state):
+    if cand.pdf_url in state.seen_urls:
+        return None
+    state.seen_urls.add(cand.pdf_url)
+    key = cand.pdf_url.rsplit("/", 1)[1][:-4]
+    data = PDFS[key]
+    sha = hashlib.sha256(data).hexdigest()
+    fname = f"openalex_{key}_{sha[:8]}.pdf"
+    (pdf_dir / fname).write_bytes(data)
+    return {"filename": fname, "title": cand.title, "doi": cand.doi, "url": cand.pdf_url,
+            "year": cand.year, "source": cand.source, "sha256": sha, "query": cand.query, "bytes": len(data)}
+
+
+FAIL_KEYS = set()
+
+
+def fake_extract(pdf_bytes, filename, api_key):
+    key = next(k for k in ORDER if f"_{k}_" in filename)
+    EXTRACTED.append(key)
+    if key in FAIL_KEYS:
+        return extraction.Extraction(doc_status="empty_extraction", tokens_in=500, tokens_out=100,
+                                     tokens_thinking=60)
+    val = "1500" if key != "other" else "1300"
+    mat = extraction.Material(
+        material_name=f"HexPly {key}", material_abbreviation=f"HP-{key}", material_class="Composite",
+        matrix="Epoxy", fiber="Carbon", fiber_volume_fraction="",
+        properties=[extraction.Property(section="Mechanical", property_name="Tensile Strength",
+                                        value_raw=val, value_num=float(val), unit="MPa",
+                                        test_condition="23 C", source_quote=f"tensile strength {val} MPa",
+                                        page=1)])
+    ext = extraction.Extraction(materials=[mat], doc_status="ok")
+    ext.tokens_in, ext.tokens_out, ext.tokens_thinking = 30000, 8000, 5000
+    return ext
+
+
+pdf_crawler.search_openalex = fake_search_openalex
+pdf_crawler.search_arxiv = lambda q, n: iter(())
+pdf_crawler.download_pdf = fake_download
+batch_ingest.extract_from_pdf = fake_extract
+
+
+def q(sql, params=()):
+    conn = agdb.connect()
+    try:
+        with conn.cursor() as cur:
+            cur.execute(sql, params) if params else cur.execute(sql)
+            try:
+                return cur.fetchall()
+            except Exception:
+                return None
+    finally:
+        conn.commit()
+        conn.close()
+
+
+orchestrator.bootstrap()
+cfg_now = agdb.get_config(agdb.connect())
+check("defaults: text model gemini-3.5-flash, vision = same, dedupe 0.9",
+      cfg_now["gemini_model"] == "gemini-3.5-flash" and cfg_now["gemini_vision_model"] == ""
+      and close(cfg_now["dedupe_text_similarity"], 0.9))
+conn = agdb.connect()
+agdb.set_config(conn, {"queries_per_cycle": 1, "max_rounds_per_cycle": 1, "max_new_pdfs_per_cycle": 5,
+                       "min_new_pdfs_per_cycle": 5, "figures_enabled": False, "link_figures": False,
+                       "expand_queries": False, "ingest_workers": 1, "use_candidate_queue": False,
+                       "gemini_model": "gemini-3.5-flash-lite",
+                       "gemini_vision_model": "gemini-3.5-flash-lite"})
+conn.close()
+
+orchestrator.run_cycle(trigger="test")
+check("node_plan applied the configured models",
+      extraction.GEMINI_MODEL == "gemini-3.5-flash-lite"
+      and extraction.GEMINI_VISION_MODEL == "gemini-3.5-flash-lite")
+row = q("SELECT status, downloaded, pdfs_ingested, duplicate_text, tokens_in, tokens_out, "
+        "tokens_thinking, text_model, cost_usd FROM agent_runs WHERE kind = 'cycle' "
+        "ORDER BY id DESC LIMIT 1")[0]
+print("   run:", row)
+check("cycle: 3 downloaded, 2 ingested, 1 near-duplicate skipped", row[:4] == ("ok", 3, 2, 1))
+check("the duplicate never reached Gemini", sorted(EXTRACTED) == ["base", "other"])
+check("tokens and thinking summed over the 2 extracted PDFs", row[4:7] == (60000, 16000, 10000))
+check("run stamped with the text model", row[7] == "gemini-3.5-flash-lite")
+check("cost stored, priced at the lite model", close(row[8], C.cost_usd(60000, 16000, "gemini-3.5-flash-lite")))
+reg = q("SELECT ingest_status FROM agent_doi_seen WHERE filename LIKE 'openalex_variant_%'")
+check("registry: the variant marked duplicate_text", reg == [("duplicate_text",)])
+check("the variant PDF was removed from disk", not list(C.PDF_DIR.glob("openalex_variant_*.pdf")))
+fps = q("SELECT count(*) FROM agent_text_fingerprints")[0][0]
+check("fingerprints: the 2 ingested documents are claimed", fps == 2)
+ev = q("SELECT count(*) FROM agent_events WHERE message LIKE '%skipped before extraction%'")[0][0]
+check("event explains the skip", ev == 1)
+done = q("SELECT message FROM agent_events WHERE message LIKE 'cycle done:%' ORDER BY id DESC LIMIT 1")[0][0]
+check("cycle-done event names thinking tokens and the model price",
+      "10000 of it thinking" in done and "gemini-3.5-flash-lite prices" in done
+      and "1 near-duplicate" in done)
+
+# A PDF that fails for good releases its claim, so a variant may stand in.
+EXTRACTED.clear()
+q("DELETE FROM agent_doi_seen"); q('DELETE FROM "Composites_materials"'); q("DELETE FROM sources")
+q("DELETE FROM agent_text_fingerprints")
+conn = agdb.connect()
+agdb.set_state(conn, "crawler_state", {"seen_urls": [], "seen_hashes": [], "failed": {}})
+agdb.set_config(conn, {"gemini_model": "gemini-3.5-flash", "gemini_vision_model": ""})
+conn.close()
+FAIL_KEYS.add("base")
+ORDER[:] = ["base", "other"]
+orchestrator.run_cycle(trigger="test")
+fps = q("SELECT count(*) FROM agent_text_fingerprints")[0][0]
+check("failed PDF's fingerprint released (only 'other' is claimed)", fps == 1)
+row = q("SELECT pdfs_ingested, tokens_in, text_model, cost_usd FROM agent_runs WHERE kind = 'cycle' "
+        "ORDER BY id DESC LIMIT 1")[0]
+check("billed-but-empty extraction still counted in cost, at 3.5 flash",
+      row[1] == 30500 and row[2] == "gemini-3.5-flash"
+      and close(row[3], C.cost_usd(30500, 8100, "gemini-3.5-flash")))
+FAIL_KEYS.clear()
+
+# Dedupe off -> nothing claimed, nothing skipped
+EXTRACTED.clear()
+q("DELETE FROM agent_doi_seen"); q('DELETE FROM "Composites_materials"'); q("DELETE FROM sources")
+q("DELETE FROM agent_text_fingerprints")
+conn = agdb.connect()
+agdb.set_state(conn, "crawler_state", {"seen_urls": [], "seen_hashes": [], "failed": {}})
+agdb.set_config(conn, {"dedupe_text_similarity": 0.0})
+conn.close()
+ORDER[:] = ["base", "variant"]
+orchestrator.run_cycle(trigger="test")
+check("dedupe off: both variants extracted", sorted(EXTRACTED) == ["base", "variant"])
+check("dedupe off: no fingerprints written", q("SELECT count(*) FROM agent_text_fingerprints")[0][0] == 0)
+
+print("\n%d checks failed" % len(fails))
+sys.exit(1 if fails else 0)

test_tokens.py

@@ -113,7 +113,7 @@ r = batch_ingest.PdfResult(pdf="x", elapsed_s=0, materials=0, extracted=0, inser
                            flagged=0, duplicates=0, material_classes=[])
 check("PdfResult has tokens_in/tokens_out defaulting to 0", (r.tokens_in, r.tokens_out) == (0, 0))
 check("summarize reports token totals",
-      batch_ingest.summarize([r])["tokens"] == {"tokens_in": 0, "tokens_out": 0})
+      {"tokens_in": 0, "tokens_out": 0}.items() <= batch_ingest.summarize([r])["tokens"].items())
 
 print("\n%d checks failed" % len(fails))
 sys.exit(1 if fails else 0)