diff --git a/.env.example b/.env.example index 64bc9459..da45ad38 100644 --- a/.env.example +++ b/.env.example @@ -222,6 +222,18 @@ 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. 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 f3e304be..8f0228e9 100644 --- a/apps/worker/src/five08/worker/config.py +++ b/apps/worker/src/five08/worker/config.py @@ -78,6 +78,21 @@ 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) + 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 76d9a6f0..d37f6509 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -233,6 +233,26 @@ 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` +- `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 +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. 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 dc0b8a9e..7a5c4ed3 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,34 @@ 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, + cached_input_tokens=decision.cached_input_tokens, + output_tokens=decision.output_tokens, + total_tokens=decision.total_tokens, + cost_usd=decision.cost_usd, ) @@ -1010,126 +965,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 +1042,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..52247e51 --- /dev/null +++ b/packages/shared/src/five08/job_lead_jev.py @@ -0,0 +1,319 @@ +"""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 + cached_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"], + cached_input_tokens=usage["cached_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"))) + 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")) + ) + total_tokens = _integer(source.get("total_tokens")) or input_tokens + output_tokens + 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), + } + + +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..4fad51c9 100644 --- a/packages/shared/src/five08/job_lead_sources.py +++ b/packages/shared/src/five08/job_lead_sources.py @@ -2,20 +2,30 @@ from __future__ import annotations +import hashlib import html import json import logging import re -from dataclasses import dataclass +import time +from collections.abc import Callable +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, + JobLeadJevRequestError, + classify_job_lead_with_jev, +) from five08.job_leads import ( JobLeadInput, existing_job_lead_external_ids, @@ -40,6 +50,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 +164,65 @@ 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 + cached_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, + "cached_input_tokens": self.cached_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 +236,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 +654,187 @@ 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) + 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._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 + 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 jev_shadow_run_summary(self) -> dict[str, Any]: + """Return the captured configuration and bounded-work state for this run.""" + + 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 { + "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, + "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, + 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 + + 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=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 + error_category = _safe_jev_shadow_error(exc) + 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((self._jev_shadow_clock() - started) * 1000), + ), + observed_at=datetime.now(timezone.utc), + 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", + error_category, + ) + return replace(classification, jev_shadow=observation) @staticmethod def _messages(comment_text: str) -> list[dict[str, str]]: @@ -671,6 +912,105 @@ 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 _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 + 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, + cached_input_tokens=decision.cached_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: + 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: return ( _clean(getattr(settings, "job_lead_classifier_model", None)) @@ -791,7 +1131,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 +1143,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 +1330,251 @@ def build_job_lead_source( raise ValueError(f"Unsupported job lead source: {source}") +def _job_lead_jev_shadow_report( + settings: SharedSettings, + leads: list[JobLeadInput], + *, + runtime_summary: dict[str, Any] | None = None, +) -> dict[str, Any]: + 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( + 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( + 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( + 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( + 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, + ) + 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) + ] + 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: + 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" + elif budget_exhaustion_reason is not None: + status = "completed_budget_limited" + 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, + "limits": { + "request_timeout_seconds": request_timeout_seconds, + "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), + "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), + "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 + ), + "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 +1592,31 @@ 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_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 {} @@ -1048,6 +1660,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..7f4bc56c 100644 --- a/packages/shared/src/five08/runtime_config.py +++ b/packages/shared/src/five08/runtime_config.py @@ -497,6 +497,84 @@ 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="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 4184e859..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 @@ -200,7 +202,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..3fe42a22 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, JobLeadJevRequestError from five08.job_lead_sources import ( HackerNewsThread, HackerNewsWhoIsHiringLeadSource, @@ -105,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 @@ -632,6 +658,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 +723,303 @@ 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 = _jev_decision() + 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, + "cached": 50, + "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 JobLeadJevRequestError( + TimeoutError("provider echoed sensitive submitted text"), + request_attempts=2, + ) + + 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.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") + + assert shadow_calls == 1 + 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_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: diff --git a/tests/unit/test_runtime_config.py b/tests/unit/test_runtime_config.py index 08f59c81..0dc08f89 100644 --- a/tests/unit/test_runtime_config.py +++ b/tests/unit/test_runtime_config.py @@ -178,6 +178,40 @@ 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" + ) + 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 + 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 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( monkeypatch: pytest.MonkeyPatch, ) -> None: