MVP Milestone 10

MVP_PROGRESS.md

@@ -775,3 +775,97 @@ Remaining risks or assumptions:
 - Candidate ID provenance is retained as structured warning metadata on converted records rather than adding a new generated output field to `ClassificationRecord` in this milestone.
 - Non-assignable concept rejection is implemented, but the current vocabulary fixture contains only assignable UUID-bearing records.
 - Redundancy cleanup is collection-level only; integration into final article result assembly remains deferred to later milestones.
+
+## Milestone 10: Batch Runner, Incremental Persistence, Caching, and Logging
+
+Status: completed
+
+Scope completed:
+- Single-article orchestration through Topic routing, Term routing, Variable descent, deterministic validation, and redundancy removal
+- Article-level `ArticleResult` generation for classified, not-classified, partial, and failed outcomes
+- Batch processing over validated article records with continuation after article failures
+- Invalid source article diagnostics without fallback DOI generation
+- Incremental JSON checkpoint persistence after each article result
+- Final consolidated machine-readable JSON output
+- File-backed cache identity and cache reuse
+- Structured logging helpers with secret redaction
+- Run summary counters for valid/invalid records, processed articles, cache hits/misses, warnings, errors, and accepted classifications
+
+Single-article pipeline behavior implemented:
+- `classify_article` coordinates existing Milestone 6-9 components.
+- No-topic results become `processing_status: completed`, `classification_outcome: not_classified`, empty classifications, and a no-classification reason.
+- Stop-at-Topic, stop-at-Term, and Variable terminal outcomes are converted only after deterministic validation succeeds.
+- Branch-level Variable descent errors are retained as structured errors while successful sibling terminal outcomes are preserved.
+- Articles with accepted classifications and branch errors become `processing_status: partial` with the accepted classifications retained.
+- Failed articles retain source fields exactly and include structured errors.
+
+Batch runner behavior implemented:
+- `run_batch` accepts an `ArticleLoadResult`, processes only valid records in source order, and reports invalid source records in the summary.
+- One article failure does not stop later valid articles.
+- Each article result is saved promptly after completion or failure.
+- Batch output avoids duplicate article files for the same DOI-safe key by overwriting the DOI checkpoint path.
+- The runner supports `force_reprocess` to bypass cache reuse.
+
+Persistence strategy selected and implemented:
+- The MVP uses one JSON checkpoint file per DOI-safe SHA-256 key plus an optional consolidated `results.json`.
+- This append/checkpoint-style representation avoids rewriting a single large output document after every article.
+- Individual article checkpoint writes and consolidated output writes use temp-file-plus-atomic-replace.
+- Temporary and destination files are created in the same destination directory, so atomic replace occurs on the same filesystem.
+- No database infrastructure was introduced.
+
+Cache identity and reuse behavior implemented:
+- `CacheIdentity` includes DOI, exact article fingerprint, vocabulary version/hash, model provider, model name, model temperature, timeout, max retries, Topic/Term/Variable prompt versions, application version, and relevant configuration hash.
+- Cache keys change when article content, vocabulary version, model settings, prompt versions, application version, or relevant configuration changes.
+- Cache hits return article results with `processing_metadata.cache_used: true`.
+- Cache hits remain completed work and are never marked `skipped`.
+- Force reprocessing bypasses cache reuse.
+- Stale cache entries are not reused when identity inputs differ.
+
+Logging and diagnostics behavior implemented:
+- `log_event` records structured events through standard logging.
+- `sanitize_log_details` redacts keys that look like API keys, tokens, authorization headers, secrets, passwords, or sensitive headers.
+- Pipeline and batch events include DOI, stage, branch/cache context, status, classification counts, and error counts when available.
+- Structured `OutputError` and `OutputWarning` models are used for article, branch, validation, redundancy, and invalid-source diagnostics.
+
+Run summary behavior implemented:
+- `RunSummary` now includes explicit `valid_article_records`, `invalid_source_records`, `processed_articles`, `cache_hits`, `cache_misses`, `total_warnings`, `total_errors`, and `duration_seconds` fields.
+- `schemas/run_summary.schema.json` was regenerated from the updated Pydantic model.
+- Summary output remains machine-readable and validates against the draft JSON Schema.
+
+Tests added:
+- `tests/test_cache.py`
+- `tests/test_persistence.py`
+- `tests/test_pipeline.py`
+
+Verification:
+- `python -m ruff check .` passed: `All checks passed!`
+- `python -m ruff format --check .` passed: `46 files already formatted`
+- `python -m pytest` passed: `261 passed in 0.77s`
+
+Live model/API status:
+- No live model API calls were made.
+- Unit tests use `FakeModelClient` only.
+
+Out of scope, not started:
+- Gradio UI
+- Hugging Face launcher
+- Root `app.py`
+- Live-model smoke tests
+- Independent semantic validation
+- Production deployment infrastructure
+
+Source-file status:
+- `data/gcmd_hierarchy.json` unchanged by this milestone.
+- `data/articles.json` unchanged by this milestone.
+- `prototype/app_hf_poc.py` unchanged by this milestone.
+
+Deviations from approved plan:
+- `src/gcmd_classifier/errors.py` did not require changes because existing typed exceptions and structured `OutputError` diagnostics were sufficient.
+- `src/gcmd_classifier/config.py` did not require changes because existing model settings already covered the cache identity inputs needed for this milestone.
+- `RunSummary` and `schemas/run_summary.schema.json` were updated to represent required Milestone 10 counters explicitly.
+
+Remaining risks or assumptions:
+- The checkpoint store uses DOI-safe SHA-256 filenames, so humans inspect consolidated JSON rather than filename-readable DOI values.
+- Cached failed and partial results can be reused when the cache identity matches exactly; future policy may choose to cache only completed results.
+- Batch processing is sequential for the MVP. Parallel execution remains deferred.
+- Run summary model-call totals are available for fake clients and providers that expose request counts or metadata; richer token/cost aggregation remains provider-dependent.

schemas/run_summary.schema.json

@@ -232,6 +232,18 @@
       "default": null,
       "title": "Average Processing Time Seconds"
     },
+    "cache_hits": {
+      "default": 0,
+      "minimum": 0,
+      "title": "Cache Hits",
+      "type": "integer"
+    },
+    "cache_misses": {
+      "default": 0,
+      "minimum": 0,
+      "title": "Cache Misses",
+      "type": "integer"
+    },
     "completed_at": {
       "anyOf": [
         {
@@ -244,6 +256,19 @@
       "default": null,
       "title": "Completed At"
     },
+    "duration_seconds": {
+      "anyOf": [
+        {
+          "minimum": 0.0,
+          "type": "number"
+        },
+        {
+          "type": "null"
+        }
+      ],
+      "default": null,
+      "title": "Duration Seconds"
+    },
     "errors": {
       "items": {
         "$ref": "#/$defs/OutputError"
@@ -277,6 +302,12 @@
       "default": null,
       "title": "Input Tokens"
     },
+    "invalid_source_records": {
+      "default": 0,
+      "minimum": 0,
+      "title": "Invalid Source Records",
+      "type": "integer"
+    },
     "output_tokens": {
       "anyOf": [
         {
@@ -290,6 +321,12 @@
       "default": null,
       "title": "Output Tokens"
     },
+    "processed_articles": {
+      "default": 0,
+      "minimum": 0,
+      "title": "Processed Articles",
+      "type": "integer"
+    },
     "reduced_classifications": {
       "default": 0,
       "minimum": 0,
@@ -319,6 +356,12 @@
       "default": null,
       "title": "Started At"
     },
+    "total_errors": {
+      "default": 0,
+      "minimum": 0,
+      "title": "Total Errors",
+      "type": "integer"
+    },
     "total_model_calls": {
       "anyOf": [
         {
@@ -332,6 +375,18 @@
       "default": null,
       "title": "Total Model Calls"
     },
+    "total_warnings": {
+      "default": 0,
+      "minimum": 0,
+      "title": "Total Warnings",
+      "type": "integer"
+    },
+    "valid_article_records": {
+      "default": 0,
+      "minimum": 0,
+      "title": "Valid Article Records",
+      "type": "integer"
+    },
     "warnings": {
       "items": {
         "$ref": "#/$defs/OutputWarning"

src/gcmd_classifier/logging_config.py

@@ -0,0 +1,33 @@
+"""Structured logging helpers for the GCMD classifier MVP."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+_SECRET_MARKERS = ("api_key", "apikey", "authorization", "token", "secret", "password", "header")
+
+
+def get_logger(name: str = "gcmd_classifier") -> logging.Logger:
+    """Return a package logger without forcing global logging configuration."""
+    return logging.getLogger(name)
+
+
+def sanitize_log_details(details: dict[str, Any]) -> dict[str, Any]:
+    """Remove values whose keys look like secrets or sensitive headers."""
+    sanitized: dict[str, Any] = {}
+    for key, value in details.items():
+        lowered = key.lower()
+        if any(marker in lowered for marker in _SECRET_MARKERS):
+            sanitized[key] = "[REDACTED]"
+        elif isinstance(value, dict):
+            sanitized[key] = sanitize_log_details(value)
+        else:
+            sanitized[key] = value
+    return sanitized
+
+
+def log_event(logger: logging.Logger, event: str, **details: Any) -> None:
+    """Emit a structured event through standard logging without secrets."""
+    safe_details = sanitize_log_details(details)
+    logger.info("gcmd_classifier_event", extra={"event": event, "details": safe_details})

src/gcmd_classifier/models.py

@@ -331,6 +331,14 @@ class RunSummary(BaseModel):
     started_at: str | None = None
     completed_at: str | None = None
     articles_received: int = Field(ge=0)
+    valid_article_records: int = Field(default=0, ge=0)
+    invalid_source_records: int = Field(default=0, ge=0)
+    processed_articles: int = Field(default=0, ge=0)
+    cache_hits: int = Field(default=0, ge=0)
+    cache_misses: int = Field(default=0, ge=0)
+    total_warnings: int = Field(default=0, ge=0)
+    total_errors: int = Field(default=0, ge=0)
+    duration_seconds: float | None = Field(default=None, ge=0.0)
     articles_completed: int = Field(default=0, ge=0)
     articles_partial: int = Field(default=0, ge=0)
     articles_failed: int = Field(default=0, ge=0)

src/gcmd_classifier/persistence/__init__.py

@@ -0,0 +1,29 @@
+"""Persistence and cache helpers for the GCMD classifier MVP."""
+
+from gcmd_classifier.persistence.cache import (
+    ArticleResultCache,
+    CachedArticleResult,
+    CacheIdentity,
+    article_fingerprint,
+    build_cache_identity,
+    configuration_hash,
+    mark_cache_used,
+)
+from gcmd_classifier.persistence.json_store import (
+    JsonResultStore,
+    PersistedRunOutput,
+    doi_to_key,
+)
+
+__all__ = [
+    "ArticleResultCache",
+    "CacheIdentity",
+    "CachedArticleResult",
+    "JsonResultStore",
+    "PersistedRunOutput",
+    "article_fingerprint",
+    "build_cache_identity",
+    "configuration_hash",
+    "doi_to_key",
+    "mark_cache_used",
+]

src/gcmd_classifier/persistence/cache.py

@@ -0,0 +1,132 @@
+"""Cache identity and file-backed result cache for article classifications."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+from pathlib import Path
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from gcmd_classifier.config import ModelSettings
+from gcmd_classifier.models import ArticleRecord, ArticleResult
+from gcmd_classifier.vocabulary.index import VocabularyIndex
+
+
+class CacheIdentity(BaseModel):
+    """All inputs that can affect reuse of a cached article result."""
+
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    DOI: str
+    article_fingerprint: str
+    vocabulary_version: str
+    model_provider: str
+    model_name: str
+    model_temperature: float
+    model_timeout_seconds: float
+    model_max_retries: int
+    prompt_version_topic: str
+    prompt_version_term: str
+    prompt_version_variable: str
+    application_version: str
+    configuration_hash: str
+
+    @property
+    def cache_key(self) -> str:
+        """Stable cache key derived from the complete identity."""
+        payload = json.dumps(self.model_dump(mode="json"), sort_keys=True, separators=(",", ":"))
+        return hashlib.sha256(payload.encode("utf-8")).hexdigest()
+
+
+class CachedArticleResult(BaseModel):
+    """Persisted cache entry containing identity and article result."""
+
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    cache_key: str = Field(min_length=1)
+    identity: CacheIdentity
+    result: ArticleResult
+
+
+class ArticleResultCache:
+    """Simple file-backed article result cache keyed by cache identity."""
+
+    def __init__(self, directory: str | Path) -> None:
+        self.directory = Path(directory)
+        self.directory.mkdir(parents=True, exist_ok=True)
+
+    def get(self, identity: CacheIdentity) -> ArticleResult | None:
+        """Return a completed cached result for the exact identity, if present."""
+        path = self._path(identity.cache_key)
+        if not path.exists():
+            return None
+        entry = CachedArticleResult.model_validate(json.loads(path.read_text()))
+        if entry.cache_key != identity.cache_key or entry.identity != identity:
+            return None
+        return mark_cache_used(entry.result)
+
+    def put(self, identity: CacheIdentity, result: ArticleResult) -> None:
+        """Persist a result for later exact-identity reuse."""
+        entry = CachedArticleResult(
+            cache_key=identity.cache_key,
+            identity=identity,
+            result=result,
+        )
+        _atomic_write_json(self._path(identity.cache_key), entry.model_dump(mode="json"))
+
+    def _path(self, cache_key: str) -> Path:
+        return self.directory / f"{cache_key}.json"
+
+
+def article_fingerprint(article: ArticleRecord) -> str:
+    """Hash exact source article values that influence classification."""
+    payload = json.dumps(article.model_dump(mode="json"), sort_keys=True, separators=(",", ":"))
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()
+
+
+def configuration_hash(config: dict[str, Any] | None = None) -> str:
+    """Hash relevant non-secret configuration values."""
+    payload = json.dumps({} if config is None else config, sort_keys=True, separators=(",", ":"))
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()
+
+
+def build_cache_identity(
+    *,
+    article: ArticleRecord,
+    vocabulary: VocabularyIndex,
+    settings: ModelSettings,
+    application_version: str,
+    relevant_config: dict[str, Any] | None = None,
+) -> CacheIdentity:
+    """Build the complete cache identity for one article classification run."""
+    return CacheIdentity(
+        DOI=article.DOI,
+        article_fingerprint=article_fingerprint(article),
+        vocabulary_version=vocabulary.vocabulary_version,
+        model_provider=settings.provider,
+        model_name=settings.model_name,
+        model_temperature=settings.temperature,
+        model_timeout_seconds=settings.timeout_seconds,
+        model_max_retries=settings.max_retries,
+        prompt_version_topic=settings.prompt_version_topic,
+        prompt_version_term=settings.prompt_version_term,
+        prompt_version_variable=settings.prompt_version_variable,
+        application_version=application_version,
+        configuration_hash=configuration_hash(relevant_config),
+    )
+
+
+def mark_cache_used(result: ArticleResult) -> ArticleResult:
+    """Return a cached copy of a result marked as completed work from cache."""
+    metadata = result.processing_metadata.model_copy(update={"cache_used": True})
+    return result.model_copy(update={"processing_metadata": metadata})
+
+
+def _atomic_write_json(path: Path, payload: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temp_path = path.with_name(f".{path.name}.tmp")
+    temp_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
+    os.replace(temp_path, path)

src/gcmd_classifier/persistence/json_store.py

@@ -0,0 +1,82 @@
+"""Incremental JSON persistence for article-level classifier results."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+from pathlib import Path
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from gcmd_classifier.models import ArticleResult, RunSummary
+
+
+class PersistedRunOutput(BaseModel):
+    """Consolidated machine-readable run output."""
+
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    summary: RunSummary
+    articles: tuple[ArticleResult, ...] = Field(default_factory=tuple)
+
+
+class JsonResultStore:
+    """Checkpoint-safe JSON store using one atomic article file per DOI.
+
+    One-file-per-DOI avoids rewriting a large consolidated document after every article.
+    A final JSON file can still be emitted by atomically replacing a temp file in the
+    destination directory, so the temp and destination files live on the same filesystem.
+    """
+
+    def __init__(self, directory: str | Path, *, consolidated_name: str = "results.json") -> None:
+        self.directory = Path(directory)
+        self.articles_dir = self.directory / "articles"
+        self.consolidated_path = self.directory / consolidated_name
+        self.articles_dir.mkdir(parents=True, exist_ok=True)
+
+    def article_path(self, doi: str) -> Path:
+        """Return the stable checkpoint path for an article DOI."""
+        return self.articles_dir / f"{doi_to_key(doi)}.json"
+
+    def save_article_result(self, result: ArticleResult) -> None:
+        """Save one article result with an atomic same-directory replace."""
+        _atomic_write_json(self.article_path(result.DOI), result.model_dump(mode="json"))
+
+    def load_article_result(self, doi: str) -> ArticleResult | None:
+        """Load a previously saved article result by DOI."""
+        path = self.article_path(doi)
+        if not path.exists():
+            return None
+        return ArticleResult.model_validate(json.loads(path.read_text()))
+
+    def load_all_results(self) -> tuple[ArticleResult, ...]:
+        """Load all checkpointed article results in deterministic filename order."""
+        results: list[ArticleResult] = []
+        for path in sorted(self.articles_dir.glob("*.json")):
+            results.append(ArticleResult.model_validate(json.loads(path.read_text())))
+        return tuple(results)
+
+    def write_consolidated(
+        self,
+        *,
+        results: tuple[ArticleResult, ...] | list[ArticleResult],
+        summary: RunSummary,
+    ) -> Path:
+        """Write a consolidated JSON output file with summary and article results."""
+        payload = PersistedRunOutput(summary=summary, articles=tuple(results))
+        _atomic_write_json(self.consolidated_path, payload.model_dump(mode="json"))
+        return self.consolidated_path
+
+
+def doi_to_key(doi: str) -> str:
+    """Create a filesystem-safe, deterministic key for a DOI."""
+    return hashlib.sha256(doi.encode("utf-8")).hexdigest()
+
+
+def _atomic_write_json(path: Path, payload: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temp_path = path.with_name(f".{path.name}.tmp")
+    temp_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
+    os.replace(temp_path, path)

src/gcmd_classifier/pipeline/__init__.py

@@ -0,0 +1,16 @@
+"""Pipeline orchestration for the GCMD classifier MVP."""
+
+from gcmd_classifier.pipeline.batch import BatchRunResult, run_batch
+from gcmd_classifier.pipeline.service import (
+    APPLICATION_VERSION,
+    classify_article,
+    failed_article_result,
+)
+
+__all__ = [
+    "APPLICATION_VERSION",
+    "BatchRunResult",
+    "classify_article",
+    "failed_article_result",
+    "run_batch",
+]

src/gcmd_classifier/pipeline/batch.py

@@ -0,0 +1,259 @@
+"""Batch orchestration for the GCMD classifier MVP."""
+
+from __future__ import annotations
+
+import logging
+import time
+from datetime import UTC, datetime
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from gcmd_classifier.config import ModelSettings
+from gcmd_classifier.llm.base import ModelClient
+from gcmd_classifier.logging_config import get_logger, log_event
+from gcmd_classifier.models import (
+    ArticleClassificationOutcome,
+    ArticleLoadResult,
+    ArticleProcessingStatus,
+    ArticleResult,
+    ArticleValidationIssue,
+    OutputError,
+    RunSummary,
+)
+from gcmd_classifier.persistence.cache import (
+    ArticleResultCache,
+    build_cache_identity,
+)
+from gcmd_classifier.persistence.json_store import JsonResultStore
+from gcmd_classifier.pipeline.service import (
+    APPLICATION_VERSION,
+    classify_article,
+    failed_article_result,
+)
+from gcmd_classifier.vocabulary.index import VocabularyIndex
+
+
+class BatchRunResult(BaseModel):
+    """Structured batch result with article outputs and run summary."""
+
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    results: tuple[ArticleResult, ...] = Field(default_factory=tuple)
+    summary: RunSummary
+
+
+def run_batch(
+    *,
+    article_load_result: ArticleLoadResult,
+    vocabulary: VocabularyIndex,
+    model_client: ModelClient,
+    settings: ModelSettings,
+    store: JsonResultStore,
+    cache: ArticleResultCache | None = None,
+    force_reprocess: bool = False,
+    run_id: str | None = None,
+    application_version: str = APPLICATION_VERSION,
+    relevant_config: dict[str, Any] | None = None,
+    logger: logging.Logger | None = None,
+) -> BatchRunResult:
+    """Process valid articles in source order while preserving completed results."""
+    active_logger = logger or get_logger(__name__)
+    started = time.perf_counter()
+    started_at = _now_iso()
+    actual_run_id = run_id or f"run-{started_at}"
+    invalid_errors = tuple(_issue_to_error(issue) for issue in article_load_result.errors)
+    results: list[ArticleResult] = []
+    cache_hits = 0
+    cache_misses = 0
+
+    log_event(
+        active_logger,
+        "batch_started",
+        stage="batch",
+        source_records=article_load_result.source_count,
+        valid_records=len(article_load_result.articles),
+        invalid_records=_invalid_record_count(article_load_result.errors),
+    )
+
+    for index, article in enumerate(article_load_result.articles):
+        try:
+            result: ArticleResult | None = None
+            identity = build_cache_identity(
+                article=article,
+                vocabulary=vocabulary,
+                settings=settings,
+                application_version=application_version,
+                relevant_config=relevant_config,
+            )
+            if cache is not None and not force_reprocess:
+                result = cache.get(identity)
+                if result is not None:
+                    cache_hits += 1
+                    log_event(
+                        active_logger,
+                        "cache_hit",
+                        DOI=article.DOI,
+                        index=index,
+                        stage="cache",
+                        cache_key=identity.cache_key,
+                    )
+                else:
+                    cache_misses += 1
+                    log_event(
+                        active_logger,
+                        "cache_miss",
+                        DOI=article.DOI,
+                        index=index,
+                        stage="cache",
+                        cache_key=identity.cache_key,
+                    )
+            elif cache is not None:
+                cache_misses += 1
+
+            if result is None:
+                result = classify_article(
+                    article=article,
+                    vocabulary=vocabulary,
+                    model_client=model_client,
+                    settings=settings,
+                    run_id=actual_run_id,
+                    application_version=application_version,
+                    relevant_config=relevant_config,
+                    logger=active_logger,
+                )
+                if cache is not None:
+                    cache.put(identity, result)
+            else:
+                result = result.model_copy(
+                    update={
+                        "processing_metadata": result.processing_metadata.model_copy(
+                            update={"run_id": actual_run_id}
+                        )
+                    }
+                )
+        except Exception as exc:
+            result = failed_article_result(
+                article=article,
+                error=OutputError(
+                    code=exc.__class__.__name__,
+                    message=str(exc),
+                    stage="batch",
+                    DOI=article.DOI,
+                ),
+                vocabulary=vocabulary,
+                settings=settings,
+                run_id=actual_run_id,
+                application_version=application_version,
+                relevant_config=relevant_config,
+            )
+            log_event(
+                active_logger,
+                "article_failed",
+                DOI=article.DOI,
+                index=index,
+                stage="batch",
+                error_code=exc.__class__.__name__,
+            )
+        store.save_article_result(result)
+        results.append(result)
+
+    summary = _run_summary(
+        run_id=actual_run_id,
+        started_at=started_at,
+        completed_at=_now_iso(),
+        duration_seconds=time.perf_counter() - started,
+        article_load_result=article_load_result,
+        results=tuple(results),
+        invalid_errors=invalid_errors,
+        cache_hits=cache_hits,
+        cache_misses=cache_misses,
+    )
+    store.write_consolidated(results=tuple(results), summary=summary)
+    log_event(
+        active_logger,
+        "batch_finished",
+        stage="batch",
+        run_id=actual_run_id,
+        processed_articles=len(results),
+        cache_hits=cache_hits,
+        cache_misses=cache_misses,
+        errors=summary.total_errors,
+    )
+    return BatchRunResult(results=tuple(results), summary=summary)
+
+
+def _run_summary(
+    *,
+    run_id: str,
+    started_at: str,
+    completed_at: str,
+    duration_seconds: float,
+    article_load_result: ArticleLoadResult,
+    results: tuple[ArticleResult, ...],
+    invalid_errors: tuple[OutputError, ...],
+    cache_hits: int,
+    cache_misses: int,
+) -> RunSummary:
+    warnings_count = sum(len(result.warnings) for result in results)
+    result_errors = tuple(error for result in results for error in result.errors)
+    errors = (*invalid_errors, *result_errors)
+    completed = sum(
+        result.processing_status is ArticleProcessingStatus.COMPLETED for result in results
+    )
+    partial = sum(result.processing_status is ArticleProcessingStatus.PARTIAL for result in results)
+    failed = sum(result.processing_status is ArticleProcessingStatus.FAILED for result in results)
+    skipped = sum(result.processing_status is ArticleProcessingStatus.SKIPPED for result in results)
+    classifications = sum(len(result.classifications) for result in results)
+    processed = len(results)
+    return RunSummary(
+        run_id=run_id,
+        started_at=started_at,
+        completed_at=completed_at,
+        articles_received=article_load_result.source_count,
+        valid_article_records=len(article_load_result.articles),
+        invalid_source_records=_invalid_record_count(article_load_result.errors),
+        processed_articles=processed,
+        cache_hits=cache_hits,
+        cache_misses=cache_misses,
+        total_warnings=warnings_count,
+        total_errors=len(errors),
+        duration_seconds=duration_seconds,
+        articles_completed=completed,
+        articles_partial=partial,
+        articles_failed=failed,
+        articles_skipped=skipped,
+        articles_not_classified=sum(
+            result.classification_outcome is ArticleClassificationOutcome.NOT_CLASSIFIED
+            for result in results
+        ),
+        accepted_classifications=classifications,
+        average_classifications_per_article=(classifications / processed if processed else None),
+        average_processing_time_seconds=(duration_seconds / processed if processed else None),
+        total_model_calls=sum(
+            result.processing_metadata.model_calls or 0
+            for result in results
+            if not result.processing_metadata.cache_used
+        ),
+        errors=errors,
+    )
+
+
+def _issue_to_error(issue: ArticleValidationIssue) -> OutputError:
+    return OutputError(
+        code=issue.code,
+        message=issue.message,
+        stage="article_loading",
+        details={"field": issue.field},
+        index=issue.index,
+        DOI=issue.DOI,
+    )
+
+
+def _invalid_record_count(issues: tuple[ArticleValidationIssue, ...]) -> int:
+    indices = {issue.index for issue in issues if issue.index is not None}
+    return len(indices)
+
+
+def _now_iso() -> str:
+    return datetime.now(UTC).isoformat()

src/gcmd_classifier/pipeline/service.py

@@ -0,0 +1,388 @@
+"""Single-article MVP classification orchestration."""
+
+from __future__ import annotations
+
+import logging
+import time
+from datetime import UTC, datetime
+from typing import Any
+
+from gcmd_classifier.classification import (
+    ClassificationCandidate,
+    TermBranchSeed,
+    candidate_from_terminal_outcome,
+    candidate_from_topic_stop,
+    remove_redundant_classifications,
+    route_terms,
+    route_topics,
+    validate_candidates,
+)
+from gcmd_classifier.config import ModelSettings
+from gcmd_classifier.llm.base import ModelClient
+from gcmd_classifier.logging_config import get_logger, log_event
+from gcmd_classifier.models import (
+    ArticleClassificationOutcome,
+    ArticleProcessingStatus,
+    ArticleRecord,
+    ArticleResult,
+    ClassificationRecord,
+    OutputError,
+    OutputWarning,
+    ProcessingMetadata,
+    ReviewStatus,
+)
+from gcmd_classifier.persistence.cache import article_fingerprint, configuration_hash
+from gcmd_classifier.vocabulary.index import VocabularyIndex
+
+APPLICATION_VERSION = "0.1.0"
+
+
+def classify_article(
+    *,
+    article: ArticleRecord,
+    vocabulary: VocabularyIndex,
+    model_client: ModelClient,
+    settings: ModelSettings,
+    run_id: str | None = None,
+    application_version: str = APPLICATION_VERSION,
+    relevant_config: dict[str, Any] | None = None,
+    logger: logging.Logger | None = None,
+) -> ArticleResult:
+    """Classify one valid article through the current MVP pipeline."""
+    active_logger = logger or get_logger(__name__)
+    started_at = _now_iso()
+    started = time.perf_counter()
+    model_calls_before = _model_request_count(model_client)
+    log_event(active_logger, "article_started", DOI=article.DOI, stage="pipeline")
+
+    candidates: list[ClassificationCandidate] = []
+    warnings: list[OutputWarning] = []
+    errors: list[OutputError] = []
+    no_classification_reason: str | None = None
+
+    try:
+        topic_result = route_topics(
+            article=article,
+            vocabulary=vocabulary,
+            model_client=model_client,
+            settings=settings,
+        )
+        if topic_result.is_no_topic:
+            no_classification_reason = (
+                topic_result.no_selection_reason or "No GCMD Topic was supported by the article."
+            )
+        for topic_branch in topic_result.branches:
+            try:
+                term_result = route_terms(
+                    article=article,
+                    topic_branch=topic_branch,
+                    vocabulary=vocabulary,
+                    model_client=model_client,
+                    settings=settings,
+                )
+            except Exception as exc:
+                errors.append(_error_from_exception(exc, stage="term_routing", DOI=article.DOI))
+                log_event(
+                    active_logger,
+                    "branch_failed",
+                    DOI=article.DOI,
+                    stage="term_routing",
+                    branch_id=topic_branch.branch_id,
+                    error_code=exc.__class__.__name__,
+                )
+                continue
+
+            if term_result.stop_at_topic is not None:
+                candidates.append(candidate_from_topic_stop(term_result.stop_at_topic))
+            for term_branch in term_result.term_branches:
+                _process_term_branch(
+                    article=article,
+                    term_branch=term_branch,
+                    vocabulary=vocabulary,
+                    model_client=model_client,
+                    settings=settings,
+                    candidates=candidates,
+                    warnings=warnings,
+                    errors=errors,
+                    logger=active_logger,
+                )
+    except Exception as exc:
+        errors.append(_error_from_exception(exc, stage="topic_routing", DOI=article.DOI))
+        log_event(
+            active_logger,
+            "article_failed",
+            DOI=article.DOI,
+            stage="topic_routing",
+            error_code=exc.__class__.__name__,
+        )
+
+    validation_result = validate_candidates(candidates, vocabulary)
+    for rejected in validation_result.rejected:
+        errors.extend(rejected.deterministic_validation.errors)
+    redundancy_result = remove_redundant_classifications(validation_result.accepted, vocabulary)
+    warnings.extend(redundancy_result.warnings)
+    classifications = redundancy_result.classifications
+
+    result = _article_result(
+        article=article,
+        classifications=classifications,
+        warnings=tuple(warnings),
+        errors=tuple(errors),
+        no_classification_reason=no_classification_reason,
+        started_at=started_at,
+        duration_seconds=time.perf_counter() - started,
+        model_calls=_model_calls_used(model_client, model_calls_before),
+        run_id=run_id,
+        vocabulary=vocabulary,
+        settings=settings,
+        application_version=application_version,
+        relevant_config=relevant_config,
+    )
+    log_event(
+        active_logger,
+        "article_finished",
+        DOI=article.DOI,
+        stage="pipeline",
+        processing_status=result.processing_status.value,
+        classification_outcome=None
+        if result.classification_outcome is None
+        else result.classification_outcome.value,
+        classifications=len(result.classifications),
+        errors=len(result.errors),
+    )
+    return result
+
+
+def _process_term_branch(
+    *,
+    article: ArticleRecord,
+    term_branch: TermBranchSeed,
+    vocabulary: VocabularyIndex,
+    model_client: ModelClient,
+    settings: ModelSettings,
+    candidates: list[ClassificationCandidate],
+    warnings: list[OutputWarning],
+    errors: list[OutputError],
+    logger: logging.Logger,
+) -> None:
+    from gcmd_classifier.classification import descend_variables
+
+    try:
+        descent_result = descend_variables(
+            article=article,
+            term_branch=term_branch,
+            vocabulary=vocabulary,
+            model_client=model_client,
+            settings=settings,
+        )
+    except Exception as exc:
+        errors.append(_error_from_exception(exc, stage="variable_descent", DOI=article.DOI))
+        log_event(
+            logger,
+            "branch_failed",
+            DOI=article.DOI,
+            stage="variable_descent",
+            branch_id=term_branch.branch_id,
+            error_code=exc.__class__.__name__,
+        )
+        return
+
+    candidates.extend(
+        candidate_from_terminal_outcome(terminal) for terminal in descent_result.terminals
+    )
+    warnings.extend(descent_result.warnings)
+    errors.extend(descent_result.diagnostics)
+    for branch_error in descent_result.errors:
+        errors.append(
+            OutputError(
+                code=branch_error.code,
+                message=branch_error.message,
+                stage="variable_descent",
+                DOI=article.DOI,
+                details={
+                    "branch_id": branch_error.branch_id,
+                    "parent_branch_id": branch_error.parent_branch_id,
+                    "parent_uuid": branch_error.parent_uuid,
+                    "parent_level": branch_error.parent_level,
+                },
+            )
+        )
+
+
+def _article_result(
+    *,
+    article: ArticleRecord,
+    classifications: tuple[ClassificationRecord, ...],
+    warnings: tuple[OutputWarning, ...],
+    errors: tuple[OutputError, ...],
+    no_classification_reason: str | None,
+    started_at: str,
+    duration_seconds: float,
+    model_calls: int | None,
+    run_id: str | None,
+    vocabulary: VocabularyIndex,
+    settings: ModelSettings,
+    application_version: str,
+    relevant_config: dict[str, Any] | None,
+) -> ArticleResult:
+    metadata = _metadata(
+        article=article,
+        started_at=started_at,
+        duration_seconds=duration_seconds,
+        model_calls=model_calls,
+        run_id=run_id,
+        vocabulary=vocabulary,
+        settings=settings,
+        application_version=application_version,
+        relevant_config=relevant_config,
+    )
+    if classifications:
+        status = ArticleProcessingStatus.PARTIAL if errors else ArticleProcessingStatus.COMPLETED
+        return ArticleResult(
+            DOI=article.DOI,
+            Title=article.Title,
+            Year=article.Year,
+            Abstract=article.Abstract,
+            processing_status=status,
+            classification_outcome=ArticleClassificationOutcome.CLASSIFIED,
+            classifications=classifications,
+            review_status=ReviewStatus.NOT_REQUIRED,
+            warnings=warnings,
+            errors=errors,
+            processing_metadata=metadata,
+        )
+    if errors:
+        return ArticleResult(
+            DOI=article.DOI,
+            Title=article.Title,
+            Year=article.Year,
+            Abstract=article.Abstract,
+            processing_status=ArticleProcessingStatus.FAILED,
+            classification_outcome=None,
+            classifications=(),
+            review_status=ReviewStatus.NOT_REQUIRED,
+            warnings=warnings,
+            errors=errors,
+            processing_metadata=metadata,
+        )
+    return ArticleResult(
+        DOI=article.DOI,
+        Title=article.Title,
+        Year=article.Year,
+        Abstract=article.Abstract,
+        processing_status=ArticleProcessingStatus.COMPLETED,
+        classification_outcome=ArticleClassificationOutcome.NOT_CLASSIFIED,
+        classifications=(),
+        no_classification_reason=no_classification_reason
+        or "No defensible GCMD classification was supported.",
+        review_status=ReviewStatus.NOT_REQUIRED,
+        warnings=warnings,
+        errors=errors,
+        processing_metadata=metadata,
+    )
+
+
+def failed_article_result(
+    *,
+    article: ArticleRecord,
+    error: OutputError,
+    vocabulary: VocabularyIndex,
+    settings: ModelSettings,
+    run_id: str | None = None,
+    application_version: str = APPLICATION_VERSION,
+    relevant_config: dict[str, Any] | None = None,
+) -> ArticleResult:
+    """Build and return a persisted failed article result for batch-level failures."""
+    started_at = _now_iso()
+    return ArticleResult(
+        DOI=article.DOI,
+        Title=article.Title,
+        Year=article.Year,
+        Abstract=article.Abstract,
+        processing_status=ArticleProcessingStatus.FAILED,
+        classification_outcome=None,
+        classifications=(),
+        review_status=ReviewStatus.NOT_REQUIRED,
+        errors=(error,),
+        processing_metadata=_metadata(
+            article=article,
+            started_at=started_at,
+            duration_seconds=0.0,
+            model_calls=None,
+            run_id=run_id,
+            vocabulary=vocabulary,
+            settings=settings,
+            application_version=application_version,
+            relevant_config=relevant_config,
+        ),
+    )
+
+
+def _metadata(
+    *,
+    article: ArticleRecord,
+    started_at: str,
+    duration_seconds: float,
+    model_calls: int | None,
+    run_id: str | None,
+    vocabulary: VocabularyIndex,
+    settings: ModelSettings,
+    application_version: str,
+    relevant_config: dict[str, Any] | None,
+) -> ProcessingMetadata:
+    completed_at = _now_iso()
+    return ProcessingMetadata(
+        run_id=run_id,
+        started_at=started_at,
+        completed_at=completed_at,
+        processed_at=completed_at,
+        model_provider=settings.provider,
+        model_name=settings.model_name,
+        model_parameters={
+            "temperature": settings.temperature,
+            "timeout_seconds": settings.timeout_seconds,
+            "max_retries": settings.max_retries,
+        },
+        prompt_versions={
+            "topic": settings.prompt_version_topic,
+            "term": settings.prompt_version_term,
+            "variable": settings.prompt_version_variable,
+        },
+        vocabulary_version=vocabulary.vocabulary_version,
+        vocabulary_hash=vocabulary.vocabulary_version,
+        application_version=application_version,
+        configuration_hash=configuration_hash(relevant_config),
+        article_fingerprint=article_fingerprint(article),
+        cache_used=False,
+        processing_time_seconds=duration_seconds,
+        model_calls=model_calls,
+        title_available=bool(article.Title),
+        abstract_available=bool(article.Abstract),
+    )
+
+
+def _error_from_exception(exc: Exception, *, stage: str, DOI: str | None = None) -> OutputError:
+    retry_count = getattr(exc, "retry_count", None)
+    return OutputError(
+        code=exc.__class__.__name__,
+        message=str(exc),
+        stage=stage,
+        DOI=DOI,
+        retry_count=retry_count if isinstance(retry_count, int) else None,
+    )
+
+
+def _model_request_count(model_client: ModelClient) -> int | None:
+    requests = getattr(model_client, "requests", None)
+    return len(requests) if isinstance(requests, list) else None
+
+
+def _model_calls_used(model_client: ModelClient, before: int | None) -> int | None:
+    after = _model_request_count(model_client)
+    if before is None or after is None:
+        return None
+    return after - before
+
+
+def _now_iso() -> str:
+    return datetime.now(UTC).isoformat()

tests/test_cache.py

@@ -0,0 +1,121 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from gcmd_classifier.config import ModelSettings
+from gcmd_classifier.models import (
+    ArticleClassificationOutcome,
+    ArticleProcessingStatus,
+    ArticleRecord,
+    ArticleResult,
+    ProcessingMetadata,
+    ReviewStatus,
+)
+from gcmd_classifier.persistence import ArticleResultCache, build_cache_identity
+from gcmd_classifier.vocabulary import load_vocabulary
+
+FIXTURE_PATH = Path("tests/fixtures/gcmd_hierarchy_small.json")
+
+
+def _article(**overrides: object) -> ArticleRecord:
+    values = {
+        "DOI": "10.example/cache",
+        "Title": "Carbon dioxide observations",
+        "Year": 2025,
+        "Abstract": "Atmospheric carbon dioxide profiles are discussed.",
+    }
+    values.update(overrides)
+    return ArticleRecord.model_validate(values)
+
+
+def _result(article: ArticleRecord) -> ArticleResult:
+    return ArticleResult(
+        DOI=article.DOI,
+        Title=article.Title,
+        Year=article.Year,
+        Abstract=article.Abstract,
+        processing_status=ArticleProcessingStatus.COMPLETED,
+        classification_outcome=ArticleClassificationOutcome.NOT_CLASSIFIED,
+        classifications=(),
+        no_classification_reason="No Topic selected.",
+        review_status=ReviewStatus.NOT_REQUIRED,
+        processing_metadata=ProcessingMetadata(cache_used=False),
+    )
+
+
+def _identity(article: ArticleRecord, settings: ModelSettings | None = None, **kwargs):
+    return build_cache_identity(
+        article=article,
+        vocabulary=load_vocabulary(FIXTURE_PATH),
+        settings=settings or ModelSettings(),
+        application_version=kwargs.pop("application_version", "0.1.0"),
+        relevant_config=kwargs.pop("relevant_config", None),
+    )
+
+
+def test_cache_key_includes_article_fingerprint() -> None:
+    original = _identity(_article())
+    changed = _identity(_article(Title="Changed title"))
+
+    assert original.article_fingerprint != changed.article_fingerprint
+    assert original.cache_key != changed.cache_key
+
+
+def test_cache_key_changes_when_article_content_changes() -> None:
+    assert (
+        _identity(_article(Abstract="A")).cache_key != _identity(_article(Abstract="B")).cache_key
+    )
+
+
+def test_cache_key_changes_when_vocabulary_hash_changes() -> None:
+    identity = _identity(_article())
+    changed = identity.model_copy(update={"vocabulary_version": "different-vocab"})
+
+    assert identity.cache_key != changed.cache_key
+
+
+def test_cache_key_changes_when_model_provider_name_or_parameters_change() -> None:
+    default = _identity(_article())
+    provider = _identity(_article(), ModelSettings(provider="other"))
+    name = _identity(_article(), ModelSettings(model_name="other-model"))
+    temperature = _identity(_article(), ModelSettings(temperature=0.2))
+
+    assert len({default.cache_key, provider.cache_key, name.cache_key, temperature.cache_key}) == 4
+
+
+def test_cache_key_changes_when_prompt_versions_change() -> None:
+    default = _identity(_article())
+    changed = _identity(_article(), ModelSettings(prompt_version_topic="topic-v2"))
+
+    assert default.cache_key != changed.cache_key
+
+
+def test_cache_key_changes_when_relevant_config_changes() -> None:
+    default = _identity(_article(), relevant_config={"candidate_limit": 100})
+    changed = _identity(_article(), relevant_config={"candidate_limit": 200})
+
+    assert default.cache_key != changed.cache_key
+
+
+def test_cache_hit_returns_cache_used_true_and_completed_status(tmp_path: Path) -> None:
+    article = _article()
+    identity = _identity(article)
+    cache = ArticleResultCache(tmp_path)
+    cache.put(identity, _result(article))
+
+    cached = cache.get(identity)
+
+    assert cached is not None
+    assert cached.processing_metadata.cache_used is True
+    assert cached.processing_status is ArticleProcessingStatus.COMPLETED
+    assert cached.processing_status is not ArticleProcessingStatus.SKIPPED
+
+
+def test_stale_cache_is_not_reused(tmp_path: Path) -> None:
+    article = _article()
+    identity = _identity(article)
+    stale_identity = _identity(_article(Title="Changed title"))
+    cache = ArticleResultCache(tmp_path)
+    cache.put(identity, _result(article))
+
+    assert cache.get(stale_identity) is None

tests/test_persistence.py

@@ -0,0 +1,122 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from gcmd_classifier.models import (
+    ArticleClassificationOutcome,
+    ArticleProcessingStatus,
+    ArticleResult,
+    OutputError,
+    ProcessingMetadata,
+    ReviewStatus,
+    RunSummary,
+)
+from gcmd_classifier.persistence import JsonResultStore
+
+
+def _result(doi: str, *, failed: bool = False) -> ArticleResult:
+    if failed:
+        return ArticleResult(
+            DOI=doi,
+            Title="Failed article",
+            Year=2025,
+            Abstract="Text.",
+            processing_status=ArticleProcessingStatus.FAILED,
+            classification_outcome=None,
+            classifications=(),
+            review_status=ReviewStatus.NOT_REQUIRED,
+            errors=(OutputError(code="MODEL_FAILED", message="Model failed."),),
+            processing_metadata=ProcessingMetadata(cache_used=False),
+        )
+    return ArticleResult(
+        DOI=doi,
+        Title="Completed article",
+        Year=2025,
+        Abstract="Text.",
+        processing_status=ArticleProcessingStatus.COMPLETED,
+        classification_outcome=ArticleClassificationOutcome.NOT_CLASSIFIED,
+        classifications=(),
+        no_classification_reason="No Topic selected.",
+        review_status=ReviewStatus.NOT_REQUIRED,
+        processing_metadata=ProcessingMetadata(cache_used=False),
+    )
+
+
+def test_article_result_is_saved_after_completion(tmp_path: Path) -> None:
+    store = JsonResultStore(tmp_path)
+    result = _result("10.example/done")
+
+    store.save_article_result(result)
+
+    loaded = store.load_article_result("10.example/done")
+    assert loaded == result
+
+
+def test_failed_article_result_is_saved(tmp_path: Path) -> None:
+    store = JsonResultStore(tmp_path)
+    result = _result("10.example/failed", failed=True)
+
+    store.save_article_result(result)
+
+    assert store.load_article_result("10.example/failed") == result
+
+
+def test_previous_results_survive_simulated_later_failure(tmp_path: Path) -> None:
+    store = JsonResultStore(tmp_path)
+    first = _result("10.example/first")
+    store.save_article_result(first)
+
+    with pytest.raises(RuntimeError):
+        raise RuntimeError("later article failed before save")
+
+    assert store.load_article_result("10.example/first") == first
+
+
+def test_loading_existing_results_works(tmp_path: Path) -> None:
+    store = JsonResultStore(tmp_path)
+    first = _result("10.example/first")
+    second = _result("10.example/second", failed=True)
+    store.save_article_result(first)
+    store.save_article_result(second)
+
+    loaded = store.load_all_results()
+
+    assert {result.DOI for result in loaded} == {"10.example/first", "10.example/second"}
+
+
+def test_duplicate_output_records_for_same_doi_are_avoided(tmp_path: Path) -> None:
+    store = JsonResultStore(tmp_path)
+    first = _result("10.example/same")
+    second = _result("10.example/same", failed=True)
+
+    store.save_article_result(first)
+    store.save_article_result(second)
+
+    loaded = store.load_all_results()
+    assert len(loaded) == 1
+    assert loaded[0].processing_status is ArticleProcessingStatus.FAILED
+
+
+def test_consolidated_output_is_valid_json(tmp_path: Path) -> None:
+    store = JsonResultStore(tmp_path)
+    result = _result("10.example/done")
+    summary = RunSummary(run_id="run-1", articles_received=1, articles_completed=1)
+
+    output_path = store.write_consolidated(results=[result], summary=summary)
+
+    payload = json.loads(output_path.read_text())
+    assert payload["summary"]["run_id"] == "run-1"
+    assert payload["articles"][0]["DOI"] == "10.example/done"
+
+
+def test_atomic_checkpoint_temp_file_is_removed_after_success(tmp_path: Path) -> None:
+    store = JsonResultStore(tmp_path)
+    result = _result("10.example/done")
+
+    store.save_article_result(result)
+
+    assert store.article_path(result.DOI).exists()
+    assert not list(store.article_path(result.DOI).parent.glob("*.tmp"))

tests/test_pipeline.py

@@ -0,0 +1,405 @@
+from __future__ import annotations
+
+import json
+import logging
+from pathlib import Path
+
+from jsonschema import Draft202012Validator
+
+from gcmd_classifier.articles import validate_article_records
+from gcmd_classifier.config import ModelSettings
+from gcmd_classifier.llm import FakeModelClient
+from gcmd_classifier.logging_config import sanitize_log_details
+from gcmd_classifier.models import (
+    ArticleClassificationOutcome,
+    ArticleProcessingStatus,
+    ArticleRecord,
+)
+from gcmd_classifier.persistence import ArticleResultCache, JsonResultStore
+from gcmd_classifier.pipeline import classify_article, run_batch
+from gcmd_classifier.vocabulary import load_vocabulary
+
+FIXTURE_PATH = Path("tests/fixtures/gcmd_hierarchy_small.json")
+FULL_ARTICLES_PATH = Path("data/articles.json")
+FULL_HIERARCHY_PATH = Path("data/gcmd_hierarchy.json")
+RUN_SUMMARY_SCHEMA_PATH = Path("schemas/run_summary.schema.json")
+
+
+def _index():
+    return load_vocabulary(FIXTURE_PATH)
+
+
+def _article(abstract: str = "Atmospheric carbon dioxide profiles are discussed.") -> ArticleRecord:
+    return ArticleRecord(
+        DOI="10.example/pipeline",
+        Title="Atmospheric carbon dioxide observations",
+        Year=2025,
+        Abstract=abstract,
+    )
+
+
+def _decision(candidate_id: str) -> dict:
+    return {
+        "candidate_id": candidate_id,
+        "confidence": 0.8,
+        "evidence": f"Evidence for {candidate_id}.",
+        "support_type": "explicit",
+        "reason": f"Reason for {candidate_id}.",
+    }
+
+
+def _no_topic(reason: str = "No supported Topic.") -> dict:
+    return {"selected": [], "no_selection_reason": reason}
+
+
+def _select_topic(candidate_id: str = "topic_0001") -> dict:
+    return {"selected": [_decision(candidate_id)]}
+
+
+def _select_term(candidate_id: str = "term_0001") -> dict:
+    return {"selected": [_decision(candidate_id)], "stop_at_parent": False}
+
+
+def _select_variable(*candidate_ids: str) -> dict:
+    return {
+        "selected": [_decision(candidate_id) for candidate_id in candidate_ids],
+        "stop_at_parent": False,
+    }
+
+
+def _stop(reason: str) -> dict:
+    return {"selected": [], "stop_at_parent": True, "stop_reason": reason}
+
+
+def test_single_article_deep_variable_classification_is_accepted() -> None:
+    client = FakeModelClient(
+        [
+            _select_topic(),
+            _select_term(),
+            _select_variable("variable_0001"),
+            _select_variable("variable_0001"),
+            _select_variable("variable_0001"),
+        ]
+    )
+
+    result = classify_article(
+        article=_article(),
+        vocabulary=_index(),
+        model_client=client,
+        settings=ModelSettings(),
+    )
+
+    assert result.processing_status is ArticleProcessingStatus.COMPLETED
+    assert result.classification_outcome is ArticleClassificationOutcome.CLASSIFIED
+    assert [record.UUID for record in result.classifications] == ["vl3-carbon-dioxide-profiles"]
+    assert result.classifications[0].deterministic_validation.valid is True
+    assert result.processing_metadata.model_calls == 5
+
+
+def test_article_stops_at_topic_and_becomes_topic_classification() -> None:
+    client = FakeModelClient([_select_topic(), _stop("Topic is the deepest supported level.")])
+
+    result = classify_article(
+        article=_article(),
+        vocabulary=_index(),
+        model_client=client,
+        settings=ModelSettings(),
+    )
+
+    assert result.classifications[0].UUID == "topic-atmosphere"
+    assert result.classifications[0].level == "Topic"
+    assert result.classifications[0].reason_for_stopping == "Topic is the deepest supported level."
+
+
+def test_article_stops_at_term_and_becomes_term_classification() -> None:
+    client = FakeModelClient(
+        [_select_topic(), _select_term(), _stop("Term is the deepest supported level.")]
+    )
+
+    result = classify_article(
+        article=_article(),
+        vocabulary=_index(),
+        model_client=client,
+        settings=ModelSettings(),
+    )
+
+    assert result.classifications[0].UUID == "term-atmospheric-chemistry"
+    assert result.classifications[0].level == "Term"
+    assert result.classifications[0].reason_for_stopping == "Term is the deepest supported level."
+
+
+def test_no_topic_selection_is_completed_not_classified() -> None:
+    result = classify_article(
+        article=_article(),
+        vocabulary=_index(),
+        model_client=FakeModelClient([_no_topic("Not Earth science.")]),
+        settings=ModelSettings(),
+    )
+
+    assert result.processing_status is ArticleProcessingStatus.COMPLETED
+    assert result.classification_outcome is ArticleClassificationOutcome.NOT_CLASSIFIED
+    assert result.classifications == ()
+    assert result.no_classification_reason == "Not Earth science."
+
+
+def test_invalid_model_candidate_produces_structured_error() -> None:
+    result = classify_article(
+        article=_article(),
+        vocabulary=_index(),
+        model_client=FakeModelClient([_select_topic("missing_topic")]),
+        settings=ModelSettings(),
+    )
+
+    assert result.processing_status is ArticleProcessingStatus.FAILED
+    assert result.classifications == ()
+    assert result.errors[0].stage == "topic_routing"
+    assert result.errors[0].code == "UnknownCandidateIDError"
+
+
+def test_partial_branch_failure_preserves_successful_sibling_outcome() -> None:
+    client = FakeModelClient(
+        [
+            _select_topic(),
+            _select_term(),
+            _select_variable("variable_0001"),
+            _select_variable("variable_0001", "variable_0002"),
+            _select_variable("missing_variable"),
+        ]
+    )
+
+    result = classify_article(
+        article=_article(),
+        vocabulary=_index(),
+        model_client=client,
+        settings=ModelSettings(),
+    )
+
+    assert result.processing_status is ArticleProcessingStatus.PARTIAL
+    assert result.classification_outcome is ArticleClassificationOutcome.CLASSIFIED
+    assert [record.UUID for record in result.classifications] == ["vl2-methane"]
+    assert result.errors
+    assert result.errors[0].stage == "variable_descent"
+
+
+def test_empty_abstract_remains_valid_and_can_be_processed() -> None:
+    result = classify_article(
+        article=_article(abstract=""),
+        vocabulary=_index(),
+        model_client=FakeModelClient([_no_topic("Title only is insufficient.")]),
+        settings=ModelSettings(),
+    )
+
+    assert result.Abstract == ""
+    assert result.processing_metadata.abstract_available is False
+    assert result.processing_status is ArticleProcessingStatus.COMPLETED
+
+
+def test_batch_processes_multiple_valid_articles_and_preserves_order(tmp_path: Path) -> None:
+    load_result = validate_article_records(
+        [
+            {"DOI": "10.example/one", "Title": "One", "Year": 2025, "Abstract": ""},
+            {"DOI": "10.example/two", "Title": "Two", "Year": 2025, "Abstract": ""},
+        ]
+    )
+    client = FakeModelClient([_no_topic("No Topic one."), _no_topic("No Topic two.")])
+
+    batch = run_batch(
+        article_load_result=load_result,
+        vocabulary=_index(),
+        model_client=client,
+        settings=ModelSettings(),
+        store=JsonResultStore(tmp_path / "results"),
+    )
+
+    assert [result.DOI for result in batch.results] == ["10.example/one", "10.example/two"]
+    assert batch.summary.processed_articles == 2
+    assert batch.summary.articles_completed == 2
+    assert batch.summary.articles_not_classified == 2
+
+
+def test_batch_continues_after_one_article_failure(tmp_path: Path) -> None:
+    load_result = validate_article_records(
+        [
+            {"DOI": "10.example/one", "Title": "One", "Year": 2025, "Abstract": ""},
+            {"DOI": "10.example/two", "Title": "Two", "Year": 2025, "Abstract": ""},
+            {"DOI": "10.example/three", "Title": "Three", "Year": 2025, "Abstract": ""},
+        ]
+    )
+    client = FakeModelClient([_no_topic(), _select_topic("missing_topic"), _no_topic()])
+
+    batch = run_batch(
+        article_load_result=load_result,
+        vocabulary=_index(),
+        model_client=client,
+        settings=ModelSettings(),
+        store=JsonResultStore(tmp_path / "results"),
+    )
+
+    assert [result.DOI for result in batch.results] == [
+        "10.example/one",
+        "10.example/two",
+        "10.example/three",
+    ]
+    assert batch.summary.articles_failed == 1
+    assert batch.summary.articles_completed == 2
+    assert batch.results[1].errors[0].code == "UnknownCandidateIDError"
+
+
+def test_batch_reports_invalid_source_records_without_inventing_doi(tmp_path: Path) -> None:
+    load_result = validate_article_records(
+        [
+            {"DOI": "10.example/valid", "Title": "Valid", "Year": 2025, "Abstract": ""},
+            {"DOI": "", "Title": "Invalid", "Year": 2025, "Abstract": ""},
+        ]
+    )
+
+    batch = run_batch(
+        article_load_result=load_result,
+        vocabulary=_index(),
+        model_client=FakeModelClient([_no_topic()]),
+        settings=ModelSettings(),
+        store=JsonResultStore(tmp_path / "results"),
+    )
+
+    assert batch.summary.articles_received == 2
+    assert batch.summary.valid_article_records == 1
+    assert batch.summary.invalid_source_records == 1
+    assert batch.summary.errors[0].DOI == ""
+    assert [result.DOI for result in batch.results] == ["10.example/valid"]
+
+
+def test_batch_cache_hit_is_completed_not_skipped_and_force_reprocess_bypasses_cache(
+    tmp_path: Path,
+) -> None:
+    load_result = validate_article_records(
+        [{"DOI": "10.example/cache", "Title": "Cache", "Year": 2025, "Abstract": ""}]
+    )
+    cache = ArticleResultCache(tmp_path / "cache")
+    store = JsonResultStore(tmp_path / "results")
+
+    first = run_batch(
+        article_load_result=load_result,
+        vocabulary=_index(),
+        model_client=FakeModelClient([_no_topic("First run.")]),
+        settings=ModelSettings(),
+        store=store,
+        cache=cache,
+    )
+    second = run_batch(
+        article_load_result=load_result,
+        vocabulary=_index(),
+        model_client=FakeModelClient([]),
+        settings=ModelSettings(),
+        store=store,
+        cache=cache,
+    )
+    forced = run_batch(
+        article_load_result=load_result,
+        vocabulary=_index(),
+        model_client=FakeModelClient([_no_topic("Forced run.")]),
+        settings=ModelSettings(),
+        store=store,
+        cache=cache,
+        force_reprocess=True,
+    )
+
+    assert first.summary.cache_misses == 1
+    assert second.summary.cache_hits == 1
+    assert second.results[0].processing_metadata.cache_used is True
+    assert second.results[0].processing_status is ArticleProcessingStatus.COMPLETED
+    assert forced.summary.cache_hits == 0
+    assert forced.summary.cache_misses == 1
+    assert forced.results[0].no_classification_reason == "Forced run."
+
+
+def test_batch_output_summary_validates_against_schema(tmp_path: Path) -> None:
+    load_result = validate_article_records(
+        [{"DOI": "10.example/schema", "Title": "Schema", "Year": 2025, "Abstract": ""}]
+    )
+    batch = run_batch(
+        article_load_result=load_result,
+        vocabulary=_index(),
+        model_client=FakeModelClient([_no_topic()]),
+        settings=ModelSettings(),
+        store=JsonResultStore(tmp_path / "results"),
+    )
+
+    schema = json.loads(RUN_SUMMARY_SCHEMA_PATH.read_text())
+    Draft202012Validator(schema).validate(batch.summary.model_dump(mode="json"))
+
+
+def test_structured_diagnostics_do_not_include_secrets() -> None:
+    details = sanitize_log_details(
+        {
+            "DOI": "10.example/redacted",
+            "api_key": "raw-secret-value",
+            "nested": {"token": "raw-secret-value"},
+        }
+    )
+
+    assert details["api_key"] == "[REDACTED]"
+    assert details["nested"]["token"] == "[REDACTED]"
+    assert "raw-secret-value" not in json.dumps(details)
+
+
+def test_model_retry_counts_and_cache_flags_appear_in_metadata(tmp_path: Path) -> None:
+    load_result = validate_article_records(
+        [{"DOI": "10.example/meta", "Title": "Meta", "Year": 2025, "Abstract": ""}]
+    )
+    batch = run_batch(
+        article_load_result=load_result,
+        vocabulary=_index(),
+        model_client=FakeModelClient([_no_topic()]),
+        settings=ModelSettings(max_retries=3),
+        store=JsonResultStore(tmp_path / "results"),
+    )
+
+    metadata = batch.results[0].processing_metadata
+    assert metadata.model_parameters["max_retries"] == 3
+    assert metadata.cache_used is False
+    assert metadata.model_calls == 1
+
+
+def test_full_data_smoke_reports_current_invalid_article_without_modifying_sources(
+    tmp_path: Path,
+) -> None:
+    hierarchy_before = FULL_HIERARCHY_PATH.read_bytes()
+    articles_before = FULL_ARTICLES_PATH.read_bytes()
+    raw_articles = json.loads(articles_before)
+    load_result = validate_article_records(raw_articles)
+    subset_load_result = load_result.model_copy(update={"articles": load_result.articles[:1]})
+
+    batch = run_batch(
+        article_load_result=subset_load_result,
+        vocabulary=load_vocabulary(FULL_HIERARCHY_PATH),
+        model_client=FakeModelClient([_no_topic()]),
+        settings=ModelSettings(),
+        store=JsonResultStore(tmp_path / "results"),
+    )
+
+    assert load_result.source_count == 468
+    assert len(load_result.articles) == 467
+    assert len({article.DOI for article in load_result.articles}) == 467
+    assert [error.index for error in load_result.errors] == [467]
+    assert load_result.errors[0].code == "EMPTY_REQUIRED_STRING"
+    assert batch.summary.processed_articles == 1
+    assert FULL_HIERARCHY_PATH.read_bytes() == hierarchy_before
+    assert FULL_ARTICLES_PATH.read_bytes() == articles_before
+
+
+def test_logging_records_processing_events_without_secrets(caplog) -> None:
+    logger = logging.getLogger("gcmd_classifier.tests.pipeline")
+    caplog.set_level(logging.INFO, logger=logger.name)
+
+    classify_article(
+        article=_article(),
+        vocabulary=_index(),
+        model_client=FakeModelClient([_no_topic()]),
+        settings=ModelSettings(),
+        logger=logger,
+        relevant_config={"api_key": "should-not-log"},
+    )
+
+    assert any(record.__dict__.get("event") == "article_started" for record in caplog.records)
+    serialized = "\n".join(str(record.__dict__) for record in caplog.records)
+    assert "should-not-log" not in serialized