Python application programming interface reference

An application programming interface (API) defines how software components communicate. Import this public API from library_of_context or context_cache.

Mkdocstrings generates the reference sections from the source code.

Model vocabulary

• A ContextEvent is an ordered source item in one governed thread. It can contain a message, instruction, or tool result.
• A ContextRecord is a searchable unit in the records table. It contains text, an embedding, metadata, origin data, and a visibility scope.
• A book is the public name and serialized view of one ContextRecord. The Library has no separate Book class or table.
• An event reserves the identifier for its derived record. A direct record write cannot replace that searchable copy.
• A ThreadKey(collection, session_id) identifies one stateful chat. The collection identifies the project, and the session identifier identifies the thread.
• A WorkingSet is the size-limited reading-desk snapshot. The Library assembles it from records that the caller can access.

Agent-facing transport views

Model Context Protocol (MCP) tools use size-limited transport views for searches and reading desks. Hypertext Transfer Protocol (HTTP) routes use the same views.

A search result contains scores, a small record reference, and a short excerpt. Treat the excerpt as untrusted data.

A desk contains the size-limited context block and small search-result references. The context block is ready for a model prompt.

These transport views omit embeddings, complete metadata, and complete record text. An embedding is a numeric representation that supports similarity searches.

Commit and protect operations return size-limited event acknowledgements. They do not return the submitted content or metadata.

Direct record administration routes return complete records.

A daemon is a background process that owns shared Library resources. The daemon versions its MCP message format separately from its SQLite database format.

A bridge rejects a daemon that uses a different message version.

Storage path

The Library requires a SQLite database file. This file supports recovery and exclusive ownership by one Library runtime.

The Library does not support the SQLite :memory: value.

Text-agent adapter

context_cache.agent.GovernedTextAgent

Wrap a stateless text-model callback with bounded prompt assembly and persistence.

The supplied callback must send exactly the messages it receives. It must not append another transcript or continue a provider-managed conversation. Structured tool calls, streaming deltas, attachments, and multimodal values need a custom adapter rather than this text-only interface.

Source code in context_cache/agent.py

class GovernedTextAgent:
    """Wrap a stateless text-model callback with bounded prompt assembly and persistence.

    The supplied callback must send exactly the ``messages`` it receives. It must not
    append another transcript or continue a provider-managed conversation. Structured
    tool calls, streaming deltas, attachments, and multimodal values need a custom
    adapter rather than this text-only interface.
    """

    def __init__(
        self,
        context: LibraryContextGovernor,
        call_model: Callable[[list[dict[str, str]]], str],
        *,
        system_prompt: str = "",
    ) -> None:
        self.context = context
        self.call_model = call_model
        self.system_prompt = system_prompt
        self.last_prompt: GovernedPrompt | None = None

    def turn(
        self,
        user_message: str,
        *,
        turn_id: str | None = None,
        focus: str | None = None,
        protected: bool = False,
        strict_freshness: bool = False,
    ) -> str:
        """Run one governed text turn and durably record the returned response.

        Pass a stable, unique ``turn_id`` when the caller may retry. The governor then
        uses deterministic user and assistant event IDs for idempotent persistence.
        """

        if turn_id is not None and not turn_id.strip():
            raise ValueError("turn_id cannot be empty")
        user_event_id = None if turn_id is None else f"{turn_id}:user"
        assistant_event_id = None if turn_id is None else f"{turn_id}:assistant"
        prompt = self.context.prepare(
            user_message,
            focus=focus,
            system_prompt=self.system_prompt,
            protected=protected,
            event_id=user_event_id,
            strict_freshness=strict_freshness,
        )
        self.last_prompt = prompt
        response = self.call_model(prompt.messages)
        if not isinstance(response, str):
            raise TypeError("call_model must return response text as str")
        if not response.strip():
            raise ValueError("call_model returned an empty response")
        self.context.commit(response, event_id=assistant_event_id)
        return response

turn

turn(
    user_message: str,
    *,
    turn_id: str | None = None,
    focus: str | None = None,
    protected: bool = False,
    strict_freshness: bool = False,
) -> str

Run one governed text turn and durably record the returned response.

Pass a stable, unique turn_id when the caller may retry. The governor then uses deterministic user and assistant event IDs for idempotent persistence.

Source code in context_cache/agent.py

def turn(
    self,
    user_message: str,
    *,
    turn_id: str | None = None,
    focus: str | None = None,
    protected: bool = False,
    strict_freshness: bool = False,
) -> str:
    """Run one governed text turn and durably record the returned response.

    Pass a stable, unique ``turn_id`` when the caller may retry. The governor then
    uses deterministic user and assistant event IDs for idempotent persistence.
    """

    if turn_id is not None and not turn_id.strip():
        raise ValueError("turn_id cannot be empty")
    user_event_id = None if turn_id is None else f"{turn_id}:user"
    assistant_event_id = None if turn_id is None else f"{turn_id}:assistant"
    prompt = self.context.prepare(
        user_message,
        focus=focus,
        system_prompt=self.system_prompt,
        protected=protected,
        event_id=user_event_id,
        strict_freshness=strict_freshness,
    )
    self.last_prompt = prompt
    response = self.call_model(prompt.messages)
    if not isinstance(response, str):
        raise TypeError("call_model must return response text as str")
    if not response.strip():
        raise ValueError("call_model returned an empty response")
    self.context.commit(response, event_id=assistant_event_id)
    return response

Context governor

context_cache.governor.LibraryContextGovernor

Own the durable record -> bounded prompt -> durable response lifecycle.

The model receives a bounded prompt assembled for every call. Full thread events stay in SQLite, recent events remain immediately visible through an in-memory overlay, and a bounded work ring indexes durable outbox events asynchronously.

Source code in context_cache/governor.py

class LibraryContextGovernor:
    """Own the durable record -> bounded prompt -> durable response lifecycle.

    The model receives a bounded prompt assembled for every call. Full thread events stay
    in SQLite, recent events remain immediately visible through an in-memory overlay,
    and a bounded work ring indexes durable outbox events asynchronously.
    """

    def __init__(
        self,
        library: "ContextCache",
        session_id: str,
        *,
        collection: str | None = None,
        token_budget: int = 12000,
        recent_token_budget: int = 4000,
        protected_token_budget: int = 2000,
        max_books: int = 12,
        start_worker: bool = True,
    ) -> None:
        if not session_id.strip():
            raise ValueError("session_id cannot be empty")
        if token_budget < 256:
            raise ValueError("token_budget must be at least 256")
        if token_budget > MAX_CONTEXT_TOKENS:
            raise ValueError(f"token_budget cannot exceed {MAX_CONTEXT_TOKENS}")
        if recent_token_budget < 64 or recent_token_budget >= token_budget:
            raise ValueError("recent_token_budget must be >= 64 and below token_budget")
        if protected_token_budget < 0 or protected_token_budget >= token_budget:
            raise ValueError("protected_token_budget must be below token_budget")
        if not 1 <= max_books <= MAX_RESULT_BOOKS:
            raise ValueError(f"max_books must be between 1 and {MAX_RESULT_BOOKS}")
        self.library = library
        self.session_id = session_id
        self.collection = library.namespace if collection is None else collection
        self.token_budget = token_budget
        self.recent_token_budget = recent_token_budget
        self.protected_token_budget = protected_token_budget
        self.max_books = max_books
        self._key = ThreadKey(self.collection, self.session_id)
        self.desk = library.runtime.swapper
        self._indexer = library.runtime.indexer
        self._last_prompt_tokens = 0
        self._last_prompt_at: float | None = None
        if start_worker:
            library.runtime.start_indexer()

    def _prompt_builder(self, state: ThreadState) -> GovernedPromptBuilder:
        return GovernedPromptBuilder(
            self.library,
            self.desk,
            state.recent,
            session_id=self.session_id,
            collection=self.collection,
            budget=PromptBudget(
                total_tokens=self.token_budget,
                recent_tokens=self.recent_token_budget,
                protected_tokens=self.protected_token_budget,
                max_books=self.max_books,
            ),
        )

    def _record_locked(
        self,
        state: ThreadState,
        role: Role,
        content: str,
        *,
        metadata: dict[str, Any] | None,
        importance: float | None,
        protected: bool | None,
        event_id: str | None,
    ) -> ContextEvent:
        if role not in _ROLES:
            raise ValueError(f"unsupported role: {role}")
        if not content.strip():
            raise ValueError("content cannot be empty")
        if importance is None:
            importance = 0.85 if role in {"system", "developer"} else 0.5
        if not 0.0 <= importance <= 1.0:
            raise ValueError("importance must be between 0 and 1")
        if protected is None:
            protected = role in {"system", "developer"}
        resolved_event_id = validate_identifier(
            "event_id",
            event_id or uuid.uuid4().hex,
        )
        record_id = hashlib.sha256(
            f"{self.collection}\x00{self.session_id}\x00{resolved_event_id}".encode(
                "utf-8"
            )
        ).hexdigest()[:24]
        event = self.library.store.append_thread_event(
            namespace=self.collection,
            session_id=self.session_id,
            event_id=resolved_event_id,
            role=role,
            content=content,
            metadata=dict(metadata or {}),
            importance=importance,
            protected=protected,
            token_count=estimate_tokens(content),
            record_id=record_id,
            created_at=time.time(),
        )
        state.recent.append(event)
        self._indexer.enqueue(event.namespace, event.event_id)
        return event

    def record(
        self,
        role: Role,
        content: str,
        *,
        metadata: dict[str, Any] | None = None,
        importance: float | None = None,
        protected: bool | None = None,
        event_id: str | None = None,
    ) -> ContextEvent:
        """Durably append one thread event and queue it for asynchronous indexing.

        System and developer roles are protected by default. A caller-supplied
        ``event_id`` makes an identical retry idempotent; reusing it with different
        event fields fails.
        """

        with self.library._operation():
            with self.library.runtime.thread_states.lease(self._key) as state:
                with state.operation_lock:
                    return self._record_locked(
                        state,
                        role,
                        content,
                        metadata=metadata,
                        importance=importance,
                        protected=protected,
                        event_id=event_id,
                    )

    def commit(
        self,
        content: str,
        *,
        role: Role = "assistant",
        metadata: dict[str, Any] | None = None,
        importance: float | None = None,
        protected: bool | None = None,
        event_id: str | None = None,
    ) -> ContextEvent:
        """Record the model or tool result after a governed model call."""

        return self.record(
            role,
            content,
            metadata=metadata,
            importance=importance,
            protected=protected,
            event_id=event_id,
        )

    def protect(
        self,
        content: str,
        *,
        role: Role = "developer",
        label: str | None = None,
        importance: float = 1.0,
        event_id: str | None = None,
    ) -> ContextEvent:
        """Append critical context that remains eligible until explicitly released."""

        metadata = {} if label is None else {"label": label}
        return self.record(
            role,
            content,
            metadata=metadata,
            importance=importance,
            protected=True,
            event_id=event_id,
        )

    def release(self, event_id: str) -> bool:
        """Remove protection without deleting the durable event or indexed record."""

        event_id = validate_identifier("event_id", event_id)
        with self.library._operation():
            with self.library.runtime.thread_states.lease(self._key) as state:
                with state.operation_lock:
                    released = self.library.set_thread_event_protected(
                        self.collection,
                        self.session_id,
                        event_id,
                        False,
                    )
                    if not released:
                        return False
                    state.recent.set_protected(event_id, False)
                    return True

    def _build_locked(
        self,
        state: ThreadState,
        *,
        focus: str | None,
        system_prompt: str,
        strict_freshness: bool,
    ) -> GovernedPrompt:
        if strict_freshness and not self.flush(timeout=10.0):
            raise TimeoutError("context index did not reach the recorded watermark")
        envelope = self._prompt_builder(state).build(
            focus=focus,
            system_prompt=system_prompt,
        )
        self._last_prompt_tokens = envelope.token_count
        self._last_prompt_at = time.time()
        return envelope

    def build_prompt(
        self,
        *,
        focus: str | None = None,
        system_prompt: str = "",
        strict_freshness: bool = False,
    ) -> GovernedPrompt:
        """Build a bounded envelope from protected, recent, and retrieved context.

        Set ``strict_freshness`` only when the caller must wait for asynchronous index
        visibility. Normal interactive calls use the recent overlay instead.
        """

        with self.library._operation():
            with self.library.runtime.thread_states.lease(self._key) as state:
                with state.operation_lock:
                    return self._build_locked(
                        state,
                        focus=focus,
                        system_prompt=system_prompt,
                        strict_freshness=strict_freshness,
                    )

    def prepare(
        self,
        user_message: str,
        *,
        focus: str | None = None,
        system_prompt: str = "",
        metadata: dict[str, Any] | None = None,
        importance: float = 0.5,
        protected: bool = False,
        event_id: str | None = None,
        strict_freshness: bool = False,
    ) -> GovernedPrompt:
        """Durably record a user turn, then construct its bounded model request."""

        with self.library._operation():
            with self.library.runtime.thread_states.lease(self._key) as state:
                with state.operation_lock:
                    self._record_locked(
                        state,
                        "user",
                        user_message,
                        metadata=metadata,
                        importance=importance,
                        protected=protected,
                        event_id=event_id,
                    )
                    return self._build_locked(
                        state,
                        focus=focus or user_message,
                        system_prompt=system_prompt,
                        strict_freshness=strict_freshness,
                    )

    def flush(self, *, timeout: float = 10.0) -> bool:
        """Wait until the thread's outbox has reached its indexed watermark."""

        with self.library._operation():
            return self._indexer.flush(
                self.collection,
                self.session_id,
                timeout=timeout,
            )

    def retry_failed(self, event_id: str) -> bool:
        """Return a quarantined event to the indexing queue."""

        event_id = validate_identifier("event_id", event_id)
        with self.library._operation():
            with self.library.runtime.thread_states.lease(self._key) as state:
                with state.operation_lock:
                    retried = self.library.store.retry_failed_outbox_event(
                        self.collection,
                        self.session_id,
                        event_id,
                    )
                    if retried:
                        self._indexer.enqueue(self.collection, event_id)
                    return retried

    def status(self) -> dict[str, Any]:
        """Return visibility watermarks, ring pressure, and worker health."""

        with self.library._operation(allow_closing=True):
            return self._status()

    def _status(self) -> dict[str, Any]:
        with self.library.runtime.thread_states.lease(self._key) as state:
            with state.operation_lock:
                watermarks = self.library.store.thread_watermarks(
                    self.collection,
                    self.session_id,
                )
                return {
                    "session_id": self.session_id,
                    "collection": self.collection,
                    "context_mode": "semantic-paging",
                    "replaces_compaction": True,
                    "watermarks": watermarks.to_dict(),
                    "failed_events": self.library.store.failed_outbox_events(
                        self.collection,
                        self.session_id,
                    ),
                    "recent_ring": state.recent.stats(),
                    "work_ring": self._indexer.status(),
                    "worker_alive": self._indexer.is_alive,
                    "last_error": self._indexer.last_error,
                    "last_prompt_tokens": self._last_prompt_tokens,
                    "last_prompt_at": self._last_prompt_at,
                }

    def close(self) -> None:
        """Release this governor handle without stopping shared runtime services."""

    def __enter__(self) -> "LibraryContextGovernor":
        return self

    def __exit__(self, *_: object) -> None:
        self.close()

record

record(
    role: Role,
    content: str,
    *,
    metadata: dict[str, Any] | None = None,
    importance: float | None = None,
    protected: bool | None = None,
    event_id: str | None = None,
) -> ContextEvent

Durably append one thread event and queue it for asynchronous indexing.

System and developer roles are protected by default. A caller-supplied event_id makes an identical retry idempotent; reusing it with different event fields fails.

Source code in context_cache/governor.py

def record(
    self,
    role: Role,
    content: str,
    *,
    metadata: dict[str, Any] | None = None,
    importance: float | None = None,
    protected: bool | None = None,
    event_id: str | None = None,
) -> ContextEvent:
    """Durably append one thread event and queue it for asynchronous indexing.

    System and developer roles are protected by default. A caller-supplied
    ``event_id`` makes an identical retry idempotent; reusing it with different
    event fields fails.
    """

    with self.library._operation():
        with self.library.runtime.thread_states.lease(self._key) as state:
            with state.operation_lock:
                return self._record_locked(
                    state,
                    role,
                    content,
                    metadata=metadata,
                    importance=importance,
                    protected=protected,
                    event_id=event_id,
                )

prepare

prepare(
    user_message: str,
    *,
    focus: str | None = None,
    system_prompt: str = "",
    metadata: dict[str, Any] | None = None,
    importance: float = 0.5,
    protected: bool = False,
    event_id: str | None = None,
    strict_freshness: bool = False,
) -> GovernedPrompt

Durably record a user turn, then construct its bounded model request.

Source code in context_cache/governor.py

def prepare(
    self,
    user_message: str,
    *,
    focus: str | None = None,
    system_prompt: str = "",
    metadata: dict[str, Any] | None = None,
    importance: float = 0.5,
    protected: bool = False,
    event_id: str | None = None,
    strict_freshness: bool = False,
) -> GovernedPrompt:
    """Durably record a user turn, then construct its bounded model request."""

    with self.library._operation():
        with self.library.runtime.thread_states.lease(self._key) as state:
            with state.operation_lock:
                self._record_locked(
                    state,
                    "user",
                    user_message,
                    metadata=metadata,
                    importance=importance,
                    protected=protected,
                    event_id=event_id,
                )
                return self._build_locked(
                    state,
                    focus=focus or user_message,
                    system_prompt=system_prompt,
                    strict_freshness=strict_freshness,
                )

commit

commit(
    content: str,
    *,
    role: Role = "assistant",
    metadata: dict[str, Any] | None = None,
    importance: float | None = None,
    protected: bool | None = None,
    event_id: str | None = None,
) -> ContextEvent

Record the model or tool result after a governed model call.

Source code in context_cache/governor.py

def commit(
    self,
    content: str,
    *,
    role: Role = "assistant",
    metadata: dict[str, Any] | None = None,
    importance: float | None = None,
    protected: bool | None = None,
    event_id: str | None = None,
) -> ContextEvent:
    """Record the model or tool result after a governed model call."""

    return self.record(
        role,
        content,
        metadata=metadata,
        importance=importance,
        protected=protected,
        event_id=event_id,
    )

protect

protect(
    content: str,
    *,
    role: Role = "developer",
    label: str | None = None,
    importance: float = 1.0,
    event_id: str | None = None,
) -> ContextEvent

Append critical context that remains eligible until explicitly released.

Source code in context_cache/governor.py

def protect(
    self,
    content: str,
    *,
    role: Role = "developer",
    label: str | None = None,
    importance: float = 1.0,
    event_id: str | None = None,
) -> ContextEvent:
    """Append critical context that remains eligible until explicitly released."""

    metadata = {} if label is None else {"label": label}
    return self.record(
        role,
        content,
        metadata=metadata,
        importance=importance,
        protected=True,
        event_id=event_id,
    )

release

release(event_id: str) -> bool

Remove protection without deleting the durable event or indexed record.

Source code in context_cache/governor.py

def release(self, event_id: str) -> bool:
    """Remove protection without deleting the durable event or indexed record."""

    event_id = validate_identifier("event_id", event_id)
    with self.library._operation():
        with self.library.runtime.thread_states.lease(self._key) as state:
            with state.operation_lock:
                released = self.library.set_thread_event_protected(
                    self.collection,
                    self.session_id,
                    event_id,
                    False,
                )
                if not released:
                    return False
                state.recent.set_protected(event_id, False)
                return True

build_prompt

build_prompt(
    *,
    focus: str | None = None,
    system_prompt: str = "",
    strict_freshness: bool = False,
) -> GovernedPrompt

Build a bounded envelope from protected, recent, and retrieved context.

Set strict_freshness only when the caller must wait for asynchronous index visibility. Normal interactive calls use the recent overlay instead.

Source code in context_cache/governor.py

def build_prompt(
    self,
    *,
    focus: str | None = None,
    system_prompt: str = "",
    strict_freshness: bool = False,
) -> GovernedPrompt:
    """Build a bounded envelope from protected, recent, and retrieved context.

    Set ``strict_freshness`` only when the caller must wait for asynchronous index
    visibility. Normal interactive calls use the recent overlay instead.
    """

    with self.library._operation():
        with self.library.runtime.thread_states.lease(self._key) as state:
            with state.operation_lock:
                return self._build_locked(
                    state,
                    focus=focus,
                    system_prompt=system_prompt,
                    strict_freshness=strict_freshness,
                )

flush

flush(*, timeout: float = 10.0) -> bool

Wait until the thread's outbox has reached its indexed watermark.

Source code in context_cache/governor.py

def flush(self, *, timeout: float = 10.0) -> bool:
    """Wait until the thread's outbox has reached its indexed watermark."""

    with self.library._operation():
        return self._indexer.flush(
            self.collection,
            self.session_id,
            timeout=timeout,
        )

retry_failed

retry_failed(event_id: str) -> bool

Return a quarantined event to the indexing queue.

Source code in context_cache/governor.py

def retry_failed(self, event_id: str) -> bool:
    """Return a quarantined event to the indexing queue."""

    event_id = validate_identifier("event_id", event_id)
    with self.library._operation():
        with self.library.runtime.thread_states.lease(self._key) as state:
            with state.operation_lock:
                retried = self.library.store.retry_failed_outbox_event(
                    self.collection,
                    self.session_id,
                    event_id,
                )
                if retried:
                    self._indexer.enqueue(self.collection, event_id)
                return retried

status

status() -> dict[str, Any]

Return visibility watermarks, ring pressure, and worker health.

Source code in context_cache/governor.py

def status(self) -> dict[str, Any]:
    """Return visibility watermarks, ring pressure, and worker health."""

    with self.library._operation(allow_closing=True):
        return self._status()

close

close() -> None

Release this governor handle without stopping shared runtime services.

Source code in context_cache/governor.py

def close(self) -> None:
    """Release this governor handle without stopping shared runtime services."""

Library facade

context_cache.library.LibraryOfContext

Bases: ContextCache

Public API for context storage, retrieval, reading desks, and governed prompts.

Source code in context_cache/library.py

class LibraryOfContext(ContextCache):
    """Public API for context storage, retrieval, reading desks, and governed prompts."""

    def shelve(
        self,
        text: str,
        *,
        book_id: str | None = None,
        collection: str | None = None,
        catalog: dict[str, Any] | None = None,
        source: str = "manual",
        importance: float = 0.5,
        shelf_life_seconds: float | None = None,
        scope: ContextScope | str = ContextScope.PROJECT,
        owner_session_id: str | None = None,
        team_id: str | None = None,
    ) -> ContextRecord:
        return self.put(
            text,
            record_id=book_id,
            namespace=collection,
            metadata=catalog,
            source=source,
            importance=importance,
            ttl_seconds=shelf_life_seconds,
            scope=scope,
            owner_session_id=owner_session_id,
            team_id=team_id,
        )

    def shelve_document(
        self,
        text: str,
        *,
        source: str,
        collection: str | None = None,
        catalog: dict[str, Any] | None = None,
        importance: float = 0.5,
        chapter_tokens: int = 450,
        overlap_tokens: int = 60,
        replace_edition: bool = False,
        scope: ContextScope | str = ContextScope.PROJECT,
        owner_session_id: str | None = None,
        team_id: str | None = None,
    ) -> list[ContextRecord]:
        return self.ingest(
            text,
            source=source,
            namespace=collection,
            metadata=catalog,
            importance=importance,
            chunk_tokens=chapter_tokens,
            overlap_tokens=overlap_tokens,
            replace_source=replace_edition,
            scope=scope,
            owner_session_id=owner_session_id,
            team_id=team_id,
        )

    def consult(
        self,
        subject: str,
        *,
        max_books: int = 8,
        collection: str | None = None,
        catalog_filters: dict[str, Any] | None = None,
        minimum_relevance: float = 0.0,
        team_ids: tuple[str, ...] = (),
    ) -> list[SearchHit]:
        return self.retrieve(
            subject,
            top_k=max_books,
            namespace=collection,
            filters=catalog_filters,
            minimum_score=minimum_relevance,
            scopes=(ContextScope.PROJECT, ContextScope.TEAM)
            if team_ids
            else (ContextScope.PROJECT,),
            team_ids=team_ids,
        )

    def promote_book(
        self,
        book_id: str,
        *,
        target_scope: ContextScope | str,
        collection: str | None = None,
        source_session_id: str | None = None,
        promoted_book_id: str | None = None,
        target_team_id: str | None = None,
    ) -> ContextRecord:
        return self.promote(
            book_id,
            target_scope=target_scope,
            namespace=collection,
            source_session_id=source_session_id,
            promoted_record_id=promoted_book_id,
            target_team_id=target_team_id,
        )

    def open_reading_desk(self) -> ReadingDesk:
        with self._operation():
            return ReadingDesk(self.runtime.swapper)

    def open_virtual_session(
        self,
        session_id: str,
        *,
        collection: str | None = None,
        token_budget: int = 12000,
        recent_token_budget: int = 4000,
    ) -> "VirtualContextSession":
        from .session import VirtualContextSession

        with self._operation():
            return VirtualContextSession(
                self,
                session_id,
                collection=collection,
                token_budget=token_budget,
                recent_token_budget=recent_token_budget,
            )

shelve

shelve(
    text: str,
    *,
    book_id: str | None = None,
    collection: str | None = None,
    catalog: dict[str, Any] | None = None,
    source: str = "manual",
    importance: float = 0.5,
    shelf_life_seconds: float | None = None,
    scope: ContextScope | str = ContextScope.PROJECT,
    owner_session_id: str | None = None,
    team_id: str | None = None,
) -> ContextRecord

Source code in context_cache/library.py

def shelve(
    self,
    text: str,
    *,
    book_id: str | None = None,
    collection: str | None = None,
    catalog: dict[str, Any] | None = None,
    source: str = "manual",
    importance: float = 0.5,
    shelf_life_seconds: float | None = None,
    scope: ContextScope | str = ContextScope.PROJECT,
    owner_session_id: str | None = None,
    team_id: str | None = None,
) -> ContextRecord:
    return self.put(
        text,
        record_id=book_id,
        namespace=collection,
        metadata=catalog,
        source=source,
        importance=importance,
        ttl_seconds=shelf_life_seconds,
        scope=scope,
        owner_session_id=owner_session_id,
        team_id=team_id,
    )

shelve_document

shelve_document(
    text: str,
    *,
    source: str,
    collection: str | None = None,
    catalog: dict[str, Any] | None = None,
    importance: float = 0.5,
    chapter_tokens: int = 450,
    overlap_tokens: int = 60,
    replace_edition: bool = False,
    scope: ContextScope | str = ContextScope.PROJECT,
    owner_session_id: str | None = None,
    team_id: str | None = None,
) -> list[ContextRecord]

Source code in context_cache/library.py

def shelve_document(
    self,
    text: str,
    *,
    source: str,
    collection: str | None = None,
    catalog: dict[str, Any] | None = None,
    importance: float = 0.5,
    chapter_tokens: int = 450,
    overlap_tokens: int = 60,
    replace_edition: bool = False,
    scope: ContextScope | str = ContextScope.PROJECT,
    owner_session_id: str | None = None,
    team_id: str | None = None,
) -> list[ContextRecord]:
    return self.ingest(
        text,
        source=source,
        namespace=collection,
        metadata=catalog,
        importance=importance,
        chunk_tokens=chapter_tokens,
        overlap_tokens=overlap_tokens,
        replace_source=replace_edition,
        scope=scope,
        owner_session_id=owner_session_id,
        team_id=team_id,
    )

consult

consult(
    subject: str,
    *,
    max_books: int = 8,
    collection: str | None = None,
    catalog_filters: dict[str, Any] | None = None,
    minimum_relevance: float = 0.0,
    team_ids: tuple[str, ...] = (),
) -> list[SearchHit]

Source code in context_cache/library.py

def consult(
    self,
    subject: str,
    *,
    max_books: int = 8,
    collection: str | None = None,
    catalog_filters: dict[str, Any] | None = None,
    minimum_relevance: float = 0.0,
    team_ids: tuple[str, ...] = (),
) -> list[SearchHit]:
    return self.retrieve(
        subject,
        top_k=max_books,
        namespace=collection,
        filters=catalog_filters,
        minimum_score=minimum_relevance,
        scopes=(ContextScope.PROJECT, ContextScope.TEAM)
        if team_ids
        else (ContextScope.PROJECT,),
        team_ids=team_ids,
    )

promote_book

promote_book(
    book_id: str,
    *,
    target_scope: ContextScope | str,
    collection: str | None = None,
    source_session_id: str | None = None,
    promoted_book_id: str | None = None,
    target_team_id: str | None = None,
) -> ContextRecord

Source code in context_cache/library.py

def promote_book(
    self,
    book_id: str,
    *,
    target_scope: ContextScope | str,
    collection: str | None = None,
    source_session_id: str | None = None,
    promoted_book_id: str | None = None,
    target_team_id: str | None = None,
) -> ContextRecord:
    return self.promote(
        book_id,
        target_scope=target_scope,
        namespace=collection,
        source_session_id=source_session_id,
        promoted_record_id=promoted_book_id,
        target_team_id=target_team_id,
    )

open_reading_desk

open_reading_desk() -> ReadingDesk

Source code in context_cache/library.py

def open_reading_desk(self) -> ReadingDesk:
    with self._operation():
        return ReadingDesk(self.runtime.swapper)

open_virtual_session

open_virtual_session(
    session_id: str,
    *,
    collection: str | None = None,
    token_budget: int = 12000,
    recent_token_budget: int = 4000,
) -> "VirtualContextSession"

Source code in context_cache/library.py

def open_virtual_session(
    self,
    session_id: str,
    *,
    collection: str | None = None,
    token_budget: int = 12000,
    recent_token_budget: int = 4000,
) -> "VirtualContextSession":
    from .session import VirtualContextSession

    with self._operation():
        return VirtualContextSession(
            self,
            session_id,
            collection=collection,
            token_budget=token_budget,
            recent_token_budget=recent_token_budget,
        )

Thread identity and visibility

context_cache.scopes.ThreadKey dataclass

Stable identity for one agent thread inside a collection.

Source code in context_cache/scopes.py

@dataclass(frozen=True, slots=True)
class ThreadKey:
    """Stable identity for one agent thread inside a collection."""

    collection: str
    session_id: str

    def __post_init__(self) -> None:
        for name, value in (
            ("collection", self.collection),
            ("session_id", self.session_id),
        ):
            validate_identifier(name, value)

context_cache.scopes.ContextScope

Bases: StrEnum

Visibility boundary assigned to an indexed context record.

Source code in context_cache/scopes.py

class ContextScope(StrEnum):
    """Visibility boundary assigned to an indexed context record."""

    THREAD = "thread"
    PROJECT = "project"
    TEAM = "team"

context_cache.scopes.ScopeSelection dataclass

Authorized record scopes for one retrieval operation.

Source code in context_cache/scopes.py

@dataclass(frozen=True, slots=True)
class ScopeSelection:
    """Authorized record scopes for one retrieval operation."""

    scopes: tuple[ContextScope, ...]
    session_id: str | None = None
    team_ids: frozenset[str] = frozenset()

    def __post_init__(self) -> None:
        if not self.scopes:
            raise ValueError("at least one context scope is required")
        if len(set(self.scopes)) != len(self.scopes):
            raise ValueError("context scopes cannot contain duplicates")
        if ContextScope.THREAD in self.scopes:
            if self.session_id is None:
                raise ValueError("thread scope requires a non-empty session_id")
            validate_identifier("session_id", self.session_id)
        if ContextScope.TEAM in self.scopes and not self.team_ids:
            raise ValueError("team scope requires at least one authorized team_id")
        for team_id in self.team_ids:
            validate_identifier("team_id", team_id)

    @classmethod
    def resolve(
        cls,
        scopes: Iterable[ContextScope | str] | None = None,
        *,
        session_id: str | None = None,
        team_ids: Iterable[str] = (),
    ) -> ScopeSelection:
        selected = (
            (ContextScope.PROJECT,)
            if scopes is None
            else tuple(ContextScope(scope) for scope in scopes)
        )
        return cls(
            selected,
            session_id=session_id,
            team_ids=frozenset(str(team_id) for team_id in team_ids),
        )

    @classmethod
    def for_thread(
        cls,
        session_id: str,
        *,
        include_project: bool = True,
        team_ids: Iterable[str] = (),
    ) -> ScopeSelection:
        scopes = [ContextScope.THREAD]
        if include_project:
            scopes.append(ContextScope.PROJECT)
        authorized_teams = frozenset(str(team_id) for team_id in team_ids)
        if authorized_teams:
            scopes.append(ContextScope.TEAM)
        return cls(
            tuple(scopes),
            session_id=session_id,
            team_ids=authorized_teams,
        )

    def cache_identity(self) -> dict[str, object]:
        return {
            "scopes": [scope.value for scope in self.scopes],
            "session_id": (
                self.session_id if ContextScope.THREAD in self.scopes else None
            ),
            "team_ids": sorted(self.team_ids),
        }

    def allows(
        self,
        scope: ContextScope | str,
        owner_session_id: str | None,
        team_id: str | None = None,
    ) -> bool:
        resolved = ContextScope(scope)
        if resolved not in self.scopes:
            return False
        if resolved is ContextScope.THREAD:
            return owner_session_id == self.session_id
        if resolved is ContextScope.TEAM:
            return team_id in self.team_ids
        return True

Shared runtime configuration

context_cache.runtime.RuntimeSettings dataclass

Bounded process-level worker and desk resources.

Source code in context_cache/runtime.py

@dataclass(frozen=True, slots=True)
class RuntimeSettings:
    """Bounded process-level worker and desk resources."""

    outbox_workers: int = 2
    outbox_capacity: int = 1024
    outbox_poll_seconds: float = 0.1
    outbox_lease_seconds: float = 120.0
    outbox_max_attempts: int = 8
    outbox_shutdown_timeout_seconds: float = 2.0
    desk_workers: int = 2
    desk_jitter_ratio: float = 0.1
    desk_shutdown_timeout_seconds: float = 2.0
    max_active_threads: int = 256
    thread_idle_ttl_seconds: float = 1800.0
    recent_ring_events: int = 256
    recent_ring_tokens: int = 8192
    max_active_desks: int = 256
    desk_idle_ttl_seconds: float = 1800.0
    http_max_connections: int = 32
    http_read_timeout_seconds: float = 15.0
    http_shutdown_timeout_seconds: float = 120.0

    def __post_init__(self) -> None:
        if self.outbox_workers < 1 or self.desk_workers < 1:
            raise ValueError("runtime worker counts must be positive")
        if self.outbox_capacity < 1:
            raise ValueError("outbox_capacity must be positive")
        if (
            self.outbox_poll_seconds <= 0
            or self.outbox_lease_seconds <= 0
            or self.outbox_shutdown_timeout_seconds <= 0
        ):
            raise ValueError("outbox timing values must be positive")
        if self.outbox_max_attempts < 1:
            raise ValueError("outbox_max_attempts must be positive")
        if self.max_active_threads < 1 or self.thread_idle_ttl_seconds <= 0:
            raise ValueError("thread registry limits must be positive")
        if self.max_active_desks < 1 or self.desk_idle_ttl_seconds <= 0:
            raise ValueError("desk registry limits must be positive")
        if self.desk_shutdown_timeout_seconds <= 0:
            raise ValueError("desk shutdown timeout must be positive")
        if (
            self.http_max_connections < 1
            or self.http_read_timeout_seconds <= 0
            or self.http_shutdown_timeout_seconds <= 0
        ):
            raise ValueError("HTTP runtime limits must be positive")

context_cache.runtime.LibraryRuntime

Own the bounded indexing and desk services for one Library process.

Source code in context_cache/runtime.py

class LibraryRuntime:
    """Own the bounded indexing and desk services for one Library process."""

    def __init__(
        self,
        cache: ContextCache,
        *,
        settings: RuntimeSettings | None = None,
    ) -> None:
        self.cache = cache
        self.settings = settings or RuntimeSettings()
        self.scheduler = DeskScheduler(
            worker_count=self.settings.desk_workers,
            jitter_ratio=self.settings.desk_jitter_ratio,
            max_tasks=self.settings.max_active_desks,
            shutdown_timeout_seconds=self.settings.desk_shutdown_timeout_seconds,
        )
        self.swapper = ContextSwapper(
            cache,
            scheduler=self.scheduler,
            max_working_sets=self.settings.max_active_desks,
            working_set_ttl_seconds=self.settings.desk_idle_ttl_seconds,
        )
        self.thread_states = ThreadStateRegistry(
            cache,
            max_entries=self.settings.max_active_threads,
            idle_ttl_seconds=self.settings.thread_idle_ttl_seconds,
            recent_events=self.settings.recent_ring_events,
            recent_tokens=self.settings.recent_ring_tokens,
        )
        self.indexer = OutboxIndexer(
            cache,
            capacity=self.settings.outbox_capacity,
            poll_seconds=self.settings.outbox_poll_seconds,
            worker_count=self.settings.outbox_workers,
            lease_seconds=self.settings.outbox_lease_seconds,
            max_attempts=self.settings.outbox_max_attempts,
            shutdown_timeout_seconds=self.settings.outbox_shutdown_timeout_seconds,
        )
        self._closed = False
        self._shutdown_complete = False
        self._close_errors: dict[str, str] = {}
        self._lock = threading.RLock()

    def start_indexer(self) -> None:
        with self._lock:
            if self._closed:
                raise RuntimeError("Library runtime is closed")
            self.indexer.start()

    def status(self, *, collection: str | None = None) -> dict[str, Any]:
        periodic_desks = [
            desk
            for desk in self.scheduler.status()
            if collection is None or desk["collection"] == collection
        ]
        return {
            "indexer": self.indexer.status(),
            "desk_scheduler": self.scheduler.health(),
            "periodic_desks": periodic_desks,
            "thread_states": self.thread_states.stats(collection=collection),
            "close_errors": dict(self._close_errors),
            "settings": {
                "outbox_workers": self.settings.outbox_workers,
                "outbox_capacity": self.settings.outbox_capacity,
                "outbox_max_attempts": self.settings.outbox_max_attempts,
                "outbox_shutdown_timeout_seconds": (
                    self.settings.outbox_shutdown_timeout_seconds
                ),
                "desk_workers": self.settings.desk_workers,
                "desk_shutdown_timeout_seconds": (
                    self.settings.desk_shutdown_timeout_seconds
                ),
                "max_active_threads": self.settings.max_active_threads,
                "max_active_desks": self.settings.max_active_desks,
                "http_max_connections": self.settings.http_max_connections,
                "http_read_timeout_seconds": self.settings.http_read_timeout_seconds,
                "http_shutdown_timeout_seconds": (
                    self.settings.http_shutdown_timeout_seconds
                ),
            },
        }

    def close(self) -> bool:
        with self._lock:
            if self._shutdown_complete:
                return True
            self._closed = True
            self._close_errors.clear()

        def attempt(name: str, callback: Any, *, expects_bool: bool = False) -> bool:
            try:
                result = callback()
            except Exception as exc:
                with self._lock:
                    self._close_errors[name] = f"{type(exc).__name__}: {exc}"
                return False
            if expects_bool and not bool(result):
                return False
            return True

        indexer_drained = attempt(
            "indexer",
            self.indexer.close,
            expects_bool=True,
        )
        swapper_closed = attempt("swapper", self.swapper.close)
        scheduler_drained = attempt(
            "scheduler",
            self.scheduler.close,
            expects_bool=True,
        )
        thread_states_closed = attempt("thread_states", self.thread_states.close)
        drained = all(
            (
                indexer_drained,
                swapper_closed,
                scheduler_drained,
                thread_states_closed,
            )
        )
        if drained:
            with self._lock:
                self._shutdown_complete = True
        return drained

Loopback daemon client

Every daemon route requires Authorization: Bearer <token>. A bearer token is a secret value that grants access to its holder.

An HTTP POST request sends a body to a route. An HTTP GET request reads data from a route.

The serve command reads or creates the owner-readable file from --auth-token-file. The default path is <database-path>.daemon-token.

A thin MCP bridge reads the same file through --daemon-token-file.

LibraryDaemonClient requires the token in its bearer_token argument. It accepts only loopback HTTP addresses and sends the token with each request.

A loopback address sends traffic only inside the local computer. The daemon rejects browser origins, cross-site requests, and non-loopback Host values.

The daemon also rejects POST bodies unless their media type is application/json. JavaScript Object Notation (JSON) is the daemon message format.

context_cache.client.LibraryDaemonClient

Call one Library daemon over its dependency-free loopback HTTP protocol.

Source code in context_cache/client.py

class LibraryDaemonClient:
    """Call one Library daemon over its dependency-free loopback HTTP protocol."""

    def __init__(
        self,
        base_url: str,
        *,
        bearer_token: str,
        timeout: float = DEFAULT_DAEMON_TIMEOUT_SECONDS,
    ) -> None:
        parsed = urllib.parse.urlsplit(base_url)
        if parsed.scheme != "http":
            raise ValueError("daemon URL must use http://")
        if not is_loopback_host(parsed.hostname):
            raise ValueError("daemon URL must identify a loopback host")
        if parsed.username is not None or parsed.password is not None:
            raise ValueError("daemon URL must not contain credentials")
        if parsed.path not in {"", "/"} or parsed.query or parsed.fragment:
            raise ValueError("daemon URL must identify the service root")
        if timeout <= 0:
            raise ValueError("daemon timeout must be positive")
        self.base_url = urllib.parse.urlunsplit(
            (parsed.scheme, parsed.netloc, "", "", "")
        ).rstrip("/")
        self.bearer_token = validate_daemon_token(bearer_token)
        self.timeout = timeout
        self._opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))

    @staticmethod
    def _error_message(value: Any) -> str:
        if isinstance(value, dict):
            error = value.get("error")
            if isinstance(error, dict):
                return str(error.get("message", error))
            if error is not None:
                return str(error)
        return str(value)

    def _request(
        self,
        method: str,
        path: str,
        body: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        payload = None
        headers = {
            "Accept": "application/json",
            "Authorization": f"Bearer {self.bearer_token}",
        }
        if body is not None:
            payload = json.dumps(body, separators=(",", ":")).encode("utf-8")
            headers["Content-Type"] = "application/json"
        request = urllib.request.Request(
            self.base_url + path,
            data=payload,
            headers=headers,
            method=method,
        )
        try:
            with self._opener.open(request, timeout=self.timeout) as response:
                status = response.status
                raw = response.read()
        except urllib.error.HTTPError as exc:
            raw = exc.read()
            try:
                detail = json.loads(raw.decode("utf-8"))
            except (UnicodeDecodeError, json.JSONDecodeError):
                detail = raw.decode("utf-8", errors="replace")
            raise DaemonRequestError(
                exc.code,
                self._error_message(detail),
            ) from exc
        except (OSError, TimeoutError, urllib.error.URLError) as exc:
            raise DaemonConnectionError(
                f"cannot reach Library daemon at {self.base_url}: {exc}"
            ) from exc
        try:
            value = json.loads(raw.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise DaemonRequestError(status, "response is not valid JSON") from exc
        if not isinstance(value, dict):
            raise DaemonRequestError(status, "response must be a JSON object")
        return value

    def health(self) -> dict[str, Any]:
        health = self._request("GET", "/health")
        protocol = health.get("protocol")
        if not isinstance(protocol, dict):
            raise DaemonProtocolError("daemon health omits the protocol contract")
        received_name = protocol.get("name")
        received_version = protocol.get("version")
        if (
            received_name != DAEMON_PROTOCOL_NAME
            or received_version != DAEMON_PROTOCOL_VERSION
        ):
            raise DaemonProtocolError(
                "daemon protocol mismatch: expected "
                f"{DAEMON_PROTOCOL_NAME}/{DAEMON_PROTOCOL_VERSION}, received "
                f"{received_name}/{received_version}"
            )
        schema = health.get("schema")
        received_schema = schema.get("mcp_tools") if isinstance(schema, dict) else None
        if received_schema != MCP_SCHEMA_VERSION:
            raise DaemonProtocolError(
                "daemon MCP schema mismatch: expected "
                f"{MCP_SCHEMA_VERSION}, received {received_schema}"
            )
        runtime = health.get("runtime")
        if not isinstance(runtime, dict) or not isinstance(runtime.get("id"), str):
            raise DaemonProtocolError("daemon health omits the runtime identity")
        return health

    def call_mcp_tool(
        self,
        name: str,
        arguments: dict[str, Any],
    ) -> dict[str, Any]:
        if not name:
            raise ValueError("MCP tool name must not be empty")
        if not isinstance(arguments, dict):
            raise TypeError("MCP tool arguments must be a JSON object")
        return self._request(
            "POST",
            "/mcp/call",
            {
                "protocol_version": DAEMON_PROTOCOL_VERSION,
                "schema_version": MCP_SCHEMA_VERSION,
                "name": name,
                "arguments": arguments,
            },
        )

    def call_tool(
        self,
        name: str,
        arguments: dict[str, Any],
    ) -> dict[str, Any]:
        return self.call_mcp_tool(name, arguments)

Reading desk

context_cache.library.ReadingDesk

Token-bounded active context whose contents are replaced during refresh.

Source code in context_cache/library.py

class ReadingDesk:
    """Token-bounded active context whose contents are replaced during refresh."""

    def __init__(self, swapper: ContextSwapper) -> None:
        self._swapper = swapper

    def refresh(
        self,
        session_id: str,
        focus: str,
        *,
        token_budget: int = 4000,
        top_k: int = 12,
        namespace: str | None = None,
        filters: dict[str, Any] | None = None,
        pinned_record_ids: list[str] | None = None,
        exclude_record_ids: list[str] | None = None,
        team_ids: Iterable[str] = (),
    ) -> WorkingSet:
        return self._swapper.refresh(
            session_id,
            focus,
            token_budget=token_budget,
            top_k=top_k,
            namespace=namespace,
            filters=filters,
            pinned_record_ids=pinned_record_ids,
            exclude_record_ids=exclude_record_ids,
            team_ids=team_ids,
        )

    def get(
        self,
        session_id: str,
        *,
        namespace: str | None = None,
    ) -> WorkingSet | None:
        return self._swapper.get(session_id, namespace=namespace)

    def start_periodic(
        self,
        session_id: str,
        focus: str,
        *,
        interval_seconds: float = 30.0,
        token_budget: int = 4000,
        top_k: int = 12,
        namespace: str | None = None,
        filters: dict[str, Any] | None = None,
        pinned_record_ids: list[str] | None = None,
        exclude_record_ids: list[str] | None = None,
        team_ids: Iterable[str] = (),
    ) -> WorkingSet:
        return self._swapper.start_periodic(
            session_id,
            focus,
            interval_seconds=interval_seconds,
            token_budget=token_budget,
            top_k=top_k,
            namespace=namespace,
            filters=filters,
            pinned_record_ids=pinned_record_ids,
            exclude_record_ids=exclude_record_ids,
            team_ids=team_ids,
        )

    def update_focus(
        self,
        session_id: str,
        focus: str,
        *,
        namespace: str | None = None,
    ) -> WorkingSet:
        return self._swapper.update_focus(session_id, focus, namespace=namespace)

    def stop_periodic(
        self,
        session_id: str,
        *,
        namespace: str | None = None,
    ) -> bool:
        return self._swapper.stop_periodic(session_id, namespace=namespace)

    def status(self, *, namespace: str | None = None) -> list[dict[str, Any]]:
        return self._swapper.status(namespace=namespace)

    def close(self) -> None:
        """Release this desk handle without stopping the shared scheduler."""

    def lay_out(
        self,
        subject: str,
        *,
        session_id: str,
        token_budget: int = 4000,
        max_books: int = 12,
        namespace: str | None = None,
        catalog_filters: dict[str, Any] | None = None,
        keep_open: list[str] | None = None,
        leave_shelved: list[str] | None = None,
        team_ids: tuple[str, ...] = (),
    ) -> WorkingSet:
        return self._swapper.refresh(
            session_id,
            subject,
            token_budget=token_budget,
            top_k=max_books,
            namespace=namespace,
            filters=catalog_filters,
            pinned_record_ids=keep_open,
            exclude_record_ids=leave_shelved,
            team_ids=team_ids,
        )

    def change_subject(
        self,
        subject: str,
        *,
        session_id: str,
        namespace: str | None = None,
    ) -> WorkingSet:
        return self._swapper.update_focus(session_id, subject, namespace=namespace)

    def current_books(
        self, *, session_id: str, namespace: str | None = None
    ) -> WorkingSet | None:
        return self._swapper.get(session_id, namespace=namespace)

lay_out

lay_out(
    subject: str,
    *,
    session_id: str,
    token_budget: int = 4000,
    max_books: int = 12,
    namespace: str | None = None,
    catalog_filters: dict[str, Any] | None = None,
    keep_open: list[str] | None = None,
    leave_shelved: list[str] | None = None,
    team_ids: tuple[str, ...] = (),
) -> WorkingSet

Source code in context_cache/library.py

def lay_out(
    self,
    subject: str,
    *,
    session_id: str,
    token_budget: int = 4000,
    max_books: int = 12,
    namespace: str | None = None,
    catalog_filters: dict[str, Any] | None = None,
    keep_open: list[str] | None = None,
    leave_shelved: list[str] | None = None,
    team_ids: tuple[str, ...] = (),
) -> WorkingSet:
    return self._swapper.refresh(
        session_id,
        subject,
        token_budget=token_budget,
        top_k=max_books,
        namespace=namespace,
        filters=catalog_filters,
        pinned_record_ids=keep_open,
        exclude_record_ids=leave_shelved,
        team_ids=team_ids,
    )

change_subject

change_subject(
    subject: str,
    *,
    session_id: str,
    namespace: str | None = None,
) -> WorkingSet

Source code in context_cache/library.py

def change_subject(
    self,
    subject: str,
    *,
    session_id: str,
    namespace: str | None = None,
) -> WorkingSet:
    return self._swapper.update_focus(session_id, subject, namespace=namespace)

current_books

current_books(
    *, session_id: str, namespace: str | None = None
) -> WorkingSet | None

Source code in context_cache/library.py

def current_books(
    self, *, session_id: str, namespace: str | None = None
) -> WorkingSet | None:
    return self._swapper.get(session_id, namespace=namespace)

Core models

context_cache.models.ContextEvent dataclass

One durable event in a governed agent thread.

Source code in context_cache/models.py

@dataclass(frozen=True, slots=True)
class ContextEvent:
    """One durable event in a governed agent thread."""

    event_id: str
    namespace: str
    session_id: str
    sequence: int
    role: str
    content: str
    metadata: dict[str, Any] = field(default_factory=dict)
    importance: float = 0.5
    protected: bool = False
    token_count: int = 0
    record_id: str = ""
    created_at: float = 0.0
    indexed_at: float | None = None

    def to_dict(self) -> dict[str, Any]:
        return asdict(self)

    @classmethod
    def from_dict(cls, value: dict[str, Any]) -> "ContextEvent":
        return cls(
            event_id=str(value["event_id"]),
            namespace=str(value["namespace"]),
            session_id=str(value["session_id"]),
            sequence=int(value["sequence"]),
            role=str(value["role"]),
            content=str(value["content"]),
            metadata=dict(value.get("metadata", {})),
            importance=float(value.get("importance", 0.5)),
            protected=bool(value.get("protected", False)),
            token_count=int(value.get("token_count", 0)),
            record_id=str(value.get("record_id", "")),
            created_at=float(value.get("created_at", 0.0)),
            indexed_at=(
                None if value.get("indexed_at") is None else float(value["indexed_at"])
            ),
        )

context_cache.models.ContextWatermarks dataclass

Durable visibility boundaries for a governed thread.

Source code in context_cache/models.py

@dataclass(slots=True)
class ContextWatermarks:
    """Durable visibility boundaries for a governed thread."""

    recorded_through: int = 0
    embedded_through: int = 0
    indexed_through: int = 0
    team_synced_through: int = 0
    pending_events: int = 0
    failed_events: int = 0

    def to_dict(self) -> dict[str, int]:
        return asdict(self)

context_cache.models.GovernedPrompt dataclass

A complete bounded model request assembled by the context governor.

Source code in context_cache/models.py

@dataclass(slots=True)
class GovernedPrompt:
    """A complete bounded model request assembled by the context governor."""

    session_id: str
    collection: str
    messages: list[dict[str, str]]
    token_count: int
    token_budget: int
    event_count: int
    recent_event_ids: list[str]
    protected_event_ids: list[str]
    desk: WorkingSet
    watermarks: ContextWatermarks
    paged_out_events: int = 0
    native_context_pressure: float = 0.0
    context_mode: str = "semantic-paging"
    replaces_compaction: bool = True

    def to_dict(self) -> dict[str, Any]:
        return {
            "session_id": self.session_id,
            "collection": self.collection,
            "messages": self.messages,
            "token_count": self.token_count,
            "token_budget": self.token_budget,
            "event_count": self.event_count,
            "recent_event_ids": self.recent_event_ids,
            "protected_event_ids": self.protected_event_ids,
            "desk": self.desk.to_dict(),
            "watermarks": self.watermarks.to_dict(),
            "paged_out_events": self.paged_out_events,
            "native_context_pressure": self.native_context_pressure,
            "context_mode": self.context_mode,
            "replaces_compaction": self.replaces_compaction,
        }

context_cache.models.ContextRecord dataclass

Source code in context_cache/models.py

@dataclass(frozen=True, slots=True)
class ContextRecord:
    id: str
    namespace: str
    text: str
    embedding: list[float]
    metadata: dict[str, Any] = field(default_factory=dict)
    source: str = "manual"
    importance: float = 0.5
    token_count: int = 0
    created_at: float = 0.0
    updated_at: float = 0.0
    accessed_at: float = 0.0
    expires_at: float | None = None
    content_hash: str = ""
    scope: ContextScope = ContextScope.PROJECT
    owner_session_id: str | None = None
    team_id: str | None = None

    def __post_init__(self) -> None:
        scope, owner_session_id, team_id = validate_record_scope(
            self.scope,
            self.owner_session_id,
            self.team_id,
        )
        object.__setattr__(self, "scope", scope)
        object.__setattr__(self, "owner_session_id", owner_session_id)
        object.__setattr__(self, "team_id", team_id)

    def to_dict(self) -> dict[str, Any]:
        return asdict(self)

    @classmethod
    def from_dict(cls, value: dict[str, Any]) -> "ContextRecord":
        return cls(
            id=str(value["id"]),
            namespace=str(value["namespace"]),
            text=str(value["text"]),
            embedding=[float(item) for item in value.get("embedding", [])],
            metadata=dict(value.get("metadata", {})),
            source=str(value.get("source", "manual")),
            importance=float(value.get("importance", 0.5)),
            token_count=int(value.get("token_count", 0)),
            created_at=float(value.get("created_at", 0.0)),
            updated_at=float(value.get("updated_at", 0.0)),
            accessed_at=float(value.get("accessed_at", 0.0)),
            expires_at=(
                None if value.get("expires_at") is None else float(value["expires_at"])
            ),
            content_hash=str(value.get("content_hash", "")),
            scope=ContextScope(value.get("scope", ContextScope.PROJECT)),
            owner_session_id=(
                None
                if value.get("owner_session_id") is None
                else str(value["owner_session_id"])
            ),
            team_id=(None if value.get("team_id") is None else str(value["team_id"])),
        )

context_cache.models.SearchHit dataclass

Source code in context_cache/models.py

@dataclass(slots=True)
class SearchHit:
    record: ContextRecord
    score: float
    vector_score: float
    lexical_score: float
    importance_score: float
    recency_score: float

    def to_dict(self) -> dict[str, Any]:
        return {
            "record": self.record.to_dict(),
            "score": self.score,
            "vector_score": self.vector_score,
            "lexical_score": self.lexical_score,
            "importance_score": self.importance_score,
            "recency_score": self.recency_score,
        }

    @classmethod
    def from_dict(cls, value: dict[str, Any]) -> "SearchHit":
        return cls(
            record=ContextRecord.from_dict(value["record"]),
            score=float(value["score"]),
            vector_score=float(value.get("vector_score", 0.0)),
            lexical_score=float(value.get("lexical_score", 0.0)),
            importance_score=float(value.get("importance_score", 0.0)),
            recency_score=float(value.get("recency_score", 0.0)),
        )

context_cache.models.WorkingSet dataclass

Source code in context_cache/models.py

@dataclass(slots=True)
class WorkingSet:
    session_id: str
    namespace: str
    focus: str
    hits: list[SearchHit]
    context: str
    token_count: int
    token_budget: int
    refreshed_at: float
    swapped_in: list[str] = field(default_factory=list)
    swapped_out: list[str] = field(default_factory=list)
    retained: list[str] = field(default_factory=list)

    def to_dict(self) -> dict[str, Any]:
        return {
            "session_id": self.session_id,
            "namespace": self.namespace,
            "focus": self.focus,
            "hits": [hit.to_dict() for hit in self.hits],
            "context": self.context,
            "token_count": self.token_count,
            "token_budget": self.token_budget,
            "refreshed_at": self.refreshed_at,
            "swapped_in": self.swapped_in,
            "swapped_out": self.swapped_out,
            "retained": self.retained,
        }

    @classmethod
    def from_dict(cls, value: dict[str, Any]) -> "WorkingSet":
        return cls(
            session_id=str(value["session_id"]),
            namespace=str(value["namespace"]),
            focus=str(value["focus"]),
            hits=[SearchHit.from_dict(item) for item in value.get("hits", [])],
            context=str(value.get("context", "")),
            token_count=int(value.get("token_count", 0)),
            token_budget=int(value.get("token_budget", 0)),
            refreshed_at=float(value.get("refreshed_at", 0.0)),
            swapped_in=[str(item) for item in value.get("swapped_in", [])],
            swapped_out=[str(item) for item in value.get("swapped_out", [])],
            retained=[str(item) for item in value.get("retained", [])],
        )