diff --git a/packages/deepctl-cmd-skills/src/deepctl_cmd_skills/startup_check.py b/packages/deepctl-cmd-skills/src/deepctl_cmd_skills/startup_check.py
index 405baaa..8d15322 100644
--- a/packages/deepctl-cmd-skills/src/deepctl_cmd_skills/startup_check.py
+++ b/packages/deepctl-cmd-skills/src/deepctl_cmd_skills/startup_check.py
@@ -107,7 +107,12 @@ def print_pending_notification() -> None:
     _thread = None
 
     if _result.get("should_prompt"):
-        sys.stderr.write(
-            "\033[36mAI coding assistants detected. "
-            "Run 'deepctl skills install' to set up integration.\033[0m\n"
-        )
+        try:
+            sys.stderr.write(
+                "\033[36mAI coding assistants detected. "
+                "Run 'deepctl skills install' to set up integration.\033[0m\n"
+            )
+        except (BrokenPipeError, OSError, ValueError):
+            # The output stream closed (e.g. an MCP host disconnected stdio
+            # after `dg mcp` finished). The prompt is purely advisory.
+            pass
diff --git a/packages/deepctl-cmd-skills/tests/unit/test_skills_command.py b/packages/deepctl-cmd-skills/tests/unit/test_skills_command.py
index 4f1372c..4094d87 100644
--- a/packages/deepctl-cmd-skills/tests/unit/test_skills_command.py
+++ b/packages/deepctl-cmd-skills/tests/unit/test_skills_command.py
@@ -55,6 +55,7 @@ def test_import(self):
             check_and_notify,
             print_pending_notification,
         )
+
         assert callable(check_and_notify)
         assert callable(print_pending_notification)
 
@@ -70,3 +71,28 @@ def test_suppressed_when_quiet(self):
 
         startup_check.check_and_notify(quiet=True)
         assert startup_check._thread is None
+
+    @pytest.mark.parametrize(
+        "error",
+        [
+            BrokenPipeError(32, "Broken pipe"),
+            ValueError("I/O operation on closed file"),
+        ],
+    )
+    def test_broken_pipe_swallowed(self, error):
+        """A closed/broken stderr (e.g. `dg mcp` host disconnect) is tolerated."""
+        import sys
+        import threading
+
+        from deepctl_cmd_skills import startup_check
+
+        startup_check._result = {"should_prompt": True}
+        startup_check._thread = threading.Thread(target=lambda: None)
+        startup_check._thread.start()
+        startup_check._thread.join()
+
+        broken_stderr = MagicMock()
+        broken_stderr.write.side_effect = error
+        with patch.object(sys, "stderr", broken_stderr):
+            # Must not raise.
+            startup_check.print_pending_notification()
diff --git a/packages/deepctl-cmd-update/src/deepctl_cmd_update/plugin_update_check.py b/packages/deepctl-cmd-update/src/deepctl_cmd_update/plugin_update_check.py
index b66d19e..e490cb3 100644
--- a/packages/deepctl-cmd-update/src/deepctl_cmd_update/plugin_update_check.py
+++ b/packages/deepctl-cmd-update/src/deepctl_cmd_update/plugin_update_check.py
@@ -256,18 +256,23 @@ def print_pending_plugin_notifications() -> None:
     if not updates:
         return
 
-    if len(updates) == 1:
-        u = updates[0]
-        sys.stderr.write(
-            f"\033[33mPlugin update available: {u['name']} {u['current']} → {u['latest']}"
-            f"  —  run 'deepctl plugin update {u['name']}' to upgrade\033[0m\n"
-        )
-    else:
-        sys.stderr.write("\033[33mPlugin updates available:\033[0m\n")
-        for u in updates:
+    try:
+        if len(updates) == 1:
+            u = updates[0]
             sys.stderr.write(
-                f"\033[33m  {u['name']} {u['current']} → {u['latest']}\033[0m\n"
+                f"\033[33mPlugin update available: {u['name']} {u['current']} → {u['latest']}"
+                f"  —  run 'deepctl plugin update {u['name']}' to upgrade\033[0m\n"
             )
-        sys.stderr.write(
-            "\033[33mRun 'deepctl plugin update <name>' to upgrade\033[0m\n"
-        )
+        else:
+            sys.stderr.write("\033[33mPlugin updates available:\033[0m\n")
+            for u in updates:
+                sys.stderr.write(
+                    f"\033[33m  {u['name']} {u['current']} → {u['latest']}\033[0m\n"
+                )
+            sys.stderr.write(
+                "\033[33mRun 'deepctl plugin update <name>' to upgrade\033[0m\n"
+            )
+    except (BrokenPipeError, OSError, ValueError):
+        # The output stream closed (e.g. an MCP host disconnected stdio
+        # after `dg mcp` finished). These notifications are purely advisory.
+        pass
diff --git a/packages/deepctl-cmd-update/src/deepctl_cmd_update/startup_check.py b/packages/deepctl-cmd-update/src/deepctl_cmd_update/startup_check.py
index 2b2afe2..889bab1 100644
--- a/packages/deepctl-cmd-update/src/deepctl_cmd_update/startup_check.py
+++ b/packages/deepctl-cmd-update/src/deepctl_cmd_update/startup_check.py
@@ -197,7 +197,13 @@ def print_pending_notification() -> None:
     latest = _result.get("latest")
     current = _result.get("current")
     if latest and current:
-        sys.stderr.write(
-            f"\n\033[33m  Update available:  {current} → {latest}"
-            f"  —  run \033[1mdg update\033[0m\033[33m to upgrade\033[0m\n\n"
-        )
+        try:
+            sys.stderr.write(
+                f"\n\033[33m  Update available:  {current} → {latest}"
+                f"  —  run \033[1mdg update\033[0m\033[33m to upgrade\033[0m\n\n"
+            )
+        except (BrokenPipeError, OSError, ValueError):
+            # The output stream closed (e.g. an MCP host disconnected stdio
+            # after `dg mcp` finished). Nothing useful to do — the notification
+            # is purely advisory.
+            pass
diff --git a/packages/deepctl-cmd-update/tests/unit/test_plugin_update_check.py b/packages/deepctl-cmd-update/tests/unit/test_plugin_update_check.py
index 1216f5c..9de3b70 100644
--- a/packages/deepctl-cmd-update/tests/unit/test_plugin_update_check.py
+++ b/packages/deepctl-cmd-update/tests/unit/test_plugin_update_check.py
@@ -66,12 +66,15 @@ def test_read_missing_returns_zero(self, tmp_path):
 
     def test_write_then_read_roundtrip(self, tmp_path):
         cache_file = tmp_path / "last_plugin_version_check"
-        with patch(
-            "deepctl_cmd_update.plugin_update_check._PLUGIN_CACHE_FILE",
-            cache_file,
-        ), patch(
-            "deepctl_cmd_update.plugin_update_check._CACHE_DIR",
-            tmp_path,
+        with (
+            patch(
+                "deepctl_cmd_update.plugin_update_check._PLUGIN_CACHE_FILE",
+                cache_file,
+            ),
+            patch(
+                "deepctl_cmd_update.plugin_update_check._CACHE_DIR",
+                tmp_path,
+            ),
         ):
             _write_plugin_cache_timestamp()
             ts = _read_plugin_cache_timestamp()
@@ -268,9 +271,7 @@ def test_suppressed_in_quiet_mode(self):
         check_plugins_and_notify(quiet=True)
         assert mod._thread is None
 
-    @patch(
-        "deepctl_cmd_update.plugin_update_check._is_ci", return_value=False
-    )
+    @patch("deepctl_cmd_update.plugin_update_check._is_ci", return_value=False)
     @patch(
         "deepctl_cmd_update.plugin_update_check._is_oneshot",
         return_value=False,
@@ -299,9 +300,7 @@ def test_fresh_cache_spawns_thread(self, *_mocks):
             assert mod._thread.daemon is True
             mod._thread.join(timeout=2.0)
 
-    @patch(
-        "deepctl_cmd_update.plugin_update_check._is_ci", return_value=False
-    )
+    @patch("deepctl_cmd_update.plugin_update_check._is_ci", return_value=False)
     @patch(
         "deepctl_cmd_update.plugin_update_check._is_oneshot",
         return_value=False,
@@ -347,7 +346,11 @@ def test_single_plugin_format(self):
 
         mod._result = {
             "updates": [
-                {"name": "deepctl-plugin-whisper", "current": "0.1.0", "latest": "0.2.0"}
+                {
+                    "name": "deepctl-plugin-whisper",
+                    "current": "0.1.0",
+                    "latest": "0.2.0",
+                }
             ]
         }
         mod._thread = threading.Thread(target=lambda: None)
@@ -369,7 +372,11 @@ def test_multiple_plugin_format(self):
 
         mod._result = {
             "updates": [
-                {"name": "deepctl-plugin-whisper", "current": "0.1.0", "latest": "0.2.0"},
+                {
+                    "name": "deepctl-plugin-whisper",
+                    "current": "0.1.0",
+                    "latest": "0.2.0",
+                },
                 {"name": "deepctl-plugin-asr", "current": "1.0.0", "latest": "1.1.0"},
             ]
         }
@@ -387,6 +394,36 @@ def test_multiple_plugin_format(self):
         assert "deepctl-plugin-asr" in output
         assert "deepctl plugin update <name>" in output
 
+    @pytest.mark.parametrize(
+        "error",
+        [
+            BrokenPipeError(32, "Broken pipe"),
+            ValueError("I/O operation on closed file"),
+        ],
+    )
+    def test_broken_pipe_swallowed(self, error):
+        """A closed/broken stderr (e.g. `dg mcp` host disconnect) is tolerated."""
+        import deepctl_cmd_update.plugin_update_check as mod
+
+        mod._result = {
+            "updates": [
+                {
+                    "name": "deepctl-plugin-whisper",
+                    "current": "0.1.0",
+                    "latest": "0.2.0",
+                }
+            ]
+        }
+        mod._thread = threading.Thread(target=lambda: None)
+        mod._thread.start()
+        mod._thread.join()
+
+        broken_stderr = MagicMock()
+        broken_stderr.write.side_effect = error
+        with patch.object(sys, "stderr", broken_stderr):
+            # Must not raise.
+            print_pending_plugin_notifications()
+
     def test_no_updates_no_output(self, capsys):
         import deepctl_cmd_update.plugin_update_check as mod
 
@@ -405,15 +442,11 @@ def test_thread_timeout_handling(self):
 
         # Create a thread that would take longer than the timeout
         mod._result = {}
-        mod._thread = threading.Thread(
-            target=lambda: time.sleep(10), daemon=True
-        )
+        mod._thread = threading.Thread(target=lambda: time.sleep(10), daemon=True)
         mod._thread.start()
 
         # Should not raise — just returns with no output
-        with patch.object(
-            mod._thread, "join", side_effect=lambda timeout=None: None
-        ):
+        with patch.object(mod._thread, "join", side_effect=lambda timeout=None: None):
             stderr_capture = io.StringIO()
             with patch.object(sys, "stderr", stderr_capture):
                 print_pending_plugin_notifications()
diff --git a/packages/deepctl-cmd-update/tests/unit/test_startup_check.py b/packages/deepctl-cmd-update/tests/unit/test_startup_check.py
index e5d1e25..8d3962c 100644
--- a/packages/deepctl-cmd-update/tests/unit/test_startup_check.py
+++ b/packages/deepctl-cmd-update/tests/unit/test_startup_check.py
@@ -220,3 +220,25 @@ def test_no_notification_when_up_to_date(self, capsys):
         print_pending_notification()
         captured = capsys.readouterr()
         assert captured.err == ""
+
+    @pytest.mark.parametrize(
+        "error",
+        [
+            BrokenPipeError(32, "Broken pipe"),
+            ValueError("I/O operation on closed file"),
+        ],
+    )
+    def test_broken_pipe_swallowed(self, error):
+        """A closed/broken stderr (e.g. `dg mcp` host disconnect) is tolerated."""
+        import deepctl_cmd_update.startup_check as mod
+
+        mod._result = {"latest": "2.0.0", "current": "1.0.0"}
+        mod._thread = threading.Thread(target=lambda: None)
+        mod._thread.start()
+        mod._thread.join()
+
+        broken_stderr = MagicMock()
+        broken_stderr.write.side_effect = error
+        with patch.object(sys, "stderr", broken_stderr):
+            # Must not raise.
+            print_pending_notification()
diff --git a/packages/deepctl-telemetry/src/deepctl_telemetry/client.py b/packages/deepctl-telemetry/src/deepctl_telemetry/client.py
index 04a4431..e7c7962 100644
--- a/packages/deepctl-telemetry/src/deepctl_telemetry/client.py
+++ b/packages/deepctl-telemetry/src/deepctl_telemetry/client.py
@@ -131,6 +131,25 @@ def _is_mcp_transient_noise(event: Event) -> bool:
     return True
 
 
+def _is_broken_pipe(event: Event) -> bool:
+    """Identify events caused by writing to a closed/broken output stream.
+
+    When `dg mcp` runs as an MCP server, the host can close stdio while the
+    CLI is still finishing up. Any residual write (an advisory notification, a
+    console error message) then raises ``BrokenPipeError`` — or rich's
+    ``ValueError: I/O operation on closed file`` when the underlying stream is
+    already closed. Neither is an actionable CLI bug, so drop them rather than
+    page the DX team.
+    """
+    for exc in (event.get("exception") or {}).get("values") or []:
+        exc_type = exc.get("type") or ""
+        if exc_type == "BrokenPipeError":
+            return True
+        if exc_type == "ValueError" and "closed file" in (exc.get("value") or ""):
+            return True
+    return False
+
+
 def _scrub_event(event: Event, _hint: Hint) -> Event | None:
     """Drop request bodies, headers, and any user-identifying data.
 
@@ -138,9 +157,10 @@ def _scrub_event(event: Event, _hint: Hint) -> Event | None:
     Auth tokens, project IDs, and file paths can still leak through
     breadcrumbs and exception messages. This is a defense-in-depth scrub.
 
-    Also drops known-noise events (see ``_is_mcp_transient_noise``).
+    Also drops known-noise events (see ``_is_mcp_transient_noise`` and
+    ``_is_broken_pipe``).
     """
-    if _is_mcp_transient_noise(event):
+    if _is_mcp_transient_noise(event) or _is_broken_pipe(event):
         return None
 
     request: dict[str, Any] = event.get("request") or {}
diff --git a/packages/deepctl-telemetry/tests/unit/test_telemetry.py b/packages/deepctl-telemetry/tests/unit/test_telemetry.py
index 9b7a3b9..8eb57f1 100644
--- a/packages/deepctl-telemetry/tests/unit/test_telemetry.py
+++ b/packages/deepctl-telemetry/tests/unit/test_telemetry.py
@@ -228,3 +228,76 @@ def test_scrub_event_still_scrubs_normal_request_headers(self) -> None:
         assert "ip_address" not in result["user"]
         assert "username" not in result["user"]
         assert result["user"]["id"] == "user-1"
+
+
+class TestBrokenPipeFilter:
+    """Drop broken/closed-stream events from `dg mcp` host disconnects.
+
+    Anchor: DX-CLI-P (BrokenPipeError on the startup notification write,
+    cascading into rich's ValueError: I/O operation on closed file).
+    """
+
+    def test_drops_broken_pipe_error(self) -> None:
+        from deepctl_telemetry.client import _is_broken_pipe
+
+        event = {
+            "exception": {
+                "values": [
+                    {
+                        "type": "BrokenPipeError",
+                        "value": "[Errno 32] Broken pipe",
+                        "mechanism": {"type": "excepthook", "handled": False},
+                    }
+                ]
+            },
+        }
+        assert _is_broken_pipe(event)
+
+    def test_drops_closed_file_value_error(self) -> None:
+        from deepctl_telemetry.client import _is_broken_pipe
+
+        event = {
+            "exception": {
+                "values": [
+                    {
+                        "type": "ValueError",
+                        "value": "I/O operation on closed file",
+                        "mechanism": {"type": "excepthook", "handled": False},
+                    }
+                ]
+            },
+        }
+        assert _is_broken_pipe(event)
+
+    def test_keeps_unrelated_value_error(self) -> None:
+        from deepctl_telemetry.client import _is_broken_pipe
+
+        event = {
+            "exception": {
+                "values": [{"type": "ValueError", "value": "bad config value"}]
+            },
+        }
+        assert not _is_broken_pipe(event)
+
+    def test_handles_missing_fields(self) -> None:
+        from deepctl_telemetry.client import _is_broken_pipe
+
+        assert not _is_broken_pipe({})
+        assert not _is_broken_pipe({"exception": None})
+        assert not _is_broken_pipe({"exception": {"values": []}})
+
+    def test_scrub_event_returns_none_for_broken_pipe(self) -> None:
+        from deepctl_telemetry.client import _scrub_event
+
+        event = {
+            "exception": {
+                "values": [
+                    {
+                        "type": "BrokenPipeError",
+                        "value": "[Errno 32] Broken pipe",
+                        "mechanism": {"type": "excepthook", "handled": False},
+                    }
+                ]
+            },
+        }
+        assert _scrub_event(event, {}) is None
diff --git a/src/deepctl/main.py b/src/deepctl/main.py
index 56c8bb9..f909dd5 100644
--- a/src/deepctl/main.py
+++ b/src/deepctl/main.py
@@ -256,6 +256,20 @@ def _telemetry_transaction() -> Iterator[None]:
         yield
 
 
+def _safe_console_print(message: str) -> None:
+    """Print to the console, tolerating a closed/broken output stream.
+
+    When `dg` runs as an MCP server, the host can close stdio before the
+    process finishes. A write to the closed stream raises ``BrokenPipeError``
+    (or rich's ``ValueError: I/O operation on closed file``); there is nowhere
+    left to report the message, so swallow it rather than crash on exit.
+    """
+    try:
+        console.print(message)
+    except (BrokenPipeError, OSError, ValueError):
+        pass
+
+
 def main() -> None:
     """Main entry point for the CLI."""
     try:
@@ -342,10 +356,10 @@ def main() -> None:
             print_timing_summary(detailed_timing)
 
     except KeyboardInterrupt:
-        console.print("\n[yellow]Operation cancelled by user[/yellow]")
+        _safe_console_print("\n[yellow]Operation cancelled by user[/yellow]")
         sys.exit(2)
     except Exception as e:
-        console.print(f"[red]Error: {e}[/red]")
+        _safe_console_print(f"[red]Error: {e}[/red]")
         sys.exit(2)
 
 
diff --git a/tests/unit/test_main.py b/tests/unit/test_main.py
index 1808891..47f36be 100644
--- a/tests/unit/test_main.py
+++ b/tests/unit/test_main.py
@@ -78,9 +78,7 @@ def test_load_commands_success(self, mock_plugin_manager_class):
 
     @patch("deepctl_core.plugin_manager.console")
     @patch("deepctl_core.plugin_manager.metadata.entry_points")
-    def test_load_commands_error_handling(
-        self, mock_entry_points, mock_console
-    ):
+    def test_load_commands_error_handling(self, mock_entry_points, mock_console):
         """Test error handling during command loading."""
         # Mock entry point that raises error
         mock_entry_point = Mock()
@@ -98,8 +96,7 @@ def test_load_commands_error_handling(
         # Check that error message was printed
         error_calls = [str(call) for call in mock_console.print.call_args_list]
         assert any(
-            "Error loading plugin broken-command" in call
-            for call in error_calls
+            "Error loading plugin broken-command" in call for call in error_calls
         )
 
     def test_cli_context_setup(self, runner):
@@ -123,9 +120,7 @@ def test_main_keyboard_interrupt(self):
         # Mock sys.argv and the cli call to raise KeyboardInterrupt
         with patch("sys.argv", ["deepctl"]):
             # Patch the cli function that's already imported at module level
-            with patch.object(
-                cli, "__call__", side_effect=KeyboardInterrupt()
-            ):
+            with patch.object(cli, "__call__", side_effect=KeyboardInterrupt()):
                 with pytest.raises(SystemExit) as exc_info:
                     main()
 
@@ -139,11 +134,59 @@ def test_main_general_exception(self):
         # Mock sys.argv and the cli call to raise an exception
         with patch("sys.argv", ["deepctl"]):
             # Patch the cli function that's already imported at module level
-            with patch.object(
-                cli, "__call__", side_effect=Exception("Test error")
-            ):
+            with patch.object(cli, "__call__", side_effect=Exception("Test error")):
                 with pytest.raises(SystemExit) as exc_info:
                     main()
 
                 # Click exits with code 2 when there's an error in standalone mode
                 assert exc_info.value.code == 2
+
+
+class TestSafeConsolePrint:
+    """The closed/broken-stream guard on the error/interrupt exit path.
+
+    Anchor: DX-CLI-P — a BrokenPipeError on the startup notification write
+    cascaded into rich's `ValueError: I/O operation on closed file` when the
+    error handler tried to print to the already-closed console, crashing the
+    process through the excepthook.
+    """
+
+    @pytest.mark.parametrize(
+        "error",
+        [
+            BrokenPipeError(32, "Broken pipe"),
+            ValueError("I/O operation on closed file"),
+            OSError("stream closed"),
+        ],
+    )
+    def test_swallows_closed_stream(self, error):
+        """A broken console does not raise out of _safe_console_print."""
+        # deepctl/__init__ re-exports the `main` function as `deepctl.main`,
+        # shadowing the submodule attribute — fetch the real module directly.
+        main_mod = sys.modules["deepctl.main"]
+
+        with patch.object(main_mod.console, "print", side_effect=error):
+            # Must not raise.
+            main_mod._safe_console_print("[red]Error: boom[/red]")
+
+    def test_main_survives_broken_console_on_error(self):
+        """The full DX-CLI-P cascade: cli raises AND the console is closed.
+
+        main() must still exit(2) cleanly rather than let rich's ValueError
+        escape to the excepthook.
+        """
+        main_mod = sys.modules["deepctl.main"]
+
+        with (
+            patch("sys.argv", ["deepctl"]),
+            patch.object(cli, "__call__", side_effect=Exception("boom")),
+            patch.object(
+                main_mod.console,
+                "print",
+                side_effect=ValueError("I/O operation on closed file"),
+            ),
+            pytest.raises(SystemExit) as exc_info,
+        ):
+            main_mod.main()
+
+        assert exc_info.value.code == 2