key_os/observer.py

from __future__ import annotations

from dataclasses import asdict, dataclass, field, replace
from typing import Any, Literal, Protocol

from .models import utc_now
from .security import redact, stable_hash


SIGNAL_SCHEMA = "observer_bus.signal/v1"
INBOX_SCHEMA = "observer_bus.signal_inbox/v1"
STATE_SCHEMA = "observer_bus.signal_state/v1"

SignalLane = Literal[
    "continuity",
    "repo",
    "tests",
    "mcp",
    "filesystem",
    "runtime",
    "space",
    "security",
    "workflow",
    "custom",
]

ACK_STATES = {"new", "seen", "deferred", "handled", "blocked", "superseded"}


@dataclass(frozen=True)
class SignalRoute:
    command: str = ""
    path: str = ""
    method: str = "GET"
    target: str = ""

    def to_dict(self) -> dict[str, Any]:
        return asdict(self)


@dataclass(frozen=True)
class SignalEvidence:
    kind: str = "metadata"
    items: list[dict[str, Any]] = field(default_factory=list)
    metadata: dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> dict[str, Any]:
        return redact(asdict(self))


@dataclass(frozen=True)
class SignalSource:
    adapter: str
    fingerprint: str
    host: str = "local"

    def to_dict(self) -> dict[str, Any]:
        return asdict(self)


@dataclass(frozen=True)
class SignalAck:
    state: str = "new"
    updated_at: str | None = None
    note: str | None = None

    def to_dict(self) -> dict[str, Any]:
        state = self.state if self.state in ACK_STATES else "new"
        return {"state": state, "updated_at": self.updated_at, "note": self.note}


@dataclass(frozen=True)
class SignalExpiry:
    expires_at: str | None = None
    stale_after_seconds: int | None = 86400
    state: str = "fresh"

    def to_dict(self) -> dict[str, Any]:
        return asdict(self)


@dataclass(frozen=True)
class Signal:
    id: str
    lane: str
    severity: str
    priority: int
    title: str
    message: str
    route: SignalRoute = field(default_factory=SignalRoute)
    evidence: SignalEvidence = field(default_factory=SignalEvidence)
    source: SignalSource = field(default_factory=lambda: SignalSource(adapter="unknown", fingerprint=""))
    timestamp: str = field(default_factory=utc_now)
    ack: SignalAck = field(default_factory=SignalAck)
    expiry: SignalExpiry = field(default_factory=SignalExpiry)
    tags: tuple[str, ...] = ()

    @classmethod
    def create(
        cls,
        lane: str,
        severity: str,
        title: str,
        message: str,
        *,
        route: SignalRoute | None = None,
        priority: int | None = None,
        evidence: SignalEvidence | None = None,
        source_adapter: str = "key_os",
        source_fingerprint: str | None = None,
        timestamp: str | None = None,
        tags: tuple[str, ...] = (),
    ) -> "Signal":
        if priority is None:
            priority = {"error": 100, "warn": 80, "info": 50, "ok": 20}.get(severity, 50)
        route = route or SignalRoute()
        evidence = evidence or SignalEvidence()
        identity = {
            "lane": lane,
            "severity": severity,
            "title": title,
            "message": message,
            "route": route.to_dict(),
            "evidence": evidence.to_dict(),
            "source_adapter": source_adapter,
        }
        signal_id = "sig_" + stable_hash(identity)[:20]
        fingerprint = source_fingerprint or stable_hash(identity)
        kwargs = {}
        if timestamp:
            kwargs["timestamp"] = timestamp
        return cls(
            id=signal_id,
            lane=lane,
            severity=severity,
            priority=max(0, min(100, int(priority))),
            title=title,
            message=message,
            route=route,
            evidence=evidence,
            source=SignalSource(adapter=source_adapter, fingerprint=fingerprint),
            tags=tags,
            **kwargs
        )

    def with_ack(self, state: dict[str, Any] | None) -> "Signal":
        if not state:
            return self
        ack_state = str(state.get("state") or "new")
        if ack_state not in ACK_STATES:
            ack_state = "new"
        return replace(
            self,
            ack=SignalAck(
                state=ack_state,
                updated_at=state.get("updated_at"),
                note=state.get("note"),
            ),
        )

    def to_dict(self) -> dict[str, Any]:
        route = self.route.to_dict()
        source = self.source.to_dict()
        data = {
            "schema": SIGNAL_SCHEMA,
            "id": self.id,
            "signal_id": self.id,
            "lane": self.lane,
            "severity": self.severity,
            "priority": self.priority,
            "title": self.title,
            "message": self.message,
            "evidence": self.evidence.to_dict(),
            "source": source,
            "source_fingerprint": source["fingerprint"],
            "timestamp": self.timestamp,
            "route": route,
            "route_command": route["command"],
            "route_path": route["path"],
            "command": route["command"],
            "path": route["path"],
            "ack": self.ack.to_dict(),
            "expiry": self.expiry.to_dict(),
            "tags": list(self.tags),
            "detail": self.evidence.metadata,
        }
        return redact(data)


@dataclass(frozen=True)
class AdapterStatus:
    name: str
    lane: str
    ok: bool
    message: str = ""
    detail: dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> dict[str, Any]:
        return redact(asdict(self))


class ObserverAdapter(Protocol):
    name: str
    lane: str
    capabilities: tuple[str, ...]

    def probe(self) -> AdapterStatus:
        ...

    def collect(self) -> list[Signal]:
        ...

    def fingerprint(self) -> str:
        ...