Python: Integrate message injection into harness agent

python/packages/core/AGENTS.md

@@ -125,7 +125,7 @@ agent_framework/
 - **Scoping** - Memories are isolated per session by default: each session writes under a working folder derived from `context.session_id`. Pass an explicit `scope` (e.g. a user id) to group memories across sessions, mirroring `FoundryMemoryProvider`'s `scope` arg.
 - **Descriptions & index** - `file_memory_write` accepts an optional `description`, stored in a companion `<stem>_description.md` sidecar. After each write/delete the provider rebuilds a capped (50-entry) `memories.md` index, and `before_run` injects that index as a `user` context message so the model knows what memories exist. Sidecars and the index are internal files hidden from `file_memory_ls`/`file_memory_grep` and rejected as write targets.
 - **`DEFAULT_FILE_MEMORY_SOURCE_ID`** / **`DEFAULT_FILE_MEMORY_INSTRUCTIONS`** - Public defaults for the provider's source id and instruction banner.
-- **Harness wiring** - `create_harness_agent` includes both `FileMemoryProvider` and `FileAccessProvider` by default. Disable via `disable_file_memory` / `disable_file_access`; override the backing store via `file_memory_store` / `file_access_store`. When no store is supplied, defaults are `FileSystemAgentFileStore` rooted at `{cwd}/agent-file-memory` (memory) and `{cwd}/working` (access), mirroring the .NET `HarnessAgent`.
+- **Harness wiring** - `create_harness_agent` includes both `FileMemoryProvider` and `FileAccessProvider` by default. Disable via `disable_file_memory` / `disable_file_access`; override the backing store via `file_memory_store` / `file_access_store`. When no store is supplied, defaults are `FileSystemAgentFileStore` rooted at `{cwd}/agent-file-memory` (memory) and `{cwd}/working` (access), mirroring the .NET `HarnessAgent`. `create_harness_agent` also wires in `MessageInjectionMiddleware` by default (mirroring the .NET harness's `UseMessageInjection`); it is always on with no opt-out because it is a no-op when no messages are queued for the session.
 
 ### Tool Approval Harness (`_harness/_tool_approval.py`)
 

python/packages/core/agent_framework/_harness/_agent.py

@@ -20,7 +20,7 @@
 from .._clients import SupportsShellTool, SupportsWebSearchTool
 from .._compaction import CompactionProvider, ContextWindowCompactionStrategy, ToolResultCompactionStrategy
 from .._feature_stage import ExperimentalFeature, experimental
-from .._sessions import ContextProvider, HistoryProvider, InMemoryHistoryProvider
+from .._sessions import ContextProvider, HistoryProvider, InMemoryHistoryProvider, MessageInjectionMiddleware
 from .._skills import SkillsProvider
 from .._types import ChatOptions
 from ._background_agents import BackgroundAgentsProvider
@@ -584,6 +584,9 @@ def create_harness_agent(
                 next_message=loop_next_message,
             ),
         )
+    # Message injection is always on. It is a no-op when no messages are queued for the session,
+    # so there is no opt-out.
+    assembled_middleware.append(MessageInjectionMiddleware())
     if middleware:
         assembled_middleware.extend(middleware)
 

python/packages/core/tests/core/test_harness_agent.py

@@ -843,6 +843,18 @@ def _rule(content: Any) -> bool:
     assert _rule in middleware.auto_approval_rules
 
 
+def test_create_harness_agent_adds_message_injection_by_default() -> None:
+    """Message injection middleware should be wired in by default (like .NET UseMessageInjection)."""
+    from agent_framework import MessageInjectionMiddleware
+
+    agent = create_harness_agent(
+        client=_FakeChatClient(),  # type: ignore[arg-type]
+        max_context_window_tokens=128_000,
+        max_output_tokens=16_384,
+    )
+    assert any(isinstance(mw, MessageInjectionMiddleware) for mw in agent.middleware or [])
+
+
 def test_create_harness_agent_tool_approval_outermost_with_user_middleware() -> None:
     """Tool approval middleware should be placed first (outermost) ahead of user middleware."""
     from agent_framework import AgentMiddleware, ToolApprovalMiddleware
@@ -865,8 +877,8 @@ async def process(self, context: Any, call_next: Any) -> None:
 
 
 def test_create_harness_agent_disable_tool_auto_approval_preserves_user_middleware() -> None:
-    """When tool approval is disabled, only user-supplied middleware should remain."""
-    from agent_framework import AgentMiddleware
+    """When tool approval is disabled, message injection plus user-supplied middleware remain."""
+    from agent_framework import AgentMiddleware, MessageInjectionMiddleware
 
     class _CustomMiddleware(AgentMiddleware):
         async def process(self, context: Any, call_next: Any) -> None:
@@ -880,18 +892,25 @@ async def process(self, context: Any, call_next: Any) -> None:
         disable_tool_auto_approval=True,
         middleware=[custom],
     )
-    assert agent.middleware == [custom]
+    # Message injection is always wired in (before user middleware); tool approval is omitted.
+    assert agent.middleware is not None
+    assert custom in agent.middleware
+    assert any(isinstance(mw, MessageInjectionMiddleware) for mw in agent.middleware)
+    assert [type(mw) for mw in agent.middleware] == [MessageInjectionMiddleware, _CustomMiddleware]
 
 
 def test_create_harness_agent_no_middleware_when_tool_approval_disabled_and_none() -> None:
-    """No middleware should be installed when tool approval is disabled and none is supplied."""
+    """Only the always-on message injection middleware remains when tool approval is disabled."""
+    from agent_framework import MessageInjectionMiddleware
+
     agent = create_harness_agent(
         client=_FakeChatClient(),  # type: ignore[arg-type]
         max_context_window_tokens=128_000,
         max_output_tokens=16_384,
         disable_tool_auto_approval=True,
     )
-    assert agent.middleware is None
+    assert agent.middleware is not None
+    assert [type(mw) for mw in agent.middleware] == [MessageInjectionMiddleware]
 
 
 # --- Loop Wiring Tests ---

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 @@
 
 import httpx
 from agent_framework import (
-    AggregatingSkillsSource,
     Agent,
     AgentModeProvider,
+    AggregatingSkillsSource,
     DeduplicatingSkillsSource,
     FileAccessProvider,
     FileSkillsSource,
@@ -84,7 +84,6 @@
 # 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 @@
 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 @@ def __init__(
         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 @@ async def start_agent_turn(
                 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 @@ async def _run_agent_loop(
         """
         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 @@ async def _run_agent_loop(
                     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 @@ async def _dispatch_update(
             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 @@ def on_text_submitted(self, event: HarnessTextInput.Submitted) -> 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 @@ def _sync_bottom_panel(self, mode: BottomPanelMode) -> 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 @@ def _sync_bottom_panel(self, mode: BottomPanelMode) -> 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 @@ def _sync_status_bar(self) -> 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 @@ def render(self) -> str:
         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 @@ def render(self) -> str:
         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 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 @@ def configure_run_options(
     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.
         """