File size: 3,505 Bytes
891669b a0533c6 891669b a0533c6 adbf39e 891669b a0533c6 891669b adbf39e 891669b a0533c6 891669b adbf39e 891669b a0533c6 adbf39e 891669b adbf39e | 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 | # ---- Changelog ----
# [2026-03-29] Chisel/TQB — Block C: WorkspaceTool
# What: map_repository_structure, get_stats extracted from RecursiveContextManager
# Why: PRD Block C — single-responsibility tool classes
# How: AST-based repo mapping and NG stats; ng instance passed for stats retrieval
# [2026-03-30] QB — Block D: Error handling hardening
# What: Specific exception types, logger replaces print()
# Why: PRD Block D — no broad except Exception, structured logging
# How: Catch OSError/SyntaxError specifically; use logger.warning for NG stats
# -------------------
import ast
import logging
import time
from pathlib import Path
from typing import Dict
logger = logging.getLogger("tools.workspace")
# Stats cache — avoid repeated full disk scans
_stats_cache = {"data": None, "expires": 0}
_STATS_TTL = 30 # seconds
class WorkspaceTool:
"""Repository mapping and statistics."""
def __init__(self, repo_path: Path, ng, policy_engine=None):
self.repo_path = repo_path
self.ng = ng
self.policy_engine = policy_engine
def map_repository_structure(self) -> str:
graph = {"nodes": [], "edges": []}
try:
file_count = 0
for file_path in self.repo_path.rglob('*.py'):
if 'venv' in str(file_path):
continue
rel_path = str(file_path.relative_to(self.repo_path))
content = file_path.read_text(errors='ignore')
file_count += 1
graph["nodes"].append({"id": rel_path, "type": "file"})
try:
tree = ast.parse(content)
for node in ast.walk(tree):
if isinstance(node, (ast.FunctionDef, ast.ClassDef)):
node_id = f"{rel_path}::{node.name}"
graph["nodes"].append({"id": node_id, "type": "function"})
except SyntaxError:
continue
return f"Map Generated: {file_count} files, {len(graph['nodes'])} nodes."
except (OSError, PermissionError) as e:
logger.error("[workspace] map_repository_structure failed: %s: %s", type(e).__name__, e, exc_info=True)
return {"status": "error", "tool": "workspace", "error": str(e), "type": type(e).__name__}
def get_stats(self) -> Dict:
now = time.time()
if _stats_cache["data"] and now < _stats_cache["expires"]:
return _stats_cache["data"]
ng_stats = {}
try:
ng_stats = self.ng.stats()
except (OSError, ValueError, AttributeError) as e:
logger.warning("NG stats retrieval failed: %s: %s", type(e).__name__, e)
# Count files excluding venv, __pycache__, .git, data/neurograph_worker checkpoints
file_count = 0
for p in self.repo_path.rglob("*"):
if p.is_file() and not any(skip in p.parts for skip in ("venv", "__pycache__", ".git")):
file_count += 1
stats = {
"total_files": file_count,
"conversations": ng_stats.get("message_count", 0),
"ng_nodes": ng_stats.get("nodes", 0),
"ng_synapses": ng_stats.get("synapses", 0),
"ng_firing_rate": ng_stats.get("firing_rate", 0.0),
"ng_prediction_accuracy": ng_stats.get("prediction_accuracy", 0.0),
}
_stats_cache["data"] = stats
_stats_cache["expires"] = now + _STATS_TTL
return stats
|