Python: add AsyncRequestBroker and shell command audit log

python/packages/core/agent_framework/__init__.py

@@ -148,6 +148,7 @@
         "MemoryTopicRecord",
     ),
     "._harness._mode": ("DEFAULT_MODE_SOURCE_ID", "AgentModeProvider", "get_agent_mode", "set_agent_mode"),
+    "._harness._request_broker": ("AsyncRequestBroker",),
     "._harness._todo": (
         "DEFAULT_TODO_SOURCE_ID",
         "TodoFileStore",
@@ -451,6 +452,7 @@
     "AgentSession",
     "AggregatingSkillsSource",
     "Annotation",
+    "AsyncRequestBroker",
     "BackgroundAgentsProvider",
     "BackgroundTaskInfo",
     "BackgroundTaskStatus",

python/packages/core/agent_framework/__init__.pyi

@@ -106,6 +106,7 @@ from ._harness._memory import (
     MemoryTopicRecord,
 )
 from ._harness._mode import DEFAULT_MODE_SOURCE_ID, AgentModeProvider, get_agent_mode, set_agent_mode
+from ._harness._request_broker import AsyncRequestBroker
 from ._harness._todo import (
     DEFAULT_TODO_SOURCE_ID,
     TodoFileStore,
@@ -406,6 +407,7 @@ __all__ = [
     "AgentSession",
     "AggregatingSkillsSource",
     "Annotation",
+    "AsyncRequestBroker",
     "BackgroundAgentsProvider",
     "BackgroundTaskInfo",
     "BackgroundTaskStatus",

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

@@ -0,0 +1,131 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""AsyncRequestBroker: a generic asyncio pause-and-wait primitive.
+
+Provides :class:`AsyncRequestBroker`, which lets any server-side code suspend
+execution on a request ID and wait asynchronously for a client to deliver the
+matching result via :meth:`AsyncRequestBroker.resolve` or
+:meth:`AsyncRequestBroker.reject`.
+
+The broker tracks recently settled request IDs in a bounded deque (default
+256 entries). A request counts as settled once it is resolved, rejected,
+timed out, or cancelled. Re-using a settled ID raises ``ValueError`` instead of
+creating a new waiter that a late response could complete with stale data.
+Late responses for recently settled IDs return ``True`` (and are ignored);
+responses for unknown or evicted IDs return ``False`` rather than raising.
+
+Usage::
+
+    broker = AsyncRequestBroker()
+
+    # Server side - suspend until answered (or timeout)
+    result = await broker.request("req-123", timeout=30.0)
+
+    # Client side - deliver the answer
+    broker.resolve("req-123", decision)
+"""
+
+from __future__ import annotations
+
+import asyncio
+import collections
+from typing import Generic, TypeVar
+
+from .._feature_stage import ExperimentalFeature, experimental
+
+__all__ = ["AsyncRequestBroker"]
+
+T = TypeVar("T")
+
+
+@experimental(feature_id=ExperimentalFeature.HARNESS)
+class AsyncRequestBroker(Generic[T]):
+    """Generic asyncio pause-and-wait broker for request/response patterns.
+
+    Allows server-side code to suspend on a request ID and await a client
+    response delivered via :meth:`resolve` or :meth:`reject`, without
+    polling or blocking a thread.
+
+    Thread-safety: designed for single-threaded asyncio use. The broker must
+    be used within a single event loop.
+    """
+
+    DEFAULT_SETTLED_DEQUE_SIZE = 256
+
+    def __init__(self, settled_deque_size: int = DEFAULT_SETTLED_DEQUE_SIZE) -> None:
+        self._pending: dict[str, asyncio.Future[T]] = {}
+        self._settled: collections.deque[str] = collections.deque(maxlen=settled_deque_size)
+
+    async def request(self, request_id: str, *, timeout: float | None = None) -> T:
+        """Suspend until the request is resolved or rejected.
+
+        Args:
+            request_id: Unique identifier for this request.
+
+        Keyword Args:
+            timeout: Maximum seconds to wait. ``None`` waits indefinitely.
+
+        Returns:
+            The result passed to :meth:`resolve`.
+
+        Raises:
+            ValueError: If the ID is already pending or was recently settled
+                (resolved, rejected, timed out, or cancelled).
+            asyncio.TimeoutError: If ``timeout`` elapses before a response arrives.
+            asyncio.CancelledError: If the waiting task is cancelled.
+            Exception: Any exception passed to :meth:`reject`.
+        """
+        if request_id in self._settled:
+            raise ValueError(f"Request '{request_id}' was already settled.")
+        if request_id in self._pending:
+            raise ValueError(f"Request '{request_id}' is already pending.")
+        loop = asyncio.get_running_loop()
+        future: asyncio.Future[T] = loop.create_future()
+        self._pending[request_id] = future
+        try:
+            return await asyncio.wait_for(asyncio.shield(future), timeout=timeout)
+        except BaseException:
+            if self._pending.get(request_id) is future:
+                del self._pending[request_id]
+                # Remember the ID so a late response cannot complete a new waiter
+                # that re-uses it, and so re-use is reported as already settled.
+                self._settled.append(request_id)
+            future.cancel()
+            raise
+
+    def resolve(self, request_id: str, result: T) -> bool:
+        """Deliver a successful result to a waiting :meth:`request` call.
+
+        Returns:
+            ``True`` if a waiter was notified or the request was recently settled
+            (the late result is ignored). ``False`` if the request ID is unknown
+            (never registered, or evicted from the settled history).
+        """
+        future = self._pending.pop(request_id, None)
+        if future is None:
+            return request_id in self._settled
+        self._settled.append(request_id)
+        if not future.done():
+            future.set_result(result)
+        return True
+
+    def reject(self, request_id: str, exception: BaseException) -> bool:
+        """Deliver an exception to a waiting :meth:`request` call.
+
+        Returns:
+            ``True`` if a waiter was notified or the request was recently settled
+            (the late exception is ignored). ``False`` if the request ID is unknown
+            (never registered, or evicted from the settled history).
+        """
+        future = self._pending.pop(request_id, None)
+        if future is None:
+            return request_id in self._settled
+        self._settled.append(request_id)
+        if not future.done():
+            future.set_exception(exception)
+        return True
+
+    @property
+    def pending_count(self) -> int:
+        """Number of outstanding requests currently awaiting a response."""
+        return len(self._pending)

python/packages/core/tests/core/test_request_broker.py

@@ -0,0 +1,109 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+
+import pytest
+
+from agent_framework import AsyncRequestBroker
+
+
+async def test_resolve_delivers_result() -> None:
+    broker: AsyncRequestBroker[str] = AsyncRequestBroker()
+    task = asyncio.create_task(broker.request("r1"))
+    await asyncio.sleep(0)
+    assert broker.resolve("r1", "value") is True
+    assert await task == "value"
+
+
+async def test_reject_delivers_exception() -> None:
+    broker: AsyncRequestBroker[str] = AsyncRequestBroker()
+    task = asyncio.create_task(broker.request("r1"))
+    await asyncio.sleep(0)
+    assert broker.reject("r1", RuntimeError("boom")) is True
+    with pytest.raises(RuntimeError, match="boom"):
+        await task
+
+
+async def test_timeout_raises() -> None:
+    broker: AsyncRequestBroker[str] = AsyncRequestBroker()
+    with pytest.raises(asyncio.TimeoutError):
+        await broker.request("r1", timeout=0.001)
+    assert broker.pending_count == 0
+
+
+async def test_resolve_returns_false_when_not_found() -> None:
+    broker: AsyncRequestBroker[str] = AsyncRequestBroker()
+    assert broker.resolve("missing", "x") is False
+    assert broker.reject("missing", RuntimeError("x")) is False
+
+
+async def test_pending_count() -> None:
+    broker: AsyncRequestBroker[int] = AsyncRequestBroker()
+    task = asyncio.create_task(broker.request("r1"))
+    await asyncio.sleep(0)
+    assert broker.pending_count == 1
+    broker.resolve("r1", 1)
+    await task
+    assert broker.pending_count == 0
+
+
+async def test_duplicate_request_raises() -> None:
+    broker: AsyncRequestBroker[int] = AsyncRequestBroker()
+    task = asyncio.create_task(broker.request("r1"))
+    await asyncio.sleep(0)
+    with pytest.raises(ValueError, match="already pending"):
+        await broker.request("r1")
+    broker.resolve("r1", 1)
+    await task
+    with pytest.raises(ValueError, match="already settled"):
+        await broker.request("r1")
+
+
+async def test_resolve_and_reject_return_true_for_recently_settled_id() -> None:
+    broker: AsyncRequestBroker[int] = AsyncRequestBroker()
+    task = asyncio.create_task(broker.request("r1"))
+    await asyncio.sleep(0)
+    assert broker.resolve("r1", 1) is True
+    assert await task == 1
+    # Late duplicate responses for a recently settled ID are acknowledged.
+    assert broker.resolve("r1", 2) is True
+    assert broker.reject("r1", RuntimeError("late")) is True
+    # Unknown IDs are still reported as not found.
+    assert broker.resolve("unknown", 1) is False
+    assert broker.reject("unknown", RuntimeError("x")) is False
+
+
+async def test_evicted_settled_id_returns_false() -> None:
+    broker: AsyncRequestBroker[int] = AsyncRequestBroker(settled_deque_size=1)
+    for request_id in ("r1", "r2"):
+        task = asyncio.create_task(broker.request(request_id))
+        await asyncio.sleep(0)
+        broker.resolve(request_id, 0)
+        await task
+    assert broker.resolve("r1", 1) is False
+    assert broker.resolve("r2", 1) is True
+
+
+async def test_timed_out_id_is_settled_and_cannot_be_reused() -> None:
+    broker: AsyncRequestBroker[str] = AsyncRequestBroker()
+    with pytest.raises(asyncio.TimeoutError):
+        await broker.request("r1", timeout=0.001)
+    # A late response for the expired request is acknowledged but not delivered.
+    assert broker.resolve("r1", "stale") is True
+    with pytest.raises(ValueError, match="already settled"):
+        await broker.request("r1")
+    assert broker.pending_count == 0
+
+
+async def test_cancelled_id_is_settled_and_cannot_be_reused() -> None:
+    broker: AsyncRequestBroker[str] = AsyncRequestBroker()
+    task = asyncio.create_task(broker.request("r1"))
+    await asyncio.sleep(0)
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+    assert broker.pending_count == 0
+    assert broker.resolve("r1", "stale") is True
+    with pytest.raises(ValueError, match="already settled"):
+        await broker.request("r1")
+    assert broker.pending_count == 0

python/packages/tools/agent_framework_tools/shell/__init__.py

@@ -4,6 +4,7 @@
 
 from __future__ import annotations
 
+from ._audit import ShellAuditLog, ShellCommandRecord
 from ._docker import (
     DEFAULT_IMAGE as DOCKER_DEFAULT_IMAGE,
 )
@@ -35,7 +36,9 @@
     "DockerNotAvailableError",
     "DockerShellTool",
     "LocalShellTool",
+    "ShellAuditLog",
     "ShellCommandError",
+    "ShellCommandRecord",
     "ShellDecision",
     "ShellEnvironmentProvider",
     "ShellEnvironmentProviderOptions",

python/packages/tools/agent_framework_tools/shell/_audit.py

@@ -0,0 +1,93 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Shell command audit log for compliance and observability."""
+
+from __future__ import annotations
+
+import datetime
+from dataclasses import dataclass, field, replace
+
+__all__ = ["ShellAuditLog", "ShellCommandRecord"]
+
+
+@dataclass
+class ShellCommandRecord:
+    """A record of a single shell command execution attempt.
+
+    Stored in the tool's audit log for compliance and observability.
+    All timestamps are UTC ISO 8601.
+    """
+
+    command: str
+    """The command string as submitted."""
+
+    policy_decision: str
+    """The :class:`~agent_framework_tools.shell.ShellDecision` outcome: ``"allow"`` or ``"deny"``."""
+
+    workdir: str | None = None
+    """Working directory used for policy evaluation and execution.
+
+    ``None`` means no working directory was configured and the command ran in
+    the host process's current working directory.
+    """
+
+    timestamp_utc: str = field(default_factory=lambda: datetime.datetime.now(datetime.timezone.utc).isoformat())
+    """UTC ISO 8601 timestamp when the command was evaluated."""
+
+    exit_code: int | None = None
+    """Process exit code, set after execution. ``None`` if the command was denied or not yet run."""
+
+    denial_reason: str = ""
+    """Reason from :class:`~agent_framework_tools.shell.ShellDecision` when denied."""
+
+
+class ShellAuditLog:
+    """An in-memory audit log of shell command decisions and outcomes.
+
+    Records every command evaluated by :class:`~agent_framework_tools.shell.LocalShellTool`,
+    including the policy decision, execution outcome, and exit code.
+
+    Access via ``LocalShellTool.audit_log``.
+
+    Examples:
+        .. code-block:: python
+
+            tool = LocalShellTool()
+            await tool.run("echo hello")
+
+            for record in tool.audit_log.records:
+                print(record.command, record.policy_decision, record.exit_code)
+    """
+
+    def __init__(self) -> None:
+        self._records: list[ShellCommandRecord] = []
+
+    def record(
+        self,
+        command: str,
+        policy_decision: str,
+        denial_reason: str = "",
+        *,
+        workdir: str | None = None,
+    ) -> ShellCommandRecord:
+        """Add a new record at evaluation time (before execution)."""
+        rec = ShellCommandRecord(
+            command=command,
+            policy_decision=policy_decision,
+            workdir=workdir,
+            denial_reason=denial_reason,
+        )
+        self._records.append(rec)
+        return rec
+
+    @property
+    def records(self) -> list[ShellCommandRecord]:
+        """Snapshots of all recorded command attempts, in chronological order.
+
+        Each call returns detached copies, so mutating a returned record does
+        not alter the stored audit history.
+        """
+        return [replace(rec) for rec in self._records]
+
+    def __len__(self) -> int:
+        return len(self._records)