ed/manual_slop
Private
Public Access
0
0
Fork 0
Code Activity
Files
0d1239601170914aa3d85c9830541ab4ce6d4665
manual_slop/docs/guide_tools.md
T

24 KiB
Raw Blame History

Tooling & IPC Technical Reference

Top | Architecture | MMA Orchestration | Simulations

The MCP Bridge: Filesystem Security

The AI's ability to interact with the filesystem is mediated by a three-layer security model in mcp_client.py. Every tool accessing the disk passes through _resolve_and_check(path) before any I/O occurs.

Global State

_allowed_paths: set[Path] = set()        # Explicit file allowlist (resolved absolutes)
_base_dirs: set[Path] = set()             # Directory roots for containment checks
_primary_base_dir: Path | None = None     # Used for resolving relative paths
perf_monitor_callback: Optional[Callable[[], dict[str, Any]]] = None

Layer 1: Allowlist Construction (configure)

Called by ai_client before each send cycle. Takes file_items (from aggregate.build_file_items()) and optional extra_base_dirs.

  1. Resets _allowed_paths and _base_dirs to empty sets on every call.
  2. Sets _primary_base_dir from extra_base_dirs[0] (resolved) or falls back to Path.cwd().
  3. Iterates all file_items, resolving each item["path"] to an absolute path. Each resolved path is added to _allowed_paths; its parent directory is added to _base_dirs.
  4. Any entries in extra_base_dirs that are valid directories are also added to _base_dirs.

Layer 2: Path Validation (_is_allowed)

Checks run in this exact order:

  1. Blacklist (hard deny): If filename is history.toml or ends with _history.toml, return False. Prevents the AI from reading conversation history.
  2. Explicit allowlist: If resolved path is in _allowed_paths, return True.
  3. CWD fallback: If _base_dirs is empty, any path under cwd() is allowed.
  4. Base directory containment: Path must be a subpath of at least one entry in _base_dirs (via relative_to()).
  5. Default deny: All other paths are rejected.

All paths are resolved (following symlinks) before comparison, preventing symlink-based traversal.

Layer 3: Resolution Gate (_resolve_and_check)

Every tool call passes through this:

  1. Convert raw path string to Path.
  2. If not absolute, prepend _primary_base_dir.
  3. Resolve to absolute.
  4. Call _is_allowed().
  5. Return (resolved_path, "") on success or (None, error_message) on failure.

The error message includes the full list of allowed base directories for debugging.

Native Tool Inventory

The dispatch function (mcp_client.py:1341) is a flat if/elif chain mapping 45 tool names to implementations. All tools are categorized below with their parameters and behavior. The count grew from the original 26 as Python structural tools (Phase 6) and C/C++ AST tools (Phase 6.5) were added; Beads tools (4) were added for the Beads mode integration.

File I/O Tools

Tool Parameters Description
read_file path UTF-8 file content extraction
list_directory path Compact table: [file/dir] name size. Applies blacklist filter to entries.
search_files path, pattern Glob pattern matching within an allowed directory. Applies blacklist filter.
get_file_slice path, start_line, end_line Returns specific line range (1-based, inclusive)
set_file_slice path, start_line, end_line, new_content Replaces a line range with new content (surgical edit)
edit_file path, old_string, new_string, replace_all Replaces an exact string match in a file. Preserves indentation and line endings. Drop-in replacement for the legacy native edit tool.
get_tree path, max_depth Directory structure up to max_depth levels

AST-Based Tools (Python only)

These use file_cache.ASTParser (tree-sitter) or stdlib ast for structural code analysis:

Tool Parameters Description
py_get_skeleton path Signatures + docstrings, bodies replaced with .... Uses tree-sitter.
py_get_code_outline path Hierarchical outline: [Class] Name (Lines X-Y) with nested methods. Uses stdlib ast.
py_get_definition path, name Full source of a specific class/function/method. Supports ClassName.method dot notation.
py_update_definition path, name, new_content Surgical replacement: locates symbol via ast, delegates to set_file_slice.
py_get_signature path, name Only the def line through the colon.
py_set_signature path, name, new_signature Replaces only the signature, preserving body.
py_get_class_summary path, name Class docstring + list of method signatures.
py_get_var_declaration path, name Module-level or class-level variable assignment line(s).
py_set_var_declaration path, name, new_declaration Surgical variable replacement.
py_find_usages path, name Exact string match search across a file or directory.
py_get_imports path Parses AST, returns strict dependency list.
py_check_syntax path Quick syntax validation via ast.parse().
py_get_hierarchy path, class_name Scans directory for subclasses of a given class.
py_get_docstring path, name Extracts docstring for module, class, or function.
py_remove_def path, name Excises a definition using AST boundaries.
py_add_def path, name, new_content, anchor_type, anchor_symbol Inserts code with automatic 1-space indentation normalization.
py_move_def src_path, dest_path, name, dest_name, anchor_type, anchor_symbol Relocates code across files/contexts.
py_region_wrap path, start_line, end_line, region_name Wraps line range in #region / #endregion tags.

C/C++ AST Tools

These use tree_sitter via src/mcp_client.py for structural analysis of C and C++ codebases. Phase 6 added these tools to support the Granular AST Control feature.

Tool Parameters Description
ts_c_get_skeleton path C/C++ function signatures and struct definitions, bodies replaced with ....
ts_cpp_get_skeleton path C++ class/struct signatures, method signatures, and inheritance info.
ts_c_get_code_outline path Hierarchical C outline: [Struct] Name (Lines X-Y) with nested members.
ts_cpp_get_code_outline path Hierarchical C++ outline with classes, methods, inheritance hierarchy.
ts_c_get_definition path, name Full source of a specific C struct or function.
ts_cpp_get_definition path, name Full source of a specific C++ class, struct, or method. Supports ClassName::method notation.
ts_c_update_definition path, name, new_content Surgical replacement for C definitions.
ts_cpp_update_definition path, name, new_content Surgical replacement for C++ definitions.
ts_c_get_signature path, name Only the function/struct declaration line.
ts_cpp_get_signature path, name Only the method/function declaration line.

Usage for Context Curation:

# Fetch outline for AST inspection modal
outline = mcp_client.ts_cpp_get_code_outline("path/to/file.hpp")

# Fetch specific definition for masked inclusion
defn = mcp_client.ts_cpp_get_definition("path/to/file.hpp", "MyClass::init")

# Apply per-symbol masking via FuzzyAnchor
from src.fuzzy_anchor import FuzzyAnchor
slice_data = FuzzyAnchor.create_slice(content, start_line, end_line)
resolved = FuzzyAnchor.resolve_slice(modified_content, slice_data)

Analysis Tools

Tool Parameters Description
get_file_summary path Heuristic summary via summarize.py: imports, classes, functions, constants for .py; table keys for .toml; headings for .md.
get_git_diff path, base_rev, head_rev Git diff output for a file or directory.
derive_code_path target, max_depth Recursively traces the execution path of a specific function or method across multiple files. Returns a multi-level call graph.

Network Tools

Tool Parameters Description
web_search query Scrapes DuckDuckGo HTML via dependency-free _DDGParser (HTMLParser subclass). Returns top 5 results with title, URL, snippet.
fetch_url url Fetches URL content, strips HTML tags via _TextExtractor.

Runtime Tools

Tool Parameters Description
get_ui_performance (none) Returns FPS, Frame Time, CPU, Input Lag via injected perf_monitor_callback. No security check (no filesystem access).

Beads Tools

These tools wrap the Beads Dolt-backed issue tracker. They are dispatched by mcp_client.dispatch (line 1474) when the tool name starts with bd_. If no active Beads workspace is detected, all Beads tools return "ERROR: no active workspace to run beads tools."

Tool Parameters Description
bd_list (none) Lists all beads in the active .beads/ repository. Returns ID: <id>, Status: <status>, Title: <title> per row.
bd_create title, description Creates a new bead in the active repository.
bd_update bead_id, status Updates the status of an existing bead (open, in_progress, closed, etc.).
bd_ready (none) Lists beads that are ready to be worked on (dependencies satisfied, not blocked).

The BeadsClient is instantiated per-call from _primary_base_dir of the MCP allowlist. The dispatch layer requires the base directory to be the project root (where .beads/ lives).

See guide_beads.md (placeholder; written in Task 10) for the full Beads client API and the integration roadmap.

Tool Implementation Patterns

AST-based read tools follow this pattern:

def py_get_skeleton(path: str) -> str:
    p, err = _resolve_and_check(path)
    if err: return err
    if not p.exists(): return f"ERROR: file not found: {path}"
    if not p.is_file() or p.suffix != ".py": return f"ERROR: not a python file: {path}"
    from file_cache import ASTParser
    code = p.read_text(encoding="utf-8")
    parser = ASTParser("python")
    return parser.get_skeleton(code)

AST-based write tools use stdlib ast (not tree-sitter) to locate symbols, then delegate to set_file_slice:

def py_update_definition(path: str, name: str, new_content: str) -> str:
    p, err = _resolve_and_check(path)
    if err: return err
    code = p.read_text(encoding="utf-8").lstrip(chr(0xFEFF))  # Strip BOM
    tree = ast.parse(code)
    node = _get_symbol_node(tree, name)          # Walks AST for matching node
    if not node: return f"ERROR: could not find definition '{name}'"
    start = getattr(node, "lineno")
    end = getattr(node, "end_lineno")
    return set_file_slice(path, start, end, new_content)

The _get_symbol_node helper supports dot notation (ClassName.method_name) by first finding the class, then searching its body for the method.

Parallel Tool Execution

Tools can be executed concurrently via async_dispatch:

async def async_dispatch(tool_name: str, tool_input: dict[str, Any]) -> str:
    """Dispatch an MCP tool call asynchronously."""
    return await asyncio.to_thread(dispatch, tool_name, tool_input)

In ai_client.py, multiple tool calls within a single AI turn are executed in parallel:

async def _execute_tool_calls_concurrently(calls, base_dir, ...):
    tasks = []
    for fc in calls:
        tasks.append(_execute_single_tool_call_async(name, args, ...))
    results = await asyncio.gather(*tasks)
    return results

This significantly reduces latency when the AI makes multiple independent file reads in a single turn.

Thread Safety Note: The configure() function resets global state. In concurrent environments, ensure configuration is complete before dispatching tools.

The Hook API: Remote Control & Telemetry

Manual Slop exposes a REST-based IPC interface on 127.0.0.1:8999 using Python's ThreadingHTTPServer. Each incoming request gets its own thread.

Server Architecture

class HookServerInstance(ThreadingHTTPServer):
    app: Any           # Reference to main App instance

class HookHandler(BaseHTTPRequestHandler):
    # Accesses self.server.app for all state

class HookServer:
    app: Any
    port: int = 8999
    server: HookServerInstance | None
    thread: threading.Thread | None

Start conditions: Only starts if app.test_hooks_enabled == True OR current provider is 'gemini_cli'. Otherwise start() silently returns.

Initialization: On start, ensures the app has _pending_gui_tasks + lock, _pending_asks + _ask_responses dicts, and _api_event_queue + lock.

GUI Thread Trampoline Pattern

The HookServer never reads GUI state directly (thread safety). For state reads, it uses a trampoline:

  1. Create a threading.Event() and a result dict.
  2. Push a custom_callback closure into _pending_gui_tasks that reads state and calls event.set().
  3. Block on event.wait(timeout=60).
  4. Return result as JSON, or 504 on timeout.

This ensures all state reads happen on the GUI main thread during _process_pending_gui_tasks.

GET Endpoints

Endpoint Thread Safety Response
GET /status Direct (stateless) {"status": "ok"}
GET /api/project Direct read {"project": <flat_config>} via project_manager.flat_config()
GET /api/session Direct read {"session": {"entries": [...]}} from app.disc_entries
GET /api/performance Direct read {"performance": <metrics>} from app.perf_monitor.get_metrics()
GET /api/events Lock-guarded drain {"events": [...]} — drains and clears _api_event_queue
GET /api/gui/value GUI trampoline {"value": <val>} — reads from _settable_fields map
GET /api/gui/value/<tag> GUI trampoline Same, via URL path param
GET /api/gui/mma_status GUI trampoline Full MMA state dict (see below)
GET /api/gui/diagnostics GUI trampoline {thinking, live, prior} booleans

/api/gui/mma_status response fields:

{
    "mma_status":                    str,   # "idle" | "planning" | "executing" | "done"
    "ai_status":                     str,   # "idle" | "sending..." | etc.
    "active_tier":                   str | None,
    "active_track":                  str,   # Track ID or raw value
    "active_tickets":                list,  # Serialized ticket dicts
    "mma_step_mode":                 bool,
    "pending_tool_approval":         bool,  # _pending_ask_dialog
    "pending_mma_step_approval":     bool,  # _pending_mma_approval is not None
    "pending_mma_spawn_approval":    bool,  # _pending_mma_spawn is not None
    "pending_approval":              bool,  # Backward compat: step OR tool
    "pending_spawn":                 bool,  # Alias for spawn approval
    "tracks":                        list,
    "proposed_tracks":               list,
    "mma_streams":                   dict,  # {stream_id: output_text}
}

/api/gui/diagnostics response fields:

{
    "thinking": bool,   # ai_status in ["sending...", "running powershell..."]
    "live":     bool,   # ai_status in ["running powershell...", "fetching url...", ...]
    "prior":    bool,   # app.is_viewing_prior_session
}

POST Endpoints

Endpoint Body Response Effect
POST /api/project {"project": {...}} {"status": "updated"} Sets app.project
POST /api/session {"session": {"entries": [...]}} {"status": "updated"} Sets app.disc_entries
POST /api/gui Any JSON dict {"status": "queued"} Appends to _pending_gui_tasks
POST /api/ask Any JSON dict {"status": "ok", "response": ...} or 504 Blocking ask dialog
POST /api/ask/respond {"request_id": ..., "response": ...} {"status": "ok"} or 404 Resolves a pending ask

The /api/ask Protocol (Synchronous HITL via HTTP)

This is the most complex endpoint — it implements a blocking request-response dialog over HTTP:

  1. Generate a UUID request_id.
  2. Create a threading.Event.
  3. Register in app._pending_asks[request_id] = event.
  4. Push an ask_received event to _api_event_queue (for client discovery).
  5. Append {"type": "ask", "request_id": ..., "data": ...} to _pending_gui_tasks.
  6. Block on event.wait(timeout=60.0).
  7. On signal: read app._ask_responses[request_id], clean up, return 200.
  8. On timeout: clean up, return 504.

The counterpart /api/ask/respond:

  1. Look up request_id in app._pending_asks.
  2. Store response in app._ask_responses[request_id].
  3. Signal the event (event.set()).
  4. Queue a clear_ask GUI task.
  5. Return 200 (or 404 if request_id not found).

ApiHookClient: The Automation Interface

api_hook_client.py provides a synchronous Python client for the Hook API, used by test scripts and external tooling.

class ApiHookClient:
    def __init__(self, base_url="http://127.0.0.1:8999", max_retries=5, retry_delay=0.2)

Connection Methods

Method Description
wait_for_server(timeout=3) Polls /status with exponential backoff until server is ready.
_make_request(method, endpoint, data, timeout) Core HTTP client with retry logic.

State Query Methods

Method Endpoint Description
get_status() GET /status Health check
get_project() GET /api/project Full project config
get_session() GET /api/session Discussion entries
get_mma_status() GET /api/gui/mma_status Full MMA orchestration state
get_performance() GET /api/performance UI metrics (FPS, CPU, etc.)
get_value(item) GET /api/gui/value/<item> Read any _settable_fields value
get_text_value(item_tag) Wraps get_value Returns string representation or None
get_events() GET /api/events Fetches and clears the event queue
get_indicator_state(tag) GET /api/gui/diagnostics Checks if an indicator is shown
get_node_status(node_tag) Two-phase: get_value then diagnostics DAG node status with fallback

GUI Manipulation Methods

Method Endpoint Description
set_value(item, value) POST /api/gui Sets any _settable_fields value; special-cases current_provider and gcli_path
click(item, *args, **kwargs) POST /api/gui Simulates button click; passes optional user_data
select_tab(tab_bar, tab) POST /api/gui Switches to a specific tab
select_list_item(listbox, item_value) POST /api/gui Selects an item in a listbox
push_event(event_type, payload) POST /api/gui Pushes event into AsyncEventQueue
post_gui(gui_data) POST /api/gui Raw task dict injection
reset_session() Clicks btn_reset_session Simulates clicking the Reset Session button

Polling Methods

Method Description
wait_for_event(event_type, timeout=5) Polls get_events() until a matching event type appears.
wait_for_value(item, expected, timeout=5) Polls get_value(item) until it equals expected.

HITL Method

Method Description
request_confirmation(tool_name, args) Sends to /api/ask, blocks until user responds via the GUI dialog.

Parallel Tool Execution

Tool calls are executed concurrently within a single AI turn using asyncio.gather. This significantly reduces latency when multiple independent tools need to be called.

async_dispatch Implementation

async def async_dispatch(tool_name: str, tool_input: dict[str, Any]) -> str:
    """
    Dispatch an MCP tool call by name asynchronously. 
    Returns the result as a string.
    """
    # Run blocking I/O bound tools in a thread to allow parallel execution
    return await asyncio.to_thread(dispatch, tool_name, tool_input)

All tools are wrapped in asyncio.to_thread() to prevent blocking the event loop. This enables ai_client.py to execute multiple tools via asyncio.gather():

results = await asyncio.gather(
    async_dispatch("read_file", {"path": "src/module_a.py"}),
    async_dispatch("read_file", {"path": "src/module_b.py"}),
    async_dispatch("get_file_summary", {"path": "src/module_c.py"}),
)

Concurrency Benefits

Scenario Sequential Parallel
3 file reads (100ms each) 300ms ~100ms
5 file reads + 1 web fetch (200ms each) 1200ms ~200ms
Mixed I/O operations Sum of all Max of all

The parallel execution model is particularly effective for:

  • Reading multiple source files simultaneously
  • Fetching URLs while performing local file operations
  • Running syntax checks across multiple files

Synthetic Context Refresh

To minimize token churn and redundant read_file calls, the ai_client performs a post-tool-execution context refresh. See guide_architecture.md for the full algorithm.

Summary:

  1. Detection: Triggered after the final tool call in each reasoning round.
  2. Collection: Re-reads all project-tracked files, comparing mtimes.
  3. Injection: Changed files are diffed and appended as [SYSTEM: FILES UPDATED] to the last tool output.
  4. Pruning: Older [FILES UPDATED] blocks are stripped from history in subsequent rounds.

Session Logging

session_logger.py opens timestamped log files at GUI startup and keeps them open for the process lifetime.

File Layout

logs/sessions/<session_id>/
    comms.log       # JSON-L: every API interaction (direction, kind, payload)
    toolcalls.log   # Markdown: sequential tool invocation records
    apihooks.log    # API hook invocations
    clicalls.log    # JSON-L: CLI subprocess details (command, stdin, stdout, stderr, latency)

scripts/generated/
    <ts>_<seq:04d>.ps1   # Each AI-generated PowerShell script, preserved in order

Logging Functions

Function Target Format
log_comms(entry) comms.log JSON-L line per entry
log_tool_call(script, result, script_path) toolcalls.log + scripts/generated/ Markdown record + preserved .ps1 file
log_api_hook(method, path, body) apihooks.log Timestamped text line
log_cli_call(command, stdin, stdout, stderr, latency) clicalls.log JSON-L with latency tracking

Lifecycle

  • open_session(label): Called once at GUI startup. Idempotent (checks if already open). Registers atexit.register(close_session).
  • close_session(): Flushes and closes all file handles.

Shell Runner

shell_runner.py executes PowerShell scripts with environment configuration, timeout handling, and optional QA integration.

Environment Configuration via mcp_env.toml

[path]
prepend = ["C:/custom/bin", "C:/other/tools"]

[env]
MY_VAR = "some_value"
EXPANDED = "${HOME}/subdir"

_build_subprocess_env() copies os.environ, prepends [path].prepend entries to PATH, and sets [env] key-value pairs with ${VAR} expansion.

run_powershell(script, base_dir, qa_callback=None) -> str

  1. Prepends Set-Location -LiteralPath '<base_dir>' (with escaped single quotes).
  2. Locates PowerShell: tries powershell.exe, pwsh.exe, powershell, pwsh in order.
  3. Runs via subprocess.Popen([exe, "-NoProfile", "-NonInteractive", "-Command", full_script]).
  4. process.communicate(timeout=60) — 60-second hard timeout.
  5. On TimeoutExpired: kills process tree via taskkill /F /T /PID, returns "ERROR: timed out after 60s".
  6. Returns combined output: STDOUT:\n<out>\nSTDERR:\n<err>\nEXIT CODE: <code>.
  7. If qa_callback provided and command failed: appends QA ANALYSIS:\n<qa_callback(stderr)> — integrates Tier 4 QA error analysis directly.
View Git Blame Copy Permalink