Nested Agent Streaming Design

class ToolContext:
    """Injected into tools that declare a ToolContext parameter."""

    def __init__(self, agent_name: str, queue: asyncio.Queue[StreamEvent]):
        self._queue = queue
        self.agent_name = agent_name

    def emit(self, event: StreamEvent) -> None:
        """Push an event to the parent agent's stream."""
        self._queue.put_nowait(event)

@tool
async def research(query: str, ctx: ToolContext) -> str:
    parts = []
    async for event in inner_agent.stream(query):
        ctx.emit(event)
        if isinstance(event, TextEvent):
            parts.append(event.text)
    return "".join(parts)

# runner.py — after _execute_tools(), before ToolResultEvent yield

# Drain inner agent events pushed by tools via ToolContext.emit()
while not agent._event_queue.empty():
    try:
        inner_event = agent._event_queue.get_nowait()
        if _passes_filter(inner_event):
            yield inner_event
    except Exception:
        break

# Existing MCP progress drain follows (unchanged)

class RalphIterationEvent(BaseModel):
    """Emitted at the start/end of each Ralph loop iteration."""
    model_config = {"frozen": True}
    type: Literal["ralph_iteration"] = "ralph_iteration"
    iteration: int
    status: Literal["started", "completed", "failed"]
    scores: dict[str, float] = {}
    agent_name: str = ""

class RalphStopEvent(BaseModel):
    """Emitted when the Ralph loop terminates."""
    model_config = {"frozen": True}
    type: Literal["ralph_stop"] = "ralph_stop"
    stop_type: str
    reason: str
    iterations: int
    final_scores: dict[str, float] = {}
    agent_name: str = ""

@classmethod
def from_agent(cls, agent, scorers, **kwargs) -> RalphRunner:
    async def _execute(input: str) -> str:
        result = await run(agent, input)
        return result.output

    async def _stream(input: str) -> AsyncIterator[StreamEvent]:
        async for event in run.stream(agent, input):
            yield event

    return cls(execute_fn=_execute, stream_execute_fn=_stream, scorers=scorers, **kwargs)

async def stream(self, input: str, *, name: str = "ralph") -> AsyncIterator[StreamEvent]:
    state = LoopState()
    current_input = input

    while True:
        state.iteration += 1
        yield RalphIterationEvent(iteration=state.iteration, status="started", agent_name=name)

        # Phase 1: Run — stream the execute_fn, forward all inner events
        output_parts = []
        success = True
        try:
            async for event in self._stream_execute_fn(current_input):
                yield event
                if isinstance(event, TextEvent):
                    output_parts.append(event.text)
            output = "".join(output_parts)
            state.record_success()
        except Exception as exc:
            output = str(exc)
            state.record_failure()
            success = False

        # Phase 2: Analyze
        scores = await self._analyze(output, current_input, state) if success else {}

        yield RalphIterationEvent(
            iteration=state.iteration,
            status="completed" if success else "failed",
            scores=scores,
            agent_name=name,
        )

        # Phase 3: Learn
        reflection = await self._learn(current_input, output, scores, success, state)

        # Phase 4: Plan
        current_input = self._plan(input, reflection)

        # Phase 5: Halt
        decision = await self._halt(state)
        if decision.should_stop:
            yield RalphStopEvent(
                stop_type=decision.stop_type.value,
                reason=decision.reason,
                iterations=state.iteration,
                final_scores=scores,
                agent_name=name,
            )
            return

class SwarmNode:
    # existing __init__, run(), describe() unchanged

    async def stream(
        self,
        input: str,
        *,
        messages: Sequence[Message] | None = None,
        provider: Any = None,
        detailed: bool = False,
        max_steps: int | None = None,
    ) -> AsyncIterator[StreamEvent]:
        """Stream inner swarm events with context isolation."""
        async for event in self._swarm.stream(
            input, provider=provider, detailed=detailed, max_steps=max_steps,
        ):
            yield event

class RalphNode:
    """Wraps a RalphRunner so it can be used as a node in a Swarm."""

    def __init__(self, *, runner: RalphRunner, name: str = "ralph"):
        self._runner = runner
        self.name = name
        self.is_group = True  # triggers Swarm's .stream() delegation path

    async def run(self, input: str, *, messages=None, provider=None, **kw) -> RunResult:
        result = await self._runner.run(input)
        return RunResult(output=result.output)

    async def stream(
        self, input: str, *, messages=None, provider=None,
        detailed=False, max_steps=None,
    ) -> AsyncIterator[StreamEvent]:
        async for event in self._runner.stream(input, name=self.name):
            yield event

researcher = Agent(name="researcher", ...)
ralph = RalphNode(
    runner=RalphRunner.from_agent(researcher, scorers=[quality_scorer]),
    name="research_loop",
)
summarizer = Agent(name="summarizer", ...)

swarm = Swarm(
    agents=[ralph, summarizer],
    flow="research_loop >> summarizer",
)

async for event in swarm.stream("analyze market trends"):
    print(event)