diff --git a/loopx/capabilities/performance_diagnosis/cli.py b/loopx/capabilities/performance_diagnosis/cli.py index a27d775e16..723f26e91a 100644 --- a/loopx/capabilities/performance_diagnosis/cli.py +++ b/loopx/capabilities/performance_diagnosis/cli.py @@ -7,8 +7,6 @@ from pathlib import Path from typing import Any, Callable -from ...control_plane.effect_runtime import effect_runtime_result - MAX_INPUT_BYTES = 16 * 1024 * 1024 AddFormat = Callable[[argparse.ArgumentParser], None] OutputFormat = Callable[..., str] @@ -46,6 +44,8 @@ def _render(payload: dict[str, Any]) -> str: def handle_performance_diagnosis_command(args: argparse.Namespace, *, output_format: OutputFormat, print_payload: PrintPayload) -> int | None: if args.command != "performance-diagnosis": return None + from ...control_plane.effect_runtime import effect_runtime_result + try: if args.performance_diagnosis_action == "plan": params = {"tool": args.tool, "command": _load(args.command_json), diff --git a/loopx/cli.py b/loopx/cli.py index 7fd25b2f08..c346f98989 100644 --- a/loopx/cli.py +++ b/loopx/cli.py @@ -89,7 +89,6 @@ register_external_evidence_commands, ) from .capabilities.performance_diagnosis.cli import ( - handle_performance_diagnosis_command, register_performance_diagnosis_commands, ) from .cli_commands import ( @@ -708,12 +707,6 @@ def _main(argv: list[str] | None = None) -> int: if external_evidence_result is not None: return external_evidence_result - performance_diagnosis_result = handle_performance_diagnosis_command( - args, output_format=output_format, print_payload=print_payload, - ) - if performance_diagnosis_result is not None: - return performance_diagnosis_result - registry_admin_result = handle_registry_admin_command( args, registry_path=registry_path, diff --git a/loopx/cli_runtime.py b/loopx/cli_runtime.py index 8983fe03f6..454a8ddf1d 100644 --- a/loopx/cli_runtime.py +++ b/loopx/cli_runtime.py @@ -76,6 +76,7 @@ "manager-inbox", "agent-context", "agent-directory", "agent-capabilities", "handoff", "handoff-mode", + "performance-diagnosis", } @@ -385,6 +386,10 @@ def _build_selected_parser(command: str) -> LoopXArgumentParser: from .help_surface import register_command_reference register_command_reference(subparsers) + elif command == "performance-diagnosis": + from .capabilities.performance_diagnosis.cli import register_performance_diagnosis_commands + + register_performance_diagnosis_commands(subparsers, add_subcommand_format) else: # pragma: no cover - caller guards the private interface raise ValueError(f"unsupported selected command: {command}") return parser @@ -396,6 +401,12 @@ def _dispatch_common_command( registry_path: Path, allow_missing_registry: bool, ) -> int | None: + if args.command == "performance-diagnosis": + from .capabilities.performance_diagnosis.cli import handle_performance_diagnosis_command + + return handle_performance_diagnosis_command( + args, output_format=output_format, print_payload=print_payload, + ) if args.command in {"agent-context", "agent-directory", "agent-capabilities"}: from .paths import effective_runtime_root diff --git a/tests/capabilities/test_performance_diagnosis.py b/tests/capabilities/test_performance_diagnosis.py index 35faa5a8d2..f5665d62c4 100644 --- a/tests/capabilities/test_performance_diagnosis.py +++ b/tests/capabilities/test_performance_diagnosis.py @@ -12,7 +12,7 @@ def command(tmp_path: Path, *args: str) -> subprocess.CompletedProcess: - return subprocess.run([sys.executable, "-m", "loopx.cli", "--format", "json", *args], + return subprocess.run([sys.executable, "-m", "loopx.entrypoint", "--format", "json", *args], env={**os.environ, "TMPDIR": str(tmp_path)}, text=True, capture_output=True, timeout=45) diff --git a/tests/capabilities/test_performance_diagnosis_cli.py b/tests/capabilities/test_performance_diagnosis_cli.py new file mode 100644 index 0000000000..0384f941bd --- /dev/null +++ b/tests/capabilities/test_performance_diagnosis_cli.py @@ -0,0 +1,146 @@ +"""Preserve CLI compatibility while loading only the requested command.""" +from __future__ import annotations + +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + + +ROOT = Path(__file__).resolve().parents[2] + + +def run_cli( + module: str, args: list[str], *, environment: dict[str, str] | None = None, + forbidden_modules: tuple[str, ...] = (), +) -> subprocess.CompletedProcess[str]: + # A common program name makes argparse output directly comparable. + script = f""" +import sys +sys.argv[0] = 'loopx' +from {module} import main +code = main() +for module in {forbidden_modules!r}: + assert module not in sys.modules, 'unexpected import: ' + module +raise SystemExit(code) +""" + return subprocess.run( + [sys.executable, "-c", script, *args], + cwd=ROOT, env={**os.environ, **(environment or {}), "LOOPX_USAGE_PING": "0"}, + capture_output=True, text=True, timeout=45, + ) + + +@pytest.mark.parametrize("args,code", [ + (["performance-diagnosis", "--help"], 0), + (["performance-diagnosis", "plan", "--help"], 0), + (["performance-diagnosis", "inspect", "--help"], 0), + (["--format=json", "performance-diagnosis", "plan", "--help"], 0), + (["--registry=missing.json", "--runtime-root=missing", "--cli-bin=loopx", "performance-diagnosis", "inspect", "--help"], 0), + (["performance-diagnosis"], 2), + (["performance-diagnosis", "unknown"], 2), + (["performance-diagnosis", "plan"], 2), + (["performance-diagnosis", "plan", "--tool", "node-cpu"], 2), + (["performance-diagnosis", "inspect"], 2), + (["performance-diagnosis", "inspect", "--profile-json", "missing.json", "--top", "invalid"], 2), + (["performance-diagnosis", "inspect", "--profile-json", "missing.json", "--unknown"], 2), + (["performance-diagnosis", "inspect", "--profile-json", "missing.json", "--format", "invalid"], 2), + (["performance-diagnosis", "inspect", "--profile", "missing.json"], 2), + (["--format", "invalid", "performance-diagnosis", "inspect", "--help"], 2), + (["--registry"], 2), +]) +def test_parser_matches_full_cli(args: list[str], code: int) -> None: + full = run_cli("loopx.cli", args) + entry = run_cli("loopx.entrypoint", args) + assert full.returncode == entry.returncode == code + assert (entry.stdout, entry.stderr) == (full.stdout, full.stderr) + + +@pytest.mark.parametrize("action", ["plan", "inspect"]) +@pytest.mark.parametrize("global_format,local_format", [("json", None), ("markdown", None), ("markdown", "json")]) +def test_real_dispatch_matches_full_cli(tmp_path: Path, action, global_format, local_format) -> None: + source = tmp_path / "input.json" + if action == "plan": + source.write_text(json.dumps(["node", "work.js"]), encoding="utf-8") + options = ["--tool=node-cpu", f"--command-json={source}", f"--output-directory={tmp_path / 'capture'}"] + else: + source.write_text(json.dumps({ + "startTime": 0, "endTime": 1000, + "nodes": [{"id": 1, "callFrame": {"functionName": "work"}}], + "samples": [1], "timeDeltas": [1000], + }), encoding="utf-8") + options = [f"--profile-json={source}", "--top=1"] + args = ["--registry", str(tmp_path / "absent-registry.json"), + "--runtime-root", str(tmp_path / "absent-runtime"), "--cli-bin", "loopx", + "--format", global_format, "performance-diagnosis", action, *options] + if local_format: + args += ["--format", local_format] + full = run_cli("loopx.cli", args) + entry = run_cli("loopx.entrypoint", args, forbidden_modules=("loopx.cli", "loopx.chat")) + assert full.returncode == entry.returncode == 0, full.stderr + entry.stderr + assert (entry.stdout, entry.stderr) == (full.stdout, full.stderr) + assert not (tmp_path / "absent-registry.json").exists() + assert not (tmp_path / "capture").exists() + payload = json.loads(entry.stdout) if (local_format or global_format) == "json" else None + if payload is not None: + claim = "execution_performed" if action == "plan" else "root_cause_proven" + assert payload[claim] is False + + +@pytest.mark.parametrize("content", [None, "{invalid", "{}"]) +def test_input_errors_match_full_cli(tmp_path: Path, content) -> None: + source = tmp_path / "profile.json" + if content is not None: + source.write_text(content, encoding="utf-8") + args = ["performance-diagnosis", "inspect", "--profile-json", str(source), "--format=json"] + full = run_cli("loopx.cli", args) + entry = run_cli("loopx.entrypoint", args) + assert full.returncode == entry.returncode == 1 + assert (entry.stdout, entry.stderr) == (full.stdout, full.stderr) + assert json.loads(entry.stdout)["ok"] is False + + +def test_missing_node_has_same_actionable_failure(tmp_path: Path) -> None: + source = tmp_path / "argv.json" + source.write_text('["node", "work.js"]', encoding="utf-8") + args = ["performance-diagnosis", "plan", "--tool=node-cpu", "--command-json", str(source), + "--output-directory", str(tmp_path / "capture"), "--format=json"] + environment = {"PATH": str(tmp_path / "empty-path"), "TMPDIR": str(tmp_path)} + full = run_cli("loopx.cli", args, environment=environment) + entry = run_cli("loopx.entrypoint", args, environment=environment) + assert full.returncode == entry.returncode == 1 + assert (entry.stdout, entry.stderr) == (full.stdout, full.stderr) + payload = json.loads(entry.stdout) + assert payload["ok"] is False + assert "Node.js" in payload["error"] + + +@pytest.mark.parametrize("action", [[], ["plan"], ["inspect"]]) +def test_help_needs_no_runtime_or_optional_stack(tmp_path: Path, action) -> None: + script = f""" +import importlib.abc +import sys + +forbidden = ('loopx.cli', 'loopx.control_plane.effect_runtime', + 'loopx.chat', 'loopx.capabilities.auto_research') +class BlockOptionalImports(importlib.abc.MetaPathFinder): + def find_spec(self, fullname, path=None, target=None): + if any(fullname == name or fullname.startswith(name + '.') for name in forbidden): + raise AssertionError('help imported ' + fullname) +sys.meta_path.insert(0, BlockOptionalImports()) +def audit(event, args): + if event in ('subprocess.Popen', 'os.system', 'os.posix_spawn', 'socket.connect', 'socket.getaddrinfo'): + raise AssertionError('help attempted ' + event) +sys.addaudithook(audit) +from loopx.entrypoint import main +raise SystemExit(main(['performance-diagnosis', *{action!r}, '--help'])) +""" + result = subprocess.run( + [sys.executable, "-c", script], cwd=ROOT, capture_output=True, text=True, timeout=15, + env={**os.environ, "PATH": str(tmp_path / "no-executables"), "LOOPX_USAGE_PING": "0"}, + ) + assert result.returncode == 0, result.stderr + assert "usage:" in result.stdout