symfonic.capabilities.human.tokens

tokens

The pause-token service: mint, authenticate, bind, consume.

What this module owns is an ordering and a bound. Authenticity, key lifecycle and the pin-less legacy-artifact policy are consumed from T2.3.6 through :class:~.contracts.EnvelopeSignerPort -- no second signature scheme here, and a suite asserts the package imports neither hmac nor secrets.

The ordering, in the one place it exists: verify before any claim is read, so a forged token cannot steer its own decoding; decode, refusing a body this build cannot account for; expire, where a closed window is "ask again" and never "denied"; bind the four axes under the registration's posture; and claim, last and exactly once -- the one step an honest caller cannot retry.

PauseTokenService

PauseTokenService(*, signer: Any, ttl: TTLPolicy, consumption: Any = None, ledger: Any = None, clock: Callable[[], float] = time.time, pinless_policy: Any = None, binder: ScopeBinder | None = None)

Mint, validate, and consume pause tokens against exactly one winner-seam.

Source code in src/symfonic/capabilities/human/tokens.py

def __init__(
    self,
    *,
    signer: Any,
    ttl: TTLPolicy,
    consumption: Any = None,
    ledger: Any = None,
    clock: Callable[[], float] = time.time,
    pinless_policy: Any = None,
    binder: ScopeBinder | None = None,
) -> None:
    for verb in ("mint", "verify"):
        if not callable(getattr(signer, verb, None)):
            raise InteractionConfigurationError(
                f"the envelope signer must offer {verb}(...); got "
                f"{type(signer).__name__}. Authenticity is consumed, not remade"
            )
    if (consumption is None) == (ledger is None):
        raise InteractionConfigurationError(
            "exactly one consumption seam is required: an issuance ledger "
            "(operated platform) or a conditional-write store (library mode). "
            "Two seams are two places that each believe they decide the "
            "winner; none is no single-use enforcement at all"
        )
    self._signer = signer
    self._ttl = ttl
    self._ledger = ledger
    self._consumption = ledger if ledger is not None else consumption
    self._clock = clock
    self._pinless = pinless_policy
    self._binder = binder or ScopeBinder()

operated property

operated: bool

Whether an authoritative issuance ledger backs this service.

authenticate

authenticate(envelope: Any, *, at: float | None = None) -> ValidatedPause

Steps 1–3: verify, decode, expire. No state is touched.

Verification precedes decoding so a forged envelope cannot choose how it is read; expiry is checked after decoding because the expiry claim is part of what the signature covers, so trusting it earlier would let a forgery declare itself fresh.

Source code in src/symfonic/capabilities/human/tokens.py

def authenticate(self, envelope: Any, *, at: float | None = None) -> ValidatedPause:
    """Steps 1–3: verify, decode, expire. No state is touched.

    Verification precedes decoding so a forged envelope cannot choose how it
    is read; expiry is checked after decoding because the expiry claim is
    part of what the signature covers, so trusting it earlier would let a
    forgery declare itself fresh.
    """
    try:
        pin = self._signer.verify(
            envelope, pinless_policy=self._pinless, schema_id=self.SCHEMA_ID
        )
    except Exception as exc:  # every refusal denies; ``refusals.py`` says why
        raise PauseTokenUnauthorizedError.from_signer_refusal(exc) from exc
    # The signer only consults schema_id on its pin-less branch, so the
    # family is this reader's: same keyset, another schema, still not a pause.
    presented = getattr(envelope, "schema_id", "")
    if presented != self.SCHEMA_ID:
        raise PauseTokenUnauthorizedError(
            f"this envelope travels under schema {presented!r}, not "
            f"{self.SCHEMA_ID!r}; a token from another payload family is not "
            "redeemable here however well it verifies"
        )
    claims = PauseClaims.decode(envelope.payload)
    now = self._clock() if at is None else at
    if claims.expired(now):
        raise PauseTokenExpiredError(
            "this pause token has expired; the question was not answered "
            "inside its lifetime, so ask it again rather than resuming"
        )
    return ValidatedPause(claims=claims, pin=pin)

bind

bind(claims: PauseClaims, *, scope: Any, session_id: str | None = None, run_id: str | None = None, call_id: str | None = None, registration: Any = None) -> bool

Step 4 — the four axes (HK2). Returns whether scopes were crossed.

None is "not stated" for the last three; the binder says why each is separately refusable. The registration is the only thing that can open cross-scope redemption, and it is passed in -- rather than looked up, or reduced to a boolean a caller could pass -- so a caller cannot bind against a different posture than the one it validates the answer with.

Source code in src/symfonic/capabilities/human/tokens.py

def bind(
    self,
    claims: PauseClaims,
    *,
    scope: Any,
    session_id: str | None = None,
    run_id: str | None = None,
    call_id: str | None = None,
    registration: Any = None,
) -> bool:
    """Step 4 — the four axes (HK2). Returns whether scopes were crossed.

    ``None`` is "not stated" for the last three; the binder says why each is
    separately refusable. The registration is the *only* thing that can open
    cross-scope redemption, and it is passed in -- rather than looked up, or
    reduced to a boolean a caller could pass -- so a caller cannot bind
    against a different posture than the one it validates the answer with.
    """
    return self._binder.check(
        claims,
        scope=scope,
        session_id=session_id,
        run_id=run_id,
        call_id=call_id,
        cross_scope_allowed=bool(
            getattr(registration, "cross_scope_allowed", False)
        ),
    )

consume async

consume(claims: PauseClaims) -> None

Step 5 — the single atomic claim. Losing it is a replay, not a fault.

Source code in src/symfonic/capabilities/human/tokens.py

async def consume(self, claims: PauseClaims) -> None:
    """Step 5 — the single atomic claim. Losing it is a replay, not a fault."""
    won = await self._consumption.claim(
        claims.jti, scope_hash=claims.scope_hash, name=claims.name
    )
    if not won:
        raise PauseTokenReplayedError(
            "this pause token has already been redeemed; single use is enforced "
            "by one atomic operation, so exactly one caller wins"
        )

drain_proof async

drain_proof(now: float | None = None) -> Any

CUT-AIR-3 — proof that legacy-pinned tokens have drained.

Source code in src/symfonic/capabilities/human/tokens.py

async def drain_proof(self, now: float | None = None) -> Any:
    """CUT-AIR-3 — proof that legacy-pinned tokens have drained."""
    return await self._require_ledger("a drain proof").drain_proof(now)

mint async

mint(*, pin: Any, scope: Any, run_id: str, session_id: str, thread_id: str, checkpoint_id: str, payload: Any, root_run_id: str = '', name: str = ASK_USER, tool_call_id: str = '', interrupt_id: str | None = None, ttl_seconds: float | None = None, legacy_pinned: bool = False) -> MintedPause

Bind a pause to this scope, session, and request, and sign it.

Source code in src/symfonic/capabilities/human/tokens.py

async def mint(
    self,
    *,
    pin: Any,
    scope: Any,
    run_id: str,
    session_id: str,
    thread_id: str,
    checkpoint_id: str,
    payload: Any,
    root_run_id: str = "",
    name: str = ASK_USER,
    tool_call_id: str = "",
    interrupt_id: str | None = None,
    ttl_seconds: float | None = None,
    legacy_pinned: bool = False,
) -> MintedPause:
    """Bind a pause to this scope, session, and request, and sign it."""
    now = self._clock()
    lifetime = self._ttl.resolve(ttl_seconds)
    if interrupt_id is None:
        # ``i-<12 hex>`` -- the legacy correlation-id shape consumers match on.
        interrupt_id = "" if name == ASK_USER else f"i-{uuid.uuid4().hex[:12]}"
    claims = PauseClaims(
        run_id=run_id,
        root_run_id=root_run_id or run_id,
        session_id=session_id,
        scope_hash=self._binder.scope_hash(scope),
        thread_id=thread_id,
        checkpoint_id=checkpoint_id,
        request_hash=self._binder.request_hash(payload),
        exp=int(now + lifetime),
        jti=uuid.uuid4().hex,
        tool_call_id=tool_call_id,
        name=name,
        interrupt_id=interrupt_id,
        issued_at=now,
        legacy_pinned=legacy_pinned,
    )
    envelope = self._signer.mint(
        pin=pin, payload=claims.encode(), schema_id=self.SCHEMA_ID
    )
    if self._ledger is not None:
        await self._ledger.record_issuance(
            IssuedToken(
                jti=claims.jti,
                scope_hash=claims.scope_hash,
                name=claims.name,
                issued_at=now,
                expires_at=float(claims.exp),
                legacy_pinned=legacy_pinned,
                vector_hash=getattr(pin, "vector_hash", ""),
            )
        )
    return MintedPause(envelope=envelope, claims=claims)

record_retirement_horizon async

record_retirement_horizon(at: float, *, reason: str) -> None

SCP-FRZ-2 — the date past which nothing may be extended.

Source code in src/symfonic/capabilities/human/tokens.py

async def record_retirement_horizon(self, at: float, *, reason: str) -> None:
    """SCP-FRZ-2 — the date past which nothing may be extended."""
    await self._require_ledger("a retirement horizon").record_retirement_horizon(
        at, reason=reason
    )

scope_hash

scope_hash(scope: Any) -> str

The binding hash, so a caller can record what was presented.

Source code in src/symfonic/capabilities/human/tokens.py

def scope_hash(self, scope: Any) -> str:
    """The binding hash, so a caller can record what was presented."""
    return self._binder.scope_hash(scope)

validate async

validate(envelope: Any, *, scope: Any, session_id: str | None = None, run_id: str | None = None, call_id: str | None = None, at: float | None = None, registration: Any = None) -> ValidatedPause

Authenticate, then bind. Still consumes nothing.

Source code in src/symfonic/capabilities/human/tokens.py

async def validate(
    self,
    envelope: Any,
    *,
    scope: Any,
    session_id: str | None = None,
    run_id: str | None = None,
    call_id: str | None = None,
    at: float | None = None,
    registration: Any = None,
) -> ValidatedPause:
    """Authenticate, then bind. Still consumes nothing."""
    validated = self.authenticate(envelope, at=at)
    crossed = self.bind(
        validated.claims,
        scope=scope,
        session_id=session_id,
        run_id=run_id,
        call_id=call_id,
        registration=registration,
    )
    return ValidatedPause(validated.claims, validated.pin, crossed)

validate_and_consume async

validate_and_consume(envelope: Any, *, scope: Any, session_id: str | None = None, run_id: str | None = None, call_id: str | None = None, at: float | None = None, registration: Any = None) -> ValidatedPause

The whole ordering, for a caller with nothing to check in between.

Source code in src/symfonic/capabilities/human/tokens.py

async def validate_and_consume(
    self,
    envelope: Any,
    *,
    scope: Any,
    session_id: str | None = None,
    run_id: str | None = None,
    call_id: str | None = None,
    at: float | None = None,
    registration: Any = None,
) -> ValidatedPause:
    """The whole ordering, for a caller with nothing to check in between."""
    validated = await self.validate(
        envelope,
        scope=scope,
        session_id=session_id,
        run_id=run_id,
        call_id=call_id,
        at=at,
        registration=registration,
    )
    await self.consume(validated.claims)
    return validated