Skip to content

Checkpointing

The checkpoint model and the checkpointer implementations. See the Checkpointing guide.

agentflow.checkpoint

Checkpointing: durable run state, resume, and time-travel.

Checkpoint dataclass

One persisted point in a run's history.

next is the frontier of node names to run when resuming. An empty next with interrupted=False means the run finished. When interrupted is True the run is suspended: interrupt_node requested it and interrupt_payload is what a human must resolve; next still holds the frontier to resume once an answer is supplied.

A suspend has a kind (:attr:suspend_kind). "human" waits for an external :meth:resume with a value (human-in-the-loop). "timer" waits only for the clock: :attr:wake_at is the earliest ISO time the run should be resumed, and resuming re-runs the suspending node (which polls again). Both share the interrupt machinery — a timer wait is an interrupt whose resolver is the clock, not a person.

Source code in agentflow/checkpoint/base.py 
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
@dataclass(frozen=True, slots=True)
class Checkpoint:
    """One persisted point in a run's history.

    ``next`` is the frontier of node names to run when resuming. An empty
    ``next`` with ``interrupted=False`` means the run finished. When
    ``interrupted`` is True the run is suspended: ``interrupt_node`` requested
    it and ``interrupt_payload`` is what a human must resolve; ``next`` still
    holds the frontier to resume once an answer is supplied.

    A suspend has a *kind* (:attr:`suspend_kind`). ``"human"`` waits for an
    external :meth:`resume` with a value (human-in-the-loop). ``"timer"`` waits
    only for the clock: :attr:`wake_at` is the earliest ISO time the run should
    be resumed, and resuming re-runs the suspending node (which polls again).
    Both share the interrupt machinery — a timer wait is an interrupt whose
    resolver is the clock, not a person.
    """

    thread: str
    step: int
    state: Mapping[str, Any]
    next: tuple[str, ...] = ()
    parent: str | None = None
    ts: str = ""
    interrupted: bool = False
    interrupt_node: str | None = None
    interrupt_payload: Any = None
    #: ISO-8601 UTC time the run should wake, for a ``"timer"`` suspend; None
    #: for a ``"human"`` suspend (which waits for an explicit resume value).
    wake_at: str | None = None
    extra: dict[str, Any] = field(default_factory=dict)
    #: Monotonic write count for this (thread, step). Set by the store on read;
    #: pass the value you read back as ``put(..., if_revision=)`` to guard a
    #: read-modify-write against a concurrent resume (optimistic concurrency).
    #: 0 means "never written" / not tracked by this backend.
    revision: int = 0

    @property
    def done(self) -> bool:
        return not self.next and not self.interrupted

    @property
    def suspend_kind(self) -> str | None:
        """``"human"`` or ``"timer"`` while suspended, else None.

        Derived so the on-disk shape stays compatible: a suspend with
        ``wake_at`` set is a timer wait; a suspend without it waits for a human.
        """
        if not self.interrupted:
            return None
        return "timer" if self.wake_at is not None else "human"

suspend_kind property 

suspend_kind: str | None

"human" or "timer" while suspended, else None.

Derived so the on-disk shape stays compatible: a suspend with wake_at set is a timer wait; a suspend without it waits for a human.

Checkpointer

Bases: Protocol

Persists and retrieves checkpoints by thread and step.

Source code in agentflow/checkpoint/base.py 
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
@runtime_checkable
class Checkpointer(Protocol):
    """Persists and retrieves checkpoints by thread and step."""

    async def put(self, cp: Checkpoint, *, if_revision: int | None = None) -> str:
        """Persist ``cp`` and return an id (thread + step is sufficient).

        When ``if_revision`` is given, the write is conditional: it succeeds
        only if the currently stored revision for ``(cp.thread, cp.step)``
        equals ``if_revision`` (or nothing is stored and ``if_revision`` is 0),
        otherwise it raises :class:`~agentflow.errors.CheckpointConflict`. This
        gives optimistic concurrency for two resumes racing the same step.
        ``None`` (default) is an unconditional upsert.
        """
        ...

    async def get(self, thread: str, step: int | None = None) -> Checkpoint | None:
        """Return the checkpoint at ``step``, or the latest when ``step`` is
        None. Returns None if the thread has no checkpoints. The returned
        checkpoint's ``revision`` reflects how many times that step was
        written (0 if the backend doesn't track it)."""
        ...

    def history(self, thread: str) -> AsyncIterator[Checkpoint]:
        """Yield every checkpoint for ``thread`` in ascending step order."""
        ...

    async def list_threads(self, *, limit: int = 100, offset: int = 0) -> list[ThreadInfo]:
        """Return a summary of each stored thread (latest step + status),
        ordered by thread id. Enables control-plane enumeration of runs
        without reading full state. ``limit``/``offset`` paginate."""
        ...

    async def delete_thread(self, thread: str) -> None:
        """Remove all checkpoints for ``thread`` (retention/cleanup)."""
        ...

    async def prune(
        self, thread: str, *, before_step: int | None = None, older_than: str | None = None
    ) -> int:
        """Delete checkpoints for ``thread`` older than a bound; return the
        count removed. ``before_step`` deletes steps strictly less than it;
        ``older_than`` is an ISO timestamp compared against each checkpoint's
        ``ts``. At least one bound should be given."""
        ...

put async 

put(cp: Checkpoint, *, if_revision: int | None = None) -> str

Persist cp and return an id (thread + step is sufficient).

When if_revision is given, the write is conditional: it succeeds only if the currently stored revision for (cp.thread, cp.step) equals if_revision (or nothing is stored and if_revision is 0), otherwise it raises :class:~agentflow.errors.CheckpointConflict. This gives optimistic concurrency for two resumes racing the same step. None (default) is an unconditional upsert.

Source code in agentflow/checkpoint/base.py 
106
107
108
109
110
111
112
113
114
115
116
async def put(self, cp: Checkpoint, *, if_revision: int | None = None) -> str:
    """Persist ``cp`` and return an id (thread + step is sufficient).

    When ``if_revision`` is given, the write is conditional: it succeeds
    only if the currently stored revision for ``(cp.thread, cp.step)``
    equals ``if_revision`` (or nothing is stored and ``if_revision`` is 0),
    otherwise it raises :class:`~agentflow.errors.CheckpointConflict`. This
    gives optimistic concurrency for two resumes racing the same step.
    ``None`` (default) is an unconditional upsert.
    """
    ...

get async 

get(thread: str, step: int | None = None) -> Checkpoint | None

Return the checkpoint at step, or the latest when step is None. Returns None if the thread has no checkpoints. The returned checkpoint's revision reflects how many times that step was written (0 if the backend doesn't track it).

Source code in agentflow/checkpoint/base.py 
118
119
120
121
122
123
async def get(self, thread: str, step: int | None = None) -> Checkpoint | None:
    """Return the checkpoint at ``step``, or the latest when ``step`` is
    None. Returns None if the thread has no checkpoints. The returned
    checkpoint's ``revision`` reflects how many times that step was
    written (0 if the backend doesn't track it)."""
    ...

history 

history(thread: str) -> AsyncIterator[Checkpoint]

Yield every checkpoint for thread in ascending step order.

Source code in agentflow/checkpoint/base.py 
125
126
127
def history(self, thread: str) -> AsyncIterator[Checkpoint]:
    """Yield every checkpoint for ``thread`` in ascending step order."""
    ...

list_threads async 

list_threads(*, limit: int = 100, offset: int = 0) -> list[ThreadInfo]

Return a summary of each stored thread (latest step + status), ordered by thread id. Enables control-plane enumeration of runs without reading full state. limit/offset paginate.

Source code in agentflow/checkpoint/base.py 
129
130
131
132
133
async def list_threads(self, *, limit: int = 100, offset: int = 0) -> list[ThreadInfo]:
    """Return a summary of each stored thread (latest step + status),
    ordered by thread id. Enables control-plane enumeration of runs
    without reading full state. ``limit``/``offset`` paginate."""
    ...

delete_thread async 

delete_thread(thread: str) -> None

Remove all checkpoints for thread (retention/cleanup).

Source code in agentflow/checkpoint/base.py 
135
136
137
async def delete_thread(self, thread: str) -> None:
    """Remove all checkpoints for ``thread`` (retention/cleanup)."""
    ...

prune async 

prune(thread: str, *, before_step: int | None = None, older_than: str | None = None) -> int

Delete checkpoints for thread older than a bound; return the count removed. before_step deletes steps strictly less than it; older_than is an ISO timestamp compared against each checkpoint's ts. At least one bound should be given.

Source code in agentflow/checkpoint/base.py 
139
140
141
142
143
144
145
146
async def prune(
    self, thread: str, *, before_step: int | None = None, older_than: str | None = None
) -> int:
    """Delete checkpoints for ``thread`` older than a bound; return the
    count removed. ``before_step`` deletes steps strictly less than it;
    ``older_than`` is an ISO timestamp compared against each checkpoint's
    ``ts``. At least one bound should be given."""
    ...

ThreadInfo dataclass

A one-line summary of a thread for control-plane listing.

Derived from the thread's latest checkpoint so a caller can enumerate runs (running / interrupted / finished) without reading full state.

Source code in agentflow/checkpoint/base.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
@dataclass(frozen=True, slots=True)
class ThreadInfo:
    """A one-line summary of a thread for control-plane listing.

    Derived from the thread's latest checkpoint so a caller can enumerate runs
    (running / interrupted / finished) without reading full state.
    """

    thread: str
    latest_step: int
    interrupted: bool
    done: bool
    ts: str = ""

MemoryCheckpointer

Keeps checkpoints in a dict keyed by thread. Not durable.

Source code in agentflow/checkpoint/memory.py 
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
class MemoryCheckpointer:
    """Keeps checkpoints in a dict keyed by thread. Not durable."""

    def __init__(self) -> None:
        self._threads: dict[str, dict[int, Checkpoint]] = {}

    async def put(self, cp: Checkpoint, *, if_revision: int | None = None) -> str:
        steps = self._threads.setdefault(cp.thread, {})
        existing = steps.get(cp.step)
        current_rev = existing.revision if existing else 0
        if if_revision is not None and if_revision != current_rev:
            raise CheckpointConflict(cp.thread, cp.step, if_revision, current_rev)
        steps[cp.step] = dataclasses.replace(cp, revision=current_rev + 1)
        return f"{cp.thread}:{cp.step}"

    async def get(self, thread: str, step: int | None = None) -> Checkpoint | None:
        steps = self._threads.get(thread)
        if not steps:
            return None
        if step is None:
            step = max(steps)
        return steps.get(step)

    async def history(self, thread: str) -> AsyncIterator[Checkpoint]:
        steps = self._threads.get(thread, {})
        for step in sorted(steps):
            yield steps[step]

    async def delete_thread(self, thread: str) -> None:
        self._threads.pop(thread, None)

    async def prune(
        self, thread: str, *, before_step: int | None = None, older_than: str | None = None
    ) -> int:
        steps = self._threads.get(thread)
        if not steps:
            return 0
        to_remove = [
            s
            for s, cp in steps.items()
            if (before_step is not None and s < before_step)
            or (older_than is not None and cp.ts and cp.ts < older_than)
        ]
        for s in to_remove:
            del steps[s]
        return len(to_remove)

    async def list_threads(self, *, limit: int = 100, offset: int = 0) -> list[ThreadInfo]:
        infos: list[ThreadInfo] = []
        for thread in sorted(self._threads):
            steps = self._threads[thread]
            if not steps:
                continue
            cp = steps[max(steps)]
            infos.append(
                ThreadInfo(
                    thread=thread,
                    latest_step=cp.step,
                    interrupted=cp.interrupted,
                    done=cp.done,
                    ts=cp.ts,
                )
            )
        return infos[offset : offset + limit]

FileCheckpointer

Persist checkpoints under root/<thread>/<step>.json.

Single-writer contract: this checkpointer is safe for one process writing a given thread at a time. Individual file writes are atomic (temp + replace), so a reader never sees a partial file, but there is no cross-process lock or revision check — two processes writing the same thread can interleave steps and corrupt history. For multi-process or concurrent-runner durability use :class:~agentflow.checkpoint.sqlite.SqliteCheckpointer, which enforces atomic per-(thread, step) revisions in a transaction.

Files and directories are created owner-only (0o600/0o700) by default so a persisted run's state is not world-readable. Pass secure_permissions= False to disable (e.g. on filesystems that don't support it).

redact optionally masks sensitive values in the persisted state before writing. WARNING: a redacted checkpoint cannot be resumed to the exact original state — masked values are lost. Use redaction only for archival/debug copies, not for the checkpointer you resume from. Default is no redaction.

Source code in agentflow/checkpoint/file.py 
 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
class FileCheckpointer:
    """Persist checkpoints under ``root/<thread>/<step>.json``.

    Single-writer contract: this checkpointer is safe for one process writing a
    given thread at a time. Individual file writes are atomic (temp + replace),
    so a *reader* never sees a partial file, but there is no cross-process lock
    or revision check — two processes writing the same thread can interleave
    steps and corrupt history. For multi-process or concurrent-runner durability
    use :class:`~agentflow.checkpoint.sqlite.SqliteCheckpointer`, which enforces
    atomic per-(thread, step) revisions in a transaction.

    Files and directories are created owner-only (0o600/0o700) by default so a
    persisted run's state is not world-readable. Pass ``secure_permissions=
    False`` to disable (e.g. on filesystems that don't support it).

    ``redact`` optionally masks sensitive values in the persisted ``state``
    before writing. WARNING: a redacted checkpoint cannot be resumed to the
    exact original state — masked values are lost. Use redaction only for
    archival/debug copies, not for the checkpointer you resume from. Default is
    no redaction.
    """

    def __init__(
        self,
        root: Path | str = ".runs",
        *,
        redact: Redactor | None = None,
        secure_permissions: bool = True,
    ) -> None:
        self.root = Path(root)
        self._redact = redact or redact_none
        self._secure = secure_permissions

    def _thread_dir(self, thread: str) -> Path:
        # Keep thread ids filesystem-safe without inventing a scheme: reject
        # separators rather than silently mangling them.
        if "/" in thread or "\\" in thread or thread in ("", ".", ".."):
            raise CheckpointError(f"unsafe thread id {thread!r}")
        return self.root / thread

    async def put(self, cp: Checkpoint, *, if_revision: int | None = None) -> str:
        directory = self._thread_dir(cp.thread)
        directory.mkdir(parents=True, exist_ok=True)
        if self._secure:
            _chmod(directory, 0o700)
            _chmod(self.root, 0o700)
        path = directory / f"{cp.step:06d}.json"
        current = _read_revision(path)
        if if_revision is not None and current != if_revision:
            raise CheckpointConflict(cp.thread, cp.step, if_revision, current)
        payload = to_payload(cp)
        payload["state"] = self._redact(payload["state"])
        payload["interrupt_payload"] = self._redact(payload["interrupt_payload"])
        envelope = {"revision": current + 1, "payload": payload}
        try:
            data = json.dumps(envelope, ensure_ascii=False, indent=2)
        except TypeError as exc:
            raise CheckpointError(
                f"checkpoint state for thread {cp.thread!r} is not JSON-serializable: {exc}"
            ) from exc
        _atomic_write(path, data)
        if self._secure:
            _chmod(path, 0o600)
        return f"{cp.thread}:{cp.step}"

    async def get(self, thread: str, step: int | None = None) -> Checkpoint | None:
        directory = self._thread_dir(thread)
        if not directory.is_dir():
            return None
        files = sorted(directory.glob("*.json"))
        if not files:
            return None
        if step is None:
            target = files[-1]
        else:
            target = directory / f"{step:06d}.json"
            if not target.exists():
                return None
        return _load(target)

    async def history(self, thread: str) -> AsyncIterator[Checkpoint]:
        directory = self._thread_dir(thread)
        if not directory.is_dir():
            return
        for path in sorted(directory.glob("*.json")):
            yield _load(path)

    async def delete_thread(self, thread: str) -> None:
        directory = self._thread_dir(thread)
        if not directory.is_dir():
            return
        for path in directory.glob("*.json"):
            path.unlink(missing_ok=True)
        with contextlib.suppress(OSError):
            directory.rmdir()

    async def prune(
        self, thread: str, *, before_step: int | None = None, older_than: str | None = None
    ) -> int:
        directory = self._thread_dir(thread)
        if not directory.is_dir():
            return 0
        removed = 0
        for path in sorted(directory.glob("*.json")):
            step = int(path.stem)
            if before_step is not None and step >= before_step:
                continue
            if older_than is not None:
                env = json.loads(path.read_text(encoding="utf-8"))
                ts = env.get("payload", env).get("ts", "")
                if ts >= older_than:
                    continue
            path.unlink(missing_ok=True)
            removed += 1
        return removed

    async def list_threads(self, *, limit: int = 100, offset: int = 0) -> list[ThreadInfo]:
        if not self.root.is_dir():
            return []
        infos: list[ThreadInfo] = []
        for directory in sorted(p for p in self.root.iterdir() if p.is_dir()):
            files = sorted(directory.glob("*.json"))
            if not files:
                continue
            cp = _load(files[-1])
            infos.append(
                ThreadInfo(
                    thread=directory.name,
                    latest_step=cp.step,
                    interrupted=cp.interrupted,
                    done=cp.done,
                    ts=cp.ts,
                )
            )
        return infos[offset : offset + limit]

SqliteCheckpointer

Durable checkpointer backed by a SQLite database file.

Source code in agentflow/checkpoint/sqlite.py 
 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
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
259
260
261
class SqliteCheckpointer:
    """Durable checkpointer backed by a SQLite database file."""

    def __init__(
        self,
        path: Path | str = ".runs/checkpoints.db",
        *,
        redact: Redactor | None = None,
    ) -> None:
        self.path = Path(path)
        self._redact = redact or redact_none
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self._init_done = False
        self._lock = asyncio.Lock()

    # ------------------------------------------------------------------

    def _connect(self) -> sqlite3.Connection:
        conn = sqlite3.connect(self.path, isolation_level=None, timeout=30.0)
        conn.execute("PRAGMA journal_mode=WAL")
        conn.execute("PRAGMA busy_timeout=30000")
        return conn

    async def _ensure_schema(self) -> None:
        if self._init_done:
            return
        async with self._lock:
            if self._init_done:
                return
            await asyncio.to_thread(self._init_schema)
            self._init_done = True

    def _init_schema(self) -> None:
        conn = self._connect()
        try:
            conn.executescript(_SCHEMA)
        finally:
            conn.close()

    # ------------------------------------------------------------------

    async def put(self, cp: Checkpoint, *, if_revision: int | None = None) -> str:
        await self._ensure_schema()
        payload = to_payload(cp)
        payload["state"] = self._redact(payload["state"])
        payload["interrupt_payload"] = self._redact(payload["interrupt_payload"])
        try:
            blob = json.dumps(payload, ensure_ascii=False)
        except TypeError as exc:
            raise CheckpointError(
                f"checkpoint state for thread {cp.thread!r} is not JSON-serializable: {exc}"
            ) from exc
        await asyncio.to_thread(self._put_row, cp.thread, cp.step, blob, cp.ts, if_revision)
        return f"{cp.thread}:{cp.step}"

    def _put_row(self, thread: str, step: int, blob: str, ts: str, if_revision: int | None) -> None:
        conn = self._connect()
        try:
            # One transaction: check-and-set the revision, then upsert with an
            # atomic revision bump. A re-run of the same step (resume)
            # overwrites and increments revision.
            conn.execute("BEGIN IMMEDIATE")
            if if_revision is not None:
                cur = conn.execute(
                    "SELECT revision FROM checkpoints WHERE thread=? AND step=?",
                    (thread, step),
                )
                row = cur.fetchone()
                current = row[0] if row else 0
                if current != if_revision:
                    conn.execute("ROLLBACK")
                    raise CheckpointConflict(thread, step, if_revision, current)
            conn.execute(
                """
                INSERT INTO checkpoints (thread, step, revision, payload, ts)
                VALUES (?, ?, 1, ?, ?)
                ON CONFLICT(thread, step) DO UPDATE SET
                    revision = revision + 1,
                    payload = excluded.payload,
                    ts = excluded.ts
                """,
                (thread, step, blob, ts),
            )
            conn.execute("COMMIT")
        except sqlite3.Error as exc:
            conn.execute("ROLLBACK")
            raise CheckpointError(f"sqlite put failed: {exc}") from exc
        finally:
            conn.close()

    async def get(self, thread: str, step: int | None = None) -> Checkpoint | None:
        await self._ensure_schema()
        row = await asyncio.to_thread(self._get_row, thread, step)
        if row is None:
            return None
        blob, revision = row
        return dataclasses.replace(from_payload(json.loads(blob)), revision=revision)

    def _get_row(self, thread: str, step: int | None) -> tuple[str, int] | None:
        conn = self._connect()
        try:
            if step is None:
                cur = conn.execute(
                    "SELECT payload, revision FROM checkpoints WHERE thread=? "
                    "ORDER BY step DESC LIMIT 1",
                    (thread,),
                )
            else:
                cur = conn.execute(
                    "SELECT payload, revision FROM checkpoints WHERE thread=? AND step=?",
                    (thread, step),
                )
            row = cur.fetchone()
            return (row[0], row[1]) if row else None
        finally:
            conn.close()

    async def history(self, thread: str) -> AsyncIterator[Checkpoint]:
        await self._ensure_schema()
        rows = await asyncio.to_thread(self._history_rows, thread)
        for blob, revision in rows:
            yield dataclasses.replace(from_payload(json.loads(blob)), revision=revision)

    def _history_rows(self, thread: str) -> list[tuple[str, int]]:
        conn = self._connect()
        try:
            cur = conn.execute(
                "SELECT payload, revision FROM checkpoints WHERE thread=? ORDER BY step ASC",
                (thread,),
            )
            return [(r[0], r[1]) for r in cur.fetchall()]
        finally:
            conn.close()

    async def delete_thread(self, thread: str) -> None:
        await self._ensure_schema()
        await asyncio.to_thread(
            self._exec_write, "DELETE FROM checkpoints WHERE thread=?", (thread,)
        )

    async def prune(
        self, thread: str, *, before_step: int | None = None, older_than: str | None = None
    ) -> int:
        await self._ensure_schema()
        return await asyncio.to_thread(self._prune_rows, thread, before_step, older_than)

    def _exec_write(self, sql: str, params: tuple) -> int:
        conn = self._connect()
        try:
            cur = conn.execute(sql, params)
            return cur.rowcount
        finally:
            conn.close()

    def _prune_rows(self, thread: str, before_step: int | None, older_than: str | None) -> int:
        clauses = ["thread=?"]
        params: list[Any] = [thread]
        if before_step is not None:
            clauses.append("step < ?")
            params.append(before_step)
        if older_than is not None:
            clauses.append("ts < ?")
            params.append(older_than)
        sql = f"DELETE FROM checkpoints WHERE {' AND '.join(clauses)}"
        return self._exec_write(sql, tuple(params))

    async def list_threads(self, *, limit: int = 100, offset: int = 0) -> list[ThreadInfo]:
        await self._ensure_schema()
        rows = await asyncio.to_thread(self._list_thread_rows, limit, offset)
        infos: list[ThreadInfo] = []
        for thread, step, blob in rows:
            cp = from_payload(json.loads(blob))
            infos.append(
                ThreadInfo(
                    thread=thread,
                    latest_step=step,
                    interrupted=cp.interrupted,
                    done=cp.done,
                    ts=cp.ts,
                )
            )
        return infos

    def _list_thread_rows(self, limit: int, offset: int) -> list[tuple[str, int, str]]:
        conn = self._connect()
        try:
            # Latest row per thread: join each thread's max(step) back to payload.
            cur = conn.execute(
                """
                SELECT c.thread, c.step, c.payload
                FROM checkpoints c
                JOIN (
                    SELECT thread, MAX(step) AS step FROM checkpoints GROUP BY thread
                ) m ON c.thread = m.thread AND c.step = m.step
                ORDER BY c.thread ASC
                LIMIT ? OFFSET ?
                """,
                (limit, offset),
            )
            return [(r[0], r[1], r[2]) for r in cur.fetchall()]
        finally:
            conn.close()

PostgresCheckpointer

Durable checkpointer backed by a PostgreSQL table.

Pass either a dsn (postgresql://user:pw@host:port/db) or an existing asyncpg pool via pool. table names the checkpoints table (it is created on first use). redact optionally masks sensitive values in the persisted state (see :class:~agentflow.checkpoint.file.FileCheckpointer for the resume-fidelity caveat).

Source code in agentflow/checkpoint/postgres.py 
 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
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
class PostgresCheckpointer:
    """Durable checkpointer backed by a PostgreSQL table.

    Pass either a ``dsn`` (``postgresql://user:pw@host:port/db``) or an existing
    ``asyncpg`` pool via ``pool``. ``table`` names the checkpoints table (it is
    created on first use). ``redact`` optionally masks sensitive values in the
    persisted state (see :class:`~agentflow.checkpoint.file.FileCheckpointer`
    for the resume-fidelity caveat).
    """

    def __init__(
        self,
        dsn: str = "postgresql://postgres@localhost:5432/postgres",
        *,
        pool: asyncpg.Pool | None = None,
        table: str = "checkpoints",
        redact: Redactor | None = None,
    ) -> None:
        if not table.isidentifier():
            raise ValueError(f"unsafe table name {table!r}")
        self._dsn = dsn
        self._pool = pool
        self._owns_pool = pool is None
        self.table = table
        self._redact = redact or redact_none
        self._init_done = False
        self._schema_lock = asyncio.Lock()
        self._pool_lock = asyncio.Lock()

    # ------------------------------------------------------------------

    async def _get_pool(self) -> asyncpg.Pool:
        if self._pool is None:
            async with self._pool_lock:
                if self._pool is None:
                    self._pool = await asyncpg.create_pool(self._dsn)
        await self._ensure_schema()
        return self._pool

    async def _ensure_schema(self) -> None:
        if self._init_done:
            return
        assert self._pool is not None
        async with self._schema_lock:
            if self._init_done:
                return

            async def create(conn: asyncpg.Connection) -> None:
                await conn.execute(
                    f"""
                    CREATE TABLE IF NOT EXISTS {self.table} (
                        thread   TEXT    NOT NULL,
                        step     INTEGER NOT NULL,
                        revision INTEGER NOT NULL DEFAULT 1,
                        payload  JSONB   NOT NULL,
                        ts       TEXT    NOT NULL DEFAULT '',
                        PRIMARY KEY (thread, step)
                    )
                    """
                )

            await ensure_schema(self._pool, self.table, create)
            self._init_done = True

    async def close(self) -> None:
        """Close the connection pool (only if this checkpointer created it)."""
        if self._pool is not None and self._owns_pool:
            await self._pool.close()
            self._pool = None

    # ------------------------------------------------------------------

    def _serialize(self, cp: Checkpoint) -> str:
        payload = to_payload(cp)
        payload["state"] = self._redact(payload["state"])
        payload["interrupt_payload"] = self._redact(payload["interrupt_payload"])
        try:
            return json.dumps(payload, ensure_ascii=False)
        except TypeError as exc:
            raise CheckpointError(
                f"checkpoint state for thread {cp.thread!r} is not JSON-serializable: {exc}"
            ) from exc

    async def put(self, cp: Checkpoint, *, if_revision: int | None = None) -> str:
        pool = await self._get_pool()
        blob = self._serialize(cp)
        async with pool.acquire() as conn, conn.transaction():
            if if_revision is not None:
                row = await conn.fetchrow(
                    f"SELECT revision FROM {self.table} WHERE thread=$1 AND step=$2 FOR UPDATE",
                    cp.thread,
                    cp.step,
                )
                current = row["revision"] if row else 0
                if current != if_revision:
                    raise CheckpointConflict(cp.thread, cp.step, if_revision, current)
            await conn.execute(
                f"""
                INSERT INTO {self.table} (thread, step, revision, payload, ts)
                VALUES ($1, $2, 1, $3::jsonb, $4)
                ON CONFLICT (thread, step) DO UPDATE SET
                    revision = {self.table}.revision + 1,
                    payload = EXCLUDED.payload,
                    ts = EXCLUDED.ts
                """,
                cp.thread,
                cp.step,
                blob,
                cp.ts,
            )
        return f"{cp.thread}:{cp.step}"

    def _row_to_cp(self, row: asyncpg.Record) -> Checkpoint:
        payload = row["payload"]
        if isinstance(payload, str):
            payload = json.loads(payload)
        return dataclasses.replace(from_payload(payload), revision=row["revision"])

    async def get(self, thread: str, step: int | None = None) -> Checkpoint | None:
        pool = await self._get_pool()
        if step is None:
            row = await pool.fetchrow(
                f"SELECT payload, revision FROM {self.table} "
                "WHERE thread=$1 ORDER BY step DESC LIMIT 1",
                thread,
            )
        else:
            row = await pool.fetchrow(
                f"SELECT payload, revision FROM {self.table} WHERE thread=$1 AND step=$2",
                thread,
                step,
            )
        return self._row_to_cp(row) if row is not None else None

    async def history(self, thread: str) -> AsyncIterator[Checkpoint]:
        pool = await self._get_pool()
        rows = await pool.fetch(
            f"SELECT payload, revision FROM {self.table} WHERE thread=$1 ORDER BY step ASC",
            thread,
        )
        for row in rows:
            yield self._row_to_cp(row)

    async def delete_thread(self, thread: str) -> None:
        pool = await self._get_pool()
        await pool.execute(f"DELETE FROM {self.table} WHERE thread=$1", thread)

    async def prune(
        self, thread: str, *, before_step: int | None = None, older_than: str | None = None
    ) -> int:
        pool = await self._get_pool()
        clauses = ["thread=$1"]
        params: list[object] = [thread]
        if before_step is not None:
            params.append(before_step)
            clauses.append(f"step < ${len(params)}")
        if older_than is not None:
            params.append(older_than)
            clauses.append(f"ts < ${len(params)}")
        sql = f"DELETE FROM {self.table} WHERE {' AND '.join(clauses)}"
        result = await pool.execute(sql, *params)
        # asyncpg returns a tag like "DELETE 3".
        return int(result.split()[-1]) if result else 0

    async def list_threads(self, *, limit: int = 100, offset: int = 0) -> list[ThreadInfo]:
        pool = await self._get_pool()
        # DISTINCT ON picks the latest row per thread (highest step).
        rows = await pool.fetch(
            f"SELECT DISTINCT ON (thread) thread, step, payload "
            f"FROM {self.table} "
            "ORDER BY thread, step DESC "
            "LIMIT $1 OFFSET $2",
            limit,
            offset,
        )
        infos: list[ThreadInfo] = []
        for row in rows:
            payload = row["payload"]
            if isinstance(payload, str):
                payload = json.loads(payload)
            cp = from_payload(payload)
            infos.append(
                ThreadInfo(
                    thread=row["thread"],
                    latest_step=row["step"],
                    interrupted=cp.interrupted,
                    done=cp.done,
                    ts=cp.ts,
                )
            )
        return infos

close async 

close() -> None

Close the connection pool (only if this checkpointer created it).

Source code in agentflow/checkpoint/postgres.py 
119
120
121
122
123
async def close(self) -> None:
    """Close the connection pool (only if this checkpointer created it)."""
    if self._pool is not None and self._owns_pool:
        await self._pool.close()
        self._pool = None

RedisCheckpointer

Durable checkpointer backed by a Redis server.

Pass either a url (redis://host:port/db) or an existing async client via client. prefix namespaces all keys.

Source code in agentflow/checkpoint/redis.py 
 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
class RedisCheckpointer:
    """Durable checkpointer backed by a Redis server.

    Pass either a ``url`` (``redis://host:port/db``) or an existing async
    client via ``client``. ``prefix`` namespaces all keys.
    """

    def __init__(
        self,
        url: str = "redis://localhost:6379/0",
        *,
        client: aioredis.Redis | None = None,
        prefix: str = "agentflow",
        redact: Redactor | None = None,
    ) -> None:
        self._client = client or aioredis.from_url(url, decode_responses=True)
        self.prefix = prefix
        self._redact = redact or redact_none

    # ------------------------------------------------------------------

    def _cp_key(self, thread: str, step: int) -> str:
        return f"{self.prefix}:{thread}:cp:{step}"

    def _index_key(self, thread: str) -> str:
        return f"{self.prefix}:{thread}:steps"

    def _threads_key(self) -> str:
        return f"{self.prefix}:threads"

    async def close(self) -> None:
        """Close the underlying client (if we own it)."""
        await self._client.aclose()

    # ------------------------------------------------------------------

    async def put(self, cp: Checkpoint, *, if_revision: int | None = None) -> str:
        payload = to_payload(cp)
        payload["state"] = self._redact(payload["state"])
        payload["interrupt_payload"] = self._redact(payload["interrupt_payload"])
        key = self._cp_key(cp.thread, cp.step)

        if if_revision is None:
            revision = await self._current_revision(cp.thread, cp.step) + 1
            blob = self._dump_envelope(cp.thread, revision, payload)
            pipe = self._client.pipeline(transaction=True)
            pipe.set(key, blob)
            pipe.zadd(self._index_key(cp.thread), {str(cp.step): cp.step})
            pipe.sadd(self._threads_key(), cp.thread)
            await pipe.execute()
            return f"{cp.thread}:{cp.step}"

        # Conditional (compare-and-set) write via WATCH/MULTI optimistic txn.
        async with self._client.pipeline(transaction=True) as pipe:
            await pipe.watch(key)
            raw = await pipe.get(key)
            current = self._envelope_revision(raw)
            if current != if_revision:
                await pipe.reset()
                raise CheckpointConflict(cp.thread, cp.step, if_revision, current)
            blob = self._dump_envelope(cp.thread, current + 1, payload)
            pipe.multi()
            pipe.set(key, blob)
            pipe.zadd(self._index_key(cp.thread), {str(cp.step): cp.step})
            pipe.sadd(self._threads_key(), cp.thread)
            await pipe.execute()
        return f"{cp.thread}:{cp.step}"

    def _dump_envelope(self, thread: str, revision: int, payload: dict) -> str:
        try:
            return json.dumps({"revision": revision, "payload": payload}, ensure_ascii=False)
        except TypeError as exc:
            raise CheckpointError(
                f"checkpoint state for thread {thread!r} is not JSON-serializable: {exc}"
            ) from exc

    @staticmethod
    def _envelope_revision(raw: bytes | str | None) -> int:
        if raw is None:
            return 0
        return int(json.loads(raw).get("revision", 0))

    async def _current_revision(self, thread: str, step: int) -> int:
        raw = await self._client.get(self._cp_key(thread, step))
        return self._envelope_revision(raw)

    @staticmethod
    def _load(raw: bytes | str) -> Checkpoint:
        env = json.loads(raw)
        return dataclasses.replace(
            from_payload(env["payload"]), revision=int(env.get("revision", 0))
        )

    async def get(self, thread: str, step: int | None = None) -> Checkpoint | None:
        if step is None:
            latest = await self._client.zrange(self._index_key(thread), -1, -1)
            if not latest:
                return None
            step = int(str(latest[0]))
        blob = await self._client.get(self._cp_key(thread, step))
        return self._load(blob) if blob is not None else None

    async def history(self, thread: str) -> AsyncIterator[Checkpoint]:
        steps = await self._client.zrange(self._index_key(thread), 0, -1)
        for s in steps:
            blob = await self._client.get(self._cp_key(thread, int(str(s))))
            if blob is not None:
                yield self._load(blob)

    async def delete_thread(self, thread: str) -> None:
        steps = await self._client.zrange(self._index_key(thread), 0, -1)
        pipe = self._client.pipeline(transaction=True)
        for s in steps:
            pipe.delete(self._cp_key(thread, int(str(s))))
        pipe.delete(self._index_key(thread))
        pipe.srem(self._threads_key(), thread)
        await pipe.execute()

    async def prune(
        self, thread: str, *, before_step: int | None = None, older_than: str | None = None
    ) -> int:
        steps = [int(str(s)) for s in await self._client.zrange(self._index_key(thread), 0, -1)]
        removed = 0
        for step in steps:
            if before_step is not None and step >= before_step:
                continue
            if older_than is not None:
                blob = await self._client.get(self._cp_key(thread, step))
                if blob is None:
                    continue
                ts = json.loads(blob).get("payload", {}).get("ts", "")
                if ts >= older_than:
                    continue
            pipe = self._client.pipeline(transaction=True)
            pipe.delete(self._cp_key(thread, step))
            pipe.zrem(self._index_key(thread), str(step))
            await pipe.execute()
            removed += 1
        return removed

    async def list_threads(self, *, limit: int = 100, offset: int = 0) -> list[ThreadInfo]:
        threads = sorted(str(t) for t in await self._client.smembers(self._threads_key()))
        infos: list[ThreadInfo] = []
        for thread in threads[offset : offset + limit]:
            cp = await self.get(thread)
            if cp is None:
                continue
            infos.append(
                ThreadInfo(
                    thread=thread,
                    latest_step=cp.step,
                    interrupted=cp.interrupted,
                    done=cp.done,
                    ts=cp.ts,
                )
            )
        return infos

close async 

close() -> None

Close the underlying client (if we own it).

Source code in agentflow/checkpoint/redis.py 
80
81
82
async def close(self) -> None:
    """Close the underlying client (if we own it)."""
    await self._client.aclose()