Skip to content

Cache

cache

Per-agent model roster cache for the ACP host-proxy.

Probing an agent (initialize + session/new + read configOptions) is expensive and the result is stable for the lifetime of an authenticated session. The cache is keyed (image_id, auth_identity, agent_id): same image, same auth, same agent ⇒ same model list.

The cache is populated lazily on the first session/new after a new auth, and never re-probed mid-session. invalidate_auth lets workflows flush an entire identity's worth of entries when credentials change (today auth is global so this is rarely useful; the hook exists for future per-project auth).

GLOBAL_CACHE = AgentRosterCache() module-attribute

CacheKey(image_id, auth_identity, agent_id) dataclass

Composite key for one agent's roster within one auth scope.

auth_identity is the constant "global" today (terok auth is process-wide); the field exists from day one so per-project auth can slot in without a key-schema migration.

image_id instance-attribute

auth_identity instance-attribute

agent_id instance-attribute

AgentRosterCache()

Thread-safe map from CacheKey to a tuple of model ids.

Models are stored as a tuple so cache entries are immutable once inserted — callers can return them directly without defensive copying. Empty tuples are valid and signal "probe ran but yielded nothing" (saved to avoid hammering a misconfigured agent on every session).

Source code in src/terok_executor/acp/cache.py 
47
48
49
def __init__(self) -> None:
    self._models: dict[CacheKey, tuple[str, ...]] = {}
    self._lock = threading.Lock()

get(key)

Return cached models for key, or None if not yet probed.

Source code in src/terok_executor/acp/cache.py 
51
52
53
54
def get(self, key: CacheKey) -> tuple[str, ...] | None:
    """Return cached models for *key*, or ``None`` if not yet probed."""
    with self._lock:
        return self._models.get(key)

put(key, models)

Store models under key, replacing any existing entry.

Source code in src/terok_executor/acp/cache.py 
56
57
58
59
def put(self, key: CacheKey, models: tuple[str, ...]) -> None:
    """Store *models* under *key*, replacing any existing entry."""
    with self._lock:
        self._models[key] = models

invalidate_auth(auth_identity)

Drop every entry tied to auth_identity.

Used when credentials for an identity rotate — the next session/new re-probes affected agents.

Source code in src/terok_executor/acp/cache.py 
61
62
63
64
65
66
67
68
69
70
def invalidate_auth(self, auth_identity: str) -> None:
    """Drop every entry tied to *auth_identity*.

    Used when credentials for an identity rotate — the next
    ``session/new`` re-probes affected agents.
    """
    with self._lock:
        self._models = {
            k: v for k, v in self._models.items() if k.auth_identity != auth_identity
        }

__len__()

Return the number of cached entries (for tests / introspection).

Source code in src/terok_executor/acp/cache.py 
72
73
74
75
def __len__(self) -> int:
    """Return the number of cached entries (for tests / introspection)."""
    with self._lock:
        return len(self._models)