Tool result cache

rmote.cache.Cache stores results under caller-chosen string keys. It does not know about facts, collectors or remote connections. Values retain their Python types and identities. There is no automatic freezing or defensive deep copy.

from rmote.cache import Cache

cache = Cache[object]()
result = {"running": True}
cache.update({"service-status": result})
assert cache["service-status"] is result
assert cache.data is cache.data
assert cache.stale(keys=["missing-result"]) == ("missing-result",)

The data property returns a read-only mapping of keys in O(1). Replacing a key creates a new outer mapping; existing key snapshots continue to refer to their previous values. Mutating a retained object affects every reader holding it. The tool or caller owns that decision. For an independent deeply immutable value, explicitly use freeze before inserting it.

Cache[T] describes the value type, such as Cache[str] or Cache[MyResult]. A heterogeneous cache can use Cache[object] with type narrowing, or Cache[Any]. Facts-specific validation and FactsData belong to rmote.tools.facts.validate, not to the cache.

stale(keys=..., max_age=...) reports missing, expired or version-mismatched entries. Zero forces refresh; None accepts any age. With no keys supplied it checks stored keys and keys declared in the optional versions mapping. An empty unconfigured cache cannot guess which tool results the caller needs. Versions default to 1; configure expected versions with Cache(versions={"service-status": 2}).

Persistence is explicit:

from pathlib import Path
from tempfile import TemporaryDirectory
import asyncio

from rmote.cache import Cache, JSONCacheDir

async def example(directory):
    backend = JSONCacheDir(Path(directory))
    cache = Cache[str]()
    cache.update({"command-output": "ready"})
    await cache.save(backend, namespace="user@server")
    restored = Cache[str]()
    await restored.load(backend, namespace="user@server")
    assert restored["command-output"] == "ready"

with TemporaryDirectory() as directory:
    asyncio.run(example(directory))

The first load/save binds the cache to one namespace. It can identify a host, job, environment or another caller-defined scope. load(keys=None) reads all stored entries in that namespace; an explicit key iterable selects a subset. invalidate() affects memory only; use the backend’s delete() for storage. Older observations cannot replace newer ones. Backends merge batches atomically; JSON uses one document replacement, and SQLite uses one transaction. JSON file names use escaped namespaces with one-space indentation.

Keep values passed to save, set or set_many unchanged until completion. Replacing a cache branch does not alter an already-started batch. Cancelling an await may leave a worker finishing its atomic write. The in-memory cache accepts arbitrary values; each backend defines its serialization limits. Failed serialization does not partially commit a batch.

Both bundled backends use the JSON hooks for dataclasses and base64 bytes. Other values follow standard JSON rules. This is a storage representation choice, not a restriction on the in-memory cache.

class rmote.cache.Cache(*, versions=None)

Bases: Mapping[str, T], Generic[T]

Store named results without knowing which tool produced them.

Optional versions declare expected versions and keys to check for freshness. Other keys remain valid and default to version 1. The first load/save binds this instance to one namespace (for example an SSH host). Read-only key snapshots share the original values; they do not promise deep immutability. Use freeze explicitly when that is required.

bind(namespace)

Return type:

None

property data: Mapping[str, T]

Return the current key snapshot in O(1), sharing retained values.

invalidate(*, keys=None)

Forget selected in-memory entries without deleting persistent data.

Return type:

None

async load(backend, *, namespace, keys=None)

Merge a backend snapshot, omitting incompatible versions.

None selects every persisted key in this namespace. Values are retained as returned by the backend, without copying or model validation. A load failure leaves in-memory entries unchanged.

Return type:

None

merge(entries)

Return type:

None

async save(backend, *, namespace, keys=None)

Persist selected entries atomically, leaving other stored keys intact.

Do not mutate their values until this await completes. Replacing a branch in this cache does not change an already-started batch.

Return type:

None

select(keys)

Return type:

tuple[Text, ...]

stale(*, keys=None, max_age=300)

Return missing, incompatible or expired keys without I/O.

With no keys supplied, inspect configured version keys and stored keys. Zero forces refresh; None accepts any age. Future timestamps are stale with a finite TTL.

Return type:

tuple[Text, ...]

update(values, *, collected_at=None)

Replace complete branches, preserving newer observations.

Values are retained by reference. Invalid keys or timestamps leave the entire cache unchanged. No remote calls or storage I/O are performed.

Return type:

None

static validate_key(key)

Return type:

None

Persistent caches with atomic batches and per-value write ordering.

class rmote.cache.backends.CacheBackend(*args, **kwargs)

Bases: Protocol

Async storage contract shared by persistent cache backends.

set_many atomically replaces selected branches, leaving other keys intact. It ignores writes older than the stored collection start time. Operations must be safe across instances and processes. Missing keys return None from get and are omitted by get_many; I/O failures and corrupt data propagate rather than becoming cache misses. Blocking storage work must not run on the caller’s event loop.

async delete(namespace, key)

Remove a branch if present. Does not cancel an in-flight collection.

Return type:

None

async get(namespace, key)

Read one branch, including metadata, without checking its age.

Return type:

Optional[CacheEntry[Any]]

async get_many(namespace, keys)

Read selected branches from one snapshot, omitting missing keys.

Return type:

WSGIEnvironment[Text, CacheEntry[Any]]

async set(namespace, key, entry)

Replace a branch unless a newer collection is already stored.

Return type:

None

async set_many(namespace, entries)

Atomically merge a batch, ignoring older writes independently per key.

Return type:

None

class rmote.cache.backends.CacheEntry(data, collected_at, version=1)

Bases: Generic[T]

One complete value result.

Parameters:

• data (TypeVar(T)) – Any tool result supported by the selected backend.

• collected_at (getdouble) – Controller observation timestamp, in Unix seconds. Also orders writes: an older collection cannot overwrite a newer cached result.

• version (getint) – Positive value schema version; changed versions force refresh.

The bundled persistent backends reconstruct values on load. Treat entries passed to set or set_many as immutable until the call completes. Ordering assumes the controller processes share a reasonably stable wall clock.

collected_at: float

data: T

version: int = 1

class rmote.cache.backends.JSONCacheDir(path)

Bases: object

Store one JSON document per namespace in a local directory (POSIX).

Documents use namespace names, for example admin@server.json, and JSON indentation of one space. Path separators and percent signs are encoded so namespace identities cannot escape the directory or alias escaped names. A stable per-namespace flock protects read/modify/write across processes; writes use fsync and atomic replacement. Lock files are retained to keep every process locking the same inode. Use a local filesystem with working POSIX locks and atomic rename.

The directory is created lazily. New directories and files are private to the current user. No persistent handles need closing. Disk operations run in worker threads; cancellation may leave an already-started write running.

async delete(namespace, key)

Delete one branch while preserving every other value.

Return type:

None

async get(namespace, key)

Read a value entry or return None when absent.

Return type:

Optional[CacheEntry[Any]]

async get_many(namespace, keys)

Read all selected branches with one lock and one document parse.

Return type:

WSGIEnvironment[Text, CacheEntry[Any]]

locked(namespace)

Lock a namespace document for a short filesystem operation.

Return type:

Iterator[Path]

static parse_document(text, *, decode=True)

Parse a document, with or without rebuilding the stored models.

Without decode the branches stay plain JSON containers. A merge and an age comparison need only numbers and opaque branches, and rebuilding the models of a whole document, then encoding them back, costs more than everything else a write does.

Return type:

WSGIEnvironment[Text, Any]

static read_document(path, *, decode=True)

Read a document, treating only an absent file as an empty cache.

Return type:

WSGIEnvironment[Text, Any]

read_entries(namespace, keys)

Return type:

WSGIEnvironment[Text, CacheEntry[Any]]

read_entry(namespace, key)

Return type:

Optional[CacheEntry[Any]]

static read_text(path)

Return the text of a document. An absent file reads as empty.

Return type:

Text

async set(namespace, key, entry)

Atomically replace one value branch if it is not older.

Return type:

None

async set_many(namespace, entries)

Encode and atomically merge a batch with one document replacement.

Return type:

None

write_entries(namespace, entries)

Return type:

None

write_entry(namespace, key, payload)

Return type:

None

class rmote.cache.backends.JSONEntryCodec

Bases: object

JSON representation shared by the bundled persistent backends.

static decode(value)

Restore an entry directly from a parsed document.

Return type:

CacheEntry[Any]

static dumps(entry)

Return type:

Text

static encode(entry)

Wrap an entry without copying or pre-serializing its data.

Return type:

WSGIEnvironment[Text, Any]

static loads(payload)

Return type:

CacheEntry[Any]

static revive(value)

Rebuild the models of one branch that was read without them.

The containers are walked bottom-up, exactly as the parser applies the hook, because encoding the branch back to text and parsing it again costs more than the walk.

Return type:

Any

static stored_at(value)

Return the collection time of a stored branch, leaving its data alone.

A merge only orders branches against each other, so it checks the format and the numbers and never rebuilds the models of the data.

Return type:

getdouble

static validate(entry)

Return type:

None

class rmote.cache.backends.SQLiteCache(path, *, timeout=5.0)

Bases: object

Store JSON value entries in a local SQLite database.

The primary key is (namespace, key). Conditional UPSERT prevents older collections from overwriting newer results. Each operation opens a short connection in a worker thread; a batch shares one transaction. Encoding and decoding also run in that thread; transactions never span remote collection. Concurrent writers wait up to timeout seconds before SQLite raises. The parent directory must exist. :memory: is intentionally unsupported; this backend persists across connections and runs. No close is needed.

connection()

Return type:

Iterator[Connection]

async delete(namespace, key)

Delete one branch, doing nothing when it is absent.

Return type:

None

execute(query, parameters)

Return type:

Any

async get(namespace, key)

Read a value entry or return None when absent.

Return type:

Optional[CacheEntry[Any]]

async get_many(namespace, keys)

Read selected branches from one transaction.

Return type:

WSGIEnvironment[Text, CacheEntry[Any]]

read_entries(namespace, keys)

Return type:

WSGIEnvironment[Text, CacheEntry[Any]]

async set(namespace, key, entry)

Atomically replace one branch unless a newer collection exists.

Return type:

None

async set_many(namespace, entries)

Validate the batch, then merge it in one transaction.

Return type:

None

write_entries(namespace, entries)

Return type:

None