feat: replace BackgroundTask with durable SQLite inspection queue (#132)

* feat: replace BackgroundTask with durable SQLite inspection queue

Introduces a SQLite-backed inspection queue (WAL mode) with crash
recovery, replacing FastAPI BackgroundTasks for the /trigger endpoint.

- Add InspectionEvent domain model and IInspectionQueue port
- Add SqliteInspectionQueue adapter (aiosqlite, WAL mode, recover_interrupted)
- Add InspectionWorker coroutine started in FastAPI lifespan
- POST /trigger now enqueues an event and returns item_id immediately
- Fix pre-existing bug: get_item_binary_names/get_item_binaries now
  filter by item_id instead of globbing all files in the storage directory
- Update functional test steps to poll for state=DONE (async behavior)
- 12 new unit tests covering queue operations and worker lifecycle

* feat: introduce ITriggerAdapter port and RestTriggerAdapter

Adds the trigger adapter abstraction as a hexagonal port.
RestTriggerAdapter is the reference passive implementation:
enqueueing is handled by the POST /trigger endpoint, so run()
simply stays alive until cancelled.

Active adapters (MQTT, OPC UA, gRPC) will implement real event
polling in run() and ship as optional Python extras.

Author: bojeanson

edge_orchestrator/pyproject.toml

@@ -22,6 +22,7 @@ classifiers = [
 ]
 dependencies = [
     "aiohttp==3.11.11",
+    "aiosqlite==0.21.0",
     "fastapi[all]==0.115.6",
     "numpy==1.24.2",
     "opencv_python==4.11.0.86",

edge_orchestrator/src/edge_orchestrator/application/workers/__init__.py

@@ -0,0 +1,42 @@
+import asyncio
+import logging
+
+from edge_orchestrator.application.config.config_manager import ConfigManager
+from edge_orchestrator.application.use_cases.supervisor import Supervisor
+from edge_orchestrator.domain.models.item import Item
+from edge_orchestrator.domain.ports.inspection_queue.i_inspection_queue import (
+    IInspectionQueue,
+)
+
+logger = logging.getLogger(__name__)
+
+_POLL_INTERVAL = 0.05  # seconds
+
+
+async def run_inspection_worker(
+    queue: IInspectionQueue,
+    supervisor: Supervisor,
+    config_manager: ConfigManager,
+) -> None:
+    recovered = await queue.recover_interrupted()
+    if recovered:
+        logger.info("Recovered %d interrupted inspection(s)", recovered)
+
+    while True:
+        event = await queue.dequeue()
+        if event is None:
+            await asyncio.sleep(_POLL_INTERVAL)
+            continue
+
+        station_config = config_manager.get_config()
+        if station_config is None:
+            await queue.mark_failed(event.id, "no active station config")
+            continue
+
+        item = Item(id=event.item_id)
+        try:
+            await supervisor.inspect(item, station_config)
+            await queue.mark_done(event.id)
+        except Exception as exc:
+            logger.exception("Inspection failed for event %s", event.id)
+            await queue.mark_failed(event.id, str(exc))

edge_orchestrator/src/edge_orchestrator/application/workers/inspection_worker.py

@@ -0,0 +1,24 @@
+from datetime import datetime
+from enum import Enum
+from typing import Optional
+from uuid import UUID, uuid4
+
+from pydantic import BaseModel, Field
+
+
+class InspectionEventStatus(str, Enum):
+    PENDING = "pending"
+    RUNNING = "running"
+    DONE = "done"
+    FAILED = "failed"
+
+
+class InspectionEvent(BaseModel):
+    id: UUID = Field(default_factory=uuid4)
+    item_id: UUID = Field(default_factory=uuid4)
+    station_name: str
+    status: InspectionEventStatus = InspectionEventStatus.PENDING
+    created_at: datetime = Field(default_factory=datetime.now)
+    updated_at: datetime = Field(default_factory=datetime.now)
+    retry_count: int = 0
+    error: Optional[str] = None

edge_orchestrator/src/edge_orchestrator/domain/models/inspection_event.py

@@ -0,0 +1,23 @@
+from abc import ABC, abstractmethod
+from typing import Optional
+from uuid import UUID
+
+from edge_orchestrator.domain.models.inspection_event import InspectionEvent
+
+
+class IInspectionQueue(ABC):
+    @abstractmethod
+    async def enqueue(self, station_name: str) -> InspectionEvent: ...
+
+    @abstractmethod
+    async def dequeue(self) -> Optional[InspectionEvent]: ...
+
+    @abstractmethod
+    async def mark_done(self, event_id: UUID) -> None: ...
+
+    @abstractmethod
+    async def mark_failed(self, event_id: UUID, error: str) -> None: ...
+
+    @abstractmethod
+    async def recover_interrupted(self) -> int:
+        """Reset RUNNING → PENDING at startup. Returns count of recovered events."""

edge_orchestrator/src/edge_orchestrator/domain/ports/inspection_queue/__init__.py

@@ -0,0 +1,19 @@
+from abc import ABC, abstractmethod
+
+from edge_orchestrator.domain.ports.inspection_queue.i_inspection_queue import IInspectionQueue
+
+
+class ITriggerAdapter(ABC):
+    @abstractmethod
+    async def run(self, queue: IInspectionQueue) -> None:
+        """Main loop: listen for trigger events and enqueue them.
+
+        For active adapters (MQTT, OPC UA) this is a polling loop.
+        For passive adapters (REST) this is a sentinel that stays alive until cancelled.
+        Runs until cancelled via asyncio.CancelledError.
+        """
+
+    @classmethod
+    @abstractmethod
+    def config_schema(cls) -> dict:
+        """Return the JSON Schema for this adapter's configuration block."""

edge_orchestrator/src/edge_orchestrator/domain/ports/inspection_queue/i_inspection_queue.py

@@ -39,7 +39,7 @@ def get_item_binaries(self, item_id: UUID) -> Dict[str, bytes]:
         path = self._storing_path_manager.get_storing_path()
         item_binaries = {}
         for extension in ImageExtension:
-            for binary_path in path.glob(f"*.{extension.value}"):
+            for binary_path in path.glob(f"{item_id}_*.{extension.value}"):
                 with binary_path.open("rb") as f:
                     item_binaries[binary_path.stem] = f.read()
         return item_binaries
@@ -48,7 +48,7 @@ def get_item_binary_names(self, item_id: UUID) -> List[str]:
         path = self._storing_path_manager.get_storing_path()
         item_binaries = []
         for extension in ImageExtension:
-            for binary_path in path.glob(f"*.{extension.value}"):
+            for binary_path in path.glob(f"{item_id}_*.{extension.value}"):
                 item_binaries.append(binary_path.name)
         return item_binaries