ed
dc397db7ed
TIER-3 READ AGENTS.md + conductor/workflow.md + conductor/code_styleguides/error_handling.md + the 4 source files + 3 test files before this commit. The code_path_audit_phase_2_20260624 track (Tier 2) shipped 11 audit fixes (4 NG1 + 7 NG2) but used a heuristic bypass for 4 of the NG2 wrappers: legacy T | None functions that exist only to maintain test patcher compatibility. Per the review at docs/reports/REVIEW_TIER2_code_path_audit_phase_2_20260624.md Finding 8, this track eliminates the legacy wrappers properly. 11 wrappers eliminated (8 main + 3 _legacy_compat inner): - src/ai_client.py: get_current_tier (1 src + 1 test consumer) - src/ai_client.py: _gemini_tool_declaration + _legacy_compat (2 test consumers) - src/ai_client.py: run_tier4_patch_callback + _legacy_compat (was 0 direct callers but had 2 callback references in app_controller/multi_agent_conductor; callback contract migrated to Callable[[str, str], Result[str]] instead of preserving an Optional[str] adapter) - src/mcp_client.py: _get_symbol_node + _legacy_compat (8 in-file consumers) - src/mcp_client.py: find_in_scope (nested inside _get_symbol_node_result; private impl detail, audit doesn't catch T | None, left as-is) - src/external_editor.py: launch_diff (1 src + 3 test + 1 live_gui test consumer) - src/external_editor.py: launch_editor (no consumers; deleted) - src/session_logger.py: log_tool_output (2 src + 3 test consumers) - src/project_manager.py: parse_ts (no consumers; deleted) For each consumer: replace legacy_fn(args) with legacy_fn_result(args).data. For T | None checks: replace if x is None: with if not result.ok: or if not result.ok or not isinstance(result.data, ...) (depending on pattern). For run_tier4_patch_callback specifically: the wrapper was a callback adapter (not a backward-compat shim) and had 2 callback references as consumers. Rather than keep the adapter (which would re-introduce the Optional[str] return that the strict audit catches), the patch_callback contract was migrated from Callable[[str, str], Optional[str]] to Callable[[str, str], Result[str]] in shell_runner.py + app_controller.py + 9 _send_<vendor>_result signatures in ai_client.py. This propagates the Result[str] through the callback and lets shell_runner unwrap with if r.ok and r.data instead of if patch_text. Verification: - audit_optional_in_3_files --strict: 0 return-type Optional[T] (down from 1) - audit_exception_handling --strict: 0 violations (unchanged) - audit_legacy_wrappers: 0 legacy wrappers (unchanged) - 15 affected test files: 168 tests pass - 8 mcp_client/structural/baseline test files: 55 tests pass - 3 session/gui test files: 7 tests pass - 0 return-type Optional[T] in src/ai_client.py (was 1: run_tier4_patch_callback)
106 lines
4.5 KiB
Python
106 lines
4.5 KiB
Python
# shell_runner.py
|
|
"""
|
|
Shell Runner - Execution engine for PowerShell scripts.
|
|
|
|
This module provides utilities to run PowerShell scripts in a subprocess,
|
|
configuring the environment via mcp_env.toml. It handles timeouts,
|
|
logging, and optional QA/patch callbacks for error recovery.
|
|
"""
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
|
|
from pathlib import Path
|
|
from typing import Callable, Optional
|
|
|
|
from src.result_types import ErrorInfo, ErrorKind, Result
|
|
|
|
try:
|
|
import tomllib
|
|
except ImportError:
|
|
import tomli as tomllib # type: ignore[no-redef]
|
|
|
|
TIMEOUT_SECONDS: int = 60
|
|
_ENV_CONFIG: dict = {}
|
|
|
|
|
|
def _load_env_config() -> dict:
|
|
"""Load mcp_env.toml from project root or environment variable."""
|
|
env_path = os.environ.get("SLOP_MCP_ENV")
|
|
if env_path and Path(env_path).exists():
|
|
with open(env_path, "rb") as f:
|
|
return tomllib.load(f)
|
|
candidates = [
|
|
Path(__file__).parent / "mcp_env.toml",
|
|
Path(__file__).parent.parent / "mcp_env.toml",
|
|
]
|
|
for p in candidates:
|
|
if p.exists():
|
|
with open(p, "rb") as f:
|
|
return tomllib.load(f)
|
|
return {}
|
|
|
|
def _build_subprocess_env() -> dict[str, str]:
|
|
"""Build environment dictionary for subprocess with overrides from mcp_env.toml."""
|
|
global _ENV_CONFIG
|
|
if not _ENV_CONFIG:
|
|
_ENV_CONFIG = _load_env_config()
|
|
env = os.environ.copy()
|
|
# Apply [path].prepend entries
|
|
prepend_dirs = _ENV_CONFIG.get("path", {}).get("prepend", [])
|
|
if prepend_dirs:
|
|
env["PATH"] = os.pathsep.join(prepend_dirs) + os.pathsep + env.get("PATH", "")
|
|
# Apply [env] key-value pairs, expanding ${VAR} references
|
|
for key, val in _ENV_CONFIG.get("env", {}).items():
|
|
env[key] = os.path.expandvars(str(val))
|
|
return env
|
|
|
|
def run_powershell(script: str, base_dir: str, qa_callback: Optional[Callable[[str], str]] = None, patch_callback: Optional[Callable[[str, str], Result[str]]] = None) -> str:
|
|
"""
|
|
Run a PowerShell script with working directory set to base_dir.
|
|
Returns a string combining stdout, stderr, and exit code.
|
|
Environment is configured via mcp_env.toml (project root).
|
|
If qa_callback is provided and the command fails or has stderr,
|
|
the callback is called with the stderr content and its result is appended.
|
|
If patch_callback is provided, it receives (error, file_context) and returns patch text.
|
|
[C: tests/test_tier4_interceptor.py:test_run_powershell_no_qa_callback_on_success, tests/test_tier4_interceptor.py:test_run_powershell_optional_qa_callback, tests/test_tier4_interceptor.py:test_run_powershell_qa_callback_on_failure, tests/test_tier4_interceptor.py:test_run_powershell_qa_callback_on_stderr_only]
|
|
"""
|
|
safe_dir: str = str(base_dir).replace("'", "''")
|
|
full_script: str = f"Set-Location -LiteralPath '{safe_dir}'\n{script}"
|
|
exe: Optional[str] = next((x for x in ["powershell.exe", "pwsh.exe", "powershell", "pwsh"] if shutil.which(x)), None)
|
|
if not exe: return "ERROR: Neither powershell nor pwsh found in PATH"
|
|
try:
|
|
process = subprocess.Popen(
|
|
[exe, "-NoProfile", "-NonInteractive", "-Command", full_script],
|
|
stdin = subprocess.DEVNULL,
|
|
stdout = subprocess.PIPE, stderr=subprocess.PIPE, text=True,
|
|
cwd = base_dir, env=_build_subprocess_env(),
|
|
)
|
|
stdout, stderr = process.communicate(timeout=TIMEOUT_SECONDS)
|
|
parts: list[str] = []
|
|
if stdout.strip(): parts.append(f"STDOUT:\n{stdout.strip()}")
|
|
if stderr.strip(): parts.append(f"STDERR:\n{stderr.strip()}")
|
|
parts.append(f"EXIT CODE: {process.returncode}")
|
|
if (process.returncode != 0 or stderr.strip()) and qa_callback:
|
|
qa_analysis: Optional[str] = qa_callback(stderr.strip())
|
|
if qa_analysis:
|
|
parts.append(f"\nQA ANALYSIS:\n{qa_analysis}")
|
|
if patch_callback and (process.returncode != 0 or stderr.strip()):
|
|
patch_result = patch_callback(stderr.strip(), base_dir)
|
|
if patch_result.ok and patch_result.data:
|
|
parts.append(f"\nAUTO_PATCH:\n{patch_result.data}")
|
|
return "\n".join(parts)
|
|
except subprocess.TimeoutExpired:
|
|
if 'process' in locals() and process:
|
|
subprocess.run(["taskkill", "/F", "/T", "/PID", str(process.pid)], capture_output=True)
|
|
return f"ERROR: timed out after {TIMEOUT_SECONDS}s"
|
|
except KeyboardInterrupt:
|
|
if 'process' in locals() and process:
|
|
subprocess.run(["taskkill", "/F", "/T", "/PID", str(process.pid)], capture_output=True)
|
|
raise
|
|
except (OSError, subprocess.SubprocessError) as e:
|
|
_shell_err = Result(data=None, errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=f"shell run failed: {e}", source="shell_runner.run_powershell", original=e)])
|
|
if 'process' in locals() and process:
|
|
subprocess.run(["taskkill", "/F", "/T", "/PID", str(process.pid)], capture_output=True)
|
|
return f"ERROR: {e}"
|