diff --git a/amplifier_module_provider_github_copilot/config/_sdk_protection.py b/amplifier_module_provider_github_copilot/config/_sdk_protection.py index 1085304..03bb737 100644 --- a/amplifier_module_provider_github_copilot/config/_sdk_protection.py +++ b/amplifier_module_provider_github_copilot/config/_sdk_protection.py @@ -209,6 +209,14 @@ class SdkConfig: # Allows operators to enable debug without code changes. log_level_env_var: str = "COPILOT_SDK_LOG_LEVEL" + # Timeout for CopilotClient.stop() inside CopilotClientWrapper.close(). + # stop() tears down the ~500MB Electron subprocess; an unresponsive or + # wedged subprocess leaves that await pending forever, and close() is on + # the mount()-cleanup path -- so an unbounded stop hangs Amplifier's + # session cleanup for the whole process. Bound it, warn, and abandon. + # Contract: sdk-protection:Subprocess:MUST:8 + close_timeout_seconds: float = 5.0 + # Pre-warm SDK subprocess at mount() time. # When true, subprocess spawn (~2s) happens in background during mount(), # so first complete() has ~200ms latency instead of ~2000ms. diff --git a/amplifier_module_provider_github_copilot/provider.py b/amplifier_module_provider_github_copilot/provider.py index 0c89f01..9a5b549 100644 --- a/amplifier_module_provider_github_copilot/provider.py +++ b/amplifier_module_provider_github_copilot/provider.py @@ -1492,7 +1492,26 @@ async def cancel_emit_tasks(self) -> None: for task in tasks_to_cancel: task.cancel() if tasks_to_cancel: - await asyncio.gather(*tasks_to_cancel, return_exceptions=True) + # Bounded: cancelling a task is a REQUEST, not a guarantee. A task + # that swallows CancelledError (or is blocked in a shielded await) + # never completes, and this gather is on the mount()-cleanup path + # -- so an unbounded drain hangs session cleanup exactly like an + # unclosable client does. Abandon the stragglers and move on. + # Contract: sdk-protection:Subprocess:MUST:8 + drain_timeout = load_sdk_protection_config().sdk.close_timeout_seconds + try: + await asyncio.wait_for( + asyncio.shield(asyncio.gather(*tasks_to_cancel, return_exceptions=True)), + timeout=drain_timeout, + ) + except TimeoutError: + logger.warning( + "[PROVIDER] %d emit task(s) did not finish cancelling " + "within %.1fs; abandoning them. Raise " + "'sdk.close_timeout_seconds' if a slow drain is expected.", + sum(1 for t in tasks_to_cancel if not t.done()), + drain_timeout, + ) self._pending_emit_tasks.clear() async def close(self) -> None: diff --git a/amplifier_module_provider_github_copilot/sdk_adapter/client.py b/amplifier_module_provider_github_copilot/sdk_adapter/client.py index 388bb6c..0eace0c 100644 --- a/amplifier_module_provider_github_copilot/sdk_adapter/client.py +++ b/amplifier_module_provider_github_copilot/sdk_adapter/client.py @@ -717,17 +717,51 @@ async def session( ) async def close(self) -> None: - """Clean up owned client resources. Safe to call multiple times.""" + """Clean up owned client resources, within a time bound. + + Safe to call multiple times. + + Contract: sdk-protection:Subprocess:MUST:8 -- bound the SDK stop. + + ``stop()`` tears down the SDK's Electron subprocess. A wedged or + unresponsive subprocess leaves that await pending forever, and this + method sits on the mount()-cleanup path (via the provider's close() + and the shared-client refcount release), so an unbounded stop hangs + Amplifier's session cleanup for the whole process. + + ``asyncio.shield`` lets the stop run to completion even if the + *enclosing* task is cancelled; ``asyncio.wait_for`` caps how long we + are willing to wait for it. On timeout we log a WARNING naming this + wrapper and the abandoned client, then return -- a slow teardown must + never become a hung session. The SDK's own graceful-shutdown bound + (v1.0.2's ``_RUNTIME_SHUTDOWN_TIMEOUT_SECONDS``) is the SDK's promise, + not ours; this is the guarantee we can actually make. + + ``self._owned_client`` is cleared before the await, so a client that + timed out or raised is never retried on a second close(). + """ self._stopped = True # Mark as stopped so is_healthy() returns False - if self._owned_client is not None: - try: - logger.info("[CLIENT] Stopping owned Copilot client...") - await self._owned_client.stop() - logger.info("[CLIENT] Copilot client stopped") - except Exception as e: - logger.warning(*self._get_safe_log()("[CLIENT] Error stopping client: %s", e)) - finally: - self._owned_client = None + owned_client = self._owned_client + if owned_client is None: + return + self._owned_client = None + close_timeout = load_sdk_protection_config().sdk.close_timeout_seconds + try: + logger.info("[CLIENT] Stopping owned Copilot client...") + await asyncio.wait_for(asyncio.shield(owned_client.stop()), timeout=close_timeout) + logger.info("[CLIENT] Copilot client stopped") + except TimeoutError: + logger.warning( + "[CLIENT] %s: Copilot client stop did not complete within " + "%.1fs; abandoning client %r. Its SDK subprocess may survive " + "until the process exits. Raise " + "'sdk.close_timeout_seconds' if a slow stop is expected.", + type(self).__name__, + close_timeout, + owned_client, + ) + except Exception as e: + logger.warning(*self._get_safe_log()("[CLIENT] Error stopping client: %s", e)) async def list_models(self) -> list[Any]: """Fetch available models from SDK backend. diff --git a/contracts/sdk-protection.md b/contracts/sdk-protection.md index aa95553..2fac52f 100644 --- a/contracts/sdk-protection.md +++ b/contracts/sdk-protection.md @@ -7,6 +7,7 @@ - **Status:** Defensive Enhancement - **Created:** 2026-03-21 — Defense-in-depth layer for SDK interaction - **Updated:** 2026-03-31 — Added Subprocess Management Invariants (MUST-5,6,7) +- **Updated:** 2026-09-06 — Added MUST-8: bound client stop and emit-task drain --- @@ -105,6 +106,19 @@ The provider MUST validate `sdk.log_level` against the allowlist defined in conf - YAML: Validation in `load_sdk_protection_config()` - ENV: Validation in `_resolve_sdk_log_level()` +### MUST-8: Bound Client Stop and Emit-Task Drain + +Every await on the mount()-cleanup path MUST be bounded by `sdk.close_timeout_seconds`. Specifically: + +- `CopilotClientWrapper.close()` MUST bound `CopilotClient.stop()`. +- `GitHubCopilotProvider.cancel_emit_tasks()` MUST bound the `asyncio.gather` that drains cancelled emit tasks. + +On timeout, log a WARNING naming the abandoned resource and return. Cleanup MUST NOT raise, and MUST NOT wait indefinitely. + +**Rationale:** `stop()` tears down a ~500MB Electron subprocess; a wedged or unresponsive subprocess leaves that await pending forever, hanging Amplifier's session cleanup for the whole process. Likewise, cancelling a task is a *request*, not a guarantee -- a task that swallows `CancelledError` never completes and wedges the drain. The SDK's own graceful-shutdown bound (v1.0.2 `_RUNTIME_SHUTDOWN_TIMEOUT_SECONDS`) is the SDK's promise, not this provider's; MUST-8 is the guarantee this provider makes regardless of SDK version. + +**Implementation:** `asyncio.wait_for(asyncio.shield(), timeout=sdk.close_timeout_seconds)`. `shield` lets the operation finish even when the enclosing task is cancelled; `wait_for` caps the wait. Clear the resource reference *before* the await so a timed-out resource is never retried. + --- ## Architectural Notes @@ -139,6 +153,7 @@ Deduplication uses O(n) set membership check where n = number of captured tools. | `sdk-protection:Subprocess:MUST:5` | Prewarm task tracking | `tests/test_client_lifecycle.py` | | `sdk-protection:Subprocess:MUST:6` | Guard re-init after stop | `tests/test_client_lifecycle.py` | | `sdk-protection:Subprocess:MUST:7` | Validate SDK config | `tests/test_sdk_protection.py` | +| `sdk-protection:Subprocess:MUST:8` | Bound client stop + emit drain | `tests/test_client_lifecycle.py`, `tests/test_provider_close.py` | --- @@ -157,6 +172,7 @@ Policy values are defined in `config/sdk_protection.yaml`. The Python code loads | `sdk.log_level` | str | "info" | SDK subprocess log level | | `sdk.log_level_env_var` | str | "COPILOT_SDK_LOG_LEVEL" | Env var override | | `sdk.prewarm_subprocess` | bool | false | Spawn subprocess at mount() | +| `sdk.close_timeout_seconds` | float | 5.0 | Ceiling on client stop / emit drain | | `sdk.valid_log_levels` | list | see below | Allowlist for validation | **Valid log levels:** `["none", "error", "warning", "info", "debug", "all"]` diff --git a/tests/test_client_lifecycle.py b/tests/test_client_lifecycle.py index 06ef04a..faece45 100644 --- a/tests/test_client_lifecycle.py +++ b/tests/test_client_lifecycle.py @@ -659,6 +659,98 @@ async def test_close_idempotent(self) -> None: await wrapper.close() assert mock_owned.stop.call_count == 1 # Still just 1 call + @pytest.mark.asyncio + async def test_close_is_bounded_when_stop_never_returns(self, caplog) -> None: + """A client whose stop() never returns must not hang cleanup. + + Regression guard: close() previously awaited ``_owned_client.stop()`` + with no ceiling. close() is on the mount()-cleanup path (via the + provider's close() and the shared-client refcount release), so a + wedged SDK subprocess hung session cleanup for the whole process. + + Contract: sdk-protection:Subprocess:MUST:8 + """ + import asyncio + import logging + import time + from unittest.mock import patch + + from amplifier_module_provider_github_copilot.config._sdk_protection import ( + SdkProtectionConfig, + ) + from amplifier_module_provider_github_copilot.sdk_adapter.client import ( + CopilotClientWrapper, + ) + + config = SdkProtectionConfig() + config.sdk.close_timeout_seconds = 0.05 + + wrapper = CopilotClientWrapper() + release = asyncio.Event() + + class _UnstoppableClient: + async def stop(self) -> None: + # Never returns until the test explicitly releases it. + await release.wait() + + owned = _UnstoppableClient() + wrapper._owned_client = owned # pyright: ignore[reportPrivateUsage,reportAttributeAccessIssue] + + with patch( + "amplifier_module_provider_github_copilot.sdk_adapter.client.load_sdk_protection_config", + return_value=config, + ): + started = time.monotonic() + with caplog.at_level(logging.WARNING): + await wrapper.close() # must not raise, must not hang + elapsed = time.monotonic() - started + + assert elapsed < 2.0, f"close() took {elapsed:.2f}s; expected ~0.05s" + assert "did not complete within" in caplog.text + assert "abandoning client" in caplog.text + # Reference dropped so a second close() does not retry the wedged stop. + assert wrapper._owned_client is None # pyright: ignore[reportPrivateUsage] + assert wrapper.is_healthy() is False + + # Let the abandoned stop task finish so the loop shuts down clean. + release.set() + await asyncio.sleep(0) + + @pytest.mark.asyncio + async def test_close_normal_stop_logs_no_warning(self, caplog) -> None: + """A well-behaved client stops once, quietly, and is released. + + Contract: sdk-protection:Subprocess:MUST:8 + """ + import logging + + from amplifier_module_provider_github_copilot.sdk_adapter.client import ( + CopilotClientWrapper, + ) + + wrapper = CopilotClientWrapper() + mock_owned = AsyncMock(spec=_MockSDKClient) + mock_owned.stop = AsyncMock() + wrapper._owned_client = mock_owned # pyright: ignore[reportPrivateUsage] + + with caplog.at_level(logging.WARNING): + await wrapper.close() + + mock_owned.stop.assert_awaited_once() + assert caplog.text == "" + assert wrapper._owned_client is None # pyright: ignore[reportPrivateUsage] + + def test_close_timeout_defaults_to_five_seconds(self) -> None: + """The shipped policy default is a 5.0s ceiling. + + Contract: sdk-protection:Subprocess:MUST:8 + """ + from amplifier_module_provider_github_copilot.config._sdk_protection import ( + SdkProtectionConfig, + ) + + assert SdkProtectionConfig().sdk.close_timeout_seconds == 5.0 + @pytest.mark.asyncio async def test_close_before_any_session(self) -> None: """close() called before any session() is safe. diff --git a/tests/test_provider_close.py b/tests/test_provider_close.py index bfb0f88..f97386c 100644 --- a/tests/test_provider_close.py +++ b/tests/test_provider_close.py @@ -11,6 +11,88 @@ from amplifier_module_provider_github_copilot.provider import GitHubCopilotProvider +class TestCancelEmitTasksIsBounded: + """cancel_emit_tasks() must not hang on a task that ignores cancellation. + + Contract: sdk-protection:Subprocess:MUST:8 + """ + + @pytest.mark.asyncio + async def test_cancel_emit_tasks_is_bounded(self, caplog): + """A task that swallows CancelledError must not wedge the drain. + + Cancelling a task is a REQUEST, not a guarantee. Before the fix this + gather was unbounded, so one uncooperative emit task hung mount() + cleanup for the whole process. + """ + import asyncio + import logging + import time + from unittest.mock import patch + + from amplifier_module_provider_github_copilot.config._sdk_protection import ( + SdkProtectionConfig, + ) + + config = SdkProtectionConfig() + config.sdk.close_timeout_seconds = 0.05 + + release = asyncio.Event() + + async def _ignores_cancellation(): + while True: + try: + await release.wait() + return + except asyncio.CancelledError: + # Deliberately uncooperative: swallow and keep waiting. + continue + + provider = GitHubCopilotProvider() + task = asyncio.ensure_future(_ignores_cancellation()) + await asyncio.sleep(0) # let it reach the await + provider._pending_emit_tasks = [task] # type: ignore[reportPrivateUsage] + + with patch( + "amplifier_module_provider_github_copilot.provider.load_sdk_protection_config", + return_value=config, + ): + started = time.monotonic() + with caplog.at_level(logging.WARNING): + await provider.cancel_emit_tasks() # must not hang + elapsed = time.monotonic() - started + + assert elapsed < 2.0, f"drain took {elapsed:.2f}s; expected ~0.05s" + assert "did not finish cancelling" in caplog.text + assert provider._pending_emit_tasks == [] # type: ignore[reportPrivateUsage] + + # Let the abandoned task finish so the loop shuts down clean. + release.set() + await asyncio.sleep(0) + task.cancel() + + @pytest.mark.asyncio + async def test_cancel_emit_tasks_cooperative_logs_no_warning(self, caplog): + """A well-behaved task drains quietly, well inside the bound.""" + import asyncio + import logging + + async def _cooperative(): + await asyncio.sleep(3600) + + provider = GitHubCopilotProvider() + task = asyncio.ensure_future(_cooperative()) + await asyncio.sleep(0) + provider._pending_emit_tasks = [task] # type: ignore[reportPrivateUsage] + + with caplog.at_level(logging.WARNING): + await provider.cancel_emit_tasks() + + assert caplog.text == "" + assert task.cancelled() + assert provider._pending_emit_tasks == [] # type: ignore[reportPrivateUsage] + + class TestProviderCloseWiring: """Verify provider.close() delegates to client.close()."""