manual_slop/docs/guide_tools.md

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 (line 806) is a flat if/elif chain mapping 26 tool names to implementations. All tools are categorized below with their parameters and behavior.

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

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.

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

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.

---

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.

---

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.