diff --git a/MCP_Server/cache/browser.py b/MCP_Server/cache/browser.py index 500dfa8..3201e3f 100644 --- a/MCP_Server/cache/browser.py +++ b/MCP_Server/cache/browser.py @@ -11,7 +11,7 @@ import time import logging import threading -from typing import Dict, Any, List +from typing import Dict, Any, List, Optional from collections import deque import MCP_Server.state as state @@ -166,7 +166,10 @@ def load_browser_cache_from_disk() -> bool: # Live browser scan # --------------------------------------------------------------------------- -def populate_browser_cache(force: bool = False) -> bool: +def populate_browser_cache( + force: bool = False, + stop_event: Optional[threading.Event] = None, +) -> bool: """Scan Ableton's browser tree and cache all items for instant search. Uses a breadth-first walk up to depth 3 across 11 browser categories. @@ -176,7 +179,7 @@ def populate_browser_cache(force: bool = False) -> bool: Uses a **dedicated TCP connection** to avoid corrupting the shared global connection when the BFS scan sends many rapid commands. """ - from MCP_Server.connections.ableton import AbletonConnection + from MCP_Server.connections.ableton import AbletonConnection, CommandCancelled now = time.time() with state.browser_cache_lock: @@ -191,6 +194,8 @@ def populate_browser_cache(force: bool = False) -> bool: ableton = AbletonConnection(host="localhost", port=9877) try: + if stop_event is not None and stop_event.is_set(): + return False try: if not ableton.connect(): logger.warning("Browser cache: cannot connect to Ableton") @@ -199,12 +204,17 @@ def populate_browser_cache(force: bool = False) -> bool: logger.warning("Browser cache: cannot connect to Ableton: %s", e) return False + if stop_event is not None and stop_event.is_set(): + return False + logger.info("Browser cache: starting scan...") flat_items: List[Dict[str, Any]] = [] by_display: Dict[str, List[Dict[str, Any]]] = {} total = 0 for path_root, display_name in BROWSER_CATEGORIES: + if stop_event is not None and stop_event.is_set(): + return False category_items: List[Dict[str, Any]] = [] cat_count = 0 @@ -212,14 +222,30 @@ def populate_browser_cache(force: bool = False) -> bool: queue = deque([(path_root, 0)]) while queue and cat_count < BROWSER_CACHE_MAX_ITEMS: + if stop_event is not None and stop_event.is_set(): + return False current_path, depth = queue.popleft() try: - result = ableton.send_command("get_browser_items_at_path", {"path": current_path}, timeout=60.0) + result = ableton.send_command( + "get_browser_items_at_path", + {"path": current_path}, + timeout=60.0, + stop_event=stop_event, + ) + except CommandCancelled: + logger.info("Browser cache scan cancelled during shutdown") + return False except Exception as e: + if stop_event is not None and stop_event.is_set(): + return False logger.warning("Browser cache: failed to read '%s': %s", current_path, e) # Try to re-establish connection before continuing - time.sleep(2) + if stop_event is not None: + if stop_event.wait(2.0): + return False + else: + time.sleep(2.0) try: ableton.disconnect() if not ableton.connect(): @@ -262,13 +288,19 @@ def populate_browser_cache(force: bool = False) -> bool: queue.append((item_path, depth + 1)) # Rate-limit to avoid overwhelming Ableton's socket handler - time.sleep(0.01) + if stop_event is not None: + if stop_event.wait(0.01): + return False + else: + time.sleep(0.01) by_display[display_name] = category_items logger.info("Browser cache: '%s' — %d items", display_name, len(category_items)) - device_map = build_device_uri_map(flat_items) + if stop_event is not None and stop_event.is_set(): + return False + device_map = build_device_uri_map(flat_items) with state.browser_cache_lock: state.browser_cache_flat = flat_items state.browser_cache_by_category = by_display diff --git a/MCP_Server/connections/ableton.py b/MCP_Server/connections/ableton.py index f0fcd78..c24ce2b 100644 --- a/MCP_Server/connections/ableton.py +++ b/MCP_Server/connections/ableton.py @@ -1,8 +1,9 @@ """AbletonConnection — TCP socket connection to the Ableton Remote Script.""" -import socket import json import logging +import select +import socket import time import threading from dataclasses import dataclass @@ -24,6 +25,10 @@ ]) +class CommandCancelled(RuntimeError): + """Raised when cooperative shutdown cancels an in-flight command.""" + + @dataclass class AbletonConnection: host: str @@ -35,6 +40,7 @@ class AbletonConnection: def connect(self) -> bool: """Connect to the Ableton Remote Script socket server""" if self.sock: + self._last_socket_open = True return True try: @@ -42,6 +48,7 @@ def connect(self) -> bool: self.sock.settimeout(5.0) self.sock.connect((self.host, self.port)) self._recv_buffer = "" # Clear buffer on new connection + self._last_socket_open = True logger.info("Connected to Ableton at %s:%s", self.host, self.port) return True except Exception as e: @@ -52,6 +59,7 @@ def connect(self) -> bool: except Exception: pass self.sock = None + self._last_socket_open = False return False def disconnect(self): @@ -63,6 +71,7 @@ def disconnect(self): logger.error("Error disconnecting from Ableton: %s", e) finally: self.sock = None + self._last_socket_open = False if self._udp_sock: try: self._udp_sock.close() @@ -72,8 +81,48 @@ def disconnect(self): self._udp_sock = None def __post_init__(self): + """Initialize per-connection receive buffering and send serialization.""" self._recv_buffer = "" self._send_lock = threading.Lock() + self._last_socket_open = self.sock is not None + + def is_connected(self) -> bool: + """Passively check whether the Remote Script socket is still open. + + The check never sends or consumes protocol data. If a command currently + owns the send lock, return the last verified socket result so a status + request cannot interfere with its response handling. + """ + sock = self.sock + if sock is None: + self._last_socket_open = False + return False + + try: + sock.getpeername() + except OSError: + self._last_socket_open = False + return False + + if not self._send_lock.acquire(blocking=False): + return self._last_socket_open + + try: + readable, _writable, _exceptional = select.select([sock], [], [], 0) + if not readable: + self._last_socket_open = True + return True + try: + self._last_socket_open = bool(sock.recv(1, socket.MSG_PEEK)) + except (BlockingIOError, socket.timeout): + self._last_socket_open = True + except OSError: + self._last_socket_open = False + except (OSError, ValueError): + self._last_socket_open = False + finally: + self._send_lock.release() + return self._last_socket_open def _ensure_udp_socket(self): """Create a UDP socket for real-time parameter sending if not already open.""" @@ -95,12 +144,22 @@ def send_udp_command(self, command_type: str, params: Dict[str, Any] = None): sock.sendto(payload, (self.host, self._udp_port)) logger.debug("Sent UDP command: %s", command_type) - def receive_full_response(self, sock, buffer_size=8192, timeout=15.0): + def receive_full_response( + self, + sock, + buffer_size=8192, + timeout=15.0, + stop_event: Optional[threading.Event] = None, + ): """Receive a complete newline-delimited JSON response and return the parsed object""" - sock.settimeout(timeout) + deadline = time.monotonic() + timeout + sock.settimeout(min(timeout, 0.25) if stop_event is not None else timeout) try: while True: + if stop_event is not None and stop_event.is_set(): + raise CommandCancelled("Ableton command cancelled during shutdown") + # Check if we already have a complete line in the buffer if '\n' in self._recv_buffer: line, self._recv_buffer = self._recv_buffer.split('\n', 1) @@ -115,18 +174,30 @@ def receive_full_response(self, sock, buffer_size=8192, timeout=15.0): return result try: + if stop_event is not None: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise socket.timeout() + sock.settimeout(min(0.25, remaining)) chunk = sock.recv(buffer_size) if not chunk: raise Exception("Connection closed before receiving any data") self._recv_buffer += chunk.decode('utf-8') except socket.timeout: + if stop_event is not None: + if stop_event.is_set(): + raise CommandCancelled( + "Ableton command cancelled during shutdown" + ) from None + if time.monotonic() < deadline: + continue logger.warning("Socket timeout during receive") raise except (ConnectionError, BrokenPipeError, ConnectionResetError) as e: logger.error("Socket connection error during receive: %s", e) raise - except (socket.timeout, json.JSONDecodeError): + except (socket.timeout, json.JSONDecodeError, CommandCancelled): raise except Exception as e: logger.error("Error during receive: %s", e) @@ -139,7 +210,13 @@ def _reconnect(self) -> bool: self._recv_buffer = "" return self.connect() - def send_command(self, command_type: str, params: Dict[str, Any] = None, timeout: Optional[float] = None) -> Dict[str, Any]: + def send_command( + self, + command_type: str, + params: Dict[str, Any] = None, + timeout: Optional[float] = None, + stop_event: Optional[threading.Event] = None, + ) -> Dict[str, Any]: """Send a command to Ableton and return the response. Includes automatic retry: if the first attempt fails due to a @@ -165,8 +242,13 @@ def send_command(self, command_type: str, params: Dict[str, Any] = None, timeout for attempt in range(1, max_attempts + 1): with self._send_lock: + if stop_event is not None and stop_event.is_set(): + raise CommandCancelled("Ableton command cancelled during shutdown") if not self.sock and not self.connect(): raise ConnectionError("Not connected to Ableton") + if stop_event is not None and stop_event.is_set(): + self.disconnect() + raise CommandCancelled("Ableton command cancelled during shutdown") command = { "type": command_type, @@ -181,7 +263,13 @@ def send_command(self, command_type: str, params: Dict[str, Any] = None, timeout # Pre-delay: give Ableton time to process before we read the response if pre_delay: - time.sleep(pre_delay) + if stop_event is not None: + if stop_event.wait(pre_delay): + raise CommandCancelled( + "Ableton command cancelled during shutdown" + ) + else: + time.sleep(pre_delay) # Set timeout based on command type (caller override takes priority) if timeout is None: @@ -190,7 +278,11 @@ def send_command(self, command_type: str, params: Dict[str, Any] = None, timeout command_type, 15.0 if is_modifying else 10.0 ) # Receive the response (already parsed by receive_full_response) - response = self.receive_full_response(self.sock, timeout=timeout) + response = self.receive_full_response( + self.sock, + timeout=timeout, + stop_event=stop_event, + ) logger.debug("Response status: %s", response.get('status', 'unknown')) if response.get("status") == "error": @@ -199,10 +291,20 @@ def send_command(self, command_type: str, params: Dict[str, Any] = None, timeout # Post-delay: let Ableton settle before the next command if post_delay: - time.sleep(post_delay) + if stop_event is not None: + if stop_event.wait(post_delay): + raise CommandCancelled( + "Ableton command cancelled during shutdown" + ) + else: + time.sleep(post_delay) return response.get("result", {}) + except CommandCancelled: + self.disconnect() + self._recv_buffer = "" + raise except Exception as e: logger.error("Command '%s' attempt %d failed: %s", command_type, attempt, e) # Close the broken socket and clear buffer @@ -211,12 +313,20 @@ def send_command(self, command_type: str, params: Dict[str, Any] = None, timeout if attempt < max_attempts: # Wait briefly then retry with a fresh connection - time.sleep(0.1) + if stop_event is not None: + if stop_event.wait(0.1): + raise CommandCancelled( + "Ableton command cancelled during shutdown" + ) from None + else: + time.sleep(0.1) if not self.connect(): - raise ConnectionError("Failed to reconnect to Ableton") + raise ConnectionError("Failed to reconnect to Ableton") from e logger.info("Reconnected, retrying command...") else: - raise Exception(f"Command '{command_type}' failed after {max_attempts} attempts: {e}") + raise Exception( + f"Command '{command_type}' failed after {max_attempts} attempts: {e}" + ) from e def get_ableton_connection(): @@ -224,11 +334,9 @@ def get_ableton_connection(): if state.ableton_connection is not None: try: - # Test if the socket is still connected - if state.ableton_connection.sock is None: - raise ConnectionError("Socket is None") - state.ableton_connection.sock.settimeout(1.0) - state.ableton_connection.sock.getpeername() # raises if disconnected + if not state.ableton_connection.is_connected(): + raise ConnectionError("Socket is no longer connected") + state.ableton_connected_event.set() return state.ableton_connection except Exception as e: logger.warning("Existing connection is no longer valid: %s", e) diff --git a/MCP_Server/connections/m4l.py b/MCP_Server/connections/m4l.py index 12065e5..29e1f87 100644 --- a/MCP_Server/connections/m4l.py +++ b/MCP_Server/connections/m4l.py @@ -666,50 +666,77 @@ def _check_bridge_version(ping_result: Dict[str, Any]): ) -def get_m4l_connection() -> M4LConnection: +def get_m4l_connection() -> M4LConnection: """Get or create a connection to the M4L bridge device. Always attempts a fresh connection if the existing one is dead. Uses a cached ping result to avoid a full UDP round trip on every call. """ - # If we have a connected instance, verify it still works - if state.m4l_connection is not None and state.m4l_connection._connected: - # Use cached ping result if recent enough (avoids ~50-200ms round trip) - now = time.time() - if (now - state.m4l_ping_cache["timestamp"]) < state.M4L_PING_CACHE_TTL: - if state.m4l_ping_cache["result"]: - return state.m4l_connection - # Cache expired or stale, do a live ping - if state.m4l_connection.ping(): - state.m4l_ping_cache["result"] = True - state.m4l_ping_cache["timestamp"] = now - return state.m4l_connection - # Ping failed -- tear down and try fresh - logger.warning("M4L bridge ping failed on existing connection, reconnecting...") - state.m4l_connection.disconnect() - state.m4l_connection = None - - # Create a fresh connection - state.m4l_connection = M4LConnection() - if not state.m4l_connection.connect(): - state.m4l_connection = None - raise ConnectionError( - "Could not initialise M4L bridge UDP sockets. " - "Check that port 9879 is not already in use." - ) - - # Quick ping to verify the device is actually responding - if not state.m4l_connection.ping(): - logger.warning("M4L UDP sockets ready but bridge device is not responding.") - # Keep the sockets open -- the device might be loaded later - # Don't tear down, so the next call can retry the ping - raise ConnectionError( - "M4L bridge device is not responding. " - "Make sure the AbletonBridge M4L device is loaded on a track in Ableton." - ) - - logger.info("M4L bridge connection established and verified.") - return state.m4l_connection + with state.m4l_connection_lock: + connection = state.m4l_connection + + # If we have a connected instance, verify it still works. + if connection is not None and connection._connected: + # Use cached ping result if recent enough (avoids ~50-200ms round trip) + now = time.time() + if (now - state.m4l_ping_cache["timestamp"]) < state.M4L_PING_CACHE_TTL: + if state.m4l_ping_cache["result"]: + state.m4l_status_snapshot = (True, True) + return connection + # Cache expired or stale, do a live ping. + if connection.ping(): + state.m4l_ping_cache = { + "result": True, + "timestamp": time.time(), + } + state.m4l_status_snapshot = (True, True) + return connection + # Ping failed -- tear down and try fresh. + logger.warning("M4L bridge ping failed on existing connection, reconnecting...") + state.m4l_status_snapshot = (False, False) + connection.disconnect() + if state.m4l_connection is connection: + state.m4l_connection = None + state.m4l_ping_cache = {"result": False, "timestamp": 0.0} + + # Invalidate the observable generation before constructing a private + # replacement. Lock-free status readers can only see this coherent, + # pessimistic snapshot while the transaction is in progress. + state.m4l_status_snapshot = (False, False) + state.m4l_ping_cache = {"result": False, "timestamp": 0.0} + + # Create and connect privately before publishing the replacement. + connection = M4LConnection() + if not connection.connect(): + raise ConnectionError( + "Could not initialise M4L bridge UDP sockets. " + "Check that port 9879 is not already in use." + ) + + state.m4l_connection = connection + state.m4l_status_snapshot = (True, False) + + # Quick ping to verify the device is actually responding. + if not connection.ping(): + state.m4l_ping_cache = { + "result": False, + "timestamp": time.time(), + } + state.m4l_status_snapshot = (bool(connection._connected), False) + logger.warning("M4L UDP sockets ready but bridge device is not responding.") + # Keep the sockets open -- the device might be loaded later. + raise ConnectionError( + "M4L bridge device is not responding. " + "Make sure the AbletonBridge M4L device is loaded on a track in Ableton." + ) + + state.m4l_ping_cache = { + "result": True, + "timestamp": time.time(), + } + state.m4l_status_snapshot = (True, True) + logger.info("M4L bridge connection established and verified.") + return connection def _m4l_batch_set_params( diff --git a/MCP_Server/dashboard/server.py b/MCP_Server/dashboard/server.py index f582041..62b5f5a 100644 --- a/MCP_Server/dashboard/server.py +++ b/MCP_Server/dashboard/server.py @@ -13,7 +13,9 @@ from typing import Any, Dict, List import MCP_Server.state as state +import MCP_Server.ownership as ownership from MCP_Server.dashboard.html import DASHBOARD_HTML +from MCP_Server.status import build_connection_status logger = logging.getLogger("AbletonBridge") @@ -31,6 +33,7 @@ class DashboardLogHandler(logging.Handler): """ def emit(self, record): + """Append one safely formatted log record to the dashboard buffer.""" try: with state.server_log_lock: state.server_log_buffer.append( @@ -72,38 +75,9 @@ def get_server_version() -> str: return __version__ -def get_m4l_status() -> tuple: - """Return (sockets_ready, bridge_responding) with cached ping.""" - sockets_ready = bool(state.m4l_connection and state.m4l_connection._connected) - if not sockets_ready: - return False, False - - now = time.time() - if now - state.m4l_ping_cache["timestamp"] < state.M4L_PING_CACHE_TTL: - return sockets_ready, state.m4l_ping_cache["result"] - - try: - result = state.m4l_connection.ping() - except Exception as e: - logger.debug("Dashboard M4L ping failed: %s", e) - result = False - - state.m4l_ping_cache["result"] = result - state.m4l_ping_cache["timestamp"] = now - return sockets_ready, result - - def build_status_json() -> dict: """Collect all dashboard status data into a JSON-serializable dict.""" - ableton_connected = False - if state.ableton_connection and state.ableton_connection.sock: - try: - state.ableton_connection.sock.getpeername() - ableton_connected = True - except Exception: - pass - - m4l_sockets_ready, m4l_connected = get_m4l_status() + connections = build_connection_status(ownership.get_status()) with state.tool_call_lock: recent = list(state.tool_call_log) @@ -124,9 +98,7 @@ def build_status_json() -> dict: return { "version": get_server_version(), "uptime_seconds": round(time.time() - state.server_start_time, 1) if state.server_start_time else 0, - "ableton_connected": ableton_connected, - "m4l_connected": m4l_connected, - "m4l_sockets_ready": m4l_sockets_ready, + **connections, "store_counts": { "snapshots": len(state.snapshot_store), "macros": len(state.macro_store), @@ -152,9 +124,11 @@ def start_dashboard_server(): import uvicorn async def dashboard_page(request): + """Serve the embedded dashboard application.""" return HTMLResponse(DASHBOARD_HTML) async def api_status(request): + """Serve the dashboard's current JSON status snapshot.""" return JSONResponse(build_status_json()) app = Starlette(routes=[ @@ -169,21 +143,70 @@ async def api_status(request): log_level="warning", access_log=False, ) - state.dashboard_server = uvicorn.Server(config) + server = uvicorn.Server(config) + state.dashboard_server = server def _run(): + """Run Uvicorn on a dedicated event loop and clear exited state.""" loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) - loop.run_until_complete(state.dashboard_server.serve()) - - thread = threading.Thread(target=_run, daemon=True, name="dashboard-http") - thread.start() - logger.info("Dashboard started at http://127.0.0.1:%d", state.DASHBOARD_PORT) - - -def stop_dashboard_server(): - """Signal the dashboard server to shut down.""" - if state.dashboard_server: - state.dashboard_server.should_exit = True + try: + loop.run_until_complete(server.serve()) + finally: + loop.close() + if state.dashboard_server is server: + state.dashboard_server = None + if state.dashboard_thread is threading.current_thread(): + state.dashboard_thread = None + + thread = threading.Thread(target=_run, daemon=True, name="dashboard-http") + state.dashboard_thread = thread + try: + thread.start() + except Exception: + server.should_exit = True + if state.dashboard_server is server: + state.dashboard_server = None + if state.dashboard_thread is thread: + state.dashboard_thread = None + raise + logger.info("Dashboard started at http://127.0.0.1:%d", state.DASHBOARD_PORT) + + +def stop_dashboard_server() -> bool: + """Signal shutdown and report whether the dashboard thread has exited.""" + server = state.dashboard_server + thread = state.dashboard_thread + if server: + server.should_exit = True + stopped = True + if thread is threading.current_thread(): + stopped = False + elif thread: + # Teardown may overlap the narrow window after the thread is + # published to shared state but before start() runs. join() raises + # RuntimeError for an unstarted thread, so keep cleanup best-effort. + if thread.ident is None: + stopped = False + else: + try: + thread.join(timeout=3.0) + except RuntimeError as exc: + logger.warning("Could not join dashboard thread: %s", exc) + stopped = False + else: + stopped = not thread.is_alive() + + if not stopped: + logger.warning( + "Dashboard shutdown is incomplete; retaining server and thread state" + ) + return False + + if state.dashboard_server is server: state.dashboard_server = None + if state.dashboard_thread is thread: + state.dashboard_thread = None + if server or thread: logger.info("Dashboard server stopped") + return True diff --git a/MCP_Server/instructions.py b/MCP_Server/instructions.py index 3596f6c..03e8331 100644 --- a/MCP_Server/instructions.py +++ b/MCP_Server/instructions.py @@ -5,11 +5,13 @@ """ SERVER_INSTRUCTIONS = """ -AbletonBridge provides 341 tools for controlling Ableton Live sessions. This guidance covers cross-tool relationships, sequencing, and constraints not documented on individual tools. +AbletonBridge provides 347 tools for controlling Ableton Live sessions. This guidance covers cross-tool relationships, sequencing, and constraints not documented on individual tools. ## Startup -Call get_server_capabilities first in every session. It reports ableton_connected, m4l_connected, browser cache state, and tool count. If ableton_connected is false, most tools will fail. +Call get_server_capabilities first in every session. It reports control_role, control_availability, owner process metadata, connection state, browser cache state, and tool count. This status call does not claim control. Connection booleans are true or false only for the owner; a standby reports null because its backend is not running locally. features.m4l_bridge mirrors m4l_connected, including null on standby. Interpret ableton_connection_state and m4l_connection_state instead: not_started means the first normal tool can attempt to claim and connect, owned_elsewhere means another task owns control and its connection health is not visible here, and unknown means port 9881 is occupied by an unrelated process. + +The first normal Ableton tool call automatically claims control when control_availability is "available". If another task owns control, tools return a structured ownership error instead of disappearing. Ask the owning task to call release_ableton_control when an intentional handoff is needed. Never assume control can be stolen or released by a standby task. ## Compound Tools @@ -83,7 +85,7 @@ ## M4L Bridge -Check m4l_connected from get_server_capabilities before calling any M4L tool. If false, use standard get_device_parameters / set_device_parameter instead. +When this task owns control, check m4l_connected from get_server_capabilities before calling any M4L tool. If false, use standard get_device_parameters / set_device_parameter instead. A standby value of null does not mean the M4L bridge is disconnected; inspect m4l_connection_state and ownership first. Key M4L subsystems: - **Note surgery**: get_clip_notes_with_ids → modify_clip_notes / remove_clip_notes_by_id for in-place, non-destructive editing (Live 11+). diff --git a/MCP_Server/ownership.py b/MCP_Server/ownership.py new file mode 100644 index 0000000..8002072 --- /dev/null +++ b/MCP_Server/ownership.py @@ -0,0 +1,575 @@ +"""Single-owner coordination for Ableton backend resources. + +Every MCP process can expose the complete tool surface, but only one process +may own the Live, M4L, and dashboard connections. Ownership is represented by +an exclusive loopback listener. The same listener also exposes a tiny JSON +status response so standby processes can describe the current owner. +""" + +from __future__ import annotations + +import errno +import json +import logging +import os +import socket +import threading +import uuid +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Callable, Mapping, Optional, Union + +import MCP_Server.state as state + +logger = logging.getLogger("AbletonBridge") + +_STATUS_PROTOCOL_VERSION = 1 +_STATUS_TIMEOUT_SECONDS = 0.25 +_Port = Union[int, Callable[[], int]] + + +@dataclass(frozen=True) +class ClaimResult: + """Result of an automatic ownership claim.""" + + acquired: bool + control: dict + error: Optional[str] = None + claim_token: Optional[str] = None + + +@dataclass(frozen=True) +class ReleaseResult: + """Result of releasing ownership held by this process.""" + + released: bool + control: dict + error: Optional[str] = None + + +class OwnershipManager: + """Coordinate one backend owner across multiple local MCP processes.""" + + def __init__( + self, + port: _Port, + *, + host: str = "127.0.0.1", + environment: Optional[Mapping[str, str]] = None, + ) -> None: + """Initialize process-local coordination for one ownership port.""" + self._port = port + self._host = host + self._environment = environment if environment is not None else os.environ + self._instance_id = str(uuid.uuid4()) + self._lock = threading.RLock() + self._transition_lock = threading.Lock() + self._listener: Optional[socket.socket] = None + self._responder_stop: Optional[threading.Event] = None + self._responder_thread: Optional[threading.Thread] = None + self._owner: Optional[dict] = None + self._claim_token: Optional[str] = None + self._phase = "standby" + self._active_operations = 0 + self._start_backend: Optional[Callable[[], None]] = None + self._stop_backend: Optional[Callable[[], Optional[bool]]] = None + + @property + def port(self) -> int: + """Return the configured ownership and status-listener port.""" + return self._port() if callable(self._port) else self._port + + def configure_backend( + self, + start_backend: Callable[[], None], + stop_backend: Callable[[], Optional[bool]], + ) -> None: + """Configure lifecycle callbacks used when ownership changes.""" + with self._lock: + self._start_backend = start_backend + self._stop_backend = stop_backend + + def unconfigure_backend(self) -> None: + """Remove lifecycle callbacks after the MCP lifespan ends.""" + with self._lock: + self._start_backend = None + self._stop_backend = None + + def is_configured(self) -> bool: + """Return whether both backend lifecycle callbacks are installed.""" + with self._lock: + return self._start_backend is not None and self._stop_backend is not None + + def ensure_control(self, *, client_name: Optional[str] = None) -> ClaimResult: + """Return local ownership, claiming and starting the backend if free.""" + # Backend startup and teardown must never overlap. The state lock is + # deliberately released while callbacks run, so a separate transition + # lock serializes those callbacks without blocking status responses. + with self._transition_lock: + return self._ensure_control(client_name=client_name) + + def _ensure_control(self, *, client_name: Optional[str] = None) -> ClaimResult: + """Claim control while the lifecycle transition lock is held.""" + with self._lock: + if self._listener is not None: + if self._phase == "owner": + self._record_client_name(client_name) + return ClaimResult(True, self._local_status_locked()) + return ClaimResult( + False, + self._local_status_locked(), + "Ableton control is currently changing state. Try again shortly.", + ) + + if not self.is_configured(): + return ClaimResult( + False, + self._standby_status("available"), + "Ableton control lifecycle is not configured.", + ) + + try: + listener = self._bind_listener() + except OSError: + listener = None + + if listener is None: + control = self.status() + if control["control_availability"] == "owned": + message = "Ableton control is owned by another task." + else: + message = ( + f"Ableton control is unavailable because loopback port " + f"{self.port} is occupied by an unknown process." + ) + return ClaimResult(False, control, message) + + owner = self._build_owner_metadata(client_name) + try: + responder_stop, responder_thread = ( + self._start_status_responder_locked(listener) + ) + except Exception as exc: + try: + listener.close() + except OSError: + pass + logger.error("Ableton owner-status responder startup failed: %s", exc) + return ClaimResult( + False, + self.status(), + f"Could not start the Ableton owner-status responder: {exc}", + ) + + self._listener = listener + self._owner = owner + claim_token = str(uuid.uuid4()) + self._claim_token = claim_token + self._phase = "starting" + self._responder_stop = responder_stop + self._responder_thread = responder_thread + start_backend = self._start_backend + + try: + assert start_backend is not None + start_backend() + except Exception as exc: + logger.error("Ableton control startup failed: %s", exc) + cleanup_complete, cleanup_error = self._cleanup_failed_start() + error = f"Could not start the Ableton control backend: {exc}" + if not cleanup_complete: + error += ( + " Cleanup is incomplete, so ownership was retained; " + "retry release after owner resources stop." + ) + if cleanup_error: + error += f" Cleanup error: {cleanup_error}" + return ClaimResult( + False, + self.status(), + error, + ) + + with self._lock: + self._phase = "owner" + logger.info( + "Ableton control acquired on port %d by process %d", + self.port, + os.getpid(), + ) + return ClaimResult( + True, + self._local_status_locked(), + claim_token=claim_token, + ) + + def release_if_current_claim(self, claim_token: str) -> ReleaseResult: + """Release only the ownership generation created by one claim call.""" + with self._transition_lock: + with self._lock: + is_current = ( + self._listener is not None + and self._claim_token == claim_token + ) + if not is_current: + return ReleaseResult(False, self.status()) + return self._release() + + def release(self, *, force: bool = False) -> ReleaseResult: + """Release local ownership; never release another process's ownership.""" + with self._transition_lock: + return self._release(force=force) + + def _release(self, *, force: bool = False) -> ReleaseResult: + """Stop owner resources while the lifecycle transition lock is held.""" + with self._lock: + if self._listener is None: + return ReleaseResult(False, self.status()) + + if self._phase not in {"owner", "cleanup_failed"} and not force: + return ReleaseResult( + False, + self._local_status_locked(), + "Ableton control is currently changing state. Try again shortly.", + ) + + if self._active_operations and not force: + count = self._active_operations + return ReleaseResult( + False, + self._local_status_locked(), + f"Cannot release Ableton control while {count} operation(s) are still running.", + ) + + self._phase = "releasing" + + cleanup_complete, cleanup_error = self._stop_backend_once() + if not cleanup_complete: + with self._lock: + self._phase = "cleanup_failed" + control = self._local_status_locked() + message = cleanup_error or ( + "Ableton control cleanup is incomplete; owner resources are " + "still stopping. Retry release shortly." + ) + logger.error("Ableton control cleanup incomplete: %s", message) + return ReleaseResult(False, control, message) + + with self._lock: + if self._active_operations: + count = self._active_operations + self._phase = "cleanup_failed" + control = self._local_status_locked() + else: + count = 0 + control = None + + if control is not None: + message = ( + "Ableton control cleanup is incomplete because " + f"{count} operation(s) are still running. Retry release shortly." + ) + logger.error("Ableton control cleanup incomplete: %s", message) + return ReleaseResult(False, control, message) + + self._close_local_ownership() + + logger.info("Ableton control released by process %d", os.getpid()) + return ReleaseResult(True, self.status()) + + def shutdown(self) -> None: + """Best-effort automatic release during MCP process shutdown.""" + with self._lock: + if self._listener is None: + return + self.release(force=True) + + def begin_operation(self) -> bool: + """Register backend work, refusing work that starts after release.""" + with self._lock: + if not self.is_configured(): + return True + if self._listener is None or self._phase != "owner": + return False + self._active_operations += 1 + return True + + def end_operation(self) -> None: + """Mark a registered backend operation complete.""" + with self._lock: + if self._active_operations: + self._active_operations -= 1 + + def status(self) -> dict: + """Return local role plus best-effort metadata for a remote owner.""" + with self._lock: + if self._listener is not None: + return self._local_status_locked() + return self._probe_remote_owner() + + def _bind_listener(self) -> socket.socket: + """Bind the exclusive loopback listener that represents ownership.""" + listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + try: + if hasattr(socket, "SO_EXCLUSIVEADDRUSE"): + listener.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1) + else: + listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + listener.bind((self._host, self.port)) + listener.listen(8) + listener.settimeout(_STATUS_TIMEOUT_SECONDS) + return listener + except Exception: + listener.close() + raise + + def _start_status_responder_locked( + self, + listener: socket.socket, + ) -> tuple[threading.Event, threading.Thread]: + """Start an owner-status responder before publishing local state.""" + stop_event = threading.Event() + thread = threading.Thread( + target=self._serve_status, + args=(listener, stop_event), + daemon=True, + name="ableton-owner-status", + ) + try: + thread.start() + except Exception: + stop_event.set() + raise + return stop_event, thread + + def _serve_status( + self, + listener: socket.socket, + stop_event: threading.Event, + ) -> None: + """Serve owner metadata until ownership is released.""" + while not stop_event.is_set(): + try: + client, _address = listener.accept() + except socket.timeout: + continue + except OSError: + break + + try: + client.settimeout(_STATUS_TIMEOUT_SECONDS) + with self._lock: + payload = { + "service": "AbletonBridge", + "protocol": _STATUS_PROTOCOL_VERSION, + "owner": dict(self._owner) if self._owner else None, + "active_operations": self._active_operations, + } + client.sendall((json.dumps(payload) + "\n").encode("utf-8")) + except OSError: + pass + finally: + client.close() + + def _probe_remote_owner(self) -> dict: + """Classify the loopback-port occupant and read known owner metadata.""" + try: + with socket.create_connection( + (self._host, self.port), + timeout=_STATUS_TIMEOUT_SECONDS, + ) as client: + client.settimeout(_STATUS_TIMEOUT_SECONDS) + chunks = [] + size = 0 + while size < 65536: + chunk = client.recv(65536 - size) + if not chunk: + break + chunks.append(chunk) + size += len(chunk) + if b"\n" in chunk: + break + raw = b"".join(chunks) + except OSError as exc: + if isinstance(exc, ConnectionRefusedError) or exc.errno == errno.ECONNREFUSED: + return self._standby_status("available") + return self._standby_status("occupied_unknown") + + try: + payload = json.loads(raw.decode("utf-8").strip()) + except (UnicodeDecodeError, json.JSONDecodeError): + return self._standby_status("occupied_unknown") + + if ( + not isinstance(payload, dict) + or payload.get("service") != "AbletonBridge" + or payload.get("protocol") != _STATUS_PROTOCOL_VERSION + or not isinstance(payload.get("owner"), dict) + ): + return self._standby_status("occupied_unknown") + + return { + "control_role": "standby", + "control_availability": "owned", + "owner": payload["owner"], + "active_operations": payload.get("active_operations", 0), + } + + def _standby_status(self, availability: str) -> dict: + """Build a status envelope for this process's standby role.""" + return { + "control_role": "standby", + "control_availability": availability, + "owner": None, + "active_operations": 0, + } + + def _local_status_locked(self) -> dict: + """Build the local owner status while coordination state is locked.""" + return { + "control_role": "owner", + "control_availability": "owned", + "owner": dict(self._owner) if self._owner else None, + "active_operations": self._active_operations, + } + + def _build_owner_metadata(self, client_name: Optional[str]) -> dict: + """Capture reliable process identity and best-effort client metadata.""" + owner = { + "process_id": os.getpid(), + "parent_process_id": os.getppid(), + "instance_id": self._instance_id, + "claimed_at": datetime.now(timezone.utc).isoformat(), + } + if client_name: + owner["client_name"] = client_name + task_id = self._environment.get("CODEX_THREAD_ID") + if task_id: + owner["task_id"] = task_id + return owner + + def _record_client_name(self, client_name: Optional[str]) -> None: + """Fill previously unavailable client metadata without replacing it.""" + if client_name and self._owner and "client_name" not in self._owner: + self._owner["client_name"] = client_name + + def _stop_backend_once(self) -> tuple[bool, Optional[str]]: + """Run teardown once, treating ``False`` as incomplete cleanup.""" + with self._lock: + stop_backend = self._stop_backend + try: + result = stop_backend() if stop_backend is not None else True + except Exception as exc: + return False, str(exc) + if result is False: + return False, None + return True, None + + def _cleanup_failed_start(self) -> tuple[bool, Optional[str]]: + """Clean up a failed startup and retain ownership if cleanup stalls.""" + complete, error = self._stop_backend_once() + if complete: + with self._lock: + active_operations = self._active_operations + if active_operations: + complete = False + error = ( + "Backend cleanup finished, but " + f"{active_operations} owner operation(s) are still running." + ) + if complete: + self._close_local_ownership() + else: + with self._lock: + self._phase = "cleanup_failed" + logger.warning( + "Cleanup after backend startup failure is incomplete%s", + f": {error}" if error else "", + ) + return complete, error + + def _close_local_ownership(self) -> None: + """Stop the responder and return the manager to standby state.""" + with self._lock: + stop_event = self._responder_stop + responder = self._responder_thread + listener = self._listener + if stop_event is not None: + stop_event.set() + if listener is not None: + try: + listener.close() + except OSError: + pass + + if ( + responder is not None + and responder is not threading.current_thread() + and responder.ident is not None + ): + responder.join(timeout=1.0) + + with self._lock: + self._listener = None + self._responder_stop = None + self._responder_thread = None + self._owner = None + self._claim_token = None + self._phase = "standby" + self._active_operations = 0 + + +_manager = OwnershipManager(lambda: state.SINGLETON_LOCK_PORT) + + +def configure_backend( + start_backend: Callable[[], None], + stop_backend: Callable[[], Optional[bool]], +) -> None: + """Configure lifecycle callbacks on the process-wide ownership manager.""" + _manager.configure_backend(start_backend, stop_backend) + + +def unconfigure_backend() -> None: + """Remove lifecycle callbacks from the process-wide ownership manager.""" + _manager.unconfigure_backend() + + +def is_configured() -> bool: + """Return whether the process-wide ownership manager is configured.""" + return _manager.is_configured() + + +def ensure_control(*, client_name: Optional[str] = None) -> ClaimResult: + """Automatically acquire process-wide Ableton control when available.""" + return _manager.ensure_control(client_name=client_name) + + +def release_control(*, force: bool = False) -> ReleaseResult: + """Release only the Ableton control owned by this MCP process.""" + return _manager.release(force=force) + + +def release_if_current_claim(claim_token: str) -> ReleaseResult: + """Release an abandoned claim only if its ownership is still current.""" + return _manager.release_if_current_claim(claim_token) + + +def shutdown() -> None: + """Best-effort release of process-wide ownership during shutdown.""" + _manager.shutdown() + + +def begin_operation() -> bool: + """Register owner-dependent work with the process-wide manager.""" + return _manager.begin_operation() + + +def end_operation() -> None: + """Mark process-wide owner-dependent work as complete.""" + _manager.end_operation() + + +def get_status() -> dict: + """Return the process-wide local or remote ownership status.""" + return _manager.status() diff --git a/MCP_Server/server.py b/MCP_Server/server.py index f5caa28..f7acf8d 100644 --- a/MCP_Server/server.py +++ b/MCP_Server/server.py @@ -14,15 +14,11 @@ import asyncio import concurrent.futures import logging -import os -import socket -import sys import time import threading from contextlib import asynccontextmanager from typing import Any, AsyncIterator, Dict from datetime import datetime, timezone -from collections import deque # --------------------------------------------------------------------------- # MCP framework @@ -33,7 +29,9 @@ # Internal modules # --------------------------------------------------------------------------- import MCP_Server.state as state -from MCP_Server.connections.ableton import AbletonConnection, get_ableton_connection +import MCP_Server.ownership as ownership +from MCP_Server.status import build_connection_status +from MCP_Server.connections.ableton import get_ableton_connection from MCP_Server.connections.m4l import M4LConnection from MCP_Server.cache.browser import load_browser_cache_from_disk, populate_browser_cache from MCP_Server.dashboard.server import ( @@ -54,88 +52,82 @@ logger = logging.getLogger("AbletonBridge") -# =================================================================== -# Singleton lock — prevent duplicate server instances -# =================================================================== - -def _acquire_singleton_lock() -> socket.socket: - """Acquire an exclusive TCP port lock to prevent duplicate server instances. - - Returns the bound socket (caller must keep it alive for the server's - lifetime). Raises RuntimeError if another instance already holds the lock. - """ - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - try: - if hasattr(socket, "SO_EXCLUSIVEADDRUSE"): - sock.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1) - sock.bind(("127.0.0.1", state.SINGLETON_LOCK_PORT)) - sock.listen(1) - logger.info("Singleton lock acquired on port %d", state.SINGLETON_LOCK_PORT) - return sock - except OSError as e: - sock.close() - raise RuntimeError( - f"Another AbletonBridge server instance is already running " - f"(port {state.SINGLETON_LOCK_PORT} is in use). " - f"Stop the other instance first." - ) from e - - -def _release_singleton_lock(sock: socket.socket): - """Release the singleton lock by closing the lock socket.""" - if sock: - try: - sock.close() - logger.info("Singleton lock released") - except Exception: - pass - - # =================================================================== # M4L auto-connect (background thread) # =================================================================== -def _m4l_auto_connect(): - """Background thread: create UDP sockets once, retry ping until M4L responds.""" - # Create sockets once — don't tear them down between retries - conn = M4LConnection() - if not conn.connect(): - logger.warning("M4L auto-connect: could not bind UDP sockets") - return - - state.m4l_connection = conn +def _m4l_auto_connect(stop_event: threading.Event): + """Background thread: create UDP sockets once, retry ping until M4L responds.""" + if stop_event.is_set(): + return + + # Publish one connection-and-cache generation atomically. Normal tools use + # the same lock when verifying or replacing this connection. + with state.m4l_connection_lock: + if stop_event.is_set(): + return + if state.m4l_connection is not None: + conn = state.m4l_connection + if not conn._connected: + return + else: + state.m4l_status_snapshot = (False, False) + state.m4l_ping_cache = {"result": False, "timestamp": 0.0} + conn = M4LConnection() + if not conn.connect(): + logger.warning("M4L auto-connect: could not bind UDP sockets") + return + if stop_event.is_set(): + conn.disconnect() + return + state.m4l_connection = conn + state.m4l_status_snapshot = (True, False) # Build a raw OSC ping packet ping_id = "autocon" ping_osc = M4LConnection._build_osc_message("/ping", [("s", ping_id)]) - for attempt in range(1, 16): # 15 attempts, ~2 s apart - try: - # Drain stale data - conn._drain_recv_socket() - conn.recv_sock.settimeout(2.0) - - # Send ping - conn.send_sock.sendto(ping_osc, (conn.send_host, conn.send_port)) - - # Wait for response - data, _addr = conn.recv_sock.recvfrom(65535) - result = conn._parse_m4l_response(data) - if result.get("status") == "success": - logger.info("M4L bridge auto-connected on attempt %d", attempt) - state.m4l_ping_cache["result"] = True - state.m4l_ping_cache["timestamp"] = time.time() - # Check bridge version compatibility - M4LConnection._check_bridge_version(result) - return - except socket.timeout: - logger.info( - "M4L auto-connect %d/15: no response (timeout), retrying...", - attempt, + for attempt in range(1, 16): # 15 attempts, ~2 s apart + if stop_event.is_set(): + return + try: + with state.m4l_connection_lock: + if stop_event.is_set() or state.m4l_connection is not conn: + return + with conn._send_lock: + # Drain stale data + conn._drain_recv_socket() + conn.recv_sock.settimeout(2.0) + + # Send ping + conn.send_sock.sendto(ping_osc, (conn.send_host, conn.send_port)) + + # Wait for response + data, _addr = conn.recv_sock.recvfrom(65535) + result = conn._parse_m4l_response(data) + if result.get("status") == "success": + logger.info("M4L bridge auto-connected on attempt %d", attempt) + state.m4l_ping_cache = { + "result": True, + "timestamp": time.time(), + } + state.m4l_status_snapshot = (True, True) + # Check bridge version compatibility + M4LConnection._check_bridge_version(result) + return + except TimeoutError: + if stop_event.is_set(): + return + logger.info( + "M4L auto-connect %d/15: no response (timeout), retrying...", + attempt, ) except Exception as e: + if stop_event.is_set(): + return logger.info("M4L auto-connect %d/15: %s", attempt, e) - time.sleep(2) + if stop_event.wait(2.0): + return logger.warning( "M4L bridge not available after 15 attempts — will retry when needed" @@ -146,7 +138,7 @@ def _m4l_auto_connect(): # Browser cache warmup (background thread) # =================================================================== -def _browser_cache_warmup(): +def _browser_cache_warmup(stop_event: threading.Event): """Background thread: load disk cache instantly, then refresh from Ableton.""" from MCP_Server.constants import BROWSER_DISK_CACHE_MAX_AGE @@ -164,35 +156,147 @@ def _browser_cache_warmup(): ) # Step 2: Wait for Ableton connection, then do a live scan to refresh - state.ableton_connected_event.wait(timeout=30.0) + deadline = time.monotonic() + 30.0 + while not state.ableton_connected_event.is_set(): + if stop_event.wait(0.1) or time.monotonic() >= deadline: + return if not (state.ableton_connection and state.ableton_connection.sock): logger.warning( "Browser cache warmup: Ableton not connected after 30s, skipping live scan" ) return - time.sleep(0.5) # brief settle after connection confirmed + if stop_event.wait(0.5): # brief settle after connection confirmed + return try: - populate_browser_cache() + populate_browser_cache(stop_event=stop_event) except Exception as e: logger.warning("Browser cache warmup failed: %s", e) +# =================================================================== +# Control-owner backend lifecycle +# =================================================================== + +def _start_control_backend(): + """Start resources that must exist in exactly one MCP process.""" + logger.info("Starting Ableton control backend") + stop_event = threading.Event() + state.control_stop_event = stop_event + state.control_background_threads = [] + state.ableton_connected_event.clear() + + # Live connectivity is required for a successful ownership claim. + get_ableton_connection() + + try: + start_dashboard_server() + except Exception as e: + logger.warning("Dashboard failed to start: %s", e) + + for target, name in ( + (_m4l_auto_connect, "m4l-auto-connect"), + (_browser_cache_warmup, "browser-cache-warmup"), + ): + thread = threading.Thread( + target=target, + args=(stop_event,), + daemon=True, + name=name, + ) + state.control_background_threads.append(thread) + try: + thread.start() + except Exception: + state.control_background_threads = [ + worker + for worker in state.control_background_threads + if worker is not thread + ] + raise + + +def _stop_control_backend() -> bool: + """Stop owner-only resources and report when handoff is safe.""" + cleanup_complete = True + stop_event = state.control_stop_event + if stop_event is not None: + stop_event.set() + + try: + dashboard_stopped = stop_dashboard_server() + except Exception as exc: + dashboard_stopped = False + logger.warning("Dashboard shutdown failed during release: %s", exc) + if dashboard_stopped is False: + cleanup_complete = False + + ableton_connection = state.ableton_connection + if ableton_connection: + logger.info("Disconnecting from Ableton") + try: + ableton_connection.disconnect() + except Exception as exc: + cleanup_complete = False + logger.warning("Ableton disconnect failed during release: %s", exc) + else: + if state.ableton_connection is ableton_connection: + state.ableton_connection = None + + remaining_threads = [] + for thread in list(state.control_background_threads): + stopped = False + if thread is not threading.current_thread() and thread.ident is not None: + try: + thread.join(timeout=3.0) + except RuntimeError as exc: + logger.warning( + "Could not join control background thread %s: %s", + thread.name, + exc, + ) + else: + stopped = not thread.is_alive() + if not stopped: + cleanup_complete = False + remaining_threads.append(thread) + logger.warning( + "Control background thread %s is still stopping", + thread.name, + ) + + state.control_background_threads = remaining_threads + + # M4L is published by its warmup worker, so re-read it only after joining + # workers. This catches a connection created just as cancellation began. + with state.m4l_connection_lock: + state.m4l_status_snapshot = (False, False) + m4l_connection = state.m4l_connection + if m4l_connection: + logger.info("Disconnecting M4L bridge") + try: + m4l_connection.disconnect() + except Exception as exc: + cleanup_complete = False + logger.warning("M4L disconnect failed during release: %s", exc) + else: + if state.m4l_connection is m4l_connection: + state.m4l_connection = None + state.m4l_ping_cache = {"result": False, "timestamp": 0.0} + + if cleanup_complete: + state.control_stop_event = None + state.ableton_connected_event.clear() + return cleanup_complete + + # =================================================================== -# Server lifespan — startup / shutdown +# Server lifespan — MCP availability is independent of backend ownership # =================================================================== @asynccontextmanager async def server_lifespan(server: FastMCP) -> AsyncIterator[Dict[str, Any]]: """Manage server startup and shutdown lifecycle.""" try: - # Singleton guard - try: - state.singleton_lock_sock = _acquire_singleton_lock() - except RuntimeError as e: - logger.error(str(e)) - logger.error("Exiting to avoid conflicts.") - sys.exit(1) - logger.info("AbletonBridge server starting up") state.server_start_time = time.time() @@ -205,29 +309,7 @@ async def server_lifespan(server: FastMCP) -> AsyncIterator[Dict[str, Any]]: concurrent.futures.ThreadPoolExecutor(max_workers=8) ) - # Connect to Ableton (Remote Script TCP) - try: - ableton = get_ableton_connection() - logger.info("Successfully connected to Ableton on startup") - except Exception as e: - logger.warning("Could not connect to Ableton on startup: %s", e) - logger.warning("Make sure the Ableton Remote Script is running") - - # Auto-connect M4L bridge in background - threading.Thread( - target=_m4l_auto_connect, daemon=True, name="m4l-auto-connect" - ).start() - - # Start web dashboard on background thread - try: - start_dashboard_server() - except Exception as e: - logger.warning("Dashboard failed to start: %s", e) - - # Pre-populate browser cache in background - threading.Thread( - target=_browser_cache_warmup, daemon=True, name="browser-cache-warmup" - ).start() + ownership.configure_backend(_start_control_backend, _stop_control_backend) # Load saved effect chain templates from disk try: @@ -239,21 +321,8 @@ async def server_lifespan(server: FastMCP) -> AsyncIterator[Dict[str, Any]]: yield {} finally: - # Shutdown sequence - stop_dashboard_server() - - if state.ableton_connection: - logger.info("Disconnecting from Ableton on shutdown") - state.ableton_connection.disconnect() - state.ableton_connection = None - - if state.m4l_connection: - logger.info("Disconnecting M4L bridge on shutdown") - state.m4l_connection.disconnect() - state.m4l_connection = None - - _release_singleton_lock(state.singleton_lock_sock) - state.singleton_lock_sock = None + ownership.shutdown() + ownership.unconfigure_backend() logger.info("AbletonBridge server shut down") @@ -289,36 +358,51 @@ async def server_lifespan(server: FastMCP) -> AsyncIterator[Dict[str, Any]]: @mcp.resource("ableton://session") def resource_session() -> str: """Current Ableton session info (tempo, tracks, transport state).""" - import json - try: - ableton = get_ableton_connection() - result = ableton.send_command("get_session_info") - return json.dumps(result) - except Exception as e: - return json.dumps({"error": str(e)}) + return _run_controlled_resource("get_session_info") @mcp.resource("ableton://tracks") def resource_tracks() -> str: """All track information including devices, clips, and routing.""" + return _run_controlled_resource("get_all_tracks_info") + + +def _run_controlled_resource(command: str) -> str: + """Run an Ableton resource read under the same ownership contract as tools.""" import json + + claim = ownership.ensure_control() + if not claim.acquired: + return json.dumps({ + "status": "error", + "message": claim.error, + "data": {"control": claim.control}, + }) + if not ownership.begin_operation(): + return json.dumps({ + "status": "error", + "message": "Ableton control was released before the resource read began.", + "data": {"control": ownership.get_status()}, + }) try: ableton = get_ableton_connection() - result = ableton.send_command("get_all_tracks_info") - return json.dumps(result) + return json.dumps(ableton.send_command(command)) except Exception as e: - return json.dumps({"error": str(e)}) + return json.dumps({"status": "error", "message": str(e)}) + finally: + ownership.end_operation() @mcp.resource("ableton://capabilities") def resource_capabilities() -> str: - """Server capabilities, connection status, and version info.""" + """Return version, ownership, and role-aware connection status.""" import json from MCP_Server import __version__ + control = ownership.get_status() result = { "server_version": __version__, - "ableton_connected": bool(state.ableton_connection and state.ableton_connection.sock), - "m4l_connected": bool(state.m4l_connection and state.m4l_connection._connected), + **control, + **build_connection_status(control), "m4l_bridge_version": state.m4l_bridge_version or "unknown", "browser_cache_ready": state.browser_cache_ready.is_set(), "browser_cache_items": len(state.browser_cache_flat), diff --git a/MCP_Server/state.py b/MCP_Server/state.py index e2a150c..dbcd52f 100644 --- a/MCP_Server/state.py +++ b/MCP_Server/state.py @@ -9,7 +9,6 @@ """ import os -import socket import threading from collections import deque from typing import Any, Dict, List, Optional @@ -19,6 +18,8 @@ # --------------------------------------------------------------------------- ableton_connection: Optional[Any] = None # AbletonConnection | None m4l_connection: Optional[Any] = None # M4LConnection | None +m4l_connection_lock: threading.RLock = threading.RLock() +m4l_status_snapshot: tuple[bool, bool] = (False, False) # --------------------------------------------------------------------------- # Feature stores (in-memory, lost on restart) @@ -37,6 +38,7 @@ tool_call_counts: Dict[str, int] = {} tool_call_lock: threading.Lock = threading.Lock() dashboard_server: Optional[Any] = None # uvicorn.Server | None +dashboard_thread: Optional[threading.Thread] = None server_log_buffer: deque = deque(maxlen=1000) server_log_lock: threading.Lock = threading.Lock() @@ -74,9 +76,10 @@ SINGLETON_LOCK_PORT: int = int(os.environ.get("ABLETON_BRIDGE_LOCK_PORT", "9881")) # --------------------------------------------------------------------------- -# Singleton lock +# Control-owner backend lifecycle # --------------------------------------------------------------------------- -singleton_lock_sock: Optional[socket.socket] = None +control_stop_event: Optional[threading.Event] = None +control_background_threads: List[threading.Thread] = [] # --------------------------------------------------------------------------- # MCP server instance (set by server.py after creating the FastMCP object) diff --git a/MCP_Server/status.py b/MCP_Server/status.py new file mode 100644 index 0000000..01f1604 --- /dev/null +++ b/MCP_Server/status.py @@ -0,0 +1,117 @@ +"""Ownership-aware connection status helpers for AbletonBridge.""" + +import logging +import time +from typing import Any + +import MCP_Server.ownership as ownership +import MCP_Server.state as state + + +logger = logging.getLogger("AbletonBridge") + + +def build_connection_status(control: dict[str, Any]) -> dict[str, Any]: + """Build truthful connection fields for one ownership snapshot. + + Only the owner process has a local backend to inspect. Standby processes + therefore return ``None`` for connection booleans and use explicit states + to distinguish an unstarted backend from unobservable remote ownership. + """ + if control.get("control_role") != "owner": + availability = control.get("control_availability") + if availability == "available": + connection_state = "not_started" + elif availability == "owned": + connection_state = "owned_elsewhere" + else: + connection_state = "unknown" + + return { + "ableton_connection_state": connection_state, + "ableton_connected": None, + "m4l_connection_state": connection_state, + "m4l_connected": None, + "m4l_sockets_ready": None, + } + + ableton_connected = _ableton_socket_connected() + m4l_sockets_ready, m4l_connected = get_m4l_status() + if m4l_connected: + m4l_connection_state = "connected" + elif m4l_sockets_ready: + m4l_connection_state = "sockets_ready" + else: + m4l_connection_state = "disconnected" + + return { + "ableton_connection_state": ( + "connected" if ableton_connected else "disconnected" + ), + "ableton_connected": ableton_connected, + "m4l_connection_state": m4l_connection_state, + "m4l_connected": m4l_connected, + "m4l_sockets_ready": m4l_sockets_ready, + } + + +def get_m4l_status() -> tuple[bool, bool]: + """Return local M4L socket readiness and cached bridge responsiveness.""" + # A live ping may reconnect its UDP sockets. Protect it like any other + # owner-local operation so manual release cannot tear the backend down + # underneath the status request. The connection lock also makes a status + # ping atomic with normal-tool connection replacement. + track_operation = ownership.is_configured() + if track_operation and not ownership.begin_operation(): + return _m4l_cached_snapshot() + + try: + if not state.m4l_connection_lock.acquire(blocking=False): + return _m4l_cached_snapshot() + try: + connection = state.m4l_connection + sockets_ready = bool(connection and connection._connected) + if not sockets_ready: + state.m4l_status_snapshot = (False, False) + return state.m4l_status_snapshot + + now = time.time() + if now - state.m4l_ping_cache["timestamp"] < state.M4L_PING_CACHE_TTL: + state.m4l_status_snapshot = ( + sockets_ready, + bool(state.m4l_ping_cache["result"]), + ) + return state.m4l_status_snapshot + + try: + result = connection.ping() + except Exception as exc: + logger.debug("M4L status ping failed: %s", exc) + result = False + + # ping() may reconnect its UDP sockets. Re-read readiness after the + # probe so a failed reconnect cannot publish sockets_ready=True. + sockets_ready = bool( + state.m4l_connection is connection and connection._connected + ) + connected = bool(result) if sockets_ready else False + state.m4l_ping_cache["result"] = connected + state.m4l_ping_cache["timestamp"] = time.time() + state.m4l_status_snapshot = (sockets_ready, connected) + return state.m4l_status_snapshot + finally: + state.m4l_connection_lock.release() + finally: + if track_operation: + ownership.end_operation() + + +def _m4l_cached_snapshot() -> tuple[bool, bool]: + """Return a non-mutating M4L snapshot while lifecycle state is busy.""" + return state.m4l_status_snapshot + + +def _ableton_socket_connected() -> bool: + """Check the local Ableton socket without sending a protocol command.""" + connection = state.ableton_connection + return bool(connection and connection.is_connected()) diff --git a/MCP_Server/tools/_base.py b/MCP_Server/tools/_base.py index 35079cb..153e928 100644 --- a/MCP_Server/tools/_base.py +++ b/MCP_Server/tools/_base.py @@ -3,6 +3,9 @@ import functools import json import logging +import threading + +import MCP_Server.ownership as ownership logger = logging.getLogger("AbletonBridge") @@ -11,20 +14,47 @@ # This prevents thread pool exhaustion and ensures orderly command dispatch. _ableton_semaphore = asyncio.Semaphore(1) -# Absolute timeout for any single tool call (prevents a stuck tool from -# blocking the semaphore indefinitely). -_TOOL_TIMEOUT_SECONDS = 120.0 +# Client response deadline, including time queued behind another control tool. +# An already-running worker keeps the semaphore until it actually exits because +# Python worker threads cannot be cancelled safely. +_TOOL_TIMEOUT_SECONDS = 120.0 + +_INVOCATION_ABANDONED = object() + + +class _ControlReleasedError(RuntimeError): + """Raised when queued backend work starts after ownership was released.""" + + +class _InvocationGate: + """Atomically decide whether client-abandoned work may begin.""" + + def __init__(self) -> None: + """Create a request that is initially allowed to start.""" + self._lock = threading.Lock() + self._abandoned = False + + def abandon(self) -> None: + """Prevent the tool body from starting after its caller has left.""" + with self._lock: + self._abandoned = True + def try_start(self) -> bool: + """Return false when timeout or cancellation won the start race.""" + with self._lock: + return not self._abandoned -def _tool_handler(error_prefix: str): + +def _tool_handler(error_prefix: str, *, requires_control: bool = True): """Decorator that wraps tool functions with standard error handling. Runs the synchronous tool function in a thread pool via asyncio.to_thread() so it doesn't block the FastMCP async event loop during TCP/UDP I/O. - An asyncio.Semaphore gates entry so that only one tool occupies the thread - pool (and the shared TCP socket) at a time. An outer timeout ensures a - stuck tool releases the semaphore after _TOOL_TIMEOUT_SECONDS. + An asyncio.Semaphore gates entry so that only one tool occupies the shared + backend at a time. A response deadline bounds both queueing and execution + from the caller's perspective. If a worker has already started, its lease + remains held until it exits so later tools cannot overlap its socket work. All plain-string returns are wrapped in tool_success() for consistent JSON envelope. Returns that are already JSON (start with '{' or '[') pass through. @@ -34,14 +64,84 @@ def _tool_handler(error_prefix: str): Exception -> tool_error("Error {prefix}: ...") """ def decorator(func): + """Decorate one synchronous tool function with the shared contract.""" @functools.wraps(func) async def wrapper(*args, **kwargs): - try: - async with _ableton_semaphore: - result = await asyncio.wait_for( - asyncio.to_thread(func, *args, **kwargs), - timeout=_TOOL_TIMEOUT_SECONDS, + """Execute the wrapped tool through ownership and timeout guards.""" + invocation = _InvocationGate() + + async def invoke(): + """Claim control when required and run one guarded tool call.""" + track_control = requires_control and ownership.is_configured() + claim_token = None + if track_control: + claim = await asyncio.to_thread( + ownership.ensure_control, + client_name=_get_client_name(args, kwargs), ) + if not claim.acquired: + return tool_error( + claim.error or "Ableton control is unavailable.", + {"control": claim.control}, + ) + claim_token = claim.claim_token + + result = await asyncio.to_thread( + _run_sync_tool, + func, + args, + kwargs, + track_control, + invocation, + ) + if result is _INVOCATION_ABANDONED: + if claim_token is not None: + release = await asyncio.to_thread( + ownership.release_if_current_claim, + claim_token, + ) + if not release.released and release.error: + logger.warning( + "Could not roll back abandoned Ableton claim: %s", + release.error, + ) + return None + return result + + deadline = asyncio.get_running_loop().time() + _TOOL_TIMEOUT_SECONDS + semaphore = None + if requires_control: + # Queueing is part of the caller's response deadline. Once the + # lease is acquired, it follows real work rather than the caller. + semaphore = _ableton_semaphore + try: + await asyncio.wait_for( + semaphore.acquire(), + timeout=max( + 0.0, + deadline - asyncio.get_running_loop().time(), + ), + ) + except asyncio.TimeoutError: + logger.error( + "Tool timed out after %ds waiting for serialized control: %s", + _TOOL_TIMEOUT_SECONDS, + error_prefix, + ) + return tool_error( + f"Tool timed out after {_TOOL_TIMEOUT_SECONDS}s: {error_prefix}" + ) + + task = asyncio.create_task(invoke()) + release_deferred = False + try: + result = await asyncio.wait_for( + asyncio.shield(task), + timeout=max( + 0.0, + deadline - asyncio.get_running_loop().time(), + ), + ) if isinstance(result, str): stripped = result.strip() if stripped.startswith(("{", "[")): @@ -49,19 +149,105 @@ async def wrapper(*args, **kwargs): return tool_success(result) return result except asyncio.TimeoutError: + # Worker threads cannot be cancelled. Keep the semaphore + # leased until the actual claim/tool work finishes. + invocation.abandon() + task.add_done_callback(_consume_background_result) + if semaphore is not None: + task.add_done_callback( + functools.partial( + _release_ableton_semaphore, + semaphore=semaphore, + ) + ) + release_deferred = True logger.error("Tool timed out after %ds: %s", _TOOL_TIMEOUT_SECONDS, error_prefix) return tool_error(f"Tool timed out after {_TOOL_TIMEOUT_SECONDS}s: {error_prefix}") + except asyncio.CancelledError: + invocation.abandon() + task.add_done_callback(_consume_background_result) + if semaphore is not None: + task.add_done_callback( + functools.partial( + _release_ableton_semaphore, + semaphore=semaphore, + ) + ) + release_deferred = True + raise except ValueError as e: return tool_error(f"Invalid input: {e}") except ConnectionError as e: return tool_error(f"M4L bridge not available: {e}") + except _ControlReleasedError as e: + control = await asyncio.to_thread(ownership.get_status) + return tool_error(str(e), {"control": control}) except Exception as e: logger.error("Error %s: %s", error_prefix, e) return tool_error(f"Error {error_prefix}: {e}") + finally: + if semaphore is not None and not release_deferred: + semaphore.release() return wrapper return decorator +def _run_sync_tool( + func, + args: tuple, + kwargs: dict, + track_control: bool, + invocation: _InvocationGate, +): + """Run a sync tool while tracking work that may outlive its async timeout.""" + if track_control and not ownership.begin_operation(): + raise _ControlReleasedError( + "Ableton control was released before this operation began. Try again." + ) + try: + if not invocation.try_start(): + return _INVOCATION_ABANDONED + return func(*args, **kwargs) + finally: + if track_control: + ownership.end_operation() + + +def _consume_background_result(task: asyncio.Task) -> None: + """Retrieve a timed-out task's result so late exceptions are not leaked.""" + try: + exc = task.exception() + except asyncio.CancelledError: + return + except Exception as exc: + logger.warning("Could not inspect timed-out tool task: %s", exc) + else: + if exc is not None: + logger.warning("Timed-out tool task finished with error: %s", exc) + + +def _release_ableton_semaphore( + _task: asyncio.Task, + *, + semaphore: asyncio.Semaphore, +) -> None: + """Release a tool's captured semaphore after shielded work completes.""" + semaphore.release() + + +def _get_client_name(args: tuple, kwargs: dict) -> str | None: + """Read the MCP initialize client name from a tool Context when available.""" + for candidate in (*args, *kwargs.values()): + try: + params = candidate.session.client_params + name = params.clientInfo.name if params and params.clientInfo else None + except (AttributeError, RuntimeError, ValueError): + continue + if isinstance(name, str) and name: + return name + return None + + def _m4l_result(result: dict) -> dict: """Extract result data from M4L response, or raise on error.""" if result.get("status") == "success": @@ -78,9 +264,12 @@ def tool_success(message: str, data: dict = None) -> str: return json.dumps(result) -def tool_error(message: str) -> str: +def tool_error(message: str, data: dict = None) -> str: """Create a standardized error response.""" - return json.dumps({"status": "error", "message": message}) + result = {"status": "error", "message": message} + if data: + result["data"] = data + return json.dumps(result) def _report_progress(ctx, current: float, total: float, message: str = None): diff --git a/MCP_Server/tools/session.py b/MCP_Server/tools/session.py index 2750eca..b617b1d 100644 --- a/MCP_Server/tools/session.py +++ b/MCP_Server/tools/session.py @@ -1,39 +1,35 @@ """Session & transport tool handlers for AbletonBridge.""" import json from mcp.server.fastmcp import Context -from MCP_Server.tools._base import _tool_handler, _m4l_result +from MCP_Server.tools._base import _tool_handler, _m4l_result, tool_error, tool_success from MCP_Server.connections.ableton import get_ableton_connection from MCP_Server.connections.m4l import get_m4l_connection from MCP_Server.validation import _validate_index, _validate_index_allow_negative, _validate_range import MCP_Server.state as state -from MCP_Server.dashboard.server import get_m4l_status +import MCP_Server.ownership as ownership +from MCP_Server.status import build_connection_status def register_tools(mcp): """Register session & transport tools with the MCP server.""" @mcp.tool() - @_tool_handler("getting server capabilities") + @_tool_handler("getting server capabilities", requires_control=False) def get_server_capabilities(ctx: Context) -> str: - """Report server version, connection status, available feature sets, and tool count. + """Report server version, ownership-aware connections, features, and tools. Call this first in any session to understand what features are available. - Returns JSON with connection status, M4L availability, browser cache state, etc. + Owner connection booleans are true or false. Standby connection booleans + are null, with explicit connection-state fields explaining why. """ from MCP_Server import __version__ - m4l_sockets_ready, m4l_connected = get_m4l_status() - ableton_connected = bool(state.ableton_connection and state.ableton_connection.sock) - try: - if ableton_connected: - state.ableton_connection.sock.getpeername() - except Exception: - ableton_connected = False + control = ownership.get_status() + connections = build_connection_status(control) return json.dumps({ "server_version": __version__, - "ableton_connected": ableton_connected, - "m4l_connected": m4l_connected, - "m4l_sockets_ready": m4l_sockets_ready, + **control, + **connections, "browser_cache_ready": state.browser_cache_ready.is_set(), "browser_cache_items": len(state.browser_cache_flat), "tool_count": len(mcp._tool_manager._tools) if hasattr(mcp, '_tool_manager') else 0, @@ -43,7 +39,7 @@ def get_server_capabilities(ctx: Context) -> str: "macros": True, "param_maps": True, "dashboard": state.dashboard_server is not None, - "m4l_bridge": m4l_connected, + "m4l_bridge": connections["m4l_connected"], }, "store_counts": { "snapshots": len(state.snapshot_store), @@ -53,6 +49,29 @@ def get_server_capabilities(ctx: Context) -> str: }) + @mcp.tool() + @_tool_handler("releasing Ableton control", requires_control=False) + def release_ableton_control(ctx: Context) -> str: + """Release Ableton control owned by this MCP process. + + This never steals or releases another task's ownership. If this task is + already on standby, the response is a harmless no-op. + """ + result = ownership.release_control() + data = { + "released": result.released, + "control": result.control, + } + if result.error and not result.released: + return tool_error(result.error, data) + if result.released: + message = "Ableton control released." + if result.error: + message += f" Backend cleanup warning: {result.error}" + return tool_success(message, data) + return tool_success("This task did not own Ableton control.", data) + + @mcp.tool() @_tool_handler("getting session info") def get_session_info(ctx: Context) -> str: diff --git a/README.md b/README.md index 8875f51..e623bd3 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # AbletonBridge -**359 tools connecting Claude AI to Ableton Live** (340 core + 19 optional ElevenLabs voice/SFX tools) +**366 tools connecting Claude AI to Ableton Live** (347 core + 19 optional ElevenLabs voice/SFX tools) AbletonBridge gives Claude direct control over your Ableton Live session through the Model Context Protocol. Create tracks, write MIDI, design sounds, mix, automate, browse instruments, snapshot presets, and navigate deep into device chains and modulation matrices — all through natural language conversation. @@ -31,14 +31,16 @@ Claude AI <--MCP--> MCP Server <--TCP:9877--> Ableton Remote Script +---<--HTTP:9880--> Web Status Dashboard MCP Server (modular architecture): - server.py — slim orchestrator (~300 lines) - state.py — centralized global state + locks + server.py — MCP orchestrator + owner-only backend lifecycle + ownership.py — process ownership, status, and safe handoff + status.py — ownership-aware Ableton and M4L connection status + state.py — centralized global state + locks constants.py — command tiers, browser categories validation.py — input validation + size limits connections/ — ableton.py (TCP), m4l.py (UDP/OSC) cache/ — browser.py (cache + disk persistence) dashboard/ — html.py, server.py (Starlette) - tools/ — 15 modules (340 tools) + tools/ — 16 modules (347 tools) prompts.py — 4 MCP prompt templates instructions.py — server instructions (cross-tool guidance) ``` @@ -53,28 +55,42 @@ MCP Server (modular architecture): --- -## Tool Overview (340 core + 19 optional = 359 total) +## Multiple MCP Clients + +Every MCP client receives the complete tool set, while one local process owns Ableton control at a time: + +- `get_server_capabilities` reports whether this process is the owner or a standby, plus owner process metadata when available. Connection booleans are `true` or `false` only for the owner; standbys return `null` with `not_started`, `owned_elsewhere`, or `unknown` connection states so an unstarted local backend is never mistaken for a disconnected Live instance. `features.m4l_bridge` mirrors the same tri-state M4L value. +- The first normal Ableton tool call automatically claims control when it is free. +- Standby tools return a structured ownership error instead of terminating the MCP server. +- `release_ableton_control` hands control back explicitly. Ownership is also released when the owning MCP process shuts down. +- Release returns `released: false` while a foreground control call or live M4L status probe is active. Port `9881` also stays owned if any dashboard, connection, or background worker has not stopped, so cleanup can be retried safely. +- Control is never stolen and has no idle timeout. + +--- + +## Tool Overview (347 core + 19 optional = 366 total) | Area | Examples | Count | |---|---|---| -| Session & Transport | tempo, play/record, capture, Link, punch, playback position | ~53 | -| Tracks & Mixing | create/rename tracks, routing, monitoring, groups, implicit arm | ~29 | -| Clips & Scenes | create/edit clips, follow actions, warp markers | ~54 | -| Scenes | create/delete/duplicate, fire, name, color, tempo, follow actions | ~10 | -| Mixer | unified set_mixer, batch_set_mixer, sends, crossfader | ~13 | -| Devices & Parameters | load/configure, rack chains, rack macros, sidechain, plugin info | ~45 | -| Browser & Presets | search/load instruments, presets, device presets | ~12 | -| Automation | clip/track automation, envelopes, curves | ~12 | -| Arrangement | arrangement clips, time editing, composition analysis | ~17 | -| Creative Generation | Euclidean rhythms, chords, drums, arpeggios, bass, transforms | ~17 | -| Deep Access (M4L) | hidden params, chain internals, audio analysis, note surgery | ~40 | -| Snapshots & Macros | snapshot/restore, morph, macros, parameter maps | ~18 | -| Audio Analysis | audio clip info, track meters, input meters | ~3 | -| Grid Notation | ASCII drum/melodic pattern I/O | ~2 | -| Compound Workflows | create instrument/drum track, batch mixer, effect chains | ~11 | -| **Core subtotal** | | **340** | +| Session & Transport | ownership, tempo, play/record, capture, Link, punch, playback position | 52 | +| Tracks | create/rename tracks, routing, monitoring, groups, implicit arm | 29 | +| Clips | create/edit clips, notes, follow actions, warp markers | 56 | +| Scenes | create/delete/duplicate, fire, name, color, tempo, follow actions | 10 | +| Mixer | unified set_mixer, sends, crossfader | 13 | +| Devices & Parameters | load/configure, rack chains, rack macros, sidechain, plugin info | 50 | +| Browser & Presets | search/load instruments, presets, device presets | 12 | +| Automation | clip/track automation, envelopes, curves | 12 | +| Arrangement | arrangement clips, time editing, composition analysis | 17 | +| Creative Generation | Euclidean rhythms, chords, drums, arpeggios, bass, transforms | 17 | +| Deep Access (M4L) | hidden params, chain internals, audio analysis, note surgery | 40 | +| Snapshots & Macros | snapshot/restore, morph, macros, parameter maps | 19 | +| Audio Analysis | audio clip info, track meters, input meters | 3 | +| Grid Notation | ASCII drum/melodic pattern I/O | 2 | +| Compound Workflows | create instrument/drum track, batch mixer, effect chains | 10 | +| MIDI CC | mapped plugin control, channel assignment, raw CC | 5 | +| **Core subtotal** | | **347** | | ElevenLabs (optional) | voice generation, SFX, cloning, transcription | 19 | -| **Total** | | **359** | +| **Total** | | **366** | See [CHANGELOG.md](CHANGELOG.md) for the complete per-tool breakdown. @@ -92,26 +108,26 @@ AbletonBridge is built to handle real-world sessions without crashing Ableton: - **Fire-and-forget writes** — no post-set readback (the #1 crash pattern) - **Command-specific timeouts** — per-command timeouts (e.g., freeze_track → 60s, load_instrument → 30s) instead of fixed 10s/15s - **Socket drain** — clears stale UDP responses before each command -- **Singleton guard** — exclusive port lock prevents duplicate server instances +- **Single backend owner** — every MCP process stays available while one process exclusively owns Live, M4L, and dashboard connections - **Disk-persisted cache** — 6,400+ browser items in gzip; instant startup (~50ms) - **Auto-reconnect** — exponential backoff for TCP and UDP connections - **Tiered command delays** — 3-tier system (0ms/10ms/20ms) eliminates unnecessary waits for property setters - **Async tool handlers** — all tools run via `asyncio.to_thread()`, preventing sync I/O from blocking the event loop - **Concurrency control** — async semaphore serializes tool dispatch; threading locks protect TCP and UDP sockets from corruption -- **Tool execution timeout** — 120s hard timeout prevents stuck tools from blocking the entire pipeline +- **Tool response deadline** — callers receive a structured timeout after 120s, including queue time; already-running work stays serialized until it exits because worker threads cannot be cancelled safely - **Bounded thread pool** — explicit 8-worker limit prevents resource exhaustion during rapid tool call bursts -- **Standardized responses** — all 340 tools return consistent `tool_success()`/`tool_error()` JSON envelopes via decorator +- **Standardized responses** — all 347 tools return consistent `tool_success()`/`tool_error()` JSON envelopes via decorator - **Chunk reassembly hardening** — duplicate detection, progress logging, missing chunk index reporting - **Parameter resolution cache** — 500-entry FIFO cache for brute-force display→value resolution (O(1) after first call) - **Effect chain persistence** — saved templates survive server restarts via `~/.ableton-bridge/chain_templates.json` -- **214 tests** — 11 test files covering connections, M4L, cache, creative tools, workflows, and validation edge cases +- **Automated regression suite** — covers ownership, multi-client stdio, connections, M4L, cache, creative tools, workflows, and validation edge cases --- ## Flexibility - **Any MCP client** — Claude Desktop, Cursor, Claude Code, or any MCP-compatible tool -- **300 tools without Max for Live** — full session control via TCP/UDP Remote Script; M4L is optional +- **307 tools without Max for Live** — full session control via TCP/UDP Remote Script; M4L is optional - **+40 deep-access tools with M4L** — hidden parameters, rack internals, audio analysis, event monitoring - **+19 optional ElevenLabs tools** — AI voice generation, sound effects, cloning, transcription - **Ableton Live 10, 11, and 12** — graceful API fallbacks for version-specific features diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 8e3aed3..4cc5a3a 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -2,7 +2,7 @@ ## Overview -AbletonBridge is a 3-layer system connecting AI assistants to Ableton Live through the Model Context Protocol (MCP). The MCP Server layer is modularized into 20+ focused modules. +AbletonBridge is a 3-layer system connecting AI assistants to Ableton Live through the Model Context Protocol (MCP). The MCP Server layer is modularized into 20+ focused modules. Each stdio MCP process exposes the complete tool surface, while a single process owns the shared Ableton backend resources at a time. ## System Architecture @@ -16,7 +16,7 @@ AbletonBridge is a 3-layer system connecting AI assistants to Ableton Live throu │ MCP Server │ │ ┌─────────┐ ┌────────────┐ ┌──────────────┐ │ │ │ server.py│ │ tools/* │ │ prompts.py │ │ -│ │(orchestr)│ │(15 modules)│ │(4 workflows) │ │ +│ │(orchestr)│ │(16 modules)│ │(4 workflows) │ │ │ └────┬─────┘ └─────┬──────┘ └──────────────┘ │ │ │ │ ┌──────────────┐ │ │ │ │ │instructions. │ │ @@ -56,11 +56,16 @@ AbletonBridge is a 3-layer system connecting AI assistants to Ableton Live throu ``` MCP_Server/ ├── __init__.py # Package init, __version__, re-exports -├── server.py # Slim orchestrator (~300 lines) -│ # - singleton lock, lifespan, MCP instance +├── server.py # MCP orchestrator and owner-only backend lifecycle +│ # - lifespan, backend start/stop, MCP instance │ # - tool/prompt/resource registration │ # - call instrumentation for dashboard -├── state.py # ALL global mutable state +├── ownership.py # Cross-process Ableton control coordination +│ # - atomic owner lock + status responder (:9881) +│ # - owner metadata, release, active-operation tracking +├── status.py # Ownership-aware Ableton and M4L connection status +│ # - tri-state standby fields, passive owner checks +├── state.py # ALL global mutable state │ # - connections, stores, caches, locks │ # - threading events, config, MCP instance ref ├── constants.py # Pure constants (no mutations) @@ -79,10 +84,11 @@ MCP_Server/ │ ├── connections/ │ ├── __init__.py # Re-exports -│ ├── ableton.py # AbletonConnection (TCP :9877) -│ │ # - tiered send_command() with per-tier delays -│ │ # - NON_IDEMPOTENT_COMMANDS (no retry for create/delete) -│ │ # - get_ableton_connection() singleton +│ ├── ableton.py # AbletonConnection (TCP :9877) +│ │ # - tiered send_command() with per-tier delays +│ │ # - NON_IDEMPOTENT_COMMANDS (no retry for create/delete) +│ │ # - passive socket liveness without consuming data +│ │ # - get_ableton_connection() singleton │ └── m4l.py # M4LConnection (UDP/OSC :9878/:9879) │ # - OSC message building, chunked response reassembly │ # - send_command_with_retry() (3 attempts, exponential backoff) @@ -104,25 +110,26 @@ MCP_Server/ │ # - DashboardLogHandler (pipes logs to buffer) │ └── tools/ - ├── __init__.py # register_all_tools(mcp) — calls 15 modules + ├── __init__.py # register_all_tools(mcp) — calls 16 modules ├── _base.py # Shared infrastructure │ # - _tool_handler (semaphore + asyncio.to_thread + timeout) │ # - _m4l_result(), tool_success(), tool_error() - ├── session.py # 56 tools: transport, tempo, recording, views, playback + ├── session.py # 52 tools: status, release, transport, recording, views ├── tracks.py # 29 tools: track CRUD, routing, monitoring, implicit arm - ├── clips.py # 54 tools: clip CRUD, notes, loop, follow actions - ├── mixer.py # 22 tools: volume, pan, sends, set_mixer - ├── devices.py # 44 tools: device params, racks, sidechain + ├── clips.py # 56 tools: clip CRUD, notes, loop, follow actions + ├── mixer.py # 13 tools: volume, pan, sends, set_mixer + ├── devices.py # 50 tools: device params, racks, sidechain ├── browser.py # 12 tools: search, load, presets ├── automation.py # 12 tools: clip/track automation - ├── arrangement.py # 12 tools: arrangement clips, time editing + ├── arrangement.py # 17 tools: arrangement clips, time editing, analysis ├── scenes.py # 10 tools: scene CRUD, fire, follow actions, tempo ├── creative.py # 17 tools: chords, drums, arpeggios, euclidean ├── m4l_tools.py # 40 tools: M4L bridge (hidden params, chains) - ├── snapshots.py # 18 tools: snapshot/macro/param_map stores + ├── snapshots.py # 19 tools: snapshot/macro/param_map stores ├── audio.py # 3 tools: audio analysis, input meters ├── grid.py # 2 tools: grid notation I/O - └── workflows.py # 10 tools: compound workflow tools + ├── workflows.py # 10 tools: compound workflow tools + └── midi_cc.py # 5 tools: mapped and raw MIDI CC control ``` ### Remote Script (`AbletonBridge_Remote_Script/`) @@ -159,21 +166,23 @@ The module import graph is strictly acyclic: Level 0 (no internal imports): state.py, constants.py, validation.py, grid_notation.py, instructions.py -Level 1 (imports Level 0 only): - connections/ableton.py → state, constants - connections/m4l.py → state - -Level 2 (imports Levels 0-1): - cache/browser.py → state, constants, connections.ableton - dashboard/server.py → state, connections.ableton - -Level 3 (imports Levels 0-2): - tools/_base.py → (standalone: asyncio, logging, json) - tools/*.py → _base, connections, validation, state, cache +Level 1 (imports Level 0 only): + ownership.py → state + connections/ableton.py → state, constants + connections/m4l.py → state + +Level 2 (imports Levels 0-1): + status.py → state, ownership + cache/browser.py → state, constants, connections.ableton + tools/_base.py → ownership + +Level 3 (imports Levels 0-2): + dashboard/server.py → state, ownership, status + tools/*.py → _base, connections, validation, state, status, cache prompts.py → (standalone: just receives mcp instance) -Level 4 (imports everything): - server.py → state, connections, cache, dashboard, tools, prompts, instructions +Level 4 (imports everything): + server.py → state, ownership, status, connections, cache, dashboard, tools, prompts, instructions ``` **Rule:** No module at Level N imports from Level N or higher. This prevents circular imports. @@ -220,6 +229,43 @@ Recv (9879): OSC response with result (possibly chunked) - Real-time tool call metrics and server logs - Auto-refreshes every 3 seconds +## Control Ownership Lifecycle + +MCP stdio connections are process-private: clients such as Codex may launch one AbletonBridge process per task, but the Live TCP connection, M4L receive socket, dashboard port, and related runtime state must have one owner. Tool availability is therefore separated from backend ownership. + +| Event | Behaviour | +|------|-----------| +| MCP process starts | Registers all tools immediately and begins in `standby` without connecting to Live. | +| First normal tool call | Atomically binds loopback port `9881`, starts the backend resources, and becomes `owner`. | +| Another process already owns `9881` | Remains healthy in `standby`; tools return a structured ownership error instead of terminating MCP initialization. | +| Status call | `get_server_capabilities` reports ownership plus role-aware connection states without claiming control. Owner values come from local checks; standby values are `null`. | +| Explicit release | `release_ableton_control` closes `9881` only after every owner resource has stopped; incomplete cleanup returns `released: false` and retains ownership for a safe retry. | +| MCP shutdown | Performs the same release automatically. | +| Backend startup fails | MCP remains healthy in standby, but owner-only services do not start. Partial resources are cleaned up and `9881` is released only after cleanup is confirmed. | + +The `9881` owner socket also serves a loopback-only JSON status response. Reusing the lock socket avoids a stale metadata file or another management port. If an unrelated process occupies the port, availability is reported as `occupied_unknown`. + +Connection status is scoped to the calling MCP process. Only the owner has backend sockets it can inspect, so standby processes never report `false` for Ableton or M4L connectivity: + +| Control state | Connection state | Connection booleans | +|---------------|------------------|---------------------| +| Standby, control available | `not_started` | `null` | +| Standby, known AbletonBridge owner | `owned_elsewhere` | `null` | +| Standby, unknown port occupant | `unknown` | `null` | +| Owner, healthy local connection | `connected` | `true` | +| Owner, failed local connection | `disconnected` | `false` | +| Owner, M4L sockets awaiting a bridge response | `sockets_ready` | M4L connected `false` | + +`null` means the local backend is absent and the connection cannot be evaluated from this process. It does not mean Live is disconnected. Remote-owner health is intentionally not added to the `9881` responder, avoiding stale distributed health state and preserving the ownership protocol. + +`features.m4l_bridge` mirrors the tri-state `m4l_connected` value: `true` or `false` for the owner and `null` for standby processes. + +Ableton status is passive: it never sends or consumes protocol data. If a command owns the socket lock, status preserves the last verified socket result instead of interfering with the response stream. An expired M4L status cache may perform a live ping; that probe is registered as an owner operation, so release refuses rather than disconnecting or reconnecting M4L underneath it. + +M4L connection replacement, ping-cache updates, warmup publication, and teardown share `state.m4l_connection_lock`. A status request acquires this lock without waiting; if another connection transaction is active, it returns one immutable last-known snapshot instead of combining fields from different connection generations or probing a stale connection. The lock order is ownership operation, M4L state, then M4L socket. + +Ownership has no idle timeout and cannot be stolen. Manual release is refused while a foreground tool, controlled resource, or live status probe is active. A call that times out or is cancelled before its tool body starts is abandoned; an already-running worker keeps its operation lease until it exits. Owner background services such as M4L connection and browser warmup are cooperatively cancelled and joined during release; port `9881` remains owned if any worker fails to stop. Forced shutdown follows the same retention rule. These constraints keep handoff explicit and prevent two processes from using backend resources during a transition. + ## Command Delay Tiers | Tier | Pre-Delay | Post-Delay | Example Commands | @@ -239,7 +285,7 @@ Defined in `instructions.py` and passed to `FastMCP(instructions=...)`. Automati ### Resources (3) - `ableton://session` — current session state - `ableton://tracks` — all track information -- `ableton://capabilities` — server version, connections, cache +- `ableton://capabilities` — server version, ownership-aware connections, cache ### Prompts (4) - `create_beat` — guided drum pattern creation @@ -247,16 +293,14 @@ Defined in `instructions.py` and passed to `FastMCP(instructions=...)`. Automati - `sound_design` — parameter exploration guide - `arrange_section` — arrangement section builder -### Tools (334 core + 19 optional) +### Tools (347 core + 19 optional ElevenLabs) All tools use the `@_tool_handler` decorator which: -1. Gates execution via `asyncio.Semaphore(1)` — only one tool runs at a time, preventing thread pool exhaustion and TCP socket corruption -2. Wraps sync functions in `asyncio.to_thread()` for non-blocking execution -3. Enforces a 120-second timeout via `asyncio.wait_for()` — prevents stuck tools from blocking the semaphore indefinitely -4. Catches `asyncio.TimeoutError` → "Tool timed out" responses -5. Catches `ValueError` → "Invalid input" responses -6. Catches `ConnectionError` → "M4L bridge not available" responses -7. Catches generic exceptions → logged + returned as error strings -8. Auto-wraps plain-string returns in `tool_success()` JSON envelopes +1. Gates owner-dependent execution via `asyncio.Semaphore(1)` to prevent TCP socket corruption; claim-free status and release remain available as recovery paths +2. Automatically claims Ableton control for normal tools with a bounded wait; status and release are explicitly exempt +3. Wraps sync functions in `asyncio.to_thread()` for non-blocking execution +4. Abandons calls that time out before their tool body starts, rolling back ownership only when that invocation created the still-current claim +5. Enforces a 120-second caller response deadline across queueing and execution; already-running workers keep their serialization lease until they exit +6. Returns consistent structured success, validation, connection, ownership, timeout, and generic error responses ## Testing @@ -267,7 +311,9 @@ tests/ ├── test_grid_notation.py # 7 tests: parse/format round-trips ├── test_constants.py # 4 tests: tier disjointness, completeness ├── test_state.py # 5 tests: thread-safety, events, stores -└── test_tool_handler.py # 11 tests: async decorator, error handling +├── test_tool_handler.py # async decorator, errors, ownership guard, timeout safety +├── test_ownership.py # claims, release, metadata, collisions, two stdio clients +└── test_status.py # tri-state connections, public surfaces, dashboard status ``` Run tests: @@ -296,3 +342,5 @@ pytest tests/ -v - 5ms inter-command delay in Remote Script — defense-in-depth against scheduler flooding This layered approach replaced the original large delays (50-200ms) with proper synchronization primitives, achieving both faster throughput and better stability. + +8. **Available MCP, single-owner backend** — every stdio process completes MCP initialization and exposes all tools. Port `9881` coordinates one owner for Live, M4L, dashboard, and background services. This preserves the existing stdio deployment model while removing the failure mode where a singleton collision caused later clients to receive no Ableton tools. diff --git a/tests/conftest.py b/tests/conftest.py index 99c6803..3862a8e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -26,15 +26,19 @@ def mock_m4l(): @pytest.fixture(autouse=True) def reset_state(): """Reset global state between tests.""" - original_ableton = state.ableton_connection - original_m4l = state.m4l_connection + original_ableton = state.ableton_connection + original_m4l = state.m4l_connection + original_m4l_ping_cache = state.m4l_ping_cache.copy() + original_m4l_status_snapshot = state.m4l_status_snapshot original_snapshots = state.snapshot_store.copy() original_macros = state.macro_store.copy() original_param_maps = state.param_map_store.copy() original_chains = state.effect_chain_store.copy() yield - state.ableton_connection = original_ableton - state.m4l_connection = original_m4l + state.ableton_connection = original_ableton + state.m4l_connection = original_m4l + state.m4l_ping_cache = original_m4l_ping_cache + state.m4l_status_snapshot = original_m4l_status_snapshot state.snapshot_store = original_snapshots state.macro_store = original_macros state.param_map_store = original_param_maps diff --git a/tests/test_browser_cache.py b/tests/test_browser_cache.py index 2a011bd..359c279 100644 --- a/tests/test_browser_cache.py +++ b/tests/test_browser_cache.py @@ -3,13 +3,15 @@ import gzip import os import tempfile +import threading from unittest.mock import patch, MagicMock import MCP_Server.state as state from MCP_Server.cache.browser import ( build_device_uri_map, save_browser_cache_to_disk, - load_browser_cache_from_disk, resolve_device_uri, + load_browser_cache_from_disk, populate_browser_cache, resolve_device_uri, get_browser_cache, ) +from MCP_Server.connections.ableton import CommandCancelled class TestBuildDeviceUriMap: @@ -92,3 +94,47 @@ def test_unknown_name_returns_input(self): state.browser_cache_ready.set() result = resolve_device_uri("NonexistentDevice") assert result == "NonexistentDevice" + + +def test_live_scan_honors_shutdown_and_preserves_existing_cache(monkeypatch): + """A cancelled live scan must preserve the last usable browser cache.""" + stop_event = threading.Event() + existing = [{"name": "Existing"}] + calls = [] + + class FakeConnection: + def __init__(self, **_kwargs): + """Create a no-op connection double.""" + pass + + def connect(self): + """Pretend the browser-scan connection succeeds.""" + return True + + def send_command(self, _command, _params, *, timeout, stop_event): + """Cancel the scan during its first browser command.""" + calls.append((timeout, stop_event)) + stop_event.set() + raise CommandCancelled("stopping") + + def disconnect(self): + """Record cleanup of the scan connection.""" + calls.append("disconnect") + + monkeypatch.setattr( + "MCP_Server.connections.ableton.AbletonConnection", + FakeConnection, + ) + monkeypatch.setattr( + "MCP_Server.cache.browser.BROWSER_CATEGORIES", + [("instruments", "Instruments")], + ) + monkeypatch.setattr(state, "browser_cache_flat", existing) + monkeypatch.setattr(state, "browser_cache_timestamp", 0.0) + monkeypatch.setattr(state, "browser_cache_populating", False) + + assert populate_browser_cache(force=True, stop_event=stop_event) is False + assert calls[0] == (60.0, stop_event) + assert calls[-1] == "disconnect" + assert state.browser_cache_flat is existing + assert state.browser_cache_populating is False diff --git a/tests/test_connections.py b/tests/test_connections.py index 2474b11..2dc20c0 100644 --- a/tests/test_connections.py +++ b/tests/test_connections.py @@ -2,8 +2,14 @@ import json import socket import time +import threading from unittest.mock import MagicMock, patch, PropertyMock, call -from MCP_Server.connections.ableton import AbletonConnection, get_ableton_connection, NON_IDEMPOTENT_COMMANDS +from MCP_Server.connections.ableton import ( + AbletonConnection, + CommandCancelled, + NON_IDEMPOTENT_COMMANDS, + get_ableton_connection, +) from MCP_Server.constants import TIER_0_COMMANDS, TIER_1_COMMANDS, TIER_2_COMMANDS import MCP_Server.state as state @@ -38,6 +44,7 @@ def test_idempotent_retry_on_failure(self): conn._recv_buffer = "" call_count = [0] def side_effect(*args, **kwargs): + """Fail the first receive and succeed after reconnection.""" call_count[0] += 1 if call_count[0] == 1: raise socket.error("connection reset") @@ -75,24 +82,132 @@ def test_tier_membership(self): assert len(TIER_1_COMMANDS & TIER_2_COMMANDS) == 0 assert len(TIER_0_COMMANDS & TIER_2_COMMANDS) == 0 + def test_receive_full_response_honors_shutdown(self): + """Socket receive cancellation should be prompt and suppress timeout context.""" + conn = AbletonConnection(host="localhost", port=9877) + client, server = socket.socketpair() + stop_event = threading.Event() + timer = threading.Timer(0.05, stop_event.set) + timer.start() + started = time.monotonic() + try: + with pytest.raises(CommandCancelled) as exc_info: + conn.receive_full_response( + client, + timeout=5.0, + stop_event=stop_event, + ) + assert time.monotonic() - started < 1.0 + assert exc_info.value.__suppress_context__ is True + finally: + timer.cancel() + client.close() + server.close() + + def test_retry_delay_cancellation_suppresses_failure_context(self): + """Retry cancellation should not present the triggering command error as its cause.""" + conn = AbletonConnection(host="localhost", port=9877) + conn.sock = MagicMock() + stop_event = MagicMock() + stop_event.is_set.return_value = False + stop_event.wait.return_value = True + + with patch.object( + conn, + "receive_full_response", + side_effect=RuntimeError("command failed"), + ): + with pytest.raises(CommandCancelled) as exc_info: + conn.send_command("get_session_info", stop_event=stop_event) + + assert exc_info.value.__suppress_context__ is True + + +class TestAbletonConnectionLiveness: + def test_open_peer_is_connected(self): + """An open idle peer should remain connected without receiving data.""" + client, server = socket.socketpair() + connection = AbletonConnection("localhost", 9877, sock=client) + try: + assert connection.is_connected() is True + finally: + client.close() + server.close() + + def test_closed_peer_is_disconnected(self): + """A clean peer shutdown should be visible without sending a command.""" + client, server = socket.socketpair() + connection = AbletonConnection("localhost", 9877, sock=client) + try: + server.shutdown(socket.SHUT_RDWR) + server.close() + assert connection.is_connected() is False + finally: + client.close() + + def test_liveness_check_does_not_consume_pending_data(self): + """Peeking at a readable socket must leave protocol bytes untouched.""" + client, server = socket.socketpair() + connection = AbletonConnection("localhost", 9877, sock=client) + try: + server.sendall(b"response") + assert connection.is_connected() is True + assert client.recv(8) == b"response" + finally: + client.close() + server.close() + + def test_busy_connection_is_not_disturbed(self): + """Status should reuse the last healthy result while a command is active.""" + client, server = socket.socketpair() + connection = AbletonConnection("localhost", 9877, sock=client) + connection._send_lock.acquire() + try: + assert connection.is_connected() is True + assert connection._send_lock.locked() + finally: + connection._send_lock.release() + client.close() + server.close() + + def test_busy_connection_preserves_last_disconnected_result(self): + """A busy status check should not turn a known dead socket back to true.""" + client, server = socket.socketpair() + connection = AbletonConnection("localhost", 9877, sock=client) + try: + server.shutdown(socket.SHUT_RDWR) + server.close() + assert connection.is_connected() is False + + connection._send_lock.acquire() + assert connection.is_connected() is False + assert connection._send_lock.locked() + finally: + if connection._send_lock.locked(): + connection._send_lock.release() + client.close() + class TestGetAbletonConnection: - def test_returns_existing_valid_connection(self): - """Should return existing connection if socket is valid.""" + def test_returns_existing_valid_connection(self, monkeypatch): + """A reused valid connection should also signal backend readiness.""" mock_conn = MagicMock() mock_conn.sock = MagicMock() - mock_conn.sock.getpeername.return_value = ("localhost", 9877) + mock_conn.is_connected.return_value = True mock_conn.send_command.return_value = {"status": "success"} + connected_event = threading.Event() state.ableton_connection = mock_conn + monkeypatch.setattr(state, "ableton_connected_event", connected_event) with patch('MCP_Server.connections.ableton.AbletonConnection'): result = get_ableton_connection() assert result == mock_conn + assert connected_event.is_set() def test_reconnects_on_dead_socket(self): """Should create new connection if existing socket is dead.""" mock_conn = MagicMock() mock_conn.sock = MagicMock() - mock_conn.sock.getpeername.side_effect = socket.error("not connected") + mock_conn.is_connected.return_value = False state.ableton_connection = mock_conn new_conn = MagicMock() new_conn.connect.return_value = True diff --git a/tests/test_ownership.py b/tests/test_ownership.py new file mode 100644 index 0000000..865df7e --- /dev/null +++ b/tests/test_ownership.py @@ -0,0 +1,1045 @@ +"""Ownership coordination and multi-client MCP availability tests.""" + +import json +import os +import socket +import sys +import tempfile +import threading +from pathlib import Path + +import pytest + +from MCP_Server.ownership import OwnershipManager + + +def _configured_manager(port, *, start=None, stop=None, environment=None): + """Build an ownership manager with lightweight lifecycle callbacks.""" + manager = OwnershipManager(port, environment=environment) + manager.configure_backend(start or (lambda: None), stop or (lambda: None)) + return manager + + +def test_one_owner_and_standby_metadata(unused_tcp_port): + """A standby should report reliable metadata for the current owner.""" + first = _configured_manager( + unused_tcp_port, + environment={"CODEX_THREAD_ID": "task-owner"}, + ) + second = _configured_manager(unused_tcp_port) + try: + claimed = first.ensure_control(client_name="Codex") + standby = second.ensure_control(client_name="Other client") + + assert claimed.acquired is True + assert standby.acquired is False + assert standby.control["control_role"] == "standby" + assert standby.control["control_availability"] == "owned" + assert standby.control["owner"]["instance_id"] == claimed.control["owner"]["instance_id"] + assert standby.control["owner"]["client_name"] == "Codex" + assert standby.control["owner"]["task_id"] == "task-owner" + assert standby.control["owner"]["process_id"] == os.getpid() + finally: + first.shutdown() + second.shutdown() + + +def test_release_allows_standby_to_claim(unused_tcp_port): + """Manual release should let a waiting process become the next owner.""" + starts = [] + stops = [] + first = _configured_manager( + unused_tcp_port, + start=lambda: starts.append("first"), + stop=lambda: stops.append("first"), + ) + second = _configured_manager( + unused_tcp_port, + start=lambda: starts.append("second"), + stop=lambda: stops.append("second"), + ) + try: + assert first.ensure_control().acquired is True + assert second.ensure_control().acquired is False + assert first.release().released is True + assert second.ensure_control().acquired is True + assert starts == ["first", "second"] + assert stops == ["first"] + finally: + first.shutdown() + second.shutdown() + + +def test_abandoned_claim_release_is_generation_safe(unused_tcp_port): + """A stale claim token must not release a newer local ownership generation.""" + manager = _configured_manager(unused_tcp_port) + try: + first = manager.ensure_control() + assert first.claim_token is not None + assert manager.release().released is True + + second = manager.ensure_control() + assert second.claim_token is not None + assert second.claim_token != first.claim_token + + stale = manager.release_if_current_claim(first.claim_token) + assert stale.released is False + assert manager.status()["control_role"] == "owner" + + current = manager.release_if_current_claim(second.claim_token) + assert current.released is True + assert manager.status()["control_role"] == "standby" + finally: + manager.shutdown() + + +def test_shutdown_automatically_releases_control(unused_tcp_port): + """Normal process shutdown should release control for another manager.""" + owner = _configured_manager(unused_tcp_port) + next_owner = _configured_manager(unused_tcp_port) + try: + assert owner.ensure_control().acquired is True + owner.shutdown() + assert next_owner.ensure_control().acquired is True + finally: + owner.shutdown() + next_owner.shutdown() + + +def test_simultaneous_claim_has_exactly_one_winner(unused_tcp_port): + """Concurrent claims should produce exactly one backend owner.""" + managers = [ + _configured_manager(unused_tcp_port), + _configured_manager(unused_tcp_port), + ] + barrier = threading.Barrier(3) + results = [] + + def claim(manager): + """Attempt one claim after all contenders reach the barrier.""" + barrier.wait() + results.append(manager.ensure_control().acquired) + + threads = [threading.Thread(target=claim, args=(manager,)) for manager in managers] + try: + for thread in threads: + thread.start() + barrier.wait() + for thread in threads: + thread.join(timeout=2.0) + + assert sorted(results) == [False, True] + finally: + for manager in managers: + manager.shutdown() + + +def test_backend_start_failure_releases_port(unused_tcp_port): + """A cleanly handled startup failure should make the port claimable again.""" + stopped = [] + + def fail_start(): + """Simulate Ableton being unavailable during backend startup.""" + raise RuntimeError("Live unavailable") + + failing = _configured_manager( + unused_tcp_port, + start=fail_start, + stop=lambda: stopped.append(True), + ) + replacement = _configured_manager(unused_tcp_port) + try: + result = failing.ensure_control() + + assert result.acquired is False + assert "Live unavailable" in result.error + assert stopped == [True] + assert replacement.ensure_control().acquired is True + finally: + failing.shutdown() + replacement.shutdown() + + +def test_status_responder_start_failure_rolls_back_ownership( + monkeypatch, + unused_tcp_port, +): + """A responder start exception must leave no local ownership to release.""" + starts = [] + stops = [] + manager = _configured_manager( + unused_tcp_port, + start=lambda: starts.append(True), + stop=lambda: stops.append(True), + ) + + def fail_start(_thread): + """Fail before the responder thread reaches a started state.""" + raise RuntimeError("thread creation unavailable") + + monkeypatch.setattr(threading.Thread, "start", fail_start) + + result = manager.ensure_control() + + assert result.acquired is False + assert "thread creation unavailable" in result.error + assert result.control["control_role"] == "standby" + assert result.control["control_availability"] == "available" + assert starts == [] + assert stops == [] + assert manager._listener is None + assert manager._owner is None + assert manager._responder_stop is None + assert manager._responder_thread is None + assert manager.release().released is False + manager.shutdown() + + +def test_status_responder_start_failure_can_retry(monkeypatch, unused_tcp_port): + """A later claim on the same manager should retry responder startup.""" + original_start = threading.Thread.start + attempts = [] + starts = [] + manager = _configured_manager( + unused_tcp_port, + start=lambda: starts.append(True), + ) + + def fail_once(thread): + """Fail only the first responder start, then use the real implementation.""" + attempts.append(thread.name) + if len(attempts) == 1: + raise RuntimeError("transient thread failure") + return original_start(thread) + + monkeypatch.setattr(threading.Thread, "start", fail_once) + try: + first = manager.ensure_control() + second = manager.ensure_control() + + assert first.acquired is False + assert second.acquired is True + assert attempts == ["ableton-owner-status", "ableton-owner-status"] + assert starts == [True] + finally: + manager.shutdown() + + +def test_failed_start_retains_port_until_partial_cleanup_finishes(unused_tcp_port): + """Failed startup should retain ownership while cleanup remains incomplete.""" + cleanup_ready = {"value": False} + + def fail_start(): + """Simulate startup failing after owner resources may have begun.""" + raise RuntimeError("Live unavailable") + + failing = _configured_manager( + unused_tcp_port, + start=fail_start, + stop=lambda: cleanup_ready["value"], + ) + replacement = _configured_manager(unused_tcp_port) + try: + result = failing.ensure_control() + assert result.acquired is False + assert "ownership was retained" in result.error + assert result.control["control_role"] == "owner" + assert replacement.ensure_control().acquired is False + + cleanup_ready["value"] = True + assert failing.release().released is True + assert replacement.ensure_control().acquired is True + finally: + cleanup_ready["value"] = True + failing.shutdown() + replacement.shutdown() + + +def test_startup_and_shutdown_transitions_are_serialized(unused_tcp_port): + """Shutdown should wait for an in-progress startup transition.""" + start_entered = threading.Event() + allow_start = threading.Event() + shutdown_entered = threading.Event() + shutdown_done = threading.Event() + claims = [] + stops = [] + + def start(): + """Pause backend startup until the shutdown race is observable.""" + start_entered.set() + allow_start.wait(timeout=1.0) + + def shutdown(): + """Request shutdown and record when the transition completes.""" + shutdown_entered.set() + manager.shutdown() + shutdown_done.set() + + manager = _configured_manager( + unused_tcp_port, + start=start, + stop=lambda: stops.append(True), + ) + claim_thread = threading.Thread( + target=lambda: claims.append(manager.ensure_control()), + ) + shutdown_thread = threading.Thread(target=shutdown) + try: + claim_thread.start() + assert start_entered.wait(timeout=1.0) + shutdown_thread.start() + assert shutdown_entered.wait(timeout=1.0) + + # Shutdown must wait for startup instead of tearing resources down + # underneath it and then allowing startup to publish a stale owner. + assert not shutdown_done.wait(timeout=0.05) + allow_start.set() + claim_thread.join(timeout=1.0) + shutdown_thread.join(timeout=1.0) + + assert len(claims) == 1 + assert claims[0].acquired is True + assert stops == [True] + assert manager.status()["control_role"] == "standby" + assert manager.status()["control_availability"] == "available" + finally: + allow_start.set() + manager.shutdown() + + +def test_incomplete_cleanup_retains_port_ownership(unused_tcp_port): + """A false cleanup result should retain ownership until a later retry.""" + cleanup_ready = {"value": False} + owner = _configured_manager( + unused_tcp_port, + stop=lambda: cleanup_ready["value"], + ) + standby = _configured_manager(unused_tcp_port) + try: + assert owner.ensure_control().acquired is True + + incomplete = owner.release() + assert incomplete.released is False + assert "still stopping" in incomplete.error + assert incomplete.control["control_role"] == "owner" + assert standby.ensure_control().acquired is False + + cleanup_ready["value"] = True + assert owner.release().released is True + assert standby.ensure_control().acquired is True + finally: + cleanup_ready["value"] = True + owner.shutdown() + standby.shutdown() + + +def test_cleanup_exception_retains_port_ownership(unused_tcp_port): + """A cleanup exception should retain ownership and surface its reason.""" + cleanup_raises = {"value": True} + + def stop(): + """Fail cleanup until the test allows a successful retry.""" + if cleanup_raises["value"]: + raise RuntimeError("dashboard still bound") + return True + + owner = _configured_manager(unused_tcp_port, stop=stop) + standby = _configured_manager(unused_tcp_port) + try: + assert owner.ensure_control().acquired is True + + failed = owner.release() + assert failed.released is False + assert "dashboard still bound" in failed.error + assert standby.ensure_control().acquired is False + + cleanup_raises["value"] = False + assert owner.release().released is True + assert standby.ensure_control().acquired is True + finally: + cleanup_raises["value"] = False + owner.shutdown() + standby.shutdown() + + +def test_release_refuses_active_operation(unused_tcp_port): + """Manual release should refuse while an owner-dependent operation runs.""" + manager = _configured_manager(unused_tcp_port) + try: + assert manager.ensure_control().acquired is True + assert manager.begin_operation() is True + + busy = manager.release() + assert busy.released is False + assert "still running" in busy.error + assert busy.control["active_operations"] == 1 + + manager.end_operation() + assert manager.release().released is True + finally: + manager.shutdown() + + +def test_operation_registration_waits_for_backend_owner_phase(unused_tcp_port): + """Owner operations must not begin while backend activation is incomplete.""" + startup_entered = threading.Event() + finish_startup = threading.Event() + claims = [] + + def start(): + """Hold the manager in its starting phase for the assertion.""" + startup_entered.set() + finish_startup.wait(timeout=1.0) + + manager = _configured_manager(unused_tcp_port, start=start) + claimant = threading.Thread( + target=lambda: claims.append(manager.ensure_control()), + ) + try: + claimant.start() + assert startup_entered.wait(timeout=1.0) + assert manager.begin_operation() is False + + finish_startup.set() + claimant.join(timeout=1.0) + assert not claimant.is_alive() + assert claims[0].acquired is True + assert manager.release().released is True + finally: + finish_startup.set() + if claimant.ident is not None: + claimant.join(timeout=1.0) + manager.shutdown() + + +def test_failed_start_retains_owner_while_operation_is_active(unused_tcp_port): + """Defensive failed-start cleanup must not hand off across a live operation.""" + manager = None + + def fail_with_active_operation(): + """Inject an operation count that survived an abnormal startup path.""" + with manager._lock: + manager._active_operations = 1 + raise RuntimeError("activation failed") + + manager = _configured_manager( + unused_tcp_port, + start=fail_with_active_operation, + stop=lambda: True, + ) + contender = _configured_manager(unused_tcp_port) + try: + failed = manager.ensure_control() + + assert failed.acquired is False + assert failed.control["control_role"] == "owner" + assert failed.control["active_operations"] == 1 + assert "still running" in failed.error + assert contender.ensure_control().acquired is False + + manager.end_operation() + assert manager.release().released is True + assert contender.ensure_control().acquired is True + finally: + manager.shutdown() + contender.shutdown() + + +def test_release_cancels_cooperative_background_worker( + unused_tcp_port, +): + """Cancellable owner services should not block an intentional handoff.""" + stop_event = threading.Event() + started = threading.Event() + workers = [] + + def background(stop): + """Wait until backend teardown requests cancellation.""" + started.set() + stop.wait(timeout=1.0) + + def start(): + """Start one representative owner background service.""" + worker = threading.Thread( + target=background, + args=(stop_event,), + ) + workers.append(worker) + worker.start() + + def stop(): + """Cooperatively stop and join every background service.""" + stop_event.set() + for worker in workers: + worker.join(timeout=1.0) + return all(not worker.is_alive() for worker in workers) + + manager = _configured_manager(unused_tcp_port, start=start, stop=stop) + try: + assert manager.ensure_control().acquired is True + assert started.wait(timeout=1.0) + assert manager.status()["active_operations"] == 0 + + released = manager.release() + + assert released.released is True + assert stop_event.is_set() + assert all(not worker.is_alive() for worker in workers) + finally: + stop_event.set() + for worker in workers: + worker.join(timeout=1.0) + manager.shutdown() + + +def test_release_retains_control_for_uncooperative_background_worker( + unused_tcp_port, +): + """A service that ignores cancellation should retain ownership until stopped.""" + stop_event = threading.Event() + started = threading.Event() + finish = threading.Event() + workers = [] + + def background(_stop): + """Ignore cooperative cancellation until explicitly released by the test.""" + started.set() + finish.wait(timeout=1.0) + + def start(): + """Start one simulated stuck owner background service.""" + worker = threading.Thread( + target=background, + args=(stop_event,), + ) + workers.append(worker) + worker.start() + + def stop(): + """Request cancellation and report whether the worker actually stopped.""" + stop_event.set() + for worker in workers: + worker.join(timeout=0.01) + return all(not worker.is_alive() for worker in workers) + + owner = _configured_manager(unused_tcp_port, start=start, stop=stop) + standby = _configured_manager(unused_tcp_port) + try: + assert owner.ensure_control().acquired is True + assert started.wait(timeout=1.0) + + incomplete = owner.release() + + assert incomplete.released is False + assert stop_event.is_set() + assert owner.status()["control_role"] == "owner" + assert standby.ensure_control().acquired is False + + finish.set() + assert owner.release().released is True + assert standby.ensure_control().acquired is True + finally: + stop_event.set() + finish.set() + for worker in workers: + worker.join(timeout=1.0) + owner.shutdown() + standby.shutdown() + + +def test_shutdown_retains_control_until_active_operation_finishes(unused_tcp_port): + """Forced shutdown must retain the port while timed-out tool work remains.""" + stops = [] + owner = _configured_manager( + unused_tcp_port, + stop=lambda: stops.append(True), + ) + standby = _configured_manager(unused_tcp_port) + try: + assert owner.ensure_control().acquired is True + assert owner.begin_operation() is True + + owner.shutdown() + + status = owner.status() + assert status["control_role"] == "owner" + assert status["active_operations"] == 1 + assert standby.ensure_control().acquired is False + + owner.end_operation() + owner.shutdown() + + assert stops == [True, True] + assert standby.ensure_control().acquired is True + finally: + owner.end_operation() + owner.shutdown() + standby.shutdown() + + +def test_unrelated_listener_is_reported_as_unknown(unused_tcp_port): + """An unrelated process on the lock port should be classified as unknown.""" + listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + listener.bind(("127.0.0.1", unused_tcp_port)) + listener.listen(1) + manager = _configured_manager(unused_tcp_port) + try: + status = manager.status() + claim = manager.ensure_control() + + assert status["control_role"] == "standby" + assert status["control_availability"] == "occupied_unknown" + assert status["owner"] is None + assert claim.acquired is False + assert "unknown process" in claim.error + finally: + listener.close() + manager.shutdown() + + +def test_non_object_status_payload_is_reported_as_unknown(unused_tcp_port): + """Valid non-object JSON must not be mistaken for bridge owner metadata.""" + listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + listener.bind(("127.0.0.1", unused_tcp_port)) + listener.listen(1) + + def respond(): + """Return a syntactically valid but structurally invalid payload.""" + client, _address = listener.accept() + try: + client.sendall(b"[]\n") + finally: + client.close() + + responder = threading.Thread(target=respond) + responder.start() + manager = _configured_manager(unused_tcp_port) + try: + status = manager.status() + assert status["control_role"] == "standby" + assert status["control_availability"] == "occupied_unknown" + assert status["owner"] is None + finally: + listener.close() + responder.join(timeout=1.0) + manager.shutdown() + + +def test_dashboard_shutdown_tolerates_unstarted_thread(monkeypatch): + """A startup/shutdown race must not abort dashboard cleanup.""" + import MCP_Server.state as state + from MCP_Server.dashboard.server import stop_dashboard_server + + class FakeServer: + should_exit = False + + server = FakeServer() + observed_stop = [] + pending = threading.Thread( + target=lambda: observed_stop.append(server.should_exit), + name="pending-dashboard", + ) + monkeypatch.setattr(state, "dashboard_server", server) + monkeypatch.setattr(state, "dashboard_thread", pending) + + assert stop_dashboard_server() is False + + assert server.should_exit is True + assert state.dashboard_server is server + assert state.dashboard_thread is pending + + # If start() wins after teardown, the stop signal is already visible and + # the pending server can exit instead of becoming a leaked owner resource. + pending.start() + pending.join(timeout=1.0) + assert observed_stop == [True] + assert stop_dashboard_server() is True + assert state.dashboard_server is None + assert state.dashboard_thread is None + + +def test_dashboard_start_failure_clears_published_state(monkeypatch): + """A failed dashboard thread start must not poison later owner cleanup.""" + import MCP_Server.dashboard.server as dashboard + import MCP_Server.state as state + + monkeypatch.setattr(state, "dashboard_server", None) + monkeypatch.setattr(state, "dashboard_thread", None) + + def fail_start(_thread): + """Simulate a runtime failure before a dashboard thread starts.""" + raise RuntimeError("cannot start thread") + + monkeypatch.setattr( + dashboard.threading.Thread, + "start", + fail_start, + ) + + with pytest.raises(RuntimeError, match="cannot start thread"): + dashboard.start_dashboard_server() + + assert state.dashboard_server is None + assert state.dashboard_thread is None + assert dashboard.stop_dashboard_server() is True + + +def test_dashboard_shutdown_retains_state_after_join_timeout(monkeypatch): + """Dashboard state should remain published while its thread is alive.""" + import MCP_Server.state as state + from MCP_Server.dashboard.server import stop_dashboard_server + + class FakeServer: + should_exit = False + + class FakeThread: + ident = 123 + name = "slow-dashboard" + + def __init__(self): + """Start as a simulated live dashboard thread.""" + self.alive = True + self.joined = False + + def join(self, timeout): + """Record the bounded join without completing the thread.""" + assert timeout == 3.0 + self.joined = True + + def is_alive(self): + """Return the test-controlled liveness state.""" + return self.alive + + server = FakeServer() + thread = FakeThread() + monkeypatch.setattr(state, "dashboard_server", server) + monkeypatch.setattr(state, "dashboard_thread", thread) + + assert stop_dashboard_server() is False + assert thread.joined is True + assert state.dashboard_server is server + assert state.dashboard_thread is thread + + thread.alive = False + assert stop_dashboard_server() is True + assert state.dashboard_server is None + assert state.dashboard_thread is None + + +def test_backend_shutdown_continues_past_unstarted_thread(monkeypatch): + """An unstarted worker must not prevent the remaining owner cleanup.""" + import MCP_Server.server as server_module + import MCP_Server.state as state + + cleanup = [] + stop_event = threading.Event() + pending = threading.Thread( + target=lambda: cleanup.append(("pending", stop_event.is_set())), + name="pending-control-worker", + ) + + class FakeConnection: + def __init__(self, name): + """Name a backend connection double for cleanup ordering.""" + self.name = name + + def disconnect(self): + """Record disconnection during best-effort cleanup.""" + cleanup.append((self.name, True)) + + connected = threading.Event() + connected.set() + monkeypatch.setattr( + server_module, + "stop_dashboard_server", + lambda: cleanup.append(("dashboard", True)), + ) + monkeypatch.setattr(state, "control_stop_event", stop_event) + monkeypatch.setattr(state, "control_background_threads", [pending]) + monkeypatch.setattr(state, "ableton_connection", FakeConnection("ableton")) + monkeypatch.setattr(state, "m4l_connection", FakeConnection("m4l")) + monkeypatch.setattr(state, "ableton_connected_event", connected) + monkeypatch.setattr(state, "m4l_ping_cache", {"result": True, "timestamp": 1.0}) + + assert server_module._stop_control_backend() is False + + assert stop_event.is_set() + assert cleanup == [ + ("dashboard", True), + ("ableton", True), + ("m4l", True), + ] + assert state.control_background_threads == [pending] + assert state.control_stop_event is stop_event + assert state.ableton_connection is None + assert state.m4l_connection is None + assert not state.ableton_connected_event.is_set() + assert state.m4l_ping_cache == {"result": False, "timestamp": 0.0} + + # A delayed start still sees cancellation and does not restore resources. + pending.start() + pending.join(timeout=1.0) + assert cleanup[-1] == ("pending", True) + assert server_module._stop_control_backend() is True + assert state.control_background_threads == [] + assert state.control_stop_event is None + + +def test_backend_start_failure_removes_unstarted_worker(monkeypatch): + """A worker start exception should leave only genuinely started workers.""" + import MCP_Server.server as server_module + import MCP_Server.state as state + + started = threading.Event() + + def cooperative_worker(stop_event): + """Keep the first worker alive until rollback signals cancellation.""" + started.set() + stop_event.wait(timeout=1.0) + + original_start = threading.Thread.start + + def fail_browser_start(thread): + """Start M4L normally, then fail the second owner-only worker.""" + if thread.name == "browser-cache-warmup": + raise RuntimeError("worker thread unavailable") + return original_start(thread) + + monkeypatch.setattr(server_module, "get_ableton_connection", lambda: None) + monkeypatch.setattr(server_module, "start_dashboard_server", lambda: None) + monkeypatch.setattr(server_module, "stop_dashboard_server", lambda: True) + monkeypatch.setattr(server_module, "_m4l_auto_connect", cooperative_worker) + monkeypatch.setattr(server_module, "_browser_cache_warmup", cooperative_worker) + monkeypatch.setattr(server_module.threading.Thread, "start", fail_browser_start) + monkeypatch.setattr(state, "ableton_connection", None) + monkeypatch.setattr(state, "m4l_connection", None) + + with pytest.raises(RuntimeError, match="worker thread unavailable"): + server_module._start_control_backend() + + assert started.wait(timeout=1.0) + assert [thread.name for thread in state.control_background_threads] == [ + "m4l-auto-connect" + ] + assert server_module._stop_control_backend() is True + assert state.control_background_threads == [] + assert state.control_stop_event is None + + +def test_backend_shutdown_closes_m4l_published_during_join(monkeypatch): + """Teardown must close an M4L connection published after cancellation.""" + import MCP_Server.server as server_module + import MCP_Server.state as state + + second_check_sampled = threading.Event() + + class CoordinatedStopEvent: + """Pause the worker after sampling a clear cancellation flag.""" + + def __init__(self): + self._event = threading.Event() + self._checks = 0 + + def is_set(self): + self._checks += 1 + sampled = self._event.is_set() + if self._checks == 2: + second_check_sampled.set() + self._event.wait(timeout=1.0) + return sampled + + def set(self): + self._event.set() + + def wait(self, timeout=None): + return self._event.wait(timeout) + + class FakeM4LConnection: + """Represent a bound M4L socket without touching the live ports.""" + + instance = None + + def __init__(self): + self.disconnected = False + FakeM4LConnection.instance = self + + def connect(self): + return True + + def disconnect(self): + self.disconnected = True + + @staticmethod + def _build_osc_message(_address, _args): + return b"ping" + + stop_event = CoordinatedStopEvent() + worker = threading.Thread( + target=server_module._m4l_auto_connect, + args=(stop_event,), + name="late-m4l-publisher", + ) + monkeypatch.setattr(server_module, "M4LConnection", FakeM4LConnection) + monkeypatch.setattr(server_module, "stop_dashboard_server", lambda: True) + monkeypatch.setattr(state, "control_stop_event", stop_event) + monkeypatch.setattr(state, "control_background_threads", [worker]) + monkeypatch.setattr(state, "ableton_connection", None) + monkeypatch.setattr(state, "m4l_connection", None) + monkeypatch.setattr(state, "ableton_connected_event", threading.Event()) + + worker.start() + assert second_check_sampled.wait(timeout=1.0) + + assert server_module._stop_control_backend() is True + assert not worker.is_alive() + assert FakeM4LConnection.instance is not None + assert FakeM4LConnection.instance.disconnected is True + assert state.m4l_connection is None + assert state.control_background_threads == [] + assert state.control_stop_event is None + + +def test_backend_shutdown_retains_live_worker_after_join_timeout(monkeypatch): + """Backend state should retain a worker that survives its join timeout.""" + import MCP_Server.server as server_module + import MCP_Server.state as state + + class FakeThread: + ident = 456 + name = "slow-cache-worker" + + def __init__(self): + """Start as a simulated live cache worker.""" + self.alive = True + + def join(self, timeout): + """Accept the bounded join while remaining alive.""" + assert timeout == 3.0 + + def is_alive(self): + """Return the test-controlled liveness state.""" + return self.alive + + stop_event = threading.Event() + worker = FakeThread() + monkeypatch.setattr(server_module, "stop_dashboard_server", lambda: True) + monkeypatch.setattr(state, "control_stop_event", stop_event) + monkeypatch.setattr(state, "control_background_threads", [worker]) + monkeypatch.setattr(state, "ableton_connection", None) + monkeypatch.setattr(state, "m4l_connection", None) + monkeypatch.setattr(state, "ableton_connected_event", threading.Event()) + monkeypatch.setattr(state, "m4l_ping_cache", {"result": True, "timestamp": 1.0}) + + assert server_module._stop_control_backend() is False + assert stop_event.is_set() + assert state.control_background_threads == [worker] + assert state.control_stop_event is stop_event + + worker.alive = False + assert server_module._stop_control_backend() is True + assert state.control_background_threads == [] + assert state.control_stop_event is None + + +@pytest.mark.asyncio +async def test_two_stdio_clients_keep_tools_while_control_is_owned(unused_tcp_port_factory): + """Regression proof: a lock collision must not abort either MCP handshake.""" + from mcp import ClientSession, StdioServerParameters + from mcp.client.stdio import stdio_client + + lock_port = unused_tcp_port_factory() + dashboard_port = unused_tcp_port_factory() + owner = _configured_manager(lock_port) + try: + assert owner.ensure_control(client_name="integration-owner").acquired is True + + root = Path(__file__).resolve().parents[1] + env = dict(os.environ) + env["ABLETON_BRIDGE_LOCK_PORT"] = str(lock_port) + env["ABLETON_BRIDGE_DASHBOARD_PORT"] = str(dashboard_port) + params = StdioServerParameters( + command=sys.executable, + args=["-m", "MCP_Server.server"], + cwd=root, + env=env, + ) + with tempfile.TemporaryFile(mode="w+") as errors: + async with stdio_client(params, errlog=errors) as (read_a, write_a): + async with ClientSession(read_a, write_a) as client_a: + await client_a.initialize() + async with stdio_client(params, errlog=errors) as (read_b, write_b): + async with ClientSession(read_b, write_b) as client_b: + await client_b.initialize() + + tools_a = await client_a.list_tools() + tools_b = await client_b.list_tools() + names_a = {tool.name for tool in tools_a.tools} + names_b = {tool.name for tool in tools_b.tools} + + assert names_a == names_b + assert len(names_a) > 300 + assert "release_ableton_control" in names_a + + status_result = await client_b.call_tool("get_server_capabilities", {}) + status = json.loads(status_result.content[0].text) + assert status["control_role"] == "standby" + assert status["control_availability"] == "owned" + assert status["owner"]["client_name"] == "integration-owner" + assert status["ableton_connection_state"] == "owned_elsewhere" + assert status["ableton_connected"] is None + assert status["m4l_connection_state"] == "owned_elsewhere" + assert status["m4l_connected"] is None + assert status["m4l_sockets_ready"] is None + + release_result = await client_b.call_tool("release_ableton_control", {}) + release = json.loads(release_result.content[0].text) + assert release["status"] == "ok" + assert release["data"]["released"] is False + assert owner.status()["control_role"] == "owner" + finally: + owner.shutdown() + + +@pytest.mark.asyncio +async def test_stdio_status_reports_unstarted_backend_without_claiming( + unused_tcp_port_factory, +): + """A free standby should report null connections and leave control free.""" + from mcp import ClientSession, StdioServerParameters + from mcp.client.stdio import stdio_client + + lock_port = unused_tcp_port_factory() + dashboard_port = unused_tcp_port_factory() + root = Path(__file__).resolve().parents[1] + env = dict(os.environ) + env["ABLETON_BRIDGE_LOCK_PORT"] = str(lock_port) + env["ABLETON_BRIDGE_DASHBOARD_PORT"] = str(dashboard_port) + params = StdioServerParameters( + command=sys.executable, + args=["-m", "MCP_Server.server"], + cwd=root, + env=env, + ) + + with tempfile.TemporaryFile(mode="w+") as errors: + async with stdio_client(params, errlog=errors) as (read, write): + async with ClientSession(read, write) as client: + await client.initialize() + tools = await client.list_tools() + status_result = await client.call_tool("get_server_capabilities", {}) + status = json.loads(status_result.content[0].text) + + assert len(tools.tools) > 300 + assert status["control_role"] == "standby" + assert status["control_availability"] == "available" + assert status["ableton_connection_state"] == "not_started" + assert status["ableton_connected"] is None + assert status["m4l_connection_state"] == "not_started" + assert status["m4l_connected"] is None + assert status["m4l_sockets_ready"] is None + + contender = _configured_manager(lock_port) + try: + assert contender.ensure_control().acquired is True + finally: + contender.shutdown() diff --git a/tests/test_status.py b/tests/test_status.py new file mode 100644 index 0000000..26d3d0c --- /dev/null +++ b/tests/test_status.py @@ -0,0 +1,416 @@ +"""Ownership-aware connection status tests.""" + +import json +import threading +from unittest.mock import MagicMock + +import pytest + +import MCP_Server.state as state +import MCP_Server.status as connection_status +from MCP_Server.ownership import OwnershipManager + + +def _control(role: str, availability: str) -> dict: + """Build the ownership fields needed by the connection status helper.""" + return { + "control_role": role, + "control_availability": availability, + "owner": None, + "active_operations": 0, + } + + +@pytest.mark.parametrize( + ("availability", "expected_state"), + [ + ("available", "not_started"), + ("owned", "owned_elsewhere"), + ("occupied_unknown", "unknown"), + ], +) +def test_standby_connections_are_unknown_without_local_probes( + monkeypatch, + availability, + expected_state, +): + """Standby status should explain nulls without inspecting backend state.""" + monkeypatch.setattr( + connection_status, + "_ableton_socket_connected", + lambda: pytest.fail("standby inspected the local Ableton socket"), + ) + monkeypatch.setattr( + connection_status, + "get_m4l_status", + lambda: pytest.fail("standby pinged the local M4L bridge"), + ) + + result = connection_status.build_connection_status( + _control("standby", availability) + ) + + assert result == { + "ableton_connection_state": expected_state, + "ableton_connected": None, + "m4l_connection_state": expected_state, + "m4l_connected": None, + "m4l_sockets_ready": None, + } + + +def test_owner_reports_connected_backends(monkeypatch): + """An owner should report verified local Ableton and M4L connections.""" + ableton = MagicMock() + ableton.is_connected.return_value = True + m4l = MagicMock(_connected=True) + m4l.ping.return_value = True + monkeypatch.setattr(state, "ableton_connection", ableton) + monkeypatch.setattr(state, "m4l_connection", m4l) + monkeypatch.setattr(state, "m4l_ping_cache", {"result": False, "timestamp": 0.0}) + + result = connection_status.build_connection_status( + _control("owner", "owned") + ) + + assert result == { + "ableton_connection_state": "connected", + "ableton_connected": True, + "m4l_connection_state": "connected", + "m4l_connected": True, + "m4l_sockets_ready": True, + } + m4l.ping.assert_called_once_with() + + +def test_owner_reports_dead_ableton_socket_and_missing_m4l(monkeypatch): + """An owner should use false only after local backend inspection.""" + ableton = MagicMock() + ableton.is_connected.return_value = False + monkeypatch.setattr(state, "ableton_connection", ableton) + monkeypatch.setattr(state, "m4l_connection", None) + + result = connection_status.build_connection_status( + _control("owner", "owned") + ) + + assert result == { + "ableton_connection_state": "disconnected", + "ableton_connected": False, + "m4l_connection_state": "disconnected", + "m4l_connected": False, + "m4l_sockets_ready": False, + } + + +def test_owner_distinguishes_m4l_sockets_from_bridge_response(monkeypatch): + """Bound M4L sockets without a ping response should remain distinguishable.""" + ableton = MagicMock() + ableton.is_connected.return_value = True + m4l = MagicMock(_connected=True) + m4l.ping.return_value = False + monkeypatch.setattr(state, "ableton_connection", ableton) + monkeypatch.setattr(state, "m4l_connection", m4l) + monkeypatch.setattr(state, "m4l_ping_cache", {"result": False, "timestamp": 0.0}) + + result = connection_status.build_connection_status( + _control("owner", "owned") + ) + + assert result["ableton_connection_state"] == "connected" + assert result["m4l_connection_state"] == "sockets_ready" + assert result["m4l_connected"] is False + assert result["m4l_sockets_ready"] is True + + +def test_m4l_status_rechecks_sockets_after_failed_ping_reconnect(monkeypatch): + """A ping that loses its sockets must not preserve pre-ping readiness.""" + m4l = MagicMock(_connected=True) + + def lose_sockets(): + """Model send_command exhausting its reconnect attempt.""" + m4l._connected = False + return False + + m4l.ping.side_effect = lose_sockets + monkeypatch.setattr(state, "m4l_connection", m4l) + monkeypatch.setattr(state, "m4l_ping_cache", {"result": False, "timestamp": 0.0}) + monkeypatch.setattr(state, "m4l_status_snapshot", (True, False)) + monkeypatch.setattr(connection_status.ownership, "is_configured", lambda: False) + + assert connection_status.get_m4l_status() == (False, False) + assert state.m4l_status_snapshot == (False, False) + + +def test_live_m4l_status_probe_prevents_concurrent_release( + monkeypatch, + unused_tcp_port, +): + """Release must not disconnect M4L while an owner status ping is active.""" + manager = OwnershipManager(unused_tcp_port) + manager.configure_backend(lambda: None, lambda: True) + started = threading.Event() + finish = threading.Event() + results = [] + m4l = MagicMock(_connected=True) + + def blocking_ping(): + """Hold the status probe until release has observed the operation.""" + started.set() + finish.wait(timeout=1.0) + return True + + m4l.ping.side_effect = blocking_ping + monkeypatch.setattr(state, "m4l_connection", m4l) + monkeypatch.setattr(state, "m4l_ping_cache", {"result": False, "timestamp": 0.0}) + monkeypatch.setattr(connection_status.ownership, "is_configured", manager.is_configured) + monkeypatch.setattr(connection_status.ownership, "begin_operation", manager.begin_operation) + monkeypatch.setattr(connection_status.ownership, "end_operation", manager.end_operation) + + worker = threading.Thread( + target=lambda: results.append(connection_status.get_m4l_status()), + ) + try: + assert manager.ensure_control().acquired is True + worker.start() + assert started.wait(timeout=1.0) + + release = manager.release() + assert release.released is False + assert release.control["active_operations"] == 1 + + finish.set() + worker.join(timeout=1.0) + assert not worker.is_alive() + assert results == [(True, True)] + assert manager.release().released is True + finally: + finish.set() + if worker.ident is not None: + worker.join(timeout=1.0) + manager.shutdown() + + +def test_m4l_status_does_not_ping_after_release_starts(monkeypatch): + """A failed operation lease should fall back to the last cached result.""" + m4l = MagicMock(_connected=True) + monkeypatch.setattr(state, "m4l_connection", m4l) + monkeypatch.setattr(state, "m4l_ping_cache", {"result": True, "timestamp": 0.0}) + monkeypatch.setattr(state, "m4l_status_snapshot", (True, True)) + monkeypatch.setattr(connection_status.ownership, "is_configured", lambda: True) + monkeypatch.setattr(connection_status.ownership, "begin_operation", lambda: False) + monkeypatch.setattr( + connection_status.ownership, + "end_operation", + lambda: pytest.fail("unacquired operation was ended"), + ) + + assert connection_status.get_m4l_status() == (True, True) + m4l.ping.assert_not_called() + + +def test_status_ping_serializes_m4l_connection_replacement(monkeypatch): + """A status ping must finish before a normal tool replaces its connection.""" + import MCP_Server.connections.m4l as m4l_module + + ping_started = threading.Event() + finish_ping = threading.Event() + replacement_done = threading.Event() + status_results = [] + replacement_results = [] + calls = [] + old = MagicMock(_connected=True) + new = MagicMock(_connected=True) + new.connect.return_value = True + new.ping.return_value = True + + def old_ping(): + """Block only the status ping; the replacement verification then fails.""" + calls.append("ping") + if len(calls) == 1: + ping_started.set() + finish_ping.wait(timeout=1.0) + return False + + def replace_connection(): + """Run the normal connection replacement path in a competing thread.""" + try: + replacement_results.append(m4l_module.get_m4l_connection()) + finally: + replacement_done.set() + + old.ping.side_effect = old_ping + monkeypatch.setattr(state, "m4l_connection", old) + monkeypatch.setattr(state, "m4l_ping_cache", {"result": False, "timestamp": 0.0}) + monkeypatch.setattr(connection_status.ownership, "is_configured", lambda: False) + monkeypatch.setattr(m4l_module, "M4LConnection", lambda: new) + + status_worker = threading.Thread( + target=lambda: status_results.append(connection_status.get_m4l_status()), + ) + replacement_worker = threading.Thread(target=replace_connection) + try: + status_worker.start() + assert ping_started.wait(timeout=1.0) + + replacement_worker.start() + assert replacement_done.wait(timeout=0.05) is False + assert state.m4l_connection is old + + finish_ping.set() + status_worker.join(timeout=1.0) + replacement_worker.join(timeout=1.0) + + assert not status_worker.is_alive() + assert not replacement_worker.is_alive() + assert status_results == [(True, False)] + assert replacement_results == [new] + assert state.m4l_connection is new + assert old.disconnect.call_count == 1 + assert state.m4l_ping_cache["result"] is True + assert state.m4l_ping_cache["timestamp"] > 0.0 + finally: + finish_ping.set() + if status_worker.ident is not None: + status_worker.join(timeout=1.0) + if replacement_worker.ident is not None: + replacement_worker.join(timeout=1.0) + + +def test_m4l_status_uses_cache_while_connection_transaction_is_busy(monkeypatch): + """Status should remain responsive instead of waiting for M4L replacement.""" + lock_held = threading.Event() + release_lock = threading.Event() + status_done = threading.Event() + results = [] + m4l = MagicMock(_connected=True) + monkeypatch.setattr(state, "m4l_connection", m4l) + monkeypatch.setattr(state, "m4l_ping_cache", {"result": True, "timestamp": 0.0}) + monkeypatch.setattr(state, "m4l_status_snapshot", (True, True)) + monkeypatch.setattr(connection_status.ownership, "is_configured", lambda: False) + + def hold_connection_transaction(): + """Keep the state lock busy until status has taken its fallback path.""" + with state.m4l_connection_lock: + lock_held.set() + release_lock.wait(timeout=1.0) + + def read_status(): + """Record completion without ever waiting for the held state lock.""" + try: + results.append(connection_status.get_m4l_status()) + finally: + status_done.set() + + holder = threading.Thread(target=hold_connection_transaction) + reader = threading.Thread(target=read_status) + try: + holder.start() + assert lock_held.wait(timeout=1.0) + reader.start() + + assert status_done.wait(timeout=0.2) + assert results == [(True, True)] + m4l.ping.assert_not_called() + finally: + release_lock.set() + if holder.ident is not None: + holder.join(timeout=1.0) + if reader.ident is not None: + reader.join(timeout=1.0) + + +def test_m4l_status_fallback_never_combines_connection_generations(monkeypatch): + """A busy transaction exposes one immutable snapshot, never hybrid state.""" + lock_held = threading.Event() + release_lock = threading.Event() + status_done = threading.Event() + results = [] + new_connection = MagicMock(_connected=True) + monkeypatch.setattr(state, "m4l_connection", None) + monkeypatch.setattr(state, "m4l_ping_cache", {"result": True, "timestamp": 1.0}) + monkeypatch.setattr(state, "m4l_status_snapshot", (False, False)) + monkeypatch.setattr(connection_status.ownership, "is_configured", lambda: False) + + def publish_partial_generation(): + """Pause on deliberately inconsistent raw fields inside the state lock.""" + with state.m4l_connection_lock: + state.m4l_connection = new_connection + lock_held.set() + release_lock.wait(timeout=1.0) + + def read_status(): + """Read the last atomic snapshot without inspecting partial fields.""" + try: + results.append(connection_status.get_m4l_status()) + finally: + status_done.set() + + publisher = threading.Thread(target=publish_partial_generation) + reader = threading.Thread(target=read_status) + try: + publisher.start() + assert lock_held.wait(timeout=1.0) + reader.start() + + assert status_done.wait(timeout=0.2) + assert results == [(False, False)] + new_connection.ping.assert_not_called() + finally: + release_lock.set() + if publisher.ident is not None: + publisher.join(timeout=1.0) + if reader.ident is not None: + reader.join(timeout=1.0) + + +@pytest.mark.asyncio +async def test_capabilities_tool_and_resource_share_standby_contract(monkeypatch): + """Both public capability surfaces should expose identical tri-state fields.""" + from mcp.server.fastmcp import FastMCP + import MCP_Server.server as server_module + import MCP_Server.tools.session as session_tools + + control = _control("standby", "available") + monkeypatch.setattr(session_tools.ownership, "get_status", lambda: control) + monkeypatch.setattr(server_module.ownership, "get_status", lambda: control) + + mcp = FastMCP("status-test") + session_tools.register_tools(mcp) + tool = mcp._tool_manager._tools["get_server_capabilities"] + tool_result = json.loads(await tool.fn(MagicMock())) + resource_result = json.loads(server_module.resource_capabilities()) + + expected = { + "ableton_connection_state": "not_started", + "ableton_connected": None, + "m4l_connection_state": "not_started", + "m4l_connected": None, + "m4l_sockets_ready": None, + } + for key, value in expected.items(): + assert tool_result[key] == value + assert resource_result[key] == value + assert tool_result["features"]["m4l_bridge"] is None + + +def test_dashboard_status_keeps_owner_booleans(monkeypatch): + """The owner-only dashboard should retain booleans and add explicit states.""" + import MCP_Server.dashboard.server as dashboard + + ableton = MagicMock() + ableton.is_connected.return_value = True + m4l = MagicMock(_connected=True) + m4l.ping.return_value = True + monkeypatch.setattr(dashboard.ownership, "get_status", lambda: _control("owner", "owned")) + monkeypatch.setattr(state, "ableton_connection", ableton) + monkeypatch.setattr(state, "m4l_connection", m4l) + monkeypatch.setattr(state, "m4l_ping_cache", {"result": False, "timestamp": 0.0}) + + result = dashboard.build_status_json() + + assert result["ableton_connection_state"] == "connected" + assert result["ableton_connected"] is True + assert result["m4l_connection_state"] == "connected" + assert result["m4l_connected"] is True + assert result["m4l_sockets_ready"] is True diff --git a/tests/test_tool_handler.py b/tests/test_tool_handler.py index 0c0db5c..b172276 100644 --- a/tests/test_tool_handler.py +++ b/tests/test_tool_handler.py @@ -1,14 +1,19 @@ import asyncio import json +import threading import pytest +import MCP_Server.tools._base as tool_base +from MCP_Server.ownership import ClaimResult, OwnershipManager from MCP_Server.tools._base import _tool_handler, tool_success, tool_error, _m4l_result class TestToolHandler: @pytest.mark.asyncio async def test_basic_success(self): + """Plain tool results should use the shared success envelope.""" @_tool_handler("test operation") def my_tool(): + """Return a successful plain-string tool result.""" return "success" result = await my_tool() @@ -18,8 +23,10 @@ def my_tool(): @pytest.mark.asyncio async def test_value_error_caught(self): + """Value errors should become structured invalid-input responses.""" @_tool_handler("test operation") def my_tool(): + """Raise the input-validation failure under test.""" raise ValueError("bad input") result = await my_tool() @@ -30,8 +37,10 @@ def my_tool(): @pytest.mark.asyncio async def test_connection_error_caught(self): + """Connection errors should become structured bridge responses.""" @_tool_handler("test operation") def my_tool(): + """Raise the connection failure under test.""" raise ConnectionError("no connection") result = await my_tool() @@ -41,8 +50,10 @@ def my_tool(): @pytest.mark.asyncio async def test_generic_exception_caught(self): + """Unexpected exceptions should include the operation context.""" @_tool_handler("doing stuff") def my_tool(): + """Raise the unexpected tool failure under test.""" raise RuntimeError("something broke") result = await my_tool() @@ -52,8 +63,10 @@ def my_tool(): @pytest.mark.asyncio async def test_with_args(self): + """The wrapper should preserve positional arguments.""" @_tool_handler("test") def my_tool(a, b): + """Combine the positional arguments for assertion.""" return f"{a}+{b}" result = await my_tool(1, 2) @@ -66,6 +79,7 @@ async def test_json_passthrough(self): """Responses already in JSON format should pass through unwrapped.""" @_tool_handler("test") def my_tool(): + """Return an already structured JSON response.""" return json.dumps({"tracks": [1, 2, 3]}) result = await my_tool() @@ -73,20 +87,495 @@ def my_tool(): assert parsed["tracks"] == [1, 2, 3] assert "status" not in parsed + @pytest.mark.asyncio + async def test_control_exempt_tool_does_not_claim(self, monkeypatch): + """A control-exempt status tool should never trigger auto-claiming.""" + monkeypatch.setattr(tool_base.ownership, "is_configured", lambda: True) + + def unexpected_claim(**kwargs): + """Fail if the exempt tool incorrectly attempts ownership.""" + raise AssertionError("status tool attempted to claim control") + + monkeypatch.setattr(tool_base.ownership, "ensure_control", unexpected_claim) + + @_tool_handler("checking status", requires_control=False) + def status_tool(): + """Return a standby result without requiring control.""" + return "standby" + + result = json.loads(await status_tool()) + assert result["message"] == "standby" + + @pytest.mark.asyncio + async def test_unbound_client_context_does_not_fail_tool(self, monkeypatch): + """Unavailable best-effort client metadata must not block control tools.""" + claims = [] + ended = [] + semaphore = asyncio.Semaphore(1) + + class UnboundContext: + """Model a FastMCP Context outside an active request scope.""" + + @property + def session(self): + """Raise the error FastMCP uses for unavailable session state.""" + raise RuntimeError("request context is unavailable") + + monkeypatch.setattr(tool_base, "_ableton_semaphore", semaphore) + monkeypatch.setattr(tool_base.ownership, "is_configured", lambda: True) + monkeypatch.setattr( + tool_base.ownership, + "ensure_control", + lambda **kwargs: ( + claims.append(kwargs) + or ClaimResult(True, {"control_role": "owner"}) + ), + ) + monkeypatch.setattr(tool_base.ownership, "begin_operation", lambda: True) + monkeypatch.setattr( + tool_base.ownership, + "end_operation", + lambda: ended.append(True), + ) + + @_tool_handler("using optional client metadata") + def guarded_tool(_ctx): + """Return normally when client metadata is unavailable.""" + return "success" + + result = json.loads(await guarded_tool(UnboundContext())) + + assert result["status"] == "ok" + assert claims == [{"client_name": None}] + assert ended == [True] + + @pytest.mark.asyncio + async def test_control_exempt_tool_bypasses_ableton_semaphore(self, monkeypatch): + """Status and release tools should remain callable during owner work.""" + semaphore = asyncio.Semaphore(1) + await semaphore.acquire() + monkeypatch.setattr(tool_base, "_ableton_semaphore", semaphore) + + @_tool_handler("checking status", requires_control=False) + def status_tool(): + """Return status while the controlled-tool semaphore is occupied.""" + return "standby" + + try: + result = json.loads(await asyncio.wait_for(status_tool(), timeout=0.2)) + assert result["message"] == "standby" + finally: + semaphore.release() + + @pytest.mark.asyncio + async def test_ownership_claim_is_time_bounded( + self, + monkeypatch, + unused_tcp_port, + ): + """A timed-out new claim should roll back without running the tool.""" + started = threading.Event() + finish = threading.Event() + executions = [] + semaphore = asyncio.Semaphore(1) + manager = OwnershipManager(unused_tcp_port) + monkeypatch.setattr(tool_base, "_ableton_semaphore", semaphore) + + def slow_start(): + """Hold a real ownership startup beyond the response deadline.""" + started.set() + finish.wait(timeout=1.0) + + manager.configure_backend(slow_start, lambda: None) + monkeypatch.setattr(tool_base.ownership, "is_configured", manager.is_configured) + monkeypatch.setattr(tool_base.ownership, "ensure_control", manager.ensure_control) + monkeypatch.setattr( + tool_base.ownership, + "release_if_current_claim", + manager.release_if_current_claim, + ) + monkeypatch.setattr(tool_base.ownership, "begin_operation", manager.begin_operation) + monkeypatch.setattr(tool_base.ownership, "end_operation", manager.end_operation) + monkeypatch.setattr(tool_base, "_TOOL_TIMEOUT_SECONDS", 0.02) + + @_tool_handler("claiming control") + def guarded_tool(): + """Record any execution after the ownership claim returns.""" + executions.append("ran") + return "unexpected" + + try: + result = json.loads(await guarded_tool()) + assert started.is_set() + assert "timed out" in result["message"] + assert semaphore.locked() + + finish.set() + for _ in range(50): + if ( + not semaphore.locked() + and manager.status()["control_role"] == "standby" + ): + break + await asyncio.sleep(0.01) + + assert not semaphore.locked() + assert executions == [] + assert manager.status()["control_role"] == "standby" + finally: + finish.set() + manager.shutdown() + + @pytest.mark.asyncio + async def test_waiting_for_control_semaphore_uses_response_deadline( + self, + monkeypatch, + ): + """A queued control call should time out without starting background work.""" + executions = [] + semaphore = asyncio.Semaphore(1) + await semaphore.acquire() + monkeypatch.setattr(tool_base, "_ableton_semaphore", semaphore) + monkeypatch.setattr(tool_base, "_TOOL_TIMEOUT_SECONDS", 0.02) + + @_tool_handler("waiting for control") + def guarded_tool(): + """Record any execution after the queueing deadline expires.""" + executions.append("ran") + return "unexpected" + + try: + result = json.loads(await guarded_tool()) + assert "timed out" in result["message"] + assert executions == [] + assert semaphore.locked() + finally: + semaphore.release() + + @pytest.mark.asyncio + async def test_cancelled_ownership_claim_does_not_run_tool_later(self, monkeypatch): + """Cancellation during a claim must abandon work that has not started.""" + started = threading.Event() + finish = threading.Event() + executions = [] + semaphore = asyncio.Semaphore(1) + monkeypatch.setattr(tool_base, "_ableton_semaphore", semaphore) + + def slow_claim(**_kwargs): + """Hold ownership startup until after the caller is cancelled.""" + started.set() + finish.wait(timeout=1.0) + return ClaimResult( + acquired=True, + control={"control_role": "owner"}, + ) + + monkeypatch.setattr(tool_base.ownership, "is_configured", lambda: True) + monkeypatch.setattr(tool_base.ownership, "ensure_control", slow_claim) + + def unexpected_release(_claim_token): + """Never release ownership that predated the abandoned invocation.""" + raise AssertionError("pre-existing ownership was released") + + monkeypatch.setattr( + tool_base.ownership, + "release_if_current_claim", + unexpected_release, + ) + + @_tool_handler("claiming control") + def guarded_tool(): + """Record any execution after the ownership claim returns.""" + executions.append("ran") + return "unexpected" + + call = asyncio.create_task(guarded_tool()) + try: + for _ in range(50): + if started.is_set(): + break + await asyncio.sleep(0.01) + assert started.is_set() + + call.cancel() + with pytest.raises(asyncio.CancelledError): + await call + assert semaphore.locked() + finally: + finish.set() + for _ in range(50): + if not semaphore.locked(): + break + await asyncio.sleep(0.01) + + assert not semaphore.locked() + assert executions == [] + + @pytest.mark.asyncio + async def test_timeout_during_operation_registration_does_not_run_tool( + self, + monkeypatch, + ): + """Timeout while registering an operation must abandon the tool body.""" + registration_started = threading.Event() + finish_registration = threading.Event() + executions = [] + ended = [] + semaphore = asyncio.Semaphore(1) + monkeypatch.setattr(tool_base, "_ableton_semaphore", semaphore) + monkeypatch.setattr(tool_base, "_TOOL_TIMEOUT_SECONDS", 0.02) + monkeypatch.setattr(tool_base.ownership, "is_configured", lambda: True) + monkeypatch.setattr( + tool_base.ownership, + "ensure_control", + lambda **_kwargs: ClaimResult( + acquired=True, + control={"control_role": "owner"}, + ), + ) + + def slow_begin_operation(): + """Hold operation registration beyond the client timeout.""" + registration_started.set() + finish_registration.wait(timeout=1.0) + return True + + monkeypatch.setattr( + tool_base.ownership, + "begin_operation", + slow_begin_operation, + ) + monkeypatch.setattr( + tool_base.ownership, + "end_operation", + lambda: ended.append(True), + ) + + @_tool_handler("registering operation") + def guarded_tool(): + """Record any execution after operation registration completes.""" + executions.append("ran") + return "unexpected" + + try: + result = json.loads(await guarded_tool()) + assert registration_started.is_set() + assert "timed out" in result["message"] + assert semaphore.locked() + finally: + finish_registration.set() + for _ in range(50): + if not semaphore.locked(): + break + await asyncio.sleep(0.01) + + assert not semaphore.locked() + assert executions == [] + assert ended == [True] + + @pytest.mark.asyncio + async def test_control_release_status_probe_runs_off_event_loop(self, monkeypatch): + """Released-control status probing should not block the event loop.""" + event_loop_thread = threading.get_ident() + status_threads = [] + monkeypatch.setattr(tool_base.ownership, "is_configured", lambda: True) + monkeypatch.setattr( + tool_base.ownership, + "ensure_control", + lambda **_kwargs: ClaimResult( + acquired=True, + control={"control_role": "owner"}, + ), + ) + monkeypatch.setattr(tool_base.ownership, "begin_operation", lambda: False) + monkeypatch.setattr( + tool_base.ownership, + "get_status", + lambda: ( + status_threads.append(threading.get_ident()) + or {"control_role": "standby"} + ), + ) + + @_tool_handler("changing Live") + def guarded_tool(): + """Fail if owner-only work runs after control is released.""" + raise AssertionError("released control executed owner-only work") + + result = json.loads(await guarded_tool()) + assert result["status"] == "error" + assert status_threads and status_threads[0] != event_loop_thread + + @pytest.mark.asyncio + async def test_standby_tool_returns_owner_details( + self, + monkeypatch, + unused_tcp_port, + ): + """Standby errors should identify the process that currently owns control.""" + owner = OwnershipManager(unused_tcp_port) + standby = OwnershipManager(unused_tcp_port) + owner.configure_backend(lambda: None, lambda: None) + standby.configure_backend(lambda: None, lambda: None) + assert owner.ensure_control(client_name="owner-client").acquired is True + + monkeypatch.setattr(tool_base.ownership, "is_configured", standby.is_configured) + monkeypatch.setattr(tool_base.ownership, "ensure_control", standby.ensure_control) + + @_tool_handler("changing Live") + def guarded_tool(): + """Fail if a standby process executes owner-only work.""" + raise AssertionError("standby executed owner-only work") + + try: + result = json.loads(await guarded_tool()) + assert result["status"] == "error" + control = result["data"]["control"] + assert control["control_role"] == "standby" + assert control["owner"]["client_name"] == "owner-client" + finally: + owner.shutdown() + standby.shutdown() + + @pytest.mark.asyncio + async def test_timeout_keeps_control_busy_until_worker_finishes( + self, + monkeypatch, + unused_tcp_port, + ): + """A timed-out tool should retain ownership and serialization until exit.""" + manager = OwnershipManager(unused_tcp_port) + manager.configure_backend(lambda: None, lambda: None) + started = threading.Event() + finish = threading.Event() + follower_started = threading.Event() + semaphore = asyncio.Semaphore(1) + + monkeypatch.setattr(tool_base.ownership, "is_configured", manager.is_configured) + monkeypatch.setattr(tool_base.ownership, "ensure_control", manager.ensure_control) + monkeypatch.setattr(tool_base.ownership, "begin_operation", manager.begin_operation) + monkeypatch.setattr(tool_base.ownership, "end_operation", manager.end_operation) + monkeypatch.setattr(tool_base.ownership, "get_status", manager.status) + monkeypatch.setattr(tool_base, "_ableton_semaphore", semaphore) + monkeypatch.setattr(tool_base, "_TOOL_TIMEOUT_SECONDS", 0.02) + + @_tool_handler("waiting") + def slow_tool(): + """Hold one controlled worker beyond the client timeout.""" + started.set() + finish.wait(timeout=1.0) + return "finished" + + @_tool_handler("following") + def follower_tool(): + """Record when a second controlled tool actually begins.""" + follower_started.set() + return "followed" + + try: + result = json.loads(await slow_tool()) + assert "timed out" in result["message"] + assert started.is_set() + assert semaphore.locked() + + busy = manager.release() + assert busy.released is False + assert busy.control["active_operations"] == 1 + + follower = asyncio.create_task(follower_tool()) + await asyncio.sleep(0.03) + assert not follower_started.is_set() + follower_result = json.loads(await follower) + assert "timed out" in follower_result["message"] + + finish.set() + for _ in range(50): + if ( + manager.status()["active_operations"] == 0 + and not semaphore.locked() + ): + break + await asyncio.sleep(0.01) + + assert manager.status()["active_operations"] == 0 + assert json.loads(await follower_tool())["message"] == "followed" + assert follower_started.is_set() + assert manager.release().released is True + finally: + finish.set() + manager.shutdown() + + @pytest.mark.asyncio + async def test_cancelled_tool_keeps_semaphore_until_worker_finishes( + self, + monkeypatch, + ): + """Caller cancellation must not release serialization before worker exit.""" + started = threading.Event() + finish = threading.Event() + semaphore = asyncio.Semaphore(1) + monkeypatch.setattr(tool_base, "_ableton_semaphore", semaphore) + + @_tool_handler("cancellable") + def slow_tool(): + """Block the worker until the cancellation assertion is complete.""" + started.set() + finish.wait(timeout=1.0) + return "finished" + + call = asyncio.create_task(slow_tool()) + try: + for _ in range(50): + if started.is_set(): + break + await asyncio.sleep(0.01) + assert started.is_set() + + call.cancel() + with pytest.raises(asyncio.CancelledError): + await call + assert semaphore.locked() + finally: + finish.set() + for _ in range(50): + if not semaphore.locked(): + break + await asyncio.sleep(0.01) + + assert not semaphore.locked() + + @pytest.mark.asyncio + async def test_timed_out_task_logs_late_exception(self, caplog): + """Late failures should remain observable after the client times out.""" + async def fail_late(): + """Raise after the client-facing task has already detached.""" + raise RuntimeError("late worker failure") + + task = asyncio.create_task(fail_late()) + await asyncio.sleep(0) + + with caplog.at_level("WARNING", logger="AbletonBridge"): + tool_base._consume_background_result(task) + + assert "late worker failure" in caplog.text + class TestToolSuccess: def test_basic(self): + """Success responses should contain status and message fields.""" result = json.loads(tool_success("Done")) assert result["status"] == "ok" assert result["message"] == "Done" def test_with_data(self): + """Success responses should include optional structured data.""" result = json.loads(tool_success("Done", {"count": 5})) assert result["data"]["count"] == 5 class TestToolError: def test_basic(self): + """Error responses should contain status and message fields.""" result = json.loads(tool_error("Failed")) assert result["status"] == "error" assert result["message"] == "Failed" @@ -94,9 +583,11 @@ def test_basic(self): class TestM4lResult: def test_success(self): + """Successful M4L envelopes should return their result payload.""" result = _m4l_result({"status": "success", "result": {"value": 42}}) assert result["value"] == 42 def test_error_raises(self): + """Failed M4L envelopes should raise with the bridge message.""" with pytest.raises(Exception, match="M4L bridge error"): _m4l_result({"status": "error", "message": "device not found"})