fix: keep headless Claude and Codex stdout clean

src/ucode/agents/__init__.py

@@ -38,6 +38,7 @@
     prompt_yes_no,
     prompt_yes_no_default,
     spinner,
+    status_subprocess_stdout,
 )
 
 from . import claude, codex, copilot, gemini, opencode, pi
@@ -139,7 +140,7 @@ def _update_installed_tool_binary(tool: str, version: str | None = None) -> bool
         # Detach potentially incompatible metadata until the next validated refresh.
         codex.detach_app_model_catalog()
     try:
-        subprocess_cross_os.run(command, check=True, timeout=300)
+        subprocess_cross_os.run(command, check=True, timeout=300, stdout=status_subprocess_stdout())
     except (OSError, subprocess.CalledProcessError, subprocess.TimeoutExpired):
         print_warning(f"Could not update {spec['display']}; continuing.")
         return False
@@ -231,7 +232,12 @@ def install_tool_binary(
     if tool == "codex":
         codex.detach_app_model_catalog()
     try:
-        subprocess_cross_os.run(["npm", "install", "-g", package], check=True, timeout=300)
+        subprocess_cross_os.run(
+            ["npm", "install", "-g", package],
+            check=True,
+            timeout=300,
+            stdout=status_subprocess_stdout(),
+        )
     except (OSError, subprocess.CalledProcessError, subprocess.TimeoutExpired) as exc:
         message = f"Failed to install {spec['display']} automatically."
         if strict:

src/ucode/cli.py

@@ -2515,13 +2515,16 @@ def _configure_managed_skills(managed: dict | None) -> None:
 
 
 def _child_owns_stdout(tool: str, tool_args: list[str]) -> bool:
-    """True when the forwarded agent command speaks a stdio protocol on stdout.
+    """Reserve stdout when forwarded arguments contain a headless-mode token.
 
-    ``codex app-server`` puts its JSON-RPC stream on stdout, so ug's status
-    output must move to stderr for that launch; the file descriptor stays
-    untouched for the agent process.
+    Only scan before ``--``. Matching option values also move ug diagnostics to
+    stderr; argument forwarding and the agent's stdout descriptor stay untouched.
     """
-    return tool == "codex" and tool_args[:1] == ["app-server"]
+    if "--" in tool_args:
+        tool_args = tool_args[: tool_args.index("--")]
+    if tool == "claude":
+        return any(arg in {"-p", "--print"} for arg in tool_args)
+    return tool == "codex" and any(arg in {"exec", "e", "app-server"} for arg in tool_args)
 
 
 def _should_launch_smart_routing(

src/ucode/databricks.py

@@ -47,6 +47,7 @@
     print_success,
     print_warning,
     spinner,
+    status_subprocess_stdout,
 )
 
 UNIX_DATABRICKS_INSTALL_URL = (
@@ -656,6 +657,7 @@ def run(
         text=text,
         env=env,
         timeout=timeout,
+        **({"stdout": status_subprocess_stdout()} if not capture_output else {}),
     )
 
 

src/ucode/ui/__init__.py

@@ -38,15 +38,20 @@
 def redirect_output_to_stderr() -> None:
     """Move status output to stderr because a child process now owns stdout.
 
-    ``codex app-server`` speaks its JSON-RPC protocol on stdout, so any ug line
-    printed there corrupts the stream. Rich resolves ``console.file`` to
-    ``sys.stdout`` at print time, so rebinding the module attribute sends every
+    Claude print mode and Codex exec/app-server can emit structured output on
+    stdout, so any ug line printed there corrupts the stream. Rich resolves
+    ``console.file`` to ``sys.stdout`` at print time, so rebinding the module attribute sends every
     status print to stderr while ``os.execvp`` still hands the launched agent
     the untouched stdout file descriptor.
     """
     sys.stdout = sys.stderr
 
 
+def status_subprocess_stdout() -> int | None:
+    """Route setup subprocess output to stderr when the agent owns stdout."""
+    return 2 if sys.stdout is sys.stderr else None
+
+
 # Past this many options the choice list is pinned to a fixed-height scrolling viewport (see
 # `_cap_choice_viewport`) rather than growing to fill the terminal, and the pickers append a
 # "↑/↓ scroll" note to their instruction line. The value is both the boundary and the number of

tests/README.md

@@ -42,6 +42,20 @@ These are component checks, not live Windows coverage for every agent.
 Agent configuration tests also verify `ug` auth/MCP helper commands, including
 quoted executable paths and replacement of legacy `ucode` routing/web-search helpers.
 
+`TestChildStdoutLaunch` in `test_cli.py` checks Claude `-p`/`--print` and Codex `exec`
+(plain and `--json`) and `app-server` status/error output on stderr, argument forwarding with
+and without ug's `--`, and an unchanged stdout descriptor. Detection checks exact tokens before
+the agent's `--`, including `e` and tokens after unknown or variadic options, without parsing
+Codex's options. An option value or prompt token equal to `exec`, `e`, or `app-server` also
+routes UG diagnostics to stderr; arguments and native agent stdout remain unchanged.
+Launches without a matching token retain stdout for status messages.
+`TestBootstrapStdout` in `test_agents_init.py` substitutes local Python processes
+for Databricks and agent installers/upgraders, asserting real stdout/stderr routing
+on success and failure, preserved version-output capture, and clean native handoff.
+These are component checks, not live installer, Isaac, or inference coverage.
+The live Claude/Codex headless prompt-argument and stdin journeys also require raw stdout to
+parse as JSON/JSONL without stripping UG messages; those assertions still require a live run.
+
 `test_mcp_web_search.py` and `test_agent_claude.py` cover custom OAuth search
 registration, stale registration repair, SDK cache reuse/refresh, CLI profile
 selection, and errors without browser consent through the MCP handler. These

tests/integration/README.md

@@ -18,6 +18,16 @@ The existing unit tests keep their fixtures. Integration has an independent
 pytest configuration and uses `--confcutdir` so those fixtures cannot leak in.
 It is not collected by the default `uv run pytest` command.
 
+`TestChildStdoutLaunch` in `../test_cli.py` covers clean Claude print-mode and
+Codex exec/app-server stdout, early launch errors, and forwarding through ug's `--`.
+Token membership before the agent's `--` intentionally also matches option values and prompt
+tokens; component tests verify this only moves UG diagnostics, preserving agent arguments/stdout.
+`TestBootstrapStdout` in `../test_agents_init.py` exercises real subprocess streams
+with substituted installer/upgrader commands, including failures and native handoff.
+These component checks do not establish live installer, Isaac, or inference coverage.
+The live Claude/Codex headless prompt-argument and stdin journeys require raw stdout to parse
+as JSON/JSONL, without discarding non-JSON lines before checking the completed file task.
+
 The `smart_defaults` wire schema, legacy `spend_tiers` cache reads, and recommendation
 request gating are covered by unit/component tests listed in `../README.md`. This suite
 does not yet assert live `recommendModel` request counts for configs with and without tiers.

tests/integration/test_ug_claude_headless.py

@@ -13,7 +13,7 @@ def test_ug_claude_headless_prompt_argument(live_session, workspace):
     """Scenario: configure claude and submit a headless prompt via argument.
 
     Expected: the real agent reads the fixture and returns its unknown value in
-    its structured completed answer, with exit code zero.
+    its structured completed answer, with exit code zero and JSON-only stdout.
     """
     session = live_session
     task = FileTask(session)
@@ -39,6 +39,7 @@ def test_ug_claude_headless_prompt_argument(live_session, workspace):
         "Read",
         timeout=180,
     )
+    assert json.loads(result.stdout)["type"] == "result"
     task.assert_headless_answer("claude", result)
     session.assert_not_routed()
 
@@ -47,7 +48,7 @@ def test_ug_claude_headless_prompt_stdin(live_session, workspace):
     """Scenario: configure claude and submit a headless prompt via stdin.
 
     Expected: the real agent reads the fixture and returns its unknown value in
-    its structured completed answer, with exit code zero.
+    its structured completed answer, with exit code zero and JSON-only stdout.
     """
     session = live_session
     task = FileTask(session)
@@ -73,6 +74,7 @@ def test_ug_claude_headless_prompt_stdin(live_session, workspace):
         timeout=180,
         input_text=task.prompt + "\n",
     )
+    assert json.loads(result.stdout)["type"] == "result"
     task.assert_headless_answer("claude", result)
     session.assert_not_routed()
 

tests/integration/test_ug_codex_headless.py

@@ -1,5 +1,7 @@
 """CUJs for using codex from scripts through installed ug."""
 
+import json
+
 import pytest
 from utils.constants import CODEX_TEST_MODEL
 from utils.evidence import FileTask
@@ -12,7 +14,7 @@ def test_ug_codex_headless_prompt_argument(live_session, workspace):
     """Scenario: configure codex and submit a headless prompt via argument.
 
     Expected: the real agent reads the fixture and returns its unknown value in
-    its structured completed answer, with exit code zero.
+    its structured completed answer, with exit code zero and JSONL-only stdout.
     """
     session = live_session
     task = FileTask(session)
@@ -38,6 +40,8 @@ def test_ug_codex_headless_prompt_argument(live_session, workspace):
         task.prompt,
         timeout=180,
     )
+    events = [json.loads(line) for line in result.stdout.splitlines()]
+    assert events and all(isinstance(event, dict) for event in events)
     task.assert_headless_answer("codex", result)
     session.assert_not_routed()
 
@@ -46,7 +50,7 @@ def test_ug_codex_headless_prompt_stdin(live_session, workspace):
     """Scenario: configure codex and submit a headless prompt via stdin.
 
     Expected: the real agent reads the fixture and returns its unknown value in
-    its structured completed answer, with exit code zero.
+    its structured completed answer, with exit code zero and JSONL-only stdout.
     """
     session = live_session
     task = FileTask(session)
@@ -73,6 +77,8 @@ def test_ug_codex_headless_prompt_stdin(live_session, workspace):
         timeout=180,
         input_text=task.prompt + "\n",
     )
+    events = [json.loads(line) for line in result.stdout.splitlines()]
+    assert events and all(isinstance(event, dict) for event in events)
     task.assert_headless_answer("codex", result)
     session.assert_not_routed()
 

tests/test_agents_init.py

@@ -3,12 +3,14 @@
 from __future__ import annotations
 
 import subprocess
-from contextlib import contextmanager
+import sys
+from contextlib import contextmanager, nullcontext, redirect_stdout
 from unittest.mock import MagicMock
 
 import pytest
 
 import ucode.agents as agents_mod
+import ucode.databricks as db_mod
 from ucode.agents import (
     DEFAULT_TOOL,
     TOOL_SPECS,
@@ -25,6 +27,7 @@
 )
 from ucode.agents.args import has_explicit_model_arg
 from ucode.managed_config import ManagedConfigResult
+from ucode.ui import redirect_output_to_stderr
 
 
 class TestModelArgumentParsing:
@@ -587,6 +590,104 @@ def _boom(*a, **k):  # pragma: no cover - must never run
         assert (model, error) == ("gemini-3.5-flash", None)
 
 
+class TestBootstrapStdout:
+    @pytest.mark.parametrize("tool", ["claude", "codex"])
+    @pytest.mark.parametrize("installed", [False, True], ids=["install", "upgrade"])
+    @pytest.mark.parametrize("child_owns_stdout", [False, True], ids=["interactive", "protocol"])
+    @pytest.mark.parametrize("failure", [None, "databricks", "agent"])
+    def test_setup_children_respect_stdout_ownership(
+        self, monkeypatch, capfd, tool, installed, child_owns_stdout, failure
+    ):
+        """Real subprocess streams exercise simulated installers and native handoff."""
+        available = {"databricks": installed, tool: installed}
+        updated = set()
+        commands = []
+
+        def which(binary):
+            return (
+                f"/tools/{binary}" if available.get(binary) or binary in {"npm", "brew"} else None
+            )
+
+        def run_dependency(command, **kwargs):
+            commands.append(command)
+            if command in (["databricks", "--version"], ["/tools/databricks", "--version"]):
+                version = "1.17.0" if "databricks" in updated else "0.1.0"
+                script = f"print('Databricks CLI v{version}')"
+            else:
+                dependency = "databricks" if command[0] == "brew" else "agent"
+                returncode = int(failure == dependency)
+                script = (
+                    f"import sys; print('{dependency} setup stdout'); "
+                    f"print('{dependency} setup stderr', file=sys.stderr); sys.exit({returncode})"
+                )
+                if returncode == 0:
+                    binary = "databricks" if dependency == "databricks" else tool
+                    available[binary] = True
+                    updated.add(binary)
+            kwargs["timeout"] = 10
+            return subprocess.run(
+                [sys.executable, "-c", script], stdin=subprocess.DEVNULL, **kwargs
+            )
+
+        monkeypatch.setattr(agents_mod.shutil, "which", which)
+        monkeypatch.setattr(
+            db_mod,
+            "_iter_databricks_executables",
+            lambda: ["/tools/databricks"] if available["databricks"] else [],
+        )
+        monkeypatch.setattr(db_mod.platform, "system", lambda: "Darwin")
+        monkeypatch.setattr(agents_mod, "_too_new_downgrade", lambda _tool: None)
+        monkeypatch.setattr(
+            agents_mod,
+            "_minimum_version_error",
+            lambda _tool: None if tool in updated else "must upgrade",
+        )
+        monkeypatch.setattr(agents_mod, "prompt_yes_no_default", lambda *_args, **_kwargs: True)
+        monkeypatch.setattr(agents_mod.subprocess_cross_os, "run", run_dependency)
+
+        with redirect_stdout(sys.stdout):
+            if child_owns_stdout:
+                redirect_output_to_stderr()
+            with pytest.raises(RuntimeError) if failure else nullcontext():
+                agents_mod.ensure_bootstrap_dependencies(tool)
+                subprocess.run(
+                    [sys.executable, "-c", 'print(\'{"result":"native output"}\')'],
+                    stdin=subprocess.DEVNULL,
+                    check=True,
+                    timeout=10,
+                )
+
+        captured = capfd.readouterr()
+        assert "databricks setup stderr" in captured.err
+        assert not {"Databricks CLI v0.1.0", "Databricks CLI v1.17.0"}.intersection(
+            (captured.out + captured.err).splitlines()
+        )
+        assert [
+            "brew",
+            "upgrade" if installed else "install",
+            "databricks/tap/databricks",
+        ] in commands
+        if failure != "databricks":
+            agent_command = (
+                agents_mod._NATIVE_UPGRADE_COMMANDS[tool]
+                if installed
+                else ["npm", "install", "-g", TOOL_SPECS[tool]["package"]]
+            )
+            assert agent_command in commands
+            assert "agent setup stderr" in captured.err
+        if child_owns_stdout:
+            assert captured.out == ("" if failure else '{"result":"native output"}\n')
+            assert "databricks setup stdout" in captured.err
+            if failure != "databricks":
+                assert "agent setup stdout" in captured.err
+        else:
+            assert "databricks setup stdout" in captured.out
+            assert "setup stdout" not in captured.err
+            if failure != "databricks":
+                assert "agent setup stdout" in captured.out
+            assert ('{"result":"native output"}' in captured.out) is (failure is None)
+
+
 class TestInstallToolBinary:
     @staticmethod
     def _seed_codex_catalog_reference(catalog_ref: str | None = None):