NODEDC_MISSION_CORE/src/k1link/web/device_lifecycle.py

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()}"