[codex] split command adapters and enforce public contract checks

CONTRIBUTING.md

@@ -11,6 +11,24 @@ uv run ableton-cli --help
 uv run ableton-cli --version
 ```
 
+## Public Contract Boundary
+
+The following are treated as public/stable contracts:
+
+- README command surface (`song`, `transport`, `track`, `tracks`, `clip`, `arrangement`, `browser`, `device`, `synth`, `effect`, `batch`, `setup`)
+- JSON output envelope shape (`ok`, `command`, `args`, `result`, `error`)
+- fixed CLI exit codes and documented error codes
+- TCP JSONL protocol request/response envelope and protocol versioning rules
+- `tests/snapshots/public_contract_snapshot.json`
+
+Before merge, regenerate snapshot only when you intentionally change a public contract:
+
+```bash
+uv run python tools/update_public_contract_snapshot.py
+```
+
+Internal modules (for example, `src/ableton_cli/commands/_*.py` and quality harness internals) can be refactored freely as long as public contracts and tests stay green.
+
 ## Commit Hook (Ruff)
 
 Enable repository-managed git hooks to run Ruff on every commit:

remote_script/AbletonCliRemote/command_backend.py

@@ -15,6 +15,9 @@
     REMOTE_SCRIPT_VERSION,
     CommandBackend,
     CommandError,
+    RemoteErrorCode,
+    RemoteErrorReason,
+    details_with_reason,
 )
 from .command_backend_registry import dispatch_command
 
@@ -33,5 +36,8 @@
     "NOTE_VELOCITY_MAX",
     "CommandError",
     "CommandBackend",
+    "RemoteErrorCode",
+    "RemoteErrorReason",
+    "details_with_reason",
     "dispatch_command",
 ]

remote_script/AbletonCliRemote/command_backend_contract.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+from enum import Enum
 from typing import Any, Protocol
 
 PROTOCOL_VERSION = 2
@@ -18,9 +19,29 @@
 NOTE_KEYS = frozenset({"pitch", "start_time", "duration", "velocity", "mute"})
 
 
+class RemoteErrorCode(str, Enum):
+    INVALID_ARGUMENT = "INVALID_ARGUMENT"
+    PROTOCOL_VERSION_MISMATCH = "PROTOCOL_VERSION_MISMATCH"
+    TIMEOUT = "TIMEOUT"
+    REMOTE_BUSY = "REMOTE_BUSY"
+    BATCH_STEP_FAILED = "BATCH_STEP_FAILED"
+    INTERNAL_ERROR = "INTERNAL_ERROR"
+
+
+class RemoteErrorReason(str, Enum):
+    NOT_SUPPORTED_BY_LIVE_API = "not_supported_by_live_api"
+
+
+def details_with_reason(reason: RemoteErrorReason, /, **details: Any) -> dict[str, Any]:
+    return {
+        "reason": reason.value,
+        **details,
+    }
+
+
 @dataclass(slots=True)
 class CommandError(Exception):
-    code: str
+    code: RemoteErrorCode | str
     message: str
     hint: str | None = None
     details: dict[str, Any] | None = None

remote_script/AbletonCliRemote/command_backend_registry.py

@@ -4,7 +4,7 @@
 from collections.abc import Callable
 from typing import Any
 
-from .command_backend_contract import CommandBackend, CommandError
+from .command_backend_contract import CommandBackend, CommandError, RemoteErrorCode
 from .command_backend_handlers_batch import make_execute_batch_handler
 from .command_backend_handlers_browser import BROWSER_HANDLERS
 from .command_backend_handlers_devices import DEVICE_HANDLERS
@@ -48,7 +48,7 @@ def dispatch_command(backend: CommandBackend, name: str, args: dict[str, Any]) -
     handler = _HANDLERS.get(name)
     if handler is None:
         raise CommandError(
-            code="INVALID_ARGUMENT",
+            code=RemoteErrorCode.INVALID_ARGUMENT,
             message=f"Unknown command: {name}",
             hint="Use a supported command name.",
         )

remote_script/AbletonCliRemote/live_backend_parts/base.py

@@ -2,7 +2,14 @@
 
 from typing import Any
 
-from ..command_backend import PROTOCOL_VERSION, REMOTE_SCRIPT_VERSION, CommandError
+from ..command_backend import (
+    PROTOCOL_VERSION,
+    REMOTE_SCRIPT_VERSION,
+    CommandError,
+    RemoteErrorCode,
+    RemoteErrorReason,
+    details_with_reason,
+)
 from ..effect_specs import (
     SUPPORTED_EFFECT_TYPES,
     canonicalize_effect_type,
@@ -18,15 +25,15 @@
 
 
 def _invalid_argument(message: str, hint: str) -> CommandError:
-    return CommandError(code="INVALID_ARGUMENT", message=message, hint=hint)
+    return CommandError(code=RemoteErrorCode.INVALID_ARGUMENT, message=message, hint=hint)
 
 
 def _not_supported_by_live_api(message: str, hint: str) -> CommandError:
     return CommandError(
-        code="INVALID_ARGUMENT",
+        code=RemoteErrorCode.INVALID_ARGUMENT,
         message=message,
         hint=hint,
-        details={"reason": "not_supported_by_live_api"},
+        details=details_with_reason(RemoteErrorReason.NOT_SUPPORTED_BY_LIVE_API),
     )
 
 

remote_script/AbletonCliRemote/server.py

@@ -7,7 +7,7 @@
 from dataclasses import dataclass
 from typing import Any
 
-from .command_backend import PROTOCOL_VERSION
+from .command_backend import PROTOCOL_VERSION, RemoteErrorCode
 
 
 @dataclass(slots=True)
@@ -74,7 +74,7 @@ def handle(self) -> None:
             self._write_response(
                 _error(
                     request_id="unknown",
-                    code="PROTOCOL_VERSION_MISMATCH",
+                    code=RemoteErrorCode.PROTOCOL_VERSION_MISMATCH.value,
                     message="Invalid JSON payload",
                     hint="Send one JSON object per line.",
                 )
@@ -86,14 +86,14 @@ def handle(self) -> None:
         try:
             if payload.get("type") != "command":
                 raise CommandExecutionError(
-                    code="INVALID_ARGUMENT",
+                    code=RemoteErrorCode.INVALID_ARGUMENT.value,
                     message="type must be 'command'",
                     hint="Use the CLI protocol request shape.",
                 )
             protocol_version = payload.get("protocol_version")
             if protocol_version != PROTOCOL_VERSION:
                 raise CommandExecutionError(
-                    code="PROTOCOL_VERSION_MISMATCH",
+                    code=RemoteErrorCode.PROTOCOL_VERSION_MISMATCH.value,
                     message=(
                         "Protocol version mismatch "
                         f"(remote={PROTOCOL_VERSION}, request={protocol_version})"
@@ -108,14 +108,14 @@ def handle(self) -> None:
             args = payload.get("args", {})
             if not isinstance(args, dict):
                 raise CommandExecutionError(
-                    code="INVALID_ARGUMENT",
+                    code=RemoteErrorCode.INVALID_ARGUMENT.value,
                     message="args must be an object",
                     hint="Pass a JSON object for args.",
                 )
             meta = payload.get("meta", {})
             if not isinstance(meta, dict):
                 raise CommandExecutionError(
-                    code="INVALID_ARGUMENT",
+                    code=RemoteErrorCode.INVALID_ARGUMENT.value,
                     message="meta must be an object",
                     hint="Pass a JSON object for meta.",
                 )
@@ -137,7 +137,7 @@ def handle(self) -> None:
         except Exception as exc:  # noqa: BLE001
             response = _error(
                 request_id=request_id,
-                code="INTERNAL_ERROR",
+                code=RemoteErrorCode.INTERNAL_ERROR.value,
                 message=f"Remote internal error: {exc}",
                 hint="Check Remote Script logs.",
             )

src/ableton_cli/capabilities.py

@@ -3,7 +3,7 @@
 import hashlib
 from typing import Any
 
-from .errors import AppError, ExitCode
+from .errors import AppError, ErrorCode, ExitCode
 
 _REQUIRED_REMOTE_COMMANDS = frozenset(
     {
@@ -165,7 +165,7 @@ def parse_supported_commands(ping_payload: dict[str, Any]) -> set[str]:
     raw_commands = ping_payload.get("supported_commands")
     if not isinstance(raw_commands, list):
         raise AppError(
-            error_code="REMOTE_SCRIPT_INCOMPATIBLE",
+            error_code=ErrorCode.REMOTE_SCRIPT_INCOMPATIBLE,
             message="Remote Script ping payload is missing supported_commands",
             hint="Reinstall Remote Script and restart Ableton Live.",
             exit_code=ExitCode.PROTOCOL_MISMATCH,
@@ -175,7 +175,7 @@ def parse_supported_commands(ping_payload: dict[str, Any]) -> set[str]:
     for index, value in enumerate(raw_commands):
         if not isinstance(value, str) or not value.strip():
             raise AppError(
-                error_code="REMOTE_SCRIPT_INCOMPATIBLE",
+                error_code=ErrorCode.REMOTE_SCRIPT_INCOMPATIBLE,
                 message=f"supported_commands[{index}] must be a non-empty string",
                 hint="Reinstall Remote Script and restart Ableton Live.",
                 exit_code=ExitCode.PROTOCOL_MISMATCH,
@@ -184,7 +184,7 @@ def parse_supported_commands(ping_payload: dict[str, Any]) -> set[str]:
 
     if not commands:
         raise AppError(
-            error_code="REMOTE_SCRIPT_INCOMPATIBLE",
+            error_code=ErrorCode.REMOTE_SCRIPT_INCOMPATIBLE,
             message="Remote Script reported no supported commands",
             hint="Reinstall Remote Script and restart Ableton Live.",
             exit_code=ExitCode.PROTOCOL_MISMATCH,
@@ -193,15 +193,15 @@ def parse_supported_commands(ping_payload: dict[str, Any]) -> set[str]:
     remote_hash = ping_payload.get("command_set_hash")
     if not isinstance(remote_hash, str) or not remote_hash.strip():
         raise AppError(
-            error_code="REMOTE_SCRIPT_INCOMPATIBLE",
+            error_code=ErrorCode.REMOTE_SCRIPT_INCOMPATIBLE,
             message="Remote Script ping payload is missing command_set_hash",
             hint="Reinstall Remote Script and restart Ableton Live.",
             exit_code=ExitCode.PROTOCOL_MISMATCH,
         )
     expected_hash = compute_command_set_hash(commands)
     if remote_hash != expected_hash:
         raise AppError(
-            error_code="REMOTE_SCRIPT_INCOMPATIBLE",
+            error_code=ErrorCode.REMOTE_SCRIPT_INCOMPATIBLE,
             message="Remote Script command_set_hash does not match supported_commands",
             hint="Reinstall Remote Script and restart Ableton Live.",
             exit_code=ExitCode.PROTOCOL_MISMATCH,

src/ableton_cli/client/_client_core.py

@@ -4,9 +4,8 @@
 
 from ..capabilities import read_only_remote_commands
 from ..config import Settings
-from ..errors import AppError, ExitCode, remote_error_to_app_error
-from .protocol import make_request, parse_response
-from .transport import RecordingTransport, ReplayTransport, TcpJsonlTransport
+from ..errors import AppError, ErrorCode, ExitCode
+from .backends import ClientBackend, LiveBackendClient, RecordingClient, ReplayClient
 
 
 class _AbletonClientCore:
@@ -23,61 +22,43 @@ def __init__(
         self._read_only_commands = read_only_remote_commands()
         if record_path is not None and replay_path is not None:
             raise AppError(
-                error_code="INVALID_ARGUMENT",
+                error_code=ErrorCode.INVALID_ARGUMENT,
                 message="--record and --replay cannot be used together",
                 hint="Choose exactly one of --record or --replay.",
                 exit_code=ExitCode.INVALID_ARGUMENT,
             )
 
-        base_transport = TcpJsonlTransport(
-            host=settings.host,
-            port=settings.port,
-            timeout_ms=settings.timeout_ms,
+        self._backend = self._build_backend(
+            settings=settings,
+            record_path=record_path,
+            replay_path=replay_path,
         )
+        # Keep direct transport access available for tests and fixtures.
+        self.transport = self._backend.transport
+
+    @staticmethod
+    def _build_backend(
+        *,
+        settings: Settings,
+        record_path: str | None,
+        replay_path: str | None,
+    ) -> ClientBackend:
         if replay_path is not None:
-            self.transport = ReplayTransport(path=replay_path)
-        elif record_path is not None:
-            self.transport = RecordingTransport(inner=base_transport, path=record_path)
-        else:
-            self.transport = base_transport
+            return ReplayClient(settings, path=replay_path)
+        if record_path is not None:
+            return RecordingClient(settings, path=record_path)
+        return LiveBackendClient(settings)
 
     def _dispatch(self, name: str, args: dict[str, Any]) -> dict[str, Any]:
         if self.read_only and name not in self._read_only_commands:
             raise AppError(
-                error_code="READ_ONLY_VIOLATION",
+                error_code=ErrorCode.READ_ONLY_VIOLATION,
                 message=f"Command '{name}' is blocked in read-only mode",
                 hint="Run without --read-only to execute write commands.",
                 exit_code=ExitCode.EXECUTION_FAILED,
                 details={"command": name},
             )
-
-        request = make_request(
-            name=name,
-            args=args,
-            protocol_version=self.settings.protocol_version,
-            meta={"request_timeout_ms": self.settings.timeout_ms},
-        )
-        raw_response = self.transport.send(request.to_dict())
-        response = parse_response(
-            payload=raw_response,
-            expected_request_id=request.request_id,
-            expected_protocol=self.settings.protocol_version,
-        )
-
-        if response.ok:
-            if response.result is None:
-                return {}
-            return response.result
-
-        if response.error is None:
-            raise AppError(
-                error_code="INTERNAL_ERROR",
-                message="Remote command failed without structured error payload",
-                hint="Update Remote Script error handling.",
-                exit_code=ExitCode.EXECUTION_FAILED,
-            )
-
-        raise remote_error_to_app_error(response.error)
+        return self._backend.dispatch(name, args)
 
     def _call(self, name: str, args: dict[str, Any] | None = None) -> dict[str, Any]:
         payload = {} if args is None else dict(args)