From 46211a49ebe5c9e4f5b525c7e98e2b5b0d43b1f0 Mon Sep 17 00:00:00 2001 From: Anton Volkov Date: Sat, 14 Nov 2026 19:59:06 +0000 Subject: [PATCH 1/3] fix(novelty): atoms, probes, grades and batches stop at a run-wide provider error (#708) Extraction, the faithfulness check, probe wording generation, probe answers and grades, the value and conflict batches (`ask_in_batches`) and the retention adjudication each caught every `Exception` per item. With no credentials, a permission error or an unknown model each of them went on and made, and recorded as failed, every remaining call. Each now stops at the first error for which `routing_stop.is_run_wide` is true, the test `novelty routing` uses (#677), and raises it as a `SpendCapReached` with `reason` `permanent_error`, which the cap-stop paths already keep as a partial result: extraction marks the unread spans and sets `report.partial`, `run_measurement` counts the jobs not run in `not_run` and carries the stop (an answer already paid for stays, ungraded), `analyze` writes its partial record, and `ask_in_batches` returns the chunks answered so far with the stop on `BatchResult.stop`, which `judge_value` and `detect_conflicts` raise. A refusal, an unparseable reply and a context-length rejection stay one item's failure. `is_retryable` alone would stop on those too. The readers of a partial file and the rerun hints no longer advise a higher `--max-usd` after this stop, and `novelty.v1.json` lists `permanent_error` as a `partial.reason`. `test_a_recovered_parse_error_does_not_hide_the_failure_that_lost_spans` used an `AuthenticationError` as its example of a lost batch: it is a `BadRequestError` there now, since a bad key stops the run. Signed-off-by: Anton Volkov --- .../20261114_130000_novelty-permanent-stop.md | 3 + dokimi/docs/skill-novelty.md | 6 +- dokimi/dokimi/artifacts.py | 1 + dokimi/dokimi/commands/novelty_atoms_file.py | 3 +- dokimi/dokimi/commands/novelty_budget.py | 1 + dokimi/dokimi/novelty/atoms.py | 14 +- dokimi/dokimi/novelty/batching.py | 24 +- dokimi/dokimi/novelty/classify.py | 4 + dokimi/dokimi/novelty/measure.py | 16 +- dokimi/dokimi/novelty/probes.py | 2 + dokimi/dokimi/novelty/retention.py | 2 + dokimi/dokimi/novelty/routing_stop.py | 23 +- dokimi/dokimi/schemas/novelty.v1.json | 4 +- dokimi/tests/test_novelty_atoms.py | 13 +- dokimi/tests/test_novelty_permanent_stop.py | 275 ++++++++++++++++++ 15 files changed, 370 insertions(+), 21 deletions(-) create mode 100644 dokimi/changelog.d/20261114_130000_novelty-permanent-stop.md create mode 100644 dokimi/tests/test_novelty_permanent_stop.py diff --git a/dokimi/changelog.d/20261114_130000_novelty-permanent-stop.md b/dokimi/changelog.d/20261114_130000_novelty-permanent-stop.md new file mode 100644 index 000000000..95ac1a649 --- /dev/null +++ b/dokimi/changelog.d/20261114_130000_novelty-permanent-stop.md @@ -0,0 +1,3 @@ +### Fixed + +- `skill novelty atoms`, `probe` and `analyze` and `skill shorten` stop at the first authentication, permission, unknown-model, unsupported-parameter or no-credentials error (extraction, the faithfulness check, probe wordings, answers and grades, the value and conflict batches, retention adjudication) instead of failing every remaining call and recording each one as failed; the partial result is kept as at the cap, with `reason` `permanent_error`. A refusal, an unparseable reply or a request that is too long is still one failed item. (#708) diff --git a/dokimi/docs/skill-novelty.md b/dokimi/docs/skill-novelty.md index d83306ff5..df0310648 100644 --- a/dokimi/docs/skill-novelty.md +++ b/dokimi/docs/skill-novelty.md @@ -838,7 +838,7 @@ so count the queries first, or pass `--max-usd` to stop the run at a spend limit A stored response in `DOKIMI_RESPONSE_CACHE` is replayed without a call, so a rerun can make fewer calls than the printed count. -**A permanent provider error that every call would meet stops a routing run the way the cap does.** An authentication or permission error, an unknown model ID (`not found`), a parameter the provider does not take (every call sends the same ones) or missing credentials (with `claude-cli`: not logged in, a usage limit, no `claude` binary) would fail every later call alike, so the run stops at that call, keeps the rows made before it and writes the partial result below, with `"reason": "permanent_error"` and the error text as the message (`--max-usd` is not involved; the exit line gives `calls_done` of `calls_planned`). Any other failure is one failed row and the run goes on: a refusal, a request one query makes too long, an unparseable reply, an error the client does not recognise, a transient error that ran out of retries. +**A permanent provider error that every call would meet stops a routing run the way the cap does.** An authentication or permission error, an unknown model ID (`not found`), a parameter the provider does not take (every call sends the same ones) or missing credentials (with `claude-cli`: not logged in, a usage limit, no `claude` binary) would fail every later call alike, so the run stops at that call, keeps the rows made before it and writes the partial result below, with `"reason": "permanent_error"` and the error text as the message (`--max-usd` is not involved; the exit line gives `calls_done` of `calls_planned`). Any other failure is one failed row and the run goes on: a refusal, a request one query makes too long, an unparseable reply, an error the client does not recognise, a transient error that ran out of retries. `skill novelty atoms`, `probe` and `analyze` and `skill shorten` stop the same way at the first such error in an atom extraction call, the faithfulness check, a probe wording, an answer, a grade, a value or conflict batch or a retention adjudication, and keep what they paid for as they do at the cap, with `"reason": "permanent_error"`; a refusal, an unparseable reply or a request one item makes too long is still one failed item. **A routing run stopped by `--max-usd` keeps what it paid for.** `skill routing --model`, single artifact or `catalog routing`, writes the result of the calls made before the cap (or a call with no @@ -1151,8 +1151,8 @@ the cap stops the run). A partial file is for inspection; rerun to get a finished one. `analyze` does the same beside its `novelty.json` (`.partial.json`, never over a finished file; the `--out` file, `novelty.json` in `--out-dir` by default). Its partial file has -`"status": "partial"` and a `partial` block: `reason` (`spend_cap`, or `no_price` for a call with -no known price), `stage` (`extraction`, `verify`, `need_graph`, `probes`, `measurement`, `classify` +`"status": "partial"` and a `partial` block: `reason` (`spend_cap`, `no_price` for a call with +no known price, or `permanent_error`), `stage` (`extraction`, `verify`, `need_graph`, `probes`, `measurement`, `classify` or `attribution`: the stage that was running), `message`, `cause` (the cap's own text, with what was spent), and `spans_read` / `spans_total` for an extraction stop or `calls_done` / `calls_planned` (probe answers) once the measurement started. It keeps the `extracted_atoms` and `source_spans` diff --git a/dokimi/dokimi/artifacts.py b/dokimi/dokimi/artifacts.py index 06789f1e3..8d8e633f7 100644 --- a/dokimi/dokimi/artifacts.py +++ b/dokimi/dokimi/artifacts.py @@ -356,6 +356,7 @@ def load_artifact(kind: str, path: Path | str, *, validate: bool = True) -> dict raise InputError(f"{path} is a partial novelty artifact: the run was stopped before it " "finished, so it is not a finished analysis. " + ( "Drop --max-usd to run unmetered" if reason == "no_price" else + "Fix the provider error and rerun" if reason == "permanent_error" else "Rerun with a higher --max-usd, or without the cap")) if validate: try: diff --git a/dokimi/dokimi/commands/novelty_atoms_file.py b/dokimi/dokimi/commands/novelty_atoms_file.py index fde485555..1b6471624 100644 --- a/dokimi/dokimi/commands/novelty_atoms_file.py +++ b/dokimi/dokimi/commands/novelty_atoms_file.py @@ -148,7 +148,8 @@ def _verdicts_hint(oracle, cache_dir, stop) -> str: stored = max(0, oracle.escalations - oracle.errors - 1) dropped = stop.reply is not None - rerun = "no --max-usd" if stop.reason == "no_price" else "a higher --max-usd" + rerun = ("no --max-usd" if stop.reason == "no_price" else + "the error fixed" if stop.reason == "permanent_error" else "a higher --max-usd") tail = "; the call with no price was paid for and its verdict was not stored" if dropped else "" if cache_dir and stored: return (f"the retention verdicts adjudicated before the stop ({stored}) are stored in " diff --git a/dokimi/dokimi/commands/novelty_budget.py b/dokimi/dokimi/commands/novelty_budget.py index cc70622ab..cbe5e4a44 100644 --- a/dokimi/dokimi/commands/novelty_budget.py +++ b/dokimi/dokimi/commands/novelty_budget.py @@ -70,6 +70,7 @@ def replay_hint(reason: str = "spend_cap") -> str: if response_cache_active(): rerun = ("a higher --max-usd" if reason == "spend_cap" + else "the error fixed," if reason == "permanent_error" else "the model priced, or without --max-usd,") return (f". Rerun with {rerun} and the same {RESPONSE_CACHE_ENV}: the calls " "already paid for replay at $0 (a cut-off or unusable reply is asked again)") diff --git a/dokimi/dokimi/novelty/atoms.py b/dokimi/dokimi/novelty/atoms.py index 240593780..6e6581e45 100644 --- a/dokimi/dokimi/novelty/atoms.py +++ b/dokimi/dokimi/novelty/atoms.py @@ -29,7 +29,7 @@ from dokimi.utils import InputError, read_json_input -from .batching import retry_wording +from .batching import or_stop, retry_wording from .client import LLMClient, OutputTruncated, SpendCapReached, accepting from .spans import Span @@ -419,7 +419,7 @@ def parse(reply: str) -> list: unparseable = False try: report.calls += 1 - reply = client.text(prompt, system=_EXTRACT_SYSTEM, accept=usable) + reply = or_stop(client.text, prompt, system=_EXTRACT_SYSTEM, accept=usable) # a run-wide error stops (#708) if not reply.strip(): # Its own cause, not a failed call: the fix is a higher output ceiling. lost = ValueError( @@ -571,8 +571,8 @@ def verify_atoms( continue source = "\n".join(by_id[s].text for s in atom.span_ids if s in by_id) try: - verdict = str(client.json_object( - _VERIFY_PROMPT.format(statement=atom.statement, source=source), + verdict = str(or_stop( + client.json_object, _VERIFY_PROMPT.format(statement=atom.statement, source=source), usable=_names_a_faithfulness_verdict, ).get("verdict", "")).strip() except Exception: @@ -786,9 +786,9 @@ def load_atoms_record(path) -> tuple[list[Atom], list[Span], dict]: if isinstance(data, dict) and ("partial" in data or data.get("status") == "partial"): if not isinstance(partial := data.get("partial"), dict) or not partial: # no run writes this raise InputError(f"{path} is marked partial with no `partial` block: truncated or hand-edited") - if partial.get("reason") == "no_price": # a call with no price: a higher cap will not help - raise InputError(f"{path} is a partial atoms file: {partial.get('message', '')}, so " - "it is not a finished atom set. Drop --max-usd to run unmetered") + if (reason := partial.get("reason")) in ("no_price", "permanent_error"): # a higher cap will not help + raise InputError(f"{path} is a partial atoms file: {partial.get('message', '')}, so it is not a finished atom set. " + + ("Drop --max-usd to run unmetered" if reason == "no_price" else "Fix the error and rerun")) raise InputError(f"{path} is a partial atoms file: --max-usd stopped the run before it " "finished, so it is not a finished atom set. Rerun with a higher " "--max-usd, or without the cap") diff --git a/dokimi/dokimi/novelty/batching.py b/dokimi/dokimi/novelty/batching.py index 0169deef0..7bf6a18a9 100644 --- a/dokimi/dokimi/novelty/batching.py +++ b/dokimi/dokimi/novelty/batching.py @@ -21,7 +21,8 @@ from functools import partial from typing import Any, Generic, TypeVar -from .client import LLMClient, OutputTruncated, accepting +from .client import LLMClient, OutputTruncated, SpendCapReached, accepting +from .routing_stop import is_run_wide, permanent_stop, stop_if_run_wide T = TypeVar("T") @@ -46,6 +47,18 @@ def retry_wording(retried: int, unusable: int) -> tuple[str, str]: return label, ("; a higher --max-tokens avoids the extra calls" if cut else "") +def or_stop(call: Callable[..., T], *args: Any, **kwargs: Any) -> T: + """``call(*args, **kwargs)``; a run-wide provider error raises its :func:`permanent_stop` instead. + + Any other error passes through unchanged, for the caller's per-item handling. + """ + try: + return call(*args, **kwargs) + except Exception as exc: + stop_if_run_wide(exc) + raise + + @dataclass class BatchResult(Generic[T]): """What came back, and what did not. @@ -68,6 +81,10 @@ class BatchResult(Generic[T]): an unusable reply is counted in ``unusable`` instead. Errors that a split recovered from are not recorded, so this never blames a cause that lost nothing. + stop: Set when a run-wide provider error (:func:`~dokimi.novelty.routing_stop.is_run_wide`) + ended the run at its first call that met it: ``reason`` ``permanent_error``. The + chunks answered before it stay in ``answered``; the chunk that met it and those not + yet asked are in ``failed``, with no ``last_error``. A caller raises it, as a cap stop is. """ answered: list[tuple[list[T], Any]] = field(default_factory=list) @@ -78,6 +95,7 @@ class BatchResult(Generic[T]): unusable_retries: int = 0 calls: int = 0 last_error: str = "" + stop: SpendCapReached | None = None def ask_in_batches( @@ -128,6 +146,10 @@ def read(chunk: list[T], reply: str) -> Any: replied = True result.answered.append((chunk, read(chunk, reply))) except Exception as exc: + if is_run_wide(exc): # every later call meets it too: stop, the rest is unanswered + result.stop = permanent_stop(exc) + result.failed.extend(item for rest in [chunk, *pending] for item in rest) + break half = len(chunk) // 2 # A reply cut at the ceiling is re-asked smaller, like one that will not parse. if (replied or isinstance(exc, OutputTruncated)) and half >= max(min_size, 1): diff --git a/dokimi/dokimi/novelty/classify.py b/dokimi/dokimi/novelty/classify.py index 5ab53b8d2..2a326c07b 100644 --- a/dokimi/dokimi/novelty/classify.py +++ b/dokimi/dokimi/novelty/classify.py @@ -283,6 +283,8 @@ def judge_value( prompt=lambda chunk: _VALUE_PROMPT.format(task=task, statements=_numbered(chunk)), parse=extract_json_array, check=_judge_check, ) + if result.stop: # a run-wide provider error ends the classification as a cap stop does (#708) + raise result.stop report.retried_batches += result.retried_batches report.unusable_retries += result.unusable_retries report.calls += result.calls @@ -364,6 +366,8 @@ def detect_conflicts( prompt=lambda chunk: _CONFLICT_PROMPT.format(statements=_numbered(chunk)), parse=_conflict_rows, min_size=2, check=_conflict_check, ) + if result.stop: + raise result.stop report.retried_batches += result.retried_batches report.unusable_retries += result.unusable_retries report.calls += result.calls diff --git a/dokimi/dokimi/novelty/measure.py b/dokimi/dokimi/novelty/measure.py index 7dbf5ef16..973fb3c9d 100644 --- a/dokimi/dokimi/novelty/measure.py +++ b/dokimi/dokimi/novelty/measure.py @@ -24,8 +24,10 @@ from __future__ import annotations import random +from collections.abc import Callable from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field +from typing import Any from dokimi.utils import InputError from .atoms import Atom @@ -33,6 +35,7 @@ from .controls import (REPEAT_ARM, clustered_ci, control_fields, reversal_by_arm, reversal_paired) from .probes import Probe +from .routing_stop import PERMANENT_REASON, stop_if_run_wide # Judge verdict scale. Deliberately finer than the collapses below so a rater # disagreement is visible as a confusion rather than hidden inside a boolean. @@ -378,6 +381,7 @@ def one(job: tuple[Probe, str]) -> ProbeRun: except SpendCapReached as exc: # a call with no price: its paid reply is kept, ungraded return _unpriced_answer_run(probe, arm, exc, stops, ungraded) except Exception as exc: # a provider failure is a missing datum, not a wrong answer + stop_if_run_wide(exc) # but one every later call meets stops the measurement (#708) response, confidence = f"", None verdict = ERRORED answered = False @@ -403,7 +407,7 @@ def guarded(job: tuple[Probe, str]) -> ProbeRun | None: return None with ThreadPoolExecutor(max_workers=max(1, concurrency)) as pool: - answered = list(pool.map(guarded, jobs)) + answered = list(pool.map(_not_after_permanent_stop(stops, guarded), jobs)) runs = [run for run in answered if run is not None] stop = (stops[0].reason, str(stops[0])) if stops else None return MeasureResult(runs=runs, arms=tuple(arms), with_arm="full", @@ -411,6 +415,15 @@ def guarded(job: tuple[Probe, str]) -> ProbeRun | None: probes=list(probes)) +def _not_after_permanent_stop(stops: list[SpendCapReached], job: Callable[[Any], Any]) -> Callable[[Any], Any]: + """``job``, which is not run (``None``) once a run-wide provider error is among ``stops`` (#708).""" + + def run(item: Any) -> Any: + return None if any(held.reason == PERMANENT_REASON for held in stops) else job(item) + + return run + + def _capped_grade_run(probe: Probe, arm: str, response: str, confidence: float | None, exc: SpendCapReached, stops: list[SpendCapReached], ungraded: list[ProbeRun]) -> ProbeRun: @@ -531,6 +544,7 @@ def _grade(probe: Probe, response: str, judge: LLMClient) -> str: try: raw = judge.json_object(prompt, usable=_names_a_verdict).get("verdict") except Exception: + stop_if_run_wide() # raises a permanent stop; any other failure is an ERRORED grade return ERRORED verdict = raw.strip() if isinstance(raw, str) else "" if not verdict: diff --git a/dokimi/dokimi/novelty/probes.py b/dokimi/dokimi/novelty/probes.py index 828c3df8a..c03ce0dd5 100644 --- a/dokimi/dokimi/novelty/probes.py +++ b/dokimi/dokimi/novelty/probes.py @@ -40,6 +40,7 @@ from dokimi.utils import InputError from .atoms import Atom from .client import LLMClient, SpendCapReached +from .routing_stop import stop_if_run_wide PROBE_KINDS = ("direct", "paraphrase", "reversal") @@ -493,6 +494,7 @@ def _generate(client: LLMClient, template: str, lines: list[str]) -> list[str]: try: rows = client.json_array(template.format(**{key: numbered})) except Exception: + stop_if_run_wide() # raises a permanent stop; any other failure is a chunk of rejected wordings rows = [] return _aligned(rows, len(lines)) diff --git a/dokimi/dokimi/novelty/retention.py b/dokimi/dokimi/novelty/retention.py index 7c6f331e2..404411713 100644 --- a/dokimi/dokimi/novelty/retention.py +++ b/dokimi/dokimi/novelty/retention.py @@ -26,6 +26,7 @@ from .atoms import Atom from .client import LLMClient +from .routing_stop import stop_if_run_wide RETAINED = "retained" ABSENT = "absent" @@ -170,6 +171,7 @@ def _adjudicate(self, atom: Atom, text: str) -> str | None: accept=lambda raw: parse_verdict(raw) is not None, ) except Exception: + stop_if_run_wide() # raises a permanent stop; any other failure is an unusable verdict return None return parse_verdict(reply) diff --git a/dokimi/dokimi/novelty/routing_stop.py b/dokimi/dokimi/novelty/routing_stop.py index 53f4e378d..b905920b7 100644 --- a/dokimi/dokimi/novelty/routing_stop.py +++ b/dokimi/dokimi/novelty/routing_stop.py @@ -1,11 +1,16 @@ # Copyright (C) 2026 Intel Corporation # SPDX-License-Identifier: Apache-2.0 -"""Which provider errors stop a routing run (#677); split from ``routing.py``, which stays under its size cap.""" +"""Which provider errors stop a run (#677, #708); split from ``routing.py``, which stays under its size cap.""" from __future__ import annotations +import sys + from dokimi.llm import _NON_RETRYABLE_MESSAGES, is_retryable +from .client import SpendCapReached + PERMANENT_STOP = "permanent error: " +PERMANENT_REASON = "permanent_error" # Errors that fail every later call alike: no key, no permission, an unknown model, a parameter every # call sends. Any other error (a context-length rejection, a refusal, an odd reply) may be one query's. _RUN_WIDE_ERRORS = frozenset({"authenticationerror", "permissiondeniederror", "notfounderror", @@ -21,3 +26,19 @@ def is_run_wide(exc: Exception) -> bool: return run_wide return type(exc).__name__.lower() in _RUN_WIDE_ERRORS or any( marker in str(exc).lower() for marker in _NO_CREDENTIALS) + + +def permanent_stop(exc: Exception) -> SpendCapReached: + """The stop a run-wide error makes, in the form a cap stop has: ``reason`` ``permanent_error``.""" + return SpendCapReached(PERMANENT_STOP + (str(exc)[:200] or type(exc).__name__), reason=PERMANENT_REASON) + + +def stop_if_run_wide(exc: BaseException | None = None) -> None: + """Raise :func:`permanent_stop` when *exc* (default: the exception being handled) is run-wide. + + Called first in the handler of a per-item ``except Exception``, so that the one failure every + later call would meet ends the run as a cap stop does, and any other failure stays one item's. + """ + exc = exc if exc is not None else sys.exc_info()[1] + if isinstance(exc, Exception) and is_run_wide(exc): + raise permanent_stop(exc) from exc diff --git a/dokimi/dokimi/schemas/novelty.v1.json b/dokimi/dokimi/schemas/novelty.v1.json index f974f8e49..24af02e9c 100644 --- a/dokimi/dokimi/schemas/novelty.v1.json +++ b/dokimi/dokimi/schemas/novelty.v1.json @@ -526,9 +526,9 @@ "partial": { "type": "object", "required": ["reason", "stage", "message"], - "description": "Why and where the run stopped. reason: spend_cap or no_price. stage: extraction, verify, need_graph, probes, measurement, classify or attribution. extraction adds spans_read and spans_total, measurement calls_done and calls_planned (probe answers paid for, the ungraded ones included; the message says how many), spend_cap a cause (what was spent). A partial record holds no paired, arms or view summaries: the cut leaves the arms uneven.", + "description": "Why and where the run stopped. reason: spend_cap, no_price or permanent_error (a provider error every later call would meet). stage: extraction, verify, need_graph, probes, measurement, classify or attribution. extraction adds spans_read and spans_total, measurement calls_done and calls_planned (probe answers paid for, the ungraded ones included; the message says how many), spend_cap a cause (what was spent). A partial record holds no paired, arms or view summaries: the cut leaves the arms uneven.", "properties": { - "reason": {"enum": ["spend_cap", "no_price"]}, + "reason": {"enum": ["spend_cap", "no_price", "permanent_error"]}, "stage": {"enum": ["extraction", "verify", "need_graph", "probes", "measurement", "classify", "attribution"]}, "message": {"type": "string"}, "cause": {"type": "string"}, diff --git a/dokimi/tests/test_novelty_atoms.py b/dokimi/tests/test_novelty_atoms.py index 428fcc929..ed289918b 100644 --- a/dokimi/tests/test_novelty_atoms.py +++ b/dokimi/tests/test_novelty_atoms.py @@ -114,17 +114,20 @@ def flaky(prompt, system): def test_a_recovered_parse_error_does_not_hide_the_failure_that_lost_spans(): - """Batch 1 is lost to a bad key; batch 2 is truncated and fully recovered.""" + """Batch 1 is lost to a request the model rejects; batch 2 is truncated and fully recovered. + + A bad key is not this case: it stops the run (test_novelty_permanent_stop.py). + """ spans = segment(DOC) first_batch = {s.id for s in spans[:5]} - class AuthenticationError(Exception): + class BadRequestError(Exception): pass def transport(prompt, system): ids = re.findall(r"^\[(s\d+)\]", prompt, flags=re.M) if first_batch & set(ids): - raise AuthenticationError("invalid api key") + raise BadRequestError("invalid request") reply = json.dumps(_rows(prompt)) return (reply[:-3] if len(ids) > 2 else reply), None @@ -133,9 +136,9 @@ def transport(prompt, system): atoms = extract_atoms(spans, client=client, batch_spans=5, report=report) assert len(atoms) == 5 and report.retried_batches > 0 lost = report.warnings(len(spans))[0] - assert "AuthenticationError: invalid api key" in lost + assert "BadRequestError: invalid request" in lost assert "did not parse" not in lost and "JSONDecodeError" not in lost - assert report.last_error == "AuthenticationError: invalid api key" + assert report.last_error == "BadRequestError: invalid request" def test_recovered_replies_that_assert_nothing_are_not_an_extraction_failure(): diff --git a/dokimi/tests/test_novelty_permanent_stop.py b/dokimi/tests/test_novelty_permanent_stop.py new file mode 100644 index 000000000..cfbe3067b --- /dev/null +++ b/dokimi/tests/test_novelty_permanent_stop.py @@ -0,0 +1,275 @@ +# Copyright (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 +"""A provider error every later call would meet stops atoms, probes, grades and batches (#708). + +Each site gets a mocked transport that raises an authentication error at its second call, and +a counterexample: a context-length rejection of one item is that item's failure and the run goes on. +""" +from __future__ import annotations + +import json +import re + +import pytest + +from dokimi.metrics.usage import UsageRecord +from dokimi.novelty.atoms import Atom, ExtractionReport, extract_atoms, verify_atoms +from dokimi.novelty.batching import ask_in_batches +from dokimi.novelty.classify import detect_conflicts, judge_value +from dokimi.novelty.client import LLMClient, SpendCapReached +from dokimi.novelty.measure import ERRORED, run_measurement +from dokimi.novelty.probes import Probe, ProbeSet, build_probe_set +from dokimi.novelty.retention import RetentionOracle +from dokimi.novelty.spans import segment + + +class AuthenticationError(Exception): + """Named as litellm's, which is_retryable and is_run_wide read by class name.""" + + +class BadRequestError(Exception): + """A context-length rejection of one oversized item.""" + + +class Calls: + """A transport that counts its calls and raises ``error`` at call number ``at``.""" + + def __init__(self, reply, *, error=None, at=2): + self.reply, self.error, self.at, self.n = reply, error, at, 0 + + def __call__(self, prompt, system=None): + self.n += 1 + if self.error is not None and self.n == self.at: + raise self.error + reply = self.reply(prompt) if callable(self.reply) else self.reply + return reply, UsageRecord(cost_usd=0.0) + + +def _client(transport, model="m") -> LLMClient: + return LLMClient(model=model, role="x", transport=transport) + + +AUTH = AuthenticationError("Unable to locate credentials") +TOO_LONG = BadRequestError("maximum context length is 8192 tokens") +ROWS = "\n\n".join(f"Setting number {i} defaults to {i * 10}." for i in range(1, 101)) + "\n" + + +def _atoms(n: int) -> list[Atom]: + return [Atom(statement=f"Setting {i} defaults to {i}.", question=f"What is setting {i}?", + answer=str(i), span_ids=(f"s{i}",)) for i in range(1, n + 1)] + + +def _one_atom_per_span(prompt): + return json.dumps([{"statement": f"{sid} holds a default.", "question": f"What of {sid}?", + "answer": "a value", "kind": "proposition", "tags": [], "span_ids": [sid]} + for sid in re.findall(r"^\[(s\d+)\]", prompt, flags=re.M)]) + + +def test_extraction_stops_at_the_first_run_wide_error_and_keeps_the_atoms_read(): + spans = segment(ROWS) + transport = Calls(_one_atom_per_span, error=AUTH) + report = ExtractionReport() + atoms = extract_atoms(spans, client=_client(transport), report=report) + assert transport.n == 2 # four batches of 25 spans + assert len(atoms) == 25 + assert report.partial["reason"] == "permanent_error" and report.partial["spans_read"] == 25 + assert report.partial["message"].endswith("permanent error: Unable to locate credentials") + assert not any(span.unread for span in spans[:25]) and all(span.unread for span in spans[25:]) + assert report.call_error == "" # not recorded as one failed batch + + +def test_extraction_goes_on_after_a_batch_the_model_rejects(): + spans = segment(ROWS) + transport = Calls(_one_atom_per_span, error=TOO_LONG) + report = ExtractionReport() + atoms = extract_atoms(spans, client=_client(transport), report=report) + assert transport.n == 4 and len(atoms) == 75 and report.partial is None + assert report.call_error.startswith("BadRequestError") + + +def test_the_faithfulness_check_stops_at_the_first_run_wide_error(): + atoms = _atoms(4) + transport = Calls('{"verdict": "supported"}', error=AUTH) + with pytest.raises(SpendCapReached) as stop: + verify_atoms(atoms, segment(ROWS), client=_client(transport)) + assert transport.n == 2 and stop.value.reason == "permanent_error" + assert atoms[0].faithfulness == "supported" and atoms[2].faithfulness == "unchecked" + + +def test_the_faithfulness_check_leaves_one_atom_unchecked_on_a_rejected_request(): + atoms = _atoms(4) + transport = Calls('{"verdict": "supported"}', error=TOO_LONG) + tally = verify_atoms(atoms, segment(ROWS), client=_client(transport)) + assert transport.n == 4 and tally == {"supported": 3, "distorted": 0, "unsupported": 0, + "unchecked": 1} + + +def test_ask_in_batches_stops_at_a_run_wide_error_and_says_why(): + transport = Calls("[1]", error=AUTH) + result = ask_in_batches(_client(transport), list(range(6)), size=2, + prompt=str, parse=json.loads) + assert transport.n == 2 + assert result.stop is not None and result.stop.reason == "permanent_error" + assert str(result.stop) == "permanent error: Unable to locate credentials" + assert result.answered == [([0, 1], [1])] and result.failed == [2, 3, 4, 5] + assert result.last_error == "" + + +def test_ask_in_batches_goes_on_after_a_rejected_request(): + transport = Calls("[1]", error=TOO_LONG) + result = ask_in_batches(_client(transport), list(range(6)), size=2, + prompt=str, parse=json.loads) + assert transport.n == 3 and result.stop is None + assert result.failed == [2, 3] and result.last_error.startswith("BadRequestError") + assert len(result.answered) == 2 + + +def test_the_value_judge_stops_at_a_run_wide_error(): + transport = Calls('[{"n": 1, "needed": true}]', error=AUTH) + with pytest.raises(SpendCapReached) as stop: + judge_value(_atoms(3), client=_client(transport), task="t", batch=1) + assert transport.n == 2 and stop.value.reason == "permanent_error" + + +def test_the_value_judge_leaves_one_batch_unjudged_on_a_rejected_request(): + transport = Calls('[{"n": 1, "needed": true}]', error=TOO_LONG) + verdicts = judge_value(_atoms(3), client=_client(transport), task="t", batch=1) + assert transport.n == 3 and len(verdicts) == 2 + + +def test_the_conflict_check_stops_at_a_run_wide_error(): + transport = Calls('{"conflicts": []}', error=AUTH) + with pytest.raises(SpendCapReached) as stop: + detect_conflicts(_atoms(6), client=_client(transport), batch=2) + assert transport.n == 2 and stop.value.reason == "permanent_error" + + +def test_the_conflict_check_leaves_one_batch_unchecked_on_a_rejected_request(): + transport = Calls('{"conflicts": []}', error=TOO_LONG) + detect_conflicts(_atoms(6), client=_client(transport), batch=2) + assert transport.n == 3 + + +def test_wording_generation_stops_at_a_run_wide_error_and_keeps_the_probes_built(): + transport = Calls('["Reworded?"]', error=AUTH) + built = ProbeSet(probes=[]) + with pytest.raises(SpendCapReached) as stop: + build_probe_set(_atoms(3), client=_client(transport), batch=1, into=built) + assert transport.n == 2 and stop.value.reason == "permanent_error" + assert [probe.kind for probe in built.probes[:3]] == ["direct"] * 3 + + +def test_wording_generation_counts_a_rejected_request_as_rejected_wordings(): + transport = Calls('["Reworded?"]', error=TOO_LONG) + build_probe_set(_atoms(3), client=_client(transport), batch=1) + assert transport.n == 6 # three paraphrase and three reversal calls + + +def test_the_retention_oracle_stops_at_a_run_wide_error(): + oracle = RetentionOracle(client=_client(Calls('{"verdict": "retained"}', error=AUTH)), + low=-1.0, high=2.0) + atoms = _atoms(4) + assert oracle.verdict(atoms[0], "some text").retained + with pytest.raises(SpendCapReached) as stop: + oracle.verdict(atoms[1], "some text") + assert stop.value.reason == "permanent_error" and oracle.errors == 0 + + +def test_the_retention_oracle_counts_a_rejected_request_as_one_error(): + transport = Calls('{"verdict": "retained"}', error=TOO_LONG) + oracle = RetentionOracle(client=_client(transport), low=-1.0, high=2.0) + for atom in _atoms(3): + oracle.verdict(atom, "some text") + assert transport.n == 3 and oracle.errors == 1 + + +def _probes(n: int) -> list[Probe]: + return [Probe(atom_id=f"a{i}", kind="direct", question=f"Q{i}?", expected="x") + for i in range(n)] + + +_ANSWER = "It is 10.\nconfidence: 0.9" +_GRADE = '{"verdict": "correct_complete"}' + + +def _measure(answer: Calls, grade: Calls, **kw): + return run_measurement([], _probes(4), "artifact text", answerer=_client(answer, "a"), + judge=_client(grade, "j"), concurrency=1, **kw) + + +def test_the_measurement_stops_at_a_run_wide_error_in_an_answer(): + answer, grade = Calls(_ANSWER, error=AUTH), Calls(_GRADE) + result = _measure(answer, grade) # 4 probes in 2 arms + assert answer.n == 2 and grade.n == 1 + assert len(result.runs) == 1 and result.not_run == 7 and result.ungraded == 0 + assert result.stop[0] == "permanent_error" and "credentials" in result.stop[1] + + +def test_the_measurement_records_a_rejected_answer_request_and_goes_on(): + answer, grade = Calls(_ANSWER, error=TOO_LONG), Calls(_GRADE) + result = _measure(answer, grade) + assert answer.n == 8 and len(result.runs) == 8 and result.not_run == 0 and result.stop is None + assert [run.verdict for run in result.runs].count(ERRORED) == 1 + + +def test_the_measurement_stops_at_a_run_wide_error_in_a_grade_and_keeps_the_answer(): + answer, grade = Calls(_ANSWER), Calls(_GRADE, error=AUTH, at=1) + result = _measure(answer, grade) + assert answer.n == 1 and grade.n == 1 + assert len(result.runs) == 1 and result.ungraded == 1 and result.not_run == 7 + assert result.runs[0].response == _ANSWER and result.runs[0].verdict == ERRORED + assert result.stop[0] == "permanent_error" + + +def test_the_measurement_records_a_rejected_grade_request_and_goes_on(): + answer, grade = Calls(_ANSWER), Calls(_GRADE, error=TOO_LONG, at=1) + result = _measure(answer, grade) + assert answer.n == 8 and len(result.runs) == 8 and result.stop is None and result.ungraded == 0 + assert [run.verdict for run in result.runs].count(ERRORED) == 1 + + +def test_analyze_keeps_a_partial_record_when_the_answerer_has_no_credentials(tmp_path): + from dokimi.novelty.analyze import AnalyzeConfig, analyze, load_artifact_text + from tests.test_novelty_analyze import SKILL, _clients, _scripted + + path = tmp_path / "accel-tuning" / "SKILL.md" + path.parent.mkdir(parents=True) + path.write_text(SKILL, encoding="utf-8") + kind, name, description, frontmatter, body = load_artifact_text(f"skill:{path}") + config = AnalyzeConfig(artifact_ref=f"skill:{path}", artifact_kind=kind, name=name, + description=description, frontmatter=frontmatter, concurrency=1) + extractor, _answerer, judge = _clients() + scripted = _scripted("informed") + answers = Calls(lambda prompt: scripted(prompt, None)[0], error=AUTH, at=3) + result = analyze(body, config, extractor=extractor, answerer=_client(answers, "answerer-m"), + judge=judge) + assert answers.n == 3 + assert result.partial["reason"] == "permanent_error" and result.partial["stage"] == "measurement" + assert result.partial["message"].endswith("permanent error: Unable to locate credentials") + assert "cause" not in result.partial + assert result.artifact["status"] == "partial" and len(result.artifact["runs"]) == 2 + + +def test_a_partial_file_of_a_provider_error_is_refused_with_advice_that_helps(tmp_path): + from dokimi.artifacts import load_artifact + from dokimi.novelty.atoms import load_atoms_record + from dokimi.utils import InputError + + block = {"reason": "permanent_error", "message": "permanent error: bad key"} + atoms, novelty = tmp_path / "atoms.partial.json", tmp_path / "novelty.partial.json" + atoms.write_text(json.dumps({"status": "partial", "partial": block}), encoding="utf-8") + novelty.write_text(json.dumps({"status": "partial", "partial": block}), encoding="utf-8") + with pytest.raises(InputError, match="permanent error: bad key, so it is not a finished atom set. Fix"): + load_atoms_record(atoms) + with pytest.raises(InputError, match="Fix the provider error and rerun") as refused: + load_artifact("novelty", novelty) + assert "--max-usd" not in str(refused.value) + + +def test_the_rerun_hint_of_a_provider_error_does_not_advise_a_higher_cap(monkeypatch, tmp_path): + from dokimi.caching import RESPONSE_CACHE_ENV + from dokimi.commands.novelty_budget import replay_hint + + monkeypatch.setenv(RESPONSE_CACHE_ENV, str(tmp_path)) + hint = replay_hint("permanent_error") + assert "Rerun with the error fixed, and the same" in hint and "--max-usd" not in hint From ee3e5863ce76527c6aaf655d752a6bc1efbb92e8 Mon Sep 17 00:00:00 2001 From: Anton Volkov Date: Sat, 14 Nov 2026 21:43:52 +0000 Subject: [PATCH 2/3] fix(novelty): a reply that quotes a credential message is no provider error; a late answer is not graded after a stop Settles the two review threads on #767. A reply that would not parse (a ValueError from the JSON extractors) quotes the reply in its message, so a model answering about "Unable to locate credentials" matched the no-credentials marker and stopped the run. is_run_wide now treats a ValueError as one failed item. With concurrency above 1, a worker whose answer came back after another worker recorded a permanent stop went on to grade it. run_measurement now keeps that answer ungraded, through the path a capped grade takes, and sends the judge nothing more. Signed-off-by: Anton Volkov --- .../20261114_130000_novelty-permanent-stop.md | 2 +- dokimi/dokimi/novelty/measure.py | 17 ++++++- dokimi/dokimi/novelty/routing_stop.py | 2 + dokimi/tests/test_novelty_permanent_stop.py | 51 +++++++++++++++++++ 4 files changed, 69 insertions(+), 3 deletions(-) diff --git a/dokimi/changelog.d/20261114_130000_novelty-permanent-stop.md b/dokimi/changelog.d/20261114_130000_novelty-permanent-stop.md index 95ac1a649..671252aa6 100644 --- a/dokimi/changelog.d/20261114_130000_novelty-permanent-stop.md +++ b/dokimi/changelog.d/20261114_130000_novelty-permanent-stop.md @@ -1,3 +1,3 @@ ### Fixed -- `skill novelty atoms`, `probe` and `analyze` and `skill shorten` stop at the first authentication, permission, unknown-model, unsupported-parameter or no-credentials error (extraction, the faithfulness check, probe wordings, answers and grades, the value and conflict batches, retention adjudication) instead of failing every remaining call and recording each one as failed; the partial result is kept as at the cap, with `reason` `permanent_error`. A refusal, an unparseable reply or a request that is too long is still one failed item. (#708) +- `skill novelty atoms`, `probe` and `analyze` and `skill shorten` stop at the first authentication, permission, unknown-model, unsupported-parameter or no-credentials error (extraction, the faithfulness check, probe wordings, answers and grades, the value and conflict batches, retention adjudication) instead of failing every remaining call and recording each one as failed; the partial result is kept as at the cap, with `reason` `permanent_error`. A refusal, an unparseable reply (also one that quotes a no-credentials message) or a request that is too long is still one failed item. With `--concurrency` above 1, an answer that returns after another worker met the error is kept ungraded, not sent to the judge. (#708) diff --git a/dokimi/dokimi/novelty/measure.py b/dokimi/dokimi/novelty/measure.py index 973fb3c9d..8bda764f6 100644 --- a/dokimi/dokimi/novelty/measure.py +++ b/dokimi/dokimi/novelty/measure.py @@ -387,7 +387,7 @@ def one(job: tuple[Probe, str]) -> ProbeRun: answered = False else: try: - verdict = _grade(probe, response, judge) + verdict = _grade_unless_stopped(stops, probe, response, judge) except SpendCapReached as exc: return _capped_grade_run(probe, arm, response, confidence, exc, stops, ungraded) run = ProbeRun( @@ -415,11 +415,24 @@ def guarded(job: tuple[Probe, str]) -> ProbeRun | None: probes=list(probes)) +def _held_permanent_stop(stops: list[SpendCapReached]) -> SpendCapReached | None: + """The first run-wide provider error among ``stops``, or None (#708).""" + return next((held for held in stops if held.reason == PERMANENT_REASON), None) + + +def _grade_unless_stopped(stops: list[SpendCapReached], probe: Probe, response: str, judge: LLMClient) -> str: + """:func:`_grade`, unless another worker met a run-wide error while this answer was in flight: + then that stop is raised and the caller keeps the answer ungraded, with no more judge calls (#708).""" + if (held := _held_permanent_stop(stops)) is not None: + raise held + return _grade(probe, response, judge) + + def _not_after_permanent_stop(stops: list[SpendCapReached], job: Callable[[Any], Any]) -> Callable[[Any], Any]: """``job``, which is not run (``None``) once a run-wide provider error is among ``stops`` (#708).""" def run(item: Any) -> Any: - return None if any(held.reason == PERMANENT_REASON for held in stops) else job(item) + return None if _held_permanent_stop(stops) else job(item) return run diff --git a/dokimi/dokimi/novelty/routing_stop.py b/dokimi/dokimi/novelty/routing_stop.py index 2bf26f7e1..8fda25390 100644 --- a/dokimi/dokimi/novelty/routing_stop.py +++ b/dokimi/dokimi/novelty/routing_stop.py @@ -21,6 +21,8 @@ def is_run_wide(exc: Exception) -> bool: """Whether *exc* is a permanent provider error that every later call would meet too.""" if is_retryable(exc): return False + if isinstance(exc, ValueError): # a reply that would not parse quotes the reply, which may name a credential + return False if isinstance(run_wide := getattr(exc, "run_wide", None), bool): # set by an error that knows (CliCallError) return run_wide return type(exc).__name__.lower() in _RUN_WIDE_ERRORS or any( diff --git a/dokimi/tests/test_novelty_permanent_stop.py b/dokimi/tests/test_novelty_permanent_stop.py index cfbe3067b..93ba703ee 100644 --- a/dokimi/tests/test_novelty_permanent_stop.py +++ b/dokimi/tests/test_novelty_permanent_stop.py @@ -228,6 +228,57 @@ def test_the_measurement_records_a_rejected_grade_request_and_goes_on(): assert [run.verdict for run in result.runs].count(ERRORED) == 1 +def test_a_reply_that_quotes_a_credential_message_is_not_a_provider_error(): + from dokimi.llm import extract_json_array + + replies = iter(["Unable to locate credentials: see the AWS docs.", "[1]", "[2]"]) + transport = Calls(lambda prompt: next(replies)) + result = ask_in_batches(_client(transport), [0, 1], size=2, prompt=str, parse=extract_json_array) + assert result.stop is None and result.failed == [] and len(result.answered) == 2 + + +def test_a_grade_reply_that_quotes_a_credential_message_is_one_errored_grade(): + answer = Calls(_ANSWER) + grade = Calls("Unable to locate credentials: see the AWS docs.") + result = _measure(answer, grade) + assert result.stop is None and result.not_run == 0 and len(result.runs) == 8 + assert {run.verdict for run in result.runs} == {ERRORED} + + +def test_a_worker_whose_answer_lands_after_a_run_wide_grade_error_does_not_grade(monkeypatch): + import threading + + from dokimi.novelty import measure + + stopped = threading.Event() + first = threading.Lock() + + second_started = threading.Event() + + def answer(prompt, system=None): + if first.acquire(blocking=False): # the first worker answers once the second is in flight + assert second_started.wait(10) + return _ANSWER, UsageRecord(cost_usd=0.0) + second_started.set() # the second worker answers only after the first one's stop is recorded + assert stopped.wait(10) + return _ANSWER, UsageRecord(cost_usd=0.0) + + grade = Calls(_GRADE, error=AUTH, at=1) + capped = measure._capped_grade_run + + def capped_then_signal(*args): + run = capped(*args) + stopped.set() + return run + + monkeypatch.setattr(measure, "_capped_grade_run", capped_then_signal) + result = run_measurement([], _probes(4), "artifact text", answerer=_client(answer, "a"), + judge=_client(grade, "j"), concurrency=2) + assert grade.n == 1 # the second worker's answer is kept, not graded + assert len(result.runs) == 2 and result.ungraded == 2 and result.not_run == 6 + assert result.stop[0] == "permanent_error" + + def test_analyze_keeps_a_partial_record_when_the_answerer_has_no_credentials(tmp_path): from dokimi.novelty.analyze import AnalyzeConfig, analyze, load_artifact_text from tests.test_novelty_analyze import SKILL, _clients, _scripted From f0e07a67bbda3e26778a92de8af1617fc4968798 Mon Sep 17 00:00:00 2001 From: vchamarthi <123653452+vchamarthi@users.noreply.github.com> Date: Sat, 14 Nov 2026 16:05:31 -0600 Subject: [PATCH 3/3] test(novelty): assert unchecked atoms and rejected wordings on a rejected request The conflict-check and wording-generation rejected-request tests checked only the call count. The conflict test now asserts the failed batch's atoms land in report.unchecked. The wording test replies with valid paraphrases and asserts that only the failed request is rejected and the other two are kept. Signed-off-by: vchamarthi <123653452+vchamarthi@users.noreply.github.com> --- dokimi/tests/test_novelty_permanent_stop.py | 21 ++++++++++++++++----- 1 file changed, 16 insertions(+), 5 deletions(-) diff --git a/dokimi/tests/test_novelty_permanent_stop.py b/dokimi/tests/test_novelty_permanent_stop.py index 93ba703ee..e3834c13a 100644 --- a/dokimi/tests/test_novelty_permanent_stop.py +++ b/dokimi/tests/test_novelty_permanent_stop.py @@ -15,7 +15,7 @@ from dokimi.metrics.usage import UsageRecord from dokimi.novelty.atoms import Atom, ExtractionReport, extract_atoms, verify_atoms from dokimi.novelty.batching import ask_in_batches -from dokimi.novelty.classify import detect_conflicts, judge_value +from dokimi.novelty.classify import ClassifyReport, detect_conflicts, judge_value from dokimi.novelty.client import LLMClient, SpendCapReached from dokimi.novelty.measure import ERRORED, run_measurement from dokimi.novelty.probes import Probe, ProbeSet, build_probe_set @@ -146,8 +146,10 @@ def test_the_conflict_check_stops_at_a_run_wide_error(): def test_the_conflict_check_leaves_one_batch_unchecked_on_a_rejected_request(): transport = Calls('{"conflicts": []}', error=TOO_LONG) - detect_conflicts(_atoms(6), client=_client(transport), batch=2) + report = ClassifyReport() + detect_conflicts(_atoms(6), client=_client(transport), batch=2, report=report) assert transport.n == 3 + assert report.unchecked == [a.id for a in _atoms(6)[2:4]] def test_wording_generation_stops_at_a_run_wide_error_and_keeps_the_probes_built(): @@ -160,9 +162,18 @@ def test_wording_generation_stops_at_a_run_wide_error_and_keeps_the_probes_built def test_wording_generation_counts_a_rejected_request_as_rejected_wordings(): - transport = Calls('["Reworded?"]', error=TOO_LONG) - build_probe_set(_atoms(3), client=_client(transport), batch=1) - assert transport.n == 6 # three paraphrase and three reversal calls + def reply(prompt): + asked = re.search(r"1\. What is setting (\d+)\?", prompt) + return f'["Tell me the value of setting {asked[1]}?"]' if asked else "[]" + + transport = Calls(reply, error=TOO_LONG) + built = build_probe_set(_atoms(3), client=_client(transport), batch=1, reversals=False) + assert transport.n == 3 + row = built.fidelity["paraphrase"] + assert (row["generated"], row["kept"], row["rejected"]) == (3, 2, 1) + assert row["reasons"] == {"empty": 1} + assert [p.atom_id for p in built.probes if p.kind == "paraphrase"] == [ + a.id for a in _atoms(3) if a.id != _atoms(3)[1].id] def test_the_retention_oracle_stops_at_a_run_wide_error():