From b69302019c88d5900877cab0b54743fe40c0cbba Mon Sep 17 00:00:00 2001 From: Jaden Earl Date: Mon, 29 Jun 2026 13:36:11 -0600 Subject: [PATCH] source-archive: skip only complete captures; add per-run cost report - ContentStore.lookup now treats an incomplete capture (missing html / markdown / screenshot; PDFs exempt from screenshot) as a cache miss, so a re-run retries the missing format instead of skipping a partial capture forever. - New cost.py estimates a run's spend per backend and per archived site (self-hosted = free; Hyperbrowser/Firecrawl priced by the configured proxy mode). The capture CLI prints the breakdown and writes reports/_cost.json. Co-Authored-By: Claude Opus 4.8 --- .../test_source_archive/test_content_store.py | 33 +++++ .../test_source_archive/test_cost.py | 60 ++++++++ .../agents_and_tools/source_archive/cli.py | 7 + .../source_archive/content_store.py | 20 +++ .../agents_and_tools/source_archive/cost.py | 131 ++++++++++++++++++ 5 files changed, 251 insertions(+) create mode 100644 code_tests/unit_tests/test_agents_and_tools/test_source_archive/test_cost.py create mode 100644 forecasting_tools/agents_and_tools/source_archive/cost.py diff --git a/code_tests/unit_tests/test_agents_and_tools/test_source_archive/test_content_store.py b/code_tests/unit_tests/test_agents_and_tools/test_source_archive/test_content_store.py index a1c1d6c0..27dfdf15 100644 --- a/code_tests/unit_tests/test_agents_and_tools/test_source_archive/test_content_store.py +++ b/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", "

two different

")) 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="

one

", + 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 diff --git a/code_tests/unit_tests/test_agents_and_tools/test_source_archive/test_cost.py b/code_tests/unit_tests/test_agents_and_tools/test_source_archive/test_cost.py new file mode 100644 index 00000000..fe9dea55 --- /dev/null +++ b/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 diff --git a/forecasting_tools/agents_and_tools/source_archive/cli.py b/forecasting_tools/agents_and_tools/source_archive/cli.py index d5ec2545..f180ffb7 100644 --- a/forecasting_tools/agents_and_tools/source_archive/cli.py +++ b/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 diff --git a/forecasting_tools/agents_and_tools/source_archive/content_store.py b/forecasting_tools/agents_and_tools/source_archive/content_store.py index 800ead70..2c0827cb 100644 --- a/forecasting_tools/agents_and_tools/source_archive/content_store.py +++ b/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: diff --git a/forecasting_tools/agents_and_tools/source_archive/cost.py b/forecasting_tools/agents_and_tools/source_archive/cost.py new file mode 100644 index 00000000..a87e9cc0 --- /dev/null +++ b/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/_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