diff --git a/runtime/cli/interactive.py b/runtime/cli/interactive.py index c4f5e80..ccf3e98 100644 --- a/runtime/cli/interactive.py +++ b/runtime/cli/interactive.py @@ -448,6 +448,7 @@ def _cmd_export(args: str) -> None: lines.append(m.content) lines.append("") + path.parent.mkdir(parents=True, exist_ok=True) path.write_text("\n".join(lines), encoding="utf-8") console.print(f"[green]Exported to {path}[/]") @@ -584,7 +585,10 @@ def start() -> None: session = _create_session() if session is None: - console.print("[dim](Tab completion unavailable in this terminal)[/]\n") + console.print( + "[dim](Tab completion not available in Git Bash / mintty. " + "Use cmd.exe, Windows Terminal, or PowerShell for full features.)[/]\n" + ) while True: try: diff --git a/runtime/orchestrator/workflows/test_coordinator.py b/runtime/orchestrator/workflows/test_coordinator.py index 8db930d..d4d9abc 100644 --- a/runtime/orchestrator/workflows/test_coordinator.py +++ b/runtime/orchestrator/workflows/test_coordinator.py @@ -67,7 +67,7 @@ class TestCoordinatorPipeline: ] def run(self, target: str) -> PipelineResult: - """Execute the full pipeline. Returns PipelineResult with step details.""" + """Execute the full pipeline per skills/test-coordinator.md.""" result = PipelineResult(ok=True) run_id = f"tc-{int(time.time())}" console.print(f"[bold]Test Coordinator Pipeline[/] ({run_id})") @@ -83,17 +83,19 @@ def run(self, target: str) -> PipelineResult: result.summary = f"Target outside workspace: {target}" return result - # Phase 0: Pre-flight - missing = self._preflight() + # Step 0: Pre-flight checklist (test-coordinator.md Step 0) + platform_hints = self._detect_platform(target) + missing = self._preflight(platform_hints) if missing: - console.print(f"[red]Pre-flight failed: {', '.join(missing)}[/]") + self._print_checklist(platform_hints, missing) result.ok = False result.aborted_at = "preflight" result.summary = f"Missing: {', '.join(missing)}" return result - # Phase 1: PRD load + route - routing = self._route_target(target) + # Step 1: PRD load + platform identification (test-coordinator.md Step 1) + prd_text = self._load_prd(target) + routing = self._route_target(prd_text or target) console.print(f"[dim]Router → {routing}[/dim]") # Execute each step @@ -103,7 +105,11 @@ def run(self, target: str) -> PipelineResult: console.print(f" [{i}/{len(self.SEQUENCE)}] {name}...", end=" ") try: - outcome = self._execute_node(name, kind, target) + # Step 4 (env-manager): retry on failure per test-coordinator.md + if name == "env-manager": + outcome = self._execute_with_retry(name, kind, target) + else: + outcome = self._execute_node(name, kind, target) step.status = "ok" if outcome.get("ok", True) else "failed" step.output = str(outcome.get("stdout", ""))[:200] step.duration_ms = outcome.get("duration_ms", 0) @@ -134,20 +140,108 @@ def run(self, target: str) -> PipelineResult: result.summary = self._build_summary(result) console.print() console.print(f"[bold]{result.summary}[/]") + + # Step 10+: Notification (best-effort) + self._notify(result.summary) + return result - def _preflight(self) -> list[str]: - """Check required env vars and tools. Returns list of missing items.""" + def _preflight(self, platform_hints: list[str] | None = None) -> list[str]: + """Step 0: Pre-flight checklist per test-coordinator.md.""" missing = [] - # Check Python version import sys if sys.version_info < (3, 10): missing.append("Python 3.10+ required") - # Check workspace exists if not _WORKSPACE.is_dir(): missing.append(f"workspace directory not found: {_WORKSPACE}") + + hints = set(platform_hints or []) + # Platform-specific checks from test-coordinator.md Step 0 + if "desktop_windows" in hints: + if not os.environ.get("WIN_APP_PATH"): + missing.append("WIN_APP_PATH (.env) — EXE完整路径") + try: + import pyautogui # noqa: F401 + except ImportError: + missing.append("pip install pyautogui (desktop test)") + if "mobile_android" in hints or "mobile_ios" in hints: + if not os.environ.get("ANDROID_HOME") and "android" in str(hints): + missing.append("ANDROID_HOME (.env)") + if "api" in hints or "web" in hints: + if not os.environ.get("TEST_APP_URL"): + missing.append("TEST_APP_URL (.env)") return missing + def _detect_platform(self, target: str) -> list[str]: + """Simple keyword-based platform detection for preflight checklist.""" + text = target.lower() + hints = [] + if any(w in text for w in ("exe", "windows", "desktop", "win32", "pywinauto")): + hints.append("desktop_windows") + if any(w in text for w in ("android", "apk", "adb")): + hints.append("mobile_android") + if any(w in text for w in ("ios", "ipa", "xcode")): + hints.append("mobile_ios") + if any(w in text for w in ("api", "rest", "graphql", "endpoint", "http")): + hints.append("api") + if any(w in text for w in ("web", "browser", "playwright", "selenium", "page")): + hints.append("web") + if any(w in text for w in ("can", "automotive", "adas", "ota", "ecu")): + hints.append("automotive") + return hints + + def _print_checklist(self, platform_hints: list[str], missing: list[str]) -> None: + """Print pre-flight checklist per test-coordinator.md Step 0.""" + from rich.panel import Panel + detected = ", ".join(platform_hints) if platform_hints else "generic" + lines = [f"Detected: {detected}", ""] + lines.append("[bold]Required:[/]") + for m in missing: + lines.append(f" [red]✗[/] {m}") + lines.append("") + lines.append("[dim]Fix missing items and re-run.[/]") + console.print(Panel("\n".join(lines), title="Pre-flight Checklist", title_align="left")) + + def _load_prd(self, target: str) -> str | None: + """Step 1: Load PRD via prd_loader per test-coordinator.md.""" + try: + from utils.prd_loader import load_prd, suggest_agents + text, meta = load_prd(target) + if text: + agents = suggest_agents(text) + console.print(f"[dim]PRD loaded: {len(text)} chars, agents: {agents}[/]") + return text[:5000] # cap for LLM context + except ImportError: + pass + except Exception: + pass + return None + + def _execute_with_retry(self, name: str, kind: str, target: str) -> dict[str, Any]: + """Step 4 (env-manager): retry on failure per test-coordinator.md. + Retry delays: 10s → 20s → 40s. Abort after 3 failures. + """ + delays = [10, 20, 40] + for attempt, delay in enumerate(delays, 1): + outcome = self._execute_node(name, kind, target) + if outcome.get("ok"): + return outcome + console.print(f"[yellow]retry {attempt}/{len(delays)} in {delay}s...[/]", end=" ") + time.sleep(delay) + console.print("[red]env-manager failed after 3 retries[/]") + return {"ok": False, "stdout": "env-manager exhausted retries", "duration_ms": 0} + + def _notify(self, summary: str) -> None: + """Post-pipeline notification per test-coordinator.md Step 10.""" + webhook = os.environ.get("TAGENT_NOTIFY_URL", "") + if not webhook: + return + try: + import requests + requests.post(webhook, json={"text": summary}, timeout=5) + except Exception: + pass # notification is best-effort + def _route_target(self, target: str) -> str: """Quick routing: what does the router want for this target?""" try: diff --git a/runtime/tests/test_interactive_commands.py b/runtime/tests/test_interactive_commands.py index 5f38453..ae95972 100644 --- a/runtime/tests/test_interactive_commands.py +++ b/runtime/tests/test_interactive_commands.py @@ -147,3 +147,74 @@ def test_common_typos_corrected(self): def test_very_different_no_match(self): from runtime.cli.interactive import _closest_command assert _closest_command("zzzpq") is None + + +class TestCostEstimation: + def test_empty_memory_minimal_tokens(self): + from runtime.cli.interactive import _estimate_cost, _get_memory + mem = _get_memory() + mem.clear() + tokens, cost = _estimate_cost(mem) + assert tokens >= 0 + assert cost >= 0.0 + + def test_cost_scales_with_messages(self): + from runtime.cli.interactive import _estimate_cost, _get_memory + mem = _get_memory() + mem.clear() + mem.add("user", "x" * 400) + mem.add("assistant", "y" * 400) + tokens, cost = _estimate_cost(mem) + assert tokens >= 100 + assert cost >= 0.0 # may be 0 if provider=stub + + def test_ollama_pricing_is_zero(self): + from runtime.cli.interactive import _PRICE_PER_1K + in_p, out_p = _PRICE_PER_1K["ollama"] + assert in_p == 0 + assert out_p == 0 + + def test_claude_pricing_positive(self): + from runtime.cli.interactive import _PRICE_PER_1K + in_p, out_p = _PRICE_PER_1K["claude"] + assert in_p > 0 + assert out_p > 0 + + +class TestCompact: + def test_compact_too_few_messages(self): + from runtime.cli.interactive import _cmd_compact, _get_memory + mem = _get_memory() + mem.clear() + mem.add("user", "a") + mem.add("assistant", "b") + _cmd_compact("") # should not crash with 2 messages + + def test_compact_on_six_messages(self): + from runtime.cli.interactive import _cmd_compact, _get_memory + mem = _get_memory() + mem.clear() + for i in range(6): + mem.add("user" if i % 2 == 0 else "assistant", f"msg {i}") + _cmd_compact("") + assert len(mem.messages) < 6 # compacted + + +class TestSessions: + def test_sessions_no_crash(self): + from runtime.cli.interactive import _cmd_sessions + _cmd_sessions("") # should not crash even with no sessions + + +class TestExport: + def test_export_empty_memory(self): + from runtime.cli.interactive import _cmd_export, _get_memory + mem = _get_memory() + mem.clear() + _cmd_export("") # should not crash + + def test_export_with_messages(self): + from runtime.cli.interactive import _cmd_export, _get_memory + mem = _get_memory() + mem.add("user", "hello") + _cmd_export("") # should create export file diff --git a/runtime/tests/test_test_coordinator_workflow.py b/runtime/tests/test_test_coordinator_workflow.py index 7ab9acc..b06a487 100644 --- a/runtime/tests/test_test_coordinator_workflow.py +++ b/runtime/tests/test_test_coordinator_workflow.py @@ -38,7 +38,6 @@ def test_preflight_checks_python_version(self): from runtime.orchestrator.workflows.test_coordinator import TestCoordinatorPipeline p = TestCoordinatorPipeline() missing = p._preflight() - # Python 3.10+ on all modern systems → should be empty assert isinstance(missing, list) def test_preflight_returns_list(self): @@ -47,6 +46,47 @@ def test_preflight_returns_list(self): result = p._preflight() assert isinstance(result, list) + def test_preflight_desktop_hints(self): + from runtime.orchestrator.workflows.test_coordinator import TestCoordinatorPipeline + p = TestCoordinatorPipeline() + missing = p._preflight(["desktop_windows"]) + assert isinstance(missing, list) + + +class TestPlatformDetection: + def test_detect_desktop_windows(self): + from runtime.orchestrator.workflows.test_coordinator import TestCoordinatorPipeline + p = TestCoordinatorPipeline() + hints = p._detect_platform("test this Windows EXE program") + assert "desktop_windows" in hints + + def test_detect_api(self): + from runtime.orchestrator.workflows.test_coordinator import TestCoordinatorPipeline + p = TestCoordinatorPipeline() + hints = p._detect_platform("test the REST API endpoint") + assert "api" in hints + + def test_detect_web(self): + from runtime.orchestrator.workflows.test_coordinator import TestCoordinatorPipeline + p = TestCoordinatorPipeline() + hints = p._detect_platform("browser based web application") + assert "web" in hints + + def test_detect_multiple(self): + from runtime.orchestrator.workflows.test_coordinator import TestCoordinatorPipeline + p = TestCoordinatorPipeline() + hints = p._detect_platform("test the API backend and web frontend") + assert "api" in hints + assert "web" in hints + + +class TestPRDLoader: + def test_load_prd_handles_missing_file(self): + from runtime.orchestrator.workflows.test_coordinator import TestCoordinatorPipeline + p = TestCoordinatorPipeline() + result = p._load_prd("/nonexistent/path.md") + assert result is None + class TestPipelineResult: def test_pipeline_result_defaults(self):