Skip to content

Session Cache API

Session Cache

Module: lean_interact.sessioncache

This module implements the session cache classes responsible for storing and retrieving Lean proof states and environments. Session cache is used internally by the AutoLeanServer class. It enables efficient resumption of proofs and environments after server restarts, timeouts, and automated recover from crashes. Additionally, ReplaySessionCache and PickleSessionCache are thread-safe and can be used to share session states between multiple AutoLeanServer instances within the same process. While by default AutoLeanServer instantiates a fresh ReplaySessionCache instance, you can also use a custom one. It can be useful to implement more advanced caching strategies, shareable cache across processes / compute nodes, ...

Examples:

from lean_interact.sessioncache import PickleSessionCache, ReplaySessionCache
from lean_interact.server import AutoLeanServer

# Create a session cache
replay_cache = ReplaySessionCache()
pickle_cache = PickleSessionCache(working_dir="./cache")

# Create Lean servers with a given cache
server = AutoLeanServer(config=..., session_cache=replay_cache)
legacy = AutoLeanServer(config=..., session_cache=pickle_cache)

ReplaySessionCache 

ReplaySessionCache(lazy: bool = True)

Bases: BaseSessionCache

Automatically replays cached Lean commands to restore proof states and environments when needed.

Parameters:

Name Type Description Default
lazy bool

When True (default) cached states are re-materialized on demand the next time they are requested. When False, reload() eagerly replays every cached command for the target LeanServer, which can reduce latency after restarts at the cost of upfront work.

 True
Source code in src/lean_interact/sessioncache.py 
166
167
168
169
170
def __init__(self, lazy: bool = True):
    super().__init__()
    self._cache: dict[int, ReplaySessionState] = {}
    self._state_counter = 0
    self._lazy = lazy

add 

add(
    lean_server: LeanServer,
    request: BaseREPLQuery,
    response: BaseREPLResponse,
    verbose: bool = False,
) -> int
Source code in src/lean_interact/sessioncache.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
def add(
    self,
    lean_server: "LeanServer",
    request: BaseREPLQuery,
    response: BaseREPLResponse,
    verbose: bool = False,
) -> int:
    if isinstance(response, ProofStepResponse):
        repl_id = response.proof_state
        is_proof_state = True
    elif isinstance(response, CommandResponse):
        repl_id = response.env
        is_proof_state = False
    else:
        raise NotImplementedError(
            f"Cannot store the session state for unsupported response of type {type(response).__name__}."
        )

    request_copy = request.model_copy(deep=True)
    with self._lock:
        self._state_counter -= 1
        session_id = self._state_counter
        self._cache[session_id] = ReplaySessionState(
            session_id=session_id,
            is_proof_state=is_proof_state,
            request=request_copy,
        )
    self._set_state_repl_id(self._cache[session_id], lean_server, repl_id)
    return session_id

remove 

remove(
    session_state_id: int, verbose: bool = False
) -> None
Source code in src/lean_interact/sessioncache.py 
249
250
251
def remove(self, session_state_id: int, verbose: bool = False) -> None:
    with self._lock:
        self._cache.pop(session_state_id, None)

reload 

reload(
    lean_server: LeanServer,
    timeout_per_state: int | float | None,
    verbose: bool = False,
) -> None
Source code in src/lean_interact/sessioncache.py 
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
def reload(
    self,
    lean_server: "LeanServer",
    timeout_per_state: int | float | None,
    verbose: bool = False,
) -> None:
    with self._lock:
        states = list(self._cache.values())
    for state in states:
        self._set_state_repl_id(state, lean_server, None)
        if not self._lazy:
            self._materialize_state(
                lean_server,
                state,
                timeout=timeout_per_state,
                verbose=verbose,
            )

is_empty 

is_empty() -> bool
Source code in src/lean_interact/sessioncache.py 
271
272
273
def is_empty(self) -> bool:
    with self._lock:
        return len(self._cache) == 0

clear 

clear(verbose: bool = False) -> None
Source code in src/lean_interact/sessioncache.py 
275
276
277
def clear(self, verbose: bool = False) -> None:
    with self._lock:
        self._cache.clear()

keys 

keys() -> list[int]
Source code in src/lean_interact/sessioncache.py 
291
292
293
def keys(self) -> list[int]:
    with self._lock:
        return list(self._cache.keys())

get_repl_id 

get_repl_id(
    session_state_id: int, lean_server: LeanServer
) -> int | None
Source code in src/lean_interact/sessioncache.py 
295
296
297
298
299
300
301
def get_repl_id(self, session_state_id: int, lean_server: "LeanServer") -> int | None:
    state = self.__getitem__(session_state_id)
    repl_id = self._get_state_repl_id(state, lean_server)
    if repl_id is None:
        self._materialize_state(lean_server, state)
        repl_id = self._get_state_repl_id(state, lean_server)
    return repl_id

PickleSessionCache 

PickleSessionCache(working_dir: str | PathLike)

Bases: BaseSessionCache

A session cache based on the local file storage and the REPL pickle feature.

Warning

Pickled Lean states are not fully reliable yet and are not always reloaded correctly. Prefer ReplaySessionCache unless you explicitly need serialized states on disk.

Source code in src/lean_interact/sessioncache.py 
317
318
319
320
321
def __init__(self, working_dir: str | PathLike):
    super().__init__()
    self._cache: dict[int, PickleSessionState] = {}
    self._state_counter = 0
    self._working_dir = Path(working_dir)

add 

add(
    lean_server: LeanServer,
    request: BaseREPLQuery,
    response: BaseREPLResponse,
    verbose: bool = False,
) -> int
Source code in src/lean_interact/sessioncache.py 
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
def add(
    self, lean_server: "LeanServer", request: BaseREPLQuery, response: BaseREPLResponse, verbose: bool = False
) -> int:
    with self._lock:
        self._state_counter -= 1
        session_id = self._state_counter
    process_id = os.getpid()  # use process id to avoid conflicts in multiprocessing
    hash_key = f"request_{type(request).__name__}_{id(request)}"
    pickle_file = (
        self._working_dir / "session_cache" / f"{hashlib.sha256(hash_key.encode()).hexdigest()}_{process_id}.olean"
    )
    pickle_file.parent.mkdir(parents=True, exist_ok=True)
    if isinstance(response, ProofStepResponse):
        repl_id = response.proof_state
        is_proof_state = True
        request = PickleProofState(proof_state=response.proof_state, pickle_to=str(pickle_file))
    elif isinstance(response, CommandResponse):
        repl_id = response.env
        is_proof_state = False
        request = PickleEnvironment(env=response.env, pickle_to=str(pickle_file))
    else:
        raise NotImplementedError(
            f"Cannot pickle the session state for unsupported request of type {type(request).__name__}."
        )

    # Use file lock when accessing the pickle file to prevent cache invalidation
    # from concurrent access
    with FileLock(f"{pickle_file}.lock", timeout=60):
        response_pickle = lean_server.run(request, verbose=verbose)
        if isinstance(response_pickle, LeanError):
            raise ValueError(
                f"Could not store the result in the session cache. The Lean server returned an error: {response_pickle.message}"
            )

        with self._lock:
            self._cache[session_id] = PickleSessionState(
                session_id=session_id,
                pickle_file=str(pickle_file),
                is_proof_state=is_proof_state,
            )
        self._set_state_repl_id(self._cache[session_id], lean_server, repl_id)
    return session_id

remove 

remove(
    session_state_id: int, verbose: bool = False
) -> None
Source code in src/lean_interact/sessioncache.py 
374
375
376
377
378
379
380
381
def remove(self, session_state_id: int, verbose: bool = False) -> None:
    with self._lock:
        state_cache = self._cache.pop(session_state_id, None)
    if state_cache is not None:
        pickle_file = state_cache.pickle_file
        with FileLock(f"{pickle_file}.lock", timeout=60):
            if os.path.exists(pickle_file):
                os.remove(pickle_file)

reload 

reload(
    lean_server: LeanServer,
    timeout_per_state: int | float | None,
    verbose: bool = False,
) -> None
Source code in src/lean_interact/sessioncache.py 
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
def reload(self, lean_server: "LeanServer", timeout_per_state: int | float | None, verbose: bool = False) -> None:
    with self._lock:
        state_snapshot = list(self._cache.values())
    for state_data in state_snapshot:
        # Use file lock when accessing the pickle file to prevent cache invalidation
        # from multiple concurrent processes
        with FileLock(
            f"{state_data.pickle_file}.lock", timeout=float(timeout_per_state) if timeout_per_state else -1
        ):
            if state_data.is_proof_state:
                cmd = UnpickleProofState(
                    unpickle_proof_state_from=state_data.pickle_file,
                    env=self._get_state_repl_id(state_data, lean_server),
                )
            else:
                cmd = UnpickleEnvironment(unpickle_env_from=state_data.pickle_file)
            result = lean_server.run(
                cmd,
                verbose=verbose,
                timeout=timeout_per_state,
            )
            if isinstance(result, LeanError):
                raise ValueError(
                    f"Could not reload the session cache. The Lean server returned an error: {result.message}"
                )
            elif isinstance(result, CommandResponse):
                self._set_state_repl_id(state_data, lean_server, result.env)
            elif isinstance(result, ProofStepResponse):
                self._set_state_repl_id(state_data, lean_server, result.proof_state)
            else:
                raise ValueError(
                    f"Could not reload the session cache. The Lean server returned an unexpected response: {result}"
                )

is_empty 

is_empty() -> bool
Source code in src/lean_interact/sessioncache.py 
417
418
419
def is_empty(self) -> bool:
    with self._lock:
        return len(self._cache) == 0

clear 

clear(verbose: bool = False) -> None
Source code in src/lean_interact/sessioncache.py 
421
422
423
424
425
426
427
def clear(self, verbose: bool = False) -> None:
    with self._lock:
        state_snapshot = list(self._cache.values())
    for state_data in state_snapshot:
        self.remove(session_state_id=state_data.session_id, verbose=verbose)
    with self._lock:
        assert not self._cache, f"Cache is not empty after clearing: {self._cache}"

keys 

keys() -> list[int]
Source code in src/lean_interact/sessioncache.py 
441
442
443
def keys(self) -> list[int]:
    with self._lock:
        return list(self._cache.keys())

get_repl_id 

get_repl_id(
    session_state_id: int, lean_server: LeanServer
) -> int | None
Source code in src/lean_interact/sessioncache.py 
445
446
447
def get_repl_id(self, session_state_id: int, lean_server: "LeanServer") -> int | None:
    state = self.__getitem__(session_state_id)
    return self._get_state_repl_id(state, lean_server)

SessionState dataclass 

SessionState(
    *,
    session_id: int,
    is_proof_state: bool,
    repl_ids: dict[str, int | None] = dict(),
)

session_id instance-attribute 

session_id: int

is_proof_state instance-attribute 

is_proof_state: bool

repl_ids class-attribute instance-attribute 

repl_ids: dict[str, int | None] = field(
    default_factory=dict
)

ReplaySessionState dataclass 

ReplaySessionState(
    *,
    session_id: int,
    is_proof_state: bool,
    repl_ids: dict[str, int | None] = dict(),
    request: BaseREPLQuery,
    _materializing_servers: set[str] = set(),
)

Bases: SessionState

request instance-attribute 

request: BaseREPLQuery

session_id instance-attribute 

session_id: int

is_proof_state instance-attribute 

is_proof_state: bool

repl_ids class-attribute instance-attribute 

repl_ids: dict[str, int | None] = field(
    default_factory=dict
)

PickleSessionState dataclass 

PickleSessionState(
    *,
    session_id: int,
    is_proof_state: bool,
    repl_ids: dict[str, int | None] = dict(),
    pickle_file: str,
)

Bases: SessionState

pickle_file instance-attribute 

pickle_file: str

session_id instance-attribute 

session_id: int

is_proof_state instance-attribute 

is_proof_state: bool

repl_ids class-attribute instance-attribute 

repl_ids: dict[str, int | None] = field(
    default_factory=dict
)