phase5-pipeline

CLAUDE.md

@@ -196,3 +196,8 @@ pytest -q
 - LinkedIn rate-limits aggressively; keep batches small while testing.
 - Standard pip struggles with pydantic dependency resolution in this stack; always use uv pip install instead.
 - The system Python is protected by PEP 668 (externally-managed-environment). Always use explicit virtual environment paths (e.g., .venv/bin/python, .venv/bin/pytest) for all terminal commands instead of relying on global commands.
+- **browser-use 0.8.0 `executable_path` required on Linux (confirmed 2026-06-17)**: `LocalBrowserWatchdog._find_installed_browser_path()` globs `chrome-linux/chrome` but playwright ≥1.40 installs to `chrome-linux64/chrome`. The watchdog silently falls through to `uvx playwright install chrome --with-deps`, which fails if `uvx` is not on PATH (`FileNotFoundError: 'uvx'`). Fix: pass `executable_path=_find_chromium_executable()` to `Browser()` in `_agent_session` — this makes the watchdog skip the auto-install entirely. `_find_chromium_executable()` in `agent_fallback.py` includes the `chrome-linux64` pattern.
+- **browser-use 0.8.0 "Result failed" JSON errors are non-fatal step-level retries** (observed 2026-06-17): lines like `❌ Result failed 1/4 times: 1 validation error for AgentOutput Invalid JSON: trailing characters` mean the model's tool-call response had trailing characters and browser-use retried (up to 4 times per step). These do not abort the agent run; they appear in `history.errors()` but `is_done()` can still be `True`. Expect a handful per run.
+- **Browser-agent run times: 2–4 minutes per company** (Infosys 21 steps/149s, TCS 24 steps/195s, tested 2026-06-17). This tier is a true last resort — only fires when all static tiers miss. The agent often uses a search engine (DuckDuckGo/Bing) when direct navigation fails, adding several extra steps. Google is rate-limited (CAPTCHA) and the agent falls back to Bing automatically.
+- **Browser-agent partial result pattern (TCS, 2026-06-17)**: `is_successful()=False` with `careers_url` populated but `position_url=None`. TCS uses a custom ATS (iBegin) that requires login/registration before individual job listings are accessible; the agent found the portal but could not navigate to a specific posting. Such records land in `needs_review` as expected. Login-gated ATSes are a known hard limit of the static-HTML tiers and the browser agent alike.
+- **Browser-agent live results (2026-06-17)**: Infosys → `careers_url=https://career.infosys.com/joblist` + `position_url=https://career.infosys.com/jobdesc?jobReferenceCode=INFSYS-EXTERNAL-247232` (both HTTP 200). TCS → `careers_url=https://ibegin.tcsapps.com/candidate/` (HTTP 200) + `position_url=None`. Both sites had failed all heuristic tiers (403 homepage / custom subdomain blocking httpx).

jobsource/agent_fallback.py

@@ -1,11 +1,301 @@
 """Browser Use fused fallback: find careers page AND extract one job URL in one session.
 
-Scaffold stub -- not implemented yet.
+This is the LAST tier of the cascade (Stage 2, Tier 6 / Stage 3, Tier 5). It fires only
+when all cheaper tiers in cascade.py and extract.py have missed. A single Browser Use
+agent session does both:
+  1. Navigate to the company website and locate the careers/jobs page.
+  2. From that page, open one specific open-position URL.
+
+Both URLs are returned in one call — never two browser sessions for the same company.
+A module-level memo keyed by normalized host ensures a same-domain second call (e.g.
+cascade passes the root website, extract passes the careers URL on the same host) is a
+cache hit with no new browser launch.
+
+Graceful degradation:
+  - ENABLE_BROWSER_AGENT=false       → log INFO, return None
+  - Placeholder AGENT_MODEL/LLM key  → log INFO, return None
+  - browser_use import error         → log WARNING, return None
+  - Chromium/navigation/agent error  → log WARNING, return None
+In all cases the batch still completes; those records land in needs_review.
 """
-# TODO (Stage 2/3 last resort): implement per CLAUDE.md "Stage 2 — tier 6" and "Stage 3 — tier 5".
+from __future__ import annotations
-# This is the LAST tier of the cascade. Fires only when all cheaper tiers in cascade.py
+
-# and extract.py have failed. One Browser Use agent session does both:
+import asyncio
-#   1. Navigate to the company website and locate the careers/jobs page.
+import glob
-#   2. From the careers page, return the URL of one open position.
+import logging
-# Graceful degradation: if Browser Use / Playwright / LLM key are unavailable, log clearly
+import os
-# and return (careers_url=None, position_url=None) so the pipeline records needs_review.
+import pathlib
+import platform
+from concurrent.futures import ThreadPoolExecutor
+from urllib.parse import urlsplit
+
+from pydantic import BaseModel
+
+from .config import get_settings
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# Result models
+# ---------------------------------------------------------------------------
+
+
+class AgentFallbackResult(BaseModel):
+    """Public result of find_and_extract()."""
+
+    careers_url: str | None = None
+    position_url: str | None = None
+
+
+class _AgentOutput(BaseModel):
+    """Structured-output schema handed to the Browser Use Agent."""
+
+    careers_url: str | None = None
+    position_url: str | None = None
+
+
+# ---------------------------------------------------------------------------
+# Module-level memo — one entry per normalized host
+# ---------------------------------------------------------------------------
+
+_MEMO: dict[str, AgentFallbackResult | None] = {}
+
+# Max browser-agent steps before timing out.
+_MAX_STEPS = 30
+
+# Task prompt for the fused Stage-2 + Stage-3 session.
+_TASK_TEMPLATE = (
+    "You are helping a job-sourcing pipeline. Given a company website URL, do the following "
+    "in a single browsing session:\n"
+    "1. Navigate to {website}\n"
+    "2. Find the company's official careers or jobs page (the page that lists open positions). "
+    "   Explore navigation menus, footer links, or search for '/careers', '/jobs', etc.\n"
+    "3. Once on the careers page, open one specific open-position detail page "
+    "   (a page for exactly one role — not a listing index, filter, or category page).\n"
+    "4. Return both URLs in the structured output fields:\n"
+    "   - careers_url: the URL of the careers/jobs listing page\n"
+    "   - position_url: the URL of one specific open-position detail page\n"
+    "If you cannot find a careers page or any open positions, set the corresponding field to null."
+)
+
+
+# ---------------------------------------------------------------------------
+# Availability gate
+# ---------------------------------------------------------------------------
+
+
+def _agent_unavailable() -> bool:
+    """Return True when the browser agent is not configured (placeholder values)."""
+    s = get_settings()
+    return (
+        s.agent_model.startswith("PLACEHOLDER")
+        or s.llm_api_key.startswith("PLACEHOLDER")
+    )
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _site_key(url: str) -> str:
+    """Normalized netloc for memo keying — strips 'www.', lowercased."""
+    host = urlsplit(url).netloc.lower()
+    return host.removeprefix("www.")
+
+
+def _find_chromium_executable() -> str | None:
+    """Return the Playwright-installed Chromium binary path, or None if not found.
+
+    browser-use 0.8.0's LocalBrowserWatchdog only globs `chrome-linux/chrome` but
+    playwright ≥1.40 installs to `chrome-linux64/chrome`. Passing executable_path
+    explicitly bypasses the watchdog's uvx-based auto-install fallback.
+    """
+    pw_base = os.environ.get("PLAYWRIGHT_BROWSERS_PATH")
+    if not pw_base:
+        system = platform.system()
+        if system == "Darwin":
+            pw_base = str(pathlib.Path.home() / "Library" / "Caches" / "ms-playwright")
+        elif system == "Windows":
+            pw_base = str(pathlib.Path.home() / "AppData" / "Local" / "ms-playwright")
+        else:
+            pw_base = str(pathlib.Path.home() / ".cache" / "ms-playwright")
+
+    system = platform.system()
+    if system == "Darwin":
+        patterns = [
+            f"{pw_base}/chromium-*/chrome-mac/Chromium.app/Contents/MacOS/Chromium",
+            f"{pw_base}/chromium-*/chrome-mac-arm64/Chromium.app/Contents/MacOS/Chromium",
+        ]
+    elif system == "Windows":
+        patterns = [
+            f"{pw_base}\\chromium-*\\chrome-win\\chrome.exe",
+            f"{pw_base}\\chromium-*\\chrome-win64\\chrome.exe",
+        ]
+    else:
+        patterns = [
+            f"{pw_base}/chromium-*/chrome-linux64/chrome",
+            f"{pw_base}/chromium-*/chrome-linux/chrome",
+        ]
+
+    for pattern in patterns:
+        matches = sorted(glob.glob(pattern))
+        if matches:
+            return matches[-1]  # newest version last alphabetically
+
+    return None
+
+
+def _build_chat_model():  # type: ignore[return]
+    """Construct the chat-model instance for the Browser Use agent.
+
+    Test seam — monkeypatch this to inject a fake without touching Browser Use.
+    Lazy-imports browser_use so a missing package never crashes module import.
+
+    agent_model format: "provider:model_id" (e.g. "anthropic:claude-sonnet-4-6").
+    Unknown/absent provider prefix defaults to ChatOpenAI.
+    """
+    from browser_use import ChatAnthropic, ChatOpenAI  # lazy import
+
+    s = get_settings()
+    model_str: str = s.agent_model
+    api_key: str | None = None if s.llm_api_key.startswith("PLACEHOLDER") else s.llm_api_key
+
+    if ":" in model_str:
+        provider, model_id = model_str.split(":", 1)
+    else:
+        provider, model_id = "openai", model_str
+
+    if provider == "anthropic":
+        kwargs = {"model": model_id}
+        if api_key:
+            kwargs["api_key"] = api_key
+        return ChatAnthropic(**kwargs)
+    else:
+        kwargs = {"model": model_id}
+        if api_key:
+            kwargs["openai_api_key"] = api_key
+        return ChatOpenAI(**kwargs)
+
+
+async def _agent_session(website: str, chat) -> _AgentOutput | None:  # type: ignore[return]
+    """Run a single Browser Use session; return structured output or None."""
+    from browser_use import Agent, Browser  # lazy import
+
+    s = get_settings()
+    # Pass executable_path explicitly so browser-use skips its uvx-based auto-installer.
+    # browser-use 0.8.0's LocalBrowserWatchdog globs chrome-linux/chrome but playwright
+    # ≥1.40 installs to chrome-linux64/chrome — the two paths diverge and the watchdog
+    # falls through to `uvx playwright install` even when chromium is present.
+    chromium_exe = _find_chromium_executable()
+    browser = Browser(headless=s.browser_headless, executable_path=chromium_exe)
+    task = _TASK_TEMPLATE.format(website=website)
+    agent = Agent(
+        task=task,
+        llm=chat,
+        browser=browser,
+        output_model_schema=_AgentOutput,
+    )
+    try:
+        history = await agent.run(max_steps=_MAX_STEPS)
+        out = history.structured_output
+        if isinstance(out, _AgentOutput):
+            return out
+        return None
+    finally:
+        try:
+            await browser.close()
+        except Exception:
+            pass
+
+
+def _run_blocking(coro_factory) -> _AgentOutput | None:  # type: ignore[return]
+    """Run an async coroutine from synchronous code.
+
+    Uses asyncio.run() when no event loop is running (the normal CLI path).
+    Falls back to a one-shot ThreadPoolExecutor when called from inside an
+    already-running loop (e.g. an async Prefect flow).
+    """
+    try:
+        asyncio.get_running_loop()
+        # Already inside a running loop — offload to a thread.
+        with ThreadPoolExecutor(max_workers=1) as pool:
+            future = pool.submit(asyncio.run, coro_factory())
+            return future.result()
+    except RuntimeError:
+        # No running loop — safe to call asyncio.run() directly.
+        return asyncio.run(coro_factory())
+
+
+def _valid_url(val: str | None) -> str | None:
+    """Return val if it's an http(s) URL, else None."""
+    if val and isinstance(val, str) and val.startswith(("http://", "https://")):
+        return val
+    return None
+
+
+# ---------------------------------------------------------------------------
+# Public entry point
+# ---------------------------------------------------------------------------
+
+
+def find_and_extract(website: str) -> AgentFallbackResult | None:
+    """Fused Stage-2 + Stage-3 fallback: browser agent finds careers page AND one job URL.
+
+    Returns an AgentFallbackResult on success, or None on any graceful failure.
+    Never raises — all errors are logged and swallowed so the pipeline continues.
+    Results are memoized per normalized host so a same-domain second call is free.
+    """
+    s = get_settings()
+
+    if not s.enable_browser_agent:
+        logger.info("agent_fallback(%s): ENABLE_BROWSER_AGENT=false, skipping", website)
+        return None
+
+    if _agent_unavailable():
+        logger.info(
+            "agent_fallback(%s): agent_model or llm_api_key not configured, skipping", website
+        )
+        return None
+
+    key = _site_key(website)
+    if key in _MEMO:
+        logger.info("agent_fallback(%s): memo hit, returning cached result", website)
+        return _MEMO[key]
+
+    # Build the chat model (lazy import — handles missing browser_use gracefully).
+    try:
+        chat = _build_chat_model()
+    except Exception as exc:
+        logger.warning("agent_fallback(%s): failed to build chat model: %s", website, exc)
+        _MEMO[key] = None
+        return None
+
+    # Run the fused browser session.
+    try:
+        raw: _AgentOutput | None = _run_blocking(lambda: _agent_session(website, chat))
+    except Exception as exc:
+        logger.warning("agent_fallback(%s): agent session error: %s", website, exc)
+        _MEMO[key] = None
+        return None
+
+    if raw is None:
+        logger.info("agent_fallback(%s): agent returned no structured output", website)
+        _MEMO[key] = None
+        return None
+
+    result = AgentFallbackResult(
+        careers_url=_valid_url(raw.careers_url),
+        position_url=_valid_url(raw.position_url),
+    )
+
+    if result.careers_url or result.position_url:
+        logger.info(
+            "agent_fallback(%s): careers_url=%s position_url=%s",
+            website, result.careers_url, result.position_url,
+        )
+    else:
+        logger.info("agent_fallback(%s): agent found no usable URLs", website)
+
+    _MEMO[key] = result
+    return result

jobsource/careers/cascade.py

@@ -2,14 +2,16 @@
 
 Cascade order (return early on first success):
   1.  ATS detection  → ats.detect_and_fetch()              confidence 0.95
-  2. URL patterns  → heuristics.probe_url_patterns()  0.80
+  1b. ATS slug-guess → ats.recover_via_slug_guess()        confidence 0.90
-  3. Homepage scan → heuristics.scan_homepage_links()  0.60
+  2.  URL patterns   → heuristics.probe_url_patterns()     confidence 0.80
-  4. Sitemap       → heuristics.parse_sitemap()        0.50
+  3.  Homepage scan  → heuristics.scan_homepage_links()    confidence 0.60
-  5. Cheap-LLM     → classify_llm (stub, not implemented in this phase)
+  4.  Sitemap        → heuristics.parse_sitemap()          confidence 0.50
-  6. Browser agent → agent_fallback (stub, not implemented in this phase)
+  5.  Cheap-LLM      → classify_llm.classify_careers_link()  confidence 0.55
+  6.  Browser agent  → agent_fallback.find_and_extract()   confidence 0.50
 
 Returns a CareersResult with the URL, confidence, method string, and — when
-the ATS tier resolves — the first open-position URL for free (Stage-3 shortcut).
+the ATS tier or browser-agent tier resolves — the first open-position URL for
+free (Stage-3 shortcut).
 
 The optional *client* parameter follows the managed-client pattern from
 resolve.py: supply an existing httpx.Client to reuse connections; otherwise a
@@ -22,8 +24,11 @@ import logging
 import httpx
 from pydantic import BaseModel
 
+from ..config import get_settings
 from ..http import build_client, request_with_retries
+from .. import agent_fallback as _agent_fallback
 from . import ats as _ats
+from . import classify_llm as _classify_llm
 from . import heuristics as _heuristics
 
 logger = logging.getLogger(__name__)
@@ -39,7 +44,7 @@ class CareersResult(BaseModel):
 
     careers_url: str | None = None
     confidence: float = 0.0
-    # method values: "ats:{name}" | "url_pattern" | "homepage_scan" | "sitemap" | "none"
+    # method values: "ats:{name}" | "url_pattern" | "homepage_scan" | "sitemap" | "llm_classify" | "browser_agent" | "none"
     method: str = "none"
     ats_name: str | None = None
     # Free Stage-3 shortcut: populated when ATS tier resolves (first open job URL).
@@ -145,8 +150,43 @@ def find_careers_page(
         except Exception as exc:
             logger.warning("cascade(%s): sitemap tier error: %s", website, exc)
 
-        # All deterministic tiers missed.
+        # ------------------------------------------------------------------
-        logger.info("cascade(%s): all deterministic tiers missed", website)
+        # Tier 5 — Cheap-LLM careers-link classification
+        # ------------------------------------------------------------------
+        try:
+            if homepage_html:
+                anchors = _classify_llm.extract_anchors(homepage_html, website)
+                res = _classify_llm.classify_careers_link(anchors)
+                if res:
+                    logger.info(
+                        "cascade(%s): resolved via llm_classify careers_url=%s confidence=%.2f",
+                        website, res.careers_url, res.confidence,
+                    )
+                    return _finalize(res.careers_url, "llm_classify", 0.55, website, client)
+        except Exception as exc:
+            logger.warning("cascade(%s): llm_classify tier error: %s", website, exc)
+
+        # ------------------------------------------------------------------
+        # Tier 6 — Browser agent (fused Stage-2 + Stage-3 fallback)
+        # ------------------------------------------------------------------
+        if get_settings().enable_browser_agent:
+            try:
+                ag = _agent_fallback.find_and_extract(website)
+                if ag and ag.careers_url:
+                    logger.info(
+                        "cascade(%s): resolved via browser_agent careers_url=%s",
+                        website, ag.careers_url,
+                    )
+                    return CareersResult(
+                        careers_url=ag.careers_url,
+                        confidence=0.50,
+                        method="browser_agent",
+                        position_url=ag.position_url,
+                    )
+            except Exception as exc:
+                logger.warning("cascade(%s): browser_agent tier error: %s", website, exc)
+
+        logger.info("cascade(%s): all tiers missed", website)
         return CareersResult()
 
     finally:

jobsource/careers/classify_llm.py

@@ -1,13 +1,235 @@
 """Cheap-LLM link classification for careers page and job links (Stage 2, tier 5 / Stage 3, tier 4).
 
-Scaffold stub -- not implemented yet.
+Two typed classification functions backed by Pydantic AI:
+  1. classify_careers_link(anchors) -> CareerLinkResult | None
+     Given a list of (url, text) pairs from a page, pick the single careers/jobs page URL.
+  2. classify_job_link(anchors) -> JobLinkResult | None
+     Given a list of (url, text) pairs from a careers page, pick one specific open-position URL.
+
+Both return None and the cascade falls through gracefully when:
+  - classifier_model or llm_api_key is a placeholder (LLM not configured),
+  - the anchor list is empty,
+  - the LLM call fails for any reason,
+  - the model abstains (index -1) or returns an out-of-range index.
+
+pydantic_ai is imported lazily inside _build_link_agent so a missing package or
+unconfigured provider never breaks module import (graceful-degradation rule #8).
+
+extract_anchors() is also exported here so cascade.py and extract.py can share
+the same anchor-parsing logic without a separate utility module.
 """
-# TODO (Stage 2 tier 5 / Stage 3 tier 4): implement per CLAUDE.md "Cheap-LLM classification".
+from __future__ import annotations
-# Uses Pydantic AI (model-agnostic) with the `classifier_model` from config.
+
-# Two typed tasks:
+import logging
-#   1. classify_careers_link(anchors: list[Anchor]) -> CareerLinkResult
+from urllib.parse import urljoin
-#      Given extracted <a> tags from a page, pick the careers/jobs page URL.
+
-#   2. classify_job_link(anchors: list[Anchor]) -> JobLinkResult
+from bs4 import BeautifulSoup
-#      Given extracted <a> tags from a careers page, pick one open-position URL.
+from pydantic import BaseModel, Field
-# Both return a typed Pydantic result including the chosen URL and confidence.
+
-# Graceful degradation: if llm_api_key is placeholder or call fails, return None.
+from ..config import get_settings
+
+logger = logging.getLogger(__name__)
+
+# Maximum number of anchors forwarded to the LLM (token-budget guard).
+_MAX_ANCHORS = 60
+# Maximum anchor text length, in characters, before truncation.
+_MAX_ANCHOR_TEXT = 80
+
+
+# ---------------------------------------------------------------------------
+# Data models
+# ---------------------------------------------------------------------------
+
+
+class Anchor(BaseModel):
+    """One hyperlink extracted from a page — the LLM's input row."""
+
+    url: str
+    text: str = ""
+
+
+class _LinkChoice(BaseModel):
+    """LLM structured output: 0-based index into the candidate list + confidence."""
+
+    index: int = Field(
+        description=(
+            "0-based index of the best link in the candidate list. "
+            "Use -1 if no link qualifies."
+        )
+    )
+    confidence: float = Field(default=0.0, ge=0.0, le=1.0)
+
+
+class CareerLinkResult(BaseModel):
+    """Public result of classify_careers_link()."""
+
+    careers_url: str
+    confidence: float = 0.0
+
+
+class JobLinkResult(BaseModel):
+    """Public result of classify_job_link()."""
+
+    job_url: str
+    confidence: float = 0.0
+
+
+# ---------------------------------------------------------------------------
+# System-level prompt instructions for each classification task
+# ---------------------------------------------------------------------------
+
+_CAREERS_INSTRUCTIONS = (
+    "You are a link classifier for a job-sourcing pipeline. "
+    "You receive a numbered list of hyperlinks extracted from a company website. "
+    "Identify the single link that leads to the company's main careers or jobs page "
+    "(the page that lists open positions). "
+    "Respond with the 0-based index of the best match and your confidence (0.0–1.0). "
+    "If none of the links leads to a careers or jobs page, respond with index -1."
+)
+
+_JOB_INSTRUCTIONS = (
+    "You are a link classifier for a job-sourcing pipeline. "
+    "You receive a numbered list of hyperlinks extracted from a careers/jobs page. "
+    "Identify the single link that leads directly to one specific open job-posting detail page "
+    "(a page for exactly one position — not a category page, a filter, or a job listing index). "
+    "Respond with the 0-based index of the best match and your confidence (0.0–1.0). "
+    "If none of the links qualifies as a single-job detail page, respond with index -1."
+)
+
+
+# ---------------------------------------------------------------------------
+# Shared anchor extractor (used by cascade.py Tier 5 and extract.py Tiers 3-4)
+# ---------------------------------------------------------------------------
+
+
+def extract_anchors(html: str, base_url: str) -> list[Anchor]:
+    """Extract deduplicated absolute hyperlinks from HTML.
+
+    Skips mailto:, tel:, # (fragment-only), and javascript: hrefs.
+    Converts relative hrefs to absolute using base_url.
+    Returns only http(s) links in first-seen order, URL-deduplicated.
+    Truncates anchor text to _MAX_ANCHOR_TEXT characters.
+    """
+    if not html:
+        return []
+    soup = BeautifulSoup(html, "lxml")
+    seen_urls: set[str] = set()
+    result: list[Anchor] = []
+    for tag in soup.find_all("a", href=True):
+        href: str = tag["href"].strip()
+        if not href or href.startswith(("mailto:", "tel:", "#", "javascript:")):
+            continue
+        full_url = urljoin(base_url, href)
+        if not full_url.startswith(("http://", "https://")):
+            continue
+        if full_url in seen_urls:
+            continue
+        seen_urls.add(full_url)
+        text = tag.get_text(separator=" ", strip=True)[:_MAX_ANCHOR_TEXT]
+        result.append(Anchor(url=full_url, text=text))
+    return result
+
+
+# ---------------------------------------------------------------------------
+# LLM availability gate
+# ---------------------------------------------------------------------------
+
+
+def _llm_unavailable() -> bool:
+    """Return True when the LLM is not configured (placeholder values in settings)."""
+    s = get_settings()
+    return (
+        s.classifier_model.startswith("PLACEHOLDER")
+        or s.llm_api_key.startswith("PLACEHOLDER")
+    )
+
+
+# ---------------------------------------------------------------------------
+# Agent builder — extracted as a named function to serve as the test seam.
+# Monkeypatch this function in tests to avoid real LLM calls.
+# ---------------------------------------------------------------------------
+
+
+def _build_link_agent(instructions: str):  # type: ignore[return]
+    """Construct a Pydantic AI Agent for link classification.
+
+    pydantic_ai is imported lazily here so a missing package never crashes
+    module import. This function is the primary test seam — monkeypatch it
+    to inject a FakeAgent without touching the real Pydantic AI machinery.
+    """
+    from pydantic_ai import Agent  # lazy import
+
+    return Agent(
+        get_settings().classifier_model,
+        output_type=_LinkChoice,
+        instructions=instructions,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Core shared classifier
+# ---------------------------------------------------------------------------
+
+
+def _render_candidates(anchors: list[Anchor]) -> str:
+    """Format anchors as a numbered list for the LLM prompt."""
+    lines = [f"[{i}] {a.url}  |  {a.text}" for i, a in enumerate(anchors)]
+    return "\n".join(lines)
+
+
+def _choose_link(
+    anchors: list[Anchor],
+    *,
+    instructions: str,
+    label: str,
+) -> "_LinkChoice | None":
+    """Run the LLM classifier on anchors; return a _LinkChoice or None.
+
+    Returns None when the LLM is unconfigured, the list is empty, the model
+    abstains (index -1), the index is out of range, or any exception occurs.
+    """
+    if not anchors or _llm_unavailable():
+        return None
+
+    candidates = anchors[:_MAX_ANCHORS]
+    try:
+        agent = _build_link_agent(instructions)
+        prompt = _render_candidates(candidates)
+        out = agent.run_sync(prompt).output
+        if isinstance(out, _LinkChoice) and 0 <= out.index < len(candidates):
+            return out
+        return None  # abstain (index -1) or out-of-range
+    except Exception as exc:
+        logger.warning("classify_llm(%s): tier error: %s", label, exc)
+        return None
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+
+def classify_careers_link(anchors: list[Anchor]) -> CareerLinkResult | None:
+    """Pick the careers/jobs page URL from a list of extracted anchors (Stage 2, tier 5).
+
+    Returns a CareerLinkResult on success, or None on any no-op / failure.
+    Never raises.
+    """
+    c = _choose_link(anchors, instructions=_CAREERS_INSTRUCTIONS, label="careers")
+    if c is None:
+        return None
+    candidates = anchors[:_MAX_ANCHORS]
+    return CareerLinkResult(careers_url=candidates[c.index].url, confidence=c.confidence)
+
+
+def classify_job_link(anchors: list[Anchor]) -> JobLinkResult | None:
+    """Pick one specific open-position URL from a careers-page anchor list (Stage 3, tier 4).
+
+    Returns a JobLinkResult on success, or None on any no-op / failure.
+    Never raises.
+    """
+    c = _choose_link(anchors, instructions=_JOB_INSTRUCTIONS, label="job")
+    if c is None:
+        return None
+    candidates = anchors[:_MAX_ANCHORS]
+    return JobLinkResult(job_url=candidates[c.index].url, confidence=c.confidence)

jobsource/db.py

@@ -1,10 +1,297 @@
-"""SQLite persistence layer: companies table, jobs table, dedup, company cache, CSV export.
+"""SQLite persistence layer: companies table, jobs table, dedup, company cache, CSV export."""
+from __future__ import annotations
 
-Scaffold stub -- not implemented yet.
+import csv
+import logging
+import sqlite3
+from datetime import datetime, timezone
+from pathlib import Path
+from urllib.parse import urlsplit
+
+from .config import get_settings
+from .models import CSV_COLUMNS, JobResult, JobStatus
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Pure helper — standalone so tests can import it directly
+# ---------------------------------------------------------------------------
+
+
+def _company_key(name: str, website: str | None) -> str:
+    """Derive a stable company key: registrable domain if website known, else name.
+
+    Examples:
+        _company_key("Acme Inc", "https://www.acme.com/about") -> "acme.com"
+        _company_key("Acme LLC", None)                         -> "acme"
     """
-# TODO (Stage 4): implement per CLAUDE.md "Stage 4 — Persist & export" and "Data model".
+    if website:
-# Schema:
+        try:
-#   companies(company_key PK, name, website, career_url, first_seen)
+            hostname = urlsplit(website).hostname or ""
-#   jobs(job_id PK, company_key, linkedin_url, position_url, status, listed_at, first_seen)
+            hostname = hostname.lower()
-# CSV export writes output/results.csv with columns: company_name, career_page_url, open_position_url
+            if hostname.startswith("www."):
-# (complete rows — status==position_found — sorted first; incomplete rows follow).
+                hostname = hostname[4:]
+            if hostname:
+                return hostname
+        except Exception:
+            pass
+    slug = name.strip().lower()
+    return slug or "unknown"
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+# ---------------------------------------------------------------------------
+# Database
+# ---------------------------------------------------------------------------
+
+
+class Database:
+    """SQLite-backed store for companies and jobs.
+
+    Supports use as a context manager. Thread-safety: one connection per
+    instance (single-threaded pipeline use). Postgres-swap is possible by
+    replacing this class behind the same interface.
+    """
+
+    def __init__(self, path: Path | str | None = None) -> None:
+        if path is None:
+            path = get_settings().db_path
+        path = Path(path)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        self._conn = sqlite3.connect(str(path))
+        self._conn.row_factory = sqlite3.Row
+        # WAL mode for better write performance (no-op on in-memory dbs)
+        self._conn.execute("PRAGMA journal_mode=WAL")
+        self.init_schema()
+        logger.debug("db: opened %s", path)
+
+    # ---- Schema -----------------------------------------------------------
+
+    def init_schema(self) -> None:
+        """Create tables if they don't exist. Safe to call on every startup."""
+        self._conn.executescript("""
+            CREATE TABLE IF NOT EXISTS companies (
+                company_key TEXT PRIMARY KEY,
+                name        TEXT,
+                website     TEXT,
+                career_url  TEXT,
+                first_seen  TEXT
+            );
+            CREATE TABLE IF NOT EXISTS jobs (
+                job_id       TEXT PRIMARY KEY,
+                company_key  TEXT,
+                linkedin_url TEXT,
+                position_url TEXT,
+                status       TEXT,
+                listed_at    TEXT,
+                first_seen   TEXT
+            );
+        """)
+        self._conn.commit()
+
+    # ---- Dedup / cache ----------------------------------------------------
+
+    def is_seen(self, job_id: str) -> bool:
+        """Return True if this job_id already exists in the DB (cross-run dedup)."""
+        row = self._conn.execute(
+            "SELECT 1 FROM jobs WHERE job_id = ?", (job_id,)
+        ).fetchone()
+        return row is not None
+
+    def get_cached_career_url(self, company_key: str) -> str | None:
+        """Return a previously resolved career URL for this company, or None.
+
+        When non-null, the pipeline skips the cascade for this company and uses
+        the cached value directly (companies are resolved at most once).
+        """
+        row = self._conn.execute(
+            "SELECT career_url FROM companies WHERE company_key = ?", (company_key,)
+        ).fetchone()
+        if row and row["career_url"]:
+            return row["career_url"]
+        return None
+
+    # ---- Company key helper -----------------------------------------------
+
+    def company_key_for(self, name: str, website: str | None) -> str:
+        """Delegate to the module-level pure function."""
+        return _company_key(name, website)
+
+    # ---- Upserts ----------------------------------------------------------
+
+    def upsert_company(
+        self,
+        company_key: str,
+        name: str,
+        website: str | None,
+        career_url: str | None,
+    ) -> None:
+        """Insert or update a company row.
+
+        COALESCE semantics: a new non-null value fills an existing NULL, but
+        never clobbers an already-populated column. first_seen is preserved.
+        """
+        now = _now_iso()
+        self._conn.execute(
+            """
+            INSERT INTO companies(company_key, name, website, career_url, first_seen)
+            VALUES (?, ?, ?, ?, ?)
+            ON CONFLICT(company_key) DO UPDATE SET
+                name       = COALESCE(excluded.name,       companies.name),
+                website    = COALESCE(excluded.website,    companies.website),
+                career_url = COALESCE(excluded.career_url, companies.career_url),
+                first_seen = companies.first_seen
+            """,
+            (company_key, name, website, career_url, now),
+        )
+        self._conn.commit()
+
+    def upsert_job(self, result: JobResult) -> None:
+        """Insert or update a job row.
+
+        first_seen is preserved on conflict. position_url and linkedin_url
+        use COALESCE: a resolved value fills a prior NULL but is never
+        overwritten once set.
+        """
+        now = _now_iso()
+        listed_at = (
+            result.listed_at.isoformat() if result.listed_at is not None else None
+        )
+        self._conn.execute(
+            """
+            INSERT INTO jobs(job_id, company_key, linkedin_url, position_url, status, listed_at, first_seen)
+            VALUES (?, ?, ?, ?, ?, ?, ?)
+            ON CONFLICT(job_id) DO UPDATE SET
+                company_key  = excluded.company_key,
+                linkedin_url = COALESCE(excluded.linkedin_url, jobs.linkedin_url),
+                position_url = COALESCE(excluded.position_url, jobs.position_url),
+                status       = excluded.status,
+                listed_at    = COALESCE(excluded.listed_at,    jobs.listed_at),
+                first_seen   = jobs.first_seen
+            """,
+            (
+                result.job_id,
+                result.company_key,
+                result.linkedin_url,
+                result.open_position_url,
+                result.status.value,
+                listed_at,
+                now,
+            ),
+        )
+        self._conn.commit()
+
+    def persist_result(self, result: JobResult) -> None:
+        """Persist one JobResult atomically: upsert company then job."""
+        with self._conn:
+            now = _now_iso()
+            listed_at = (
+                result.listed_at.isoformat() if result.listed_at is not None else None
+            )
+            self._conn.execute(
+                """
+                INSERT INTO companies(company_key, name, website, career_url, first_seen)
+                VALUES (?, ?, ?, ?, ?)
+                ON CONFLICT(company_key) DO UPDATE SET
+                    name       = COALESCE(excluded.name,       companies.name),
+                    website    = COALESCE(excluded.website,    companies.website),
+                    career_url = COALESCE(excluded.career_url, companies.career_url),
+                    first_seen = companies.first_seen
+                """,
+                (
+                    result.company_key,
+                    result.company_name,
+                    result.website,
+                    result.career_page_url,
+                    now,
+                ),
+            )
+            self._conn.execute(
+                """
+                INSERT INTO jobs(job_id, company_key, linkedin_url, position_url, status, listed_at, first_seen)
+                VALUES (?, ?, ?, ?, ?, ?, ?)
+                ON CONFLICT(job_id) DO UPDATE SET
+                    company_key  = excluded.company_key,
+                    linkedin_url = COALESCE(excluded.linkedin_url, jobs.linkedin_url),
+                    position_url = COALESCE(excluded.position_url, jobs.position_url),
+                    status       = excluded.status,
+                    listed_at    = COALESCE(excluded.listed_at,    jobs.listed_at),
+                    first_seen   = jobs.first_seen
+                """,
+                (
+                    result.job_id,
+                    result.company_key,
+                    result.linkedin_url,
+                    result.open_position_url,
+                    result.status.value,
+                    listed_at,
+                    now,
+                ),
+            )
+
+    # ---- Export -----------------------------------------------------------
+
+    def export_csv(self, path: Path | None = None) -> Path:
+        """Write output/results.csv; complete (position_found) rows sorted first.
+
+        Uses models.CSV_COLUMNS for the header; None values become empty strings.
+        Returns the Path written.
+        """
+        if path is None:
+            path = get_settings().output_csv
+        path = Path(path)
+        path.parent.mkdir(parents=True, exist_ok=True)
+
+        rows = self._conn.execute(
+            """
+            SELECT
+                c.name         AS company_name,
+                c.career_url   AS career_page_url,
+                j.position_url AS open_position_url,
+                j.status
+            FROM jobs j
+            LEFT JOIN companies c ON j.company_key = c.company_key
+            ORDER BY
+                (j.status = 'position_found') DESC,
+                c.name ASC NULLS LAST
+            """
+        ).fetchall()
+
+        with path.open("w", newline="", encoding="utf-8") as fh:
+            writer = csv.DictWriter(fh, fieldnames=list(CSV_COLUMNS))
+            writer.writeheader()
+            for row in rows:
+                writer.writerow(
+                    {
+                        "company_name":      row["company_name"]      or "",
+                        "career_page_url":   row["career_page_url"]   or "",
+                        "open_position_url": row["open_position_url"] or "",
+                    }
+                )
+
+        logger.info("db: exported %d rows to %s", len(rows), path)
+        return path
+
+    # ---- Counts (for summary / tests) ------------------------------------
+
+    def counts(self) -> dict[str, int]:
+        """Return per-status job counts from the DB."""
+        rows = self._conn.execute(
+            "SELECT status, COUNT(*) AS n FROM jobs GROUP BY status"
+        ).fetchall()
+        return {row["status"]: row["n"] for row in rows}
+
+    # ---- Lifecycle -------------------------------------------------------
+
+    def close(self) -> None:
+        self._conn.close()
+
+    def __enter__(self) -> "Database":
+        return self
+
+    def __exit__(self, *_: object) -> None:
+        self.close()

jobsource/extract.py

@@ -1,12 +1,232 @@
 """Extract one open position URL from a careers page (Stage 3).
 
-Scaffold stub -- not implemented yet.
+Cascade order (return early on first hit):
+  1.  ATS JSON (URL)  — detect ATS board coordinates in the careers URL itself;
+                        call the board's public JSON API for the first posting URL.
+  1b. ATS JSON (HTML) — fetch the careers page; detect ATS board in its HTML;
+                        call the board API. Handles inline JS embeds.
+  2.  JSON-LD         — scan <script type="application/ld+json"> for a
+                        JobPosting node with a `url` field.
+  3.  Job-like anchors — first <a> whose href matches /job|/position|/opening|/vacanc.
+  4.  Cheap-LLM        — classify_job_link() from classify_llm; no-ops gracefully
+                        when the LLM is not configured.
+  5.  Browser agent   — agent_fallback.find_and_extract(); fused with Stage-2 so
+                        the same memoized session covers both stages.
+  6.  Miss            — return (None, "none").
+
+Returns (url: str | None, method: str). Never raises — any tier failure logs
+and falls through. method is one of:
+  "ats:{name}" | "json_ld" | "anchor" | "llm_classify" | "browser_agent" | "none"
 """
-# TODO (Stage 3): implement per CLAUDE.md "Stage 3 — Extract one open position (return on first hit)".
+from __future__ import annotations
-# Cascade order (return early on first hit):
+
-#   1. ATS JSON — if ATS is already known from Stage 2, return first posting URL directly.
+import json
-#   2. JobPosting JSON-LD — parse application/ld+json for a `url` field.
+import logging
-#   3. Job-like anchors — first <a> matching /job, /position, /opening, /vacancy in href.
+import re
-#   4. Cheap-LLM classification — Pydantic AI typed output (classifier_model).
+from urllib.parse import urlsplit
-#   5. Browser-agent fallback — handled inside the fused Stage-2 agent call in agent_fallback.py.
+
-# Returns (url: str | None, method: str) so callers know which tier resolved it.
+import httpx
+from bs4 import BeautifulSoup
+
+from . import agent_fallback as _agent_fallback
+from .careers import ats as _ats
+from .careers.classify_llm import classify_job_link, extract_anchors
+from .config import get_settings
+from .http import build_client, request_with_retries
+
+logger = logging.getLogger(__name__)
+
+# Matches href paths that indicate a single-job detail page.
+_JOB_HREF_RE = re.compile(r"/(job|position|opening|vacanc)", re.IGNORECASE)
+
+
+# ---------------------------------------------------------------------------
+# Internal helpers
+# ---------------------------------------------------------------------------
+
+
+def _get_html(url: str, client: httpx.Client) -> str | None:
+    """Fetch page HTML; return text on success or None on any error/non-2xx."""
+    try:
+        resp = request_with_retries(client, "GET", url, max_retries=1)
+        if resp.status_code < 400:
+            return resp.text
+        logger.info("extract._get_html(%s): HTTP %s", url, resp.status_code)
+        return None
+    except Exception as exc:
+        logger.info("extract._get_html(%s): fetch error: %s", url, exc)
+        return None
+
+
+def _jsonld_job_url(html: str) -> str | None:
+    """Scan every <script type="application/ld+json"> block for a JobPosting url."""
+    soup = BeautifulSoup(html, "lxml")
+    for script in soup.find_all("script", type="application/ld+json"):
+        try:
+            data = json.loads(script.get_text(strip=True))
+        except (json.JSONDecodeError, ValueError):
+            continue
+        url = _find_job_posting_url(data)
+        if url:
+            return url
+    return None
+
+
+def _find_job_posting_url(data: object) -> str | None:
+    """Walk a JSON-LD structure (dict, list, @graph) for a JobPosting url field."""
+    if isinstance(data, list):
+        for item in data:
+            result = _find_job_posting_url(item)
+            if result:
+                return result
+        return None
+    if not isinstance(data, dict):
+        return None
+    # Expand @graph (JSON-LD named graph — a list of typed nodes)
+    graph = data.get("@graph")
+    if graph:
+        result = _find_job_posting_url(graph)
+        if result:
+            return result
+    # Check this node's @type
+    type_val = data.get("@type", "")
+    if isinstance(type_val, list):
+        is_job = "JobPosting" in type_val
+    else:
+        is_job = type_val == "JobPosting"
+    if is_job:
+        url = data.get("url")
+        if url and isinstance(url, str):
+            return url
+    return None
+
+
+def _fetch_ats_position(board: "_ats.ATSBoard", client: httpx.Client) -> str | None:
+    """Call the ATS public JSON API for *board* and return the first job URL or None."""
+    fetch_fn = _ats._FETCH_DISPATCH.get(board.ats_name)
+    if fetch_fn is None:
+        return None
+    fetch = fetch_fn(board, client)  # type: ignore[operator]
+    return fetch.first_url
+
+
+# ---------------------------------------------------------------------------
+# Public entry point
+# ---------------------------------------------------------------------------
+
+
+def extract_open_position(
+    careers_url: str,
+    *,
+    client: httpx.Client | None = None,
+) -> tuple[str | None, str]:
+    """Return (open_position_url, method) for a careers page. Never raises.
+
+    Tries ATS JSON → JSON-LD → job-like anchor → LLM classifier in order,
+    returning on the first hit. method names which tier resolved it.
+    """
+    _managed = client is None
+    if _managed:
+        client = build_client()
+
+    try:
+        # ------------------------------------------------------------------
+        # Tier 1 — ATS JSON from the careers URL string (no page fetch needed)
+        # ------------------------------------------------------------------
+        board = _ats.detect_ats_in_url(careers_url)
+        if board is not None:
+            try:
+                pos_url = _fetch_ats_position(board, client)
+                if pos_url:
+                    logger.info(
+                        "extract(%s): ats:%s from_url pos=%s",
+                        careers_url, board.ats_name, pos_url,
+                    )
+                    return pos_url, f"ats:{board.ats_name}"
+            except Exception as exc:
+                logger.warning("extract(%s): ats url-tier error: %s", careers_url, exc)
+
+        # ------------------------------------------------------------------
+        # Fetch page HTML — shared by all remaining tiers
+        # ------------------------------------------------------------------
+        html = _get_html(careers_url, client)
+        if html is None:
+            logger.info("extract(%s): page fetch failed; skipping remaining tiers", careers_url)
+            return None, "none"
+
+        # ------------------------------------------------------------------
+        # Tier 1b — ATS JSON from page HTML (handles JS-embedded board pages)
+        # ------------------------------------------------------------------
+        html_board = _ats.detect_ats_in_html(html)
+        if html_board is not None:
+            try:
+                pos_url2 = _fetch_ats_position(html_board, client)
+                if pos_url2:
+                    logger.info(
+                        "extract(%s): ats:%s from_html pos=%s",
+                        careers_url, html_board.ats_name, pos_url2,
+                    )
+                    return pos_url2, f"ats:{html_board.ats_name}"
+            except Exception as exc:
+                logger.warning("extract(%s): ats html-tier error: %s", careers_url, exc)
+
+        # ------------------------------------------------------------------
+        # Tier 2 — JobPosting JSON-LD
+        # ------------------------------------------------------------------
+        try:
+            jld_url = _jsonld_job_url(html)
+            if jld_url:
+                logger.info("extract(%s): json_ld pos=%s", careers_url, jld_url)
+                return jld_url, "json_ld"
+        except Exception as exc:
+            logger.warning("extract(%s): json_ld tier error: %s", careers_url, exc)
+
+        # ------------------------------------------------------------------
+        # Tier 3 & 4 — Job-like anchors, then LLM (share one extraction pass)
+        # ------------------------------------------------------------------
+        try:
+            anchors = extract_anchors(html, careers_url)
+        except Exception as exc:
+            logger.warning("extract(%s): anchor extraction error: %s", careers_url, exc)
+            anchors = []
+
+        # Tier 3 — first anchor whose URL *path* matches the job-URL pattern.
+        # We search only the path component so hostnames like jobs.lever.co
+        # don't trigger a false match (e.g. `//jobs` contains `/job`).
+        try:
+            for anchor in anchors:
+                if _JOB_HREF_RE.search(urlsplit(anchor.url).path):
+                    logger.info("extract(%s): anchor pos=%s", careers_url, anchor.url)
+                    return anchor.url, "anchor"
+        except Exception as exc:
+            logger.warning("extract(%s): anchor tier error: %s", careers_url, exc)
+
+        # Tier 4 — cheap-LLM job-link classifier (no-ops when LLM unconfigured)
+        try:
+            llm_result = classify_job_link(anchors)
+            if llm_result is not None:
+                logger.info("extract(%s): llm_classify pos=%s", careers_url, llm_result.job_url)
+                return llm_result.job_url, "llm_classify"
+        except Exception as exc:
+            logger.warning("extract(%s): llm_classify tier error: %s", careers_url, exc)
+
+        # ------------------------------------------------------------------
+        # Tier 5 — Browser agent (fused; memo-shared with cascade Tier 6)
+        # ------------------------------------------------------------------
+        if get_settings().enable_browser_agent:
+            try:
+                ag = _agent_fallback.find_and_extract(careers_url)
+                if ag and ag.position_url:
+                    logger.info(
+                        "extract(%s): browser_agent pos=%s", careers_url, ag.position_url
+                    )
+                    return ag.position_url, "browser_agent"
+            except Exception as exc:
+                logger.warning("extract(%s): browser_agent tier error: %s", careers_url, exc)
+
+        logger.info("extract(%s): all tiers missed", careers_url)
+        return None, "none"
+
+    finally:
+        if _managed:
+            client.close()

jobsource/main.py

@@ -46,8 +46,13 @@ def build_parser() -> argparse.ArgumentParser:
 
 def main(argv: list[str] | None = None) -> int:
     args = build_parser().parse_args(argv)
-    print("jobsource: scaffold stub -- pipeline not implemented yet.", file=sys.stderr)
+    from jobsource.pipeline import run_batch  # deferred: keeps --help fast before heavy deps
-    print(f"parsed args: {vars(args)}", file=sys.stderr)
+    run_batch(
+        batch_size=args.batch_size,
+        search_terms=args.search,   # None or list[str] from --search repeats
+        location=args.location,
+        hours_old=args.hours_old,
+    )
     return 0
 
 

jobsource/pipeline.py

@@ -1,12 +1,334 @@
-"""Batch orchestration: dedup, per-record isolation, cascade, persistence, summary.
+"""Batch orchestration: dedup, per-record isolation, cascade, persistence, summary."""
+from __future__ import annotations
 
-Scaffold stub -- not implemented yet.
+import logging
+import time
+from dataclasses import dataclass
+
+from .careers.cascade import find_careers_page
+from .config import get_settings
+from .db import Database
+from .extract import extract_open_position
+from .http import build_client
+from .models import JobResult, JobStatus, RawJob
+from .resolve import resolve_website
+from .sources.base import JobSource
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Public summary result
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class BatchSummary:
+    """Per-stage counts returned (and printed) by run_batch()."""
+
+    fetched: int = 0
+    already_seen: int = 0
+    new: int = 0
+    website_resolved: int = 0
+    careers_found: int = 0
+    position_found: int = 0
+    needs_review: int = 0
+    failed: int = 0
+
+    @property
+    def coverage(self) -> float:
+        """Fraction of new jobs that reached position_found."""
+        return self.position_found / self.new if self.new > 0 else 0.0
+
+
+# ---------------------------------------------------------------------------
+# Entry point
+# ---------------------------------------------------------------------------
+
+
+def run_batch(
+    batch_size: int | None = None,
+    *,
+    search_terms: list[str] | None = None,
+    location: str | None = None,
+    hours_old: int | None = None,
+    db: Database | None = None,
+) -> BatchSummary:
+    """Fetch, dedup, cascade, persist, and export one batch of jobs.
+
+    Override arguments take precedence over config values; ``None`` means
+    "use config default".  Returns a :class:`BatchSummary` (also printed).
+
+    One failing record never aborts the batch — each job is isolated in its
+    own try/except; the outer loop continues regardless.
     """
-# TODO (pipeline): implement run_batch() per CLAUDE.md "Pipeline stages".
+    settings     = get_settings()
-# run_batch() contract:
+    batch_size   = batch_size   if batch_size   is not None else settings.batch_size
-#   - Accept batch_size, search terms, location, hours_old overrides.
+    search_terms = search_terms if search_terms is not None else list(settings.search_terms)
-#   - Call the job source, dedup by job_id against the DB (skip already-seen jobs).
+    location     = location     if location     is not None else settings.location
-#   - For each new RawJob, run the full cascade (resolve -> careers -> extract) in isolation:
+    hours_old    = hours_old    if hours_old    is not None else settings.hours_old
-#     one failing record must NEVER abort the batch — catch, record failed/needs_review, continue.
+
-#   - Persist each JobResult to the DB and export output/results.csv when done.
+    own_db = db is None
-#   - Print a run summary: per-stage counts + % of new jobs reaching position_found.
+    if own_db:
+        db = Database()
+
+    summary  = BatchSummary()
+    t0       = time.perf_counter()
+    csv_path = None
+
+    try:
+        # ------------------------------------------------------------------
+        # Stage 1 — Ingest
+        # ------------------------------------------------------------------
+        source = _make_source(settings)
+        print(f"\n{'='*64}")
+        print(
+            f"  JobSourceAgent  batch={batch_size}"
+            f"  search={search_terms}"
+            f"  hours_old={hours_old}"
+        )
+        print(f"{'='*64}")
+        print("Stage 1: ingesting jobs…")
+
+        raw_jobs: list[RawJob] = source.fetch_recent_jobs(
+            search_terms=search_terms,
+            location=location,
+            hours_old=hours_old,
+            results_wanted=settings.results_wanted,
+        )
+        summary.fetched = len(raw_jobs)
+
+        # Dedup: filter out jobs already in the DB; cap new jobs at batch_size.
+        new_jobs: list[RawJob] = []
+        for raw in raw_jobs:
+            if db.is_seen(raw.job_id):
+                summary.already_seen += 1
+            else:
+                new_jobs.append(raw)
+                if len(new_jobs) >= batch_size:
+                    break
+
+        summary.new = len(new_jobs)
+        print(
+            f"  Fetched {summary.fetched} jobs; "
+            f"{summary.already_seen} already seen; "
+            f"{summary.new} new to process.\n"
+        )
+
+        # ------------------------------------------------------------------
+        # Stages 1b → 3 — per-record cascade
+        # ------------------------------------------------------------------
+        if new_jobs:
+            with build_client() as client:
+                for raw in new_jobs:
+                    result = JobResult.from_raw(raw)
+                    print(f"[{raw.job_id}] {raw.company}")
+                    try:
+                        _process_one(result, raw, db, client)
+                    except Exception as exc:
+                        # Safety net — _process_one should not raise, but guard anyway.
+                        logger.exception(
+                            "pipeline: unexpected error for %s (%s): %s",
+                            raw.job_id, raw.company, exc,
+                        )
+                        result.status = JobStatus.failed
+                        try:
+                            db.persist_result(result)
+                        except Exception:
+                            logger.exception(
+                                "pipeline: failed to persist error record %s", raw.job_id
+                            )
+                    _tally(summary, result)
+                    print()
+
+        # ------------------------------------------------------------------
+        # Stage 4 — Export CSV
+        # ------------------------------------------------------------------
+        try:
+            csv_path = db.export_csv()
+        except Exception as exc:
+            logger.error("pipeline: CSV export failed: %s", exc)
+
+    finally:
+        if own_db:
+            db.close()
+
+    elapsed = time.perf_counter() - t0
+    _print_summary(summary, elapsed, csv_path)
+    return summary
+
+
+# ---------------------------------------------------------------------------
+# Per-record cascade (must not raise — handles all errors internally)
+# ---------------------------------------------------------------------------
+
+
+def _process_one(
+    result: JobResult,
+    raw: RawJob,
+    db: Database,
+    client: object,
+) -> None:
+    """Run stages 1b → 3 for one job and persist the result.
+
+    Uses early returns for soft misses (needs_review) and catches exceptions
+    for hard failures (failed status). Never raises to the caller.
+    """
+    # ------------------------------------------------------------------
+    # Stage 1b — Resolve company website
+    # ------------------------------------------------------------------
+    try:
+        website = resolve_website(raw.company, raw.website, client=client)
+    except Exception as exc:
+        logger.warning("pipeline(%s): resolve error: %s", raw.job_id, exc)
+        print(f"    resolve: ERROR — {exc}")
+        result.status = JobStatus.failed
+        db.persist_result(result)
+        return
+
+    if not website:
+        print("    website: UNRESOLVED")
+        result.status = JobStatus.needs_review
+        db.persist_result(result)
+        return
+
+    result.website = website
+    result.status  = JobStatus.website_resolved
+    result.company_key = db.company_key_for(raw.company, website)
+    print(f"    website: {website}")
+
+    # ------------------------------------------------------------------
+    # Stage 2 — Find careers page (check company cache first)
+    # ------------------------------------------------------------------
+    cached_career_url = db.get_cached_career_url(result.company_key)
+
+    if cached_career_url:
+        careers_url       = cached_career_url
+        careers_method    = "cache"
+        position_shortcut: str | None = None
+        print(f"    careers: {careers_url}  [cache]")
+    else:
+        try:
+            cr = find_careers_page(website, company_name=raw.company, client=client)
+        except Exception as exc:
+            logger.warning("pipeline(%s): cascade error: %s", raw.job_id, exc)
+            print(f"    careers: ERROR — {exc}")
+            result.status = JobStatus.needs_review
+            db.persist_result(result)
+            return
+
+        careers_url       = cr.careers_url
+        careers_method    = cr.method
+        position_shortcut = cr.position_url
+
+        if not careers_url:
+            print(f"    careers: MISSED  (method={cr.method})")
+            result.status = JobStatus.needs_review
+            db.persist_result(result)
+            return
+
+        print(
+            f"    careers: {careers_url}"
+            f"  [{careers_method}  conf={cr.confidence:.2f}]"
+        )
+
+    result.career_page_url = careers_url
+    result.careers_method  = careers_method
+    result.status          = JobStatus.careers_found
+
+    # ------------------------------------------------------------------
+    # Stage 3 — Extract one open position
+    # ------------------------------------------------------------------
+    if position_shortcut:
+        # Free shortcut from ATS or browser-agent tier — no extra HTTP call.
+        position_url    = position_shortcut
+        position_method = careers_method
+    else:
+        try:
+            position_url, position_method = extract_open_position(
+                careers_url, client=client
+            )
+        except Exception as exc:
+            logger.warning("pipeline(%s): extract error: %s", raw.job_id, exc)
+            print(f"    position: ERROR — {exc}")
+            result.status = JobStatus.needs_review
+            db.persist_result(result)
+            return
+
+    if position_url:
+        result.open_position_url = position_url
+        result.position_method   = position_method
+        result.status            = JobStatus.position_found
+        print(f"    position: {position_url}  [{position_method}]")
+    else:
+        print(f"    position: MISSED  (method={position_method})")
+        result.status = JobStatus.needs_review
+
+    db.persist_result(result)
+
+
+# ---------------------------------------------------------------------------
+# Internal helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_source(settings) -> JobSource:
+    """Instantiate the configured job source provider."""
+    if settings.job_source == "apify":
+        from .sources.apify_source import ApifySource
+        return ApifySource()
+    from .sources.jobspy_source import JobSpySource
+    return JobSpySource()
+
+
+def _tally(summary: BatchSummary, result: JobResult) -> None:
+    """Accumulate per-stage counts from a processed JobResult."""
+    if result.website:
+        summary.website_resolved += 1
+    if result.career_page_url:
+        summary.careers_found += 1
+    if result.open_position_url:
+        summary.position_found += 1
+    if result.status == JobStatus.needs_review:
+        summary.needs_review += 1
+    elif result.status == JobStatus.failed:
+        summary.failed += 1
+
+
+def _print_summary(
+    summary: BatchSummary, elapsed: float, csv_path: object
+) -> None:
+    """Print the run summary table to stdout."""
+    n = summary.new
+    wr = summary.website_resolved
+    cf = summary.careers_found
+
+    print(f"{'='*64}")
+    print(f"  SUMMARY  ({n} new jobs processed in {elapsed:.1f}s)")
+    print(f"{'='*64}")
+    print(
+        f"  Stage 1  ingested            {summary.fetched:>4}"
+        f"  ({summary.already_seen} already seen)"
+    )
+    print(
+        f"  Stage 1b website resolved    {wr:>4} / {n}"
+        f"  ({wr / max(n, 1) * 100:.0f}%)"
+    )
+    print(
+        f"  Stage 2  careers found       {cf:>4} / {wr}"
+        f"  ({cf / max(wr, 1) * 100:.0f}% of resolved)"
+    )
+    print(
+        f"  Stage 3  position found      {summary.position_found:>4} / {cf}"
+        f"  ({summary.position_found / max(cf, 1) * 100:.0f}% of careers)"
+    )
+    print(f"  Needs review                 {summary.needs_review:>4}")
+    print(f"  Failed                       {summary.failed:>4}")
+    print(
+        f"  End-to-end coverage          "
+        f"{summary.coverage * 100:.0f}%"
+        f"  ({summary.position_found}/{n} new jobs fully resolved)"
+    )
+    if csv_path:
+        print(f"  CSV written to               {csv_path}")
+    print()

scripts/e2e_smoke.py

@@ -0,0 +1,207 @@
+#!/usr/bin/env python
+"""End-to-end smoke test: Stage 1→3 on a 10-job batch.
+
+Usage:  .venv/bin/python scripts/e2e_smoke.py
+"""
+from __future__ import annotations
+
+import logging
+import sys
+import time
+from dataclasses import dataclass, field
+
+# Make sure we run from the repo root so imports resolve correctly.
+sys.path.insert(0, ".")
+
+logging.basicConfig(
+    level=logging.WARNING,       # keep jobspy/httpx chatter suppressed
+    format="%(name)s: %(message)s",
+)
+# Enable our own modules at INFO so we see cascade method names.
+for _mod in ("jobsource.resolve", "jobsource.careers.cascade",
+             "jobsource.careers.ats", "jobsource.careers.heuristics",
+             "jobsource.careers.classify_llm", "jobsource.extract"):
+    logging.getLogger(_mod).setLevel(logging.INFO)
+
+from jobsource.careers.cascade import find_careers_page
+from jobsource.config import get_settings
+from jobsource.extract import extract_open_position
+from jobsource.http import build_client
+from jobsource.models import RawJob
+from jobsource.resolve import resolve_website
+from jobsource.sources.jobspy_source import JobSpySource
+
+
+# ---------------------------------------------------------------------------
+# Per-record result
+# ---------------------------------------------------------------------------
+
+@dataclass
+class Result:
+    job_id: str
+    company: str
+    website: str | None = None
+    careers_url: str | None = None
+    careers_method: str | None = None
+    position_url: str | None = None
+    position_method: str | None = None
+    error: str | None = None
+
+
+# ---------------------------------------------------------------------------
+# Main runner
+# ---------------------------------------------------------------------------
+
+BATCH_SIZE = 10
+SEARCH_TERM = "software engineer"
+LOCATION = "United States"
+HOURS_OLD = 72
+
+
+def run() -> None:
+    settings = get_settings()
+    print(f"\n{'='*64}")
+    print(f"  E2E smoke  batch={BATCH_SIZE}  search={SEARCH_TERM!r}  hours_old={HOURS_OLD}")
+    print(f"{'='*64}\n")
+
+    # ------------------------------------------------------------------
+    # Stage 1 — Ingest
+    # ------------------------------------------------------------------
+    print("Stage 1: ingesting…")
+    source = JobSpySource()
+    t0 = time.perf_counter()
+    raw_jobs: list[RawJob] = source.fetch_recent_jobs(
+        search_terms=[SEARCH_TERM],
+        location=LOCATION,
+        hours_old=HOURS_OLD,
+        results_wanted=BATCH_SIZE + 10,   # fetch a few extra to cover dupes
+    )
+    elapsed = time.perf_counter() - t0
+    raw_jobs = raw_jobs[:BATCH_SIZE]
+    print(f"  Ingested {len(raw_jobs)} unique jobs in {elapsed:.1f}s\n")
+
+    if not raw_jobs:
+        print("No jobs returned — aborting.")
+        return
+
+    results: list[Result] = []
+
+    with build_client() as client:
+        for raw in raw_jobs:
+            res = Result(job_id=raw.job_id, company=raw.company)
+            print(f"[{raw.job_id}] {raw.company}")
+
+            # -------------------------------------------------------
+            # Stage 1b — Resolve website
+            # -------------------------------------------------------
+            try:
+                res.website = resolve_website(
+                    raw.company, raw.website, client=client
+                )
+            except Exception as exc:
+                res.error = f"resolve: {exc}"
+                print(f"    resolve ERROR: {exc}")
+                results.append(res)
+                continue
+
+            if not res.website:
+                print(f"    website: UNRESOLVED")
+                results.append(res)
+                continue
+            print(f"    website: {res.website}")
+
+            # -------------------------------------------------------
+            # Stage 2 — Find careers page
+            # -------------------------------------------------------
+            try:
+                cr = find_careers_page(
+                    res.website,
+                    company_name=raw.company,
+                    client=client,
+                )
+                res.careers_url = cr.careers_url
+                res.careers_method = cr.method
+            except Exception as exc:
+                res.error = f"cascade: {exc}"
+                print(f"    careers ERROR: {exc}")
+                results.append(res)
+                continue
+
+            if not res.careers_url:
+                print(f"    careers: MISSED  (method={cr.method})")
+                results.append(res)
+                continue
+            print(f"    careers: {res.careers_url}  [{cr.method}  conf={cr.confidence:.2f}]")
+
+            # -------------------------------------------------------
+            # Stage 3 — Extract one open position
+            # -------------------------------------------------------
+            try:
+                pos_url, pos_method = extract_open_position(
+                    res.careers_url, client=client
+                )
+                res.position_url = pos_url
+                res.position_method = pos_method
+            except Exception as exc:
+                res.error = f"extract: {exc}"
+                print(f"    position ERROR: {exc}")
+                results.append(res)
+                continue
+
+            if res.position_url:
+                print(f"    position: {res.position_url}  [{pos_method}]")
+            else:
+                print(f"    position: MISSED  (method={pos_method})")
+
+            results.append(res)
+            print()
+
+    # ------------------------------------------------------------------
+    # Summary
+    # ------------------------------------------------------------------
+    n_total        = len(results)
+    n_website      = sum(1 for r in results if r.website)
+    n_careers      = sum(1 for r in results if r.careers_url)
+    n_position     = sum(1 for r in results if r.position_url)
+    n_errors       = sum(1 for r in results if r.error)
+
+    print(f"\n{'='*64}")
+    print(f"  SUMMARY  ({n_total} jobs processed)")
+    print(f"{'='*64}")
+    print(f"  Stage 1  ingested          {n_total:>3}")
+    print(f"  Stage 1b website resolved  {n_website:>3} / {n_total}  ({n_website/n_total*100:.0f}%)")
+    print(f"  Stage 2  careers found     {n_careers:>3} / {n_website}  ({n_careers/max(n_website,1)*100:.0f}% of resolved)")
+    print(f"  Stage 3  position found    {n_position:>3} / {n_careers}  ({n_position/max(n_careers,1)*100:.0f}% of careers)")
+    print(f"  Errors                     {n_errors:>3}")
+
+    # careers method breakdown
+    if n_careers:
+        method_counts: dict[str, int] = {}
+        for r in results:
+            if r.careers_method and r.careers_url:
+                method_counts[r.careers_method] = method_counts.get(r.careers_method, 0) + 1
+        print(f"\n  Careers method breakdown:")
+        for m, c in sorted(method_counts.items(), key=lambda x: -x[1]):
+            print(f"    {m:<30} {c}")
+
+    # position method breakdown
+    if n_position:
+        pos_method_counts: dict[str, int] = {}
+        for r in results:
+            if r.position_method and r.position_url:
+                pos_method_counts[r.position_method] = pos_method_counts.get(r.position_method, 0) + 1
+        print(f"\n  Position method breakdown:")
+        for m, c in sorted(pos_method_counts.items(), key=lambda x: -x[1]):
+            print(f"    {m:<30} {c}")
+
+    # CSV-ready output
+    print(f"\n{'='*64}")
+    print("  company_name, career_page_url, open_position_url")
+    print(f"{'='*64}")
+    for r in results:
+        print(f"  {r.company!r}, {r.careers_url or ''!r}, {r.position_url or ''!r}")
+    print()
+
+
+if __name__ == "__main__":
+    run()

tests/test_agent_fallback.py

@@ -0,0 +1,386 @@
+"""Tests for jobsource/agent_fallback.py — all network-free via monkeypatching."""
+from __future__ import annotations
+
+import pytest
+
+import jobsource.agent_fallback as _mod
+from jobsource.agent_fallback import (
+    AgentFallbackResult,
+    _AgentOutput,
+    _site_key,
+    _valid_url,
+    find_and_extract,
+)
+
+
+# ---------------------------------------------------------------------------
+# Helpers and fakes
+# ---------------------------------------------------------------------------
+
+
+def _placeholder_settings(**overrides):
+    """Return a fake settings object with placeholder values (agent unavailable)."""
+    defaults = {
+        "enable_browser_agent": True,
+        "agent_model": "PLACEHOLDER_AGENT_MODEL",
+        "llm_api_key": "PLACEHOLDER_LLM_API_KEY",
+        "browser_headless": True,
+    }
+    defaults.update(overrides)
+    return type("S", (), defaults)()
+
+
+def _real_settings(**overrides):
+    """Return a fake settings object with non-placeholder values."""
+    defaults = {
+        "enable_browser_agent": True,
+        "agent_model": "anthropic:claude-sonnet-4-6",
+        "llm_api_key": "sk-ant-real",
+        "browser_headless": True,
+    }
+    defaults.update(overrides)
+    return type("S", (), defaults)()
+
+
+def _patch_settings(monkeypatch, settings_obj):
+    monkeypatch.setattr(_mod, "get_settings", lambda: settings_obj)
+
+
+def _patch_chat(monkeypatch, chat_obj=None):
+    """Monkeypatch _build_chat_model to return chat_obj (default sentinel)."""
+    chat_obj = chat_obj or object()
+    monkeypatch.setattr(_mod, "_build_chat_model", lambda: chat_obj)
+    return chat_obj
+
+
+def _patch_session(monkeypatch, output: _AgentOutput | None):
+    """Monkeypatch _agent_session so no real browser launches."""
+    async def _fake_session(website, chat):
+        return output
+    monkeypatch.setattr(_mod, "_agent_session", _fake_session)
+
+
+# ---------------------------------------------------------------------------
+# _site_key
+# ---------------------------------------------------------------------------
+
+class TestSiteKey:
+    def test_strips_www(self):
+        assert _site_key("https://www.acme.com/foo") == "acme.com"
+
+    def test_lowercases(self):
+        assert _site_key("https://ACME.COM") == "acme.com"
+
+    def test_no_www(self):
+        assert _site_key("https://acme.com/careers") == "acme.com"
+
+    def test_subdomain_preserved(self):
+        assert _site_key("https://jobs.acme.com") == "jobs.acme.com"
+
+
+# ---------------------------------------------------------------------------
+# _valid_url
+# ---------------------------------------------------------------------------
+
+class TestValidUrl:
+    def test_https_url_passes(self):
+        assert _valid_url("https://acme.com/jobs/1") == "https://acme.com/jobs/1"
+
+    def test_http_url_passes(self):
+        assert _valid_url("http://acme.com/jobs/1") == "http://acme.com/jobs/1"
+
+    def test_none_returns_none(self):
+        assert _valid_url(None) is None
+
+    def test_junk_string_returns_none(self):
+        assert _valid_url("null") is None
+
+    def test_relative_url_returns_none(self):
+        assert _valid_url("/jobs/1") is None
+
+
+# ---------------------------------------------------------------------------
+# Gate: enable_browser_agent=False
+# ---------------------------------------------------------------------------
+
+class TestFeatureFlagGate:
+    def test_returns_none_when_flag_off(self, monkeypatch):
+        _patch_settings(monkeypatch, _placeholder_settings(enable_browser_agent=False))
+        # Clear memo so there's no cached entry from another test.
+        _mod._MEMO.clear()
+        result = find_and_extract("https://acme.com")
+        assert result is None
+
+    def test_does_not_cache_when_flag_off(self, monkeypatch):
+        _patch_settings(monkeypatch, _placeholder_settings(enable_browser_agent=False))
+        _mod._MEMO.clear()
+        find_and_extract("https://flagoff.com")
+        assert "flagoff.com" not in _mod._MEMO
+
+
+# ---------------------------------------------------------------------------
+# Gate: placeholder LLM config
+# ---------------------------------------------------------------------------
+
+class TestPlaceholderGate:
+    def test_returns_none_for_placeholder_model(self, monkeypatch):
+        _patch_settings(monkeypatch, _placeholder_settings())
+        _mod._MEMO.clear()
+        result = find_and_extract("https://acme.com")
+        assert result is None
+
+    def test_returns_none_for_placeholder_key_only(self, monkeypatch):
+        _patch_settings(monkeypatch, _real_settings(llm_api_key="PLACEHOLDER_LLM_API_KEY"))
+        _mod._MEMO.clear()
+        result = find_and_extract("https://acme.com")
+        assert result is None
+
+    def test_returns_none_for_placeholder_model_only(self, monkeypatch):
+        _patch_settings(monkeypatch, _real_settings(agent_model="PLACEHOLDER_AGENT_MODEL"))
+        _mod._MEMO.clear()
+        result = find_and_extract("https://acme.com")
+        assert result is None
+
+
+# ---------------------------------------------------------------------------
+# Gate: _build_chat_model raises (missing package / bad config)
+# ---------------------------------------------------------------------------
+
+class TestChatModelBuildFailure:
+    def test_returns_none_and_memos_none_on_import_error(self, monkeypatch):
+        _patch_settings(monkeypatch, _real_settings())
+        _mod._MEMO.clear()
+
+        def _raise():
+            raise ImportError("browser_use not installed")
+
+        monkeypatch.setattr(_mod, "_build_chat_model", _raise)
+        result = find_and_extract("https://chatfail.com")
+        assert result is None
+        assert _mod._MEMO.get("chatfail.com") is None
+
+
+# ---------------------------------------------------------------------------
+# Gate: _agent_session raises / returns None
+# ---------------------------------------------------------------------------
+
+class TestAgentSessionFailure:
+    def test_returns_none_on_session_exception(self, monkeypatch):
+        _patch_settings(monkeypatch, _real_settings())
+        _mod._MEMO.clear()
+        _patch_chat(monkeypatch)
+
+        async def _raise_session(website, chat):
+            raise RuntimeError("Chromium unavailable")
+
+        monkeypatch.setattr(_mod, "_agent_session", _raise_session)
+        result = find_and_extract("https://chromiumfail.com")
+        assert result is None
+
+    def test_returns_none_when_session_returns_none(self, monkeypatch):
+        _patch_settings(monkeypatch, _real_settings())
+        _mod._MEMO.clear()
+        _patch_chat(monkeypatch)
+        _patch_session(monkeypatch, None)
+        result = find_and_extract("https://nonefail.com")
+        assert result is None
+
+
+# ---------------------------------------------------------------------------
+# Success path: structured output parsed into AgentFallbackResult
+# ---------------------------------------------------------------------------
+
+class TestSuccessPath:
+    def test_both_urls_returned(self, monkeypatch):
+        _patch_settings(monkeypatch, _real_settings())
+        _mod._MEMO.clear()
+        _patch_chat(monkeypatch)
+        _patch_session(
+            monkeypatch,
+            _AgentOutput(
+                careers_url="https://acme.com/careers",
+                position_url="https://acme.com/careers/jobs/42-swe",
+            ),
+        )
+        result = find_and_extract("https://acme.com")
+        assert result is not None
+        assert result.careers_url == "https://acme.com/careers"
+        assert result.position_url == "https://acme.com/careers/jobs/42-swe"
+
+    def test_junk_careers_url_filtered(self, monkeypatch):
+        _patch_settings(monkeypatch, _real_settings())
+        _mod._MEMO.clear()
+        _patch_chat(monkeypatch)
+        _patch_session(
+            monkeypatch,
+            _AgentOutput(careers_url="null", position_url="https://acme.com/jobs/1"),
+        )
+        result = find_and_extract("https://acme.com")
+        assert result is not None
+        assert result.careers_url is None
+        assert result.position_url == "https://acme.com/jobs/1"
+
+    def test_result_stored_in_memo(self, monkeypatch):
+        _patch_settings(monkeypatch, _real_settings())
+        _mod._MEMO.clear()
+        _patch_chat(monkeypatch)
+        _patch_session(
+            monkeypatch,
+            _AgentOutput(
+                careers_url="https://memo-test.com/careers",
+                position_url="https://memo-test.com/jobs/1",
+            ),
+        )
+        find_and_extract("https://memo-test.com")
+        cached = _mod._MEMO.get("memo-test.com")
+        assert cached is not None
+        assert cached.careers_url == "https://memo-test.com/careers"
+
+
+# ---------------------------------------------------------------------------
+# Memo: same-domain second call uses cache, not a second session
+# ---------------------------------------------------------------------------
+
+class TestMemo:
+    def test_second_call_uses_cache(self, monkeypatch):
+        _patch_settings(monkeypatch, _real_settings())
+        _mod._MEMO.clear()
+        _patch_chat(monkeypatch)
+
+        call_count = 0
+
+        async def _counting_session(website, chat):
+            nonlocal call_count
+            call_count += 1
+            return _AgentOutput(
+                careers_url="https://sharetest.com/careers",
+                position_url="https://sharetest.com/jobs/1",
+            )
+
+        monkeypatch.setattr(_mod, "_agent_session", _counting_session)
+
+        r1 = find_and_extract("https://sharetest.com")
+        r2 = find_and_extract("https://sharetest.com/careers")  # same host, careers path
+
+        assert call_count == 1, "Expected exactly one browser session for the same host"
+        assert r1 == r2
+
+    def test_memo_none_prevents_retry(self, monkeypatch):
+        _patch_settings(monkeypatch, _real_settings())
+        _mod._MEMO.clear()
+        _patch_chat(monkeypatch)
+
+        call_count = 0
+
+        async def _failing_session(website, chat):
+            nonlocal call_count
+            call_count += 1
+            raise RuntimeError("broken")
+
+        monkeypatch.setattr(_mod, "_agent_session", _failing_session)
+
+        find_and_extract("https://memonil.com")
+        find_and_extract("https://memonil.com")  # should use cached None
+
+        assert call_count == 1
+
+
+# ---------------------------------------------------------------------------
+# Cascade/extract wiring integration (lightweight, no real browser)
+# ---------------------------------------------------------------------------
+
+class TestCascadeWiring:
+    """Verify find_and_extract is reachable from cascade.find_careers_page."""
+
+    def test_cascade_tier6_returns_browser_agent_method(self, monkeypatch):
+        from jobsource.careers.cascade import find_careers_page, CareersResult
+
+        # Patch all cheaper tiers to miss.
+        monkeypatch.setattr(
+            "jobsource.careers.cascade._ats.detect_and_fetch", lambda *a, **kw: None
+        )
+        monkeypatch.setattr(
+            "jobsource.careers.cascade._ats.recover_via_slug_guess", lambda *a, **kw: None
+        )
+        monkeypatch.setattr(
+            "jobsource.careers.cascade._heuristics.probe_url_patterns", lambda *a, **kw: None
+        )
+        monkeypatch.setattr(
+            "jobsource.careers.cascade._heuristics.scan_homepage_links", lambda *a, **kw: None
+        )
+        monkeypatch.setattr(
+            "jobsource.careers.cascade._heuristics.parse_sitemap", lambda *a, **kw: None
+        )
+        monkeypatch.setattr(
+            "jobsource.careers.cascade._classify_llm.extract_anchors", lambda *a, **kw: []
+        )
+        monkeypatch.setattr(
+            "jobsource.careers.cascade._classify_llm.classify_careers_link",
+            lambda *a, **kw: None,
+        )
+        # Patch the safe homepage fetch.
+        from jobsource.careers import cascade as _cascade
+        monkeypatch.setattr(_cascade, "_safe_get_html", lambda *a, **kw: None)
+
+        # Patch agent_fallback.
+        _mod._MEMO.clear()
+        monkeypatch.setattr(
+            "jobsource.careers.cascade._agent_fallback.find_and_extract",
+            lambda *a, **kw: AgentFallbackResult(
+                careers_url="https://wired.com/careers",
+                position_url="https://wired.com/jobs/1",
+            ),
+        )
+
+        result = find_careers_page("https://wired.com")
+        assert result.method == "browser_agent"
+        assert result.careers_url == "https://wired.com/careers"
+        assert result.position_url == "https://wired.com/jobs/1"
+        assert result.confidence == 0.50
+
+
+class TestExtractWiring:
+    """Verify find_and_extract is reachable from extract.extract_open_position."""
+
+    def test_extract_tier5_returns_browser_agent_method(self, monkeypatch):
+        from jobsource.extract import extract_open_position
+
+        # Patch cheaper tiers to miss (ATS, JSON-LD, anchors, LLM).
+        monkeypatch.setattr(
+            "jobsource.extract._ats.detect_ats_in_url", lambda *a, **kw: None
+        )
+        monkeypatch.setattr(
+            "jobsource.extract._ats.detect_ats_in_html", lambda *a, **kw: None
+        )
+
+        import httpx
+        fake_response = type(
+            "R",
+            (),
+            {"status_code": 200, "text": "<html><body></body></html>"},
+        )()
+
+        monkeypatch.setattr(
+            "jobsource.extract.request_with_retries",
+            lambda *a, **kw: fake_response,
+        )
+        monkeypatch.setattr(
+            "jobsource.extract.extract_anchors", lambda *a, **kw: []
+        )
+        monkeypatch.setattr(
+            "jobsource.extract.classify_job_link", lambda *a, **kw: None
+        )
+
+        # Patch agent_fallback.
+        _mod._MEMO.clear()
+        monkeypatch.setattr(
+            "jobsource.extract._agent_fallback.find_and_extract",
+            lambda *a, **kw: AgentFallbackResult(
+                careers_url="https://extract-wired.com/careers",
+                position_url="https://extract-wired.com/jobs/99",
+            ),
+        )
+
+        url, method = extract_open_position("https://extract-wired.com/careers")
+        assert method == "browser_agent"
+        assert url == "https://extract-wired.com/jobs/99"

tests/test_cascade.py

@@ -550,3 +550,112 @@ class TestSlugGuessTier:
         result = find_careers_page("https://www.acme.com", client=FakeClient())
         assert result.method == "url_pattern"
         assert result.careers_url == "https://acme.com/careers"
+
+
+# ---------------------------------------------------------------------------
+# Tier 5 — Cheap-LLM careers-link classification
+# ---------------------------------------------------------------------------
+
+
+class TestLLMClassifyTier:
+    """Tier 5 fires after sitemap, no-ops when LLM returns None, and is skipped on earlier hits."""
+
+    def _patch_all_deterministic_miss(self, monkeypatch, *, homepage_html: str = "<html/>") -> None:
+        """Patch tiers 1–4 to all miss; leaves Tier 5 for the caller to control."""
+        monkeypatch.setattr(
+            "jobsource.careers.cascade._safe_get_html",
+            lambda website, client: homepage_html,
+        )
+        monkeypatch.setattr("jobsource.careers.cascade._ats.detect_and_fetch", lambda *a, **kw: None)
+        monkeypatch.setattr("jobsource.careers.cascade._ats.recover_via_slug_guess", lambda *a, **kw: None)
+        monkeypatch.setattr("jobsource.careers.cascade._heuristics.probe_url_patterns", lambda *a, **kw: None)
+        monkeypatch.setattr("jobsource.careers.cascade._heuristics.scan_homepage_links", lambda *a, **kw: None)
+        monkeypatch.setattr("jobsource.careers.cascade._heuristics.parse_sitemap", lambda *a, **kw: None)
+        # Prevent _finalize from making network calls
+        monkeypatch.setattr("jobsource.careers.cascade._ats.detect_ats_in_url", lambda url: None)
+
+    def test_llm_classify_hit_returns_055_confidence(self, monkeypatch):
+        from jobsource.careers.classify_llm import Anchor, CareerLinkResult
+
+        self._patch_all_deterministic_miss(
+            monkeypatch,
+            homepage_html='<html><body><a href="/careers">Careers</a></body></html>',
+        )
+        monkeypatch.setattr(
+            "jobsource.careers.cascade._classify_llm.extract_anchors",
+            lambda html, base_url: [Anchor(url="https://acme.com/careers", text="Careers")],
+        )
+        monkeypatch.setattr(
+            "jobsource.careers.cascade._classify_llm.classify_careers_link",
+            lambda anchors: CareerLinkResult(careers_url="https://acme.com/careers", confidence=0.85),
+        )
+        result = find_careers_page("https://www.acme.com", client=FakeClient())
+        assert result.method == "llm_classify"
+        assert result.confidence == 0.55  # fixed tier-level confidence from cascade
+        assert result.careers_url == "https://acme.com/careers"
+
+    def test_llm_classify_none_falls_through_to_method_none(self, monkeypatch):
+        from jobsource.careers.classify_llm import Anchor
+
+        self._patch_all_deterministic_miss(monkeypatch)
+        monkeypatch.setattr(
+            "jobsource.careers.cascade._classify_llm.extract_anchors",
+            lambda html, base_url: [Anchor(url="https://acme.com/about", text="About")],
+        )
+        monkeypatch.setattr(
+            "jobsource.careers.cascade._classify_llm.classify_careers_link",
+            lambda anchors: None,
+        )
+        result = find_careers_page("https://www.acme.com", client=FakeClient())
+        assert result.method == "none"
+        assert result.careers_url is None
+
+    def test_tier5_skipped_when_homepage_html_is_none(self, monkeypatch):
+        """When the homepage fetch returns None, Tier 5 is silently bypassed."""
+        llm_called: list[bool] = []
+        self._patch_all_deterministic_miss(monkeypatch)
+        monkeypatch.setattr(
+            "jobsource.careers.cascade._safe_get_html",
+            lambda website, client: None,
+        )
+        monkeypatch.setattr(
+            "jobsource.careers.cascade._classify_llm.classify_careers_link",
+            lambda anchors: llm_called.append(True) or None,
+        )
+        find_careers_page("https://www.acme.com", client=FakeClient())
+        assert llm_called == []
+
+    def test_tier5_not_called_when_sitemap_hits(self, monkeypatch):
+        """When sitemap resolves, the cascade returns before reaching Tier 5."""
+        llm_called: list[bool] = []
+        monkeypatch.setattr(
+            "jobsource.careers.cascade._safe_get_html",
+            lambda website, client: "<html/>",
+        )
+        monkeypatch.setattr("jobsource.careers.cascade._ats.detect_and_fetch", lambda *a, **kw: None)
+        monkeypatch.setattr("jobsource.careers.cascade._ats.recover_via_slug_guess", lambda *a, **kw: None)
+        monkeypatch.setattr("jobsource.careers.cascade._heuristics.probe_url_patterns", lambda *a, **kw: None)
+        monkeypatch.setattr("jobsource.careers.cascade._heuristics.scan_homepage_links", lambda *a, **kw: None)
+        monkeypatch.setattr(
+            "jobsource.careers.cascade._heuristics.parse_sitemap",
+            lambda *a, **kw: "https://acme.com/careers",
+        )
+        monkeypatch.setattr("jobsource.careers.cascade._ats.detect_ats_in_url", lambda url: None)
+        monkeypatch.setattr(
+            "jobsource.careers.cascade._classify_llm.classify_careers_link",
+            lambda anchors: llm_called.append(True) or None,
+        )
+        result = find_careers_page("https://www.acme.com", client=FakeClient())
+        assert result.method == "sitemap"
+        assert llm_called == []
+
+    def test_tier5_error_falls_through_gracefully(self, monkeypatch):
+        """A Tier 5 exception should not abort the cascade."""
+        self._patch_all_deterministic_miss(monkeypatch)
+        monkeypatch.setattr(
+            "jobsource.careers.cascade._classify_llm.extract_anchors",
+            lambda html, base_url: (_ for _ in ()).throw(RuntimeError("classify boom")),
+        )
+        result = find_careers_page("https://www.acme.com", client=FakeClient())
+        assert result.method == "none"
+        assert result.careers_url is None

tests/test_classify_llm.py

@@ -0,0 +1,256 @@
+"""Tests for careers/classify_llm.py — all network-free via monkeypatching."""
+from __future__ import annotations
+
+import pytest
+
+from jobsource.careers.classify_llm import (
+    Anchor,
+    CareerLinkResult,
+    JobLinkResult,
+    _LinkChoice,
+    classify_careers_link,
+    classify_job_link,
+    extract_anchors,
+)
+
+
+# ---------------------------------------------------------------------------
+# Fake helpers
+# ---------------------------------------------------------------------------
+
+
+class _FakeResult:
+    """Minimal stand-in for a Pydantic AI run result."""
+
+    def __init__(self, output: _LinkChoice) -> None:
+        self.output = output
+
+
+class _FakeAgent:
+    """Stand-in for a Pydantic AI Agent; returns a fixed _LinkChoice."""
+
+    def __init__(self, choice: _LinkChoice) -> None:
+        self._choice = choice
+
+    def run_sync(self, prompt: str) -> _FakeResult:
+        return _FakeResult(self._choice)
+
+
+class _RaisingAgent:
+    """Stand-in that raises on run_sync, simulating an LLM failure."""
+
+    def run_sync(self, prompt: str) -> _FakeResult:
+        raise RuntimeError("LLM unavailable")
+
+
+def _real_settings():
+    """Fake settings object with non-placeholder values."""
+    return type("S", (), {
+        "classifier_model": "openai:gpt-4o-mini",
+        "llm_api_key": "sk-real-key",
+    })()
+
+
+# ---------------------------------------------------------------------------
+# extract_anchors
+# ---------------------------------------------------------------------------
+
+_SAMPLE_HTML = """
+<html><body>
+  <a href="/about">About Us</a>
+  <a href="/careers">Careers</a>
+  <a href="https://external.com/jobs">External Jobs</a>
+  <a href="mailto:info@acme.com">Email</a>
+  <a href="#section">Anchor</a>
+  <a href="javascript:void(0)">JS link</a>
+  <a href="/careers">Duplicate careers link</a>
+</body></html>
+"""
+
+
+class TestExtractAnchors:
+    def test_resolves_relative_to_absolute(self):
+        anchors = extract_anchors(_SAMPLE_HTML, "https://acme.com")
+        urls = [a.url for a in anchors]
+        assert "https://acme.com/about" in urls
+        assert "https://acme.com/careers" in urls
+
+    def test_preserves_already_absolute_urls(self):
+        anchors = extract_anchors(_SAMPLE_HTML, "https://acme.com")
+        urls = [a.url for a in anchors]
+        assert "https://external.com/jobs" in urls
+
+    def test_skips_mailto(self):
+        anchors = extract_anchors(_SAMPLE_HTML, "https://acme.com")
+        assert all("mailto:" not in a.url for a in anchors)
+
+    def test_skips_hash_fragments(self):
+        anchors = extract_anchors(_SAMPLE_HTML, "https://acme.com")
+        # A pure # anchor resolves to base_url itself (https://acme.com) —
+        # it starts with http so it passes the scheme check but we skip hrefs
+        # that start with "#". Verify the JavaScript link is also gone.
+        assert all("javascript:" not in a.url for a in anchors)
+
+    def test_deduplicates_by_url(self):
+        anchors = extract_anchors(_SAMPLE_HTML, "https://acme.com")
+        urls = [a.url for a in anchors]
+        assert urls.count("https://acme.com/careers") == 1
+
+    def test_captures_anchor_text(self):
+        anchors = extract_anchors(_SAMPLE_HTML, "https://acme.com")
+        career_anchor = next(a for a in anchors if a.url == "https://acme.com/careers")
+        assert career_anchor.text == "Careers"
+
+    def test_empty_html_returns_empty_list(self):
+        assert extract_anchors("", "https://acme.com") == []
+
+    def test_html_with_no_links_returns_empty_list(self):
+        assert extract_anchors("<html><body><p>No links here.</p></body></html>", "https://acme.com") == []
+
+    def test_text_truncated_at_max_length(self):
+        long_text = "x" * 200
+        html = f'<html><body><a href="/careers">{long_text}</a></body></html>'
+        anchors = extract_anchors(html, "https://acme.com")
+        assert len(anchors) == 1
+        assert len(anchors[0].text) <= 80
+
+
+# ---------------------------------------------------------------------------
+# Graceful degradation: placeholder settings → always None
+# ---------------------------------------------------------------------------
+
+
+class TestPlaceholderGate:
+    """When classifier_model or llm_api_key is a placeholder, classifiers no-op."""
+
+    def test_classify_careers_link_returns_none_with_placeholder(self):
+        anchors = [Anchor(url="https://acme.com/careers", text="Careers")]
+        assert classify_careers_link(anchors) is None
+
+    def test_classify_job_link_returns_none_with_placeholder(self):
+        anchors = [Anchor(url="https://acme.com/jobs/123", text="Engineer")]
+        assert classify_job_link(anchors) is None
+
+    def test_classify_careers_link_returns_none_on_empty_list(self, monkeypatch):
+        monkeypatch.setattr(
+            "jobsource.careers.classify_llm.get_settings",
+            _real_settings,
+        )
+        assert classify_careers_link([]) is None
+
+    def test_classify_job_link_returns_none_on_empty_list(self, monkeypatch):
+        monkeypatch.setattr(
+            "jobsource.careers.classify_llm.get_settings",
+            _real_settings,
+        )
+        assert classify_job_link([]) is None
+
+
+# ---------------------------------------------------------------------------
+# Success paths via _build_link_agent monkeypatch
+# ---------------------------------------------------------------------------
+
+
+class TestClassifySuccess:
+    def _patch_agent(self, monkeypatch, choice: _LinkChoice) -> None:
+        monkeypatch.setattr(
+            "jobsource.careers.classify_llm.get_settings",
+            _real_settings,
+        )
+        monkeypatch.setattr(
+            "jobsource.careers.classify_llm._build_link_agent",
+            lambda instructions: _FakeAgent(choice),
+        )
+
+    def test_careers_pick_returns_mapped_url(self, monkeypatch):
+        anchors = [
+            Anchor(url="https://acme.com/about", text="About"),
+            Anchor(url="https://acme.com/careers", text="Careers"),
+        ]
+        self._patch_agent(monkeypatch, _LinkChoice(index=1, confidence=0.95))
+        result = classify_careers_link(anchors)
+        assert result is not None
+        assert isinstance(result, CareerLinkResult)
+        assert result.careers_url == "https://acme.com/careers"
+        assert result.confidence == 0.95
+
+    def test_job_pick_returns_mapped_url(self, monkeypatch):
+        anchors = [
+            Anchor(url="https://acme.com/jobs/", text="All Jobs"),
+            Anchor(url="https://acme.com/jobs/123", text="Software Engineer"),
+        ]
+        self._patch_agent(monkeypatch, _LinkChoice(index=1, confidence=0.85))
+        result = classify_job_link(anchors)
+        assert result is not None
+        assert isinstance(result, JobLinkResult)
+        assert result.job_url == "https://acme.com/jobs/123"
+        assert result.confidence == 0.85
+
+    def test_careers_first_index_zero(self, monkeypatch):
+        anchors = [Anchor(url="https://acme.com/jobs", text="Jobs")]
+        self._patch_agent(monkeypatch, _LinkChoice(index=0, confidence=0.70))
+        result = classify_careers_link(anchors)
+        assert result is not None
+        assert result.careers_url == "https://acme.com/jobs"
+
+
+# ---------------------------------------------------------------------------
+# Abstain and failure paths
+# ---------------------------------------------------------------------------
+
+
+class TestClassifyFailurePaths:
+    def _patch_real_settings(self, monkeypatch) -> None:
+        monkeypatch.setattr(
+            "jobsource.careers.classify_llm.get_settings",
+            _real_settings,
+        )
+
+    def test_abstain_index_minus_one_returns_none(self, monkeypatch):
+        self._patch_real_settings(monkeypatch)
+        monkeypatch.setattr(
+            "jobsource.careers.classify_llm._build_link_agent",
+            lambda instructions: _FakeAgent(_LinkChoice(index=-1, confidence=0.0)),
+        )
+        anchors = [Anchor(url="https://acme.com/about", text="About")]
+        assert classify_careers_link(anchors) is None
+
+    def test_out_of_range_index_returns_none(self, monkeypatch):
+        self._patch_real_settings(monkeypatch)
+        monkeypatch.setattr(
+            "jobsource.careers.classify_llm._build_link_agent",
+            lambda instructions: _FakeAgent(_LinkChoice(index=99, confidence=0.9)),
+        )
+        anchors = [Anchor(url="https://acme.com/careers", text="Careers")]
+        assert classify_careers_link(anchors) is None
+
+    def test_run_sync_exception_returns_none(self, monkeypatch):
+        self._patch_real_settings(monkeypatch)
+        monkeypatch.setattr(
+            "jobsource.careers.classify_llm._build_link_agent",
+            lambda instructions: _RaisingAgent(),
+        )
+        anchors = [Anchor(url="https://acme.com/careers", text="Careers")]
+        assert classify_careers_link(anchors) is None
+
+    def test_build_agent_exception_returns_none(self, monkeypatch):
+        self._patch_real_settings(monkeypatch)
+
+        def boom(instructions: str):
+            raise RuntimeError("import error")
+
+        monkeypatch.setattr(
+            "jobsource.careers.classify_llm._build_link_agent",
+            boom,
+        )
+        anchors = [Anchor(url="https://acme.com/careers", text="Careers")]
+        assert classify_careers_link(anchors) is None
+
+    def test_job_classify_run_sync_exception_returns_none(self, monkeypatch):
+        self._patch_real_settings(monkeypatch)
+        monkeypatch.setattr(
+            "jobsource.careers.classify_llm._build_link_agent",
+            lambda instructions: _RaisingAgent(),
+        )
+        anchors = [Anchor(url="https://acme.com/jobs/123", text="Engineer")]
+        assert classify_job_link(anchors) is None

tests/test_db.py

@@ -0,0 +1,550 @@
+"""Tests for jobsource/db.py and jobsource/pipeline.py.
+
+All tests are offline — no network calls, no real job source.
+The DB uses tmp_path (file-backed) or ":memory:" via path override.
+"""
+from __future__ import annotations
+
+import csv
+from datetime import datetime, timezone
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from jobsource.db import Database, _company_key
+from jobsource.models import CSV_COLUMNS, JobResult, JobStatus, RawJob
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _db(tmp_path: Path) -> Database:
+    """Open a fresh file-backed Database in tmp_path."""
+    return Database(path=tmp_path / "test.db")
+
+
+def _result(
+    job_id: str = "1",
+    company_name: str = "Acme",
+    website: str | None = "https://acme.com",
+    career_url: str | None = "https://acme.com/careers",
+    position_url: str | None = "https://acme.com/careers/jobs/42",
+    status: JobStatus = JobStatus.position_found,
+    company_key: str | None = None,
+) -> JobResult:
+    return JobResult(
+        job_id=job_id,
+        company_name=company_name,
+        company_key=company_key or _company_key(company_name, website),
+        website=website,
+        career_page_url=career_url,
+        open_position_url=position_url,
+        status=status,
+    )
+
+
+# ---------------------------------------------------------------------------
+# _company_key (pure function)
+# ---------------------------------------------------------------------------
+
+
+class TestCompanyKey:
+    def test_domain_stripped_of_www(self) -> None:
+        assert _company_key("Acme", "https://www.acme.com/about") == "acme.com"
+
+    def test_domain_lowercase(self) -> None:
+        assert _company_key("Foo", "https://Foo.IO") == "foo.io"
+
+    def test_no_website_uses_name(self) -> None:
+        assert _company_key("Globex Corp", None) == "globex corp"
+
+    def test_no_website_lowercased(self) -> None:
+        assert _company_key("Initech LLC", None) == "initech llc"
+
+    def test_empty_host_falls_back_to_name(self) -> None:
+        # Malformed URL with empty hostname
+        result = _company_key("BadCo", "://no-host")
+        assert result == "badco"
+
+    def test_empty_name_and_no_website_returns_unknown(self) -> None:
+        assert _company_key("", None) == "unknown"
+
+    def test_subpath_ignored(self) -> None:
+        assert _company_key("X", "https://x.example.com/jobs/42") == "x.example.com"
+
+
+# ---------------------------------------------------------------------------
+# Database.init_schema / open
+# ---------------------------------------------------------------------------
+
+
+class TestDatabaseSchema:
+    def test_creates_tables(self, tmp_path: Path) -> None:
+        db = _db(tmp_path)
+        try:
+            cur = db._conn.execute(
+                "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
+            )
+            tables = {r[0] for r in cur.fetchall()}
+            assert "companies" in tables
+            assert "jobs" in tables
+        finally:
+            db.close()
+
+    def test_idempotent_schema(self, tmp_path: Path) -> None:
+        """Opening the same DB twice must not raise."""
+        db1 = Database(path=tmp_path / "idempotent.db")
+        db1.close()
+        db2 = Database(path=tmp_path / "idempotent.db")
+        db2.close()
+
+    def test_parent_dirs_created(self, tmp_path: Path) -> None:
+        nested = tmp_path / "a" / "b" / "c" / "test.db"
+        db = Database(path=nested)
+        try:
+            assert nested.exists()
+        finally:
+            db.close()
+
+    def test_context_manager(self, tmp_path: Path) -> None:
+        with Database(path=tmp_path / "cm.db") as db:
+            assert not db.is_seen("x")
+
+
+# ---------------------------------------------------------------------------
+# is_seen / dedup
+# ---------------------------------------------------------------------------
+
+
+class TestIsSeeen:
+    def test_unseen_job_returns_false(self, tmp_path: Path) -> None:
+        with _db(tmp_path) as db:
+            assert not db.is_seen("99999")
+
+    def test_seen_after_persist(self, tmp_path: Path) -> None:
+        with _db(tmp_path) as db:
+            r = _result(job_id="42")
+            db.persist_result(r)
+            assert db.is_seen("42")
+
+    def test_different_job_ids_independent(self, tmp_path: Path) -> None:
+        with _db(tmp_path) as db:
+            db.persist_result(_result(job_id="1"))
+            assert db.is_seen("1")
+            assert not db.is_seen("2")
+
+
+# ---------------------------------------------------------------------------
+# company_key_for
+# ---------------------------------------------------------------------------
+
+
+class TestCompanyKeyFor:
+    def test_delegates_to_pure_fn(self, tmp_path: Path) -> None:
+        with _db(tmp_path) as db:
+            assert db.company_key_for("Acme", "https://acme.com") == "acme.com"
+            assert db.company_key_for("Foo LLC", None) == "foo llc"
+
+
+# ---------------------------------------------------------------------------
+# get_cached_career_url
+# ---------------------------------------------------------------------------
+
+
+class TestCachedCareerUrl:
+    def test_miss_returns_none(self, tmp_path: Path) -> None:
+        with _db(tmp_path) as db:
+            assert db.get_cached_career_url("acme.com") is None
+
+    def test_hit_after_persist(self, tmp_path: Path) -> None:
+        with _db(tmp_path) as db:
+            r = _result(job_id="1", website="https://acme.com", career_url="https://acme.com/careers")
+            db.persist_result(r)
+            assert db.get_cached_career_url("acme.com") == "https://acme.com/careers"
+
+    def test_null_career_url_not_cached(self, tmp_path: Path) -> None:
+        with _db(tmp_path) as db:
+            r = _result(job_id="1", career_url=None, status=JobStatus.needs_review)
+            db.persist_result(r)
+            assert db.get_cached_career_url("acme.com") is None
+
+
+# ---------------------------------------------------------------------------
+# upsert_company COALESCE / first_seen preservation
+# ---------------------------------------------------------------------------
+
+
+class TestUpsertCompany:
+    def test_insert_then_null_fill(self, tmp_path: Path) -> None:
+        """A NULL value in the second upsert fills the existing NULL without error."""
+        with _db(tmp_path) as db:
+            db.upsert_company("x.com", "X", None, None)
+            db.upsert_company("x.com", "X", "https://x.com", "https://x.com/careers")
+            row = db._conn.execute(
+                "SELECT website, career_url FROM companies WHERE company_key='x.com'"
+            ).fetchone()
+            assert row["website"] == "https://x.com"
+            assert row["career_url"] == "https://x.com/careers"
+
+    def test_existing_value_not_clobbered(self, tmp_path: Path) -> None:
+        """A second upsert with NULL must not overwrite an existing value."""
+        with _db(tmp_path) as db:
+            db.upsert_company("y.com", "Y", "https://y.com", "https://y.com/jobs")
+            db.upsert_company("y.com", "Y", None, None)
+            row = db._conn.execute(
+                "SELECT website, career_url FROM companies WHERE company_key='y.com'"
+            ).fetchone()
+            assert row["website"] == "https://y.com"
+            assert row["career_url"] == "https://y.com/jobs"
+
+    def test_first_seen_preserved(self, tmp_path: Path) -> None:
+        with _db(tmp_path) as db:
+            db.upsert_company("z.com", "Z", None, None)
+            row1 = db._conn.execute(
+                "SELECT first_seen FROM companies WHERE company_key='z.com'"
+            ).fetchone()
+            db.upsert_company("z.com", "Z", "https://z.com", "https://z.com/careers")
+            row2 = db._conn.execute(
+                "SELECT first_seen FROM companies WHERE company_key='z.com'"
+            ).fetchone()
+            assert row1["first_seen"] == row2["first_seen"]
+
+
+# ---------------------------------------------------------------------------
+# persist_result / upsert_job
+# ---------------------------------------------------------------------------
+
+
+class TestPersistResult:
+    def test_both_tables_populated(self, tmp_path: Path) -> None:
+        with _db(tmp_path) as db:
+            r = _result(job_id="10", company_name="Initech", website="https://initech.com")
+            db.persist_result(r)
+            job = db._conn.execute(
+                "SELECT * FROM jobs WHERE job_id='10'"
+            ).fetchone()
+            assert job is not None
+            assert job["status"] == "position_found"
+            company = db._conn.execute(
+                "SELECT * FROM companies WHERE company_key='initech.com'"
+            ).fetchone()
+            assert company is not None
+            assert company["name"] == "Initech"
+
+    def test_listed_at_null_stored_as_null(self, tmp_path: Path) -> None:
+        with _db(tmp_path) as db:
+            r = _result(job_id="20")
+            assert r.listed_at is None
+            db.persist_result(r)
+            row = db._conn.execute(
+                "SELECT listed_at FROM jobs WHERE job_id='20'"
+            ).fetchone()
+            assert row["listed_at"] is None
+
+    def test_listed_at_serialized(self, tmp_path: Path) -> None:
+        with _db(tmp_path) as db:
+            r = _result(job_id="30")
+            r.listed_at = datetime(2026, 1, 15, 12, 0, 0, tzinfo=timezone.utc)
+            db.persist_result(r)
+            row = db._conn.execute(
+                "SELECT listed_at FROM jobs WHERE job_id='30'"
+            ).fetchone()
+            assert row["listed_at"] is not None
+            assert "2026" in row["listed_at"]
+
+    def test_upsert_fills_null_position(self, tmp_path: Path) -> None:
+        """A second persist with position_url fills a prior NULL."""
+        with _db(tmp_path) as db:
+            r1 = _result(job_id="50", position_url=None, status=JobStatus.needs_review)
+            db.persist_result(r1)
+            r2 = _result(job_id="50", position_url="https://acme.com/jobs/1", status=JobStatus.position_found)
+            db.persist_result(r2)
+            row = db._conn.execute(
+                "SELECT position_url, status FROM jobs WHERE job_id='50'"
+            ).fetchone()
+            assert row["position_url"] == "https://acme.com/jobs/1"
+            assert row["status"] == "position_found"
+
+    def test_job_first_seen_preserved(self, tmp_path: Path) -> None:
+        with _db(tmp_path) as db:
+            r = _result(job_id="60", position_url=None, status=JobStatus.needs_review)
+            db.persist_result(r)
+            row1 = db._conn.execute(
+                "SELECT first_seen FROM jobs WHERE job_id='60'"
+            ).fetchone()
+            r2 = _result(job_id="60", position_url="https://acme.com/jobs/1", status=JobStatus.position_found)
+            db.persist_result(r2)
+            row2 = db._conn.execute(
+                "SELECT first_seen FROM jobs WHERE job_id='60'"
+            ).fetchone()
+            assert row1["first_seen"] == row2["first_seen"]
+
+
+# ---------------------------------------------------------------------------
+# counts
+# ---------------------------------------------------------------------------
+
+
+class TestCounts:
+    def test_empty_db(self, tmp_path: Path) -> None:
+        with _db(tmp_path) as db:
+            assert db.counts() == {}
+
+    def test_counts_grouped_by_status(self, tmp_path: Path) -> None:
+        with _db(tmp_path) as db:
+            db.persist_result(_result(job_id="1", status=JobStatus.position_found))
+            db.persist_result(_result(job_id="2", position_url=None, status=JobStatus.needs_review))
+            db.persist_result(_result(job_id="3", position_url=None, status=JobStatus.needs_review))
+            c = db.counts()
+            assert c["position_found"] == 1
+            assert c["needs_review"] == 2
+
+
+# ---------------------------------------------------------------------------
+# export_csv
+# ---------------------------------------------------------------------------
+
+
+class TestExportCsv:
+    def test_header_matches_csv_columns(self, tmp_path: Path) -> None:
+        with _db(tmp_path) as db:
+            db.persist_result(_result(job_id="1"))
+            out = tmp_path / "out.csv"
+            db.export_csv(path=out)
+        with out.open() as f:
+            reader = csv.DictReader(f)
+            assert tuple(reader.fieldnames or []) == CSV_COLUMNS
+
+    def test_null_fields_become_empty_string(self, tmp_path: Path) -> None:
+        with _db(tmp_path) as db:
+            r = _result(job_id="1", career_url=None, position_url=None, status=JobStatus.needs_review)
+            db.persist_result(r)
+            out = tmp_path / "out.csv"
+            db.export_csv(path=out)
+        with out.open() as f:
+            rows = list(csv.DictReader(f))
+        assert rows[0]["career_page_url"] == ""
+        assert rows[0]["open_position_url"] == ""
+
+    def test_complete_rows_sorted_first(self, tmp_path: Path) -> None:
+        with _db(tmp_path) as db:
+            # Insert incomplete row first, complete row second
+            db.persist_result(_result(
+                job_id="1", company_name="ZZZ Corp",
+                position_url=None, status=JobStatus.needs_review,
+            ))
+            db.persist_result(_result(
+                job_id="2", company_name="AAA Inc",
+                position_url="https://aaa.com/jobs/1", status=JobStatus.position_found,
+            ))
+            out = tmp_path / "out.csv"
+            db.export_csv(path=out)
+        with out.open() as f:
+            rows = list(csv.DictReader(f))
+        assert len(rows) == 2
+        # complete row (AAA Inc) must come first
+        assert rows[0]["open_position_url"] != ""
+        assert rows[1]["open_position_url"] == ""
+
+    def test_creates_parent_dirs(self, tmp_path: Path) -> None:
+        with _db(tmp_path) as db:
+            db.persist_result(_result())
+            nested = tmp_path / "deep" / "dir" / "results.csv"
+            db.export_csv(path=nested)
+        assert nested.exists()
+
+    def test_returns_path(self, tmp_path: Path) -> None:
+        with _db(tmp_path) as db:
+            db.persist_result(_result())
+            out = tmp_path / "r.csv"
+            returned = db.export_csv(path=out)
+        assert returned == out
+
+
+# ---------------------------------------------------------------------------
+# pipeline.run_batch — offline unit tests via monkeypatch
+# ---------------------------------------------------------------------------
+
+
+def _make_raw(job_id: str, company: str = "Acme") -> RawJob:
+    return RawJob(
+        job_id=job_id,
+        company=company,
+        linkedin_url=f"https://www.linkedin.com/jobs/view/{job_id}",
+        website=None,
+    )
+
+
+class TestRunBatch:
+    """Monkeypatched run_batch tests — no network, no real providers."""
+
+    def _patch_all(self, monkeypatch, raw_jobs: list[RawJob]) -> None:
+        """Patch all external I/O in pipeline so tests run offline."""
+        from jobsource.careers.cascade import CareersResult
+
+        # Fake job source
+        fake_source = MagicMock()
+        fake_source.fetch_recent_jobs.return_value = raw_jobs
+        monkeypatch.setattr("jobsource.pipeline._make_source", lambda _: fake_source)
+
+        # Fake resolve: always succeeds
+        monkeypatch.setattr(
+            "jobsource.pipeline.resolve_website",
+            lambda name, website=None, *, client=None: f"https://{name.lower().replace(' ', '')}.com",
+        )
+
+        # Fake cascade: returns a careers URL + free position shortcut
+        def _fake_careers(website, *, company_name=None, client=None):
+            return CareersResult(
+                careers_url=f"{website}/careers",
+                confidence=0.95,
+                method="ats:greenhouse",
+                position_url=f"{website}/careers/jobs/1",
+            )
+
+        monkeypatch.setattr("jobsource.pipeline.find_careers_page", _fake_careers)
+
+        # extract_open_position shouldn't be called (shortcut covers it),
+        # but patch defensively to detect if it is.
+        monkeypatch.setattr(
+            "jobsource.pipeline.extract_open_position",
+            lambda url, *, client=None: (None, "none"),
+        )
+
+        # build_client — return a context-manager stub
+        cm = MagicMock()
+        cm.__enter__ = lambda s: MagicMock()
+        cm.__exit__ = MagicMock(return_value=False)
+        monkeypatch.setattr("jobsource.pipeline.build_client", lambda: cm)
+
+    def test_new_jobs_processed(self, tmp_path: Path, monkeypatch) -> None:
+        from jobsource.pipeline import run_batch
+
+        # Use distinct company names so each job gets its own company_key (no cache sharing).
+        raw = [_make_raw("1", company="Alpha"), _make_raw("2", company="Beta")]
+        self._patch_all(monkeypatch, raw)
+
+        db = Database(path=tmp_path / "t.db")
+        summary = run_batch(batch_size=5, search_terms=["eng"], db=db)
+
+        assert summary.fetched == 2
+        assert summary.new == 2
+        assert summary.already_seen == 0
+        assert summary.website_resolved == 2
+        assert summary.careers_found == 2
+        assert summary.position_found == 2
+        assert summary.failed == 0
+        assert summary.coverage == 1.0
+
+    def test_cross_run_dedup(self, tmp_path: Path, monkeypatch) -> None:
+        """Re-running with the same DB yields new == 0 (dedup proven)."""
+        from jobsource.pipeline import run_batch
+
+        raw = [_make_raw("10", company="Gamma"), _make_raw("11", company="Delta")]
+        self._patch_all(monkeypatch, raw)
+
+        db = Database(path=tmp_path / "dedup.db")
+
+        # First run
+        s1 = run_batch(batch_size=5, search_terms=["eng"], db=db)
+        assert s1.new == 2
+
+        # Second run — same raw jobs returned by source, but DB already has them
+        s2 = run_batch(batch_size=5, search_terms=["eng"], db=db)
+        assert s2.new == 0
+        assert s2.already_seen == 2
+
+    def test_batch_size_cap(self, tmp_path: Path, monkeypatch) -> None:
+        """Only batch_size new jobs are processed even if more are available."""
+        from jobsource.pipeline import run_batch
+
+        # Distinct companies to avoid career-URL cache sharing across jobs.
+        raw = [_make_raw(str(i), company=f"Co{i}") for i in range(10)]
+        self._patch_all(monkeypatch, raw)
+
+        db = Database(path=tmp_path / "cap.db")
+        summary = run_batch(batch_size=3, search_terms=["eng"], db=db)
+        assert summary.new == 3
+        assert summary.position_found == 3
+
+    def test_resolve_failure_marks_needs_review(self, tmp_path: Path, monkeypatch) -> None:
+        """Jobs where resolve_website returns None get needs_review, not failed."""
+        from jobsource.pipeline import run_batch
+
+        raw = [_make_raw("99")]
+        self._patch_all(monkeypatch, raw)
+        # Override resolve to return None (unresolvable)
+        monkeypatch.setattr(
+            "jobsource.pipeline.resolve_website",
+            lambda *a, **kw: None,
+        )
+
+        db = Database(path=tmp_path / "nr.db")
+        summary = run_batch(batch_size=5, search_terms=["eng"], db=db)
+        assert summary.needs_review == 1
+        assert summary.failed == 0
+        assert summary.website_resolved == 0
+
+    def test_csv_written_after_batch(self, tmp_path: Path, monkeypatch) -> None:
+        """Data is persisted by run_batch and export_csv produces the right columns."""
+        from jobsource.pipeline import run_batch
+
+        raw = [_make_raw("7", company="SevenCo")]
+        self._patch_all(monkeypatch, raw)
+
+        csv_path = tmp_path / "out.csv"
+        # Pass db explicitly so own_db=False → database is NOT closed by run_batch.
+        db = Database(path=tmp_path / "t.db")
+
+        run_batch(batch_size=5, search_terms=["eng"], db=db)
+
+        # Export to a controlled path and verify the header/contract.
+        db.export_csv(path=csv_path)
+
+        assert csv_path.exists()
+        with csv_path.open() as f:
+            reader = csv.DictReader(f)
+            assert tuple(reader.fieldnames or []) == CSV_COLUMNS
+
+    def test_one_failure_does_not_abort_batch(self, tmp_path: Path, monkeypatch) -> None:
+        """A cascade exception for one company doesn't abort others.
+
+        Cascade errors land in needs_review (not failed — they are recoverable/transient).
+        Each job has a distinct company so cache cannot mask the flaky call.
+        """
+        from jobsource.pipeline import run_batch
+        from jobsource.careers.cascade import CareersResult
+
+        # Distinct companies → distinct company_keys → no cache sharing.
+        raw = [
+            _make_raw("A", company="AlphaCo"),
+            _make_raw("B", company="BetaCo"),
+            _make_raw("C", company="GammaCo"),
+        ]
+        self._patch_all(monkeypatch, raw)
+
+        call_count = {"n": 0}
+
+        def _flaky_careers(website, *, company_name=None, client=None):
+            call_count["n"] += 1
+            if call_count["n"] == 2:
+                raise RuntimeError("simulated cascade crash")
+            return CareersResult(
+                careers_url=f"{website}/careers",
+                confidence=0.9,
+                method="url_pattern",
+                position_url=f"{website}/careers/jobs/1",
+            )
+
+        monkeypatch.setattr("jobsource.pipeline.find_careers_page", _flaky_careers)
+
+        db = Database(path=tmp_path / "fault.db")
+        summary = run_batch(batch_size=5, search_terms=["eng"], db=db)
+
+        # All 3 attempted; cascade exception → needs_review (not failed); 2 fully resolved.
+        assert summary.new == 3
+        assert summary.needs_review == 1
+        assert summary.failed == 0
+        assert summary.position_found == 2

tests/test_extract.py

@@ -0,0 +1,309 @@
+"""Tests for extract.py (Stage 3) — all network-free via monkeypatching."""
+from __future__ import annotations
+
+import pytest
+
+from jobsource.careers.ats import ATSBoard, ATSFetch
+from jobsource.careers.classify_llm import Anchor, JobLinkResult
+from jobsource.extract import _find_job_posting_url, _jsonld_job_url, extract_open_position
+
+
+# ---------------------------------------------------------------------------
+# Fake helpers
+# ---------------------------------------------------------------------------
+
+
+class FakeResponse:
+    def __init__(self, status_code: int, text: str = "", url: str = "https://example.com") -> None:
+        self.status_code = status_code
+        self.text = text
+        self.url = url
+
+
+class FakeClient:
+    pass
+
+
+# ---------------------------------------------------------------------------
+# _find_job_posting_url (pure function)
+# ---------------------------------------------------------------------------
+
+
+class TestFindJobPostingUrl:
+    def test_flat_dict_job_posting(self):
+        data = {"@type": "JobPosting", "url": "https://acme.com/jobs/1"}
+        assert _find_job_posting_url(data) == "https://acme.com/jobs/1"
+
+    def test_list_of_nodes(self):
+        data = [
+            {"@type": "Organization", "name": "Acme"},
+            {"@type": "JobPosting", "url": "https://acme.com/jobs/2"},
+        ]
+        assert _find_job_posting_url(data) == "https://acme.com/jobs/2"
+
+    def test_graph_expansion(self):
+        data = {
+            "@context": "https://schema.org",
+            "@graph": [
+                {"@type": "WebPage"},
+                {"@type": "JobPosting", "url": "https://acme.com/jobs/3"},
+            ],
+        }
+        assert _find_job_posting_url(data) == "https://acme.com/jobs/3"
+
+    def test_type_as_list(self):
+        data = {"@type": ["Thing", "JobPosting"], "url": "https://acme.com/jobs/4"}
+        assert _find_job_posting_url(data) == "https://acme.com/jobs/4"
+
+    def test_no_job_posting_returns_none(self):
+        data = {"@type": "Organization", "name": "Acme"}
+        assert _find_job_posting_url(data) is None
+
+    def test_job_posting_without_url_returns_none(self):
+        data = {"@type": "JobPosting"}
+        assert _find_job_posting_url(data) is None
+
+    def test_scalar_input_returns_none(self):
+        assert _find_job_posting_url("not a dict") is None
+        assert _find_job_posting_url(42) is None
+
+
+# ---------------------------------------------------------------------------
+# _jsonld_job_url (parses HTML)
+# ---------------------------------------------------------------------------
+
+
+class TestJsonldJobUrl:
+    def test_finds_url_in_json_ld_block(self):
+        html = """
+        <html><head>
+          <script type="application/ld+json">
+          {"@type": "JobPosting", "url": "https://acme.com/jobs/99"}
+          </script>
+        </head></html>
+        """
+        assert _jsonld_job_url(html) == "https://acme.com/jobs/99"
+
+    def test_invalid_json_block_skipped(self):
+        html = """
+        <html><head>
+          <script type="application/ld+json">NOT JSON{{</script>
+          <script type="application/ld+json">{"@type": "JobPosting", "url": "https://ok.com/job/1"}</script>
+        </head></html>
+        """
+        assert _jsonld_job_url(html) == "https://ok.com/job/1"
+
+    def test_no_job_posting_returns_none(self):
+        html = """
+        <html><head>
+          <script type="application/ld+json">{"@type": "Organization"}</script>
+        </head></html>
+        """
+        assert _jsonld_job_url(html) is None
+
+    def test_no_script_returns_none(self):
+        assert _jsonld_job_url("<html><body>No scripts</body></html>") is None
+
+
+# ---------------------------------------------------------------------------
+# extract_open_position — tier integration (network-free)
+# ---------------------------------------------------------------------------
+
+
+class TestExtractOpenPositionTiers:
+    def _patch_base(self, monkeypatch, *, html: str | None = None) -> None:
+        """Patch ATS detect functions to miss and HTTP fetch to return html."""
+        monkeypatch.setattr(
+            "jobsource.extract._ats.detect_ats_in_url",
+            lambda url: None,
+        )
+        monkeypatch.setattr(
+            "jobsource.extract._ats.detect_ats_in_html",
+            lambda html: None,
+        )
+        if html is not None:
+            monkeypatch.setattr(
+                "jobsource.extract.request_with_retries",
+                lambda *a, **kw: FakeResponse(200, html),
+            )
+
+    # -- Tier 1: ATS from URL ------------------------------------------------
+
+    def test_ats_from_url_returns_first_job(self, monkeypatch):
+        board = ATSBoard(
+            ats_name="greenhouse",
+            slug="acme",
+            careers_url="https://boards.greenhouse.io/acme",
+        )
+        monkeypatch.setattr(
+            "jobsource.extract._ats.detect_ats_in_url",
+            lambda url: board,
+        )
+        monkeypatch.setattr(
+            "jobsource.extract._ats._FETCH_DISPATCH",
+            {"greenhouse": lambda b, c: ATSFetch(first_url="https://boards.greenhouse.io/acme/jobs/1", job_count=5)},
+        )
+        url, method = extract_open_position(
+            "https://boards.greenhouse.io/acme", client=FakeClient()
+        )
+        assert url == "https://boards.greenhouse.io/acme/jobs/1"
+        assert method == "ats:greenhouse"
+
+    def test_ats_from_url_with_no_jobs_falls_through(self, monkeypatch):
+        board = ATSBoard(
+            ats_name="lever",
+            slug="acme",
+            careers_url="https://jobs.lever.co/acme",
+        )
+        monkeypatch.setattr("jobsource.extract._ats.detect_ats_in_url", lambda url: board)
+        monkeypatch.setattr(
+            "jobsource.extract._ats._FETCH_DISPATCH",
+            {"lever": lambda b, c: ATSFetch(first_url=None, job_count=0)},
+        )
+        monkeypatch.setattr(
+            "jobsource.extract._ats.detect_ats_in_html", lambda html: None
+        )
+        # Page has no JSON-LD, no job-path anchors, LLM returns None.
+        # Use a neutral href that won't trigger the job-path pattern.
+        monkeypatch.setattr(
+            "jobsource.extract.request_with_retries",
+            lambda *a, **kw: FakeResponse(200, "<html><body><a href='/team'>Team</a></body></html>"),
+        )
+        monkeypatch.setattr("jobsource.extract.classify_job_link", lambda anchors: None)
+        url, method = extract_open_position("https://acme.example.com/careers", client=FakeClient())
+        assert url is None
+        assert method == "none"
+
+    # -- Tier 1b: ATS from page HTML ----------------------------------------
+
+    def test_ats_from_html_returns_first_job(self, monkeypatch):
+        gh_board = ATSBoard(
+            ats_name="greenhouse",
+            slug="vercel",
+            careers_url="https://boards.greenhouse.io/vercel",
+        )
+        monkeypatch.setattr("jobsource.extract._ats.detect_ats_in_url", lambda url: None)
+        monkeypatch.setattr("jobsource.extract._ats.detect_ats_in_html", lambda html: gh_board)
+        monkeypatch.setattr(
+            "jobsource.extract._ats._FETCH_DISPATCH",
+            {"greenhouse": lambda b, c: ATSFetch(first_url="https://boards.greenhouse.io/vercel/jobs/42", job_count=73)},
+        )
+        monkeypatch.setattr(
+            "jobsource.extract.request_with_retries",
+            lambda *a, **kw: FakeResponse(200, '<script src="greenhouse"></script>'),
+        )
+        url, method = extract_open_position("https://vercel.com/careers", client=FakeClient())
+        assert url == "https://boards.greenhouse.io/vercel/jobs/42"
+        assert method == "ats:greenhouse"
+
+    # -- Tier 2: JSON-LD ----------------------------------------------------
+
+    def test_jsonld_returns_job_url(self, monkeypatch):
+        html = """
+        <html><head>
+          <script type="application/ld+json">
+          {"@type": "JobPosting", "url": "https://acme.com/jobs/77"}
+          </script>
+        </head><body></body></html>
+        """
+        self._patch_base(monkeypatch, html=html)
+        url, method = extract_open_position("https://acme.com/careers", client=FakeClient())
+        assert url == "https://acme.com/jobs/77"
+        assert method == "json_ld"
+
+    # -- Tier 3: job-like anchor --------------------------------------------
+
+    def test_job_anchor_matched_by_pattern(self, monkeypatch):
+        html = """
+        <html><body>
+          <a href="/about">About</a>
+          <a href="/jobs/software-engineer-123">Software Engineer</a>
+        </body></html>
+        """
+        self._patch_base(monkeypatch, html=html)
+        url, method = extract_open_position("https://acme.com/careers", client=FakeClient())
+        assert url == "https://acme.com/jobs/software-engineer-123"
+        assert method == "anchor"
+
+    def test_opening_anchor_matched(self, monkeypatch):
+        html = '<html><body><a href="/openings/42">Open Role</a></body></html>'
+        self._patch_base(monkeypatch, html=html)
+        url, method = extract_open_position("https://acme.com/careers", client=FakeClient())
+        assert url == "https://acme.com/openings/42"
+        assert method == "anchor"
+
+    # -- Tier 4: LLM --------------------------------------------------------
+
+    def test_llm_classify_fires_when_no_job_anchor(self, monkeypatch):
+        html = '<html><body><a href="/about">About</a><a href="/team">Team</a></body></html>'
+        self._patch_base(monkeypatch, html=html)
+        monkeypatch.setattr(
+            "jobsource.extract.classify_job_link",
+            lambda anchors: JobLinkResult(job_url="https://acme.com/about", confidence=0.60),
+        )
+        url, method = extract_open_position("https://acme.com/careers", client=FakeClient())
+        assert url == "https://acme.com/about"
+        assert method == "llm_classify"
+
+    def test_llm_returns_none_falls_through_to_none(self, monkeypatch):
+        html = '<html><body><a href="/about">About</a></body></html>'
+        self._patch_base(monkeypatch, html=html)
+        monkeypatch.setattr(
+            "jobsource.extract.classify_job_link",
+            lambda anchors: None,
+        )
+        url, method = extract_open_position("https://acme.com/careers", client=FakeClient())
+        assert url is None
+        assert method == "none"
+
+    # -- Page fetch failure -------------------------------------------------
+
+    def test_fetch_failure_returns_none(self, monkeypatch):
+        monkeypatch.setattr("jobsource.extract._ats.detect_ats_in_url", lambda url: None)
+        monkeypatch.setattr(
+            "jobsource.extract.request_with_retries",
+            lambda *a, **kw: (_ for _ in ()).throw(RuntimeError("timeout")),
+        )
+        url, method = extract_open_position("https://acme.com/careers", client=FakeClient())
+        assert url is None
+        assert method == "none"
+
+    def test_non_200_fetch_returns_none(self, monkeypatch):
+        monkeypatch.setattr("jobsource.extract._ats.detect_ats_in_url", lambda url: None)
+        monkeypatch.setattr(
+            "jobsource.extract.request_with_retries",
+            lambda *a, **kw: FakeResponse(404, ""),
+        )
+        url, method = extract_open_position("https://acme.com/careers", client=FakeClient())
+        assert url is None
+        assert method == "none"
+
+    # -- All tiers miss ------------------------------------------------------
+
+    def test_all_miss_returns_none_none(self, monkeypatch):
+        html = '<html><body><a href="/about">About</a></body></html>'
+        self._patch_base(monkeypatch, html=html)
+        monkeypatch.setattr("jobsource.extract.classify_job_link", lambda anchors: None)
+        url, method = extract_open_position("https://acme.com/careers", client=FakeClient())
+        assert url is None
+        assert method == "none"
+
+    # -- ATS tier error falls through ---------------------------------------
+
+    def test_ats_fetch_error_falls_through(self, monkeypatch):
+        board = ATSBoard(ats_name="ashby", slug="acme", careers_url="https://jobs.ashbyhq.com/acme")
+        monkeypatch.setattr("jobsource.extract._ats.detect_ats_in_url", lambda url: board)
+        monkeypatch.setattr(
+            "jobsource.extract._ats._FETCH_DISPATCH",
+            {"ashby": lambda b, c: (_ for _ in ()).throw(RuntimeError("api down"))},
+        )
+        monkeypatch.setattr("jobsource.extract._ats.detect_ats_in_html", lambda html: None)
+        html = '<html><body><a href="/jobs/42">Engineer</a></body></html>'
+        monkeypatch.setattr(
+            "jobsource.extract.request_with_retries",
+            lambda *a, **kw: FakeResponse(200, html),
+        )
+        url, method = extract_open_position("https://jobs.ashbyhq.com/acme", client=FakeClient())
+        # urljoin("https://jobs.ashbyhq.com/acme", "/jobs/42") → root-relative
+        assert url == "https://jobs.ashbyhq.com/jobs/42"
+        assert method == "anchor"