diff --git a/dokimi/changelog.d/20261112_120000_analyze-cap-partial.md b/dokimi/changelog.d/20261112_120000_analyze-cap-partial.md
new file mode 100644
index 000000000..08386d158
--- /dev/null
+++ b/dokimi/changelog.d/20261112_120000_analyze-cap-partial.md
@@ -0,0 +1,3 @@
+### Fixed
+
+- **`skill novelty analyze` keeps what a `--max-usd` stop paid for** — a run stopped by the cap (or by a call with no known price) used to exit with `nothing was written`. It now writes `<out stem>.partial.json` beside `novelty.json` (never over a finished file) with `status: "partial"`, the stage it reached, the extracted atoms, the spans and the probe answers and judgments so far, and exits non-zero naming the file and how a rerun with `DOKIMI_RESPONSE_CACHE` replays the paid calls at $0. `novelty view`, `--novelty-artifact` and every other reader of a `novelty.json` refuse a partial file; the `novelty.v1` schema accepts it. A record with `status: "partial"` and no `partial` block is refused too, by the `novelty.json` readers and by `novelty items` on a probe file, a stop on a call with no known price tells the user to price the model or drop `--max-usd` (a higher cap does not help it), and the partial record keeps the generated probe wordings in `probes.items`, also when the cap falls between the paraphrase call and the reversal call. The replay promise is withheld when the cache directory is not writable or a cache write failed during the run (a full disk), and `dokimi runs show` reports a finished `novelty.json` that lacks `paired`, `k0` or `spans` as invalid, as the loaders do. The partial record counts `atoms.n` as a finished one does (the probed sample), a stop in extraction keeps `run_config` (its warnings stay empty, and `unread_span_ids` names the spans not read), and `save_artifact("novelty", ...)` refuses a record with `status` or `partial`. `skill shorten` still writes nothing at the cap.
diff --git a/dokimi/docs/skill-novelty.md b/dokimi/docs/skill-novelty.md
index 045a114eb..8376dc4af 100644
--- a/dokimi/docs/skill-novelty.md
+++ b/dokimi/docs/skill-novelty.md
@@ -846,8 +846,8 @@ between repeats keeps its picks but is left out of `noise`. The lexical arm is c
 Treat a `partial` file as a prefix of the run, not a measurement of the whole field: the skills
 and queries reached first are over-represented. To finish, raise `--max-usd` and rerun with
 `DOKIMI_RESPONSE_CACHE` set (it must also have been set on the capped run), which replays the
-calls already paid for at no cost. `skill novelty analyze` and `skill shorten` still write nothing
-when the cap stops them; `atoms` and `probe` keep the calls they made as a partial result.
+calls already paid for at no cost. `skill shorten` still writes nothing when the cap stops it;
+`atoms`, `probe` and `analyze` keep the calls they made as a partial result.
 
 Three properties worth knowing before quoting either:
 
@@ -1110,7 +1110,7 @@ from the network before it falls back to the copy bundled with the package; on a
 with no network that is a pause of a few seconds and a warning. Setting
 `LITELLM_LOCAL_MODEL_COST_MAP=True` skips the fetch and uses the bundled prices.
 
-**Spend cap on `atoms` and `probe`.** `--max-usd` stops these at the limit and keeps the work it
+**Spend cap on `atoms`, `probe` and `analyze`.** `--max-usd` stops these at the limit and keeps the work it
 paid for. `atoms` writes `<stem>.partial.json` beside the `--out` file (`atoms.json` by default;
 the spans read so far, the rest marked unread), and `probe` does the same beside its `--out`
 file (`probes.json` by default; the answers so far). Both exit with a message naming the file. A partial file is never read as a
@@ -1120,7 +1120,29 @@ the summaries computed over the answers the cap left (`paired`, `arms`, `abstent
 confidence blocks, and `judge2`: the second judge shares the cap, so it cannot regrade at all once
 the cap stops the run).
 A partial file is for inspection; rerun to get a finished one.
-`analyze` and `shorten` still lose their work at the cap (a follow-up). Before its first paid
+`analyze` does the same beside its `novelty.json` (`<out stem>.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`
+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`
+(the unread spans are in `extraction.unread_span_ids`), the `probes` once built (with the
+wordings and answer keys in `probes.items`, so a stop before the first answer still keeps them) and
+the `runs` (probe answers and judgments) so far; it has no `paired`, `arms`, `k0` or view summaries, and no
+`novelty.md` or annotated view is written. The exit is non-zero (1), the message names the file and
+how to get the paid calls back: with `DOKIMI_RESPONSE_CACHE` set to a writable directory on the capped run, a rerun with a
+higher `--max-usd` and the same cache replays them at $0 (a reply that was cut off or unusable is
+asked again; after a `no_price` stop the message says to price the model or drop `--max-usd`
+instead, since a higher cap does not help); without it, or when it could not be written, the message says the calls were not cached. `novelty view`, `subjects`
+(`--novelty-artifact`) and every other reader of a `novelty.json` refuse a partial file by name,
+and `runs list` does not list it (it opens with `status`, not `schema_version`). A record
+with `status: "partial"` but no `partial` block is refused too, and so is saving one with
+`save_artifact("novelty", ...)`. Its `atoms.n` counts the probed sample once the cap's sample is
+chosen, as a finished record does, and every extracted atom for a stop in extraction (`atoms.sampling`
+is then `none`, and `run_config` holds the settings of the run but not its tier). A library caller of
+`dokimi.novelty.analyze()` checks `result.partial` before it saves `result.artifact`.
+`shorten` still loses its work at the cap (a follow-up). Before its first paid
 call `analyze` prints the run's expected cost at list prices to stderr.
 
 The estimate payload has `expected_calls`, the point figure between
diff --git a/dokimi/dokimi/artifacts.py b/dokimi/dokimi/artifacts.py
index 756845455..beb48fde3 100644
--- a/dokimi/dokimi/artifacts.py
+++ b/dokimi/dokimi/artifacts.py
@@ -212,9 +212,19 @@ def schema_errors(schema: str | Path, instance: Any) -> list[str]:
     ]
 
 
+def summary_errors(kind: str, data: dict) -> list[str]:
+    """What the schema file of ``kind`` leaves unchecked: a finished ``novelty`` record (one without
+    ``status``) needs its ``paired``, ``k0`` and ``spans`` summaries, which only a partial record lacks."""
+    if kind != "novelty" or "status" in data:
+        return []
+    return [f"<root>: {name!r} is a required property"
+            for name in ("paired", "k0", "spans") if name not in data]
+
+
 def validate_artifact(kind: str, data: dict) -> None:
     """Validate ``data`` against the schema for ``kind`` (raises on failure)."""
     errors = schema_errors(_schema_path(kind, schema_version_for(kind, data)), data)
+    errors += summary_errors(kind, data)
     if errors:
         details = "\n".join(f"  - {e}" for e in errors)
         raise ArtifactValidationError(f"{kind} artifact validation failed:\n{details}")
@@ -238,6 +248,9 @@ def save_artifact(kind: str, data: dict, path: Path | str, *, ensure_ascii: bool
       input kind. Values already in ``data`` under these keys are replaced: the file records the
       process that last saved it, so an artifact built from a loaded one takes the new run's id.
       ``data`` itself is not mutated.
+    - A ``novelty`` payload with ``status`` or ``partial`` (what a ``--max-usd`` stop keeps) is
+      refused with :class:`~dokimi.utils.InputError`, as :func:`load_artifact` refuses it: nothing
+      is written, since ``runs`` would list the file as a finished run.
     - The payload is validated first. :class:`ArtifactValidationError` means nothing was
       written at ``path`` (a file already there is untouched); the rejected payload is kept
       at ``<name>.rejected`` so a paid run is not lost to a schema problem. A later valid
@@ -261,6 +274,10 @@ def save_artifact(kind: str, data: dict, path: Path | str, *, ensure_ascii: bool
     """
     if kind not in SCHEMA_VERSIONS:
         raise KeyError(f"Unknown artifact kind: {kind!r}")
+    if _is_partial_novelty(kind, data):  # `runs` would list it as a finished run
+        raise InputError(f"not written to {path}: a partial novelty record (`status` or `partial` "
+                         "set) is not a finished analysis; `skill novelty analyze` keeps it in "
+                         "<out stem>.partial.json")
     path = Path(path)
     path.parent.mkdir(parents=True, exist_ok=True)
     try:
@@ -306,6 +323,12 @@ def _write_target(path: Path) -> Path:
     return target
 
 
+def _is_partial_novelty(kind: str, data: dict) -> bool:
+    """Whether ``data`` is what `skill novelty analyze` keeps when --max-usd stops it: not a finished
+    analysis. Either marker alone is enough: the schema lets a record carry one without the other."""
+    return kind == "novelty" and ("partial" in data or data.get("status") == "partial")
+
+
 def load_artifact(kind: str, path: Path | str, *, validate: bool = True) -> dict:
     """Read and (optionally) validate an artifact of ``kind`` from ``path``.
 
@@ -315,6 +338,13 @@ def load_artifact(kind: str, path: Path | str, *, validate: bool = True) -> dict
     data = read_json_input(path)
     if not isinstance(data, dict):
         raise InputError(f"{path} is not a {kind} artifact: expected a JSON object, got {type(data).__name__}")
+    if _is_partial_novelty(kind, data):
+        block = data.get("partial")
+        reason = block.get("reason") if isinstance(block, dict) else None
+        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
+                             "Rerun with a higher --max-usd, or without the cap"))
     if validate:
         try:
             validate_artifact(kind, data)
diff --git a/dokimi/dokimi/caching.py b/dokimi/dokimi/caching.py
index 7bd315c30..2214113db 100644
--- a/dokimi/dokimi/caching.py
+++ b/dokimi/dokimi/caching.py
@@ -455,4 +455,16 @@ def store_response(key: str, text: str, usage: UsageRecord) -> None:
         # half-written entry.
         tmp.replace(directory / f"{key}.json")
     except OSError as exc:
+        _failed_stores.add(directory)
         logger.warning("Could not write response cache entry: %s", exc)
+
+
+#: Cache directories a write failed in during this process (a full disk, a quota): a directory
+#: that passes ``os.access`` can still have kept nothing.
+_failed_stores: set[Path] = set()
+
+
+def response_store_failed(directory: Path) -> bool:
+    """Whether a :func:`store_response` write into ``directory`` failed in this process, so some
+    call it paid for may not be in the cache."""
+    return directory in _failed_stores
diff --git a/dokimi/dokimi/commands/novelty.py b/dokimi/dokimi/commands/novelty.py
index df0d5fb36..04710075e 100644
--- a/dokimi/dokimi/commands/novelty.py
+++ b/dokimi/dokimi/commands/novelty.py
@@ -19,7 +19,7 @@
 
 from dokimi.commands.inputs import placebo_text_for_arms_or_exit
 from dokimi.commands.novelty_run import (add_judge2_args, add_spend_cap, judge2_client,
-                                         stamp, stops_at_cap)
+                                         exit_if_analyze_partial, stamp, stops_at_cap)
 from dokimi.commands.novelty_routing_inputs import (
     checked_out_path,
     load_all_vs_all_field,
@@ -139,10 +139,10 @@ def cmd_novelty_analyze(args: argparse.Namespace) -> None:
         raise InputError(str(exc)) from exc
     except (ExtractionError, ValueError) as exc:  # a model failure under way
         raise SystemExit(str(exc)) from None
-
-    stamp(result.artifact, args, result.measure_result, answerer=answerer, judge=judge, judge2=judge2)
     out_dir = Path(args.out_dir) if args.out_dir else _default_dir(args.artifact)
     out_json = Path(args.out_json) if args.out_json else out_dir / "novelty.json"
+    exit_if_analyze_partial(result, out_json)  # a run the cap stopped keeps what it paid for
+    stamp(result.artifact, args, result.measure_result, answerer=answerer, judge=judge, judge2=judge2)
     save_artifact("novelty", result.artifact, out_json)
     out_md = out_json.with_suffix(".md")
     out_md.write_text(render_novelty_markdown(result.artifact), encoding="utf-8")
diff --git a/dokimi/dokimi/commands/novelty_budget.py b/dokimi/dokimi/commands/novelty_budget.py
index d67c2d616..3b04d78da 100644
--- a/dokimi/dokimi/commands/novelty_budget.py
+++ b/dokimi/dokimi/commands/novelty_budget.py
@@ -3,15 +3,16 @@
 
 """What a run that ``--max-usd`` stops leaves behind, and what a run is expected to cost.
 
-A stopped atoms or probe run is kept as ``<out stem>.partial.json``, beside the file it would
-have written: a partial result never replaces a finished one, and the readers that take an
-atoms file refuse a partial one.
+A stopped atoms, probe or analyze run is kept as ``<out stem>.partial.json``, beside the file it
+would have written: a partial result never replaces a finished one, and the readers that take an
+atoms file, a probe file or a ``novelty.json`` refuse a partial one.
 """
 
 from __future__ import annotations
 
 import argparse
 import json
+import os
 import sys
 from collections.abc import Callable
 from pathlib import Path
@@ -42,9 +43,48 @@ def partial_write(out: Path, write: Callable[[Path], object]) -> Path:
     raise RuntimeError("unreachable: the sidecar names never run out")
 
 
-def exit_partial(summary: str, path: Path) -> NoReturn:
+def exit_partial(summary: str, path: Path, note: str = "") -> NoReturn:
     """Stop with a message that names the partial file, as a run the cap stopped must."""
-    raise SystemExit(f"{summary}; the partial result is in {path}")
+    raise SystemExit(f"{summary}; the partial result is in {path}{note}")
+
+
+def replay_hint(reason: str = "spend_cap") -> str:
+    """How a rerun gets the calls this run paid for back at $0: the response cache, which must
+    have been set on the run that paid (every novelty call goes through it). A higher
+    ``--max-usd`` is advised only for a ``spend_cap`` stop: a call with no price is refused
+    whatever the cap, so that stop is cured by a price or by dropping the cap."""
+    from dokimi.caching import RESPONSE_CACHE_ENV, response_cache_dir, response_store_failed
+
+    try:
+        cache = response_cache_dir()
+    except ValueError:  # the variable names a file: no call was cached
+        cache = None
+    # Every paid call has gone through the cache by now: a dir missing or read-only kept none,
+    # and one whose write failed (a full disk) may have lost some.
+    active = (cache is not None and os.access(cache, os.W_OK)
+              and not response_store_failed(cache))
+    if active:
+        rerun = ("a higher --max-usd" if reason == "spend_cap"
+                 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)")
+    return (f". {RESPONSE_CACHE_ENV} was not set on this run or could not be written, so a "
+            "rerun pays for these calls again; set it to a writable directory before the next "
+            "run to replay them at $0")
+
+
+def save_analyze_partial(result, out: Path) -> NoReturn:
+    """Keep what a stopped ``analyze`` paid for beside ``out`` and exit, naming the file.
+
+    ``result`` is a :class:`~dokimi.novelty.analyze.AnalyzeResult` with ``partial`` set. The
+    record goes to the sidecar rules of ``atoms`` and ``probe`` (never over a finished
+    ``novelty.json``).
+    """
+    partial = result.partial
+    cause = partial.get("cause")
+    summary = f"{partial['message']} ({cause})" if cause else partial["message"]
+    exit_partial(summary, write_run(result.artifact, out, partial),
+                 replay_hint(partial["reason"]))
 
 
 def write_run(payload: dict, out: Path, partial: dict | None = None) -> Path:
diff --git a/dokimi/dokimi/commands/novelty_run.py b/dokimi/dokimi/commands/novelty_run.py
index 13d2ed072..2d2bd1b56 100644
--- a/dokimi/dokimi/commands/novelty_run.py
+++ b/dokimi/dokimi/commands/novelty_run.py
@@ -22,12 +22,20 @@ def run(args: argparse.Namespace) -> None:
         try:
             command(args)
         except SpendCapReached as exc:
-            # A command that keeps its partial result catches the cap itself and never gets here.
+            # A command that keeps its partial result (atoms, probe, analyze) saves it itself.
             raise SystemExit(f"{exc}; nothing was written") from None
 
     return run
 
 
+def exit_if_analyze_partial(result, out) -> None:
+    """Save and exit when the cap stopped ``analyze`` (``result.partial``); a finished run returns."""
+    from dokimi.commands.novelty_budget import save_analyze_partial
+
+    if result.partial:
+        save_analyze_partial(result, out)
+
+
 def usd_amount(text: str) -> float:
     """A positive, finite USD amount (``0``, negatives, ``nan`` and ``inf`` would disable the cap)."""
     value = float(text)
diff --git a/dokimi/dokimi/novelty/analyze.py b/dokimi/dokimi/novelty/analyze.py
index 9d6579c2c..3b08bd3c4 100644
--- a/dokimi/dokimi/novelty/analyze.py
+++ b/dokimi/dokimi/novelty/analyze.py
@@ -25,6 +25,7 @@
 from . import classify, items as items_mod
 from . import measure, needs, probes, provenance, report, retention, shorten, spans
 from .artifact import load_artifact_text  # noqa: F401  (public path: novelty.analyze)
+from .analyze_partial import Progress, stopped_record
 from .client import LLMClient, SpendCapReached, roll_up_cost, same_client, warn_extractor_judge
 from .retention import RetentionOracle
 from .tiers import DEFAULT_SETTINGS, RunSettings
@@ -108,6 +109,9 @@ class AnalyzeResult:
     spans: list = field(default_factory=list)
     measure_result: object | None = None
     extracted_atoms: list = field(default_factory=list)
+    #: Set when ``--max-usd`` (or a call with no price) stopped the run: ``artifact`` is then the
+    #: partial record (:mod:`.analyze_partial`) and this is its ``partial`` block.
+    partial: dict | None = None
 
 
 def kind_agreement_warnings(rows: dict[str, dict]) -> list[str]:
@@ -135,13 +139,12 @@ def kind_agreement_warnings(rows: dict[str, dict]) -> list[str]:
 
 
 def _refuse_cut_short(measure_result: measure.MeasureResult) -> None:
-    """A measurement the cap cut short is not a finished report, and analyze has no partial
-    result to keep it in: stop before the paid classify and attribution stages."""
+    """A measurement the cap cut short is not a finished report: stop before the paid classify
+    and attribution stages. ``analyze`` keeps what was paid for as a partial record."""
     if measure_result.not_run or measure_result.ungraded:
         # An answer the judge never graded counts as not finished, like one never asked.
         not_run = measure_result.not_run + measure_result.ungraded
         reason, cause = measure_result.stop or ("spend_cap", "")
-        # stops_at_cap adds "nothing was written" to the exit, once.
         if reason == "spend_cap":
             # The cap's own message (the cause) names the spend and the advice: say it once.
             detail = cause or ("spend cap reached. Raise --max-usd, or measure less (a cheaper "
@@ -156,6 +159,13 @@ def _refuse_cut_short(measure_result: measure.MeasureResult) -> None:
         )
 
 
+def _raise_if_extraction_stopped(extraction: atoms_mod.ExtractionReport) -> None:
+    """A cap that stopped extraction left a ``partial`` record on the report: stop here."""
+    stopped = extraction.partial
+    if stopped:
+        raise SpendCapReached(stopped["message"], reason=stopped["reason"])
+
+
 def analyze(
     text: str,
     config: AnalyzeConfig,
@@ -169,6 +179,13 @@ def analyze(
 ) -> AnalyzeResult:
     """Run the full pipeline over one artifact's text.
 
+    A spend cap (or a call with no price) that stops it does not lose what it paid for: the
+    result then has ``partial`` set and ``artifact`` is the partial record
+    (:func:`~dokimi.novelty.analyze_partial.stopped_record`): the extracted atoms, the spans, the
+    probe set once built and the probe answers and judgments so far, and no summary over them.
+    A caller must check ``result.partial`` before it saves or reports ``artifact`` as a finished
+    analysis: :func:`~dokimi.artifacts.save_artifact` refuses a partial ``novelty`` record.
+
     Args:
         text: The artifact body (frontmatter already split off into ``config``).
         config: What to run.
@@ -187,17 +204,55 @@ def analyze(
             file it came from, which is the cut an author needs to decide what belongs in
             the body. Without it the run is entry-point-only and says so.
     """
+    progress = Progress()
+    try:
+        return _analyze(text, config, extractor=extractor, answerer=answerer, judge=judge,
+                        ref_digest=ref_digest, artifact_bundle=artifact_bundle, tier=tier,
+                        progress=progress)
+    except SpendCapReached as stop:
+        if ref_digest is None and artifact_bundle is not None:
+            ref_digest = artifact_bundle.measured_digest()
+        record = stopped_record(
+            progress, stop, artifact_kind=config.artifact_kind, artifact_ref=config.artifact_ref,
+            ref_digest=ref_digest, model=answerer.model, provider=answerer.provider,
+            judge_model=judge.model, clients=[extractor, answerer, judge],
+            provenance=provenance.provenance(
+                artifact_digest=ref_digest, answerer_model=answerer.model,
+                judge_model=judge.model, context_framing=config.context_framing),
+            bundle=artifact_bundle.summary() if artifact_bundle is not None else {})
+        return AnalyzeResult(
+            artifact=record, atoms=progress.atoms, spans=progress.spans,
+            measure_result=progress.measure_result, extracted_atoms=progress.extracted,
+            partial=record["partial"])
+
+
+def _analyze(
+    text: str,
+    config: AnalyzeConfig,
+    *,
+    extractor: LLMClient,
+    answerer: LLMClient,
+    judge: LLMClient,
+    ref_digest: str | None,
+    artifact_bundle: "bundle_mod.Bundle | None",
+    tier: dict | None,
+    progress: Progress,
+) -> AnalyzeResult:
+    """The pipeline of :func:`analyze`; ``progress`` records each stage as it starts and what it
+    produced, so a spend stop anywhere can be saved."""
     measure.check_self_judge(judge, answerer, config.allow_self_judge)
     relation = warn_extractor_judge(extractor.model, judge, extractor.provider)
-    warnings: list[str] = []
-    span_list = spans.segment(text)
-    extraction = atoms_mod.ExtractionReport()
-    extracted = atoms_mod.extract_atoms(
+    warnings: list[str] = progress.warnings
+    span_list = progress.spans = spans.segment(text)
+    extraction = progress.extraction = atoms_mod.ExtractionReport()
+    extracted = progress.extracted = atoms_mod.extract_atoms(
         span_list, client=extractor, cap=config.max_atoms, report=extraction
     )
+    progress.run_config = _run_config(config)  # a stop in extraction keeps this much of it
+    _raise_if_extraction_stopped(extraction)
     # Spans, attribution and shortening's structure index every atom; the paid stages see the
     # cap's sample: https://github.com/intel/dokimi/issues/371
-    atom_list = atoms_mod.sample_atoms(extracted, span_list, config.max_atoms)
+    atom_list = progress.atoms = atoms_mod.sample_atoms(extracted, span_list, config.max_atoms)
     warnings.extend(extraction.warnings(len(span_list)))
     settings = RunSettings.of(config)
     if tier:
@@ -209,7 +264,9 @@ def analyze(
     warnings.extend(settings.warnings(sampled=capped))
     # A label for a sample only when the cap left atoms out; a cap the atoms meet is a full run.
     # Older runs recorded `cap-per-artifact:<n>`, the first n atoms in document order (#293).
-    sampling = f"cap-spread:{config.max_atoms}" if capped else "full"
+    sampling = progress.sampling = f"cap-spread:{config.max_atoms}" if capped else "full"
+    progress.run_config = {**_run_config(config), "tier": settings.tier_record(sampled=capped),
+                           "extractor_judge": relation}
     if extraction.sampled:  # set only when the cap left atoms out
         warnings.append(report.cap_warning(extraction.as_dict()))
     if not atom_list:
@@ -222,8 +279,10 @@ def analyze(
             "`faithfulness` is empty, so distortion by the extractor is unbounded for "
             "this run"
         )
+    progress.stage = "verify"
     if config.verify_faithfulness and atom_list:
-        faithfulness = atoms_mod.verify_atoms(atom_list, span_list, client=extractor)
+        faithfulness = progress.faithfulness = atoms_mod.verify_atoms(
+            atom_list, span_list, client=extractor)
         distorted = faithfulness.get("distorted", 0) + faithfulness.get("unsupported", 0)
         if distorted:
             warnings.append(
@@ -231,6 +290,7 @@ def analyze(
                 "stay in the set and are visible in atoms.faithfulness"
             )
 
+    progress.stage = "need_graph"
     graph = needs.build_need_graph(
         config.name or config.artifact_ref, config.description, client=extractor
     )
@@ -241,9 +301,12 @@ def analyze(
         )
     relevance = needs.match_atoms(atom_list, graph)
 
+    progress.stage = "probes"
+    progress.probe_set = probes.ProbeSet(probes=[])
     probe_set = probes.build_probe_set(
         atom_list, client=extractor,
         paraphrases=config.paraphrases, reversals=config.reversals,
+        into=progress.probe_set,
     )
     probe_list = probe_set.probes
     # A generated wording that lost the version bound, or a "reversal" that never
@@ -258,13 +321,15 @@ def analyze(
                 f"({', '.join(sorted(row['reasons']))}); {kind} probes cover "
                 f"{row['kept']} of {len(atom_list)} atoms"
             )
-    measure_result = measure.run_measurement(
+    progress.stage = "measurement"
+    measure_result = progress.measure_result = measure.run_measurement(
         atom_list, probe_list, text,
         answerer=answerer, judge=judge, arms=config.arms,
         concurrency=config.concurrency, allow_self_judge=config.allow_self_judge,
         placebo_text=config.placebo_text, seed=config.seed, framing=config.context_framing,
     )
     _refuse_cut_short(measure_result)
+    progress.stage = "classify"
     if config.allow_self_judge and same_client(judge, answerer):
         warnings.append(
             "judge and answerer are the same model (--allow-self-judge): scores are inflated"
@@ -305,6 +370,7 @@ def analyze(
     elif not config.classify_content:
         warnings.append("content categories were not judged (--no-classify)")
 
+    progress.stage = "attribution"
     attributions = attribute.attribute_spans(
         span_list, extracted, outcomes=outcomes, categories=categories
     )
diff --git a/dokimi/dokimi/novelty/analyze_partial.py b/dokimi/dokimi/novelty/analyze_partial.py
new file mode 100644
index 000000000..2f54c9af8
--- /dev/null
+++ b/dokimi/dokimi/novelty/analyze_partial.py
@@ -0,0 +1,111 @@
+# Copyright (C) 2026 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+"""What ``skill novelty analyze`` keeps when ``--max-usd`` (or a call with no price) stops it.
+
+:func:`stopped_record` builds a partial ``novelty.v1`` record from the stages that finished: the
+extracted atoms and the spans always, the probe set once it was built, and the probe answers and
+judgments made so far. It holds none of the summaries over the runs (``paired``, ``arms``, the
+views): the cut leaves the arms uneven, so they would read as results next to ``status: partial``.
+The record opens with ``status``, so ``runs`` does not list it as a finished run, and every reader
+of a finished ``novelty.json`` refuses it (:func:`dokimi.artifacts.load_artifact`).
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+
+from .atoms import ExtractionReport
+from .client import SpendCapReached, roll_up_cost
+from .measure import MeasureResult
+from .probes import ProbeSet
+
+#: Where the cut fell, as the partial record names it, and how the message says it.
+STAGES = {
+    "extraction": "extraction",
+    "verify": "the faithfulness check",
+    "need_graph": "the need graph",
+    "probes": "the probe wordings",
+    "measurement": "the probes",
+    "classify": "the content classification",
+    "attribution": "the span attribution and shortening",
+}
+
+
+@dataclass
+class Progress:
+    """What an analysis had produced when it stopped; filled in by ``analyze`` as it goes."""
+
+    stage: str = "extraction"
+    spans: list = field(default_factory=list)
+    extraction: ExtractionReport | None = None
+    extracted: list = field(default_factory=list)
+    atoms: list = field(default_factory=list)
+    sampling: str = "none"  # until the run has chosen the atoms it probes
+    run_config: dict = field(default_factory=dict)
+    warnings: list = field(default_factory=list)
+    faithfulness: dict | None = None
+    probe_set: ProbeSet | None = None
+    measure_result: MeasureResult | None = None
+
+
+def stop_block(progress: Progress, stop: SpendCapReached) -> dict:
+    """The ``partial`` block: ``reason``, the ``stage`` reached, ``calls_done`` / ``calls_planned``
+    where the stage counts calls (the probes) or the spans read (extraction), and ``message``."""
+    extraction = progress.extraction
+    if progress.stage == "extraction" and extraction is not None and extraction.partial:
+        return dict(extraction.partial)  # spans_read / spans_total, as `atoms` records them
+    stage = STAGES[progress.stage]
+    block: dict = {"reason": stop.reason, "stage": progress.stage}
+    result = progress.measure_result
+    where = stage
+    if result is not None:  # the measurement started: its answers are kept, whatever stage stopped
+        done = len(result.runs)
+        block.update(calls_done=done, calls_planned=done + result.not_run)
+        left = f"; {result.ungraded} of them not graded" if result.ungraded else ""
+        if progress.stage == "measurement":
+            where = f"the probes with {done} of {done + result.not_run} answers{left}"
+    block["message"] = (f"--max-usd stopped {where}" if stop.reason == "spend_cap"
+                        else f"analyze stopped at {where}: {stop}")
+    if stop.reason == "spend_cap":  # what was spent and what to do (a no-price message is the message)
+        block["cause"] = str(stop)
+    return block
+
+
+def stopped_record(progress: Progress, stop: SpendCapReached, *, artifact_kind: str,
+                   artifact_ref: str, ref_digest: str | None, model: str, provider: str,
+                   judge_model: str, clients: list, provenance: dict, bundle: dict) -> dict:
+    """The partial record: it validates as ``novelty.v1`` and says ``status: partial``."""
+    result = progress.measure_result
+    extraction = progress.extraction
+    return {
+        "status": "partial",
+        "partial": stop_block(progress, stop),
+        "schema_version": "novelty.v1",
+        "generated_at": datetime.now(timezone.utc).replace(tzinfo=None).isoformat() + "Z",
+        "artifact": {"kind": artifact_kind, "ref": artifact_ref, "ref_digest": ref_digest,
+                     "tokens": sum(span.tokens for span in progress.spans)},
+        "model": model,
+        "provider": provider,
+        "judge_model": judge_model,
+        "atoms": {
+            # The probed sample once it is chosen, as a finished record counts; before, all extracted.
+            "n": len(progress.extracted if progress.sampling == "none" else progress.atoms),
+            "n_probed": len({row.atom_id for row in result.runs}) if result else 0,
+            "sampling": progress.sampling,
+            "faithfulness": progress.faithfulness or {},
+        },
+        "bundle": bundle,
+        "run_config": progress.run_config,
+        "provenance": provenance,
+        "extraction": extraction.as_dict() if extraction is not None else {},
+        "cost": roll_up_cost(clients),
+        "warnings": sorted(set(progress.warnings)),
+        "extracted_atoms": [atom.as_dict() for atom in progress.extracted],
+        "source_spans": [span.as_dict() for span in progress.spans],
+        **({"probes": {**progress.probe_set.as_dict(),
+                       "items": [probe.as_dict() for probe in progress.probe_set.probes]}}
+           if progress.probe_set is not None else {}),
+        **({"runs": [row.as_dict() for row in result.runs]} if result is not None else {}),
+    }
diff --git a/dokimi/dokimi/novelty/atoms.py b/dokimi/dokimi/novelty/atoms.py
index 8ac192067..9c579e5bb 100644
--- a/dokimi/dokimi/novelty/atoms.py
+++ b/dokimi/dokimi/novelty/atoms.py
@@ -320,8 +320,7 @@ def _unread_at_cap(stop: SpendCapReached, batch: list[Span], parse: Callable[[st
 
 def _stop_at_cap(spans: list[Span], batch: list[Span], pending: list, report: ExtractionReport,
                  stop: SpendCapReached) -> None:
-    """The cap ends extraction: ``batch`` and the batches queued behind it go unread, and the
-    atoms read so far stand (the caller saves them as a partial result)."""
+    """The cap ends extraction: ``batch`` and those queued go unread; the atoms read stand."""
     report.pre_cap = (len(report.unread_span_ids), report.unread_tokens)
     for rest in [batch, *(queued for queued, _is_half in pending)]:
         _mark_unread(rest, report)
@@ -331,7 +330,8 @@ def _stop_at_cap(spans: list[Span], batch: list[Span], pending: list, report: Ex
                if stop.reason == "spend_cap" else
                f"extraction stopped with {read} of {len(spans)} spans read: {stop}")
     report.partial = {"reason": stop.reason, "stage": "extraction", "spans_read": read,
-                      "spans_total": len(spans), "message": message}
+                      "spans_total": len(spans), "message": message,
+                      **({"cause": str(stop)} if stop.reason == "spend_cap" else {})}
 
 
 def _keep_atoms(rows: list, valid_ids: set[str], seen: set[str], atoms: list[Atom]) -> None:
diff --git a/dokimi/dokimi/novelty/items.py b/dokimi/dokimi/novelty/items.py
index a2cdcaafb..e67a853e2 100644
--- a/dokimi/dokimi/novelty/items.py
+++ b/dokimi/dokimi/novelty/items.py
@@ -268,7 +268,7 @@ def _check_run_strings(raw: str, row: dict) -> None:
 
 def _refuse_partial(raw: str, data: dict) -> None:
     """A partial probe result (``--max-usd`` stopped the measurement) is not a finished one."""
-    if "partial" in data:
+    if "partial" in data or data.get("status") == "partial":
         raise ProbeFileError(f"{raw} is a partial probe result (--max-usd stopped the measurement); "
                              "it is not a finished measurement")
 
diff --git a/dokimi/dokimi/novelty/probes.py b/dokimi/dokimi/novelty/probes.py
index e0715991f..903c07330 100644
--- a/dokimi/dokimi/novelty/probes.py
+++ b/dokimi/dokimi/novelty/probes.py
@@ -34,10 +34,11 @@
 from __future__ import annotations
 
 import re
+from collections.abc import Iterator
 from dataclasses import dataclass, field
 
 from .atoms import Atom
-from .client import LLMClient
+from .client import LLMClient, SpendCapReached
 
 PROBE_KINDS = ("direct", "paraphrase", "reversal")
 
@@ -384,6 +385,7 @@ def build_probe_set(
     paraphrases: int = 1,
     reversals: bool = True,
     batch: int = 20,
+    into: ProbeSet | None = None,
 ) -> ProbeSet:
     """Build the probe set for a list of atoms.
 
@@ -396,15 +398,19 @@ def build_probe_set(
             generator returns one alternative per call).
         reversals: Whether to add the planted-inversion probe.
         batch: Atoms per wording-generation call.
+        into: An empty set to fill in place and return, so a caller whose generator raises
+            part-way (a spend cap) still holds the probes accepted so far.
     """
-    probes: list[Probe] = [
+    built = into if into is not None else ProbeSet(probes=[])
+    probes = built.probes
+    probes.extend(
         Probe(atom_id=a.id, kind="direct", question=a.question, expected=a.answer)
         for a in atoms
-    ]
+    )
     if client is None:
-        return ProbeSet(probes=probes)
+        return built
 
-    fidelity: dict[str, dict] = {}
+    fidelity = built.fidelity
 
     def tally(kind: str, reasons: list[str]) -> bool:
         row = fidelity.setdefault(
@@ -420,8 +426,7 @@ def tally(kind: str, reasons: list[str]) -> bool:
         return True
 
     if paraphrases:
-        for chunk in _chunks(atoms, batch):
-            worded = _generate(client, _PARAPHRASE_PROMPT, [a.question for a in chunk])
+        for chunk, worded in _worded(client, _PARAPHRASE_PROMPT, atoms, batch, "question"):
             for atom, question in zip(chunk, worded, strict=False):
                 reasons = check_wording(
                     "paraphrase", source=atom.question, candidate=question,
@@ -431,8 +436,7 @@ def tally(kind: str, reasons: list[str]) -> bool:
                     probes.append(Probe(atom_id=atom.id, kind="paraphrase",
                                         question=question.strip(), expected=atom.answer))
     if reversals:
-        for chunk in _chunks(atoms, batch):
-            worded = _generate(client, _REVERSAL_PROMPT, [a.statement for a in chunk])
+        for chunk, worded in _worded(client, _REVERSAL_PROMPT, atoms, batch, "statement"):
             for atom, question in zip(chunk, worded, strict=False):
                 reasons = check_wording(
                     "reversal", source=atom.statement, candidate=question,
@@ -443,7 +447,7 @@ def tally(kind: str, reasons: list[str]) -> bool:
                         expected=f"Refuse the premise. The truth is: {atom.statement}",
                         expect_refusal=True,
                     ))
-    return ProbeSet(probes=probes, fidelity=fidelity)
+    return built
 
 
 def _chunks(items: list, size: int) -> list[list]:
@@ -465,10 +469,37 @@ def _generate(client: LLMClient, template: str, lines: list[str]) -> list[str]:
         rows = client.json_array(template.format(**{key: numbered}))
     except Exception:
         rows = []
+    return _aligned(rows, len(lines))
+
+
+def _worded(client: LLMClient, template: str, atoms: list, batch: int,
+            field_name: str) -> Iterator[tuple[list, list[str]]]:
+    """Each chunk of ``atoms`` with its wordings. A call with no price was paid for: its reply's
+    wordings are yielded, and the stop is raised once the caller has kept them."""
+    from dokimi.llm_json import extract_json_array
+
+    for chunk in _chunks(atoms, batch):
+        lines = [getattr(atom, field_name) for atom in chunk]
+        try:
+            worded = _generate(client, template, lines)
+        except SpendCapReached as stop:
+            if not stop.reply:
+                raise
+            try:
+                rows = extract_json_array(stop.reply)
+            except ValueError:
+                rows = []
+            yield chunk, _aligned(rows, len(lines))
+            raise stop
+        yield chunk, worded
+
+
+def _aligned(rows: list, n: int) -> list[str]:
+    """``n`` wordings from a reply's rows, ``""`` for a row that is missing or not text."""
     worded = [
         # A generator that echoes the "1. " list numbering would otherwise add an anchor.
         re.sub(r"^\d+\.\s+", "", str(row).strip())
         if isinstance(row, (str, int, float)) else ""
         for row in rows
-    ][: len(lines)]
-    return worded + [""] * (len(lines) - len(worded))
+    ][:n]
+    return worded + [""] * (n - len(worded))
diff --git a/dokimi/dokimi/runs/show.py b/dokimi/dokimi/runs/show.py
index 92a728f0d..8e413f017 100644
--- a/dokimi/dokimi/runs/show.py
+++ b/dokimi/dokimi/runs/show.py
@@ -12,7 +12,7 @@
 from dataclasses import dataclass, field
 from pathlib import Path
 
-from dokimi.artifacts import is_run_id, schema_errors, schema_version_for
+from dokimi.artifacts import is_run_id, schema_errors, schema_version_for, summary_errors
 from dokimi.runs.kinds import VERSION_KINDS, contents
 from dokimi.runs.scan import (
     Artifact,
@@ -144,7 +144,8 @@ def show(root: Path, address: str) -> Shown:
     except (OSError, ValueError, RecursionError) as exc:
         raise InputError(f"{plain(address)}: {read_error_reason(exc)}") from exc
     schema = schema_version_for(kind, data)
-    shown = Shown(artifact=artifact, schema=schema, errors=schema_errors(schema, data),
+    shown = Shown(artifact=artifact, schema=schema,
+                  errors=schema_errors(schema, data) + summary_errors(kind, data),
                   contents=contents(kind, data))
     _siblings(shown, path, os.path.realpath(root))
     return shown
diff --git a/dokimi/dokimi/schemas/novelty.v1.json b/dokimi/dokimi/schemas/novelty.v1.json
index d67f9978e..f974f8e49 100644
--- a/dokimi/dokimi/schemas/novelty.v1.json
+++ b/dokimi/dokimi/schemas/novelty.v1.json
@@ -9,11 +9,9 @@
     "artifact",
     "model",
     "judge_model",
-    "atoms",
-    "paired",
-    "k0",
-    "spans"
+    "atoms"
   ],
+  "$comment": "paired, k0 and spans are required of every record without `status` (a finished run); dokimi.artifacts.validate_artifact enforces it, since a schema keyword for it would narrow this version in place.",
   "properties": {
     "schema_version": {
       "const": "novelty.v1"
@@ -81,7 +79,7 @@
         },
         "sampling": {
           "type": "string",
-          "description": "How the probed subset was chosen: full, every extracted atom (also a cap the atoms meet or fall under), or cap-spread:<n> (e.g. cap-spread:120), only when the cap left atoms out: n atoms spread over every extracted atom in document order (extraction.sampled). Runs before cap-spread wrote cap-per-artifact:<n>, the first n atoms in document order. Present so a capped run is never read as a full sweep."
+          "description": "How the probed subset was chosen: none, not chosen yet (a partial record of a stop in extraction, whose atoms.n counts every extracted atom); full, every extracted atom (also a cap the atoms meet or fall under), or cap-spread:<n> (e.g. cap-spread:120), only when the cap left atoms out: n atoms spread over every extracted atom in document order (extraction.sampled). Runs before cap-spread wrote cap-per-artifact:<n>, the first n atoms in document order. Present so a capped run is never read as a full sweep."
         },
         "faithfulness": {
           "type": ["object", "null"],
@@ -197,6 +195,11 @@
         "n_probes": {"type": "integer", "minimum": 0},
         "malformed_rate": {"type": "number", "minimum": 0, "maximum": 1},
         "checks": {"type": "string"},
+        "items": {
+          "type": "array",
+          "items": {"type": "object"},
+          "description": "Partial record only: every probe built before the stop (atom_id, kind, question, expected, expect_refusal), so the wordings paid for are kept."
+        },
         "by_kind": {
           "type": "object",
           "additionalProperties": {
@@ -515,6 +518,40 @@
     "judge2": {
       "type": "object",
       "description": "With --judge2: a seeded --judge2-frac subset of the graded runs regraded by a second judge; lenient_agreement, kappa_lenient, exact_agreement over n_compared, and that judge's cost."
+    },
+    "status": {
+      "const": "partial",
+      "description": "Present only on a partial record: `skill novelty analyze` stopped by --max-usd (or a call with no known price) keeps what it paid for in <out stem>.partial.json. Every reader of a finished novelty.json refuses it."
+    },
+    "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.",
+      "properties": {
+        "reason": {"enum": ["spend_cap", "no_price"]},
+        "stage": {"enum": ["extraction", "verify", "need_graph", "probes", "measurement", "classify", "attribution"]},
+        "message": {"type": "string"},
+        "cause": {"type": "string"},
+        "spans_read": {"type": "integer", "minimum": 0},
+        "spans_total": {"type": "integer", "minimum": 0},
+        "calls_done": {"type": "integer", "minimum": 0},
+        "calls_planned": {"type": "integer", "minimum": 0}
+      }
+    },
+    "extracted_atoms": {
+      "type": "array",
+      "items": {"type": "object"},
+      "description": "Partial record only: every atom extracted before the stop, as `skill novelty atoms` writes them."
+    },
+    "source_spans": {
+      "type": "array",
+      "items": {"type": "object"},
+      "description": "Partial record only: the artifact's spans, which extracted_atoms cite; the ones extraction did not read are named in extraction.unread_span_ids."
+    },
+    "runs": {
+      "type": "array",
+      "items": {"type": "object"},
+      "description": "Partial record only: the probe answers and judgments made before the stop, as `skill novelty probe` writes them."
     }
   },
   "$defs": {
diff --git a/dokimi/scripts/growth_baseline.json b/dokimi/scripts/growth_baseline.json
index fcbde4ba2..7470bdb28 100644
--- a/dokimi/scripts/growth_baseline.json
+++ b/dokimi/scripts/growth_baseline.json
@@ -45,7 +45,7 @@
     "dokimi/harnesses/skill_treatment.py::stream_stats": {"complexity": 16},
     "dokimi/harnesses/tasks.py::harbor_task_problem": {"complexity": 11},
     "dokimi/mcp/context7.py::Context7Client.get_library_docs": {"complexity": 12},
-    "dokimi/novelty/analyze.py::analyze": {"complexity": 21},
+    "dokimi/novelty/analyze.py::_analyze": {"complexity": 21, "rationale": "the pipeline body of analyze, renamed (#624) so analyze can wrap it and keep a partial record at a spend stop; its complexity is unchanged"},
     "dokimi/novelty/atoms.py::extract_atoms": {"complexity": 13},
     "dokimi/novelty/bundle.py::load_bundle": {"complexity": 14},
     "dokimi/novelty/items.py::discrimination": {"complexity": 11},
diff --git a/dokimi/tests/test_novelty_analyze_cap_partial.py b/dokimi/tests/test_novelty_analyze_cap_partial.py
new file mode 100644
index 000000000..6153f88c1
--- /dev/null
+++ b/dokimi/tests/test_novelty_analyze_cap_partial.py
@@ -0,0 +1,492 @@
+# Copyright (C) 2026 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+"""`skill novelty analyze` stopped by --max-usd keeps what it paid for, beside the finished file.
+
+The cap is hit in extraction, in the faithfulness check, in the measurement and in the paid
+classify stage; each leaves ``<out stem>.partial.json`` with ``status: partial``, the extracted
+atoms and, once measured, the probe answers. A finished ``novelty.json`` is never replaced, and
+every reader of one refuses the partial file.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+
+import pytest
+
+from dokimi.artifacts import load_artifact, validate_artifact
+from dokimi.cli import build_parser
+from dokimi.metrics.usage import UsageRecord
+from dokimi.novelty import cost as novelty_cost
+from dokimi.novelty.analyze import AnalyzeConfig, analyze, load_artifact_text
+from dokimi.novelty.client import LLMClient, SpendCap
+from dokimi.utils import InputError
+from tests.cli_paths import novelty_argv
+from tests.test_novelty_analyze import SKILL, _scripted
+
+ROWS = "\n\n".join(f"Setting number {i} defaults to {i * 10}." for i in range(1, 201)) + "\n"
+
+
+def _priced(cost_of):
+    """The scripted model of ``test_novelty_analyze``, each reply billed ``cost_of(prompt)``."""
+    base = _scripted("informed")
+
+    def transport(prompt, system=None):
+        reply, _usage = base(prompt, system)
+        return reply, UsageRecord(cost_usd=cost_of(prompt))
+
+    return transport
+
+
+def _clients(cost_of, limit=1.0):
+    cap = SpendCap(limit_usd=limit)
+    transport = _priced(cost_of)
+    return tuple(LLMClient(model=f"{role}-m", role=role, transport=transport, cap=cap)
+                 for role in ("extractor", "answerer", "judge"))
+
+
+def _skill(tmp_path):
+    path = tmp_path / "accel-tuning" / "SKILL.md"
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(SKILL, encoding="utf-8")
+    return path
+
+
+def _run(tmp_path, cost_of, limit=1.0, **options):
+    path = _skill(tmp_path)
+    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,
+                           **options)
+    extractor, answerer, judge = _clients(cost_of, limit)
+    return analyze(body, config, extractor=extractor, answerer=answerer, judge=judge)
+
+
+def _free(prompt):
+    return 0.0
+
+
+def _answers_cost(prompt):
+    # Each probe answer is 0.6 USD under a 1.00 cap: two fit, the third is refused.
+    return 0.6 if prompt.startswith(("Reference material:", "Question:")) else 0.0
+
+
+def _after(marker):
+    """The call that carries ``marker`` spends the whole cap; the next call, whoever's, is refused."""
+    return lambda prompt: 100.0 if marker in prompt else 0.0
+
+
+def test_a_finished_analyze_has_no_partial_marker(tmp_path):
+    result = _run(tmp_path, _free)
+    assert result.partial is None
+    assert "status" not in result.artifact and "partial" not in result.artifact
+
+
+def test_a_record_without_status_still_needs_its_summaries(tmp_path):
+    from dokimi.artifacts import ArtifactValidationError
+
+    record = dict(_run(tmp_path, _answers_cost).artifact)
+    del record["status"], record["partial"]
+    with pytest.raises(ArtifactValidationError, match="'paired' is a required property"):
+        validate_artifact("novelty", record)
+
+
+def test_a_cap_in_measurement_keeps_the_atoms_and_the_answers_paid_for(tmp_path):
+    result = _run(tmp_path, _answers_cost)
+    record = result.artifact
+    assert record["status"] == "partial"
+    assert record["partial"]["stage"] == "measurement" and record["partial"]["reason"] == "spend_cap"
+    assert record["partial"]["calls_done"] == 2 and record["partial"]["calls_planned"] > 2
+    # The cap spent on the second answer, so the judge's call for it was refused: kept ungraded.
+    assert re.fullmatch(r"--max-usd stopped the probes with 2 of \d+ answers; 1 of them not graded",
+                        record["partial"]["message"])
+    assert "spend cap reached" in record["partial"]["cause"]
+    assert len(record["extracted_atoms"]) == 2 and record["atoms"]["n"] == 2
+    assert len(record["runs"]) == 2 and all(run["verdict"] for run in record["runs"])
+    assert record["probes"]["n_probes"] > 0 and record["source_spans"]
+    # Nothing summarises the uneven arms: no paired figures, no views.
+    assert not {"paired", "k0", "arms", "views", "shortening"} & set(record)
+    # What the stopped run paid and how it was configured are kept for its rerun to compare.
+    assert (record["cost"]["usd"], record["cost"]["llm_calls"]) == (1.2, 9)
+    assert record["run_config"]["prompts_sha256"] and record["atoms"]["sampling"] == "full"
+    assert record["atoms"]["faithfulness"]
+    validate_artifact("novelty", record)
+
+
+def test_a_capped_run_stopped_in_measurement_counts_the_probed_sample_as_a_finished_one_does(tmp_path):
+    stopped = _run(tmp_path, _answers_cost, max_atoms=1).artifact
+    finished = _run(tmp_path, _free, max_atoms=1).artifact
+    assert stopped["partial"]["stage"] == "measurement"
+    assert len(stopped["extracted_atoms"]) == 2
+    assert stopped["atoms"]["sampling"] == finished["atoms"]["sampling"] == "cap-spread:1"
+    assert stopped["atoms"]["n"] == finished["atoms"]["n"] == 1
+    assert stopped["atoms"]["n_probed"] == 1
+
+
+def test_a_cap_in_the_need_graph_counts_the_sample_not_the_extraction(tmp_path):
+    record = _run(tmp_path, lambda prompt: None if "TASK NAME:" in prompt else 0.0,
+                  max_atoms=1).artifact
+    assert record["partial"]["stage"] == "need_graph"
+    assert (record["atoms"]["n"], len(record["extracted_atoms"])) == (1, 2)
+
+
+def test_a_cap_after_the_wordings_keeps_every_probe_before_the_first_answer(tmp_path):
+    record = _run(tmp_path, _after("Rewrite each statement below")).artifact
+    assert record["partial"]["stage"] == "measurement" and record["partial"]["calls_done"] == 0
+    assert record["runs"] == []
+    items = record["probes"]["items"]
+    assert len(items) == record["probes"]["n_probes"] > 0
+    assert all(item["question"] and item["expected"] and item["atom_id"] for item in items)
+    assert {"paraphrase", "reversal"} <= {item["kind"] for item in items}
+    validate_artifact("novelty", record)
+
+
+def test_a_cap_between_the_two_wording_calls_keeps_the_paraphrases_already_paid_for(tmp_path):
+    record = _run(tmp_path, _after("Reword each question")).artifact
+    assert record["partial"]["stage"] == "probes" and record["partial"]["reason"] == "spend_cap"
+    items = record["probes"]["items"]
+    kinds = {item["kind"] for item in items}
+    assert kinds == {"direct", "paraphrase"}
+    assert len(items) == record["probes"]["n_probes"]
+    assert record["probes"]["by_kind"]["paraphrase"]["kept"] > 0
+    assert "reversal" not in record["probes"]["by_kind"]
+    validate_artifact("novelty", record)
+
+
+@pytest.mark.parametrize("marker, kind", [("Reword each question", "paraphrase"),
+                                          ("Rewrite each statement below", "reversal")])
+def test_a_no_price_wording_call_keeps_the_wordings_its_reply_carried(tmp_path, marker, kind):
+    record = _run(tmp_path, lambda prompt: None if marker in prompt else 0.0).artifact
+    assert record["partial"]["stage"] == "probes" and record["partial"]["reason"] == "no_price"
+    items = record["probes"]["items"]
+    assert kind in {item["kind"] for item in items}
+    assert record["probes"]["by_kind"][kind]["kept"] > 0
+    assert len(items) == record["probes"]["n_probes"]
+    validate_artifact("novelty", record)
+
+
+@pytest.mark.parametrize("marker, kind", [("Reword each question", "paraphrase"),
+                                          ("Rewrite each statement below", "reversal")])
+def test_a_no_price_wording_reply_that_is_not_json_keeps_no_wording_and_still_stops(tmp_path, marker, kind):
+    base = _scripted("informed")
+
+    def transport(prompt, system=None):
+        if marker in prompt:
+            return "Sorry, I cannot do that.", UsageRecord(cost_usd=None)
+        reply, _usage = base(prompt, system)
+        return reply, UsageRecord(cost_usd=0.0)
+
+    cap = SpendCap(limit_usd=1.0)
+    path = _skill(tmp_path)
+    art_kind, name, description, frontmatter, body = load_artifact_text(f"skill:{path}")
+    config = AnalyzeConfig(artifact_ref=f"skill:{path}", artifact_kind=art_kind, name=name,
+                           description=description, frontmatter=frontmatter, concurrency=1)
+    extractor, answerer, judge = (LLMClient(model=f"{role}-m", role=role, transport=transport, cap=cap)
+                                  for role in ("extractor", "answerer", "judge"))
+    record = analyze(body, config, extractor=extractor, answerer=answerer, judge=judge).artifact
+    assert record["partial"]["stage"] == "probes" and record["partial"]["reason"] == "no_price"
+    # The reply carried no array: the stop is still raised, no wording of that kind is kept, and
+    # each atom asked for is tallied as a rejected (empty) wording.
+    assert kind not in {item["kind"] for item in record["probes"]["items"]}
+    tally = record["probes"]["by_kind"][kind]
+    assert tally["kept"] == 0 and tally["generated"] == tally["rejected"] > 0
+    assert tally["reasons"] == {"empty": tally["rejected"]}
+    validate_artifact("novelty", record)
+
+
+def test_a_cap_in_the_paid_classify_stage_keeps_the_complete_measurement(tmp_path):
+    result = _run(tmp_path, _after("judge how it functions"))
+    partial = result.artifact["partial"]
+    assert partial["stage"] == "classify"
+    assert partial["calls_done"] == partial["calls_planned"] == len(result.artifact["runs"]) > 0
+    validate_artifact("novelty", result.artifact)
+
+
+def test_a_cap_in_the_faithfulness_check_keeps_the_extracted_atoms(tmp_path):
+    result = _run(tmp_path, _after("Does the source text below support"))
+    record = result.artifact
+    assert record["partial"]["stage"] == "verify"
+    assert len(record["extracted_atoms"]) == 2 and "runs" not in record and "probes" not in record
+    validate_artifact("novelty", record)
+
+
+def test_a_cap_in_extraction_keeps_the_atoms_read_and_marks_the_rest_unread():
+    def one_atom_per_span(prompt, system=None):
+        rows = [{"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)]
+        return json.dumps(rows), UsageRecord(cost_usd=0.6)
+
+    cap = SpendCap(limit_usd=1.0)
+    extractor = LLMClient(model="e", role="extractor", transport=one_atom_per_span, cap=cap)
+    idle = LLMClient(model="a", role="answerer", transport=one_atom_per_span, cap=cap)
+    judge = LLMClient(model="j", role="judge", transport=one_atom_per_span, cap=cap)
+    config = AnalyzeConfig(artifact_ref="text:rows", artifact_kind="text", concurrency=1,
+                           seed=7, max_atoms=30)
+    record = analyze(ROWS, config, extractor=extractor, answerer=idle, judge=judge).artifact
+    assert record["partial"]["stage"] == "extraction"
+    # The settings of the run are kept; the sample is not chosen yet, so `atoms` counts the
+    # extracted atoms and says `none`.
+    assert (record["run_config"]["seed"], record["run_config"]["max_atoms"]) == (7, 30)
+    assert record["run_config"]["prompts_sha256"] and record["run_config"]["arms"]
+    assert (record["atoms"]["n"], record["atoms"]["sampling"]) == (50, "none")
+    assert (record["partial"]["spans_read"], record["partial"]["spans_total"]) == (50, 200)
+    assert record["partial"]["cause"].startswith("spend cap reached: $1.20 of $1.00")
+    assert len(record["extracted_atoms"]) == 50 and "runs" not in record
+    assert len(record["extraction"]["unread_span_ids"]) == 150
+    assert len(record["source_spans"]) == 200
+    validate_artifact("novelty", record)
+
+
+def test_a_call_with_no_price_stops_analyze_and_is_named_as_such(tmp_path):
+    result = _run(tmp_path, lambda prompt: None)
+    partial = result.artifact["partial"]
+    assert partial["reason"] == "no_price" and "cause" not in partial
+    assert partial["stage"] == "extraction"
+    assert "no price is known" in partial["message"]
+    validate_artifact("novelty", result.artifact)
+
+
+def test_a_stop_in_the_need_graph_is_named_as_that_stage(tmp_path):
+    record = _run(tmp_path, lambda prompt: None if "TASK NAME:" in prompt else 0.0).artifact
+    assert record["partial"]["stage"] == "need_graph"
+    assert record["partial"]["message"].startswith("analyze stopped at the need graph: ")
+    validate_artifact("novelty", record)
+
+
+# --- the command ---------------------------------------------------------------------------
+
+def _command(tmp_path, monkeypatch, cost_of, *extra):
+    path = _skill(tmp_path)
+    clients = _clients(cost_of)
+    monkeypatch.setattr(novelty_cost, "clients", lambda args: clients)
+    out_dir = tmp_path / "run"
+    args = build_parser().parse_args([*novelty_argv("analyze"), "--artifact", f"skill:{path}",
+                                      "--model", "m", "--out-dir", str(out_dir), "--max-usd", "1",
+                                      "--concurrency", "1", *extra])
+    return args, out_dir
+
+
+def test_the_command_writes_the_partial_file_beside_novelty_json_and_names_it(tmp_path, monkeypatch):
+    monkeypatch.delenv("DOKIMI_RESPONSE_CACHE", raising=False)
+    args, out_dir = _command(tmp_path, monkeypatch, _answers_cost)
+    with pytest.raises(SystemExit) as stop:
+        args.func(args)
+    message = str(stop.value)
+    partial = out_dir / "novelty.partial.json"
+    assert f"the partial result is in {partial}" in message
+    assert message.startswith("--max-usd stopped the probes with 2 of ")
+    assert "spend cap reached" in message  # what was spent, from the cap itself
+    assert "DOKIMI_RESPONSE_CACHE was not set on this run" in message
+    assert sorted(path.name for path in out_dir.iterdir()) == ["novelty.partial.json"]
+    record = json.loads(partial.read_text(encoding="utf-8"))
+    assert record["status"] == "partial" and record["partial"]["stage"] == "measurement"
+    validate_artifact("novelty", record)
+
+
+def test_the_message_says_a_rerun_replays_the_paid_calls_when_the_cache_was_set(tmp_path,
+                                                                              monkeypatch):
+    (tmp_path / "cache").mkdir()  # the scripted transport bypasses the cache
+    monkeypatch.setenv("DOKIMI_RESPONSE_CACHE", str(tmp_path / "cache"))
+    args, _out_dir = _command(tmp_path, monkeypatch, _answers_cost)
+    with pytest.raises(SystemExit, match="same DOKIMI_RESPONSE_CACHE: the calls already paid for "
+                                         "replay at \\$0"):
+        args.func(args)
+
+
+def test_a_no_price_stop_does_not_advise_a_higher_cap(tmp_path, monkeypatch):
+    (tmp_path / "cache").mkdir()  # the scripted transport bypasses the cache
+    monkeypatch.setenv("DOKIMI_RESPONSE_CACHE", str(tmp_path / "cache"))
+    args, _out_dir = _command(tmp_path, monkeypatch, lambda prompt: None)
+    with pytest.raises(SystemExit) as stop:
+        args.func(args)
+    message = str(stop.value)
+    assert "Rerun with the model priced, or without --max-usd, and the same DOKIMI_RESPONSE_CACHE" \
+        in message
+    assert "higher --max-usd" not in message
+
+
+def test_the_message_does_not_claim_a_replay_when_the_cache_variable_names_a_file(tmp_path,
+                                                                               monkeypatch):
+    not_a_directory = tmp_path / "cache-file"
+    not_a_directory.write_text("x", encoding="utf-8")
+    monkeypatch.setenv("DOKIMI_RESPONSE_CACHE", str(not_a_directory))
+    args, _out_dir = _command(tmp_path, monkeypatch, _answers_cost)
+    with pytest.raises(SystemExit) as stop:
+        args.func(args)
+    message = str(stop.value)
+    assert "DOKIMI_RESPONSE_CACHE was not set on this run" in message
+    assert "a rerun pays for these calls again" in message
+    assert "already paid for replay" not in message
+
+
+def _replay_hint(monkeypatch, cache):
+    from dokimi.commands.novelty_budget import replay_hint
+
+    monkeypatch.setenv("DOKIMI_RESPONSE_CACHE", str(cache))
+    return replay_hint("spend_cap")
+
+
+def test_no_replay_is_promised_when_the_cache_path_lies_under_a_file(tmp_path, monkeypatch):
+    (tmp_path / "afile").write_text("x", encoding="utf-8")
+    hint = _replay_hint(monkeypatch, tmp_path / "afile" / "cache")
+    assert "already paid for replay" not in hint
+    assert "could not be written, so a rerun pays for these calls again" in hint
+
+
+@pytest.mark.skipif(os.name == "nt" or getattr(os, "geteuid", lambda: 1)() == 0,
+                    reason="a read-only directory is still writable here")
+def test_no_replay_is_promised_when_the_cache_dir_is_read_only(tmp_path, monkeypatch):
+    cache = tmp_path / "cache"
+    cache.mkdir()
+    cache.chmod(0o555)
+    try:
+        hint = _replay_hint(monkeypatch, cache)
+    finally:
+        cache.chmod(0o755)
+    assert "already paid for replay" not in hint
+    assert "a rerun pays for these calls again" in hint
+
+
+def test_no_replay_is_promised_when_a_cache_write_failed_on_a_full_disk(tmp_path, monkeypatch):
+    from pathlib import Path
+
+    from dokimi import caching
+
+    cache = tmp_path / "cache"
+    cache.mkdir()
+    monkeypatch.setenv("DOKIMI_RESPONSE_CACHE", str(cache))
+
+    def disk_full(self, *args, **kwargs):
+        raise OSError(28, "No space left on device")
+
+    monkeypatch.setattr(Path, "write_text", disk_full)
+    caching.store_response("k", "text", UsageRecord(cost_usd=0.5))
+    monkeypatch.undo()
+    monkeypatch.setenv("DOKIMI_RESPONSE_CACHE", str(cache))
+    assert os.access(cache, os.W_OK) and not list(cache.iterdir())  # the directory looks fine
+    hint = _replay_hint(monkeypatch, cache)
+    assert "already paid for replay" not in hint
+    assert "a rerun pays for these calls again" in hint
+
+
+def _through_the_cache(cost_of, limit, sent):
+    """Three clients on the real ``LLM`` path (response cache included) over the scripted model;
+    every prompt that reaches the model is appended to ``sent``."""
+    from dokimi import llm
+
+    base = _scripted("informed")
+
+    def adapter(request):
+        content = request.messages[-1]["content"]
+        prompt = content if isinstance(content, str) else "".join(part["text"] for part in content)
+        sent.append(prompt)
+        reply, _usage = base(prompt, None)
+        return llm.Reply(text=reply, usage=UsageRecord(cost_usd=cost_of(prompt)))
+
+    cap = SpendCap(limit_usd=limit)
+    clients = tuple(LLMClient(model=f"{role}-m", role=role, cap=cap)
+                    for role in ("extractor", "answerer", "judge"))
+    return cap, clients, llm.use_adapter(adapter)
+
+
+def test_a_rerun_with_a_higher_cap_and_the_same_cache_replays_the_paid_calls_at_zero(
+        tmp_path, monkeypatch):
+    monkeypatch.setenv("DOKIMI_RESPONSE_CACHE", str(tmp_path / "cache"))
+    monkeypatch.setenv("OPENAI_API_KEY", "sk-test")
+    path = _skill(tmp_path)
+    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)
+
+    first_sent: list[str] = []
+    cap, (extractor, answerer, judge), adapter = _through_the_cache(_answers_cost, 1.0, first_sent)
+    with adapter:
+        stopped = analyze(body, config, extractor=extractor, answerer=answerer, judge=judge)
+    assert stopped.partial and stopped.partial["reason"] == "spend_cap"
+    assert first_sent and cap.spent_usd == pytest.approx(1.2)
+
+    second_sent: list[str] = []
+    cap, (extractor, answerer, judge), adapter = _through_the_cache(_answers_cost, 100.0, second_sent)
+    with adapter:
+        finished = analyze(body, config, extractor=extractor, answerer=answerer, judge=judge)
+    assert finished.partial is None and "status" not in finished.artifact
+    assert not set(first_sent) & set(second_sent)  # nothing paid for is asked again
+    assert cap.spent_usd == pytest.approx(0.6 * sum(_answers_cost(p) > 0 for p in second_sent))
+    assert finished.artifact["atoms"]["n"] == len(stopped.artifact["extracted_atoms"]) == 2
+
+
+def test_a_partial_result_never_replaces_a_finished_novelty_json(tmp_path, monkeypatch):
+    args, out_dir = _command(tmp_path, monkeypatch, _answers_cost)
+    out_dir.mkdir()
+    finished = out_dir / "novelty.json"
+    finished.write_text('{"schema_version": "novelty.v1", "finished": true}', encoding="utf-8")
+    with pytest.raises(SystemExit):
+        args.func(args)
+    assert json.loads(finished.read_text(encoding="utf-8"))["finished"] is True
+    assert json.loads((out_dir / "novelty.partial.json").read_text(encoding="utf-8"))["status"] \
+        == "partial"
+    assert not (out_dir / "novelty.md").exists()
+
+
+def test_a_partial_analyze_file_is_refused_by_the_readers_of_a_finished_one(tmp_path,
+                                                                           monkeypatch):
+    from dokimi.commands.novelty_view_inputs import load_view_inputs
+    from dokimi.subjects.novelty_track import load_novelty_artifacts
+
+    args, out_dir = _command(tmp_path, monkeypatch, _answers_cost)
+    with pytest.raises(SystemExit):
+        args.func(args)
+    partial = out_dir / "novelty.partial.json"
+    record = json.loads(partial.read_text(encoding="utf-8"))
+    del record["partial"]  # the schema lets `status` stand alone
+    status_only = tmp_path / "status-only.json"
+    status_only.write_text(json.dumps(record), encoding="utf-8")
+    for source in (partial, status_only):
+        for validate in (True, False):
+            with pytest.raises(InputError, match="partial novelty artifact"):
+                load_artifact("novelty", source, validate=validate)
+    for read in (lambda: load_artifact("novelty", partial),
+                 lambda: load_artifact("novelty", partial, validate=False),
+                 lambda: load_view_inputs(str(partial), str(tmp_path / "atoms.json")),
+                 lambda: load_novelty_artifacts([partial])):
+        with pytest.raises(InputError, match="partial novelty artifact") as refused:
+            read()
+        assert str(partial) in str(refused.value)
+
+
+@pytest.mark.parametrize("marker", ["both", "status", "partial"])
+def test_save_artifact_refuses_a_partial_novelty_record_and_writes_nothing(tmp_path, marker):
+    from dokimi.artifacts import save_artifact
+
+    record = dict(_run(tmp_path, _answers_cost).artifact)
+    if marker == "status":
+        del record["partial"]
+    elif marker == "partial":
+        del record["status"]
+    out = tmp_path / "novelty.json"
+    with pytest.raises(InputError, match="partial novelty record") as refused:
+        save_artifact("novelty", record, out)
+    assert str(out) in str(refused.value)
+    assert not out.exists() and not list(tmp_path.glob("novelty.json.*"))
+
+
+def test_save_artifact_still_saves_a_finished_novelty_record(tmp_path):
+    from dokimi.artifacts import save_artifact
+
+    out = tmp_path / "novelty.json"
+    saved = save_artifact("novelty", _run(tmp_path, _free).artifact, out)
+    assert json.loads(out.read_text(encoding="utf-8"))["schema_version"] == saved["schema_version"]
+
+
+def test_a_partial_analyze_file_is_not_listed_as_a_finished_run(tmp_path, monkeypatch):
+    from dokimi.runs.scan import scan
+
+    args, out_dir = _command(tmp_path, monkeypatch, _answers_cost)
+    with pytest.raises(SystemExit):
+        args.func(args)
+    result = scan(out_dir)
+    assert result.artifacts == [] and result.skipped["no_leading_schema_version"] == 1
diff --git a/dokimi/tests/test_novelty_cap_partial.py b/dokimi/tests/test_novelty_cap_partial.py
index fef76d2fe..afbbd8c8b 100644
--- a/dokimi/tests/test_novelty_cap_partial.py
+++ b/dokimi/tests/test_novelty_cap_partial.py
@@ -337,11 +337,10 @@ def capped(sample, spans, client):
 
 
 @pytest.mark.parametrize("not_run", [1, 3])
-def test_analyze_stops_rather_than_publish_a_measurement_the_cap_cut_short(tmp_path, monkeypatch,
-                                                                          not_run):
+def test_analyze_keeps_a_partial_record_rather_than_publish_a_measurement_the_cap_cut_short(
+        tmp_path, monkeypatch, not_run):
     from dokimi.novelty import measure
     from dokimi.novelty.analyze import AnalyzeConfig, analyze, load_artifact_text
-    from dokimi.novelty.client import SpendCapReached
 
     from tests.test_novelty_analyze import SKILL, _clients
 
@@ -362,10 +361,11 @@ def cut_short(*a, **kw):
 
     monkeypatch.setattr(measure, "run_measurement", cut_short)
     extractor, answerer, judge = _clients()
-    # The match is a prefix: the CLI adds the rest of the exit message.
-    with pytest.raises(SpendCapReached,
-                       match=f"{not_run} probe answers were not run or graded"):
-        analyze(body, config, extractor=extractor, answerer=answerer, judge=judge)
+    result = analyze(body, config, extractor=extractor, answerer=answerer, judge=judge)
+    assert result.partial["stage"] == "measurement"
+    assert f"{not_run} probe answers were not run or graded" in result.partial["cause"]
+    assert (result.partial["calls_done"], result.partial["calls_planned"]) == (1, 1 + not_run)
+    assert result.artifact["status"] == "partial" and "paired" not in result.artifact
 
 
 @pytest.mark.parametrize("reason", ["spend_cap", "no_price"])
@@ -374,7 +374,6 @@ def test_analyze_refuses_answers_the_judge_never_graded(tmp_path, monkeypatch, r
     # cut, so only the ungraded half of the gate decides. The stop must still name the count.
     from dokimi.novelty import measure
     from dokimi.novelty.analyze import AnalyzeConfig, analyze, load_artifact_text
-    from dokimi.novelty.client import SpendCapReached
 
     from tests.test_novelty_analyze import SKILL, _clients
 
@@ -396,8 +395,11 @@ def ungraded_only(*a, **kw):
 
     monkeypatch.setattr(measure, "run_measurement", ungraded_only)
     extractor, answerer, judge = _clients()
-    with pytest.raises(SpendCapReached, match="2 probe answers (were )?not run or graded"):
-        analyze(body, config, extractor=extractor, answerer=answerer, judge=judge)
+    partial = analyze(body, config, extractor=extractor, answerer=answerer, judge=judge).partial
+    assert partial["reason"] == reason and "2 of them not graded" in partial["message"]
+    # A spend stop keeps the cap's own text as the cause; an unpriced one has it in the message.
+    assert re.search("2 probe answers (were )?not run or graded",
+                     partial["cause"] if reason == "spend_cap" else partial["message"])
 
 
 def test_save_probe_measure_keeps_an_ungraded_answer_in_the_partial_file(tmp_path):
@@ -426,6 +428,17 @@ def test_novelty_items_refuses_a_partial_probe_file_by_name(tmp_path):
     assert str(partial) in str(refused.value)
 
 
+def test_novelty_items_refuses_a_probe_file_marked_partial_by_status_alone(tmp_path):
+    from dokimi.novelty.items import ProbeFileError, load_probe_cells
+
+    run = {"atom_id": "a", "probe_kind": "direct", "arm": "model", "verdict": "yes"}
+    marked = tmp_path / "probes.json"
+    marked.write_text(json.dumps({"status": "partial", "runs": [run]}), encoding="utf-8")
+    with pytest.raises(ProbeFileError, match="partial probe result") as refused:
+        load_probe_cells([str(marked)])
+    assert str(marked) in str(refused.value)
+
+
 def test_an_earlier_file_that_is_not_json_is_kept_and_not_taken_for_a_partial_one(tmp_path):
     from dokimi.commands.novelty_budget import partial_destination
 
@@ -514,7 +527,6 @@ def test_an_unpriced_reply_that_fails_part_way_keeps_no_atoms_and_its_batch_stay
 def test_analyze_names_a_missing_price_when_the_measurement_stops_on_one(tmp_path, monkeypatch):
     from dokimi.novelty import measure
     from dokimi.novelty.analyze import AnalyzeConfig, analyze, load_artifact_text
-    from dokimi.novelty.client import SpendCapReached
 
     from tests.test_novelty_analyze import SKILL, _clients
 
@@ -536,11 +548,10 @@ def stopped_on_a_price(*a, **kw):
 
     monkeypatch.setattr(measure, "run_measurement", stopped_on_a_price)
     extractor, answerer, judge = _clients()
-    with pytest.raises(SpendCapReached) as refused:
-        analyze(body, config, extractor=extractor, answerer=answerer, judge=judge)
-    assert refused.value.reason == "no_price"
-    assert "Raise --max-usd" not in str(refused.value)
-    assert "no price is known for judge" in str(refused.value)
+    partial = analyze(body, config, extractor=extractor, answerer=answerer, judge=judge).partial
+    assert partial["reason"] == "no_price"
+    assert "Raise --max-usd" not in partial["message"] and "cause" not in partial
+    assert "no price is known for judge" in partial["message"]
 
 
 def test_an_unpriced_partial_atoms_file_is_refused_with_its_cause(tmp_path):
diff --git a/dokimi/tests/test_runs_show.py b/dokimi/tests/test_runs_show.py
index 95056c064..accb67715 100644
--- a/dokimi/tests/test_runs_show.py
+++ b/dokimi/tests/test_runs_show.py
@@ -143,6 +143,22 @@ def test_an_artifact_that_fails_its_schema_lists_the_errors_and_exits_0(monkeypa
     assert lines[5] == "subject:    onetbb"
 
 
+def test_a_finished_novelty_record_without_its_summaries_is_invalid_but_a_partial_one_is_not(
+        monkeypatch, capsys, root):
+    data = {"schema_version": "novelty.v1", **_fixture("novelty.v1")}  # the version leads the file
+    for name in ("paired", "k0", "spans"):
+        del data[name]
+    (root / "novelty.json").write_text(json.dumps(data, indent=2), encoding="utf-8")
+    partial = {**data, "status": "partial",
+               "partial": {"reason": "spend_cap", "stage": "measurement", "message": "stopped"}}
+    (root / "novelty.partial.json").write_text(json.dumps(partial, indent=2), encoding="utf-8")
+
+    finished = _show(monkeypatch, capsys, "novelty.json").out
+    assert "schema:     invalid, 3 errors" in finished
+    assert "  - <root>: 'paired' is a required property" in finished
+    assert "schema:     valid" in _show(monkeypatch, capsys, "novelty.partial.json").out
+
+
 def test_a_time_the_artifact_does_not_record_is_shown_as_the_files_mtime(monkeypatch, capsys, root):
     stamp = 1_780_000_000
     os.utime(root / "arms" / "plugin_delta.json", (stamp, stamp))