From e662c14a1cd1680e0fdce87fd2709c80f9aea638 Mon Sep 17 00:00:00 2001 From: westey <164392973+westey-m@users.noreply.github.com> Date: Thu, 9 Jul 2026 17:34:22 +0000 Subject: [PATCH 1/3] Integrate message injection into harness agent and sample console --- .../core/agent_framework/_harness/_agent.py | 5 +- .../core/tests/core/test_harness_agent.py | 29 +++++-- .../claw_step01_meet_your_claw.py | 2 + .../claw_step02_working_with_data.py | 6 ++ .../claw_step03_scaling_capabilities.py | 3 +- .../02-agents/harness/console/agent_runner.py | 86 ++++++++++++++++++- .../samples/02-agents/harness/console/app.py | 17 +++- .../console/components/agent_status.py | 6 +- .../harness/console/observers/base.py | 11 +-- .../console/observers/planning_output.py | 54 ++++++++++-- .../02-agents/harness/console/state_driver.py | 17 ++++ .../security/email_security_example.py | 2 +- 12 files changed, 209 insertions(+), 29 deletions(-) diff --git a/python/packages/core/agent_framework/_harness/_agent.py b/python/packages/core/agent_framework/_harness/_agent.py index 4a434dba09..d4dd8d0ae1 100644 --- a/python/packages/core/agent_framework/_harness/_agent.py +++ b/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) diff --git a/python/packages/core/tests/core/test_harness_agent.py b/python/packages/core/tests/core/test_harness_agent.py index 4bf2d1fe15..9cb6d94a37 100644 --- a/python/packages/core/tests/core/test_harness_agent.py +++ b/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 --- diff --git a/python/samples/02-agents/harness/build_your_own_claw/claw_step01_meet_your_claw.py b/python/samples/02-agents/harness/build_your_own_claw/claw_step01_meet_your_claw.py index bd50743932..2de9173ba8 100644 --- a/python/samples/02-agents/harness/build_your_own_claw/claw_step01_meet_your_claw.py +++ b/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(), } + + # diff --git a/python/samples/02-agents/harness/build_your_own_claw/claw_step02_working_with_data.py b/python/samples/02-agents/harness/build_your_own_claw/claw_step02_working_with_data.py index 8e1542572f..ecd9caf4bb 100644 --- a/python/samples/02-agents/harness/build_your_own_claw/claw_step02_working_with_data.py +++ b/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(), } + + # @@ -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}." + + # @@ -217,6 +221,8 @@ async def _maybe_enable_foundry_memory(stack: AsyncExitStack) -> FoundryMemoryPr ) print(f"Foundry memory enabled (store: {store_name}).") return provider + + # diff --git a/python/samples/02-agents/harness/build_your_own_claw/claw_step03_scaling_capabilities.py b/python/samples/02-agents/harness/build_your_own_claw/claw_step03_scaling_capabilities.py index 9c2d31254f..dbab85a8b5 100644 --- a/python/samples/02-agents/harness/build_your_own_claw/claw_step03_scaling_capabilities.py +++ b/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 diff --git a/python/samples/02-agents/harness/console/agent_runner.py b/python/samples/02-agents/harness/console/agent_runner.py index 743ad1e132..3fe876d32f 100644 --- a/python/samples/02-agents/harness/console/agent_runner.py +++ b/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,56 @@ 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) + + consumed_count = len(self._last_pending_messages) - len(pending) + for i in range(min(consumed_count, len(self._last_pending_messages))): + self._ux.write_user_input_echo(self._last_pending_messages[i].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 +184,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 +209,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 +373,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, diff --git a/python/samples/02-agents/harness/console/app.py b/python/samples/02-agents/harness/console/app.py index e2260eb200..9a1e687923 100644 --- a/python/samples/02-agents/harness/console/app.py +++ b/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.""" diff --git a/python/samples/02-agents/harness/console/components/agent_status.py b/python/samples/02-agents/harness/console/components/agent_status.py index 34ac521a52..95225aa2fd 100644 --- a/python/samples/02-agents/harness/console/components/agent_status.py +++ b/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) diff --git a/python/samples/02-agents/harness/console/observers/base.py b/python/samples/02-agents/harness/console/observers/base.py index 40169ed4ae..cabcb45824 100644 --- a/python/samples/02-agents/harness/console/observers/base.py +++ b/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. """ diff --git a/python/samples/02-agents/harness/console/observers/planning_output.py b/python/samples/02-agents/harness/console/observers/planning_output.py index 3a91ec76c4..1243ac2dc0 100644 --- a/python/samples/02-agents/harness/console/observers/planning_output.py +++ b/python/samples/02-agents/harness/console/observers/planning_output.py @@ -25,7 +25,7 @@ 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 @@ def __init__( 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 @@ def configure_run_options( 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 @@ async def on_stream_complete( """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 @@ def _is_planning_mode_from_ux(self, ux: IUXStateDriver) -> bool: 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, diff --git a/python/samples/02-agents/harness/console/state_driver.py b/python/samples/02-agents/harness/console/state_driver.py index 959c8757ba..d4bc01f663 100644 --- a/python/samples/02-agents/harness/console/state_driver.py +++ b/python/samples/02-agents/harness/console/state_driver.py @@ -191,6 +191,18 @@ def write_user_input_echo(self, text: str) -> None: """ ... + 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,11 @@ def update_last_entry(self, entry_type, new_text: str) -> None: 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)}]") + def request_shutdown(self) -> None: """Request application shutdown.""" print("[Shutdown requested]") diff --git a/python/samples/02-agents/security/email_security_example.py b/python/samples/02-agents/security/email_security_example.py index 802ce25d6b..75fe0fc4a1 100644 --- a/python/samples/02-agents/security/email_security_example.py +++ b/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:") From 73401e6f662ec1a634fb68921c119c3dfb711fbd Mon Sep 17 00:00:00 2001 From: westey <164392973+westey-m@users.noreply.github.com> Date: Thu, 9 Jul 2026 17:41:06 +0000 Subject: [PATCH 2/3] Add agents.md update. --- python/packages/core/AGENTS.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/python/packages/core/AGENTS.md b/python/packages/core/AGENTS.md index 3d7b587a43..31124b70df 100644 --- a/python/packages/core/AGENTS.md +++ b/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 `_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`) From 87998a72a4dbf8879ad74362be33de9711467f4e Mon Sep 17 00:00:00 2001 From: westey <164392973+westey-m@users.noreply.github.com> Date: Thu, 9 Jul 2026 18:12:32 +0000 Subject: [PATCH 3/3] Address PR comments --- .../02-agents/harness/console/agent_runner.py | 12 +++++++++--- .../02-agents/harness/console/state_driver.py | 2 ++ 2 files changed, 11 insertions(+), 3 deletions(-) diff --git a/python/samples/02-agents/harness/console/agent_runner.py b/python/samples/02-agents/harness/console/agent_runner.py index 3fe876d32f..938a29b109 100644 --- a/python/samples/02-agents/harness/console/agent_runner.py +++ b/python/samples/02-agents/harness/console/agent_runner.py @@ -158,9 +158,15 @@ def _sync_queued_message_display(self, session: AgentSession | None) -> None: pending = self._message_injector.get_pending_messages(session) - consumed_count = len(self._last_pending_messages) - len(pending) - for i in range(min(consumed_count, len(self._last_pending_messages))): - self._ux.write_user_input_echo(self._last_pending_messages[i].text or "") + # 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]) diff --git a/python/samples/02-agents/harness/console/state_driver.py b/python/samples/02-agents/harness/console/state_driver.py index d4bc01f663..c0290fcb56 100644 --- a/python/samples/02-agents/harness/console/state_driver.py +++ b/python/samples/02-agents/harness/console/state_driver.py @@ -341,6 +341,8 @@ 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."""