Skip to content

Control plane

The registry, run queue, and workers. See The control plane guide.

agentflow.controlplane

Control plane: manage graph runs outside the process that created them.

An execution plane (CompiledGraph.invoke/stream/resume) runs a graph inline in your code. The control plane adds a durable :class:RunQueue of run requests, a :class:GraphRegistry (name -> compiled graph), and a :class:Worker pool that claims requests and executes them — so runs survive a process restart, scale across workers, and can be listed, cancelled, and resumed from anywhere sharing the same queue and registry.

GraphRegistry

Maps graph names to factories, caching one compiled instance per name.

The factory is called at most once per name; the resulting :class:CompiledGraph is reused (it is immutable and safe to share across concurrent runs on different threads).

Source code in agentflow/controlplane/registry.py 
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
class GraphRegistry:
    """Maps graph names to factories, caching one compiled instance per name.

    The factory is called at most once per name; the resulting
    :class:`CompiledGraph` is reused (it is immutable and safe to share across
    concurrent runs on different threads).
    """

    def __init__(self) -> None:
        self._factories: dict[str, GraphFactory] = {}
        self._cache: dict[str, CompiledGraph] = {}

    def register(self, name: str, factory: GraphFactory) -> None:
        if not name:
            raise ValueError("graph name must be non-empty")
        if name in self._factories:
            raise ValueError(f"graph {name!r} is already registered")
        self._factories[name] = factory

    def get(self, name: str) -> CompiledGraph:
        cached = self._cache.get(name)
        if cached is not None:
            return cached
        factory = self._factories.get(name)
        if factory is None:
            raise KeyError(f"no graph registered under {name!r}")
        graph = factory()
        self._cache[name] = graph
        return graph

    def names(self) -> list[str]:
        return sorted(self._factories)

    def __contains__(self, name: object) -> bool:
        return name in self._factories

RunQueue

Bases: Protocol

Persists run requests and hands them to workers under a lease.

Source code in agentflow/controlplane/queue.py 
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
@runtime_checkable
class RunQueue(Protocol):
    """Persists run requests and hands them to workers under a lease."""

    async def enqueue(
        self, graph: str, input: Mapping[str, Any] | None = None, *, thread: str | None = None
    ) -> RunRecord:
        """Add a new run for ``graph`` and return its record (status queued).
        ``thread`` defaults to the generated run id."""
        ...

    async def get(self, run_id: str) -> RunRecord | None:
        """Return the run record, or None if unknown."""
        ...

    async def list(
        self,
        *,
        status: str | None = None,
        graph: str | None = None,
        limit: int = 100,
        offset: int = 0,
    ) -> list[RunRecord]:
        """Return run records, newest first, optionally filtered."""
        ...

    async def claim(self, *, lease_seconds: float = 60.0) -> RunRecord | None:
        """Atomically take one runnable request and mark it running under a
        fresh lease. Runnable means: queued, running with an expired lease, or
        waiting (parked on ``ctx.wait``) with its ``wake_at`` now past. Returns
        None if nothing is available. Safe for concurrent workers."""
        ...

    async def heartbeat(self, run_id: str, *, lease_seconds: float = 60.0) -> None:
        """Extend the lease on a claimed run (call periodically while working)."""
        ...

    async def complete(
        self, run_id: str, *, status: str, error: str | None = None, wake_at: str | None = None
    ) -> None:
        """Move a claimed run to a terminal-ish status (succeeded / failed /
        interrupted / cancelled), clearing its lease. For a ``waiting`` status
        (parked on ``ctx.wait``), pass ``wake_at`` so the run becomes claimable
        again once that time passes."""
        ...

    async def request_cancel(self, run_id: str) -> bool:
        """Request cooperative cancellation; return True if the run exists and
        is not already terminal. A queued run is cancelled immediately; a
        running one is flagged for the worker to stop at the next boundary."""
        ...

    async def enqueue_resume(self, run_id: str, value: Any = None) -> RunRecord:
        """Re-queue an interrupted run with the human ``value`` so a worker
        resumes it. Returns the updated record (status queued)."""
        ...

    async def stats(self) -> QueueStats:
        """Return a point-in-time snapshot of queue depth by status plus the
        count of expired-lease (stranded) runs — for health/monitoring."""
        ...

enqueue async 

enqueue(graph: str, input: Mapping[str, Any] | None = None, *, thread: str | None = None) -> RunRecord

Add a new run for graph and return its record (status queued). thread defaults to the generated run id.

Source code in agentflow/controlplane/queue.py 
38
39
40
41
42
43
async def enqueue(
    self, graph: str, input: Mapping[str, Any] | None = None, *, thread: str | None = None
) -> RunRecord:
    """Add a new run for ``graph`` and return its record (status queued).
    ``thread`` defaults to the generated run id."""
    ...

get async 

get(run_id: str) -> RunRecord | None

Return the run record, or None if unknown.

Source code in agentflow/controlplane/queue.py 
45
46
47
async def get(self, run_id: str) -> RunRecord | None:
    """Return the run record, or None if unknown."""
    ...

list async 

list(*, status: str | None = None, graph: str | None = None, limit: int = 100, offset: int = 0) -> list[RunRecord]

Return run records, newest first, optionally filtered.

Source code in agentflow/controlplane/queue.py 
49
50
51
52
53
54
55
56
57
58
async def list(
    self,
    *,
    status: str | None = None,
    graph: str | None = None,
    limit: int = 100,
    offset: int = 0,
) -> list[RunRecord]:
    """Return run records, newest first, optionally filtered."""
    ...

claim async 

claim(*, lease_seconds: float = 60.0) -> RunRecord | None

Atomically take one runnable request and mark it running under a fresh lease. Runnable means: queued, running with an expired lease, or waiting (parked on ctx.wait) with its wake_at now past. Returns None if nothing is available. Safe for concurrent workers.

Source code in agentflow/controlplane/queue.py 
60
61
62
63
64
65
async def claim(self, *, lease_seconds: float = 60.0) -> RunRecord | None:
    """Atomically take one runnable request and mark it running under a
    fresh lease. Runnable means: queued, running with an expired lease, or
    waiting (parked on ``ctx.wait``) with its ``wake_at`` now past. Returns
    None if nothing is available. Safe for concurrent workers."""
    ...

heartbeat async 

heartbeat(run_id: str, *, lease_seconds: float = 60.0) -> None

Extend the lease on a claimed run (call periodically while working).

Source code in agentflow/controlplane/queue.py 
67
68
69
async def heartbeat(self, run_id: str, *, lease_seconds: float = 60.0) -> None:
    """Extend the lease on a claimed run (call periodically while working)."""
    ...

complete async 

complete(run_id: str, *, status: str, error: str | None = None, wake_at: str | None = None) -> None

Move a claimed run to a terminal-ish status (succeeded / failed / interrupted / cancelled), clearing its lease. For a waiting status (parked on ctx.wait), pass wake_at so the run becomes claimable again once that time passes.

Source code in agentflow/controlplane/queue.py 
71
72
73
74
75
76
77
78
async def complete(
    self, run_id: str, *, status: str, error: str | None = None, wake_at: str | None = None
) -> None:
    """Move a claimed run to a terminal-ish status (succeeded / failed /
    interrupted / cancelled), clearing its lease. For a ``waiting`` status
    (parked on ``ctx.wait``), pass ``wake_at`` so the run becomes claimable
    again once that time passes."""
    ...

request_cancel async 

request_cancel(run_id: str) -> bool

Request cooperative cancellation; return True if the run exists and is not already terminal. A queued run is cancelled immediately; a running one is flagged for the worker to stop at the next boundary.

Source code in agentflow/controlplane/queue.py 
80
81
82
83
84
async def request_cancel(self, run_id: str) -> bool:
    """Request cooperative cancellation; return True if the run exists and
    is not already terminal. A queued run is cancelled immediately; a
    running one is flagged for the worker to stop at the next boundary."""
    ...

enqueue_resume async 

enqueue_resume(run_id: str, value: Any = None) -> RunRecord

Re-queue an interrupted run with the human value so a worker resumes it. Returns the updated record (status queued).

Source code in agentflow/controlplane/queue.py 
86
87
88
89
async def enqueue_resume(self, run_id: str, value: Any = None) -> RunRecord:
    """Re-queue an interrupted run with the human ``value`` so a worker
    resumes it. Returns the updated record (status queued)."""
    ...

stats async 

stats() -> QueueStats

Return a point-in-time snapshot of queue depth by status plus the count of expired-lease (stranded) runs — for health/monitoring.

Source code in agentflow/controlplane/queue.py 
91
92
93
94
async def stats(self) -> QueueStats:
    """Return a point-in-time snapshot of queue depth by status plus the
    count of expired-lease (stranded) runs — for health/monitoring."""
    ...

MemoryRunQueue

Keeps run records in a dict, guarded by a lock so claim is atomic.

Source code in agentflow/controlplane/memory.py 
 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
class MemoryRunQueue:
    """Keeps run records in a dict, guarded by a lock so ``claim`` is atomic."""

    def __init__(self) -> None:
        self._runs: dict[str, RunRecord] = {}
        self._lock = asyncio.Lock()

    async def enqueue(
        self, graph: str, input: Mapping[str, Any] | None = None, *, thread: str | None = None
    ) -> RunRecord:
        run_id = uuid.uuid4().hex
        ts = _iso(_now())
        rec = RunRecord(
            run_id=run_id,
            graph=graph,
            thread=thread or run_id,
            status=RunStatus.QUEUED,
            input=dict(input or {}),
            created_at=ts,
            updated_at=ts,
        )
        async with self._lock:
            self._runs[run_id] = rec
        return rec

    async def get(self, run_id: str) -> RunRecord | None:
        return self._runs.get(run_id)

    async def list(
        self,
        *,
        status: str | None = None,
        graph: str | None = None,
        limit: int = 100,
        offset: int = 0,
    ) -> list[RunRecord]:
        recs = [
            r
            for r in self._runs.values()
            if (status is None or r.status == status) and (graph is None or r.graph == graph)
        ]
        recs.sort(key=lambda r: r.created_at, reverse=True)
        return recs[offset : offset + limit]

    async def claim(self, *, lease_seconds: float = 60.0) -> RunRecord | None:
        now = _now()
        async with self._lock:
            candidates = [
                r
                for r in self._runs.values()
                if r.status == RunStatus.QUEUED
                or (
                    r.status == RunStatus.RUNNING
                    and r.lease_until is not None
                    and datetime.fromisoformat(r.lease_until) <= now
                )
                or (
                    # A parked (ctx.wait) run becomes claimable once its wake
                    # time passes — the timer-based sibling of lease expiry.
                    r.status == RunStatus.WAITING
                    and r.wake_at is not None
                    and datetime.fromisoformat(r.wake_at) <= now
                )
            ]
            if not candidates:
                return None
            candidates.sort(key=lambda r: r.created_at)
            chosen = candidates[0]
            claimed = dataclasses.replace(
                chosen,
                status=RunStatus.RUNNING,
                lease_until=_iso(now + timedelta(seconds=lease_seconds)),
                wake_at=None,
                attempt=chosen.attempt + 1,
                updated_at=_iso(now),
            )
            self._runs[chosen.run_id] = claimed
            return claimed

    async def heartbeat(self, run_id: str, *, lease_seconds: float = 60.0) -> None:
        async with self._lock:
            rec = self._require(run_id)
            self._runs[run_id] = dataclasses.replace(
                rec,
                lease_until=_iso(_now() + timedelta(seconds=lease_seconds)),
                updated_at=_iso(_now()),
            )

    async def complete(
        self, run_id: str, *, status: str, error: str | None = None, wake_at: str | None = None
    ) -> None:
        async with self._lock:
            rec = self._require(run_id)
            self._runs[run_id] = dataclasses.replace(
                rec,
                status=status,
                error=error,
                lease_until=None,
                # wake_at is meaningful only for a parked (waiting) run; any
                # other completion clears it.
                wake_at=wake_at if status == RunStatus.WAITING else None,
                updated_at=_iso(_now()),
            )

    async def request_cancel(self, run_id: str) -> bool:
        async with self._lock:
            rec = self._runs.get(run_id)
            if rec is None or rec.status in RunStatus.TERMINAL:
                return False
            if rec.status == RunStatus.QUEUED:
                # Not yet claimed — cancel outright.
                self._runs[run_id] = dataclasses.replace(
                    rec, status=RunStatus.CANCELLED, lease_until=None, updated_at=_iso(_now())
                )
            else:
                # Running / interrupted — flag for the worker to stop.
                self._runs[run_id] = dataclasses.replace(
                    rec, cancel_requested=True, updated_at=_iso(_now())
                )
            return True

    async def enqueue_resume(self, run_id: str, value: Any = None) -> RunRecord:
        async with self._lock:
            rec = self._require(run_id)
            updated = dataclasses.replace(
                rec,
                status=RunStatus.QUEUED,
                resume_value=value,
                error=None,
                lease_until=None,
                cancel_requested=False,
                updated_at=_iso(_now()),
            )
            self._runs[run_id] = updated
            return updated

    async def stats(self) -> QueueStats:
        now = _now()
        by_status: dict[str, int] = {}
        expired = 0
        for rec in self._runs.values():
            by_status[rec.status] = by_status.get(rec.status, 0) + 1
            if (
                rec.status == RunStatus.RUNNING
                and rec.lease_until is not None
                and datetime.fromisoformat(rec.lease_until) <= now
            ):
                expired += 1
        return QueueStats(total=len(self._runs), by_status=by_status, expired_leases=expired)

    def _require(self, run_id: str) -> RunRecord:
        rec = self._runs.get(run_id)
        if rec is None:
            raise RunNotFound(run_id)
        return rec

PostgresRunQueue

Durable run queue backed by a PostgreSQL table.

Pass either a dsn or an existing asyncpg pool via pool. table names the runs table (created on first use).

Source code in agentflow/controlplane/postgres.py 
 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
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
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
class PostgresRunQueue:
    """Durable run queue backed by a PostgreSQL table.

    Pass either a ``dsn`` or an existing ``asyncpg`` pool via ``pool``.
    ``table`` names the runs table (created on first use).
    """

    def __init__(
        self,
        dsn: str = "postgresql://postgres@localhost:5432/postgres",
        *,
        pool: asyncpg.Pool | None = None,
        table: str = "runs",
    ) -> 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._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} (
                        run_id           TEXT        PRIMARY KEY,
                        graph            TEXT        NOT NULL,
                        thread           TEXT        NOT NULL,
                        status           TEXT        NOT NULL,
                        input            JSONB       NOT NULL DEFAULT '{{}}'::jsonb,
                        resume_value     JSONB,
                        error            TEXT,
                        attempt          INTEGER     NOT NULL DEFAULT 0,
                        created_at       TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                        updated_at       TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                        lease_until      TIMESTAMPTZ,
                        wake_at          TIMESTAMPTZ,
                        cancel_requested BOOLEAN     NOT NULL DEFAULT FALSE
                    )
                    """
                )
                await conn.execute(
                    f"CREATE INDEX IF NOT EXISTS idx_{self.table}_status "
                    f"ON {self.table} (status, created_at)"
                )

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

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

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

    @staticmethod
    def _iso(dt: datetime | None) -> str | None:
        return dt.isoformat() if dt is not None else None

    def _row_to_record(self, row: asyncpg.Record) -> RunRecord:
        def _load(v: Any) -> Any:
            return json.loads(v) if isinstance(v, str) else v

        return RunRecord(
            run_id=row["run_id"],
            graph=row["graph"],
            thread=row["thread"],
            status=row["status"],
            input=_load(row["input"]) or {},
            resume_value=_load(row["resume_value"]),
            error=row["error"],
            attempt=row["attempt"],
            created_at=self._iso(row["created_at"]) or "",
            updated_at=self._iso(row["updated_at"]) or "",
            lease_until=self._iso(row["lease_until"]),
            wake_at=self._iso(row["wake_at"]),
            cancel_requested=row["cancel_requested"],
        )

    _COLS = (
        "run_id, graph, thread, status, input, resume_value, error, attempt, "
        "created_at, updated_at, lease_until, wake_at, cancel_requested"
    )

    async def enqueue(
        self, graph: str, input: Mapping[str, Any] | None = None, *, thread: str | None = None
    ) -> RunRecord:
        pool = await self._get_pool()
        run_id = uuid.uuid4().hex
        row = await pool.fetchrow(
            f"INSERT INTO {self.table} (run_id, graph, thread, status, input) "
            f"VALUES ($1, $2, $3, $4, $5::jsonb) RETURNING {self._COLS}",
            run_id,
            graph,
            thread or run_id,
            RunStatus.QUEUED,
            json.dumps(dict(input or {}), ensure_ascii=False),
        )
        return self._row_to_record(row)

    async def get(self, run_id: str) -> RunRecord | None:
        pool = await self._get_pool()
        row = await pool.fetchrow(f"SELECT {self._COLS} FROM {self.table} WHERE run_id=$1", run_id)
        return self._row_to_record(row) if row is not None else None

    async def list(
        self,
        *,
        status: str | None = None,
        graph: str | None = None,
        limit: int = 100,
        offset: int = 0,
    ) -> list[RunRecord]:
        pool = await self._get_pool()
        clauses = []
        params: list[Any] = []
        if status is not None:
            params.append(status)
            clauses.append(f"status = ${len(params)}")
        if graph is not None:
            params.append(graph)
            clauses.append(f"graph = ${len(params)}")
        where = f"WHERE {' AND '.join(clauses)}" if clauses else ""
        params.extend([limit, offset])
        rows = await pool.fetch(
            f"SELECT {self._COLS} FROM {self.table} {where} "
            f"ORDER BY created_at DESC LIMIT ${len(params) - 1} OFFSET ${len(params)}",
            *params,
        )
        return [self._row_to_record(r) for r in rows]

    async def claim(self, *, lease_seconds: float = 60.0) -> RunRecord | None:
        pool = await self._get_pool()
        # Pick one runnable row (queued, or running with an expired lease),
        # locking it and skipping any row a competing worker already holds.
        row = await pool.fetchrow(
            f"""
            UPDATE {self.table} SET
                status = '{RunStatus.RUNNING}',
                attempt = attempt + 1,
                lease_until = NOW() + make_interval(secs => $1),
                wake_at = NULL,
                updated_at = NOW()
            WHERE run_id = (
                SELECT run_id FROM {self.table}
                WHERE status = '{RunStatus.QUEUED}'
                   OR (status = '{RunStatus.RUNNING}' AND lease_until < NOW())
                   OR (status = '{RunStatus.WAITING}' AND wake_at <= NOW())
                ORDER BY created_at
                FOR UPDATE SKIP LOCKED
                LIMIT 1
            )
            RETURNING {self._COLS}
            """,
            float(lease_seconds),
        )
        return self._row_to_record(row) if row is not None else None

    async def heartbeat(self, run_id: str, *, lease_seconds: float = 60.0) -> None:
        pool = await self._get_pool()
        result = await pool.execute(
            f"UPDATE {self.table} SET "
            "lease_until = NOW() + make_interval(secs => $2), updated_at = NOW() "
            "WHERE run_id = $1",
            run_id,
            float(lease_seconds),
        )
        if result.split()[-1] == "0":
            raise RunNotFound(run_id)

    async def complete(
        self, run_id: str, *, status: str, error: str | None = None, wake_at: str | None = None
    ) -> None:
        pool = await self._get_pool()
        # wake_at applies only to a parked (waiting) run; any other completion
        # clears it so a terminal/queued row never carries a stale wake time.
        wake_iso = wake_at if status == RunStatus.WAITING else None
        # asyncpg binds a timestamptz parameter from a datetime, not a str, so
        # parse the ISO wake time before binding.
        wake = datetime.fromisoformat(wake_iso) if wake_iso else None
        result = await pool.execute(
            f"UPDATE {self.table} SET "
            "status = $2, error = $3, lease_until = NULL, "
            "wake_at = $4, updated_at = NOW() "
            "WHERE run_id = $1",
            run_id,
            status,
            error,
            wake,
        )
        if result.split()[-1] == "0":
            raise RunNotFound(run_id)

    async def request_cancel(self, run_id: str) -> bool:
        pool = await self._get_pool()
        async with pool.acquire() as conn, conn.transaction():
            row = await conn.fetchrow(
                f"SELECT status FROM {self.table} WHERE run_id=$1 FOR UPDATE", run_id
            )
            if row is None or row["status"] in RunStatus.TERMINAL:
                return False
            if row["status"] == RunStatus.QUEUED:
                await conn.execute(
                    f"UPDATE {self.table} SET status=$2, lease_until=NULL, updated_at=NOW() "
                    "WHERE run_id=$1",
                    run_id,
                    RunStatus.CANCELLED,
                )
            else:
                await conn.execute(
                    f"UPDATE {self.table} SET cancel_requested=TRUE, updated_at=NOW() "
                    "WHERE run_id=$1",
                    run_id,
                )
            return True

    async def enqueue_resume(self, run_id: str, value: Any = None) -> RunRecord:
        pool = await self._get_pool()
        row = await pool.fetchrow(
            f"""
            UPDATE {self.table} SET
                status = '{RunStatus.QUEUED}',
                resume_value = $2::jsonb,
                error = NULL,
                lease_until = NULL,
                cancel_requested = FALSE,
                updated_at = NOW()
            WHERE run_id = $1
            RETURNING {self._COLS}
            """,
            run_id,
            json.dumps(value, ensure_ascii=False),
        )
        if row is None:
            raise RunNotFound(run_id)
        return self._row_to_record(row)

    async def stats(self) -> QueueStats:
        pool = await self._get_pool()
        rows = await pool.fetch(f"SELECT status, COUNT(*) AS n FROM {self.table} GROUP BY status")
        by_status = {r["status"]: r["n"] for r in rows}
        expired = await pool.fetchval(
            f"SELECT COUNT(*) FROM {self.table} "
            f"WHERE status = '{RunStatus.RUNNING}' AND lease_until < NOW()"
        )
        return QueueStats(
            total=sum(by_status.values()),
            by_status=by_status,
            expired_leases=int(expired or 0),
        )

close async 

close() -> None

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

Source code in agentflow/controlplane/postgres.py 
122
123
124
125
126
async def close(self) -> None:
    """Close the connection pool (only if this queue created it)."""
    if self._pool is not None and self._owns_pool:
        await self._pool.close()
        self._pool = None

Worker

Executes runs claimed from a :class:RunQueue.

lease_seconds is the claim lease; heartbeat_interval how often it is renewed (should be well under the lease). poll_interval is how long run_forever waits when the queue is empty.

Source code in agentflow/controlplane/worker.py 
 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
class Worker:
    """Executes runs claimed from a :class:`RunQueue`.

    ``lease_seconds`` is the claim lease; ``heartbeat_interval`` how often it is
    renewed (should be well under the lease). ``poll_interval`` is how long
    ``run_forever`` waits when the queue is empty.
    """

    def __init__(
        self,
        queue: RunQueue,
        registry: GraphRegistry,
        *,
        lease_seconds: float = 60.0,
        heartbeat_interval: float = 15.0,
        poll_interval: float = 1.0,
    ) -> None:
        self.queue = queue
        self.registry = registry
        self.lease_seconds = lease_seconds
        self.heartbeat_interval = heartbeat_interval
        self.poll_interval = poll_interval
        self._stopping = False
        self._busy = False

    @property
    def busy(self) -> bool:
        """True while this worker is executing a claimed run."""
        return self._busy

    @property
    def stopping(self) -> bool:
        return self._stopping

    async def run_once(self) -> bool:
        """Claim and execute at most one run. Return True if one was handled."""
        rec = await self.queue.claim(lease_seconds=self.lease_seconds)
        if rec is None:
            return False
        self._busy = True
        try:
            await self._execute(rec)
        finally:
            self._busy = False
        return True

    async def run_forever(self) -> None:
        """Loop claiming and executing runs until :meth:`stop` is called."""
        self._stopping = False
        while not self._stopping:
            did = await self.run_once()
            if not did:
                await asyncio.sleep(self.poll_interval)

    def stop(self) -> None:
        self._stopping = True

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

    async def _execute(self, rec: RunRecord) -> None:
        try:
            graph = self.registry.get(rec.graph)
        except KeyError as exc:
            await self.queue.complete(rec.run_id, status=RunStatus.FAILED, error=str(exc))
            return

        heartbeat = asyncio.create_task(self._heartbeat_loop(rec.run_id))
        try:
            work = self._start_work(graph, rec, await self._is_resume(graph, rec))
            await self._race_with_cancel(rec.run_id, work)
        except _Cancelled:
            await self.queue.complete(rec.run_id, status=RunStatus.CANCELLED)
            return
        except InterruptError:
            # Should not escape the runtime, but treat defensively as suspended.
            await self.queue.complete(rec.run_id, status=RunStatus.INTERRUPTED)
            return
        except Exception as exc:  # noqa: BLE001 - report any node/graph failure
            await self.queue.complete(
                rec.run_id, status=RunStatus.FAILED, error=f"{type(exc).__name__}: {exc}"
            )
            return
        finally:
            heartbeat.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await heartbeat

        # Read the checkpoint the run just wrote to tell finished from
        # suspended, and a human interrupt from a timer wait (ctx.wait).
        status, wake_at = await self._final_status(graph, rec.thread)
        await self.queue.complete(rec.run_id, status=status, wake_at=wake_at)

    def _start_work(self, graph, rec: RunRecord, is_resume: bool):
        if is_resume:
            return graph.resume(rec.thread, rec.resume_value)
        return graph.invoke(dict(rec.input), thread=rec.thread)

    async def _is_resume(self, graph, rec: RunRecord) -> bool:
        """A claim is a resume iff the thread already sits on an interrupted,
        not-yet-finished checkpoint. Authoritative regardless of attempt count
        or whether the human answer happened to be None."""
        if graph.checkpointer is None:
            return False
        cp = await graph.get_state(rec.thread)
        return cp is not None and cp.interrupted and not cp.done

    async def _final_status(self, graph, thread: str) -> tuple[str, str | None]:
        cp = await graph.get_state(thread) if graph.checkpointer is not None else None
        if cp is None or not cp.interrupted:
            return RunStatus.SUCCEEDED, None
        # A timer wait (ctx.wait) parks the run until wake_at; a human interrupt
        # waits for an explicit resume. Same checkpoint flag, different queue
        # state and resolver.
        if cp.suspend_kind == "timer":
            return RunStatus.WAITING, cp.wake_at
        return RunStatus.INTERRUPTED, None

    async def _heartbeat_loop(self, run_id: str) -> None:
        while True:
            await asyncio.sleep(self.heartbeat_interval)
            await self.queue.heartbeat(run_id, lease_seconds=self.lease_seconds)

    async def _race_with_cancel(self, run_id: str, work) -> None:
        """Run ``work`` while polling for a cancel request; raise _Cancelled."""
        work_task = asyncio.ensure_future(work)
        while True:
            done, _ = await asyncio.wait({work_task}, timeout=self.poll_interval)
            if work_task in done:
                work_task.result()  # propagate exception / completion
                return
            rec = await self.queue.get(run_id)
            if rec is not None and rec.cancel_requested:
                work_task.cancel()
                with contextlib.suppress(asyncio.CancelledError, Exception):
                    await work_task
                raise _Cancelled

busy property 

busy: bool

True while this worker is executing a claimed run.

run_once async 

run_once() -> bool

Claim and execute at most one run. Return True if one was handled.

Source code in agentflow/controlplane/worker.py 
74
75
76
77
78
79
80
81
82
83
84
async def run_once(self) -> bool:
    """Claim and execute at most one run. Return True if one was handled."""
    rec = await self.queue.claim(lease_seconds=self.lease_seconds)
    if rec is None:
        return False
    self._busy = True
    try:
        await self._execute(rec)
    finally:
        self._busy = False
    return True

run_forever async 

run_forever() -> None

Loop claiming and executing runs until :meth:stop is called.

Source code in agentflow/controlplane/worker.py 
86
87
88
89
90
91
92
async def run_forever(self) -> None:
    """Loop claiming and executing runs until :meth:`stop` is called."""
    self._stopping = False
    while not self._stopping:
        did = await self.run_once()
        if not did:
            await asyncio.sleep(self.poll_interval)

WorkerPool

Runs concurrency workers over one queue+registry.

Source code in agentflow/controlplane/worker.py 
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
class WorkerPool:
    """Runs ``concurrency`` workers over one queue+registry."""

    def __init__(
        self,
        queue: RunQueue,
        registry: GraphRegistry,
        *,
        concurrency: int = 4,
        lease_seconds: float = 60.0,
        poll_interval: float = 1.0,
    ) -> None:
        self._workers = [
            Worker(
                queue,
                registry,
                lease_seconds=lease_seconds,
                poll_interval=poll_interval,
            )
            for _ in range(concurrency)
        ]
        self._tasks: list[asyncio.Task] = []

    async def start(self) -> None:
        self._tasks = [asyncio.create_task(w.run_forever()) for w in self._workers]

    async def stop(self) -> None:
        """Signal workers to stop after their current run, then await them."""
        for w in self._workers:
            w.stop()
        for t in self._tasks:
            with contextlib.suppress(asyncio.CancelledError):
                await t
        self._tasks = []

    def health(self) -> PoolHealth:
        """Snapshot of the pool: configured size, live loops, and busy count.

        A loop is "alive" if it was started and its task has not finished
        (neither stopped cleanly nor crashed). ``busy`` counts workers mid-run.
        Use ``health().healthy`` as a readiness signal — False means a worker
        loop died and the pool is running degraded."""
        alive = sum(1 for t in self._tasks if not t.done())
        busy = sum(1 for w in self._workers if w.busy)
        return PoolHealth(workers=len(self._workers), alive=alive, busy=busy)

stop async 

stop() -> None

Signal workers to stop after their current run, then await them.

Source code in agentflow/controlplane/worker.py 
204
205
206
207
208
209
210
211
async def stop(self) -> None:
    """Signal workers to stop after their current run, then await them."""
    for w in self._workers:
        w.stop()
    for t in self._tasks:
        with contextlib.suppress(asyncio.CancelledError):
            await t
    self._tasks = []

health 

health() -> PoolHealth

Snapshot of the pool: configured size, live loops, and busy count.

A loop is "alive" if it was started and its task has not finished (neither stopped cleanly nor crashed). busy counts workers mid-run. Use health().healthy as a readiness signal — False means a worker loop died and the pool is running degraded.

Source code in agentflow/controlplane/worker.py 
213
214
215
216
217
218
219
220
221
222
def health(self) -> PoolHealth:
    """Snapshot of the pool: configured size, live loops, and busy count.

    A loop is "alive" if it was started and its task has not finished
    (neither stopped cleanly nor crashed). ``busy`` counts workers mid-run.
    Use ``health().healthy`` as a readiness signal — False means a worker
    loop died and the pool is running degraded."""
    alive = sum(1 for t in self._tasks if not t.done())
    busy = sum(1 for w in self._workers if w.busy)
    return PoolHealth(workers=len(self._workers), alive=alive, busy=busy)

RunRecord dataclass

One run's queue entry and status.

graph names a factory in a :class:~agentflow.controlplane.registry. GraphRegistry; thread is the checkpointer thread the run executes on (defaults to run_id). resume_value carries the human answer when an interrupted run is re-queued. lease_until is the ISO time a worker's claim holds until; an expired lease lets another worker re-claim (crash recovery). cancel_requested is a cooperative flag a worker honors at a super-step boundary.

Source code in agentflow/controlplane/records.py 
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
@dataclass(frozen=True, slots=True)
class RunRecord:
    """One run's queue entry and status.

    ``graph`` names a factory in a :class:`~agentflow.controlplane.registry.
    GraphRegistry`; ``thread`` is the checkpointer thread the run executes on
    (defaults to ``run_id``). ``resume_value`` carries the human answer when an
    interrupted run is re-queued. ``lease_until`` is the ISO time a worker's
    claim holds until; an expired lease lets another worker re-claim (crash
    recovery). ``cancel_requested`` is a cooperative flag a worker honors at a
    super-step boundary.
    """

    run_id: str
    graph: str
    thread: str
    status: str
    input: Mapping[str, Any] = field(default_factory=dict)
    resume_value: Any = None
    error: str | None = None
    attempt: int = 0
    created_at: str = ""
    updated_at: str = ""
    lease_until: str | None = None
    #: For a ``waiting`` run (parked on ``ctx.wait``): the earliest ISO time it
    #: should be re-claimed. None otherwise.
    wake_at: str | None = None
    cancel_requested: bool = False
    extra: dict[str, Any] = field(default_factory=dict)

RunStatus

The lifecycle states of a run (string constants, JSON-friendly).

queued -> running -> succeeded | interrupted | waiting | failed | cancelled. An interrupted run returns to queued via enqueue_resume; a waiting run (parked on ctx.wait) becomes claimable again on its own once wake_at passes; a queued/running run can be cancelled.

Source code in agentflow/controlplane/records.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
class RunStatus:
    """The lifecycle states of a run (string constants, JSON-friendly).

    ``queued`` -> ``running`` -> ``succeeded`` | ``interrupted`` | ``waiting``
    | ``failed`` | ``cancelled``. An ``interrupted`` run returns to ``queued``
    via ``enqueue_resume``; a ``waiting`` run (parked on ``ctx.wait``) becomes
    claimable again on its own once ``wake_at`` passes; a ``queued``/``running``
    run can be cancelled.
    """

    QUEUED: Final = "queued"
    RUNNING: Final = "running"
    SUCCEEDED: Final = "succeeded"
    INTERRUPTED: Final = "interrupted"
    WAITING: Final = "waiting"
    FAILED: Final = "failed"
    CANCELLED: Final = "cancelled"

    #: Terminal states — a run in one of these will not run again on its own.
    TERMINAL: Final = frozenset({SUCCEEDED, FAILED, CANCELLED})
    #: States a worker may claim (queued, or running with an expired lease).
    #: ``waiting`` is claimable too, but only once ``wake_at`` passes, so the
    #: time gate lives in the queue's claim logic, not this static set.
    CLAIMABLE: Final = frozenset({QUEUED, RUNNING})

QueueStats dataclass

A point-in-time snapshot of a run queue's depth, for monitoring.

by_status counts runs in every lifecycle state. expired_leases is the number of running runs whose lease is past due — i.e. work stranded by a crashed/stalled worker and awaiting re-claim. A rising value there is the signal that workers are dying or falling behind their heartbeat.

Source code in agentflow/controlplane/records.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@dataclass(frozen=True, slots=True)
class QueueStats:
    """A point-in-time snapshot of a run queue's depth, for monitoring.

    ``by_status`` counts runs in every lifecycle state. ``expired_leases`` is
    the number of ``running`` runs whose lease is past due — i.e. work stranded
    by a crashed/stalled worker and awaiting re-claim. A rising value there is
    the signal that workers are dying or falling behind their heartbeat.
    """

    total: int = 0
    by_status: Mapping[str, int] = field(default_factory=dict)
    expired_leases: int = 0

    @property
    def queued(self) -> int:
        return self.by_status.get(RunStatus.QUEUED, 0)

    @property
    def running(self) -> int:
        return self.by_status.get(RunStatus.RUNNING, 0)

PoolHealth dataclass

Health snapshot of a :class:~agentflow.controlplane.worker.WorkerPool.

workers is the configured size; alive counts worker loops still running (not stopped and not crashed); busy counts workers currently executing a run; idle = alive - busy. healthy is True when every configured worker loop is alive.

Source code in agentflow/controlplane/records.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
@dataclass(frozen=True, slots=True)
class PoolHealth:
    """Health snapshot of a :class:`~agentflow.controlplane.worker.WorkerPool`.

    ``workers`` is the configured size; ``alive`` counts worker loops still
    running (not stopped and not crashed); ``busy`` counts workers currently
    executing a run; ``idle`` = alive - busy. ``healthy`` is True when every
    configured worker loop is alive.
    """

    workers: int = 0
    alive: int = 0
    busy: int = 0

    @property
    def idle(self) -> int:
        return max(self.alive - self.busy, 0)

    @property
    def healthy(self) -> bool:
        return self.alive == self.workers and self.workers > 0