Refactor Cortex namespace/event contracts for KempOS reliability

bridge_kempos.py

@@ -0,0 +1,131 @@
+"""Bridge mixin for explicit Cortex reliability contracts."""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+from capabilities import CapabilityRegistry
+from events import EventStore
+from health import doctor_namespace
+from recommendations_store import RecommendationStore
+
+
+class KempOSContractsMixin:
+    """Expose namespace/event/capability contracts through CortexBridge.
+
+    The methods are generic despite the name: KempOS is the reference workload,
+    not a special case in the storage layer.
+    """
+
+    def _contract_config_dir(self):
+        config = getattr(self, "config", None)
+        return getattr(config, "config_dir", None)
+
+    def capabilities(self) -> dict[str, dict]:
+        return CapabilityRegistry(config_dir=self._contract_config_dir()).list()
+
+    def require_capability(self, name: str) -> dict:
+        return CapabilityRegistry(config_dir=self._contract_config_dir()).require(name).to_dict()
+
+    def append_event(
+        self,
+        namespace: str,
+        event_type: str,
+        payload: dict[str, Any],
+        visibility: str = "private",
+    ) -> dict[str, Any]:
+        return EventStore(config_dir=self._contract_config_dir()).append(
+            namespace=namespace,
+            event_type=event_type,
+            payload=payload,
+            visibility=visibility,
+        )
+
+    def list_events(
+        self,
+        namespace: str,
+        event_type: str | None = None,
+        limit: int = 50,
+    ) -> list[dict[str, Any]]:
+        return EventStore(config_dir=self._contract_config_dir()).list(
+            namespace=namespace,
+            event_type=event_type,
+            limit=limit,
+        )
+
+    def doctor_namespace(self, namespace: str) -> dict[str, Any]:
+        return doctor_namespace(namespace, config_dir=self._contract_config_dir())
+
+    def inject_recommendation(
+        self,
+        title: str,
+        rationale: str,
+        priority: str = "medium",
+        type: str = "ai_suggestion",
+        effort: str = "Unknown",
+        related_project: str = "",
+        namespace: str = "default",
+        visibility: str = "private",
+    ) -> bool:
+        """Inject recommendation with namespace-aware storage.
+
+        For namespace='default', preserve the legacy external_recommendations.json
+        behavior closely. For any other namespace, store under
+        ~/.cortex/namespaces/<namespace>/recommendations.json.
+        """
+        if namespace != "default":
+            try:
+                RecommendationStore(config_dir=self._contract_config_dir()).add(
+                    namespace=namespace,
+                    title=title,
+                    rationale=rationale,
+                    priority=priority,
+                    type=type,
+                    effort=effort,
+                    related_project=related_project,
+                    visibility=visibility,
+                )
+                return True
+            except Exception:
+                return False
+
+        rec_data = {
+            "id": f"bridge_{datetime.now().timestamp()}",
+            "title": title,
+            "type": type,
+            "priority": priority,
+            "rationale": rationale,
+            "estimated_effort": effort,
+            "estimated_impact": priority,
+            "confidence": 0.95,
+            "related_projects": [related_project] if related_project else [],
+            "description": f"Injected via Cortex Bridge.\nRationale: {rationale}",
+            "created_at": datetime.now().isoformat(),
+            "source": "CortexBridge",
+            "namespace": namespace,
+            "visibility": visibility,
+        }
+
+        if (self.root_dir / "bridge.py").exists():
+            external_file = self.root_dir / "external_recommendations.json"
+        else:
+            external_file = self.root_dir / "cortex" / "external_recommendations.json"
+
+        try:
+            current_recs = []
+            if external_file.exists():
+                content = external_file.read_text()
+                if content.strip():
+                    try:
+                        current_recs = json.loads(content)
+                    except json.JSONDecodeError:
+                        current_recs = []
+
+            current_recs.append(rec_data)
+            external_file.write_text(json.dumps(current_recs, indent=2))
+            return True
+        except Exception:
+            return False

capabilities.py

@@ -0,0 +1,115 @@
+"""Explicit Cortex capability registry.
+
+This module makes degraded/missing behavior visible to callers instead of
+requiring them to infer it from optional imports or silent fallbacks.
+"""
+
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Literal
+
+from events import EventStore
+from namespaces import ensure_namespace
+
+CapabilityState = Literal["available", "degraded", "missing"]
+
+
+@dataclass(frozen=True)
+class CapabilityStatus:
+    name: str
+    status: CapabilityState
+    reason: str | None = None
+    fallback: str | None = None
+    checked_at: str | None = None
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+
+class CapabilityRegistry:
+    """Small explicit registry for reliability contracts.
+
+    Pass 1 is intentionally conservative: filesystem-backed namespace/events
+    are available; optional higher-order systems are degraded or missing until
+    specific contracts are implemented.
+    """
+
+    def __init__(self, config_dir: Path | str | None = None):
+        self.config_dir = Path(config_dir).expanduser().resolve() if config_dir else None
+
+    def _now(self) -> str:
+        return datetime.now(timezone.utc).isoformat()
+
+    def _available(self, name: str, reason: str | None = None) -> CapabilityStatus:
+        return CapabilityStatus(name=name, status="available", reason=reason, checked_at=self._now())
+
+    def _degraded(self, name: str, reason: str, fallback: str | None = None) -> CapabilityStatus:
+        return CapabilityStatus(
+            name=name,
+            status="degraded",
+            reason=reason,
+            fallback=fallback,
+            checked_at=self._now(),
+        )
+
+    def _missing(self, name: str, reason: str) -> CapabilityStatus:
+        return CapabilityStatus(name=name, status="missing", reason=reason, checked_at=self._now())
+
+    def _events_available(self, capability_name: str) -> CapabilityStatus:
+        try:
+            ensure_namespace("healthcheck", config_dir=self.config_dir)
+            store = EventStore(config_dir=self.config_dir)
+            event = store.append(
+                namespace="healthcheck",
+                event_type="capability_check",
+                payload={"capability": capability_name},
+            )
+            if store.get("healthcheck", event["id"]):
+                return self._available(capability_name, "filesystem event store write/read ok")
+            return self._degraded(capability_name, "event write succeeded but readback failed")
+        except Exception as exc:  # pragma: no cover - exact failure platform-dependent
+            return self._missing(capability_name, str(exc))
+
+    def get(self, name: str) -> CapabilityStatus:
+        if name in {"events.write", "events.read", "namespace.isolation", "health.doctor"}:
+            if name == "namespace.isolation":
+                try:
+                    ensure_namespace("healthcheck", config_dir=self.config_dir)
+                    return self._available(name, "namespaced private state directory available")
+                except Exception as exc:
+                    return self._missing(name, str(exc))
+            if name == "health.doctor":
+                return self._available(name, "doctor_namespace is filesystem-backed")
+            return self._events_available(name)
+
+        if name == "recommendations.inject":
+            return self._degraded(name, "namespaced recommendation contract pending Pass 2", "events.write")
+        if name in {"memory.read", "memory.write"}:
+            return self._degraded(name, "optional memory subsystem not part of Pass 1 contract", "events.jsonl")
+        if name in {"scheduler.status", "scheduler.run"}:
+            return self._missing(name, "scheduler contract deferred to later phase")
+
+        return self._missing(name, "unknown capability")
+
+    def list(self) -> dict[str, dict]:
+        names = [
+            "memory.read",
+            "memory.write",
+            "events.read",
+            "events.write",
+            "recommendations.inject",
+            "scheduler.status",
+            "scheduler.run",
+            "health.doctor",
+            "namespace.isolation",
+        ]
+        return {name: self.get(name).to_dict() for name in names}
+
+    def require(self, name: str) -> CapabilityStatus:
+        status = self.get(name)
+        if status.status == "missing":
+            raise RuntimeError(f"Required capability missing: {name} ({status.reason})")
+        return status