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

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

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

resume_stream async

resume_stream(token: str, answer: Any, *, scope: Any | None = None, admission: Callable[[Any, Any], Any] | None = None, owner_guard: Callable[[Any, str, str, bool], Awaitable[None]] | None = None, on_claimed: Callable[[], None] | None = None) -> AsyncIterator[Any]

Redeem once, then stream the recovered turn's typed events.

on_claimed is called synchronously once this call has won the single-use claim, so a caller can tell a failure after the token was spent from a refusal that left it redeemable.

Source code in src/symfonic/platform/human_continuation.py

async def resume_stream(
    self, token: str, answer: Any, *, scope: Any | None = None,
    admission: Callable[[Any, Any], Any] | None = None,
    owner_guard: Callable[[Any, str, str, bool], Awaitable[None]] | None = None,
    on_claimed: Callable[[], None] | None = None,
) -> AsyncIterator[Any]:
    """Redeem once, then stream the recovered turn's typed events.

    ``on_claimed`` is called synchronously once this call has won the
    single-use claim, so a caller can tell a failure after the token was
    spent from a refusal that left it redeemable.
    """
    agent, continuation, (admitted, _root) = await self._redeem(
        token, answer, scope, admission, owner_guard, on_claimed,
    )
    source = (
        agent.continue_recovered_typed(continuation)
        if admitted is None else agent.continue_recovered_typed(
            continuation, prepared_plan=admitted.prepared_plan,
            event_guard=admitted.event_guard,
        )
    )
    async with aclosing(source):
        async for event in source:
            yield event

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

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)