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._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._closed:
        return
    self._closed = True
    await self.registry.aclose()
    if self._started and self._resources is not None:
        await self._resources.aclose()

resume async

resume(token: str, answer: Any) -> 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) -> 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=...)"
        )
    return await self._continuation.resume(token, answer)

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._started or self._closed:
        return
    # Set before awaiting, so a second caller arriving mid-open does not
    # start a second one. Two opens is worse than one wait: it is two pools.
    self._started = True
    if self._open is not None:
        await self._open()

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()