diff --git a/python/packages/github_copilot/README.md b/python/packages/github_copilot/README.md index 4d07cabcb8..6e24d19029 100644 --- a/python/packages/github_copilot/README.md +++ b/python/packages/github_copilot/README.md @@ -50,6 +50,34 @@ agent = GitHubCopilotAgent( > Note: with the default (deny-all) permission handler, an `always_require` tool is denied > unless you wire an approving `on_permission_request`. +### Approving for the rest of the session + +`PermissionDecisionApproveForSession` scopes its approval with either an `approval` (tool +prompts) or a `domain` (URL prompts). Both are optional, so a bare +`PermissionDecisionApproveForSession()` carries no scope at all and the Copilot CLI cannot +interpret it. + +`GitHubCopilotAgent` therefore scopes such a decision automatically, using the request that +triggered it — a shell prompt becomes an approval for that prompt's command identifiers, an +MCP prompt an approval for that server and tool, a URL prompt an approval for that URL's +domain, and so on: + +```python +from copilot.generated.rpc import PermissionDecisionApproveForSession + + +def on_permission_request(request, invocation): + # Scoped to `request` automatically; approves that kind of call for the whole session. + return PermissionDecisionApproveForSession() +``` + +The decision is only ever narrowed, never widened. When the prompt reports that it cannot +offer session-scoped approval (`can_offer_session_approval=False`), or the request kind has +no session-scoped approval at all (such as a `hook` prompt), the decision is downgraded to a +single-use approval and a warning is logged. Pass an explicit `approval=` or `domain=` when +you want to approve something other than the request being handled — decisions that already +specify a scope are forwarded unchanged. + ### Deprecated: `on_function_approval` The `on_function_approval` callback is **deprecated**. It still works (and is still enforced diff --git a/python/packages/github_copilot/agent_framework_github_copilot/_agent.py b/python/packages/github_copilot/agent_framework_github_copilot/_agent.py index 883810f9fc..726a345bf0 100644 --- a/python/packages/github_copilot/agent_framework_github_copilot/_agent.py +++ b/python/packages/github_copilot/agent_framework_github_copilot/_agent.py @@ -10,6 +10,7 @@ import warnings from collections.abc import AsyncIterable, Awaitable, Callable, Mapping, MutableMapping, Sequence from typing import Any, ClassVar, Generic, Literal, TypedDict, cast, overload +from urllib.parse import urlparse from agent_framework import ( AgentMiddlewareLayer, @@ -52,7 +53,20 @@ try: from copilot import CopilotClient, CopilotSession, RuntimeConnection - from copilot.generated.rpc import PermissionDecisionUserNotAvailable + from copilot.generated.rpc import ( + PermissionDecisionApproveForSession, + PermissionDecisionApproveForSessionApproval, + PermissionDecisionApproveForSessionApprovalCommands, + PermissionDecisionApproveForSessionApprovalCustomTool, + PermissionDecisionApproveForSessionApprovalExtensionManagement, + PermissionDecisionApproveForSessionApprovalExtensionPermissionAccess, + PermissionDecisionApproveForSessionApprovalMCP, + PermissionDecisionApproveForSessionApprovalMemory, + PermissionDecisionApproveForSessionApprovalRead, + PermissionDecisionApproveForSessionApprovalWrite, + PermissionDecisionApproveOnce, + PermissionDecisionUserNotAvailable, + ) from copilot.session import ( Attachment, BlobAttachment, @@ -64,7 +78,21 @@ SessionHooks, SystemMessageConfig, ) - from copilot.session_events import AssistantUsageData, PermissionRequest, SessionEvent, SessionEventType + from copilot.session_events import ( + AssistantUsageData, + PermissionRequest, + PermissionRequestCustomTool, + PermissionRequestExtensionManagement, + PermissionRequestExtensionPermissionAccess, + PermissionRequestMcp, + PermissionRequestMemory, + PermissionRequestRead, + PermissionRequestShell, + PermissionRequestUrl, + PermissionRequestWrite, + SessionEvent, + SessionEventType, + ) from copilot.tools import Tool as CopilotTool from copilot.tools import ToolInvocation, ToolResult except ImportError as _copilot_import_error: @@ -81,6 +109,9 @@ ] """Type for permission request handlers. Supports both sync and async callbacks.""" +AsyncPermissionHandlerType = Callable[[PermissionRequest, dict[str, str]], "Awaitable[PermissionRequestResult]"] +"""Type for permission request handlers that are always asynchronous.""" + FunctionApprovalCallback = Callable[[Content], "bool | Awaitable[bool]"] """Deprecated approval callback for ``FunctionTool`` instances declared with @@ -140,6 +171,173 @@ def _deny_all_permissions( return PermissionDecisionUserNotAvailable() +def _derive_session_approval(request: PermissionRequest) -> PermissionDecisionApproveForSessionApproval | None: + """Build the session-scoped approval implied by ``request``. + + ``PermissionDecisionApproveForSession.approval`` describes *what* is being approved for + the remainder of the session. Its shape is dictated by the prompt that triggered it, so + it can be reconstructed from the request itself. + + Args: + request: The permission request the decision is responding to. + + Returns: + The approval covering ``request``, or ``None`` for request kinds that have no + session-scoped approval representation (such as ``hook`` prompts). + """ + if isinstance(request, PermissionRequestShell): + return PermissionDecisionApproveForSessionApprovalCommands( + command_identifiers=[command.identifier for command in request.commands] + ) + if isinstance(request, PermissionRequestRead): + return PermissionDecisionApproveForSessionApprovalRead() + if isinstance(request, PermissionRequestWrite): + return PermissionDecisionApproveForSessionApprovalWrite() + if isinstance(request, PermissionRequestMcp): + return PermissionDecisionApproveForSessionApprovalMCP( + server_name=request.server_name, tool_name=request.tool_name + ) + if isinstance(request, PermissionRequestCustomTool): + return PermissionDecisionApproveForSessionApprovalCustomTool(tool_name=request.tool_name) + if isinstance(request, PermissionRequestMemory): + return PermissionDecisionApproveForSessionApprovalMemory() + if isinstance(request, PermissionRequestExtensionManagement): + return PermissionDecisionApproveForSessionApprovalExtensionManagement(operation=request.operation) + if isinstance(request, PermissionRequestExtensionPermissionAccess): + return PermissionDecisionApproveForSessionApprovalExtensionPermissionAccess( + extension_name=request.extension_name + ) + return None + + +# Characters the WHATWG URL parser (used by the Copilot CLI) treats specially for +# special-scheme URLs in ways that can move the authority boundary: backslashes are +# normalized to forward slashes, and tabs/newlines/carriage returns are stripped before +# parsing. Python's ``urlparse`` does none of this, so a URL containing any of them may +# resolve to a different host than the CLI actually contacts. +_WHATWG_AMBIGUOUS_URL_CHARS = ("\\", "\t", "\n", "\r") + + +def _derive_url_session_domain(url: str) -> str | None: + """Return the domain to persist for a URL prompt, or ``None`` when it is unsafe to. + + The persisted domain must match the host the Copilot CLI actually contacts, but the CLI + parses URLs with WHATWG semantics while this runs on Python's ``urlparse``. The two + disagree on crafted authorities -- a backslash before the ``@`` in + ``https://example.com@evil.com`` resolves to ``example.com`` under the CLI + but ``evil.com`` under ``urlparse`` -- so trusting ``urlparse`` here could persist a + session-wide approval for an unrelated, attacker-chosen domain. + + To keep the "narrow, never widen" guarantee, the domain is only returned when the URL + contains none of the characters the two parsers handle differently; any ambiguity (or a + URL with no derivable host) yields ``None`` so the caller can approve the single request + without persisting a domain. + + Args: + url: The URL from the permission request. + + Returns: + The lower-cased host to approve for the session, or ``None`` when the URL is + parser-ambiguous or has no host. + """ + if any(char in url for char in _WHATWG_AMBIGUOUS_URL_CHARS): + return None + return urlparse(url).hostname or None + + +def _normalize_permission_decision( + decision: PermissionRequestResult, + request: PermissionRequest, +) -> PermissionRequestResult: + """Fill in the missing scope of an under-specified ``approve-for-session`` decision. + + ``PermissionDecisionApproveForSession`` carries an optional ``approval`` (tool prompts) + and an optional ``domain`` (URL prompts), so ``PermissionDecisionApproveForSession()`` + is constructible with neither. That serializes to ``{"kind": "approve-for-session"}``, + which the Copilot CLI cannot interpret -- it crashes with ``Cannot read properties of + undefined (reading 'commandIdentifiers')``, taking the whole run down with it. This + reconstructs the intended scope from ``request``. + + The decision is only ever narrowed, never widened: when the prompt does not offer + session-scoped approval, or the request kind has no session approval representation, + the decision is downgraded to a single-use approval. + + Args: + decision: The decision returned by the caller's permission handler. + request: The permission request the decision is responding to. + + Returns: + ``decision`` unchanged unless it is an ``approve-for-session`` decision missing both + ``approval`` and ``domain``, in which case an equivalent fully-scoped decision (or a + narrower single-use approval) is returned. The input is never mutated. + """ + if not isinstance(decision, PermissionDecisionApproveForSession): + return decision + if decision.approval is not None or decision.domain is not None: + return decision + + try: + if isinstance(request, PermissionRequestUrl): + domain = _derive_url_session_domain(request.url) + if domain: + return PermissionDecisionApproveForSession(domain=domain) + logger.warning( + "Permission handler returned an unscoped 'approve-for-session' decision for a URL prompt, " + "but no unambiguous domain could be derived from '%s'. Approving this request only. Return " + "PermissionDecisionApproveForSession(domain=...) to approve a domain for the session.", + request.url, + ) + return PermissionDecisionApproveOnce() + + # Only shell and write prompts advertise this; other kinds always allow session approval. + if not getattr(request, "can_offer_session_approval", True): + logger.warning( + "Permission handler returned an 'approve-for-session' decision for a '%s' prompt that does not " + "offer session-scoped approval. Approving this request only.", + request.kind, + ) + return PermissionDecisionApproveOnce() + + approval = _derive_session_approval(request) + except Exception: + logger.exception( + "Failed to derive the session approval for a '%s' permission prompt. Approving this request only.", + getattr(request, "kind", "unknown"), + ) + return PermissionDecisionApproveOnce() + + if approval is None: + logger.warning( + "Permission handler returned an unscoped 'approve-for-session' decision for a '%s' prompt, which has " + "no session-scoped approval. Approving this request only.", + request.kind, + ) + return PermissionDecisionApproveOnce() + return PermissionDecisionApproveForSession(approval=approval) + + +def _with_normalized_permission_decisions(handler: PermissionHandlerType) -> AsyncPermissionHandlerType: + """Wrap a permission handler so its decisions are normalized before reaching the SDK. + + Exceptions raised by ``handler`` deliberately propagate: the SDK already catches them + and denies the request, and preserving that keeps the secure-by-default behavior. + + Args: + handler: The caller-supplied permission handler. May be sync or async. + + Returns: + An async handler delegating to ``handler`` and normalizing its result. + """ + + async def normalized_handler(request: PermissionRequest, invocation: dict[str, str]) -> PermissionRequestResult: + result = handler(request, invocation) + if inspect.isawaitable(result): + result = await result + return _normalize_permission_decision(result, request) + + return normalized_handler + + class GitHubCopilotSettings(TypedDict, total=False): """GitHub Copilot model settings. @@ -1201,9 +1399,10 @@ def _build_session_kwargs( the Copilot SDK, so any ``create_session`` parameter is supported without a dedicated mapping here (an unknown name surfaces as a ``TypeError`` from the SDK). A few keys are handled specially because they need a secure default - (``on_permission_request`` defaults to denying all requests) or transforming: - ``tools`` are merged with the agent's tools and converted to SDK tools, and - approval callbacks are turned into ``hooks``. + (``on_permission_request`` defaults to denying all requests, and is wrapped so + under-specified ``approve-for-session`` decisions are scoped to the request that + triggered them) or transforming: ``tools`` are merged with the agent's tools and + converted to SDK tools, and approval callbacks are turned into ``hooks``. Args: streaming: Whether to enable streaming for the session. @@ -1227,7 +1426,7 @@ def _build_session_kwargs( # back to the resolved setting (which carries the default_options / env model). if not kwargs.get("model"): kwargs["model"] = self._settings.get("model") or None - kwargs["on_permission_request"] = ( + kwargs["on_permission_request"] = _with_normalized_permission_decisions( opts.get("on_permission_request") or self._permission_handler or _deny_all_permissions ) kwargs["hooks"] = self._build_session_hooks(all_tools, kwargs) diff --git a/python/packages/github_copilot/tests/test_github_copilot_agent.py b/python/packages/github_copilot/tests/test_github_copilot_agent.py index 5e4ff52e05..8d65c2ca83 100644 --- a/python/packages/github_copilot/tests/test_github_copilot_agent.py +++ b/python/packages/github_copilot/tests/test_github_copilot_agent.py @@ -31,6 +31,9 @@ from copilot.session_events import ( AssistantUsageData, Data, + PermissionRequestShell, + PermissionRequestShellCommand, + PermissionRequestWrite, SessionEvent, SessionEventType, ToolExecutionCompleteError, @@ -58,6 +61,34 @@ def pre_tool_use_input(tool_name: str) -> PreToolUseHookInput: } +def shell_request( + command_identifiers: Sequence[str], can_offer_session_approval: bool = True +) -> PermissionRequestShell: + """Build a shell permission request covering the given command identifiers.""" + return PermissionRequestShell( + can_offer_session_approval=can_offer_session_approval, + commands=[ + PermissionRequestShellCommand(identifier=identifier, read_only=True) for identifier in command_identifiers + ], + full_command_text=" && ".join(command_identifiers), + has_write_file_redirection=False, + intention="run commands", + possible_paths=[], + possible_urls=[], + ) + + +def write_request(can_offer_session_approval: bool = True) -> PermissionRequestWrite: + """Build a write permission request.""" + return PermissionRequestWrite( + can_offer_session_approval=can_offer_session_approval, + diff="+ hello", + file_name="a.txt", + intention="write a file", + new_file_contents="hello", + ) + + def create_session_event( event_type: SessionEventType, content: str | None = None, @@ -2540,6 +2571,329 @@ async def test_session_config_uses_deny_all_when_no_permission_handler_set( assert config["on_permission_request"] is not None +class TestNormalizeApproveForSession: + """Regression tests for issue #7553. + + A bare ``PermissionDecisionApproveForSession()`` serializes to + ``{"kind": "approve-for-session"}``. The Copilot CLI cannot interpret that and crashes + with ``Cannot read properties of undefined (reading 'commandIdentifiers')``, so the + agent scopes such decisions to the request that triggered them. + """ + + @staticmethod + async def normalize(request: Any, decision: Any) -> Any: + """Run ``decision`` through the agent's permission-handler wrapper.""" + from agent_framework_github_copilot._agent import _with_normalized_permission_decisions + + handler = _with_normalized_permission_decisions(lambda _request, _invocation: decision) + return await handler(request, {"session_id": "test-session"}) + + async def test_shell_request_derives_command_identifiers(self) -> None: + """A shell prompt yields a commands approval covering every command in the request.""" + from copilot.generated.rpc import ( + PermissionDecisionApproveForSession, + PermissionDecisionApproveForSessionApprovalCommands, + ) + + result = await self.normalize(shell_request(["ls", "cat"]), PermissionDecisionApproveForSession()) + + assert isinstance(result, PermissionDecisionApproveForSession) + assert isinstance(result.approval, PermissionDecisionApproveForSessionApprovalCommands) + assert result.approval.command_identifiers == ["ls", "cat"] + + async def test_normalized_shell_decision_serializes_with_command_identifiers(self) -> None: + """The serialized payload carries the key whose absence crashed the CLI.""" + from copilot.generated.rpc import PermissionDecisionApproveForSession + + result = await self.normalize(shell_request(["ls"]), PermissionDecisionApproveForSession()) + + payload = result.to_dict() + assert payload["kind"] == "approve-for-session" + assert payload["approval"]["commandIdentifiers"] == ["ls"] + + async def test_unnormalized_decision_is_missing_approval(self) -> None: + """Guard the premise of this fix: the bare decision really does omit ``approval``.""" + from copilot.generated.rpc import PermissionDecisionApproveForSession + + assert PermissionDecisionApproveForSession().to_dict() == {"kind": "approve-for-session"} + + async def test_read_request_derives_read_approval(self) -> None: + """A read prompt yields a read approval.""" + from copilot.generated.rpc import ( + PermissionDecisionApproveForSession, + PermissionDecisionApproveForSessionApprovalRead, + ) + from copilot.session_events import PermissionRequestRead + + request = PermissionRequestRead(intention="read it", path="/tmp/a.txt") + result = await self.normalize(request, PermissionDecisionApproveForSession()) + + assert isinstance(result.approval, PermissionDecisionApproveForSessionApprovalRead) + + async def test_write_request_derives_write_approval(self) -> None: + """A write prompt yields a write approval.""" + from copilot.generated.rpc import ( + PermissionDecisionApproveForSession, + PermissionDecisionApproveForSessionApprovalWrite, + ) + + result = await self.normalize(write_request(), PermissionDecisionApproveForSession()) + + assert isinstance(result.approval, PermissionDecisionApproveForSessionApprovalWrite) + + async def test_mcp_request_derives_server_and_tool(self) -> None: + """An MCP prompt yields an approval naming the server and tool.""" + from copilot.generated.rpc import ( + PermissionDecisionApproveForSession, + PermissionDecisionApproveForSessionApprovalMCP, + ) + from copilot.session_events import PermissionRequestMcp + + request = PermissionRequestMcp( + read_only=True, server_name="my-server", tool_name="my-tool", tool_title="My Tool", args={} + ) + result = await self.normalize(request, PermissionDecisionApproveForSession()) + + assert isinstance(result.approval, PermissionDecisionApproveForSessionApprovalMCP) + assert result.approval.server_name == "my-server" + assert result.approval.tool_name == "my-tool" + + async def test_custom_tool_request_derives_tool_name(self) -> None: + """A custom-tool prompt yields an approval naming the tool.""" + from copilot.generated.rpc import ( + PermissionDecisionApproveForSession, + PermissionDecisionApproveForSessionApprovalCustomTool, + ) + from copilot.session_events import PermissionRequestCustomTool + + request = PermissionRequestCustomTool(tool_description="does a thing", tool_name="my_tool", args={}) + result = await self.normalize(request, PermissionDecisionApproveForSession()) + + assert isinstance(result.approval, PermissionDecisionApproveForSessionApprovalCustomTool) + assert result.approval.tool_name == "my_tool" + + async def test_memory_request_derives_memory_approval(self) -> None: + """A memory prompt yields a memory approval.""" + from copilot.generated.rpc import ( + PermissionDecisionApproveForSession, + PermissionDecisionApproveForSessionApprovalMemory, + ) + from copilot.session_events import PermissionRequestMemory + + request = PermissionRequestMemory(fact="the sky is blue") + result = await self.normalize(request, PermissionDecisionApproveForSession()) + + assert isinstance(result.approval, PermissionDecisionApproveForSessionApprovalMemory) + + async def test_extension_management_request_preserves_operation(self) -> None: + """An extension-management prompt yields an approval carrying the request's operation.""" + from copilot.generated.rpc import ( + PermissionDecisionApproveForSession, + PermissionDecisionApproveForSessionApprovalExtensionManagement, + ) + from copilot.session_events import PermissionRequestExtensionManagement + + request = PermissionRequestExtensionManagement(operation="enable", extension_name="my-ext") + result = await self.normalize(request, PermissionDecisionApproveForSession()) + + assert isinstance(result, PermissionDecisionApproveForSession) + assert isinstance(result.approval, PermissionDecisionApproveForSessionApprovalExtensionManagement) + assert result.approval.operation == "enable" + assert result.to_dict()["approval"] == {"kind": "extension-management", "operation": "enable"} + + async def test_extension_permission_access_request_preserves_extension_name(self) -> None: + """An extension-permission-access prompt yields an approval carrying the extension name.""" + from copilot.generated.rpc import ( + PermissionDecisionApproveForSession, + PermissionDecisionApproveForSessionApprovalExtensionPermissionAccess, + ) + from copilot.session_events import PermissionRequestExtensionPermissionAccess + + request = PermissionRequestExtensionPermissionAccess(capabilities=["read"], extension_name="my-ext") + result = await self.normalize(request, PermissionDecisionApproveForSession()) + + assert isinstance(result, PermissionDecisionApproveForSession) + assert isinstance(result.approval, PermissionDecisionApproveForSessionApprovalExtensionPermissionAccess) + assert result.approval.extension_name == "my-ext" + assert result.to_dict()["approval"] == {"kind": "extension-permission-access", "extensionName": "my-ext"} + + async def test_url_request_derives_domain_instead_of_approval(self) -> None: + """A URL prompt is scoped by ``domain``; URL prompts have no ``approval``.""" + from copilot.generated.rpc import PermissionDecisionApproveForSession + from copilot.session_events import PermissionRequestUrl + + request = PermissionRequestUrl(intention="fetch", url="https://example.com/some/path?q=1") + result = await self.normalize(request, PermissionDecisionApproveForSession()) + + assert isinstance(result, PermissionDecisionApproveForSession) + assert result.domain == "example.com" + assert result.approval is None + + async def test_url_request_without_derivable_domain_falls_back_to_approve_once(self) -> None: + """A URL with no host cannot be scoped, so the decision narrows to a single approval.""" + from copilot.generated.rpc import PermissionDecisionApproveForSession, PermissionDecisionApproveOnce + from copilot.session_events import PermissionRequestUrl + + request = PermissionRequestUrl(intention="fetch", url="not-a-url") + result = await self.normalize(request, PermissionDecisionApproveForSession()) + + assert isinstance(result, PermissionDecisionApproveOnce) + + @pytest.mark.parametrize( + "url", + [ + "https://example.com\\@evil.com/a", # backslash moves the authority boundary under WHATWG + "https://example.com\t@evil.com/a", # tab is stripped by WHATWG before parsing + "https://example.com\n@evil.com/a", # newline is stripped by WHATWG before parsing + "https://example.com\r@evil.com/a", # carriage return is stripped by WHATWG before parsing + ], + ) + async def test_parser_ambiguous_url_falls_back_to_approve_once(self, url: str) -> None: + """A URL whose host Python and the CLI parse differently must not persist a domain. + + The CLI (WHATWG) contacts ``example.com`` for these URLs while ``urlparse`` derives + ``evil.com``. Persisting ``evil.com`` would widen approval to an unrelated, + attacker-chosen domain, so the decision must narrow to a single-use approval. + """ + from copilot.generated.rpc import PermissionDecisionApproveForSession, PermissionDecisionApproveOnce + from copilot.session_events import PermissionRequestUrl + + request = PermissionRequestUrl(intention="fetch", url=url) + result = await self.normalize(request, PermissionDecisionApproveForSession()) + + assert isinstance(result, PermissionDecisionApproveOnce) + + async def test_unambiguous_url_with_userinfo_derives_real_host(self) -> None: + """Userinfo without ambiguous characters is safe; the real host is persisted.""" + from copilot.generated.rpc import PermissionDecisionApproveForSession + from copilot.session_events import PermissionRequestUrl + + request = PermissionRequestUrl(intention="fetch", url="https://user:pass@example.com/a") + result = await self.normalize(request, PermissionDecisionApproveForSession()) + + assert isinstance(result, PermissionDecisionApproveForSession) + assert result.domain == "example.com" + + async def test_shell_request_that_cannot_offer_session_approval_narrows_to_approve_once(self) -> None: + """Never fabricate a session approval the prompt said it could not offer.""" + from copilot.generated.rpc import PermissionDecisionApproveForSession, PermissionDecisionApproveOnce + + request = shell_request(["rm"], can_offer_session_approval=False) + result = await self.normalize(request, PermissionDecisionApproveForSession()) + + assert isinstance(result, PermissionDecisionApproveOnce) + + async def test_write_request_that_cannot_offer_session_approval_narrows_to_approve_once(self) -> None: + """The same narrowing applies to write prompts.""" + from copilot.generated.rpc import PermissionDecisionApproveForSession, PermissionDecisionApproveOnce + + request = write_request(can_offer_session_approval=False) + result = await self.normalize(request, PermissionDecisionApproveForSession()) + + assert isinstance(result, PermissionDecisionApproveOnce) + + async def test_hook_request_without_session_approval_narrows_to_approve_once(self) -> None: + """A hook prompt has no session-scoped approval representation.""" + from copilot.generated.rpc import PermissionDecisionApproveForSession, PermissionDecisionApproveOnce + from copilot.session_events import PermissionRequestHook + + request = PermissionRequestHook(tool_name="t", hook_message="nope", tool_args={}) + result = await self.normalize(request, PermissionDecisionApproveForSession()) + + assert isinstance(result, PermissionDecisionApproveOnce) + + async def test_fully_specified_approval_is_passed_through_untouched(self) -> None: + """A decision that already names its scope must not be rewritten.""" + from copilot.generated.rpc import ( + PermissionDecisionApproveForSession, + PermissionDecisionApproveForSessionApprovalRead, + ) + + decision = PermissionDecisionApproveForSession(approval=PermissionDecisionApproveForSessionApprovalRead()) + result = await self.normalize(shell_request(["ls"]), decision) + + assert result is decision + + async def test_explicit_domain_is_passed_through_untouched(self) -> None: + """A decision scoped by ``domain`` must not be rewritten either.""" + from copilot.generated.rpc import PermissionDecisionApproveForSession + from copilot.session_events import PermissionRequestUrl + + decision = PermissionDecisionApproveForSession(domain="contoso.com") + request = PermissionRequestUrl(intention="fetch", url="https://example.com/a") + result = await self.normalize(request, decision) + + assert result is decision + assert result.domain == "contoso.com" + + @pytest.mark.parametrize("decision_name", ["PermissionDecisionApproveOnce", "PermissionDecisionUserNotAvailable"]) + async def test_other_decision_kinds_are_passed_through_untouched(self, decision_name: str) -> None: + """Only ``approve-for-session`` decisions are eligible for normalization.""" + import copilot.generated.rpc as rpc + + decision = getattr(rpc, decision_name)() + result = await self.normalize(shell_request(["ls"]), decision) + + assert result is decision + + async def test_async_handlers_are_supported(self) -> None: + """The wrapper awaits async handlers before normalizing.""" + from copilot.generated.rpc import ( + PermissionDecisionApproveForSession, + PermissionDecisionApproveForSessionApprovalCommands, + ) + + from agent_framework_github_copilot._agent import _with_normalized_permission_decisions + + async def async_handler(_request: Any, _invocation: Any) -> Any: + return PermissionDecisionApproveForSession() + + handler = _with_normalized_permission_decisions(async_handler) + result = await handler(shell_request(["ls"]), {"session_id": "test-session"}) + + assert isinstance(result, PermissionDecisionApproveForSession) + assert isinstance(result.approval, PermissionDecisionApproveForSessionApprovalCommands) + + async def test_handler_exceptions_propagate(self) -> None: + """Handler failures must keep reaching the SDK, which denies the request.""" + from agent_framework_github_copilot._agent import _with_normalized_permission_decisions + + def failing_handler(_request: Any, _invocation: Any) -> Any: + raise RuntimeError("handler exploded") + + handler = _with_normalized_permission_decisions(failing_handler) + + with pytest.raises(RuntimeError, match="handler exploded"): + await handler(shell_request(["ls"]), {"session_id": "test-session"}) + + async def test_agent_wires_the_normalizer_into_the_session( + self, + mock_client: MagicMock, + mock_session: MagicMock, + ) -> None: + """End to end: the handler reaching create_session normalizes decisions.""" + from copilot.generated.rpc import ( + PermissionDecisionApproveForSession, + PermissionDecisionApproveForSessionApprovalCommands, + ) + + def approve_for_session(_request: Any, _invocation: Any) -> Any: + return PermissionDecisionApproveForSession() + + agent = GitHubCopilotAgent( + client=mock_client, + default_options=copilot_options({"on_permission_request": approve_for_session}), + ) + await agent.start() + await agent._get_or_create_session(AgentSession()) # type: ignore[reportPrivateUsage] + + handler = mock_client.create_session.call_args.kwargs["on_permission_request"] + result = await handler(shell_request(["ls"]), {"session_id": "test-session"}) + + assert isinstance(result.approval, PermissionDecisionApproveForSessionApprovalCommands) + assert result.approval.command_identifiers == ["ls"] + + class SpyContextProvider(ContextProvider): """A context provider that records whether its hooks are called."""