symfonic.agent.middleware.pause_token_store

pause_token_store

Pause-token consumption stores (v7.1.1).

A PauseTokenStore records which pause-token JTIs have been redeemed so the single-use property of the ask_user flow is enforceable. Two implementations ship:

• :class:InMemoryPauseTokenStore -- the v7.1.0 set-based fallback. Single process only; tokens leak across worker restarts and are NOT safe in a horizontally-scaled deployment. Used implicitly when no Postgres pool is wired (e.g. unit tests using MemorySaver or SqliteSaver).

• :class:PostgresPauseTokenStore -- the production path. Uses INSERT ... ON CONFLICT DO NOTHING RETURNING jti in a single SQL round-trip so the first consumer always wins atomically, even across two workers redeeming the same token simultaneously.

The split mirrors the convention used by postgres_graph.py: DDL is a module-level constant, the backend exposes an ensure_schema() method, and the asyncpg pool is injected at construction time.

InMemoryPauseTokenStore

InMemoryPauseTokenStore()

Set-based in-process store.

Backwards-compatible fallback for deployments without a Postgres pool (typical: unit tests with MemorySaver / SqliteSaver). Emits a warning the first time consume is called so production operators know they are running on a single-process token cache that does NOT survive worker restart and does NOT coordinate across workers.

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

def __init__(self) -> None:
    self._consumed: set[str] = set()
    self._lock = asyncio.Lock()
    self._warned = False

clear

clear() -> None

Reset the consumed set (test-only helper).

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

def clear(self) -> None:
    """Reset the consumed set (test-only helper)."""
    self._consumed.clear()
    self._warned = False

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.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.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.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

PauseTokenStore

Bases: Protocol

Single-use token consumption store.

Implementations MUST guarantee that consume(jti, scope_hash) returns True exactly once for any given jti across the lifetime of the store. Concurrent callers competing for the same jti MUST see exactly one True and the rest False.

Roadmap Item 9 (v7.1.x): the name kwarg is an opt-in telemetry field carrying the interrupt's registered name ("ask_user" for the built-in case). Stores that persist rows MAY record it; the in-memory store ignores it.

consume async

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

Mark jti consumed atomically.

Returns:

Type	Description
bool	True if this caller is the first to consume the token;
bool	False if it was already consumed (replay attempt).

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

async def consume(
    self, jti: str, scope_hash: str, *, name: str = "ask_user"
) -> bool:
    """Mark ``jti`` consumed atomically.

    Returns:
        ``True`` if this caller is the first to consume the token;
        ``False`` if it was already consumed (replay attempt).
    """
    ...

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.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  # 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.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.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