github-featured/liaohch3--claude-tap
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

fix: preserve wrapped client stdout (#431)

Browse Source 
* fix: preserve machine-readable client stdout

* fix: detect documented Codex JSON modes

* fix: keep prompt exports on stdout

* fix: preserve Codex stdio server output

* fix: detect Codex protocol output modes

* fix: route operational output to stderr

* fix: apply output policy to async entry point

* fix: avoid process-wide stdout redirection

* docs: remove unrelated dashboard evidence

---------

Co-authored-by: monitor-bot <monitor@example.invalid>
This commit is contained in:
Pan Ding
2026-08-18 00:37:14 +08:00
committed by GitHub
parent 901b856ffd
commit 8d2a680560
13 changed files with 250 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+3
View File
@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
### Fixed
- Preserve wrapped-client stdout for every mode by always routing claude-tap operational output to stderr.
claude_tap/cli.py
+71 -58
View File
@@ -17,6 +17,7 @@ import sys
import threading
import time
import webbrowser
from contextvars import ContextVar
from pathlib import Path
from urllib.parse import urlparse
@@ -50,6 +51,7 @@ from claude_tap.cli_clients import (
_toml_dotted_key_segment,
run_client,
)
from claude_tap.cli_output import print_status as _print
from claude_tap.cli_update import (
_build_update_command,
_detect_installer,
@@ -74,6 +76,8 @@ from claude_tap.trace import TraceWriter, create_trace_writer
from claude_tap.trace_log_handler import SQLiteLogHandler
from claude_tap.trace_store import TraceStore, get_trace_store, resolve_db_path
_COMMAND_STDOUT = ContextVar("command_stdout", default=None)
# Force UTF-8 + line-buffered stdout/stderr so emoji output works on Windows
# consoles (GBK/cp936) and `uv tool` doesn't fully buffer our progress prints.
if hasattr(sys.stdout, "reconfigure"):
@@ -263,29 +267,29 @@ def _extract_wrapped_client_command(client: str, args: list[str]) -> tuple[str |
def _trust_ca_for_current_user(ca_cert_path: Path) -> int:
"""Trust the forward-proxy CA in the current user's macOS login keychain."""
if sys.platform != "darwin":
print("--tap-trust-ca is currently only supported on macOS.", file=sys.stderr)
print(f"CA certificate: {ca_cert_path}", file=sys.stderr)
_print("--tap-trust-ca is currently only supported on macOS.", file=sys.stderr)
_print(f"CA certificate: {ca_cert_path}", file=sys.stderr)
return 1
if is_macos_ca_trusted(ca_cert_path):
print(f"🔐 CA already trusted in the macOS login keychain: {ca_cert_path}")
_print(f"🔐 CA already trusted in the macOS login keychain: {ca_cert_path}")
return 0
result = trust_macos_ca(ca_cert_path)
if result.returncode != 0:
details = (result.stderr or result.stdout or "").strip()
print("Error: failed to trust claude-tap CA in the macOS login keychain.", file=sys.stderr)
_print("Error: failed to trust claude-tap CA in the macOS login keychain.", file=sys.stderr)
if details:
print(details, file=sys.stderr)
print("This command does not use sudo; macOS may require unlocking your login keychain.", file=sys.stderr)
_print(details, file=sys.stderr)
_print("This command does not use sudo; macOS may require unlocking your login keychain.", file=sys.stderr)
return result.returncode or 1
if not is_macos_ca_trusted(ca_cert_path):
print("Error: macOS did not report the claude-tap CA as trusted after installation.", file=sys.stderr)
print(f"CA certificate: {ca_cert_path}", file=sys.stderr)
_print("Error: macOS did not report the claude-tap CA as trusted after installation.", file=sys.stderr)
_print(f"CA certificate: {ca_cert_path}", file=sys.stderr)
return 1
print(f"🔐 Trusted claude-tap CA in the current user's macOS login keychain: {ca_cert_path}")
_print(f"🔐 Trusted claude-tap CA in the current user's macOS login keychain: {ca_cert_path}")
return 0
@@ -304,19 +308,19 @@ def _ensure_ca_trust_for_forward_proxy(args: argparse.Namespace, ca_cert_path: P
if is_macos_ca_trusted(ca_cert_path):
return 0
print(f"🔐 {cfg.label} needs the claude-tap CA trusted in your macOS login keychain.")
print(" Installing for the current user only; no sudo or System keychain write is used.")
_print(f"🔐 {cfg.label} needs the claude-tap CA trusted in your macOS login keychain.")
_print(" Installing for the current user only; no sudo or System keychain write is used.")
return _trust_ca_for_current_user(ca_cert_path)
async def async_main(args: argparse.Namespace):
async def _async_main(args: argparse.Namespace) -> int:
output_dir = Path(args.output_dir)
output_dir.mkdir(parents=True, exist_ok=True)
if not args.live_viewer:
try:
migrate_legacy_traces(output_dir)
except sqlite3.Error as exc:
print(
_print(
f"claude-tap: legacy trace migration skipped because storage is unavailable ({exc})",
file=sys.stderr,
)
@@ -378,11 +382,11 @@ async def async_main(args: argparse.Namespace):
open_browser_fn=_open_browser,
)
if spawned:
print(f"🌐 Dashboard: {dashboard_url_value}")
_print(f"🌐 Dashboard: {dashboard_url_value}")
else:
print(f"🌐 Dashboard: {dashboard_url_value} (shared)")
_print(f"🌐 Dashboard: {dashboard_url_value} (shared)")
except (RuntimeError, sqlite3.Error) as exc:
print(f"⚠️ {exc}", file=sys.stderr)
_print(f"⚠️ {exc}", file=sys.stderr)
# Proxy logs go to SQLite, not terminal (avoids polluting Claude TUI)
sqlite_handler: SQLiteLogHandler | None = None
@@ -410,12 +414,12 @@ async def async_main(args: argparse.Namespace):
exit_code = 0
capture_only = bool(getattr(args, "export_prompt", None))
if capture_only and not transcript_only:
print("📝 Prompt export mode: upstream calls are skipped after capture.")
_print("📝 Prompt export mode: upstream calls are skipped after capture.")
try:
if transcript_only:
print(f"🔍 claude-tap v{__version__} watching Cursor agent-transcripts")
print(" Mode: one tap session per Cursor conversation JSONL")
print(f"🗄️ Trace database: {resolve_db_path()}")
_print(f"🔍 claude-tap v{__version__} watching Cursor agent-transcripts")
_print(" Mode: one tap session per Cursor conversation JSONL")
_print(f"🗄️ Trace database: {resolve_db_path()}")
cursor_watcher = CursorTranscriptWatcher(
since=watch_since,
model=model_from_cursor_args(args.claude_args),
@@ -443,7 +447,7 @@ async def async_main(args: argparse.Namespace):
except asyncio.CancelledError:
pass
else:
print("\n--tap-no-launch: watching local Cursor transcripts only. Press Ctrl+C to stop.")
_print("\n--tap-no-launch: watching local Cursor transcripts only. Press Ctrl+C to stop.")
try:
while True:
await asyncio.sleep(3600)
@@ -486,8 +490,8 @@ async def async_main(args: argparse.Namespace):
capture_only=capture_only,
)
actual_port = await forward_server.start()
print(f"🔍 claude-tap v{__version__} forward proxy on http://{args.host}:{actual_port}")
print(f" CA cert: {ca_cert_path}")
_print(f"🔍 claude-tap v{__version__} forward proxy on http://{args.host}:{actual_port}")
_print(f" CA cert: {ca_cert_path}")
else:
assert session is not None
assert writer is not None
@@ -518,10 +522,10 @@ async def async_main(args: argparse.Namespace):
actual_port = site._server.sockets[0].getsockname()[1]
except (AttributeError, IndexError, OSError):
actual_port = args.port
print(f"🔍 claude-tap v{__version__} listening on http://{args.host}:{actual_port}")
_print(f"🔍 claude-tap v{__version__} listening on http://{args.host}:{actual_port}")
print(f"📁 Trace session: {session_id}")
print(f"🗄️ Trace database: {resolve_db_path()}")
_print(f"📁 Trace session: {session_id}")
_print(f"🗄️ Trace database: {resolve_db_path()}")
if not args.no_launch:
try:
@@ -539,7 +543,7 @@ async def async_main(args: argparse.Namespace):
except asyncio.CancelledError:
pass
else:
print("\n--no-launch mode: proxy running. Press Ctrl+C to stop.")
_print("\n--no-launch mode: proxy running. Press Ctrl+C to stop.")
try:
while True:
await asyncio.sleep(3600)
@@ -560,8 +564,8 @@ async def async_main(args: argparse.Namespace):
if turns == 0 and imported:
turns = int(imported)
if turns or cursor_session_ids:
print(f" Cursor transcript turns: {turns}")
print(f" Cursor conversations: {len(cursor_session_ids)}")
_print(f" Cursor transcript turns: {turns}")
_print(f" Cursor conversations: {len(cursor_session_ids)}")
if forward_server:
try:
await asyncio.wait_for(forward_server.stop(), timeout=10)
@@ -600,10 +604,10 @@ async def async_main(args: argparse.Namespace):
protected_session_ids=protected_ids or None,
)
except sqlite3.Error as exc:
print(f"\nclaude-tap: trace cleanup skipped because storage is unavailable ({exc})", file=sys.stderr)
_print(f"\nclaude-tap: trace cleanup skipped because storage is unavailable ({exc})", file=sys.stderr)
else:
if cleaned:
print(f"\n🧹 Cleaned up {cleaned} old trace session(s)")
_print(f"\n🧹 Cleaned up {cleaned} old trace session(s)")
# Print summary with cost estimation
if cursor_watcher is not None:
@@ -620,30 +624,30 @@ async def async_main(args: argparse.Namespace):
"models_used": {},
"has_error": False,
}
print("\n📊 Trace summary:")
print(f" API calls: {stats['api_calls']}")
_print("\n📊 Trace summary:")
_print(f" API calls: {stats['api_calls']}")
if stats.get("trace_storage_errors"):
print(f" Trace storage errors: {stats['trace_storage_errors']}")
print(f" Dropped trace records: {stats.get('dropped_trace_records', 0)}")
_print(f" Trace storage errors: {stats['trace_storage_errors']}")
_print(f" Dropped trace records: {stats.get('dropped_trace_records', 0)}")
# Token breakdown
total_tokens = stats["input_tokens"] + stats["output_tokens"]
if total_tokens > 0:
print(f" Tokens: {stats['input_tokens']:,} in / {stats['output_tokens']:,} out", end="")
_print(f" Tokens: {stats['input_tokens']:,} in / {stats['output_tokens']:,} out", end="")
if stats["cache_read_tokens"] > 0:
print(f" / {stats['cache_read_tokens']:,} cache_read", end="")
_print(f" / {stats['cache_read_tokens']:,} cache_read", end="")
if stats["cache_create_tokens"] > 0:
print(f" / {stats['cache_create_tokens']:,} cache_write", end="")
print()
_print(f" / {stats['cache_create_tokens']:,} cache_write", end="")
_print()
if cursor_session_ids:
print(f" Sessions: {len(cursor_session_ids)} (one per Cursor conversation)")
_print(f" Sessions: {len(cursor_session_ids)} (one per Cursor conversation)")
elif session_id is not None:
print(f" Session: {session_id}")
print(f" Database: {resolve_db_path()}")
_print(f" Session: {session_id}")
_print(f" Database: {resolve_db_path()}")
if dashboard_url_value:
print(f" Dashboard: {dashboard_url_value}")
print(f" Stop dashboard: {_dashboard_stop_command(dashboard_host, dashboard_port)}")
_print(f" Dashboard: {dashboard_url_value}")
_print(f" Stop dashboard: {_dashboard_stop_command(dashboard_host, dashboard_port)}")
if prompt_export_rc is not None:
if prompt_export_rc != 0:
@@ -658,20 +662,20 @@ def _export_prompt_from_session(store, session_id: str, output: str) -> int:
try:
text = render_prompt_markdown(snapshot_from_records(store.load_records(session_id)))
except ValueError as exc:
print(f"Error: {exc}", file=sys.stderr)
_print(f"Error: {exc}", file=sys.stderr)
return 1
if output == "-":
print(text, end="")
_print(text, end="", file=_COMMAND_STDOUT.get())
return 0
path = Path(output).expanduser()
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(text, encoding="utf-8")
print(f"📝 Prompt snapshot: {path}")
_print(f"📝 Prompt snapshot: {path}")
trace_path = _prompt_trace_path(path)
trace_path.write_text(store.export_jsonl(session_id), encoding="utf-8")
print(f"🧾 Raw trace: {trace_path}")
_print(f"🧾 Raw trace: {trace_path}")
return 0
@@ -1016,6 +1020,15 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
return args
async def async_main(args: argparse.Namespace) -> int:
"""Run claude-tap while keeping operational output off command stdout."""
command_stdout = _COMMAND_STDOUT.set(sys.stdout)
try:
return await _async_main(args)
finally:
_COMMAND_STDOUT.reset(command_stdout)
def parse_dashboard_args(argv: list[str] | None = None) -> argparse.Namespace:
"""Parse arguments for the standalone dashboard command."""
parser = argparse.ArgumentParser(
@@ -1065,12 +1078,12 @@ async def dashboard_main(args: argparse.Namespace) -> int:
port = resolve_dashboard_port(args.live_port)
if args.command in {"stop", "quit"}:
if not await is_dashboard_healthy(host, port, require_current_db=False):
print(f"claude-tap dashboard is not running on {dashboard_url(host, port)}")
_print(f"claude-tap dashboard is not running on {dashboard_url(host, port)}")
return 1
if not await stop_dashboard_service(host, port):
print(f"Unable to stop claude-tap dashboard on {dashboard_url(host, port)}")
_print(f"Unable to stop claude-tap dashboard on {dashboard_url(host, port)}")
return 1
print(f"Stopped claude-tap dashboard on {dashboard_url(host, port)}")
_print(f"Stopped claude-tap dashboard on {dashboard_url(host, port)}")
return 0
output_dir.mkdir(parents=True, exist_ok=True)
@@ -1078,8 +1091,8 @@ async def dashboard_main(args: argparse.Namespace) -> int:
if await _is_dashboard_reusable(host, port):
migrate_legacy_traces(output_dir)
url = dashboard_url(host, port)
print(f"🌐 claude-tap dashboard already running: {url}")
print(f"🗄️ Trace database: {resolve_db_path()}")
_print(f"🌐 claude-tap dashboard already running: {url}")
_print(f"🗄️ Trace database: {resolve_db_path()}")
if args.open_viewer:
_open_browser(url)
return 0
@@ -1099,16 +1112,16 @@ async def dashboard_main(args: argparse.Namespace) -> int:
if await _is_dashboard_reusable(host, port):
migrate_legacy_traces(output_dir)
url = dashboard_url(host, port)
print(f"🌐 claude-tap dashboard already running: {url}")
_print(f"🌐 claude-tap dashboard already running: {url}")
if args.open_viewer:
_open_browser(url)
return 0
raise
print(f"🌐 claude-tap dashboard: {server.url}")
print(f"🗄️ Trace database: {resolve_db_path()}")
_print(f"🌐 claude-tap dashboard: {server.url}")
_print(f"🗄️ Trace database: {resolve_db_path()}")
if output_dir.exists():
print(f"📁 Legacy import dir: {output_dir}")
print("Press Ctrl+C to stop.")
_print(f"📁 Legacy import dir: {output_dir}")
_print("Press Ctrl+C to stop.")
if args.open_viewer:
_open_browser(server.url)
claude_tap/cli_clients.py
+28 -26
View File
@@ -19,6 +19,8 @@ from dataclasses import dataclass
from pathlib import Path
from typing import Iterable, Sequence
from claude_tap.cli_output import print_status as _print
_BEDROCK_HOST_RE = re.compile(
r"(^|\.)("
r"(bedrock-runtime|bedrock-runtime-fips)"
@@ -260,17 +262,17 @@ async def _prepare_codex_app_forward_launch() -> CodexAppLaunchPlan:
profile_dir = _codex_app_isolated_profile_dir()
profile_dir.mkdir(parents=True, exist_ok=True)
if processes:
print("\n⚠️ Codex/ChatGPT App is already running.")
print(" Launching an isolated second instance with a dedicated profile so the")
print(" current window keeps working and the new one inherits HTTPS_PROXY/CA.")
_print("\n⚠️ Codex/ChatGPT App is already running.")
_print(" Launching an isolated second instance with a dedicated profile so the")
_print(" current window keeps working and the new one inherits HTTPS_PROXY/CA.")
for line in processes[:3]:
print(f" {line}")
_print(f" {line}")
if len(processes) > 3:
print(f" ... {len(processes) - 3} more process(es)")
_print(f" ... {len(processes) - 3} more process(es)")
else:
print("\nℹ️ Using isolated Codex/ChatGPT profile from CODEX_APP_USER_DATA_DIR.")
print(f" Isolated profile: {profile_dir}")
print(" You may need to sign in again inside the tapped window.")
_print("\nℹ️ Using isolated Codex/ChatGPT profile from CODEX_APP_USER_DATA_DIR.")
_print(f" Isolated profile: {profile_dir}")
_print(" You may need to sign in again inside the tapped window.")
return CodexAppLaunchPlan(proceed=True, user_data_dir=profile_dir)
@@ -653,16 +655,16 @@ async def run_client(
resolved_cmd = _resolve_client_executable(client, cfg, client_cmd)
if resolved_cmd is None:
if client_cmd:
print(f"\nError: '{client_cmd}' command not found.\nPlease check the wrapper-provided {cfg.label} path.\n")
_print(f"\nError: '{client_cmd}' command not found.\nPlease check the wrapper-provided {cfg.label} path.\n")
elif client == "codexapp":
print(
_print(
"\nError: Codex desktop app executable not found.\n"
"Install Codex.app or ChatGPT.app (bundle id com.openai.codex) in "
"/Applications, or set "
f"{_CODEX_APP_EXECUTABLE_ENV}=/path/to/App.app/Contents/MacOS/<Executable>.\n"
)
else:
print(cfg.missing_help)
_print(cfg.missing_help)
return 1
resolved_cmd = _prefer_windows_command_shim(resolved_cmd)
if client == "codexapp" and proxy_mode == "forward" and not codex_app_preflighted:
@@ -685,7 +687,7 @@ async def run_client(
if inject_proxy and proxy_mode == "forward":
if client == "dsh" and not _node_supports_env_proxy(env):
print(
_print(
"\nError: DeepSeek Harness forward capture requires a Node runtime "
"with --use-env-proxy support.\n"
"Upgrade Node until `node --use-env-proxy --version` succeeds, or use "
@@ -799,22 +801,22 @@ async def run_client(
env.pop(key, None)
cmd = [resolved_cmd] + cmd_args
print(f"\n🚀 Starting {cfg.label}: {' '.join([display_cmd, *cmd_args])}")
_print(f"\n🚀 Starting {cfg.label}: {' '.join([display_cmd, *cmd_args])}")
if cfg.transcript_only:
print(" Mode: local agent-transcripts (no HTTPS_PROXY)")
_print(" Mode: local agent-transcripts (no HTTPS_PROXY)")
elif proxy_mode == "forward":
print(f" HTTPS_PROXY=http://127.0.0.1:{port}")
_print(f" HTTPS_PROXY=http://127.0.0.1:{port}")
for env_key in cfg.forward_base_url_envs:
print(f" {env_key}={cfg.reverse_base_url(port)}")
_print(f" {env_key}={cfg.reverse_base_url(port)}")
if ca_cert_path:
print(f" NODE_EXTRA_CA_CERTS={ca_cert_path}")
_print(f" NODE_EXTRA_CA_CERTS={ca_cert_path}")
elif client == "kimi-code":
print(f" KIMI_CODE_HOME={env.get('KIMI_CODE_HOME', '')}")
print(f" KIMI_CODE_BASE_URL={env.get('KIMI_CODE_BASE_URL', '')}")
_print(f" KIMI_CODE_HOME={env.get('KIMI_CODE_HOME', '')}")
_print(f" KIMI_CODE_BASE_URL={env.get('KIMI_CODE_BASE_URL', '')}")
else:
for env_key, base_url in cfg.reverse_base_url_env_map(port).items():
print(f" {env_key}={base_url}")
print()
_print(f" {env_key}={base_url}")
_print()
# Give TUI children their own process group and make them the foreground
# group so they have full terminal control (e.g. Cmd+Delete, Ctrl+U).
@@ -863,7 +865,7 @@ async def run_client(
if sigint_count == 1:
if proc.returncode is None:
proc.terminate()
print(f"\n⏳ Shutting down {cfg.label}... (Ctrl+C again to force)")
_print(f"\n⏳ Shutting down {cfg.label}... (Ctrl+C again to force)")
else:
if proc.returncode is None:
proc.kill()
@@ -871,7 +873,7 @@ async def run_client(
def _handle_sigtstp():
if proc.returncode is None:
proc.terminate()
print(f"\n⏳ Shutting down {cfg.label}...")
_print(f"\n⏳ Shutting down {cfg.label}...")
try:
loop.add_signal_handler(signal.SIGINT, _handle_sigint)
@@ -925,9 +927,9 @@ async def run_client(
pass
elapsed = loop.time() - started_at
print(f"\n📋 {cfg.label} exited with code {code}")
_print(f"\n📋 {cfg.label} exited with code {code}")
if client == "codexapp" and proxy_mode == "forward" and code == 0 and elapsed < _CODEX_APP_FAST_EXIT_HINT_SECONDS:
print(
_print(
" Codex App exited immediately. If macOS printed something like "
"'opening in an existing browser session', an already-running "
"Codex/ChatGPT App handled the launch and did not inherit "
@@ -971,7 +973,7 @@ def _maybe_rewrite_hermes_gateway_start(client: str, cmd_args: list[str]) -> lis
continue
break
if i + 1 < len(cmd_args) and cmd_args[i] == "gateway" and cmd_args[i + 1] == "start":
print(
_print(
"ℹ️ Rewriting `hermes gateway start` to `hermes gateway run` so the "
"gateway runs in the foreground under claude-tap. Recent hermes "
"versions delegate `gateway start` to systemd / launchd, which spawns "
claude_tap/cli_output.py
+13
View File
@@ -0,0 +1,13 @@
"""Output helpers shared by claude-tap command-line modules."""
from __future__ import annotations
import builtins
import sys
from typing import Any
def print_status(*values: object, **kwargs: Any) -> None:
"""Print operational output to stderr unless a stream is explicit."""
kwargs.setdefault("file", sys.stderr)
builtins.print(*values, **kwargs)
tests/e2e/test_real_proxy.py
+4 -4
View File
@@ -198,8 +198,8 @@ class TestRealProxy:
html_content = html_files[0].read_text()
assert "EMBEDDED_TRACE_COMPACT_DATA" in html_content, "HTML viewer should contain EMBEDDED_TRACE_COMPACT_DATA"
# Verify Dashboard: line in stdout
assert "Dashboard:" in result.stdout, "Expected 'Dashboard:' URL in stdout"
# Verify Dashboard: line in stderr
assert "Dashboard:" in result.stderr, "Expected 'Dashboard:' URL in stderr"
@pytest.mark.timeout(180)
def test_api_key_redaction(self, claude_env):
@@ -257,5 +257,5 @@ class TestRealProxy:
result = _run_claude_tap(env, trace_dir, "Reply with exactly: SUMMARY_CHECK", proxy_mode=proxy_mode)
assert result.returncode == 0
assert "Trace summary" in result.stdout, f"Expected 'Trace summary' in stdout:\n{result.stdout[:500]}"
assert "API calls:" in result.stdout, f"Expected 'API calls:' in stdout:\n{result.stdout[:500]}"
assert "Trace summary" in result.stderr, f"Expected 'Trace summary' in stderr:\n{result.stderr[:500]}"
assert "API calls:" in result.stderr, f"Expected 'API calls:' in stderr:\n{result.stderr[:500]}"
tests/test_client_config_framework.py
+2 -2
View File
@@ -239,7 +239,7 @@ async def test_run_client_reverse_sets_all_base_url_envs_and_settings(
)
assert cmd[3:] == ("--flag",)
out = capsys.readouterr().out
out = capsys.readouterr().err
assert out.count("PRIMARY_BASE_URL=http://127.0.0.1:43123/v1") == 1
assert out.count("SECONDARY_BASE_URL=http://127.0.0.1:43123/v1") == 1
@@ -559,6 +559,6 @@ async def test_run_client_agy_forward_sets_proxy_ca_and_cloud_code_url(
assert "AGY_BASE_URL" not in env
assert captured["cmd"] == ("/tmp/agy", "--print", "ok")
out = capsys.readouterr().out
out = capsys.readouterr().err
assert "HTTPS_PROXY=http://127.0.0.1:43123" in out
assert "CLOUD_CODE_URL=http://127.0.0.1:43123" in out
tests/test_codex_app_launch.py
+4 -4
View File
@@ -348,7 +348,7 @@ async def test_prepare_codex_app_forward_launch_uses_isolated_profile_when_alrea
assert plan.proceed is True
assert plan.user_data_dir == profile
assert profile.is_dir()
out = capsys.readouterr().out
out = capsys.readouterr().err
assert "already running" in out
assert "isolated second instance" in out
assert str(profile) in out
@@ -381,7 +381,7 @@ async def test_prepare_codex_app_forward_launch_forces_isolated_profile_from_env
assert plan.proceed is True
assert plan.user_data_dir == profile
assert profile.is_dir()
out = capsys.readouterr().out
out = capsys.readouterr().err
assert "CODEX_APP_USER_DATA_DIR" in out
assert str(profile) in out
@@ -430,7 +430,7 @@ async def test_run_client_codexapp_forward_launches_app_with_proxy_env(
assert captured["stdin"] == subprocess.DEVNULL
assert captured["stdout"] == subprocess.DEVNULL
assert captured["stderr"] == subprocess.DEVNULL
out = capsys.readouterr().out
out = capsys.readouterr().err
assert "Codex App exited immediately" in out
assert "already-running Codex/ChatGPT App" in out
@@ -469,7 +469,7 @@ async def test_run_client_codexapp_forward_launches_isolated_instance_when_app_r
"--proxy-server=http://127.0.0.1:43123",
)
assert profile.is_dir()
out = capsys.readouterr().out
out = capsys.readouterr().err
assert "isolated second instance" in out
tests/test_codex_launch.py
+88
View File
@@ -1,6 +1,7 @@
from __future__ import annotations
import asyncio
import os
from pathlib import Path
import pytest
@@ -59,6 +60,93 @@ def _custom_codex_http_args(provider: str, *tail: str) -> tuple[str, ...]:
)
@pytest.mark.asyncio
async def test_output_policy_keeps_client_stdout_and_routes_wrapper_status_to_stderr(monkeypatch, capfd) -> None:
from claude_tap import cli
async def fake_async_main(_args) -> int:
cli._print("wrapper status")
os.write(1, b"client output\n")
return 0
args = parse_args(["--tap-client", "codex", "exec", "hello"])
monkeypatch.setattr(cli, "_async_main", fake_async_main)
code = await cli.async_main(args)
captured = capfd.readouterr()
assert code == 0
assert captured.out == "client output\n"
assert captured.err == "wrapper status\n"
@pytest.mark.asyncio
async def test_output_policy_does_not_redirect_concurrent_host_output(monkeypatch, capsys) -> None:
from claude_tap import cli
started = asyncio.Event()
release = asyncio.Event()
async def fake_async_main(_args) -> int:
cli._print("wrapper status")
started.set()
await release.wait()
return 0
args = parse_args(["--tap-client", "codex", "exec", "hello"])
monkeypatch.setattr(cli, "_async_main", fake_async_main)
task = asyncio.create_task(cli.async_main(args))
await started.wait()
print("host output")
release.set()
assert await task == 0
captured = capsys.readouterr()
assert captured.out == "host output\n"
assert captured.err == "wrapper status\n"
@pytest.mark.asyncio
async def test_output_policy_keeps_prompt_export_payload_on_stdout(monkeypatch, capsys) -> None:
from claude_tap import cli
class FakeStore:
def load_records(self, _session_id):
return [
{
"request": {
"body": {
"model": "gpt-5",
"instructions": "system instructions",
}
}
}
]
async def fake_async_main(_args) -> int:
cli._print("wrapper status")
return cli._export_prompt_from_session(FakeStore(), "session", "-")
args = parse_args(
[
"--tap-client",
"codex",
"--tap-export-prompt",
"-",
]
)
monkeypatch.setattr(cli, "_async_main", fake_async_main)
code = await cli.async_main(args)
captured = capsys.readouterr()
assert code == 0
assert "# System Prompt\n\nsystem instructions" in captured.out
assert "wrapper status" not in captured.out
assert captured.err == "wrapper status\n"
@pytest.mark.asyncio
async def test_run_client_codex_reverse_forces_builtin_provider_to_http(monkeypatch) -> None:
captured: dict[str, object] = {}
tests/test_cursor_launch.py
+7 -4
View File
@@ -695,9 +695,10 @@ async def test_async_main_cursor_transcript_only_skips_proxy(monkeypatch, tmp_pa
assert code == 0
assert ca_calls == []
assert client_calls and client_calls[0].get("ca_cert_path") is None
out = capsys.readouterr().out
assert "watching Cursor agent-transcripts" in out
assert "Cursor transcript turns: 3" in out
captured = capsys.readouterr()
assert captured.out == ""
assert "watching Cursor agent-transcripts" in captured.err
assert "Cursor transcript turns: 3" in captured.err
@pytest.mark.asyncio
@@ -767,7 +768,9 @@ async def test_async_main_cursor_no_launch_watch_only(monkeypatch, tmp_path: Pat
assert FakeWatcher.last_since is not None
assert dashboard_started_at["t"] is not None
assert FakeWatcher.last_since <= dashboard_started_at["t"]
assert "watching local Cursor transcripts only" in capsys.readouterr().out
captured = capsys.readouterr()
assert captured.out == ""
assert "watching local Cursor transcripts only" in captured.err
@pytest.mark.asyncio
tests/test_dsh_launch.py
+1 -1
View File
@@ -93,7 +93,7 @@ async def test_run_client_dsh_forward_rejects_node_without_env_proxy_support(
code = await run_client(43123, [], client="dsh", proxy_mode="forward")
assert code == 1
output = capsys.readouterr().out
output = capsys.readouterr().err
assert "requires a Node runtime with --use-env-proxy support" in output
assert "--tap-proxy-mode reverse" in output
tests/test_e2e.py
+24 -24
View File
@@ -345,8 +345,8 @@ def _run_test(upstream_port, store_stream_events=False):
print(" ✅ Turn 2 (streaming, SSE reassembly without raw event storage): OK")
# ── Terminal output is clean ──
assert "Trace summary" in proc.stdout
assert "API calls: 2" in proc.stdout
assert "Trace summary" in proc.stderr
assert "API calls: 2" in proc.stderr
assert "[Turn" not in proc.stdout, "Proxy logs leaked to stdout!"
print(" ✅ Terminal output: clean")
@@ -354,7 +354,7 @@ def _run_test(upstream_port, store_stream_events=False):
assert "[Turn 1]" in log_content
assert "[Turn 2]" in log_content
print(" ✅ Proxy log: has Turn details")
assert "Session:" in proc.stdout or "Trace session:" in proc.stdout
assert "Session:" in proc.stderr or "Trace session:" in proc.stderr
print(" ✅ SQLite session persisted")
print("\n✅ E2E test PASSED")
@@ -621,8 +621,8 @@ for suffix, stream in [(":rawPredict", False), (":streamRawPredict", True)]:
)
assert proc.returncode == 0, f"vertex e2e failed: stdout={proc.stdout} stderr={proc.stderr}"
assert "ANTHROPIC_VERTEX_BASE_URL=http://127.0.0.1:" in proc.stdout
assert "ANTHROPIC_BASE_URL=http://127.0.0.1:" in proc.stdout
assert "ANTHROPIC_VERTEX_BASE_URL=http://127.0.0.1:" in proc.stderr
assert "ANTHROPIC_BASE_URL=http://127.0.0.1:" in proc.stderr
assert received_paths == [
"/v1/projects/test-project/locations/us-east5/publishers/anthropic/models/claude-opus-4-7:rawPredict",
"/v1/projects/test-project/locations/us-east5/publishers/anthropic/models/claude-opus-4-7:streamRawPredict",
@@ -830,8 +830,8 @@ def test_upstream_error():
print(" OK: 500 status recorded correctly in trace")
# The proxy should still produce summary output
assert "Trace summary" in proc.stdout
assert "API calls: 1" in proc.stdout
assert "Trace summary" in proc.stderr
assert "API calls: 1" in proc.stderr
print(" OK: proxy summary output present")
print("\n test_upstream_error PASSED")
@@ -980,7 +980,7 @@ def test_malformed_sse():
assert body["content"][0]["text"] == "partial"
print(" OK: reconstructed body has 'partial' text from valid events")
assert "Trace summary" in proc.stdout
assert "Trace summary" in proc.stderr
print(" OK: summary present")
print("\n test_malformed_sse PASSED")
@@ -1107,8 +1107,8 @@ def test_large_payload():
assert reported_len > 100_000, f"Upstream only received {reported_len} chars"
print(f" OK: upstream received full payload ({reported_len} chars)")
assert "Trace summary" in proc.stdout
assert "API calls: 1" in proc.stdout
assert "Trace summary" in proc.stderr
assert "API calls: 1" in proc.stderr
print(" OK: summary present")
payload_size = sum(len(json.dumps(record)) for record in records)
@@ -1272,8 +1272,8 @@ def test_concurrent_requests():
assert len(set(req_ids)) == 5, f"Expected 5 unique request IDs, got {len(set(req_ids))}"
print(" OK: all request IDs are unique")
assert "Trace summary" in proc.stdout
assert "API calls: 5" in proc.stdout
assert "Trace summary" in proc.stderr
assert "API calls: 5" in proc.stderr
print(" OK: summary present")
print("\n test_concurrent_requests PASSED")
@@ -1601,7 +1601,7 @@ async def test_async_main_live_viewer_default_opens_when_allowed(monkeypatch, tm
assert len(opened_urls) == 1
assert all(url.startswith("http://127.0.0.1:") for url in opened_urls)
assert migration_calls == []
output = capsys.readouterr().out
output = capsys.readouterr().err
assert "Stop dashboard: claude-tap dashboard stop" in output
@@ -1634,7 +1634,7 @@ async def test_async_main_stop_hint_includes_custom_dashboard_address(monkeypatc
code = await async_main(args)
assert code == 0
output = capsys.readouterr().out
output = capsys.readouterr().err
assert "Stop dashboard: claude-tap dashboard stop --tap-live-port 3000 --tap-host 0.0.0.0" in output
@@ -1943,7 +1943,7 @@ def test_codex_client_reverse_proxy():
assert record["request"]["path"] == "/v1/messages"
assert record["upstream_base_url"] == "http://127.0.0.1:19242"
assert record["request"]["body"]["model"] == "gpt-5-codex"
assert "OPENAI_BASE_URL=http://127.0.0.1:" in proc.stdout
assert "OPENAI_BASE_URL=http://127.0.0.1:" in proc.stderr
finally:
stop()
_cleanup(trace_dir, fake_bin_dir, "codex")
@@ -2082,7 +2082,7 @@ def test_grok_client_reverse_proxy():
assert records[1]["request"]["body"] == {"event": "repository_bundle_uploaded"}
assert records[2]["request"]["body"]["model"] == "grok-build"
assert records[2]["response"]["body"]["output"][0]["content"][0]["text"] == "HELLO_GROK"
assert "GROK_CLI_CHAT_PROXY_BASE_URL=http://127.0.0.1:" in proc.stdout
assert "GROK_CLI_CHAT_PROXY_BASE_URL=http://127.0.0.1:" in proc.stderr
finally:
stop()
_cleanup(trace_dir, fake_bin_dir, "grok")
@@ -2213,7 +2213,7 @@ def test_dsh_client_forward_proxy_captures_local_gateway():
assert record["response"]["body"]["content"][0]["type"] == "thinking"
assert record["response"]["body"]["content"][1]["text"] == "HELLO_DSH"
assert record["response"]["body"]["usage"]["input_tokens"] == 21
assert "forward proxy" in proc.stdout
assert "forward proxy" in proc.stderr
finally:
stop()
_cleanup(trace_dir, fake_bin_dir, "dsh")
@@ -2320,7 +2320,7 @@ def test_kimi_client_reverse_proxy():
assert record["response"]["body"]["content"][1]["text"] == "HELLO_KIMI"
assert record["response"]["body"]["usage"]["input_tokens"] == 13
assert record["response"]["body"]["usage"]["cache_read_input_tokens"] == 5
assert "KIMI_BASE_URL=http://127.0.0.1:" in proc.stdout
assert "KIMI_BASE_URL=http://127.0.0.1:" in proc.stderr
finally:
stop()
_cleanup(trace_dir, fake_bin_dir, "kimi")
@@ -2574,7 +2574,7 @@ def test_kimi_multiturn_tool_calls_reverse_proxy():
expected_tool_names = {"read_file", "search_code", "list_dir", "run_tests", "inspect_git", "parse_json"}
assert total_tool_calls == 10
assert unique_tool_names == expected_tool_names
assert "KIMI_BASE_URL=http://127.0.0.1:" in proc.stdout
assert "KIMI_BASE_URL=http://127.0.0.1:" in proc.stderr
finally:
stop()
_cleanup(trace_dir, fake_bin_dir, "kimi_multiturn")
@@ -2680,8 +2680,8 @@ def test_kimi_code_client_reverse_proxy():
assert record["request"]["path"] == "/chat/completions"
assert record["upstream_base_url"] == "http://127.0.0.1:19246"
assert record["response"]["body"]["content"][1]["text"] == "HELLO_KIMI_CODE"
assert "KIMI_CODE_HOME=" in proc.stdout
assert "KIMI_CODE_BASE_URL=http://127.0.0.1:" in proc.stdout
assert "KIMI_CODE_HOME=" in proc.stderr
assert "KIMI_CODE_BASE_URL=http://127.0.0.1:" in proc.stderr
finally:
stop()
_cleanup(trace_dir, fake_bin_dir, "kimi_code")
@@ -3062,7 +3062,7 @@ def test_upstream_unreachable():
print(f"[test_upstream_unreachable] stderr:\n{proc.stderr.rstrip()}")
# The proxy should still produce summary output
assert "Trace summary" in proc.stdout
assert "Trace summary" in proc.stderr
print(" OK: proxy did not crash")
# No trace records (502 returned in-process, not from upstream)
@@ -3213,7 +3213,7 @@ def test_startup_does_not_contact_pypi():
assert proc.returncode == 0, proc.stderr
assert "[fake-claude] noop exit" in proc.stdout
assert "Update available" not in proc.stdout
assert "Update available" not in proc.stderr
assert requests == []
print(" test_startup_does_not_contact_pypi PASSED")
except subprocess.TimeoutExpired as exc:
@@ -3348,7 +3348,7 @@ def test_e2e_with_cleanup():
print(f"[test_e2e_with_cleanup] stdout:\n{proc.stdout.rstrip()}")
assert proc.returncode == 0
assert "Cleaned up" in proc.stdout, f"Expected cleanup message in stdout:\n{proc.stdout}"
assert "Cleaned up" in proc.stderr, f"Expected cleanup message in stderr:\n{proc.stderr}"
reset_trace_store()
os.environ["CLOUDTAP_DB"] = str(db_path)
assert len(get_trace_store().list_session_rows()) == 3
tests/test_export.py
+1 -1
View File
@@ -458,7 +458,7 @@ def test_export_prompt_from_session_also_writes_raw_trace(trace_db, tmp_path, ca
assert trace_path.exists()
assert trace_path.read_text(encoding="utf-8") == store.export_jsonl(session_id)
assert json.loads(trace_path.read_text(encoding="utf-8"))["request_id"] == "req_1"
output = capsys.readouterr().out
output = capsys.readouterr().err
assert f"Prompt snapshot: {prompt_path}" in output
assert f"Raw trace: {trace_path}" in output
tests/test_trace_store_locking.py
+4 -4
View File
@@ -277,7 +277,7 @@ def test_real_proxy_continues_when_database_is_locked_at_startup(tmp_path: Path)
stop_upstream()
assert process.returncode == 0, process.stderr
assert "API calls: 1" in process.stdout
assert "API calls: 1" in process.stderr
assert "legacy trace migration skipped" in process.stderr
assert "continuing without blocking proxy" in process.stderr
@@ -290,9 +290,9 @@ def test_real_proxy_continues_when_database_locks_during_request(tmp_path: Path)
stop_upstream()
assert process.returncode == 0, process.stderr
assert "API calls: 1" in process.stdout
assert "Trace storage errors: 1" in process.stdout
assert "Dropped trace records: 1" in process.stdout
assert "API calls: 1" in process.stderr
assert "Trace storage errors: 1" in process.stderr
assert "Dropped trace records: 1" in process.stderr
assert "continuing without blocking proxy" in process.stderr
conn = sqlite3.connect(db_path)