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: ...