Stop the CLI server and close all active sessions. This method performs graceful cleanup: 1. Closes all active sessions (releases in-memory resources) 2. Requests runtime shutdown for SDK-owned CLI processes 3. Closes the JSON-RPC connection 4. Termi
(self)
| 1467 | raise |
| 1468 | |
| 1469 | async def stop(self) -> None: |
| 1470 | """ |
| 1471 | Stop the CLI server and close all active sessions. |
| 1472 | |
| 1473 | This method performs graceful cleanup: |
| 1474 | 1. Closes all active sessions (releases in-memory resources) |
| 1475 | 2. Requests runtime shutdown for SDK-owned CLI processes |
| 1476 | 3. Closes the JSON-RPC connection |
| 1477 | 4. Terminates the CLI server process (if spawned by this client) |
| 1478 | |
| 1479 | Note: session data on disk is preserved, so sessions can be resumed |
| 1480 | later. To permanently remove session data before stopping, call |
| 1481 | :meth:`delete_session` for each session first. |
| 1482 | |
| 1483 | Raises: |
| 1484 | ExceptionGroup[StopError]: If any errors occurred during cleanup. |
| 1485 | |
| 1486 | Example: |
| 1487 | >>> try: |
| 1488 | ... await client.stop() |
| 1489 | ... except* StopError as eg: |
| 1490 | ... for error in eg.exceptions: |
| 1491 | ... print(f"Cleanup error: {error.message}") |
| 1492 | """ |
| 1493 | errors: list[StopError] = [] |
| 1494 | |
| 1495 | # Atomically take ownership of all sessions and clear the dict |
| 1496 | # so no other thread can access them |
| 1497 | with self._sessions_lock: |
| 1498 | sessions_to_destroy = list(self._sessions.values()) |
| 1499 | self._sessions.clear() |
| 1500 | |
| 1501 | for session in sessions_to_destroy: |
| 1502 | try: |
| 1503 | await session.disconnect() |
| 1504 | except Exception as e: |
| 1505 | logger.debug( |
| 1506 | "Error while cleaning up Copilot session %s", |
| 1507 | session.session_id, |
| 1508 | exc_info=True, |
| 1509 | ) |
| 1510 | errors.append( |
| 1511 | StopError(message=f"Failed to disconnect session {session.session_id}: {e}") |
| 1512 | ) |
| 1513 | |
| 1514 | if self._rpc is not None and self._cli_process is not None and not self._is_external_server: |
| 1515 | runtime_shutdown_start = time.perf_counter() |
| 1516 | try: |
| 1517 | await self._rpc.runtime.shutdown(timeout=_RUNTIME_SHUTDOWN_TIMEOUT_SECONDS) |
| 1518 | log_timing( |
| 1519 | logger, |
| 1520 | logging.DEBUG, |
| 1521 | "CopilotClient.stop runtime shutdown complete", |
| 1522 | runtime_shutdown_start, |
| 1523 | ) |
| 1524 | except Exception as e: |
| 1525 | log_timing( |
| 1526 | logger, |