fix(core): keep stdout pure when -o json/yaml/csv is requested

packages/deepctl-cmd-billing/src/deepctl_cmd_billing/command.py

@@ -11,12 +11,13 @@
     Config,
     DeepgramClient,
 )
-from rich.console import Console
+from deepctl_core.output import get_status_console
 from rich.table import Table
 
 from .models import BalanceInfo, BillingResult
 
-console = Console()
+# Human/status output — routed to stderr when -o json/yaml/csv owns stdout
+console = get_status_console()
 
 
 class BillingCommand(BaseCommand):

packages/deepctl-cmd-keys/src/deepctl_cmd_keys/command.py

@@ -11,12 +11,13 @@
     Config,
     DeepgramClient,
 )
-from rich.console import Console
+from deepctl_core.output import get_status_console
 from rich.table import Table
 
 from .models import KeyCreatedInfo, KeyInfo, KeysResult
 
-console = Console()
+# Human/status output — routed to stderr when -o json/yaml/csv owns stdout
+console = get_status_console()
 
 
 class KeysCommand(BaseCommand):

packages/deepctl-cmd-members/src/deepctl_cmd_members/command.py

@@ -11,12 +11,13 @@
     Config,
     DeepgramClient,
 )
-from rich.console import Console
+from deepctl_core.output import get_status_console
 from rich.table import Table
 
 from .models import InviteInfo, MemberInfo, MembersResult
 
-console = Console()
+# Human/status output — routed to stderr when -o json/yaml/csv owns stdout
+console = get_status_console()
 
 
 class MembersCommand(BaseCommand):

packages/deepctl-cmd-models/src/deepctl_cmd_models/command.py

@@ -11,12 +11,13 @@
     Config,
     DeepgramClient,
 )
-from rich.console import Console
+from deepctl_core.output import get_status_console
 from rich.table import Table
 
 from .models import ModelInfo, ModelsResult
 
-console = Console()
+# Human/status output — routed to stderr when -o json/yaml/csv owns stdout
+console = get_status_console()
 
 
 class ModelsCommand(BaseCommand):

packages/deepctl-cmd-projects/src/deepctl_cmd_projects/command.py

@@ -9,11 +9,12 @@
     Config,
     DeepgramClient,
 )
-from rich.console import Console
+from deepctl_core.output import get_status_console
 
 from .models import ProjectInfo, ProjectsResult
 
-console = Console()
+# Human/status output — routed to stderr when -o json/yaml/csv owns stdout
+console = get_status_console()
 
 
 class ProjectsCommand(BaseCommand):

packages/deepctl-cmd-read/src/deepctl_cmd_read/command.py

@@ -13,11 +13,12 @@
     Config,
     DeepgramClient,
 )
-from rich.console import Console
+from deepctl_core.output import get_status_console
 
 from .models import ReadResult
 
-console = Console()
+# Human/status output — routed to stderr when -o json/yaml/csv owns stdout
+console = get_status_console()
 
 
 class ReadCommand(BaseCommand):

packages/deepctl-cmd-requests/src/deepctl_cmd_requests/command.py

@@ -12,12 +12,13 @@
     Config,
     DeepgramClient,
 )
-from rich.console import Console
+from deepctl_core.output import get_status_console
 from rich.table import Table
 
 from .models import RequestInfo, RequestsResult
 
-console = Console()
+# Human/status output — routed to stderr when -o json/yaml/csv owns stdout
+console = get_status_console()
 
 
 class RequestsCommand(BaseCommand):

packages/deepctl-cmd-usage/src/deepctl_cmd_usage/command.py

@@ -10,12 +10,13 @@
     Config,
     DeepgramClient,
 )
+from deepctl_core.output import get_status_console
 from deepctl_shared_utils import validate_date_format
-from rich.console import Console
 
 from .models import UsageBucket, UsageResult
 
-console = Console()
+# Human/status output — routed to stderr when -o json/yaml/csv owns stdout
+console = get_status_console()
 
 
 class UsageCommand(BaseCommand):

packages/deepctl-core/src/deepctl_core/output.py

@@ -70,10 +70,6 @@ def is_agentic() -> bool:
 
 _agentic = is_agentic()
 
-# Global console instance — plain when agentic, rich when interactive
-console = Console(no_color=_agentic, highlight=not _agentic)
-stderr_console = Console(stderr=True, no_color=_agentic, highlight=not _agentic)
-
 # Global output configuration
 _output_config: dict[str, Any] = {
     "format": "default",
@@ -84,6 +80,51 @@ def is_agentic() -> bool:
 }
 
 
+# Formats where stdout carries a machine-readable payload and must not be
+# polluted by human-facing status text, tables or progress.
+MACHINE_FORMATS = frozenset({"json", "yaml", "csv"})
+
+
+class StatusConsole(Console):
+    """Console for human-facing output that yields stdout to the payload.
+
+    Commands print status lines and tables for humans, while the framework
+    writes the serialised result to stdout. When a machine-readable format is
+    requested those two collide: `dg -o json projects | jq` sees
+    "Fetching projects..." before the JSON and fails. Resolving the target at
+    write time (rather than at construction) lets status output move to stderr
+    for exactly those formats, without commands having to ask.
+    """
+
+    @property
+    def file(self) -> Any:
+        if _output_config["format"] in MACHINE_FORMATS:
+            return sys.stderr
+        return self._file or sys.stdout
+
+    @file.setter
+    def file(self, new_file: Any) -> None:
+        self._file = new_file
+
+
+# Global console instance — plain when agentic, rich when interactive.
+# `console` is the human/status channel: it steps aside to stderr whenever a
+# machine-readable format owns stdout. `stdout_console` is the payload channel
+# and always writes to stdout.
+console = StatusConsole(no_color=_agentic, highlight=not _agentic)
+stdout_console = Console(no_color=_agentic, highlight=not _agentic)
+stderr_console = Console(stderr=True, no_color=_agentic, highlight=not _agentic)
+
+
+def get_status_console() -> Console:
+    """Get the shared human/status console.
+
+    Prefer this over a module-level ``Console()`` so status output is routed to
+    stderr automatically when ``-o json``/``yaml``/``csv`` owns stdout.
+    """
+    return console
+
+
 def setup_output(
     format_type: str = "json", quiet: bool = False, verbose: bool = False
 ) -> None:
@@ -309,23 +350,23 @@ def print_output(data: Any, format_type: str | None = None) -> None:
         if isinstance(data, str):
             try:
                 parsed = json.loads(data)
-                console.print(JSON.from_data(parsed))
+                stdout_console.print(JSON.from_data(parsed))
             except json.JSONDecodeError:
-                console.print(data)
+                stdout_console.print(data)
         else:
-            console.print(JSON.from_data(data))
+            stdout_console.print(JSON.from_data(data))
     elif format_type == "yaml":
         # Use Rich's syntax highlighting for YAML
         yaml_str = formatter.format(data)
         syntax = Syntax(yaml_str, "yaml", theme="monokai", line_numbers=False)
-        console.print(syntax)
+        stdout_console.print(syntax)
     elif format_type == "table":
         # Table is already formatted for Rich
         formatted = formatter.format(data)
-        console.print(formatted, end="")
+        stdout_console.print(formatted, end="")
     else:
         # CSV and other formats
-        console.print(formatter.format(data))
+        stdout_console.print(formatter.format(data))
 
 
 def print_success(message: str) -> None:

packages/deepctl-core/tests/unit/test_output.py

@@ -1,13 +1,16 @@
 """Tests for the output utilities."""
 
 import json
+import sys
 from unittest.mock import MagicMock, Mock, patch
 
 import pytest
 import yaml
 from deepctl_core.output import (
+    MACHINE_FORMATS,
     OutputFormatter,
     get_console,
+    get_status_console,
     is_agentic,
     print_error,
     print_info,
@@ -387,7 +390,7 @@ def test_print_info(self, mock_console):
 class TestPrintOutput:
     """Test print_output function."""
 
-    @patch("deepctl_core.output.console")
+    @patch("deepctl_core.output.stdout_console")
     @patch(
         "deepctl_core.output._output_config",
         {"quiet": False, "format": "json"},
@@ -400,7 +403,7 @@ def test_print_output_json_dict(self, mock_console):
         # Should use Rich's JSON display
         mock_console.print.assert_called_once()
 
-    @patch("deepctl_core.output.console")
+    @patch("deepctl_core.output.stdout_console")
     @patch(
         "deepctl_core.output._output_config", {"quiet": True, "format": "json"}
     )
@@ -411,7 +414,7 @@ def test_print_output_quiet(self, mock_console):
         # Should not print in quiet mode
         mock_console.print.assert_not_called()
 
-    @patch("deepctl_core.output.console")
+    @patch("deepctl_core.output.stdout_console")
     @patch(
         "deepctl_core.output._output_config",
         {"quiet": False, "format": "yaml"},
@@ -424,7 +427,7 @@ def test_print_output_yaml(self, mock_console):
         # Should use syntax highlighting
         mock_console.print.assert_called_once()
 
-    @patch("deepctl_core.output.console")
+    @patch("deepctl_core.output.stdout_console")
     @patch(
         "deepctl_core.output._output_config",
         {"quiet": False, "format": "table"},
@@ -437,7 +440,7 @@ def test_print_output_table(self, mock_console):
         # Table formatting uses capture which results in multiple print calls
         assert mock_console.print.call_count >= 1
 
-    @patch("deepctl_core.output.console")
+    @patch("deepctl_core.output.stdout_console")
     @patch(
         "deepctl_core.output._output_config", {"quiet": False, "format": "csv"}
     )
@@ -460,3 +463,47 @@ def test_get_console(self):
         # Should return Rich Console instance
         assert console is not None
         assert hasattr(console, "print")
+
+
+class TestStatusConsoleRouting:
+    """Status output must yield stdout to machine-readable payloads.
+
+    Regression tests for the `-o json` pollution bug: commands printed status
+    lines and tables to stdout, so `dg -o json projects | jq` received
+    "Fetching projects..." ahead of the JSON and failed to parse.
+    """
+
+    @pytest.mark.parametrize("fmt", sorted(MACHINE_FORMATS))
+    def test_status_console_writes_to_stderr_for_machine_formats(self, fmt):
+        with patch(
+            "deepctl_core.output._output_config",
+            {"format": fmt, "quiet": False, "agentic": False},
+        ):
+            assert get_status_console().file is sys.stderr
+
+    @pytest.mark.parametrize("fmt", ["default", "table"])
+    def test_status_console_writes_to_stdout_for_human_formats(self, fmt):
+        with patch(
+            "deepctl_core.output._output_config",
+            {"format": fmt, "quiet": False, "agentic": False},
+        ):
+            assert get_status_console().file is sys.stdout
+
+    def test_machine_formats_membership(self):
+        # `table` is a human rendering and must keep stdout
+        assert "table" not in MACHINE_FORMATS
+        assert "default" not in MACHINE_FORMATS
+        assert {"json", "yaml", "csv"} == set(MACHINE_FORMATS)
+
+    def test_payload_console_is_not_the_status_console(self):
+        """The payload must never be diverted along with status output."""
+        from deepctl_core.output import stdout_console
+
+        assert stdout_console is not get_status_console()
+        with patch(
+            "deepctl_core.output._output_config",
+            {"format": "json", "quiet": False, "agentic": False},
+        ):
+            # status steps aside, payload keeps stdout
+            assert get_status_console().file is sys.stderr
+            assert stdout_console.file is sys.stdout