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
6 changes: 5 additions & 1 deletion runtime/cli/interactive.py
Original file line number Diff line number Diff line change
Expand Up @@ -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}[/]")

Expand Down Expand Up @@ -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:
Expand Down
116 changes: 105 additions & 11 deletions runtime/orchestrator/workflows/test_coordinator.py
Original file line number Diff line number Diff line change
Expand Up @@ -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})")
Expand All @@ -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
Expand All @@ -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)
Expand Down Expand Up @@ -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:
Expand Down
71 changes: 71 additions & 0 deletions runtime/tests/test_interactive_commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
42 changes: 41 additions & 1 deletion runtime/tests/test_test_coordinator_workflow.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand All @@ -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):
Expand Down