hermes_agent.py

import os
import json
import subprocess
from typing import Any, Dict, List
import httpx
from .base import BaseAgent, AgentContext
from .registry import AgentRegistry


@AgentRegistry.register("assistant")
class HermesCodingAgent(BaseAgent):
    def __init__(self, agent_id: str = None, name: str = None):
        super().__init__(agent_id, name)
        self.base_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "../.."))
        self.config_path = os.path.join(self.base_dir, "config/model-controller.json")
        self.load_config()

    def load_config(self) -> None:
        with open(self.config_path, "r", encoding="utf-8") as f:
            self.cfg = json.load(f)

    async def execute(self, context: AgentContext, prompt: str = "", tool: str = None, **kwargs) -> Any:
        if tool == "file_read":
            return self.file_read(kwargs.get("file_path", ""))
        if tool == "file_write":
            return self.file_write(kwargs.get("file_path", ""), kwargs.get("content", ""))
        if tool == "terminal":
            return self.terminal(kwargs.get("command", ""), kwargs.get("args", []))
        
        return await self.chat(prompt, kwargs.get("memory_files", []))

    async def chat(self, prompt: str, memory_files: List[str]) -> Dict[str, str]:
        sys_path = os.path.join(self.base_dir, self.cfg["systemPromptPath"])
        rules_path = os.path.join(self.base_dir, self.cfg["projectRulesPath"])
        
        with open(sys_path, "r", encoding="utf-8") as f:
            system_prompt = f.read()
        with open(rules_path, "r", encoding="utf-8") as f:
            project_rules = f.read()

        memory_content = self.load_project_memory(memory_files)
        full_prompt = f"{project_rules}

{memory_content}

{prompt}"

        async with httpx.AsyncClient(timeout=120.0) as client:
            response = await client.post(
                f"{self.cfg['baseUrl']}/v1/chat/completions",
                json={
                    "model": self.cfg["modelName"],
                    "temperature": self.cfg["temperature"],
                    "top_p": self.cfg["top_p"],
                    "max_tokens": self.cfg["max_tokens"],
                    "messages": [
                        {"role": "system", "content": system_prompt},
                        {"role": "user", "content": full_prompt}
                    ]
                }
            )
            response.raise_for_status()
            data = response.json()
            return {"response": data["choices"][0]["message"]["content"]}

    def safe_resolve(self, base_root: str, file_path: str) -> str:
        resolved_base = os.path.abspath(os.path.join(self.base_dir, base_root))
        resolved_target = os.path.abspath(os.path.join(resolved_base, file_path))
        if resolved_target == resolved_base or resolved_target.startswith(resolved_base + os.sep):
            return resolved_target
        raise ValueError("Invalid file path / Path traversal detected")

    def load_project_memory(self, requested_files: List[str]) -> str:
        memory_content = ""
        allowed_extensions = {".md", ".txt", ".json", ".ts", ".tsx", ".js", ".prisma"}
        blocked_filenames = {".env", ".env.local", ".env.production"}
        blocked_extensions = {".pem", ".key", ".p12", ".sqlite", ".db"}
        memory_root = self.cfg["projectMemoryRoot"]

        for file in requested_files:
            _, ext = os.path.splitext(file)
            if ext not in allowed_extensions:
                continue
            basename = os.path.basename(file)
            if basename in blocked_filenames or ext in blocked_extensions:
                continue

            try:
                file_path = self.safe_resolve(memory_root, file)
                if os.path.exists(file_path):
                    if os.path.getsize(file_path) <= 1024 * 1024:
                        with open(file_path, "r", encoding="utf-8") as f:
                            memory_content += f.read() + "

"
            except ValueError:
                continue
        return memory_content

    def file_read(self, file_path: str) -> Dict[str, Any]:
        if "file_read" not in self.cfg["allowedTools"]:
            return {"error": "File read not allowed"}
        
        try:
            allowed_path = self.safe_resolve(self.cfg["repoRoot"], file_path)
        except ValueError as e:
            return {"error": str(e)}

        if not os.path.exists(allowed_path):
            return {"error": "File not found"}

        blocked_filenames = {".env", ".env.local", ".env.production"}
        blocked_extensions = {".pem", ".key", ".p12", ".sqlite", ".db"}
        allowed_extensions = {".md", ".txt", ".json", ".ts", ".tsx", ".js", ".prisma"}
        basename = os.path.basename(allowed_path)
        _, extname = os.path.splitext(allowed_path)

        if basename in blocked_filenames or extname in blocked_extensions:
            return {"error": "File read not allowed"}
        if extname not in allowed_extensions:
            return {"error": "File extension not allowed"}

        with open(allowed_path, "r", encoding="utf-8") as f:
            return {"content": f.read()}

    def file_write(self, file_path: str, content: str) -> Dict[str, Any]:
        if "file_write" not in self.cfg["allowedTools"]:
            return {"error": "File write not allowed"}

        try:
            allowed_path = self.safe_resolve(self.cfg["repoRoot"], file_path)
        except ValueError as e:
            return {"error": str(e)}

        normalized_path = file_path.replace("\", "/").lstrip("./")
        frontend_paths = ["frontend/", "client/", "web/", "app/", "pages/", "components/", "src/frontend/"]
        if not self.cfg["allowFrontendFiles"] and any(normalized_path.startswith(p) for p in frontend_paths):
            return {"error": "Frontend file write not allowed"}

        blocked_filenames = {".env", ".env.local", ".env.production"}
        blocked_extensions = {".pem", ".key", ".p12", ".sqlite", ".db"}
        allowed_extensions = {".md", ".txt", ".json", ".ts", ".tsx", ".js", ".prisma"}
        basename = os.path.basename(allowed_path)
        _, extname = os.path.splitext(allowed_path)

        if basename in blocked_filenames or extname in blocked_extensions:
            return {"error": "File write not allowed"}
        if extname not in allowed_extensions:
            return {"error": "File extension not allowed"}

        os.makedirs(os.path.dirname(allowed_path), exist_ok=True)
        with open(allowed_path, "w", encoding="utf-8") as f:
            f.write(content)
        return {"success": True}

    def terminal(self, command: str, args: List[str]) -> Dict[str, Any]:
        if "terminal" not in self.cfg["allowedTools"]:
            return {"error": "Terminal not allowed"}

        rules = self.cfg["allowedTerminalCommands"].get(command)
        if not rules:
            return {"error": "Command not allowed"}

        is_allowed = False
        for rule in rules:
            rule_args = rule["args"]
            prefix_matches = len(rule_args) <= len(args) and all(args[i] == rule_args[i] for i in range(len(rule_args)))
            if not prefix_matches:
                continue
            if rule["allowExtraArgs"] or len(rule_args) == len(args):
                is_allowed = True
                break

        if not is_allowed:
            return {"error": "Command arguments not allowed"}

        try:
            resolved_root = self.safe_resolve(self.cfg["repoRoot"], "")
            res = subprocess.run(
                [command] + args,
                cwd=resolved_root,
                capture_output=True,
                text=True,
                shell=False
            )
            return {"stdout": res.stdout, "stderr": res.stderr, "exitCode": res.returncode}
        except Exception as e:
            return {"error": f"Command execution failed: {str(e)}"}