From f78fe738352deda03a293be9ec872f1c8ea04f2b Mon Sep 17 00:00:00 2001 From: Sean Phan Date: Mon, 21 Sep 2026 02:52:27 +0000 Subject: [PATCH 1/2] feat(search): add djev-spark decision lane with validated /v1/systemone adapter Run the existing refined-ask workflow against a self-hosted djev-spark server, which speaks the same documented /v1/systemone contract as hosted System One. An explicit AV_DJEV_ENDPOINT selects the lane; no default endpoint ships, and the endpoint wins over hosted Jev when both are set. - DjevClient mirrors the hosted client's timeout/retry/backoff semantics, passes a reproducible seed, and visibly rejects malformed responses: missing or unknown answers, type mismatches, choices outside the offered options, legend/probability mismatches, out-of-range or non-summing probabilities, scores outside the zero-based legend range or inconsistent with sum(i*p), null answers for questions without ask_if, and responses that do not identify the served model and engine. - Every successful decision records the served model, engine, and redacted endpoint host into usage receipts and ask metadata, so a djev answer is never presented as a Jev measurement; confidence_basis strings are lane-derived. - Boundary judging resolves single-candidate sides locally instead of sending one-option questions, which structured providers reject. - Offline deterministic tests cover the wire contract, validation, identity enforcement, lane selection, and ask-path outputs; cookbook recipe documents configuration, self-host hardening, and the pending live comparison. Upstream audited at mmastrac/djev-spark 1444f3e927f83ba508e5b28a4fd4fdd9ecd0976b. --- CLAUDE.md | 10 + cookbook/README.md | 1 + cookbook/djev-clipping/README.md | 121 ++++++ src/av/cli/config_cmd.py | 4 + src/av/core/config.py | 19 + src/av/search/djev.py | 290 ++++++++++++++ src/av/search/rag.py | 42 ++- src/av/search/refine.py | 125 ++++--- tests/test_ask_refinement.py | 73 ++-- tests/test_djev_adapter.py | 520 ++++++++++++++++++++++++++ tests/test_natural_language_search.py | 4 +- 11 files changed, 1130 insertions(+), 79 deletions(-) create mode 100644 cookbook/djev-clipping/README.md create mode 100644 src/av/search/djev.py create mode 100644 tests/test_djev_adapter.py 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..47a39fb --- /dev/null +++ b/src/av/search/djev.py @@ -0,0 +1,290 @@ +"""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() + if answer.get("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..1503d0f 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 @@ -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..8944602 --- /dev/null +++ b/tests/test_djev_adapter.py @@ -0,0 +1,520 @@ +"""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", + ), + ], +) +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 From 7dc30cbad59e10cc0422b685b7f2e32b7d8acf0c Mon Sep 17 00:00:00 2001 From: Sean Phan Date: Mon, 21 Sep 2026 02:59:34 +0000 Subject: [PATCH 2/2] fix(search): reject non-string choice values in djev validation Set membership raises TypeError on an unhashable answer value (a JSON array or object), which escaped RefinementError handling and crashed past ask's sanitized fallback. Require a string before comparing the reported choice against the offered options, with fake-session regressions for array and object choices, and rename an unused local flagged by the targeted lint check. --- src/av/search/djev.py | 5 ++++- tests/test_ask_refinement.py | 2 +- tests/test_djev_adapter.py | 9 ++++++++- 3 files changed, 13 insertions(+), 3 deletions(-) diff --git a/src/av/search/djev.py b/src/av/search/djev.py index 47a39fb..3831a2e 100644 --- a/src/av/search/djev.py +++ b/src/av/search/djev.py @@ -215,7 +215,10 @@ def _validate_choice( ) -> None: criteria = question.get("criteria") names = set(criteria) if isinstance(criteria, dict) else set() - if answer.get("choice") not in names: + # 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: diff --git a/tests/test_ask_refinement.py b/tests/test_ask_refinement.py index 1503d0f..a62709c 100644 --- a/tests/test_ask_refinement.py +++ b/tests/test_ask_refinement.py @@ -179,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, diff --git a/tests/test_djev_adapter.py b/tests/test_djev_adapter.py index 8944602..02340a1 100644 --- a/tests/test_djev_adapter.py +++ b/tests/test_djev_adapter.py @@ -246,6 +246,14 @@ def test_success_without_served_identity_is_rejected(removed: str) -> None: ), "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: @@ -257,7 +265,6 @@ def test_malformed_responses_are_rejected_visibly(mutate, fragment) -> None: _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"):