From beff6908a985005f9fb00b5685953204aa271d1b Mon Sep 17 00:00:00 2001
From: jessekemp1 <137220111+jessekemp1@users.noreply.github.com>
Date: Thu, 9 Jul 2026 01:40:52 -0400
Subject: [PATCH 01/15] Add namespace isolation primitives for KempOS contracts

---
 namespaces.py | 76 +++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 76 insertions(+)
 create mode 100644 namespaces.py

diff --git a/namespaces.py b/namespaces.py
new file mode 100644
index 0000000..73e462c
--- /dev/null
+++ b/namespaces.py
@@ -0,0 +1,76 @@
+"""Namespaced private state helpers for Cortex.
+
+These helpers keep application-specific/private state outside the repository
+working tree and under the Cortex state directory, e.g.:
+
+    ~/.cortex/namespaces/<namespace>/
+
+The first reference workload is KempOS, but this module is intentionally generic.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+from pathlib import Path
+
+
+VALID_NAMESPACE_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$")
+
+
+class InvalidNamespaceError(ValueError):
+    """Raised when a namespace is invalid or unsafe."""
+
+
+def validate_namespace(namespace: str) -> str:
+    """Validate and return a safe namespace name."""
+    if not isinstance(namespace, str):
+        raise InvalidNamespaceError("Namespace must be a string")
+
+    candidate = namespace.strip()
+    if not candidate:
+        raise InvalidNamespaceError("Namespace cannot be empty")
+
+    if not VALID_NAMESPACE_RE.fullmatch(candidate):
+        raise InvalidNamespaceError(
+            "Namespace must match ^[a-z0-9][a-z0-9_-]{0,63}$"
+        )
+
+    return candidate
+
+
+def cortex_config_dir(config_dir: Path | str | None = None) -> Path:
+    """Return the Cortex state/config directory.
+
+    Tests can inject a temp config_dir. Production defaults to ~/.cortex or
+    CORTEX_CONFIG_DIR when explicitly set.
+    """
+    if config_dir is not None:
+        return Path(config_dir).expanduser().resolve()
+    return Path(os.environ.get("CORTEX_CONFIG_DIR", Path.home() / ".cortex")).expanduser().resolve()
+
+
+def namespace_dir(namespace: str, config_dir: Path | str | None = None) -> Path:
+    """Resolve the directory for a validated namespace."""
+    safe_namespace = validate_namespace(namespace)
+    root = cortex_config_dir(config_dir)
+    namespaces_root = (root / "namespaces").resolve()
+    path = (namespaces_root / safe_namespace).resolve()
+
+    if namespaces_root not in path.parents:
+        raise InvalidNamespaceError("Resolved namespace path escapes Cortex config dir")
+
+    return path
+
+
+def ensure_namespace(namespace: str, config_dir: Path | str | None = None) -> Path:
+    """Create and return the namespace directory with private permissions."""
+    path = namespace_dir(namespace, config_dir=config_dir)
+    path.mkdir(parents=True, exist_ok=True)
+    try:
+        os.chmod(path, 0o700)
+    except OSError:
+        # chmod may fail on some platforms/filesystems. Directory creation still
+        # succeeds and health checks can surface permission concerns later.
+        pass
+    return path

From 52e113acb1968a34452d30aa5e943fc5d515fd57 Mon Sep 17 00:00:00 2001
From: jessekemp1 <137220111+jessekemp1@users.noreply.github.com>
Date: Thu, 9 Jul 2026 01:41:55 -0400
Subject: [PATCH 02/15] Add append-only namespaced event store

---
 events.py | 121 ++++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 121 insertions(+)
 create mode 100644 events.py

diff --git a/events.py b/events.py
new file mode 100644
index 0000000..d49f63e
--- /dev/null
+++ b/events.py
@@ -0,0 +1,121 @@
+"""Append-only namespaced event store for Cortex private workloads."""
+
+from __future__ import annotations
+
+import json
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from namespaces import ensure_namespace, validate_namespace
+
+
+VALID_VISIBILITIES = {"private", "internal", "public"}
+
+
+class EventWriteError(RuntimeError):
+    """Raised when an event cannot be written."""
+
+
+class EventReadError(RuntimeError):
+    """Raised when events cannot be read."""
+
+
+class InvalidVisibilityError(ValueError):
+    """Raised when event visibility is invalid."""
+
+
+class EventStore:
+    """Simple append-only JSONL event store.
+
+    This intentionally avoids optional Cortex subsystems. It is the boring
+    reliability substrate KempOS can depend on first.
+    """
+
+    def __init__(self, config_dir: Path | str | None = None):
+        self.config_dir = Path(config_dir).expanduser().resolve() if config_dir else None
+
+    def _events_path(self, namespace: str) -> Path:
+        ns_dir = ensure_namespace(namespace, config_dir=self.config_dir)
+        return ns_dir / "events.jsonl"
+
+    def append(
+        self,
+        namespace: str,
+        event_type: str,
+        payload: dict[str, Any],
+        visibility: str = "private",
+    ) -> dict[str, Any]:
+        """Append one event and return the stored event envelope."""
+        safe_namespace = validate_namespace(namespace)
+        if not isinstance(event_type, str) or not event_type.strip():
+            raise ValueError("event_type must be a non-empty string")
+        if not isinstance(payload, dict):
+            raise TypeError("payload must be a dict")
+        if visibility not in VALID_VISIBILITIES:
+            raise InvalidVisibilityError(
+                f"visibility must be one of {sorted(VALID_VISIBILITIES)}"
+            )
+
+        event = {
+            "id": f"evt_{uuid.uuid4().hex}",
+            "namespace": safe_namespace,
+            "type": event_type.strip(),
+            "visibility": visibility,
+            "created_at": datetime.now(timezone.utc).isoformat(),
+            "payload": payload,
+        }
+
+        path = self._events_path(safe_namespace)
+        try:
+            with path.open("a", encoding="utf-8") as f:
+                f.write(json.dumps(event, sort_keys=True) + "\n")
+        except OSError as exc:
+            raise EventWriteError(f"Failed to append event: {exc}") from exc
+
+        return event
+
+    def list(
+        self,
+        namespace: str,
+        event_type: str | None = None,
+        limit: int = 50,
+    ) -> list[dict[str, Any]]:
+        """List events newest-first, optionally filtered by type."""
+        safe_namespace = validate_namespace(namespace)
+        if limit <= 0:
+            return []
+
+        path = self._events_path(safe_namespace)
+        if not path.exists():
+            return []
+
+        events: list[dict[str, Any]] = []
+        try:
+            with path.open("r", encoding="utf-8") as f:
+                for line in f:
+                    if not line.strip():
+                        continue
+                    try:
+                        event = json.loads(line)
+                    except json.JSONDecodeError:
+                        # Deterministic skip for corrupt rows. Future pass can add
+                        # structured warnings if needed.
+                        continue
+                    if event.get("namespace") != safe_namespace:
+                        continue
+                    if event_type is not None and event.get("type") != event_type:
+                        continue
+                    events.append(event)
+        except OSError as exc:
+            raise EventReadError(f"Failed to read events: {exc}") from exc
+
+        return list(reversed(events))[:limit]
+
+    def get(self, namespace: str, event_id: str) -> dict[str, Any] | None:
+        """Return one event by id, or None."""
+        for event in self.list(namespace=namespace, limit=100000):
+            if event.get("id") == event_id:
+                return event
+        return None

From 5c1a464c9ed113468ecccf057e2f513c9fa55394 Mon Sep 17 00:00:00 2001
From: jessekemp1 <137220111+jessekemp1@users.noreply.github.com>
Date: Thu, 9 Jul 2026 01:42:45 -0400
Subject: [PATCH 03/15] Add explicit capability registry

---
 capabilities.py | 115 ++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 115 insertions(+)
 create mode 100644 capabilities.py

diff --git a/capabilities.py b/capabilities.py
new file mode 100644
index 0000000..2105543
--- /dev/null
+++ b/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

From 4db7301df50450d1e8048025fc3dd66083a64ef2 Mon Sep 17 00:00:00 2001
From: jessekemp1 <137220111+jessekemp1@users.noreply.github.com>
Date: Thu, 9 Jul 2026 01:43:35 -0400
Subject: [PATCH 04/15] Add namespace doctor health checks

---
 health.py | 113 ++++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 113 insertions(+)
 create mode 100644 health.py

diff --git a/health.py b/health.py
new file mode 100644
index 0000000..e3f1630
--- /dev/null
+++ b/health.py
@@ -0,0 +1,113 @@
+"""Health checks for Cortex namespaced private state."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+from capabilities import CapabilityRegistry
+from events import EventStore
+from namespaces import ensure_namespace, validate_namespace
+
+
+def _check(name: str, status: str, detail: str | None = None) -> dict[str, str | None]:
+    return {"name": name, "status": status, "detail": detail}
+
+
+def _path_inside(path: Path, root: Path) -> bool:
+    path = path.resolve()
+    root = root.resolve()
+    return path == root or root in path.parents
+
+
+def doctor_namespace(namespace: str, config_dir: Path | str | None = None) -> dict[str, Any]:
+    """Validate basic namespaced Cortex storage reliability.
+
+    This is filesystem-only in Pass 1. It should run without API keys,
+    scheduler, MCP, vector DBs, or optional intelligence modules.
+    """
+    checks: list[dict[str, str | None]] = []
+    recommendations: list[str] = []
+
+    try:
+        safe_namespace = validate_namespace(namespace)
+        checks.append(_check("namespace.validate", "pass"))
+    except Exception as exc:
+        return {
+            "namespace": namespace,
+            "status": "unhealthy",
+            "checks": [_check("namespace.validate", "fail", str(exc))],
+            "recommendations": ["Use a lowercase safe namespace like kempos."],
+        }
+
+    try:
+        ns_dir = ensure_namespace(safe_namespace, config_dir=config_dir)
+        checks.append(_check("namespace.ensure", "pass", str(ns_dir)))
+    except Exception as exc:
+        checks.append(_check("namespace.ensure", "fail", str(exc)))
+        return {
+            "namespace": safe_namespace,
+            "status": "unhealthy",
+            "checks": checks,
+            "recommendations": ["Fix Cortex config directory permissions."],
+        }
+
+    # Make sure private namespace state is not accidentally under the repo cwd.
+    try:
+        cwd = Path.cwd().resolve()
+        if _path_inside(ns_dir, cwd):
+            checks.append(_check("namespace.private_path", "fail", str(ns_dir)))
+            recommendations.append("Namespace state is inside repo working tree; move config_dir outside repo.")
+        else:
+            checks.append(_check("namespace.private_path", "pass", str(ns_dir)))
+    except Exception as exc:
+        checks.append(_check("namespace.private_path", "warn", str(exc)))
+
+    store = EventStore(config_dir=config_dir)
+    event_id = None
+    try:
+        event = store.append(
+            namespace=safe_namespace,
+            event_type="doctor_check",
+            payload={"check": "write_read"},
+            visibility="private",
+        )
+        event_id = event["id"]
+        checks.append(_check("events.write", "pass", event_id))
+    except Exception as exc:
+        checks.append(_check("events.write", "fail", str(exc)))
+
+    if event_id:
+        try:
+            readback = store.get(safe_namespace, event_id)
+            if readback:
+                checks.append(_check("events.read", "pass", event_id))
+            else:
+                checks.append(_check("events.read", "fail", "event not found after write"))
+        except Exception as exc:
+            checks.append(_check("events.read", "fail", str(exc)))
+
+    try:
+        registry = CapabilityRegistry(config_dir=config_dir)
+        capability_report = registry.list()
+        checks.append(_check("capabilities.list", "pass"))
+    except Exception as exc:
+        capability_report = {}
+        checks.append(_check("capabilities.list", "fail", str(exc)))
+
+    failed = [c for c in checks if c["status"] == "fail"]
+    warned = [c for c in checks if c["status"] == "warn"]
+    if failed:
+        status = "unhealthy"
+    elif warned:
+        status = "degraded"
+    else:
+        status = "healthy"
+
+    return {
+        "namespace": safe_namespace,
+        "status": status,
+        "checks": checks,
+        "capabilities": capability_report,
+        "recommendations": recommendations,
+    }

From 2410f972a5f4cc5a114c16250ebf86586fa7101d Mon Sep 17 00:00:00 2001
From: jessekemp1 <137220111+jessekemp1@users.noreply.github.com>
Date: Thu, 9 Jul 2026 01:43:56 -0400
Subject: [PATCH 05/15] Add namespace isolation tests

---
 tests/test_namespaces.py | 29 +++++++++++++++++++++++++++++
 1 file changed, 29 insertions(+)
 create mode 100644 tests/test_namespaces.py

diff --git a/tests/test_namespaces.py b/tests/test_namespaces.py
new file mode 100644
index 0000000..d402516
--- /dev/null
+++ b/tests/test_namespaces.py
@@ -0,0 +1,29 @@
+from pathlib import Path
+
+import pytest
+
+from namespaces import InvalidNamespaceError, ensure_namespace, namespace_dir, validate_namespace
+
+
+def test_valid_namespace_accepted():
+    assert validate_namespace("kempos") == "kempos"
+    assert validate_namespace("kempos_01") == "kempos_01"
+    assert validate_namespace("kemp-os") == "kemp-os"
+
+
+@pytest.mark.parametrize("value", ["", "KempOS", "../kempos", "kempos/private", ".kempos", "kempos space"])
+def test_invalid_namespace_rejected(value):
+    with pytest.raises(InvalidNamespaceError):
+        validate_namespace(value)
+
+
+def test_namespace_dir_under_config(tmp_path):
+    path = namespace_dir("kempos", config_dir=tmp_path)
+    assert path == (tmp_path / "namespaces" / "kempos").resolve()
+
+
+def test_ensure_namespace_creates_directory(tmp_path):
+    path = ensure_namespace("kempos", config_dir=tmp_path)
+    assert path.exists()
+    assert path.is_dir()
+    assert Path(tmp_path).resolve() in path.parents

From d32baf40578411f0435fec6132e235c0c0655992 Mon Sep 17 00:00:00 2001
From: jessekemp1 <137220111+jessekemp1@users.noreply.github.com>
Date: Thu, 9 Jul 2026 01:44:35 -0400
Subject: [PATCH 06/15] Add namespaced event store tests

---
 tests/test_events.py | 50 ++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 50 insertions(+)
 create mode 100644 tests/test_events.py

diff --git a/tests/test_events.py b/tests/test_events.py
new file mode 100644
index 0000000..18225ee
--- /dev/null
+++ b/tests/test_events.py
@@ -0,0 +1,50 @@
+import pytest
+
+from events import EventStore, EventWriteError, InvalidVisibilityError
+
+
+def test_event_append_returns_envelope(tmp_path):
+    store = EventStore(config_dir=tmp_path)
+    event = store.append("kempos", "evidence", {"signal": "test"})
+
+    assert event["id"].startswith("evt_")
+    assert event["namespace"] == "kempos"
+    assert event["type"] == "evidence"
+    assert event["visibility"] == "private"
+    assert event["payload"] == {"signal": "test"}
+
+
+def test_event_list_filters_by_type_newest_first(tmp_path):
+    store = EventStore(config_dir=tmp_path)
+    first = store.append("kempos", "evidence", {"n": 1})
+    second = store.append("kempos", "review", {"n": 2})
+    third = store.append("kempos", "evidence", {"n": 3})
+
+    evidence = store.list("kempos", event_type="evidence")
+    assert [e["id"] for e in evidence] == [third["id"], first["id"]]
+
+    all_events = store.list("kempos")
+    assert [e["id"] for e in all_events] == [third["id"], second["id"], first["id"]]
+
+
+def test_event_get_by_id(tmp_path):
+    store = EventStore(config_dir=tmp_path)
+    event = store.append("kempos", "evidence", {"signal": "test"})
+
+    assert store.get("kempos", event["id"])["payload"]["signal"] == "test"
+    assert store.get("kempos", "evt_missing") is None
+
+
+def test_event_rejects_invalid_visibility(tmp_path):
+    store = EventStore(config_dir=tmp_path)
+    with pytest.raises(InvalidVisibilityError):
+        store.append("kempos", "evidence", {}, visibility="secret")
+
+
+def test_event_append_failure_raises(tmp_path):
+    config_file = tmp_path / "not_a_dir"
+    config_file.write_text("x")
+    store = EventStore(config_dir=config_file)
+
+    with pytest.raises(EventWriteError):
+        store.append("kempos", "evidence", {"signal": "test"})

From 15601cf311e6f1c818b4f19db366b150084fd15b Mon Sep 17 00:00:00 2001
From: jessekemp1 <137220111+jessekemp1@users.noreply.github.com>
Date: Thu, 9 Jul 2026 01:44:59 -0400
Subject: [PATCH 07/15] Add capability registry tests

---
 tests/test_capabilities.py | 34 ++++++++++++++++++++++++++++++++++
 1 file changed, 34 insertions(+)
 create mode 100644 tests/test_capabilities.py

diff --git a/tests/test_capabilities.py b/tests/test_capabilities.py
new file mode 100644
index 0000000..fe1cf34
--- /dev/null
+++ b/tests/test_capabilities.py
@@ -0,0 +1,34 @@
+import pytest
+
+from capabilities import CapabilityRegistry
+
+
+def test_capability_registry_returns_explicit_statuses(tmp_path):
+    registry = CapabilityRegistry(config_dir=tmp_path)
+    report = registry.list()
+
+    assert report["events.write"]["status"] == "available"
+    assert report["events.read"]["status"] == "available"
+    assert report["namespace.isolation"]["status"] == "available"
+    assert report["health.doctor"]["status"] == "available"
+    assert report["memory.read"]["status"] == "degraded"
+    assert report["memory.write"]["status"] == "degraded"
+    assert report["scheduler.run"]["status"] == "missing"
+
+
+def test_require_allows_degraded_but_not_missing(tmp_path):
+    registry = CapabilityRegistry(config_dir=tmp_path)
+
+    assert registry.require("events.write").status == "available"
+    assert registry.require("memory.write").status == "degraded"
+
+    with pytest.raises(RuntimeError):
+        registry.require("scheduler.run")
+
+
+def test_unknown_capability_missing(tmp_path):
+    registry = CapabilityRegistry(config_dir=tmp_path)
+    status = registry.get("does.not.exist")
+
+    assert status.status == "missing"
+    assert status.reason == "unknown capability"

From e565d3465cdeb2b662fb47a6d2d6bd610ff787bc Mon Sep 17 00:00:00 2001
From: jessekemp1 <137220111+jessekemp1@users.noreply.github.com>
Date: Thu, 9 Jul 2026 01:45:22 -0400
Subject: [PATCH 08/15] Add namespace doctor tests

---
 tests/test_namespace_doctor.py | 22 ++++++++++++++++++++++
 1 file changed, 22 insertions(+)
 create mode 100644 tests/test_namespace_doctor.py

diff --git a/tests/test_namespace_doctor.py b/tests/test_namespace_doctor.py
new file mode 100644
index 0000000..0cbd77b
--- /dev/null
+++ b/tests/test_namespace_doctor.py
@@ -0,0 +1,22 @@
+from health import doctor_namespace
+
+
+def test_doctor_namespace_healthy_for_temp_namespace(tmp_path):
+    report = doctor_namespace("kempos", config_dir=tmp_path)
+
+    assert report["namespace"] == "kempos"
+    assert report["status"] == "healthy"
+    checks = {c["name"]: c["status"] for c in report["checks"]}
+    assert checks["namespace.validate"] == "pass"
+    assert checks["namespace.ensure"] == "pass"
+    assert checks["events.write"] == "pass"
+    assert checks["events.read"] == "pass"
+    assert checks["capabilities.list"] == "pass"
+
+
+def test_doctor_namespace_rejects_invalid_namespace(tmp_path):
+    report = doctor_namespace("../kempos", config_dir=tmp_path)
+
+    assert report["status"] == "unhealthy"
+    assert report["checks"][0]["name"] == "namespace.validate"
+    assert report["checks"][0]["status"] == "fail"

From f6cb39a57c7668d0d604b78bafd702b3e96a396c Mon Sep 17 00:00:00 2001
From: jessekemp1 <137220111+jessekemp1@users.noreply.github.com>
Date: Thu, 9 Jul 2026 01:46:26 -0400
Subject: [PATCH 09/15] Add KempOS contracts roadmap

---
 docs/kempos_contracts_roadmap.md | 203 +++++++++++++++++++++++++++++++
 1 file changed, 203 insertions(+)
 create mode 100644 docs/kempos_contracts_roadmap.md

diff --git a/docs/kempos_contracts_roadmap.md b/docs/kempos_contracts_roadmap.md
new file mode 100644
index 0000000..d9536d0
--- /dev/null
+++ b/docs/kempos_contracts_roadmap.md
@@ -0,0 +1,203 @@
+# KempOS Reliability Contracts Roadmap
+
+KempOS is a private reference workload that forces Cortex to become reliable enough for long-running personal/team intelligence without leaking private data into the OSS core.
+
+This roadmap intentionally separates the reliability substrate from higher-level coaching features.
+
+## Phase 0: Current Assumption
+
+KempOS must operate reliably on Cortex. If it cannot, Cortex is not robust enough yet.
+
+Cortex remains the generic engine. KempOS remains a private application layer.
+
+## Phase 1: Minimal Reliable Substrate
+
+Status: initial branch `refactor/kempos-contracts`.
+
+Build:
+
+- `namespaces.py`
+- `events.py`
+- `capabilities.py`
+- `health.py`
+- tests for each
+
+Purpose:
+
+- private state isolation
+- append-only events
+- explicit capability reporting
+- namespace doctor checks
+
+Definition of done:
+
+```bash
+pytest tests/test_namespaces.py tests/test_events.py tests/test_capabilities.py tests/test_namespace_doctor.py
+```
+
+Must pass without:
+
+- Anthropic key
+- ChromaDB
+- MCP
+- scheduler
+- optional memory/intelligence modules
+- network
+
+## Phase 2: Bridge Integration
+
+Build:
+
+- `CortexBridge.capabilities()`
+- `CortexBridge.require_capability(name)`
+- `CortexBridge.append_event(namespace, event_type, payload, visibility="private")`
+- `CortexBridge.list_events(namespace, event_type=None, limit=50)`
+- `CortexBridge.doctor_namespace(namespace)`
+
+Extend `inject_recommendation` with optional:
+
+- `namespace="default"`
+- `visibility="private"`
+
+For non-default namespaces, write recommendations to:
+
+```text
+~/.cortex/namespaces/<namespace>/recommendations.json
+```
+
+Definition of done:
+
+- existing bridge behavior remains backward compatible
+- namespaced event/recommendation calls work through the bridge
+- invalid namespaces are rejected
+- tests use temp config dirs only
+
+## Phase 3: KempOS Reference Workload Test
+
+Add:
+
+```text
+tests/reference_workloads/test_kempos_contract.py
+```
+
+Scenario:
+
+1. create namespace `kempos`
+2. append daily evidence event
+3. append weekly review event
+4. inject next-action recommendation
+5. run namespace doctor
+6. inspect capability registry
+7. assert all private paths are outside repo working tree
+
+No real personal data. Fake payloads only.
+
+Definition of done:
+
+- reference workload passes locally with filesystem-only dependencies
+- no network
+- no LLM key
+- no scheduler
+
+## Phase 4: CLI Surface
+
+Add commands only after Python contracts are stable.
+
+Target commands:
+
+```bash
+cortex capabilities
+cortex namespace doctor kempos
+cortex event append --namespace kempos --type evidence --payload sample.json
+cortex event list --namespace kempos --type evidence
+```
+
+Definition of done:
+
+- CLI returns JSON with `--json`
+- errors are explicit
+- no silent fallbacks
+
+## Phase 5: Scheduler Contract
+
+Build scheduler after the event/recommendation substrate is stable.
+
+Target API:
+
+```python
+scheduler.register(namespace, job)
+scheduler.status(namespace)
+scheduler.run_due(namespace=None)
+scheduler.history(namespace)
+```
+
+Definition of done:
+
+- daily/weekly/monthly KempOS jobs can be registered
+- job history writes to namespace-local logs or global metrics with namespace field
+- missed jobs are visible
+- failures are explicit
+
+## Phase 6: KempOS Private App MVP
+
+Private repo or server folder, not Cortex OSS.
+
+Build:
+
+- daily capture command
+- weekly review generator
+- monthly CEO review generator
+- experiment tracker
+- deterministic pattern rules
+- Cortex bridge adapter
+
+Definition of done:
+
+- KempOS stores evidence through Cortex events
+- weekly review references actual evidence only
+- no evidence means coach asks for evidence instead of hallucinating
+- personal data remains private
+
+## Phase 7: Pattern Intelligence
+
+Only after 4+ weeks of evidence.
+
+Build:
+
+- repeated avoidance detection
+- visible-signal trends
+- shipment trends
+- health consistency correlations
+- experiment result summaries
+
+Definition of done:
+
+- pattern reports cite source events/reviews
+- deterministic rules first
+- LLM synthesis optional and clearly marked
+
+## Phase 8: Dashboard / Reports
+
+Build after the evidence model stabilizes.
+
+Outputs:
+
+- weekly compound score
+- monthly trend report
+- active experiments
+- role scorecards
+- privacy-scrubbed public artifacts if desired
+
+Definition of done:
+
+- dashboard is read-only first
+- no private data leaves the server by default
+- exports are explicitly scrubbed
+
+## Non-Negotiables
+
+- No private KempOS content in Cortex OSS.
+- No silent success on write failures.
+- All private data paths must be namespace-scoped.
+- Every new reliability contract needs tests.
+- Filesystem-only MVP first. Clever later.

From 6c057016f6a0e3a8c1890590390af5c76326f055 Mon Sep 17 00:00:00 2001
From: jessekemp1 <137220111+jessekemp1@users.noreply.github.com>
Date: Thu, 9 Jul 2026 06:50:50 -0400
Subject: [PATCH 10/15] Add namespaced recommendation store

---
 recommendations_store.py | 81 ++++++++++++++++++++++++++++++++++++++++
 1 file changed, 81 insertions(+)
 create mode 100644 recommendations_store.py

diff --git a/recommendations_store.py b/recommendations_store.py
new file mode 100644
index 0000000..750168f
--- /dev/null
+++ b/recommendations_store.py
@@ -0,0 +1,81 @@
+"""Namespaced recommendation storage for Cortex private workloads."""
+
+from __future__ import annotations
+
+import json
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from namespaces import ensure_namespace, validate_namespace
+
+
+class RecommendationWriteError(RuntimeError):
+    """Raised when a recommendation cannot be written."""
+
+
+class RecommendationStore:
+    """Simple JSON-backed namespaced recommendation store."""
+
+    def __init__(self, config_dir: Path | str | None = None):
+        self.config_dir = Path(config_dir).expanduser().resolve() if config_dir else None
+
+    def _path(self, namespace: str) -> Path:
+        ns_dir = ensure_namespace(namespace, config_dir=self.config_dir)
+        return ns_dir / "recommendations.json"
+
+    def add(
+        self,
+        namespace: str,
+        title: str,
+        rationale: str,
+        priority: str = "medium",
+        type: str = "ai_suggestion",
+        effort: str = "Unknown",
+        related_project: str = "",
+        visibility: str = "private",
+    ) -> dict[str, Any]:
+        safe_namespace = validate_namespace(namespace)
+        rec = {
+            "id": f"rec_{uuid.uuid4().hex}",
+            "namespace": safe_namespace,
+            "visibility": visibility,
+            "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(timezone.utc).isoformat(),
+            "source": "CortexBridge",
+        }
+
+        path = self._path(safe_namespace)
+        try:
+            current: list[dict[str, Any]] = []
+            if path.exists() and path.read_text(encoding="utf-8").strip():
+                current = json.loads(path.read_text(encoding="utf-8"))
+            current.append(rec)
+            path.write_text(json.dumps(current, indent=2, sort_keys=True), encoding="utf-8")
+        except (OSError, json.JSONDecodeError) as exc:
+            raise RecommendationWriteError(f"Failed to write recommendation: {exc}") from exc
+
+        return rec
+
+    def list(self, namespace: str) -> list[dict[str, Any]]:
+        safe_namespace = validate_namespace(namespace)
+        path = self._path(safe_namespace)
+        if not path.exists():
+            return []
+        try:
+            text = path.read_text(encoding="utf-8").strip()
+            if not text:
+                return []
+            items = json.loads(text)
+        except (OSError, json.JSONDecodeError):
+            return []
+        return [item for item in items if item.get("namespace") == safe_namespace]

From da0bb9fc137b4b5de8381b43f4f5d9e9dd218e63 Mon Sep 17 00:00:00 2001
From: jessekemp1 <137220111+jessekemp1@users.noreply.github.com>
Date: Thu, 9 Jul 2026 06:52:08 -0400
Subject: [PATCH 11/15] Add KempOS contract bridge mixin

---
 bridge_kempos.py | 131 +++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 131 insertions(+)
 create mode 100644 bridge_kempos.py

diff --git a/bridge_kempos.py b/bridge_kempos.py
new file mode 100644
index 0000000..d2bdcea
--- /dev/null
+++ b/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

From 9b225641916302d0d6721af8a1896378c0c84003 Mon Sep 17 00:00:00 2001
From: jessekemp1 <137220111+jessekemp1@users.noreply.github.com>
Date: Thu, 9 Jul 2026 06:57:45 -0400
Subject: [PATCH 12/15] Wire KempOS contract mixin into CortexBridge

---
 bridge.py | 487 +-----------------------------------------------------
 1 file changed, 5 insertions(+), 482 deletions(-)

diff --git a/bridge.py b/bridge.py
index 36a952d..a7202ac 100644
--- a/bridge.py
+++ b/bridge.py
@@ -196,19 +196,22 @@
 try:
     from cortex.bridge_intelligence import IntelligenceMixin
     from cortex.bridge_system import SystemMixin
+    from cortex.bridge_kempos import KempOSContractsMixin
 except ImportError:
     from bridge_intelligence import IntelligenceMixin  # type: ignore[no-redef]
     from bridge_system import SystemMixin  # type: ignore[no-redef]
+    from bridge_kempos import KempOSContractsMixin  # type: ignore[no-redef]
 
 
-class CortexBridge(IntelligenceMixin, SystemMixin):
+class CortexBridge(KempOSContractsMixin, IntelligenceMixin, SystemMixin):
     """Universal interface for AI agents to interact with Cortex.
 
     Composed from domain mixins for maintainability:
+    - KempOSContractsMixin: capabilities, namespace doctor, events, namespaced recommendations
     - IntelligenceMixin: Context, recommendations, specs, feedback, analysis
     - SystemMixin: V2 Prime, health, dependencies, batch, planning, warnings
 
-    See bridge_intelligence.py and bridge_system.py for method implementations.
+    See bridge_kempos.py, bridge_intelligence.py, and bridge_system.py for method implementations.
     """
 
     def __init__(self, root_dir: Optional[str | Path] = None):
@@ -517,483 +520,3 @@ def main():
     # trigger
     trig_parser = subparsers.add_parser("trigger", help="Trigger agent")
     trig_parser.add_argument("agent")
-
-    # portfolio
-    port_parser = subparsers.add_parser("portfolio", help="Portfolio operations")
-    port_sub = port_parser.add_subparsers(dest="subcommand", help="Portfolio subcommand")
-
-    # portfolio patterns
-    patterns_parser = port_sub.add_parser("patterns", help="Get cross-project patterns")
-    patterns_parser.add_argument("--type", help="Pattern type filter")
-
-    # portfolio lessons
-    lessons_parser = port_sub.add_parser("lessons", help="Get lessons learned")
-    lessons_parser.add_argument("--project", help="Filter by project")
-    lessons_parser.add_argument("--pattern", help="Filter by pattern")
-
-    # portfolio project
-    project_parser = port_sub.add_parser("project", help="Get project context")
-    project_parser.add_argument("name", help="Project name")
-
-    # portfolio stats
-    port_sub.add_parser("stats", help="Get portfolio statistics")
-
-    # intelligence
-    intel_parser = subparsers.add_parser("intelligence", help="Query unified intelligence")
-    intel_parser.add_argument("request", help="User request")
-    intel_parser.add_argument(
-        "--project", default=None, help="Project name (auto-detected from cwd if omitted)"
-    )
-    intel_parser.add_argument(
-        "--type", default="spec", help="Query type (spec/impl/analysis/research)"
-    )
-
-    # similar-work
-    similar_parser = subparsers.add_parser("similar-work", help="Find similar work")
-    similar_parser.add_argument("domain", help="Domain/topic")
-    similar_parser.add_argument(
-        "--project", default=None, help="Project name (auto-detected from cwd if omitted)"
-    )
-    similar_parser.add_argument("--limit", type=int, default=5, help="Max results")
-
-    # session-context
-    session_parser = subparsers.add_parser("session-context", help="Get session context")
-    session_parser.add_argument(
-        "--format",
-        choices=["json", "terminal", "compact"],
-        default="json",
-        help="Output format",
-    )
-    session_parser.add_argument(
-        "--max-chars", type=int, default=450, help="Max characters for compact format"
-    )
-
-    # index-spec
-    index_parser = subparsers.add_parser("index-spec", help="Index a spec")
-    index_parser.add_argument("path", help="Path to spec file")
-    index_parser.add_argument("--project", required=True, help="Project name")
-    index_parser.add_argument("--domain", help="Domain tag")
-
-    # health - Project health analysis
-    health_parser = subparsers.add_parser("health", help="Project health analysis")
-    health_sub = health_parser.add_subparsers(dest="health_command", help="Health command")
-
-    # health summary
-    summary_parser = health_sub.add_parser("summary", help="Portfolio health summary")
-    summary_parser.add_argument("--days", type=int, default=7, help="Days to analyze")
-
-    # health project
-    project_health_parser = health_sub.add_parser("project", help="Detailed project health")
-    project_health_parser.add_argument("name", help="Project name")
-    project_health_parser.add_argument("--days", type=int, default=7, help="Days to analyze")
-
-    # health compare
-    compare_parser = health_sub.add_parser("compare", help="Compare two projects")
-    compare_parser.add_argument("project1", help="First project")
-    compare_parser.add_argument("project2", help="Second project")
-    compare_parser.add_argument("--days", type=int, default=7, help="Days to analyze")
-
-    # health trends
-    trends_parser = health_sub.add_parser("trends", help="Health trends for project")
-    trends_parser.add_argument("name", help="Project name")
-
-    # plan - Planning and execution
-    plan_parser = subparsers.add_parser("plan", help="Plan creation and execution")
-    plan_sub = plan_parser.add_subparsers(dest="plan_command", help="Plan command")
-
-    # plan create
-    create_plan_parser = plan_sub.add_parser("create", help="Create a plan from recommendations")
-    create_plan_parser.add_argument("project", help="Project name")
-    create_plan_parser.add_argument("--title", help="Plan title")
-
-    # plan list
-    list_plans_parser = plan_sub.add_parser("list", help="List all plans")
-    list_plans_parser.add_argument(
-        "--status",
-        choices=["draft", "active", "completed", "cancelled"],
-        help="Filter by status",
-    )
-
-    # plan show
-    show_plan_parser = plan_sub.add_parser("show", help="Show plan details")
-    show_plan_parser.add_argument("plan_id", help="Plan ID")
-    show_plan_parser.add_argument(
-        "--format",
-        choices=["json", "markdown"],
-        default="markdown",
-        help="Output format",
-    )
-
-    # plan start
-    start_plan_parser = plan_sub.add_parser("start", help="Start plan execution")
-    start_plan_parser.add_argument("plan_id", help="Plan ID")
-
-    # plan complete
-    complete_step_parser = plan_sub.add_parser("complete", help="Complete a step")
-    complete_step_parser.add_argument("step_id", help="Step ID")
-    complete_step_parser.add_argument("--notes", default="", help="Completion notes")
-
-    # plan progress
-    plan_sub.add_parser("progress", help="Show active plan progress")
-
-    # profile - Project profiling (Layer 1)
-    profile_parser = subparsers.add_parser(
-        "profile", help="Analyze project structure and tech stack"
-    )
-    profile_parser.add_argument("project", help="Project name")
-
-    # patterns - Pattern search (Layer 2)
-    patterns_parser = subparsers.add_parser(
-        "patterns", help="Find similar work from other projects"
-    )
-    patterns_parser.add_argument("project", help="Current project name")
-    patterns_parser.add_argument("task", help="Task description")
-    patterns_parser.add_argument("--limit", type=int, default=5, help="Maximum results")
-
-    # mine-anti-patterns - Extract anti-patterns from failed outcomes
-    mine_parser = subparsers.add_parser(
-        "mine-anti-patterns", help="Extract anti-patterns from failed outcomes"
-    )
-    mine_parser.add_argument("--days", type=int, default=30, help="Days to analyze")
-    mine_parser.add_argument("--dry-run", action="store_true", help="Don't write to files")
-
-    # batch
-    batch_parser = subparsers.add_parser("batch", help="Batch API operations")
-    batch_sub = batch_parser.add_subparsers(dest="batch_command", help="Batch subcommand")
-
-    # batch research
-    batch_research_parser = batch_sub.add_parser("research", help="Submit research batch")
-    batch_research_parser.add_argument(
-        "--file", required=True, help="JSON file with research items"
-    )
-
-    # batch briefing
-    batch_briefing_parser = batch_sub.add_parser("briefing", help="Submit briefing batch")
-    batch_briefing_parser.add_argument(
-        "--file", required=True, help="JSON file with briefing contexts"
-    )
-
-    # batch status
-    batch_status_parser = batch_sub.add_parser("status", help="Get batch status")
-    batch_status_parser.add_argument("batch_id", help="Batch ID to check")
-
-    # recommendations - Smart recommendations
-    rec_parser = subparsers.add_parser("recommendations", help="Get smart recommendations")
-    rec_sub = rec_parser.add_subparsers(dest="rec_command", help="Recommendation command")
-
-    # recommendations report - full smart recommendations report
-    rec_sub.add_parser("report", help="Get full recommendations report")
-
-    # recommendations next - single next action
-    rec_sub.add_parser("next", help="Get recommended next action")
-
-    # recommendations alerts - risk alerts
-    rec_sub.add_parser("alerts", help="Get risk alerts")
-
-    # recommendations priorities - priority projects
-    rec_priorities_parser = rec_sub.add_parser("priorities", help="Get priority projects")
-    rec_priorities_parser.add_argument("--limit", type=int, default=5, help="Maximum projects")
-
-    # recommendations get (legacy)
-    rec_get_parser = rec_sub.add_parser("get", help="Get prioritized recommendations")
-    rec_get_parser.add_argument("project", help="Project name")
-    rec_get_parser.add_argument("--limit", type=int, default=10, help="Maximum recommendations")
-
-    # recommendations dashboard (legacy)
-    rec_dashboard_parser = rec_sub.add_parser("dashboard", help="Get recommendation dashboard")
-    rec_dashboard_parser.add_argument("project", help="Project name")
-    rec_dashboard_parser.add_argument(
-        "--limit", type=int, default=10, help="Maximum recommendations"
-    )
-
-    args = parser.parse_args()
-    bridge = CortexBridge()
-
-    if args.command == "context":
-        print(json.dumps(bridge.get_context(args.query, project=args.project), indent=2))
-    elif args.command == "inject":
-        success = bridge.inject_recommendation(args.title, args.rationale, priority=args.priority)
-        print(json.dumps({"success": success}))
-    elif args.command == "trigger":
-        print(json.dumps(bridge.trigger_action(args.agent)))
-    elif args.command == "portfolio":
-        if args.subcommand == "patterns":
-            result = bridge.get_patterns(pattern_type=getattr(args, "type", None))
-            print(json.dumps(result, indent=2))
-        elif args.subcommand == "lessons":
-            result = bridge.get_lessons(
-                project=getattr(args, "project", None),
-                pattern=getattr(args, "pattern", None),
-            )
-            print(json.dumps(result, indent=2))
-        elif args.subcommand == "project":
-            result = bridge.get_portfolio_context(args.name)
-            print(json.dumps(result, indent=2))
-        elif args.subcommand == "stats":
-            result = bridge.get_portfolio_stats()
-            print(json.dumps(result, indent=2))
-        else:
-            port_parser.print_help()
-    elif args.command == "intelligence":
-        project = args.project or bridge._detect_current_project()
-        result = bridge.query_intelligence(args.request, project, getattr(args, "type", "spec"))
-        print(json.dumps(result, indent=2, default=str))
-    elif args.command == "similar-work":
-        project = args.project or bridge._detect_current_project()
-        result = bridge.find_similar_work(args.domain, project, getattr(args, "limit", 5))
-        print(json.dumps(result, indent=2, default=str))
-    elif args.command == "session-context":
-        result = bridge.get_session_context()
-        format_type = getattr(args, "format", "json")
-
-        # Handle compact format for inject_context hook (<450 chars)
-        if format_type == "compact":
-            if "error" in result:
-                # Fallback to empty context
-                print("")
-                sys.exit(0)
-
-            max_chars = getattr(args, "max_chars", 450)
-            parts = []
-
-            # Project
-            project = result.get("project", "Unknown")
-            parts.append(f"Project: {project}")
-
-            # Focus (truncate if needed)
-            focus = result.get("current_focus", "No active focus")
-            if len(focus) > 50:
-                focus = focus[:47] + "..."
-            parts.append(f"Focus: {focus}")
-
-            # First goal if available
-            if result.get("active_goals"):
-                goal = result["active_goals"][0]
-                if len(goal) > 40:
-                    goal = goal[:37] + "..."
-                parts.append(f"Goal: {goal}")
-
-            # Build compact string
-            compact = " | ".join(parts)
-
-            # Enforce max_chars limit
-            if len(compact) > max_chars:
-                compact = compact[: max_chars - 3] + "..."
-
-            print(compact)
-
-        # Handle terminal format for shell startup display
-        elif format_type == "terminal":
-            if "error" in result:
-                # Fail silently for startup hook
-                sys.exit(0)
-
-            print("\n🧠 Cortex Session Intelligence\n")
-
-            # Handle nested project structure
-            project_info = result.get("project", {})
-            if isinstance(project_info, dict):
-                project_name = project_info.get("name", "Unknown")
-            else:
-                project_name = project_info
-            print(f"📂 Project: {project_name}")
-
-            print(
-                f"🎯 Focus: {result.get('focus', result.get('current_focus', 'No active focus'))}"
-            )
-
-            # Handle goals (could be in 'goals' or 'active_goals')
-            goals = result.get("goals", result.get("active_goals", []))
-            if goals:
-                goals_display = goals[:3]  # Show max 3 goals
-                print(f"✅ Goals: {', '.join(goals_display)}")
-
-            # Handle recent work (could be in git.recent_commits or recent_work)
-            git_info = result.get("git", {})
-            recent_commits = git_info.get("recent_commits", result.get("recent_work", []))
-            if recent_commits:
-                print("\n📝 Recent Work:")
-                for work in recent_commits[:3]:  # Show max 3 items
-                    print(f"   • {work.get('summary', work.get('commit', 'Unknown'))}")
-
-            print()  # Empty line for spacing
-        else:
-            # JSON format (default)
-            print(json.dumps(result, indent=2, default=str))
-    elif args.command == "index-spec":
-        result = bridge.index_spec(args.path, args.project, getattr(args, "domain", None))
-        print(json.dumps(result, indent=2, default=str))
-    elif args.command == "health":
-        import subprocess
-
-        # Delegate to data agent CLI
-        cortex_root = Path(__file__).parent
-
-        if args.health_command == "summary":
-            subprocess.run(
-                [
-                    sys.executable,
-                    "-m",
-                    "agents.data_agent.cli",
-                    "summary",
-                    str(getattr(args, "days", 7)),
-                ],
-                cwd=cortex_root,
-            )
-        elif args.health_command == "project":
-            subprocess.run(
-                [
-                    sys.executable,
-                    "-m",
-                    "agents.data_agent.cli",
-                    "project",
-                    args.name,
-                    str(getattr(args, "days", 7)),
-                ],
-                cwd=cortex_root,
-            )
-        elif args.health_command == "compare":
-            subprocess.run(
-                [
-                    sys.executable,
-                    "-m",
-                    "agents.data_agent.cli",
-                    "compare",
-                    args.project1,
-                    args.project2,
-                    str(getattr(args, "days", 7)),
-                ],
-                cwd=cortex_root,
-            )
-        elif args.health_command == "trends":
-            subprocess.run(
-                [sys.executable, "-m", "agents.data_agent.cli", "trends", args.name],
-                cwd=cortex_root,
-            )
-        else:
-            health_parser.print_help()
-    elif args.command == "plan":
-        if args.plan_command == "create":
-            result = bridge.create_plan(args.project, title=getattr(args, "title", None))
-            print(json.dumps(result, indent=2))
-        elif args.plan_command == "list":
-            result = bridge.list_plans(status=getattr(args, "status", None))
-            print(json.dumps(result, indent=2))
-        elif args.plan_command == "show":
-            result = bridge.get_plan(args.plan_id, format=getattr(args, "format", "json"))
-            if getattr(args, "format", "json") == "markdown":
-                print(result.get("markdown", "No plan found"))
-            else:
-                print(json.dumps(result, indent=2))
-        elif args.plan_command == "start":
-            result = bridge.start_plan(args.plan_id)
-            print(json.dumps(result, indent=2))
-        elif args.plan_command == "complete":
-            result = bridge.complete_step(args.step_id, notes=getattr(args, "notes", ""))
-            print(json.dumps(result, indent=2))
-        elif args.plan_command == "progress":
-            result = bridge.get_plan_progress()
-            print(json.dumps(result, indent=2))
-        else:
-            plan_parser.print_help()
-    elif args.command == "profile":
-        result = bridge.get_project_profile(args.project)
-        print(json.dumps(result, indent=2))
-    elif args.command == "patterns":
-        result = bridge.find_similar_work(args.project, args.task, limit=getattr(args, "limit", 5))
-        print(json.dumps(result, indent=2))
-    elif args.command == "mine-anti-patterns":
-        import subprocess
-
-        script_path = os.environ.get("CORTEX_ANTI_PATTERNS_SCRIPT")
-        if not script_path:
-            print(
-                "Error: CORTEX_ANTI_PATTERNS_SCRIPT env var not set.\n"
-                "Point it to your mine_anti_patterns.py script path.",
-                file=sys.stderr,
-            )
-            sys.exit(1)
-        cmd = ["python3", script_path]
-        cmd.extend(["--days", str(getattr(args, "days", 30))])
-        if getattr(args, "dry_run", False):
-            cmd.append("--dry-run")
-        result = subprocess.run(cmd, capture_output=False)
-        sys.exit(result.returncode)
-    elif args.command == "recommendations":
-        if args.rec_command == "report":
-            result = bridge.get_recommendations()
-            print(json.dumps(result, indent=2, default=str))
-        elif args.rec_command == "next":
-            result = bridge.get_next_action()
-            # Pretty print for terminal
-            print("\n🎯 Recommended Next Action")
-            print(f"   Priority: {result.get('priority', 'Unknown')}")
-            print(f"   Action: {result.get('action', 'Unknown')}")
-            if result.get("project"):
-                print(f"   Project: {result['project']}")
-            print()
-        elif args.rec_command == "alerts":
-            alerts = bridge.get_risk_alerts()
-            if not alerts or (len(alerts) == 1 and "error" in alerts[0]):
-                print("\n✅ No risk alerts\n")
-            else:
-                print(f"\n⚠️  Risk Alerts ({len(alerts)})")
-                for alert in alerts:
-                    icon = (
-                        "🔴"
-                        if alert.get("severity") == "HIGH"
-                        else "🟡"
-                        if alert.get("severity") == "MEDIUM"
-                        else "🟢"
-                    )
-                    print(
-                        f"   {icon} [{alert.get('severity', 'Unknown')}] {alert.get('message', 'Unknown')}"
-                    )
-                    print(f"      → {alert.get('recommendation', '')}")
-                print()
-        elif args.rec_command == "priorities":
-            priorities = bridge.get_priority_projects(limit=getattr(args, "limit", 5))
-            print(f"\n📋 Priority Projects ({len(priorities)})")
-            for i, p in enumerate(priorities, 1):
-                health = f" [{p.get('health_score')}/100]" if p.get("health_score") else ""
-                print(
-                    f"   {i}. [{p.get('priority', 'Unknown')}] {p.get('project', 'Unknown')}{health}"
-                )
-                print(f"      {p.get('reason', '')}")
-            print()
-        elif args.rec_command == "get":
-            result = bridge.get_smart_recommendations(
-                args.project, limit=getattr(args, "limit", 10)
-            )
-            print(json.dumps(result, indent=2))
-        elif args.rec_command == "dashboard":
-            result = bridge.get_recommendation_dashboard(
-                args.project, limit=getattr(args, "limit", 10)
-            )
-            print(json.dumps(result, indent=2))
-        else:
-            rec_parser.print_help()
-    elif args.command == "batch":
-        if args.batch_command == "research":
-            # Load research items from JSON file
-            with open(args.file, "r") as f:
-                research_items = json.load(f)
-            result = bridge.submit_research_batch(research_items)
-            print(json.dumps(result, indent=2))
-        elif args.batch_command == "briefing":
-            # Load briefing contexts from JSON file
-            with open(args.file, "r") as f:
-                contexts = json.load(f)
-            result = bridge.submit_briefing_batch(contexts)
-            print(json.dumps(result, indent=2))
-        elif args.batch_command == "status":
-            result = bridge.get_batch_status(args.batch_id)
-            print(json.dumps(result, indent=2))
-        else:
-            batch_parser.print_help()
-    else:
-        parser.print_help()
-
-
-if __name__ == "__main__":
-    main()

From 23ecd70846846abda000140bd081434e22bbf8f6 Mon Sep 17 00:00:00 2001
From: jessekemp1 <137220111+jessekemp1@users.noreply.github.com>
Date: Thu, 9 Jul 2026 06:58:35 -0400
Subject: [PATCH 13/15] Add bridge KempOS contract tests

---
 tests/test_bridge_kempos_contract.py | 74 ++++++++++++++++++++++++++++
 1 file changed, 74 insertions(+)
 create mode 100644 tests/test_bridge_kempos_contract.py

diff --git a/tests/test_bridge_kempos_contract.py b/tests/test_bridge_kempos_contract.py
new file mode 100644
index 0000000..adb94d3
--- /dev/null
+++ b/tests/test_bridge_kempos_contract.py
@@ -0,0 +1,74 @@
+from pathlib import Path
+
+from bridge import CortexBridge
+
+
+class DummyConfig:
+    def __init__(self, config_dir):
+        self.config_dir = config_dir
+        self.context_optimizer_enabled = False
+        self.implicit_feedback_enabled = False
+        self.tiered_memory_enabled = False
+        self.hybrid_retrieval_enabled = False
+        self.defensive_prompting_enabled = False
+        self.prompt_versioning_enabled = False
+
+
+def make_bridge(tmp_path):
+    bridge = CortexBridge(root_dir=tmp_path)
+    bridge.config = DummyConfig(tmp_path / ".cortex")
+    return bridge
+
+
+def test_bridge_appends_and_lists_events(tmp_path):
+    bridge = make_bridge(tmp_path)
+
+    event = bridge.append_event("kempos", "evidence", {"signal": "test"})
+    events = bridge.list_events("kempos", event_type="evidence")
+
+    assert event["namespace"] == "kempos"
+    assert events[0]["id"] == event["id"]
+
+
+def test_bridge_capabilities_include_events(tmp_path):
+    bridge = make_bridge(tmp_path)
+    report = bridge.capabilities()
+
+    assert report["events.write"]["status"] == "available"
+    assert report["events.read"]["status"] == "available"
+
+
+def test_bridge_doctor_namespace(tmp_path):
+    bridge = make_bridge(tmp_path)
+    report = bridge.doctor_namespace("kempos")
+
+    assert report["status"] == "healthy"
+    assert report["namespace"] == "kempos"
+
+
+def test_bridge_namespaced_recommendation_writes_namespace_file(tmp_path):
+    bridge = make_bridge(tmp_path)
+
+    ok = bridge.inject_recommendation(
+        title="Ship one small artifact",
+        rationale="KempOS weekly loop requires visible evidence.",
+        type="kempos_next_action",
+        namespace="kempos",
+        visibility="private",
+    )
+
+    assert ok is True
+    rec_path = tmp_path / ".cortex" / "namespaces" / "kempos" / "recommendations.json"
+    assert rec_path.exists()
+    assert "Ship one small artifact" in rec_path.read_text()
+
+
+def test_bridge_invalid_namespace_rejected_for_events(tmp_path):
+    bridge = make_bridge(tmp_path)
+
+    try:
+        bridge.append_event("../kempos", "evidence", {})
+    except Exception as exc:
+        assert "Namespace" in str(exc) or "namespace" in str(exc)
+    else:
+        raise AssertionError("invalid namespace should have been rejected")

From ebe3d53af7b291c229e6d1709c9d132314496d26 Mon Sep 17 00:00:00 2001
From: jessekemp1 <137220111+jessekemp1@users.noreply.github.com>
Date: Thu, 9 Jul 2026 10:35:25 -0400
Subject: [PATCH 14/15] Wrap event path failures as write errors

---
 events.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/events.py b/events.py
index d49f63e..bfc7311 100644
--- a/events.py
+++ b/events.py
@@ -67,8 +67,8 @@ def append(
             "payload": payload,
         }
 
-        path = self._events_path(safe_namespace)
         try:
+            path = self._events_path(safe_namespace)
             with path.open("a", encoding="utf-8") as f:
                 f.write(json.dumps(event, sort_keys=True) + "\n")
         except OSError as exc:

From 63d0e6d03c6df920d1d969797c24780327f0cee2 Mon Sep 17 00:00:00 2001
From: jessekemp1 <137220111+jessekemp1@users.noreply.github.com>
Date: Thu, 9 Jul 2026 10:44:29 -0400
Subject: [PATCH 15/15] Wrap recommendation path failures as write errors

---
 recommendations_store.py             |  2 +-
 tests/test_bridge_kempos_contract.py | 16 ++++++++++++++++
 2 files changed, 17 insertions(+), 1 deletion(-)

diff --git a/recommendations_store.py b/recommendations_store.py
index 750168f..7563746 100644
--- a/recommendations_store.py
+++ b/recommendations_store.py
@@ -54,8 +54,8 @@ def add(
             "source": "CortexBridge",
         }
 
-        path = self._path(safe_namespace)
         try:
+            path = self._path(safe_namespace)
             current: list[dict[str, Any]] = []
             if path.exists() and path.read_text(encoding="utf-8").strip():
                 current = json.loads(path.read_text(encoding="utf-8"))
diff --git a/tests/test_bridge_kempos_contract.py b/tests/test_bridge_kempos_contract.py
index adb94d3..b3dd1da 100644
--- a/tests/test_bridge_kempos_contract.py
+++ b/tests/test_bridge_kempos_contract.py
@@ -1,6 +1,9 @@
 from pathlib import Path
 
+import pytest
+
 from bridge import CortexBridge
+from recommendations_store import RecommendationStore, RecommendationWriteError
 
 
 class DummyConfig:
@@ -63,6 +66,19 @@ def test_bridge_namespaced_recommendation_writes_namespace_file(tmp_path):
     assert "Ship one small artifact" in rec_path.read_text()
 
 
+def test_recommendation_write_failure_raises(tmp_path):
+    config_file = tmp_path / "not_a_dir"
+    config_file.write_text("x")
+    store = RecommendationStore(config_dir=config_file)
+
+    with pytest.raises(RecommendationWriteError):
+        store.add(
+            namespace="kempos",
+            title="Ship one small artifact",
+            rationale="KempOS weekly loop requires visible evidence.",
+        )
+
+
 def test_bridge_invalid_namespace_rejected_for_events(tmp_path):
     bridge = make_bridge(tmp_path)