fix: keep stdout parseable, correct advertised examples and the auto-JSON claim

README.md

@@ -316,8 +316,11 @@ dg keys --list -o csv
 dg usage --last-week -o yaml
 ```
 
-When running in a non-TTY environment (pipes, CI, or AI coding tools), the CLI
-automatically switches to structured JSON output with plain-text status messages.
+In CI and AI coding tools, the CLI detects the context and automatically
+switches to structured JSON output with plain-text status messages. A plain
+pipe on its own does not trigger this — `dg projects | jq` still gets the
+human-readable table, so pass `-o json` explicitly when you are piping from an
+interactive shell.
 
 ### Exit codes
 

packages/deepctl-cmd-debug/src/deepctl_cmd_debug/command.py

@@ -27,7 +27,7 @@ class DebugCommand(BaseGroupCommand):
         "dg debug audio -f recording.wav",
         "dg debug network",
         "dg debug browser",
-        "dg debug stream",
+        "dg debug probe",
     ]
     agent_help = (
         "Diagnostic utilities for troubleshooting Deepgram integrations. "

packages/deepctl-cmd-login/src/deepctl_cmd_login/command.py

@@ -538,7 +538,7 @@ class ProfilesCommand(BaseCommand):
 
     examples = [
         "dg profiles --list",
-        "dg profiles --show default",
+        "dg profiles --current",
         "dg profiles --switch staging",
     ]
     agent_help = (

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

@@ -37,8 +37,8 @@ class UsageCommand(BaseCommand):
 
     examples = [
         "dg usage",
-        "dg usage --days 30",
-        "dg usage --start 2025-01-01 --end 2025-01-31",
+        "dg usage --last-month",
+        "dg usage --start-date 2025-01-01 --end-date 2025-01-31",
     ]
     agent_help = (
         "View Deepgram API usage statistics for the current project. "

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

@@ -9,14 +9,17 @@
 import httpx
 import keyring
 from pydantic import BaseModel
-from rich.console import Console
 from rich.progress import Progress, SpinnerColumn, TextColumn
 
 from .client import _split_base_url
 from .config import Config
 from .models import ProfileInfo, ProfilesResult
+from .output import stderr_console
 
-console = Console()
+# Diagnostics only: everything printed here is status or error chrome, so
+# it goes to stderr and can never corrupt the machine-readable payload a
+# command writes to stdout. See get_status_console() in output.py.
+console = stderr_console
 
 # Auth provider base URL (dx-id OIDC provider)
 AUTH_BASE_URL = os.getenv("DEEPGRAM_CLI_BASE_URL", "https://id.dx.deepgram.com")

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

@@ -4,15 +4,20 @@
 from typing import Any, ClassVar
 
 import click
-from rich.console import Console
 
 from .auth import AuthManager
 from .client import DeepgramClient
 from .config import Config
+from .models import ErrorResult
 from .output import _agentic, print_error, print_info, print_warning, stderr_console
+from .output import console as stdout_console
 from .timing import TimingContext
 
-console = Console()
+# The PAYLOAD console -- stdout, deliberately. Tables, JSON and raw text
+# written by _output_* are the machine-readable result and belong there.
+# Shared instance rather than a bare Console() so it honours the agentic
+# no-color/highlight settings; diagnostics use stderr_console instead.
+console = stdout_console
 
 
 class BaseCommand(ABC):
@@ -105,10 +110,22 @@ def execute(self, ctx: click.Context, **kwargs: Any) -> None:
                                 else:
                                     print_warning("No project ID specified")
 
-                    except Exception:
-                        # guard() already printed helpful error messages;
-                        # exit without duplicating them.
-                        raise SystemExit(1)
+                    except Exception as auth_error:
+                        # guard() already wrote the human-readable diagnosis to
+                        # stderr, so don't duplicate it -- but stdout must still
+                        # carry a parseable payload in a machine-readable
+                        # format. A CI step doing `dg -o json ... > out.json`
+                        # and parsing the result should get a structured error,
+                        # not an empty file. output_result is a no-op in
+                        # default mode, so this adds nothing for humans.
+                        self._tag_telemetry_status("error")
+                        try:
+                            self.output_result(
+                                ErrorResult(error=str(auth_error)), config
+                            )
+                        except (BrokenPipeError, OSError):
+                            pass
+                        raise SystemExit(1) from auth_error
 
             # Check project ID if required
             if self.requires_project:

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

@@ -3,14 +3,17 @@
 from typing import Any
 
 import click
-from rich.console import Console
 
 from .auth import AuthManager
 from .base_command import BaseCommand
 from .client import DeepgramClient
 from .config import Config
+from .output import stderr_console
 
-console = Console()
+# Diagnostics only: everything printed here is status or error chrome, so
+# it goes to stderr and can never corrupt the machine-readable payload a
+# command writes to stdout. See get_status_console() in output.py.
+console = stderr_console
 
 
 class BaseGroupCommand(BaseCommand):

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

@@ -9,7 +9,8 @@
 from deepgram import DeepgramClient as DGClient
 from deepgram import DeepgramClientEnvironment
 from deepgram.core.api_error import ApiError
-from rich.console import Console
+
+from .output import stderr_console
 
 if TYPE_CHECKING:
     from collections.abc import Iterator
@@ -18,7 +19,10 @@
     from .auth import AuthManager
     from .config import Config
 
-console = Console()
+# Diagnostics only: everything printed here is status or error chrome, so
+# it goes to stderr and can never corrupt the machine-readable payload a
+# command writes to stdout. See get_status_console() in output.py.
+console = stderr_console
 
 
 def _split_base_url(base_url: str) -> tuple[str, str, str]:

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

@@ -7,9 +7,12 @@
 from threading import local
 from typing import Any
 
-from rich.console import Console
+from .output import stderr_console
 
-console = Console()
+# Diagnostics only: everything printed here is status or error chrome, so
+# it goes to stderr and can never corrupt the machine-readable payload a
+# command writes to stdout. See get_status_console() in output.py.
+console = stderr_console
 
 # Thread-local storage for timing data
 _timing_data = local()

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

@@ -0,0 +1,152 @@
+"""Which stream each console writes to, pinned.
+
+Regression cover for #104: `deepctl_core` modules declared bare `Console()`
+instances bound to stdout and printed diagnostics through them, so
+`dg -o json <cmd>` on an auth failure wrote English prose to stdout and left
+stderr empty -- unparseable for the CI step that redirects stdout and parses
+it. `get_status_console()` warns about exactly this in its own docstring; the
+tests below turn that warning into a gate.
+
+The distinction these tests protect is *what the console carries*, not the
+module it lives in:
+
+- diagnostics (errors, status, progress, timing chrome) -> stderr, always
+- the payload (JSON/YAML/table/CSV a command produces) -> stdout, always
+"""
+
+from __future__ import annotations
+
+import ast
+import json
+from pathlib import Path
+
+import pytest
+from deepctl_core import output
+
+CORE_SRC = Path(output.__file__).parent
+
+# Modules whose module-level `console` carries diagnostics only.
+DIAGNOSTIC_MODULES = [
+    "auth",
+    "client",
+    "timing",
+    "base_group_command",
+    "plugin_manager",
+]
+
+
+class TestConsoleBindings:
+    """Every module-level console must be one of the two shared instances."""
+
+    @pytest.mark.parametrize("module_name", DIAGNOSTIC_MODULES)
+    def test_diagnostic_console_is_the_shared_stderr_console(
+        self, module_name: str
+    ) -> None:
+        import importlib
+
+        module = importlib.import_module(f"deepctl_core.{module_name}")
+
+        assert module.console is output.stderr_console
+        assert module.console.stderr is True
+
+    def test_base_command_console_is_the_shared_stdout_console(self) -> None:
+        """The payload console stays on stdout -- deliberately.
+
+        Tables, JSON and raw text written by `_output_*` are the
+        machine-readable result and belong on stdout. The requirement is that
+        it is the *shared* instance, so it honours the agentic no-color and
+        highlight settings a bare Console() would silently miss.
+        """
+        from deepctl_core import base_command
+
+        assert base_command.console is output.console
+        assert base_command.console.stderr is False
+
+
+class TestNoBareConsoleInCore:
+    """A bare `Console()` in deepctl_core is how #104 happened."""
+
+    def test_no_module_declares_a_bare_console(self) -> None:
+        offenders: list[str] = []
+
+        for path in sorted(CORE_SRC.glob("*.py")):
+            # encoding is explicit because read_text() otherwise uses the
+            # locale codec -- cp1252 on Windows, which cannot decode the
+            # emoji in timing.py and fails the whole matrix.
+            source = path.read_text(encoding="utf-8")
+            tree = ast.parse(source, filename=str(path))
+            for node in ast.walk(tree):
+                if not isinstance(node, ast.Call):
+                    continue
+                func = node.func
+                name = getattr(func, "id", None) or getattr(func, "attr", None)
+                if name != "Console":
+                    continue
+                # output.py is where the two shared instances are built.
+                if path.name == "output.py":
+                    continue
+                offenders.append(f"{path.name}:{node.lineno}")
+
+        assert not offenders, (
+            "bare Console() in deepctl_core reintroduces the #104 stdout "
+            "pollution -- import `console` or `stderr_console` from .output "
+            f"instead. Found at: {', '.join(offenders)}"
+        )
+
+
+class TestAuthFailurePayload:
+    """The #104 reproduction, as a test."""
+
+    def _run_guard_failure(self, capsys, output_format: str):
+        """Drive a requires_auth command whose guard() raises."""
+        from unittest.mock import MagicMock, patch
+
+        import click
+        from deepctl_core.auth import AuthenticationError
+        from deepctl_core.base_command import BaseCommand
+        from deepctl_core.config import Config
+
+        class NeedsAuth(BaseCommand):
+            name = "needs-auth"
+            help = "test command"
+            requires_auth = True
+
+            def handle(self, config, auth_manager, client, **kwargs):  # type: ignore[no-untyped-def]
+                raise AssertionError("handle must not run when guard() fails")
+
+        command = NeedsAuth()
+        ctx = MagicMock(spec=click.Context)
+        ctx.obj = {"config": Config()}
+
+        auth_manager = MagicMock()
+        auth_manager.guard.side_effect = AuthenticationError(
+            "Invalid API key - authentication failed"
+        )
+
+        with (
+            patch(
+                "deepctl_core.base_command.AuthManager", return_value=auth_manager
+            ),
+            patch("deepctl_core.base_command.DeepgramClient"),
+            patch(
+                "deepctl_core.output.get_output_format", return_value=output_format
+            ),
+        ):
+            with pytest.raises(SystemExit) as exc:
+                command.execute(ctx)
+
+        assert exc.value.code == 1
+        return capsys.readouterr()
+
+    def test_json_failure_writes_parseable_payload_to_stdout(self, capsys) -> None:
+        captured = self._run_guard_failure(capsys, "json")
+
+        payload = json.loads(captured.out)
+        assert payload["status"] == "error"
+        assert "Invalid API key" in payload["error"]
+
+    def test_default_mode_writes_nothing_to_stdout(self, capsys) -> None:
+        """Human mode must not gain a duplicate of the stderr diagnosis."""
+        captured = self._run_guard_failure(capsys, "default")
+
+        assert captured.out == ""