Merge pull request #237 from fairagro/feature/review_harvesting_api

Feature/review harvesting api

Author: Zalfsten

.agents/skills/create-specifica-feature/SKILL.md

@@ -76,14 +76,19 @@ condition and the expected output or side-effect.
 - One behaviour per checkbox — if you need "and" it is two requirements.
 - State the outcome, not the implementation (`→ return 404` not `→ use Flask abort()`).
 - Every edge case ends with a concrete outcome — no open-ended statements.
-- **Avoid naming concrete classes, methods, or modules in `spec.md`** unless it
-  is a deliberate act. Naming a class in a requirement couples that requirement to
-  a design decision — which is fine when the decision is intentional and should be
-  locked in by the spec, but must be recognised as such. Class names, method names,
-  and module paths that are not yet decided belong in `design.md`, not `spec.md`.
-  *Exception:* simple classes whose name fully describes their behaviour — in
-  particular exception classes (e.g. `InvalidJsonSemanticError`) — may be
-  referenced freely in `spec.md` without implying a design constraint.
+- **`spec.md` must not reference function names, method names, or variable names
+  unless that element was previously explicitly specified earlier in the same spec.**
+  Naming an implementation detail in a requirement couples it to a design decision —
+  such decisions belong in `design.md`. Class names and module paths that are not yet
+  decided also belong in `design.md`, not `spec.md`.
+  *Exemptions:*
+  - Exception classes whose name fully describes their behaviour (e.g.
+    `InvalidJsonSemanticError`, `DocumentConflictError`) may be referenced freely.
+  - String or numeric literals that belong to an external protocol (e.g. CouchDB's
+    `_rev`, `_id`) are not implementation names and may be used freely.
+  - Domain-level identifiers that also appear as code names (e.g. `arc_id`,
+    `harvest_id`) may be used when they function as domain terms rather than as
+    references to a specific code symbol.
 
 ---
 

middleware/api/spec/document-store/spec.md

@@ -22,8 +22,9 @@ prefix, not by separate databases.
       exist if they do not yet exist.
 - [ ] Handle `412 Precondition Failed` (database already exists) as a success
       during initialization — parallel service startups must not cause crashes.
-- [ ] Store ARC documents keyed by `arc_id`; return `is_new` and `has_changes`
-      flags based on content hash comparison.
+- [ ] Store ARC documents keyed by `arc_id`; return flags indicating whether the
+      document was newly created and whether its content changed, based on content
+      hash comparison.
 - [ ] Support harvest run lifecycle operations: create, retrieve, increment
       statistics counters, and finalize a harvest run.
 - [ ] Append event records to an ARC document's event log.
@@ -34,7 +35,13 @@ prefix, not by separate databases.
 Parallel service startup (two containers connecting simultaneously) → both attempt
 database creation; `412 Precondition Failed` is treated as success, not an error.
 
-ARC document already exists with identical content → `has_changes = False`; no
-CouchDB write performed for the body; only timestamp fields may be updated.
+ARC document already exists with identical content → the content-changed flag is
+`false`; no CouchDB write performed for the body; only timestamp fields may be updated.
 
-`get_harvest` for unknown ID → returns `None`; callers raise `ResourceNotFoundError`.
+Concurrent writes to the same ARC document (e.g. two harvest workers submitting the
+same ARC simultaneously) → the store strips the stale `_rev` from the payload,
+re-fetches the current revision on each attempt, and retries up to a configurable
+maximum (default 3) on `ConflictError` before raising `DocumentConflictError`.
+
+Fetching a harvest by an unknown ID → the store returns nothing; callers raise
+`ResourceNotFoundError`.

middleware/api/spec/harvest-manager/design.md

@@ -11,6 +11,7 @@ API endpoint
     └─→ HarvestManager
             ├─→ DocumentStore.create_harvest
             ├─→ DocumentStore.get_harvest
+            ├─→ DocumentStore.update_harvest          (transition_harvest)
             ├─→ DocumentStore.increment_harvest_statistics
             └─→ DocumentStore.finalize_harvest
 ```
@@ -32,3 +33,26 @@ API endpoint
    `HarvestConfig` (a Pydantic model). Application code reads them from the
    config object rather than hardcoding values, making them overridable via
    environment variables or YAML without a code change.
+
+4. **`transition_harvest` accepts a pre-fetched `HarvestDocument`, not a `harvest_id`**
+   — The router always fetches the harvest before calling into the service layer
+   (for RDI auth and existence checks). Passing the already-fetched document
+   avoids a redundant round-trip to CouchDB and removes the dead
+   `pre_fetched or await get_harvest()` fallback path. Responsibility for
+   handling a missing harvest therefore stays in the router (HTTP 404), while
+   `transition_harvest` focuses solely on ownership validation, the RUNNING
+   guard, and the DB write.
+
+5. **Single `transition_harvest` replaces separate `complete_harvest` / `cancel_harvest` / `fail_harvest` methods**
+   — All three terminal transitions share the same shape: ownership check →
+   RUNNING guard → `DocumentStore.update_harvest`. A single generic method
+   parameterised over `target_status` avoids duplicating that logic three
+   times. The only special case is `COMPLETED`: it also persists the current
+   statistics snapshot (same behaviour as the old `complete_harvest`).
+
+6. **`PATCH /v3/harvests/{harvest_id}` as the canonical state-transition endpoint**
+   — A single PATCH endpoint with `{"status": "..."}` in the body covers all
+   terminal transitions. The legacy `DELETE /{harvest_id}` (cancel) and
+   `POST /{harvest_id}/complete` endpoints are kept for backward compatibility
+   but are not the preferred path for new clients. The API client's
+   `cancel_harvest()` and `fail_harvest()` methods both call PATCH internally.

middleware/api/spec/harvest-manager/spec.md

@@ -15,6 +15,10 @@ finalization — and enforce client ownership of harvest resources.
 - [ ] Increment per-harvest statistics (new-ARC counter and changed-ARC counter)
       via `increment_harvest_statistics` for each processed ARC.
 - [ ] Finalize a harvest run, marking it complete and recording the final statistics.
+- [ ] Transition a harvest run to a target terminal status (`COMPLETED`, `CANCELLED`,
+      or `FAILED`) via an explicit state-transition operation.
+- [ ] Enforce the transition guard: only a `RUNNING` harvest may transition to a
+      terminal status; raise `ConflictError` otherwise.
 
 ## Edge Cases
 
@@ -25,3 +29,5 @@ finalization — and enforce client ownership of harvest resources.
 `expected_datasets` not provided → harvest document is created without a progress denominator; progress reporting shows raw counts only.
 
 Finalize called before all ARCs arrive → harvest is marked complete with whatever statistics are current; no enforcement of `expected_datasets`.
+
+Transition called on a non-`RUNNING` harvest → raise `ConflictError` with the current status in the message.

middleware/api/src/middleware/api/api/fastapi_app.py

@@ -58,7 +58,6 @@
         __version__ = "0.0.0"
 
 
-loaded_config = None
 if "pytest" in sys.modules:
     # pytest is executing this file during a test discovery run.
     # No config file is available, so we create a dummy config so that pytest does not fail.
@@ -88,9 +87,27 @@
         sys.exit(1)
 
 logging.basicConfig(
-    level=getattr(logging, loaded_config.log_level), format="%(asctime)s %(levelname)s %(name)s: %(message)s"
+    level=getattr(logging, loaded_config.log_level),
+    format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+    force=True,
 )
 
+
+def _configure_uvicorn_loggers(log_level: str) -> None:
+    """Ensure uvicorn loggers use the same format as the middleware logs."""
+    root_handlers = logging.root.handlers[:]
+    if not root_handlers:
+        return
+
+    for logger_name in ("uvicorn.access", "uvicorn.error"):
+        uvicorn_logger = logging.getLogger(logger_name)
+        uvicorn_logger.handlers = root_handlers[:]
+        uvicorn_logger.setLevel(getattr(logging, log_level))
+        uvicorn_logger.propagate = False
+
+
+_configure_uvicorn_loggers(loaded_config.log_level)
+
 logger = logging.getLogger("middleware_api")
 
 
@@ -151,11 +168,14 @@ def __init__(self, app_config: Config) -> None:
 
         logger.debug("API configuration: %s", self._config.model_dump())
 
-        # Apply polling log filter to uvicorn access logger
-        logging.getLogger("uvicorn.access").addFilter(PollingLogFilter())
-
         @asynccontextmanager
         async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
+            # Re-apply our log format after uvicorn has finished its own log setup.
+            # uvicorn.main() installs its own handlers on uvicorn.access/uvicorn.error
+            # *after* this module is imported, so we must reconfigure here.
+            _configure_uvicorn_loggers(self._config.log_level)
+            logging.getLogger("uvicorn.access").addFilter(PollingLogFilter())
+
             # Initialize business logic and its stores
             try:
                 try:

middleware/api/src/middleware/api/api/v3/harvests.py

@@ -13,7 +13,8 @@
     get_common_deps,
     get_content_type,
 )
-from middleware.api.business_logic import BusinessLogic
+from middleware.api.business_logic import BusinessLogic, ConflictError
+from middleware.api.business_logic.exceptions import DuplicateArcInHarvestError
 from middleware.api.document_store.harvest_document import HarvestDocument
 from middleware.shared.api_models.v3 import models as v3_models
 
@@ -88,42 +89,77 @@ async def get_harvest(
     return _map_harvest(harvest)
 
 
-@router.post("/{harvest_id}/complete", response_model=v3_models.HarvestResponse)
+@router.post("/{harvest_id}/complete", response_model=v3_models.HarvestResponse, deprecated=True)
 async def complete_harvest(  # noqa: PLR0913, PLR0917
     request: Request,
     harvest_id: str,
     bl: Annotated[BusinessLogic, Depends(get_business_logic)],
     deps: Annotated[CommonApiDependencies, Depends(get_common_deps)],
     client_id: Annotated[str | None, Depends(get_client_id)],
 ) -> v3_models.HarvestResponse:
-    """Mark a harvest as completed."""
+    """Mark a harvest as completed. [DEPRECATED].
+
+    Use ``PATCH /v3/harvests/{harvest_id}`` with ``status=COMPLETED`` instead.
+    """
     # Fetch once — used for both RDI auth and passed to complete_harvest (C2).
     harvest = await bl.harvest_manager.get_harvest(harvest_id)
     if not harvest:
         raise HTTPException(status_code=HTTPStatus.NOT_FOUND, detail="Harvest not found")
 
     await deps.validate_rdi_authorized(harvest.rdi, request)
 
-    harvest = await bl.harvest_manager.complete_harvest(harvest_id, client_id=client_id, pre_fetched=harvest)
+    harvest = await bl.harvest_manager.complete_harvest(harvest, client_id=client_id)
     return _map_harvest(harvest)
 
 
-@router.delete("/{harvest_id}", status_code=HTTPStatus.NO_CONTENT)
+@router.patch("/{harvest_id}", response_model=v3_models.HarvestResponse)
+async def patch_harvest_status(  # noqa: PLR0913, PLR0917
+    request: Request,
+    harvest_id: str,
+    request_body: v3_models.PatchHarvestRequest,
+    bl: Annotated[BusinessLogic, Depends(get_business_logic)],
+    deps: Annotated[CommonApiDependencies, Depends(get_common_deps)],
+    client_id: Annotated[str | None, Depends(get_client_id)],
+    _content_type: Annotated[None, Depends(get_content_type)],
+) -> v3_models.HarvestResponse:
+    """Transition a harvest run to a terminal status (COMPLETED, CANCELLED, or FAILED).
+
+    Only a ``RUNNING`` harvest may be transitioned; any other current status returns
+    409 Conflict.
+    """
+    harvest = await bl.harvest_manager.get_harvest(harvest_id)
+    if not harvest:
+        raise HTTPException(status_code=HTTPStatus.NOT_FOUND, detail="Harvest not found")
+
+    await deps.validate_rdi_authorized(harvest.rdi, request)
+
+    try:
+        harvest = await bl.harvest_manager.transition_harvest(harvest, request_body.status, client_id=client_id)
+    except ConflictError as exc:
+        raise HTTPException(status_code=HTTPStatus.CONFLICT, detail=str(exc)) from exc
+
+    return _map_harvest(harvest)
+
+
+@router.delete("/{harvest_id}", status_code=HTTPStatus.NO_CONTENT, deprecated=True)
 async def cancel_harvest(
     request: Request,
     harvest_id: str,
     bl: Annotated[BusinessLogic, Depends(get_business_logic)],
     deps: Annotated[CommonApiDependencies, Depends(get_common_deps)],
     client_id: Annotated[str | None, Depends(get_client_id)],
 ) -> None:
-    """Cancel a harvest run."""
+    """Cancel a harvest run. [DEPRECATED].
+
+    Use ``PATCH /v3/harvests/{harvest_id}`` with ``status=CANCELLED`` instead.
+    """
     # Fetch once — used for both RDI auth and passed to cancel_harvest (C2).
     harvest = await bl.harvest_manager.get_harvest(harvest_id)
     if not harvest:
         raise HTTPException(status_code=HTTPStatus.NOT_FOUND, detail="Harvest not found")
 
     await deps.validate_rdi_authorized(harvest.rdi, request)
-    await bl.harvest_manager.cancel_harvest(harvest_id, client_id=client_id, pre_fetched=harvest)
+    await bl.harvest_manager.cancel_harvest(harvest, client_id=client_id)
 
 
 @router.post("/{harvest_id}/arcs", response_model=v3_models.ArcResponse)
@@ -146,7 +182,10 @@ async def submit_arc_in_harvest(  # noqa: PLR0913, PLR0917
     rdi = harvest.rdi
     await deps.validate_rdi_authorized(rdi, request)
 
-    result = await bl.create_or_update_arc(rdi, request_body.arc, client_id, harvest_id=harvest_id)
+    try:
+        result = await bl.create_or_update_arc(rdi, request_body.arc, client_id, harvest_id=harvest_id)
+    except DuplicateArcInHarvestError as exc:
+        raise HTTPException(status_code=HTTPStatus.CONFLICT, detail=str(exc)) from exc
 
     arc_id = result.arc.id
     metadata = await bl.get_metadata(arc_id)

middleware/api/src/middleware/api/business_logic/__init__.py

@@ -9,6 +9,7 @@
     AccessDeniedError,
     BusinessLogicError,
     ConflictError,
+    DuplicateArcInHarvestError,
     InvalidJsonSemanticError,
     ResourceNotFoundError,
     SetupError,
@@ -23,6 +24,7 @@
     "BusinessLogic",
     "BusinessLogicError",
     "ConflictError",
+    "DuplicateArcInHarvestError",
     "InvalidJsonSemanticError",
     "ResourceNotFoundError",
     "SetupError",

middleware/api/src/middleware/api/business_logic/arc_manager.py

@@ -9,12 +9,12 @@
 from opentelemetry import trace
 
 from middleware.api.arc_store import ArcStore, ArcStoreTransientError
-from middleware.api.document_store import DocumentStore
+from middleware.api.document_store import DocumentStore, DuplicateArcError
 from middleware.api.document_store.arc_document import ArcEvent, ArcEventType
 from middleware.api.utils import calculate_arc_id, extract_identifier
 from middleware.shared.api_models.common.models import ArcOperationResult, ArcResponse, ArcStatus
 
-from .exceptions import BusinessLogicError, InvalidJsonSemanticError, TransientError
+from .exceptions import BusinessLogicError, DuplicateArcInHarvestError, InvalidJsonSemanticError, TransientError
 from .ports import TaskDispatcher
 from .task_payloads import ArcSyncTask
 
@@ -101,13 +101,6 @@ async def create_or_update_arc(
                 has_changes = doc_result.has_changes
                 should_trigger_git = is_new or has_changes
 
-                if harvest_id:
-                    await self._doc_store.increment_harvest_statistics(
-                        harvest_id,
-                        is_new=is_new,
-                        has_changes=has_changes,
-                    )
-
                 logger.info(
                     "[%s] Stored ARC %s in CouchDB: is_new=%s, has_changes=%s, trigger_git=%s",
                     client_id,
@@ -147,6 +140,8 @@ async def create_or_update_arc(
                     raise
                 if isinstance(e, BusinessLogicError):
                     raise
+                if isinstance(e, DuplicateArcError):
+                    raise DuplicateArcInHarvestError(str(e)) from e
                 raise BusinessLogicError(f"unexpected error encountered: {str(e)}") from e
 
     async def sync_to_gitlab(self, rdi: str, arc: dict[str, Any]) -> None:

middleware/api/src/middleware/api/business_logic/exceptions.py

@@ -17,6 +17,10 @@ class ConflictError(BusinessLogicError):
     """Arises when the request conflicts with the current resource state."""
 
 
+class DuplicateArcInHarvestError(ConflictError):
+    """Arises when the same ARC is submitted more than once within a harvest run."""
+
+
 class InvalidJsonSemanticError(BusinessLogicError):
     """Arises when the ARC JSON syntax is valid but semantically incorrect.