223 lines
8.8 KiB
Python
223 lines
8.8 KiB
Python
import os
|
|
import subprocess
|
|
import re
|
|
import glob as glob_module
|
|
|
|
# Global state to track the agent's current working directory
|
|
# This ensures that subsequent bash commands run in the correct directory
|
|
CURRENT_WORKSPACE = os.getcwd()
|
|
|
|
def get_current_workspace():
|
|
global CURRENT_WORKSPACE
|
|
return CURRENT_WORKSPACE
|
|
|
|
def set_current_workspace(path):
|
|
global CURRENT_WORKSPACE
|
|
CURRENT_WORKSPACE = path
|
|
os.chdir(path)
|
|
|
|
schema_read_file = {
|
|
"type": "function",
|
|
"function": {
|
|
"name" : "read_file",
|
|
"description" : "Read the contents of a file. Returns the file content with line numbers prefixed.",
|
|
"parameters" : {
|
|
"type" : "object",
|
|
"properties" : {
|
|
"path": {"type": "string", "description": "Absolute path to the file to read"}
|
|
},
|
|
"required" : ["path"]
|
|
}
|
|
}
|
|
}
|
|
|
|
def read_file(path):
|
|
try:
|
|
# Use absolute path if it's relative to CURRENT_WORKSPACE
|
|
full_path = os.path.join(get_current_workspace(), path) if not os.path.isabs(path) else path
|
|
with open(full_path, 'r', encoding='utf-8', errors='replace') as f:
|
|
lines = f.readlines()
|
|
return ''.join(f"{i+1}: {line}" for i, line in enumerate(lines))
|
|
except Exception as e:
|
|
return f"Error reading file: {str(e)}"
|
|
|
|
schema_write_file = {
|
|
"type": "function",
|
|
"function": {
|
|
"name" : "write_file",
|
|
"description" : "Write content to a file. Overwrites the file if it exists.",
|
|
"parameters" : {
|
|
"type" : "object",
|
|
"properties" : {
|
|
"path": {"type": "string", "description": "Absolute path to the file to write"},
|
|
"content": {"type": "string", "description": "Content to write to the file"}
|
|
},
|
|
"required" : ["path", "content"]
|
|
}
|
|
}
|
|
}
|
|
|
|
def write_file(path, content):
|
|
try:
|
|
full_path = os.path.join(get_current_workspace(), path) if not os.path.isabs(path) else path
|
|
with open(full_path, 'w', encoding='utf-8') as f:
|
|
f.write(content)
|
|
return f"Successfully wrote to {full_path}"
|
|
except Exception as e:
|
|
return f"Error writing file: {str(e)}"
|
|
|
|
schema_edit_file = {
|
|
"type": "function",
|
|
"function": {
|
|
"name" : "edit_file",
|
|
"description" : "Replace old_string with new_string in a file. If old_string is not found, returns error. If multiple matches, replaces all.",
|
|
"parameters" : {
|
|
"type" : "object",
|
|
"properties" : {
|
|
"path": {"type": "string", "description": "Absolute path to the file to edit"},
|
|
"old_string": {"type": "string", "description": "String to replace"},
|
|
"new_string": {"type": "string", "description": "Replacement string"}
|
|
},
|
|
"required" : ["path", "old_string", "new_string"]
|
|
}
|
|
}
|
|
}
|
|
|
|
def edit_file(path, old_string, new_string):
|
|
try:
|
|
full_path = os.path.join(get_current_workspace(), path) if not os.path.isabs(path) else path
|
|
with open(full_path, 'r', encoding='utf-8', errors='replace') as f:
|
|
content = f.read()
|
|
if old_string not in content:
|
|
return f"Error: old_string not found in {full_path}"
|
|
new_content = content.replace(old_string, new_string)
|
|
with open(full_path, 'w', encoding='utf-8') as f:
|
|
f.write(new_content)
|
|
return f"Successfully edited {full_path}"
|
|
except Exception as e:
|
|
return f"Error editing file: {str(e)}"
|
|
|
|
schema_run_bash = {
|
|
"type": "function",
|
|
"function": {
|
|
"name" : "run_bash",
|
|
"description" : "Run a bash command. Returns stdout, stderr, and return code. Commands are executed relative to the current workspace.",
|
|
"parameters" : {
|
|
"type" : "object",
|
|
"properties" : {
|
|
"command": {"type": "string", "description": "Bash command to execute"}
|
|
},
|
|
"required" : ["command"]
|
|
}
|
|
}
|
|
}
|
|
|
|
def run_bash(command):
|
|
try:
|
|
# Prepend 'cd' to the command to ensure it runs in the correct workspace
|
|
full_command = f"cd {get_current_workspace()} && {command}"
|
|
result = subprocess.run(full_command, shell=True, capture_output=True, text=True, timeout=30)
|
|
return f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}\nreturn code: {result.returncode}"
|
|
except Exception as e:
|
|
return f"Error running command: {str(e)}"
|
|
|
|
schema_search_code = {
|
|
"type": "function",
|
|
"function": {
|
|
"name": "search_code",
|
|
"description": "Search for files or content in files. Use type 'glob' to find files matching a pattern, 'content' to search file contents for a regex pattern.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"pattern": {"type": "string", "description": "Glob pattern (for type 'glob') or regex pattern (for type 'content')"},
|
|
"path": {"type": "string", "description": "Directory to search in (default: current working directory)", "default": "."},
|
|
"search_type": {"type": "string", "enum": ["glob", "content"], "description": "Type of search: 'glob' for files, 'content' for file contents"}
|
|
},
|
|
"required": ["pattern", "search_type"]
|
|
}
|
|
}
|
|
}
|
|
|
|
def search_code(pattern, search_type, path="."):
|
|
try:
|
|
# Resolve path relative to workspace
|
|
search_path = os.path.join(get_current_workspace(), path) if not os.path.isabs(path) else path
|
|
|
|
if search_type == "glob":
|
|
files = glob_module.glob(f"{search_path}/**/{pattern}", recursive=True)
|
|
return "\n".join(files) if files else "No files found"
|
|
elif search_type == "content":
|
|
results = []
|
|
for root, dirs, files in os.walk(search_path):
|
|
for file in files:
|
|
file_path = os.path.join(root, file)
|
|
try:
|
|
with open(file_path, 'r', encoding='utf-8', errors='replace') as f:
|
|
for i, line in enumerate(f.readlines(), 1):
|
|
if re.search(pattern, line):
|
|
results.append(f"{file_path}:{i}: {line.strip()}")
|
|
except:
|
|
continue
|
|
return "\n".join(results[:50]) if results else "No matches found"
|
|
else:
|
|
return "Invalid search_type. Use 'glob' or 'content'."
|
|
except Exception as e:
|
|
return f"Error searching: {str(e)}"
|
|
|
|
schema_git_operation = {
|
|
"type": "function",
|
|
"function": {
|
|
"name": "git_operation",
|
|
"description": "Run a git command. Pass the git arguments as a list (e.g., ['status', '--short'] for 'git status --short'). "
|
|
"POLICY: Never run 'git add' or 'git commit' without explicit user permission. "
|
|
"Safe to run without asking: git status, git diff, git log. "
|
|
"Always ask first before committing.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"args": {"type": "array", "items": {"type": "string"}, "description": "List of git command arguments (without 'git' prefix)"}
|
|
},
|
|
"required": ["args"]
|
|
}
|
|
}
|
|
}
|
|
|
|
def git_operation(args):
|
|
try:
|
|
# Execute git command from the current workspace
|
|
full_command = ["git", *args]
|
|
result = subprocess.run(full_command, cwd=get_current_workspace(), capture_output=True, text=True, timeout=10)
|
|
return f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}\nreturn code: {result.returncode}"
|
|
except Exception as e:
|
|
return f"Error running git command: {str(e)}"
|
|
|
|
schema_workspace_change = {
|
|
"type": "function",
|
|
"function": {
|
|
"name": "workspace_change",
|
|
"description": "Change the current working directory of the agent session. Supports tilde (~) expansion for the user home directory (/home/aji/).",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"path": {"type": "string", "description": "The target directory path to change to."}
|
|
},
|
|
"required": ["path"]
|
|
}
|
|
}
|
|
}
|
|
|
|
def workspace_change(path):
|
|
try:
|
|
if path.startswith("~"):
|
|
target_path = os.path.expanduser(path)
|
|
else:
|
|
target_path = os.path.abspath(path)
|
|
|
|
if os.path.exists(target_path) and os.path.isdir(target_path):
|
|
set_current_workspace(target_path)
|
|
return f"Successfully changed workspace to: {target_path}"
|
|
else:
|
|
return f"Error: The path {target_path} does not exist or is not a directory."
|
|
except Exception as e:
|
|
return f"Error changing workspace: {str(e)}"
|