Blablablab/audio-classification

1"""2Coding Agent Backend Abstraction3 4Defines the interface for coding agent backends and common event types.5Backends implement the agent loop (LLM + tool execution) and yield6events that the CodingAgentRunner consumes.7 8Available backends:9- anthropic_tool_use: Custom agent loop using Anthropic API10- ollama_tool_use: Custom agent loop using Ollama (fully local, no API key)11- openai_tool_use: Custom agent loop using any OpenAI-compatible server12  (OpenAI, vLLM, llama.cpp, ...) with tool calling13- claude_sdk: Claude Agent SDK (subprocess with JSON-lines IPC)14- subprocess: Generic CLI agent (Phase 4)15- opencode: OpenCode SDK (Phase 4)16"""17 18import logging19import os20from abc import ABC, abstractmethod21from dataclasses import dataclass, field22from enum import Enum23from typing import Any, Dict, Iterator, List, Optional24 25logger = logging.getLogger(__name__)26 27 28class CodingAgentEventType(str, Enum):29    """Event types emitted by coding agent backends."""30    THINKING = "thinking"31    TOOL_CALL_START = "tool_call_start"32    TOOL_CALL_END = "tool_call_end"33    TURN_END = "turn_end"34    ERROR = "error"35    COMPLETE = "complete"36 37 38@dataclass39class CodingAgentEvent:40    """Single event from a coding agent backend."""41    event_type: CodingAgentEventType42    timestamp: float = 0.043    data: Dict[str, Any] = field(default_factory=dict)44 45    def to_dict(self) -> Dict[str, Any]:46        return {47            "event_type": self.event_type.value,48            "timestamp": self.timestamp,49            "data": self.data,50        }51 52 53# Tool definitions for custom tool-use backends54CODING_TOOLS = [55    {56        "name": "Read",57        "description": "Read a file from the filesystem. Returns the file contents.",58        "input_schema": {59            "type": "object",60            "properties": {61                "file_path": {"type": "string", "description": "Absolute or relative path to the file"},62            },63            "required": ["file_path"],64        },65    },66    {67        "name": "Edit",68        "description": "Replace a specific string in a file with a new string.",69        "input_schema": {70            "type": "object",71            "properties": {72                "file_path": {"type": "string", "description": "Path to the file to edit"},73                "old_string": {"type": "string", "description": "The exact text to find and replace"},74                "new_string": {"type": "string", "description": "The replacement text"},75            },76            "required": ["file_path", "old_string", "new_string"],77        },78    },79    {80        "name": "Write",81        "description": "Create or overwrite a file with the given content.",82        "input_schema": {83            "type": "object",84            "properties": {85                "file_path": {"type": "string", "description": "Path to the file to write"},86                "content": {"type": "string", "description": "The full file content"},87            },88            "required": ["file_path", "content"],89        },90    },91    {92        "name": "Bash",93        "description": "Execute a bash command and return its output.",94        "input_schema": {95            "type": "object",96            "properties": {97                "command": {"type": "string", "description": "The command to execute"},98            },99            "required": ["command"],100        },101    },102    {103        "name": "Grep",104        "description": "Search for a pattern in files. Returns matching lines with file paths.",105        "input_schema": {106            "type": "object",107            "properties": {108                "pattern": {"type": "string", "description": "Regex pattern to search for"},109                "path": {"type": "string", "description": "Directory or file to search in"},110            },111            "required": ["pattern"],112        },113    },114    {115        "name": "Glob",116        "description": "Find files matching a glob pattern.",117        "input_schema": {118            "type": "object",119            "properties": {120                "pattern": {"type": "string", "description": "Glob pattern (e.g. '**/*.py')"},121            },122            "required": ["pattern"],123        },124    },125]126 127# Ollama-compatible tool format (OpenAI function calling style)128CODING_TOOLS_OLLAMA = [129    {130        "type": "function",131        "function": {132            "name": t["name"],133            "description": t["description"],134            "parameters": t["input_schema"],135        },136    }137    for t in CODING_TOOLS138]139 140 141def execute_tool(tool_name: str, tool_input: dict, working_dir: str) -> str:142    """Execute a coding tool in the working directory.143 144    Args:145        tool_name: Tool name (Read, Edit, Write, Bash, Grep, Glob)146        tool_input: Tool input parameters147        working_dir: Working directory for file operations148 149    Returns:150        Tool output as a string151    """152    import glob as glob_module153    import subprocess154 155    try:156        if tool_name == "Read":157            file_path = tool_input["file_path"]158            abs_path = os.path.join(working_dir, file_path) if not os.path.isabs(file_path) else file_path159            with open(abs_path, "r", encoding="utf-8", errors="replace") as f:160                return f.read()161 162        elif tool_name == "Edit":163            file_path = tool_input["file_path"]164            abs_path = os.path.join(working_dir, file_path) if not os.path.isabs(file_path) else file_path165            old_string = tool_input["old_string"]166            new_string = tool_input["new_string"]167            with open(abs_path, "r", encoding="utf-8") as f:168                content = f.read()169            if old_string not in content:170                return f"Error: old_string not found in {file_path}"171            content = content.replace(old_string, new_string, 1)172            with open(abs_path, "w", encoding="utf-8") as f:173                f.write(content)174            return "Edit applied successfully."175 176        elif tool_name == "Write":177            file_path = tool_input["file_path"]178            abs_path = os.path.join(working_dir, file_path) if not os.path.isabs(file_path) else file_path179            os.makedirs(os.path.dirname(abs_path), exist_ok=True)180            with open(abs_path, "w", encoding="utf-8") as f:181                f.write(tool_input["content"])182            return f"File written: {file_path}"183 184        elif tool_name == "Bash":185            command = tool_input["command"]186            result = subprocess.run(187                command, shell=True, capture_output=True, text=True,188                cwd=working_dir, timeout=60,189            )190            output = result.stdout191            if result.stderr:192                output += "\n" + result.stderr193            if result.returncode != 0:194                output += f"\n[exit code: {result.returncode}]"195            return output.strip() or "(no output)"196 197        elif tool_name == "Grep":198            pattern = tool_input["pattern"]199            path = tool_input.get("path", ".")200            abs_path = os.path.join(working_dir, path) if not os.path.isabs(path) else path201            result = subprocess.run(202                ["grep", "-rn", pattern, abs_path],203                capture_output=True, text=True, cwd=working_dir, timeout=30,204            )205            return result.stdout.strip() or "(no matches)"206 207        elif tool_name == "Glob":208            pattern = tool_input["pattern"]209            matches = sorted(glob_module.glob(210                os.path.join(working_dir, pattern), recursive=True211            ))212            # Make paths relative to working_dir213            rel_matches = [os.path.relpath(m, working_dir) for m in matches]214            return "\n".join(rel_matches) or "(no matches)"215 216        else:217            return f"Unknown tool: {tool_name}"218 219    except FileNotFoundError as e:220        return f"Error: File not found: {e}"221    except PermissionError as e:222        return f"Error: Permission denied: {e}"223    except subprocess.TimeoutExpired:224        return "Error: Command timed out (60s limit)"225    except Exception as e:226        return f"Error: {type(e).__name__}: {e}"227 228 229class CodingAgentBackend(ABC):230    """Abstract interface for coding agent backends."""231 232    @abstractmethod233    def start(self, task: str, working_dir: str, system_prompt: str = "") -> None:234        """Start the agent with a task description."""235        ...236 237    @abstractmethod238    def get_events(self) -> Iterator[CodingAgentEvent]:239        """Yield events as the agent works. Blocks until next event or completion."""240        ...241 242    @abstractmethod243    def pause(self) -> None:244        """Pause the agent between tool executions."""245        ...246 247    @abstractmethod248    def resume(self) -> None:249        """Resume a paused agent."""250        ...251 252    @abstractmethod253    def inject_instruction(self, text: str) -> None:254        """Send an instruction to the agent (appended as user message)."""255        ...256 257    @abstractmethod258    def stop(self) -> None:259        """Stop the agent."""260        ...261 262    @abstractmethod263    def get_conversation_history(self) -> List[Dict]:264        """Get the full conversation history."""265        ...266 267    @abstractmethod268    def get_state(self) -> str:269        """Get the current state: running, paused, completed, error."""270        ...271 272    def truncate_history(self, to_step: int) -> None:273        """Truncate conversation history to the given step (for rollback)."""274        pass  # Optional, backends that support rollback override this275 276 277# Backend registry278BACKEND_REGISTRY: Dict[str, type] = {}279 280 281def register_backend(name: str, cls: type) -> None:282    """Register a backend implementation."""283    BACKEND_REGISTRY[name] = cls284 285 286def create_backend(backend_type: str, config: dict) -> CodingAgentBackend:287    """Create a backend instance from config."""288    if backend_type not in BACKEND_REGISTRY:289        available = ", ".join(sorted(BACKEND_REGISTRY.keys()))290        raise ValueError(291            f"Unknown backend type '{backend_type}'. Available: {available}"292        )293    cls = BACKEND_REGISTRY[backend_type]294    return cls(config)295 296 297def _register_builtin_backends():298    """Register built-in backends. Called on import."""299    try:300        from .coding_agent_backends.anthropic_backend import AnthropicToolUseBackend301        register_backend("anthropic_tool_use", AnthropicToolUseBackend)302    except ImportError:303        logger.debug("Anthropic backend not available (missing anthropic package)")304 305    try:306        from .coding_agent_backends.ollama_backend import OllamaToolUseBackend307        register_backend("ollama_tool_use", OllamaToolUseBackend)308    except ImportError:309        logger.debug("Ollama backend not available")310 311    try:312        from .coding_agent_backends.openai_backend import OpenAIToolUseBackend313        register_backend("openai_tool_use", OpenAIToolUseBackend)314    except ImportError:315        logger.debug("OpenAI backend not available (missing openai package)")316 317    try:318        from .coding_agent_backends.claude_sdk_backend import ClaudeSDKBackend319        register_backend("claude_sdk", ClaudeSDKBackend)320    except ImportError:321        logger.debug("Claude SDK backend not available (missing claude-agent-sdk)")322 323 324_register_builtin_backends()325