Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions apps/backend/agents/tools_pkg/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
# =============================================================================
Expand Down Expand Up @@ -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",
},
Expand All @@ -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",
},
Expand Down
2 changes: 2 additions & 0 deletions apps/backend/agents/tools_pkg/registry.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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))

Expand Down
2 changes: 2 additions & 0 deletions apps/backend/agents/tools_pkg/tools/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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",
]
139 changes: 139 additions & 0 deletions apps/backend/agents/tools_pkg/tools/cli_harness.py
Original file line number Diff line number Diff line change
@@ -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:

Check warning on line 30 in apps/backend/agents/tools_pkg/tools/cli_harness.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Remove the unused function parameter "spec_dir".

See more on https://sonarcloud.io/project/issues?id=OBenner_Auto-Coding&issues=AZ9U8o7k1sKQ6MLEVzgW&open=AZ9U8o7k1sKQ6MLEVzgW&pullRequest=415
"""
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"]
Loading
Loading