`symfonic.agent.middleware.pause_token`¶

pause_token ¶

Pause token middleware — JWT-based single-use tokens for elicitation resume.

The single-use enforcement is delegated to a :class:PauseTokenStore (see pause_token_store.py). v7.1.1 ships two stores:

:class:InMemoryPauseTokenStore -- the default, used when no Postgres pool is wired (test/dev) and the one previously inlined here as a class attribute _consumed_jtis. Not safe across worker restarts.
:class:PostgresPauseTokenStore -- the production path, atomic across workers via INSERT ... ON CONFLICT DO NOTHING RETURNING jti.

Ownership model (v7.1.3 — Item 10.2 fix)¶

The store is owned by a :class:PauseTokenManager instance held on the SymfonicAgent. Two agents in the same process therefore have two independent stores and cannot clobber each other (the prior class-level ClassVar singleton was a latent multi-tenant bug).

The legacy classmethod API on :class:PauseToken (PauseToken.consume / PauseToken.set_store / PauseToken.reset_store_to_default) is preserved as a thin facade over a process-wide default manager so existing test fixtures and direct callers continue to work unchanged. New code (engine boot wiring, multi-agent harnesses) should use a per-agent manager instead.

PauseToken ¶

Opaque token management for elicitation resume.

Pure helper methods (hash_scope, hash_request, encode, decode, jti_to_timestamp, load_original_request) carry no instance state and live as @staticmethod / @classmethod on this class.

Single-use consumption (consume / set_store / reset_store_to_default) is now owned by :class:PauseTokenManager; the methods below remain as a thin facade over the process-wide default manager so existing test fixtures and direct callers continue to work unchanged.

consume `async` `classmethod` ¶

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

Mark a token as consumed (single-use enforcement).

Legacy facade: delegates to the process-wide default :class:PauseTokenManager. Multi-agent deployments should hold a per-agent PauseTokenManager and call its instance method instead -- two agents sharing the process-wide default still clobber each other.

Returns True if consumption succeeded, False if already used.

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

@classmethod
async def consume(
    cls,
    jti: str,
    scope: FrameworkTenantScope,
    *,
    name: str = "ask_user",
) -> bool:
    """Mark a token as consumed (single-use enforcement).

    **Legacy facade**: delegates to the process-wide default
    :class:`PauseTokenManager`. Multi-agent deployments should hold
    a per-agent ``PauseTokenManager`` and call its instance method
    instead -- two agents sharing the process-wide default still
    clobber each other.

    Returns True if consumption succeeded, False if already used.
    """
    return await _default_manager.consume(jti, scope, name=name)

decode `classmethod` ¶

decode(
    token: str, config: FrameworkConfig
) -> PauseTokenClaims

Decode and verify a pause token.

Requires python-jose (symfonic-core[ask-user]); raises :class:ImportError with the install hint when absent.

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

@classmethod
def decode(cls, token: str, config: FrameworkConfig) -> PauseTokenClaims:
    """Decode and verify a pause token.

    Requires ``python-jose`` (``symfonic-core[ask-user]``); raises
    :class:`ImportError` with the install hint when absent.
    """
    _require_jose()
    # Re-bind ``JWTError`` from the now-imported jose module so the
    # ``except`` clause catches the real type (the module-level
    # symbol is ``None`` when jose was absent at import time, but
    # ``_require_jose()`` above guarantees the import succeeded
    # by the time we reach this line). Mirrors the engine.py:3400
    # lazy-import precedent.
    from jose import JWTError as _JWTError

    secret = os.environ.get("JWT_SECRET", "framework-default-secret-change-me")
    try:
        payload = jwt.decode(token, secret, algorithms=["HS256"])
        return PauseTokenClaims(**payload)
    except _JWTError as exc:
        raise SymfonicAgentError(f"Invalid pause token: {exc}", code="unauthorized") from exc

encode `classmethod` ¶

encode(
    claims: PauseTokenClaims, config: FrameworkConfig
) -> str

Mint a HMAC-signed JWT token.

Requires python-jose (symfonic-core[ask-user]); raises :class:ImportError with the install hint when absent.

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

@classmethod
def encode(cls, claims: PauseTokenClaims, config: FrameworkConfig) -> str:
    """Mint a HMAC-signed JWT token.

    Requires ``python-jose`` (``symfonic-core[ask-user]``); raises
    :class:`ImportError` with the install hint when absent.
    """
    _require_jose()
    secret = os.environ.get("JWT_SECRET", "framework-default-secret-change-me")
    # Use HS256 for symmetric signing
    return jwt.encode(claims.model_dump(), secret, algorithm="HS256")

hash_payload `staticmethod` ¶

hash_payload(payload: Any) -> str

Deterministic hash of an arbitrary Pydantic payload.

Used by Roadmap Item 9 generic interrupts: any registered payload_schema (BaseModel subclass) can be hashed for the request_hash claim. The hash protects against payload tampering between mint and resume the same way :meth:hash_request protects ask_user.

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

@staticmethod
def hash_payload(payload: Any) -> str:
    """Deterministic hash of an arbitrary Pydantic payload.

    Used by Roadmap Item 9 generic interrupts: any registered
    ``payload_schema`` (BaseModel subclass) can be hashed for the
    ``request_hash`` claim. The hash protects against payload
    tampering between mint and resume the same way
    :meth:`hash_request` protects ``ask_user``.
    """
    if hasattr(payload, "model_dump_json"):
        raw = payload.model_dump_json()
    else:
        import json as _json

        raw = _json.dumps(payload, sort_keys=True, default=str)
    return hashlib.sha256(raw.encode()).hexdigest()

hash_request `staticmethod` ¶

hash_request(request: AskUserRequest) -> str

Deterministic hash of the server-stamped request.

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

@staticmethod
def hash_request(request: AskUserRequest) -> str:
    """Deterministic hash of the server-stamped request."""
    return hashlib.sha256(request.model_dump_json().encode()).hexdigest()

hash_scope `staticmethod` ¶

hash_scope(scope: FrameworkTenantScope) -> str

Deterministic hash of the tenant scope for binding.

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

@staticmethod
def hash_scope(scope: FrameworkTenantScope) -> str:
    """Deterministic hash of the tenant scope for binding."""
    raw = f"{scope.tenant_id}:{scope.sub_tenant_id or '_'}:{scope.namespace or '_'}"
    return hashlib.sha256(raw.encode()).hexdigest()[:16]

jti_to_timestamp `staticmethod` ¶

jti_to_timestamp(jti: str) -> float

Extract issuance timestamp from JTI if encoded there (fallback to now).

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

@staticmethod
def jti_to_timestamp(jti: str) -> float:
    """Extract issuance timestamp from JTI if encoded there (fallback to now)."""
    # Our JTI is uuid4.hex, so no timestamp. Fallback to now for time-to-answer.
    return time.time()

load_original_request `async` `staticmethod` ¶

load_original_request(
    checkpointer: Any, claims: PauseTokenClaims
) -> AskUserRequest

Load the original stamped AskUserRequest from the checkpoint.

Tries two paths in order: 1. Checkpoint pending-writes metadata keyed by ask_user_request:{tool_call_id} (written by _mint_pause_token via checkpointer.aput_writes). 2. AIMessage tool-call args in the checkpoint state — the args include the full stamped request dict (id included) because the elicitation node passed the stamped request into interrupt().

On either path, verifies the recovered request's hash matches claims.request_hash to detect tampering.

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

@staticmethod
async def load_original_request(
    checkpointer: Any, claims: PauseTokenClaims
) -> AskUserRequest:
    """Load the original stamped AskUserRequest from the checkpoint.

    Tries two paths in order:
    1. Checkpoint pending-writes metadata keyed by ``ask_user_request:{tool_call_id}``
       (written by ``_mint_pause_token`` via ``checkpointer.aput_writes``).
    2. AIMessage tool-call args in the checkpoint state — the args include
       the full stamped request dict (id included) because the elicitation
       node passed the stamped request into ``interrupt()``.

    On either path, verifies the recovered request's hash matches
    ``claims.request_hash`` to detect tampering.
    """
    import json as _json

    if checkpointer is None:
        raise SymfonicAgentError("No checkpointer available to resume", code="error")

    cfg = {
        "configurable": {
            "thread_id": claims.thread_id,
            "checkpoint_id": claims.checkpoint_id,
        }
    }
    checkpoint_tuple = await checkpointer.aget_tuple(cfg)

    if not checkpoint_tuple:
        # When MemorySaver is the active checkpointer and the worker
        # restarts between question-emit and resume, the checkpoint
        # vanishes from process memory. The HMAC + scope binding still
        # holds, but the original request cannot be reconstructed --
        # surface a clear, actionable message so callers do not chase
        # a perceived token problem.
        if _is_non_persistent_checkpointer(checkpointer):
            raise SymfonicAgentError(
                "Checkpoint lost on restart; non-persistent "
                "checkpointers (MemorySaver) are not supported with "
                "ask_user_enabled=True. Configure AsyncPostgresSaver "
                "or AsyncSqliteSaver for production. "
                f"(thread={claims.thread_id}, "
                f"checkpoint={claims.checkpoint_id})",
                code="failed_dependency",
            )
        raise SymfonicAgentError(
            f"Checkpoint {claims.checkpoint_id} not found for "
            f"thread {claims.thread_id}",
            code="not_found",
        )

    # -- Path 1: metadata written by _mint_pause_token -------------------
    meta_key = f"ask_user_request:{claims.tool_call_id}"
    pending_writes = getattr(checkpoint_tuple, "pending_writes", None) or []
    for _task_id, key, value in (pending_writes or []):
        if key == meta_key:
            try:
                raw = value if isinstance(value, str) else _json.dumps(value)
                request = AskUserRequest.model_validate_json(raw)
                if PauseToken.hash_request(request) == claims.request_hash:
                    return request
            except Exception:
                pass

    # -- Path 2: AIMessage tool-call args in checkpoint state ------------
    # The elicitation node passes stamped_request.model_dump() into
    # interrupt(), so the interrupt payload carries the full request
    # including server-stamped IDs.  LangGraph stores interrupt payloads
    # in ``checkpoint.pending_sends`` or in the channel state depending
    # on the version.  We also check the raw messages for the AIMessage
    # whose tool_calls contain the matching id.
    channel_values = checkpoint_tuple.checkpoint.get("channel_values", {})
    messages = (
        channel_values.get("messages", [])
        or checkpoint_tuple.checkpoint.get("messages", [])
    )

    for msg in reversed(messages):
        if not hasattr(msg, "tool_calls"):
            continue
        for tc in msg.tool_calls:
            if tc.get("id") != claims.tool_call_id:
                continue
            raw_req = tc["args"].get("request", {})
            # The args from the elicitation node's interrupt include id fields.
            questions = [AskUserQuestion(**q_data) for q_data in raw_req.get("questions", [])]
            try:
                request = AskUserRequest(questions=questions)
                if PauseToken.hash_request(request) == claims.request_hash:
                    return request
            except Exception:
                pass

    raise SymfonicAgentError(
        "Original request not found or hash mismatch — possible tampering",
        code="bad_request",
    )

reset_store_to_default `classmethod` ¶

reset_store_to_default() -> None

Restore the in-memory store on the process-wide default manager.

Legacy facade: test fixtures use this for isolation. New tests should construct their own PauseTokenManager to avoid cross-test global state.

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

@classmethod
def reset_store_to_default(cls) -> None:
    """Restore the in-memory store on the process-wide default manager.

    **Legacy facade**: test fixtures use this for isolation. New
    tests should construct their own ``PauseTokenManager`` to avoid
    cross-test global state.
    """
    _default_manager.reset_store_to_default()

set_store `classmethod` ¶

set_store(store: PauseTokenStore) -> None

Swap the consumption store on the process-wide default manager.

Legacy facade: mutates a global. Engine boot wiring at engine.py now configures a per-agent manager (self._pause_token_manager) instead of touching this global. Kept for backward compatibility with existing test fixtures.

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

@classmethod
def set_store(cls, store: PauseTokenStore) -> None:
    """Swap the consumption store on the process-wide default manager.

    **Legacy facade**: mutates a global. Engine boot wiring at
    ``engine.py`` now configures a per-agent manager
    (``self._pause_token_manager``) instead of touching this global.
    Kept for backward compatibility with existing test fixtures.
    """
    _default_manager.set_store(store)

PauseTokenManager ¶

PauseTokenManager(store: PauseTokenStore | None = None)

Per-agent owner of a :class:PauseTokenStore.

Held by SymfonicAgent at construction time so each agent in the same process has its own consumption store. Replaces the old class- level PauseToken._store ClassVar that two agents in the same process would clobber.

The store can be swapped at any time via :meth:set_store (engine boot wiring does this when a Postgres pool is available) or restored to the in-memory default via :meth:reset_store_to_default (test fixtures).

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

def __init__(self, store: PauseTokenStore | None = None) -> None:
    self._store: PauseTokenStore = store or InMemoryPauseTokenStore()

store `property` ¶

store: PauseTokenStore

Expose the active store (read-only) for diagnostics.

consume `async` ¶

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

Mark jti consumed for scope (single-use enforcement).

Returns True for the first caller, False on replay. Concurrent callers competing on the same jti see exactly one True.

Roadmap Item 9: name is forwarded to the store as a telemetry field. Defaults to "ask_user" so existing callers observe no behaviour change.

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

async def consume(
    self,
    jti: str,
    scope: FrameworkTenantScope,
    *,
    name: str = "ask_user",
) -> bool:
    """Mark ``jti`` consumed for ``scope`` (single-use enforcement).

    Returns True for the first caller, False on replay. Concurrent
    callers competing on the same jti see exactly one True.

    Roadmap Item 9: ``name`` is forwarded to the store as a
    telemetry field. Defaults to ``"ask_user"`` so existing callers
    observe no behaviour change.
    """
    scope_hash = PauseToken.hash_scope(scope)
    return await self._store.consume(jti, scope_hash, name=name)

reset_store_to_default ¶

reset_store_to_default() -> None

Restore an :class:InMemoryPauseTokenStore.

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

def reset_store_to_default(self) -> None:
    """Restore an :class:`InMemoryPauseTokenStore`."""
    self._store = InMemoryPauseTokenStore()

set_store ¶

set_store(store: PauseTokenStore) -> None

Swap the consumption store.

Idempotent; safe to call again. Does not close/teardown the previous store -- the caller owns store lifetime.

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

def set_store(self, store: PauseTokenStore) -> None:
    """Swap the consumption store.

    Idempotent; safe to call again. Does not close/teardown the
    previous store -- the caller owns store lifetime.
    """
    self._store = store

get_default_pause_token_manager ¶

get_default_pause_token_manager() -> PauseTokenManager

Return the process-wide default manager (legacy facade backing).

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

def get_default_pause_token_manager() -> PauseTokenManager:
    """Return the process-wide default manager (legacy facade backing)."""
    return _default_manager

symfonic.agent.middleware.pause_token¶

pause_token ¶

Ownership model (v7.1.3 — Item 10.2 fix)¶

PauseToken ¶

consume async classmethod ¶

decode classmethod ¶

encode classmethod ¶

hash_payload staticmethod ¶

hash_request staticmethod ¶

hash_scope staticmethod ¶

jti_to_timestamp staticmethod ¶

load_original_request async staticmethod ¶

reset_store_to_default classmethod ¶

set_store classmethod ¶

PauseTokenManager ¶

store property ¶

consume async ¶

reset_store_to_default ¶

set_store ¶

get_default_pause_token_manager ¶

`symfonic.agent.middleware.pause_token`¶

consume `async` `classmethod` ¶

decode `classmethod` ¶

encode `classmethod` ¶

hash_payload `staticmethod` ¶

hash_request `staticmethod` ¶

hash_scope `staticmethod` ¶

jti_to_timestamp `staticmethod` ¶

load_original_request `async` `staticmethod` ¶

reset_store_to_default `classmethod` ¶

set_store `classmethod` ¶

store `property` ¶

consume `async` ¶