Coverage for src / osiris_cli / tools.py: 0%
1210 statements
« prev ^ index » next coverage.py v7.13.0, created at 2025-12-31 05:01 +0200
1import ast
2import asyncio
3import fnmatch
4import json
5import os
6import pathspec
7import re
8import shlex
9import shutil
10import subprocess
11import xml.etree.ElementTree as ET
12from pathlib import Path
13from typing import Any, Dict, List, Optional, Tuple
14from urllib.parse import quote_plus, urljoin, urlparse
16import requests
17from bs4 import BeautifulSoup
18from rich.console import Console
19from rich.panel import Panel
20from rich.syntax import Syntax
22from .activity_monitor import activity_monitor, monitor_tool_call
23from .config import CONFIG_DIR
24from .session import session
25from .shell_session import persistent_shell
26from .neural_tools import neural_tools
28StructuredResult = Dict[str, Any]
31def normalize_tool_result(tool_name: str, raw_result: Any) -> StructuredResult:
32 """Normalize raw tool output into the shared StructuredResult contract."""
33 def coerce_text(value: Any) -> str:
34 return "" if value is None else str(value)
36 status = "success"
37 message = ""
38 stdout = ""
39 stderr = ""
40 exit_code = None
41 meta: Dict[str, Any] = {}
43 if isinstance(raw_result, dict):
44 status = raw_result.get("status") or status
45 message = coerce_text(raw_result.get("message") or raw_result.get("summary") or "")
46 stdout = coerce_text(raw_result.get("stdout"))
47 stderr = coerce_text(raw_result.get("stderr"))
48 exit_code = raw_result.get("exit_code")
49 meta = dict(raw_result.get("meta") or {})
50 for key in ("pid", "url", "detach_hint", "command", "duration"):
51 if key in raw_result and raw_result.get(key) is not None:
52 meta.setdefault(key, raw_result[key])
53 else:
54 value = coerce_text(raw_result)
55 message = value
56 stdout = value
57 stderr = ""
59 if exit_code not in (None, 0) and status not in {"running", "timeout"}:
60 status = "error"
62 return {
63 "tool": tool_name,
64 "status": status,
65 "message": message,
66 "stdout": stdout,
67 "stderr": stderr,
68 "exit_code": exit_code,
69 "meta": meta,
70 }
72from .tools_registry import tools_registry
74# GLOBAL IGNORE PATTERNS
75IGNORE_DIRS = {".git", "node_modules", ".venv", "__pycache__", "Library", "dist", "build"}
77class ToolRegistry:
78 """Unified bridge to the global tools_registry."""
79 def __init__(self):
80 self.home_dir = Path.home().resolve()
81 self._ignore_cache = {}
82 # We now use the global tools_registry instance
83 self._register_core_tools()
85 def register(self, name: str, func, description: str, parameters: Dict, category: str = "general"):
86 """Register a tool in the unified global registry."""
87 tools_registry.register(
88 name=name,
89 description=description,
90 parameters=parameters,
91 function=func,
92 category=category
93 )
95 def _replace_content(self, path: str, old_str: str, new_str: str) -> str:
96 """Surgically replace content in a file."""
97 if not self._is_safe_path(path):
98 return "Error: Access denied."
100 p = Path(path)
101 if not p.exists():
102 return f"Error: File {path} not found."
104 try:
105 content = p.read_text(encoding='utf-8')
107 # Check for unique match
108 count = content.count(old_str)
109 if count == 0:
110 return f"Error: Could not find the exact string to replace in {path}. Ensure whitespace and indentation match exactly."
111 if count > 1:
112 return f"Error: Found {count} occurrences of the target string. Please provide more context to make the match unique."
114 new_content = content.replace(old_str, new_str)
115 p.write_text(new_content, encoding='utf-8')
117 # Return a small diff-like summary
118 return f"Successfully updated {path}. Replaced 1 occurrence."
119 except Exception as e:
120 return f"Error: {str(e)}"
122 def _get_cpp_symbols(self, path: str) -> str:
123 """Regex-based C++ symbol extraction (Classes, Structs, Methods)."""
124 p = Path(path)
125 if not p.exists():
126 return f"Error: File {path} not found."
128 try:
129 content = p.read_text(encoding="utf-8")
130 symbols = []
132 # 1. Classes & Structs
133 class_matches = re.finditer(r'(class|struct)\s+(\w+)\s*[:{]', content)
134 for m in class_matches:
135 symbols.append(f"{m.group(1).upper()}: {m.group(2)}")
137 # 2. Functions/Methods (Simple signature regex)
138 # Matches: type name(args) { or ;
139 func_pattern = r'([\w:*&<>]+)\s+([\w:~]+)\s*\(([^)]*)\)\s*[{;]'
140 func_matches = re.finditer(func_pattern, content)
141 for m in func_matches:
142 ret_type, name, args = m.groups()
143 # Filter out keywords that look like functions
144 if name in {"if", "while", "for", "switch", "catch", "return"}: continue
145 symbols.append(f" def {name}({args}) -> {ret_type}")
147 if not symbols:
148 return "No C++ symbols detected."
150 return "C++ Symbols Found:\n" + "\n".join(symbols[:300])
151 except Exception as e:
152 return f"Error parsing C++: {str(e)}"
154 def _generate_repo_map(self, path: str = ".", include_private: bool = False) -> str:
155 """Generate a high-density map of project symbols (classes/functions)."""
156 root = Path(path or ".").resolve()
157 map_lines = []
159 for p in root.rglob("*.py"):
160 if self._is_ignored(root, p):
161 continue
163 rel_path = p.relative_to(root)
164 try:
165 content = p.read_text(encoding="utf-8")
166 tree = ast.parse(content)
168 file_symbols = []
169 for node in tree.body:
170 if isinstance(node, ast.ClassDef):
171 if not include_private and node.name.startswith("_"): continue
172 file_symbols.append(f"class {node.name}")
173 for item in node.body:
174 if isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef)):
175 if not include_private and item.name.startswith("_"): continue
176 file_symbols.append(f" def {item.name}")
177 elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
178 if not include_private and node.name.startswith("_"): continue
179 file_symbols.append(f"def {node.name}")
181 if file_symbols:
182 map_lines.append(f"### {rel_path}")
183 map_lines.extend(file_symbols)
184 map_lines.append("")
185 except:
186 continue
188 return "\n".join(map_lines[:500]) or "No symbols found."
190 def _search_archival_memory(self, query: str) -> str:
191 """Search past session summaries for relevant context."""
192 memory_file = Path(CONFIG_DIR) / "archival_memory.jsonl"
193 if not memory_file.exists():
194 return "No archival memory found. This is a new agent."
196 results = []
197 try:
198 with open(memory_file, "r") as f:
199 for line in f:
200 if query.lower() in line.lower():
201 results.append(line.strip())
203 if not results:
204 return f"No matches found for '{query}' in archival memory."
206 return "\n".join(results[-10:]) # Return last 10 matches
207 except Exception as e:
208 return f"Error searching memory: {str(e)}"
210 def _scan_code_safety(self, path: str) -> str:
211 """AST-based security scan for dangerous Python patterns."""
212 p = Path(path)
213 if not p.exists():
214 return f"Error: File {path} not found."
216 try:
217 code = p.read_text(encoding="utf-8")
218 tree = ast.parse(code)
220 dangerous_calls = []
221 # Patterns from your provided snippet: os.system, eval, etc.
222 restricted = {
223 "os.system", "os.popen", "os.spawn", "subprocess.call",
224 "subprocess.run", "subprocess.Popen", "eval", "exec",
225 "__import__", "pickle.load"
226 }
228 for node in ast.walk(tree):
229 if isinstance(node, ast.Call):
230 func_name = ""
231 if isinstance(node.func, ast.Attribute):
232 if isinstance(node.func.value, ast.Name):
233 func_name = f"{node.func.value.id}.{node.func.attr}"
234 elif isinstance(node.func, ast.Name):
235 func_name = node.func.id
237 if func_name in restricted:
238 dangerous_calls.append({
239 "function": func_name,
240 "line": node.lineno,
241 "col": node.col_offset
242 })
244 if not dangerous_calls:
245 return "✅ No common dangerous patterns detected in AST."
247 report = ["⚠️ SECURITY ALERT: Dangerous patterns found!"]
248 for call in dangerous_calls:
249 report.append(f"- Found '{call['function']}' at line {call['line']}")
251 return "\n".join(report)
252 except Exception as e:
253 return f"Error scanning safety: {str(e)}"
255 def _learn_skill(self, name: str, code: str, description: str, parameters: Dict[str, Any], validation_code: str) -> str:
256 """Create and VALIDATE a new persistent tool from provided Python code."""
257 skills_dir = Path(CONFIG_DIR) / "skills"
258 skills_dir.mkdir(parents=True, exist_ok=True)
260 file_path = skills_dir / f"{name}.py"
261 test_path = skills_dir / f"test_{name}.py"
263 # Build the skill module content
264 skill_template = f"""
265import json
266import os
267import subprocess
268from typing import Any, Dict
270def {name}(**kwargs):
271 \"\"\" {description} \"\"\"
272{code}
274def register_skill(registry):
275 registry.register(
276 name="{name}",
277 description="{description}",
278 parameters={json.dumps(parameters)},
279 function={name},
280 category="skill"
281 )
282"""
283 try:
284 # 1. Write temp file for validation
285 file_path.write_text(skill_template, encoding="utf-8")
287 # 2. Run Validation Script (isolated)
288 # The agent provides code that imports and calls the function
289 full_test_code = f"from {name} import {name}\n{validation_code}"
290 test_path.write_text(full_test_code, encoding="utf-8")
292 # Run test in sub-process
293 env = os.environ.copy()
294 env["PYTHONPATH"] = str(skills_dir)
295 result = subprocess.run([sys.executable, str(test_path)], capture_output=True, text=True, env=env)
297 if result.returncode != 0:
298 # Cleanup failed attempt
299 file_path.unlink(missing_ok=True)
300 test_path.unlink(missing_ok=True)
301 return f"Skill Validation FAILED: The provided test code failed.\nSTDERR: {result.stderr}"
303 # 3. Success: Register
304 tools_registry.load_skills_from_dir(str(skills_dir))
305 test_path.unlink(missing_ok=True) # Cleanup test, keep skill
306 return f"SUCCESS: New skill '{name}' verified and integrated into your body."
308 except Exception as e:
309 return f"Error learning skill: {str(e)}"
311 def _check_container_engine(self) -> str:
312 """Check if Docker or Finch is installed and running."""
313 engines = [
314 ("docker", ["docker", "--version"]),
315 ("finch", ["finch", "--version"])
316 ]
317 results = []
318 for name, cmd in engines:
319 try:
320 proc = subprocess.run(cmd, capture_output=True, text=True, check=False)
321 if proc.returncode == 0:
322 results.append(f"✅ {name.upper()} is active: {proc.stdout.strip()}")
323 except:
324 continue
326 return "\n".join(results) if results else "❌ No container engine (Docker/Finch) detected in PATH."
328 def _build_container_image(self, dockerfile_path: str, context_path: str = ".", tag: str = "latest") -> str:
329 """Build a container image using Docker or Finch."""
330 # Detect engine
331 engine = "docker" if shutil.which("docker") else "finch" if shutil.which("finch") else None
332 if not engine:
333 return "Error: No container engine found (Docker/Finch)."
335 if not self._is_safe_path(dockerfile_path) or not self._is_safe_path(context_path):
336 return "Error: Access denied to paths."
338 cmd = [engine, "build", "-f", dockerfile_path, "-t", tag, context_path]
339 try:
340 result = subprocess.run(cmd, capture_output=True, text=True, timeout=600)
341 if result.returncode == 0:
342 return f"Successfully built image '{tag}' using {engine}.\n{result.stdout}"
343 return f"Failed to build image using {engine}:\n{result.stderr}"
344 except Exception as e:
345 return f"Error during container build: {str(e)}"
347 def _finch_build_image(self, dockerfile_path: str, tag: str, context_path: str = ".") -> str:
348 """High-end container build using Finch (2025 DevOps Spec)."""
349 if not shutil.which("finch"):
350 return "Error: Finch is not installed on this system."
352 # 1. Verify Finch VM status
353 try:
354 vm_check = subprocess.run(["finch", "vm", "status"], capture_output=True, text=True, check=False)
355 if "Running" not in vm_check.stdout:
356 return "Error: Finch VM is not running. Start it with 'finch vm start'."
357 except: pass
359 # 2. Path Validation
360 if not self._is_safe_path(dockerfile_path):
361 return "Error: Access denied."
363 # 3. Build Command (Brutalist Hardening)
364 # We force linux/amd64 for ECR/Dev compatibility as seen in your snippets
365 cmd = [
366 "finch", "build",
367 "--platform", "linux/amd64",
368 "-f", str(dockerfile_path),
369 "-t", tag,
370 str(context_path)
371 ]
373 # Check if tag is an ECR reference (high-end heuristic)
374 if "dkr.ecr" in tag:
375 console.print(f" [dim]DevOps › Detected ECR destination: {tag}[/dim]")
377 try:
378 result = subprocess.run(cmd, capture_output=True, text=True, timeout=600)
379 if result.returncode == 0:
380 return f"SUCCESS: Image '{tag}' built natively with Finch.\nSTDOUT: {result.stdout[-500:]}"
381 return f"FAILED: Finch build exited {result.returncode}.\nSTDERR: {result.stderr}"
382 except Exception as e:
383 return f"FATAL: Finch bridge error: {str(e)}"
385 def _find_tests(self, source_path: str) -> str:
386 """Heuristically find relevant test files for a source file."""
387 src_p = Path(source_path)
388 src_name = src_p.stem
389 root = Path.cwd()
391 # Look in common test locations
392 test_dirs = [root / "tests", root / "test", src_p.parent / "tests", src_p.parent]
394 candidates = []
395 patterns = [f"test_{src_name}.py", f"{src_name}_test.py", f"test_{src_name}.js", f"test_{src_name}.ts"]
397 for d in test_dirs:
398 if not d.exists(): continue
399 for p in patterns:
400 # Direct match
401 match = d / p
402 if match.exists(): candidates.append(str(match))
403 # Recursive match
404 for found in d.rglob(p):
405 candidates.append(str(found))
407 if not candidates:
408 return f"No direct tests found for {source_path}. You may need to run the full suite or write a new test."
410 return "Relevant tests found:\n" + "\n".join(list(set(candidates)))
412 def _verify_health(self) -> str:
413 """Sovereign Self-Heal: Check for critical binary dependencies."""
414 deps = ["git", "npm", "node", "python3", "docker", "finch", "googler", "rg"]
415 report = ["### ENVIRONMENT HEALTH REPORT"]
416 missing = []
418 for d in deps:
419 path = shutil.which(d)
420 if path:
421 report.append(f"- ✅ {d.upper()}: Found at {path}")
422 else:
423 report.append(f"- ❌ {d.upper()}: MISSING")
424 missing.append(d)
426 if missing:
427 report.append(f"\n⚠️ CRITICAL GAP: {len(missing)} dependencies missing. Agent should use 'run_shell_command' to install or 'learn_skill' to bypass.")
429 return "\n".join(report)
431 def _register_core_tools(self):
432 # Neural & Cognitive Tools
433 self.register(
434 name="scratchpad",
435 func=neural_tools.scratchpad,
436 description="Manage a persistent project-level scratchpad for planning and temporary notes.",
437 parameters={
438 "type": "object",
439 "properties": {
440 "action": {"type": "string", "enum": ["read", "write", "append"]},
441 "content": {"type": "string", "description": "The content to write or append."}
442 },
443 "required": ["action"]
444 },
445 category="cognitive"
446 )
447 self.register(
448 name="dump_state",
449 func=neural_tools.dump_state,
450 description="Persist internal neural model state for debugging or migration.",
451 parameters={
452 "type": "object",
453 "properties": {
454 "state_obj": {"type": "object", "description": "The state dictionary to persist."},
455 "label": {"type": "string", "description": "A unique label for this state dump."}
456 },
457 "required": ["state_obj"]
458 },
459 category="cognitive"
460 )
461 self.register(
462 name="behavioral_cloning",
463 func=neural_tools.behavioral_cloning,
464 description="Archive current interaction sequence for model alignment and cloning.",
465 parameters={
466 "type": "object",
467 "properties": {
468 "mission_id": {"type": "string", "description": "Identifier for the current task."},
469 "success": {"type": "boolean", "description": "Whether the mission was accomplished."}
470 },
471 "required": ["mission_id"]
472 },
473 category="cognitive"
474 )
475 self.register(
476 name="pad_outputs",
477 func=neural_tools.pad_outputs,
478 description="Standardized padding for neural sequence outputs (tensors or token lists).",
479 parameters={
480 "type": "object",
481 "properties": {
482 "outputs": {"type": "array", "items": {"type": "array"}, "description": "List of sequences to pad."},
483 "max_length": {"type": "integer", "description": "Target length for all sequences."},
484 "pad_value": {"type": "integer", "description": "Value to use for padding (default 0)."}
485 },
486 "required": ["outputs", "max_length"]
487 },
488 category="cognitive"
489 )
491 # ... (keep existing tools) ...
492 self.register(
493 name="check_container_engine",
494 func=self._check_container_engine,
495 description="Check if Docker or Finch is installed and running. Use this before attempting to build containers.",
496 parameters={"type": "object", "properties": {}}
497 )
498 self.register(
499 name="resolve_logic_symbol",
500 func=self._resolve_symbol,
501 description="Find the closest functional match for a class or function name in the project. Use this to resolve NameErrors or confirm symbol nomenclature.",
502 parameters={
503 "type": "object",
504 "properties": {
505 "symbol_name": {"type": "string", "description": "The name of the missing or ambiguous symbol."},
506 "path": {"type": "string", "description": "Search root path."}
507 },
508 "required": ["symbol_name"]
509 },
510 category="sensory"
511 )
512 self.register(
513 name="collect_env_grounding",
514 func=self._collect_env_grounding,
515 description="Gather high-fidelity system context (OS version, Shell, User, Binaries). Use this at the start of a mission to ground your actions in reality.",
516 parameters={"type": "object", "properties": {}},
517 category="sensory"
518 )
519 self.register(
520 name="map_external_resources",
521 func=self._map_resources,
522 description="Identify external dependencies (API keys, DB urls, endpoints) by scanning configuration and source files. Use this to orient yourself before interacting with external systems.",
523 parameters={"type": "object", "properties": {"path": {"type": "string"}}},
524 category="sensory"
525 )
526 self.register(
527 name="handoff_to_branch",
528 func=self._handoff,
529 description="Transfer the current mission to a new specialized branch.",
530 parameters={
531 "type": "object",
532 "properties": {
533 "target_branch": {"type": "string", "description": "Name of the target branch."},
534 "mission_summary": {"type": "string", "description": "Condensed state and goals."}
535 },
536 "required": ["target_branch", "mission_summary"]
537 },
538 category="swarm"
539 )
540 self.register(
541 name="create_artifact",
542 func=self._create_artifact,
543 description="Create a high-value deliverable (Code, Diagram, or Report).",
544 parameters={
545 "type": "object",
546 "properties": {
547 "identifier": {"type": "string"},
548 "title": {"type": "string"},
549 "type": {"type": "string"},
550 "content": {"type": "string"}
551 },
552 "required": ["identifier", "title", "type", "content"]
553 },
554 category="motor"
555 )
556 self.register(
557 name="run_tests",
558 func=self._run_tests,
559 description="Auto-detect and run tests in the current project.",
560 parameters={"type": "object", "properties": {"path": {"type": "string"}}},
561 category="sensory"
562 )
563 self.register(
564 name="search_documentation",
565 func=self._search_docs,
566 description="Search high-quality technical documentation (MDN, StackOverflow, GitHub). Use this to find specific code examples or API usage patterns.",
567 parameters={
568 "type": "object",
569 "properties": {
570 "query": {"type": "string", "description": "The technical topic or error message to search for."}
571 },
572 "required": ["query"]
573 },
574 category="research"
575 )
576 self.register(
577 name="verify_environment_health",
578 func=self._verify_health,
579 description="Check for critical system dependencies (git, npm, docker, etc.). Use this to diagnose environment-level failures.",
580 parameters={"type": "object", "properties": {}},
581 category="sensory"
582 )
583 self.register(
584 name="find_relevant_tests",
585 func=self._find_tests,
586 description="Find test files associated with a specific source file. Use this to fulfill your 'Mandatory Verification' mandate after modifying code.",
587 parameters={
588 "type": "object",
589 "properties": {
590 "source_path": {"type": "string", "description": "The path to the file you modified."}
591 },
592 "required": ["source_path"]
593 },
594 category="sensory"
595 )
596 self.register(
597 name="finch_build_image",
598 func=self._finch_build_image,
599 description="Build a container image using AWS Finch. Optimized for linux/amd64 and ECR workflows.",
600 parameters={
601 "type": "object",
602 "properties": {
603 "dockerfile_path": {"type": "string", "description": "Path to the Dockerfile."},
604 "tag": {"type": "string", "description": "Tag for the resulting image."},
605 "context_path": {"type": "string", "description": "Build context root (default '.')"}
606 },
607 "required": ["dockerfile_path", "tag"]
608 },
609 category="devops"
610 )
611 self.register(
612 name="build_container_image",
613 func=self._build_container_image,
614 description="Build a container image using Docker or Finch. Requires a Dockerfile.",
615 parameters={
616 "type": "object",
617 "properties": {
618 "dockerfile_path": {"type": "string", "description": "Path to the Dockerfile."},
619 "context_path": {"type": "string", "description": "Build context directory (default '.')"},
620 "tag": {"type": "string", "description": "Tag for the image."}
621 },
622 "required": ["dockerfile_path"]
623 },
624 category="devops"
625 )
626 self.register(
627 name="learn_skill",
628 func=self._learn_skill,
629 description="Acquire a new permanent capability by writing a Python tool. You MUST provide a validation script to prove the tool works.",
630 parameters={
631 "type": "object",
632 "properties": {
633 "name": {"type": "string", "description": "The unique name of the new tool (e.g. 'extract_financials')."},
634 "description": {"type": "string", "description": "A clear explanation of what the tool does."},
635 "code": {"type": "string", "description": "The Python body of the function. Use indentation correctly."},
636 "parameters": {"type": "object", "description": "OpenAI-style parameters schema for the new tool."},
637 "validation_code": {"type": "string", "description": "A Python script that imports and tests your new function. Must exit with 0."}
638 },
639 "required": ["name", "description", "code", "parameters", "validation_code"]
640 },
641 category="motor"
642 )
643 self.register(
644 name="scan_code_safety",
645 func=self._scan_code_safety,
646 description="Scan a Python file for dangerous patterns like 'os.system' or 'eval'. Use this before executing unknown code.",
647 parameters={
648 "type": "object",
649 "properties": {
650 "path": {"type": "string", "description": "Path to the Python file."}
651 },
652 "required": ["path"]
653 }
654 )
655 self.register(
656 name="search_archival_memory",
657 func=self._search_archival_memory,
658 description="Search through the history of past sessions and summarized interactions. Use this when the user refers to a previous task you don't remember.",
659 parameters={
660 "type": "object",
661 "properties": {
662 "query": {"type": "string", "description": "The search term or topic."}
663 },
664 "required": ["query"]
665 }
666 )
667 self.register(
668 name="get_cpp_symbols",
669 func=self._get_cpp_symbols,
670 description="Parse a C++ file and return a list of classes, structs, and methods. Use this to understand C++ logic signatures without reading the full file.",
671 parameters={
672 "type": "object",
673 "properties": {
674 "path": {"type": "string", "description": "Path to the C++ file."}
675 },
676 "required": ["path"]
677 }
678 )
679 self.register(
680 name="generate_repo_map",
681 func=self._generate_repo_map,
682 description="Generate a compressed map of the entire project's Python symbols (classes and functions). Use this to orient yourself in a new codebase.",
683 parameters={
684 "type": "object",
685 "properties": {
686 "path": {"type": "string", "description": "Root path to scan. Default '.'"},
687 "include_private": {"type": "boolean", "description": "Include symbols starting with '_'. Default false."}
688 }
689 }
690 )
691 self.register(
692 name="replace_content",
693 func=self._replace_content,
694 description="Surgically replace a specific string in a file. Provide enough context in 'old_str' to ensure it is unique within the file.",
695 parameters={
696 "type": "object",
697 "properties": {
698 "path": {"type": "string", "description": "The path to the file."},
699 "old_str": {"type": "string", "description": "The exact string to be replaced."},
700 "new_str": {"type": "string", "description": "The new string to replace it with."}
701 },
702 "required": ["path", "old_str", "new_str"]
703 }
704 )
705 self.register(
706 name="run_shell_command",
707 func=self._run_shell,
708 description="Execute a shell command on the local system (Linux). Use this to list files, run scripts, install packages, etc.",
709 parameters={
710 "type": "object",
711 "properties": {
712 "command": {"type": "string", "description": "The command to execute."},
713 "shell": {"type": "boolean", "description": "Run the command through a shell interpreter (default: true)."},
714 "timeout_sec": {"type": "integer", "description": "Maximum runtime in seconds (default: 120)."},
715 "long_running": {"type": "boolean", "description": "Treat as a long-running command (no timeout)."},
716 "detach": {"type": "boolean", "description": "Detach the command and return immediately with PID information."}
717 },
718 "required": ["command"]
719 }
720 )
721 self.register(
722 name="read_file",
723 func=self._read_file,
724 description="Read the contents of a file. Supports reading specific line ranges for large files.",
725 parameters={
726 "type": "object",
727 "properties": {
728 "path": {"type": "string", "description": "The absolute or relative path to the file."},
729 "start_line": {"type": "integer", "description": "Optional: 1-based start line number."},
730 "end_line": {"type": "integer", "description": "Optional: 1-based end line number."}
731 },
732 "required": ["path"]
733 }
734 )
735 self.register(
736 name="write_file",
737 func=self._write_file,
738 description="Write content to a file. Supports overwrite and diff modes.",
739 parameters={
740 "type": "object",
741 "properties": {
742 "path": {"type": "string", "description": "The path to the file."},
743 "content": {"type": "string", "description": "The text content to write."},
744 "mode": {"type": "string", "description": "Write mode: 'overwrite' or 'diff'. Default: 'overwrite'."}
745 },
746 "required": ["path", "content"]
747 }
748 )
749 self.register(
750 name="list_directory",
751 func=self._list_dir,
752 description="List files and directories in a given path.",
753 parameters={
754 "type": "object",
755 "properties": {
756 "path": {"type": "string", "description": "The directory path. Defaults to current directory."}
757 }
758 }
759 )
760 self.register(
761 name="get_file_tree",
762 func=self._get_tree,
763 description="Get a recursive tree view of the directory structure. Useful for understanding project layout.",
764 parameters={
765 "type": "object",
766 "properties": {
767 "path": {"type": "string", "description": "The root path. Defaults to current directory."},
768 "max_depth": {"type": "integer", "description": "Max depth to traverse. Default 3."}
769 }
770 }
771 )
772 self.register(
773 name="web_search",
774 func=self._web_search,
775 description="Search the internet using a CLI backend with structured fallbacks.",
776 parameters={
777 "type": "object",
778 "properties": {
779 "query": {"type": "string", "description": "The search query."},
780 "num_results": {"type": "integer", "description": "Number of results to return. Default 5."}
781 },
782 "required": ["query"]
783 }
784 )
785 self.register(
786 name="read_url",
787 func=self._read_url,
788 description="Fetch and extract text content from a URL (Web Scraping).",
789 parameters={
790 "type": "object",
791 "properties": {
792 "url": {"type": "string", "description": "The URL to visit."}
793 },
794 "required": ["url"]
795 }
796 )
797 self.register(
798 name="site_report",
799 func=self._site_report,
800 description="Crawl a site (depth-limited) and return a structured markdown report with sources.",
801 parameters={
802 "type": "object",
803 "properties": {
804 "url": {"type": "string", "description": "The starting URL or domain."},
805 "max_pages": {"type": "integer", "description": "Max pages to crawl (default 6)."},
806 "max_depth": {"type": "integer", "description": "Max crawl depth (default 2)."}
807 },
808 "required": ["url"]
809 }
810 )
811 self.register(
812 name="search_codebase",
813 func=self._search_codebase,
814 description="Search for a string or pattern in the current directory (recursive). Useful for finding where functions/classes are defined.",
815 parameters={
816 "type": "object",
817 "properties": {
818 "pattern": {"type": "string", "description": "The regex or string to search for."},
819 "path": {"type": "string", "description": "The root path. Defaults to current directory."}
820 },
821 "required": ["pattern"]
822 }
823 )
824 # --- NEW TOOLS (Wave 2) ---
825 self.register(
826 name="get_lsp_diagnostics",
827 func=self._get_lsp_diagnostics,
828 description="Run static analysis (pyright/ruff/flake8) to find potential errors, type mismatches, or stylistic issues in the project.",
829 parameters={
830 "type": "object",
831 "properties": {
832 "path": {"type": "string", "description": "Path to scan. Defaults to current directory."}
833 }
834 }
835 )
836 self.register(
837 name="delegate_to_subagent",
838 func=self._delegate_subagent,
839 description="Delegate a complex sub-task to a specialized sub-agent. The sub-agent will work in the background and return a summary of its work.",
840 parameters={
841 "type": "object",
842 "properties": {
843 "task": {"type": "string", "description": "Comprehensive task description for the sub-agent."},
844 "agent_name": {"type": "string", "description": "Optional: Name for the sub-agent (affects its memory). Defaults to 'sub-agent'."}
845 },
846 "required": ["task"]
847 }
848 )
849 self.register(
850 name="get_code_symbols",
851 func=self._get_symbols,
852 description="Parse a Python file and return a list of its classes, functions, and methods (AST analysis). Use this to understand code structure without reading the whole file.",
853 parameters={
854 "type": "object",
855 "properties": {
856 "path": {"type": "string", "description": "Path to the Python file."}
857 },
858 "required": ["path"]
859 }
860 )
861 self.register(
862 name="check_syntax",
863 func=self._check_syntax,
864 description="Check a Python file for syntax errors without executing it.",
865 parameters={
866 "type": "object",
867 "properties": {
868 "path": {"type": "string", "description": "Path to the Python file."}
869 },
870 "required": ["path"]
871 }
872 )
873 self.register(
874 name="mcp_call",
875 func=self._mcp_call,
876 description="Call an MCP server with a JSON payload (requires MCP enabled in config).",
877 parameters={
878 "type": "object",
879 "properties": {
880 "server": {"type": "string", "description": "Server name from config"},
881 "payload": {"type": "object", "description": "JSON payload to send"}
882 },
883 "required": ["server", "payload"]
884 }
885 )
886 self.register(
887 name="lsp_diagnostics",
888 func=self._lsp_diagnostics,
889 description="Run LSP diagnostics for a file (configured in settings.lsp_servers).",
890 parameters={
891 "type": "object",
892 "properties": {
893 "path": {"type": "string", "description": "File path to check"},
894 "lang": {"type": "string", "description": "Language key (e.g., python, typescript). Optional if inferred by extension."}
895 },
896 "required": ["path"]
897 }
898 )
899 self.register(
900 name="summarize_repo_structure",
901 func=self._summarize_repo_structure,
902 description="Produce a trimmed directory overview for the current repository.",
903 parameters={
904 "type": "object",
905 "properties": {
906 "path": {"type": "string", "description": "Directory to summarize (default: current directory)"},
907 "max_depth": {"type": "integer", "description": "Maximum depth to traverse (default: 2)"}
908 }
909 }
910 )
911 self.register(
912 name="collect_project_insights",
913 func=self._collect_project_insights,
914 description="Return repository statistics including python/test counts and documentation presence.",
915 parameters={
916 "type": "object",
917 "properties": {
918 "path": {"type": "string", "description": "Directory to inspect (default: current directory)"}
919 }
920 }
921 )
923 def get_definitions(self) -> Dict:
924 """Return tools as dict for display"""
925 return tools_registry.get_definitions()
927 def get_openai_tools(self) -> List[Dict]:
928 """Return tools in OpenAI format"""
929 return tools_registry.get_openai_format()
931 # --- Implementations ---
932 # ... (Existing implementations _run_shell, _read_file, etc.) ...
934 def _run_shell(
935 self,
936 command: Any,
937 shell: Optional[bool] = None,
938 timeout_sec: Optional[int] = None,
939 long_running: Optional[bool] = None,
940 detach: Optional[bool] = None,
941 ) -> Dict[str, Any]:
942 from .main import settings, console
944 SAFE_COMMANDS = {
945 'ls', 'pwd', 'cd', 'cat', 'head', 'tail', 'grep', 'find', 'wc',
946 'python', 'python3', 'pip', 'pip3', 'npm', 'node', 'git',
947 'mkdir', 'touch', 'cp', 'mv', 'chmod', 'chown', 'ps', 'top',
948 'df', 'du', 'free', 'uname', 'whoami', 'date', 'echo',
949 'curl', 'wget', 'ssh', 'scp', 'rsync', 'tar', 'zip', 'unzip',
950 'pytest', 'python -m pytest', 'python3 -m pytest', 'make', 'cmake',
951 'gcc', 'g++', 'clang', 'cargo', 'go', 'java', 'javac', 'timedatectl'
952 }
954 DANGEROUS_PATTERNS = [
955 'rm -rf /', 'sudo rm', 'chmod 777', 'dd if=',
956 'mkfs', 'fdisk', 'format', 'del /f', 'rmdir /s',
957 'shutdown', 'reboot', 'halt', 'poweroff',
958 'passwd', 'su -', 'sudo su', 'curl | sh', 'wget | sh',
959 'eval', 'exec', 'system(', 'subprocess.call(',
960 '__import__', 'compile(', 'open(', 'file('
961 ]
963 SAFE_OPERATORS = ['&&', '||', ';', '|', '>', '>>', '<']
965 params = {}
966 if isinstance(command, dict):
967 params.update(command)
968 else:
969 params["command"] = command
971 if shell is not None:
972 params["shell"] = shell
973 else:
974 params.setdefault("shell", True)
976 if timeout_sec is not None:
977 params["timeout_sec"] = timeout_sec
979 if long_running is not None:
980 params["long_running"] = long_running
981 else:
982 params.setdefault("long_running", False)
984 if detach is not None:
985 params["detach"] = detach
986 else:
987 params.setdefault("detach", False)
989 cmd = str(params.get("command", "")).strip()
990 if not cmd:
991 return self._build_structured_shell_result(
992 command=cmd,
993 stdout="",
994 stderr="Error: No command provided.",
995 exit_code=None,
996 status="error",
997 )
999 shell_mode = params.get("shell", True)
1000 timeout = params.get("timeout_sec")
1001 long_running = bool(params.get("long_running"))
1002 detach = bool(params.get("detach"))
1003 timeout = timeout if timeout is not None else (
1004 None if long_running or detach else persistent_shell.default_timeout
1005 )
1007 allowed_by_session = False
1008 if hasattr(settings, 'session_allowlist') and settings.session_allowlist:
1009 for allowed_cmd in settings.session_allowlist:
1010 if allowed_cmd and allowed_cmd in cmd:
1011 console.print(f"[dim]✅ Allowed by session allowlist[/dim]")
1012 allowed_by_session = True
1013 break
1015 cmd_lower = cmd.lower()
1016 dangerous_found = [pattern for pattern in DANGEROUS_PATTERNS if pattern in cmd_lower]
1017 if dangerous_found and not (settings.yolo_mode or allowed_by_session):
1018 console.print(f"[bold red]⚠️ SECURITY ALERT[/bold red]")
1019 console.print(f"Command contains potentially dangerous pattern(s): {', '.join(dangerous_found)}")
1020 console.print(f"[yellow]Command: {cmd}[/yellow]")
1021 base_msg = f"❌ Blocked: Dangerous command detected ({', '.join(dangerous_found)})"
1022 message = base_msg
1023 if settings.safety_profile == "strict":
1024 message += ". Strict safety profile prevents execution."
1025 return self._build_structured_shell_result(
1026 command=cmd,
1027 stdout="",
1028 stderr=message,
1029 exit_code=None,
1030 status="error",
1031 message=message,
1032 )
1033 message += ". Re-run with YOLO mode or add the command to the session allowlist."
1034 return self._build_structured_shell_result(
1035 command=cmd,
1036 stdout="",
1037 stderr=message,
1038 exit_code=None,
1039 status="error",
1040 message=message,
1041 )
1043 try:
1044 tokens = shlex.split(cmd)
1045 except ValueError:
1046 tokens = cmd.split()
1048 base_cmd = tokens[0].lower() if tokens else ""
1049 if base_cmd and base_cmd not in SAFE_COMMANDS and not allowed_by_session:
1050 if settings.yolo_mode:
1051 console.print(f"[dim]⚠️ YOLO mode: allowing '{base_cmd}'[/dim]")
1052 else:
1053 message = f"Error: Command '{base_cmd}' is not in the safe allowlist. Enable YOLO mode or add it to the session allowlist."
1054 return self._build_structured_shell_result(
1055 command=cmd,
1056 stdout="",
1057 stderr=message,
1058 exit_code=None,
1059 status="error",
1060 message=message,
1061 )
1063 has_operators = any(op in cmd for op in SAFE_OPERATORS)
1064 if has_operators and not (settings.yolo_mode or allowed_by_session or base_cmd in SAFE_COMMANDS):
1065 message = "Error: Shell operators are blocked unless YOLO mode is enabled or the command is allowlisted."
1066 return self._build_structured_shell_result(
1067 command=cmd,
1068 stdout="",
1069 stderr=message,
1070 exit_code=None,
1071 status="error",
1072 message=message,
1073 )
1074 if not (settings.yolo_mode or allowed_by_session):
1075 if "$(" in cmd or "`" in cmd:
1076 message = "Error: Command substitution is blocked unless YOLO mode is enabled or the command is allowlisted."
1077 return self._build_structured_shell_result(
1078 command=cmd,
1079 stdout="",
1080 stderr=message,
1081 exit_code=None,
1082 status="error",
1083 message=message,
1084 )
1085 if has_operators and base_cmd and base_cmd not in SAFE_COMMANDS and not allowed_by_session:
1086 message = f"Error: Shell operators not allowed with unsafe command '{base_cmd}'."
1087 return self._build_structured_shell_result(
1088 command=cmd,
1089 stdout="",
1090 stderr=message,
1091 exit_code=None,
1092 status="error",
1093 message=message,
1094 )
1096 cd_target = None
1097 if cmd.startswith("cd ") and "&&" in cmd:
1098 cd_target = self._extract_cd_target(cmd)
1100 use_persistent = not has_operators and not long_running and not detach
1101 if use_persistent:
1102 try:
1103 persistent_result = persistent_shell.run(cmd, timeout=timeout)
1104 return self._build_structured_shell_result(
1105 command=cmd,
1106 stdout=persistent_result.get("stdout", ""),
1107 stderr=persistent_result.get("stderr", ""),
1108 exit_code=persistent_result.get("exit_code"),
1109 status=persistent_result.get("status", "success"),
1110 )
1111 except subprocess.TimeoutExpired:
1112 return self._build_structured_shell_result(
1113 command=cmd,
1114 stdout="",
1115 stderr="Error: Command timed out.",
1116 exit_code=None,
1117 status="timeout",
1118 )
1119 except Exception as exc:
1120 return self._build_structured_shell_result(
1121 command=cmd,
1122 stdout="",
1123 stderr=f"Error executing command: {exc}",
1124 exit_code=None,
1125 status="error",
1126 )
1127 finally:
1128 if cd_target:
1129 persistent_shell.cwd = cd_target
1131 if long_running or detach:
1132 result = self._start_long_running_process(
1133 cmd, shell_mode=shell_mode, detach=detach
1134 )
1135 if cd_target:
1136 persistent_shell.cwd = cd_target
1137 return self._build_structured_shell_result(
1138 command=cmd,
1139 stdout="",
1140 stderr="",
1141 exit_code=None,
1142 status="running",
1143 pid=result.get("pid"),
1144 url=result.get("url"),
1145 message=result.get("message"),
1146 detach_hint=result.get("detach_hint"),
1147 )
1149 try:
1150 completed = subprocess.run(
1151 cmd,
1152 shell=shell_mode,
1153 cwd=str(persistent_shell.cwd),
1154 env=persistent_shell.env,
1155 capture_output=True,
1156 text=True,
1157 timeout=timeout,
1158 )
1159 return self._build_structured_shell_result(
1160 command=cmd,
1161 stdout=completed.stdout,
1162 stderr=completed.stderr,
1163 exit_code=completed.returncode,
1164 status="success" if completed.returncode == 0 else "error",
1165 )
1166 except subprocess.TimeoutExpired:
1167 return self._build_structured_shell_result(
1168 command=cmd,
1169 stdout="",
1170 stderr="Error: Command timed out.",
1171 exit_code=None,
1172 status="timeout",
1173 )
1174 except Exception as exc:
1175 return self._build_structured_shell_result(
1176 command=cmd,
1177 stdout="",
1178 stderr=f"Error executing command: {exc}",
1179 exit_code=None,
1180 status="error",
1181 )
1182 finally:
1183 if cd_target:
1184 persistent_shell.cwd = cd_target
1186 def _summarize_shell_output(self, stdout: str, stderr: str, message: Optional[str]) -> str:
1187 if message:
1188 return message
1189 parts = []
1190 cleaned_out = stdout.strip()
1191 cleaned_err = stderr.strip()
1192 if cleaned_out:
1193 parts.append(cleaned_out)
1194 if cleaned_err:
1195 parts.append(f"STDERR: {cleaned_err}")
1196 if parts:
1197 return "\n".join(parts)
1198 return "(Command ran with no output)"
1200 def _build_structured_shell_result(
1201 self,
1202 command: str,
1203 stdout: str = "",
1204 stderr: str = "",
1205 exit_code: Optional[int] = None,
1206 status: str = "success",
1207 message: Optional[str] = None,
1208 pid: Optional[int] = None,
1209 url: Optional[str] = None,
1210 detach_hint: Optional[str] = None,
1211 ) -> Dict[str, Any]:
1212 return {
1213 "command": command,
1214 "stdout": stdout.strip(),
1215 "stderr": stderr.strip(),
1216 "exit_code": exit_code,
1217 "status": status,
1218 "pid": pid,
1219 "url": url,
1220 "detach_hint": detach_hint,
1221 "message": message,
1222 "summary": self._summarize_shell_output(stdout, stderr, message),
1223 "meta": {
1224 "pid": pid,
1225 "url": url,
1226 "detach_hint": detach_hint,
1227 "command": command,
1228 },
1229 }
1231 def _extract_cd_target(self, command: str) -> Optional[Path]:
1232 parts = command.split("&&", 1)
1233 if not parts:
1234 return None
1235 cd_segment = parts[0].strip()
1236 try:
1237 tokens = shlex.split(cd_segment)
1238 except ValueError:
1239 tokens = cd_segment.split()
1240 if len(tokens) < 2:
1241 return None
1242 target = tokens[1]
1243 expanded = os.path.expanduser(target)
1244 new_path = (
1245 (persistent_shell.cwd / expanded).resolve()
1246 if not os.path.isabs(expanded)
1247 else Path(expanded).resolve()
1248 )
1249 if new_path.exists() and new_path.is_dir():
1250 return new_path
1251 return None
1253 def _start_long_running_process(self, command: str, shell_mode: bool, detach: bool) -> Dict[str, Any]:
1254 env = persistent_shell.env
1255 proc = subprocess.Popen(
1256 command,
1257 shell=shell_mode,
1258 cwd=str(persistent_shell.cwd),
1259 env=env,
1260 stdout=subprocess.PIPE,
1261 stderr=subprocess.PIPE,
1262 text=True,
1263 )
1264 self.running_processes[proc.pid] = proc
1265 url = self._infer_server_url(command)
1266 return {
1267 "command": command,
1268 "pid": proc.pid,
1269 "url": url,
1270 "message": f"RUNNING: {command}",
1271 "detach_hint": f"Stop with `kill {proc.pid}`" if detach else None,
1272 }
1274 def _infer_server_url(self, command: str) -> Optional[str]:
1275 match = re.search(r"-p\s+(\d+)", command)
1276 if not match:
1277 match = re.search(r"http\.server\s+(\d+)", command)
1278 port = int(match.group(1)) if match else 8000
1279 return f"http://localhost:{port}"
1281 def _summarize_repo_structure(self, path: str = ".", max_depth: int = 2) -> str:
1282 """Return a lightweight directory tree for the repository."""
1283 root = Path(path or ".").resolve()
1284 if not root.exists():
1285 raise FileNotFoundError(f"Path not found: {root}")
1286 if not root.is_dir():
1287 raise ValueError("summarize_repo_structure requires a directory path")
1289 max_depth = max(1, min(max_depth or 2, 5))
1290 lines: List[str] = []
1292 for current_path, dirs, files in os.walk(root):
1293 rel = Path(current_path).relative_to(root)
1294 depth = len(rel.parts) if rel.parts else 0
1295 if depth > max_depth:
1296 dirs[:] = []
1297 continue
1299 indent = " " * depth
1300 name = root.name if depth == 0 else rel.parts[-1]
1301 lines.append(f"{indent}- {name}/")
1303 shown_files = 0
1304 for file_name in sorted(files):
1305 if file_name.startswith("."):
1306 continue
1307 lines.append(f"{indent} - {file_name}")
1308 shown_files += 1
1309 if shown_files >= 10:
1310 remaining = len([f for f in files if not f.startswith(".")]) - shown_files
1311 if remaining > 0:
1312 lines.append(f"{indent} - ... ({remaining} more)")
1313 break
1315 return "\n".join(lines[:400])
1317 def _collect_project_insights(self, path: str = ".") -> str:
1318 """Return repository statistics that help the agent reason about the project."""
1319 root = Path(path or ".").resolve()
1320 if not root.exists():
1321 raise FileNotFoundError(f"Path not found: {root}")
1323 python_files = list(root.rglob("*.py"))
1324 test_files = [p for p in python_files if "tests" in p.parts]
1325 markdown_files = list(root.glob("*.md"))
1327 summary = {
1328 "root": str(root),
1329 "python_files": len(python_files),
1330 "test_files": len(test_files),
1331 "markdown_files": len(markdown_files),
1332 "has_readme": any(p.name.lower() == "readme.md" for p in markdown_files),
1333 "sample_tests": [str(p) for p in test_files[:5]],
1334 }
1335 return json.dumps(summary, indent=2)
1337 def _is_safe_path(self, path: str) -> bool:
1338 """Validate path to prevent dangerous access while allowing normal usage."""
1339 try:
1340 p = Path(path).expanduser().resolve()
1341 except Exception:
1342 return False
1344 blocked_patterns = ["/etc/passwd", "/etc/shadow", "/root/.ssh", "/home/*/.ssh"]
1345 for pattern in blocked_patterns:
1346 if fnmatch.fnmatch(str(p), pattern):
1347 return False
1349 for allowed_root in self._allowed_roots():
1350 if p == allowed_root or allowed_root in p.parents:
1351 return True
1353 return False
1355 def _allowed_roots(self) -> List[Path]:
1356 roots = []
1357 try:
1358 roots.append(Path.cwd().resolve())
1359 except Exception:
1360 pass
1362 try:
1363 roots.append(Path(CONFIG_DIR).resolve())
1364 except Exception:
1365 pass
1367 return roots
1369 def _read_file(self, path: str, start_line: Optional[int] = None, end_line: Optional[int] = None) -> str:
1370 # SECURITY: Path validation to prevent traversal
1371 if not self._is_safe_path(path):
1372 return "Error: Access denied. Path is outside allowed directory or contains traversal patterns."
1374 try:
1375 p = Path(path)
1376 if not p.exists():
1377 return f"Error: File {path} does not exist."
1379 # Add file size limit for security
1380 MAX_FILE_SIZE = 10_000_000 # 10MB limit
1381 if p.stat().st_size > MAX_FILE_SIZE:
1382 return f"Error: File too large. Maximum size is {MAX_FILE_SIZE:,} bytes."
1384 content = p.read_text(encoding='utf-8', errors='replace')
1386 if start_line is not None or end_line is not None:
1387 lines = content.splitlines()
1388 total_lines = len(lines)
1389 start = (start_line - 1) if start_line else 0
1390 end = end_line if end_line else total_lines
1392 # Clamp values
1393 start = max(0, start)
1394 end = min(total_lines, end)
1396 if start >= end:
1397 return f"Error: Invalid line range {start_line}-{end_line} (File has {total_lines} lines)."
1399 selected_lines = lines[start:end]
1400 snippet = "\n".join(selected_lines)
1401 return f"--- Lines {start+1}-{end} of {total_lines} ---\n{snippet}"
1403 return content
1404 except Exception as e:
1405 return f"Error reading file: {str(e)}"
1407 def _write_file(self, path: str, content: str, mode: str = "overwrite") -> str:
1408 # SECURITY: Path validation to prevent traversal
1409 if not self._is_safe_path(path):
1410 return "Error: Access denied. Path is outside allowed directory or contains traversal patterns."
1412 try:
1413 p = Path(path)
1415 # Enhanced Secret/DLP scan
1416 secret_hits = self._scan_secrets(content)
1417 if secret_hits:
1418 return f"Blocked write: potential secrets detected ({', '.join(secret_hits[:5])})."
1420 # Add content size limit for security
1421 MAX_CONTENT_SIZE = 10_000_000 # 10MB limit
1422 if len(content.encode('utf-8')) > MAX_CONTENT_SIZE:
1423 return f"Error: Content too large. Maximum size is {MAX_CONTENT_SIZE:,} bytes."
1425 p.parent.mkdir(parents=True, exist_ok=True)
1427 if mode == "diff" or mode == "patch":
1428 # Diff-based editing
1429 if p.exists():
1430 existing_content = p.read_text(encoding='utf-8')
1431 # Simple diff application - replace content between markers
1432 if "<<<" in content and ">>>" in content:
1433 # Extract diff markers
1434 parts = content.split("<<<")
1435 if len(parts) > 1:
1436 before_marker = parts[0]
1437 after_parts = parts[1].split(">>>")
1438 if len(after_parts) > 1:
1439 replacement = after_parts[0]
1440 after_marker = ">>>".join(after_parts[1:])
1442 # Apply the diff
1443 new_content = existing_content.replace(replacement.strip(), before_marker.strip() + replacement.strip() + after_marker.strip())
1444 p.write_text(new_content, encoding='utf-8')
1445 return f"Successfully applied diff patch to {path}"
1446 else:
1447 return "Error: Diff mode requires <<<OLD>>> and <<<NEW>>> markers"
1448 else:
1449 return f"Error: Cannot apply diff to non-existent file {path}"
1451 # Standard overwrite mode
1452 p.write_text(content, encoding='utf-8')
1453 return f"Successfully wrote to {path}"
1454 except Exception as e:
1455 return f"Error writing file: {str(e)}"
1457 def _list_dir(self, path: str = ".") -> str:
1458 try:
1459 p = Path(path).resolve()
1460 if p == self.home_dir or p in self.home_dir.parents:
1461 return "Error: Scanning the user home directory (or above) is blocked for safety. Please target a specific project subdirectory."
1463 entries = os.listdir(path)
1464 # Add trailing slash to dirs
1465 formatted = []
1466 for e in entries:
1467 full = os.path.join(path, e)
1468 if self._is_ignored(p, Path(full)): # Use p as root
1469 continue
1470 if os.path.isdir(full):
1471 formatted.append(f"{e}/")
1472 else:
1473 formatted.append(e)
1474 return "\n".join(sorted(formatted))
1475 except Exception as e:
1476 return f"Error listing directory: {str(e)}"
1478 def _get_tree(self, path: str = ".", max_depth: int = 3) -> str:
1479 """Generate a tree view of the directory structure with smart exploration."""
1480 output = []
1481 root = Path(path).resolve()
1483 if root == self.home_dir or root in self.home_dir.parents:
1484 return "Error: Recursive tree scan of the user home directory (or above) is blocked for performance. Please target a specific project folder."
1486 # 2025 Performance Guard
1487 try:
1488 # Optimized count: stop counting if we exceed 500
1489 total_files = 0
1490 for _ in root.rglob('*'):
1491 if any(p in _.parts for p in IGNORE_DIRS):
1492 continue
1493 total_files += 1
1494 if total_files > 500:
1495 break
1497 if total_files > 500:
1498 max_depth = 2 # Capped depth for large repos
1499 output.append(f"⚠️ HIGH DENSITY SECTOR (>500 files). Depth restricted to level 2 for performance.")
1500 except: pass
1502 try:
1503 # Smart exploration: analyze repository size to choose strategy
1504 total_files_count = 0
1505 total_dirs_count = 0
1507 def count_items(current_path: Path, depth: int = 0) -> Tuple[int, int]:
1508 if depth >= 1: # Even shallower count for home-dir adjacent folders
1509 return 0, 0
1511 files, dirs = 0, 0
1512 try:
1513 for entry in current_path.iterdir():
1514 if self._is_ignored(root, entry) or any(p in entry.parts for p in IGNORE_DIRS):
1515 continue
1516 if entry.is_dir():
1517 dirs += 1
1518 else:
1519 files += 1
1520 except PermissionError:
1521 pass
1523 return files, dirs
1525 total_files_count, total_dirs_count = count_items(root)
1527 # Smart depth selection based on repository size
1528 if total_files_count > 1000 or total_dirs_count > 100: # Large repository
1529 effective_max_depth = min(max_depth, 2) # Shallow exploration
1530 strategy = "shallow (large repository)"
1531 elif total_files_count > 100 or total_dirs_count > 20: # Medium repository
1532 effective_max_depth = min(max_depth, 3) # Medium exploration
1533 strategy = "medium (moderate repository)"
1534 else: # Small repository
1535 effective_max_depth = max_depth # Full exploration
1536 strategy = "deep (small repository)"
1538 output.append(f"Repository Analysis: {total_files_count} files, {total_dirs_count} directories")
1539 output.append(f"Exploration Strategy: {strategy}")
1540 output.append("")
1542 def walk_tree(current_path: Path, depth: int = 0):
1543 if depth >= effective_max_depth:
1544 return
1546 indent = " " * depth
1548 try:
1549 entries = sorted(current_path.iterdir())
1550 except PermissionError:
1551 output.append(f"{indent}{current_path.name}/ [Permission Denied]")
1552 return
1554 for entry in entries:
1555 # Skip ignored files
1556 if self._is_ignored(root, entry) or any(p in entry.parts for p in IGNORE_DIRS):
1557 continue
1559 if entry.is_dir():
1560 # Count items in directory for smart display
1561 try:
1562 dir_files = len([f for f in entry.iterdir() if not self._is_ignored(root, f) and f.is_file() and not any(p in f.parts for p in IGNORE_DIRS)])
1563 dir_dirs = len([d for d in entry.iterdir() if not self._is_ignored(root, d) and d.is_dir() and not any(p in d.parts for p in IGNORE_DIRS)])
1564 if dir_files > 0 or dir_dirs > 0:
1565 output.append(f"{indent}{entry.name}/ ({dir_files} files, {dir_dirs} dirs)")
1566 else:
1567 output.append(f"{indent}{entry.name}/")
1568 except PermissionError:
1569 output.append(f"{indent}{entry.name}/ [Permission Denied]")
1571 walk_tree(entry, depth + 1)
1572 else:
1573 # Show file size for large files
1574 try:
1575 size = entry.stat().st_size
1576 if size > 1024 * 1024: # > 1MB
1577 size_mb = size / (1024 * 1024)
1578 output.append(f"{indent}{entry.name} ({size_mb:.1f}MB)")
1579 elif size > 1024: # > 1KB
1580 size_kb = size / 1024
1581 output.append(f"{indent}{entry.name} ({size_kb:.0f}KB)")
1582 else:
1583 output.append(f"{indent}{entry.name}")
1584 except:
1585 output.append(f"{indent}{entry.name}")
1587 walk_tree(root, 0)
1588 return "\n".join(output) if output else "Directory is empty or all files are ignored."
1590 except Exception as e:
1591 return f"Error generating tree: {e}"
1593 def _web_search(self, query: str, num_results: int = 5) -> str:
1594 try:
1595 results = []
1596 errors = []
1597 source = None
1598 def add_result(title: str, url: str, desc: str = "") -> None:
1599 safe_title = str(title) if title else "No Title"
1600 safe_url = str(url) if url else "#"
1601 safe_desc = str(desc) if desc else ""
1602 results.append(f"- [{safe_title}]({safe_url}): {safe_desc}")
1604 def handle_item(item) -> None:
1605 if isinstance(item, dict):
1606 add_result(item.get("title", "No Title"), item.get("url", "#"), item.get("abstract", ""))
1607 return
1608 if isinstance(item, list):
1609 if not item:
1610 return
1611 if all(isinstance(elem, dict) for elem in item):
1612 for elem in item:
1613 handle_item(elem)
1614 return
1615 if len(item) >= 2:
1616 add_result(item[0], item[1], item[2] if len(item) > 2 else "")
1617 return
1618 # Prefer googler CLI (if installed)
1619 try:
1620 if shutil.which("googler"):
1621 cmd = ["googler", "--json", "-n", str(num_results), query]
1622 proc = subprocess.run(cmd, capture_output=True, text=True, check=False)
1623 if proc.returncode == 0 and proc.stdout.strip():
1624 stdout = proc.stdout.strip()
1625 try:
1626 parsed = json.loads(stdout)
1627 items = parsed if isinstance(parsed, list) else [parsed]
1628 for item in items:
1629 handle_item(item)
1630 if results:
1631 source = "googler"
1632 except json.JSONDecodeError:
1633 for line in stdout.splitlines():
1634 try:
1635 item = json.loads(line)
1636 handle_item(item)
1637 except json.JSONDecodeError:
1638 continue
1639 if results:
1640 source = "googler"
1641 else:
1642 if proc.stderr.strip():
1643 errors.append(f"googler: {proc.stderr.strip()}")
1644 else:
1645 errors.append("googler: not installed")
1646 except Exception as e:
1647 errors.append(f"googler: {e}")
1649 # Fallback to Bing RSS (structured, no external deps)
1650 if not results:
1651 try:
1652 rss_url = f"https://www.bing.com/search?format=rss&q={quote_plus(query)}"
1653 resp = requests.get(rss_url, timeout=10)
1654 if resp.status_code == 200 and resp.text.strip():
1655 root = ET.fromstring(resp.text)
1656 items = root.findall(".//item")[:num_results]
1657 for item in items:
1658 title = item.findtext("title") or "No Title"
1659 link = item.findtext("link") or "#"
1660 desc = item.findtext("description") or ""
1661 results.append(f"- [{title}]({link}): {desc}")
1662 if results:
1663 source = "bing_rss"
1664 except Exception as e:
1665 errors.append(f"bing_rss: {e}")
1667 # Fallback to lynx if googler fails or returns nothing
1668 if not results and shutil.which("lynx"):
1669 try:
1670 url = f"https://www.bing.com/search?q={quote_plus(query)}"
1671 cmd = ["lynx", "-dump", "-listonly", "-nonumbers", url]
1672 proc = subprocess.run(cmd, capture_output=True, text=True, check=False)
1673 if proc.returncode == 0 and proc.stdout.strip():
1674 urls = []
1675 for line in proc.stdout.splitlines():
1676 line = line.strip()
1677 if not line.startswith("http"):
1678 continue
1679 if line not in urls:
1680 urls.append(line)
1681 if len(urls) >= num_results:
1682 break
1683 for url in urls:
1684 results.append(f"- {url}")
1685 if results:
1686 source = "lynx"
1687 else:
1688 if proc.stderr.strip():
1689 errors.append(f"lynx: {proc.stderr.strip()}")
1690 except Exception as e:
1691 errors.append(f"lynx: {e}")
1693 if not results:
1694 if errors:
1695 return f"Error searching web (all backends failed): {', '.join(errors)}"
1696 return "No results found."
1698 notes = ""
1699 if errors:
1700 notes = f"\n\nNotes: {', '.join(errors[:2])}"
1701 return f"Results (source: {source or 'unknown'})\n" + "\n".join(results) + notes
1702 except Exception as e:
1703 return f"Error in web search tool: {e}"
1705 def _is_safe_url(self, url: str) -> bool:
1706 """Validate URL to prevent SSRF and dangerous requests."""
1707 try:
1708 from urllib.parse import urlparse
1709 parsed = urlparse(url)
1711 # Block dangerous schemes
1712 if parsed.scheme not in ['http', 'https']:
1713 return False
1715 # Block private/internal ranges
1716 if parsed.hostname and parsed.hostname.lower() in ['localhost', '127.0.0.1', '0.0.0.0', '::1']:
1717 return False
1719 return True
1720 except Exception:
1721 return False
1723 def _read_url(self, url: str) -> str:
1724 # SECURITY: URL validation to prevent SSRF
1725 if not self._is_safe_url(url):
1726 return "Error: URL blocked for security reasons. Only external HTTP/HTTPS URLs are allowed."
1728 try:
1729 # Enhanced scraping with security headers
1730 headers = {
1731 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
1732 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
1733 'Accept-Language': 'en-US,en;q=0.5',
1734 'Accept-Encoding': 'gzip, deflate',
1735 'DNT': '1',
1736 'Connection': 'keep-alive',
1737 'Upgrade-Insecure-Requests': '1'
1738 }
1740 response = requests.get(url, headers=headers, timeout=10, stream=True)
1741 response.raise_for_status()
1743 # Check content type
1744 content_type = response.headers.get('content-type', '').lower()
1745 if not content_type.startswith('text/'):
1746 return f"Error: Blocked non-text content type: {content_type}"
1748 # Read content with size limit
1749 content = b''
1750 total_size = 0
1751 MAX_CONTENT_SIZE = 10_000_000 # 10MB limit
1752 for chunk in response.iter_content(chunk_size=8192):
1753 total_size += len(chunk)
1754 if total_size > MAX_CONTENT_SIZE:
1755 return f"Error: Content too large. Maximum size is {MAX_CONTENT_SIZE:,} bytes."
1756 content += chunk
1758 from bs4 import XMLParsedAsHTMLWarning
1759 import warnings
1760 warnings.filterwarnings("ignore", category=XMLParsedAsHTMLWarning)
1761 soup = BeautifulSoup(content.decode('utf-8', errors='replace'), 'html.parser')
1763 # Strip script/style
1764 for script in soup(["script", "style"]):
1765 script.extract()
1767 text = soup.get_text()
1769 # Clean lines
1770 lines = (line.strip() for line in text.splitlines())
1771 chunks = (phrase.strip() for line in lines for phrase in line.split(" "))
1772 text = '\n'.join(chunk for chunk in chunks if chunk)
1774 return text[:10000] + ("\n... (truncated)" if len(text) > 10000 else "")
1775 except Exception as e:
1776 return f"Error reading URL: {e}"
1778 def _normalize_url(self, url: str) -> str:
1779 trimmed = url.strip()
1780 if not trimmed.startswith(("http://", "https://")):
1781 trimmed = "https://" + trimmed
1782 return trimmed
1784 def _extract_text(self, soup: BeautifulSoup) -> str:
1785 for script in soup(["script", "style", "noscript"]):
1786 script.extract()
1787 text = soup.get_text()
1788 lines = (line.strip() for line in text.splitlines())
1789 chunks = (phrase.strip() for line in lines for phrase in line.split(" "))
1790 cleaned = "\n".join(chunk for chunk in chunks if chunk)
1791 return cleaned
1793 def _site_report(self, url: str, max_pages: int = 6, max_depth: int = 2) -> str:
1794 target = self._normalize_url(url)
1795 if not self._is_safe_url(target):
1796 return "Error: URL blocked for security reasons. Only external HTTP/HTTPS URLs are allowed."
1797 max_pages = max(1, min(max_pages, 12))
1798 max_depth = max(1, min(max_depth, 3))
1800 headers = {
1801 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
1802 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
1803 'Accept-Language': 'en-US,en;q=0.5',
1804 }
1806 parsed = urlparse(target)
1807 base_host = parsed.netloc.lower()
1808 visited = set()
1809 queue: List[Tuple[str, int]] = [(target, 0)]
1810 pages: List[Dict[str, Any]] = []
1812 # Try sitemap for quick coverage
1813 sitemap_url = f"{parsed.scheme}://{base_host}/sitemap.xml"
1814 try:
1815 resp = requests.get(sitemap_url, timeout=8, headers=headers)
1816 if resp.status_code == 200 and resp.text.strip().startswith("<?xml"):
1817 root = ET.fromstring(resp.text)
1818 urls = [loc.text for loc in root.findall(".//{*}loc") if loc.text]
1819 for loc in urls[:max_pages]:
1820 loc_parsed = urlparse(loc)
1821 if loc_parsed.netloc.lower() == base_host and loc not in visited:
1822 queue.append((loc, 1))
1823 except Exception:
1824 pass
1826 while queue and len(pages) < max_pages:
1827 current, depth = queue.pop(0)
1828 if current in visited or depth > max_depth:
1829 continue
1830 visited.add(current)
1831 try:
1832 resp = requests.get(current, timeout=10, headers=headers)
1833 if resp.status_code != 200:
1834 continue
1835 content_type = resp.headers.get("content-type", "").lower()
1836 if "text/html" not in content_type:
1837 continue
1838 from bs4 import XMLParsedAsHTMLWarning
1839 import warnings
1840 warnings.filterwarnings("ignore", category=XMLParsedAsHTMLWarning)
1841 soup = BeautifulSoup(resp.text, "html.parser")
1842 title = soup.title.get_text(strip=True) if soup.title else "Untitled"
1843 headings = []
1844 for tag in soup.find_all(["h1", "h2"]):
1845 text = tag.get_text(strip=True)
1846 if text and text not in headings:
1847 headings.append(text)
1848 if len(headings) >= 6:
1849 break
1850 text = self._extract_text(soup)
1851 excerpt = text[:600].replace("\n", " ").strip()
1852 pages.append({
1853 "url": current,
1854 "title": title,
1855 "headings": headings,
1856 "excerpt": excerpt,
1857 })
1859 if depth < max_depth:
1860 for link in soup.find_all("a", href=True):
1861 href = link.get("href", "").strip()
1862 if not href or href.startswith(("mailto:", "tel:", "javascript:", "#")):
1863 continue
1864 absolute = urljoin(current, href)
1865 parsed_link = urlparse(absolute)
1866 if parsed_link.scheme not in ("http", "https"):
1867 continue
1868 if parsed_link.netloc.lower() != base_host:
1869 continue
1870 if absolute not in visited:
1871 queue.append((absolute, depth + 1))
1872 except Exception:
1873 continue
1875 if not pages:
1876 return "No pages could be crawled. Try a different URL or check connectivity."
1878 lines = []
1879 lines.append(f"# Site Report: {base_host}")
1880 lines.append("")
1881 lines.append(f"- Seed URL: `{target}`")
1882 lines.append(f"- Pages crawled: `{len(pages)}`")
1883 lines.append(f"- Max depth: `{max_depth}`")
1884 lines.append("")
1885 lines.append("## Pages")
1886 lines.append("")
1887 lines.append("| Page | Title | Top headings |")
1888 lines.append("|------|-------|--------------|")
1889 for page in pages:
1890 headings = ", ".join(page["headings"][:3]) if page["headings"] else "n/a"
1891 lines.append(f"| {page['url']} | {page['title']} | {headings} |")
1892 lines.append("")
1893 lines.append("## Key excerpts")
1894 lines.append("")
1895 for page in pages:
1896 lines.append(f"- `{page['url']}`: {page['excerpt'] or 'n/a'}")
1897 lines.append("")
1898 lines.append("## Sources")
1899 for page in pages:
1900 lines.append(f"- {page['url']}")
1902 return "\n".join(lines)
1904 def _search_codebase(self, pattern: str, path: str = ".") -> str:
1905 try:
1906 rg = shutil.which("rg")
1907 ignore_args = []
1908 # prefer ripgrep with ignore files if available
1909 if rg:
1910 gitignore = Path(path) / ".gitignore"
1911 osirisignore = Path(path) / ".osirisignore"
1912 if osirisignore.exists():
1913 ignore_args.extend(["--ignore-file", str(osirisignore)])
1914 if gitignore.exists():
1915 ignore_args.extend(["--ignore-file", str(gitignore)])
1917 cmd = [rg, "-n", "--context", "2", "--hidden"]
1918 cmd.extend(ignore_args)
1919 cmd.extend([pattern, str(path)])
1920 result = subprocess.run(
1921 cmd,
1922 capture_output=True,
1923 text=True,
1924 timeout=30,
1925 )
1926 if result.returncode == 0:
1927 output = result.stdout
1928 if len(output) > 10000:
1929 return output[:10000] + "\n... (truncated)"
1930 return output
1931 elif result.returncode == 1:
1932 return "No matches found."
1933 else:
1934 return f"Search error: {result.stderr}"
1935 else:
1936 # Fallback to grep (no ignore awareness). SECURITY FIX: Use list args.
1937 cmd = ["grep", "-rnC", "2", pattern, str(Path(path).resolve())]
1938 result = subprocess.run(
1939 cmd,
1940 shell=False, # Explicitly disable shell for security
1941 capture_output=True,
1942 text=True,
1943 timeout=30
1944 )
1945 if result.returncode == 0:
1946 output = result.stdout
1947 if len(output) > 10000:
1948 return output[:10000] + "\n... (truncated)"
1949 return output
1950 elif result.returncode == 1:
1951 return "No matches found."
1952 else:
1953 return f"Grep error: {result.stderr}"
1954 except Exception as e:
1955 return f"Error searching codebase: {e}"
1957 def _get_symbols(self, path: str) -> str:
1958 try:
1959 p = Path(path)
1960 if not p.exists():
1961 return f"Error: File {path} not found."
1963 code = p.read_text(encoding="utf-8")
1964 tree = ast.parse(code)
1966 symbols = []
1967 for node in ast.walk(tree):
1968 if isinstance(node, ast.ClassDef):
1969 symbols.append(f"CLASS: {node.name} (Line {node.lineno})")
1970 # Methods
1971 for item in node.body:
1972 if isinstance(item, ast.FunctionDef):
1973 symbols.append(f" METHOD: {item.name} (Line {item.lineno})")
1974 elif isinstance(node, ast.FunctionDef):
1975 # Only top level functions if we want to avoid duplicates from walking class body?
1976 # ast.walk visits everything. We can just check indent or parent?
1977 # Simplification: Just list all FunctionDefs that are not methods?
1978 # Hard to know parent in walk.
1979 # Let's just list them. User can infer.
1980 symbols.append(f"FUNC: {node.name} (Line {node.lineno})")
1982 if not symbols:
1983 return "No classes or functions found."
1985 # Deduplication (Naive) - ast.walk yields class then methods.
1986 # If we print methods inside class block above, we might print them again as standalone Funcs?
1987 # Yes. Let's refine.
1989 output = []
1990 for node in tree.body:
1991 if isinstance(node, ast.ClassDef):
1992 output.append(f"class {node.name}:")
1993 for item in node.body:
1994 if isinstance(item, ast.FunctionDef):
1995 output.append(f" def {item.name}")
1996 elif isinstance(item, ast.AsyncFunctionDef):
1997 output.append(f" async def {item.name}")
1998 elif isinstance(node, ast.FunctionDef):
1999 output.append(f"def {node.name}")
2000 elif isinstance(node, ast.AsyncFunctionDef):
2001 output.append(f"async def {node.name}")
2003 return "\n".join(output)
2004 except Exception as e:
2005 return f"Error parsing AST: {e}"
2007 def _check_syntax(self, path: str) -> str:
2008 try:
2009 p = Path(path)
2010 if not p.exists():
2011 return f"Error: File {path} not found."
2013 code = p.read_text(encoding="utf-8")
2014 ast.parse(code)
2015 return "✅ Syntax is valid."
2016 except SyntaxError as e:
2017 return f"❌ Syntax Error at line {e.lineno}, offset {e.offset}:\n{e.text}\n{e.msg}"
2018 except Exception as e:
2019 return f"Error checking syntax: {e}"
2021 async def _async_mcp_call(self, server: str, payload: dict):
2022 from .mcp import mcp_client
2023 return await mcp_client.call(server, payload)
2025 def _mcp_call(self, server: str, payload: dict) -> str:
2026 try:
2027 loop = asyncio.new_event_loop()
2028 asyncio.set_event_loop(loop)
2029 result = loop.run_until_complete(self._async_mcp_call(server, payload))
2030 loop.close()
2031 return json.dumps(result, indent=2)
2032 except Exception as e:
2033 return f"Error calling MCP server: {e}"
2035 def _lsp_diagnostics(self, path: str, lang: Optional[str] = None) -> str:
2036 from .config import settings
2037 if not settings.lsp_enabled:
2038 return "LSP diagnostics disabled. Enable lsp_enabled in config."
2039 servers = settings.lsp_servers or {}
2040 if not lang:
2041 ext = Path(path).suffix.lower().lstrip(".")
2042 lang = ext if ext else "default"
2043 cfg = servers.get(lang)
2044 if not cfg:
2045 return f"No LSP server configured for '{lang}'."
2046 cmd = cfg.get("command")
2047 args = cfg.get("args", [])
2048 if not cmd:
2049 return f"No command defined for LSP '{lang}'."
2050 try:
2051 proc = subprocess.run(
2052 [cmd, *args],
2053 input=Path(path).read_bytes(),
2054 capture_output=True,
2055 timeout=20
2056 )
2057 out = proc.stdout.decode(errors="ignore")
2058 err = proc.stderr.decode(errors="ignore")
2059 if proc.returncode == 0:
2060 return out[:8000] or "No diagnostics reported."
2061 return f"Diagnostics exited {proc.returncode}:\n{(out + err)[:8000]}"
2062 except Exception as e:
2063 return f"Error running LSP diagnostics: {e}"
2065 def _resolve_symbol(self, symbol_name: str, path: str = ".") -> str:
2066 """Sovereign Symbol Recovery: Fuzzy match logic symbols across the project."""
2067 from difflib import get_close_matches
2068 root = Path(path).resolve()
2069 all_symbols = []
2071 # 1. Collect all likely symbols (Heuristic)
2072 for f in root.rglob("*.py"):
2073 if any(x in f.name for x in [".venv", "node_modules", ".git"]): continue
2074 try:
2075 content = f.read_text(errors="ignore")
2076 # Find defs and classes
2077 matches = re.findall(r'(?:def|class)\s+([\w_]+)', content)
2078 all_symbols.extend(matches)
2079 except: pass
2081 if not all_symbols: return "No symbols found in project."
2083 # 2. Fuzzy Match
2084 matches = get_close_matches(symbol_name, list(set(all_symbols)), n=5, cutoff=0.5)
2085 if matches:
2086 return f"Symbol '{symbol_name}' not found. Did you mean:\n" + "\n".join([f"- {m}" for m in matches])
2088 return f"Could not resolve symbol '{symbol_name}'. Try 'generate_repo_map' for a full scan."
2090 def _collect_env_grounding(self) -> str:
2091 """Sovereign Grounding: Gather deep system/environment context."""
2092 import platform
2093 import os
2095 details = [
2096 "### SYSTEM GROUNDING DETAILS",
2097 f"- OS KERNEL: {platform.platform()}",
2098 f"- MACHINE: {platform.machine()}",
2099 f"- SHELL: {os.getenv('SHELL', 'unknown')}",
2100 f"- USER: {os.getenv('USER', 'unknown') or os.getenv('USERNAME')}",
2101 f"- PATH ENTRIES: {len(os.getenv('PATH', '').split(':'))}",
2102 f"- CONDA/VENV: {os.getenv('CONDA_DEFAULT_ENV', 'None')}/{os.getenv('VIRTUAL_ENV', 'None')}"
2103 ]
2105 # Check for critical tools availability
2106 critical_bins = ["git", "docker", "npm", "python3", "rg", "fd"]
2107 available = [b for b in critical_bins if shutil.which(b)]
2108 details.append(f"- RELEVANT BINARIES: {', '.join(available)}")
2110 return "\n".join(details)
2112 def _map_resources(self, path: str = ".") -> str:
2113 """Sovereign Tactical Map: Identify external dependencies and configs."""
2114 root = Path(path).resolve()
2115 findings = []
2117 # 1. Look for Configs & Secrets (Names only)
2118 configs = list(root.rglob("*.env*")) + list(root.rglob("config.y*ml"))
2119 if configs:
2120 findings.append("### CONFIGURATION ASSETS")
2121 for c in configs[:5]:
2122 findings.append(f"- Found config: {c.relative_to(root)}")
2124 # 2. Look for Database/API Logic
2125 keywords = ["connect", "database_url", "api_key", "endpoint", "base_url"]
2126 found_logic = []
2127 # We search source files for connection patterns
2128 for ext in ["*.py", "*.js", "*.ts", "*.go"]:
2129 for f in root.rglob(ext):
2130 if any(x in f.name for x in ["node_modules", ".venv", ".git"]): continue
2131 try:
2132 content = f.read_text(errors="ignore")
2133 for k in keywords:
2134 if k in content.lower():
2135 found_logic.append(f"- {f.relative_to(root)} (Ref: {k})")
2136 break
2137 except: pass
2138 if len(found_logic) > 10: break
2140 if found_logic:
2141 findings.append("\n### EXTERNAL INTERFACE LOGIC")
2142 findings.extend(found_logic)
2144 return "\n".join(findings) or "No external resources detected in current scope."
2146 def _handoff(self, target_branch: str, mission_summary: str) -> str:
2147 """Sovereign Handoff: Transfer mission to a specialized branch."""
2148 current_branch = getattr(self, "branch", "main")
2149 console.print(f"\n[bold yellow]>>> MISSION HANDOFF: {current_branch} \u2794 {target_branch}[/bold yellow]")
2151 # We trigger a delegate call but with 'handoff' semantics
2152 return self._delegate_subagent(
2153 task=f"MISSION HANDOFF FROM {current_branch}. SUMMARY: {mission_summary}",
2154 agent_name=target_branch
2155 )
2157 def _create_artifact(self, identifier: str, title: str, type: str, content: str, path: Optional[str] = None) -> str:
2158 """Create a high-value sovereign artifact (Code, Document, or Diagram)."""
2159 # Type can be: 'code', 'markdown', 'mermaid', 'report'
2160 final_path = path or f"artifacts/{identifier}.{type if type != 'mermaid' else 'md'}"
2161 p = Path(final_path)
2163 try:
2164 p.parent.mkdir(parents=True, exist_ok=True)
2165 p.write_text(content, encoding="utf-8")
2167 # Signal the UI via a structured result
2168 return {
2169 "status": "success",
2170 "message": f"Artifact '{title}' created successfully at {final_path}.",
2171 "meta": {
2172 "artifact_id": identifier,
2173 "artifact_title": title,
2174 "artifact_type": type,
2175 "artifact_path": str(p)
2176 }
2177 }
2178 except Exception as e:
2179 return f"Error creating artifact: {str(e)}"
2181 def _run_tests(self, path: str = ".") -> str:
2182 """Sovereign Test Suite: Auto-detect and run project tests."""
2183 root = Path(path).resolve()
2185 # 1. Detection Logic
2186 runners = []
2188 if (root / "pytest.ini").exists() or (root / "tests").exists():
2189 runners.append(("pytest", ["pytest", "--maxfail=3", "-v"]))
2190 if (root / "package.json").exists():
2191 try:
2192 pkg = json.loads((root / "package.json").read_text())
2193 scripts = pkg.get("scripts", {})
2194 if "test" in scripts:
2195 runners.append(("npm test", ["npm", "test"]))
2196 except: pass
2198 if not runners:
2199 return "Error: No recognized test runner (pytest, npm test) found in this path."
2201 # 2. Execution (We pick the first match for now)
2202 name, cmd = runners[0]
2203 console.print(f" [b cyan]TESTING[/b] \u203a Running {name}...")
2205 try:
2206 # We use a short timeout for safety
2207 result = subprocess.run(cmd, capture_output=True, text=True, timeout=120, cwd=str(root))
2209 if result.returncode == 0:
2210 return {
2211 "status": "success",
2212 "message": f"\u2705 TESTS PASSED ({name})\n{result.stdout[-500:]}",
2213 "stdout": result.stdout,
2214 "stderr": result.stderr,
2215 "exit_code": 0
2216 }
2217 else:
2218 # Capture the failures specifically
2219 return {
2220 "status": "error",
2221 "message": f"\u274c TESTS FAILED ({name}, exit={result.returncode})\nERRORS:\n{result.stderr or result.stdout[-1000:]}",
2222 "stdout": result.stdout,
2223 "stderr": result.stderr,
2224 "exit_code": result.returncode
2225 }
2226 except Exception as e:
2227 return f"FATAL: Test runner {name} crashed: {str(e)}"
2229 def _search_docs(self, query: str) -> str:
2230 """Search technical documentation and StackOverflow for code patterns."""
2231 tech_sites = ["stackoverflow.com", "github.com", "docs.python.org", "developer.mozilla.org", "react.dev", "nextjs.org"]
2232 site_query = " OR ".join([f"site:{s}" for s in tech_sites])
2233 full_query = f"({query}) ({site_query})"
2235 return self._web_search(full_query, num_results=5)
2237 def _load_ignore_spec(self, root: Path) -> Optional[pathspec.PathSpec]:
2238 root = root.resolve()
2239 if root in self._ignore_cache:
2240 return self._ignore_cache[root]
2242 patterns = []
2243 for fname in [".osirisignore", ".gitignore"]:
2244 f = root / fname
2245 if f.exists():
2246 try:
2247 file_patterns = f.read_text().splitlines()
2248 patterns.extend([p for p in file_patterns if p.strip()])
2249 except Exception:
2250 continue
2252 spec = None
2253 if patterns:
2254 try:
2255 spec = pathspec.PathSpec.from_lines("gitwildmatch", patterns)
2256 except Exception:
2257 pass
2259 self._ignore_cache[root] = spec
2260 return spec
2262 def _is_ignored(self, root: Path, target: Path) -> bool:
2263 spec = self._load_ignore_spec(root)
2264 if not spec:
2265 return False
2266 try:
2267 rel = target.relative_to(root)
2268 rel_str = str(rel)
2269 except Exception:
2270 rel_str = str(target)
2271 try:
2272 return spec.match_file(rel_str)
2273 except Exception:
2274 return False
2276 def _scan_secrets(self, text: str) -> List[str]:
2277 """
2278 Lightweight secret detection. Returns list of matched prefixes.
2279 """
2280 patterns = {
2281 "sk-": r"sk-[A-Za-z0-9]{16,}",
2282 "gsk_": r"gsk_[A-Za-z0-9]{16,}",
2283 "api_key": r"(?i)api[_-]?key[:=]\\s*[A-Za-z0-9\\-_]{12,}",
2284 "token": r"(?i)token[:=]\\s*[A-Za-z0-9\\-_]{12,}",
2285 }
2286 hits = []
2287 for label, regex in patterns.items():
2288 if re.search(regex, text):
2289 hits.append(label)
2290 return hits
2291 def _delegate_subagent(self, task: str, agent_name: str = "sub-agent") -> str:
2292 """Call ourselves in non-interactive mode for a specific task with Context Pass-through."""
2293 try:
2294 import subprocess
2295 import sys
2296 import tempfile
2298 # 2025 Sovereign Swarm: Context Pass-through
2299 # We serialize the current 'Memory Map' to a temporary file
2300 with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as tmp:
2301 tmp_path = Path(tmp.name)
2302 session.dump_context(tmp_path)
2304 # Use same python executable and find osiris entry point
2305 # We pass the context-file to the child
2306 cmd = [
2307 sys.executable, "-m", "osiris_cli.main",
2308 "-p", task,
2309 "-a", agent_name,
2310 "--yolo",
2311 "--context-file", str(tmp_path)
2312 ]
2313 console.print(f" [dim]Spawning sub-agent '{agent_name}' with parent context...[/dim]")
2315 result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
2317 # Cleanup temp file
2318 tmp_path.unlink(missing_ok=True)
2320 if result.returncode == 0:
2321 output = result.stdout.strip()
2322 return f"Sub-agent session complete.\n\nSummary of Output:\n{output}"
2323 else:
2324 return f"Sub-agent failed with code {result.returncode}.\nError: {result.stderr}"
2325 except Exception as e:
2326 return f"Error delegating to sub-agent: {e}"
2328 def _get_lsp_diagnostics(self, path: str = ".") -> str:
2329 # Check for common analyzers
2330 checks = [
2331 ("ruff", ["ruff", "check", path]),
2332 ("flake8", ["flake8", path]),
2333 ("pyright", ["pyright", path]),
2334 ("mypy", ["mypy", path])
2335 ]
2337 results = []
2338 for name, cmd in checks:
2339 if shutil.which(cmd[0]):
2340 try:
2341 res = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
2342 if res.stdout.strip():
2343 results.append(f"--- {name} ---\n{res.stdout.strip()}")
2344 except:
2345 continue
2347 if not results:
2348 return "No static analysis errors found with available tools (ruff, flake8, pyright, mypy)."
2350 return "\n\n".join(results)
2352tools = ToolRegistry()
2355# Helper functions for easy access
2356def get_tools():
2357 """Get tools in OpenAI format"""
2358 base_tools = tools.get_openai_tools()
2360 # Temporarily disable new tools due to format issues
2361 # TODO: Fix database and cloud tools format
2362 # from .database_tools_letta import DATABASE_TOOLS_LETTA
2363 # from .cloud_tools_letta import CLOUD_TOOLS_LETTA
2364 # base_tools.extend(DATABASE_TOOLS_LETTA)
2365 # base_tools.extend(CLOUD_TOOLS_LETTA)
2367 return base_tools
2369@monitor_tool_call
2370def execute_tool(tool_name: str, arguments: dict):
2371 """Execute a tool by name with given arguments"""
2372 if tool_name not in tools.tools:
2373 raise ValueError(f"Unknown tool: {tool_name}")
2375 session.enforce_tool_policy(tool_name)
2377 tool_func = tools.tools[tool_name]['func']
2378 return tool_func(**arguments)