manual_slop/docs/guide_tools.md

# Tooling & IPC Technical Reference

[Top](../Readme.md) | [Architecture](guide_architecture.md) | [MMA Orchestration](guide_mma.md) | [Simulations](guide_simulations.md)

---

## 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

```python
_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:**

```python
# 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](https://github.com/steveyegge/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](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:
```python
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`:
```python
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`:

```python
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:

```python
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

```python
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:**

```python
{
    "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:**

```python
{
    "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.

```python
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

```python
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()`:

```python
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](guide_architecture.md#context-refresh-mechanism) 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`

```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.