PR4: Storage-protocol hygiene & guard

CHANGELOG.md

@@ -5,6 +5,34 @@ All notable changes to the Context Intelligence Server are recorded here.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [7.3.0]
+
+### Added
+
+- **`lease_store` package.** Writer-lease persistence now lives behind a
+  backend-neutral `LeaseStore` protocol (`protocol` + `filesystem` + `factory`),
+  the fourth storage backend alongside `blob_store`, `queue_manager`, and
+  `identity_store`. The writer-lease detector keeps all policy (staleness,
+  conflict, the bounded single-thread I/O executor) and reaches the lease only
+  through the store, so the same detector runs unchanged against any backend.
+- **`QueueManager.session_keys()`.** A backend-neutral way to enumerate every
+  persisted session key. Boot reclaim sweeps the queue through this method
+  instead of globbing the queue directory, so the sweep works unchanged against
+  any queue backend.
+- **Storage-boundary guard test.** A standing AST tripwire asserts no module
+  outside the four storage backend packages performs a storage-artifact file
+  operation (glob/unlink/scandir) or reads a storage root path; it now also
+  catches a raw `queues_dir` glob/path-join.
+
+### Changed
+
+- **`queues_dir` removed from the `QueueManager` Protocol.** A caller enumerates
+  sessions via `session_keys()` and never learns the on-disk layout. The single
+  sanctioned exception (`registry.queues_dir_path`, used by the WriterLease boot
+  detector) resolves the directory straight from settings. The `Batch`
+  docstring now states its offsets are opaque queue-produced cursors, matching
+  `Record`'s contract.
+
 ## [7.2.0]
 
 ### Added

context_intelligence_server/lease_store/__init__.py

@@ -0,0 +1,8 @@
+"""Writer-lease persistence behind a backend-neutral Protocol."""
+
+from __future__ import annotations
+
+from context_intelligence_server.lease_store.factory import create_lease_store
+from context_intelligence_server.lease_store.protocol import LeaseRecord, LeaseStore
+
+__all__ = ["LeaseRecord", "LeaseStore", "create_lease_store"]

context_intelligence_server/lease_store/factory.py

@@ -0,0 +1,19 @@
+"""Backend selection for the writer-lease store.
+
+One backend today (filesystem). A future backend is added here and nowhere
+else -- the detector never learns which one it got.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from pathlib import Path
+
+from context_intelligence_server.lease_store.filesystem import FileSystemLeaseStore
+from context_intelligence_server.lease_store.protocol import LeaseStore
+
+
+def create_lease_store(dir_source: Callable[[], Path]) -> LeaseStore:
+    """Build the lease store. *dir_source* is resolved lazily per operation, so
+    nothing is constructed and no path is read at build time."""
+    return FileSystemLeaseStore(dir_source)

context_intelligence_server/lease_store/filesystem.py

@@ -0,0 +1,89 @@
+"""Filesystem-backed writer-lease store.
+
+The lease is one atomically-replaced ``.writer.lease`` file in a directory
+resolved lazily via *dir_source* -- resolved per operation (a cheap attribute
+read, zero syscalls) so the store, like the detector it serves, constructs
+nothing at build time and reflects a directory the tests may re-point.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from collections.abc import Callable
+from pathlib import Path
+
+from context_intelligence_server.lease_store.protocol import LeaseRecord
+
+LEASE_FILENAME = ".writer.lease"
+LEASE_TMP_FILENAME = ".writer.lease.tmp"
+_LEASE_VERSION = 1
+
+
+class FileSystemLeaseStore:
+    """A ``LeaseStore`` backed by a single atomically-written file on disk."""
+
+    def __init__(self, dir_source: Callable[[], Path]) -> None:
+        self._dir_source = dir_source
+
+    def _path(self) -> Path:
+        return self._dir_source() / LEASE_FILENAME
+
+    def read(self) -> LeaseRecord | None:
+        try:
+            text = self._path().read_text(encoding="utf-8")
+        except FileNotFoundError:
+            # A missing lease means "free directory", not a share fault.
+            return None
+        try:
+            data = json.loads(text.strip())
+            return LeaseRecord(
+                owner=str(data["owner"]),
+                host=str(data.get("host", "")),
+                pid=int(data.get("pid", 0)),
+                started_at=float(data.get("started_at", 0.0)),
+                heartbeat=float(data["heartbeat"]),
+                revision=data.get("revision"),
+                server_version=str(data.get("server_version", "")),
+                lease_version=int(data.get("lease_version", -1)),
+            )
+        except (json.JSONDecodeError, KeyError, TypeError, ValueError):
+            # Torn/malformed lease is treated as fresh-and-foreign, same
+            # strength as a genuine live peer.
+            return LeaseRecord(
+                owner="",
+                host="",
+                pid=0,
+                started_at=0.0,
+                heartbeat=0.0,
+                revision=None,
+                server_version="",
+                lease_version=-1,
+                unreadable=True,
+            )
+
+    def write(self, record: LeaseRecord) -> None:
+        directory = self._dir_source()
+        payload = {
+            "lease_version": record.lease_version,
+            "owner": record.owner,
+            "host": record.host,
+            "pid": record.pid,
+            "started_at": record.started_at,
+            "heartbeat": record.heartbeat,
+            "revision": record.revision,
+            "server_version": record.server_version,
+        }
+        tmp = directory / LEASE_TMP_FILENAME
+        tmp.write_text(
+            json.dumps(payload, separators=(",", ":")) + "\n", encoding="utf-8"
+        )
+        os.replace(tmp, directory / LEASE_FILENAME)
+
+    def delete_if_owned(self, owner: str) -> None:
+        rec = self.read()
+        if rec is not None and not rec.unreadable and rec.owner == owner:
+            try:
+                self._path().unlink()
+            except FileNotFoundError:
+                pass

context_intelligence_server/lease_store/protocol.py

@@ -0,0 +1,66 @@
+"""The writer-lease persistence boundary.
+
+A single named lease record (owner, heartbeat, identity) persisted somewhere
+durable. The writer-lease DETECTOR (``writer_lease.py``) owns all policy --
+staleness, conflict, the bounded-thread I/O executor -- and reaches the lease
+only through this Protocol, so the same detector runs unchanged against any
+backend (a filesystem file today, a blob lease or a row tomorrow).
+
+The methods are synchronous by contract: the detector runs each one on its own
+private single-thread executor to bound a hung mount to a single leaked thread,
+which ``asyncio.to_thread`` (shared pool) cannot guarantee. A backend whose I/O
+is natively async wraps itself to satisfy this sync surface.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+from typing import Protocol
+
+
+@dataclasses.dataclass
+class LeaseRecord:
+    """Parsed view of one persisted lease record.
+
+    ``unreadable=True`` marks a synthetic record standing in for a torn or
+    hand-mangled lease (decode error / missing key / wrong type / unknown
+    ``lease_version``) -- treated at fresh-foreign strength, never at face
+    value.
+    """
+
+    owner: str
+    host: str
+    pid: int
+    started_at: float
+    heartbeat: float
+    revision: str | None
+    server_version: str
+    lease_version: int
+    unreadable: bool = False
+
+
+class LeaseStore(Protocol):
+    """Persistence for exactly one writer-lease record.
+
+    All three operations may raise ``OSError`` (a share fault); the detector
+    absorbs that as "not armed", never as a conflict.
+    """
+
+    def read(self) -> LeaseRecord | None:
+        """Return the current lease record, or ``None`` when no lease exists
+        (a free directory). A torn/malformed record returns a ``LeaseRecord``
+        with ``unreadable=True`` rather than ``None``."""
+        ...
+
+    def write(self, record: LeaseRecord) -> None:
+        """Persist *record* as the current lease, atomically (a reader never
+        observes a half-written record)."""
+        ...
+
+    def delete_if_owned(self, owner: str) -> None:
+        """Delete the lease only if it is still owned by *owner*.
+
+        Never deletes a foreign lease: if a peer took it over, removing theirs
+        would actively hand the directory to a third writer. Best-effort: a
+        lease already gone is not an error."""
+        ...

context_intelligence_server/main.py

@@ -429,9 +429,9 @@ async def _boot_reclaim() -> None:
     # sync with test monkeypatches bound to the same object.
     settings = _settings
     boot_state.reclaim_enabled = settings.reclaim_enabled
-    # Iterate the QueueManager's own directory, not settings.queues_path --
-    # the two can differ (tests do this routinely).
-    keys = sorted(p.stem for p in qm.queues_dir.glob("*.log"))
+    # Enumerate the queue through the backend-neutral protocol method, not a
+    # raw directory glob, so the sweep works unchanged against any backend.
+    keys = await qm.session_keys()
     reclaimed = 0
     reclaimed_bytes = 0
     kept = 0
@@ -469,9 +469,8 @@ async def _boot_reclaim() -> None:
         # gated: they can act on a log whose offset was merely unreadable.
         if c.verdict.value != "drained" and not settings.reclaim_enabled:
             logger.warning(
-                "boot_reclaimed reason=%s path=%s session=%s bytes=%d action=dry_run",
+                "boot_reclaimed reason=%s session=%s bytes=%d action=dry_run",
                 c.reason,
-                qm.queues_dir / f"{key}.log",
                 key,
                 c.size,
             )

context_intelligence_server/queue_manager/filesystem.py

@@ -1708,6 +1708,21 @@ def _scan() -> list[str]:
 
         return await asyncio.to_thread(_scan)
 
+    async def session_keys(self) -> list[str]:
+        """Return every persisted session key (sorted).
+
+        A session key is present whenever the backend holds a queue log for it,
+        regardless of drain state or live-worker status. Callers that need to
+        sweep the whole queue (e.g. boot reclaim) enumerate here rather than
+        reaching into the backend's on-disk layout, so the sweep works
+        unchanged against any backend.
+        """
+
+        def _scan() -> list[str]:
+            return sorted(log.stem for log in self._dir.glob("*.log"))
+
+        return await asyncio.to_thread(_scan)
+
     async def is_fully_drained(self, session_id: str) -> bool:
         """True iff the session has no undrained log data left.
 

context_intelligence_server/queue_manager/protocol.py

@@ -41,10 +41,12 @@ class Batch:
         records: Queue-produced ``Record``s -- each carries its own opaque
             ``start``/``end`` cursor. The queue produces these offsets; a
             caller (the registry) only ever hands them back via ``commit``.
-        start_offset: Byte position in the log where this batch begins.
-        end_offset: Byte position in the log AFTER the last returned record.
-            This is the value passed to ``commit``. When no complete records
-            are available, ``end_offset == start_offset``.
+        start_offset: Opaque queue-produced cursor where this batch begins.
+        end_offset: Opaque queue-produced cursor AFTER the last returned
+            record -- the value handed back to ``commit``. Like ``Record``'s
+            ``start``/``end``, callers MUST NOT compute it or assume it is a
+            byte position; that framing is the queue's private invariant. When
+            no complete records are available, ``end_offset == start_offset``.
     """
 
     session_id: str
@@ -66,14 +68,12 @@ def lines(self) -> list[bytes]:
 class QueueManager(Protocol):
     """Durable, per-session append-only queue.
 
-    The method set mirrors the on-disk backend's public surface. A backend
-    reports its own queue root via ``queues_dir``; every other on-disk detail
-    stays private to the implementation.
+    The method set mirrors the on-disk backend's public surface. No ``Path`` or
+    on-disk-layout detail appears here: a caller enumerates sessions via
+    ``session_keys`` and never learns where (or whether) they live on a disk,
+    so the same consumers run unchanged against any backend.
     """
 
-    @property
-    def queues_dir(self) -> Any: ...
-
     async def heal_torn_tails(self) -> dict[str, int]: ...
 
     async def append(self, session_id: str, raw: bytes) -> None: ...
@@ -110,6 +110,8 @@ async def reclaim_orphans(
 
     async def active_sessions(self) -> list[str]: ...
 
+    async def session_keys(self) -> list[str]: ...
+
     async def recover(self) -> list[str]: ...
 
     async def derive_all_stats(self) -> dict[str, Any]: ...

context_intelligence_server/registry.py

@@ -132,15 +132,15 @@ def queue_manager(self) -> QueueManager:
 
     @property
     def queues_dir_path(self) -> Path:
-        """Queue directory path, resolved without constructing a QueueManager.
-
-        Unlike ``queue_manager``, never calls ``_ensure_infra`` -- avoids a
-        race where an observer builds a second QueueManager for the same
-        directory. Falls back to the same expression ``_ensure_infra`` uses,
-        so the two can never disagree.
+        """Queue directory path for the WriterLease boot detector.
+
+        Resolved straight from settings -- the single sanctioned exception to
+        the storage-boundary rule -- so the lease can locate the queue
+        directory WITHOUT constructing a QueueManager (avoiding a race where an
+        observer builds a second one for the same directory). This is the same
+        expression ``_ensure_infra`` feeds the queue-manager factory, so the
+        detector and the constructed queue can never disagree.
         """
-        if self._queue_manager is not None:
-            return self._queue_manager.queues_dir
         return Path(get_settings().queues_path)
 
     def _ensure_neo4j_driver(self) -> Any: