Skip to content

symfonic.platform.agent_boundary

agent_boundary

Keep the compatibility facade out of native AgentHost composition.

UnsupportedAgentComposer

Bases: TypeError

An AgentHost composer returned a result the native boundary refuses.

refuse_awaitable_agent 

refuse_awaitable_agent(agent: Any) -> None

Refuse an awaitable composer result without ever driving it.

AgentComposer is synchronous; an awaitable hides its agent from the boundary check. A future is cancelled and a coroutine closed; any other awaitable is borrowed and left alone. An Exception during cleanup does not mask the refusal; cancellation and process interrupts still propagate.

Source code in src/symfonic/platform/agent_boundary.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def refuse_awaitable_agent(agent: Any) -> None:
    """Refuse an awaitable composer result without ever driving it.

    AgentComposer is synchronous; an awaitable hides its agent from the
    boundary check. A future is cancelled and a coroutine closed; any other
    awaitable is borrowed and left alone. An Exception during cleanup does
    not mask the refusal; cancellation and process interrupts still propagate.
    """
    if not inspect.isawaitable(agent):
        return
    with suppress(Exception):
        if isinstance(agent, asyncio.Future):
            agent.cancel()
        elif inspect.iscoroutine(agent):
            agent.close()
    raise UnsupportedAgentComposer("AgentHost composer must be synchronous")

require_supported_agent 

require_supported_agent(agent: Any) -> None

Reject the known incompatible producer without importing the engine.

A constructed SymfonicAgent necessarily has its defining module loaded; looking it up here keeps the engine off native imports and also catches subclasses. Other structural host agents remain supported.

Source code in src/symfonic/platform/agent_boundary.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def require_supported_agent(agent: Any) -> None:
    """Reject the known incompatible producer without importing the engine.

    A constructed SymfonicAgent necessarily has its defining module loaded;
    looking it up here keeps the engine off native imports and also catches
    subclasses. Other structural host agents remain supported.
    """
    engine = sys.modules.get("symfonic.agent.engine")
    legacy_type = getattr(engine, "SymfonicAgent", None)
    if isinstance(legacy_type, type) and isinstance(agent, legacy_type):
        raise UnsupportedAgentComposer(
            "AgentHost composer returned SymfonicAgent; native routes require "
            "an AgentEvent-producing agent (use symfonic.Agent)"
        )