ldy/JobSourceAgent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

agent-fallback

Browse Source
This commit is contained in:
ldy
2026-06-17 23:32:14 -04:00
parent 41037fb2f5
commit ed3ec1d2c6
5 changed files with 735 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
CLAUDE.md
View File

@@ -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).

306
jobsource/agent_fallback.py
View File

@@ -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".
# 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:
# 1. Navigate to the company website and locate the careers/jobs page.
# 2. From the careers page, return the URL of one open position.
# Graceful degradation: if Browser Use / Playwright / LLM key are unavailable, log clearly
# and return (careers_url=None, position_url=None) so the pipeline records needs_review.
from __future__ import annotations
import asyncio
import glob
import logging
import os
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

29
jobsource/careers/cascade.py
View File

@@ -7,10 +7,11 @@ Cascade order (return early on first success):
3. Homepage scan → heuristics.scan_homepage_links() confidence 0.60
4. Sitemap → heuristics.parse_sitemap() confidence 0.50
5. Cheap-LLM → classify_llm.classify_careers_link() confidence 0.55
6. Browser agent → agent_fallback (not implemented in this phase)
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
@@ -23,7 +24,9 @@ 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
@@ -41,7 +44,7 @@ class CareersResult(BaseModel):
careers_url: str | None = None
confidence: float = 0.0
# method values: "ats:{name}" | "url_pattern" | "homepage_scan" | "sitemap" | "llm_classify" | "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).
@@ -163,6 +166,26 @@ def find_careers_page(
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()

22
jobsource/extract.py
View File

@@ -10,11 +10,13 @@ Cascade order (return early on first hit):
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. Miss — return (None, "none").
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" | "none"
"ats:{name}" | "json_ld" | "anchor" | "llm_classify" | "browser_agent" | "none"
"""
from __future__ import annotations
@@ -26,8 +28,10 @@ from urllib.parse import urlsplit
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__)
@@ -206,6 +210,20 @@ def extract_open_position(
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"

386
tests/test_agent_fallback.py Normal file
View File

@@ -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"