symfonic.platform.agent_host

agent_host

The host and registry, implemented (task-1-2-2).

Contracts in :mod:symfonic.platform.host; this is the one implementation the generated scaffold builds. Nothing here takes a turn or inspects one.

AgentHost

AgentHost(*, composer: Any, resources: Any, open_resources: Any = None, continuation_factory: Any = None)

Owns the process's shared resources and the registry over them.

Built once by a composition root and closed once. open_resources is a separate callable rather than a method on the resources object because opening is not part of :class:~symfonic.platform.host.SharedResources: an adopter may hand over something already open, and requiring an aopen would make that a lie.

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

def __init__(
    self,
    *,
    composer: Any,
    resources: Any,
    open_resources: Any = None,
    continuation_factory: Any = None,
) -> None:
    self._resources = resources
    self._open = open_resources
    self._opening: asyncio.Task[None] | None = None
    self._closing: asyncio.Task[None] | None = None
    self._started = False
    self._closed = False
    self.registry = ScopedAgentRegistry(composer=composer, resources=resources)
    self._continuation = (
        None if continuation_factory is None else continuation_factory(self.agent_for)
    )

aclose async

aclose() -> None

Close the registry, then the resources, exactly once.

In that order: an agent holding a closed pool is a worse failure than a pool held a moment longer, because the first surfaces mid-turn and the second surfaces nowhere.

Resources are closed only if they were opened. A host whose construction failed before start is still closed by the handler, and closing what was never opened is how a shutdown path raises during a shutdown that was already going badly.

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

async def aclose(self) -> None:
    """Close the registry, then the resources, exactly once.

    In that order: an agent holding a closed pool is a worse failure than
    a pool held a moment longer, because the first surfaces mid-turn and
    the second surfaces nowhere.

    Resources are closed only if they were opened. A host whose
    construction failed before ``start`` is still closed by the handler,
    and closing what was never opened is how a shutdown path raises during
    a shutdown that was already going badly.
    """
    if self._closing is None:
        self._closed = True
        self._closing = asyncio.create_task(self._close_once())
    await asyncio.shield(self._closing)

resume async

resume(token: str, answer: Any, *, scope: Any | None = None) -> Any

Continue a durable human pause through this host's scoped agents.

A host with no configured continuation fails early instead of accepting a token it cannot safely bind to a scope. Compose human_continuation(...) with a verified claim-to-scope resolver.

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

async def resume(self, token: str, answer: Any, *, scope: Any | None = None) -> Any:
    """Continue a durable human pause through this host's scoped agents.

    A host with no configured continuation fails early instead of accepting
    a token it cannot safely bind to a scope.  Compose
    ``human_continuation(...)`` with a verified claim-to-scope resolver.
    """
    if self._closed:
        raise HostClosed("the host is closed; no continuation can be served")
    if self._continuation is None:
        from symfonic.platform.continuation import ContinuationUnavailable

        raise ContinuationUnavailable(
            "this AgentHost has no continuation service; compose "
            "human_continuation(capability=..., scope_for_claims=...)"
        )
    # Keep the original continuation port callable for roots that have no
    # request-authenticated scope (CLI or direct API use). The HTTP
    # adapter supplies a scope only for the explicit cross-scope binding
    # path; unconditionally forwarding ``scope=None`` would break a
    # pre-existing structural ContinuationService implementation.
    if scope is None:
        return await self._continuation.resume(token, answer)
    return await self._continuation.resume(token, answer, scope=scope)

resume_stream async

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

Stream a continuation from the scoped agent after one redemption.

on_claimed is accepted only with owner_guard (the guarded path).

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

async def resume_stream(
    self, token: str, answer: Any, *, scope: Any | None = None,
    admission: Any = None, owner_guard: Any = None, on_claimed: Any = None,
) -> AsyncIterator[Any]:
    """Stream a continuation from the scoped agent after one redemption.

    ``on_claimed`` is accepted only with ``owner_guard`` (the guarded path).
    """
    if self._closed:
        raise HostClosed("the host is closed; no continuation can be served")
    if self._continuation is None:
        from symfonic.platform.continuation import ContinuationUnavailable

        raise ContinuationUnavailable("this AgentHost has no continuation service")
    if ((admission is not None or owner_guard is not None)
            and not self._supports_resume_admission()):
        from symfonic.platform.continuation import ContinuationUnavailable

        reason = "budgeted" if owner_guard is None else "guarded"
        raise ContinuationUnavailable(f"continuation service cannot admit {reason} resume")
    resume_stream = getattr(self._continuation, "resume_stream", None)
    if not callable(resume_stream):
        # The published structural service only requires ``resume``.
        # Keep those roots working while the supported human adapter
        # supplies the incremental path.
        result = await self.resume(token, answer, scope=scope)
        if hasattr(result, "__aiter__"):
            async with aclosing(result):
                async for event in result:
                    yield event
        else:
            yield result
        return
    if on_claimed is not None and owner_guard is None:
        raise TypeError("on_claimed is reported only on the guarded resume path")
    if owner_guard is not None:
        source = resume_stream(
            token, answer, scope=scope, admission=admission, owner_guard=owner_guard,
            **({"on_claimed": on_claimed} if on_claimed is not None else {}),
        )
    else:
        source = self._continuation.resume_stream(
            token, answer, scope=scope, admission=admission,
        ) if admission is not None else (
            resume_stream(token, answer)
            if scope is None
            else resume_stream(token, answer, scope=scope)
        )
    async with aclosing(source):
        async for event in source:
            yield event

start async

start() -> None

Open shared resources once, however many times this is called.

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

async def start(self) -> None:
    """Open shared resources once, however many times this is called."""
    if self._closed and self._opening is None:
        return
    if self._opening is None and not self._closed:
        self._started = True
        if self._open is not None:
            self._opening = asyncio.create_task(self._open())
    if self._opening is not None:
        # Every caller observes the same completion or failure. A caller
        # cancelling its wait must not cancel opening for other callers.
        await asyncio.shield(self._opening)

ticket async

ticket(token: str) -> Any

Recover a continuation's bound scope and answer schema.

This verifies the opaque token but does not redeem it. HTTP adapters use the returned schema to refuse malformed answers before invoking :meth:resume, preserving the capability's single-use ordering.

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

async def ticket(self, token: str) -> Any:
    """Recover a continuation's bound scope and answer schema.

    This verifies the opaque token but does not redeem it.  HTTP adapters
    use the returned schema to refuse malformed answers before invoking
    :meth:`resume`, preserving the capability's single-use ordering.
    """
    if self._closed:
        raise HostClosed("the host is closed; no continuation can be served")
    if self._continuation is None:
        from symfonic.platform.continuation import ContinuationUnavailable

        raise ContinuationUnavailable(
            "this AgentHost has no continuation service; compose "
            "human_continuation(capability=..., scope_for_claims=...)"
        )
    return await self._continuation.ticket(token)

ScopedAgentRegistry

ScopedAgentRegistry(*, composer: Any, resources: Any)

One agent per scope, composed on first use.

Concurrency is handled with a per-scope lock rather than one global lock, so a slow composition for one tenant does not serialise every other tenant's first request. The double check inside the lock is what makes the composer run exactly once: the first waiter composes, the rest find the entry already there.

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

def __init__(self, *, composer: Any, resources: Any) -> None:
    self._composer = composer
    self._resources = resources
    self._agents: dict[str, Any] = {}
    self._locks: dict[str, asyncio.Lock] = {}
    # Guards the lock map itself. Without it two coroutines racing on a
    # brand-new scope each build their own lock and then each compose --
    # the race this class exists to prevent, moved one level down.
    self._guard = asyncio.Lock()
    self._closed = False

aclose async

aclose() -> None

Drop every agent. Idempotent, and does not touch shared resources.

Ownership runs one way: the registry borrows what the host opened, so closing it here would close a pool other registries may still hold -- and would do it from whichever one happened to close first.

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

async def aclose(self) -> None:
    """Drop every agent. Idempotent, and does not touch shared resources.

    Ownership runs one way: the registry borrows what the host opened, so
    closing it here would close a pool other registries may still hold --
    and would do it from whichever one happened to close first.
    """
    self._closed = True
    self._agents.clear()
    self._locks.clear()