from __future__ import annotations import threading from collections.abc import Callable, Iterable, Mapping from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta from typing import Any, Literal from uuid import UUID, uuid4 OperationStatus = Literal[ "accepted", "running", "operator_action_required", "succeeded", "failed", "cancelled", "timed_out", "interrupted", ] AcquisitionState = Literal[ "preparing", "prepared", "awaiting_external_start", "starting", "acquiring", "awaiting_external_stop", "stopping", "finalizing", "completed", "failed", "aborted", "interrupted", ] ControlMode = Literal["operator-manual", "plugin-commanded", "observe-only"] TERMINAL_OPERATION_STATUSES: frozenset[OperationStatus] = frozenset( {"succeeded", "failed", "cancelled", "timed_out", "interrupted"} ) TERMINAL_ACQUISITION_STATES: frozenset[AcquisitionState] = frozenset( {"completed", "failed", "aborted", "interrupted"} ) def utc_now() -> datetime: return datetime.now(UTC) def _iso(value: datetime | None) -> str | None: return value.isoformat().replace("+00:00", "Z") if value is not None else None def _identifier(value: str | None, *, prefix: str) -> str: if value is None: return f"{prefix}-{uuid4()}" candidate = value.strip() if not candidate or len(candidate) > 128: raise ValueError(f"{prefix} id must contain 1..128 characters") try: UUID(candidate.removeprefix(f"{prefix}-")) except ValueError as exc: raise ValueError(f"{prefix} id must be a generated UUID identifier") from exc return candidate @dataclass(slots=True) class OperationRecord: operation_id: str action: str status: OperationStatus accepted_at: datetime device_id: str | None = None device_session_id: str | None = None idempotency_key: str | None = None deadline_at: datetime | None = None stage_code: str = "accepted" message_code: str = "operation.accepted" sequence: int = 1 state_revision: int = 1 completed_at: datetime | None = None cancellable: bool = False cancel_requested: bool = False result: dict[str, Any] | None = None error: dict[str, Any] | None = None evidence_refs: tuple[str, ...] = () context: dict[str, Any] = field(default_factory=dict) events: list[dict[str, Any]] = field(default_factory=list) # A keyed, non-reversible digest supplied by the service. It is deliberately # excluded from API snapshots: callers only need mismatch detection, while # the journal must never retain action inputs or secret material. request_fingerprint: str | None = field(default=None, repr=False) def as_dict(self) -> dict[str, Any]: return { "schema_version": "missioncore.operation-snapshot/v1alpha2", "operation_id": self.operation_id, "action": self.action, "status": self.status, "accepted_at": _iso(self.accepted_at), "completed_at": _iso(self.completed_at), "deadline_at": _iso(self.deadline_at), "device_id": self.device_id, "device_session_id": self.device_session_id, "idempotency_key": self.idempotency_key, "stage_code": self.stage_code, "message_code": self.message_code, "sequence": self.sequence, "state_revision": self.state_revision, "cancellable": self.cancellable, "cancel_requested": self.cancel_requested, "result": dict(self.result) if self.result is not None else None, "error": dict(self.error) if self.error is not None else None, "evidence_refs": list(self.evidence_refs), "context": dict(self.context), "events": [dict(event) for event in self.events], } class OperationJournal: """Bounded, secret-free operation journal for one in-process plugin runtime. The journal deliberately stores lifecycle metadata only. Action inputs, BLE frames, MQTT payloads and credentials never enter operation events. """ def __init__( self, *, max_records: int = 128, clock: Callable[[], datetime] = utc_now, ) -> None: if max_records < 1: raise ValueError("max_records must be positive") self._max_records = max_records self._clock = clock self._lock = threading.Lock() self._records: dict[str, OperationRecord] = {} self._order: list[str] = [] self._idempotency: dict[str, str] = {} def begin( self, action: str, *, operation_id: str | None = None, idempotency_key: str | None = None, device_id: str | None = None, device_session_id: str | None = None, deadline_seconds: float | None = None, cancellable: bool = False, request_fingerprint: str | None = None, context: Mapping[str, Any] | None = None, ) -> tuple[OperationRecord, bool]: action = action.strip() if not action: raise ValueError("operation action cannot be blank") if idempotency_key is not None: idempotency_key = idempotency_key.strip() if not idempotency_key or len(idempotency_key) > 160: raise ValueError("idempotency key must contain 1..160 characters") if deadline_seconds is not None and not 0 < deadline_seconds <= 86_400: raise ValueError("operation deadline must be within 1..86400 seconds") with self._lock: if idempotency_key is not None and idempotency_key in self._idempotency: existing_idempotent = self._records[self._idempotency[idempotency_key]] if existing_idempotent.action != action: raise ValueError("idempotency key is already bound to another action") if existing_idempotent.request_fingerprint != request_fingerprint: raise ValueError("idempotency key is already bound to a different request") return existing_idempotent, False resolved_id = _identifier(operation_id, prefix="op") existing_by_id = self._records.get(resolved_id) if existing_by_id is not None: if existing_by_id.action != action: raise ValueError("operation id is already bound to another action") if existing_by_id.request_fingerprint != request_fingerprint: raise ValueError("operation id is already bound to a different request") return existing_by_id, False now = self._clock() record = OperationRecord( operation_id=resolved_id, action=action, status="accepted", accepted_at=now, device_id=device_id, device_session_id=device_session_id, idempotency_key=idempotency_key, deadline_at=( now + timedelta(seconds=deadline_seconds) if deadline_seconds is not None else None ), cancellable=cancellable, request_fingerprint=request_fingerprint, context=dict(context or {}), ) self._append_event_locked(record) self._records[resolved_id] = record self._order.append(resolved_id) if idempotency_key is not None: self._idempotency[idempotency_key] = resolved_id self._trim_locked() return record, True def transition( self, operation_id: str, status: OperationStatus, *, stage_code: str, message_code: str, result: Mapping[str, Any] | None = None, error: Mapping[str, Any] | None = None, evidence_refs: Iterable[str] = (), ) -> OperationRecord: with self._lock: record = self._require_locked(operation_id) if record.status in TERMINAL_OPERATION_STATUSES: if record.status == status: return record raise ValueError(f"operation {operation_id} is already terminal") record.status = status record.stage_code = stage_code record.message_code = message_code record.sequence += 1 record.state_revision += 1 record.result = dict(result) if result is not None else None record.error = dict(error) if error is not None else None record.evidence_refs = tuple(evidence_refs) if status in TERMINAL_OPERATION_STATUSES: record.completed_at = self._clock() self._trim_locked() self._append_event_locked(record) return record def request_cancel(self, operation_id: str) -> OperationRecord: with self._lock: record = self._require_locked(operation_id) if record.status in TERMINAL_OPERATION_STATUSES: return record if not record.cancellable: raise ValueError(f"operation {operation_id} is not cancellable") record.cancel_requested = True record.sequence += 1 record.state_revision += 1 record.stage_code = "cancellation-requested" record.message_code = "operation.cancellation_requested" self._append_event_locked(record) return record def transition_if_pending( self, operation_id: str | None, status: OperationStatus, *, stage_code: str, message_code: str, result: Mapping[str, Any] | None = None, error: Mapping[str, Any] | None = None, evidence_refs: Iterable[str] = (), ) -> OperationRecord | None: """Atomically transition an existing non-terminal operation. Lifecycle reconciliation and explicit stop/abort paths can race. This helper makes terminalization idempotent without exposing mutable journal records or turning an already-completed operation into an error. """ with self._lock: if operation_id is None: return None record = self._records.get(operation_id) if record is None or record.status in TERMINAL_OPERATION_STATUSES: return record record.status = status record.stage_code = stage_code record.message_code = message_code record.sequence += 1 record.state_revision += 1 record.result = dict(result) if result is not None else None record.error = dict(error) if error is not None else None record.evidence_refs = tuple(evidence_refs) if status in TERMINAL_OPERATION_STATUSES: record.completed_at = self._clock() self._trim_locked() self._append_event_locked(record) return record def get(self, operation_id: str) -> OperationRecord: with self._lock: return self._require_locked(operation_id) def deadline_reached(self, operation_id: str | None) -> bool: """Evaluate one operation deadline on the journal-owned server clock. Device lifecycle reducers use this instead of browser timers or a second wall-clock source. Terminal records deliberately retain the same answer so an idempotent local-cleanup retry can continue after the operation outcome itself has already been sealed. """ if operation_id is None: return False with self._lock: record = self._records.get(operation_id) return bool( record is not None and record.deadline_at is not None and self._clock() >= record.deadline_at ) def latest(self) -> OperationRecord | None: with self._lock: return self._records[self._order[-1]] if self._order else None def snapshot(self, *, limit: int = 20) -> list[dict[str, Any]]: if limit < 1: return [] with self._lock: return [self._records[item].as_dict() for item in self._order[-limit:]] def _require_locked(self, operation_id: str) -> OperationRecord: try: return self._records[operation_id] except KeyError as exc: raise KeyError(f"unknown operation: {operation_id}") from exc def _append_event_locked(self, record: OperationRecord) -> None: """Append one bounded, secret-free stage fact for operator diagnosis.""" error = record.error or {} result = record.result or {} side_effect_status = error.get("side_effect_status", result.get("side_effect_status")) event: dict[str, Any] = { "schema_version": "missioncore.operation-event/v1", "sequence": record.sequence, "status": record.status, "stage_code": record.stage_code, "message_code": record.message_code, "observed_at": _iso(self._clock()), "side_effect_status": ( side_effect_status if isinstance(side_effect_status, str) else None ), "error_code": error.get("code") if isinstance(error.get("code"), str) else None, "safe_to_retry": ( error.get("safe_to_retry") if isinstance(error.get("safe_to_retry"), bool) else None ), "automatic_retry": False, } record.events.append(event) def _trim_locked(self) -> None: while len(self._order) > self._max_records: oldest_id = next( ( operation_id for operation_id in self._order if self._records[operation_id].status in TERMINAL_OPERATION_STATUSES ), None, ) # Never evict an operation that still needs reconciliation. A brief # overrun is safer than turning a later device observation into an # unknown-operation failure. if oldest_id is None: return self._order.remove(oldest_id) oldest = self._records.pop(oldest_id) if oldest.idempotency_key is not None: self._idempotency.pop(oldest.idempotency_key, None) @dataclass(slots=True) class AcquisitionRecord: acquisition_id: str device_id: str device_session_id: str compatibility_profile_id: str control_mode: ControlMode requested_streams: tuple[str, ...] target_host: str duration_seconds: float | None evidence_policy: Literal["required", "best-effort", "disabled"] state: AcquisitionState = "preparing" state_revision: int = 1 created_at: datetime = field(default_factory=utc_now) updated_at: datetime = field(default_factory=utc_now) message_code: str = "acquisition.preparing" operator_instructions: tuple[str, ...] = () result: dict[str, Any] | None = None def transition( self, state: AcquisitionState, *, message_code: str, operator_instructions: Iterable[str] = (), result: Mapping[str, Any] | None = None, ) -> None: if self.state in TERMINAL_ACQUISITION_STATES: if self.state == state: return raise ValueError(f"acquisition {self.acquisition_id} is already terminal") self.state = state self.state_revision += 1 self.updated_at = utc_now() self.message_code = message_code self.operator_instructions = tuple(operator_instructions) self.result = dict(result) if result is not None else None def as_dict(self) -> dict[str, Any]: return { "schema_version": "missioncore.acquisition-snapshot/v1alpha2", "acquisition_id": self.acquisition_id, "device_id": self.device_id, "device_session_id": self.device_session_id, "compatibility_profile_id": self.compatibility_profile_id, "control_mode": self.control_mode, "requested_streams": list(self.requested_streams), "target_host": self.target_host, "duration_seconds": self.duration_seconds, "evidence_policy": self.evidence_policy, "state": self.state, "state_revision": self.state_revision, "created_at": _iso(self.created_at), "updated_at": _iso(self.updated_at), "message_code": self.message_code, "operator_instructions": list(self.operator_instructions), "result": dict(self.result) if self.result is not None else None, } def new_acquisition_id() -> str: return f"acq-{uuid4()}" def new_device_id() -> str: return f"device-{uuid4()}" def new_device_session_id() -> str: return f"device-session-{uuid4()}"