fix(mcp): swallow broken/closed-pipe on dg mcp startup notifications and error path

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

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

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

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

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

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

packages/deepctl-telemetry/src/deepctl_telemetry/client.py

@@ -131,16 +131,36 @@ 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.
 
     Sentry SDK already filters most PII via send_default_pii=False, but
     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 {}

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