From 99dbe57acfe4e9585975a1c6acf3e676ad0d62a5 Mon Sep 17 00:00:00 2001 From: sohail0992 Date: Mon, 10 Aug 2026 17:04:28 +0200 Subject: [PATCH 1/2] Cap claude CLI calls per run to protect the 5-hour usage window The CLI path had no ceiling on how many subprocess calls a single `applypilot run` could make. Score, judge, tailor, and cover all share one LLMClient, so a bad batch (e.g. a pile of junk scraped "jobs") could burn through the entire subscription window before the first stage even finished, with no way to stop it short of killing the process. - LLMClient tracks claude_cli_call_count against a CLAUDE_CLI_CALL_BUDGET env var (default 40); _chat_claude_cli raises ClaudeUsageLimitError before spawning another subprocess once the budget is hit, so the cap applies uniformly to every stage without each one having to check it. - scorer.py and cover_letter.py were catching ClaudeUsageLimitError inside a blind `except Exception`, converting it into a fake per-job error and continuing -- so even a real limit hit never stopped the loop, it just burned through every remaining job. Both now let it propagate and break the loop, flushing whatever's already done (same pattern tailor.py already had). - Bumped the CLI subprocess timeout from 120s to 240s -- the tailoring prompt (skills boundary, hard rules, full resume) is large enough that a cold `claude -p` call was hitting the old timeout on legitimate requests, not just rate limits. --- src/applypilot/llm.py | 70 ++++++++++++++++++++++++-- src/applypilot/scoring/cover_letter.py | 14 +++++- src/applypilot/scoring/scorer.py | 19 +++++-- src/applypilot/scoring/tailor.py | 66 +++++++++++++++++------- 4 files changed, 142 insertions(+), 27 deletions(-) diff --git a/src/applypilot/llm.py b/src/applypilot/llm.py index 5c5d7b8b3..5c9fb5217 100644 --- a/src/applypilot/llm.py +++ b/src/applypilot/llm.py @@ -22,6 +22,34 @@ log = logging.getLogger(__name__) + +class ClaudeUsageLimitError(RuntimeError): + """Raised when the `claude` CLI reports a usage/rate limit, not a transient failure. + + Distinguished from other CLI failures (bad prompt, malformed output, timeout) + so callers can stop a batch run immediately instead of burning retries into + a wall that won't clear until the limit window resets. + """ + + +# Substrings the `claude` CLI itself checks for when classifying provider +# errors (see its own "Anthropic API:" error list) — matched case-insensitively +# against stderr/result text to tell a real limit from any other failure. +_USAGE_LIMIT_INDICATORS = ( + "usage limit reached", + "rate limited", + "rate limit", + "credit balance too low", + "overloaded", + "429", + "529", +) + + +def _is_usage_limit_message(text: str) -> bool: + lowered = text.lower() + return any(indicator in lowered for indicator in _USAGE_LIMIT_INDICATORS) + # --------------------------------------------------------------------------- # Provider detection # --------------------------------------------------------------------------- @@ -87,6 +115,14 @@ def _detect_provider() -> tuple[str, str, str]: _MAX_RETRIES = 5 _TIMEOUT = 120 # seconds +# Hard cap on `claude` CLI calls for a single `applypilot run` process, +# shared across every stage (score/judge/tailor/cover all route through the +# same LLMClient singleton). Once hit, the next call raises instead of +# spawning a subprocess, so a bad batch (e.g. junk scraped "jobs") can't +# burn through the whole 5-hour subscription window unattended. There's no +# real usage/quota API to read, so this is a local proxy, not a true count. +_CLAUDE_CLI_CALL_BUDGET = int(os.environ.get("CLAUDE_CLI_CALL_BUDGET", "40")) + # Base wait on first 429/503 (doubles each retry, caps at 60s). # Gemini free tier is 15 RPM = 4s minimum between requests; 10s gives headroom. _RATE_LIMIT_BASE_WAIT = 10 @@ -96,7 +132,7 @@ def _detect_provider() -> tuple[str, str, str]: _GEMINI_NATIVE_BASE = "https://generativelanguage.googleapis.com/v1beta" _ANTHROPIC_BASE = "https://api.anthropic.com/v1" _CLAUDE_CLI_MARKER = "claude-cli" # not a real URL — signals the subprocess path -_CLAUDE_CLI_TIMEOUT = 120 # seconds +_CLAUDE_CLI_TIMEOUT = 240 # seconds class LLMClient: @@ -118,6 +154,8 @@ def __init__(self, base_url: str, model: str, api_key: str) -> None: self._is_gemini: bool = base_url.startswith(_GEMINI_COMPAT_BASE) self._is_anthropic: bool = base_url == _ANTHROPIC_BASE self._is_claude_cli: bool = base_url == _CLAUDE_CLI_MARKER + self.claude_cli_call_count = 0 + self.claude_cli_call_budget = _CLAUDE_CLI_CALL_BUDGET # -- Native Gemini API -------------------------------------------------- @@ -225,6 +263,13 @@ def _chat_claude_cli( single-turn). `--disallowedTools *` keeps this a pure text completion — no file/bash access. """ + if self.claude_cli_call_count >= self.claude_cli_call_budget: + raise ClaudeUsageLimitError( + f"Local call budget exhausted ({self.claude_cli_call_count}/{self.claude_cli_call_budget} " + "calls this run) -- stopping before another CLI call to protect the 5-hour usage window. " + "Set CLAUDE_CLI_CALL_BUDGET to raise it." + ) + system_text = "\n".join(msg.get("content", "") for msg in messages if msg["role"] == "system") turns = [msg for msg in messages if msg["role"] != "system"] prompt = ( @@ -233,23 +278,40 @@ def _chat_claude_cli( else "\n\n".join(f"{msg['role']}: {msg['content']}" for msg in turns) ) + # --disallowedTools takes a variadic list ("") and swallows + # every bare argument after it, including a trailing prompt — so the + # prompt is sent via stdin instead of as a positional arg. cmd = ["claude", "-p", "--model", self.model, "--output-format", "json", "--disallowedTools", "*"] if system_text: cmd += ["--system-prompt", system_text] - cmd.append(prompt) + + # ANTHROPIC_API_KEY in the subprocess env makes the CLI prefer that + # (metered) auth source over the claude.ai subscription login this + # path is meant to use, which it then refuses to do — unset it here. + cli_env = os.environ.copy() + cli_env.pop("ANTHROPIC_API_KEY", None) result = subprocess.run( cmd, + input=prompt, capture_output=True, text=True, timeout=_CLAUDE_CLI_TIMEOUT, check=False, + env=cli_env, ) if result.returncode != 0: - raise RuntimeError(f"claude CLI exited {result.returncode}: {result.stderr[:300]}") + msg = f"claude CLI exited {result.returncode}: {result.stderr[:300]}" + if _is_usage_limit_message(result.stderr): + raise ClaudeUsageLimitError(msg) + raise RuntimeError(msg) data = json.loads(result.stdout) if data.get("is_error"): - raise RuntimeError(f"claude CLI error: {data.get('result', 'unknown')}") + msg = f"claude CLI error: {data.get('result', 'unknown')}" + if _is_usage_limit_message(str(data.get("result", ""))): + raise ClaudeUsageLimitError(msg) + raise RuntimeError(msg) + self.claude_cli_call_count += 1 return data["result"] # -- OpenAI-compat API -------------------------------------------------- diff --git a/src/applypilot/scoring/cover_letter.py b/src/applypilot/scoring/cover_letter.py index c16cdd5f7..5d04d39ca 100644 --- a/src/applypilot/scoring/cover_letter.py +++ b/src/applypilot/scoring/cover_letter.py @@ -13,7 +13,7 @@ from applypilot.config import COVER_LETTER_DIR, RESUME_PATH, load_profile from applypilot.database import get_connection, get_jobs_by_stage -from applypilot.llm import get_client +from applypilot.llm import ClaudeUsageLimitError, get_client from applypilot.scoring.validator import ( BANNED_WORDS, LLM_LEAK_PHRASES, @@ -230,6 +230,7 @@ def run_cover_letters(min_score: int = 7, limit: int = 20, completed = 0 results: list[dict] = [] error_count = 0 + stopped_on_limit = False for job in jobs: completed += 1 @@ -268,6 +269,10 @@ def run_cover_letters(min_score: int = 7, limit: int = 20, "%d/%d [OK] | %.1f jobs/min | %s", completed, len(jobs), rate * 60, result["title"][:40], ) + except ClaudeUsageLimitError as e: + log.error("%d/%d [USAGE LIMIT] stopping run early -- %s", completed, len(jobs), e) + stopped_on_limit = True + break except Exception as e: result = { "url": job["url"], "title": job["title"], "site": job["site"], @@ -296,10 +301,15 @@ def run_cover_letters(min_score: int = 7, limit: int = 20, conn.commit() elapsed = time.time() - t0 - log.info("Cover letters done in %.1fs: %d generated, %d errors", elapsed, saved, error_count) + log.info( + "Cover letters done in %.1fs: %d generated, %d errors%s", + elapsed, saved, error_count, + " (stopped early: usage limit hit)" if stopped_on_limit else "", + ) return { "generated": saved, "errors": error_count, "elapsed": elapsed, + "stopped_on_limit": stopped_on_limit, } diff --git a/src/applypilot/scoring/scorer.py b/src/applypilot/scoring/scorer.py index 97692d5f7..a67397f7a 100644 --- a/src/applypilot/scoring/scorer.py +++ b/src/applypilot/scoring/scorer.py @@ -13,7 +13,7 @@ from applypilot.config import RESUME_PATH, load_profile from applypilot.database import get_connection, get_jobs_by_stage -from applypilot.llm import get_client +from applypilot.llm import ClaudeUsageLimitError, get_client log = logging.getLogger(__name__) @@ -96,6 +96,8 @@ def score_job(resume_text: str, job: dict) -> dict: client = get_client() response = client.chat(messages, max_tokens=512, temperature=0.2) return _parse_score_response(response) + except ClaudeUsageLimitError: + raise except Exception as e: log.error("LLM error scoring job '%s': %s", job.get("title", "?"), e) return {"score": 0, "keywords": "", "reasoning": f"LLM error: {e}"} @@ -137,8 +139,14 @@ def run_scoring(limit: int = 0, rescore: bool = False) -> dict: errors = 0 results: list[dict] = [] + stopped_on_limit = False for job in jobs: - result = score_job(resume_text, job) + try: + result = score_job(resume_text, job) + except ClaudeUsageLimitError as e: + log.error("[%d/%d] [USAGE LIMIT] stopping run early -- %s", completed, len(jobs), e) + stopped_on_limit = True + break result["url"] = job["url"] completed += 1 @@ -162,7 +170,11 @@ def run_scoring(limit: int = 0, rescore: bool = False) -> dict: conn.commit() elapsed = time.time() - t0 - log.info("Done: %d scored in %.1fs (%.1f jobs/sec)", len(results), elapsed, len(results) / elapsed if elapsed > 0 else 0) + log.info( + "Done: %d scored in %.1fs (%.1f jobs/sec)%s", + len(results), elapsed, len(results) / elapsed if elapsed > 0 else 0, + " (stopped early: usage limit hit)" if stopped_on_limit else "", + ) # Score distribution dist = conn.execute(""" @@ -177,4 +189,5 @@ def run_scoring(limit: int = 0, rescore: bool = False) -> dict: "errors": errors, "elapsed": elapsed, "distribution": distribution, + "stopped_on_limit": stopped_on_limit, } diff --git a/src/applypilot/scoring/tailor.py b/src/applypilot/scoring/tailor.py index 352fb5ff9..a8d8ef86b 100644 --- a/src/applypilot/scoring/tailor.py +++ b/src/applypilot/scoring/tailor.py @@ -18,7 +18,7 @@ from applypilot.config import RESUME_PATH, TAILORED_DIR, load_profile from applypilot.database import get_connection, get_jobs_by_stage -from applypilot.llm import get_client +from applypilot.llm import ClaudeUsageLimitError, get_client from applypilot.scoring.validator import ( BANNED_WORDS, FABRICATION_WATCHLIST, @@ -484,6 +484,33 @@ def run_tailoring(min_score: int = 7, limit: int = 20, results: list[dict] = [] stats: dict[str, int] = {"approved": 0, "failed_validation": 0, "failed_judge": 0, "error": 0} + # Flush to DB every BATCH_SIZE jobs (instead of one commit for the whole + # run) so an interrupt or a hit rate limit only loses the in-progress + # batch, not everything already tailored. + BATCH_SIZE = 5 + _success_statuses = {"approved", "approved_with_judge_warning"} + pending: list[dict] = [] + + def _flush(batch: list[dict]) -> None: + if not batch: + return + now = datetime.now(timezone.utc).isoformat() + for r in batch: + if r["status"] in _success_statuses: + conn.execute( + "UPDATE jobs SET tailored_resume_path=?, tailored_at=?, " + "tailor_attempts=COALESCE(tailor_attempts,0)+1 WHERE url=?", + (r["path"], now, r["url"]), + ) + else: + conn.execute( + "UPDATE jobs SET tailor_attempts=COALESCE(tailor_attempts,0)+1 WHERE url=?", + (r["url"],), + ) + conn.commit() + + stopped_on_limit = False + for job in jobs: completed += 1 try: @@ -534,6 +561,19 @@ def run_tailoring(min_score: int = 7, limit: int = 20, "status": report["status"], "attempts": report["attempts"], } + except ClaudeUsageLimitError as e: + # Not a per-job failure -- retrying would just burn more calls + # into the same wall. Flush what's already done and stop; this + # job's tailor_attempts is left untouched so it's retried fresh + # once the limit window resets. + log.error( + "%d/%d [USAGE LIMIT] stopping run early -- %s", + completed, len(jobs), e, + ) + _flush(pending) + pending = [] + stopped_on_limit = True + break except Exception as e: result = { "url": job["url"], "title": job["title"], "site": job["site"], @@ -542,6 +582,7 @@ def run_tailoring(min_score: int = 7, limit: int = 20, log.error("%d/%d [ERROR] %s -- %s", completed, len(jobs), job["title"][:40], e) results.append(result) + pending.append(result) stats[result.get("status", "error")] = stats.get(result.get("status", "error"), 0) + 1 elapsed = time.time() - t0 @@ -555,31 +596,19 @@ def run_tailoring(min_score: int = 7, limit: int = 20, result["title"][:40], ) - # Persist to DB: increment attempt counter for ALL, save path only for approved - now = datetime.now(timezone.utc).isoformat() - _success_statuses = {"approved", "approved_with_judge_warning"} - for r in results: - if r["status"] in _success_statuses: - conn.execute( - "UPDATE jobs SET tailored_resume_path=?, tailored_at=?, " - "tailor_attempts=COALESCE(tailor_attempts,0)+1 WHERE url=?", - (r["path"], now, r["url"]), - ) - else: - conn.execute( - "UPDATE jobs SET tailor_attempts=COALESCE(tailor_attempts,0)+1 WHERE url=?", - (r["url"],), - ) - conn.commit() + if len(pending) >= BATCH_SIZE or completed == len(jobs): + _flush(pending) + pending = [] elapsed = time.time() - t0 log.info( - "Tailoring done in %.1fs: %d approved, %d failed_validation, %d failed_judge, %d errors", + "Tailoring done in %.1fs: %d approved, %d failed_validation, %d failed_judge, %d errors%s", elapsed, stats.get("approved", 0), stats.get("failed_validation", 0), stats.get("failed_judge", 0), stats.get("error", 0), + " (stopped early: usage limit hit)" if stopped_on_limit else "", ) return { @@ -587,4 +616,5 @@ def run_tailoring(min_score: int = 7, limit: int = 20, "failed": stats.get("failed_validation", 0) + stats.get("failed_judge", 0), "errors": stats.get("error", 0), "elapsed": elapsed, + "stopped_on_limit": stopped_on_limit, } From 473175398c6e89ec1157b4346c46f8d19d90ff9a Mon Sep 17 00:00:00 2001 From: sohail0992 Date: Mon, 10 Aug 2026 17:04:43 +0200 Subject: [PATCH 2/2] Stop discovery from capturing non-job API noise as jobs collect_page_intelligence()'s response listener captures anything with "/api/" in the URL, which also catches telemetry, auth, and consent endpoints a job board's own frontend fires on every page load (e.g. talent.com's /api/auth/get-session and /api/telemetry/web-vitals, onetrust.com's geolocation lookup). These were reaching the LLM judge and, on judge error, getting kept and stored as "jobs" -- so a chunk of every run's LLM budget (see previous commit) went toward scoring/judging garbage that was never a job posting. - Added a denylist (_is_noise_url) applied at capture time, before a response is even considered for judging, and again as a backstop right before the DB insert in case anything slips through. - judge_api_responses() was also fail-open: on any error (LLM or otherwise) it kept the response rather than dropping it, which is how a real usage-limit hit turned into "keep everything for the rest of the batch." Now it fails closed and stops the judging pass entirely on a usage-limit hit instead of grinding through the rest. --- src/applypilot/discovery/smartextract.py | 32 +++++++++++++++++++++--- 1 file changed, 29 insertions(+), 3 deletions(-) diff --git a/src/applypilot/discovery/smartextract.py b/src/applypilot/discovery/smartextract.py index cf49a9a2d..0284a3a18 100644 --- a/src/applypilot/discovery/smartextract.py +++ b/src/applypilot/discovery/smartextract.py @@ -31,10 +31,29 @@ from applypilot import config from applypilot.config import CONFIG_DIR from applypilot.database import get_connection, init_db, store_jobs, get_stats -from applypilot.llm import get_client +from applypilot.llm import ClaudeUsageLimitError, get_client log = logging.getLogger(__name__) +# Non-job noise that job-board frontends fire on every page load -- these +# match the broad "/api/" response-capture heuristic below but are never +# job data, so they're rejected before ever reaching the LLM judge. +_NOISE_URL_PATTERNS = ( + "/api/auth/", + "/api/telemetry/", + "/api/analytics/", + "get-session", + "web-vitals", + "geolocation", + "onetrust.com", + "cookieconsent", +) + + +def _is_noise_url(url: str) -> bool: + lowered = url.lower() + return any(p in lowered for p in _NOISE_URL_PATTERNS) + # Fix Windows encoding -- prevents charmap errors on emoji/unicode in job titles if sys.stdout.encoding and sys.stdout.encoding.lower() != "utf-8": try: @@ -103,6 +122,9 @@ def _store_jobs_filtered( url = job.get("url") if not url: continue + if _is_noise_url(url): + filtered += 1 + continue if not _location_ok(job.get("location"), accept_locs, reject_locs): filtered += 1 continue @@ -145,6 +167,8 @@ def on_response(response): rurl = response.url if any(ext in rurl for ext in [".js", ".css", ".png", ".jpg", ".svg", ".woff", ".ico", ".gif", ".webp"]): return + if _is_noise_url(rurl): + return if "json" in ct or "/api/" in rurl or "algolia" in rurl or "graphql" in rurl: try: body = response.text() @@ -401,9 +425,11 @@ def judge_api_responses(api_responses: list[dict]) -> list[dict]: "KEEP" if is_relevant else "DROP", reason) if is_relevant: relevant.append(resp) + except ClaudeUsageLimitError as e: + log.error("Judge stopped -- usage limit hit: %s", e) + break except Exception as e: - log.warning("Judge ERROR for %s: %s -- keeping", resp.get("url", "?")[:80], e) - relevant.append(resp) + log.warning("Judge ERROR for %s: %s -- dropping", resp.get("url", "?")[:80], e) return relevant