symfonic.capabilities.human.binding

binding

Scope, session, and request binding.

The two hash formulas below are frozen. They are not chosen here — they were chosen by the engine that minted every token currently in flight, and a token minted before this extraction must validate after it. Both are reproduced character for character, including the _ placeholder and the truncation to sixteen hex digits.

Neither hash is a signature. Authenticity is the envelope's job (T2.3.6); these answer "is this the same scope / the same request?" once authenticity is already established.

ScopeBinder

Checks the bindings a pause token carries. Holds no state.

call_id_of staticmethod

call_id_of(claims: PauseClaims) -> str

Which id this pause correlates on -- the split, in one expression.

ask_user joins on the reserved tool_call_id and a registered interrupt on its own interrupt_id; a claim carries whichever its family minted. Derived from the claims rather than from which subsystem is asking, exactly as :func:~symfonic.capabilities.human.checkpoints.payload_key derives the metadata key.

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

@staticmethod
def call_id_of(claims: PauseClaims) -> str:
    """Which id this pause correlates on -- the split, in one expression.

    ``ask_user`` joins on the reserved ``tool_call_id`` and a registered
    interrupt on its own ``interrupt_id``; a claim carries whichever its
    family minted. Derived from the claims rather than from which subsystem
    is asking, exactly as
    :func:`~symfonic.capabilities.human.checkpoints.payload_key` derives the
    metadata key.
    """
    return claims.tool_call_id or claims.interrupt_id

check

check(claims: PauseClaims, *, scope: Any, session_id: str | None = None, run_id: str | None = None, call_id: str | None = None, cross_scope_allowed: bool = False) -> bool

Return whether this redemption crossed scopes. Raise when it may not.

Four axes, checked one at a time (HK2). Scope says which tenant, session says which conversation, run says which turn of it, and the call id says which of that turn's questions. They are four independent facts and each is separately refusable, because a redemption that satisfies three of them is a real, reachable mistake rather than a hypothetical: an operator with two paused runs open answers the wrong tab (run), an onboarding agent that asked twice gets the second answer filed against the first question (call), and a shared-inbox admin answers a colleague's session (session). Only the tenancy axis has ever had a bypass.

None means "not stated" for the last three, and a caller that does not state them gets no check for them. That is why the resume route states all four rather than trusting this default: the default is what a transport left out, and a transport that forgets an axis must not be the thing that decides the axis does not matter.

The session, run and call checks run regardless of cross_scope_allowed: crossing scopes is about who redeems (a tenant admin approving a sub-tenant's action), never about which run gets resumed. Relaxing them all at once would let one opt-in reach every paused run in the deployment.

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

def check(
    self,
    claims: PauseClaims,
    *,
    scope: Any,
    session_id: str | None = None,
    run_id: str | None = None,
    call_id: str | None = None,
    cross_scope_allowed: bool = False,
) -> bool:
    """Return whether this redemption crossed scopes. Raise when it may not.

    **Four axes, checked one at a time (HK2).** Scope says which tenant,
    session says which conversation, run says which turn of it, and the call
    id says which of that turn's questions. They are four independent facts
    and each is separately refusable, because a redemption that satisfies
    three of them is a real, reachable mistake rather than a hypothetical:
    an operator with two paused runs open answers the wrong tab (run), an
    onboarding agent that asked twice gets the second answer filed against
    the first question (call), and a shared-inbox admin answers a colleague's
    session (session). Only the tenancy axis has ever had a bypass.

    ``None`` means "not stated" for the last three, and a caller that does
    not state them gets no check for them. That is why the resume route
    states all four rather than trusting this default: the default is what a
    *transport* left out, and a transport that forgets an axis must not be
    the thing that decides the axis does not matter.

    The session, run and call checks run regardless of
    ``cross_scope_allowed``: crossing scopes is about *who* redeems (a
    tenant admin approving a sub-tenant's action), never about which run
    gets resumed. Relaxing them all at once would let one opt-in reach every
    paused run in the deployment.
    """
    presented = self.scope_hash(scope)
    cross_scope = presented != claims.scope_hash
    if cross_scope and not cross_scope_allowed:
        raise ScopeBindingError(
            "this pause token was minted under a different tenant scope; "
            "redeeming it here would cross a tenancy boundary the "
            "registration did not open"
        )
    if session_id is not None and session_id != claims.session_id:
        raise SessionBindingError(
            f"this pause token belongs to session {claims.session_id!r}, "
            f"not {session_id!r}"
        )
    if run_id is not None and run_id != claims.run_id:
        raise RunBindingError(
            f"this pause token belongs to run {claims.run_id!r}, not "
            f"{run_id!r}; it answers a question a different turn asked"
        )
    if call_id is not None and call_id != self.call_id_of(claims):
        raise CallBindingError(
            f"this pause token belongs to call {self.call_id_of(claims)!r}, "
            f"not {call_id!r}; a run that paused more than once has one live "
            "token per question and they are not interchangeable"
        )
    return cross_scope

check_payload

check_payload(claims: PauseClaims, *, payload: Any) -> None

SEC-PTK-5 — the paused request is the one the token was minted for.

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

def check_payload(self, claims: PauseClaims, *, payload: Any) -> None:
    """SEC-PTK-5 — the paused request is the one the token was minted for."""
    if self.request_hash(payload) != claims.request_hash:
        raise PayloadBindingError(
            "the recovered request does not hash to the token's "
            "request_hash; the paused request was altered between mint and "
            "resume, and this refuses rather than resuming the new one"
        )

payload_body_matches

payload_body_matches(claims: PauseClaims, *, body: str) -> bool

Whether a recorded body hashes to the token's request_hash.

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

def payload_body_matches(self, claims: PauseClaims, *, body: str) -> bool:
    """Whether a recorded body hashes to the token's ``request_hash``."""
    return hash_payload_body(body) == claims.request_hash

request_body

request_body(payload: Any) -> str

What to record, so that what is recorded is what was hashed.

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

def request_body(self, payload: Any) -> str:
    """What to record, so that what is recorded is what was hashed."""
    return payload_body(payload)

hash_payload

hash_payload(payload: Any) -> str

The legacy PauseToken.hash_payload formula, unchanged.

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

def hash_payload(payload: Any) -> str:
    """The legacy ``PauseToken.hash_payload`` formula, unchanged."""
    return hash_payload_body(payload_body(payload))

hash_payload_body

hash_payload_body(body: str) -> str

Hash an already-serialised payload body.

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

def hash_payload_body(body: str) -> str:
    """Hash an already-serialised payload body."""
    return hashlib.sha256(body.encode()).hexdigest()

hash_scope

hash_scope(scope: Any) -> str

The legacy PauseToken.hash_scope formula, unchanged.

Read structurally — the tenancy scope type lives outside this layer and must not be imported into it. An object with no tenant_id is refused rather than hashed as the empty tenant, which would make every unauthenticated caller share one scope.

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

def hash_scope(scope: Any) -> str:
    """The legacy ``PauseToken.hash_scope`` formula, unchanged.

    Read structurally — the tenancy scope type lives outside this layer and must
    not be imported into it. An object with no ``tenant_id`` is refused rather
    than hashed as the empty tenant, which would make every unauthenticated
    caller share one scope.
    """
    tenant_id = getattr(scope, "tenant_id", None)
    if not tenant_id:
        raise ScopeBindingError(
            "cannot bind a pause token to a scope with no tenant_id; an "
            "unattributed token is one every caller could redeem"
        )
    sub_tenant_id = getattr(scope, "sub_tenant_id", None) or _ABSENT
    namespace = getattr(scope, "namespace", None) or _ABSENT
    raw = f"{tenant_id}:{sub_tenant_id}:{namespace}"
    return hashlib.sha256(raw.encode()).hexdigest()[:16]

payload_body

payload_body(payload: Any) -> str

The exact text :func:hash_payload hashes.

A model serialises itself; anything else goes through canonical JSON with default=str so an unserialisable value degrades to a stable string rather than raising in the middle of a mint.

Split out from the hash so the recorded payload can be the same bytes the hashed payload was. Recording a differently-serialised body and re-hashing whatever survives the round trip is how a model payload — and every token the pre-extraction engine minted, which stored request.model_dump_json() — ends up answering "the paused request was altered" at resume time.

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

def payload_body(payload: Any) -> str:
    """The exact text :func:`hash_payload` hashes.

    A model serialises itself; anything else goes through canonical JSON with
    ``default=str`` so an unserialisable value degrades to a stable string
    rather than raising in the middle of a mint.

    Split out from the hash so the *recorded* payload can be the same bytes the
    *hashed* payload was. Recording a differently-serialised body and re-hashing
    whatever survives the round trip is how a model payload — and every token the
    pre-extraction engine minted, which stored ``request.model_dump_json()`` —
    ends up answering "the paused request was altered" at resume time.
    """
    dump = getattr(payload, "model_dump_json", None)
    return dump() if callable(dump) else json.dumps(payload, sort_keys=True, default=str)