Skip to content

teff.checkpoint

teff.checkpoint

Modules:

Name Description
base

Checkpointing primitives for durable graph execution.
file

JSON-file checkpointing — zero dependencies, atomic via tempfile + rename.
from_config

Build a :class:Checkpointer from a declarative {type, ...} config.
history

Time-travel checkpoints: keep every per-iteration snapshot.
pg

PostgreSQL checkpointing — requires asyncpg (teff[pg-checkpoint]).
sqlite

SQLite checkpointing — stdlib only, shared file format with the RAG store.

Classes:

Name Description
Checkpoint

A snapshot of execution that can be resumed from.
Checkpointer

Interface for persisting execution checkpoints.
JSONFileCheckpointer

Store checkpoints as one JSON file per (owner, checkpoint ID).
PGCheckpointer

Store checkpoints in a PostgreSQL table.
PGHistoryCheckpointer

PostgreSQL checkpointer that also keeps the full per-step history.
SQLiteCheckpointer

Store checkpoints in a SQLite database.
SQLiteHistoryCheckpointer

SQLite checkpointer that also keeps the full per-step history.

Functions:

Name Description
checkpoint_from_dict

Reconstruct a checkpoint from a dict produced by :func:checkpoint_to_dict.
checkpoint_to_dict

Convert a checkpoint to a JSON-serializable dict.

Checkpoint dataclass

A snapshot of execution that can be resumed from.

Attributes:

Name Type Description
state dict

Workflow state data (JSON-serializable dict).
next_node_id str | None

ID of the node to execute on resume. None means the graph completed.
iteration int

Number of completed node executions so far.
Source code in teff/checkpoint/base.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
@dataclass
class Checkpoint:
    """A snapshot of execution that can be resumed from.

    Attributes:
        state: Workflow state data (JSON-serializable dict).
        next_node_id: ID of the node to execute on resume.
            ``None`` means the graph completed.
        iteration: Number of completed node executions so far.
    """

    state: dict
    next_node_id: str | None
    iteration: int

Checkpointer

Bases: Protocol

Interface for persisting execution checkpoints.

Implementations must be safe to call concurrently for different checkpoint IDs and must persist atomically enough that a crash never leaves a partially-written checkpoint.

The optional owner scopes a checkpoint to a user/session/tenant. Two checkpoints with the same ID but different owners never collide. When omitted the default owner (:data:DEFAULT_OWNER, "default") is used, keeping single-tenant callers unchanged.

Use a distinct owner per end-user (e.g. a user id or session id) so every tenant's runs are isolated from the others and can be listed with :meth:list.

Methods:

Name Description
cleanup

Delete stale checkpoints; returns how many were removed.
delete

Remove a saved checkpoint. No-op if it does not exist.
list

Return all checkpoint IDs persisted for owner.
load

Return the saved checkpoint for owner, or None if never saved.
save

Persist checkpoint under checkpoint_id for owner.
Source code in teff/checkpoint/base.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
class Checkpointer(Protocol):
    """Interface for persisting execution checkpoints.

    Implementations must be safe to call concurrently for different
    checkpoint IDs and must persist *atomically* enough that a crash
    never leaves a partially-written checkpoint.

    The optional *owner* scopes a checkpoint to a user/session/tenant.
    Two checkpoints with the same ID but different owners never collide.
    When omitted the default owner (:data:`DEFAULT_OWNER`, ``"default"``)
    is used, keeping single-tenant callers unchanged.

    Use a distinct owner per end-user (e.g. a user id or session id) so
    every tenant's runs are isolated from the others and can be listed
    with :meth:`list`.
    """

    async def save(
        self, checkpoint_id: str, checkpoint: Checkpoint, *, owner: str = DEFAULT_OWNER
    ) -> None:
        """Persist *checkpoint* under *checkpoint_id* for *owner*."""
        ...

    async def load(
        self, checkpoint_id: str, *, owner: str = DEFAULT_OWNER
    ) -> Checkpoint | None:
        """Return the saved checkpoint for *owner*, or ``None`` if never saved."""
        ...

    async def delete(self, checkpoint_id: str, *, owner: str = DEFAULT_OWNER) -> None:
        """Remove a saved checkpoint. No-op if it does not exist."""
        ...

    async def list(self, owner: str = DEFAULT_OWNER) -> list[str]:
        """Return all checkpoint IDs persisted for *owner*."""
        ...

    async def cleanup(
        self,
        *,
        owner: str | None = None,
        max_age: float | None = None,
        keep_last: int | None = None,
    ) -> int:
        """Delete stale checkpoints; returns how many were removed.

        ``owner=None`` cleans up every owner; otherwise only that owner.
        ``max_age`` removes checkpoints last written more than that many
        seconds ago.  ``keep_last`` retains the *N* most recently written
        checkpoints per owner (after any ``max_age`` pruning) and deletes
        the rest.  When both are omitted nothing is deleted.
        """
        ...

cleanup async 

cleanup(*, owner=None, max_age=None, keep_last=None)

Delete stale checkpoints; returns how many were removed.

owner=None cleans up every owner; otherwise only that owner. max_age removes checkpoints last written more than that many seconds ago. keep_last retains the N most recently written checkpoints per owner (after any max_age pruning) and deletes the rest. When both are omitted nothing is deleted.

Source code in teff/checkpoint/base.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
async def cleanup(
    self,
    *,
    owner: str | None = None,
    max_age: float | None = None,
    keep_last: int | None = None,
) -> int:
    """Delete stale checkpoints; returns how many were removed.

    ``owner=None`` cleans up every owner; otherwise only that owner.
    ``max_age`` removes checkpoints last written more than that many
    seconds ago.  ``keep_last`` retains the *N* most recently written
    checkpoints per owner (after any ``max_age`` pruning) and deletes
    the rest.  When both are omitted nothing is deleted.
    """
    ...

delete async 

delete(checkpoint_id, *, owner=DEFAULT_OWNER)

Remove a saved checkpoint. No-op if it does not exist.

Source code in teff/checkpoint/base.py 
59
60
61
async def delete(self, checkpoint_id: str, *, owner: str = DEFAULT_OWNER) -> None:
    """Remove a saved checkpoint. No-op if it does not exist."""
    ...

list async 

list(owner=DEFAULT_OWNER)

Return all checkpoint IDs persisted for owner.

Source code in teff/checkpoint/base.py 
63
64
65
async def list(self, owner: str = DEFAULT_OWNER) -> list[str]:
    """Return all checkpoint IDs persisted for *owner*."""
    ...

load async 

load(checkpoint_id, *, owner=DEFAULT_OWNER)

Return the saved checkpoint for owner, or None if never saved.

Source code in teff/checkpoint/base.py 
53
54
55
56
57
async def load(
    self, checkpoint_id: str, *, owner: str = DEFAULT_OWNER
) -> Checkpoint | None:
    """Return the saved checkpoint for *owner*, or ``None`` if never saved."""
    ...

save async 

save(checkpoint_id, checkpoint, *, owner=DEFAULT_OWNER)

Persist checkpoint under checkpoint_id for owner.

Source code in teff/checkpoint/base.py 
47
48
49
50
51
async def save(
    self, checkpoint_id: str, checkpoint: Checkpoint, *, owner: str = DEFAULT_OWNER
) -> None:
    """Persist *checkpoint* under *checkpoint_id* for *owner*."""
    ...

JSONFileCheckpointer

Bases: Checkpointer

Store checkpoints as one JSON file per (owner, checkpoint ID).

Writes go to a temp file in the same directory and are atomically renamed over the target, so a crash never leaves a corrupt file. Each owner gets its own subdirectory, so IDs only need to be unique within an owner. See :class:~teff.checkpoint.Checkpointer for how to pick an owner.

Methods:

Name Description
cleanup

Delete stale checkpoints; returns how many were removed.
list

Return all checkpoint IDs persisted for owner.
Source code in teff/checkpoint/file.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
class JSONFileCheckpointer(Checkpointer):
    """Store checkpoints as one JSON file per (owner, checkpoint ID).

    Writes go to a temp file in the same directory and are atomically
    renamed over the target, so a crash never leaves a corrupt file.
    Each *owner* gets its own subdirectory, so IDs only need to be
    unique within an owner.  See :class:`~teff.checkpoint.Checkpointer`
    for how to pick an owner.
    """

    def __init__(self, directory: str, suffix: str = ".json"):
        self._directory = Path(directory)
        self._directory.mkdir(parents=True, exist_ok=True)
        self._suffix = suffix

    def _path(self, checkpoint_id: str, owner: str = DEFAULT_OWNER) -> Path:
        safe = checkpoint_id.replace(os.sep, "_").replace("/", "_")
        owner_dir = self._directory / self._safe_owner(owner)
        owner_dir.mkdir(parents=True, exist_ok=True)
        return owner_dir / f"{safe}{self._suffix}"

    @staticmethod
    def _safe_owner(owner: str) -> str:
        return owner.replace(os.sep, "_").replace("/", "_").replace(".", "_")

    async def save(
        self,
        checkpoint_id: str,
        checkpoint: Checkpoint,
        *,
        owner: str = DEFAULT_OWNER,
    ) -> None:
        target = self._path(checkpoint_id, owner)
        tmp = target.with_suffix(f"{self._suffix}.tmp")

        def _save() -> None:
            tmp.write_text(
                json.dumps(checkpoint_to_dict(checkpoint), ensure_ascii=False),
                encoding="utf-8",
            )
            os.replace(tmp, target)

        await asyncio.to_thread(_save)

    async def load(
        self, checkpoint_id: str, *, owner: str = DEFAULT_OWNER
    ) -> Checkpoint | None:
        path = self._path(checkpoint_id, owner)

        def _load() -> Checkpoint | None:
            if not path.exists():
                return None
            data = json.loads(path.read_text(encoding="utf-8"))
            return checkpoint_from_dict(data)

        return await asyncio.to_thread(_load)

    async def delete(self, checkpoint_id: str, *, owner: str = DEFAULT_OWNER) -> None:
        path = self._path(checkpoint_id, owner)

        def _delete() -> None:
            if path.exists():
                path.unlink()

        await asyncio.to_thread(_delete)

    async def list(self, owner: str = DEFAULT_OWNER) -> list[str]:
        """Return all checkpoint IDs persisted for *owner*."""
        base = self._directory / self._safe_owner(owner)

        def _list() -> list[str]:
            if not base.exists():
                return []
            return sorted(
                p.name[: -len(self._suffix)]
                for p in base.glob(f"*{self._suffix}")
                if not p.name.endswith(f"{self._suffix}.tmp")
            )

        return await asyncio.to_thread(_list)

    def _owners(self) -> List[str]:
        if not self._directory.exists():
            return []
        return sorted(p.name for p in self._directory.iterdir() if p.is_dir())

    def _owner_checkpoints(self, owner: str) -> List[Tuple[str, float]]:
        """Return ``(checkpoint_id, mtime)`` pairs for one owner."""
        base = self._directory / self._safe_owner(owner)
        if not base.exists():
            return []
        pairs = []
        for p in base.glob(f"*{self._suffix}"):
            if p.name.endswith(f"{self._suffix}.tmp"):
                continue
            pairs.append((p.name[: -len(self._suffix)], p.stat().st_mtime))
        pairs.sort(key=lambda item: item[1], reverse=True)
        return pairs

    async def cleanup(
        self,
        *,
        owner: str | None = None,
        max_age: float | None = None,
        keep_last: int | None = None,
    ) -> int:
        """Delete stale checkpoints; returns how many were removed."""
        if max_age is None and keep_last is None:
            return 0

        def _cleanup() -> int:
            removed = 0
            owners = [owner] if owner is not None else self._owners()
            now = time.time()
            for own in owners:
                pairs = self._owner_checkpoints(own)
                to_delete: List[str] = []
                for idx, (cid, mtime) in enumerate(pairs):
                    if max_age is not None and now - mtime > max_age:
                        to_delete.append(cid)
                    elif keep_last is not None and idx >= keep_last:
                        to_delete.append(cid)
                for cid in to_delete:
                    path = self._path(cid, own)
                    if path.exists():
                        path.unlink()
                        removed += 1
            return removed

        return await asyncio.to_thread(_cleanup)

cleanup async 

cleanup(*, owner=None, max_age=None, keep_last=None)

Delete stale checkpoints; returns how many were removed.

Source code in teff/checkpoint/file.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
async def cleanup(
    self,
    *,
    owner: str | None = None,
    max_age: float | None = None,
    keep_last: int | None = None,
) -> int:
    """Delete stale checkpoints; returns how many were removed."""
    if max_age is None and keep_last is None:
        return 0

    def _cleanup() -> int:
        removed = 0
        owners = [owner] if owner is not None else self._owners()
        now = time.time()
        for own in owners:
            pairs = self._owner_checkpoints(own)
            to_delete: List[str] = []
            for idx, (cid, mtime) in enumerate(pairs):
                if max_age is not None and now - mtime > max_age:
                    to_delete.append(cid)
                elif keep_last is not None and idx >= keep_last:
                    to_delete.append(cid)
            for cid in to_delete:
                path = self._path(cid, own)
                if path.exists():
                    path.unlink()
                    removed += 1
        return removed

    return await asyncio.to_thread(_cleanup)

list async 

list(owner=DEFAULT_OWNER)

Return all checkpoint IDs persisted for owner.

Source code in teff/checkpoint/file.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
async def list(self, owner: str = DEFAULT_OWNER) -> list[str]:
    """Return all checkpoint IDs persisted for *owner*."""
    base = self._directory / self._safe_owner(owner)

    def _list() -> list[str]:
        if not base.exists():
            return []
        return sorted(
            p.name[: -len(self._suffix)]
            for p in base.glob(f"*{self._suffix}")
            if not p.name.endswith(f"{self._suffix}.tmp")
        )

    return await asyncio.to_thread(_list)

PGCheckpointer

Bases: Checkpointer

Store checkpoints in a PostgreSQL table.

Requires asyncpg (install via teff[pg-checkpoint]). The table checkpoints is created lazily on first use. Connections are drawn from a lazily-created async connection pool, so checkpoint saves reuse warm connections instead of paying the handshake per operation.

Parameters:

Name Type Description Default
dsn str

PostgreSQL connection string.

 required
table str

Table name (default "checkpoints").

 'checkpoints'
pool_size int

Maximum pooled connections (default 5).

 5

Methods:

Name Description
cleanup

Delete stale checkpoints; returns how many were removed.
close

Close the connection pool (idempotent).
list

Return all checkpoint IDs persisted for owner.
Source code in teff/checkpoint/pg.py 
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
class PGCheckpointer(Checkpointer):
    """Store checkpoints in a PostgreSQL table.

    Requires ``asyncpg`` (install via ``teff[pg-checkpoint]``). The
    table ``checkpoints`` is created lazily on first use.  Connections are
    drawn from a lazily-created async connection pool, so checkpoint saves
    reuse warm connections instead of paying the handshake per operation.

    Args:
        dsn: PostgreSQL connection string.
        table: Table name (default ``"checkpoints"``).
        pool_size: Maximum pooled connections (default 5).
    """

    def __init__(self, dsn: str, table: str = "checkpoints", pool_size: int = 5):
        import importlib.util

        if importlib.util.find_spec("asyncpg") is None:
            raise ImportError("install asyncpg for PGCheckpointer")
        self._dsn = dsn
        self._table = table
        self._pool_size = max(1, pool_size)
        self._pool = None

    async def _ensure_pool(self):
        """Lazily create the connection pool and the table."""
        if self._pool is None:
            import asyncpg

            pool = await asyncpg.create_pool(
                self._dsn, min_size=1, max_size=self._pool_size
            )
            async with pool.acquire() as conn:
                await conn.execute(
                    f"""
                    CREATE TABLE IF NOT EXISTS {self._table} (
                        owner TEXT NOT NULL DEFAULT 'default',
                        checkpoint_id TEXT NOT NULL,
                        state JSONB NOT NULL,
                        next_node_id TEXT,
                        iteration INTEGER NOT NULL,
                        updated_at DOUBLE PRECISION,
                        PRIMARY KEY (owner, checkpoint_id)
                    )
                    """
                )
            self._pool = pool
        return self._pool

    async def close(self) -> None:
        """Close the connection pool (idempotent)."""
        pool, self._pool = self._pool, None
        if pool is not None:
            await pool.close()

    async def save(
        self,
        checkpoint_id: str,
        checkpoint: Checkpoint,
        *,
        owner: str = DEFAULT_OWNER,
    ) -> None:
        pool = await self._ensure_pool()
        async with pool.acquire() as conn:
            await conn.execute(
                f"""
                INSERT INTO {self._table} (owner, checkpoint_id, state, next_node_id, iteration, updated_at)
                VALUES ($1, $2, $3::jsonb, $4, $5, $6)
                ON CONFLICT(owner, checkpoint_id) DO UPDATE SET
                    state = EXCLUDED.state,
                    next_node_id = EXCLUDED.next_node_id,
                    iteration = EXCLUDED.iteration,
                    updated_at = EXCLUDED.updated_at
                """,
                owner,
                checkpoint_id,
                json.dumps(checkpoint.state, ensure_ascii=False),
                checkpoint.next_node_id,
                checkpoint.iteration,
                time.time(),
            )

    async def load(
        self, checkpoint_id: str, *, owner: str = DEFAULT_OWNER
    ) -> Checkpoint | None:
        pool = await self._ensure_pool()
        async with pool.acquire() as conn:
            row = await conn.fetchrow(
                f"SELECT state, next_node_id, iteration FROM {self._table} "
                f"WHERE owner = $1 AND checkpoint_id = $2",
                owner,
                checkpoint_id,
            )
            if row is None:
                return None
            return Checkpoint(
                state=json.loads(row["state"]),
                next_node_id=row["next_node_id"],
                iteration=row["iteration"],
            )

    async def delete(self, checkpoint_id: str, *, owner: str = DEFAULT_OWNER) -> None:
        pool = await self._ensure_pool()
        async with pool.acquire() as conn:
            await conn.execute(
                f"DELETE FROM {self._table} WHERE owner = $1 AND checkpoint_id = $2",
                owner,
                checkpoint_id,
            )

    async def list(self, owner: str = DEFAULT_OWNER) -> list[str]:
        """Return all checkpoint IDs persisted for *owner*."""
        pool = await self._ensure_pool()
        async with pool.acquire() as conn:
            rows = await conn.fetch(
                f"SELECT checkpoint_id FROM {self._table} "
                f"WHERE owner = $1 ORDER BY checkpoint_id",
                owner,
            )
            return [r["checkpoint_id"] for r in rows]

    async def cleanup(
        self,
        *,
        owner: str | None = None,
        max_age: float | None = None,
        keep_last: int | None = None,
    ) -> int:
        """Delete stale checkpoints; returns how many were removed."""
        if max_age is None and keep_last is None:
            return 0
        removed = 0
        now = time.time()
        pool = await self._ensure_pool()
        async with pool.acquire() as conn:
            if owner is not None:
                owners = [owner]
            else:
                rows = await conn.fetch(f"SELECT DISTINCT owner FROM {self._table}")
                owners = [r["owner"] for r in rows]
            for own in owners:
                if max_age is not None:
                    result = await conn.execute(
                        f"DELETE FROM {self._table} WHERE owner = $1 AND "
                        f"COALESCE(updated_at, 0) < $2",
                        own,
                        now - max_age,
                    )
                    tag = result.split(" ", 1)
                    removed += int(tag[1]) if len(tag) == 2 else 0
                if keep_last is not None:
                    stale = await conn.fetch(
                        f"SELECT checkpoint_id FROM {self._table} WHERE owner = $1 "
                        f"ORDER BY COALESCE(updated_at, 0) DESC OFFSET $2",
                        own,
                        keep_last,
                    )
                    for row in stale:
                        await conn.execute(
                            f"DELETE FROM {self._table} "
                            f"WHERE owner = $1 AND checkpoint_id = $2",
                            own,
                            row["checkpoint_id"],
                        )
                        removed += 1
        return removed

cleanup async 

cleanup(*, owner=None, max_age=None, keep_last=None)

Delete stale checkpoints; returns how many were removed.

Source code in teff/checkpoint/pg.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
async def cleanup(
    self,
    *,
    owner: str | None = None,
    max_age: float | None = None,
    keep_last: int | None = None,
) -> int:
    """Delete stale checkpoints; returns how many were removed."""
    if max_age is None and keep_last is None:
        return 0
    removed = 0
    now = time.time()
    pool = await self._ensure_pool()
    async with pool.acquire() as conn:
        if owner is not None:
            owners = [owner]
        else:
            rows = await conn.fetch(f"SELECT DISTINCT owner FROM {self._table}")
            owners = [r["owner"] for r in rows]
        for own in owners:
            if max_age is not None:
                result = await conn.execute(
                    f"DELETE FROM {self._table} WHERE owner = $1 AND "
                    f"COALESCE(updated_at, 0) < $2",
                    own,
                    now - max_age,
                )
                tag = result.split(" ", 1)
                removed += int(tag[1]) if len(tag) == 2 else 0
            if keep_last is not None:
                stale = await conn.fetch(
                    f"SELECT checkpoint_id FROM {self._table} WHERE owner = $1 "
                    f"ORDER BY COALESCE(updated_at, 0) DESC OFFSET $2",
                    own,
                    keep_last,
                )
                for row in stale:
                    await conn.execute(
                        f"DELETE FROM {self._table} "
                        f"WHERE owner = $1 AND checkpoint_id = $2",
                        own,
                        row["checkpoint_id"],
                    )
                    removed += 1
    return removed

close async 

close()

Close the connection pool (idempotent).

Source code in teff/checkpoint/pg.py 
58
59
60
61
62
async def close(self) -> None:
    """Close the connection pool (idempotent)."""
    pool, self._pool = self._pool, None
    if pool is not None:
        await pool.close()

list async 

list(owner=DEFAULT_OWNER)

Return all checkpoint IDs persisted for owner.

Source code in teff/checkpoint/pg.py 
119
120
121
122
123
124
125
126
127
128
async def list(self, owner: str = DEFAULT_OWNER) -> list[str]:
    """Return all checkpoint IDs persisted for *owner*."""
    pool = await self._ensure_pool()
    async with pool.acquire() as conn:
        rows = await conn.fetch(
            f"SELECT checkpoint_id FROM {self._table} "
            f"WHERE owner = $1 ORDER BY checkpoint_id",
            owner,
        )
        return [r["checkpoint_id"] for r in rows]

PGHistoryCheckpointer

Bases: _HistoryMixin, PGCheckpointer

PostgreSQL checkpointer that also keeps the full per-step history.

Requires asyncpg (install via teff[pg-checkpoint]). Mirrors :class:PGCheckpointer but appends every save to a checkpoint_history table, exposing history / load_at for time travel in production.

Parameters:

Name Type Description Default
dsn str

PostgreSQL connection string.

 required
table str

Table name for the current checkpoints (default "checkpoints"); the history table is checkpoint_history.

 'checkpoints'
Source code in teff/checkpoint/history.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
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
248
249
250
251
252
253
254
255
256
257
258
class PGHistoryCheckpointer(_HistoryMixin, PGCheckpointer):
    """PostgreSQL checkpointer that also keeps the full per-step history.

    Requires ``asyncpg`` (install via ``teff[pg-checkpoint]``).  Mirrors
    :class:`PGCheckpointer` but appends every ``save`` to a
    ``checkpoint_history`` table, exposing ``history`` / ``load_at`` for
    time travel in production.

    Args:
        dsn: PostgreSQL connection string.
        table: Table name for the *current* checkpoints (default
            ``"checkpoints"``); the history table is ``checkpoint_history``.
    """

    async def _ensure_pool(self):
        """Create the pool plus the history table alongside the base tables."""
        pool = await super()._ensure_pool()
        async with pool.acquire() as conn:
            await conn.execute(_HISTORY_DDL_PG)
        return pool

    async def _history_insert(
        self, owner: str, checkpoint_id: str, checkpoint: Checkpoint
    ) -> None:
        pool = await self._ensure_pool()
        async with pool.acquire() as conn:
            await conn.execute(
                """
                INSERT INTO checkpoint_history
                (owner, checkpoint_id, iteration, state, next_node_id)
                VALUES ($1, $2, $3, $4::jsonb, $5)
                ON CONFLICT(owner, checkpoint_id, iteration) DO UPDATE SET
                    state = EXCLUDED.state,
                    next_node_id = EXCLUDED.next_node_id
                """,
                owner,
                checkpoint_id,
                checkpoint.iteration,
                json.dumps(checkpoint.state, ensure_ascii=False),
                checkpoint.next_node_id,
            )

    async def _history_rows(
        self, owner: str, checkpoint_id: str
    ) -> list[tuple[int, str | None]]:
        pool = await self._ensure_pool()
        async with pool.acquire() as conn:
            rows = await conn.fetch(
                "SELECT iteration, next_node_id FROM checkpoint_history "
                "WHERE owner = $1 AND checkpoint_id = $2 ORDER BY iteration",
                owner,
                checkpoint_id,
            )
            return [(r["iteration"], r["next_node_id"]) for r in rows]

    async def _history_row_at(
        self, owner: str, checkpoint_id: str, iteration: int
    ) -> tuple | None:
        pool = await self._ensure_pool()
        async with pool.acquire() as conn:
            row = await conn.fetchrow(
                "SELECT state, next_node_id FROM checkpoint_history "
                "WHERE owner = $1 AND checkpoint_id = $2 AND iteration = $3",
                owner,
                checkpoint_id,
                iteration,
            )
            if row is None:
                return None
            return (row["state"], row["next_node_id"])

SQLiteCheckpointer

Bases: Checkpointer

Store checkpoints in a SQLite database.

Uses one row per (owner, checkpoint_id) pair — a composite primary key, so the same ID can belong to different owners (users/tenants) without colliding. Each save is a single INSERT .. ON CONFLICT REPLACE transaction, so a crash leaves either the old or the new row, never a mix. Existing single-owner databases are migrated in place: their rows move under :data:~teff.checkpoint.DEFAULT_OWNER, and an updated_at column is added for TTL cleanup.

All database work runs in a worker thread (asyncio.to_thread) behind a lock, so checkpoint saves never block the event loop — important when many parallel branches checkpoint through the same store.

Parameters:

Name Type Description Default
path str

Path to the SQLite database file.

 required

Methods:

Name Description
cleanup

Delete stale checkpoints; returns how many were removed.
close

Close the underlying SQLite connection.
list

Return all checkpoint IDs persisted for owner.
Source code in teff/checkpoint/sqlite.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
class SQLiteCheckpointer(Checkpointer):
    """Store checkpoints in a SQLite database.

    Uses one row per ``(owner, checkpoint_id)`` pair — a composite primary
    key, so the same ID can belong to different owners (users/tenants)
    without colliding.  Each ``save`` is a single ``INSERT .. ON CONFLICT
    REPLACE`` transaction, so a crash leaves either the old or the new row,
    never a mix.  Existing single-owner databases are migrated in place:
    their rows move under :data:`~teff.checkpoint.DEFAULT_OWNER`, and an
    ``updated_at`` column is added for TTL cleanup.

    All database work runs in a worker thread (``asyncio.to_thread``) behind
    a lock, so checkpoint saves never block the event loop — important when
    many parallel branches checkpoint through the same store.

    Args:
        path: Path to the SQLite database file.
    """

    def __init__(self, path: str):
        self._path = path
        Path(path).parent.mkdir(parents=True, exist_ok=True)
        self._lock = threading.Lock()
        self._conn = sqlite3.connect(path, check_same_thread=False)
        self._migrate()
        self._conn.execute(
            """
            CREATE TABLE IF NOT EXISTS checkpoints (
                owner TEXT NOT NULL DEFAULT 'default',
                checkpoint_id TEXT NOT NULL,
                state TEXT NOT NULL,
                next_node_id TEXT,
                iteration INTEGER NOT NULL,
                updated_at REAL,
                PRIMARY KEY (owner, checkpoint_id)
            )
            """
        )
        self._conn.commit()

    def _migrate(self) -> None:
        """Migrate a legacy single-owner table to the owner-scoped schema."""
        row = self._conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table' AND name='checkpoints'"
        ).fetchone()
        if row is None:
            return
        cols = [r[1] for r in self._conn.execute("PRAGMA table_info(checkpoints)")]
        if "owner" not in cols:
            self._conn.execute("ALTER TABLE checkpoints RENAME TO checkpoints_legacy")
            self._conn.execute(
                """
                CREATE TABLE checkpoints (
                    owner TEXT NOT NULL DEFAULT 'default',
                    checkpoint_id TEXT NOT NULL,
                    state TEXT NOT NULL,
                    next_node_id TEXT,
                    iteration INTEGER NOT NULL,
                    updated_at REAL,
                    PRIMARY KEY (owner, checkpoint_id)
                )
                """
            )
            self._conn.execute(
                """
                INSERT INTO checkpoints (owner, checkpoint_id, state, next_node_id, iteration)
                SELECT 'default', checkpoint_id, state, next_node_id, iteration
                FROM checkpoints_legacy
                """
            )
            self._conn.execute("DROP TABLE checkpoints_legacy")
            self._conn.commit()
            return
        if "updated_at" not in cols:
            self._conn.execute("ALTER TABLE checkpoints ADD COLUMN updated_at REAL")
            self._conn.commit()

    def close(self) -> None:
        """Close the underlying SQLite connection."""
        with self._lock:
            self._conn.close()

    async def _run(self, fn, *args, **kwargs):
        """Run a sync DB call in a worker thread, serialised by the lock."""

        def _call():
            with self._lock:
                return fn(*args, **kwargs)

        return await asyncio.to_thread(_call)

    async def save(
        self,
        checkpoint_id: str,
        checkpoint: Checkpoint,
        *,
        owner: str = DEFAULT_OWNER,
    ) -> None:
        def _save():
            self._conn.execute(
                """
                INSERT INTO checkpoints (owner, checkpoint_id, state, next_node_id, iteration, updated_at)
                VALUES (?, ?, ?, ?, ?, ?)
                ON CONFLICT(owner, checkpoint_id) DO UPDATE SET
                    state = excluded.state,
                    next_node_id = excluded.next_node_id,
                    iteration = excluded.iteration,
                    updated_at = excluded.updated_at
                """,
                (
                    owner,
                    checkpoint_id,
                    json.dumps(checkpoint.state, ensure_ascii=False),
                    checkpoint.next_node_id,
                    checkpoint.iteration,
                    time.time(),
                ),
            )
            self._conn.commit()

        await self._run(_save)

    async def load(
        self, checkpoint_id: str, *, owner: str = DEFAULT_OWNER
    ) -> Checkpoint | None:
        def _load():
            return self._conn.execute(
                "SELECT state, next_node_id, iteration FROM checkpoints "
                "WHERE owner = ? AND checkpoint_id = ?",
                (owner, checkpoint_id),
            ).fetchone()

        row = await self._run(_load)
        if row is None:
            return None
        return Checkpoint(
            state=json.loads(row[0]),
            next_node_id=row[1],
            iteration=row[2],
        )

    async def delete(self, checkpoint_id: str, *, owner: str = DEFAULT_OWNER) -> None:
        def _delete():
            self._conn.execute(
                "DELETE FROM checkpoints WHERE owner = ? AND checkpoint_id = ?",
                (owner, checkpoint_id),
            )
            self._conn.commit()

        await self._run(_delete)

    async def list(self, owner: str = DEFAULT_OWNER) -> list[str]:
        """Return all checkpoint IDs persisted for *owner*."""

        def _list():
            return self._conn.execute(
                "SELECT checkpoint_id FROM checkpoints WHERE owner = ? ORDER BY checkpoint_id",
                (owner,),
            ).fetchall()

        rows = await self._run(_list)
        return [r[0] for r in rows]

    async def cleanup(
        self,
        *,
        owner: str | None = None,
        max_age: float | None = None,
        keep_last: int | None = None,
    ) -> int:
        """Delete stale checkpoints; returns how many were removed."""

        def _cleanup():
            if max_age is None and keep_last is None:
                return 0
            removed = 0
            now = time.time()
            if owner is not None:
                owners = [owner]
            else:
                owners = [
                    r[0]
                    for r in self._conn.execute(
                        "SELECT DISTINCT owner FROM checkpoints"
                    ).fetchall()
                ]
            for own in owners:
                if max_age is not None:
                    cur = self._conn.execute(
                        "DELETE FROM checkpoints WHERE owner = ? AND "
                        "COALESCE(updated_at, 0) < ?",
                        (own, now - max_age),
                    )
                    removed += cur.rowcount
                if keep_last is not None:
                    stale = [
                        r[0]
                        for r in self._conn.execute(
                            "SELECT checkpoint_id FROM checkpoints WHERE owner = ? "
                            "ORDER BY COALESCE(updated_at, 0) DESC LIMIT -1 OFFSET ?",
                            (own, keep_last),
                        ).fetchall()
                    ]
                    for cid in stale:
                        self._conn.execute(
                            "DELETE FROM checkpoints WHERE owner = ? AND checkpoint_id = ?",
                            (own, cid),
                        )
                        removed += 1
            self._conn.commit()
            return removed

        return await self._run(_cleanup)

cleanup async 

cleanup(*, owner=None, max_age=None, keep_last=None)

Delete stale checkpoints; returns how many were removed.

Source code in teff/checkpoint/sqlite.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
async def cleanup(
    self,
    *,
    owner: str | None = None,
    max_age: float | None = None,
    keep_last: int | None = None,
) -> int:
    """Delete stale checkpoints; returns how many were removed."""

    def _cleanup():
        if max_age is None and keep_last is None:
            return 0
        removed = 0
        now = time.time()
        if owner is not None:
            owners = [owner]
        else:
            owners = [
                r[0]
                for r in self._conn.execute(
                    "SELECT DISTINCT owner FROM checkpoints"
                ).fetchall()
            ]
        for own in owners:
            if max_age is not None:
                cur = self._conn.execute(
                    "DELETE FROM checkpoints WHERE owner = ? AND "
                    "COALESCE(updated_at, 0) < ?",
                    (own, now - max_age),
                )
                removed += cur.rowcount
            if keep_last is not None:
                stale = [
                    r[0]
                    for r in self._conn.execute(
                        "SELECT checkpoint_id FROM checkpoints WHERE owner = ? "
                        "ORDER BY COALESCE(updated_at, 0) DESC LIMIT -1 OFFSET ?",
                        (own, keep_last),
                    ).fetchall()
                ]
                for cid in stale:
                    self._conn.execute(
                        "DELETE FROM checkpoints WHERE owner = ? AND checkpoint_id = ?",
                        (own, cid),
                    )
                    removed += 1
        self._conn.commit()
        return removed

    return await self._run(_cleanup)

close 

close()

Close the underlying SQLite connection.

Source code in teff/checkpoint/sqlite.py 
90
91
92
93
def close(self) -> None:
    """Close the underlying SQLite connection."""
    with self._lock:
        self._conn.close()

list async 

list(owner=DEFAULT_OWNER)

Return all checkpoint IDs persisted for owner.

Source code in teff/checkpoint/sqlite.py 
164
165
166
167
168
169
170
171
172
173
174
async def list(self, owner: str = DEFAULT_OWNER) -> list[str]:
    """Return all checkpoint IDs persisted for *owner*."""

    def _list():
        return self._conn.execute(
            "SELECT checkpoint_id FROM checkpoints WHERE owner = ? ORDER BY checkpoint_id",
            (owner,),
        ).fetchall()

    rows = await self._run(_list)
    return [r[0] for r in rows]

SQLiteHistoryCheckpointer

Bases: _HistoryMixin, SQLiteCheckpointer

SQLite checkpointer that also keeps the full per-step history.

A drop-in for :class:teff.checkpoint.SQLiteCheckpointer that, on every save, additionally appends the snapshot to a checkpoint_history table — so the current checkpoint can be overwritten without losing the earlier ones. history / load_at expose the timeline for time travel.

Parameters:

Name Type Description Default
path str

Path to the SQLite database file.

 required
Source code in teff/checkpoint/history.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
class SQLiteHistoryCheckpointer(_HistoryMixin, SQLiteCheckpointer):
    """SQLite checkpointer that also keeps the full per-step history.

    A drop-in for :class:`teff.checkpoint.SQLiteCheckpointer` that, on every
    ``save``, additionally appends the snapshot to a ``checkpoint_history``
    table — so the current checkpoint can be overwritten without losing the
    earlier ones.  ``history`` / ``load_at`` expose the timeline for time
    travel.

    Args:
        path: Path to the SQLite database file.
    """

    def __init__(self, path: str):
        super().__init__(path)
        self._history_ensure()

    def _history_ensure(self) -> None:
        with self._lock:
            self._conn.execute(_HISTORY_DDL)
            self._conn.commit()

    async def _history_insert(
        self, owner: str, checkpoint_id: str, checkpoint: Checkpoint
    ) -> None:
        def _insert():
            with self._lock:
                self._conn.execute(
                    "INSERT OR REPLACE INTO checkpoint_history "
                    "(owner, checkpoint_id, iteration, state, next_node_id) VALUES (?, ?, ?, ?, ?)",
                    (
                        owner,
                        checkpoint_id,
                        checkpoint.iteration,
                        json.dumps(checkpoint.state, ensure_ascii=False),
                        checkpoint.next_node_id,
                    ),
                )
                self._conn.commit()

        import asyncio

        await asyncio.to_thread(_insert)

    async def _history_rows(
        self, owner: str, checkpoint_id: str
    ) -> list[tuple[int, str | None]]:
        def _rows():
            with self._lock:
                rows = self._conn.execute(
                    "SELECT iteration, next_node_id FROM checkpoint_history "
                    "WHERE owner = ? AND checkpoint_id = ? ORDER BY iteration",
                    (owner, checkpoint_id),
                ).fetchall()
                return [(r[0], r[1]) for r in rows]

        import asyncio

        return await asyncio.to_thread(_rows)

    async def _history_row_at(
        self, owner: str, checkpoint_id: str, iteration: int
    ) -> tuple | None:
        def _row_at():
            with self._lock:
                return self._conn.execute(
                    "SELECT state, next_node_id FROM checkpoint_history "
                    "WHERE owner = ? AND checkpoint_id = ? AND iteration = ?",
                    (owner, checkpoint_id, iteration),
                ).fetchone()

        import asyncio

        return await asyncio.to_thread(_row_at)

checkpoint_from_dict 

checkpoint_from_dict(data)

Reconstruct a checkpoint from a dict produced by :func:checkpoint_to_dict.

Source code in teff/checkpoint/base.py 
 94
 95
 96
 97
 98
 99
100
def checkpoint_from_dict(data: dict[str, Any]) -> Checkpoint:
    """Reconstruct a checkpoint from a dict produced by :func:`checkpoint_to_dict`."""
    return Checkpoint(
        state=data.get("state", {}),
        next_node_id=data.get("next_node_id"),
        iteration=data.get("iteration", 0),
    )

checkpoint_to_dict 

checkpoint_to_dict(cp)

Convert a checkpoint to a JSON-serializable dict.

Source code in teff/checkpoint/base.py 
85
86
87
88
89
90
91
def checkpoint_to_dict(cp: Checkpoint) -> dict[str, Any]:
    """Convert a checkpoint to a JSON-serializable dict."""
    return {
        "state": cp.state,
        "next_node_id": cp.next_node_id,
        "iteration": cp.iteration,
    }