Python: add AsyncRequestBroker and shell command audit log

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

@@ -28,6 +28,7 @@
 )
 
 logger = logging.getLogger(__name__)
+_audit_logger = logging.getLogger("agent_framework.shell.audit")
 
 
 def _quote_posix(value: str) -> str:
@@ -143,6 +144,15 @@ class LocalShellTool:
             will execute any command the model emits.
         on_command: Optional audit hook called with the command string for
             every command that passes policy. Use for logging / telemetry.
+
+    Note:
+        Shell execution events are emitted through
+        ``logging.getLogger("agent_framework.shell.audit")`` with ``shell.policy``
+        (policy decisions) and ``shell.execution`` (outcomes) messages.
+        Configure a handler on that logger to collect, store, or forward events.
+        Commands are not included in log records by default because they may
+        contain secrets; use the ``on_command`` hook if you need them in a
+        controlled environment.
     """
 
     def __init__(
@@ -236,6 +246,9 @@ async def __aexit__(self, *_exc: object) -> None:
 
     # ------------------------------------------------------------------ core run
 
+    def _effective_workdir(self) -> str:
+        return os.path.abspath(self._workdir) if self._workdir is not None else os.getcwd()
+
     async def run(self, command: str, *, timeout: float | None = None) -> ShellResult:
         """Execute ``command`` directly and return its :class:`ShellResult`.
 
@@ -256,6 +269,19 @@ async def run(self, command: str, *, timeout: float | None = None) -> ShellResul
         mark_feature_used(FeatureIndex.TOOLS_SHELL)
         request = ShellRequest(command=command, workdir=self._workdir)
         decision = self._policy.evaluate(request)
+        # Resolve the effective cwd so records never report ``None``.
+        # In persistent mode the shell's real cwd may differ if
+        # ``confine_workdir=False``; this reports the configured location.
+        effective_workdir = self._effective_workdir()
+        _audit_logger.info(
+            "shell.policy",
+            extra={
+                "policy_decision": decision.decision,
+                "denial_reason": decision.reason if decision.decision == "deny" else "",
+                "workdir": effective_workdir,
+                # command is intentionally omitted -- it may contain secrets.
+            },
+        )
         if decision.decision == "deny":
             raise ShellCommandError(f"Command rejected by policy: {decision.reason}")
         if self._on_command is not None:
@@ -266,22 +292,43 @@ async def run(self, command: str, *, timeout: float | None = None) -> ShellResul
 
         effective_timeout = self._timeout if timeout is None else timeout
 
-        if self._mode == "persistent":
-            if self._session is None:
-                await self.start()
-            if self._session is None:
-                raise RuntimeError("LocalShellTool session failed to start")
-            effective = self._maybe_reanchor(command)
-            return await self._session.run(effective, timeout=effective_timeout)
-
-        return await run_stateless(
-            self._stateless_argv,
-            command,
-            workdir=self._workdir,
-            env=self._env,
-            timeout=effective_timeout,
-            max_output_bytes=self._max_output_bytes,
-        )
+        exit_code: int | None = None
+        timed_out = False
+        outcome = "unknown"
+        try:
+            if self._mode == "persistent":
+                if self._session is None:
+                    await self.start()
+                if self._session is None:
+                    raise RuntimeError("LocalShellTool session failed to start")
+                effective = self._maybe_reanchor(command)
+                result = await self._session.run(effective, timeout=effective_timeout)
+            else:
+                result = await run_stateless(
+                    self._stateless_argv,
+                    command,
+                    workdir=self._workdir,
+                    env=self._env,
+                    timeout=effective_timeout,
+                    max_output_bytes=self._max_output_bytes,
+                )
+            exit_code = result.exit_code
+            timed_out = bool(getattr(result, "timed_out", False))
+            outcome = "timeout" if timed_out else ("success" if exit_code == 0 else "failure")
+            return result
+        except BaseException as exc:
+            outcome = "cancelled" if isinstance(exc, asyncio.CancelledError) else "exception"
+            raise
+        finally:
+            _audit_logger.info(
+                "shell.execution",
+                extra={
+                    "exit_code": exit_code,
+                    "timed_out": timed_out,
+                    "outcome": outcome,
+                    "workdir": effective_workdir,
+                },
+            )
 
     # ------------------------------------------------------------------ AF wiring
 

python/packages/tools/tests/test_local_shell_tool.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import json
+import logging
 import os
 import shlex
 import shutil
@@ -846,3 +847,68 @@ async def test_persistent_confines_workdir_by_default_powershell(tmp_path: os.Pa
         await tool.run(f"Set-Location -LiteralPath '{subdir}'")
         pwd = await tool.run("(Get-Location).Path")
         assert os.path.realpath(pwd.stdout.strip()) == os.path.realpath(str(tmp_path))
+
+
+def _audit_records(caplog: pytest.LogCaptureFixture, message: str) -> list[Any]:
+    return [r for r in caplog.records if r.name == "agent_framework.shell.audit" and r.message == message]
+
+
+async def test_shell_logging_records_policy_decision_for_allowed_command(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    tool = LocalShellTool(mode="stateless", approval_mode="never_require", acknowledge_unsafe=True)
+    cmd = "Write-Output hi" if sys.platform == "win32" else "echo hi"
+    with caplog.at_level(logging.INFO, logger="agent_framework.shell.audit"):
+        await tool.run(cmd)
+    records = _audit_records(caplog, "shell.policy")
+    assert len(records) == 1
+    assert records[0].__dict__["policy_decision"] == "allow"
+    assert records[0].__dict__["denial_reason"] == ""
+    assert "command" not in records[0].__dict__
+
+
+async def test_shell_logging_records_execution_outcome(caplog: pytest.LogCaptureFixture) -> None:
+    tool = LocalShellTool(mode="stateless", approval_mode="never_require", acknowledge_unsafe=True)
+    cmd = "Write-Output hi" if sys.platform == "win32" else "echo hi"
+    with caplog.at_level(logging.INFO, logger="agent_framework.shell.audit"):
+        await tool.run(cmd)
+    records = _audit_records(caplog, "shell.execution")
+    assert len(records) == 1
+    assert records[0].__dict__["exit_code"] == 0
+    assert records[0].__dict__["timed_out"] is False
+    assert records[0].__dict__["outcome"] == "success"
+
+
+async def test_shell_logging_records_denial(caplog: pytest.LogCaptureFixture) -> None:
+    tool = LocalShellTool(
+        mode="stateless",
+        approval_mode="never_require",
+        acknowledge_unsafe=True,
+        policy=ShellPolicy(denylist=[r".*"]),
+    )
+    with caplog.at_level(logging.INFO, logger="agent_framework.shell.audit"), pytest.raises(ShellCommandError):
+        await tool.run("echo hi")
+    policy_records = _audit_records(caplog, "shell.policy")
+    assert len(policy_records) == 1
+    assert policy_records[0].__dict__["policy_decision"] == "deny"
+    assert policy_records[0].__dict__["denial_reason"]
+    assert _audit_records(caplog, "shell.execution") == []
+
+
+async def test_shell_logging_effective_workdir_resolves_none(caplog: pytest.LogCaptureFixture) -> None:
+    """With no workdir configured, the logged workdir is the process cwd, not None."""
+    tool = LocalShellTool(mode="stateless", approval_mode="never_require", acknowledge_unsafe=True)
+    cmd = "Write-Output hi" if sys.platform == "win32" else "echo hi"
+    with caplog.at_level(logging.INFO, logger="agent_framework.shell.audit"):
+        await tool.run(cmd)
+    records = _audit_records(caplog, "shell.policy")
+    assert records[0].__dict__["workdir"] == os.getcwd()
+
+
+async def test_shell_logging_records_configured_workdir(caplog: pytest.LogCaptureFixture, tmp_path: Any) -> None:
+    tool = LocalShellTool(mode="stateless", approval_mode="never_require", acknowledge_unsafe=True, workdir=tmp_path)
+    cmd = "Write-Output hi" if sys.platform == "win32" else "echo hi"
+    with caplog.at_level(logging.INFO, logger="agent_framework.shell.audit"):
+        await tool.run(cmd)
+    assert _audit_records(caplog, "shell.policy")[0].__dict__["workdir"] == os.path.abspath(tmp_path)
+    assert _audit_records(caplog, "shell.execution")[0].__dict__["workdir"] == os.path.abspath(tmp_path)

python/samples/02-agents/harness/README.md

@@ -30,6 +30,7 @@ Each feature can be disabled or customized via keyword arguments.
 |------|-------------|
 | `harness_research.py` | Interactive research assistant with web search, a plan/execute workflow, and an execute-mode loop that re-invokes the agent until every todo is complete |
 | `harness_data_processing.py` | Data-processing assistant over a folder of CSV files, demonstrating file-access tools and tool approval |
+| `async_request_broker.py` | MAF approval pause-and-resume: agent pauses for shell-command approval, application delivers the request to a simulated client, and resumes the agent with the decision |
 | [`build_your_own_claw/`](./build_your_own_claw/README.md) | *Build your own claw* blog series — a personal finance assistant built step by step |
 
 ## Running