feat(crawler): captcha solving + per-attempt billing (Phase 3d)

Wire captchatools as the StealthyFetcher-tier page_action to detect, harvest
(egressing from the crawl's own proxy IP), inject, and submit reCAPTCHA v2/v3
and hCaptcha tokens. Opt-in and off by default (zero attempts, zero cost).

Licensing split:
- Apache-2 app/utils/captcha/ holds the generic, vendor-agnostic config
  (CaptchaConfig + captcha_enabled() = flag AND key present).
- Proprietary app/proprietary/web_crawler/captcha.py holds the bypass logic
  (detect/harvest/inject) plus a process-wide solver latch that halts solving
  on unrecoverable errors (no balance / bad key).

Crawler: CrawlOutcome gains captcha_attempts/captcha_solved, surfaced via a
per-call captcha_state dict threaded crawl_url -> _crawl_with_stealthy(_sync)
and stamped onto every stealth terminal outcome. The stealth tier captures the
proxy once and reuses it for both the fetch and the solver (IP-coherence).

Billing: WebCrawlCreditService gains captcha_billing_enabled,
captcha_solves_to_micros, charge_captcha, and a generic check_balance, sharing
a single _apply_debit path. The indexer accumulates attempts (even on failed
crawls), runs a combined crawl+captcha pre-flight, and posts a per-attempt
owner charge as usage_type="web_crawl_captcha". The captcha worst-case is only
reserved when solving is actually enabled, so a solving-off deployment is never
blocked for captcha that can never run. Both chat scrape tools fold attempts
into the current turn before the success/fail branch.

Fully config-driven prices; no migration. New unit tests cover the config,
factory (detection/latch/timeout/cap), credit service, indexer wiring, and the
chat fold.

Co-authored-by: Cursor <cursoragent@cursor.com>

surfsense_backend/tests/unit/connector_indexers/test_web_crawl_billing.py

@@ -53,7 +53,12 @@ def _make_session(owner_id, balance_micros, reserved_micros=0):
     return session, fake_user
 
 
-def _outcome(success: bool, content: str = "Hello content"):
+def _outcome(
+    success: bool,
+    content: str = "Hello content",
+    captcha_attempts: int = 0,
+    captcha_solved: bool = False,
+):
     o = MagicMock()
     if success:
         o.status = CrawlOutcomeStatus.SUCCESS
@@ -67,6 +72,9 @@ def _outcome(success: bool, content: str = "Hello content"):
         o.status = CrawlOutcomeStatus.FAILED
         o.result = None
         o.error = "blocked"
+    # Real ints/bools (not MagicMock) so 03d accumulation arithmetic is safe.
+    o.captcha_attempts = captcha_attempts
+    o.captcha_solved = captcha_solved
     return o
 
 
@@ -188,3 +196,111 @@ async def test_disabled_skips_billing_but_still_crawls(indexer_env, monkeypatch)
     indexer_env["crawler"].crawl_url.assert_awaited_once()
     indexer_env["record_usage"].assert_not_awaited()
     assert user.credit_micros_balance == 100_000
+
+
+# ===================================================================
+# Phase 3d — captcha per-attempt billing in the indexer
+# ===================================================================
+
+
+async def test_captcha_attempts_charged_to_owner(indexer_env, monkeypatch):
+    """Captcha billed per attempt to the OWNER as a separate web_crawl_captcha
+    unit. Crawl billing OFF here to isolate the captcha charge."""
+    monkeypatch.setattr(config, "WEB_CRAWL_CREDIT_BILLING_ENABLED", False)
+    monkeypatch.setattr(config, "WEB_CRAWL_CAPTCHA_BILLING_ENABLED", True)
+    monkeypatch.setattr(config, "WEB_CRAWL_CAPTCHA_MICROS_PER_SOLVE", 3000)
+    session, user = _make_session(_OWNER_USER, balance_micros=100_000)
+    indexer_env["crawler"].crawl_url.side_effect = [
+        _outcome(True, captcha_attempts=1, captcha_solved=True),
+        _outcome(True, captcha_attempts=1, captcha_solved=True),
+    ]
+
+    total, _warning = await _run(
+        indexer_env, session, ["https://a.com", "https://b.com"]
+    )
+
+    assert total == 2
+    # Owner debited 2 attempts * 3000 (no crawl charge — crawl billing off).
+    assert user.credit_micros_balance == 100_000 - 6000
+    indexer_env["record_usage"].assert_awaited_once()
+    kwargs = indexer_env["record_usage"].await_args.kwargs
+    assert kwargs["usage_type"] == "web_crawl_captcha"
+    assert kwargs["user_id"] == _OWNER_USER
+    assert kwargs["cost_micros"] == 6000
+    assert kwargs["call_details"]["attempts"] == 2
+    assert kwargs["call_details"]["solved"] == 2
+
+
+async def test_captcha_attempt_billed_even_when_crawl_fails(indexer_env, monkeypatch):
+    """A failed solve still cost solver money → bill the attempt anyway."""
+    monkeypatch.setattr(config, "WEB_CRAWL_CREDIT_BILLING_ENABLED", False)
+    monkeypatch.setattr(config, "WEB_CRAWL_CAPTCHA_BILLING_ENABLED", True)
+    monkeypatch.setattr(config, "WEB_CRAWL_CAPTCHA_MICROS_PER_SOLVE", 3000)
+    session, user = _make_session(_OWNER_USER, balance_micros=100_000)
+    indexer_env["crawler"].crawl_url.side_effect = [
+        _outcome(False, captcha_attempts=1, captcha_solved=False)
+    ]
+
+    total, _warning = await _run(indexer_env, session, ["https://a.com"])
+
+    assert total == 0  # crawl failed → not indexed
+    assert user.credit_micros_balance == 100_000 - 3000  # but solve attempt billed
+    kwargs = indexer_env["record_usage"].await_args.kwargs
+    assert kwargs["usage_type"] == "web_crawl_captcha"
+    assert kwargs["call_details"]["attempts"] == 1
+    assert kwargs["call_details"]["solved"] == 0
+
+
+async def test_captcha_billing_on_but_solving_off_does_not_overreserve(
+    indexer_env, monkeypatch
+):
+    """Regression: captcha billing ON but solving OFF must NOT reserve the
+    worst-case captcha budget in pre-flight.
+
+    With solving off, attempts can never happen, so a balance below the
+    (would-be) worst-case reservation must still let the run proceed rather than
+    being wrongly blocked for captcha that never runs.
+    """
+    monkeypatch.setattr(config, "WEB_CRAWL_CREDIT_BILLING_ENABLED", False)
+    monkeypatch.setattr(config, "WEB_CRAWL_CAPTCHA_BILLING_ENABLED", True)
+    monkeypatch.setattr(config, "WEB_CRAWL_CAPTCHA_MICROS_PER_SOLVE", 3000)
+    monkeypatch.setattr(config, "CAPTCHA_MAX_ATTEMPTS_PER_URL", 3)
+    # Solving OFF → captcha_enabled() is False (no flag + no key).
+    monkeypatch.setattr(config, "CAPTCHA_SOLVING_ENABLED", False)
+    monkeypatch.setattr(config, "CAPTCHA_SOLVER_API_KEY", "")
+
+    # Balance (5000) is below the buggy worst-case reservation
+    # (2 URLs * 3 attempts * 3000 = 18000) but the run must still proceed.
+    session, user = _make_session(_OWNER_USER, balance_micros=5000)
+    indexer_env["crawler"].crawl_url.side_effect = [
+        _outcome(True, captcha_attempts=0),
+        _outcome(True, captcha_attempts=0),
+    ]
+
+    total, warning = await _run(
+        indexer_env, session, ["https://a.com", "https://b.com"]
+    )
+
+    assert total == 2
+    assert warning is None
+    assert indexer_env["crawler"].crawl_url.await_count == 2
+    # Nothing billed: crawl billing off, no captcha attempts.
+    indexer_env["record_usage"].assert_not_awaited()
+    assert user.credit_micros_balance == 5000
+
+
+async def test_captcha_disabled_skips_captcha_billing(indexer_env, monkeypatch):
+    """Captcha billing OFF (default) → attempts on the outcome are ignored."""
+    monkeypatch.setattr(config, "WEB_CRAWL_CAPTCHA_BILLING_ENABLED", False)
+    session, user = _make_session(_OWNER_USER, balance_micros=100_000)
+    indexer_env["crawler"].crawl_url.side_effect = [
+        _outcome(True, captcha_attempts=5, captcha_solved=True)
+    ]
+
+    total, _warning = await _run(indexer_env, session, ["https://a.com"])
+
+    assert total == 1
+    # Only the crawl unit billed (1 * 1000); captcha ignored entirely.
+    assert user.credit_micros_balance == 100_000 - 1000
+    kwargs = indexer_env["record_usage"].await_args.kwargs
+    assert kwargs["usage_type"] == "web_crawl"

surfsense_backend/tests/unit/proprietary/web_crawler/test_captcha.py

@@ -0,0 +1,237 @@
+"""Unit tests for the proprietary captcha page_action factory (Phase 3d).
+
+The browser/solver boundary is mocked: a fake Playwright ``page`` and a
+monkeypatched ``_harvest_token`` / injection. We assert the glue logic —
+detection, proxy reformatting, attempt counting + state surfacing, the per-URL
+cap, and the no-balance process latch.
+"""
+
+import pytest
+
+from app.proprietary.web_crawler import captcha as cap
+from app.utils.captcha import CaptchaConfig
+
+pytestmark = pytest.mark.unit
+
+
+def _cfg(**overrides) -> CaptchaConfig:
+    base = dict(
+        enabled=True,
+        solving_site="capsolver",
+        api_key="key-123",
+        max_attempts_per_url=1,
+        timeout_s=30,
+        captcha_type_default="v2",
+        v3_min_score=0.7,
+        v3_action="verify",
+    )
+    base.update(overrides)
+    return CaptchaConfig(**base)
+
+
+class _FakeEl:
+    def __init__(self, attrs: dict[str, str]):
+        self._attrs = attrs
+
+    def get_attribute(self, name: str) -> str | None:
+        return self._attrs.get(name)
+
+
+class _FakePage:
+    """Minimal sync-Playwright-ish page for detection/injection tests."""
+
+    def __init__(self, widgets=None, iframes=None, scripts=None, url="https://t.test/p"):
+        self._widgets = widgets or {}  # selector -> _FakeEl
+        self._iframes = iframes or []  # list[_FakeEl]
+        self._scripts = scripts or []  # list[_FakeEl]
+        self.url = url
+        self.evaluated: list = []
+
+    def query_selector(self, selector: str):
+        return self._widgets.get(selector)
+
+    def query_selector_all(self, selector: str):
+        if selector == "iframe[src]":
+            return self._iframes
+        if selector == "script[src]":
+            return self._scripts
+        return []
+
+    def evaluate(self, js, arg=None):
+        self.evaluated.append((js, arg))
+        if "navigator.userAgent" in js:
+            return "UA/1.0"
+        return None
+
+    def wait_for_load_state(self, *_a, **_k):
+        return None
+
+
+@pytest.fixture(autouse=True)
+def _clear_latch():
+    cap.reset_solver_latch()
+    yield
+    cap.reset_solver_latch()
+
+
+# --- proxy reformat --------------------------------------------------------
+
+
+class TestProxyReformat:
+    def test_with_auth(self):
+        assert (
+            cap.proxy_url_to_captchatools("http://user:pass@1.2.3.4:8080")
+            == "1.2.3.4:8080:user:pass"
+        )
+
+    def test_without_auth(self):
+        assert cap.proxy_url_to_captchatools("http://1.2.3.4:8080") == "1.2.3.4:8080"
+
+    def test_none_and_garbage(self):
+        assert cap.proxy_url_to_captchatools(None) is None
+        assert cap.proxy_url_to_captchatools("not-a-url") is None
+
+
+# --- detection -------------------------------------------------------------
+
+
+class TestDetect:
+    def test_recaptcha_v2_widget(self):
+        page = _FakePage(
+            widgets={".g-recaptcha[data-sitekey]": _FakeEl({"data-sitekey": "SK_V2"})}
+        )
+        assert cap.detect_challenge(page, _cfg()) == ("v2", "SK_V2")
+
+    def test_hcaptcha_widget(self):
+        page = _FakePage(
+            widgets={".h-captcha[data-sitekey]": _FakeEl({"data-sitekey": "SK_H"})}
+        )
+        assert cap.detect_challenge(page, _cfg()) == ("hcaptcha", "SK_H")
+
+    def test_iframe_fallback(self):
+        page = _FakePage(
+            iframes=[_FakeEl({"src": "https://www.google.com/recaptcha/api2?k=SK_IF"})]
+        )
+        assert cap.detect_challenge(page, _cfg()) == ("v2", "SK_IF")
+
+    def test_v3_render_param_when_default_v3(self):
+        page = _FakePage(
+            scripts=[_FakeEl({"src": "https://www.google.com/recaptcha/api.js?render=SK_V3"})]
+        )
+        assert cap.detect_challenge(page, _cfg(captcha_type_default="v3")) == (
+            "v3",
+            "SK_V3",
+        )
+
+    def test_no_challenge(self):
+        assert cap.detect_challenge(_FakePage(), _cfg()) is None
+
+
+# --- factory / page_action -------------------------------------------------
+
+
+class TestFactoryGating:
+    def test_returns_none_when_disabled(self):
+        state = {"attempts": 0, "solved": False}
+        assert build_action(state, _cfg(enabled=False)) is None
+
+    def test_returns_none_when_latched(self):
+        cap._latch_solver("test")
+        state = {"attempts": 0, "solved": False}
+        assert build_action(state, _cfg()) is None
+
+
+def build_action(state, cfg, proxy="http://u:p@1.2.3.4:9000"):
+    return cap.build_captcha_page_action(state, proxy, cfg)
+
+
+class TestPageAction:
+    def test_solves_and_records(self, monkeypatch):
+        captured = {}
+
+        def _fake_harvest(cfg, ctype, sitekey, page_url, proxy, ua):
+            captured.update(
+                ctype=ctype, sitekey=sitekey, page_url=page_url, proxy=proxy, ua=ua
+            )
+            return "TOKEN"
+
+        injected = {}
+        monkeypatch.setattr(cap, "_harvest_token", _fake_harvest)
+        monkeypatch.setattr(
+            cap,
+            "_inject_and_submit",
+            lambda page, ctype, token: injected.update(ctype=ctype, token=token),
+        )
+
+        state = {"attempts": 0, "solved": False}
+        action = build_action(state, _cfg())
+        page = _FakePage(
+            widgets={".g-recaptcha[data-sitekey]": _FakeEl({"data-sitekey": "SK"})}
+        )
+        action(page)
+
+        assert state == {"attempts": 1, "solved": True}
+        # Solver egressed from the crawl's proxy, reformatted, with the page UA.
+        assert captured["proxy"] == "1.2.3.4:9000:u:p"
+        assert captured["ua"] == "UA/1.0"
+        assert captured["ctype"] == "v2"
+        assert injected == {"ctype": "v2", "token": "TOKEN"}
+
+    def test_no_challenge_no_attempt(self, monkeypatch):
+        monkeypatch.setattr(
+            cap, "_harvest_token", lambda *a, **k: pytest.fail("should not solve")
+        )
+        state = {"attempts": 0, "solved": False}
+        build_action(state, _cfg())(_FakePage())
+        assert state == {"attempts": 0, "solved": False}
+
+    def test_per_url_attempt_cap(self, monkeypatch):
+        calls = {"n": 0}
+
+        def _h(*_a, **_k):
+            calls["n"] += 1
+            return "TOKEN"
+
+        monkeypatch.setattr(cap, "_harvest_token", _h)
+        monkeypatch.setattr(cap, "_inject_and_submit", lambda *a, **k: None)
+
+        # Already at the cap → no further solve.
+        state = {"attempts": 1, "solved": False}
+        page = _FakePage(
+            widgets={".g-recaptcha[data-sitekey]": _FakeEl({"data-sitekey": "SK"})}
+        )
+        build_action(state, _cfg(max_attempts_per_url=1))(page)
+        assert calls["n"] == 0
+        assert state["attempts"] == 1
+
+    def test_empty_token_counts_attempt_but_not_solved(self, monkeypatch):
+        monkeypatch.setattr(cap, "_harvest_token", lambda *a, **k: None)
+        monkeypatch.setattr(
+            cap, "_inject_and_submit", lambda *a, **k: pytest.fail("no inject")
+        )
+        state = {"attempts": 0, "solved": False}
+        page = _FakePage(
+            widgets={".g-recaptcha[data-sitekey]": _FakeEl({"data-sitekey": "SK"})}
+        )
+        build_action(state, _cfg())(page)
+        assert state == {"attempts": 1, "solved": False}
+
+    def test_no_balance_latches_and_stops(self, monkeypatch):
+        class ErrNoBalance(Exception):
+            pass
+
+        def _boom(*_a, **_k):
+            raise ErrNoBalance("out of balance")
+
+        monkeypatch.setattr(cap, "_harvest_token", _boom)
+        state = {"attempts": 0, "solved": False}
+        page = _FakePage(
+            widgets={".g-recaptcha[data-sitekey]": _FakeEl({"data-sitekey": "SK"})}
+        )
+        build_action(state, _cfg(max_attempts_per_url=3))(page)
+
+        # Attempt counted, not solved, and the process latch is tripped so the
+        # next factory build refuses to wire solving at all (no retry loop).
+        assert state == {"attempts": 1, "solved": False}
+        assert cap.solver_latched() is True
+        assert build_action({"attempts": 0, "solved": False}, _cfg()) is None

surfsense_backend/tests/unit/proprietary/web_crawler/test_connector.py

@@ -47,7 +47,7 @@ async def test_static_tier_success_short_circuits(
         later_calls.append("dynamic")
         return None
 
-    async def _record_stealthy(_url: str) -> None:
+    async def _record_stealthy(_url: str, *_args) -> None:
         later_calls.append("stealthy")
         return None
 
@@ -89,7 +89,7 @@ async def test_all_tiers_empty_is_empty(monkeypatch: pytest.MonkeyPatch) -> None
     """Every tier fetched but extracted nothing -> EMPTY (not billable)."""
     crawler = WebCrawlerConnector()
 
-    async def _empty(_url: str) -> None:
+    async def _empty(_url: str, *_args) -> None:
         return None
 
     monkeypatch.setattr(crawler, "_crawl_with_async_fetcher", _empty)
@@ -106,7 +106,7 @@ async def test_all_tiers_raise_is_failed(monkeypatch: pytest.MonkeyPatch) -> Non
     """Every tier raising (none reachable) -> FAILED with aggregated errors."""
     crawler = WebCrawlerConnector()
 
-    async def _boom(_url: str) -> None:
+    async def _boom(_url: str, *_args) -> None:
         raise RuntimeError("fetch exploded")
 
     monkeypatch.setattr(crawler, "_crawl_with_async_fetcher", _boom)
@@ -153,7 +153,7 @@ async def test_proxy_error_no_retry_when_single_endpoint(
         static_calls["n"] += 1
         raise RuntimeError("connection refused by upstream proxy")
 
-    async def _empty(_url: str) -> None:
+    async def _empty(_url: str, *_args) -> None:
         return None
 
     monkeypatch.setattr(crawler, "_crawl_with_async_fetcher", _proxy_err)
@@ -167,6 +167,77 @@ async def test_proxy_error_no_retry_when_single_endpoint(
     assert outcome.status is CrawlOutcomeStatus.EMPTY
 
 
+async def test_captcha_defaults_zero_on_non_stealthy_success(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """03d: a static success never touches captcha → fields stay 0/False."""
+    crawler = WebCrawlerConnector()
+
+    async def _static(_url: str) -> dict:
+        return _result("scrapling-static")
+
+    monkeypatch.setattr(crawler, "_crawl_with_async_fetcher", _static)
+    outcome = await crawler.crawl_url("https://example.com")
+
+    assert outcome.status is CrawlOutcomeStatus.SUCCESS
+    assert outcome.captcha_attempts == 0
+    assert outcome.captcha_solved is False
+
+
+async def test_captcha_state_surfaced_onto_outcome(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """03d: the stealthy tier's captcha_state is stamped onto the outcome,
+    even when the lower tiers missed and stealthy itself succeeds."""
+    crawler = WebCrawlerConnector()
+
+    async def _empty(_url: str, *_args) -> None:
+        return None
+
+    async def _stealthy(_url: str, captcha_state: dict) -> dict:
+        # Simulate the page_action having solved a captcha mid-fetch.
+        captcha_state["attempts"] = 2
+        captcha_state["solved"] = True
+        return _result("scrapling-stealthy")
+
+    monkeypatch.setattr(crawler, "_crawl_with_async_fetcher", _empty)
+    monkeypatch.setattr(crawler, "_crawl_with_dynamic", _empty)
+    monkeypatch.setattr(crawler, "_crawl_with_stealthy", _stealthy)
+
+    outcome = await crawler.crawl_url("https://example.com")
+
+    assert outcome.status is CrawlOutcomeStatus.SUCCESS
+    assert outcome.tier == "scrapling-stealthy"
+    assert outcome.captcha_attempts == 2
+    assert outcome.captcha_solved is True
+
+
+async def test_captcha_attempts_surface_even_when_crawl_fails(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """03d: attempts are billed per-attempt → must surface on a FAILED outcome."""
+    crawler = WebCrawlerConnector()
+
+    async def _empty(_url: str, *_args) -> None:
+        return None
+
+    async def _stealthy_attempt_then_empty(_url: str, captcha_state: dict) -> None:
+        captcha_state["attempts"] = 1  # solve attempted but crawl still empty
+        return None
+
+    monkeypatch.setattr(crawler, "_crawl_with_async_fetcher", _empty)
+    monkeypatch.setattr(crawler, "_crawl_with_dynamic", _empty)
+    monkeypatch.setattr(
+        crawler, "_crawl_with_stealthy", _stealthy_attempt_then_empty
+    )
+
+    outcome = await crawler.crawl_url("https://example.com")
+
+    assert outcome.status is CrawlOutcomeStatus.EMPTY
+    assert outcome.captcha_attempts == 1
+    assert outcome.captcha_solved is False
+
+
 async def test_non_proxy_error_never_retries(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -178,7 +249,7 @@ async def test_non_proxy_error_never_retries(
         calls["n"] += 1
         raise RuntimeError("totally unrelated failure")
 
-    async def _empty(_url: str) -> None:
+    async def _empty(_url: str, *_args) -> None:
         return None
 
     monkeypatch.setattr(crawler, "_crawl_with_async_fetcher", _boom)

surfsense_backend/tests/unit/services/test_web_crawl_credit_service.py

@@ -213,3 +213,112 @@ class TestChatScrapeFold:
         # Run in a fresh context so the turn ContextVar resolves to its default
         # (None) regardless of other tests — must not raise.
         contextvars.Context().run(self._import_helper())
+
+
+# ===================================================================
+# F) Captcha per-attempt unit (Phase 3d)
+# ===================================================================
+
+
+@pytest.fixture
+def _enable_captcha_billing(monkeypatch):
+    monkeypatch.setattr(config, "WEB_CRAWL_CAPTCHA_BILLING_ENABLED", True)
+    monkeypatch.setattr(config, "WEB_CRAWL_CAPTCHA_MICROS_PER_SOLVE", 3000)
+
+
+class TestCaptchaStatics:
+    def test_billing_enabled_reads_flag(self, monkeypatch):
+        monkeypatch.setattr(config, "WEB_CRAWL_CAPTCHA_BILLING_ENABLED", True)
+        assert WebCrawlCreditService.captcha_billing_enabled() is True
+        monkeypatch.setattr(config, "WEB_CRAWL_CAPTCHA_BILLING_ENABLED", False)
+        assert WebCrawlCreditService.captcha_billing_enabled() is False
+
+    def test_solves_to_micros_is_config_driven(self, monkeypatch):
+        monkeypatch.setattr(config, "WEB_CRAWL_CAPTCHA_MICROS_PER_SOLVE", 3000)
+        assert WebCrawlCreditService.captcha_solves_to_micros(1) == 3000
+        assert WebCrawlCreditService.captcha_solves_to_micros(1000) == 3_000_000
+        monkeypatch.setattr(config, "WEB_CRAWL_CAPTCHA_MICROS_PER_SOLVE", 5000)
+        assert WebCrawlCreditService.captcha_solves_to_micros(2) == 10_000
+
+
+class TestChargeCaptcha:
+    async def test_debits_per_attempt(self, _enable_captcha_billing):
+        session, user = _make_session(balance_micros=100_000)
+        svc = WebCrawlCreditService(session)
+        new_balance = await svc.charge_captcha(_USER_ID, attempts=2)
+        assert user.credit_micros_balance == 100_000 - 2 * 3000
+        assert new_balance == 94_000
+        session.commit.assert_awaited()
+
+    async def test_zero_attempts_is_noop(self, _enable_captcha_billing):
+        session, user = _make_session(balance_micros=100_000)
+        svc = WebCrawlCreditService(session)
+        assert await svc.charge_captcha(_USER_ID, attempts=0) is None
+        assert user.credit_micros_balance == 100_000
+        session.commit.assert_not_awaited()
+
+    async def test_disabled_is_noop(self, monkeypatch):
+        monkeypatch.setattr(config, "WEB_CRAWL_CAPTCHA_BILLING_ENABLED", False)
+        session, user = _make_session(balance_micros=100_000)
+        svc = WebCrawlCreditService(session)
+        assert await svc.charge_captcha(_USER_ID, attempts=5) is None
+        assert user.credit_micros_balance == 100_000
+        session.execute.assert_not_called()
+
+
+class TestCheckBalance:
+    """Combined (crawl + captcha) pre-flight primitive — ungated."""
+
+    async def test_passes_when_affordable(self):
+        session, _user = _make_session(balance_micros=10_000)
+        await WebCrawlCreditService(session).check_balance(_USER_ID, 5_000)
+
+    async def test_raises_when_short(self):
+        session, _user = _make_session(balance_micros=3_000)
+        with pytest.raises(InsufficientCreditsError):
+            await WebCrawlCreditService(session).check_balance(_USER_ID, 5_000)
+
+    async def test_zero_required_is_noop(self):
+        session, _user = _make_session(balance_micros=0)
+        await WebCrawlCreditService(session).check_balance(_USER_ID, 0)
+        session.execute.assert_not_called()
+
+
+class TestCaptchaChatFold:
+    """The scrape tools fold captcha attempts into the live turn accumulator."""
+
+    def _import_helper(self):
+        from app.agents.chat.multi_agent_chat.main_agent.tools.scrape_webpage import (
+            _bill_captcha_attempts,
+        )
+
+        return _bill_captcha_attempts
+
+    def _outcome(self, attempts):
+        o = MagicMock()
+        o.captcha_attempts = attempts
+        return o
+
+    def test_folds_attempts_when_enabled(self, _enable_captcha_billing):
+        from app.services.token_tracking_service import start_turn
+
+        acc = start_turn()
+        self._import_helper()(self._outcome(2))
+        assert acc.total_cost_micros == 6000
+        assert len(acc.calls) == 1
+        assert acc.calls[0].call_kind == "web_crawl_captcha"
+
+    def test_noop_when_zero_attempts(self, _enable_captcha_billing):
+        from app.services.token_tracking_service import start_turn
+
+        acc = start_turn()
+        self._import_helper()(self._outcome(0))
+        assert acc.calls == []
+
+    def test_noop_when_billing_disabled(self, monkeypatch):
+        from app.services.token_tracking_service import start_turn
+
+        monkeypatch.setattr(config, "WEB_CRAWL_CAPTCHA_BILLING_ENABLED", False)
+        acc = start_turn()
+        self._import_helper()(self._outcome(3))
+        assert acc.calls == []

surfsense_backend/tests/unit/utils/test_captcha_config.py

@@ -0,0 +1,47 @@
+"""Unit tests for the generic captcha config gate (Phase 3d, Apache-2 layer)."""
+
+import pytest
+
+from app.config import config
+from app.utils.captcha import captcha_enabled, get_captcha_config
+
+pytestmark = pytest.mark.unit
+
+
+class TestCaptchaEnabled:
+    def test_off_by_default(self, monkeypatch):
+        monkeypatch.setattr(config, "CAPTCHA_SOLVING_ENABLED", False)
+        monkeypatch.setattr(config, "CAPTCHA_SOLVER_API_KEY", "key")
+        assert captcha_enabled() is False
+
+    def test_flag_on_but_no_key_is_disabled(self, monkeypatch):
+        monkeypatch.setattr(config, "CAPTCHA_SOLVING_ENABLED", True)
+        monkeypatch.setattr(config, "CAPTCHA_SOLVER_API_KEY", None)
+        assert captcha_enabled() is False
+
+    def test_flag_on_with_key_is_enabled(self, monkeypatch):
+        monkeypatch.setattr(config, "CAPTCHA_SOLVING_ENABLED", True)
+        monkeypatch.setattr(config, "CAPTCHA_SOLVER_API_KEY", "key")
+        assert captcha_enabled() is True
+
+
+class TestGetCaptchaConfig:
+    def test_snapshot_reflects_config(self, monkeypatch):
+        monkeypatch.setattr(config, "CAPTCHA_SOLVING_ENABLED", True)
+        monkeypatch.setattr(config, "CAPTCHA_SOLVER_API_KEY", "key")
+        monkeypatch.setattr(config, "CAPTCHA_SOLVER_PROVIDER", "2captcha")
+        monkeypatch.setattr(config, "CAPTCHA_MAX_ATTEMPTS_PER_URL", 3)
+        monkeypatch.setattr(config, "CAPTCHA_SOLVE_TIMEOUT_S", 90)
+        monkeypatch.setattr(config, "CAPTCHA_TYPE_DEFAULT", "hcaptcha")
+
+        cfg = get_captcha_config()
+        assert cfg.enabled is True
+        assert cfg.solving_site == "2captcha"
+        assert cfg.max_attempts_per_url == 3
+        assert cfg.timeout_s == 90
+        assert cfg.captcha_type_default == "hcaptcha"
+
+    def test_keyless_config_is_not_enabled(self, monkeypatch):
+        monkeypatch.setattr(config, "CAPTCHA_SOLVING_ENABLED", True)
+        monkeypatch.setattr(config, "CAPTCHA_SOLVER_API_KEY", None)
+        assert get_captcha_config().enabled is False