Plugin refactor + classifier support + idempotency

README.md

@@ -12,7 +12,7 @@ improv has three storage layers:
 - **Columnar store** — queryable image metadata, provenance records, and plugin index tables (DuckDB+Parquet or VAST DB)
 - **OLTP database** — mutable organizing metadata: instruments, samples, datasets, ingest tasks (PostgreSQL or SQLite)
 
-A **plugin system** extends provenance handling. Each plugin handles a specific provenance `kind` and optionally maintains an index table for fast querying. Plugins can be generic (geolocation, sample context) or instrument-specific (IFCB morphometric features, IFCB CNN classification scores).
+A **plugin system** extends provenance handling. Plugins register via dependency injection — the core never imports them at load time. Each plugin handles a specific provenance `kind` and optionally maintains an index table for fast querying. Plugins are generic and parameterized (geolocation, sample context, machine classification); instrument-specific presets live under `improv.plugins.ifcb` (IFCB morphometric features, IFCB CNN classification), pinning a `kind`/`index_table` onto a generic plugin.
 
 ## Access patterns
 
@@ -24,16 +24,25 @@ A **plugin system** extends provenance handling. Each plugin handles a specific
 
 ## REST API
 
-improv exposes a FastAPI service with endpoints for image data and metadata, provenance, instruments, samples, datasets, and ingest task tracking. A thin HTTP client (`improv.client.ImprovClient`) is provided for ingest scripts that need OLTP access without direct database credentials.
+improv exposes a FastAPI service with endpoints for image data and metadata, provenance, instruments, samples, datasets, and ingest task tracking. Classifier support adds taxonomy registration/lookup and two decode paths: a stateless decode (caller supplies a vector) and a decoded read that fetches an image's classification provenance and resolves each record against its own `model_version`. A thin HTTP client (`improv.client.ImprovClient`) is provided for ingest scripts that need OLTP access without direct database credentials, including taxonomy registration.
 
 ## Ingest architecture
 
 Batch producers (ingest pipelines, classifiers) use a hybrid approach:
 
-- **OLTP operations** (register instruments, samples, ingest tasks) go through the REST API
+- **OLTP operations** (register instruments, samples, ingest tasks, classifier taxonomies) go through the REST API
 - **High-volume writes** (image metadata, provenance, index tables, image bytes) go directly to the columnar store and object store
 
-This avoids coupling ingest scripts to the database while keeping high-throughput writes off the HTTP path.
+This avoids coupling ingest scripts to the database while keeping high-throughput writes off the HTTP path. Low-volume producers that prefer not to depend on the columnar store directly can post small provenance batches over REST (`POST /images/provenance/batch`, one instrument per batch).
+
+## Idempotency
+
+The provenance log is append-only, so writes are never overwritten — idempotency is achieved by **appending, then deduplicating at read time**. This works identically on every backend (VAST DB, DuckDB+Parquet, and any future one), because append is the only operation they all share.
+
+- A **provenance record** is identified by `(image_id, kind, source, data_hash)`, where `data_hash` is a canonical [RFC 8785 (JCS)](https://www.rfc-editor.org/rfc/rfc8785) hash of the `data` payload, stamped server-side. Re-posting a byte-identical record re-appends a row that collapses to one on read; a genuinely different payload (e.g. a new `model_version`) hashes differently and is retained. Canonicalization normalizes key order and number formatting (`1.0` == `1`) across producers in different languages, and rejects NaN/Infinity.
+- **Index records** are deterministic projections deduplicated on their full column tuple.
+
+**Client contract:** a record's `timestamp` is the **event time** (when the image was collected, or the classifier result produced) — a property of the observed fact, captured once. It is *not* the time of the HTTP request. Retries must resend the identical record, so put real event time in `timestamp`, never wall-clock-at-send; otherwise each attempt looks distinct and will not deduplicate. (The server separately records its own write time.)
 
 ## Install
 
@@ -51,5 +60,6 @@ pip install '.[service]' # adds FastAPI, CLI, migrations
 | `amplify-storage-utils` | Object storage (HashdirStore / S3) |
 | `pydantic` | Models and validation |
 | `pyarrow` | Columnar data exchange |
+| `rfc8785` | Canonical (JCS) hashing of provenance payloads for idempotency |
 | `httpx` | Thin ingest client |
 | `fastapi`, `sqlalchemy`, `alembic` | Service extras |

pyproject.toml

@@ -10,6 +10,7 @@ requires-python = ">=3.10"
 dependencies = [
     "pydantic>=2.0",
     "pyarrow>=14.0",
+    "rfc8785>=0.1",
     "httpx",
     "amplify-db-utils @ git+https://github.com/WHOIGit/amplify-db-utils.git",
     "amplify-storage-utils @ git+https://github.com/WHOIGit/amplify-storage-utils.git@v1.7.0",
@@ -20,6 +21,9 @@ db = [
     "sqlalchemy>=2.0",
     "psycopg2-binary",
 ]
+vastdb = [
+    "amplify-db-utils[vastdb] @ git+https://github.com/WHOIGit/amplify-db-utils.git",
+]
 service = [
     "fastapi",
     "uvicorn[standard]",
@@ -29,7 +33,7 @@ service = [
 ]
 dev = [
     "pytest",
-    "httpx",
+    "improv[service]",
 ]
 
 [project.scripts]

src/improv/api/app.py

@@ -21,7 +21,7 @@ def create_app(config: "ImprovConfig") -> FastAPI:
     from sqlalchemy import create_engine
     from sqlalchemy.orm import sessionmaker
 
-    from improv.api.routers import blobs, datasets, images, ingest_tasks, instruments, provenance, samples
+    from improv.api.routers import blobs, classification, datasets, images, ingest_tasks, instruments, provenance, samples
     from improv.oltp.models import Base
     from improv.service import ImageService
     from improv.store.tables import register_service_tables
@@ -59,5 +59,6 @@ async def lifespan(app: FastAPI):
     app.include_router(blobs.router)
     app.include_router(datasets.router)
     app.include_router(ingest_tasks.router)
+    app.include_router(classification.router)
 
     return app

src/improv/api/routers/classification.py

@@ -0,0 +1,165 @@
+"""Classifier taxonomy + classification-decode endpoints.
+
+Thin routing over existing ImageService methods. No decode/taxonomy business
+logic lives here — the service owns it (see ImageService.decode_classification
+and register/get_classifier_taxonomy). Decoding always uses a record's own
+model_version, never "latest" (which is display / new-work only).
+"""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Depends, HTTPException, status
+
+from improv.api.deps import get_service
+from improv.api.schemas import (
+    DecodedClassificationResponse,
+    DecodeRequest,
+    TaxonomyCreate,
+    TaxonomyResponse,
+)
+from improv.service import ImageService
+
+router = APIRouter(tags=["classification"])
+
+
+def _taxonomy_response(tax) -> TaxonomyResponse:
+    return TaxonomyResponse(
+        classifier=tax.classifier,
+        model_version=tax.model_version,
+        class_names=tax.class_names,
+        created_at=tax.created_at,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Taxonomy registration + lookup
+# ---------------------------------------------------------------------------
+
+@router.post(
+    "/classifiers/{classifier}/taxonomies",
+    response_model=TaxonomyResponse,
+    status_code=status.HTTP_201_CREATED,
+)
+def register_taxonomy(
+    classifier: str,
+    body: TaxonomyCreate,
+    service: ImageService = Depends(get_service),
+) -> TaxonomyResponse:
+    taxonomy, created = service.register_classifier_taxonomy(
+        classifier, body.model_version, body.class_names
+    )
+    if not created:
+        raise HTTPException(
+            status_code=409,
+            detail=f"Taxonomy for {classifier!r}/{body.model_version!r} already exists.",
+        )
+    return _taxonomy_response(taxonomy)
+
+
+# Literal /latest registered before /{model_version} so it isn't captured as a
+# version (same pattern the provenance router documents).
+@router.get(
+    "/classifiers/{classifier}/taxonomies/latest",
+    response_model=TaxonomyResponse,
+)
+def get_latest_taxonomy(
+    classifier: str,
+    service: ImageService = Depends(get_service),
+) -> TaxonomyResponse:
+    taxonomy = service.get_latest_classifier_taxonomy(classifier)
+    if taxonomy is None:
+        raise HTTPException(
+            status_code=404, detail=f"No taxonomy registered for classifier {classifier!r}."
+        )
+    return _taxonomy_response(taxonomy)
+
+
+@router.get(
+    "/classifiers/{classifier}/taxonomies/{model_version}",
+    response_model=TaxonomyResponse,
+)
+def get_taxonomy(
+    classifier: str,
+    model_version: str,
+    service: ImageService = Depends(get_service),
+) -> TaxonomyResponse:
+    taxonomy = service.get_classifier_taxonomy(classifier, model_version)
+    if taxonomy is None:
+        raise HTTPException(
+            status_code=404,
+            detail=f"No taxonomy for {classifier!r}/{model_version!r}.",
+        )
+    return _taxonomy_response(taxonomy)
+
+
+# ---------------------------------------------------------------------------
+# Decoding — (A) stateless, (B) decoded classification read
+# ---------------------------------------------------------------------------
+
+def _decode(
+    service: ImageService,
+    classifier: str,
+    model_version: str,
+    scores: list[float],
+    winner_index: int,
+) -> DecodedClassificationResponse:
+    """Decode one vector, mapping the service's ValueErrors to HTTP status.
+
+    Pre-checks taxonomy existence (→404) so a missing taxonomy is distinguished
+    from a length/range mismatch (→422); decode_classification raises an
+    undifferentiated ValueError for all three.
+    """
+    if service.get_classifier_taxonomy(classifier, model_version) is None:
+        raise HTTPException(
+            status_code=404,
+            detail=f"No taxonomy for {classifier!r}/{model_version!r}.",
+        )
+    try:
+        decoded = service.decode_classification(
+            classifier, model_version, scores, winner_index
+        )
+    except ValueError as exc:
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
+    return DecodedClassificationResponse(**decoded)
+
+
+@router.post(
+    "/classifiers/{classifier}/decode",
+    response_model=DecodedClassificationResponse,
+)
+def decode(
+    classifier: str,
+    body: DecodeRequest,
+    service: ImageService = Depends(get_service),
+) -> DecodedClassificationResponse:
+    return _decode(
+        service, classifier, body.model_version, body.scores, body.winner_index
+    )
+
+
+@router.get(
+    "/images/{image_id}/classification",
+    response_model=list[DecodedClassificationResponse],
+)
+def get_decoded_classification(
+    image_id: str,
+    kind: str,
+    instrument: str | None = None,
+    service: ImageService = Depends(get_service),
+) -> list[DecodedClassificationResponse]:
+    """Read an image's classification-kind provenance, decoded.
+
+    `classifier` is the plugin `kind`. Each record decodes against its OWN
+    model_version (from the stored payload), never "latest".
+    """
+    records = service.get_provenance(image_id, kind=kind, instrument=instrument)
+    return [
+        _decode(
+            service,
+            kind,
+            r.data["model_version"],
+            r.data["scores"],
+            r.data["winner_index"],
+        )
+        for r in records
+    ]

src/improv/api/routers/provenance.py

@@ -24,6 +24,8 @@ def _to_response(env: ProvenanceEnvelope) -> ProvenanceResponse:
         timestamp=env.timestamp,
         data=env.data,
         instrument=env.instrument,
+        year=env.year,
+        month=env.month,
     )
 
 
@@ -39,17 +41,37 @@ def ingest_provenance_batch(
 ) -> dict:
     envelopes = [
         ProvenanceEnvelope(
-            image_id="",  # overridden per record below — batch has no shared image_id
+            image_id=r.image_id,
             kind=r.kind,
             source=r.source,
             timestamp=r.timestamp,
             data=r.data,
-            instrument=instrument,
+            instrument=instrument,  # uniform fallback hint; parser wins per record
         )
         for r in body.records
     ]
-    service.ingest_provenance(envelopes)
-    return {"ingested": len(envelopes)}
+
+    # A batch is assumed single-instrument. Resolve each record's instrument up
+    # front and reject (before any write) if the batch spans more than one, or if
+    # a record's image_id neither parses nor has the fallback hint.
+    try:
+        enriched = service.enrich_envelopes(envelopes)
+    except ValueError as exc:
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
+
+    instruments = {e.instrument for e in enriched}
+    if len(instruments) > 1:
+        raise HTTPException(
+            status_code=422,
+            detail=f"Batch spans multiple instruments {sorted(instruments)}; "
+            "post one instrument's records at a time.",
+        )
+
+    try:
+        service.ingest_provenance(enriched)
+    except ValueError as exc:
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
+    return {"ingested": len(enriched)}
 
 
 @router.get(
@@ -97,5 +119,8 @@ def ingest_provenance(
         data=body.data,
         instrument=instrument,
     )
-    service.ingest_provenance([envelope])
+    try:
+        service.ingest_provenance([envelope])
+    except ValueError as exc:
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
     return {"ingested": 1}

src/improv/api/schemas.py

@@ -54,6 +54,8 @@ class ProvenanceResponse(BaseModel):
     timestamp: datetime
     data: dict
     instrument: str | None = None
+    year: int | None = None
+    month: int | None = None
 
 
 class ProvenanceIngest(BaseModel):
@@ -63,8 +65,45 @@ class ProvenanceIngest(BaseModel):
     data: dict
 
 
+class ProvenanceBatchRecord(BaseModel):
+    # Batch records carry their own image_id (unlike single-record ingest, where
+    # image_id comes from the path).
+    image_id: str
+    kind: str
+    source: str
+    timestamp: datetime
+    data: dict
+
+
 class ProvenanceBatchIngest(BaseModel):
-    records: list[ProvenanceIngest]
+    records: list[ProvenanceBatchRecord]
+
+
+# ---------------------------------------------------------------------------
+# Classification / taxonomy
+# ---------------------------------------------------------------------------
+
+class TaxonomyCreate(BaseModel):
+    model_version: str
+    class_names: list[str]
+
+
+class TaxonomyResponse(BaseModel):
+    classifier: str
+    model_version: str
+    class_names: list[str]
+    created_at: datetime
+
+
+class DecodeRequest(BaseModel):
+    model_version: str
+    scores: list[float]
+    winner_index: int
+
+
+class DecodedClassificationResponse(BaseModel):
+    winner: str
+    scores: dict[str, float]
 
 
 # ---------------------------------------------------------------------------