diff --git a/src/ucode/agents/__init__.py b/src/ucode/agents/__init__.py index 242c43f57..3091a948f 100644 --- a/src/ucode/agents/__init__.py +++ b/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: diff --git a/src/ucode/cli.py b/src/ucode/cli.py index 6c377cd35..177805e03 100644 --- a/src/ucode/cli.py +++ b/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( diff --git a/src/ucode/databricks.py b/src/ucode/databricks.py index 970f51ecc..1c1f0f23e 100644 --- a/src/ucode/databricks.py +++ b/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 {}), ) diff --git a/src/ucode/ui/__init__.py b/src/ucode/ui/__init__.py index f98981d72..976c744a2 100644 --- a/src/ucode/ui/__init__.py +++ b/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 diff --git a/tests/README.md b/tests/README.md index f9e9a8130..84636732c 100644 --- a/tests/README.md +++ b/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 diff --git a/tests/integration/README.md b/tests/integration/README.md index 5f631a5eb..caf0116e0 100644 --- a/tests/integration/README.md +++ b/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. diff --git a/tests/integration/test_ug_claude_headless.py b/tests/integration/test_ug_claude_headless.py index ae94538e1..9ffba1e77 100644 --- a/tests/integration/test_ug_claude_headless.py +++ b/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() diff --git a/tests/integration/test_ug_codex_headless.py b/tests/integration/test_ug_codex_headless.py index 2ea6df191..b58f5dcd8 100644 --- a/tests/integration/test_ug_codex_headless.py +++ b/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() diff --git a/tests/test_agents_init.py b/tests/test_agents_init.py index ac94a8e0a..2e1736c45 100644 --- a/tests/test_agents_init.py +++ b/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): diff --git a/tests/test_cli.py b/tests/test_cli.py index 1c44bed56..06c6f9c9c 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -8,6 +8,7 @@ import os import re import subprocess +import sys import time import tomllib from importlib import metadata @@ -5539,26 +5540,244 @@ def test_still_logs_in_when_nothing_external_is_set(self, monkeypatch): assert self._run(monkeypatch) == ["https://ws.cloud.databricks.com"] -class TestStdioProtocolLaunch: - """`codex app-server` owns stdout, so ug's status output moves to stderr.""" +class TestChildStdoutLaunch: + """Claude print mode and Codex exec/app-server reserve stdout for the child.""" + + @pytest.fixture(autouse=True) + def preserve_runner_stdout(self, monkeypatch): + """Keep CliRunner's temporary stdout alive when ug rebinds sys.stdout.""" + original_isolation = runner.isolation + + @contextlib.contextmanager + def isolation(*args, **kwargs): + with original_isolation(*args, **kwargs) as streams: + with contextlib.redirect_stdout(sys.stdout): + yield streams + + monkeypatch.setattr(runner, "isolation", isolation) + + @pytest.mark.parametrize("separator", [[], ["--"]], ids=["direct", "separator"]) + @pytest.mark.parametrize( + ("tool", "tool_args", "display"), + [ + ("claude", ["-p", "say hello", "--output-format", "json"], "Claude Code"), + ( + "claude", + ["--output-format", "stream-json", "--verbose", "--print", "say hello"], + "Claude Code", + ), + ("codex", ["app-server", "--listen", "stdio://"], "Codex"), + ("codex", ["exec", "say hello"], "Codex"), + ("codex", ["exec", "--json", "say hello"], "Codex"), + ("codex", ["e", "say hello"], "Codex"), + ("codex", ["e", "--json", "say hello"], "Codex"), + ("codex", ["--model", "exec"], "Codex"), + ("codex", ["--cd", "app-server"], "Codex"), + ("codex", ["--profile", "e"], "Codex"), + ("codex", ["--no-alt-screen", "explain", "exec"], "Codex"), + ("codex", ["--image", "one.png", "two.png", "exec", "hello"], "Codex"), + ( + "codex", + ["-c", 'model_reasoning_effort="low"', "exec", "--json", "hello"], + "Codex", + ), + ("codex", ["--config", 'model="example"', "e", "hello"], "Codex"), + ("codex", ['--config=model="example"', "exec", "hello"], "Codex"), + ("codex", ['-cmodel="example"', "exec", "--json", "hello"], "Codex"), + ("codex", ["-m", "example", "exec", "--json", "hello"], "Codex"), + ("codex", ["--model", "exec", "e", "--json", "hello"], "Codex"), + ("codex", ["--model=example", "exec", "hello"], "Codex"), + ("codex", ["-C", "dir with spaces", "exec", "--json", "hello"], "Codex"), + ("codex", ["--cd", "exec", "e", "hello"], "Codex"), + ("codex", ["--cd=example", "app-server", "--listen", "stdio://"], "Codex"), + ( + "codex", + ["--strict-config", "--no-alt-screen", "--search", "exec", "--json", "hello"], + "Codex", + ), + ( + "codex", + ["-c", 'model="example"', "--config", "features.web_search=true", "exec", "hello"], + "Codex", + ), + ], + ) + def test_status_goes_to_stderr_and_arguments_are_preserved( + self, capfd, separator, tool, tool_args, display + ): + child_output = b'{"result":"child output"}\n' + if tool == "codex" and tool_args[0] in {"exec", "e"} and "--json" not in tool_args: + child_output = b"child output\n" + + def bootstrap_status(*_args, **_kwargs): + print("Checking agent dependencies") + cli_mod.print_warning("Agent setup warning") + + with ( + _launch_policy_patches(None) as calls, + patch("ucode.cli.ensure_bootstrap_dependencies", side_effect=bootstrap_status), + ): + calls["launch"].side_effect = lambda *_args, **_kwargs: os.write(1, child_output) + result = runner.invoke(app, [tool, *separator, *tool_args]) + + captured = capfd.readouterr() + assert captured.out == child_output.decode() + assert captured.err == "" + assert result.exit_code == 0, result.output + assert result.stdout == "" + stderr = _strip_ansi(result.stderr) + assert "Checking agent dependencies" in stderr + assert "Agent setup warning" in stderr + assert "No managed coding agent config found" in stderr + assert f"Starting {display}" in stderr + calls["launch"].assert_called_once() + assert calls["launch"].call_args.args[0] == tool + assert calls["launch"].call_args.args[2] == tool_args + + @pytest.mark.parametrize("separator", [[], ["--"]], ids=["direct", "separator"]) + @pytest.mark.parametrize( + ("tool", "tool_args"), + [ + ("claude", ["-p", "hello"]), + ("claude", ["--print", "hello"]), + ("codex", ["app-server"]), + ("codex", ["exec", "hello"]), + ("codex", ["exec", "--json", "hello"]), + ("codex", ["e", "hello"]), + ("codex", ["e", "--json", "hello"]), + ("codex", ["-c", 'model_reasoning_effort="low"', "exec", "--json", "hello"]), + ("codex", ["--no-alt-screen", "--model", "example", "e", "hello"]), + ("codex", ["--cd", "example", "app-server"]), + ], + ) + def test_early_launch_failure_keeps_stdout_clean(self, separator, tool, tool_args): + def fail_bootstrap(*_args, **_kwargs): + print("Checking agent dependencies") + raise RuntimeError("Agent setup failed") + + with ( + _launch_policy_patches(None) as calls, + patch("ucode.cli.ensure_bootstrap_dependencies", side_effect=fail_bootstrap), + ): + result = runner.invoke(app, [tool, *separator, *tool_args]) + + assert result.exit_code == 1 + assert result.stdout == "" + stderr = _strip_ansi(result.stderr) + assert "Checking agent dependencies" in stderr + assert "Agent setup failed" in stderr + calls["launch"].assert_not_called() + + @pytest.mark.parametrize( + ("tool", "tool_args", "display"), + [ + ("claude", [], "Claude Code"), + ("claude", ["explain --print and -p"], "Claude Code"), + ("claude", ["--", "-p"], "Claude Code"), + ("codex", [], "Codex"), + ("codex", ["explain exec and e"], "Codex"), + ("codex", ["--", "exec"], "Codex"), + ("codex", ["--model=exec"], "Codex"), + ("codex", ["-Cexec"], "Codex"), + ("codex", ["-c", 'model="exec"'], "Codex"), + ("codex", ["--model", "example", "--", "exec"], "Codex"), + ("codex", ['--config=model="example"', "--", "e"], "Codex"), + ("codex", ["--no-alt-screen", "--", "app-server"], "Codex"), + ], + ) + def test_other_launches_keep_status_on_stdout(self, tool, tool_args, display): + with _launch_policy_patches(None) as calls: + result = runner.invoke(app, [tool, "--", *tool_args]) + + assert result.exit_code == 0, result.output + assert f"Starting {display}" in _strip_ansi(result.stdout) + assert result.stderr == "" + calls["launch"].assert_called_once() + assert calls["launch"].call_args.args[2] == tool_args def test_app_server_subcommand_owns_stdout(self): assert cli_mod._child_owns_stdout("codex", ["app-server", "--listen", "stdio://"]) is True def test_other_codex_launches_keep_stdout(self): assert cli_mod._child_owns_stdout("codex", []) is False - assert cli_mod._child_owns_stdout("codex", ["exec", "--json", "hi"]) is False + assert cli_mod._child_owns_stdout("codex", ["--", "exec"]) is False + assert cli_mod._child_owns_stdout("codex", ["--model=exec"]) is False - def test_other_agents_never_own_stdout(self): + def test_other_agent_commands_keep_stdout(self): assert cli_mod._child_owns_stdout("claude", ["app-server"]) is False assert cli_mod._child_owns_stdout("gemini", []) is False - def test_redirect_rebinds_stdout_without_touching_the_descriptor(self): - import sys + @pytest.mark.parametrize( + "option", + [ + "--strict-config", + "--oss", + "--approve-for-me", + "--dangerously-bypass-approvals-and-sandbox", + "--dangerously-bypass-hook-trust", + "--worktree", + "--search", + "--no-alt-screen", + "--no-daemon", + ], + ) + def test_codex_boolean_options_before_exec(self, option): + assert cli_mod._child_owns_stdout("codex", [option, "exec", "hello"]) + + @pytest.mark.parametrize( + "option", + [ + "-c", + "--config", + "-m", + "--model", + "-C", + "--cd", + "-p", + "--profile", + "-s", + "--sandbox", + "-a", + "--ask-for-approval", + "--enable", + "--disable", + "--remote", + "--remote-auth-token-env", + "--local-provider", + "--add-dir", + ], + ) + def test_codex_token_membership_also_matches_option_values(self, option): + assert cli_mod._child_owns_stdout("codex", [option, "exec"]) + assert cli_mod._child_owns_stdout("codex", [f"{option}=exec"]) is False + assert cli_mod._child_owns_stdout("codex", [option, "exec", "e", "hello"]) + assert cli_mod._child_owns_stdout("codex", [option, "exec", "--", "app-server"]) + assert cli_mod._child_owns_stdout("codex", [option, "example", "--", "exec"]) is False + + @pytest.mark.parametrize( + ("tool_args", "expected"), + [ + (["--unknown", "exec"], True), + (["--image", "one.png", "two.png", "e"], True), + (["--model"], False), + (["--model", "--", "exec"], False), + (["--help", "exec"], True), + (["--version", "exec"], True), + ], + ) + def test_codex_token_membership_does_not_parse_options(self, tool_args, expected): + assert cli_mod._child_owns_stdout("codex", tool_args) is expected + def test_redirect_rebinds_stdout_without_touching_the_descriptor(self, capfd): real_stdout = sys.stdout try: cli_mod.redirect_output_to_stderr() assert sys.stdout is sys.stderr + cli_mod.print_note("Gateway status") + os.write(1, b'{"result":"child output"}\n') finally: sys.stdout = real_stdout + + captured = capfd.readouterr() + assert captured.out == '{"result":"child output"}\n' + assert "Gateway status" in captured.err