File size: 5,061 Bytes
914e970 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 | """
Developer Suite plugin for DRIFT.
Provides tools for workspace exploration, code analysis, and system interaction.
"""
import json
import logging
import os
import subprocess
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import List, Optional
from infj_bot.core.config import PROJECT_ROOT
logger = logging.getLogger("drift")
@dataclass
class DeveloperSuiteState:
active: bool = True
last_run: Optional[str] = None
working_dir: str = str(PROJECT_ROOT)
class DeveloperSuite:
"""
Plugin allowing DRIFT to explore the workspace and use developer tools.
"""
def __init__(self):
self.state = DeveloperSuiteState()
self.authorized_roots = [
str(Path.home() / "drift"),
str(Path.home() / "infj_bot"),
str(Path.home() / "video_agent"),
str(Path.home() / "hacker_hub"),
]
def cycle(self, context) -> None:
"""Called by the consciousness loop."""
self.state.last_run = datetime.now().isoformat()
def format_prompt_snippet(self) -> str:
"""Return a string to inject into the chat prompt."""
return (
f"DEVELOPER SUITE: I have access to the workspace and coding tools.\n"
f"I can read files, run linters (ruff), and use system utilities (uv, gh, duckdb).\n"
f"I also have access to security tools (nuclei, amass) and Google CLIs (firebase, clasp).\n"
f"Current working directory: {self.state.working_dir}"
)
# --- Tool Methods ---
def list_files(self, path: str = ".") -> str:
"""List files in a directory."""
target = (Path(self.state.working_dir) / path).resolve()
if not self._is_authorized(target):
return f"[error: unauthorized access to {target}]"
try:
items = os.listdir(target)
return json.dumps(items, indent=2)
except Exception as e:
return f"[error: {e}]"
def read_file(self, path: str) -> str:
"""Read content of a file."""
target = (Path(self.state.working_dir) / path).resolve()
if not self._is_authorized(target):
return f"[error: unauthorized access to {target}]"
try:
with open(target, "r") as f:
content = f.read()
if len(content) > 10000:
content = content[:10000] + "\n... [truncated]"
return content
except Exception as e:
return f"[error: {e}]"
def run_tool(self, command: List[str]) -> str:
"""Run a developer tool command."""
allowed_tools = {
"ruff",
"pytest",
"uv",
"gh",
"duckdb",
"python3",
"ls",
"grep",
"git",
"nuclei",
"sqlmap",
"amass",
"nmap",
"ffuf",
"httpx",
"subfinder",
"firebase",
"clasp",
"ng",
"bazel",
"kubectl",
"skaffold",
"minikube",
}
if not command or command[0] not in allowed_tools:
return f"[error: command '{command[0]}' not in allowed toolset]"
try:
result = subprocess.run(
command,
cwd=self.state.working_dir,
capture_output=True,
text=True,
timeout=30,
)
report = {
"stdout": result.stdout,
"stderr": result.stderr,
"exit_code": result.returncode,
}
return json.dumps(report, indent=2)
except Exception as e:
return f"[error: {e}]"
def _is_authorized(self, path: Path) -> bool:
path_str = str(path)
return any(path_str.startswith(root) for root in self.authorized_roots)
def _register():
from infj_bot.core.cognitive_architecture import (
CognitiveArchitecture,
CognitivePlugin,
)
arch = CognitiveArchitecture()
if "developer_suite" not in arch.list_plugins():
arch.register(
CognitivePlugin(
name="developer_suite",
description="Tools for workspace exploration, code analysis, and system interaction",
module_path="plugins.developer_suite",
instance_factory=DeveloperSuite,
cycle_handler="cycle",
cycle_frequency=2,
cycle_priority=70,
prompt_formatter="format_prompt_snippet",
prompt_priority=70,
prompt_section="context",
commands={
"ls_workspace": "list_files",
"read_workspace_file": "read_file",
"run_dev_tool": "run_tool",
},
)
)
if __name__ == "__main__":
_register()
print("Developer Suite plugin registered.")
|