close(): bound client close so an unclosable HTTP client cannot hang session cleanup

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.

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:

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.

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(<awaitable>), 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"]`

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.

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()."""