ed/manual_slop
Private
Public Access
0
0
Fork 0
Code Activity

Revert "merge: tier2/phase2_4_5_call_site_completion_20260621 (parent + follow-up + Phase 6e analysis)"

Browse Source 
This reverts commit f914b2bcd4, reversing
changes made to 7fef95cc87.
This commit is contained in:
Edward R. Gonzalez
2026-06-21 22:39:14 -04:00
parent f32e4fd268
commit 751b94d4e8
81 changed files with 2683 additions and 5005 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ai_client.py
+11 -19
View File
@@ -39,8 +39,6 @@ from typing import Optional, Callable, Any, List, Union, cast, Iterable
from src import project_manager
from src import file_cache
from src import mcp_client
from src import mcp_tool_specs
from src.openai_schemas import UsageStats
from src import mma_prompts
from src import performance_monitor
from src import project_manager
@@ -559,7 +557,7 @@ def _set_tool_preset_result(preset_name: Optional[str]) -> Result[None]:
if preset_name in presets:
preset = presets[preset_name]
_active_tool_preset = preset
new_tools = {name: False for name in mcp_tool_specs.tool_names()}
new_tools = {name: False for name in mcp_client.TOOL_NAMES}
new_tools[TOOL_NAME] = False
for cat in preset.categories.values():
for tool in cat:
@@ -581,7 +579,7 @@ def set_tool_preset(preset_name: Optional[str]) -> None:
_tool_approval_modes = {}
if not preset_name or preset_name == "None":
# Enable all tools if no preset
_agent_tools = {name: True for name in mcp_tool_specs.tool_names()}
_agent_tools = {name: True for name in mcp_client.TOOL_NAMES}
_agent_tools[TOOL_NAME] = True
_active_tool_preset = None
else:
@@ -1011,7 +1009,7 @@ async def _execute_single_tool_call_async(
tool_executed = True
if not tool_executed:
is_native = name in mcp_tool_specs.tool_names()
is_native = name in mcp_client.TOOL_NAMES
ext_tools = mcp_client.get_external_mcp_manager().get_all_tools()
is_external = name in ext_tools
if name and (is_native or is_external):
@@ -2052,7 +2050,7 @@ def _send_gemini_cli(md_content: str, user_message: str, base_dir: str,
def _send(r_idx: int) -> NormalizedResponse:
if adapter is None:
return NormalizedResponse(text="(adapter unavailable)", tool_calls=(), usage=UsageStats(input_tokens=0, output_tokens=0), raw_response=None)
return NormalizedResponse(text="(adapter unavailable)", tool_calls=[], usage_input_tokens=0, usage_output_tokens=0, usage_cache_read_tokens=0, usage_cache_creation_tokens=0, raw_response=None)
send_result = _send_cli_round_result(r_idx, adapter, payload, safety_settings, sys_instr, stream_callback)
if not send_result.ok:
raise cast(Exception, send_result.errors[0].original) from None
@@ -2086,7 +2084,7 @@ def _send_gemini_cli(md_content: str, user_message: str, base_dir: str,
"kind": "history_add",
"payload": {"role": "AI", "content": txt}
})
return NormalizedResponse(text=txt, tool_calls=(), usage=UsageStats(input_tokens=usage.get("prompt_tokens", 0), output_tokens=usage.get("completion_tokens", 0)), raw_response=resp_data)
return NormalizedResponse(text=txt, tool_calls=calls, usage_input_tokens=usage.get("prompt_tokens", 0), usage_output_tokens=usage.get("completion_tokens", 0), usage_cache_read_tokens=0, usage_cache_creation_tokens=0, raw_response=resp_data)
def _pre_dispatch(r_idx: int, calls: list[Metadata]) -> list[Metadata]:
nonlocal payload, cumulative_tool_bytes, file_items
@@ -2570,7 +2568,6 @@ def _send_grok(md_content: str, user_message: str, base_dir: str,
Runs synchronously in the caller thread; synchronizes Grok history using _grok_history_lock.
"""
from src.openai_compatible import OpenAICompatibleRequest, _classify_openai_compatible_error
from src.openai_schemas import ChatMessage
try:
client = _ensure_grok_client()
tools: list[Metadata] | None = _get_deepseek_tools() or None
@@ -2587,9 +2584,8 @@ def _send_grok(md_content: str, user_message: str, base_dir: str,
_grok_history.append({"role": "user", "content": user_content})
def _build_grok_request(_round_idx: int) -> OpenAICompatibleRequest:
with _grok_history_lock:
history_msgs: list[ChatMessage] = [ChatMessage(role=m["role"], content=m["content"]) for m in _grok_history]
messages: list[ChatMessage] = [ChatMessage(role="system", content=f"{_get_combined_system_prompt()}\n\n<context>\n{md_content}\n</context>")]
messages.extend(history_msgs)
messages: list[Metadata] = [{"role": "system", "content": f"{_get_combined_system_prompt()}\n\n<context>\n{md_content}\n</context>"}]
messages.extend(_grok_history)
extra_body: Metadata = {}
if caps.web_search:
extra_body["search_parameters"] = {"mode": "auto"}
@@ -2657,7 +2653,6 @@ def _send_minimax(md_content: str, user_message: str, base_dir: str,
Runs synchronously in the caller thread; synchronizes MiniMax history using _minimax_history_lock.
"""
from src.openai_compatible import OpenAICompatibleRequest
from src.openai_schemas import ChatMessage
try:
_ensure_minimax_client()
tools: list[Metadata] | None = _get_deepseek_tools() or None
@@ -2668,9 +2663,8 @@ def _send_minimax(md_content: str, user_message: str, base_dir: str,
_minimax_history.append({"role": "user", "content": user_message})
def _build_minimax_request(_round_idx: int) -> OpenAICompatibleRequest:
with _minimax_history_lock:
history_msgs: list[ChatMessage] = [ChatMessage(role=m["role"], content=m["content"]) for m in _minimax_history]
messages: list[ChatMessage] = [ChatMessage(role="system", content=f"{_get_combined_system_prompt()}\n\n<context>\n{md_content}\n</context>")]
messages.extend(history_msgs)
messages: list[Metadata] = [{"role": "system", "content": f"{_get_combined_system_prompt()}\n\n<context>\n{md_content}\n</context>"}]
messages.extend(_minimax_history)
return OpenAICompatibleRequest(
messages=messages, model=_model, temperature=_temperature, top_p=_top_p,
max_tokens=min(_max_tokens, 8192), stream=stream, stream_callback=stream_callback,
@@ -2899,7 +2893,6 @@ def _send_llama(md_content: str, user_message: str, base_dir: str,
Runs synchronously in the caller thread; synchronizes history using _llama_history_lock.
"""
from src.openai_compatible import OpenAICompatibleRequest, _classify_openai_compatible_error
from src.openai_schemas import ChatMessage
try:
if "localhost" in _llama_base_url or "127.0.0.1" in _llama_base_url:
return _send_llama_native(md_content, user_message, base_dir, file_items, discussion_history, stream, pre_tool_callback, qa_callback, stream_callback, patch_callback)
@@ -2917,9 +2910,8 @@ def _send_llama(md_content: str, user_message: str, base_dir: str,
_llama_history.append({"role": "user", "content": user_content})
def _build_llama_request(_round_idx: int) -> OpenAICompatibleRequest:
with _llama_history_lock:
history_msgs: list[ChatMessage] = [ChatMessage(role=m["role"], content=m["content"]) for m in _llama_history]
messages: list[ChatMessage] = [ChatMessage(role="system", content=f"{_get_combined_system_prompt()}\n\n<context>\n{md_content}\n</context>")]
messages.extend(history_msgs)
messages: list[Metadata] = [{"role": "system", "content": f"{_get_combined_system_prompt()}\n\n<context>\n{md_content}\n</context>"}]
messages.extend(_llama_history)
return OpenAICompatibleRequest(
messages=messages, model=_model, temperature=_temperature, top_p=_top_p,
max_tokens=_max_tokens, stream=stream, stream_callback=stream_callback,
src/api_hooks.py
+6 -14
View File
@@ -10,17 +10,9 @@ import uuid
# TODO(Ed): Eliminate these?
from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler
from typing import Any
from dataclasses import dataclass
from src.module_loader import _require_warmed
from src.result_types import ErrorInfo, ErrorKind, Result
from src.type_aliases import JsonValue
@dataclass(frozen=True)
class WebSocketMessage:
channel: str
payload: JsonValue
"""
@@ -139,7 +131,7 @@ class HookServerInstance(ThreadingHTTPServer):
super().__init__(server_address, RequestHandlerClass)
self.app = app
def _serialize_for_api(obj: Any) -> JsonValue:
def _serialize_for_api(obj: Any) -> Any:
"""Serializes complex objects into API-friendly formats (dicts/lists)."""
if hasattr(obj, "to_dict"):
return obj.to_dict()
@@ -980,12 +972,12 @@ class WebSocketServer:
if self.thread:
self.thread.join(timeout=2.0)
def broadcast(self, message: WebSocketMessage) -> None:
def broadcast(self, channel: str, payload: dict[str, Any]) -> None:
"""
[C: src/app_controller.py:AppController._process_pending_gui_tasks, src/events.py:AsyncEventQueue.put, tests/test_websocket_server.py:test_websocket_subscription_and_broadcast]
"""
if not self.loop or message.channel not in self.clients:
if not self.loop or channel not in self.clients:
return
wire = json.dumps({"channel": message.channel, "payload": message.payload})
for ws in list(self.clients[message.channel]):
asyncio.run_coroutine_threadsafe(ws.send(wire), self.loop)
message = json.dumps({"channel": channel, "payload": payload})
for ws in list(self.clients[channel]):
asyncio.run_coroutine_threadsafe(ws.send(message), self.loop)
src/app_controller.py
+1 -2
View File
@@ -1841,13 +1841,12 @@ class AppController:
def _process_pending_gui_tasks(self) -> None:
"""Processes pending GUI tasks from the queue on the main render thread."""
from src.api_hooks import WebSocketMessage
now = time.time()
if hasattr(self, 'event_queue') and hasattr(self.event_queue, 'websocket_server') and self.event_queue.websocket_server:
if now - self._last_telemetry_time >= 1.0:
self._last_telemetry_time = now
metrics = self.perf_monitor.get_metrics()
self.event_queue.websocket_server.broadcast(WebSocketMessage(channel="telemetry", payload=metrics))
self.event_queue.websocket_server.broadcast("telemetry", metrics)
if not self._pending_gui_tasks: return
src/events.py
+1 -3
View File
@@ -34,8 +34,6 @@ import queue
from pathlib import Path
from typing import Callable, Any, Dict, List, Tuple, Optional
from src.api_hooks import WebSocketMessage
class EventEmitter:
"""
@@ -114,7 +112,7 @@ class AsyncEventQueue:
elif hasattr(payload, '__dict__'):
serializable_payload = vars(payload)
self.websocket_server.broadcast(WebSocketMessage(channel="events", payload={"event": event_name, "payload": serializable_payload}))
self.websocket_server.broadcast("events", {"event": event_name, "payload": serializable_payload})
def get(self) -> Tuple[str, Any]:
"""
src/log_registry.py
+46 -130
View File
@@ -43,96 +43,12 @@ import os
import tomli_w
import tomllib
from dataclasses import dataclass
from datetime import datetime
from typing import Any, Optional
from typing import Any
from src.result_types import Result, ErrorInfo, ErrorKind
@dataclass(frozen=True)
class SessionMetadata:
message_count: int = 0
errors: int = 0
size_kb: int = 0
whitelisted: bool = False
reason: str = ''
timestamp: Optional[str] = None
def to_dict(self) -> dict[str, Any]:
return {
"message_count": self.message_count,
"errors": self.errors,
"size_kb": self.size_kb,
"whitelisted": self.whitelisted,
"reason": self.reason,
"timestamp": self.timestamp,
}
@dataclass(frozen=True)
class Session:
session_id: str
path: str
start_time: str
whitelisted: bool = False
metadata: Optional[SessionMetadata] = None
def to_dict(self) -> dict[str, Any]:
d: dict[str, Any] = {
"path": self.path,
"start_time": self.start_time,
"whitelisted": self.whitelisted,
}
if self.metadata is not None:
d["metadata"] = self.metadata.to_dict()
else:
d["metadata"] = None
return d
def __getitem__(self, key: str) -> Any:
"""Backward-compat: dict-like access (e.g., session['path'])."""
if key == "path":
return self.path
if key == "start_time":
return self.start_time
if key == "whitelisted":
return self.whitelisted
if key == "metadata":
return self.metadata.to_dict() if self.metadata is not None else None
raise KeyError(key)
def get(self, key: str, default: Any = None) -> Any:
"""Backward-compat: dict.get."""
try:
return self[key]
except KeyError:
return default
@classmethod
def from_dict(cls, session_id: str, d: dict[str, Any]) -> Session:
metadata_raw = d.get("metadata")
metadata: Optional[SessionMetadata] = None
if isinstance(metadata_raw, dict):
metadata = SessionMetadata(
message_count=int(metadata_raw.get("message_count", 0)),
errors=int(metadata_raw.get("errors", 0)),
size_kb=int(metadata_raw.get("size_kb", 0)),
whitelisted=bool(metadata_raw.get("whitelisted", False)),
reason=str(metadata_raw.get("reason", "")),
timestamp=metadata_raw.get("timestamp"),
)
elif metadata_raw is not None:
metadata = metadata_raw
return cls(
session_id=session_id,
path=str(d.get("path", "")),
start_time=str(d.get("start_time", "")),
whitelisted=bool(d.get("whitelisted", False)),
metadata=metadata,
)
class LogRegistry:
"""
Manages a persistent registry of session logs using a TOML file.
@@ -142,13 +58,13 @@ class LogRegistry:
def __init__(self, registry_path: str) -> None:
"""
Initializes the LogRegistry with a path to the registry file.
Args:
registry_path (str): The file path to the TOML registry.
[C: src/mcp_client.py:_DDGParser.__init__, src/mcp_client.py:_TextExtractor.__init__]
"""
self.registry_path = registry_path
self.data: dict[str, Session] = {}
self.data: dict[str, dict[str, Any]] = {}
self.load_registry()
@property
@@ -177,7 +93,7 @@ class LogRegistry:
m = new_session_data['metadata']
if 'timestamp' in m and isinstance(m['timestamp'], datetime):
m['timestamp'] = m['timestamp'].isoformat()
self.data[session_id] = Session.from_dict(session_id, new_session_data)
self.data[session_id] = new_session_data
except Exception as e:
print(f"Error loading registry from {self.registry_path}: {e}")
self.data = {}
@@ -193,14 +109,13 @@ class LogRegistry:
try:
# Convert datetime objects to ISO format strings for TOML serialization
data_to_save: dict[str, Any] = {}
for session_id, session in self.data.items():
session_dict = session.to_dict()
filtered: dict[str, Any] = {}
for k, v in session_dict.items():
for session_id, session_data in self.data.items():
session_data_copy: dict[str, Any] = {}
for k, v in session_data.items():
if v is None:
continue
if k == 'start_time' and isinstance(v, datetime):
filtered[k] = v.isoformat()
session_data_copy[k] = v.isoformat()
elif k == 'metadata' and isinstance(v, dict):
metadata_copy: dict[str, Any] = {}
for mk, mv in v.items():
@@ -210,10 +125,10 @@ class LogRegistry:
metadata_copy[mk] = mv.isoformat()
else:
metadata_copy[mk] = mv
filtered[k] = metadata_copy
session_data_copy[k] = metadata_copy
else:
filtered[k] = v
data_to_save[session_id] = filtered
session_data_copy[k] = v
data_to_save[session_id] = session_data_copy
with open(self.registry_path, 'wb') as f:
tomli_w.dump(data_to_save, f)
return Result(data=True)
@@ -237,13 +152,12 @@ class LogRegistry:
start_time_str = start_time.isoformat()
else:
start_time_str = start_time
self.data[session_id] = Session(
session_id=session_id,
path=path,
start_time=start_time_str,
whitelisted=False,
metadata=None,
)
self.data[session_id] = {
'path': path,
'start_time': start_time_str,
'whitelisted': False,
'metadata': None
}
self.save_registry()
def update_session_metadata(self, session_id: str, message_count: int, errors: int, size_kb: int, whitelisted: bool, reason: str) -> None:
@@ -262,22 +176,21 @@ class LogRegistry:
if session_id not in self.data:
print(f"Error: Session ID '{session_id}' not found for metadata update.")
return
existing = self.data[session_id]
new_metadata = SessionMetadata(
message_count=message_count,
errors=errors,
size_kb=size_kb,
whitelisted=whitelisted,
reason=reason,
timestamp=existing.metadata.timestamp if existing.metadata else None,
)
self.data[session_id] = Session(
session_id=existing.session_id,
path=existing.path,
start_time=existing.start_time,
whitelisted=whitelisted,
metadata=new_metadata,
)
# Ensure metadata exists
if self.data[session_id].get('metadata') is None:
self.data[session_id]['metadata'] = {}
# Update fields
metadata = self.data[session_id].get('metadata')
if isinstance(metadata, dict):
metadata['message_count'] = message_count
metadata['errors'] = errors
metadata['size_kb'] = size_kb
metadata['whitelisted'] = whitelisted
metadata['reason'] = reason
# self.data[session_id]['metadata']['timestamp'] = datetime.utcnow() # Optionally add a timestamp
# Also update the top-level whitelisted flag if provided
if whitelisted is not None:
self.data[session_id]['whitelisted'] = whitelisted
self.save_registry() # Save after update
def is_session_whitelisted(self, session_id: str) -> bool:
@@ -289,12 +202,13 @@ class LogRegistry:
Returns:
bool: True if whitelisted, False otherwise.
[C: tests/test_auto_whitelist.py:test_auto_whitelist_keywords, tests/test_auto_whitelist.py:test_auto_whitelist_large_size, tests/test_auto_whitelist.py:test_auto_whitelist_message_count, tests/test_no_auto_whitelist_insignificant, tests/test_log_registry.py:TestLogRegistry.test_is_session_whitelisted, tests/test_logging_e2e.py:test_logging_e2e]
[C: tests/test_auto_whitelist.py:test_auto_whitelist_keywords, tests/test_auto_whitelist.py:test_auto_whitelist_large_size, tests/test_auto_whitelist.py:test_auto_whitelist_message_count, tests/test_auto_whitelist.py:test_no_auto_whitelist_insignificant, tests/test_log_registry.py:TestLogRegistry.test_is_session_whitelisted, tests/test_logging_e2e.py:test_logging_e2e]
"""
session = self.data.get(session_id)
if session is None:
session_data = self.data.get(session_id)
if session_data is None:
return False # Non-existent sessions are not whitelisted
return session.whitelisted
# Check the top-level 'whitelisted' flag. If it's not set or False, it's not whitelisted.
return bool(session_data.get('whitelisted', False))
def update_auto_whitelist_status(self, session_id: str) -> None:
"""
@@ -309,7 +223,7 @@ class LogRegistry:
if session_id not in self.data:
return
session_data = self.data[session_id]
session_path = session_data.path
session_path = session_data.get('path')
if not session_path or not os.path.isdir(str(session_path)):
return
total_size_bytes = 0
@@ -371,9 +285,9 @@ class LogRegistry:
[C: tests/test_log_pruner.py:test_prune_old_insignificant_logs, tests/test_log_pruning_heuristic.py:TestLogPruningHeuristic.test_get_old_non_whitelisted_sessions_includes_empty_sessions, tests/test_log_pruning_heuristic.py:TestLogPruningHeuristic.test_get_old_non_whitelisted_sessions_includes_sessions_without_metadata, tests/test_log_registry.py:TestLogRegistry.test_get_old_non_whitelisted_sessions]
"""
old_sessions = []
for session_id, session in self.data.items():
for session_id, session_data in self.data.items():
# Check if session is older than cutoff and not whitelisted
start_time_raw = session.start_time
start_time_raw = session_data.get('start_time')
if isinstance(start_time_raw, str):
try:
start_time = datetime.fromisoformat(start_time_raw)
@@ -381,20 +295,22 @@ class LogRegistry:
start_time = None
else:
start_time = start_time_raw
is_whitelisted = session.whitelisted
is_whitelisted = session_data.get('whitelisted', False)
# Heuristic: also include non-whitelisted sessions that have 0 messages or 0 KB size, or missing metadata
metadata = session.metadata
metadata = session_data.get('metadata')
if metadata is None:
is_empty = True
else:
is_empty = (metadata.message_count == 0 or metadata.size_kb == 0)
message_count = metadata.get('message_count', -1)
size_kb = metadata.get('size_kb', -1)
is_empty = (message_count == 0 or size_kb == 0)
if not is_whitelisted:
if is_empty or (start_time is not None and start_time < cutoff_datetime):
old_sessions.append({
'session_id': session_id,
'path': session.path,
'path': session_data.get('path'),
'start_time': start_time_raw
})
return old_sessions
src/mcp_client.py
+780 -7
View File
@@ -69,7 +69,6 @@ from typing import Optional, Callable, Any, cast
from scripts import py_struct_tools
from src import beads_client
from src import mcp_tool_specs
from src import models
from src import outline_tool
from src import summarize
@@ -1010,10 +1009,10 @@ def get_tree_result(path: str, max_depth: int = 2) -> Result[str]:
entries = [e for e in entries if not e.name.startswith('.') and e.name not in ('__pycache__', 'venv', 'env') and e.name != "history.toml" and not e.name.endswith("_history.toml")]
for i, entry in enumerate(entries):
is_last = (i == len(entries) - 1)
connector = "└── " if is_last else "├── "
connector = "└── " if is_last else "├── "
if entry.is_dir():
lines.append(f"{prefix}{connector}{entry.name}/")
extension = " " if is_last else "│ "
extension = " " if is_last else "│ "
lines.extend(_build_tree(entry, current_depth + 1, prefix + extension))
else:
lines.append(f"{prefix}{connector}{entry.name}")
@@ -1942,7 +1941,7 @@ async def async_dispatch(tool_name: str, tool_input: dict[str, Any]) -> str:
"""
[C: src/rag_engine.py:RAGEngine._async_search_mcp, tests/test_external_mcp.py:test_external_mcp_real_process]
"""
native_names = mcp_tool_specs.tool_names()
native_names = {t['name'] for t in MCP_TOOL_SPECS}
if tool_name in native_names:
return await asyncio.to_thread(dispatch, tool_name, tool_input)
@@ -1954,9 +1953,9 @@ async def async_dispatch(tool_name: str, tool_input: dict[str, Any]) -> str:
def get_tool_schemas() -> list[dict[str, Any]]:
"""
[C: tests/test_arch_boundary_phase2.py:TestArchBoundaryPhase2.test_mcp_client_dispatch_completeness, tests/test_external_mcp.py:test_get_tool_schemas_includes_external, tests/test_mcp_client.py:test_bd_mcp_tools]
[C: tests/test_arch_boundary_phase2.py:TestArchBoundaryPhase2.test_mcp_client_dispatch_completeness, tests/test_external_mcp.py:test_get_tool_schemas_includes_external, tests/test_mcp_client_beads.py:test_bd_mcp_tools]
"""
res = [s.to_dict() for s in mcp_tool_specs.get_tool_schemas()]
res = list(MCP_TOOL_SPECS)
manager = get_external_mcp_manager()
for tname, tinfo in manager.get_all_tools().items():
res.append({
@@ -1970,5 +1969,779 @@ def get_tool_schemas() -> list[dict[str, Any]]:
# ------------------------------------------------------------------ tool schema helpers
# These are imported by ai_client.py to build provider-specific declarations.
MCP_TOOL_SPECS: list[dict[str, Any]] = [
{
"name": "py_remove_def",
"description": "Excises a specific class or function definition from a Python file using AST-derived line ranges, preserving surrounding formatting and comments.",
"parameters": {
"type": "object",
"properties": {
"path": { "type": "string", "description": "Path to the .py file." },
"name": { "type": "string", "description": "The name of the class or function to remove. Use 'ClassName.method_name' for methods." }
},
"required": ["path", "name"]
}
},
{
"name": "py_add_def",
"description": "Inserts a new definition into a specific context (module level or within a specific class).",
"parameters": {
"type": "object",
"properties": {
"path": { "type": "string", "description": "Path to the .py file." },
"name": { "type": "string", "description": "Context path (e.g. 'ClassName' or empty for module level)." },
"new_content": { "type": "string", "description": "The code to insert." },
"anchor_type": { "type": "string", "enum": ["before", "after", "top", "bottom"], "description": "Where to insert relative to the anchor." },
"anchor_symbol": { "type": "string", "description": "Symbol name to anchor to if anchor_type is 'before' or 'after'." }
},
"required": ["path", "name", "new_content", "anchor_type"]
}
},
{
"name": "py_move_def",
"description": "Relocates a definition within a file or across different Python files.",
"parameters": {
"type": "object",
"properties": {
"src_path": { "type": "string", "description": "Path to the source .py file." },
"dest_path": { "type": "string", "description": "Path to the destination .py file." },
"name": { "type": "string", "description": "The name of the class or function to move." },
"dest_name": { "type": "string", "description": "Context path in destination file (e.g. 'ClassName' or empty)." },
"anchor_type": { "type": "string", "enum": ["before", "after", "top", "bottom"], "description": "Where to insert in destination." },
"anchor_symbol": { "type": "string", "description": "Anchor symbol in destination." }
},
"required": ["src_path", "dest_path", "name", "dest_name", "anchor_type"]
}
},
{
"name": "py_region_wrap",
"description": "Wraps a specified block of code (e.g., a set of methods) in #region: Name and #endregion: Name tags.",
"parameters": {
"type": "object",
"properties": {
"path": { "type": "string", "description": "Path to the .py file." },
"start_line": { "type": "integer", "description": "1-based start line number." },
"end_line": { "type": "integer", "description": "1-based end line number (inclusive)." },
"region_name": { "type": "string", "description": "The name of the region." }
},
"required": ["path", "start_line", "end_line", "region_name"]
}
},
{
"name": "read_file",
"description": (
"Read the full UTF-8 content of a file within the allowed project paths. "
"Use get_file_summary first to decide whether you need the full content."
),
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Absolute or relative path to the file to read.",
}
},
"required": ["path"],
},
},
{
"name": "list_directory",
"description": (
"List files and subdirectories within an allowed directory. "
"Shows name, type (file/dir), and size. Use this to explore the project structure."
),
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Absolute path to the directory to list.",
}
},
"required": ["path"],
},
},
{
"name": "search_files",
"description": (
"Search for files matching a glob pattern within an allowed directory. "
"Supports recursive patterns like '**/*.py'. "
"Use this to find files by extension or name pattern."
),
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Absolute path to the directory to search within.",
},
"pattern": {
"type": "string",
"description": "Glob pattern, e.g. '*.py', '**/*.toml', 'src/**/*.rs'.",
},
},
"required": ["path", "pattern"],
},
},
{
"name": "get_file_summary",
"description": (
"Get a compact heuristic summary of a file without reading its full content. "
"For Python: imports, classes, methods, functions, constants. "
"For TOML: table keys. For Markdown: headings. Others: line count + preview. "
"Use this before read_file to decide if you need the full content."
),
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Absolute or relative path to the file to summarise.",
}
},
"required": ["path"],
},
},
{
"name": "py_get_skeleton",
"description": (
"Get a skeleton view of a Python file. "
"This returns all classes and function signatures with their docstrings, "
"but replaces function bodies with '...'. "
"Use this to understand module interfaces without reading the full implementation."
),
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the .py file.",
}
},
"required": ["path"],
},
},
{
"name": "py_get_code_outline",
"description": (
"Get a hierarchical outline of a code file. "
"This returns classes, functions, and methods with their line ranges and brief docstrings. "
"Use this to quickly map out a file's structure before reading specific sections."
),
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the code file (currently supports .py).",
}
},
"required": ["path"],
},
},
{
"name": "ts_c_get_skeleton",
"description": (
"Get a skeleton view of a C file. "
"This returns all function signatures and structs, "
"but replaces function bodies with '...'. "
"Use this to understand C interfaces without reading the full implementation."
),
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the C file.",
}
},
"required": ["path"],
},
},
{
"name": "ts_cpp_get_skeleton",
"description": (
"Get a skeleton view of a C++ file. "
"This returns all classes, structs and function signatures, "
"but replaces function bodies with '...'. "
"Use this to understand C++ interfaces without reading the full implementation."
),
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the C++ file.",
}
},
"required": ["path"],
},
},
{
"name": "ts_c_get_code_outline",
"description": (
"Get a hierarchical outline of a C file. "
"This returns structs and functions with their line ranges. "
"Use this to quickly map out a file's structure before reading specific sections."
),
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the C file.",
}
},
"required": ["path"],
},
},
{
"name": "ts_cpp_get_code_outline",
"description": (
"Get a hierarchical outline of a C++ file. "
"This returns classes, structs and functions with their line ranges. "
"Use this to quickly map out a file's structure before reading specific sections."
),
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the C++ file.",
}
},
"required": ["path"],
},
},
{
"name": "ts_c_get_definition",
"description": (
"Get the full source code of a specific function or struct definition in a C file. "
"This is more efficient than reading the whole file if you know what you're looking for."
),
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the C file.",
},
"name": {
"type": "string",
"description": "The name of the function or struct to retrieve.",
}
},
"required": ["path", "name"],
},
},
{
"name": "ts_cpp_get_definition",
"description": (
"Get the full source code of a specific class, function, or method definition in a C++ file. "
"This is more efficient than reading the whole file if you know what you're looking for."
),
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the C++ file.",
},
"name": {
"type": "string",
"description": "The name of the class or function to retrieve. Use 'ClassName::method_name' for methods.",
}
},
"required": ["path", "name"],
},
},
{
"name": "ts_c_get_signature",
"description": "Get only the signature part of a C function.",
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the C file."
},
"name": {
"type": "string",
"description": "Name of the function."
}
},
"required": ["path", "name"]
}
},
{
"name": "ts_cpp_get_signature",
"description": "Get only the signature part of a C++ function or method.",
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the C++ file."
},
"name": {
"type": "string",
"description": "Name of the function/method (e.g. 'ClassName::method_name')."
}
},
"required": ["path", "name"]
}
},
{
"name": "ts_c_update_definition",
"description": "Surgically replace the definition of a function in a C file using AST to find line ranges.",
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the C file."
},
"name": {
"type": "string",
"description": "Name of function."
},
"new_content": {
"type": "string",
"description": "Complete new source for the definition."
}
},
"required": ["path", "name", "new_content"]
}
},
{
"name": "ts_cpp_update_definition",
"description": "Surgically replace the definition of a class or function in a C++ file using AST to find line ranges.",
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the C++ file."
},
"name": {
"type": "string",
"description": "Name of class/function/method."
},
"new_content": {
"type": "string",
"description": "Complete new source for the definition."
}
},
"required": ["path", "name", "new_content"]
}
},
{
"name": "get_file_slice",
"description": "Read a specific line range from a file. Useful for reading parts of very large files.",
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the file."
},
"start_line": {
"type": "integer",
"description": "1-based start line number."
},
"end_line": {
"type": "integer",
"description": "1-based end line number (inclusive)."
}
},
"required": ["path", "start_line", "end_line"]
}
},
{
"name": "set_file_slice",
"description": "Replace a specific line range in a file with new content. Surgical edit tool.",
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the file."
},
"start_line": {
"type": "integer",
"description": "1-based start line number."
},
"end_line": {
"type": "integer",
"description": "1-based end line number (inclusive)."
},
"new_content": {
"type": "string",
"description": "New content to insert."
}
},
"required": ["path", "start_line", "end_line", "new_content"]
}
},
{
"name": "edit_file",
"description": "Replace exact string match in a file. Preserves indentation and line endings. Drop-in replacement for native edit tool.",
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the file."
},
"old_string": {
"type": "string",
"description": "The text to replace."
},
"new_string": {
"type": "string",
"description": "The replacement text."
},
"replace_all": {
"type": "boolean",
"description": "Replace all occurrences. Default false."
}
},
"required": ["path", "old_string", "new_string"]
}
},
{
"name": "py_get_definition",
"description": (
"Get the full source code of a specific class, function, or method definition. "
"This is more efficient than reading the whole file if you know what you're looking for."
),
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the .py file.",
},
"name": {
"type": "string",
"description": "The name of the class or function to retrieve. Use 'ClassName.method_name' for methods.",
}
},
"required": ["path", "name"],
},
},
{
"name": "py_update_definition",
"description": "Surgically replace the definition of a class or function in a Python file using AST to find line ranges.",
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the .py file."
},
"name": {
"type": "string",
"description": "Name of class/function/method."
},
"new_content": {
"type": "string",
"description": "Complete new source for the definition."
}
},
"required": ["path", "name", "new_content"]
}
},
{
"name": "py_get_signature",
"description": "Get only the signature part of a Python function or method (from def until colon).",
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the .py file."
},
"name": {
"type": "string",
"description": "Name of the function/method (e.g. 'ClassName.method_name')."
}
},
"required": ["path", "name"]
}
},
{
"name": "py_set_signature",
"description": "Surgically replace only the signature of a Python function or method.",
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the .py file."
},
"name": {
"type": "string",
"description": "Name of the function/method."
},
"new_signature": {
"type": "string",
"description": "Complete new signature string (including def and trailing colon)."
}
},
"required": ["path", "name", "new_signature"]
}
},
{
"name": "py_get_class_summary",
"description": "Get a summary of a Python class, listing its docstring and all method signatures.",
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the .py file."
},
"name": {
"type": "string",
"description": "Name of the class."
}
},
"required": ["path", "name"]
}
},
{
"name": "py_get_var_declaration",
"description": "Get the assignment/declaration line for a variable.",
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the .py file."
},
"name": {
"type": "string",
"description": "Name of the variable."
}
},
"required": ["path", "name"]
}
},
{
"name": "py_set_var_declaration",
"description": "Surgically replace a variable assignment/declaration.",
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the .py file."
},
"name": {
"type": "string",
"description": "Name of the variable."
},
"new_declaration": {
"type": "string",
"description": "Complete new assignment/declaration string."
}
},
"required": ["path", "name", "new_declaration"]
}
},
{
"name": "get_git_diff",
"description": (
"Returns the git diff for a file or directory. "
"Use this to review changes efficiently without reading entire files."
),
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the file or directory.",
},
"base_rev": {
"type": "string",
"description": "Base revision (e.g. 'HEAD', 'HEAD~1', or a commit hash). Defaults to 'HEAD'.",
},
"head_rev": {
"type": "string",
"description": "Head revision (optional).",
}
},
"required": ["path"],
},
},
{
"name": "web_search",
"description": "Search the web using DuckDuckGo. Returns the top 5 search results with titles, URLs, and snippets. Chain this with fetch_url to read specific pages.",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query."
}
},
"required": ["query"]
}
},
{
"name": "fetch_url",
"description": "Fetch the full text content of a URL (stripped of HTML tags). Use this after web_search to read relevant information from the web.",
"parameters": {
"type": "object",
"properties": {
"url": {
"type": "string",
"description": "The full URL to fetch."
}
},
"required": ["url"]
}
},
{
"name": "get_ui_performance",
"description": "Get a snapshot of the current UI performance metrics, including FPS, Frame Time (ms), CPU usage (%), and Input Lag (ms). Use this to diagnose UI slowness or verify that your changes haven't degraded the user experience.",
"parameters": {
"type": "object",
"properties": {}
}
},
{
"name": "py_find_usages",
"description": "Finds exact string matches of a symbol in a given file or directory.",
"parameters": {
"type": "object",
"properties": {
"path": { "type": "string", "description": "Path to file or directory to search." },
"name": { "type": "string", "description": "The symbol/string to search for." }
},
"required": ["path", "name"]
}
},
{
"name": "py_get_imports",
"description": "Parses a file's AST and returns a strict list of its dependencies.",
"parameters": {
"type": "object",
"properties": {
"path": { "type": "string", "description": "Path to the .py file." }
},
"required": ["path"]
}
},
{
"name": "py_check_syntax",
"description": "Runs a quick syntax check on a Python file.",
"parameters": {
"type": "object",
"properties": {
"path": { "type": "string", "description": "Path to the .py file." }
},
"required": ["path"]
}
},
{
"name": "py_get_hierarchy",
"description": "Scans the project to find subclasses of a given class.",
"parameters": {
"type": "object",
"properties": {
"path": { "type": "string", "description": "Directory path to search in." },
"class_name": { "type": "string", "description": "Name of the base class." }
},
"required": ["path", "class_name"]
}
},
{
"name": "py_get_docstring",
"description": "Extracts the docstring for a specific module, class, or function.",
"parameters": {
"type": "object",
"properties": {
"path": { "type": "string", "description": "Path to the .py file." },
"name": { "type": "string", "description": "Name of symbol or 'module' for the file docstring." }
},
"required": ["path", "name"]
}
},
{
"name": "get_tree",
"description": "Returns a directory structure up to a max depth.",
"parameters": {
"type": "object",
"properties": {
"path": { "type": "string", "description": "Directory path." },
"max_depth": { "type": "integer", "description": "Maximum depth to recurse (default 2)." }
},
"required": ["path"]
}
},
{
"name": "bd_create",
"description": "Create a new Bead in the active Beads repository.",
"parameters": {
"type": "object",
"properties": {
"title": { "type": "string", "description": "Title of the Bead." },
"description": { "type": "string", "description": "Description of the Bead." }
},
"required": ["title", "description"]
}
},
{
"name": "bd_update",
"description": "Update an existing Bead.",
"parameters": {
"type": "object",
"properties": {
"bead_id": { "type": "string", "description": "ID of the Bead to update." },
"status": { "type": "string", "description": "New status for the Bead." }
},
"required": ["bead_id", "status"]
}
},
{
"name": "bd_list",
"description": "List all Beads in the active Beads repository.",
"parameters": {
"type": "object",
"properties": {}
}
},
{
"name": "bd_ready",
"description": "Check if the Beads repository is initialized in the current workspace.",
"parameters": {
"type": "object",
"properties": {}
}
},
{
"name": "derive_code_path",
"description": (
"Recursively traces the execution path of a specific function or method across multiple files. "
"Identifies call chains and data hand-offs to build an intensive technical map."
),
"parameters": {
"type": "object",
"properties": {
"target": {
"type": "string",
"description": "Fully qualified name of the target (e.g., 'src.ai_client.send') or class.method.",
},
"max_depth": {
"type": "integer",
"description": "Maximum recursion depth for the call graph (default 5).",
},
},
"required": ["target"],
},
}
]
TOOL_NAMES: set[str] = mcp_tool_specs.tool_names()
TOOL_NAMES: set[str] = {t['name'] for t in MCP_TOOL_SPECS}
src/mcp_tool_specs.py
-124
View File
@@ -1,124 +0,0 @@
"""Tool specification module for the Manual Slop MCP tool registry.
Promotes the legacy `MCP_TOOL_SPECS: list[dict[str, Any]]` from
`src/mcp_client.py` to typed dataclass instances. Follows the
`src/vendor_capabilities.py` reference pattern: `frozen=True` dataclass
+ module-level `_REGISTRY` dict + factory functions.
Each tool has:
- name (str): unique tool identifier
- description (str): human-readable purpose
- parameters (tuple[ToolParameter, ...]): the parameter schema
The legacy dict shape (JSON-compatible) is preserved via `to_dict()` so
downstream consumers (provider API requests, comms logging) can still
serialize tool specs to JSON without knowing the dataclass layout.
CONVENTION: 1-space indentation. NO COMMENTS.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
@dataclass(frozen=True)
class ToolParameter:
name: str
type: str
description: str
required: bool = False
enum: tuple[str, ...] | None = None
def to_dict(self) -> dict[str, Any]:
d: dict[str, Any] = {"type": self.type, "description": self.description}
if self.enum is not None:
d["enum"] = list(self.enum)
return d
@dataclass(frozen=True)
class ToolSpec:
name: str
description: str
parameters: tuple[ToolParameter, ...]
def to_dict(self) -> dict[str, Any]:
properties: dict[str, Any] = {p.name: p.to_dict() for p in self.parameters}
required: list[str] = [p.name for p in self.parameters if p.required]
return {
"name": self.name,
"description": self.description,
"parameters": {
"type": "object",
"properties": properties,
"required": required,
},
}
_REGISTRY: dict[str, ToolSpec] = {}
def register(spec: ToolSpec) -> None:
_REGISTRY[spec.name] = spec
def get_tool_spec(name: str) -> ToolSpec:
if name not in _REGISTRY:
raise KeyError(f"No tool registered with name {name!r}")
return _REGISTRY[name]
def get_tool_schemas() -> list[ToolSpec]:
return list(_REGISTRY.values())
def tool_names() -> set[str]:
return set(_REGISTRY.keys())
register(ToolSpec(name='py_remove_def', description='Excises a specific class or function definition from a Python file using AST-derived line ranges, preserving surrounding formatting and comments.', parameters=(ToolParameter( name='path', type='string', description='Path to the .py file.', required=True), ToolParameter( name='name', type='string', description="The name of the class or function to remove. Use 'ClassName.method_name' for methods.", required=True))))
register(ToolSpec(name='py_add_def', description='Inserts a new definition into a specific context (module level or within a specific class).', parameters=(ToolParameter( name='path', type='string', description='Path to the .py file.', required=True), ToolParameter( name='name', type='string', description="Context path (e.g. 'ClassName' or empty for module level).", required=True), ToolParameter( name='new_content', type='string', description='The code to insert.', required=True), ToolParameter( name='anchor_type', type='string', description='Where to insert relative to the anchor.', required=True, enum=('before', 'after', 'top', 'bottom',)), ToolParameter( name='anchor_symbol', type='string', description="Symbol name to anchor to if anchor_type is 'before' or 'after'."))))
register(ToolSpec(name='py_move_def', description='Relocates a definition within a file or across different Python files.', parameters=(ToolParameter( name='src_path', type='string', description='Path to the source .py file.', required=True), ToolParameter( name='dest_path', type='string', description='Path to the destination .py file.', required=True), ToolParameter( name='name', type='string', description='The name of the class or function to move.', required=True), ToolParameter( name='dest_name', type='string', description="Context path in destination file (e.g. 'ClassName' or empty).", required=True), ToolParameter( name='anchor_type', type='string', description='Where to insert in destination.', required=True, enum=('before', 'after', 'top', 'bottom',)), ToolParameter( name='anchor_symbol', type='string', description='Anchor symbol in destination.'))))
register(ToolSpec(name='py_region_wrap', description='Wraps a specified block of code (e.g., a set of methods) in #region: Name and #endregion: Name tags.', parameters=(ToolParameter( name='path', type='string', description='Path to the .py file.', required=True), ToolParameter( name='start_line', type='integer', description='1-based start line number.', required=True), ToolParameter( name='end_line', type='integer', description='1-based end line number (inclusive).', required=True), ToolParameter( name='region_name', type='string', description='The name of the region.', required=True))))
register(ToolSpec(name='read_file', description='Read the full UTF-8 content of a file within the allowed project paths. Use get_file_summary first to decide whether you need the full content.', parameters=(ToolParameter( name='path', type='string', description='Absolute or relative path to the file to read.', required=True),)))
register(ToolSpec(name='list_directory', description='List files and subdirectories within an allowed directory. Shows name, type (file/dir), and size. Use this to explore the project structure.', parameters=(ToolParameter( name='path', type='string', description='Absolute path to the directory to list.', required=True),)))
register(ToolSpec(name='search_files', description="Search for files matching a glob pattern within an allowed directory. Supports recursive patterns like '**/*.py'. Use this to find files by extension or name pattern.", parameters=(ToolParameter( name='path', type='string', description='Absolute path to the directory to search within.', required=True), ToolParameter( name='pattern', type='string', description="Glob pattern, e.g. '*.py', '**/*.toml', 'src/**/*.rs'.", required=True))))
register(ToolSpec(name='get_file_summary', description='Get a compact heuristic summary of a file without reading its full content. For Python: imports, classes, methods, functions, constants. For TOML: table keys. For Markdown: headings. Others: line count + preview. Use this before read_file to decide if you need the full content.', parameters=(ToolParameter( name='path', type='string', description='Absolute or relative path to the file to summarise.', required=True),)))
register(ToolSpec(name='py_get_skeleton', description="Get a skeleton view of a Python file. This returns all classes and function signatures with their docstrings, but replaces function bodies with '...'. Use this to understand module interfaces without reading the full implementation.", parameters=(ToolParameter( name='path', type='string', description='Path to the .py file.', required=True),)))
register(ToolSpec(name='py_get_code_outline', description="Get a hierarchical outline of a code file. This returns classes, functions, and methods with their line ranges and brief docstrings. Use this to quickly map out a file's structure before reading specific sections.", parameters=(ToolParameter( name='path', type='string', description='Path to the code file (currently supports .py).', required=True),)))
register(ToolSpec(name='ts_c_get_skeleton', description="Get a skeleton view of a C file. This returns all function signatures and structs, but replaces function bodies with '...'. Use this to understand C interfaces without reading the full implementation.", parameters=(ToolParameter( name='path', type='string', description='Path to the C file.', required=True),)))
register(ToolSpec(name='ts_cpp_get_skeleton', description="Get a skeleton view of a C++ file. This returns all classes, structs and function signatures, but replaces function bodies with '...'. Use this to understand C++ interfaces without reading the full implementation.", parameters=(ToolParameter( name='path', type='string', description='Path to the C++ file.', required=True),)))
register(ToolSpec(name='ts_c_get_code_outline', description="Get a hierarchical outline of a C file. This returns structs and functions with their line ranges. Use this to quickly map out a file's structure before reading specific sections.", parameters=(ToolParameter( name='path', type='string', description='Path to the C file.', required=True),)))
register(ToolSpec(name='ts_cpp_get_code_outline', description="Get a hierarchical outline of a C++ file. This returns classes, structs and functions with their line ranges. Use this to quickly map out a file's structure before reading specific sections.", parameters=(ToolParameter( name='path', type='string', description='Path to the C++ file.', required=True),)))
register(ToolSpec(name='ts_c_get_definition', description="Get the full source code of a specific function or struct definition in a C file. This is more efficient than reading the whole file if you know what you're looking for.", parameters=(ToolParameter( name='path', type='string', description='Path to the C file.', required=True), ToolParameter( name='name', type='string', description='The name of the function or struct to retrieve.', required=True))))
register(ToolSpec(name='ts_cpp_get_definition', description="Get the full source code of a specific class, function, or method definition in a C++ file. This is more efficient than reading the whole file if you know what you're looking for.", parameters=(ToolParameter( name='path', type='string', description='Path to the C++ file.', required=True), ToolParameter( name='name', type='string', description="The name of the class or function to retrieve. Use 'ClassName::method_name' for methods.", required=True))))
register(ToolSpec(name='ts_c_get_signature', description='Get only the signature part of a C function.', parameters=(ToolParameter( name='path', type='string', description='Path to the C file.', required=True), ToolParameter( name='name', type='string', description='Name of the function.', required=True))))
register(ToolSpec(name='ts_cpp_get_signature', description='Get only the signature part of a C++ function or method.', parameters=(ToolParameter( name='path', type='string', description='Path to the C++ file.', required=True), ToolParameter( name='name', type='string', description="Name of the function/method (e.g. 'ClassName::method_name').", required=True))))
register(ToolSpec(name='ts_c_update_definition', description='Surgically replace the definition of a function in a C file using AST to find line ranges.', parameters=(ToolParameter( name='path', type='string', description='Path to the C file.', required=True), ToolParameter( name='name', type='string', description='Name of function.', required=True), ToolParameter( name='new_content', type='string', description='Complete new source for the definition.', required=True))))
register(ToolSpec(name='ts_cpp_update_definition', description='Surgically replace the definition of a class or function in a C++ file using AST to find line ranges.', parameters=(ToolParameter( name='path', type='string', description='Path to the C++ file.', required=True), ToolParameter( name='name', type='string', description='Name of class/function/method.', required=True), ToolParameter( name='new_content', type='string', description='Complete new source for the definition.', required=True))))
register(ToolSpec(name='get_file_slice', description='Read a specific line range from a file. Useful for reading parts of very large files.', parameters=(ToolParameter( name='path', type='string', description='Path to the file.', required=True), ToolParameter( name='start_line', type='integer', description='1-based start line number.', required=True), ToolParameter( name='end_line', type='integer', description='1-based end line number (inclusive).', required=True))))
register(ToolSpec(name='set_file_slice', description='Replace a specific line range in a file with new content. Surgical edit tool.', parameters=(ToolParameter( name='path', type='string', description='Path to the file.', required=True), ToolParameter( name='start_line', type='integer', description='1-based start line number.', required=True), ToolParameter( name='end_line', type='integer', description='1-based end line number (inclusive).', required=True), ToolParameter( name='new_content', type='string', description='New content to insert.', required=True))))
register(ToolSpec(name='edit_file', description='Replace exact string match in a file. Preserves indentation and line endings. Drop-in replacement for native edit tool.', parameters=(ToolParameter( name='path', type='string', description='Path to the file.', required=True), ToolParameter( name='old_string', type='string', description='The text to replace.', required=True), ToolParameter( name='new_string', type='string', description='The replacement text.', required=True), ToolParameter( name='replace_all', type='boolean', description='Replace all occurrences. Default false.'))))
register(ToolSpec(name='py_get_definition', description="Get the full source code of a specific class, function, or method definition. This is more efficient than reading the whole file if you know what you're looking for.", parameters=(ToolParameter( name='path', type='string', description='Path to the .py file.', required=True), ToolParameter( name='name', type='string', description="The name of the class or function to retrieve. Use 'ClassName.method_name' for methods.", required=True))))
register(ToolSpec(name='py_update_definition', description='Surgically replace the definition of a class or function in a Python file using AST to find line ranges.', parameters=(ToolParameter( name='path', type='string', description='Path to the .py file.', required=True), ToolParameter( name='name', type='string', description='Name of class/function/method.', required=True), ToolParameter( name='new_content', type='string', description='Complete new source for the definition.', required=True))))
register(ToolSpec(name='py_get_signature', description='Get only the signature part of a Python function or method (from def until colon).', parameters=(ToolParameter( name='path', type='string', description='Path to the .py file.', required=True), ToolParameter( name='name', type='string', description="Name of the function/method (e.g. 'ClassName.method_name').", required=True))))
register(ToolSpec(name='py_set_signature', description='Surgically replace only the signature of a Python function or method.', parameters=(ToolParameter( name='path', type='string', description='Path to the .py file.', required=True), ToolParameter( name='name', type='string', description='Name of the function/method.', required=True), ToolParameter( name='new_signature', type='string', description='Complete new signature string (including def and trailing colon).', required=True))))
register(ToolSpec(name='py_get_class_summary', description='Get a summary of a Python class, listing its docstring and all method signatures.', parameters=(ToolParameter( name='path', type='string', description='Path to the .py file.', required=True), ToolParameter( name='name', type='string', description='Name of the class.', required=True))))
register(ToolSpec(name='py_get_var_declaration', description='Get the assignment/declaration line for a variable.', parameters=(ToolParameter( name='path', type='string', description='Path to the .py file.', required=True), ToolParameter( name='name', type='string', description='Name of the variable.', required=True))))
register(ToolSpec(name='py_set_var_declaration', description='Surgically replace a variable assignment/declaration.', parameters=(ToolParameter( name='path', type='string', description='Path to the .py file.', required=True), ToolParameter( name='name', type='string', description='Name of the variable.', required=True), ToolParameter( name='new_declaration', type='string', description='Complete new assignment/declaration string.', required=True))))
register(ToolSpec(name='get_git_diff', description='Returns the git diff for a file or directory. Use this to review changes efficiently without reading entire files.', parameters=(ToolParameter( name='path', type='string', description='Path to the file or directory.', required=True), ToolParameter( name='base_rev', type='string', description="Base revision (e.g. 'HEAD', 'HEAD~1', or a commit hash). Defaults to 'HEAD'."), ToolParameter( name='head_rev', type='string', description='Head revision (optional).'))))
register(ToolSpec(name='web_search', description='Search the web using DuckDuckGo. Returns the top 5 search results with titles, URLs, and snippets. Chain this with fetch_url to read specific pages.', parameters=(ToolParameter( name='query', type='string', description='The search query.', required=True),)))
register(ToolSpec(name='fetch_url', description='Fetch the full text content of a URL (stripped of HTML tags). Use this after web_search to read relevant information from the web.', parameters=(ToolParameter( name='url', type='string', description='The full URL to fetch.', required=True),)))
register(ToolSpec(name='get_ui_performance', description="Get a snapshot of the current UI performance metrics, including FPS, Frame Time (ms), CPU usage (%), and Input Lag (ms). Use this to diagnose UI slowness or verify that your changes haven't degraded the user experience.", parameters=()))
register(ToolSpec(name='py_find_usages', description='Finds exact string matches of a symbol in a given file or directory.', parameters=(ToolParameter( name='path', type='string', description='Path to file or directory to search.', required=True), ToolParameter( name='name', type='string', description='The symbol/string to search for.', required=True))))
register(ToolSpec(name='py_get_imports', description="Parses a file's AST and returns a strict list of its dependencies.", parameters=(ToolParameter( name='path', type='string', description='Path to the .py file.', required=True),)))
register(ToolSpec(name='py_check_syntax', description='Runs a quick syntax check on a Python file.', parameters=(ToolParameter( name='path', type='string', description='Path to the .py file.', required=True),)))
register(ToolSpec(name='py_get_hierarchy', description='Scans the project to find subclasses of a given class.', parameters=(ToolParameter( name='path', type='string', description='Directory path to search in.', required=True), ToolParameter( name='class_name', type='string', description='Name of the base class.', required=True))))
register(ToolSpec(name='py_get_docstring', description='Extracts the docstring for a specific module, class, or function.', parameters=(ToolParameter( name='path', type='string', description='Path to the .py file.', required=True), ToolParameter( name='name', type='string', description="Name of symbol or 'module' for the file docstring.", required=True))))
register(ToolSpec(name='get_tree', description='Returns a directory structure up to a max depth.', parameters=(ToolParameter( name='path', type='string', description='Directory path.', required=True), ToolParameter( name='max_depth', type='integer', description='Maximum depth to recurse (default 2).'))))
register(ToolSpec(name='bd_create', description='Create a new Bead in the active Beads repository.', parameters=(ToolParameter( name='title', type='string', description='Title of the Bead.', required=True), ToolParameter( name='description', type='string', description='Description of the Bead.', required=True))))
register(ToolSpec(name='bd_update', description='Update an existing Bead.', parameters=(ToolParameter( name='bead_id', type='string', description='ID of the Bead to update.', required=True), ToolParameter( name='status', type='string', description='New status for the Bead.', required=True))))
register(ToolSpec(name='bd_list', description='List all Beads in the active Beads repository.', parameters=()))
register(ToolSpec(name='bd_ready', description='Check if the Beads repository is initialized in the current workspace.', parameters=()))
register(ToolSpec(name='derive_code_path', description='Recursively traces the execution path of a specific function or method across multiple files. Identifies call chains and data hand-offs to build an intensive technical map.', parameters=(ToolParameter( name='target', type='string', description="Fully qualified name of the target (e.g., 'src.ai_client.send') or class.method.", required=True), ToolParameter( name='max_depth', type='integer', description='Maximum recursion depth for the call graph (default 5).'))))
src/openai_compatible.py
+46 -78
View File
@@ -1,59 +1,42 @@
"""OpenAI-compatible API client for the Manual Slop ai_client layer.
Provides `send_openai_compatible(client, request, *, capabilities)` which
calls any OpenAI-compatible chat completion endpoint and returns a
`NormalizedResponse` (re-exported from src.openai_schemas).
CONVENTION: 1-space indentation. NO COMMENTS.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Callable, Optional
from openai import (
APIConnectionError,
APIStatusError,
AuthenticationError,
BadRequestError,
OpenAIError,
PermissionDeniedError,
RateLimitError,
)
from openai import OpenAIError, RateLimitError, AuthenticationError, PermissionDeniedError, APIConnectionError, APIStatusError, BadRequestError
from src.openai_schemas import (
ChatMessage,
NormalizedResponse,
OpenAICompatibleRequest,
ToolCall,
ToolCallFunction,
UsageStats,
)
from src.result_types import ErrorInfo, ErrorKind, Result
__all__ = [
"ChatMessage",
"NormalizedResponse",
"OpenAICompatibleRequest",
"ToolCall",
"ToolCallFunction",
"UsageStats",
]
def _to_typed_tool_call(tc: Any) -> ToolCall:
return ToolCall(
id=getattr(tc, "id", "") or "",
type=getattr(tc, "type", "function"),
function=ToolCallFunction(
name=getattr(tc.function, "name", "") or "",
arguments=getattr(tc.function, "arguments", "{}") or "{}",
),
)
def _to_dict_tool_call(tc: ToolCall) -> dict[str, Any]:
return tc.to_dict()
@dataclass(frozen=True)
class NormalizedResponse:
text: str
tool_calls: list[dict[str, Any]]
usage_input_tokens: int
usage_output_tokens: int
usage_cache_read_tokens: int
usage_cache_creation_tokens: int
raw_response: Any
@dataclass
class OpenAICompatibleRequest:
messages: list[dict[str, Any]]
model: str
temperature: float = 0.0
top_p: float = 1.0
max_tokens: int = 8192
tools: Optional[list[dict[str, Any]]] = None
tool_choice: str = "auto"
stream: bool = False
stream_callback: Optional[Callable[[str], None]] = None
extra_body: Optional[dict[str, Any]] = None
def _to_dict_tool_call(tc: Any) -> dict[str, Any]:
return {
"id": getattr(tc, "id", None),
"type": getattr(tc, "type", "function"),
"function": {
"name": getattr(tc.function, "name", None),
"arguments": getattr(tc.function, "arguments", "{}"),
},
}
def _classify_openai_compatible_error(exc: Exception, source: str = "openai_compatible") -> ErrorInfo:
if isinstance(exc, RateLimitError):
@@ -76,17 +59,15 @@ def _classify_openai_compatible_error(exc: Exception, source: str = "openai_comp
return ErrorInfo(kind=ErrorKind.QUOTA, message=str(exc), source=source, original=exc)
return ErrorInfo(kind=ErrorKind.UNKNOWN, message=str(exc), source=source, original=exc)
def send_openai_compatible(
client: Any,
request: OpenAICompatibleRequest,
*,
capabilities: Any,
) -> Result[NormalizedResponse]:
messages_dicts = [m.to_dict() if hasattr(m, "to_dict") else m for m in request.messages]
kwargs: dict[str, Any] = {
"model": request.model,
"messages": messages_dicts,
"messages": request.messages,
"temperature": request.temperature,
"top_p": request.top_p,
"max_tokens": request.max_tokens,
@@ -104,32 +85,27 @@ def send_openai_compatible(
response = _send_blocking(client, kwargs)
return Result(data=response)
except OpenAIError as exc:
empty_resp = NormalizedResponse(
text="",
tool_calls=(),
usage=UsageStats(input_tokens=0, output_tokens=0),
raw_response=None,
)
empty_resp = NormalizedResponse(text="", tool_calls=[], usage_input_tokens=0, usage_output_tokens=0, usage_cache_read_tokens=0, usage_cache_creation_tokens=0, raw_response=None)
return Result(data=empty_resp, errors=[_classify_openai_compatible_error(exc, source="openai_compatible")])
def _send_blocking(client: Any, kwargs: dict[str, Any]) -> NormalizedResponse:
resp = client.chat.completions.create(**kwargs)
msg = resp.choices[0].message
tool_calls_raw = msg.tool_calls or []
tool_calls: tuple[ToolCall, ...] = tuple(_to_typed_tool_call(tc) for tc in tool_calls_raw)
tool_calls: list[dict[str, Any]] = []
for tc in tool_calls_raw:
tool_calls.append(_to_dict_tool_call(tc))
usage = getattr(resp, "usage", None)
return NormalizedResponse(
text=msg.content or "",
tool_calls=tool_calls,
usage=UsageStats(
input_tokens=int(getattr(usage, "prompt_tokens", 0) or 0),
output_tokens=int(getattr(usage, "completion_tokens", 0) or 0),
),
usage_input_tokens=int(getattr(usage, "prompt_tokens", 0) or 0),
usage_output_tokens=int(getattr(usage, "completion_tokens", 0) or 0),
usage_cache_read_tokens=0,
usage_cache_creation_tokens=0,
raw_response=resp,
)
def _send_streaming(client: Any, kwargs: dict[str, Any], callback: Optional[Callable[[str], None]]) -> NormalizedResponse:
kwargs_stream = dict(kwargs)
kwargs_stream["stream"] = True
@@ -163,20 +139,12 @@ def _send_streaming(client: Any, kwargs: dict[str, Any], callback: Optional[Call
if chunk_usage is not None:
usage_input = int(getattr(chunk_usage, "prompt_tokens", 0) or 0)
usage_output = int(getattr(chunk_usage, "completion_tokens", 0) or 0)
tool_calls_typed: tuple[ToolCall, ...] = tuple(
ToolCall(
id=acc["id"] or "",
type=acc["type"],
function=ToolCallFunction(
name=acc["function"]["name"] or "",
arguments=acc["function"]["arguments"] or "{}",
),
)
for acc in (tool_calls_acc[k] for k in sorted(tool_calls_acc.keys()))
)
return NormalizedResponse(
text="".join(text_parts),
tool_calls=tool_calls_typed,
usage=UsageStats(input_tokens=usage_input, output_tokens=usage_output),
tool_calls=[tool_calls_acc[k] for k in sorted(tool_calls_acc.keys())],
usage_input_tokens=usage_input,
usage_output_tokens=usage_output,
usage_cache_read_tokens=0,
usage_cache_creation_tokens=0,
raw_response=None,
)
src/openai_schemas.py
-105
View File
@@ -1,105 +0,0 @@
"""OpenAI-compatible dataclasses for the Manual Slop ai_client layer.
Promotes `NormalizedResponse` and `OpenAICompatibleRequest` from
`src/openai_compatible.py` to typed dataclasses. The 4 dataclasses
here model the OpenAI Chat Completion API shape:
- ToolCall: a single tool call from the model
- ToolCallFunction: the function portion of a tool call (name + JSON args)
- ChatMessage: a single message in the conversation (system/user/assistant/tool)
- UsageStats: token usage accounting (input, output, cache hits/creation)
`NormalizedResponse` and `OpenAICompatibleRequest` keep their public
shapes but consume these typed shapes internally.
CONVENTION: 1-space indentation. NO COMMENTS.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Callable, Optional
@dataclass(frozen=True)
class ToolCallFunction:
name: str
arguments: str
@dataclass(frozen=True)
class ToolCall:
id: str
function: ToolCallFunction
type: str = "function"
def to_dict(self) -> dict[str, Any]:
return {
"id": self.id,
"type": self.type,
"function": {
"name": self.function.name,
"arguments": self.function.arguments,
},
}
@dataclass(frozen=True)
class ChatMessage:
role: str
content: str
tool_calls: Optional[tuple[ToolCall, ...]] = None
tool_call_id: Optional[str] = None
name: Optional[str] = None
def to_dict(self) -> dict[str, Any]:
d: dict[str, Any] = {"role": self.role, "content": self.content}
if self.tool_calls is not None:
d["tool_calls"] = [tc.to_dict() for tc in self.tool_calls]
if self.tool_call_id is not None:
d["tool_call_id"] = self.tool_call_id
if self.name is not None:
d["name"] = self.name
return d
@dataclass(frozen=True)
class UsageStats:
input_tokens: int
output_tokens: int
cache_read_tokens: int = 0
cache_creation_tokens: int = 0
@dataclass(frozen=True)
class NormalizedResponse:
text: str
tool_calls: tuple[ToolCall, ...]
usage: UsageStats
raw_response: Any
def to_legacy_dict(self) -> dict[str, Any]:
return {
"text": self.text,
"tool_calls": [tc.to_dict() for tc in self.tool_calls],
"usage": {
"input_tokens": self.usage.input_tokens,
"output_tokens": self.usage.output_tokens,
"cache_read_tokens": self.usage.cache_read_tokens,
"cache_creation_tokens": self.usage.cache_creation_tokens,
},
"raw_response": self.raw_response,
}
@dataclass
class OpenAICompatibleRequest:
messages: list[ChatMessage]
model: str
temperature: float = 0.0
top_p: float = 1.0
max_tokens: int = 8192
tools: Optional[list[dict[str, Any]]] = None
tool_choice: str = "auto"
stream: bool = False
stream_callback: Optional[Callable[[str], None]] = None
extra_body: Optional[dict[str, Any]] = None
src/provider_state.py
-69
View File
@@ -1,69 +0,0 @@
"""Per-provider history state for the AI client layer.
Promotes 14 module globals in src/ai_client.py:
- 7x `_<provider>_history: list[Metadata]` (anthropic/deepseek/minimax/qwen/grok/llama)
- 7x `_<provider>_history_lock: threading.Lock`
To a single `_PROVIDER_HISTORIES: dict[str, ProviderHistory]` keyed by
provider name. Each `ProviderHistory` owns its own lock and message list;
the cross-provider pattern is encapsulated behind a 4-method interface.
SDK client holders (`_gemini_chat`, `_deepseek_client`, etc.) stay as
module-level `Any` variables per Pattern 3 (heterogeneous SDK types,
lazy-initialized). Only the homogeneous history aspect is unified.
CONVENTION: 1-space indentation. NO COMMENTS.
"""
from __future__ import annotations
import threading
from dataclasses import dataclass, field
from src.type_aliases import HistoryMessage, Metadata
@dataclass
class ProviderHistory:
messages: list[HistoryMessage] = field(default_factory=list)
lock: threading.Lock = field(default_factory=threading.Lock)
def append(self, message: HistoryMessage) -> None:
with self.lock:
self.messages.append(message)
def get_all(self) -> list[HistoryMessage]:
with self.lock:
return list(self.messages)
def replace_all(self, messages: list[HistoryMessage]) -> None:
with self.lock:
self.messages = list(messages)
def clear(self) -> None:
with self.lock:
self.messages = []
_PROVIDER_HISTORIES: dict[str, ProviderHistory] = {
"anthropic": ProviderHistory(),
"deepseek": ProviderHistory(),
"minimax": ProviderHistory(),
"qwen": ProviderHistory(),
"grok": ProviderHistory(),
"llama": ProviderHistory(),
}
def get_history(provider: str) -> ProviderHistory:
if provider not in _PROVIDER_HISTORIES:
raise KeyError(f"Unknown provider: {provider!r}")
return _PROVIDER_HISTORIES[provider]
def clear_all() -> None:
for h in _PROVIDER_HISTORIES.values():
h.clear()
def providers() -> tuple[str, ...]:
return tuple(_PROVIDER_HISTORIES.keys())
src/type_aliases.py
-3
View File
@@ -18,9 +18,6 @@ ToolCall: TypeAlias = Metadata
CommsLogCallback: TypeAlias = Callable[[CommsLogEntry], None]
JsonPrimitive: TypeAlias = str | int | float | bool | None
JsonValue: TypeAlias = JsonPrimitive | list["JsonValue"] | dict[str, "JsonValue"]
class FileItemsDiff(NamedTuple):
refreshed: FileItems