diff --git a/CLAUDE.md b/CLAUDE.md index f0b33e8..8219c13 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -133,6 +133,10 @@ automatically runs Jev source relevance, configurable bounded scene grouping, an synthesis, and a separate answer-support Noul. `--no-refine` preserves legacy RAG for one request. Refined responses retain answer/citations/confidence and add route, evidence, refinement, warning, inspected-window, and stage-usage metadata. +With `AV_DJEV_ENDPOINT` set, the same refinement runs against a self-hosted +[djev-spark](https://github.com/mmastrac/djev-spark) server (same +`/v1/systemone` contract); the served model and engine are recorded in usage +receipts, and a djev answer is never presented as a Jev measurement. ### `av list` / `av info ` / `av transcript ` / `av export` / `av open ` See `av --help` for details. @@ -201,6 +205,12 @@ When a capability is unavailable (e.g. Anthropic has no Whisper), the pipeline s | `AV_TYPESAFE_API_KEY` | (none) | Jev/System One key; enables ask refinement by default | | `AV_TYPESAFE_ENDPOINT` | `https://api.typesafe.ai/v1/systemone` | Explicit System One endpoint | | `AV_TYPESAFE_MODEL` | `jev-latest` | System One model | +| `AV_DJEV_ENDPOINT` | (none) | Self-hosted djev-spark `/v1/systemone` endpoint; selects the self-hosted decision lane over hosted Jev | +| `AV_DJEV_API_KEY` | (none) | Bearer key for the djev-spark server (needed only when the server sets `API_KEY`) | +| `AV_DJEV_MODEL` | (none) | Advisory request model; djev-spark ignores it and reports what it served | +| `AV_DJEV_TIMEOUT_SEC` | `180` | Per-attempt djev-spark timeout (cold reads on long states are slow) | +| `AV_DJEV_MAX_RETRIES` | `1` | Explicit retry count for djev-spark requests | +| `AV_DJEV_SEED` | `42` | Sampler seed sent with every djev-spark decision | | `AV_REFINE_RELEVANCE_MIN` | `0.5` | Minimum source-relevance Noul probability | | `AV_REFINE_SUPPORT_MIN` | `0.5` | Minimum answer-support Noul probability | | `AV_REFINE_MAX_SCENES` | `8` | Maximum merged scenes sent to synthesis | diff --git a/cookbook/README.md b/cookbook/README.md index b48de07..d7004e7 100644 --- a/cookbook/README.md +++ b/cookbook/README.md @@ -6,6 +6,7 @@ Runnable recipes for the open-source **av** CLI live here alongside the code. |---|---|---| | [Cost model](cost-model/README.md) | Separate tokens, estimates, unknown costs, ingestion, queries, failures, reservations, and cap accounting | Completed one-question comparison; no aggregate parity claim | | [Jev-refined ask](jev-refined-ask/README.md) | Build a source-bound transcript sidecar, retrieve indexed moments, refine evidence, answer, and check support | Runnable recipe; no speed, cost, or quality parity claim | +| [djev-spark refined ask](djev-clipping/README.md) | Run the same refined-ask workflow against a self-hosted djev-spark server, with served-model receipts and visible response validation | Adapter + offline compatibility tests complete; no live quality, latency, or parity claim | | [Sanitized receipts](receipts/README.md) | Completed ASR/baseline, caption smoke/abort, incompatible cap probes, Grok ingestion, and Jev-refined query | No media, transcript/caption corpus, credentials, upload URIs, or private routes; baseline question and returned answer/rationale retained | The [original public cost notebook](https://github.com/PixelML/cookbook/tree/main/agentic-video/cost-model) diff --git a/cookbook/djev-clipping/README.md b/cookbook/djev-clipping/README.md new file mode 100644 index 0000000..0ab9eaf --- /dev/null +++ b/cookbook/djev-clipping/README.md @@ -0,0 +1,121 @@ +# djev-spark refined ask — self-hosted decisions over the same contract + +This recipe runs the [Jev-refined ask](../jev-refined-ask/README.md) workflow +against a **self-hosted [djev-spark](https://github.com/mmastrac/djev-spark)** +server instead of the hosted TypeSafe System One endpoint. Both speak the same +documented `POST /v1/systemone` contract, so `av ask` refinement, evidence +grouping, and answer-support checking work unchanged — only the decision +provider changes. + +**Evidence status:** adapter and offline compatibility tests are complete; no +live djev quality or latency measurement has been made in this repository. +Everything below about runtime performance is upstream's own documentation and +is labelled as such. + +## What the adapter guarantees + +Audited upstream revision: `mmastrac/djev-spark` +`1444f3e927f83ba508e5b28a4fd4fdd9ecd0976b`. + +- **Identity honesty.** djev-spark ignores a request's `model` field and + reports the model it actually served. `DjevClient` records + `served_model`, `server_engine`, and the redacted `served_endpoint_host` + into every stage-usage receipt and into the `refinement` metadata of + `av ask` output. A djev answer is never presented as a Jev measurement. +- **Visible rejection.** Responses that are malformed or incomplete for the + questions actually asked — missing answers, answers to unasked questions, + type mismatches, choices outside the offered options, score legends that do + not match the levels, probabilities out of range or not summing to one — + fail with a `djev-spark response failed validation` error instead of being + silently coerced. A question the server skipped (`ask_if`) is preserved as + `null`. +- **Same failure semantics as the hosted lane.** Timeouts, explicit retries + (429/5xx and connection errors, exponential backoff), sanitized error + messages, and provider-accurate fallback warnings ("djev-spark refinement + was unavailable…") match `SystemOneClient` conventions. +- **Reproducibility knob.** Every request carries `AV_DJEV_SEED` + (default `42`), the server-side default seed. + +## Configuration + +No endpoint ships with av. djev-spark is software you host; an explicit +endpoint selects this lane over hosted Jev (the endpoint wins even if a +TypeSafe key is also set). + +| Variable | Example | Purpose | +|---|---|---| +| `AV_DJEV_ENDPOINT` | `http://10.1.2.3:8011/v1/systemone` | Structured server endpoint (compose default port `8011`) | +| `AV_DJEV_API_KEY` | any string | Sent as `Authorization: Bearer …`; needed only when the server sets `API_KEY` | +| `AV_DJEV_MODEL` | `dgemma` | Advisory only; the server ignores it. Omit unless you want it recorded in request logs | +| `AV_DJEV_TIMEOUT_SEC` | `180` | Cold structured reads on long states are slow (see below) | +| `AV_DJEV_MAX_RETRIES` | `1` | Retry count, mirroring the hosted lane's default | +| `AV_DJEV_SEED` | `42` | Sampler seed sent with every decision | + +Verify wiring offline: + +```bash +av config show | jq '.djev_endpoint, .djev_api_key' +``` + +Then ask exactly as in the Jev recipe: + +```bash +av ask "When does the door open?" --video-id +``` + +The response's `ask_settings.decision_provider` reads `"djev-spark"`, and +`refinement.served_model` / `refinement.server_engine` carry the identity the +server reported. If the server is unreachable, `route` becomes +`refinement_fallback`, the warning names the provider, and the answer is +produced from raw retrieval with `evidence_status: raw_unjudged`. + +## Running the server safely (read before you start it) + +These are requirements this project imposes on itself; the upstream defaults +are more permissive. + +- **Networking.** Upstream's compose file uses host networking and both + servers listen on all interfaces. Put the box behind a firewall or tailnet + and reach it over a private address. +- **Authentication.** Upstream serves POST routes with no API key unless its + `API_KEY` env var is set. Set one, and set `AV_DJEV_API_KEY` to match. + `/health` (GET) stays open by upstream design. +- **Playground.** Upstream's test page (`TEST_PAGE=1`) is off by default; + leave it off on any shared host. +- **Resources.** The model is DiffusionGemma 26B-A4B NVFP4 and wants a + compatible GPU, verified non-boot storage for weights and build caches, and + the upstream entrypoint's own headroom check. Never place weights, Docker + caches, or media on the av control plane's boot disk. Do not start this + next to workloads you do not own; do not evict anything to make room. +- **Model terms.** Gemma weight licence terms apply to the checkpoint; the + server source files carry Apache-2.0 headers. av implements the wire + protocol independently and distributes no upstream code or weights. + +## Performance claims (upstream-documented, untested here) + +Upstream's README reports roughly 0.1-second warm reads and, on its 128k +profile, a 104.94-second cold versus 0.44-second warm selection at a +110,707-token state. These are the author's numbers on their hardware — not +measurements from this repository, and not a guarantee that any full video +clips in two seconds. Measure your own cold/warm/export timings before +relying on any latency figure; record p50/p95 with sample counts when you do. + +## Comparison protocol (pending) + +A fair comparison against hosted Jev and the plain retrieval baseline must use +the same frozen corpus, queries, candidate windows, and transcript/vision text +as the sibling Jev clipping task, with independent labels — not provider +scores as ground truth. Measure relevance, recall, absent-topic false +positives, timing boundaries, duplication, and export validity separately per +lane. That evaluation is **not started here**: it is gated on the shared +fixture contract and on an authorized runtime. This recipe will gain a +results section only from actual recorded runs. + +## Known limitations + +- Image input (`data:` URLs / multipart parts) is supported by the upstream + server but not sent by this adapter; the primary comparison is text-to-text. +- Clip candidate construction and export belong to the shared clipping + contract, not to this provider adapter. +- Offered-label probabilities are diagnostics; nothing here treats them as + calibrated quality scores. diff --git a/src/av/cli/config_cmd.py b/src/av/cli/config_cmd.py index a77ca48..f80c162 100644 --- a/src/av/cli/config_cmd.py +++ b/src/av/cli/config_cmd.py @@ -73,6 +73,10 @@ def config_show() -> None: "typesafe_api_key": "***" if config.typesafe_api_key else "(not set)", "typesafe_endpoint": config.typesafe_endpoint, "typesafe_model": config.typesafe_model, + "djev_endpoint": config.djev_endpoint or "(not set)", + "djev_api_key": "***" if config.djev_api_key else "(not set)", + "djev_model": config.djev_model or "(not set)", + "djev_seed": config.djev_seed, "refine_enabled": config.refine_enabled, "refine_relevance_min": config.refine_relevance_min, "refine_support_min": config.refine_support_min, diff --git a/src/av/core/config.py b/src/av/core/config.py index 11f4895..f4fecf2 100644 --- a/src/av/core/config.py +++ b/src/av/core/config.py @@ -77,6 +77,19 @@ class AVConfig(BaseSettings): typesafe_model: str = Field(default="jev-latest") typesafe_timeout_sec: float = Field(default=30.0, gt=0) typesafe_max_retries: int = Field(default=1, ge=0, le=3) + # Optional self-hosted djev-spark decision endpoint speaking the same + # documented /v1/systemone contract. No default endpoint ships with av: + # an explicit endpoint selects this lane over hosted System One. + djev_endpoint: str = Field(default="") + djev_api_key: str = Field(default="") + # Advisory request model. djev-spark ignores it and reports the model it + # actually served; responses carry that identity into usage records. + djev_model: str = Field(default="") + # Cold structured reads are slow on long states (upstream documents ~105 s + # at 110k tokens), so the default is more generous than the hosted lane's. + djev_timeout_sec: float = Field(default=180.0, gt=0) + djev_max_retries: int = Field(default=1, ge=0, le=3) + djev_seed: int = Field(default=42, ge=0) refine_enabled: bool = Field(default=True) refine_relevance_min: float = Field(default=0.5, ge=0, le=1) refine_support_min: float = Field(default=0.5, ge=0, le=1) @@ -130,6 +143,12 @@ def get_config(db_path: Path | None = None) -> AVConfig: "typesafe_endpoint", "typesafe_model", "typesafe_timeout_sec", + "djev_endpoint", + "djev_api_key", + "djev_model", + "djev_timeout_sec", + "djev_max_retries", + "djev_seed", "typesafe_max_retries", "refine_enabled", "refine_relevance_min", diff --git a/src/av/search/djev.py b/src/av/search/djev.py new file mode 100644 index 0000000..3831a2e --- /dev/null +++ b/src/av/search/djev.py @@ -0,0 +1,293 @@ +"""djev-spark decision client: Jev-compatible structured reads on self-hosted +DiffusionGemma, speaking the same documented ``/v1/systemone`` contract as +:class:`~av.search.refine.SystemOneClient`. + +Upstream audited at ``mmastrac/djev-spark`` commit +``1444f3e927f83ba508e5b28a4fd4fdd9ecd0976b``. Properties of that server that +shape this adapter: + +* A request's ``model`` field is ignored. Every 200 response carries the model + the server actually served plus the engine in ``diagnostics`` — this client + requires both and records them into usage receipts, so a djev answer is + never presented as a Jev measurement and never carries an unknown runtime. +* An exported ``score`` is the zero-based expected level + ``sum(i * p)`` over legend indices, in ``[0, len(levels) - 1]`` + (``decide_group`` sums one-based internally; ``jev_answer`` re-exports it + zero-based from the same probabilities). +* A question whose ``ask_if`` condition failed is answered ``null``; that is + the only way an asked question can be null. Mandatory decisions may not be + skipped. +* The server may skip a question whose ``ask_if`` condition failed and report + it as ``null``. Skipped answers are preserved; consumers reject them where a + decision is mandatory. + +No default endpoint ships with av: djev-spark is self-hosted, its upstream +deployment binds all interfaces by default, and pointing at a machine by +default would couple the public package to somebody's private box. +""" + +from __future__ import annotations + +import math +import time +from typing import Any + +import requests + +from av.bench.receipts import redact_endpoint +from av.core.config import AVConfig +from av.search.refine import _RETRYABLE_STATUS, RefinementError, SystemOneClient + +# Sampled probabilities are means of per-read label distributions, so they sum +# to one only up to sampling and float noise; anything further off is a +# malformed response, not a calibration quirk. +_PROB_SUM_TOLERANCE = 0.05 +_SCORE_RANGE_SLACK = 0.01 +_SCORE_CONSISTENCY_TOLERANCE = 0.01 + + +def _finite_unit_interval(value: Any) -> bool: + return ( + not isinstance(value, bool) + and isinstance(value, (int, float)) + and math.isfinite(float(value)) + and 0 <= float(value) <= 1 + ) + + +class DjevClient: + """Same ``ask`` protocol as SystemOneClient against a djev-spark server.""" + + provider_label = "djev-spark" + + def __init__(self, config: AVConfig, *, session: requests.Session | None = None) -> None: + if not config.djev_endpoint: + raise RefinementError( + "djev-spark endpoint is not configured; set AV_DJEV_ENDPOINT " + "(self-hosted only — av ships no default endpoint)" + ) + self.endpoint = config.djev_endpoint + self.api_key = config.djev_api_key + self.model = config.djev_model # advisory: the server ignores it + self.seed = config.djev_seed + self.timeout = config.djev_timeout_sec + self.max_retries = config.djev_max_retries + self.session = session or requests.Session() + self.served_model: str | None = None + self.server_engine: str | None = None + self.served_endpoint_host: str | None = redact_endpoint(self.endpoint) + + def ask(self, state: Any, questions: dict[str, dict]) -> tuple[dict[str, dict], dict]: + payload: dict[str, Any] = {"state": state, "questions": questions, "seed": self.seed} + if self.model: + payload["model"] = self.model + headers = {"Content-Type": "application/json"} + if self.api_key: + headers["Authorization"] = f"Bearer {self.api_key}" + last_error = "request failed" + attempts = 0 + for attempt in range(self.max_retries + 1): + attempts += 1 + try: + response = self.session.post( + self.endpoint, + headers=headers, + json=payload, + timeout=self.timeout, + ) + except requests.RequestException as exc: + last_error = type(exc).__name__ + if attempt < self.max_retries: + time.sleep(min(0.25 * (2**attempt), 1.0)) + continue + raise RefinementError( + f"djev-spark unavailable ({last_error})", + attempts=attempts, + ) from exc + if response.status_code in _RETRYABLE_STATUS and attempt < self.max_retries: + time.sleep(min(0.25 * (2**attempt), 1.0)) + continue + if not response.ok: + raise RefinementError( + f"djev-spark request failed with HTTP {response.status_code}", + attempts=attempts, + ) + try: + data = response.json() + except ValueError as exc: + raise RefinementError( + "djev-spark returned invalid JSON", + attempts=attempts, + ) from exc + if not isinstance(data, dict): + raise RefinementError( + "djev-spark returned an invalid JSON document", + attempts=attempts, + ) + raw_usage = data.get("usage") if isinstance(data.get("usage"), dict) else {} + answers = data.get("answers") + if not isinstance(answers, dict): + raise RefinementError( + "djev-spark response is missing answers", + attempts=attempts, + raw_usage=raw_usage, + ) + self._capture_identity(data) + if self._missing_identity(): + raise RefinementError( + "djev-spark response did not identify the served runtime: " + "the pinned contract requires the served model and engine " + "on every success", + attempts=attempts, + raw_usage=raw_usage, + ) + self._validate_answers(questions, answers, attempts=attempts, raw_usage=raw_usage) + usage = { + **raw_usage, + "_attempts": attempts, + "served_model": self.served_model, + "server_engine": self.server_engine, + "served_endpoint_host": self.served_endpoint_host, + } + return answers, usage + raise RefinementError(f"djev-spark unavailable ({last_error})", attempts=attempts) + + def _missing_identity(self) -> bool: + return not self.served_model or not self.server_engine + + def _capture_identity(self, data: dict) -> None: + """Record the identity the server reported for itself, never the + request's. Reset per response: an identity from an earlier call must + not vouch for this one.""" + model = data.get("model") + self.served_model = model if isinstance(model, str) and model else None + diagnostics = data.get("diagnostics") + engine = diagnostics.get("engine") if isinstance(diagnostics, dict) else None + self.server_engine = engine if isinstance(engine, str) and engine else None + + def _validate_answers( + self, + questions: dict[str, dict], + answers: dict[str, Any], + *, + attempts: int, + raw_usage: dict, + ) -> None: + """Visibly reject responses that are malformed or incomplete for the + questions actually asked. ``null`` is accepted only for questions that + declare an ``ask_if`` condition — the only skip path in the pinned + contract; a mandatory decision reported as ``null`` is rejected.""" + problems: list[str] = [] + for key in answers: + if key not in questions: + problems.append(f"{key}: answer for a question that was not asked") + for qid, question in questions.items(): + if qid not in answers: + problems.append(f"{qid}: missing answer") + continue + answer = answers[qid] + if answer is None: + if not question.get("ask_if"): + problems.append(f"{qid}: mandatory question was skipped") + continue + kind = question.get("type") + if not isinstance(answer, dict) or answer.get("type") != kind: + problems.append(f"{qid}: expected a {kind} answer") + continue + if "confidence" in answer and not _finite_unit_interval(answer["confidence"]): + problems.append(f"{qid}: confidence out of range") + if kind == "noul": + if not _finite_unit_interval(answer.get("noul")): + problems.append(f"{qid}: noul probability out of range") + elif kind == "choice": + self._validate_choice(qid, question, answer, problems) + elif kind == "score": + self._validate_score(qid, question, answer, problems) + if problems: + raise RefinementError( + "djev-spark response failed validation: " + "; ".join(problems), + attempts=attempts, + raw_usage=raw_usage, + ) + + def _validate_choice( + self, qid: str, question: dict, answer: dict, problems: list[str] + ) -> None: + criteria = question.get("criteria") + names = set(criteria) if isinstance(criteria, dict) else set() + # A set membership test raises TypeError on an unhashable value (a + # JSON array or object), so require a string before comparing. + choice = answer.get("choice") + if not isinstance(choice, str) or choice not in names: + problems.append(f"{qid}: choice outside the offered options") + probabilities = answer.get("probabilities") + if not isinstance(probabilities, dict) or set(probabilities) != names: + problems.append(f"{qid}: choice probabilities do not match the offered options") + return + self._validate_distribution(qid, probabilities, problems) + + def _validate_score( + self, qid: str, question: dict, answer: dict, problems: list[str] + ) -> None: + levels = question.get("criteria") + if not isinstance(levels, list) or not levels: + problems.append(f"{qid}: score question without levels") + return + expected_legend = {str(index): level for index, level in enumerate(levels)} + if answer.get("legend") != expected_legend: + problems.append(f"{qid}: score legend does not match the offered levels") + probabilities = answer.get("probabilities") + if not isinstance(probabilities, dict) or set(probabilities) != set(expected_legend): + problems.append(f"{qid}: score probabilities do not match the legend") + return + self._validate_distribution(qid, probabilities, problems) + score = answer.get("score") + # The contract exports the zero-based expected level sum(i * p) over + # legend indices, so a three-level question scores in [0, 2]. + highest = float(len(levels) - 1) + if ( + isinstance(score, bool) + or not isinstance(score, (int, float)) + or not math.isfinite(float(score)) + or not -_SCORE_RANGE_SLACK <= float(score) <= highest + _SCORE_RANGE_SLACK + ): + problems.append(f"{qid}: score outside the legend range") + return + if all(_finite_unit_interval(p) for p in probabilities.values()): + expected = sum(int(key) * float(p) for key, p in probabilities.items()) + if abs(float(score) - expected) > _SCORE_CONSISTENCY_TOLERANCE: + problems.append(f"{qid}: score inconsistent with the legend probabilities") + + def _validate_distribution(self, qid: str, probabilities: dict, problems: list[str]) -> None: + total = 0.0 + for label, probability in probabilities.items(): + if not _finite_unit_interval(probability): + problems.append(f"{qid}: probability for {label!r} out of range") + return + total += float(probability) + if abs(total - 1.0) > _PROB_SUM_TOLERANCE: + problems.append(f"{qid}: probabilities sum to {total:.3f}") + + +def decision_provider_name(config: AVConfig) -> str | None: + """The decision provider this configuration selects, or ``None`` when no + decision lane is configured and refinement must fall back to retrieval.""" + if config.djev_endpoint: + return "djev-spark" + if config.typesafe_api_key: + return "jev" + return None + + +def open_decision_client( + config: AVConfig, *, session: requests.Session | None = None +) -> SystemOneClient | DjevClient: + """Construct the configured decision client. + + An explicitly configured djev endpoint selects the self-hosted lane over + hosted System One; both speak the same ``/v1/systemone`` contract through + the same ``ask`` protocol. + """ + if config.djev_endpoint: + return DjevClient(config, session=session) + return SystemOneClient(config, session=session) diff --git a/src/av/search/rag.py b/src/av/search/rag.py index 0f712c2..e468f29 100644 --- a/src/av/search/rag.py +++ b/src/av/search/rag.py @@ -5,10 +5,10 @@ from av.core.config import AVConfig from av.db.repository import Repository, _fmt_timestamp from av.providers.openai import OpenAILLM +from av.search.djev import decision_provider_name, open_decision_client from av.search.inspection import inspect_with_stronger_vision from av.search.refine import ( RefinementError, - SystemOneClient, judge_answer_support, refine_search_results, ) @@ -65,9 +65,17 @@ def _ask_settings(config: AVConfig) -> dict: return { "chat_model": config.chat_model, "chat_max_output_tokens": config.chat_max_output_tokens, + "decision_provider": decision_provider_name(config), } +def _confidence_basis(decision_provider: str | None, stem: str) -> str: + """Name the lane that actually produced a confidence value: a djev-spark + decision is never reported under a Jev basis label.""" + lane = "djev_spark" if decision_provider == "djev-spark" else "jev" + return f"{lane}_{stem}" + + def _llm_usage_snapshot(llm: OpenAILLM | None) -> dict: usage = getattr(llm, "usage", None) snapshot = getattr(usage, "snapshot", None) @@ -167,7 +175,8 @@ def ask( ) raw_results = search_result.get("results", []) - if not refine or not config.refine_enabled or not config.typesafe_api_key: + decision_provider = decision_provider_name(config) + if not refine or not config.refine_enabled or decision_provider is None: return _legacy_ask( question, raw_results, @@ -192,23 +201,34 @@ def ask( "confidence_basis": "no_evidence", "route": "refined_no_results", "evidence_status": "no_retrieval_hits", - "refinement": {"status": "no_retrieval_hits", "raw_count": 0, "scene_count": 0}, + "refinement": {"status": "no_retrieval_hits", "raw_count": 0, "scene_count": 0, "decision_provider": decision_provider}, "warnings": warnings, "inspected_windows": [], "stage_usage": stage_usage, "ask_settings": _ask_settings(config), } - client = SystemOneClient(config) + client = open_decision_client(config) try: results, refinement, refinement_usage = refine_search_results( question, raw_results, repo, config, client=client ) stage_usage.update(refinement_usage) + refinement["decision_provider"] = decision_provider + served_model = getattr(client, "served_model", None) + if served_model: + refinement["served_model"] = served_model + server_engine = getattr(client, "server_engine", None) + if server_engine: + refinement["server_engine"] = server_engine except RefinementError as exc: for stage, usage in exc.stage_usage.items(): stage_usage[stage] = usage - warnings.append("Jev refinement was unavailable; answering from raw retrieval without judged evidence confidence.") + decision_label = {"jev": "Jev"}.get(decision_provider, decision_provider) + warnings.append( + f"{decision_label} refinement was unavailable; answering from raw retrieval " + "without judged evidence confidence." + ) llm: OpenAILLM | None = None try: llm = OpenAILLM(config) @@ -224,7 +244,7 @@ def ask( "confidence_basis": "unknown", "route": "refinement_fallback_answer_failed", "evidence_status": "answer_unavailable", - "refinement": {"status": "provider_fallback", "raw_count": len(raw_results)}, + "refinement": {"status": "provider_fallback", "raw_count": len(raw_results), "decision_provider": decision_provider}, "warnings": warnings, "inspected_windows": [], "stage_usage": stage_usage, @@ -237,7 +257,7 @@ def ask( "confidence_basis": "retrieval_heuristic", "route": "refinement_fallback", "evidence_status": "raw_unjudged", - "refinement": {"status": "provider_fallback", "raw_count": len(raw_results)}, + "refinement": {"status": "provider_fallback", "raw_count": len(raw_results), "decision_provider": decision_provider}, "warnings": warnings, "inspected_windows": [], "stage_usage": stage_usage, @@ -249,7 +269,7 @@ def ask( "answer": "No supported evidence was found for this question in the retrieved video moments.", "citations": [], "confidence": 0.0, - "confidence_basis": "jev_relevance", + "confidence_basis": _confidence_basis(decision_provider, "relevance"), "route": "refined_no_results", "evidence_status": "all_sources_irrelevant", "refinement": refinement, @@ -298,7 +318,7 @@ def ask( "answer": answer, "citations": citations, "confidence": support_probability, - "confidence_basis": "jev_answer_support", + "confidence_basis": _confidence_basis(decision_provider, "answer_support"), "route": "refined", "evidence_status": "supported", "refinement": refinement, @@ -341,7 +361,7 @@ def ask( "answer": inspected_answer, "citations": inspected_citations, "confidence": inspected_support, - "confidence_basis": "jev_answer_support_after_sampled_frames", + "confidence_basis": _confidence_basis(decision_provider, "answer_support_after_sampled_frames"), "route": "vision_inspected", "evidence_status": "sampled_frames_supported", "refinement": refinement, @@ -361,7 +381,7 @@ def ask( "answer": uncertain, "citations": citations, "confidence": support_probability or 0.0, - "confidence_basis": "jev_answer_support" if support_probability is not None else "unknown", + "confidence_basis": _confidence_basis(decision_provider, "answer_support") if support_probability is not None else "unknown", "route": "refined_uncertain", "evidence_status": evidence_status, "refinement": refinement, diff --git a/src/av/search/refine.py b/src/av/search/refine.py index 7d062ff..3fac831 100644 --- a/src/av/search/refine.py +++ b/src/av/search/refine.py @@ -8,7 +8,6 @@ from __future__ import annotations -import json import math import time from dataclasses import dataclass, field @@ -66,19 +65,26 @@ def _record_call_usage(usage: dict, call_usage: dict[str, Any] | None) -> None: requests=attempts, ambiguous_attempts=attempts > 1, ) + # Server-reported identity travels with the receipt so a reader can tell + # which runtime actually answered (djev-spark reports the model it served). + for key in ("served_model", "server_engine", "served_endpoint_host"): + value = cleaned.get(key) if isinstance(cleaned, dict) else None + if isinstance(value, str) and value: + usage[key] = value -def _probability(value: Any, name: str) -> float: +def _probability(value: Any, name: str, label: str = "System One") -> float: if isinstance(value, bool) or not isinstance(value, (int, float)): - raise RefinementError(f"System One returned an invalid probability for {name}") + raise RefinementError(f"{label} returned an invalid probability for {name}") value = float(value) if not math.isfinite(value) or not 0 <= value <= 1: - raise RefinementError(f"System One returned an out-of-range probability for {name}") + raise RefinementError(f"{label} returned an out-of-range probability for {name}") return value class SystemOneClient: """Small synchronous client for the documented TypeSafe System One endpoint.""" + provider_label = "Jev/System One" def __init__(self, config: AVConfig, *, session: requests.Session | None = None) -> None: if not config.typesafe_api_key: @@ -341,13 +347,15 @@ def _choice_criteria(labels: list[str], candidates: dict[str, dict]) -> dict: return out -def _read_choice(answer: Any, name: str, allowed: list[str]) -> tuple[str, float]: +def _read_choice( + answer: Any, name: str, allowed: list[str], label: str = "System One" +) -> tuple[str, float]: if not isinstance(answer, dict) or answer.get("type") != "choice": - raise RefinementError(f"System One returned an invalid Choice answer for {name}") + raise RefinementError(f"{label} returned an invalid Choice answer for {name}") choice = answer.get("choice") if choice not in allowed: - raise RefinementError(f"System One returned an invalid boundary choice for {name}") - confidence = _probability(answer.get("confidence"), f"{name}.confidence") + raise RefinementError(f"{label} returned an invalid boundary choice for {name}") + confidence = _probability(answer.get("confidence"), f"{name}.confidence", label) return str(choice), confidence @@ -361,40 +369,66 @@ def _judge_bounds( ) -> tuple[float, float, float, str, str, dict]: data = _boundary_input(events, hit_index, query, window) candidates = data["surrounding_events"] - questions = { - "start": { - "type": "choice", - "instructions": "Select the earliest candidate that belongs to the same continuous video moment as `hit_event` for the user's `query`. Use e0 when earlier candidates do not belong to that moment.", - "criteria": _choice_criteria(data["start_labels"], candidates), - }, - "end": { + label = getattr(client, "provider_label", "System One") + + def side_question(side: str) -> dict: + direction, other = ("earliest", "earlier") if side == "start" else ("latest", "later") + return { "type": "choice", - "instructions": "Select the latest candidate that belongs to the same continuous video moment as `hit_event` for the user's `query`. Use e0 when later candidates do not belong to that moment.", - "criteria": _choice_criteria(data["end_labels"], candidates), - }, + "instructions": ( + f"Select the {direction} candidate that belongs to the same continuous video moment " + f"as `hit_event` for the user's `query`. Use e0 when {other} candidates do not belong " + "to that moment." + ), + "criteria": _choice_criteria(data[f"{side}_labels"], candidates), + } + + # A side whose only candidate is the hit itself has no alternative to + # choose. Structured providers reject one-option questions outright, so + # such sides are resolved locally (the scene simply cannot extend that + # way) and only sides with a real decision are sent to the provider. + forced = { + side: data[f"{side}_labels"][0] + for side in ("start", "end") + if len(data[f"{side}_labels"]) < 2 } - state = { - "query": query, - "hit_event": data["hit_event"], - "surrounding_events": candidates, - "note": "Candidates are ordered temporal events from one video. e0 contains the retrieved hit; negative labels are earlier and positive labels are later.", + questions = { + side: side_question(side) for side in ("start", "end") if side not in forced } - try: - answers, call_usage = client.ask(state, questions) - except RefinementError as exc: - _record_client_error(usage, exc) - exc.stage_usage["boundary"] = usage - raise - _record_call_usage(usage, call_usage) - try: - start_label, start_conf = _read_choice(answers.get("start"), "start", data["start_labels"]) - end_label, end_conf = _read_choice(answers.get("end"), "end", data["end_labels"]) - except RefinementError as exc: - exc.stage_usage["boundary"] = usage - raise - start_event = events[hit_index + int(start_label[1:])] - end_event = events[hit_index + int(end_label[1:])] - return start_event["start"], end_event["end"], min(start_conf, end_conf), start_label, end_label, data + resolved: dict[str, str] = dict(forced) + confidences: dict[str, float] = {side: 1.0 for side in forced} + if questions: + state = { + "query": query, + "hit_event": data["hit_event"], + "surrounding_events": candidates, + "note": "Candidates are ordered temporal events from one video. e0 contains the retrieved hit; negative labels are earlier and positive labels are later.", + } + try: + answers, call_usage = client.ask(state, questions) + except RefinementError as exc: + _record_client_error(usage, exc) + exc.stage_usage["boundary"] = usage + raise + _record_call_usage(usage, call_usage) + try: + for side in questions: + resolved[side], confidences[side] = _read_choice( + answers.get(side), side, data[f"{side}_labels"], label + ) + except RefinementError as exc: + exc.stage_usage["boundary"] = usage + raise + start_event = events[hit_index + int(resolved["start"][1:])] + end_event = events[hit_index + int(resolved["end"][1:])] + return ( + start_event["start"], + end_event["end"], + min(confidences.values()), + resolved["start"], + resolved["end"], + data, + ) def judge_relevance( @@ -405,6 +439,7 @@ def judge_relevance( batch_size: int = 10, ) -> tuple[dict[str, float], dict[str, int | None]]: usage = new_usage() + label = getattr(client, "provider_label", "System One") probabilities: dict[str, float] = {} for offset in range(0, len(results), batch_size): batch = results[offset : offset + batch_size] @@ -442,8 +477,8 @@ def judge_relevance( for key, artifact_id in ids.items(): answer = answers.get(key) if not isinstance(answer, dict) or answer.get("type") != "noul": - raise RefinementError(f"System One response is missing Noul answer {key}") - probabilities[artifact_id] = _probability(answer.get("noul"), key) + raise RefinementError(f"{label} response is missing Noul answer {key}") + probabilities[artifact_id] = _probability(answer.get("noul"), key, label) except RefinementError as exc: exc.stage_usage["relevance"] = usage raise @@ -577,6 +612,7 @@ def refine_search_results( client: SystemOneClient | None = None, ) -> tuple[list[dict], dict, dict[str, dict[str, int | None]]]: client = client or SystemOneClient(config) + label = getattr(client, "provider_label", "System One") probabilities, relevance_usage = judge_relevance( client, query, @@ -589,7 +625,7 @@ def refine_search_results( probability = probabilities.get(artifact_id) if probability is None: raise RefinementError( - "System One omitted a source relevance probability", + f"{label} omitted a source relevance probability", stage_usage={"relevance": relevance_usage}, ) if probability >= config.refine_relevance_min: @@ -674,6 +710,7 @@ def judge_answer_support( } } usage = new_usage() + label = getattr(client, "provider_label", "System One") try: answers, raw_usage = client.ask(state, questions) except RefinementError as exc: @@ -684,8 +721,8 @@ def judge_answer_support( try: answer_data = answers.get("is_supported") if not isinstance(answer_data, dict) or answer_data.get("type") != "noul": - raise RefinementError("System One response is missing the support Noul") - probability = _probability(answer_data.get("noul"), "is_supported") + raise RefinementError(f"{label} response is missing the support Noul") + probability = _probability(answer_data.get("noul"), "is_supported", label) except RefinementError as exc: exc.stage_usage["support"] = usage raise diff --git a/tests/test_ask_refinement.py b/tests/test_ask_refinement.py index 7c0a65b..a62709c 100644 --- a/tests/test_ask_refinement.py +++ b/tests/test_ask_refinement.py @@ -101,23 +101,28 @@ def ask(self, state, questions): self.calls.append((state, questions)) if "is_supported" in questions: return {"is_supported": {"type": "noul", "noul": self.support}}, self.usage - if "start" in questions and "end" in questions: + if "start" in questions or "end" in questions: self.boundary_calls += 1 - start_labels = list(questions["start"]["criteria"]) - end_labels = list(questions["end"]["criteria"]) - if self.edge_boundaries and self.boundary_calls == 1: - start = start_labels[0] - end = end_labels[-1] - elif self.edge_boundaries: - start = "e-5" if "e-5" in start_labels else start_labels[0] - end = "e4" if "e4" in end_labels else end_labels[-1] - else: - start = "e0" - end = "e0" - return { - "start": {"type": "choice", "choice": start, "confidence": 0.8}, - "end": {"type": "choice", "choice": end, "confidence": 0.7}, - }, self.usage + answers = {} + for side in ("start", "end"): + if side not in questions: + continue + labels = list(questions[side]["criteria"]) + if self.edge_boundaries and self.boundary_calls == 1: + choice = labels[0] if side == "start" else labels[-1] + elif self.edge_boundaries: + if side == "start": + choice = "e-5" if "e-5" in labels else labels[0] + else: + choice = "e4" if "e4" in labels else labels[-1] + else: + choice = "e0" + answers[side] = { + "type": "choice", + "choice": choice, + "confidence": 0.8 if side == "start" else 0.7, + } + return answers, self.usage answers = {} for index, key in enumerate(questions): position = self.relevance_offset + index @@ -174,7 +179,7 @@ def test_boundary_window_is_single_pass_and_configurable( # Put the hit in the middle so ±6 has real room on both sides. raw[0] = repo.search_fts("scene 5", limit=1, video_id="v1")[0].model_dump() fake = FakeSystemOne([0.9], edge_boundaries=True) - refined, meta, _ = refine_search_results( + _refined, meta, _ = refine_search_results( "event", raw, repo, @@ -187,6 +192,30 @@ def test_boundary_window_is_single_pass_and_configurable( assert len(boundary_questions["end"]["criteria"]) <= 5 assert meta["scene_count"] == 1 +def test_edge_hit_resolves_singleton_side_locally_without_asking( + repo: Repository, tmp_path: Path +) -> None: + """A hit at the first temporal event has only `e0` on its start side. A + structured provider rejects one-option questions, so that side must be + resolved locally and never sent — otherwise every edge hit would fall + back with a validation 422.""" + _seed_video(repo, tmp_path, "v1") + raw = [repo.search_fts("scene 0", limit=1, video_id="v1")[0].model_dump()] + fake = FakeSystemOne([0.9]) + refined, meta, _ = refine_search_results( + "event", + raw, + repo, + AVConfig(typesafe_api_key="test", embed_model=""), + client=fake, + ) + boundary = next(questions for _, questions in fake.calls if "start" in questions or "end" in questions) + assert set(boundary) == {"end"} + assert fake.boundary_calls == 1 + assert meta["scene_count"] == 1 + assert refined[0]["timestamp_sec"] == 0.0 + assert refined[0]["end_sec"] == 10.0 + def test_overlap_merge_is_same_video_only_and_ranks_probability_times_score() -> None: def scene(artifact_id: str, video: str, start: float, end: float, score: float, p: float) -> Scene: @@ -419,7 +448,7 @@ def test_valid_all_irrelevant_is_no_results_not_raw_fallback(repo: Repository, t _seed_video(repo, tmp_path, "v1", prefix="cake") fake = FakeSystemOne([0.1] * 30) config = AVConfig(typesafe_api_key="test", embed_model="", refine_relevance_min=0.5) - with patch("av.search.rag.SystemOneClient", return_value=fake), \ + with patch("av.search.rag.open_decision_client", return_value=fake), \ patch("av.search.rag.OpenAILLM", side_effect=AssertionError("answer model must not run")): result = ask("cake", repo, config, video_id="v1") assert result["route"] == "refined_no_results" @@ -434,7 +463,7 @@ def test_fast_path_support_skips_vision_and_usage_stays_unknown( _seed_video(repo, tmp_path, "v1", prefix="cake") fake = FakeSystemOne([0.9] * 30, support=0.88) config = AVConfig(typesafe_api_key="test", embed_model="") - with patch("av.search.rag.SystemOneClient", return_value=fake), \ + with patch("av.search.rag.open_decision_client", return_value=fake), \ patch("av.search.rag.OpenAILLM", FakeLLM), \ patch("av.search.rag.inspect_with_stronger_vision") as inspect: result = ask("cake", repo, config, video_id="v1") @@ -458,7 +487,7 @@ def test_relevant_sources_can_still_fail_answer_support_and_escalate( "usage": {"requests": 1, "input_tokens": None, "output_tokens": None}, "warnings": [], } - with patch("av.search.rag.SystemOneClient", return_value=fake), \ + with patch("av.search.rag.open_decision_client", return_value=fake), \ patch("av.search.rag.OpenAILLM", FakeLLM), \ patch("av.search.rag.inspect_with_stronger_vision", return_value=inspection) as inspect: result = ask("cake", repo, AVConfig(typesafe_api_key="test", embed_model=""), video_id="v1") @@ -473,7 +502,7 @@ def test_refinement_outage_falls_back_raw_without_secret_leak( ) -> None: _seed_video(repo, tmp_path, "v1", prefix="cake") config = AVConfig(typesafe_api_key="top-secret", embed_model="") - with patch("av.search.rag.SystemOneClient"), \ + with patch("av.search.rag.open_decision_client"), \ patch("av.search.rag.refine_search_results", side_effect=RefinementError("top-secret private input")), \ patch("av.search.rag.OpenAILLM", FakeLLM): result = ask("cake", repo, config, video_id="v1") @@ -550,7 +579,7 @@ def test_refinement_fallback_returns_partial_stage_usage(repo: Repository, tmp_p "output_tokens_complete": False, } error = RefinementError("failed", stage_usage={"relevance": partial}) - with patch("av.search.rag.SystemOneClient"), \ + with patch("av.search.rag.open_decision_client"), \ patch("av.search.rag.refine_search_results", side_effect=error), \ patch("av.search.rag.OpenAILLM", FakeLLM): result = ask( @@ -577,7 +606,7 @@ def complete_with_usage(self, prompt: str, context: str) -> CompletionResult: raise RuntimeError("https://private.example/v1 secret-token") fake = FakeSystemOne([0.9] * 30) - with patch("av.search.rag.SystemOneClient", return_value=fake), \ + with patch("av.search.rag.open_decision_client", return_value=fake), \ patch("av.search.rag.OpenAILLM", FailingLLM), \ patch("av.search.rag.judge_answer_support") as support, \ patch("av.search.rag.inspect_with_stronger_vision") as inspect: diff --git a/tests/test_djev_adapter.py b/tests/test_djev_adapter.py new file mode 100644 index 0000000..02340a1 --- /dev/null +++ b/tests/test_djev_adapter.py @@ -0,0 +1,527 @@ +"""Offline, deterministic compatibility tests for the djev-spark decision lane. + +The wire fixtures here mirror the audited ``/v1/systemone`` behaviour of +``mmastrac/djev-spark`` at commit ``1444f3e927f83ba508e5b28a4fd4fdd9ecd0976b`` +(derived from that source, not recorded from a live server). No test opens a +socket: every HTTP interaction goes through a fake ``requests`` session, and +the ask-path tests patch the client construction seam directly. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import pytest +import requests + +from av.core.config import AVConfig +from av.db.models import ArtifactRecord, VideoRecord +from av.db.repository import Repository +from av.providers.base import CompletionResult +from av.search.djev import DjevClient, decision_provider_name, open_decision_client +from av.search.rag import ask +from av.search.refine import RefinementError, judge_relevance + +ENDPOINT = "http://10.1.2.3:8011/v1/systemone" + + +def _questions() -> dict[str, dict]: + """The three-question Jev request shape from the upstream test suite.""" + return { + "urgent": { + "type": "noul", + "instructions": "Does the customer need a reply within the hour?", + "criteria": {"true": "needs a reply now", "false": "can wait"}, + }, + "bucket": { + "type": "choice", + "instructions": "Which team owns this?", + "criteria": {"billing": None, "outage": "service down", "feature": None}, + }, + "tone": { + "type": "score", + "instructions": "How angry is the customer?", + "criteria": ["calm", "annoyed", "furious"], + }, + } + + +def _answers() -> dict[str, dict]: + """The answer shapes ``jev_answer`` produces for a confident first read. + + The exported score is the zero-based expected level ``sum(i * p)``: + 0·0.7 + 1·0.2 + 2·0.1 = 0.4 (the upstream test expects 0.45 for its own + distribution on the same formula). + """ + return { + "urgent": {"type": "noul", "noul": 0.7}, + "bucket": { + "type": "choice", + "choice": "billing", + "probabilities": {"billing": 0.7, "outage": 0.2, "feature": 0.1}, + "confidence": 0.7, + }, + "tone": { + "type": "score", + "score": 0.4, + "legend": {"0": "calm", "1": "annoyed", "2": "furious"}, + "probabilities": {"0": 0.7, "1": 0.2, "2": 0.1}, + "confidence": 0.7, + }, + } + + +def _payload(**over: Any) -> dict: + body: dict[str, Any] = { + "model": "dgemma", + "answers": _answers(), + "usage": {"input_tokens": 321, "output_tokens": 12}, + "diagnostics": {"engine": "vllm", "timing": {"total_ms": 12.5, "reads": 1}}, + } + body.update(over) + return body + + +class FakeResponse: + def __init__(self, status_code: int = 200, payload: Any = None, broken_json: bool = False): + self.status_code = status_code + self.ok = status_code < 400 + self._payload = payload + self._broken_json = broken_json + + def json(self) -> Any: + if self._broken_json: + raise ValueError("not json") + return self._payload + + +class FakeSession: + def __init__(self, *responses: Any): + self.responses = list(responses) + self.calls: list[dict] = [] + + def post(self, url, headers=None, json=None, timeout=None): + self.calls.append( + {"url": url, "headers": headers, "json": json, "timeout": timeout} + ) + response = self.responses.pop(0) + if isinstance(response, Exception): + raise response + return response + + +def _client(session: FakeSession, **config_over: Any) -> DjevClient: + config = AVConfig(djev_endpoint=ENDPOINT, **config_over) + return DjevClient(config, session=session) + + +def test_endpoint_is_required_and_never_defaulted() -> None: + with pytest.raises(RefinementError, match="AV_DJEV_ENDPOINT"): + DjevClient(AVConfig()) + + +def test_request_shape_auth_is_optional_and_model_is_advisory() -> None: + session = FakeSession(FakeResponse(payload=_payload())) + client = _client(session, djev_api_key="test-key", djev_model="jev-latest") + client.ask({"ticket": "down"}, _questions()) + + call = session.calls[0] + assert call["url"] == ENDPOINT + assert call["headers"]["Authorization"] == "Bearer test-key" + assert call["json"]["seed"] == 42 + assert call["json"]["state"] == {"ticket": "down"} + assert set(call["json"]["questions"]) == {"urgent", "bucket", "tone"} + + # Without a configured key there is no Authorization header at all, and + # without an advisory model the request carries no model field. + anonymous = FakeSession( + FakeResponse(payload=_payload(answers={"a": {"type": "noul", "noul": 0.5}})) + ) + _client(anonymous).ask("state", {"a": {"type": "noul", "criteria": {}}}) + call = anonymous.calls[0] + assert "Authorization" not in call["headers"] + assert "model" not in call["json"] + + +def test_happy_path_records_served_identity_not_the_request_model() -> None: + session = FakeSession(FakeResponse(payload=_payload())) + client = _client(session, djev_model="jev-latest") + answers, usage = client.ask({}, _questions()) + + assert answers["urgent"] == {"type": "noul", "noul": 0.7} + assert answers["bucket"]["choice"] == "billing" + # The exported score is the zero-based expected level sum(i * p), in + # [0, len(levels) - 1]: 0·0.7 + 1·0.2 + 2·0.1 = 0.4. + assert answers["tone"]["score"] == pytest.approx(0.4) + assert answers["tone"]["score"] == pytest.approx( + sum(i * p for i, p in enumerate((0.7, 0.2, 0.1))) + ) + + # The request said "jev-latest"; the server ignored it and answered as + # dgemma. The receipt must carry what actually served. + assert client.served_model == "dgemma" + assert client.server_engine == "vllm" + assert usage["served_model"] == "dgemma" + assert usage["server_engine"] == "vllm" + assert usage["served_endpoint_host"] == "" + assert usage["input_tokens"] == 321 + assert usage["output_tokens"] == 12 + assert usage["_attempts"] == 1 + + +def test_skipped_questions_are_reported_as_null() -> None: + payload = _payload() + payload["answers"]["bucket"] = None # ask_if failed server-side + questions = _questions() + questions["bucket"]["ask_if"] = {"urgent": ["yes"]} + session = FakeSession(FakeResponse(payload=payload)) + answers, _ = _client(session).ask({}, questions) + assert answers["bucket"] is None + assert answers["urgent"]["noul"] == 0.7 + + +def test_null_answer_for_a_mandatory_question_is_rejected() -> None: + payload = _payload() + payload["answers"]["urgent"] = None # relevance judgments cannot be skipped + session = FakeSession(FakeResponse(payload=payload)) + with pytest.raises(RefinementError, match="mandatory question was skipped"): + _client(session).ask({}, _questions()) + + +@pytest.mark.parametrize("removed", ["model", "diagnostics"]) +def test_success_without_served_identity_is_rejected(removed: str) -> None: + payload = _payload() + if removed == "diagnostics": + payload["diagnostics"] = {} # engine missing from diagnostics + else: + del payload[removed] + session = FakeSession(FakeResponse(payload=payload)) + client = _client(session) + with pytest.raises(RefinementError, match="did not identify the served runtime"): + client.ask({}, _questions()) + # A failed identity check must not leave a default or stale value standing + # in for this response. + missing, present = ( + ("served_model", "server_engine") + if removed == "model" + else ("server_engine", "served_model") + ) + assert getattr(client, missing) is None + assert getattr(client, present) in (None, "vllm", "dgemma") + + +@pytest.mark.parametrize( + "mutate, fragment", + [ + (lambda p, q: p["answers"].pop("tone"), "missing answer"), + (lambda p, q: p["answers"].update(extra={"type": "noul", "noul": 0.5}), "not asked"), + (lambda p, q: p["answers"].update(urgent={"type": "choice", "choice": "billing"}), "expected a noul"), + ( + lambda p, q: p["answers"]["bucket"].update(choice="security"), + "outside the offered options", + ), + ( + lambda p, q: p["answers"]["bucket"].update(probabilities={"billing": 1.0}), + "do not match the offered options", + ), + ( + lambda p, q: p["answers"]["tone"].update(legend={"0": "calm", "1": "annoyed"}), + "legend does not match", + ), + ( + lambda p, q: p["answers"]["tone"].update(probabilities={"0": 0.5, "1": 0.5}), + "do not match the legend", + ), + (lambda p, q: p["answers"]["tone"].update(score=2.5), "outside the legend range"), + (lambda p, q: p["answers"]["tone"].update(score=-0.5), "outside the legend range"), + ( + lambda p, q: p["answers"]["tone"].update(score=0.9), + "inconsistent with the legend probabilities", + ), + (lambda p, q: p["answers"]["urgent"].update(noul=1.5), "out of range"), + ( + lambda p, q: p["answers"]["bucket"].update( + probabilities={"billing": 0.4, "outage": 0.2, "feature": 0.1} + ), + "probabilities sum to", + ), + ( + lambda p, q: p["answers"]["bucket"].update(choice=[]), + "outside the offered options", + ), + ( + lambda p, q: p["answers"]["bucket"].update(choice={}), + "outside the offered options", + ), + ], +) +def test_malformed_responses_are_rejected_visibly(mutate, fragment) -> None: + payload = _payload() + questions = _questions() + mutate(payload, questions) + session = FakeSession(FakeResponse(payload=payload)) + with pytest.raises(RefinementError, match=fragment) as excinfo: + _client(session).ask({}, questions) + assert "djev-spark" in str(excinfo.value) + +def test_http_401_is_not_retried_and_names_the_provider() -> None: + session = FakeSession(FakeResponse(status_code=401)) + with pytest.raises(RefinementError, match="djev-spark request failed with HTTP 401"): + _client(session).ask({}, _questions()) + assert len(session.calls) == 1 + + +def test_retryable_503_then_success_counts_both_attempts() -> None: + session = FakeSession( + FakeResponse(status_code=503), FakeResponse(payload=_payload()) + ) + _, usage = _client(session).ask({}, _questions()) + assert len(session.calls) == 2 + assert usage["_attempts"] == 2 + + +def test_retry_exhaustion_reports_unavailability() -> None: + session = FakeSession( + requests.ConnectionError("reset"), requests.ConnectionError("reset") + ) + with pytest.raises(RefinementError, match="djev-spark unavailable"): + _client(session, djev_max_retries=1).ask({}, _questions()) + + +@pytest.mark.parametrize( + "response", + [ + FakeResponse(broken_json=True), + FakeResponse(payload=[1, 2, 3]), + FakeResponse(payload={"usage": {}}), # no answers key + ], +) +def test_unusable_successes_are_rejected(response: FakeResponse) -> None: + session = FakeSession(response) + with pytest.raises(RefinementError): + _client(session).ask({}, _questions()) + + +def test_judge_relevance_consumes_a_djev_client() -> None: + payload = _payload() + payload["answers"] = { + "c0": {"type": "noul", "noul": 0.9}, + "c1": {"type": "noul", "noul": 0.2}, + } + session = FakeSession(FakeResponse(payload=payload)) + client = _client(session) + results = [ + {"artifact_id": f"a{i}", "video_id": "v", "timestamp_sec": i, "text": "x", "source_type": "caption"} + for i in range(2) + ] + probabilities, usage = judge_relevance(client, "query", results) + + assert probabilities == {"a0": 0.9, "a1": 0.2} + assert usage["requests"] == 1 + assert usage["served_model"] == "dgemma" + # The batched relevance questions are exactly what went on the wire. + sent = session.calls[0]["json"]["questions"] + assert set(sent) == {"c0", "c1"} + assert sent["c0"]["type"] == "noul" + + +def test_decision_lane_selection_and_gating(monkeypatch: pytest.MonkeyPatch) -> None: + hosted = AVConfig(typesafe_api_key="k") + assert decision_provider_name(hosted) == "jev" + assert type(open_decision_client(hosted)).__name__ == "SystemOneClient" + + self_hosted = AVConfig(typesafe_api_key="k", djev_endpoint=ENDPOINT) + assert decision_provider_name(self_hosted) == "djev-spark" + client = open_decision_client(self_hosted) + assert isinstance(client, DjevClient) + + unconfigured = AVConfig() + assert decision_provider_name(unconfigured) is None + + # Env var alone selects the lane with file/env priority intact. + monkeypatch.setenv("AV_DJEV_ENDPOINT", ENDPOINT) + from_env = AVConfig() + assert decision_provider_name(from_env) == "djev-spark" + + +def test_config_file_and_env_priority_for_djev_fields( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from av.core import config as config_module + + config_file = tmp_path / "config.json" + config_file.write_text( + '{"djev_endpoint": "http://file-host:8011/v1/systemone", "djev_seed": 7}' + ) + monkeypatch.setattr(config_module, "CONFIG_FILE_PATH", config_file) + + config = config_module.get_config() + assert config.djev_endpoint == "http://file-host:8011/v1/systemone" + assert config.djev_seed == 7 + + monkeypatch.setenv("AV_DJEV_ENDPOINT", "http://env-host:8011/v1/systemone") + config = config_module.get_config() + assert config.djev_endpoint == "http://env-host:8011/v1/systemone" + assert config.djev_seed == 7 # file value survives where env is silent + + +# --- ask() wiring through the djev lane ------------------------------- + + +def _video(video_id: str, path: Path) -> VideoRecord: + return VideoRecord( + id=video_id, + file_path=str(path), + file_hash=f"hash-{video_id}", + file_size_bytes=path.stat().st_size, + filename=path.name, + duration_sec=100.0, + status="complete", + ) + + +def _artifact(artifact_id: str, video_id: str, start: float, text: str) -> ArtifactRecord: + return ArtifactRecord( + id=artifact_id, + video_id=video_id, + type="caption", + start_sec=start, + end_sec=start + 10.0, + text=text, + ) + + +@pytest.fixture() +def repo(tmp_path: Path) -> Repository: + return Repository(tmp_path / "av.db") + + +def _seed_video(repo: Repository, tmp_path: Path, video_id: str = "v1") -> Path: + path = tmp_path / f"{video_id}.mp4" + path.write_bytes(b"fake-video") + repo.insert_video(_video(video_id, path)) + repo.insert_artifacts_batch([ + _artifact(f"{video_id}-{i}", video_id, i * 10.0, f"cake scene {i}") + for i in range(10) + ]) + return path + + +class FakeDjevDecision: + """Duck-typed DjevClient: scripted decisions with served identity.""" + + provider_label = "djev-spark" + served_model = "dgemma" + server_engine = "vllm" + + def __init__(self, relevance: float = 0.9, supports: list[float] | None = None) -> None: + self.relevance = relevance + self.supports = list(supports) if supports is not None else [0.95] + self.calls: list[tuple[dict, dict]] = [] + + def ask(self, state: Any, questions: dict[str, dict]) -> tuple[dict[str, dict], dict]: + self.calls.append((state, questions)) + usage = {"input_tokens": 5, "output_tokens": 1, "_attempts": 1} + if "is_supported" in questions: + return {"is_supported": {"type": "noul", "noul": self.supports.pop(0)}}, usage + answers: dict[str, dict] = {} + for key, question in questions.items(): + if question["type"] == "choice": + names = list(question["criteria"]) + share = 1.0 / len(names) + answers[key] = { + "type": "choice", + "choice": names[0], + "probabilities": {name: share for name in names}, + "confidence": share, + } + else: + answers[key] = {"type": "noul", "noul": self.relevance} + return answers, usage + + +class FakeAnswerLLM: + def __init__(self, config: AVConfig) -> None: + self.config = config + + def complete_with_usage(self, prompt: str, context: str) -> CompletionResult: + return CompletionResult("refined answer", input_tokens=None, output_tokens=None) + + +def test_ask_over_djev_lane_reports_djev_identity_and_bases( + repo: Repository, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _seed_video(repo, tmp_path) + fake = FakeDjevDecision(relevance=0.9, supports=[0.95]) + monkeypatch.setattr("av.search.rag.open_decision_client", lambda config: fake) + monkeypatch.setattr("av.search.rag.OpenAILLM", FakeAnswerLLM) + + result = ask( + "cake", repo, AVConfig(djev_endpoint=ENDPOINT, embed_model=""), video_id="v1" + ) + + assert result["route"] == "refined" + assert result["confidence_basis"] == "djev_spark_answer_support" + assert result["refinement"]["decision_provider"] == "djev-spark" + assert result["refinement"]["served_model"] == "dgemma" + assert result["refinement"]["server_engine"] == "vllm" + assert result["ask_settings"]["decision_provider"] == "djev-spark" + + +def test_ask_all_irrelevant_reports_djev_relevance_basis( + repo: Repository, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _seed_video(repo, tmp_path) + fake = FakeDjevDecision(relevance=0.1) + monkeypatch.setattr("av.search.rag.open_decision_client", lambda config: fake) + monkeypatch.setattr( + "av.search.rag.OpenAILLM", + lambda config: (_ for _ in ()).throw(AssertionError("answer model must not run")), + ) + + result = ask( + "cake", repo, AVConfig(djev_endpoint=ENDPOINT, embed_model=""), video_id="v1" + ) + + assert result["route"] == "refined_no_results" + assert result["confidence_basis"] == "djev_spark_relevance" + assert result["refinement"]["decision_provider"] == "djev-spark" + + +def test_ask_inspected_answer_reports_djev_sampled_frames_basis( + repo: Repository, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _seed_video(repo, tmp_path) + # First support judgment (plain answer) fails; the judgment on the + # vision-inspected citations succeeds. + fake = FakeDjevDecision(relevance=0.9, supports=[0.1, 0.95]) + monkeypatch.setattr("av.search.rag.open_decision_client", lambda config: fake) + monkeypatch.setattr("av.search.rag.OpenAILLM", FakeAnswerLLM) + from av.search.usage import new_usage + + monkeypatch.setattr( + "av.search.rag.inspect_with_stronger_vision", + lambda *args, **kwargs: { + "status": "supported", + "answer": "inspected answer", + "citations": [{ + "video_id": "v1", + "start_sec": 0.0, + "end_sec": 10.0, + "source_type": "caption", + "text": "cake scene 0", + }], + "windows": [], + "usage": new_usage(), + "warnings": [], + }, + ) + + result = ask( + "cake", repo, AVConfig(djev_endpoint=ENDPOINT, embed_model=""), video_id="v1" + ) + + assert result["route"] == "vision_inspected" + assert result["confidence_basis"] == "djev_spark_answer_support_after_sampled_frames" + assert result["refinement"]["served_model"] == "dgemma" diff --git a/tests/test_natural_language_search.py b/tests/test_natural_language_search.py index 208fc0b..cbf28dc 100644 --- a/tests/test_natural_language_search.py +++ b/tests/test_natural_language_search.py @@ -57,7 +57,7 @@ def test_unicode_punctuation_and_fts_operators_are_literal_candidates(repo): @pytest.mark.parametrize("question", ["What is it?", "?! () :", "Where did the submarine surface?"]) def test_empty_or_unrelated_question_returns_no_hits_or_model_calls(repo, question): - with patch("av.search.rag.SystemOneClient") as judge, patch("av.search.rag.OpenAILLM") as llm: + with patch("av.search.rag.open_decision_client") as judge, patch("av.search.rag.OpenAILLM") as llm: result = ask(question, repo, AVConfig(typesafe_api_key="test", embed_model=""), video_id="local") assert result["route"] == "refined_no_results" assert result["citations"] == [] @@ -123,7 +123,7 @@ def complete_with_usage(self, prompt, context): assert "rental" not in context return CompletionResult("The truck was blue", input_tokens=20, output_tokens=5) - with patch("av.search.rag.SystemOneClient", return_value=Judge()), \ + with patch("av.search.rag.open_decision_client", return_value=Judge()), \ patch("av.search.rag.OpenAILLM", side_effect=Answer) as llm: result = ask(question, repo, AVConfig(typesafe_api_key="test", embed_model=""), video_id="local") assert len(relevance_texts) == 2