Skip to content

symfonic.platform.human_continuation

human_continuation

The supported host composition for durable human-interaction recovery.

The public Agent is deliberately scope-bound. A pause token records a hash of that scope, never a tenant identifier, so this adapter authenticates a token before asking the deployment to resolve its scope. The resolver is an explicit composition dependency: guessing a tenant from an opaque token would turn recovery into a cross-tenant read.

HostContinuationFactory 

HostContinuationFactory(capability: Any, scope_for_claims: Callable[[PauseClaims], Any])

Bind host-owned agent lookup after a deployment has composed its ports.

Source code in src/symfonic/platform/human_continuation.py 
133
134
135
def __init__(self, capability: Any, scope_for_claims: Callable[[PauseClaims], Any]) -> None:
    self._capability = capability
    self._scope_for_claims = scope_for_claims

HumanContinuationService 

HumanContinuationService(*, capability: Any, scope_for_claims: Callable[[PauseClaims], Any], agent_for: Callable[[Any], Awaitable[Any]])

Redeem durable human pauses and route the resulting turn through a host.

scope_for_claims receives only claims that passed signature and expiry verification. It must resolve the deployment's durable scope record and may return None when the scope is no longer known; that is rejected before a capability or agent is asked to continue the turn.

Source code in src/symfonic/platform/human_continuation.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def __init__(
    self,
    *,
    capability: Any,
    scope_for_claims: Callable[[PauseClaims], Any],
    agent_for: Callable[[Any], Awaitable[Any]],
) -> None:
    if not callable(getattr(capability, "resume", None)):
        raise TypeError(
            "human continuation needs a HumanInteractionCapability with resume(); "
            f"got {type(capability).__name__}"
        )
    if not callable(getattr(capability, "decode_token", None)):
        raise TypeError(
            "human continuation needs decode_token so an opaque pause can be "
            "authenticated on a fresh host"
        )
    self._capability = capability
    self._scope_for_claims = scope_for_claims
    self._agent_for = agent_for

human_continuation 

human_continuation(*, capability: Any, scope_for_claims: Callable[[PauseClaims], Any]) -> HostContinuationFactory

Create the factory passed as AgentHost(continuation_factory=...).

The explicit scope resolver is the migration boundary for a legacy SQLite checkpoint path: it replaces an accepted-but-inert path setting with a durable token-to-scope route that the deployment owns and can test.

Source code in src/symfonic/platform/human_continuation.py 
145
146
147
148
149
150
151
152
153
154
def human_continuation(
    *, capability: Any, scope_for_claims: Callable[[PauseClaims], Any]
) -> HostContinuationFactory:
    """Create the factory passed as ``AgentHost(continuation_factory=...)``.

    The explicit scope resolver is the migration boundary for a legacy SQLite
    checkpoint path: it replaces an accepted-but-inert path setting with a
    durable token-to-scope route that the deployment owns and can test.
    """
    return HostContinuationFactory(capability, scope_for_claims)