symfonic.agent.checkpointer.factory

factory

CheckpointerFactory protocol + three concrete implementations (v7.1.1).

Closes the §2 Item 2 carryover from the 7.1.x roadmap. The agent runtime no longer peeks at MemoryLayer.SEMANTIC._graph.backend._pool to build its LangGraph checkpointer; instead it picks one of the factories below and treats the returned BaseCheckpointSaver as opaque.

Two-phase lifecycle

LangGraph's workflow.compile(checkpointer=...) runs in sync code and requires the saver object up front. Production Postgres setup, however, needs async I/O (pool open + idempotent DDL). The factory protocol therefore exposes a two-phase lifecycle:

1. build_saver_sync() — constructs the saver with zero I/O. Safe from any sync constructor; the engine calls this in __init__.
2. astart() — performs the async startup (open the pool, run saver.setup()). Idempotent. The engine calls this once per agent lifetime from the top of run/stream/stream_typed.

For convenience async def build() runs both phases in order and returns the saver — used by tests and callers that already live in an async context.

Why dual pools?

langgraph-checkpoint-postgres only accepts psycopg connections. The rest of the symfonic memory stack uses asyncpg for tenant SQL (graph/vector backends via PostgresPoolManager). The two protocols are incompatible inside langgraph.checkpoint.postgres._ainternal, so PostgresCheckpointerFactory builds its own psycopg_pool.AsyncConnectionPool from the same DSN. The pools share a database; they do not share connections.

CheckpointerFactory

Bases: Protocol

Constructs and owns a LangGraph checkpointer.

Implementations own their own connection lifecycle. The agent Runtime instantiates the factory once, calls build_saver_sync to obtain the saver for graph.compile(checkpointer=...), then calls astart() once before the first real invocation. No knowledge of memory-layer internals leaks across this boundary.

astart async

astart() -> None

Open external resources (pool, DDL). Idempotent.

Source code in src/symfonic/agent/checkpointer/factory.py

async def astart(self) -> None:
    """Open external resources (pool, DDL). Idempotent."""
    ...

build async

build() -> BaseCheckpointSaver

Convenience: build_saver_sync + astart. Returns the saver.

Source code in src/symfonic/agent/checkpointer/factory.py

async def build(self) -> BaseCheckpointSaver:
    """Convenience: build_saver_sync + astart. Returns the saver."""
    ...

build_saver_sync

build_saver_sync() -> BaseCheckpointSaver

Construct the saver with zero I/O. Safe from sync code.

Source code in src/symfonic/agent/checkpointer/factory.py

def build_saver_sync(self) -> BaseCheckpointSaver:
    """Construct the saver with zero I/O. Safe from sync code."""
    ...

close async

close() -> None

Release any external resources. Idempotent.

Source code in src/symfonic/agent/checkpointer/factory.py

async def close(self) -> None:
    """Release any external resources. Idempotent."""
    ...

MemoryCheckpointerFactory

MemoryCheckpointerFactory()

Returns a MemorySaver (in-process, non-durable).

Use for unit tests and local development where pause/resume durability is not required. astart and close are no-ops.

Source code in src/symfonic/agent/checkpointer/factory.py

def __init__(self) -> None:
    self._saver: BaseCheckpointSaver | None = None

MongoCheckpointerFactory

MongoCheckpointerFactory(
    uri: str,
    database: str = "symfonic",
    *,
    checkpoint_collection_name: str = "checkpoints",
    writes_collection_name: str = "checkpoint_writes",
)

Returns a MongoDBSaver backed by a pymongo.MongoClient.

The durable checkpointer for Mongo-backed deployments. Before this factory existed, a Mongo graph backend had no matching checkpointer, so the engine fell through to an ephemeral MemorySaver — checkpoints (interactive resume, durable transcript) died on restart.

Like PostgresCheckpointerFactory's dual-pool design, this owns a connection distinct from the graph/vector backend: MongoGraphBackend uses motor's AsyncIOMotorClient for tenant I/O, while langgraph-checkpoint-mongodb's MongoDBSaver requires a synchronous pymongo.MongoClient. Its async methods (aput / aget_tuple / alist) offload the blocking pymongo calls to a thread pool, so it drives the async graph without blocking the event loop. The two clients share a database; they do not share connections.

Lifecycle:

• build_saver_sync constructs the client with connect=False (defers all network I/O and server monitoring) and wraps it in MongoDBSaver — no I/O.
• astart is a no-op: MongoDBSaver needs no setup() (its collections/indexes are created lazily on first write).
• close closes the client.

Raises:

Type	Description
SymfonicAgentError(code='bad_config')	when no URI is provided.
ImportError	when pymongo / langgraph-checkpoint-mongodb are not installed; the message names the mongodb extra.

Source code in src/symfonic/agent/checkpointer/factory.py

def __init__(
    self,
    uri: str,
    database: str = "symfonic",
    *,
    checkpoint_collection_name: str = "checkpoints",
    writes_collection_name: str = "checkpoint_writes",
) -> None:
    if not uri:
        raise SymfonicAgentError(
            "MongoCheckpointerFactory requires a non-empty URI. "
            "Pass the same connection string used by MongoGraphBackend.",
            code="bad_config",
        )
    self._uri = uri
    self._database = database
    self._checkpoint_collection_name = checkpoint_collection_name
    self._writes_collection_name = writes_collection_name
    self._saver: BaseCheckpointSaver | None = None
    self._client: Any = None  # pymongo.MongoClient at runtime
    self._started: bool = False

uri property

uri: str

Expose the configured URI (read-only) for diagnostics.

PostgresCheckpointerFactory

PostgresCheckpointerFactory(
    dsn: str, *, min_size: int = 2, max_size: int = 10
)

Returns an AsyncPostgresSaver backed by psycopg_pool.

Builds and owns its own AsyncConnectionPool. This pool is DISTINCT from PostgresPoolManager (which is asyncpg-backed and used for graph/vector I/O); the two pools share a database but do not share connections.

Lifecycle:

• build_saver_sync constructs the pool with open=False and wraps it in AsyncPostgresSaver — no I/O.
• astart opens the pool and runs saver.setup() (idempotent CREATE-IF-NOT-EXISTS DDL).
• close closes the pool.

Raises:

Type	Description
SymfonicAgentError(code='bad_config')	when no DSN is provided.
ImportError	when psycopg or langgraph-checkpoint-postgres are not installed; the message names the postgres extra.

Source code in src/symfonic/agent/checkpointer/factory.py

def __init__(
    self,
    dsn: str,
    *,
    min_size: int = 2,
    max_size: int = 10,
) -> None:
    if not dsn:
        raise SymfonicAgentError(
            "PostgresCheckpointerFactory requires a non-empty DSN. "
            "Pass the same connection string used by PostgresPoolManager.",
            code="bad_config",
        )
    self._dsn = dsn
    self._min_size = min_size
    self._max_size = max_size
    self._saver: BaseCheckpointSaver | None = None
    self._pool: Any = None  # psycopg_pool.AsyncConnectionPool at runtime
    self._started: bool = False

dsn property

dsn: str

Expose the configured DSN (read-only) for diagnostics.

astart async

astart() -> None

Open the pool and run AsyncPostgresSaver.setup() once.

Source code in src/symfonic/agent/checkpointer/factory.py

async def astart(self) -> None:
    """Open the pool and run AsyncPostgresSaver.setup() once."""
    if self._started:
        return
    if self._saver is None or self._pool is None:
        self.build_saver_sync()
    # mypy: build_saver_sync above populates these.
    assert self._pool is not None and self._saver is not None
    await self._pool.open()
    # Idempotent CREATE-IF-NOT-EXISTS DDL — safe to invoke per boot.
    await self._saver.setup()  # type: ignore[attr-defined]
    self._started = True

SqliteCheckpointerFactory

SqliteCheckpointerFactory(path: str)

Returns an AsyncSqliteSaver backed by aiosqlite.

Accepts a filesystem path (":memory:" also valid). Holds the underlying aiosqlite.Connection so close() can release it.

Requires the ask-user extra (aiosqlite + langgraph-checkpoint-sqlite). A missing dependency raises an ImportError that explicitly names the extra.

Source code in src/symfonic/agent/checkpointer/factory.py

def __init__(self, path: str) -> None:
    self._path = path
    self._saver: BaseCheckpointSaver | None = None
    self._conn: Any = None