symfonic.agent.middleware.interrupt_minter

interrupt_minter

Generic interrupt minter — Roadmap Item 9 (v7.2-bound).

Carved out of :class:SymfonicAgent so the file stays bounded while the generic interrupt path lives in a focused module.

Responsibilities

• Mint a HMAC-signed pause token for a registered named interrupt.
• Build the public :class:InterruptEvent echoed back to stream consumers.
• Persist the original payload to checkpoint metadata under interrupt_payload:<interrupt_id> so the resume path can recover it without re-running the originating tool.

The minter intentionally does NOT know about ask_user -- that path keeps its dedicated _mint_pause_token codepath in engine.py so a regression here cannot influence the byte-identical event stream acceptance criterion (criterion 5 in the roadmap).

load_original_payload async

load_original_payload(
    *, checkpointer: Any, claims: PauseTokenClaims
) -> dict[str, Any] | None

Recover the payload stored at mint time so the hash can be reverified.

Returns the dict written under interrupt_payload:<interrupt_id> at mint time, or None if the checkpoint is gone (e.g. MemorySaver worker restart). Callers MUST handle None -- typically by surfacing a 424 failed_dependency.

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

async def load_original_payload(
    *,
    checkpointer: Any,
    claims: PauseTokenClaims,
) -> dict[str, Any] | None:
    """Recover the payload stored at mint time so the hash can be reverified.

    Returns the dict written under ``interrupt_payload:<interrupt_id>`` at
    mint time, or ``None`` if the checkpoint is gone (e.g. MemorySaver
    worker restart). Callers MUST handle ``None`` -- typically by
    surfacing a 424 ``failed_dependency``.
    """
    if checkpointer is None:
        return None

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

    import json as _json
    meta_key = f"interrupt_payload:{claims.interrupt_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)
                payload = _json.loads(raw)
                if PauseToken.hash_payload(payload) == claims.request_hash:
                    return payload  # type: ignore[no-any-return]
            except Exception:
                pass
    return None

mint_interrupt_token async

mint_interrupt_token(
    *,
    run_id: str,
    session_id: str,
    scope: FrameworkTenantScope,
    interrupt_data: dict[str, Any],
    registration: InterruptRegistration,
    config: FrameworkConfig,
    checkpointer: Any,
    ttl_seconds: int,
) -> InterruptEvent | None

Mint a pause token for a generic (non-ask_user) interrupt.

Parameters

run_id, session_id, scope Standard run identity. session_id may be empty; in that case the thread_id falls back to run_id. interrupt_data The dict surfaced by LangGraph's interrupt() call inside :class:InterruptNode. Expected keys: name, interrupt_id, payload, optionally tool_call_id. registration The :class:InterruptRegistration looked up from the agent's _registered_interrupts dict. Used here only to validate the payload one more time before minting -- defence-in-depth against a faulty interrupt-marker on the wire. config Live FrameworkConfig -- supplied so PauseToken.encode can read JWT_SECRET once via the same indirection used by the ask_user path. checkpointer Active LangGraph saver. aget_tuple is called to recover the freshly-minted checkpoint_id so the claims bind to a specific checkpoint, and aput_writes stores the payload keyed by interrupt_id for the resume path. ttl_seconds Token expiry. Mirrors ask_user_pause_ttl_seconds knob.

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

async def mint_interrupt_token(
    *,
    run_id: str,
    session_id: str,
    scope: FrameworkTenantScope,
    interrupt_data: dict[str, Any],
    registration: InterruptRegistration,
    config: FrameworkConfig,
    checkpointer: Any,
    ttl_seconds: int,
) -> InterruptEvent | None:
    """Mint a pause token for a generic (non-``ask_user``) interrupt.

    Parameters
    ----------
    run_id, session_id, scope
        Standard run identity. ``session_id`` may be empty; in that
        case the thread_id falls back to ``run_id``.
    interrupt_data
        The dict surfaced by LangGraph's ``interrupt()`` call inside
        :class:`InterruptNode`. Expected keys: ``name``,
        ``interrupt_id``, ``payload``, optionally ``tool_call_id``.
    registration
        The :class:`InterruptRegistration` looked up from the agent's
        ``_registered_interrupts`` dict.  Used here only to validate
        the payload one more time before minting -- defence-in-depth
        against a faulty interrupt-marker on the wire.
    config
        Live ``FrameworkConfig`` -- supplied so ``PauseToken.encode``
        can read ``JWT_SECRET`` once via the same indirection used
        by the ``ask_user`` path.
    checkpointer
        Active LangGraph saver. ``aget_tuple`` is called to recover
        the freshly-minted ``checkpoint_id`` so the claims bind to a
        specific checkpoint, and ``aput_writes`` stores the payload
        keyed by interrupt_id for the resume path.
    ttl_seconds
        Token expiry. Mirrors ``ask_user_pause_ttl_seconds`` knob.
    """
    name: str = interrupt_data["name"]
    interrupt_id: str = interrupt_data["interrupt_id"]
    payload: dict[str, Any] = interrupt_data.get("payload", {}) or {}
    tool_call_id: str = interrupt_data.get("tool_call_id") or ""

    # Defence-in-depth: revalidate the payload against the registered
    # schema before minting a token. A bad payload here is a framework
    # bug (the public ``SymfonicAgent.interrupt`` already validated on
    # the way in) but a stray ``state["_interrupt_pending"]`` written by
    # an unguarded caller should never get a pause token.
    try:
        registration.payload_schema.model_validate(payload)
    except Exception as exc:
        logger.error(
            "interrupt payload validation failed at mint time "
            "(name=%s interrupt_id=%s): %s",
            name, interrupt_id, exc,
        )
        return None

    thread_id = f"{scope.tenant_id}:{scope.sub_tenant_id or '_'}:{session_id}"
    if not session_id:
        thread_id = f"{scope.tenant_id}:{scope.sub_tenant_id or '_'}:{run_id}"

    checkpoint_tuple = await checkpointer.aget_tuple(
        {"configurable": {"thread_id": thread_id}}
    )
    if not checkpoint_tuple:
        logger.warning(
            "Failed to get checkpoint for interrupt thread_id=%s name=%s",
            thread_id, name,
        )
        return None

    checkpoint_id = checkpoint_tuple.config["configurable"]["checkpoint_id"]

    # Hash the payload -- this binds the token to *this specific* payload
    # so a tampered resume body cannot redeem a token minted for a
    # different request.  Mirrors the ``ask_user`` request_hash.
    request_hash = PauseToken.hash_payload(payload)

    # Persist the payload under interrupt_payload:<interrupt_id> so the
    # resume path can verify the hash without re-running the originating
    # tool. Same convention as ``ask_user_request:<tool_call_id>``.
    import json as _json
    meta_key = f"interrupt_payload:{interrupt_id}"
    try:
        await checkpointer.aput_writes(
            {"configurable": {"thread_id": thread_id, "checkpoint_id": checkpoint_id}},
            [(meta_key, _json.dumps(payload, default=str))],
            run_id,
        )
    except Exception:
        # Not all checkpointers support aput_writes; the resume path
        # falls back to re-validating against the registered schema.
        logger.debug(
            "Could not persist interrupt payload metadata for "
            "interrupt_id=%s",
            interrupt_id,
        )

    expires_at = datetime.now(UTC) + timedelta(seconds=ttl_seconds)
    claims = PauseTokenClaims(
        run_id=run_id,
        session_id=session_id,
        scope_hash=PauseToken.hash_scope(scope),
        thread_id=thread_id,
        checkpoint_id=checkpoint_id,
        tool_call_id=tool_call_id,
        request_hash=request_hash,
        exp=int(expires_at.timestamp()),
        jti=uuid.uuid4().hex,
        name=name,
        interrupt_id=interrupt_id,
    )
    pause_token = PauseToken.encode(claims, config)

    return InterruptEvent(
        name=name,
        pause_token=pause_token,
        payload=payload,
        expires_at=expires_at,
        interrupt_id=interrupt_id,
        run_id=run_id,
        session_id=session_id,
    )