trying to support intermediate streaming

Author: felixs8696

examples/tutorials/00_sync/000_hello_acp/project/acp.py

@@ -2,7 +2,7 @@
 from agentex.lib.sdk.fastacp.fastacp import FastACP
 from agentex.lib.types.acp import SendMessageParams
 
-from agentex.lib.types.task_message_updates import TaskMessageUpdate
+from agentex.types.task_message_update import TaskMessageUpdate
 from agentex.types.task_message import TaskMessageContent
 from agentex.types.task_message_content import TextContent
 from agentex.lib.utils.logging import make_logger

examples/tutorials/00_sync/010_multiturn/project/acp.py

@@ -5,7 +5,7 @@
 from agentex.lib.sdk.fastacp.fastacp import FastACP
 from agentex.lib.types.acp import SendMessageParams
 from agentex.lib.types.llm_messages import AssistantMessage, LLMConfig, SystemMessage, UserMessage
-from agentex.lib.types.task_message_updates import TaskMessageUpdate
+from agentex.types.task_message_update import TaskMessageUpdate
 from agentex.types.task_message import TaskMessageContent
 from agentex.types.task_message_content import TextContent
 from agentex.lib.utils.model_utils import BaseModel

examples/tutorials/00_sync/020_streaming/project/acp.py

@@ -5,7 +5,8 @@
 from agentex.lib.sdk.fastacp.fastacp import FastACP
 from agentex.lib.types.acp import SendMessageParams
 from agentex.lib.types.llm_messages import AssistantMessage, LLMConfig, SystemMessage, UserMessage
-from agentex.lib.types.task_message_updates import StreamTaskMessageDelta, StreamTaskMessageDone, StreamTaskMessageFull, TaskMessageUpdate, TextDelta
+from agentex.types.task_message_update import StreamTaskMessageDelta, StreamTaskMessageDone, StreamTaskMessageFull, TaskMessageUpdate
+from agentex.types.task_message_delta import TextDelta
 from agentex.lib.utils.model_utils import BaseModel
 from agentex.types.task_message_content import TaskMessageContent, TextContent
 

examples/tutorials/10_agentic/00_base/040_other_sdks/project/acp.py

@@ -8,11 +8,11 @@
 from agentex.lib.sdk.fastacp.fastacp import FastACP
 from agentex.lib.types.acp import CancelTaskParams, CreateTaskParams, SendEventParams
 from agentex.lib.types.fastacp import AgenticACPConfig
-from agentex.lib.types.task_message_updates import (
+from agentex.types.task_message_update import (
     StreamTaskMessageDelta, 
     StreamTaskMessageFull,
-    TextDelta,
 )
+from agentex.types.task_message_delta import TextDelta
 from agentex.lib.utils.logging import make_logger
 from agentex.lib.utils.model_utils import BaseModel
 from agentex.types.text_content import TextContent

src/agentex/lib/cli/templates/sync/project/acp.py.j2

@@ -2,7 +2,7 @@ from typing import AsyncGenerator, Union
 from agentex.lib.sdk.fastacp.fastacp import FastACP
 from agentex.lib.types.acp import SendMessageParams
 
-from agentex.lib.types.task_message_updates import TaskMessageUpdate
+from agentex.types.task_message_update import TaskMessageUpdate
 from agentex.types.task_message_content import TaskMessageContent
 from agentex.types.text_content import TextContent
 from agentex.lib.utils.logging import make_logger

src/agentex/lib/core/services/adk/messages.py

@@ -4,7 +4,7 @@
 from agentex import AsyncAgentex
 from agentex.lib.core.services.adk.streaming import StreamingService
 from agentex.lib.core.tracing.tracer import AsyncTracer
-from agentex.lib.types.task_message_updates import StreamTaskMessageFull, TaskMessageUpdate
+from agentex.types.task_message_update import StreamTaskMessageFull, TaskMessageUpdate
 from agentex.types.task_message import TaskMessage, TaskMessageContent
 from agentex.lib.utils.logging import make_logger
 from agentex.lib.utils.temporal import heartbeat_if_in_workflow

src/agentex/lib/core/services/adk/providers/litellm.py

@@ -8,11 +8,11 @@
     Completion,
     LLMConfig,
 )
-from agentex.lib.types.task_message_updates import (
+from agentex.types.task_message_update import (
     StreamTaskMessageDelta,
     StreamTaskMessageFull,
-    TextDelta,
 )
+from agentex.types.task_message_delta import TextDelta
 from agentex.types.task_message import TaskMessage
 from agentex.types.task_message_content import TextContent
 from agentex.lib.utils import logging

src/agentex/lib/core/services/adk/providers/openai.py

@@ -12,6 +12,10 @@
     ResponseFunctionToolCall,
     ResponseOutputItemDoneEvent,
     ResponseTextDeltaEvent,
+    ResponseReasoningSummaryTextDeltaEvent,
+    ResponseReasoningSummaryTextDoneEvent,
+    ResponseReasoningTextDeltaEvent,
+    ResponseReasoningTextDoneEvent,
 )
 from pydantic import BaseModel
 
@@ -22,14 +26,18 @@
     StreamingTaskMessageContext,
 )
 from agentex.lib.core.tracing.tracer import AsyncTracer
-from agentex.lib.types.task_message_updates import (
+from agentex.types.task_message_update import (
     StreamTaskMessageDelta,
     StreamTaskMessageFull,
+)
+from agentex.types.task_message_delta import (
     TextDelta,
+    ReasoningSummaryDelta,
+    ReasoningContentDelta,
 )
 from agentex.types.task_message_content import (
-    TextContent,
     ReasoningContent,
+    TextContent,
     ToolRequestContent,
     ToolResponseContent,
 )
@@ -649,30 +657,7 @@ async def run_agent_streamed_auto_send(
                                         ),
                                     )
 
-                            elif event.item.type == "reasoning_item":
-                                # Handle reasoning items
-                                reasoning_item = event.item.raw_item
-                                
-                                reasoning_content = ReasoningContent(
-                                    author="agent",
-                                    summary=[summary.text for summary in reasoning_item.summary],
-                                    content=[content.text for content in reasoning_item.content] if reasoning_item.content else None,
-                                )
 
-                                # Create reasoning content using streaming context (immediate completion)
-                                async with (
-                                    self.streaming_service.streaming_task_message_context(
-                                        task_id=task_id,
-                                        initial_content=reasoning_content,
-                                    ) as streaming_context
-                                ):
-                                    # The message has already been persisted, but we still need to send an update
-                                    await streaming_context.stream_update(
-                                        update=StreamTaskMessageFull(
-                                            parent_task_message=streaming_context.task_message,
-                                            content=reasoning_content,
-                                        ),
-                                    )
 
                         elif event.type == "raw_response_event":
                             if isinstance(event.data, ResponseTextDeltaEvent):
@@ -707,6 +692,100 @@ async def run_agent_streamed_auto_send(
                                     ),
                                 )
 
+                            elif isinstance(event.data, ResponseReasoningSummaryTextDeltaEvent):
+                                # Handle reasoning summary text delta
+                                item_id = event.data.item_id
+                                summary_index = event.data.summary_index
+
+                                # Check if we already have a streaming context for this reasoning item
+                                if item_id not in item_id_to_streaming_context:
+                                    # Create a new streaming context for this reasoning item
+                                    streaming_context = self.streaming_service.streaming_task_message_context(
+                                        task_id=task_id,
+                                        initial_content=ReasoningContent(
+                                            author="agent",
+                                            summary=[],
+                                            content=[],
+                                        ),
+                                    )
+                                    # Open the streaming context
+                                    item_id_to_streaming_context[
+                                        item_id
+                                    ] = await streaming_context.open()
+                                    unclosed_item_ids.add(item_id)
+                                else:
+                                    streaming_context = item_id_to_streaming_context[
+                                        item_id
+                                    ]
+
+                                # Stream the summary delta through the streaming service
+                                await streaming_context.stream_update(
+                                    update=StreamTaskMessageDelta(
+                                        parent_task_message=streaming_context.task_message,
+                                        delta=ReasoningSummaryDelta(
+                                            summary_index=summary_index,
+                                            summary_delta=event.data.delta,
+                                        ),
+                                    ),
+                                )
+
+                            elif isinstance(event.data, ResponseReasoningTextDeltaEvent):
+                                # Handle reasoning content text delta
+                                item_id = event.data.item_id
+                                content_index = event.data.content_index
+
+                                # Check if we already have a streaming context for this reasoning item
+                                if item_id not in item_id_to_streaming_context:
+                                    # Create a new streaming context for this reasoning item
+                                    streaming_context = self.streaming_service.streaming_task_message_context(
+                                        task_id=task_id,
+                                        initial_content=ReasoningContent(
+                                            author="agent",
+                                            summary=[],
+                                            content=[],
+                                        ),
+                                    )
+                                    # Open the streaming context
+                                    item_id_to_streaming_context[
+                                        item_id
+                                    ] = await streaming_context.open()
+                                    unclosed_item_ids.add(item_id)
+                                else:
+                                    streaming_context = item_id_to_streaming_context[
+                                        item_id
+                                    ]
+
+                                # Stream the content delta through the streaming service
+                                await streaming_context.stream_update(
+                                    update=StreamTaskMessageDelta(
+                                        parent_task_message=streaming_context.task_message,
+                                        delta=ReasoningContentDelta(
+                                            content_index=content_index,
+                                            content_delta=event.data.delta,
+                                        ),
+                                    ),
+                                )
+
+
+
+
+
+                            elif isinstance(event.data, ResponseReasoningSummaryTextDoneEvent):
+                                # Handle reasoning summary text completion
+                                item_id = event.data.item_id
+                                summary_index = event.data.summary_index
+
+                                # Note: We don't close the streaming context here since there might be more deltas
+                                # The context will be closed when the reasoning item is completely done
+
+                            elif isinstance(event.data, ResponseReasoningTextDoneEvent):
+                                # Handle reasoning content text completion
+                                item_id = event.data.item_id
+                                content_index = event.data.content_index
+
+                                # Note: We don't close the streaming context here since there might be more deltas
+                                # The context will be closed when the reasoning item is completely done
+
                             elif isinstance(event.data, ResponseOutputItemDoneEvent):
                                 # Handle item completion
                                 item_id = event.data.item.id

src/agentex/lib/core/services/adk/streaming.py

@@ -3,19 +3,20 @@
 
 from agentex import AsyncAgentex
 from agentex.lib.core.adapters.streams.port import StreamRepository
-from agentex.lib.types.task_message_updates import (
+from agentex.types.task_message_update import (
     TaskMessageDelta, 
     TaskMessageUpdate,
-    TextDelta,
-    DataDelta,
-    ToolRequestDelta,
-    ToolResponseDelta,
-    StreamTaskMessage,
     StreamTaskMessageStart,
     StreamTaskMessageDelta,
     StreamTaskMessageFull,
     StreamTaskMessageDone,
 )
+from agentex.types.task_message_delta import (
+    TextDelta,
+    DataDelta,
+    ToolRequestDelta,
+    ToolResponseDelta,
+)
 from agentex.lib.utils.logging import make_logger
 from agentex.types.data_content import DataContent
 from agentex.types.task_message import (
@@ -187,8 +188,8 @@ async def close(self) -> TaskMessage:
         return self.task_message
 
     async def stream_update(
-        self, update: StreamTaskMessage
-    ) -> StreamTaskMessage | None:
+        self, update: TaskMessageUpdate
+    ) -> TaskMessageUpdate | None:
         """Stream an update to the repository."""
         if self._is_closed:
             raise ValueError("Context is already done")

src/agentex/lib/core/temporal/activities/adk/streaming_activities.py

@@ -3,7 +3,7 @@
 from temporalio import activity
 
 from agentex.lib.core.services.adk.streaming import StreamingService
-from agentex.lib.types.task_message_updates import TaskMessageUpdate
+from agentex.types.task_message_update import TaskMessageUpdate
 from agentex.lib.utils.logging import make_logger
 from agentex.lib.utils.model_utils import BaseModel
 from agentex.lib.utils.temporal import heartbeat_if_in_workflow