Add Jev job-lead shadow observations

.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

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"

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.

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,14 +18,20 @@
 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,
     _classification_from_llm_response,
     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]}"