symfonic.capabilities.human.binding

binding

Scope, session, and request binding.

Scope binding v2 includes every canonical kind/id pair and the orthogonal namespace. Pre-v2 hashes are ambiguous and must be refused, even under an explicit cross-scope registration; ask the question again after upgrading. Payload hashing retains its existing wire contract. Neither hash authenticates a token: authenticity belongs to its envelope.

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.
    """
    _require_current_scope_hash(claims.scope_hash)
    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}, 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

Bind the full root-first path and namespace without importing tenancy.

Structural callers must supply the canonical path, never only the lossy tenant/sub-tenant projection. JSON preserves kind/id and namespace boundaries (including None, empty and underscore); the prefix versions this contract.

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

def hash_scope(scope: Any) -> str:
    """Bind the full root-first path and namespace without importing tenancy.

    Structural callers must supply the canonical path, never only the lossy
    tenant/sub-tenant projection. JSON preserves kind/id and namespace boundaries
    (including None, empty and underscore); the prefix versions this contract.
    """
    path = getattr(scope, "path", None)
    namespace = getattr(scope, "namespace", None)
    if not isinstance(path, (tuple, list)) or not path:
        raise ScopeBindingError("pause scope requires a non-empty canonical path")
    pairs = [(getattr(level, "kind", None), getattr(level, "id", None)) for level in path]
    if any(not isinstance(value, str) or not value.strip() for pair in pairs for value in pair):
        raise ScopeBindingError("pause scope path requires non-empty string kind/id pairs")
    if namespace is not None and not isinstance(namespace, str):
        raise ScopeBindingError("pause scope namespace must be a string or None")
    raw = json.dumps(
        ["symfonic.pause-scope.v2", pairs, namespace], ensure_ascii=True, separators=(",", ":")
    )
    return "v2:" + hashlib.sha256(raw.encode("utf-8")).hexdigest()

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)