From f89c5137082654ee939e09439f7efcc9e54fa3ca Mon Sep 17 00:00:00 2001 From: Oleg Miagkov Date: Sun, 12 Jul 2026 10:08:49 +0400 Subject: [PATCH 1/3] feat(qa): CLI E2E harness tool for terminal applications QA agents can E2E-test Electron/web apps (Electron MCP, Puppeteer) but had no way to drive terminal applications. New run_cli_session MCP tool: - Spawns the command in a pseudo-terminal on POSIX (prompts, colors, isatty() behave like a real terminal); plain pipes on Windows - Feeds scripted stdin lines, captures combined output (capped), enforces a timeout, reports found/missing expected substrings - Commands validated against the project security allowlist and always executed inside the project directory - Available to qa_reviewer and qa_fixer Co-Authored-By: Claude Fable 5 Signed-off-by: Oleg Miagkov --- apps/backend/agents/tools_pkg/models.py | 5 + apps/backend/agents/tools_pkg/registry.py | 2 + .../agents/tools_pkg/tools/__init__.py | 2 + .../agents/tools_pkg/tools/cli_harness.py | 139 +++++++++++++ apps/backend/core/cli_session.py | 190 ++++++++++++++++++ tests/test_cli_session.py | 133 ++++++++++++ 6 files changed, 471 insertions(+) create mode 100644 apps/backend/agents/tools_pkg/tools/cli_harness.py create mode 100644 apps/backend/core/cli_session.py create mode 100644 tests/test_cli_session.py diff --git a/apps/backend/agents/tools_pkg/models.py b/apps/backend/agents/tools_pkg/models.py index 665e0de7c..c95541749 100644 --- a/apps/backend/agents/tools_pkg/models.py +++ b/apps/backend/agents/tools_pkg/models.py @@ -48,6 +48,9 @@ TOOL_GET_TASK_OUTPUT = "mcp__auto-claude__get_task_output" TOOL_CANCEL_TASK = "mcp__auto-claude__cancel_task" +# CLI E2E harness (pseudo-terminal runner for terminal apps) +TOOL_RUN_CLI_SESSION = "mcp__auto-claude__run_cli_session" + # ============================================================================= # External MCP Tools # ============================================================================= @@ -282,6 +285,7 @@ def is_electron_mcp_enabled() -> bool: TOOL_UPDATE_QA_STATUS, TOOL_GET_SESSION_CONTEXT, TOOL_SEARCH_TEAM_DOCS, # Verify against team standards + TOOL_RUN_CLI_SESSION, # E2E-verify terminal apps ], "thinking_default": "high", }, @@ -294,6 +298,7 @@ def is_electron_mcp_enabled() -> bool: TOOL_GET_BUILD_PROGRESS, TOOL_UPDATE_QA_STATUS, TOOL_RECORD_GOTCHA, + TOOL_RUN_CLI_SESSION, # Verify CLI fixes end-to-end ], "thinking_default": "medium", }, diff --git a/apps/backend/agents/tools_pkg/registry.py b/apps/backend/agents/tools_pkg/registry.py index ff5c619aa..83298e1dd 100644 --- a/apps/backend/agents/tools_pkg/registry.py +++ b/apps/backend/agents/tools_pkg/registry.py @@ -17,6 +17,7 @@ from .tools import ( create_background_task_tools, + create_cli_harness_tools, create_debugging_tools, create_knowledge_base_tools, create_memory_tools, @@ -50,6 +51,7 @@ def create_all_tools(spec_dir: Path, project_dir: Path) -> list: all_tools.extend(create_qa_tools(spec_dir, project_dir)) all_tools.extend(create_statistics_tools(spec_dir, project_dir)) all_tools.extend(create_background_task_tools(spec_dir, project_dir)) + all_tools.extend(create_cli_harness_tools(spec_dir, project_dir)) all_tools.extend(create_debugging_tools(spec_dir, project_dir)) all_tools.extend(create_knowledge_base_tools(spec_dir, project_dir)) diff --git a/apps/backend/agents/tools_pkg/tools/__init__.py b/apps/backend/agents/tools_pkg/tools/__init__.py index fa123a2fa..adf6d9b92 100644 --- a/apps/backend/agents/tools_pkg/tools/__init__.py +++ b/apps/backend/agents/tools_pkg/tools/__init__.py @@ -6,6 +6,7 @@ """ from .background_task import create_background_task_tools +from .cli_harness import create_cli_harness_tools from .debugging import create_debugging_tools from .knowledge_base import create_knowledge_base_tools from .memory import create_memory_tools @@ -21,6 +22,7 @@ "create_qa_tools", "create_statistics_tools", "create_background_task_tools", + "create_cli_harness_tools", "create_debugging_tools", "create_knowledge_base_tools", ] diff --git a/apps/backend/agents/tools_pkg/tools/cli_harness.py b/apps/backend/agents/tools_pkg/tools/cli_harness.py new file mode 100644 index 000000000..223fbe0f9 --- /dev/null +++ b/apps/backend/agents/tools_pkg/tools/cli_harness.py @@ -0,0 +1,139 @@ +""" +CLI E2E Harness Tool +==================== + +MCP tool that lets QA agents verify command-line applications end-to-end: +spawn the app (in a pseudo-terminal on POSIX), feed scripted stdin lines, +capture output, and check expected substrings. + +Commands run through the same security allowlist as Bash commands and are +always executed inside the project directory. +""" + +import logging +from pathlib import Path +from typing import Any + +try: + from claude_agent_sdk import tool + + SDK_AVAILABLE = True +except ImportError: + SDK_AVAILABLE = False + +# How much output to echo back to the agent +OUTPUT_TAIL_CHARS = 4_000 + +MAX_TIMEOUT_SECONDS = 300 + + +def create_cli_harness_tools(spec_dir: Path, project_dir: Path) -> list: + """ + Create CLI harness tools bound to the given project directory. + + Args: + spec_dir: Path to the spec directory (unused, kept for registry symmetry) + project_dir: Path to the project root; commands run here + + Returns: + List of tool functions + """ + if not SDK_AVAILABLE: + return [] + + tools = [] + + @tool( + "run_cli_session", + "Run a command-line application end-to-end and verify its behavior. " + "Spawns the command inside a pseudo-terminal (so prompts, colors, and " + "isatty() checks behave like a real terminal), optionally sends " + "scripted stdin lines, captures combined output, and checks that " + "expected substrings appear. Use this to E2E-test CLI apps the same " + "way Electron MCP is used for desktop apps. The command must be " + "allowed by the project security profile and runs in the project " + "directory.", + { + "command": str, + "inputs": str, + "expect_substrings": str, + "timeout_seconds": int, + }, + ) + async def run_cli_session_tool(args: dict[str, Any]) -> dict[str, Any]: + """Run a CLI session and report the outcome.""" + command = (args.get("command") or "").strip() + if not command: + return _text_result("Error: command parameter is required") + + inputs_raw = args.get("inputs") or "" + inputs = [line for line in inputs_raw.splitlines() if line != ""] + + expect_raw = args.get("expect_substrings") or "" + expectations = [line for line in expect_raw.splitlines() if line.strip()] + + timeout_seconds = args.get("timeout_seconds") or 30 + timeout_seconds = max(1, min(int(timeout_seconds), MAX_TIMEOUT_SECONDS)) + + try: + from security import validate_command + + allowed, reason = validate_command(command, project_dir) + if not allowed: + return _text_result(f"Command blocked by security profile: {reason}") + + from core.cli_session import run_cli_session + + result = run_cli_session( + command=command, + cwd=str(project_dir), + inputs=inputs, + timeout_seconds=timeout_seconds, + ) + except ImportError as e: + return _text_result(f"Error: CLI harness not available: {e}") + except Exception as e: + logging.exception("Error during run_cli_session tool execution") + return _text_result(f"Error running CLI session: {e}") + + return _text_result(_format_report(result, expectations)) + + tools.append(run_cli_session_tool) + + return tools + + +def _format_report(result, expectations: list[str]) -> str: + """Build a human-readable verification report.""" + missing = [e for e in expectations if e not in result.output] + found = [e for e in expectations if e in result.output] + + lines = [] + if result.timed_out: + lines.append("TIMED OUT: process was killed after the timeout") + lines.append(f"Exit code: {result.exit_code}") + + if expectations: + status = "PASS" if not missing else "FAIL" + lines.append(f"Expectations: {status} ({len(found)}/{len(expectations)})") + for item in found: + lines.append(f" [found] {item}") + for item in missing: + lines.append(f" [MISSING] {item}") + + output_tail = result.output[-OUTPUT_TAIL_CHARS:] + if len(result.output) > OUTPUT_TAIL_CHARS: + lines.append(f"Output (last {OUTPUT_TAIL_CHARS} chars):") + else: + lines.append("Output:") + lines.append(output_tail if output_tail else "(no output)") + + return "\n".join(lines) + + +def _text_result(text: str) -> dict[str, Any]: + """Wrap text in the MCP tool result envelope.""" + return {"content": [{"type": "text", "text": text}]} + + +__all__ = ["create_cli_harness_tools"] diff --git a/apps/backend/core/cli_session.py b/apps/backend/core/cli_session.py new file mode 100644 index 000000000..975c89316 --- /dev/null +++ b/apps/backend/core/cli_session.py @@ -0,0 +1,190 @@ +""" +CLI Session Runner +================== + +Runs a command-line application end-to-end for QA verification: spawn the +process, optionally feed scripted stdin lines, capture combined output, and +enforce a timeout. + +On POSIX the process runs inside a pseudo-terminal so applications that +check isatty() (prompts, progress bars, colored output) behave like they +do for a real user. On Windows there is no stdlib pty; plain pipes are +used, which covers argument-driven CLIs and simple stdin protocols. +""" + +import os +import shlex +import subprocess +import time +from dataclasses import dataclass + +from core.platform import is_windows + +# Cap captured output so a chatty process cannot blow up the result +MAX_OUTPUT_CHARS = 100_000 + +# Delay between scripted stdin lines, giving the app time to prompt +INPUT_INTERVAL_SECONDS = 0.2 + + +@dataclass +class CliSessionResult: + """Outcome of a CLI session run.""" + + exit_code: int | None + output: str + timed_out: bool + + +def run_cli_session( + command: str, + cwd: str, + inputs: list[str] | None = None, + timeout_seconds: int = 30, +) -> CliSessionResult: + """ + Run a CLI application and capture its output. + + Args: + command: Command line to run (parsed with shlex, no shell) + cwd: Working directory for the process + inputs: Lines to send to stdin, in order (newline appended) + timeout_seconds: Kill the process after this many seconds + + Returns: + CliSessionResult with exit code, combined output, and timeout flag + """ + argv = shlex.split(command, posix=not is_windows()) + if not argv: + return CliSessionResult(exit_code=None, output="", timed_out=False) + + if is_windows(): + return _run_with_pipes(argv, cwd, inputs, timeout_seconds) + return _run_with_pty(argv, cwd, inputs, timeout_seconds) + + +def _run_with_pipes( + argv: list[str], + cwd: str, + inputs: list[str] | None, + timeout_seconds: int, +) -> CliSessionResult: + """Run with plain pipes (Windows fallback, no pty in stdlib).""" + stdin_data = "".join(line + "\n" for line in inputs) if inputs else None + proc = subprocess.Popen( + argv, + cwd=cwd, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + encoding="utf-8", + errors="replace", + ) + try: + output, _ = proc.communicate(input=stdin_data, timeout=timeout_seconds) + timed_out = False + except subprocess.TimeoutExpired: + proc.kill() + output, _ = proc.communicate() + timed_out = True + + return CliSessionResult( + exit_code=proc.returncode, + output=(output or "")[:MAX_OUTPUT_CHARS], + timed_out=timed_out, + ) + + +def _run_with_pty( + argv: list[str], + cwd: str, + inputs: list[str] | None, + timeout_seconds: int, +) -> CliSessionResult: + """Run inside a pseudo-terminal (POSIX).""" + import pty + import select + + master_fd, slave_fd = pty.openpty() + proc = subprocess.Popen( + argv, + cwd=cwd, + stdin=slave_fd, + stdout=slave_fd, + stderr=slave_fd, + close_fds=True, + start_new_session=True, + ) + os.close(slave_fd) + + pending = list(inputs or []) + chunks: list[bytes] = [] + total = 0 + deadline = time.monotonic() + timeout_seconds + next_write = time.monotonic() + INPUT_INTERVAL_SECONDS + timed_out = False + + try: + while True: + if time.monotonic() > deadline: + timed_out = True + proc.kill() + break + + eof = _drain_output(master_fd, chunks, total) + total = sum(len(c) for c in chunks) + if eof: + break + + if pending and time.monotonic() >= next_write: + os.write(master_fd, (pending.pop(0) + "\n").encode()) + next_write = time.monotonic() + INPUT_INTERVAL_SECONDS + + if proc.poll() is not None: + # Process exited: drain whatever is left, then stop + while not _drain_output(master_fd, chunks, total): + total = sum(len(c) for c in chunks) + break + + select.select([master_fd], [], [], 0.05) + finally: + os.close(master_fd) + if proc.poll() is None: + proc.kill() + proc.wait() + + output = b"".join(chunks).decode("utf-8", errors="replace") + return CliSessionResult( + exit_code=proc.returncode, + output=output[:MAX_OUTPUT_CHARS], + timed_out=timed_out, + ) + + +def _drain_output(master_fd: int, chunks: list[bytes], total: int) -> bool: + """ + Read available pty output without blocking. + + Returns: + True when the pty reached EOF (process side closed) + """ + import select + + while True: + readable, _, _ = select.select([master_fd], [], [], 0) + if not readable: + return False + try: + data = os.read(master_fd, 4096) + except OSError: + # EIO: slave side closed - treat as EOF + return True + if not data: + return True + if total < MAX_OUTPUT_CHARS: + chunks.append(data) + total += len(data) + + +__all__ = ["CliSessionResult", "run_cli_session", "MAX_OUTPUT_CHARS"] diff --git a/tests/test_cli_session.py b/tests/test_cli_session.py new file mode 100644 index 000000000..ae2662323 --- /dev/null +++ b/tests/test_cli_session.py @@ -0,0 +1,133 @@ +#!/usr/bin/env python3 +""" +Tests for the CLI session runner and harness tool registration. + +Tests cover: +- Output capture and exit codes +- Scripted stdin input +- Timeout enforcement +- Output size capping +- Tool registration for QA agents +""" + +# Add auto-claude to path for imports +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent.parent / "apps" / "backend")) + +from core.cli_session import MAX_OUTPUT_CHARS, run_cli_session + +PYTHON = sys.executable + + +class TestRunCliSession: + """Tests for run_cli_session.""" + + def test_captures_output_and_exit_code(self, tmp_path): + """Captures stdout and reports a zero exit code.""" + result = run_cli_session( + f"{PYTHON} -c \"print('hello from cli')\"", + cwd=str(tmp_path), + ) + + assert result.exit_code == 0 + assert result.timed_out is False + assert "hello from cli" in result.output + + def test_nonzero_exit_code(self, tmp_path): + """Reports non-zero exit codes.""" + result = run_cli_session( + f'{PYTHON} -c "import sys; sys.exit(3)"', + cwd=str(tmp_path), + ) + + assert result.exit_code == 3 + + def test_captures_stderr(self, tmp_path): + """stderr is captured in combined output.""" + result = run_cli_session( + f"{PYTHON} -c \"import sys; print('oops', file=sys.stderr)\"", + cwd=str(tmp_path), + ) + + assert "oops" in result.output + + def test_feeds_stdin_inputs(self, tmp_path): + """Scripted stdin lines reach the application.""" + script = tmp_path / "prompt.py" + script.write_text( + "name = input('name? ')\nprint('hi ' + name)\n", + encoding="utf-8", + ) + + result = run_cli_session( + f"{PYTHON} {script}", + cwd=str(tmp_path), + inputs=["world"], + timeout_seconds=15, + ) + + assert result.exit_code == 0 + assert "hi world" in result.output + + def test_timeout_kills_process(self, tmp_path): + """A hung process is killed and flagged.""" + result = run_cli_session( + f'{PYTHON} -c "import time; time.sleep(60)"', + cwd=str(tmp_path), + timeout_seconds=2, + ) + + assert result.timed_out is True + + def test_output_is_capped(self, tmp_path): + """Runaway output is truncated to the cap.""" + result = run_cli_session( + f"{PYTHON} -c \"print('x' * 300000)\"", + cwd=str(tmp_path), + timeout_seconds=30, + ) + + assert len(result.output) <= MAX_OUTPUT_CHARS + + def test_empty_command_is_noop(self, tmp_path): + """Empty command returns an empty result instead of raising.""" + result = run_cli_session("", cwd=str(tmp_path)) + + assert result.exit_code is None + assert result.output == "" + + def test_runs_in_given_cwd(self, tmp_path): + """The process runs in the requested working directory.""" + result = run_cli_session( + f'{PYTHON} -c "import os; print(os.getcwd())"', + cwd=str(tmp_path), + ) + + assert tmp_path.name in result.output + + +class TestToolRegistration: + """Tests for QA agent access to the harness tool.""" + + def test_qa_agents_have_cli_session_tool(self): + """QA reviewer and fixer configs include run_cli_session.""" + from agents.tools_pkg.models import ( + TOOL_RUN_CLI_SESSION, + get_agent_config, + ) + + for agent_type in ("qa_reviewer", "qa_fixer"): + config = get_agent_config(agent_type) + assert TOOL_RUN_CLI_SESSION in config["auto_claude_tools"] + + def test_coder_does_not_have_cli_session_tool(self): + """Coder agents are not given the harness (QA-only).""" + from agents.tools_pkg.models import ( + TOOL_RUN_CLI_SESSION, + get_agent_config, + ) + + config = get_agent_config("coder") + assert TOOL_RUN_CLI_SESSION not in config.get("auto_claude_tools", []) From 158fc4320202104396deb63ec0bffc93bd5241c1 Mon Sep 17 00:00:00 2001 From: Oleg Miagkov Date: Sun, 12 Jul 2026 10:17:39 +0400 Subject: [PATCH 2/3] fix(qa): pass raw command string to CreateProcess on Windows shlex.split(posix=False) keeps surrounding quotes, so quoted arguments reached Windows processes with literal quote characters. CreateProcess parses the command line itself; pass the string through unchanged. Co-Authored-By: Claude Fable 5 Signed-off-by: Oleg Miagkov --- apps/backend/core/cli_session.py | 18 ++++++++++++------ 1 file changed, 12 insertions(+), 6 deletions(-) diff --git a/apps/backend/core/cli_session.py b/apps/backend/core/cli_session.py index 975c89316..ca7d3d17f 100644 --- a/apps/backend/core/cli_session.py +++ b/apps/backend/core/cli_session.py @@ -46,7 +46,8 @@ def run_cli_session( Run a CLI application and capture its output. Args: - command: Command line to run (parsed with shlex, no shell) + command: Command line to run (shlex-parsed on POSIX, passed to + CreateProcess unchanged on Windows; never run through a shell) cwd: Working directory for the process inputs: Lines to send to stdin, in order (newline appended) timeout_seconds: Kill the process after this many seconds @@ -54,17 +55,22 @@ def run_cli_session( Returns: CliSessionResult with exit code, combined output, and timeout flag """ - argv = shlex.split(command, posix=not is_windows()) - if not argv: + if not command.strip(): return CliSessionResult(exit_code=None, output="", timed_out=False) if is_windows(): - return _run_with_pipes(argv, cwd, inputs, timeout_seconds) + # CreateProcess parses the command line itself; shlex quoting rules + # do not apply on Windows, so pass the string through unchanged + return _run_with_pipes(command, cwd, inputs, timeout_seconds) + + argv = shlex.split(command) + if not argv: + return CliSessionResult(exit_code=None, output="", timed_out=False) return _run_with_pty(argv, cwd, inputs, timeout_seconds) def _run_with_pipes( - argv: list[str], + command: str, cwd: str, inputs: list[str] | None, timeout_seconds: int, @@ -72,7 +78,7 @@ def _run_with_pipes( """Run with plain pipes (Windows fallback, no pty in stdlib).""" stdin_data = "".join(line + "\n" for line in inputs) if inputs else None proc = subprocess.Popen( - argv, + command, cwd=cwd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, From b76e8dee65fc07dda8596dfaf9d0d9b766853c93 Mon Sep 17 00:00:00 2001 From: Oleg Miagkov Date: Sun, 12 Jul 2026 10:29:22 +0400 Subject: [PATCH 3/3] fix(qa): non-blocking pty writes + independent cleanup in cli_session Address review comments: - os.write on a full pty buffer could block forever when the child never reads stdin, bypassing the timeout. Master fd is now non-blocking with a write buffer (_flush_input handles partial writes and EAGAIN); BlockingIOError on read treated as no-data, not EOF. - finally block steps are now independent: a failing os.close can no longer skip proc.kill()/wait() and leak a zombie process. Co-Authored-By: Claude Fable 5 Signed-off-by: Oleg Miagkov --- apps/backend/core/cli_session.py | 37 +++++++++++++++++++++++++++++--- tests/test_cli_session.py | 11 ++++++++++ 2 files changed, 45 insertions(+), 3 deletions(-) diff --git a/apps/backend/core/cli_session.py b/apps/backend/core/cli_session.py index ca7d3d17f..2ce8dfc08 100644 --- a/apps/backend/core/cli_session.py +++ b/apps/backend/core/cli_session.py @@ -113,6 +113,9 @@ def _run_with_pty( import select master_fd, slave_fd = pty.openpty() + # Non-blocking master: a child that never reads stdin must not let + # os.write block past the timeout when the pty buffer fills up + os.set_blocking(master_fd, False) proc = subprocess.Popen( argv, cwd=cwd, @@ -125,6 +128,7 @@ def _run_with_pty( os.close(slave_fd) pending = list(inputs or []) + write_buf = b"" chunks: list[bytes] = [] total = 0 deadline = time.monotonic() + timeout_seconds @@ -143,9 +147,10 @@ def _run_with_pty( if eof: break - if pending and time.monotonic() >= next_write: - os.write(master_fd, (pending.pop(0) + "\n").encode()) + if not write_buf and pending and time.monotonic() >= next_write: + write_buf = (pending.pop(0) + "\n").encode() next_write = time.monotonic() + INPUT_INTERVAL_SECONDS + write_buf = _flush_input(master_fd, write_buf) if proc.poll() is not None: # Process exited: drain whatever is left, then stop @@ -155,7 +160,11 @@ def _run_with_pty( select.select([master_fd], [], [], 0.05) finally: - os.close(master_fd) + # Independent cleanup steps: a failure in one must not skip the rest + try: + os.close(master_fd) + except OSError: + pass if proc.poll() is None: proc.kill() proc.wait() @@ -183,6 +192,9 @@ def _drain_output(master_fd: int, chunks: list[bytes], total: int) -> bool: return False try: data = os.read(master_fd, 4096) + except BlockingIOError: + # Non-blocking fd raced with select: no data right now + return False except OSError: # EIO: slave side closed - treat as EOF return True @@ -193,4 +205,23 @@ def _drain_output(master_fd: int, chunks: list[bytes], total: int) -> bool: total += len(data) +def _flush_input(master_fd: int, write_buf: bytes) -> bytes: + """ + Write as much buffered stdin as the pty accepts without blocking. + + Returns: + The unwritten remainder of the buffer + """ + if not write_buf: + return write_buf + try: + written = os.write(master_fd, write_buf) + except BlockingIOError: + return write_buf + except OSError: + # Slave side closed: nothing left to feed + return b"" + return write_buf[written:] + + __all__ = ["CliSessionResult", "run_cli_session", "MAX_OUTPUT_CHARS"] diff --git a/tests/test_cli_session.py b/tests/test_cli_session.py index ae2662323..037dcd0e1 100644 --- a/tests/test_cli_session.py +++ b/tests/test_cli_session.py @@ -81,6 +81,17 @@ def test_timeout_kills_process(self, tmp_path): assert result.timed_out is True + def test_timeout_fires_when_child_never_reads_stdin(self, tmp_path): + """Unconsumed stdin filling the pty buffer must not block the timeout.""" + result = run_cli_session( + f'{PYTHON} -c "import time; time.sleep(60)"', + cwd=str(tmp_path), + inputs=["x" * 65536] * 8, + timeout_seconds=3, + ) + + assert result.timed_out is True + def test_output_is_capped(self, tmp_path): """Runaway output is truncated to the cap.""" result = run_cli_session(