From 96c58439e13e1a0591d78a6d87033523de520a6b Mon Sep 17 00:00:00 2001 From: Michael Wu Date: Mon, 21 Sep 2026 15:29:42 +0900 Subject: [PATCH 1/4] feat: add Jev job lead shadow observations --- .env.example | 8 + apps/worker/src/five08/worker/config.py | 9 + docs/configuration.md | 17 + packages/shared/src/five08/job_lead_evals.py | 220 ++------- packages/shared/src/five08/job_lead_jev.py | 313 +++++++++++++ .../shared/src/five08/job_lead_sources.py | 424 +++++++++++++++++- packages/shared/src/five08/runtime_config.py | 56 +++ tests/unit/test_job_lead_evals.py | 2 +- tests/unit/test_job_lead_sources.py | 141 ++++++ tests/unit/test_runtime_config.py | 20 + 10 files changed, 1014 insertions(+), 196 deletions(-) create mode 100644 packages/shared/src/five08/job_lead_jev.py diff --git a/.env.example b/.env.example index 64bc9459..95ed24cb 100644 --- a/.env.example +++ b/.env.example @@ -222,6 +222,14 @@ AGENT_REASONING_API_KEY= JOB_LEAD_CLASSIFIER_ENABLED=true JOB_LEAD_CLASSIFIER_MODEL= JOB_LEAD_CLASSIFIER_TIMEOUT_SECONDS=8.0 +# Optional Jev canary. It never changes the production classification. Configure +# OPENROUTER_API_KEY, enable the shadow, and inspect the scrape job result in the +# dashboard background-task detail view. +JOB_LEAD_JEV_SHADOW_ENABLED=false +JOB_LEAD_JEV_SHADOW_MODEL=typesafe/jev-1.13 +JOB_LEAD_JEV_SHADOW_SAMPLE_RATE=0.1 +JOB_LEAD_JEV_SHADOW_CONFIDENCE_THRESHOLD=0.8 +JOB_LEAD_JEV_SHADOW_TIMEOUT_SECONDS=4.0 # Optional deterministic agent tool integrations. # GitHub Issues are the canonical todo backend. Members can read and write the # default repo; Steering Committee/Admin/Owner can work across every repository diff --git a/apps/worker/src/five08/worker/config.py b/apps/worker/src/five08/worker/config.py index f3e304be..0d70c064 100644 --- a/apps/worker/src/five08/worker/config.py +++ b/apps/worker/src/five08/worker/config.py @@ -78,6 +78,15 @@ class WorkerSettings(SharedSettings): job_lead_classifier_enabled: bool = True job_lead_classifier_model: str | None = None job_lead_classifier_timeout_seconds: float = Field(default=8.0, gt=0) + job_lead_jev_shadow_enabled: bool = False + job_lead_jev_shadow_model: str = "typesafe/jev-1.13" + job_lead_jev_shadow_sample_rate: float = Field(default=0.1, ge=0.0, le=1.0) + job_lead_jev_shadow_confidence_threshold: float = Field( + default=0.8, + ge=0.5, + le=1.0, + ) + job_lead_jev_shadow_timeout_seconds: float = Field(default=4.0, gt=0, le=30.0) resume_ai_api_key: str | None = None resume_ai_base_url: str | None = None resume_ai_model: str = "gpt-4.1-mini" diff --git a/docs/configuration.md b/docs/configuration.md index 76d9a6f0..ce855540 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -233,6 +233,23 @@ intake-completed field unset, and matches resume filenames with - `JOB_LEAD_CLASSIFIER_ENABLED` - `JOB_LEAD_CLASSIFIER_MODEL` - `JOB_LEAD_CLASSIFIER_TIMEOUT_SECONDS` +- `JOB_LEAD_JEV_SHADOW_ENABLED` +- `JOB_LEAD_JEV_SHADOW_MODEL` +- `JOB_LEAD_JEV_SHADOW_SAMPLE_RATE` +- `JOB_LEAD_JEV_SHADOW_CONFIDENCE_THRESHOLD` +- `JOB_LEAD_JEV_SHADOW_TIMEOUT_SECONDS` + +The Jev job-lead shadow is disabled by default and requires +`OPENROUTER_API_KEY`. When enabled, it deterministically samples eligible HN +posts and records Jev agreement, confidence-gate coverage, latency, token use, +cost, provider failures, and review items without changing the production +classification. View the latest data in **Dashboard → Background tasks**, open +the `scrape_job_leads_job` run, and inspect `Result → classifier_shadow`. +Disagreements, confidence fallbacks, and provider failures include only the HN +item ID and URL plus normalized decisions; raw post text and provider responses +are not retained in the shadow report. A provider failure disables further Jev +calls for that scrape run so an unavailable shadow service cannot repeatedly +delay the production job. Resume/profile LLM calls retry matching direct providers after Bifrost request failures when direct provider credentials are configured. diff --git a/packages/shared/src/five08/job_lead_evals.py b/packages/shared/src/five08/job_lead_evals.py index dc0b8a9e..7ebff597 100644 --- a/packages/shared/src/five08/job_lead_evals.py +++ b/packages/shared/src/five08/job_lead_evals.py @@ -9,7 +9,7 @@ import sys import time from collections import defaultdict -from collections.abc import Callable, Mapping, Sequence +from collections.abc import Callable, Sequence from datetime import datetime, timezone from pathlib import Path from typing import Any, Literal @@ -18,6 +18,13 @@ from openai import OpenAI from pydantic import BaseModel, ConfigDict, Field, model_validator +from five08.job_lead_jev import ( + DEFAULT_JOB_LEAD_JEV_MODEL, + OPENROUTER_DECISIONS_URL, + JobLeadJevRequestError, + classify_job_lead_with_jev, + job_lead_jev_questions, +) from five08.job_lead_sources import ( JobLeadClassifier, JobLeadLLMClassificationResponse, @@ -25,7 +32,6 @@ classify_contractor_lead_heuristic, ) from five08.model_catalog import model_chat_completion_options -from five08.tls import default_ca_bundle_path PostingType = Literal[ "part_time", @@ -39,9 +45,8 @@ "tests/evals/job-lead-classification/fixtures/v1/corpus.json" ) DEFAULT_OUTPUT_DIR = Path("tests/evals/job-lead-classification/reports") -DEFAULT_JEV_MODEL = "typesafe/jev-1.13" +DEFAULT_JEV_MODEL = DEFAULT_JOB_LEAD_JEV_MODEL DEFAULT_LLM_MODEL = "gpt-5.6-luna" -OPENROUTER_DECISIONS_URL = "https://openrouter.ai/api/alpha/decisions" OPENAI_BASE_URL = "https://api.openai.com/v1" LUNA_INPUT_COST_PER_1M = 0.20 LUNA_CACHED_INPUT_COST_PER_1M = 0.02 @@ -171,42 +176,7 @@ def load_job_lead_eval_corpus(path: Path = DEFAULT_CORPUS_PATH) -> JobLeadEvalCo def jev_questions() -> dict[str, dict[str, Any]]: """Return the stable Jev decision contract for this eval.""" - return { - "contractor_friendly": { - "type": "noul", - "instructions": ( - "Is this a direct employer or recruiter job posting that explicitly " - "offers contract, contractor, freelance, consulting, fractional, " - "1099, B2B contracting, or part-time work? Answer false for " - "full-time employee-only roles, people seeking work, replies, closed " - "roles, and company products or customer contracts." - ), - }, - "posting_type": { - "type": "choice", - "instructions": ( - "What employment arrangement does this direct job posting explicitly offer?" - ), - "criteria": { - "part_time": ( - "Contract, contractor, freelance, consulting, fractional, 1099, " - "B2B contracting, or part-time work, without a full-time option." - ), - "full_time": ( - "Full-time or permanent employee work only, with no contract or " - "part-time option." - ), - "part_time_or_full_time": ( - "Explicitly offers both full-time employment and contract, " - "freelance, consulting, or part-time work." - ), - "unknown": ( - "Not a direct current job posting, or the employment arrangement " - "is not stated clearly." - ), - }, - }, - } + return job_lead_jev_questions() def run_job_lead_eval_suite( @@ -354,7 +324,9 @@ def _run_case( latency_ms=_elapsed_ms(started), requested_model={"jev": jev_model, "luna": llm_model}.get(profile), request_attempts=( - exc.request_attempts if isinstance(exc, _RequestFailure) else 1 + exc.request_attempts + if isinstance(exc, _RequestFailure | JobLeadJevRequestError) + else 1 ), error=_safe_error(exc), ) @@ -389,51 +361,33 @@ def _run_jev( max_attempts: int, started: float, ) -> JobLeadEvalObservation: - body, attempts = _post_json_with_retries( + decision = classify_job_lead_with_jev( session=session, - url=OPENROUTER_DECISIONS_URL, api_key=api_key, - payload={ - "model": model, - "state": case.text, - "questions": jev_questions(), - }, + comment_text=case.text, + model=model, timeout_seconds=timeout_seconds, max_attempts=max_attempts, - service_name="OpenRouter", - extra_headers={"X-OpenRouter-Title": "508.dev Job Lead Eval"}, - ) - answers = _mapping(body.get("answers"), name="answers") - contractor_answer = _mapping( - answers.get("contractor_friendly"), name="answers.contractor_friendly" - ) - posting_answer = _mapping(answers.get("posting_type"), name="answers.posting_type") - contractor_probability = _probability( - contractor_answer.get("noul"), name="answers.contractor_friendly.noul" - ) - posting_type = _posting_type( - posting_answer.get("choice"), name="answers.posting_type.choice" + request_title="508.dev Job Lead Eval", ) - posting_probabilities = _probabilities( - posting_answer.get("probabilities"), name="answers.posting_type.probabilities" - ) - confidence = _optional_probability(posting_answer.get("confidence")) - usage = _usage(body.get("usage")) return _base_observation( profile="jev", case=case, repeat=repeat, requested_model=model, - resolved_model=_optional_text(body.get("model")), - provider=_optional_text(body.get("provider")), - predicted_posting_type=posting_type, - predicted_contractor_friendly=contractor_probability >= 0.5, - contractor_probability=contractor_probability, - classification_confidence=confidence, - posting_probabilities=posting_probabilities, + resolved_model=decision.resolved_model, + provider=decision.provider, + predicted_posting_type=decision.posting_type.value, + predicted_contractor_friendly=decision.is_contractor_friendly, + contractor_probability=decision.contractor_probability, + classification_confidence=decision.posting_confidence, + posting_probabilities=decision.posting_probabilities, latency_ms=_elapsed_ms(started), - request_attempts=attempts, - **usage, + request_attempts=decision.request_attempts, + input_tokens=decision.input_tokens, + output_tokens=decision.output_tokens, + total_tokens=decision.total_tokens, + cost_usd=decision.cost_usd, ) @@ -1010,126 +964,12 @@ def _openai_parse_with_retries( raise RuntimeError("OpenAI request did not produce a response") -def _post_json_with_retries( - *, - session: requests.Session, - url: str, - api_key: str, - payload: dict[str, Any], - timeout_seconds: float, - max_attempts: int, - service_name: str, - extra_headers: Mapping[str, str] | None = None, -) -> tuple[dict[str, Any], int]: - if max_attempts < 1: - raise ValueError("max_attempts must be at least 1") - response: requests.Response | None = None - for attempt in range(1, max_attempts + 1): - headers = { - "Authorization": f"Bearer {api_key}", - "Content-Type": "application/json", - } - if extra_headers: - headers.update(extra_headers) - try: - response = session.post( - url, - headers=headers, - json=payload, - timeout=timeout_seconds, - verify=default_ca_bundle_path(), - ) - except requests.RequestException as exc: - if attempt == max_attempts: - raise _RequestFailure(exc, request_attempts=attempt) from exc - time.sleep(min(float(2 ** (attempt - 1)), 8.0)) - continue - if ( - response.status_code not in _RETRYABLE_STATUS_CODES - or attempt == max_attempts - ): - break - time.sleep(_retry_delay(response, attempt)) - if response is None: - raise _RequestFailure( - RuntimeError(f"{service_name} request did not produce a response"), - request_attempts=max_attempts, - ) - try: - body = response.json() - except ValueError as exc: - cause = ValueError( - f"{service_name} returned non-JSON HTTP {response.status_code}" - ) - raise _RequestFailure(cause, request_attempts=attempt) from exc - if not response.ok: - error = body.get("error") if isinstance(body, dict) else None - if isinstance(error, dict): - message = _optional_text(error.get("message")) or "unknown error" - else: - message = _optional_text(error) or "unknown error" - raise _RequestFailure( - RuntimeError( - f"{service_name} HTTP {response.status_code}: {message[:300]}" - ), - request_attempts=attempt, - ) - if not isinstance(body, dict): - raise _RequestFailure( - ValueError(f"{service_name} response must be a JSON object"), - request_attempts=attempt, - ) - return body, attempt - - -def _retry_delay(response: requests.Response, attempt: int) -> float: - retry_after = response.headers.get("retry-after") - if retry_after: - try: - return max(0.0, min(float(retry_after), 15.0)) - except ValueError: - pass - return min(float(2 ** (attempt - 1)), 8.0) - - -def _mapping(value: Any, *, name: str) -> Mapping[str, Any]: - if not isinstance(value, dict): - raise ValueError(f"{name} must be an object") - return value - - def _posting_type(value: Any, *, name: str) -> PostingType: if value not in _POSTING_TYPES: raise ValueError(f"{name} has unsupported value: {value!r}") return value -def _probability(value: Any, *, name: str) -> float: - probability = _optional_probability(value) - if probability is None: - raise ValueError(f"{name} must be a probability") - return probability - - -def _optional_probability(value: Any) -> float | None: - if isinstance(value, bool) or not isinstance(value, int | float): - return None - probability = float(value) - return probability if 0.0 <= probability <= 1.0 else None - - -def _probabilities(value: Any, *, name: str) -> dict[str, float]: - source = _mapping(value, name=name) - probabilities = { - str(key): probability - for key, raw in source.items() - if (probability := _optional_probability(raw)) is not None - } - if not set(_POSTING_TYPES).issubset(probabilities): - raise ValueError(f"{name} must include all posting types") - return probabilities - - def _usage(value: Any) -> dict[str, Any]: source = value if isinstance(value, dict) else {} input_tokens = _integer(source.get("input_tokens", source.get("prompt_tokens"))) @@ -1201,7 +1041,7 @@ def _optional_text(value: Any) -> str | None: def _safe_error(exc: Exception) -> str: - if isinstance(exc, _RequestFailure): + if isinstance(exc, _RequestFailure | JobLeadJevRequestError): exc = exc.cause return f"{type(exc).__name__}: {str(exc)[:500]}" diff --git a/packages/shared/src/five08/job_lead_jev.py b/packages/shared/src/five08/job_lead_jev.py new file mode 100644 index 00000000..ed83f7fc --- /dev/null +++ b/packages/shared/src/five08/job_lead_jev.py @@ -0,0 +1,313 @@ +"""Shared Jev contract and OpenRouter transport for job-lead classification.""" + +from __future__ import annotations + +import time +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Any + +import requests + +from five08.job_channels import JobPostingType +from five08.tls import default_ca_bundle_path + +DEFAULT_JOB_LEAD_JEV_MODEL = "typesafe/jev-1.13" +OPENROUTER_DECISIONS_URL = "https://openrouter.ai/api/alpha/decisions" +_RETRYABLE_STATUS_CODES = frozenset({408, 409, 429, 500, 502, 503, 504, 529}) + + +class JobLeadJevRequestError(RuntimeError): + """Carry provider attempt metadata while retaining a safe root cause.""" + + def __init__(self, cause: Exception, *, request_attempts: int) -> None: + super().__init__(str(cause)) + self.cause = cause + self.request_attempts = request_attempts + + +@dataclass(frozen=True) +class JobLeadJevDecision: + """Normalized Jev decision without retaining the raw provider response.""" + + requested_model: str + resolved_model: str | None + provider: str | None + is_contractor_friendly: bool + contractor_probability: float + posting_type: JobPostingType + posting_confidence: float | None + posting_probabilities: dict[str, float] + latency_ms: int + request_attempts: int + input_tokens: int + output_tokens: int + total_tokens: int + cost_usd: float | None + + +def job_lead_jev_questions() -> dict[str, dict[str, Any]]: + """Return the versioned Jev decision contract used by eval and production.""" + + return { + "contractor_friendly": { + "type": "noul", + "instructions": ( + "Is this a direct employer or recruiter job posting that explicitly " + "offers contract, contractor, freelance, consulting, fractional, " + "1099, B2B contracting, or part-time work? Answer false for " + "full-time employee-only roles, people seeking work, replies, closed " + "roles, and company products or customer contracts." + ), + }, + "posting_type": { + "type": "choice", + "instructions": ( + "What employment arrangement does this direct job posting explicitly offer?" + ), + "criteria": { + "part_time": ( + "Contract, contractor, freelance, consulting, fractional, 1099, " + "B2B contracting, or part-time work, without a full-time option." + ), + "full_time": ( + "Full-time or permanent employee work only, with no contract or " + "part-time option." + ), + "part_time_or_full_time": ( + "Explicitly offers both full-time employment and contract, " + "freelance, consulting, or part-time work." + ), + "unknown": ( + "Not a direct current job posting, or the employment arrangement " + "is not stated clearly." + ), + }, + }, + } + + +def classify_job_lead_with_jev( + *, + session: requests.Session, + api_key: str, + comment_text: str, + model: str = DEFAULT_JOB_LEAD_JEV_MODEL, + timeout_seconds: float = 4.0, + max_attempts: int = 2, + request_title: str = "508.dev Job Lead Shadow", +) -> JobLeadJevDecision: + """Classify one job lead through OpenRouter's Jev Decisions endpoint.""" + + started = time.perf_counter() + body, attempts = _post_json_with_retries( + session=session, + api_key=api_key, + payload={ + "model": model, + "state": comment_text, + "questions": job_lead_jev_questions(), + }, + timeout_seconds=timeout_seconds, + max_attempts=max_attempts, + request_title=request_title, + ) + answers = _mapping(body.get("answers"), name="answers") + contractor_answer = _mapping( + answers.get("contractor_friendly"), + name="answers.contractor_friendly", + ) + posting_answer = _mapping( + answers.get("posting_type"), + name="answers.posting_type", + ) + contractor_probability = _probability( + contractor_answer.get("noul"), + name="answers.contractor_friendly.noul", + ) + posting_type = _posting_type( + posting_answer.get("choice"), + name="answers.posting_type.choice", + ) + posting_probabilities = _probabilities( + posting_answer.get("probabilities"), + name="answers.posting_type.probabilities", + ) + usage = _usage(body.get("usage")) + return JobLeadJevDecision( + requested_model=model, + resolved_model=_optional_text(body.get("model")), + provider=_optional_text(body.get("provider")), + is_contractor_friendly=contractor_probability >= 0.5, + contractor_probability=contractor_probability, + posting_type=posting_type, + posting_confidence=_optional_probability(posting_answer.get("confidence")), + posting_probabilities=posting_probabilities, + latency_ms=max(0, round((time.perf_counter() - started) * 1000)), + request_attempts=attempts, + input_tokens=usage["input_tokens"], + output_tokens=usage["output_tokens"], + total_tokens=usage["total_tokens"], + cost_usd=usage["cost_usd"], + ) + + +def _post_json_with_retries( + *, + session: requests.Session, + api_key: str, + payload: dict[str, Any], + timeout_seconds: float, + max_attempts: int, + request_title: str, +) -> tuple[dict[str, Any], int]: + if max_attempts < 1: + raise ValueError("max_attempts must be at least 1") + response: requests.Response | None = None + for attempt in range(1, max_attempts + 1): + try: + response = session.post( + OPENROUTER_DECISIONS_URL, + headers={ + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + "X-OpenRouter-Title": request_title, + }, + json=payload, + timeout=timeout_seconds, + verify=default_ca_bundle_path(), + ) + except requests.RequestException as exc: + if attempt == max_attempts: + raise JobLeadJevRequestError( + exc, + request_attempts=attempt, + ) from exc + time.sleep(min(float(2 ** (attempt - 1)), 8.0)) + continue + if ( + response.status_code not in _RETRYABLE_STATUS_CODES + or attempt == max_attempts + ): + break + time.sleep(_retry_delay(response, attempt)) + if response is None: + raise JobLeadJevRequestError( + RuntimeError("OpenRouter request did not produce a response"), + request_attempts=max_attempts, + ) + try: + body = response.json() + except ValueError as exc: + cause = ValueError(f"OpenRouter returned non-JSON HTTP {response.status_code}") + raise JobLeadJevRequestError(cause, request_attempts=attempt) from exc + if not response.ok: + error = body.get("error") if isinstance(body, dict) else None + if isinstance(error, dict): + message = _optional_text(error.get("message")) or "unknown error" + else: + message = _optional_text(error) or "unknown error" + raise JobLeadJevRequestError( + RuntimeError( + f"OpenRouter HTTP {response.status_code}: {message[:300]}" + ), + request_attempts=attempt, + ) + if not isinstance(body, dict): + raise JobLeadJevRequestError( + ValueError("OpenRouter response must be a JSON object"), + request_attempts=attempt, + ) + return body, attempt + + +def _retry_delay(response: requests.Response, attempt: int) -> float: + retry_after = response.headers.get("retry-after") + if retry_after: + try: + return max(0.0, min(float(retry_after), 15.0)) + except ValueError: + pass + return min(float(2 ** (attempt - 1)), 8.0) + + +def _mapping(value: Any, *, name: str) -> Mapping[str, Any]: + if not isinstance(value, dict): + raise ValueError(f"{name} must be an object") + return value + + +def _posting_type(value: Any, *, name: str) -> JobPostingType: + try: + return JobPostingType(value) + except (TypeError, ValueError) as exc: + raise ValueError(f"{name} has unsupported value: {value!r}") from exc + + +def _probability(value: Any, *, name: str) -> float: + probability = _optional_probability(value) + if probability is None: + raise ValueError(f"{name} must be a probability") + return probability + + +def _optional_probability(value: Any) -> float | None: + if isinstance(value, bool) or not isinstance(value, int | float): + return None + probability = float(value) + return probability if 0.0 <= probability <= 1.0 else None + + +def _probabilities(value: Any, *, name: str) -> dict[str, float]: + source = _mapping(value, name=name) + probabilities = { + str(key): probability + for key, raw in source.items() + if (probability := _optional_probability(raw)) is not None + } + expected = {posting_type.value for posting_type in JobPostingType} + if not expected.issubset(probabilities): + raise ValueError(f"{name} must include all posting types") + return probabilities + + +def _usage(value: Any) -> dict[str, Any]: + source = value if isinstance(value, dict) else {} + input_tokens = _integer(source.get("input_tokens", source.get("prompt_tokens"))) + output_tokens = _integer( + source.get("output_tokens", source.get("completion_tokens")) + ) + total_tokens = _integer(source.get("total_tokens")) or input_tokens + output_tokens + cost = source.get("cost") + return { + "input_tokens": input_tokens, + "output_tokens": output_tokens, + "total_tokens": total_tokens, + "cost_usd": _optional_float(cost), + } + + +def _integer(value: Any) -> int: + if isinstance(value, bool): + return 0 + try: + return max(0, int(value or 0)) + except (TypeError, ValueError): + return 0 + + +def _optional_float(value: Any) -> float | None: + if isinstance(value, bool) or value is None: + return None + try: + result = float(value) + except (TypeError, ValueError): + return None + return result if result >= 0 else None + + +def _optional_text(value: Any) -> str | None: + if not isinstance(value, str): + return None + stripped = value.strip() + return stripped or None diff --git a/packages/shared/src/five08/job_lead_sources.py b/packages/shared/src/five08/job_lead_sources.py index e1898f0c..97880f59 100644 --- a/packages/shared/src/five08/job_lead_sources.py +++ b/packages/shared/src/five08/job_lead_sources.py @@ -2,20 +2,28 @@ from __future__ import annotations +import hashlib import html import json import logging import re -from dataclasses import dataclass +import time +from dataclasses import dataclass, replace from datetime import datetime, timezone from html.parser import HTMLParser from typing import Any, Literal, Protocol from urllib.parse import quote, urlencode, urlsplit from urllib.request import Request, urlopen +import requests from pydantic import BaseModel, ConfigDict, Field from five08.job_channels import JobPostingType +from five08.job_lead_jev import ( + DEFAULT_JOB_LEAD_JEV_MODEL, + JobLeadJevDecision, + classify_job_lead_with_jev, +) from five08.job_leads import ( JobLeadInput, existing_job_lead_external_ids, @@ -40,6 +48,9 @@ HN_WHO_IS_HIRING_SOURCE_KEY = "hackernews_who_is_hiring" HN_WHO_IS_HIRING_SOURCE_TYPE = "hackernews" DEFAULT_JOB_LEAD_CLASSIFIER_MODEL = "gpt-4.1-mini" +JOB_LEAD_JEV_SHADOW_METADATA_KEY = "contractor_classification_shadow" +_JOB_LEAD_JEV_SHADOW_REPORT_VERSION = "job-lead-jev-shadow.v1" +_JOB_LEAD_JEV_SHADOW_REVIEW_LIMIT = 50 _WHO_IS_HIRING_TITLE_RE = re.compile( r"^Ask HN: Who is hiring\? \((?P[A-Za-z]+) (?P20\d\d)\)$" @@ -151,6 +162,63 @@ def collect(self) -> list[JobLeadInput]: """Return current leads from the source.""" +@dataclass(frozen=True) +class JobLeadJevShadowObservation: + """One normalized, non-authoritative Jev shadow observation.""" + + status: Literal["succeeded", "failed"] + requested_model: str + confidence_threshold: float + primary_is_contractor_friendly: bool + primary_posting_type: JobPostingType + latency_ms: int + observed_at: datetime + resolved_model: str | None = None + provider: str | None = None + predicted_is_contractor_friendly: bool | None = None + predicted_posting_type: JobPostingType | None = None + contractor_probability: float | None = None + gate_accepted: bool | None = None + agrees_with_primary: bool | None = None + request_attempts: int = 0 + input_tokens: int = 0 + output_tokens: int = 0 + total_tokens: int = 0 + cost_usd: float | None = None + error: str | None = None + + def payload(self) -> dict[str, Any]: + """Return dashboard-safe metadata without raw prompts or responses.""" + + return { + "version": _JOB_LEAD_JEV_SHADOW_REPORT_VERSION, + "status": self.status, + "requested_model": self.requested_model, + "resolved_model": self.resolved_model, + "provider": self.provider, + "confidence_threshold": self.confidence_threshold, + "primary_is_contractor_friendly": (self.primary_is_contractor_friendly), + "primary_posting_type": self.primary_posting_type.value, + "observed_at": self.observed_at.isoformat(), + "predicted_is_contractor_friendly": (self.predicted_is_contractor_friendly), + "predicted_posting_type": ( + self.predicted_posting_type.value + if self.predicted_posting_type is not None + else None + ), + "contractor_probability": self.contractor_probability, + "gate_accepted": self.gate_accepted, + "agrees_with_primary": self.agrees_with_primary, + "latency_ms": self.latency_ms, + "request_attempts": self.request_attempts, + "input_tokens": self.input_tokens, + "output_tokens": self.output_tokens, + "total_tokens": self.total_tokens, + "cost_usd": self.cost_usd, + "error": self.error, + } + + @dataclass(frozen=True) class JobLeadClassification: """Contractor-friendliness classification for one external lead.""" @@ -164,6 +232,7 @@ class JobLeadClassification: method: Literal["llm", "heuristic"] apply_url: str | None = None contact_email: str | None = None + jev_shadow: JobLeadJevShadowObservation | None = None class JobLeadLLMClassificationResponse(BaseModel): @@ -581,19 +650,117 @@ def __init__( *, settings: SharedSettings, client: Any | None = None, + jev_shadow_session: requests.Session | None = None, ) -> None: self.settings = settings self.client = client if client is not None else _build_llm_client(settings) + self._jev_shadow_enabled = bool( + getattr(settings, "job_lead_jev_shadow_enabled", False) + ) + self._jev_shadow_model = ( + _clean(getattr(settings, "job_lead_jev_shadow_model", None)) + or DEFAULT_JOB_LEAD_JEV_MODEL + ) + self._jev_shadow_api_key = _clean(getattr(settings, "openrouter_api_key", None)) + self._jev_shadow_sample_rate = _bounded_float( + getattr(settings, "job_lead_jev_shadow_sample_rate", 0.1), + minimum=0.0, + maximum=1.0, + default=0.1, + ) + self._jev_shadow_confidence_threshold = _bounded_float( + getattr(settings, "job_lead_jev_shadow_confidence_threshold", 0.8), + minimum=0.5, + maximum=1.0, + default=0.8, + ) + self._jev_shadow_timeout_seconds = _bounded_float( + getattr(settings, "job_lead_jev_shadow_timeout_seconds", 4.0), + minimum=0.1, + maximum=30.0, + default=4.0, + ) + self._owns_jev_shadow_session = False + self._jev_shadow_available = True + self._jev_shadow_session = jev_shadow_session + if ( + self._jev_shadow_session is None + and self._jev_shadow_enabled + and self._jev_shadow_api_key + ): + self._jev_shadow_session = requests.Session() + self._owns_jev_shadow_session = True def classify(self, comment_text: str) -> JobLeadClassification: if _SEEKING_WORK_RE.search(comment_text): return classify_contractor_lead_heuristic(comment_text) + classification: JobLeadClassification | None = None if self.client is not None: try: - return self._classify_with_llm(comment_text) + classification = self._classify_with_llm(comment_text) except Exception as exc: logger.warning("Job lead LLM classification failed: %s", exc) - return classify_contractor_lead_heuristic(comment_text) + if classification is None: + classification = classify_contractor_lead_heuristic(comment_text) + return self._attach_jev_shadow(comment_text, classification) + + def close(self) -> None: + """Close only the Jev session owned by this classifier.""" + + if self._owns_jev_shadow_session and self._jev_shadow_session is not None: + self._jev_shadow_session.close() + self._jev_shadow_session = None + + def _attach_jev_shadow( + self, + comment_text: str, + classification: JobLeadClassification, + ) -> JobLeadClassification: + if ( + not self._jev_shadow_enabled + or not self._jev_shadow_available + or not self._jev_shadow_api_key + or self._jev_shadow_session is None + or not _selected_for_jev_shadow( + comment_text, + self._jev_shadow_sample_rate, + ) + ): + return classification + + started = time.perf_counter() + try: + decision = classify_job_lead_with_jev( + session=self._jev_shadow_session, + api_key=self._jev_shadow_api_key, + comment_text=comment_text, + model=self._jev_shadow_model, + timeout_seconds=self._jev_shadow_timeout_seconds, + max_attempts=1, + ) + observation = _successful_jev_shadow_observation( + decision=decision, + classification=classification, + confidence_threshold=self._jev_shadow_confidence_threshold, + ) + except Exception as exc: + self._jev_shadow_available = False + observation = JobLeadJevShadowObservation( + status="failed", + requested_model=self._jev_shadow_model, + confidence_threshold=self._jev_shadow_confidence_threshold, + primary_is_contractor_friendly=(classification.is_contractor_friendly), + primary_posting_type=classification.posting_type, + latency_ms=max(0, round((time.perf_counter() - started) * 1000)), + observed_at=datetime.now(timezone.utc), + error=_safe_jev_shadow_error(exc), + ) + logger.warning( + "Jev job-lead shadow classification failed; disabling it for the " + "remainder of this scrape: %s", + exc, + ) + return replace(classification, jev_shadow=observation) @staticmethod def _messages(comment_text: str) -> list[dict[str, str]]: @@ -671,6 +838,72 @@ def _clean(value: object) -> str | None: return stripped or None +def _bounded_float( + value: object, + *, + minimum: float, + maximum: float, + default: float, +) -> float: + if isinstance(value, bool) or not isinstance(value, int | float | str): + return default + try: + numeric = float(value) + except (TypeError, ValueError): + return default + return max(minimum, min(maximum, numeric)) + + +def _selected_for_jev_shadow(comment_text: str, sample_rate: float) -> bool: + if sample_rate <= 0.0: + return False + if sample_rate >= 1.0: + return True + digest = hashlib.sha256(comment_text.encode("utf-8")).digest() + bucket = int.from_bytes(digest[:8], "big") / float(2**64) + return bucket < sample_rate + + +def _successful_jev_shadow_observation( + *, + decision: JobLeadJevDecision, + classification: JobLeadClassification, + confidence_threshold: float, +) -> JobLeadJevShadowObservation: + probability = decision.contractor_probability + gate_accepted = ( + probability >= confidence_threshold or probability <= 1.0 - confidence_threshold + ) + return JobLeadJevShadowObservation( + status="succeeded", + requested_model=decision.requested_model, + resolved_model=decision.resolved_model, + provider=decision.provider, + confidence_threshold=confidence_threshold, + primary_is_contractor_friendly=classification.is_contractor_friendly, + primary_posting_type=classification.posting_type, + observed_at=datetime.now(timezone.utc), + predicted_is_contractor_friendly=decision.is_contractor_friendly, + predicted_posting_type=decision.posting_type, + contractor_probability=probability, + gate_accepted=gate_accepted, + agrees_with_primary=( + decision.is_contractor_friendly == classification.is_contractor_friendly + ), + latency_ms=decision.latency_ms, + request_attempts=decision.request_attempts, + input_tokens=decision.input_tokens, + output_tokens=decision.output_tokens, + total_tokens=decision.total_tokens, + cost_usd=decision.cost_usd, + ) + + +def _safe_jev_shadow_error(exc: Exception) -> str: + message = " ".join(str(exc).split())[:240] + return f"{type(exc).__name__}: {message}" if message else type(exc).__name__ + + def _classifier_model(settings: SharedSettings) -> str: return ( _clean(getattr(settings, "job_lead_classifier_model", None)) @@ -791,7 +1024,7 @@ def _classification_from_llm_response( def _classification_metadata( classification: JobLeadClassification, ) -> dict[str, Any]: - return { + metadata = { "contractor_classification": { "is_contractor_friendly": classification.is_contractor_friendly, "posting_type": classification.posting_type.value, @@ -803,6 +1036,9 @@ def _classification_metadata( "contact_email": classification.contact_email, } } + if classification.jev_shadow is not None: + metadata[JOB_LEAD_JEV_SHADOW_METADATA_KEY] = classification.jev_shadow.payload() + return metadata def _lead_from_hn_comment( @@ -987,6 +1223,177 @@ def build_job_lead_source( raise ValueError(f"Unsupported job lead source: {source}") +def _job_lead_jev_shadow_report( + settings: SharedSettings, + leads: list[JobLeadInput], +) -> dict[str, Any]: + enabled = bool(getattr(settings, "job_lead_jev_shadow_enabled", False)) + api_key_configured = bool(_clean(getattr(settings, "openrouter_api_key", None))) + model = ( + _clean(getattr(settings, "job_lead_jev_shadow_model", None)) + or DEFAULT_JOB_LEAD_JEV_MODEL + ) + sample_rate = _bounded_float( + getattr(settings, "job_lead_jev_shadow_sample_rate", 0.1), + minimum=0.0, + maximum=1.0, + default=0.1, + ) + confidence_threshold = _bounded_float( + getattr(settings, "job_lead_jev_shadow_confidence_threshold", 0.8), + minimum=0.5, + maximum=1.0, + default=0.8, + ) + observed: list[tuple[JobLeadInput, dict[str, Any], dict[str, Any]]] = [] + for lead in leads: + metadata = lead.metadata if isinstance(lead.metadata, dict) else {} + primary = metadata.get("contractor_classification") + shadow = metadata.get(JOB_LEAD_JEV_SHADOW_METADATA_KEY) + if isinstance(primary, dict) and isinstance(shadow, dict): + observed.append((lead, primary, shadow)) + + successful = [item for item in observed if item[2].get("status") == "succeeded"] + failures = [item for item in observed if item[2].get("status") == "failed"] + agreements = [item for item in successful if item[2].get("agrees_with_primary")] + disagreements = [ + item for item in successful if item[2].get("agrees_with_primary") is False + ] + gate_accepted = [item for item in successful if item[2].get("gate_accepted")] + gate_fallback = [ + item for item in successful if item[2].get("gate_accepted") is False + ] + high_confidence_disagreements = [ + item for item in disagreements if item[2].get("gate_accepted") is True + ] + latencies = [ + int(item[2]["latency_ms"]) + for item in observed + if isinstance(item[2].get("latency_ms"), int | float) + and not isinstance(item[2].get("latency_ms"), bool) + ] + cost_values = [ + float(item[2]["cost_usd"]) + for item in successful + if isinstance(item[2].get("cost_usd"), int | float) + and not isinstance(item[2].get("cost_usd"), bool) + ] + + review_candidates: list[dict[str, Any]] = [] + for lead, primary, shadow in observed: + reasons: list[str] = [] + if shadow.get("status") == "failed": + reasons.append("provider_failure") + else: + if shadow.get("agrees_with_primary") is False: + reasons.append("binary_disagreement") + if shadow.get("gate_accepted") is False: + reasons.append("confidence_fallback") + if not reasons: + continue + review_candidates.append( + { + "external_id": lead.external_id, + "source_url": lead.source_url, + "reasons": reasons, + "primary": { + "method": primary.get("method"), + "is_contractor_friendly": primary.get("is_contractor_friendly"), + "posting_type": primary.get("posting_type"), + "confidence": primary.get("confidence"), + }, + "shadow": { + "status": shadow.get("status"), + "is_contractor_friendly": shadow.get( + "predicted_is_contractor_friendly" + ), + "posting_type": shadow.get("predicted_posting_type"), + "contractor_probability": shadow.get("contractor_probability"), + "gate_accepted": shadow.get("gate_accepted"), + "resolved_model": shadow.get("resolved_model"), + "provider": shadow.get("provider"), + "latency_ms": shadow.get("latency_ms"), + "observed_at": shadow.get("observed_at"), + "error": shadow.get("error"), + }, + } + ) + + if not enabled: + status = "disabled" + elif not api_key_configured: + status = "missing_openrouter_api_key" + elif sample_rate <= 0.0: + status = "paused" + elif not observed: + status = "no_sample_selected" + elif failures: + status = "completed_with_errors" + else: + status = "completed" + + attempted = len(observed) + succeeded = len(successful) + return { + "version": _JOB_LEAD_JEV_SHADOW_REPORT_VERSION, + "status": status, + "enabled": enabled, + "provider_configured": api_key_configured, + "requested_model": model, + "sample_rate": sample_rate, + "confidence_threshold": confidence_threshold, + "eligible": len(leads), + "attempted": attempted, + "not_observed": max(0, len(leads) - attempted), + "succeeded": succeeded, + "failed": len(failures), + "success_rate": round(succeeded / attempted, 4) if attempted else None, + "agreements": len(agreements), + "disagreements": len(disagreements), + "agreement_rate": round(len(agreements) / succeeded, 4) if succeeded else None, + "gate_accepted": len(gate_accepted), + "gate_fallback": len(gate_fallback), + "high_confidence_disagreements": len(high_confidence_disagreements), + "resolved_models": sorted( + { + str(item[2]["resolved_model"]) + for item in successful + if item[2].get("resolved_model") + } + ), + "providers": sorted( + {str(item[2]["provider"]) for item in successful if item[2].get("provider")} + ), + "latency_ms": { + "p50": _shadow_percentile(latencies, 0.50), + "p95": _shadow_percentile(latencies, 0.95), + "max": max(latencies) if latencies else None, + }, + "usage": { + "unit": "tokens", + "input": sum(int(item[2].get("input_tokens") or 0) for item in successful), + "output": sum( + int(item[2].get("output_tokens") or 0) for item in successful + ), + "total": sum(int(item[2].get("total_tokens") or 0) for item in successful), + "cost_usd": round(sum(cost_values), 8) if cost_values else None, + }, + "review_items": review_candidates[:_JOB_LEAD_JEV_SHADOW_REVIEW_LIMIT], + "review_items_truncated": max( + 0, + len(review_candidates) - _JOB_LEAD_JEV_SHADOW_REVIEW_LIMIT, + ), + } + + +def _shadow_percentile(values: list[int], quantile: float) -> int | None: + if not values: + return None + ordered = sorted(values) + index = round((len(ordered) - 1) * quantile) + return ordered[index] + + def scrape_job_leads( settings: SharedSettings, *, @@ -1004,7 +1411,13 @@ def scrape_job_leads( created = 0 updated = 0 lead_ids: list[str] = [] - leads = adapter.collect() + try: + leads = adapter.collect() + finally: + close_classifier = getattr(classifier, "close", None) + if callable(close_classifier): + close_classifier() + shadow_report = _job_lead_jev_shadow_report(settings, leads) collection_report_factory = getattr(adapter, "collection_report", None) raw_collection_report = ( collection_report_factory() if callable(collection_report_factory) else {} @@ -1048,6 +1461,7 @@ def scrape_job_leads( return { "source": adapter.source_key, **collection_report, + "classifier_shadow": shadow_report, "created": created, "updated": updated, "total": len(lead_ids), diff --git a/packages/shared/src/five08/runtime_config.py b/packages/shared/src/five08/runtime_config.py index 854ff230..92ed63df 100644 --- a/packages/shared/src/five08/runtime_config.py +++ b/packages/shared/src/five08/runtime_config.py @@ -497,6 +497,62 @@ class RuntimeConfigDBSnapshot: env_names=("JOB_LEAD_CLASSIFIER_TIMEOUT_SECONDS",), min_value=0.1, ), + RuntimeConfigDefinition( + key="JOB_LEAD_JEV_SHADOW_ENABLED", + attr="job_lead_jev_shadow_enabled", + label="Jev job lead shadow enabled", + category="AI", + description=( + "Observe sampled HN job leads with Jev without changing production " + "classification decisions." + ), + value_type="bool", + env_names=("JOB_LEAD_JEV_SHADOW_ENABLED",), + ), + RuntimeConfigDefinition( + key="JOB_LEAD_JEV_SHADOW_MODEL", + attr="job_lead_jev_shadow_model", + label="Jev job lead shadow model", + category="AI", + description="Pinned OpenRouter Jev model used for shadow decisions.", + env_names=("JOB_LEAD_JEV_SHADOW_MODEL",), + ), + RuntimeConfigDefinition( + key="JOB_LEAD_JEV_SHADOW_SAMPLE_RATE", + attr="job_lead_jev_shadow_sample_rate", + label="Jev job lead shadow sample rate", + category="AI", + description="Deterministic fraction of eligible HN posts sent to Jev.", + value_type="float", + env_names=("JOB_LEAD_JEV_SHADOW_SAMPLE_RATE",), + min_value=0.0, + max_value=1.0, + ), + RuntimeConfigDefinition( + key="JOB_LEAD_JEV_SHADOW_CONFIDENCE_THRESHOLD", + attr="job_lead_jev_shadow_confidence_threshold", + label="Jev job lead shadow confidence threshold", + category="AI", + description=( + "Symmetric positive or negative probability threshold used to mark " + "which Jev decisions would be accepted." + ), + value_type="float", + env_names=("JOB_LEAD_JEV_SHADOW_CONFIDENCE_THRESHOLD",), + min_value=0.5, + max_value=1.0, + ), + RuntimeConfigDefinition( + key="JOB_LEAD_JEV_SHADOW_TIMEOUT_SECONDS", + attr="job_lead_jev_shadow_timeout_seconds", + label="Jev job lead shadow timeout seconds", + category="AI", + description="Timeout for each non-authoritative Jev shadow request.", + value_type="float", + env_names=("JOB_LEAD_JEV_SHADOW_TIMEOUT_SECONDS",), + min_value=0.1, + max_value=30.0, + ), RuntimeConfigDefinition( key="RESUME_AI_API_KEY", attr="resume_ai_api_key", diff --git a/tests/unit/test_job_lead_evals.py b/tests/unit/test_job_lead_evals.py index 4184e859..922c8914 100644 --- a/tests/unit/test_job_lead_evals.py +++ b/tests/unit/test_job_lead_evals.py @@ -200,7 +200,7 @@ def test_jev_retries_transport_errors( monkeypatch: pytest.MonkeyPatch, ) -> None: session = _FlakySession() - monkeypatch.setattr("five08.job_lead_evals.time.sleep", lambda _delay: None) + monkeypatch.setattr("five08.job_lead_jev.time.sleep", lambda _delay: None) observation = _run_jev( case=_case(), diff --git a/tests/unit/test_job_lead_sources.py b/tests/unit/test_job_lead_sources.py index e4633ce9..a1f4feff 100644 --- a/tests/unit/test_job_lead_sources.py +++ b/tests/unit/test_job_lead_sources.py @@ -6,6 +6,7 @@ from types import SimpleNamespace import five08.job_lead_sources as job_lead_sources +from five08.job_lead_jev import JobLeadJevDecision from five08.job_lead_sources import ( HackerNewsThread, HackerNewsWhoIsHiringLeadSource, @@ -632,6 +633,7 @@ def update_existing(_settings: object, lead: JobLeadInput) -> str: assert result["created"] == 1 assert result["updated"] == 1 assert result["lead_ids"] == ["lead-10", "lead-11"] + assert result["classifier_shadow"]["status"] == "disabled" def test_scrape_skips_reviewed_contractor_friendly_lead(monkeypatch) -> None: @@ -696,6 +698,145 @@ def test_classifier_falls_back_without_second_llm_call_after_provider_failure() assert client.chat_create_calls == 0 +def test_jev_shadow_records_disagreement_without_changing_primary( + monkeypatch, +) -> None: + settings = SimpleNamespace( + job_lead_jev_shadow_enabled=True, + job_lead_jev_shadow_model="typesafe/jev-1.13", + job_lead_jev_shadow_sample_rate=1.0, + job_lead_jev_shadow_confidence_threshold=0.8, + job_lead_jev_shadow_timeout_seconds=4.0, + openrouter_api_key="test-key", + ) + primary = JobLeadClassification( + is_contractor_friendly=False, + posting_type=JobPostingType.FULL_TIME, + tags=["full-time"], + confidence=0.9, + confidence_label="high", + rationale="Employee-only role.", + method="llm", + ) + decision = JobLeadJevDecision( + requested_model="typesafe/jev-1.13", + resolved_model="typesafe/jev-1.13-20260917", + provider="TypeSafe", + is_contractor_friendly=True, + contractor_probability=0.91, + posting_type=JobPostingType.PART_TIME, + posting_confidence=0.98, + posting_probabilities={ + "part_time": 0.98, + "full_time": 0.01, + "part_time_or_full_time": 0.01, + "unknown": 0.0, + }, + latency_ms=420, + request_attempts=1, + input_tokens=450, + output_tokens=73, + total_tokens=523, + cost_usd=0.000019, + ) + classifier = JobLeadClassifier( + settings=settings, # type: ignore[arg-type] + client=object(), + jev_shadow_session=object(), # type: ignore[arg-type] + ) + monkeypatch.setattr(classifier, "_classify_with_llm", lambda _text: primary) + monkeypatch.setattr( + job_lead_sources, + "classify_job_lead_with_jev", + lambda **_kwargs: decision, + ) + + classification = classifier.classify("Acme | Full-time or contract | Remote") + + assert classification.is_contractor_friendly is False + assert classification.posting_type is JobPostingType.FULL_TIME + assert classification.jev_shadow is not None + assert classification.jev_shadow.predicted_is_contractor_friendly is True + assert classification.jev_shadow.gate_accepted is True + assert classification.jev_shadow.agrees_with_primary is False + metadata = job_lead_sources._classification_metadata(classification) + assert metadata["contractor_classification"]["is_contractor_friendly"] is False + assert metadata["contractor_classification_shadow"]["cost_usd"] == 0.000019 + + lead = JobLeadInput( + source_key="hackernews_who_is_hiring", + source_type="hackernews", + external_id="42", + source_url="https://news.ycombinator.com/item?id=42", + title="Acme role", + body_raw="not retained in report", + body_normalized="not retained in report", + metadata=metadata, + ) + report = job_lead_sources._job_lead_jev_shadow_report( # noqa: SLF001 + settings, # type: ignore[arg-type] + [lead], + ) + + assert report["status"] == "completed" + assert report["attempted"] == 1 + assert report["disagreements"] == 1 + assert report["high_confidence_disagreements"] == 1 + assert report["usage"] == { + "unit": "tokens", + "input": 450, + "output": 73, + "total": 523, + "cost_usd": 0.000019, + } + assert report["review_items"][0]["external_id"] == "42" + assert "body_raw" not in report["review_items"][0] + + +def test_jev_shadow_failure_never_changes_primary(monkeypatch) -> None: + settings = SimpleNamespace( + job_lead_jev_shadow_enabled=True, + job_lead_jev_shadow_sample_rate=1.0, + job_lead_jev_shadow_confidence_threshold=0.8, + job_lead_jev_shadow_timeout_seconds=4.0, + openrouter_api_key="test-key", + ) + primary = classify_contractor_lead_heuristic( + "Acme | Contract API engineer | Remote" + ) + classifier = JobLeadClassifier( + settings=settings, # type: ignore[arg-type] + client=object(), + jev_shadow_session=object(), # type: ignore[arg-type] + ) + monkeypatch.setattr(classifier, "_classify_with_llm", lambda _text: primary) + + shadow_calls = 0 + + def fail_shadow(**_kwargs: object) -> JobLeadJevDecision: + nonlocal shadow_calls + shadow_calls += 1 + raise TimeoutError("provider timed out") + + monkeypatch.setattr( + job_lead_sources, + "classify_job_lead_with_jev", + fail_shadow, + ) + + classification = classifier.classify("Acme | Contract API engineer | Remote") + + assert classification.is_contractor_friendly is True + assert classification.jev_shadow is not None + assert classification.jev_shadow.status == "failed" + assert classification.jev_shadow.error == "TimeoutError: provider timed out" + + next_classification = classifier.classify("Beta | Contract API engineer | Remote") + + assert shadow_calls == 1 + assert next_classification.jev_shadow is None + + def test_build_llm_client_uses_classifier_model_for_fireworks_direct( monkeypatch, ) -> None: diff --git a/tests/unit/test_runtime_config.py b/tests/unit/test_runtime_config.py index 08f59c81..6e5db6f9 100644 --- a/tests/unit/test_runtime_config.py +++ b/tests/unit/test_runtime_config.py @@ -178,6 +178,26 @@ def test_env_value_locks_matching_runtime_config( assert definition_is_env_locked(definition) +def test_job_lead_jev_shadow_runtime_config_has_safe_bounds() -> None: + sample_rate = runtime_config_definition_for_key("JOB_LEAD_JEV_SHADOW_SAMPLE_RATE") + threshold = runtime_config_definition_for_key( + "JOB_LEAD_JEV_SHADOW_CONFIDENCE_THRESHOLD" + ) + enabled = runtime_config_definition_for_key("JOB_LEAD_JEV_SHADOW_ENABLED") + + assert enabled is not None + assert enabled.value_type == "bool" + assert sample_rate is not None + assert sample_rate.min_value == 0.0 + assert sample_rate.max_value == 1.0 + assert threshold is not None + assert threshold.min_value == 0.5 + assert threshold.max_value == 1.0 + assert coerce_runtime_config_value(threshold, "0.8") == "0.8" + with pytest.raises(ValueError, match="greater than or equal to 0.5"): + coerce_runtime_config_value(threshold, "0.4") + + def test_outline_admin_runtime_config_supports_legacy_dashboard_values( monkeypatch: pytest.MonkeyPatch, ) -> None: From 77800d690271e32a9af2d04acb9969c30fd77d51 Mon Sep 17 00:00:00 2001 From: Michael Wu Date: Mon, 21 Sep 2026 16:26:20 +0900 Subject: [PATCH 2/4] fix: bound Jev shadow execution --- .env.example | 16 +- apps/worker/src/five08/worker/config.py | 6 + docs/configuration.md | 9 +- packages/shared/src/five08/job_lead_evals.py | 1 + packages/shared/src/five08/job_lead_jev.py | 8 + .../shared/src/five08/job_lead_sources.py | 153 ++++++++++++++++- packages/shared/src/five08/runtime_config.py | 22 +++ tests/unit/test_job_lead_evals.py | 2 + tests/unit/test_job_lead_sources.py | 156 +++++++++++++++--- tests/unit/test_runtime_config.py | 14 ++ 10 files changed, 353 insertions(+), 34 deletions(-) diff --git a/.env.example b/.env.example index 95ed24cb..da45ad38 100644 --- a/.env.example +++ b/.env.example @@ -224,12 +224,16 @@ JOB_LEAD_CLASSIFIER_MODEL= JOB_LEAD_CLASSIFIER_TIMEOUT_SECONDS=8.0 # Optional Jev canary. It never changes the production classification. Configure # OPENROUTER_API_KEY, enable the shadow, and inspect the scrape job result in the -# dashboard background-task detail view. -JOB_LEAD_JEV_SHADOW_ENABLED=false -JOB_LEAD_JEV_SHADOW_MODEL=typesafe/jev-1.13 -JOB_LEAD_JEV_SHADOW_SAMPLE_RATE=0.1 -JOB_LEAD_JEV_SHADOW_CONFIDENCE_THRESHOLD=0.8 -JOB_LEAD_JEV_SHADOW_TIMEOUT_SECONDS=4.0 +# dashboard background-task detail view. These are the code defaults; leave them +# commented to keep the corresponding dashboard controls editable. Uncommenting +# a value intentionally locks that setting to the environment. +# JOB_LEAD_JEV_SHADOW_ENABLED=false +# JOB_LEAD_JEV_SHADOW_MODEL=typesafe/jev-1.13 +# JOB_LEAD_JEV_SHADOW_SAMPLE_RATE=0.1 +# JOB_LEAD_JEV_SHADOW_CONFIDENCE_THRESHOLD=0.8 +# JOB_LEAD_JEV_SHADOW_TIMEOUT_SECONDS=4.0 +# JOB_LEAD_JEV_SHADOW_MAX_CALLS=25 +# JOB_LEAD_JEV_SHADOW_RUN_BUDGET_SECONDS=20.0 # Optional deterministic agent tool integrations. # GitHub Issues are the canonical todo backend. Members can read and write the # default repo; Steering Committee/Admin/Owner can work across every repository diff --git a/apps/worker/src/five08/worker/config.py b/apps/worker/src/five08/worker/config.py index 0d70c064..8f0228e9 100644 --- a/apps/worker/src/five08/worker/config.py +++ b/apps/worker/src/five08/worker/config.py @@ -87,6 +87,12 @@ class WorkerSettings(SharedSettings): le=1.0, ) job_lead_jev_shadow_timeout_seconds: float = Field(default=4.0, gt=0, le=30.0) + job_lead_jev_shadow_max_calls: int = Field(default=25, ge=1, le=100) + job_lead_jev_shadow_run_budget_seconds: float = Field( + default=20.0, + gt=0, + le=60.0, + ) resume_ai_api_key: str | None = None resume_ai_base_url: str | None = None resume_ai_model: str = "gpt-4.1-mini" diff --git a/docs/configuration.md b/docs/configuration.md index ce855540..d37f6509 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -238,6 +238,8 @@ intake-completed field unset, and matches resume filenames with - `JOB_LEAD_JEV_SHADOW_SAMPLE_RATE` - `JOB_LEAD_JEV_SHADOW_CONFIDENCE_THRESHOLD` - `JOB_LEAD_JEV_SHADOW_TIMEOUT_SECONDS` +- `JOB_LEAD_JEV_SHADOW_MAX_CALLS` +- `JOB_LEAD_JEV_SHADOW_RUN_BUDGET_SECONDS` The Jev job-lead shadow is disabled by default and requires `OPENROUTER_API_KEY`. When enabled, it deterministically samples eligible HN @@ -247,9 +249,10 @@ classification. View the latest data in **Dashboard → Background tasks**, open the `scrape_job_leads_job` run, and inspect `Result → classifier_shadow`. Disagreements, confidence fallbacks, and provider failures include only the HN item ID and URL plus normalized decisions; raw post text and provider responses -are not retained in the shadow report. A provider failure disables further Jev -calls for that scrape run so an unavailable shadow service cannot repeatedly -delay the production job. +are not retained in the shadow report. At most 25 calls within a 20-second +wall-clock window are allowed by default per scrape run, even if a larger sample +is selected. A provider failure disables further Jev calls for that run so an +unavailable shadow service cannot repeatedly delay the production job. Resume/profile LLM calls retry matching direct providers after Bifrost request failures when direct provider credentials are configured. diff --git a/packages/shared/src/five08/job_lead_evals.py b/packages/shared/src/five08/job_lead_evals.py index 7ebff597..7a5c4ed3 100644 --- a/packages/shared/src/five08/job_lead_evals.py +++ b/packages/shared/src/five08/job_lead_evals.py @@ -385,6 +385,7 @@ def _run_jev( latency_ms=_elapsed_ms(started), request_attempts=decision.request_attempts, input_tokens=decision.input_tokens, + cached_input_tokens=decision.cached_input_tokens, output_tokens=decision.output_tokens, total_tokens=decision.total_tokens, cost_usd=decision.cost_usd, diff --git a/packages/shared/src/five08/job_lead_jev.py b/packages/shared/src/five08/job_lead_jev.py index ed83f7fc..53b76a4b 100644 --- a/packages/shared/src/five08/job_lead_jev.py +++ b/packages/shared/src/five08/job_lead_jev.py @@ -41,6 +41,7 @@ class JobLeadJevDecision: latency_ms: int request_attempts: int input_tokens: int + cached_input_tokens: int output_tokens: int total_tokens: int cost_usd: float | None @@ -146,6 +147,7 @@ def classify_job_lead_with_jev( latency_ms=max(0, round((time.perf_counter() - started) * 1000)), request_attempts=attempts, input_tokens=usage["input_tokens"], + cached_input_tokens=usage["cached_input_tokens"], output_tokens=usage["output_tokens"], total_tokens=usage["total_tokens"], cost_usd=usage["cost_usd"], @@ -274,6 +276,11 @@ def _probabilities(value: Any, *, name: str) -> dict[str, float]: def _usage(value: Any) -> dict[str, Any]: source = value if isinstance(value, dict) else {} input_tokens = _integer(source.get("input_tokens", source.get("prompt_tokens"))) + raw_details = source.get("input_tokens_details") or source.get( + "prompt_tokens_details" + ) + details = raw_details if isinstance(raw_details, dict) else {} + cached_input_tokens = _integer(details.get("cached_tokens")) output_tokens = _integer( source.get("output_tokens", source.get("completion_tokens")) ) @@ -281,6 +288,7 @@ def _usage(value: Any) -> dict[str, Any]: cost = source.get("cost") return { "input_tokens": input_tokens, + "cached_input_tokens": cached_input_tokens, "output_tokens": output_tokens, "total_tokens": total_tokens, "cost_usd": _optional_float(cost), diff --git a/packages/shared/src/five08/job_lead_sources.py b/packages/shared/src/five08/job_lead_sources.py index 97880f59..80dd447d 100644 --- a/packages/shared/src/five08/job_lead_sources.py +++ b/packages/shared/src/five08/job_lead_sources.py @@ -8,6 +8,7 @@ import logging import re import time +from collections.abc import Callable from dataclasses import dataclass, replace from datetime import datetime, timezone from html.parser import HTMLParser @@ -182,6 +183,7 @@ class JobLeadJevShadowObservation: agrees_with_primary: bool | None = None request_attempts: int = 0 input_tokens: int = 0 + cached_input_tokens: int = 0 output_tokens: int = 0 total_tokens: int = 0 cost_usd: float | None = None @@ -212,6 +214,7 @@ def payload(self) -> dict[str, Any]: "latency_ms": self.latency_ms, "request_attempts": self.request_attempts, "input_tokens": self.input_tokens, + "cached_input_tokens": self.cached_input_tokens, "output_tokens": self.output_tokens, "total_tokens": self.total_tokens, "cost_usd": self.cost_usd, @@ -651,6 +654,7 @@ def __init__( settings: SharedSettings, client: Any | None = None, jev_shadow_session: requests.Session | None = None, + jev_shadow_clock: Callable[[], float] | None = None, ) -> None: self.settings = settings self.client = client if client is not None else _build_llm_client(settings) @@ -680,6 +684,24 @@ def __init__( maximum=30.0, default=4.0, ) + self._jev_shadow_max_calls = _bounded_int( + getattr(settings, "job_lead_jev_shadow_max_calls", 25), + minimum=1, + maximum=100, + default=25, + ) + self._jev_shadow_run_budget_seconds = _bounded_float( + getattr(settings, "job_lead_jev_shadow_run_budget_seconds", 20.0), + minimum=0.1, + maximum=60.0, + default=20.0, + ) + self._jev_shadow_clock = jev_shadow_clock or time.perf_counter + self._jev_shadow_calls_started = 0 + self._jev_shadow_run_started_at: float | None = None + self._jev_shadow_budget_exhaustion_reason: ( + Literal["max_calls", "run_budget"] | None + ) = None self._owns_jev_shadow_session = False self._jev_shadow_available = True self._jev_shadow_session = jev_shadow_session @@ -711,6 +733,42 @@ def close(self) -> None: self._jev_shadow_session.close() self._jev_shadow_session = None + def jev_shadow_run_summary(self) -> dict[str, Any]: + """Return bounded-work state for the current scrape report.""" + + elapsed_ms = 0 + if self._jev_shadow_run_started_at is not None: + elapsed_ms = max( + 0, + round( + (self._jev_shadow_clock() - self._jev_shadow_run_started_at) * 1000 + ), + ) + return { + "calls_started": self._jev_shadow_calls_started, + "max_calls": self._jev_shadow_max_calls, + "run_budget_seconds": self._jev_shadow_run_budget_seconds, + "run_elapsed_ms": elapsed_ms, + "budget_exhaustion_reason": self._jev_shadow_budget_exhaustion_reason, + } + + def _next_jev_shadow_timeout(self) -> float | None: + if self._jev_shadow_calls_started >= self._jev_shadow_max_calls: + self._jev_shadow_budget_exhaustion_reason = "max_calls" + return None + + now = self._jev_shadow_clock() + if self._jev_shadow_run_started_at is None: + self._jev_shadow_run_started_at = now + elapsed = max(0.0, now - self._jev_shadow_run_started_at) + remaining = self._jev_shadow_run_budget_seconds - elapsed + if remaining < 0.1: + self._jev_shadow_budget_exhaustion_reason = "run_budget" + return None + + self._jev_shadow_calls_started += 1 + return min(self._jev_shadow_timeout_seconds, remaining) + def _attach_jev_shadow( self, comment_text: str, @@ -728,14 +786,18 @@ def _attach_jev_shadow( ): return classification - started = time.perf_counter() + shadow_timeout_seconds = self._next_jev_shadow_timeout() + if shadow_timeout_seconds is None: + return classification + + started = self._jev_shadow_clock() try: decision = classify_job_lead_with_jev( session=self._jev_shadow_session, api_key=self._jev_shadow_api_key, comment_text=comment_text, model=self._jev_shadow_model, - timeout_seconds=self._jev_shadow_timeout_seconds, + timeout_seconds=shadow_timeout_seconds, max_attempts=1, ) observation = _successful_jev_shadow_observation( @@ -751,7 +813,10 @@ def _attach_jev_shadow( confidence_threshold=self._jev_shadow_confidence_threshold, primary_is_contractor_friendly=(classification.is_contractor_friendly), primary_posting_type=classification.posting_type, - latency_ms=max(0, round((time.perf_counter() - started) * 1000)), + latency_ms=max( + 0, + round((self._jev_shadow_clock() - started) * 1000), + ), observed_at=datetime.now(timezone.utc), error=_safe_jev_shadow_error(exc), ) @@ -854,6 +919,22 @@ def _bounded_float( return max(minimum, min(maximum, numeric)) +def _bounded_int( + value: object, + *, + minimum: int, + maximum: int, + default: int, +) -> int: + if isinstance(value, bool) or not isinstance(value, int | str): + return default + try: + numeric = int(value) + except (TypeError, ValueError): + return default + return max(minimum, min(maximum, numeric)) + + def _selected_for_jev_shadow(comment_text: str, sample_rate: float) -> bool: if sample_rate <= 0.0: return False @@ -893,6 +974,7 @@ def _successful_jev_shadow_observation( latency_ms=decision.latency_ms, request_attempts=decision.request_attempts, input_tokens=decision.input_tokens, + cached_input_tokens=decision.cached_input_tokens, output_tokens=decision.output_tokens, total_tokens=decision.total_tokens, cost_usd=decision.cost_usd, @@ -1226,6 +1308,8 @@ def build_job_lead_source( def _job_lead_jev_shadow_report( settings: SharedSettings, leads: list[JobLeadInput], + *, + runtime_summary: dict[str, Any] | None = None, ) -> dict[str, Any]: enabled = bool(getattr(settings, "job_lead_jev_shadow_enabled", False)) api_key_configured = bool(_clean(getattr(settings, "openrouter_api_key", None))) @@ -1245,6 +1329,18 @@ def _job_lead_jev_shadow_report( maximum=1.0, default=0.8, ) + max_calls = _bounded_int( + getattr(settings, "job_lead_jev_shadow_max_calls", 25), + minimum=1, + maximum=100, + default=25, + ) + run_budget_seconds = _bounded_float( + getattr(settings, "job_lead_jev_shadow_run_budget_seconds", 20.0), + minimum=0.1, + maximum=60.0, + default=20.0, + ) observed: list[tuple[JobLeadInput, dict[str, Any], dict[str, Any]]] = [] for lead in leads: metadata = lead.metadata if isinstance(lead.metadata, dict) else {} @@ -1278,6 +1374,24 @@ def _job_lead_jev_shadow_report( if isinstance(item[2].get("cost_usd"), int | float) and not isinstance(item[2].get("cost_usd"), bool) ] + runtime = runtime_summary if isinstance(runtime_summary, dict) else {} + raw_calls_started = runtime.get("calls_started") + calls_started = ( + int(raw_calls_started) + if isinstance(raw_calls_started, int) + and not isinstance(raw_calls_started, bool) + else len(observed) + ) + raw_run_elapsed_ms = runtime.get("run_elapsed_ms") + run_elapsed_ms = ( + int(raw_run_elapsed_ms) + if isinstance(raw_run_elapsed_ms, int | float) + and not isinstance(raw_run_elapsed_ms, bool) + else None + ) + budget_exhaustion_reason = runtime.get("budget_exhaustion_reason") + if budget_exhaustion_reason not in {"max_calls", "run_budget"}: + budget_exhaustion_reason = None review_candidates: list[dict[str, Any]] = [] for lead, primary, shadow in observed: @@ -1329,6 +1443,8 @@ def _job_lead_jev_shadow_report( status = "no_sample_selected" elif failures: status = "completed_with_errors" + elif budget_exhaustion_reason is not None: + status = "completed_budget_limited" else: status = "completed" @@ -1342,6 +1458,14 @@ def _job_lead_jev_shadow_report( "requested_model": model, "sample_rate": sample_rate, "confidence_threshold": confidence_threshold, + "limits": { + "max_calls": max_calls, + "run_budget_seconds": run_budget_seconds, + }, + "calls_started": calls_started, + "run_elapsed_ms": run_elapsed_ms, + "budget_exhausted": budget_exhaustion_reason is not None, + "budget_exhaustion_reason": budget_exhaustion_reason, "eligible": len(leads), "attempted": attempted, "not_observed": max(0, len(leads) - attempted), @@ -1372,6 +1496,9 @@ def _job_lead_jev_shadow_report( "usage": { "unit": "tokens", "input": sum(int(item[2].get("input_tokens") or 0) for item in successful), + "cached": sum( + int(item[2].get("cached_input_tokens") or 0) for item in successful + ), "output": sum( int(item[2].get("output_tokens") or 0) for item in successful ), @@ -1417,7 +1544,25 @@ def scrape_job_leads( close_classifier = getattr(classifier, "close", None) if callable(close_classifier): close_classifier() - shadow_report = _job_lead_jev_shadow_report(settings, leads) + shadow_runtime_summary_factory = getattr( + classifier, + "jev_shadow_run_summary", + None, + ) + raw_shadow_runtime_summary = ( + shadow_runtime_summary_factory() + if callable(shadow_runtime_summary_factory) + else None + ) + shadow_report = _job_lead_jev_shadow_report( + settings, + leads, + runtime_summary=( + raw_shadow_runtime_summary + if isinstance(raw_shadow_runtime_summary, dict) + else None + ), + ) collection_report_factory = getattr(adapter, "collection_report", None) raw_collection_report = ( collection_report_factory() if callable(collection_report_factory) else {} diff --git a/packages/shared/src/five08/runtime_config.py b/packages/shared/src/five08/runtime_config.py index 92ed63df..7f4bc56c 100644 --- a/packages/shared/src/five08/runtime_config.py +++ b/packages/shared/src/five08/runtime_config.py @@ -553,6 +553,28 @@ class RuntimeConfigDBSnapshot: min_value=0.1, max_value=30.0, ), + RuntimeConfigDefinition( + key="JOB_LEAD_JEV_SHADOW_MAX_CALLS", + attr="job_lead_jev_shadow_max_calls", + label="Jev job lead shadow maximum calls", + category="AI", + description="Maximum Jev requests allowed during one scrape run.", + value_type="int", + env_names=("JOB_LEAD_JEV_SHADOW_MAX_CALLS",), + min_value=1, + max_value=100, + ), + RuntimeConfigDefinition( + key="JOB_LEAD_JEV_SHADOW_RUN_BUDGET_SECONDS", + attr="job_lead_jev_shadow_run_budget_seconds", + label="Jev job lead shadow run budget seconds", + category="AI", + description="Total wall-clock budget for Jev requests during one scrape run.", + value_type="float", + env_names=("JOB_LEAD_JEV_SHADOW_RUN_BUDGET_SECONDS",), + min_value=0.1, + max_value=60.0, + ), RuntimeConfigDefinition( key="RESUME_AI_API_KEY", attr="resume_ai_api_key", diff --git a/tests/unit/test_job_lead_evals.py b/tests/unit/test_job_lead_evals.py index 922c8914..aaa19cfe 100644 --- a/tests/unit/test_job_lead_evals.py +++ b/tests/unit/test_job_lead_evals.py @@ -55,6 +55,7 @@ def json(self) -> dict: }, "usage": { "input_tokens": 450, + "input_tokens_details": {"cached_tokens": 50}, "output_tokens": 73, "cost": 0.000019, }, @@ -191,6 +192,7 @@ def test_jev_response_is_normalized_without_raw_provider_output() -> None: assert observation.contractor_probability == 0.91 assert observation.resolved_model == "typesafe/jev-1.13-20260917" assert observation.input_tokens == 450 + assert observation.cached_input_tokens == 50 assert observation.output_tokens == 73 assert observation.total_tokens == 523 assert observation.cost_usd == 0.000019 diff --git a/tests/unit/test_job_lead_sources.py b/tests/unit/test_job_lead_sources.py index a1f4feff..93ded543 100644 --- a/tests/unit/test_job_lead_sources.py +++ b/tests/unit/test_job_lead_sources.py @@ -106,6 +106,31 @@ def classify(self, comment_text: str) -> JobLeadClassification: ) +def _jev_decision() -> JobLeadJevDecision: + return JobLeadJevDecision( + requested_model="typesafe/jev-1.13", + resolved_model="typesafe/jev-1.13-20260917", + provider="TypeSafe", + is_contractor_friendly=True, + contractor_probability=0.91, + posting_type=JobPostingType.PART_TIME, + posting_confidence=0.98, + posting_probabilities={ + "part_time": 0.98, + "full_time": 0.01, + "part_time_or_full_time": 0.01, + "unknown": 0.0, + }, + latency_ms=420, + request_attempts=1, + input_tokens=450, + cached_input_tokens=50, + output_tokens=73, + total_tokens=523, + cost_usd=0.000019, + ) + + class _FakeClassifierHackerNewsClient(_FakeHackerNewsClient): def get_algolia_item_tree(self, item_id: int) -> dict: assert item_id == 48357725 @@ -718,27 +743,7 @@ def test_jev_shadow_records_disagreement_without_changing_primary( rationale="Employee-only role.", method="llm", ) - decision = JobLeadJevDecision( - requested_model="typesafe/jev-1.13", - resolved_model="typesafe/jev-1.13-20260917", - provider="TypeSafe", - is_contractor_friendly=True, - contractor_probability=0.91, - posting_type=JobPostingType.PART_TIME, - posting_confidence=0.98, - posting_probabilities={ - "part_time": 0.98, - "full_time": 0.01, - "part_time_or_full_time": 0.01, - "unknown": 0.0, - }, - latency_ms=420, - request_attempts=1, - input_tokens=450, - output_tokens=73, - total_tokens=523, - cost_usd=0.000019, - ) + decision = _jev_decision() classifier = JobLeadClassifier( settings=settings, # type: ignore[arg-type] client=object(), @@ -785,6 +790,7 @@ def test_jev_shadow_records_disagreement_without_changing_primary( assert report["usage"] == { "unit": "tokens", "input": 450, + "cached": 50, "output": 73, "total": 523, "cost_usd": 0.000019, @@ -837,6 +843,114 @@ def fail_shadow(**_kwargs: object) -> JobLeadJevDecision: assert next_classification.jev_shadow is None +def test_jev_shadow_stops_when_run_time_budget_is_exhausted(monkeypatch) -> None: + settings = SimpleNamespace( + job_lead_jev_shadow_enabled=True, + job_lead_jev_shadow_sample_rate=1.0, + job_lead_jev_shadow_confidence_threshold=0.8, + job_lead_jev_shadow_timeout_seconds=4.0, + job_lead_jev_shadow_max_calls=10, + job_lead_jev_shadow_run_budget_seconds=0.1, + openrouter_api_key="test-key", + ) + primary = classify_contractor_lead_heuristic( + "Acme | Contract API engineer | Remote" + ) + now = [10.0] + timeouts: list[float] = [] + classifier = JobLeadClassifier( + settings=settings, # type: ignore[arg-type] + client=object(), + jev_shadow_session=object(), # type: ignore[arg-type] + jev_shadow_clock=lambda: now[0], + ) + monkeypatch.setattr(classifier, "_classify_with_llm", lambda _text: primary) + + def classify_shadow(**kwargs: object) -> JobLeadJevDecision: + timeouts.append(float(kwargs["timeout_seconds"])) + now[0] += 0.2 + return _jev_decision() + + monkeypatch.setattr( + job_lead_sources, + "classify_job_lead_with_jev", + classify_shadow, + ) + + first = classifier.classify("Acme | Contract API engineer | Remote") + second = classifier.classify("Beta | Contract API engineer | Remote") + + assert first.jev_shadow is not None + assert second.jev_shadow is None + assert len(timeouts) == 1 + assert 0.09 <= timeouts[0] <= 0.1 + runtime_summary = classifier.jev_shadow_run_summary() + assert runtime_summary["budget_exhaustion_reason"] == "run_budget" + + lead = JobLeadInput( + source_key="hackernews_who_is_hiring", + source_type="hackernews", + external_id="budget-1", + source_url="https://news.ycombinator.com/item?id=budget-1", + title="Budgeted shadow", + body_raw="not retained in report", + body_normalized="not retained in report", + metadata=job_lead_sources._classification_metadata(first), # noqa: SLF001 + ) + report = job_lead_sources._job_lead_jev_shadow_report( # noqa: SLF001 + settings, # type: ignore[arg-type] + [lead], + runtime_summary=runtime_summary, + ) + assert report["status"] == "completed_budget_limited" + assert report["calls_started"] == 1 + assert report["budget_exhausted"] is True + assert report["budget_exhaustion_reason"] == "run_budget" + + +def test_jev_shadow_stops_at_per_run_call_cap(monkeypatch) -> None: + settings = SimpleNamespace( + job_lead_jev_shadow_enabled=True, + job_lead_jev_shadow_sample_rate=1.0, + job_lead_jev_shadow_confidence_threshold=0.8, + job_lead_jev_shadow_timeout_seconds=4.0, + job_lead_jev_shadow_max_calls=1, + job_lead_jev_shadow_run_budget_seconds=20.0, + openrouter_api_key="test-key", + ) + primary = classify_contractor_lead_heuristic( + "Acme | Contract API engineer | Remote" + ) + shadow_calls = 0 + classifier = JobLeadClassifier( + settings=settings, # type: ignore[arg-type] + client=object(), + jev_shadow_session=object(), # type: ignore[arg-type] + ) + monkeypatch.setattr(classifier, "_classify_with_llm", lambda _text: primary) + + def classify_shadow(**_kwargs: object) -> JobLeadJevDecision: + nonlocal shadow_calls + shadow_calls += 1 + return _jev_decision() + + monkeypatch.setattr( + job_lead_sources, + "classify_job_lead_with_jev", + classify_shadow, + ) + + first = classifier.classify("Acme | Contract API engineer | Remote") + second = classifier.classify("Beta | Contract API engineer | Remote") + + assert first.jev_shadow is not None + assert second.jev_shadow is None + assert shadow_calls == 1 + assert classifier.jev_shadow_run_summary()["budget_exhaustion_reason"] == ( + "max_calls" + ) + + def test_build_llm_client_uses_classifier_model_for_fireworks_direct( monkeypatch, ) -> None: diff --git a/tests/unit/test_runtime_config.py b/tests/unit/test_runtime_config.py index 6e5db6f9..0dc08f89 100644 --- a/tests/unit/test_runtime_config.py +++ b/tests/unit/test_runtime_config.py @@ -183,6 +183,10 @@ def test_job_lead_jev_shadow_runtime_config_has_safe_bounds() -> None: threshold = runtime_config_definition_for_key( "JOB_LEAD_JEV_SHADOW_CONFIDENCE_THRESHOLD" ) + max_calls = runtime_config_definition_for_key("JOB_LEAD_JEV_SHADOW_MAX_CALLS") + run_budget = runtime_config_definition_for_key( + "JOB_LEAD_JEV_SHADOW_RUN_BUDGET_SECONDS" + ) enabled = runtime_config_definition_for_key("JOB_LEAD_JEV_SHADOW_ENABLED") assert enabled is not None @@ -193,9 +197,19 @@ def test_job_lead_jev_shadow_runtime_config_has_safe_bounds() -> None: assert threshold is not None assert threshold.min_value == 0.5 assert threshold.max_value == 1.0 + assert max_calls is not None + assert max_calls.value_type == "int" + assert max_calls.min_value == 1 + assert max_calls.max_value == 100 + assert run_budget is not None + assert run_budget.value_type == "float" + assert run_budget.min_value == 0.1 + assert run_budget.max_value == 60.0 assert coerce_runtime_config_value(threshold, "0.8") == "0.8" with pytest.raises(ValueError, match="greater than or equal to 0.5"): coerce_runtime_config_value(threshold, "0.4") + with pytest.raises(ValueError, match="less than or equal to 60"): + coerce_runtime_config_value(run_budget, "61") def test_outline_admin_runtime_config_supports_legacy_dashboard_values( From e2cabae428854965cbdc263036202cb4ac1660f5 Mon Sep 17 00:00:00 2001 From: Michael Wu Date: Mon, 21 Sep 2026 16:44:07 +0900 Subject: [PATCH 3/4] fix: report captured Jev shadow configuration --- packages/shared/src/five08/job_lead_jev.py | 4 +- .../shared/src/five08/job_lead_sources.py | 57 +++++++++++++++---- tests/unit/test_job_lead_sources.py | 46 +++++++++++++++ 3 files changed, 93 insertions(+), 14 deletions(-) diff --git a/packages/shared/src/five08/job_lead_jev.py b/packages/shared/src/five08/job_lead_jev.py index 53b76a4b..52247e51 100644 --- a/packages/shared/src/five08/job_lead_jev.py +++ b/packages/shared/src/five08/job_lead_jev.py @@ -210,9 +210,7 @@ def _post_json_with_retries( else: message = _optional_text(error) or "unknown error" raise JobLeadJevRequestError( - RuntimeError( - f"OpenRouter HTTP {response.status_code}: {message[:300]}" - ), + RuntimeError(f"OpenRouter HTTP {response.status_code}: {message[:300]}"), request_attempts=attempt, ) if not isinstance(body, dict): diff --git a/packages/shared/src/five08/job_lead_sources.py b/packages/shared/src/five08/job_lead_sources.py index 80dd447d..800c8a18 100644 --- a/packages/shared/src/five08/job_lead_sources.py +++ b/packages/shared/src/five08/job_lead_sources.py @@ -734,7 +734,7 @@ def close(self) -> None: self._jev_shadow_session = None def jev_shadow_run_summary(self) -> dict[str, Any]: - """Return bounded-work state for the current scrape report.""" + """Return the captured configuration and bounded-work state for this run.""" elapsed_ms = 0 if self._jev_shadow_run_started_at is not None: @@ -745,6 +745,12 @@ def jev_shadow_run_summary(self) -> dict[str, Any]: ), ) return { + "enabled": self._jev_shadow_enabled, + "provider_configured": bool(self._jev_shadow_api_key), + "requested_model": self._jev_shadow_model, + "sample_rate": self._jev_shadow_sample_rate, + "confidence_threshold": self._jev_shadow_confidence_threshold, + "request_timeout_seconds": self._jev_shadow_timeout_seconds, "calls_started": self._jev_shadow_calls_started, "max_calls": self._jev_shadow_max_calls, "run_budget_seconds": self._jev_shadow_run_budget_seconds, @@ -1311,32 +1317,61 @@ def _job_lead_jev_shadow_report( *, runtime_summary: dict[str, Any] | None = None, ) -> dict[str, Any]: - enabled = bool(getattr(settings, "job_lead_jev_shadow_enabled", False)) - api_key_configured = bool(_clean(getattr(settings, "openrouter_api_key", None))) - model = ( - _clean(getattr(settings, "job_lead_jev_shadow_model", None)) - or DEFAULT_JOB_LEAD_JEV_MODEL + runtime = runtime_summary if isinstance(runtime_summary, dict) else {} + raw_enabled = runtime.get("enabled") + enabled = ( + raw_enabled + if isinstance(raw_enabled, bool) + else bool(getattr(settings, "job_lead_jev_shadow_enabled", False)) + ) + raw_provider_configured = runtime.get("provider_configured") + api_key_configured = ( + raw_provider_configured + if isinstance(raw_provider_configured, bool) + else bool(_clean(getattr(settings, "openrouter_api_key", None))) + ) + raw_model = ( + runtime["requested_model"] + if "requested_model" in runtime + else getattr(settings, "job_lead_jev_shadow_model", None) ) + model = _clean(raw_model) or DEFAULT_JOB_LEAD_JEV_MODEL sample_rate = _bounded_float( - getattr(settings, "job_lead_jev_shadow_sample_rate", 0.1), + runtime["sample_rate"] + if "sample_rate" in runtime + else getattr(settings, "job_lead_jev_shadow_sample_rate", 0.1), minimum=0.0, maximum=1.0, default=0.1, ) confidence_threshold = _bounded_float( - getattr(settings, "job_lead_jev_shadow_confidence_threshold", 0.8), + runtime["confidence_threshold"] + if "confidence_threshold" in runtime + else getattr(settings, "job_lead_jev_shadow_confidence_threshold", 0.8), minimum=0.5, maximum=1.0, default=0.8, ) + request_timeout_seconds = _bounded_float( + runtime["request_timeout_seconds"] + if "request_timeout_seconds" in runtime + else getattr(settings, "job_lead_jev_shadow_timeout_seconds", 4.0), + minimum=0.1, + maximum=30.0, + default=4.0, + ) max_calls = _bounded_int( - getattr(settings, "job_lead_jev_shadow_max_calls", 25), + runtime["max_calls"] + if "max_calls" in runtime + else getattr(settings, "job_lead_jev_shadow_max_calls", 25), minimum=1, maximum=100, default=25, ) run_budget_seconds = _bounded_float( - getattr(settings, "job_lead_jev_shadow_run_budget_seconds", 20.0), + runtime["run_budget_seconds"] + if "run_budget_seconds" in runtime + else getattr(settings, "job_lead_jev_shadow_run_budget_seconds", 20.0), minimum=0.1, maximum=60.0, default=20.0, @@ -1374,7 +1409,6 @@ def _job_lead_jev_shadow_report( if isinstance(item[2].get("cost_usd"), int | float) and not isinstance(item[2].get("cost_usd"), bool) ] - runtime = runtime_summary if isinstance(runtime_summary, dict) else {} raw_calls_started = runtime.get("calls_started") calls_started = ( int(raw_calls_started) @@ -1459,6 +1493,7 @@ def _job_lead_jev_shadow_report( "sample_rate": sample_rate, "confidence_threshold": confidence_threshold, "limits": { + "request_timeout_seconds": request_timeout_seconds, "max_calls": max_calls, "run_budget_seconds": run_budget_seconds, }, diff --git a/tests/unit/test_job_lead_sources.py b/tests/unit/test_job_lead_sources.py index 93ded543..c1822a9e 100644 --- a/tests/unit/test_job_lead_sources.py +++ b/tests/unit/test_job_lead_sources.py @@ -951,6 +951,52 @@ def classify_shadow(**_kwargs: object) -> JobLeadJevDecision: ) +def test_jev_shadow_report_uses_classifier_configuration_snapshot() -> None: + settings = SimpleNamespace( + job_lead_jev_shadow_enabled=True, + job_lead_jev_shadow_model="typesafe/jev-1.13", + job_lead_jev_shadow_sample_rate=0.25, + job_lead_jev_shadow_confidence_threshold=0.85, + job_lead_jev_shadow_timeout_seconds=3.0, + job_lead_jev_shadow_max_calls=7, + job_lead_jev_shadow_run_budget_seconds=12.0, + openrouter_api_key="test-key", + ) + classifier = JobLeadClassifier( + settings=settings, # type: ignore[arg-type] + client=object(), + jev_shadow_session=object(), # type: ignore[arg-type] + ) + run_snapshot = classifier.jev_shadow_run_summary() + + settings.job_lead_jev_shadow_enabled = False + settings.job_lead_jev_shadow_model = "typesafe/jev-changed" + settings.job_lead_jev_shadow_sample_rate = 0.0 + settings.job_lead_jev_shadow_confidence_threshold = 0.95 + settings.job_lead_jev_shadow_timeout_seconds = 9.0 + settings.job_lead_jev_shadow_max_calls = 99 + settings.job_lead_jev_shadow_run_budget_seconds = 55.0 + settings.openrouter_api_key = None + + report = job_lead_sources._job_lead_jev_shadow_report( # noqa: SLF001 + settings, # type: ignore[arg-type] + [], + runtime_summary=run_snapshot, + ) + + assert report["status"] == "no_sample_selected" + assert report["enabled"] is True + assert report["provider_configured"] is True + assert report["requested_model"] == "typesafe/jev-1.13" + assert report["sample_rate"] == 0.25 + assert report["confidence_threshold"] == 0.85 + assert report["limits"] == { + "request_timeout_seconds": 3.0, + "max_calls": 7, + "run_budget_seconds": 12.0, + } + + def test_build_llm_client_uses_classifier_model_for_fireworks_direct( monkeypatch, ) -> None: From 1c05c6c4d2a85ec79c09594c581f9acaf7edab78 Mon Sep 17 00:00:00 2001 From: Michael Wu Date: Mon, 21 Sep 2026 16:58:30 +0900 Subject: [PATCH 4/4] fix: sanitize Jev shadow failures --- .../shared/src/five08/job_lead_sources.py | 27 ++++++++++++++--- tests/unit/test_job_lead_sources.py | 29 +++++++++++++++++-- 2 files changed, 49 insertions(+), 7 deletions(-) diff --git a/packages/shared/src/five08/job_lead_sources.py b/packages/shared/src/five08/job_lead_sources.py index 800c8a18..4fad51c9 100644 --- a/packages/shared/src/five08/job_lead_sources.py +++ b/packages/shared/src/five08/job_lead_sources.py @@ -23,6 +23,7 @@ from five08.job_lead_jev import ( DEFAULT_JOB_LEAD_JEV_MODEL, JobLeadJevDecision, + JobLeadJevRequestError, classify_job_lead_with_jev, ) from five08.job_leads import ( @@ -813,6 +814,7 @@ def _attach_jev_shadow( ) except Exception as exc: self._jev_shadow_available = False + error_category = _safe_jev_shadow_error(exc) observation = JobLeadJevShadowObservation( status="failed", requested_model=self._jev_shadow_model, @@ -824,12 +826,13 @@ def _attach_jev_shadow( round((self._jev_shadow_clock() - started) * 1000), ), observed_at=datetime.now(timezone.utc), - error=_safe_jev_shadow_error(exc), + request_attempts=_jev_shadow_request_attempts(exc), + error=error_category, ) logger.warning( "Jev job-lead shadow classification failed; disabling it for the " "remainder of this scrape: %s", - exc, + error_category, ) return replace(classification, jev_shadow=observation) @@ -988,8 +991,24 @@ def _successful_jev_shadow_observation( def _safe_jev_shadow_error(exc: Exception) -> str: - message = " ".join(str(exc).split())[:240] - return f"{type(exc).__name__}: {message}" if message else type(exc).__name__ + cause = exc.cause if isinstance(exc, JobLeadJevRequestError) else exc + if isinstance(cause, requests.Timeout | TimeoutError): + return "provider_timeout" + if isinstance(cause, requests.ConnectionError): + return "provider_connection_error" + if isinstance(cause, requests.RequestException): + return "provider_transport_error" + if isinstance(cause, ValueError): + return "invalid_provider_response" + if isinstance(cause, RuntimeError): + return "provider_request_failed" + return "provider_error" + + +def _jev_shadow_request_attempts(exc: Exception) -> int: + if isinstance(exc, JobLeadJevRequestError): + return max(1, exc.request_attempts) + return 1 def _classifier_model(settings: SharedSettings) -> str: diff --git a/tests/unit/test_job_lead_sources.py b/tests/unit/test_job_lead_sources.py index c1822a9e..3fe42a22 100644 --- a/tests/unit/test_job_lead_sources.py +++ b/tests/unit/test_job_lead_sources.py @@ -6,7 +6,7 @@ from types import SimpleNamespace import five08.job_lead_sources as job_lead_sources -from five08.job_lead_jev import JobLeadJevDecision +from five08.job_lead_jev import JobLeadJevDecision, JobLeadJevRequestError from five08.job_lead_sources import ( HackerNewsThread, HackerNewsWhoIsHiringLeadSource, @@ -822,7 +822,10 @@ def test_jev_shadow_failure_never_changes_primary(monkeypatch) -> None: def fail_shadow(**_kwargs: object) -> JobLeadJevDecision: nonlocal shadow_calls shadow_calls += 1 - raise TimeoutError("provider timed out") + raise JobLeadJevRequestError( + TimeoutError("provider echoed sensitive submitted text"), + request_attempts=2, + ) monkeypatch.setattr( job_lead_sources, @@ -835,7 +838,27 @@ def fail_shadow(**_kwargs: object) -> JobLeadJevDecision: assert classification.is_contractor_friendly is True assert classification.jev_shadow is not None assert classification.jev_shadow.status == "failed" - assert classification.jev_shadow.error == "TimeoutError: provider timed out" + assert classification.jev_shadow.request_attempts == 2 + assert classification.jev_shadow.error == "provider_timeout" + metadata = job_lead_sources._classification_metadata(classification) # noqa: SLF001 + assert "sensitive submitted text" not in str(metadata) + + lead = JobLeadInput( + source_key="hackernews_who_is_hiring", + source_type="hackernews", + external_id="failed-shadow", + source_url="https://news.ycombinator.com/item?id=failed-shadow", + title="Failed shadow", + body_raw="not retained in report", + body_normalized="not retained in report", + metadata=metadata, + ) + report = job_lead_sources._job_lead_jev_shadow_report( # noqa: SLF001 + settings, # type: ignore[arg-type] + [lead], + ) + assert report["review_items"][0]["shadow"]["error"] == "provider_timeout" + assert "sensitive submitted text" not in str(report) next_classification = classifier.classify("Beta | Contract API engineer | Remote")