async def arun(
        self,
        system_instruction: str | None = None,
        messages: list[Message] | None = None,
        *args,
        **kwargs,
    ) -> AsyncGenerator[str, None]:
        """
        Main streaming entrypoint: returns an async generator of SSE lines.
        """
        self._reset()
        await self._setup(system_instruction)

        if messages:
            for m in messages:
                await self.conversation.add_message(m)

        # Initialize citation tracker for this run
        citation_tracker = CitationTracker()

        # Dictionary to store citation payloads by ID
        citation_payloads = {}

        # Track all citations emitted during streaming for final persistence
        self.streaming_citations: list[dict] = []

        async def sse_generator() -> AsyncGenerator[str, None]:
            pending_tool_calls = {}
            partial_text_buffer = ""
            iterations_count = 0

            try:
                # Keep streaming until we complete
                while (
                    not self._completed
                    and iterations_count < self.config.max_iterations
                ):
                    iterations_count += 1
                    # 1) Get current messages
                    msg_list = await self.conversation.get_messages()
                    gen_cfg = self.get_generation_config(
                        msg_list[-1], stream=True
                    )

                    accumulated_thinking = ""
                    thinking_signatures = {}  # Map thinking content to signatures

                    # 2) Start streaming from LLM
                    llm_stream = self.llm_provider.aget_completion_stream(
                        msg_list, gen_cfg
                    )
                    async for chunk in llm_stream:
                        delta = chunk.choices[0].delta
                        finish_reason = chunk.choices[0].finish_reason

                        if hasattr(delta, "thinking") and delta.thinking:
                            # Accumulate thinking for later use in messages
                            accumulated_thinking += delta.thinking

                            # Emit SSE "thinking" event
                            async for (
                                line
                            ) in SSEFormatter.yield_thinking_event(
                                delta.thinking
                            ):
                                yield line

                        # Add this new handler for thinking signatures
                        if hasattr(delta, "thinking_signature"):
                            thinking_signatures[accumulated_thinking] = (
                                delta.thinking_signature
                            )
                            accumulated_thinking = ""

                        # 3) If new text, accumulate it
                        if delta.content:
                            partial_text_buffer += delta.content

                            # (a) Now emit the newly streamed text as a "message" event
                            async for line in SSEFormatter.yield_message_event(
                                delta.content
                            ):
                                yield line

                            # (b) Find new citation spans in the accumulated text
                            new_citation_spans = find_new_citation_spans(
                                partial_text_buffer, citation_tracker
                            )

                            # Process each new citation span
                            for cid, spans in new_citation_spans.items():
                                for span in spans:
                                    # Check if this is the first time we've seen this citation ID
                                    is_new_citation = (
                                        citation_tracker.is_new_citation(cid)
                                    )

                                    # Get payload if it's a new citation
                                    payload = None
                                    if is_new_citation:
                                        source_obj = self.search_results_collector.find_by_short_id(
                                            cid
                                        )
                                        if source_obj:
                                            # Store payload for reuse
                                            payload = dump_obj(source_obj)
                                            citation_payloads[cid] = payload

                                    # Create citation event payload
                                    citation_data = {
                                        "id": cid,
                                        "object": "citation",
                                        "is_new": is_new_citation,
                                        "span": {
                                            "start": span[0],
                                            "end": span[1],
                                        },
                                    }

                                    # Only include full payload for new citations
                                    if is_new_citation and payload:
                                        citation_data["payload"] = payload

                                    # Add to streaming citations for final answer
                                    self.streaming_citations.append(
                                        citation_data
                                    )

                                    # Emit the citation event
                                    async for (
                                        line
                                    ) in SSEFormatter.yield_citation_event(
                                        citation_data
                                    ):
                                        yield line

                        if delta.tool_calls:
                            for tc in delta.tool_calls:
                                idx = tc.index
                                if idx not in pending_tool_calls:
                                    pending_tool_calls[idx] = {
                                        "id": tc.id,
                                        "name": tc.function.name or "",
                                        "arguments": tc.function.arguments
                                        or "",
                                    }
                                else:
                                    # Accumulate partial name/arguments
                                    if tc.function.name:
                                        pending_tool_calls[idx]["name"] = (
                                            tc.function.name
                                        )
                                    if tc.function.arguments:
                                        pending_tool_calls[idx][
                                            "arguments"
                                        ] += tc.function.arguments

                        # 5) If the stream signals we should handle "tool_calls"
                        if finish_reason == "tool_calls":
                            # Handle thinking if present
                            await self._handle_thinking(
                                thinking_signatures, accumulated_thinking
                            )

                            calls_list = []
                            for idx in sorted(pending_tool_calls.keys()):
                                cinfo = pending_tool_calls[idx]
                                calls_list.append(
                                    {
                                        "tool_call_id": cinfo["id"]
                                        or f"call_{idx}",
                                        "name": cinfo["name"],
                                        "arguments": cinfo["arguments"],
                                    }
                                )

                            # (a) Emit SSE "tool_call" events
                            for c in calls_list:
                                tc_data = self._create_tool_call_data(c)
                                async for (
                                    line
                                ) in SSEFormatter.yield_tool_call_event(
                                    tc_data
                                ):
                                    yield line

                            # (b) Add an assistant message capturing these calls
                            await self._add_tool_calls_message(
                                calls_list, partial_text_buffer
                            )

                            # (c) Execute each tool call in parallel
                            await asyncio.gather(
                                *[
                                    self.handle_function_or_tool_call(
                                        c["name"],
                                        c["arguments"],
                                        tool_id=c["tool_call_id"],
                                    )
                                    for c in calls_list
                                ]
                            )

                            # Reset buffer & calls
                            pending_tool_calls.clear()
                            partial_text_buffer = ""

                        elif finish_reason == "stop":
                            # Handle thinking if present
                            await self._handle_thinking(
                                thinking_signatures, accumulated_thinking
                            )

                            # 6) The LLM is done. If we have any leftover partial text,
                            #    finalize it in the conversation
                            if partial_text_buffer:
                                # Create the final message with metadata including citations
                                final_message = Message(
                                    role="assistant",
                                    content=partial_text_buffer,
                                    metadata={
                                        "citations": self.streaming_citations
                                    },
                                )

                                # Add it to the conversation
                                await self.conversation.add_message(
                                    final_message
                                )

                            # (a) Prepare final answer with optimized citations
                            consolidated_citations = []
                            # Group citations by ID with all their spans
                            for (
                                cid,
                                spans,
                            ) in citation_tracker.get_all_spans().items():
                                if cid in citation_payloads:
                                    consolidated_citations.append(
                                        {
                                            "id": cid,
                                            "object": "citation",
                                            "spans": [
                                                {"start": s[0], "end": s[1]}
                                                for s in spans
                                            ],
                                            "payload": citation_payloads[cid],
                                        }
                                    )

                            # Create final answer payload
                            final_evt_payload = {
                                "id": "msg_final",
                                "object": "agent.final_answer",
                                "generated_answer": partial_text_buffer,
                                "citations": consolidated_citations,
                            }

                            # Emit final answer event
                            async for (
                                line
                            ) in SSEFormatter.yield_final_answer_event(
                                final_evt_payload
                            ):
                                yield line

                            # (b) Signal the end of the SSE stream
                            yield SSEFormatter.yield_done_event()
                            self._completed = True
                            break

                # If we exit the while loop due to hitting max iterations
                if not self._completed:
                    # Generate a summary using the LLM
                    summary = await self._generate_llm_summary(
                        iterations_count
                    )

                    # Send the summary as a message event
                    async for line in SSEFormatter.yield_message_event(
                        summary
                    ):
                        yield line

                    # Add summary to conversation with citations metadata
                    await self.conversation.add_message(
                        Message(
                            role="assistant",
                            content=summary,
                            metadata={"citations": self.streaming_citations},
                        )
                    )

                    # Create and emit a final answer payload with the summary
                    final_evt_payload = {
                        "id": "msg_final",
                        "object": "agent.final_answer",
                        "generated_answer": summary,
                        "citations": consolidated_citations,
                    }

                    async for line in SSEFormatter.yield_final_answer_event(
                        final_evt_payload
                    ):
                        yield line

                    # Signal the end of the SSE stream
                    yield SSEFormatter.yield_done_event()
                    self._completed = True

            except Exception as e:
                logger.error(f"Error in streaming agent: {str(e)}")
                # Emit error event for client
                async for line in SSEFormatter.yield_error_event(
                    f"Agent error: {str(e)}"
                ):
                    yield line
                # Send done event to close the stream
                yield SSEFormatter.yield_done_event()

        # Finally, we return the async generator
        async for line in sse_generator():
            yield line

        Main streaming entrypoint: returns an async generator of SSE lines.
        
py/core/agent/base.py