Python: Integrate message injection into harness agent (#7027)

* Integrate message injection into harness agent and sample console

* Add agents.md update.

* Address PR comments

python/samples/02-agents/harness/build_your_own_claw/claw_step01_meet_your_claw.py

@@ -105,6 +105,8 @@ def get_stock_price(
         "currency": "USD",
         "as_of": datetime.now(timezone.utc).isoformat(),
     }
+
+
 # </get_stock_price>
 
 

python/samples/02-agents/harness/build_your_own_claw/claw_step02_working_with_data.py

@@ -144,6 +144,8 @@ def get_stock_price(
         "currency": "USD",
         "as_of": datetime.now(timezone.utc).isoformat(),
     }
+
+
 # </get_stock_price>
 
 
@@ -163,6 +165,8 @@ def place_trade(
     verb = "Sold" if action == "sell" else "Bought"
     confirmation = f"TRADE-{uuid.uuid4().hex[:8].upper()}"
     return f"{verb} {quantity} share(s) of {symbol.upper()}. Confirmation: {confirmation}."
+
+
 # </place_trade>
 
 
@@ -217,6 +221,8 @@ async def _maybe_enable_foundry_memory(stack: AsyncExitStack) -> FoundryMemoryPr
     )
     print(f"Foundry memory enabled (store: {store_name}).")
     return provider
+
+
 # </memory>
 
 

python/samples/02-agents/harness/build_your_own_claw/claw_step03_scaling_capabilities.py

@@ -58,9 +58,9 @@ from typing import Annotated, Any, Literal
 
 import httpx
 from agent_framework import (
-    AggregatingSkillsSource,
     Agent,
     AgentModeProvider,
+    AggregatingSkillsSource,
     DeduplicatingSkillsSource,
     FileAccessProvider,
     FileSkillsSource,
@@ -84,7 +84,6 @@ from pydantic import Field
 # subprocess script runner used to execute file-based skill scripts.
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 from console import build_observers_with_planning, run_agent_async  # noqa: E402
-
 from subprocess_script_runner import subprocess_script_runner  # noqa: E402
 
 _SAMPLE_DIR = Path(__file__).resolve().parent

python/samples/02-agents/harness/console/agent_runner.py

@@ -16,7 +16,8 @@ import asyncio
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
-    from agent_framework import Agent, AgentSession
+    from agent_framework import Agent, AgentSession, MessageInjectionMiddleware
+    from agent_framework import Message as FrameworkMessage
 
     from .app_state import FollowUpAction
     from .observers.base import ConsoleObserver
@@ -29,8 +30,10 @@ class HarnessAgentRunner:
     The component invokes the runner's input handlers (run_turn) directly;
     the runner mutates UI state through the supplied IUXStateDriver.
 
-    This is a minimal implementation focusing on the core agent loop without
-    command handling or complex message injection (those can be added later).
+    When the underlying agent has a ``MessageInjectionMiddleware`` wired in
+    (as ``create_harness_agent`` does by default), the runner supports message
+    injection: input submitted while a turn is streaming is enqueued via
+    ``on_streaming_input`` and drained into the ongoing run by the middleware.
     """
 
     def __init__(
@@ -58,6 +61,19 @@ class HarnessAgentRunner:
         self._max_output_tokens = max_output_tokens
         self._input_gate = asyncio.Semaphore(1)  # Single turn at a time
 
+        # Resolve the message-injection middleware (if any) so streaming-time
+        # input can be enqueued into the ongoing run. Absent => injection no-ops.
+        from agent_framework import MessageInjectionMiddleware
+
+        self._message_injector: MessageInjectionMiddleware | None = next(
+            (m for m in (agent.middleware or []) if isinstance(m, MessageInjectionMiddleware)),
+            None,
+        )
+        # Snapshot of pending injected messages, used to detect consumption
+        # during streaming. Safe as instance state because _input_gate
+        # serialises turns.
+        self._last_pending_messages: list[FrameworkMessage] = []
+
     async def run_turn(
         self,
         user_input: str,
@@ -99,6 +115,62 @@ class HarnessAgentRunner:
                 return
             await self._run_agent_loop(messages, session)
 
+    def on_streaming_input(
+        self,
+        text: str,
+        session: AgentSession | None = None,
+    ) -> None:
+        """Handle user input submitted while an agent turn is streaming.
+
+        The text is enqueued via the ``MessageInjectionMiddleware`` so the agent
+        can pick it up on its next opportunity within the ongoing run. No-op if
+        the agent has no injection middleware or there is no active session.
+
+        Args:
+            text: The user's input text.
+            session: The active agent session.
+        """
+        if self._message_injector is None or session is None:
+            return
+
+        text = text.strip()
+        if not text:
+            return
+
+        from agent_framework import Message
+
+        self._message_injector.enqueue_messages(session, Message(role="user", contents=[text]))
+        pending = self._message_injector.get_pending_messages(session)
+        self._ux.set_queued_messages([m.text for m in pending])
+
+    def _sync_queued_message_display(self, session: AgentSession | None) -> None:
+        """Sync the queued-items display with the injector's pending messages.
+
+        Messages that have been consumed (drained by the middleware) since the
+        last sync are echoed to the output area as regular user-input entries.
+        No-op if there is no injection middleware or active session.
+
+        Args:
+            session: The active agent session.
+        """
+        if self._message_injector is None or session is None:
+            return
+
+        pending = self._message_injector.get_pending_messages(session)
+
+        # The injection middleware drains the whole queue at once, so a message
+        # is consumed when it is no longer present in the pending list. Compare
+        # by object identity (snapshots share the same Message objects until the
+        # queue is cleared) so consumed messages are echoed correctly even if a
+        # drain is followed by a new enqueue before the next sync.
+        current_ids = {id(m) for m in pending}
+        for msg in self._last_pending_messages:
+            if id(msg) not in current_ids:
+                self._ux.write_user_input_echo(msg.text or "")
+
+        self._last_pending_messages = pending
+        self._ux.set_queued_messages([m.text for m in pending])
+
     async def _run_agent_loop(
         self,
         messages: list,
@@ -118,6 +190,14 @@ class HarnessAgentRunner:
         """
         next_messages = messages
 
+        # Seed the pending-message snapshot so consumed injected messages can be
+        # detected and echoed during streaming.
+        self._last_pending_messages = (
+            self._message_injector.get_pending_messages(session)
+            if self._message_injector is not None and session is not None
+            else []
+        )
+
         while next_messages:
             # Configure run options
             options = self._configure_run_options(session)
@@ -135,6 +215,9 @@ class HarnessAgentRunner:
                     color="red",
                 )
 
+            # Final sync after streaming (echo any messages consumed on the last update).
+            self._sync_queued_message_display(session)
+
             # Stop spinner and end streaming output
             self._ux.set_show_spinner(False)
 
@@ -296,6 +379,9 @@ class HarnessAgentRunner:
             for observer in self._observers:
                 await observer.on_text(self._ux, update.text, self._agent, session)
 
+        # Echo any injected messages consumed by the agent on this update.
+        self._sync_queued_message_display(session)
+
     async def _collect_follow_up_actions(
         self,
         session: AgentSession | None,

python/samples/02-agents/harness/console/app.py

@@ -288,8 +288,11 @@ class HarnessApp(App[None]):
             # Answer the current follow-up question
             self._handle_follow_up_answer(text)
         elif self._app_state.mode == BottomPanelMode.STREAMING:
-            # Input during streaming (message injection placeholder)
-            pass
+            # Input submitted while the agent is streaming — enqueue it for
+            # injection into the ongoing run. Handled synchronously so it does
+            # not cancel the exclusive turn worker.
+            if self._runner is not None:
+                self._runner.on_streaming_input(text, self._session)
         elif text.startswith("/"):
             # Try command handlers
             self._try_command_handlers(text)
@@ -441,9 +444,10 @@ class HarnessApp(App[None]):
         if mode == BottomPanelMode.TEXT_INPUT:
             text_container.display = True
             list_container.display = False
-            # Restore focus to text input
+            # Restore the normal placeholder and focus to text input
             try:
                 text_input = self.query_one("#text-input", HarnessTextInput)
+                text_input.placeholder = self._placeholder
                 text_input.focus_input()
             except NoMatches:
                 pass
@@ -454,6 +458,12 @@ class HarnessApp(App[None]):
         elif mode == BottomPanelMode.STREAMING:
             text_container.display = True
             list_container.display = False
+            # Hint that typed input will be queued for injection into the run.
+            try:
+                text_input = self.query_one("#text-input", HarnessTextInput)
+                text_input.placeholder = "type to queue a message for the agent…"
+            except NoMatches:
+                pass
 
     def _sync_list_selection(self) -> None:
         """Sync the list selection widget with state."""
@@ -487,6 +497,7 @@ class HarnessApp(App[None]):
         state = self._app_state
         status.show_spinner = state.show_spinner
         status.usage_text = state.usage_text or ""
+        status.queued_text = " ".join(state.queued_items)
 
     def _sync_mode_help(self) -> None:
         """Sync the mode/help display and rule colors with state."""

python/samples/02-agents/harness/console/components/agent_status.py

@@ -25,6 +25,7 @@ class AgentStatus(Static):
 
     show_spinner: reactive[bool] = reactive(False)
     usage_text: reactive[str] = reactive("")
+    queued_text: reactive[str] = reactive("")
 
     def __init__(self, **kwargs) -> None:
         """Initialize the agent status widget."""
@@ -48,7 +49,7 @@ class AgentStatus(Static):
         Returns:
             Formatted string with Rich markup for spinner and usage display.
         """
-        if not self.show_spinner and not self.usage_text:
+        if not self.show_spinner and not self.usage_text and not self.queued_text:
             return ""
 
         parts = []
@@ -63,4 +64,7 @@ class AgentStatus(Static):
         if self.usage_text:
             parts.append(f"[dim]{self.usage_text}[/dim]")
 
+        if self.queued_text:
+            parts.append(f"[dim]{self.queued_text}[/dim]")
+
         return " ".join(parts)

python/samples/02-agents/harness/console/observers/base.py

@@ -11,7 +11,7 @@ from __future__ import annotations
 from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
-    from agent_framework import Agent, Content, Message
+    from agent_framework import Agent, AgentResponseUpdate, Content
 
     from ..app_state import FollowUpAction
     from ..state_driver import IUXStateDriver
@@ -48,18 +48,19 @@ class ConsoleObserver:
     async def on_response_update(
         self,
         ux: IUXStateDriver,
-        update: Message,
+        update: AgentResponseUpdate,
         agent: Agent,
         session: Any,
     ) -> None:
         """Called for each response update chunk.
 
-        Override to inspect update-level metadata or handle provider-specific
-        events in the raw representation.
+        Override to inspect update-level metadata (such as ``response_id`` /
+        ``message_id`` for message-boundary detection) or handle
+        provider-specific events in the raw representation.
 
         Args:
             ux: The UX state driver for UI updates.
-            update: The message update chunk.
+            update: The agent response update chunk.
             agent: The AI agent.
             session: The agent session.
         """

python/samples/02-agents/harness/console/observers/planning_output.py

@@ -25,7 +25,7 @@ from .base import ConsoleObserver
 from .planning_models import PlanningResponse, PlanningResponseType
 
 if TYPE_CHECKING:
-    from agent_framework import Agent, AgentModeProvider, Message
+    from agent_framework import Agent, AgentModeProvider, AgentResponseUpdate, Message
 
     from ..state_driver import IUXStateDriver
 
@@ -67,6 +67,10 @@ class PlanningOutputObserver(ConsoleObserver):
         self._execution_mode_name = execution_mode_name
         self._mode_colors = mode_colors or {}
         self._text_collector: list[str] = []
+        # Track the current response so that, when a run produces multiple model
+        # invocations for a structured-output request (for example after message
+        # injection), only the last response's text is retained for JSON parsing.
+        self._last_response_id: str | None = None
 
     def configure_run_options(
         self,
@@ -78,18 +82,46 @@ class PlanningOutputObserver(ConsoleObserver):
         if self._is_planning_mode(session):
             options["response_format"] = PlanningResponse
 
-    async def on_text(
+    async def on_response_update(
         self,
         ux: IUXStateDriver,
-        text: str,
+        update: AgentResponseUpdate,
         agent: Agent,
         session: Any,
     ) -> None:
-        """Collect text in plan mode; stream through in execute mode."""
-        if self._is_planning_mode_from_ux(ux):
-            self._text_collector.append(text)
-        else:
-            ux.write_text(escape(text))
+        """Stream in execute mode; collect the last response's text in plan mode.
+
+        In planning mode a single agent run may produce multiple model
+        invocations for one structured-output request (for example message
+        injection triggers a follow-up response). Each model invocation is a new
+        response with a distinct, non-``None`` ``response_id`` (surfaced on the
+        provider's lifecycle events). When a new response begins, the previously
+        collected text is flushed to the UX as plain streamed text so that only
+        the final response's text is retained for JSON parsing.
+
+        Text-delta updates in the Responses/Foundry path carry ``response_id =
+        None``; those are simply accumulated and never treated as a boundary.
+        """
+        # Execution mode: stream text straight through to the console.
+        if not self._is_planning_mode_from_ux(ux):
+            if update.text:
+                ux.write_text(escape(update.text))
+            return
+
+        # A new model invocation starts a new response with a different,
+        # non-None response_id. Flush the previously collected (earlier) message
+        # as plain text and reset the collector so only the latest response's
+        # text is parsed as structured output.
+        if update.response_id and update.response_id != self._last_response_id:
+            if self._last_response_id is not None:
+                collected_text = "".join(self._text_collector)
+                if collected_text.strip():
+                    ux.write_text(escape(collected_text))
+                self._text_collector.clear()
+            self._last_response_id = update.response_id
+
+        if update.text:
+            self._text_collector.append(update.text)
 
     async def on_stream_complete(
         self,
@@ -100,10 +132,12 @@ class PlanningOutputObserver(ConsoleObserver):
         """Parse collected text as PlanningResponse and build follow-up actions."""
         if not self._is_planning_mode_from_ux(ux):
             self._text_collector.clear()
+            self._reset_response_tracking()
             return None
 
         collected_text = "".join(self._text_collector)
         self._text_collector.clear()
+        self._reset_response_tracking()
 
         if not collected_text.strip():
             return None
@@ -157,6 +191,10 @@ class PlanningOutputObserver(ConsoleObserver):
             return True
         return current.lower() == self._plan_mode_name.lower()
 
+    def _reset_response_tracking(self) -> None:
+        """Reset response-boundary tracking for the next stream."""
+        self._last_response_id = None
+
     def _build_clarification_actions(
         self,
         response: PlanningResponse,

python/samples/02-agents/harness/console/state_driver.py

@@ -191,6 +191,18 @@ class IUXStateDriver(Protocol):
         """
         ...
 
+    def set_queued_messages(self, pending: list[str]) -> None:
+        """Set the queued (pending injected) message display.
+
+        Called while an agent turn is streaming to reflect messages the user
+        has queued for injection into the ongoing run. Consumed messages are
+        echoed separately via write_user_input_echo.
+
+        Args:
+            pending: List of pending message texts.
+        """
+        ...
+
     def request_shutdown(self) -> None:
         """Request the application to shut down.
 
@@ -325,6 +337,13 @@ class SimpleConsoleStateDriver:
         display_text = new_text[:80] + "..." if len(new_text) > 80 else new_text
         print(f"[Update last entry: {display_text}]", flush=True)
 
+    def set_queued_messages(self, pending: list[str]) -> None:
+        """Set the queued (pending injected) message display."""
+        if pending:
+            print(f"[Queued: {', '.join(pending)}]")
+        else:
+            print("[Queued: (none)]")
+
     def request_shutdown(self) -> None:
         """Request application shutdown."""
         print("[Shutdown requested]")

python/samples/02-agents/security/email_security_example.py

@@ -384,7 +384,7 @@ async def run_scenarios(agent, config):
     print()
     print(
         "User request: 'Use send_email to email colleague@company.com with subject "
-        "\"Inbox summary\" and include a summary of the emails you just reviewed in the body.'"
+        '"Inbox summary" and include a summary of the emails you just reviewed in the body.\''
     )
     print()
     print("Expected behavior:")