symfonic.capabilities.human.capability

capability

The capability as one object: registrations in, pauses and resumes out.

A composition root wanting human interaction has one call to make and two verbs to use afterwards. Everything else — which metadata key a payload lives under, which claim binds which scope, which object decides the single winner, whether this deployment may be asked for a drain proof — is a rule, and a rule restated at a call site is a rule that can be restated differently at the next one.

HumanInteractionCapability

HumanInteractionCapability(*, registry: InteractionRegistry, tokens: PauseTokenService, payloads: PausePayloadStore, resumes: ResumeService, binding: Callable[[], Any] | None = None, encode_token: Callable[[Any], str] | None = None, decode_token: Callable[[str], Any] | None = None, turns: TurnCheckpointStore | None = None)

Registrations, tokens, checkpoint commands, and resume, wired once.

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

def __init__(
    self,
    *,
    registry: InteractionRegistry,
    tokens: PauseTokenService,
    payloads: PausePayloadStore,
    resumes: ResumeService,
    binding: Callable[[], Any] | None = None,
    encode_token: Callable[[Any], str] | None = None,
    decode_token: Callable[[str], Any] | None = None,
    turns: TurnCheckpointStore | None = None,
) -> None:
    self.registry = registry
    self.tokens = tokens
    self.payloads = payloads
    self.resumes = resumes
    self.binding = binding
    self.encode_token = encode_token
    #: The inverse of :attr:`encode_token`. Without it a pause taken on the
    #: kernel route has no public reader -- see ``agent.cutover.continuation``.
    self.decode_token = decode_token
    #: The checkpointer role (HK2). ``None`` is a deployment that can ask a
    #: question and never continue the run that asked it, which is what
    #: every deployment was before TA8.35 -- so it is a state, not a bug.
    self.turns = turns or TurnCheckpointStore()

active property

active: bool

True when at least one interaction is registered.

A deployment that registered none must not advertise ask_user: a tool whose only possible outcome is a refusal is an invitation to pause a run that can never be resumed.

compile classmethod

compile(registrations: Iterable[InteractionRegistration], *, signer: Any, ttl: TTLPolicy, consumption: Any = None, ledger: Any = None, checkpoints: Any = None, clock: Callable[[], float] = time.time, pinless_policy: Any = None, audit: Callable[[CrossScopeRedemption], None] | None = None, require_durable: bool = True, binding: Callable[[], Any] | None = None, encode_token: Callable[[Any], str] | None = None, decode_token: Callable[[str], Any] | None = None) -> HumanInteractionCapability

Wire the capability. Every argument is a port or a policy.

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

@classmethod
def compile(
    cls,
    registrations: Iterable[InteractionRegistration],
    *,
    signer: Any,
    ttl: TTLPolicy,
    consumption: Any = None,
    ledger: Any = None,
    checkpoints: Any = None,
    clock: Callable[[], float] = time.time,
    pinless_policy: Any = None,
    audit: Callable[[CrossScopeRedemption], None] | None = None,
    require_durable: bool = True,
    binding: Callable[[], Any] | None = None,
    encode_token: Callable[[Any], str] | None = None,
    decode_token: Callable[[str], Any] | None = None,
) -> HumanInteractionCapability:
    """Wire the capability. Every argument is a port or a policy."""
    registry = InteractionRegistry()
    for registration in registrations:
        registry.register(registration)
    tokens = PauseTokenService(
        signer=signer,
        ttl=ttl,
        consumption=consumption,
        ledger=ledger,
        clock=clock,
        pinless_policy=pinless_policy,
    )
    payloads = PausePayloadStore(commands=checkpoints)
    # One port, two records. A deployment cannot end up with durable
    # requests and volatile turn state, or the reverse, and then find out
    # only when somebody answers a question.
    turns = TurnCheckpointStore(commands=checkpoints)
    if require_durable:
        require_durable_consumption(
            consumption if consumption is not None else ledger,
            checkpoints_durable=payloads.durable,
        )
    return cls(
        registry=registry,
        tokens=tokens,
        payloads=payloads,
        resumes=ResumeService(
            tokens=tokens,
            registry=registry,
            payloads=payloads,
            clock=clock,
            audit=audit,
            turns=turns,
        ),
        binding=binding,
        encode_token=encode_token,
        decode_token=decode_token,
        turns=turns,
    )

contribute

contribute(request: CapabilityRequest) -> CapabilityContribution

Offer the pause point to the turn being compiled (HK1, TA8.34).

A stage and a tool, and :mod:~symfonic.capabilities.human.contribution holds the reasoning for why both: the tool is how the model asks, the pre-tool stage is where the run actually stops, and only the stage sees the reserved call id a pause has to be bound to.

Contributing nothing is a real answer, the same one DelegationCapability gives for a parent with no children. Three states produce it, and each is a deployment that cannot serve a pause: nothing registered, no per-run :class:PauseBinding resolver, or no way to render a minted envelope as the opaque token a consumer answers with. Advertising ask_user in any of them would invite the model to stop a run that could never be resumed -- and a run stopped by a pause nobody can answer is strictly worse than one that never stopped. The contribution still carries the capability's name, so a plan records that human interaction was folded and found nothing rather than that it was never folded.

request is read for its grants and found to need none. Minting a token, recording its payload and writing the issuance row all go through this capability's own ports, which the composition root wired; none of them is an :class:~symfonic.kernel.contracts.effects.EffectFamily the plan grants, and declaring one it was not granted is refused at fold rather than at the point of the effect (STG-8).

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

def contribute(self, request: CapabilityRequest) -> CapabilityContribution:
    """Offer the pause point to the turn being compiled (HK1, TA8.34).

    **A stage and a tool**, and :mod:`~symfonic.capabilities.human.contribution`
    holds the reasoning for why both: the tool is how the model asks, the
    ``pre-tool`` stage is where the run actually stops, and only the stage
    sees the reserved call id a pause has to be bound to.

    **Contributing nothing is a real answer**, the same one
    ``DelegationCapability`` gives for a parent with no children. Three
    states produce it, and each is a deployment that cannot serve a pause:
    nothing registered, no per-run :class:`PauseBinding` resolver, or no
    way to render a minted envelope as the opaque token a consumer answers
    with. Advertising ``ask_user`` in any of them would invite the model to
    stop a run that could never be resumed -- and a run stopped by a pause
    nobody can answer is strictly worse than one that never stopped. The
    contribution still carries the capability's name, so a plan records
    that human interaction was folded and found nothing rather than that it
    was never folded.

    ``request`` is read for its grants and found to need none. Minting a
    token, recording its payload and writing the issuance row all go
    through this capability's own ports, which the composition root wired;
    none of them is an :class:`~symfonic.kernel.contracts.effects.EffectFamily`
    the plan grants, and declaring one it was not granted is refused at
    fold rather than at the point of the effect (STG-8).
    """
    return build_contribution(
        self,
        request,
        binding=self.binding,
        encode_token=self.encode_token,
    )

pause async

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

Mint a pause, record its payload, and record the turn it stopped.

turn_state is what makes the pause redeemable (HK2). It is optional because a caller may have nothing continuable to record -- a transport minting a pause outside a kernel turn, for instance -- and the returned event then says resumable=False rather than pretending otherwise.

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

async def pause(
    self,
    *,
    pin: Any,
    scope: Any,
    run_id: str,
    session_id: str,
    thread_id: str,
    payload: Any,
    root_run_id: str = "",
    name: str = ASK_USER,
    tool_call_id: str = "",
    checkpoint_id: str | None = None,
    ttl_seconds: float | None = None,
    legacy_pinned: bool = False,
    turn_state: TurnCheckpoint | None = None,
) -> InteractionEvent:
    """Mint a pause, record its payload, and record the turn it stopped.

    ``turn_state`` is what makes the pause redeemable (HK2). It is optional
    because a caller may have nothing continuable to record -- a transport
    minting a pause outside a kernel turn, for instance -- and the returned
    event then says ``resumable=False`` rather than pretending otherwise.
    """
    registration = self.registry.get(name)
    try:
        registration.payload_schema.model_validate(payload)
    except Exception as exc:
        # The shipped minter logged this and returned ``None``, producing a
        # pause with no token and therefore no way to resume. A refusal
        # stops the run where the bug is.
        raise PayloadBindingError(
            f"the {name!r} payload does not fit its registered schema, so no "
            f"pause token is minted for it: {exc}"
        ) from exc
    # Everything a pause can be refused for is refused before the mint, so
    # a failed pause leaves no issued row the ledger will count as
    # outstanding and nothing can ever consume.
    require_payload_key_inputs(name, tool_call_id=tool_call_id)
    resolved = checkpoint_id or await self.payloads.resolve_checkpoint_id(thread_id)
    if not resolved:
        raise PauseCheckpointNotFoundError(
            f"thread {thread_id!r} has no checkpoint to pause at; a token "
            "bound to no checkpoint could never recover its own request"
        )
    minted = await self.tokens.mint(
        pin=pin,
        scope=scope,
        run_id=run_id,
        root_run_id=root_run_id,
        session_id=session_id,
        thread_id=thread_id,
        checkpoint_id=resolved,
        payload=payload,
        name=name,
        tool_call_id=tool_call_id,
        ttl_seconds=ttl_seconds,
        legacy_pinned=legacy_pinned,
    )
    await self.payloads.record(minted.claims, payload, task_id=run_id)
    # Earned, in this order and no other. The write is attempted, the write
    # reports, and only a report of success makes the pause say it can be
    # resumed. Anything that read ``resumable`` before this line would be
    # reading an intention.
    resumable = False
    if turn_state is not None:
        resumable = await self.turns.record(
            minted.claims, stamped_turn(turn_state, minted.claims)
        )
    return InteractionEvent(
        name=name,
        payload=payload,
        pause=minted,
        run_id=run_id,
        session_id=session_id,
        interrupt_id=minted.claims.interrupt_id,
        tool_call_id=minted.claims.tool_call_id,
        resumable=resumable,
    )