windlass.agent.checkpoint

checkpoint

Checkpointing — durable agent state.

An agent run is a sequence of model calls and tool executions. Checkpointing snapshots that state after every step, which buys three things:

• Resumability. A run interrupted for human approval, or by a crash, picks up where it stopped instead of starting over (and re-paying for every token already spent).
• Multi-turn conversations. A thread_id maps to accumulated state, so the next message continues the same run.
• Time travel. Every step is retained, so you can inspect exactly what the agent believed at step 3 when debugging why step 4 went wrong.

Two backends ship: in-memory (fast, per-process) and SQLite (durable, shared between processes, stdlib-only).

Example

saver = MemoryCheckpointer() saver.put("thread-1", {"step": 1, "messages": []}) saver.get("thread-1")["step"] 1

Checkpointer

Bases: ABC

Stores and retrieves agent state by thread.

Implementations must be safe to use from multiple threads, since one agent instance typically serves many concurrent conversations.

put abstractmethod

put(thread_id: str, state: dict[str, Any]) -> None

Save a state snapshot.

Parameters:

Name	Type	Description	Default
thread_id	str	Conversation / run identifier.	required
state	dict[str, Any]	JSON-serialisable state.	required

Raises:

Type	Description
SerializationError	When the state cannot be stored.

Source code in src\windlass\agent\checkpoint.py

@abc.abstractmethod
def put(self, thread_id: str, state: dict[str, Any]) -> None:
    """Save a state snapshot.

    Args:
        thread_id: Conversation / run identifier.
        state: JSON-serialisable state.

    Raises:
        SerializationError: When the state cannot be stored.
    """

get abstractmethod

get(thread_id: str) -> dict[str, Any] | None

Return the latest snapshot for a thread, or None.

Source code in src\windlass\agent\checkpoint.py

@abc.abstractmethod
def get(self, thread_id: str) -> dict[str, Any] | None:
    """Return the latest snapshot for a thread, or ``None``."""

history abstractmethod

history(thread_id: str, limit: int = 20) -> list[dict[str, Any]]

Return recent snapshots, newest first.

Parameters:

Name	Type	Description	Default
thread_id	str	Conversation identifier.	required
limit	int	Maximum snapshots to return.	20

Returns:

Type	Description
list[dict[str, Any]]	Snapshots, newest first.

Source code in src\windlass\agent\checkpoint.py

@abc.abstractmethod
def history(self, thread_id: str, limit: int = 20) -> list[dict[str, Any]]:
    """Return recent snapshots, newest first.

    Args:
        thread_id: Conversation identifier.
        limit: Maximum snapshots to return.

    Returns:
        Snapshots, newest first.
    """

delete abstractmethod

delete(thread_id: str) -> None

Discard every snapshot for a thread.

Source code in src\windlass\agent\checkpoint.py

@abc.abstractmethod
def delete(self, thread_id: str) -> None:
    """Discard every snapshot for a thread."""

threads

threads() -> list[str]

Return the known thread ids.

Source code in src\windlass\agent\checkpoint.py

def threads(self) -> list[str]:
    """Return the known thread ids."""
    return []

MemoryCheckpointer

MemoryCheckpointer(max_history: int = 50)

Bases: Checkpointer

Keeps checkpoints in a dict.

Fast and dependency-free, but per-process and lost on restart. Right for tests, notebooks and single-process services; wrong for anything that needs a resumed run to survive a deploy.

Parameters:

Name	Type	Description	Default
max_history	int	Snapshots retained per thread. Older ones are dropped.	50

Example

saver = MemoryCheckpointer() saver.put("t", {"step": 1}) saver.put("t", {"step": 2}) saver.get("t")["step"], len(saver.history("t")) (2, 2)

Source code in src\windlass\agent\checkpoint.py

def __init__(self, max_history: int = 50) -> None:
    self.max_history = max_history
    self._data: dict[str, list[dict[str, Any]]] = {}
    self._lock = threading.RLock()

put

put(thread_id: str, state: dict[str, Any]) -> None

Append a snapshot, trimming the oldest beyond max_history.

Source code in src\windlass\agent\checkpoint.py

def put(self, thread_id: str, state: dict[str, Any]) -> None:
    """Append a snapshot, trimming the oldest beyond ``max_history``."""
    with self._lock:
        bucket = self._data.setdefault(thread_id, [])
        bucket.append({**state, "_saved_at": time.time()})
        if len(bucket) > self.max_history:
            del bucket[: len(bucket) - self.max_history]

get

get(thread_id: str) -> dict[str, Any] | None

Return the newest snapshot, or None.

Source code in src\windlass\agent\checkpoint.py

def get(self, thread_id: str) -> dict[str, Any] | None:
    """Return the newest snapshot, or ``None``."""
    with self._lock:
        bucket = self._data.get(thread_id)
        return dict(bucket[-1]) if bucket else None

history

history(thread_id: str, limit: int = 20) -> list[dict[str, Any]]

Return recent snapshots, newest first.

Source code in src\windlass\agent\checkpoint.py

def history(self, thread_id: str, limit: int = 20) -> list[dict[str, Any]]:
    """Return recent snapshots, newest first."""
    with self._lock:
        bucket = self._data.get(thread_id, [])
        return [dict(s) for s in reversed(bucket[-limit:])]

delete

delete(thread_id: str) -> None

Discard a thread's snapshots.

Source code in src\windlass\agent\checkpoint.py

def delete(self, thread_id: str) -> None:
    """Discard a thread's snapshots."""
    with self._lock:
        self._data.pop(thread_id, None)

threads

threads() -> list[str]

Return the known thread ids.

Source code in src\windlass\agent\checkpoint.py

def threads(self) -> list[str]:
    """Return the known thread ids."""
    with self._lock:
        return sorted(self._data)

SQLiteCheckpointer

SQLiteCheckpointer(
    path: str | Path = ".windlass/checkpoints.db", max_history: int = 50
)

Bases: Checkpointer

Persists checkpoints to a SQLite database.

Survives restarts and is shared between processes on one machine — enough for a single-node deployment, and it needs nothing beyond the standard library.

Parameters:

Name	Type	Description	Default
path	str | Path	Database file. Parent directories are created. ":memory:" gives a transient database.	'.windlass/checkpoints.db'
max_history	int	Snapshots retained per thread.	50

Raises:

Type	Description
SerializationError	When the database cannot be opened.

Note

Each call opens its own connection. That costs a little per write and buys thread safety without a connection pool — the right trade at the rate an agent checkpoints.

Example

import tempfile, pathlib db = pathlib.Path(tempfile.mkdtemp()) / "state.db" saver = SQLiteCheckpointer(db) saver.put("t", {"step": 7}) SQLiteCheckpointer(db).get("t")["step"] 7

Source code in src\windlass\agent\checkpoint.py

def __init__(
    self, path: str | Path = ".windlass/checkpoints.db", max_history: int = 50
) -> None:
    self.path = Path(path) if str(path) != ":memory:" else Path(":memory:")
    self.max_history = max_history
    self._lock = threading.RLock()
    if str(self.path) != ":memory:":
        self.path.parent.mkdir(parents=True, exist_ok=True)
    self._shared: sqlite3.Connection | None = (
        sqlite3.connect(":memory:", check_same_thread=False)
        if str(self.path) == ":memory:"
        else None
    )
    self._init()

put

put(thread_id: str, state: dict[str, Any]) -> None

Insert a snapshot and prune old ones.

Raises:

Type	Description
SerializationError	When the state is not JSON-serialisable or the write fails.

Source code in src\windlass\agent\checkpoint.py

def put(self, thread_id: str, state: dict[str, Any]) -> None:
    """Insert a snapshot and prune old ones.

    Raises:
        SerializationError: When the state is not JSON-serialisable or the
            write fails.
    """
    try:
        payload = json.dumps(state, default=str)
    except (TypeError, ValueError) as exc:
        raise SerializationError(
            f"Agent state for thread {thread_id!r} is not JSON-serialisable: {exc}",
            hint="Keep custom objects out of agent state, or store an id instead.",
        ) from exc

    with self._lock:
        connection = self._connect()
        try:
            connection.execute(
                "INSERT INTO checkpoints (thread_id, state, saved_at) VALUES (?, ?, ?)",
                (thread_id, payload, time.time()),
            )
            connection.execute(
                """
                DELETE FROM checkpoints
                WHERE thread_id = ?
                  AND id NOT IN (
                      SELECT id FROM checkpoints
                      WHERE thread_id = ? ORDER BY id DESC LIMIT ?
                  )
                """,
                (thread_id, thread_id, self.max_history),
            )
            connection.commit()
        except sqlite3.Error as exc:
            raise SerializationError(f"Could not save the checkpoint: {exc}") from exc
        finally:
            if self._shared is None:
                connection.close()

get

get(thread_id: str) -> dict[str, Any] | None

Return the newest snapshot, or None.

Source code in src\windlass\agent\checkpoint.py

def get(self, thread_id: str) -> dict[str, Any] | None:
    """Return the newest snapshot, or ``None``."""
    rows = self.history(thread_id, limit=1)
    return rows[0] if rows else None

history

history(thread_id: str, limit: int = 20) -> list[dict[str, Any]]

Return recent snapshots, newest first.

Source code in src\windlass\agent\checkpoint.py

def history(self, thread_id: str, limit: int = 20) -> list[dict[str, Any]]:
    """Return recent snapshots, newest first."""
    with self._lock:
        connection = self._connect()
        try:
            cursor = connection.execute(
                "SELECT state FROM checkpoints WHERE thread_id = ? ORDER BY id DESC LIMIT ?",
                (thread_id, limit),
            )
            rows = cursor.fetchall()
        except sqlite3.Error as exc:
            _log.warning("Could not read checkpoints for %s: %s", thread_id, exc)
            return []
        finally:
            if self._shared is None:
                connection.close()

    snapshots: list[dict[str, Any]] = []
    for (payload,) in rows:
        try:
            snapshots.append(json.loads(payload))
        except json.JSONDecodeError:  # pragma: no cover - corrupt row
            _log.warning("Skipping a corrupt checkpoint for thread %s.", thread_id)
    return snapshots

delete

delete(thread_id: str) -> None

Discard a thread's snapshots.

Source code in src\windlass\agent\checkpoint.py

def delete(self, thread_id: str) -> None:
    """Discard a thread's snapshots."""
    with self._lock:
        connection = self._connect()
        try:
            connection.execute("DELETE FROM checkpoints WHERE thread_id = ?", (thread_id,))
            connection.commit()
        except sqlite3.Error as exc:  # pragma: no cover
            _log.warning("Could not delete checkpoints for %s: %s", thread_id, exc)
        finally:
            if self._shared is None:
                connection.close()

threads

threads() -> list[str]

Return the known thread ids.

Source code in src\windlass\agent\checkpoint.py

def threads(self) -> list[str]:
    """Return the known thread ids."""
    with self._lock:
        connection = self._connect()
        try:
            cursor = connection.execute("SELECT DISTINCT thread_id FROM checkpoints")
            return sorted(row[0] for row in cursor.fetchall())
        except sqlite3.Error:  # pragma: no cover
            return []
        finally:
            if self._shared is None:
                connection.close()

close

close() -> None

Close the shared in-memory connection, if there is one.

Source code in src\windlass\agent\checkpoint.py

def close(self) -> None:
    """Close the shared in-memory connection, if there is one."""
    if self._shared is not None:
        self._shared.close()
        self._shared = None