diff --git a/python/packages/core/agent_framework/__init__.py b/python/packages/core/agent_framework/__init__.py index 3953187578e..9720b557024 100644 --- a/python/packages/core/agent_framework/__init__.py +++ b/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", diff --git a/python/packages/core/agent_framework/__init__.pyi b/python/packages/core/agent_framework/__init__.pyi index e8d1bc6e411..265ee28a071 100644 --- a/python/packages/core/agent_framework/__init__.pyi +++ b/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", diff --git a/python/packages/core/agent_framework/_harness/_request_broker.py b/python/packages/core/agent_framework/_harness/_request_broker.py new file mode 100644 index 00000000000..abb0e8afc0a --- /dev/null +++ b/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) diff --git a/python/packages/core/tests/core/test_request_broker.py b/python/packages/core/tests/core/test_request_broker.py new file mode 100644 index 00000000000..1ba13a72d73 --- /dev/null +++ b/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 diff --git a/python/packages/tools/agent_framework_tools/shell/__init__.py b/python/packages/tools/agent_framework_tools/shell/__init__.py index 06517a96c22..0be0e982aed 100644 --- a/python/packages/tools/agent_framework_tools/shell/__init__.py +++ b/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", diff --git a/python/packages/tools/agent_framework_tools/shell/_audit.py b/python/packages/tools/agent_framework_tools/shell/_audit.py new file mode 100644 index 00000000000..005b76f862b --- /dev/null +++ b/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) diff --git a/python/packages/tools/agent_framework_tools/shell/_tool.py b/python/packages/tools/agent_framework_tools/shell/_tool.py index d85beeebd03..56cb202f8a7 100644 --- a/python/packages/tools/agent_framework_tools/shell/_tool.py +++ b/python/packages/tools/agent_framework_tools/shell/_tool.py @@ -15,6 +15,7 @@ from agent_framework._tools import SHELL_TOOL_KIND_VALUE from .._feature_usage import FeatureIndex +from ._audit import ShellAuditLog from ._executor import run_stateless from ._policy import ShellPolicy, ShellRequest from ._resolve import is_powershell, resolve_shell @@ -143,6 +144,10 @@ 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. + + Attributes: + audit_log: :class:`ShellAuditLog` recording every evaluated command + (allowed or denied) with its policy decision and exit code. """ def __init__( @@ -181,6 +186,7 @@ def __init__( self._max_output_bytes = max_output_bytes self._approval_mode: Literal["always_require", "never_require"] = approval_mode self._on_command = on_command + self.audit_log: ShellAuditLog = ShellAuditLog() merged_env: dict[str, str] | None if env is None and not clean_env: @@ -256,6 +262,12 @@ 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) + record = self.audit_log.record( + command=command, + policy_decision=decision.decision, + denial_reason=decision.reason if decision.decision == "deny" else "", + workdir=request.workdir, + ) if decision.decision == "deny": raise ShellCommandError(f"Command rejected by policy: {decision.reason}") if self._on_command is not None: @@ -272,9 +284,11 @@ async def run(self, command: str, *, timeout: float | None = None) -> ShellResul 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) + result = await self._session.run(effective, timeout=effective_timeout) + record.exit_code = result.exit_code + return result - return await run_stateless( + result = await run_stateless( self._stateless_argv, command, workdir=self._workdir, @@ -282,6 +296,8 @@ async def run(self, command: str, *, timeout: float | None = None) -> ShellResul timeout=effective_timeout, max_output_bytes=self._max_output_bytes, ) + record.exit_code = result.exit_code + return result # ------------------------------------------------------------------ AF wiring diff --git a/python/packages/tools/tests/test_local_shell_tool.py b/python/packages/tools/tests/test_local_shell_tool.py index f9dcbdf172c..34ac7662945 100644 --- a/python/packages/tools/tests/test_local_shell_tool.py +++ b/python/packages/tools/tests/test_local_shell_tool.py @@ -846,3 +846,97 @@ 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)) + + +async def test_audit_log_records_allowed_command() -> None: + tool = LocalShellTool(mode="stateless", approval_mode="never_require", acknowledge_unsafe=True) + cmd = "Write-Output hello" if sys.platform == "win32" else "echo hello" + await tool.run(cmd) + + record = tool.audit_log.records[0] + assert record.command == cmd + assert record.policy_decision == "allow" + assert record.exit_code == 0 + assert record.workdir is None + assert record.timestamp_utc + + +async def test_audit_log_records_denied_command() -> None: + tool = LocalShellTool( + mode="stateless", + policy=ShellPolicy(denylist=[r"forbidden"]), + approval_mode="never_require", + acknowledge_unsafe=True, + ) + with pytest.raises(ShellCommandError): + await tool.run("echo forbidden") + + record = tool.audit_log.records[0] + assert record.policy_decision == "deny" + assert record.exit_code is None + assert record.denial_reason + assert record.workdir is None + + +async def test_audit_log_records_workdir_for_denied_command(tmp_path: Any) -> None: + tool = LocalShellTool( + mode="stateless", + policy=ShellPolicy(denylist=[r"forbidden"]), + approval_mode="never_require", + acknowledge_unsafe=True, + workdir=tmp_path, + ) + with pytest.raises(ShellCommandError): + await tool.run("echo forbidden") + + record = tool.audit_log.records[0] + assert record.policy_decision == "deny" + assert record.workdir == str(tmp_path) + + +async def test_audit_log_records_distinct_workdirs_for_same_command(tmp_path: Any) -> None: + dir_a = tmp_path / "a" + dir_b = tmp_path / "b" + dir_a.mkdir() + dir_b.mkdir() + cmd = "Write-Output hello" if sys.platform == "win32" else "echo hello" + + tool_a = LocalShellTool(mode="stateless", approval_mode="never_require", acknowledge_unsafe=True, workdir=dir_a) + tool_b = LocalShellTool(mode="stateless", approval_mode="never_require", acknowledge_unsafe=True, workdir=dir_b) + await tool_a.run(cmd) + await tool_b.run(cmd) + + record_a = tool_a.audit_log.records[0] + record_b = tool_b.audit_log.records[0] + assert record_a.command == record_b.command == cmd + assert record_a.workdir == str(dir_a) + assert record_b.workdir == str(dir_b) + assert record_a.workdir != record_b.workdir + + +async def test_audit_log_accumulates_multiple() -> None: + tool = LocalShellTool(mode="stateless", approval_mode="never_require", acknowledge_unsafe=True) + cmd = "Write-Output hello" if sys.platform == "win32" else "echo hello" + for _ in range(3): + await tool.run(cmd) + assert len(tool.audit_log) == 3 + + +async def test_audit_log_records_are_detached_snapshots() -> None: + tool = LocalShellTool(mode="stateless", approval_mode="never_require", acknowledge_unsafe=True) + cmd = "Write-Output hello" if sys.platform == "win32" else "echo hello" + await tool.run(cmd) + + snapshot = tool.audit_log.records[0] + snapshot.command = "tampered" + snapshot.policy_decision = "deny" + snapshot.exit_code = 99 + snapshot.workdir = "tampered" + tool.audit_log.records.clear() + + record = tool.audit_log.records[0] + assert len(tool.audit_log) == 1 + assert record.command == cmd + assert record.policy_decision == "allow" + assert record.exit_code == 0 + assert record.workdir is None diff --git a/python/samples/02-agents/harness/README.md b/python/samples/02-agents/harness/README.md index 20ccc49780d..a949cd9403d 100644 --- a/python/samples/02-agents/harness/README.md +++ b/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` | `AsyncRequestBroker` basics with no LLM required: server-side code awaits a request that a client resolves, rejects, or never answers (timeout) | | [`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 diff --git a/python/samples/02-agents/harness/async_request_broker.py b/python/samples/02-agents/harness/async_request_broker.py new file mode 100644 index 00000000000..53ca49075aa --- /dev/null +++ b/python/samples/02-agents/harness/async_request_broker.py @@ -0,0 +1,110 @@ +# Copyright (c) Microsoft. All rights reserved. + +import asyncio +import warnings + +from agent_framework import AsyncRequestBroker + +"""AsyncRequestBroker: pause server-side work until a client responds. + +This sample demonstrates how to use `AsyncRequestBroker` to let server-side +code (for example a running agent or tool) stop partway through, ask a +connected client for a decision, and continue once the client answers. + +A typical case is a tool that needs human approval before it runs. The server +calls `await broker.request(request_id)` and sends the request ID to the UI +over whatever transport you use (websocket, SSE, HTTP). When the user answers, +the client handler calls `broker.resolve(request_id, value)` or +`broker.reject(request_id, exception)`, and the awaiting coroutine continues. + +The sample covers three cases: +1. The server awaits a request and the client resolves it. +2. The server awaits a request, the client rejects it, and the server handles + the error. +3. The server awaits a request with a timeout that nobody answers. + +No LLM or credentials are needed. The "client" is simulated with +`asyncio.create_task` in the same event loop. +""" + +# AsyncRequestBroker is experimental; silence the warning to keep output tidy. +warnings.filterwarnings("ignore", message=r"\[HARNESS\]") + + +class ApprovalDenied(Exception): + """Raised on the server when the client denies an approval request.""" + + +async def server_delete_files(broker: AsyncRequestBroker[bool], request_id: str, path: str) -> None: + """Server-side work that pauses for client approval before continuing.""" + print(f"[server] Requesting approval to delete '{path}' (id={request_id})") + try: + # Suspends here, without blocking the event loop, until the client answers. + approved = await broker.request(request_id) + except ApprovalDenied as ex: + print(f"[server] Request {request_id} was rejected: {ex}. Skipping.") + return + print(f"[server] Request {request_id} answered: approved={approved}. Deleting '{path}'.") + + +async def client_respond(broker: AsyncRequestBroker[bool], request_id: str, *, approve: bool) -> None: + """Simulate a client (e.g. a UI) answering the request after a short delay.""" + await asyncio.sleep(0.1) # The user thinks about it... + if approve: + print(f"[client] Approving {request_id}") + broker.resolve(request_id, True) + else: + print(f"[client] Denying {request_id}") + broker.reject(request_id, ApprovalDenied("user said no")) + + +async def main() -> None: + broker: AsyncRequestBroker[bool] = AsyncRequestBroker() + + # 1. Resolve: the client approves and the server continues with the result. + print("=== Resolve ===") + await asyncio.gather( + server_delete_files(broker, "req-1", "./tmp/cache"), + client_respond(broker, "req-1", approve=True), + ) + + # 2. Reject: the client denies and the server gets the exception. + print("\n=== Reject ===") + await asyncio.gather( + server_delete_files(broker, "req-2", "./important"), + client_respond(broker, "req-2", approve=False), + ) + + # 3. Timeout: nobody answers, so the server stops waiting. + print("\n=== Timeout ===") + try: + await broker.request("req-3", timeout=0.2) + except asyncio.TimeoutError: + print("[server] Request req-3 timed out with no response.") + + # A late reply for a settled request is ignored. resolve() returns True to + # show the ID is known but already settled. + print(f"[client] Late reply to req-3 accepted as already settled: {broker.resolve('req-3', True)}") + print(f"Pending requests: {broker.pending_count}") + + +if __name__ == "__main__": + asyncio.run(main()) + +""" +Sample output: +=== Resolve === +[server] Requesting approval to delete './tmp/cache' (id=req-1) +[client] Approving req-1 +[server] Request req-1 answered: approved=True. Deleting './tmp/cache'. + +=== Reject === +[server] Requesting approval to delete './important' (id=req-2) +[client] Denying req-2 +[server] Request req-2 was rejected: user said no. Skipping. + +=== Timeout === +[server] Request req-3 timed out with no response. +[client] Late reply to req-3 accepted as already settled: True +Pending requests: 0 +"""