source-archive: skip complete captures + per-run cost report

code_tests/unit_tests/test_agents_and_tools/test_source_archive/test_content_store.py

@@ -159,3 +159,36 @@ def test_different_content_not_aliased(tmp_path):
     b = store.store(_result("https://b.test/y", "<p>two different</p>"))
     assert b.capture.content_alias_of is None
     assert b.capture.html_key != a.capture.html_key
+
+
+def test_incomplete_capture_is_not_a_cache_hit(tmp_path):
+    # A browser capture whose screenshot failed to encode (screenshot_key=None)
+    # is not "done" — the next run should retry it to fill the missing format.
+    store = _store(tmp_path, ttl_days=14)
+    store.store(
+        CaptureResult(
+            url="https://a.test",
+            final_url="https://a.test",
+            status_code=200,
+            html="<p>one</p>",
+            markdown="md " * 50,
+            screenshot=None,
+            fetcher="cloakbrowser",
+        )
+    )
+    assert store.lookup("https://a.test") is None
+
+
+def test_pdf_capture_without_screenshot_is_complete(tmp_path):
+    # PDFs have no screenshot by nature, so markdown alone counts as complete.
+    store = _store(tmp_path, ttl_days=14)
+    store.store(
+        CaptureResult(
+            url="https://a.test/x.pdf",
+            final_url="https://a.test/x.pdf",
+            status_code=200,
+            markdown="md " * 50,
+            fetcher="pdf",
+        )
+    )
+    assert store.lookup("https://a.test/x.pdf") is not None

code_tests/unit_tests/test_agents_and_tools/test_source_archive/test_cost.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+from forecasting_tools.agents_and_tools.source_archive.config import ArchiveConfig
+from forecasting_tools.agents_and_tools.source_archive.cost import (
+    estimate_run_cost,
+    price_per_capture,
+)
+from forecasting_tools.agents_and_tools.source_archive.models import StoredCapture
+from forecasting_tools.agents_and_tools.source_archive.pipeline import (
+    CaptureOutcome,
+    PipelineSummary,
+)
+
+
+def _cap(url: str, fetcher: str) -> StoredCapture:
+    return StoredCapture(url=url, url_hash="h", content_hash="c", fetcher=fetcher)
+
+
+def _stored(url: str, fetcher: str) -> CaptureOutcome:
+    return CaptureOutcome(url=url, status="stored", stored=_cap(url, fetcher))
+
+
+def test_free_backends_cost_nothing():
+    cfg = ArchiveConfig()
+    for f in ("cloakbrowser", "playwright", "pdf", ""):
+        assert price_per_capture(f, cfg) == 0.0
+
+
+def test_paid_backends_priced_by_config():
+    cfg = ArchiveConfig(hyperbrowser_use_proxy=True, firecrawl_proxy="basic")
+    assert price_per_capture("hyperbrowser", cfg) == 10 * 0.001
+    assert price_per_capture("firecrawl", cfg) == 1 * 0.00083
+
+    cheap = ArchiveConfig(hyperbrowser_use_proxy=False, firecrawl_proxy="auto")
+    assert price_per_capture("hyperbrowser", cheap) == 1 * 0.001
+    assert price_per_capture("firecrawl", cheap) == 5 * 0.00083
+
+
+def test_estimate_run_cost_breakdown():
+    cfg = ArchiveConfig(hyperbrowser_use_proxy=True, firecrawl_proxy="basic")
+    summary = PipelineSummary(
+        outcomes=[
+            _stored("u1", "cloakbrowser"),
+            _stored("u2", "cloakbrowser"),
+            _stored("u3", "hyperbrowser"),
+            _stored("u4", "firecrawl"),
+            CaptureOutcome(
+                url="u5", status="cache_hit", stored=_cap("u5", "cloakbrowser")
+            ),
+            CaptureOutcome(url="u6", status="error", reason="boom"),
+        ]
+    )
+    rc = estimate_run_cost(summary, cfg, run_id="r1")
+
+    assert rc.archived == 5  # 4 stored + 1 cache_hit; the error doesn't count
+    assert rc.paid_captures == 2  # hyperbrowser + firecrawl
+    assert rc.total_usd == round(0.01 + 0.00083, 4)  # 0.0108 (4-dp rounding)
+    by = {b.backend: b for b in rc.by_backend}
+    assert by["cloakbrowser"].captures == 2 and by["cloakbrowser"].total_usd == 0.0
+    assert by["hyperbrowser"].captures == 1 and by["hyperbrowser"].unit_usd == 0.01

forecasting_tools/agents_and_tools/source_archive/cli.py

@@ -110,12 +110,19 @@ def _cmd_capture(args, config: ArchiveConfig) -> int:
     summary = capture_urls_concurrent(urls, store, config, build_default_fetcher)
     print(summary)
 
+    from forecasting_tools.agents_and_tools.source_archive import cost as cost_mod
+
+    run_cost = cost_mod.estimate_run_cost(summary, config, run_id=args.run_id)
+    print(run_cost)
+
     run_id = args.run_id or (records[0].run_id if records else None)
     if run_id:
         from forecasting_tools.agents_and_tools.source_archive import reports
 
         reports.write_run_report(store.blobs, run_id, summary, config)
         print(f"Wrote run outcomes -> {config.s3_prefix}/reports/{run_id}.json")
+        cost_mod.write_cost_report(store.blobs, run_id, run_cost, config)
+        print(f"Wrote cost report -> {config.s3_prefix}/reports/{run_id}_cost.json")
 
     # Failures leave no cache entry, so re-running retries exactly them. Write a
     # retry manifest (with provenance) so coming back — e.g. with hyperbrowser

forecasting_tools/agents_and_tools/source_archive/content_store.py

@@ -69,6 +69,20 @@ def _parse_iso(ts: str) -> datetime:
     return dt
 
 
+def _capture_is_complete(cap: dict) -> bool:
+    """Whether a stored capture has every format we expect for its type.
+
+    A browser capture is complete only with html + markdown + screenshot; a PDF
+    (which has no screenshot) only needs its markdown. Used by :meth:`lookup` so
+    an incomplete capture is re-fetched rather than treated as already done.
+    """
+    if (cap.get("fetcher") or "").lower() == "pdf":
+        return bool(cap.get("markdown_key"))
+    return bool(
+        cap.get("html_key") and cap.get("markdown_key") and cap.get("screenshot_key")
+    )
+
+
 class ContentStore:
     def __init__(self, blob_store: BlobStore, config: ArchiveConfig | None = None):
         self.blobs = blob_store
@@ -166,6 +180,12 @@ def lookup(self, url: str) -> StoredCapture | None:
         age = datetime.now(timezone.utc) - last_seen
         if age > timedelta(days=self.config.ttl_days):
             return None
+        # Skip only a COMPLETE capture. A partial one (e.g. a failed screenshot
+        # encode left screenshot_key=None) is treated as a miss so the next run
+        # retries the missing format instead of skipping it forever. PDFs have no
+        # screenshot by nature, so they only need their markdown.
+        if not _capture_is_complete(latest):
+            return None
         return StoredCapture.model_validate(latest)
 
     def store(self, result: CaptureResult) -> StoreResult:

forecasting_tools/agents_and_tools/source_archive/cost.py

@@ -0,0 +1,131 @@
+"""Estimate what a capture run cost — per backend and per archived site.
+
+Self-hosted browsers (CloakBrowser / Playwright) and local PDF parsing are ~free;
+the managed backends (Hyperbrowser, Firecrawl) bill per page. This turns a run's
+outcomes into a cost breakdown so an operator can see what the paid backends are
+costing per site archived.
+
+Costs are **estimates** from each vendor's public pricing applied to the
+configured proxy mode — we record the backend that produced each capture, not the
+live credit count — so treat them as close approximations, not billed amounts.
+Only *successful* captures are priced; a paid backend call that then failed the
+quality gate isn't attributed to a backend here (so this slightly under-counts).
+"""
+
+from __future__ import annotations
+
+import json
+from collections import Counter
+
+from pydantic import BaseModel
+
+from forecasting_tools.agents_and_tools.source_archive.config import ArchiveConfig
+
+# $ per vendor credit (public list pricing, 2026-06).
+_FIRECRAWL_CREDIT_USD = 0.00083
+_HYPERBROWSER_CREDIT_USD = 0.001
+
+# Backends that run on our own machine — no per-page charge.
+_FREE_BACKENDS = {"cloakbrowser", "playwright", "pdf", ""}
+
+
+def price_per_capture(fetcher: str, config: ArchiveConfig) -> float:
+    """Estimated $ for one successful capture by ``fetcher`` under ``config``."""
+    f = (fetcher or "").lower()
+    if f in _FREE_BACKENDS:
+        return 0.0
+    if f == "hyperbrowser":
+        credits = 10 if config.hyperbrowser_use_proxy else 1
+        return credits * _HYPERBROWSER_CREDIT_USD
+    if f.startswith("firecrawl"):
+        basic = (config.firecrawl_proxy or "basic").lower() in ("", "basic")
+        return (1 if basic else 5) * _FIRECRAWL_CREDIT_USD
+    return 0.0  # unknown backend — assume free rather than invent a number
+
+
+class BackendCost(BaseModel):
+    backend: str
+    captures: int
+    unit_usd: float
+    total_usd: float
+
+
+class RunCost(BaseModel):
+    run_id: str | None = None
+    archived: int = 0  # sites we now hold (stored + deduped + cache_hit)
+    paid_captures: int = 0  # captures via a paid backend this run
+    total_usd: float = 0.0
+    usd_per_archived: float = 0.0
+    by_backend: list[BackendCost] = []
+
+    def __str__(self) -> str:
+        lines = [
+            f"RunCost(run_id={self.run_id}, archived={self.archived}, "
+            f"paid_captures={self.paid_captures}, total=${self.total_usd:.4f}, "
+            f"$/archived=${self.usd_per_archived:.5f})",
+            f"  {'backend':<14}{'captures':>9}{'$/capture':>12}{'$ total':>10}",
+        ]
+        for b in self.by_backend:
+            lines.append(
+                f"  {b.backend:<14}{b.captures:>9}{b.unit_usd:>12.5f}{b.total_usd:>10.4f}"
+            )
+        return "\n".join(lines)
+
+
+def estimate_run_cost(
+    summary, config: ArchiveConfig, run_id: str | None = None
+) -> RunCost:
+    """Estimate a :class:`PipelineSummary`'s cost, broken down by backend.
+
+    Newly fetched captures (``stored`` / ``deduped``) are priced by the backend
+    that produced them; ``cache_hit`` re-uses cost nothing (no fetch happened).
+    """
+    counts: Counter[str] = Counter()
+    for o in summary.outcomes:
+        if o.status in ("stored", "deduped") and o.stored is not None:
+            counts[(o.stored.fetcher or "unknown")] += 1
+
+    by_backend: list[BackendCost] = []
+    total = 0.0
+    paid = 0
+    for backend, n in sorted(counts.items()):
+        unit = price_per_capture(backend, config)
+        sub = unit * n
+        total += sub
+        if unit > 0:
+            paid += n
+        by_backend.append(
+            BackendCost(
+                backend=backend,
+                captures=n,
+                unit_usd=round(unit, 6),
+                total_usd=round(sub, 4),
+            )
+        )
+
+    archived = sum(
+        1 for o in summary.outcomes if o.status in ("stored", "deduped", "cache_hit")
+    )
+    return RunCost(
+        run_id=run_id,
+        archived=archived,
+        paid_captures=paid,
+        total_usd=round(total, 4),
+        usd_per_archived=round(total / archived, 6) if archived else 0.0,
+        by_backend=by_backend,
+    )
+
+
+def cost_report_key(run_id: str, config: ArchiveConfig) -> str:
+    return f"{config.s3_prefix.rstrip('/')}/reports/{run_id}_cost.json"
+
+
+def write_cost_report(store, run_id: str, cost: RunCost, config: ArchiveConfig) -> str:
+    """Persist the cost breakdown next to the run report (``reports/<id>_cost.json``)."""
+    key = cost_report_key(run_id, config)
+    store.put(
+        key,
+        json.dumps(cost.model_dump(), indent=2).encode("utf-8"),
+        content_type="application/json",
+    )
+    return key