core/shell_executor.py

"""Local shell executor: run shell_call commands and return shell_call_output."""

import re
import subprocess
import yaml
from pathlib import Path
from core.config_loader import BASE_DIR, PATHS, IS_WINDOWS

SHELL_TIMEOUT = 30

# Match `type/cat foo.md` (single-line), used to short-circuit pure file reads
# back to Python — saves a cmd.exe spawn (~50-200ms on Windows).
_FILE_READ_RE = re.compile(
    r"^(?:type|cat)\s+[\"']?([^\"|']+\.md)[\"']?\s*$",
    re.IGNORECASE,
)

# {path_fragment: skill_name} — both absolute and relative forms are listed
# because the model may emit either (relative is typical: `cat skills/foo/...`).
SKILL_PATHS = {
    p: name
    for name, rel_path in PATHS.get("skills", {}).items()
    for p in (
        rel_path.replace("\\", "/"),
        str(BASE_DIR / rel_path).replace("\\", "/"),
    )
}


def parse_skill_description(skill_md_path):
    """Parse description from SKILL.md YAML frontmatter."""
    content = skill_md_path.read_text(encoding="utf-8")
    parts = content.split("---", 2)
    if len(parts) >= 3:
        frontmatter = yaml.safe_load(parts[1])
        if isinstance(frontmatter, dict) and frontmatter.get("description"):
            return frontmatter["description"]
    return None


def build_tools():
    """Build the tools list from config skill paths."""
    skills = []
    for name, rel_path in PATHS["skills"].items():
        skill_dir = BASE_DIR / rel_path
        skill_md = skill_dir / "SKILL.md"
        description = f"Skill: {name}"
        if skill_md.exists():
            description = parse_skill_description(skill_md) or description
        skills.append({
            "name": name,
            "description": description,
            "path": str(skill_dir),
        })
    return [{
        "type": "shell",
        "environment": {"type": "local", "skills": skills},
    }]



def detect_skills(commands):
    """Detect which skill(s) a set of commands reference, based on path fragments."""
    triggered = set()
    for cmd in commands:
        normalized_cmd = cmd.replace("\\", "/")
        for path_fragment, skill_name in SKILL_PATHS.items():
            if path_fragment in normalized_cmd:
                triggered.add(skill_name)
    return sorted(triggered)


def execute_shell_call(shell_call):
    """Execute a shell_call locally. Returns (shell_call_output_dict, list_of_skill_names)."""
    commands = getattr(shell_call.action, "commands", [])
    results = []

    for cmd in commands:
        # Short-circuit `type/cat *.md` → read the file directly, skipping the
        # cmd.exe / sh spawn (saves ~50-200ms per read on Windows).
        m = _FILE_READ_RE.match(cmd.strip())
        if m:
            rel = m.group(1).replace("\\", "/")
            path = (BASE_DIR / rel).resolve()
            try:
                results.append({
                    "stdout": path.read_text(encoding="utf-8", errors="replace"),
                    "stderr": "",
                    "outcome": {"type": "exit", "exit_code": 0},
                })
            except OSError as e:
                results.append({
                    "stdout": "",
                    "stderr": f"{type(e).__name__}: {e}",
                    "outcome": {"type": "exit", "exit_code": 1},
                })
            continue

        try:
            proc = subprocess.run(
                cmd,
                shell=True,
                capture_output=True,
                text=True,
                encoding="utf-8",
                errors="replace",
                timeout=SHELL_TIMEOUT,
                cwd=str(BASE_DIR),
            )
            results.append({
                # OpenAI rejects null stdout/stderr — coerce to empty string.
                "stdout": proc.stdout or "",
                "stderr": proc.stderr or "",
                "outcome": {"type": "exit", "exit_code": proc.returncode},
            })
        except subprocess.TimeoutExpired:
            results.append({
                "stdout": "",
                "stderr": f"Command timed out after {SHELL_TIMEOUT} seconds.",
                "outcome": {"type": "timeout"},
            })

    skills_used = detect_skills(commands)

    return {
        "type": "shell_call_output",
        "call_id": shell_call.call_id,
        "output": results,
    }, skills_used