symfonic.capabilities.human.values

values

The frozen values a pause is made of.

:class:PauseClaims is a compatibility contract, not a convenience. Its field names are the claim names the legacy engine put in a JWT body, character for character, so a token minted before this extraction decodes here and a token minted here decodes there. Two fields are new — issued_at and legacy_pinned — and both default, so a legacy body is missing nothing.

The claims travel inside the T2.3.6 authenticated envelope rather than in a signature of their own. This package never signs anything; it decides what is bound, and something else vouches for it.

CrossScopeRedemption dataclass

CrossScopeRedemption(name: str, interrupt_id: str, jti: str, expected_scope_hash: str, presented_scope_hash: str, at: float)

An audit record for the one relaxation the contract permits.

InteractionEvent dataclass

InteractionEvent(name: str, payload: Any, pause: MintedPause, run_id: str, session_id: str, interrupt_id: str = '', tool_call_id: str = '', resumable: bool = False)

The pause a transport serialises: it owns the wire format, this the facts.

MintedPause dataclass

MintedPause(envelope: Any, claims: PauseClaims)

A signed envelope and the claims inside it, together.

PauseClaims dataclass

PauseClaims(run_id: str, session_id: str, scope_hash: str, thread_id: str, checkpoint_id: str, request_hash: str, exp: int, jti: str, root_run_id: str = '', tool_call_id: str = '', name: str = 'ask_user', interrupt_id: str = '', issued_at: float = 0.0, legacy_pinned: bool = False)

What the token binds: who, which run, which thread, which request.

as_legacy_dict

as_legacy_dict() -> dict[str, Any]

Exactly the legacy claim names, for a reader that predates this.

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

def as_legacy_dict(self) -> dict[str, Any]:
    """Exactly the legacy claim names, for a reader that predates this."""
    return {key: getattr(self, key) for key in LEGACY_CLAIM_KEYS}

decode classmethod

decode(raw: bytes) -> PauseClaims

Read a verified body. Called only after the envelope verified.

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

@classmethod
def decode(cls, raw: bytes) -> PauseClaims:
    """Read a verified body. Called only *after* the envelope verified."""
    try:
        body = json.loads(raw)
    except (ValueError, TypeError) as exc:
        raise PauseTokenUnauthorizedError(
            "pause-token claims are unreadable; a body that cannot be parsed "
            "is refused rather than partially honoured"
        ) from exc
    return cls.from_legacy_dict(body)

encode

encode() -> bytes

The canonical wire body the envelope signs over.

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

def encode(self) -> bytes:
    """The canonical wire body the envelope signs over."""
    body = self.as_legacy_dict()
    body.update({key: getattr(self, key) for key in sorted(_ADDED_CLAIM_KEYS)})
    return json.dumps(body, separators=(",", ":"), sort_keys=True).encode("utf-8")

expired

expired(now: float) -> bool

exp is inclusive: a token is live through its expiry second.

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

def expired(self, now: float) -> bool:
    """``exp`` is inclusive: a token is live *through* its expiry second."""
    return now > self.exp

from_legacy_dict classmethod

from_legacy_dict(body: Any) -> PauseClaims

Build from a claim mapping, refusing anything it cannot account for.

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

@classmethod
def from_legacy_dict(cls, body: Any) -> PauseClaims:
    """Build from a claim mapping, refusing anything it cannot account for."""
    if not isinstance(body, dict):
        raise PauseTokenUnauthorizedError(
            f"pause-token claims must be a mapping, got {type(body).__name__}"
        )
    known = set(LEGACY_CLAIM_KEYS) | _ADDED_CLAIM_KEYS
    unknown = sorted(set(body) - known)
    if unknown:
        # A verified envelope means the producer is trusted, not that this
        # build understands everything it wrote. Ignoring an unknown claim
        # would silently drop a binding a newer producer added on purpose.
        raise PauseTokenUnauthorizedError(
            f"pause-token claims carry {unknown}, which this build does not "
            "understand; it refuses rather than dropping a binding"
        )
    missing = sorted(_REQUIRED_CLAIM_KEYS - set(body))
    if missing:
        raise PauseTokenUnauthorizedError(
            f"pause-token claims are missing {missing}; refusing to guess"
        )
    return cls(**{key: body[key] for key in body})

ResumeCommand dataclass

ResumeCommand(envelope: Any, response: Any, scope: Any, session_id: str | None = None, run_id: str | None = None, call_id: str | None = None)

One redemption attempt, in the shape a transport can build from a request.

session_id, run_id and call_id are the three axes beside scope that a redemption is checked against (HK2). Each defaults to None, meaning "this transport did not state it", and an unstated axis is not checked -- which is why :mod:~symfonic.agent.cutover.kernel_resume builds this value with all four and refuses to build one without them. The default belongs to a caller that genuinely has no such fact; it must not be how a caller that should have had one silently opts out of the check.

ResumeOutcome dataclass

ResumeOutcome(name: str, thread_id: str, checkpoint_id: str, payload: Any, response: Any, run_id: str, session_id: str, tool_call_id: str = '', interrupt_id: str = '', cross_scope: bool = False, time_to_resolve_seconds: float = 0.0, turn: Any = None)

What a graph runner needs to continue, and what telemetry needs to record.

as_configurable

as_configurable() -> dict[str, Any]

The run-config shape the resume executes against.

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

def as_configurable(self) -> dict[str, Any]:
    """The run-config shape the resume executes against."""
    return {
        "configurable": {
            "thread_id": self.thread_id,
            "checkpoint_id": self.checkpoint_id,
        }
    }

TTLPolicy dataclass

TTLPolicy(default_seconds: float, maximum_seconds: float)

The lifetime a pause may have, and the bound nothing may exceed.

The maximum is the same number the operated ledger uses to bound its drain proof, which is why refusing is the only correct answer to a request past it: a clamped token would tell the caller they have a window they do not, and would make the drain deadline a guess.

ValidatedPause dataclass

ValidatedPause(claims: PauseClaims, pin: Any, cross_scope: bool = False)

What survived authentication, expiry, and binding.