diff --git a/docs/contrib.md b/docs/contrib.md index a024f43..b2e78f1 100644 --- a/docs/contrib.md +++ b/docs/contrib.md @@ -24,6 +24,7 @@ The helpers under `acp.contrib` package recurring patterns we saw in integration - `ToolCallTracker.start()/progress()/append_stream_text()` emits canonical `ToolCallStart` / `ToolCallProgress` updates and keeps an in-memory view via `view()` or `tool_call_model()`. - `PermissionBroker.request_for()` wraps `requestPermission` RPCs. It reuses tracker state (or a provided `ToolCall`), lets you append extra content, and defaults to Approve / Approve for session / Reject options. +- Omit `options` or pass `None` to use broker defaults. An explicit empty option list raises `MissingPermissionOptionsError` before the permission request is sent; the same applies to empty `default_options` when no per-request options are supplied. - `default_permission_options()` exposes that canonical option triple if you need to customise it. > Tip: Keep one tracker near the agent event loop. Emit notifications through it and share the tracker with `PermissionBroker` so permission prompts always match the latest tool call state. diff --git a/src/acp/contrib/permissions.py b/src/acp/contrib/permissions.py index fedb00a..8bf39ed 100644 --- a/src/acp/contrib/permissions.py +++ b/src/acp/contrib/permissions.py @@ -56,7 +56,8 @@ def __init__( self._requester = requester self._tracker = tracker self._default_options = tuple( - option.model_copy(deep=True) for option in (default_options or default_permission_options()) + option.model_copy(deep=True) + for option in (default_permission_options() if default_options is None else default_options) ) async def request_for( @@ -84,7 +85,9 @@ async def request_for( existing.append(ContentToolCallContent(content=TextContentBlock(text=description))) tool_call.content = existing - option_set = tuple(option.model_copy(deep=True) for option in (options or self._default_options)) + option_set = tuple( + option.model_copy(deep=True) for option in (self._default_options if options is None else options) + ) if not option_set: raise MissingPermissionOptionsError() diff --git a/tests/contrib/test_contrib_permissions.py b/tests/contrib/test_contrib_permissions.py index 4ad9105..d705434 100644 --- a/tests/contrib/test_contrib_permissions.py +++ b/tests/contrib/test_contrib_permissions.py @@ -2,7 +2,7 @@ import pytest -from acp.contrib.permissions import PermissionBroker, default_permission_options +from acp.contrib.permissions import MissingPermissionOptionsError, PermissionBroker, default_permission_options from acp.contrib.tool_calls import ToolCallTracker from acp.schema import ( AllowedOutcome, @@ -58,6 +58,60 @@ async def requester(request: RequestPermissionRequest): assert recorded == ["allow"] +@pytest.mark.asyncio +async def test_permission_broker_none_uses_standard_options(): + tracker = ToolCallTracker(id_factory=lambda: "standard") + tracker.start("external", title="Standard options") + recorded: list[list[str]] = [] + + async def requester(request: RequestPermissionRequest): + recorded.append([option.option_id for option in request.options]) + return RequestPermissionResponse(outcome=AllowedOutcome(option_id="approve", outcome="selected")) + + broker = PermissionBroker("session", requester, tracker=tracker, default_options=None) + await broker.request_for("external", options=None) + assert recorded == [["approve", "approve_for_session", "reject"]] + + +@pytest.mark.asyncio +async def test_permission_broker_custom_default_and_override(): + tracker = ToolCallTracker(id_factory=lambda: "reject-only") + tracker.start("external", title="Reject only") + reject = PermissionOption(option_id="reject", name="Reject", kind="reject_once") + allow = PermissionOption(option_id="allow", name="Allow", kind="allow_once") + recorded: list[list[str]] = [] + + async def requester(request: RequestPermissionRequest): + recorded.append([option.option_id for option in request.options]) + return RequestPermissionResponse( + outcome=AllowedOutcome(option_id=request.options[0].option_id, outcome="selected") + ) + + broker = PermissionBroker("session", requester, tracker=tracker, default_options=[reject]) + await broker.request_for("external") + await broker.request_for("external", options=[allow]) + assert recorded == [["reject"], ["allow"]] + + +@pytest.mark.asyncio +@pytest.mark.parametrize(("default_options", "options"), [(None, []), ([], None)]) +async def test_permission_broker_rejects_empty_option_sources(default_options, options): + tracker = ToolCallTracker(id_factory=lambda: "empty") + tracker.start("external", title="No options") + recorded: list[RequestPermissionRequest] = [] + + async def requester(request: RequestPermissionRequest): + recorded.append(request) + return RequestPermissionResponse( + outcome=AllowedOutcome(option_id=request.options[0].option_id, outcome="selected") + ) + + broker = PermissionBroker("session", requester, tracker=tracker, default_options=default_options) + with pytest.raises(MissingPermissionOptionsError, match="requires at least one permission option"): + await broker.request_for("external", options=options) + assert recorded == [] + + def test_default_permission_options_shape(): options = default_permission_options() assert len(options) == 3