codenib.storage

Wiki-only database boundary.

This module exposes the pluggable :class:WikiStore contract and its supported SQLite implementation. Repository indexes remain manifest-bound file artifacts; this namespace does not provide a generic catalog, CAS, or backend registry.

Classes:

Name	Description
SQLiteWikiStore	A short-connection, transactional SQLite store for Wiki envelopes.
WikiStore	Minimal persistence boundary consumed by AgentWiki.
WikiStoreCorruptionError	Persisted Wiki data failed an integrity or decoding check.
WikiStoredEntry	One complete Wiki cache envelope and its stable identity.
WikiStoreError	Base error for a Wiki store operation.
WikiStoreSchemaError	The database does not implement the supported Wiki schema.
WikiStoreValidationError	The caller supplied an invalid Wiki entry or query.

SQLiteWikiStore

SQLiteWikiStore(path: str | PathLike[str])

A short-connection, transactional SQLite store for Wiki envelopes.

Methods:

Name	Description
generation_guard	Own one entry generation after a bounded file-lock acquisition.

Source code in codenib/wiki/sqlite_store.py

def __init__(self, path: str | os.PathLike[str]) -> None:
    resolved = self._resolve_path(path)
    self.path = resolved
    self._connection_path = resolved
    self._lock_directory = Path(f"{resolved}.locks")
    self._read_only = False
    try:
        resolved.parent.mkdir(parents=True, exist_ok=True)
        self._lock_directory.mkdir(parents=True, exist_ok=True, mode=0o700)
        # Lock acquisition linearizes schema identity and the WAL transition
        # for this database. Initialization takes no entry lock, retains no
        # owner after return, and the OS releases the file lock on exit.
        with FileLock(str(self._lock_directory / ".initialize.lock")):
            self._create_database_file_if_missing()
            self._initialize()
    except WikiStoreError:
        raise
    except OSError as exc:
        raise WikiStoreError("Wiki database initialization lock failed") from exc

generation_guard

generation_guard(entry_id: str) -> Iterator[None]

Own one entry generation after a bounded file-lock acquisition.

Successful FileLock.acquire is the cross-process linearization point. A timeout leaves the existing owner untouched, and this process releases only a lock it acquired successfully.

Source code in codenib/wiki/sqlite_store.py

@contextmanager
def generation_guard(self, entry_id: str) -> Iterator[None]:
    """Own one entry generation after a bounded file-lock acquisition.

    Successful ``FileLock.acquire`` is the cross-process linearization
    point. A timeout leaves the existing owner untouched, and this process
    releases only a lock it acquired successfully.
    """

    if self._read_only:
        raise WikiStoreError("read-only Wiki store cannot guard generation")
    entry_id = _validate_identifier(entry_id, field="entry_id")
    lock_name = hashlib.sha256(entry_id.encode("utf-8")).hexdigest() + ".lock"
    try:
        self._lock_directory.mkdir(parents=True, exist_ok=True, mode=0o700)
        lock = FileLock(str(self._lock_directory / lock_name))
        lock.acquire(timeout=_GENERATION_LOCK_TIMEOUT_SECONDS)
    except FileLockTimeout as exc:
        raise WikiStoreError("Wiki generation lock wait timed out") from exc
    except OSError as exc:
        raise WikiStoreError("Wiki generation lock failed") from exc
    try:
        yield
    finally:
        try:
            lock.release()
        except OSError as exc:
            raise WikiStoreError("Wiki generation lock release failed") from exc

WikiStore

Bases: Protocol

Minimal persistence boundary consumed by AgentWiki.

Methods:

Name	Description
read	Read one entry, returning None when it is absent.
publish	Atomically insert or replace one complete entry.
scan	Return entries in stable ID order, optionally filtered by repository.
generation_guard	Serialize generation of the same entry across processes.

read

read(entry_id: str) -> WikiStoredEntry | None

Read one entry, returning None when it is absent.

Source code in codenib/wiki/store.py

def read(self, entry_id: str) -> WikiStoredEntry | None:
    """Read one entry, returning ``None`` when it is absent."""

publish

publish(
    *,
    entry_id: str,
    repository_id: str,
    envelope: Mapping[str, Any],
    if_absent: bool = False
) -> WikiStoredEntry

Atomically insert or replace one complete entry.

Source code in codenib/wiki/store.py

def publish(
    self,
    *,
    entry_id: str,
    repository_id: str,
    envelope: Mapping[str, Any],
    if_absent: bool = False,
) -> WikiStoredEntry:
    """Atomically insert or replace one complete entry."""

scan

scan(*, repository_ids: Collection[str] | None = None) -> tuple[WikiStoredEntry, ...]

Return entries in stable ID order, optionally filtered by repository.

Source code in codenib/wiki/store.py

def scan(
    self,
    *,
    repository_ids: Collection[str] | None = None,
) -> tuple[WikiStoredEntry, ...]:
    """Return entries in stable ID order, optionally filtered by repository."""

generation_guard

generation_guard(entry_id: str) -> ContextManager[None]

Serialize generation of the same entry across processes.

Source code in codenib/wiki/store.py

def generation_guard(self, entry_id: str) -> ContextManager[None]:
    """Serialize generation of the same entry across processes."""

WikiStoreCorruptionError

Bases: WikiStoreError

Persisted Wiki data failed an integrity or decoding check.

WikiStoredEntry dataclass

WikiStoredEntry(entry_id: str, repository_id: str, envelope: Mapping[str, Any])

One complete Wiki cache envelope and its stable identity.

WikiStoreError

Bases: RuntimeError

Base error for a Wiki store operation.

WikiStoreSchemaError

Bases: WikiStoreError

The database does not implement the supported Wiki schema.

WikiStoreValidationError

Bases: WikiStoreError, ValueError

The caller supplied an invalid Wiki entry or query.