Stream

async def send(self, event: BaseEvent, context: "ConversationContext") -> None: ...

def enqueue(self, *content: "SendableMessage | Input") -> None:
    """Append a follow-up turn to this stream's inbox."""
    ...

def spawn_background(self, coro: Coroutine[Any, Any, None]) -> asyncio.Task[None]:
    """Start a fire-and-forget task in this stream's scope.

    The task is not awaited by the agent loop. Tasks deliver their results
    via ``self.enqueue(...)`` — anything enqueued while an ``ask`` is live
    feeds the next model call; anything enqueued after ``ask`` returned
    sits in ``pending_messages`` and is consumed by the next ``ask`` on
    the same stream.
    """
    ...

def where(self, condition: ClassInfo | Condition) -> "Stream": ...

def join(
    self,
    *,
    max_events: int | None = None,
) -> AbstractContextManager[AsyncIterator[BaseEvent]]: ...

def subscribe(
    self,
    func: Callable[..., Any] | None = None,
    *,
    interrupt: bool = False,
    sync_to_thread: bool = True,
    condition: Condition | None = None,
) -> Callable[[Callable[..., Any]], SubId] | SubId: ...

def unsubscribe(self, sub_id: SubId) -> None: ...

def sub_scope(
    self,
    func: Callable[..., Any],
    *,
    interrupt: bool = False,
    sync_to_thread: bool = True,
) -> AbstractContextManager[None]: ...

def get(
    self,
    condition: ClassInfo | Condition,
) -> AbstractAsyncContextManager[asyncio.Future[BaseEvent]]: ...