refactor: layered scrape API, timeout_phase telemetry, and isolation hardening (#45)

* fix(obs): tag scrape timeouts with queue/boot/work phase

Handler wait_for cancellations were always reporting attempts=0 with no stage, which made Sentry P0 timeouts impossible to triage. Track thread-visible progress and surface timeout_phase on 504 envelopes and Sentry tags.

* fix(obs): document timeout_phase and lock engine progress marks

Address review follow-ups: README/OpenAPI 504 example for the new diagnostics field, mark boot before profile setup, and assert Driver construction sees BOOT while request/browser tiers advance to WORK.

* test(obs): assert engine marks BOOT before Driver and WORK on tiers

Lock the progress.mark placement that handler-timeout telemetry depends on, so deleting engine marks cannot pass the suite via a faked execute alone.

* refactor(obs): densify timeout_phase progress and tests

Collapse duplicate terminal-error plumbing and coalesce progress once so
tier marks stay unconditional; shrink the regression suite around the
same BOOT/WORK locks.

* test(obs): lock request-tier timeout category and phase

Document that request-tier timeout exceptions mirror browser remapping,
and assert envelope timeout_phase plus Sentry fingerprint invariance.

* fix(runtime): prune orphan scrape dirs on every request

Extract runtime cleanup helpers, prune inactive dirs at session entry,
and retry profile creation after ENOSPC with another prune pass.

* refactor(api): restructure into layered packages with FastAPI DI.

Split the monolithic engine and main modules into api/domain/engine/infra layers, centralize env config in Settings, and wire scrape orchestration through ScrapeService with dependency injection.

* refactor(api): remove dead DI/tier params and dedupe service identity

* fix(engine): singleton engine and executor with isolation regression tests

* refactor(config): thread Settings through deps and drop import-time freezes

* refactor(api): single-source OpenAPI examples from model instances

* refactor(engine): typed DriverProtocol and centralized capability adapter

* refactor(config): nest Sentry settings and simplify env layout

* refactor(schemas): split into enums/request/response modules

* test: reorganize suite by layer and add scrape bench harness

Split monolithic contract tests into api/security/infra/engine layers,
extract shared fakes, and add scripts/bench_scrape.py with a lightweight
regression guard for the bench entrypoint.

* perf(engine): lazy imports and dedupe browser-tier hot path

Defer Botasaurus/CDP imports to tier entrypoints, consolidate post-bypass
page-state collection, and document SCRAPE_MAX_WORKERS tuning for low-RAM
hosts. Bench p50 wall_ms: 106.3 -> 102.9 (no regression).

* chore(types): add pyright gate and tighten engine seams

Wire pyright into make check, tighten Settings.sentry and DriverProtocol
seams, and relax test-only diagnostics via executionEnvironments.

* docs(agents): document hardening pass architecture conventions

Capture singleton lifecycle, settings threading, isolation invariants,
driver capabilities, schemas layout, bench harness, and pyright gate.

* refactor(types): use strict pyright (#46)

* chore(types): add vendor stubs and CDP TypedDicts

Introduce local .pyi stubs for Botasaurus/browser driver seams and
shared CDP log/XHR shapes so strict pyright can type vendor boundaries.

* refactor(types): type engine vendor seams end-to-end

Wire DriverProtocol/CdpTabProtocol through metadata, XHR capture, and
envelope builders so xhr_responses stays list[XhrResponse] throughout.

* refactor(types): harden API and OpenAPI module boundaries

Replace mutable OpenAPI globals with a registry/getter pattern and add
typed validation-error and Sentry callback seams for strict pyright.

* test(types): add factories and typed test support seams

Introduce scrape_request()/example_url() helpers, protocol-shaped fakes,
and an execute-side-effect proxy so tests stay strict-ready without raw
HttpUrl construction at every call site.

* chore(types): enable pyright strict mode gate

Flip typeCheckingMode=strict, add httpx/TestClient stubs and test support
helpers, document residual policy, and regenerate OpenAPI after schema tweak.

* chore(cleanup): delete spike script, dead surfaces, and unreachable paths

Remove scripts/xhr_spike.py (Phase 0 spike; its finding lives in the
XhrCollector docstring), unused tests/support/responses.py, the
uncalled ScrapeService.serialize, the assessment error_category field
that only tests asserted, dead target_url/min_free_bytes parameters,
the unreachable re-raise in ScrapeSession.prepare_profile_dirs, and
the redundant import-time configure_openapi in app.main.

* refactor(logging): single-source the service logger via get_logger()

Fourteen modules hardcoded the "botasaurus_scrape_api" logger name
while app.logging_config already owned it; route every module through
get_logger() so the name has one home.

* refactor(engine): own wall-clock budget math in one module

Move remaining_total/work seconds, step budget, elapsed_ms, and
is_timeout_exception into app/engine/budget.py so browser_tier no
longer imports from request_tier and the orchestrator no longer
reaches into browser_tier for a generic timeout predicate. Fold the
challenge/bypass double-collect into settle_page_state and dedupe the
five hand-rolled render_ms computations behind elapsed_ms.

* refactor(api): deepen ScrapeService and thin the scrape route

The route owned guardrail policy: which validations run, in what
order, and each fallback message. Move request-id resolution and SSRF
guard sequencing into ScrapeService.process so the HTTP shell only
parses and serializes; drop the never-injected resolve_scrape_request_id
dependency and narrow the service surface to process + build_timeout_error.
OpenAPI snapshot verified unchanged.

* refactor(types): typed Sentry scope seam and single readiness fact

Replace ten attr-defined ignores in ops telemetry with a SentryScope
protocol cast once at the vendor boundary. Collapse sentry_is_ready to
the _initialized fact (init already requires a DSN) and delete the
now-callerless is_sentry_enabled; drop the duplicate sample-rate clamp
that Settings validators already own. Tighten ChallengeDetector's
driver param to object and make PendingXhrMeta.request_id required so
harvest stops re-normalizing ids.

* test: layer the suite and shrink blanket pyright directives

Move root test modules into their owning layer dirs (sentry, ops
telemetry, request id, runtime cleanup -> infra; isolation -> engine)
and split the timeout-phase monolith by flight height: progress
snapshot unit (infra), build_timeout_error mapping (domain), engine
phase marking (engine), and the 504 envelope contract (api). Share one
canned botasaurus Request fake instead of three inline mock blobs,
route all HTTP tests through test_client, and drop the eleven-rule
blanket pyright directive from the seven modules that pass strict
without it (narrow reportPrivateUsage or real annotations elsewhere).

* docs: sync AGENTS.md and typing residuals with the refactor

Layout now names logging_config.py, engine/budget.py, and the layered
test tree; conventions point at get_logger(), budget.py as the only
wall-clock owner, and test_client() over ad-hoc TestClient. The contract
section reflects ScrapeService.process() (serialize() is gone), and the
typing-residuals ledger lists the six remaining per-module pyright
directives instead of a blanket-everywhere policy.

* test(api): pin SSRF guardrail at the HTTP seam

Review flagged that after folding URL guardrails into
ScrapeService.process(), no in-process HTTP test proved the 403 path;
only the Docker smoke script covered it. Pin localhost, private-IP,
and blocked-proxy targets to the 403 validation envelope with honored
request ids so a regression in the service wiring fails make check,
not just make smoke.

* fix(engine): make ENOSPC retry recreatable; pin budget math with units

Review nice-to-haves: the ENOSPC retry in ScrapeSession could never
succeed when runtime_dir was created but profile_dir hit ENOSPC --
the retry's exist_ok=False mkdir tripped on the surviving partial dir.
Remove the partial tree before the prune-and-retry pass. Add direct
unit tests for budget.py clamps (floor at 1s, tighter-constraint min)
and timeout-substring classification, and make the 504 fixture return
an inert sentinel so the test proves the envelope comes from the
handler timeout path, not the fixture.

* fix(engine): honor submission deadline and harden session/boot isolation

Derive wall-clock start from the API deadline so queue wait cannot grant a
second full budget; unregister on prepare failure; prune under the active-id
lock; map Driver() boot failures to the 502 envelope; resolve optional
methods via driver_capabilities.resolve_callable.

* fix(api): build routes after OpenAPI configure; live wait_timeout default

create_router() factories capture timeout-dependent response metadata per
app instance. wait_timeout_seconds uses a settings-aware default_factory
with a stable OpenAPI default of 15.

* chore(ci): install pyright in requirements-dev and typecheck in CI

Align HTTP contract tests on test_client() and collapse dual sentry imports.

Author: gildesmarais

.github/workflows/ci.yml

@@ -98,6 +98,9 @@ jobs:
       - name: Execute unit test suite
         run: python -m unittest discover -s tests -p "test_*.py" -v
 
+      - name: Typecheck
+        run: make typecheck
+
       - name: Verify OpenAPI snapshot
         run: make openapi-verify
 

AGENTS.md

@@ -5,11 +5,103 @@
 - Docker-first and Docker-only unless user asks otherwise.
 - Keep repo focused: stable Botasaurus scrape API wrapper, not generic framework.
 
+## Project Layout
+
+```
+app/
+  main.py                    # create_app() factory; module-level `app` for uvicorn
+  constants.py               # SERVICE_NAME and other shared literals
+  config.py                  # Settings + nested SentrySettings; single env source of truth
+  exceptions.py              # domain exceptions (e.g. RequestIdCollisionError)
+  logging_config.py          # setup_logging + get_logger(); single owner of logger name
+  api/
+    deps.py                  # FastAPI Depends: settings, engine, executor, ScrapeService
+    errors.py                # 422 + 500 handlers → scrape error envelope
+    openapi.py               # route OpenAPI metadata (configure_openapi at app creation)
+    openapi_examples.py      # OpenAPI examples built from Pydantic model instances
+    routes/                  # thin HTTP handlers (health, scrape)
+  domain/
+    scrape_service.py        # request-id resolution, URL guardrails, threadpool execution, status mapping
+  engine/
+    orchestrator.py          # ScraperEngine.execute
+    session.py               # ScrapeSession lifecycle
+    budget.py                # wall-clock budget math shared across tiers (elapsed_ms, step budgets)
+    request_tier.py          # HTTP/curl_cffi path
+    browser_tier.py          # Chromium path
+    strategies.py            # NavigationMode resolution, driver helpers
+    driver_capabilities.py   # DriverProtocol + call_if_available adapter
+    envelope.py              # success/error builders, UTF-8 HTML normalization
+  schemas/
+    enums.py                 # ExecutionMode, NavigationMode, ErrorCategory, ...
+    request.py               # ScrapeRequest and validators
+    response.py              # ScrapeSuccess, ScrapeError, HealthResponse, ...
+  infra/                     # telemetry, progress, metadata, xhr, runtime cleanup, sentry
+  security/                  # UrlGuard SSRF guardrails
+scripts/
+  bench_scrape.py            # TestClient wall-time bench for POST /scrape (request tier)
+tests/
+  api/                       # HTTP contract, request schema, 504 timeout envelope tests
+  domain/                    # ScrapeService unit tests (timeout error mapping)
+  engine/                    # ScraperEngine units, isolation regressions, timeout progress
+  infra/                     # challenge, metadata, xhr, progress, sentry, telemetry, request-id, cleanup
+  security/                  # UrlGuard tests
+  support/
+    http.py                  # test_client() context manager + dependency_overrides helper
+    fakes.py                 # shared FakeDriver, FakeRequest, fake_request_cls, ...
+    factories.py             # scrape_request(), example_url()
+  test_bench_regression.py   # lightweight guard that bench script completes (root: guards scripts/)
+```
+
+Layer rules:
+
+| Layer | May import | Must not import |
+| --- | --- | --- |
+| `api/routes` | `domain`, `api/deps`, `schemas.*` | `engine` internals, Botasaurus |
+| `domain` | `engine`, `security`, `schemas.*`, `infra` | FastAPI, Botasaurus |
+| `engine` | `infra`, `security`, `schemas.*`, `config` | FastAPI |
+| `infra` | Botasaurus, CDP (lazy at use sites) | FastAPI, routes |
+
+Conventions:
+
+- Use `tests/support/http.test_client()` for HTTP tests; it runs lifespan and manages `dependency_overrides`. No ad-hoc `TestClient(create_app())` in test modules.
+- Loggers come from `app.logging_config.get_logger()`; do not call `logging.getLogger` with a literal name.
+- Wall-clock/timeout math (elapsed, remaining, step budgets) lives in `app/engine/budget.py`; tiers must not re-derive it.
+- Config: add env vars to `Settings` in `config.py`; call `reset_settings_cache()` in tests that patch env.
+- Wire types live in `app/schemas/` submodules; import directly (`from app.schemas.request import ScrapeRequest`). No long-lived re-export shim.
+- Domain logic stays out of route handlers and Pydantic shells.
+- Typed exceptions over string-matching (`RequestIdCollisionError`, not `RuntimeError` message checks).
+- `NavigationMode` end-to-end in engine code; no raw strategy strings outside enum conversion boundaries.
+- Optional Botasaurus driver methods go through `driver_capabilities.call_if_available` / `resolve_callable` only; do not ad-hoc `getattr(driver, ...)`.
+- OpenAPI route examples come from `openapi_examples.py` model instances, not hand-typed dicts.
+- Route modules expose `create_router()` factories included from `create_app()` **after** `configure_openapi(settings)`, so timeout-dependent response metadata is not frozen on first import.
+- Botasaurus/CDP imports are lazy inside tier entrypoints (`run_request_tier`, `run_browser_tier`, XhrCollector methods), not at app import time.
+
+## Singleton + Settings Threading
+
+- **One** `ScraperEngine` and **one** `ThreadPoolExecutor` are created in `create_app()` lifespan and stored on `app.state`.
+- `get_engine` / `get_executor` / `SettingsDep` read from `request.app.state` (not per-request construction).
+- Lifespan shutdown calls `executor.shutdown(wait=False, cancel_futures=True)` on the real pool instance.
+- Do **not** freeze settings at import time in schemas or OpenAPI modules. `configure_openapi(settings)` runs during `create_app()`; `clamp_wait_timeout_seconds` reads live `get_settings()`.
+- `ScraperEngine` and tier functions require an injected `Settings` parameter; no `get_settings()` fallback in the hot path.
+
+## Isolation Invariants
+
+| Resource | Lifetime | Rule |
+| --- | --- | --- |
+| `ScraperEngine` | process (app.state) | shared |
+| `ThreadPoolExecutor` | process (app.state) | shared, sized by `SCRAPE_MAX_WORKERS` |
+| `_active_request_ids` | in-process memory | shared; collision guard |
+| runtime dir `/tmp/scrape/<request_id>` | per request | isolated; deleted in `finally` |
+| browser profile | per request | isolated; no reuse |
+| Botasaurus Driver | per request | isolated; closed in `finally` |
+
+Multi-worker uvicorn breaks in-process collision detection unless request ids are sticky to a worker. Default to single-worker for isolation semantics.
+
 ## Contract (Do Not Break)
 
 - Endpoints: `GET /health`, `POST /scrape`.
 - `openapi.yaml` is generated from `app.openapi()` via `make openapi`. Do not hand-edit. `make openapi-verify` is part of `make check`. Spectral (`make spectral`) lints the snapshot; do not add a post-processor that mutates the dump.
-- Wire types live in `app/schemas.py`. Engine imports them. Routes `model_dump()` once into `JSONResponse`.
+- Wire types live in `app/schemas/`. Engine imports them. Routes call `ScrapeService.process()` and serialize via `json_response()`.
 - OpenAPI `info.version` is `2.0.0`. Schema names: `ScrapeSuccess` (200) and `ScrapeError` (400/403/422/502/504). No `ScrapeResponse` alias.
 - Success `/scrape` fields: `url`, `final_url`, `status_code`, `headers`, `html`, `metadata_error`, `xhr_responses`, `diagnostics`.
 - When `html` is present, document `headers` `content-type` is `text/html; charset=utf-8` and `html` is UTF-8-normalized.
@@ -36,7 +128,8 @@
   - close browser driver
   - delete request runtime dir
   - remove in-memory active request id
-- Keep request-id collision/invariant guard (`_active_request_ids`) intact.
+- Before each scrape, prune orphaned runtime dirs not tied to an active request id; ENOSPC on profile creation retries after another prune pass. Optional `SCRAPE_RUNTIME_MIN_FREE_BYTES` (default 256MiB) logs when the runtime filesystem is low.
+- Keep request-id collision/invariant guard (`_active_request_ids`) intact; raises `RequestIdCollisionError`.
 - `driver.requests.get` metadata is best-effort; metadata failure must not fail HTML success.
 - Keep strategy engine behavior:
   - `auto` mode attempt order: `google_get` -> `google_get_bypass` -> `get`
@@ -46,14 +139,29 @@
   - keep `/usr/bin/google-chrome` symlink to Chromium for compatibility
 - If browser install logic changes, re-verify binary path and Botasaurus startup.
 
+## Performance
+
+- Baseline bench: `PYTHONPATH=. .venv/bin/python3 scripts/bench_scrape.py --runs 10` (TestClient, `execution_mode=request`).
+- Record p50 wall time when changing hot paths; avoid regressions vs prior baseline.
+- On low-RAM hosts (Docker `--memory=768m`, 1–2 vCPU), tune `SCRAPE_MAX_WORKERS` to `1` or `2`; higher values increase queue wait and swap without improving wall time.
+- Browser tier skips XHR harvest before Cloudflare bypass; one consolidated `collect_page_state` pass runs after bypass.
+
+## Types
+
+- `make typecheck` runs **`pyright` strict** on `app tests` and is part of `make check`.
+- Vendor seams: local stubs in `typings/` (`stubPath` in `pyproject.toml`); CDP shapes in `app/infra/cdp_types.py`.
+- Engine driver seams use `DriverProtocol` / `CdpTabProtocol` in `driver_capabilities.py`; cast vendor `Driver` at construction when needed.
+- Tests construct `ScrapeRequest` via `tests/support/factories.py` (`scrape_request`, `example_url`); fakes implement protocol shapes in `tests/support/fakes.py`.
+- Residual policy and file-level test directives: `docs/typing-residuals.md`.
+
 ## Safety
 
 - Keep SSRF guardrails: localhost/domain checks and blocked IP classes (loopback/private/link-local/multicast/reserved/unspecified).
 - Do not weaken URL validation without explicit request plus docs/tests updates.
 
 ## Done Criteria
 
-- Run `make check` before finish.
+- Run `make check` before finish (lint, test, typecheck, openapi-verify).
 - When Pydantic models or route response metadata change, run `make openapi` and commit the snapshot with the code change.
 - When API contract, Docker behavior, or error semantics change, also run `make smoke`.
 - `make smoke` must cover build, boot, `/health`, `/scrape` happy path, strategy override, retry path, isolation check, localhost guardrail.

Makefile

@@ -1,4 +1,4 @@
-.PHONY: test build serve health scrape-example smoke lint lintfix check ready openapi openapi-verify spectral
+.PHONY: test build serve health scrape-example smoke lint lintfix check ready openapi openapi-verify spectral typecheck
 
 .DEFAULT_GOAL := check
 
@@ -35,13 +35,16 @@ openapi-verify:
 	$(PYTHON) scripts/export_openapi.py --out $$tmp && \
 	diff -u $(OPENAPI_FILE) $$tmp
 
-check: lint test openapi-verify
+check: lint test typecheck openapi-verify
 
 ready: check
 
 test:
 	$(PYTHON) -m unittest discover -s tests
 
+typecheck:
+	$(PYTHON) -m pyright app tests
+
 
 build:
 	docker build -t $(IMAGE) .

README.md

@@ -312,6 +312,7 @@ Exception:
 - Browser profile/session artifacts are request-scoped only.
 - No cache/profile/driver reuse across requests.
 - Cleanup is enforced in `finally`: driver close + runtime directory delete + request-id in-memory state scrub.
+- Before each scrape, orphaned runtime dirs (not tied to an active request id) are pruned; ENOSPC triggers an extra prune-and-retry. Mount `/tmp/scrape` on tmpfs in production (see html2rss-web `docker-compose.yml`).
 
 ## Environment Variables
 
@@ -321,20 +322,21 @@ Use a **separate Sentry project** from html2rss-web (`BOTASAURUS_SENTRY_DSN` →
 
 | Variable | Default | Description |
 | :--- | :--- | :--- |
-| `SENTRY_DSN` | _(unset)_ | Project DSN. |
-| `SENTRY_ENVIRONMENT` | `production` | Deployment tag (`ENVIRONMENT` fallback). |
-| `SENTRY_RELEASE` | _(unset)_ | Release tag on events. |
-| `SENTRY_TRACES_SAMPLE_RATE` | `0.0` | APM traces (off by default; enable later if needed). |
-| `SENTRY_PROFILES_SAMPLE_RATE` | `0.0` | Profiling sample rate. |
-| `SENTRY_SEND_DEFAULT_PII` | `false` | Send default PII when `true`. |
+| `SENTRY_DSN` | _(unset)_ | Project DSN (`Settings.sentry.dsn`). |
+| `SENTRY_ENVIRONMENT` | `production` | Deployment tag (`Settings.sentry.environment`; `ENVIRONMENT` fallback). |
+| `SENTRY_RELEASE` | _(unset)_ | Release tag on events (`Settings.sentry.release`). |
+| `SENTRY_TRACES_SAMPLE_RATE` | `0.0` | APM traces (`Settings.sentry.traces_sample_rate`; off by default). |
+| `SENTRY_PROFILES_SAMPLE_RATE` | `0.0` | Profiling sample rate (`Settings.sentry.profiles_sample_rate`). |
+| `SENTRY_SEND_DEFAULT_PII` | `false` | Send default PII when `true` (`Settings.sentry.send_default_pii`). |
 
 **Signal routing:** `navigation_error` and `timeout` → grouped Sentry Issues. `challenge_block` → `scrape.challenge_block` metric only; engine stdout keeps the detailed log line. Traces stay off unless you raise `SENTRY_TRACES_SAMPLE_RATE`.
 
 | Variable | Default | Description |
 | :--- | :--- | :--- |
-| `SCRAPE_MAX_WORKERS` | `4` | Threadpool worker limit for sync browser execution. |
+| `SCRAPE_MAX_WORKERS` | `4` | Threadpool worker limit for sync browser execution. On low-RAM hosts (for example Docker with `--memory=768m` or a 1–2 vCPU VM), use `1` or `2` to limit concurrent Chromium boots; higher values increase queue wait and swap pressure without improving wall time. |
 | `SCRAPE_TIMEOUT_SECONDS` | `45` | Handler wall-clock budget in seconds (queue, browser boot, and work). |
 | `SCRAPE_WORK_TIMEOUT_SECONDS` | `30` | Post-boot navigate, selector wait, and scroll budget in seconds. |
+| `SCRAPE_RUNTIME_MIN_FREE_BYTES` | `268435456` (256 MiB) | Prune orphan runtime dirs when free space drops below this threshold. |
 
 ## Example Calls
 

app/__init__.py

@@ -0,0 +1 @@
+"""Botasaurus scrape API application package."""

app/api/__init__.py

@@ -0,0 +1 @@
+"""HTTP route registration."""

app/api/deps.py

@@ -0,0 +1,44 @@
+"""FastAPI dependency providers."""
+
+from __future__ import annotations
+
+from concurrent.futures import ThreadPoolExecutor
+from typing import Annotated
+
+from fastapi import Depends, Request
+
+from app.config import Settings
+from app.domain.scrape_service import ScrapeService
+from app.engine import ScraperEngine
+
+
+def get_app_settings(request: Request) -> Settings:
+    return request.app.state.settings
+
+
+SettingsDep = Annotated[Settings, Depends(get_app_settings)]
+
+
+def get_executor(request: Request) -> ThreadPoolExecutor:
+    return request.app.state.executor
+
+
+ExecutorDep = Annotated[ThreadPoolExecutor, Depends(get_executor)]
+
+
+def get_engine(request: Request) -> ScraperEngine:
+    return request.app.state.engine
+
+
+EngineDep = Annotated[ScraperEngine, Depends(get_engine)]
+
+
+def get_scrape_service(
+    settings: SettingsDep,
+    engine: EngineDep,
+    executor: ExecutorDep,
+) -> ScrapeService:
+    return ScrapeService(settings=settings, engine=engine, executor=executor)
+
+
+ScrapeServiceDep = Annotated[ScrapeService, Depends(get_scrape_service)]

app/api/errors.py

@@ -0,0 +1,111 @@
+"""HTTP exception handlers returning scrape error envelopes."""
+
+from __future__ import annotations
+
+from typing import Any, TypedDict, cast
+from urllib.parse import urlparse
+
+from fastapi import FastAPI, Request
+from fastapi.exceptions import RequestValidationError
+from fastapi.responses import JSONResponse
+
+from app.infra.request_id import resolve_request_id
+from app.logging_config import get_logger
+from app.schemas.response import ScrapeError, ScrapeSuccess, validation_error
+
+logger = get_logger()
+
+_NON_FIELD_LOC = {"body", "query", "path", "header"}
+
+
+class ValidationErrorItem(TypedDict, total=False):
+    loc: tuple[Any, ...] | list[Any]
+    msg: str
+    type: str
+    input: Any
+
+
+def schema_field_from_loc(loc: tuple[Any, ...] | list[Any]) -> str:
+    for part in loc:
+        if part not in _NON_FIELD_LOC:
+            return str(part)
+    return str(loc[-1]) if loc else "unknown"
+
+
+def first_schema_field(errors: list[ValidationErrorItem]) -> str:
+    if not errors:
+        return "unknown"
+    return schema_field_from_loc(errors[0].get("loc") or ())
+
+
+def url_from_validation_body(body: Any) -> str:
+    if isinstance(body, dict):
+        url_value = cast(dict[str, Any], body).get("url")
+        if url_value is not None:
+            return str(url_value)
+    return ""
+
+
+def validation_error_message(errors: list[ValidationErrorItem]) -> str:
+    if not errors:
+        return "Request schema validation failed"
+    parts: list[str] = []
+    for err in errors:
+        loc = err.get("loc") or ()
+        field = schema_field_from_loc(loc)
+        message = str(err.get("msg") or "invalid")
+        parts.append(f"{field}: {message}")
+    return "; ".join(parts)
+
+
+def json_response(
+    body: ScrapeSuccess | ScrapeError, *, status_code: int
+) -> JSONResponse:
+    return JSONResponse(status_code=status_code, content=body.model_dump(mode="json"))
+
+
+async def request_schema_validation_handler(
+    request: Request, exc: RequestValidationError
+) -> JSONResponse:
+    errors: list[ValidationErrorItem] = list(exc.errors())  # type: ignore[arg-type]
+    url = url_from_validation_body(exc.body)
+    field = first_schema_field(errors)
+    request_id, _ = resolve_request_id(
+        request.headers.get("X-Request-Id"),
+        host=urlparse(url).hostname if url else None,
+    )
+    logger.info(
+        "request_schema_422 host=%s field=%s",
+        urlparse(url).hostname if url else None,
+        field,
+    )
+    return json_response(
+        validation_error(
+            url,
+            validation_error_message(errors),
+            request_id=request_id,
+        ),
+        status_code=422,
+    )
+
+
+async def unhandled_exception_handler(request: Request, exc: Exception) -> JSONResponse:
+    del exc
+    logger.exception("unhandled_exception path=%s", request.url.path)
+    request_id, _ = resolve_request_id(request.headers.get("X-Request-Id"))
+    return json_response(
+        validation_error(
+            "",
+            "Internal server error",
+            request_id=request_id,
+        ),
+        status_code=500,
+    )
+
+
+def register_exception_handlers(app: FastAPI) -> None:
+    app.add_exception_handler(
+        RequestValidationError,
+        cast(Any, request_schema_validation_handler),
+    )
+    app.add_exception_handler(Exception, unhandled_exception_handler)