symfonic.agent.middleware.pause_token_store_backends

pause_token_store_backends

Database-backed pause-token stores: Postgres and Mongo, plus their DDL.

Split out of :mod:symfonic.agent.middleware.pause_token_store (336 lines against the 300-line budget). What stayed behind is the PauseTokenStore protocol and the in-process fallback -- the part with no datastore in it. What moved here is everything that talks to one: the table and index DDL, the forward-compatible migration, and the two production implementations that own them.

pause_token_store re-exports every name below, so from symfonic.agent.middleware.pause_token_store import PostgresPauseTokenStore and its siblings are unchanged.

MongoPauseTokenStore

MongoPauseTokenStore(db: Any, *, collection_name: str = 'interrupt_consumed_tokens')

MongoDB-backed atomic store using a unique _id insert.

The Mongo counterpart to :class:PostgresPauseTokenStore. Single-use redemption is enforced by using the token jti as the document _id: the first insert_one wins; a concurrent replay hits the unique-_id constraint and raises DuplicateKeyError -> False. The check is atomic at the document level, so exactly one caller sees True.

Reuses the graph backend's async motor database (MongoGraphBackend._db) rather than opening a parallel client — the store only needs one small collection alongside the tenant graph collections. The checkpointer, by contrast, needs its own sync pymongo client (see MongoCheckpointerFactory); this store stays fully async on motor.

Raises:

Type	Description
ImportError	when pymongo (its errors module) is not installed; the message names the mongodb extra.

Source code in src/symfonic/agent/middleware/pause_token_store_backends.py

def __init__(
    self, db: Any, *, collection_name: str = "interrupt_consumed_tokens"
) -> None:
    # Probe pymongo BEFORE touching ``self`` (mirrors PostgresPauseTokenStore).
    # ``motor`` re-raises ``pymongo.errors.DuplicateKeyError`` on a dup _id,
    # so we need pymongo's error types available for the atomic check.
    try:
        import pymongo.errors  # noqa: F401
    except ImportError as missing:
        raise ImportError(
            "pymongo is required for MongoPauseTokenStore. "
            "Install with: pip install 'symfonic-core[mongodb]'\n"
            f"Original error: {missing}"
        ) from missing

    # ``db`` is a motor ``AsyncIOMotorDatabase``; index into it for the
    # collection. Untyped to avoid coupling the middleware package to motor.
    self._col = db[collection_name]

consume async

consume(jti: str, scope_hash: str, *, name: str = 'ask_user') -> bool

Atomic single-use redemption via unique _id insert.

• First caller: insert_one succeeds -> True.
• Replay caller: duplicate _id -> DuplicateKeyError -> False.

Source code in src/symfonic/agent/middleware/pause_token_store_backends.py

async def consume(
    self, jti: str, scope_hash: str, *, name: str = "ask_user"
) -> bool:
    """Atomic single-use redemption via unique ``_id`` insert.

    * First caller: ``insert_one`` succeeds -> True.
    * Replay caller: duplicate ``_id`` -> ``DuplicateKeyError`` -> False.
    """
    from pymongo.errors import DuplicateKeyError

    try:
        await self._col.insert_one(
            {"_id": jti, "scope_hash": scope_hash, "name": name}
        )
        return True
    except DuplicateKeyError:
        return False

ensure_schema async

ensure_schema() -> None

No-op: single-use is enforced by the unique _id (jti).

Provided for parity with :class:PostgresPauseTokenStore so callers can invoke it uniformly. Mongo needs no DDL — the _id index is implicit and unique.

Source code in src/symfonic/agent/middleware/pause_token_store_backends.py

async def ensure_schema(self) -> None:
    """No-op: single-use is enforced by the unique ``_id`` (jti).

    Provided for parity with :class:`PostgresPauseTokenStore` so callers
    can invoke it uniformly. Mongo needs no DDL — the ``_id`` index is
    implicit and unique.
    """
    return None

PostgresPauseTokenStore

PostgresPauseTokenStore(pool: Any)

Postgres-backed atomic store using INSERT ... ON CONFLICT DO NOTHING.

Construction takes the same PostgresPoolManager already used by the semantic-memory backend so we re-use the existing asyncpg pool rather than opening a parallel connection. ensure_schema() must be awaited once at agent boot before the first consume call.

Raises:

Type	Description
ImportError	when asyncpg is not installed. v7.1.3 (Item 10.4) moved this check ahead of any state assignment so a failed import does not partially construct the store; the caller's set_store (or its modern per-agent equivalent) never sees a half-built object.

Source code in src/symfonic/agent/middleware/pause_token_store_backends.py

def __init__(self, pool: Any) -> None:
    # v7.1.3 (Item 10.4): probe asyncpg BEFORE touching ``self``. If
    # the extra is not installed, raise a clean ImportError naming
    # the install command -- mirroring the pattern in
    # ``PostgresCheckpointerFactory.build_saver_sync()``. The store
    # only needs asyncpg for runtime SQL; the import probe is the
    # cheapest way to surface the missing dep eagerly.
    try:
        import asyncpg  # type: ignore[import-untyped]  # noqa: F401
    except ImportError as missing:
        raise ImportError(
            "asyncpg is required for PostgresPauseTokenStore. "
            "Install with: pip install 'symfonic-core[postgres]'\n"
            f"Original error: {missing}"
        ) from missing

    # Accepts a PostgresPoolManager (preferred) OR a raw asyncpg.Pool.
    # Untyped here to avoid hard-coupling the middleware package to the
    # memory backend; engine wiring decides which to pass.
    self._pool = pool
    self._schema_ready = False
    self._schema_lock = asyncio.Lock()

consume async

consume(jti: str, scope_hash: str, *, name: str = 'ask_user') -> bool

Atomic single-use redemption.

The query semantics

INSERT ... ON CONFLICT (jti) DO NOTHING RETURNING jti

• First caller: row inserted, RETURNING yields the jti -> True.
• Replay caller: ON CONFLICT DO NOTHING suppresses the insert, RETURNING yields nothing -> False.

The check is atomic at the row level; concurrent inserts of the same jti are serialised by the PRIMARY KEY constraint -- exactly one wins, the rest see ON CONFLICT.

name defaults to "ask_user" so unchanged callers keep their semantics. Roadmap Item 9 generic interrupts populate it with the registered interrupt name (e.g. "approval_required").

Source code in src/symfonic/agent/middleware/pause_token_store_backends.py

async def consume(
    self, jti: str, scope_hash: str, *, name: str = "ask_user"
) -> bool:
    """Atomic single-use redemption.

    The query semantics:
        INSERT ... ON CONFLICT (jti) DO NOTHING RETURNING jti

    * First caller: row inserted, ``RETURNING`` yields the jti -> True.
    * Replay caller: ``ON CONFLICT DO NOTHING`` suppresses the insert,
      ``RETURNING`` yields nothing -> False.

    The check is atomic at the row level; concurrent inserts of the
    same jti are serialised by the PRIMARY KEY constraint -- exactly
    one wins, the rest see ``ON CONFLICT``.

    ``name`` defaults to ``"ask_user"`` so unchanged callers keep
    their semantics. Roadmap Item 9 generic interrupts populate it
    with the registered interrupt name (e.g. ``"approval_required"``).
    """
    await self.ensure_schema()
    conn = await self._acquire()
    try:
        row = await conn.fetchrow(
            """
            INSERT INTO interrupt_consumed_tokens (jti, scope_hash, name)
            VALUES ($1, $2, $3)
            ON CONFLICT (jti) DO NOTHING
            RETURNING jti
            """,
            jti,
            scope_hash,
            name,
        )
        return row is not None
    finally:
        await self._release(conn)

ensure_schema async

ensure_schema() -> None

Create the consumed-tokens table + index (idempotent).

Called automatically before the first consume if it has not already run; explicit invocation at agent boot is also supported for callers that want to surface DDL errors eagerly.

Source code in src/symfonic/agent/middleware/pause_token_store_backends.py

async def ensure_schema(self) -> None:
    """Create the consumed-tokens table + index (idempotent).

    Called automatically before the first ``consume`` if it has not
    already run; explicit invocation at agent boot is also supported
    for callers that want to surface DDL errors eagerly.
    """
    # Double-checked locking so concurrent first-callers only run DDL once.
    if self._schema_ready:
        return
    async with self._schema_lock:
        if self._schema_ready:
            return
        conn = await self._acquire()
        try:
            await conn.execute(INTERRUPT_CONSUMED_TOKENS_DDL)
            # Forward-compat migration for tables created by earlier
            # 7.1.x revisions (PR-3 shipped without the ``name``
            # column).  ``ADD COLUMN IF NOT EXISTS`` is idempotent
            # so re-runs on a fresh table are a no-op.
            for migration_sql in INTERRUPT_CONSUMED_TOKENS_MIGRATIONS:
                await conn.execute(migration_sql)
            for idx_sql in INTERRUPT_CONSUMED_TOKENS_INDEXES_DDL:
                await conn.execute(idx_sql)
        finally:
            await self._release(conn)
        self._schema_ready = True