diff --git a/README.md b/README.md
index 78328dd..ab7f8bb 100644
--- a/README.md
+++ b/README.md
@@ -246,14 +246,14 @@ only be noticed by polling — that is what `session watch` does. A connection c
 instead send one `"type": "subscribe"` request and become an event stream:
 
 ```json
-{"type":"subscribe","name":"events","args":{"events":["tempo","is_playing"]},"meta":{},"request_id":"...","protocol_version":2}
+{"type":"subscribe","name":"events","args":{"events":["tempo","is_playing"]},"meta":{},"request_id":"...","protocol_version":3}
 ```
 
 The remote answers with the usual response envelope (`result.subscribed` lists the
 accepted names, `result.stream` is `true`) and then pushes one line per event:
 
 ```json
-{"type":"event","protocol_version":2,"event":"tempo","ts":1730000000.0,"data":{"tempo":174.0},"dropped":0}
+{"type":"event","protocol_version":3,"event":"tempo","ts":1730000000.0,"data":{"tempo":174.0},"dropped":0}
 ```
 
 - A subscribing connection stops accepting commands, so pushed events can never
@@ -315,20 +315,38 @@ accepted names, `result.stream` is `true`) and then pushes one line per event:
   "name": "song_info",
   "args": {},
   "meta": {
-    "request_timeout_ms": 15000
+    "request_timeout_ms": 15000,
+    "idempotency_key": "0f1c2d3e4f5a6b7c8d9e0f1a2b3c4d5e"
   },
   "request_id": "8c9f9b0c1a9d4dc2abdf2d53f3a19be9",
-  "protocol_version": 2
+  "protocol_version": 3
 }
 ```
 
+`request_id` is fresh on every send and never identifies a retry.
+
+`meta.idempotency_key` is optional (max 128 characters) and is what makes a
+resend safe. The Remote Script keeps the response of the last 256 completed
+keys; a request repeating a key it has already completed is **not executed
+again** — the stored response is returned with `idempotent_replay: true` added
+to `result` (or to `error.details` for a stored failure). `batch run` and
+`batch stream` generate one key per step and reuse it for every retry attempt,
+so a step that times out and is retried cannot be applied twice.
+
+Protocol versions must match exactly. After upgrading the CLI, reinstall the
+Remote Script and re-select the Control Surface in Live:
+
+```bash
+uv run ableton-cli install-remote-script --yes
+```
+
 ### Response (success)
 
 ```json
 {
   "ok": true,
   "request_id": "8c9f9b0c1a9d4dc2abdf2d53f3a19be9",
-  "protocol_version": 2,
+  "protocol_version": 3,
   "result": {
     "tempo": 120.0
   },
@@ -342,7 +360,7 @@ accepted names, `result.stream` is `true`) and then pushes one line per event:
 {
   "ok": false,
   "request_id": "8c9f9b0c1a9d4dc2abdf2d53f3a19be9",
-  "protocol_version": 2,
+  "protocol_version": 3,
   "result": null,
   "error": {
     "code": "INVALID_ARGUMENT",
@@ -383,20 +401,19 @@ Batch steps run once unless the step opts in with a `retry` object:
 `retry.on` defaults to `["REMOTE_BUSY"]`. `REMOTE_BUSY` is rejected by the
 Remote Script before the command is queued, so a retry cannot double-apply it.
 
-`TIMEOUT` is not retried by default, and listing it in `retry.on` fails with
-`INVALID_ARGUMENT` unless the step's remote command is idempotent (read
-commands). A timed-out command may already have been applied by Live, so
-retrying it would apply it twice.
+`TIMEOUT` is not retried by default, but adding it to `retry.on` is safe for
+any command: a retrying step carries one `meta.idempotency_key` across all of
+its attempts, so the Remote Script replays its stored response rather than
+applying the command a second time. A replayed response carries
+`idempotent_replay: true`.
 
-A step that fails with `TIMEOUT` reports `error.details.may_have_executed`:
+A step that fails with `TIMEOUT` and is not retried reports
+`error.details.may_have_executed`:
 
 - `true` — Live had already started the command; inspect the set before retrying
 - `false` — the command was cancelled before it ran and can be resent safely
 - `null` — no structured answer reached the CLI; assume it may have been applied
 
-A `TIMEOUT` with `may_have_executed: true` is never retried, even when
-`retry.on` lists `TIMEOUT`.
-
 ## Exit Codes
 
 - `0` success
@@ -514,7 +531,11 @@ Development workflows (local checks, quality gates, and merge criteria) are docu
 2. Confirm Ableton is running and `AbletonCliRemote` is selected as Control Surface.
 3. Confirm host/port (`127.0.0.1:8765` by default).
 4. Try a longer timeout: `--timeout-ms 30000`.
-5. If protocol mismatches, use `--protocol-version <n>` or `uv run ableton-cli config set protocol_version <n>`.
+5. If protocol mismatches, the installed Remote Script is from a different release:
+   run `uv run ableton-cli install-remote-script --yes`, restart Ableton Live, and re-select
+   `AbletonCliRemote` as Control Surface. Use `--protocol-version <n>` or
+   `uv run ableton-cli config set protocol_version <n>` only to talk to a deliberately pinned
+   Remote Script version.
 
 ### `AbletonCliRemote` not shown in Control Surface list
 
diff --git a/remote_script/AbletonCliRemote/command_backend_contract.py b/remote_script/AbletonCliRemote/command_backend_contract.py
index a96cf94..851e2da 100644
--- a/remote_script/AbletonCliRemote/command_backend_contract.py
+++ b/remote_script/AbletonCliRemote/command_backend_contract.py
@@ -6,7 +6,7 @@
 
 from .note_fields import NOTE_FIELD_SPECS
 
-PROTOCOL_VERSION = 2
+PROTOCOL_VERSION = 3
 REMOTE_SCRIPT_VERSION = "0.5.0"
 MIN_BPM = 20.0
 MAX_BPM = 999.0
diff --git a/remote_script/AbletonCliRemote/control_surface.py b/remote_script/AbletonCliRemote/control_surface.py
index 584a1da..ec8a80e 100644
--- a/remote_script/AbletonCliRemote/control_surface.py
+++ b/remote_script/AbletonCliRemote/control_surface.py
@@ -4,6 +4,7 @@
 import queue
 import threading
 import time
+from collections import OrderedDict
 from collections.abc import Callable
 from dataclasses import dataclass, field
 from typing import Any
@@ -45,12 +46,25 @@ def schedule_message(self, _delay: int, callback: Callable[[], None]) -> None:
 DRAIN_BUDGET_S = 0.005
 
 
+#: How many completed responses are kept for idempotency-key replay.
+#: A retry follows within one request timeout, so only recent keys can ever be
+#: hit; 256 covers a deep `batch stream` burst while keeping the cache small
+#: enough to be irrelevant to Live's memory. Eviction is FIFO by insertion.
+IDEMPOTENCY_CACHE_SIZE = 256
+
+#: Reserved result/details field marking a response that was replayed from the
+#: idempotency cache instead of being executed again. No command handler may
+#: return a result key by this name.
+IDEMPOTENT_REPLAY_FIELD = "idempotent_replay"
+
+
 @dataclass(slots=True)
 class _CommandRequest:
     name: str
     args: dict[str, Any]
     timeout_ms: int
     event: threading.Event
+    idempotency_key: str | None = None
     result: dict[str, Any] | None = None
     error: Exception | None = None
     lock: threading.Lock = field(default_factory=threading.Lock)
@@ -58,6 +72,30 @@ class _CommandRequest:
     executing: bool = False
 
 
+@dataclass(slots=True)
+class _CachedResponse:
+    """A completed dispatch, kept so a retry can replay it instead of rerunning."""
+
+    result: dict[str, Any] | None = None
+    error: CommandError | None = None
+
+
+def _as_command_error(error: Exception) -> CommandError:
+    """Normalize a dispatch failure for caching.
+
+    Mirrors how ``_execute_command_from_server_thread`` renders a
+    non-``CommandError`` exception, so a replayed failure is byte-identical to
+    the original apart from the replay marker.
+    """
+    if isinstance(error, CommandError):
+        return error
+    return CommandError(
+        code=RemoteErrorCode.INTERNAL_ERROR.value,
+        message=str(error),
+        hint="Check Ableton Log.txt for details.",
+    )
+
+
 def _mark_request_timed_out(request: _CommandRequest) -> bool:
     """Cancel a request whose client-side wait timed out.
 
@@ -82,6 +120,10 @@ def __init__(self, c_instance):  # noqa: ANN001
         self._queue: queue.Queue[_CommandRequest] = queue.Queue()
         self._drain_lock = threading.Lock()
         self._drain_scheduled = False
+        # Read and written only from `_drain_requests`, i.e. only on Live's
+        # main thread, which serializes dispatch — so no lock is needed and no
+        # request can observe another one mid-flight under the same key.
+        self._response_cache: OrderedDict[str, _CachedResponse] = OrderedDict()
         remote_config = load_remote_config()
         self._auth_token = remote_config.auth_token
         self._event_broker = EventBroker()
@@ -184,6 +226,7 @@ def _execute_command_from_server_thread(
             args=args,
             timeout_ms=timeout_ms,
             event=threading.Event(),
+            idempotency_key=meta.get("idempotency_key"),
         )
         self._queue.put(request)
         self._schedule_drain()
@@ -232,6 +275,38 @@ def _scheduled_drain(self) -> None:
         if needs_reschedule:
             self._schedule_drain()
 
+    def _remember_response(self, request: _CommandRequest) -> None:
+        key = request.idempotency_key
+        if key is None:
+            return
+        if request.error is not None:
+            entry = _CachedResponse(error=_as_command_error(request.error))
+        else:
+            entry = _CachedResponse(result=request.result)
+        self._response_cache.pop(key, None)
+        self._response_cache[key] = entry
+        while len(self._response_cache) > IDEMPOTENCY_CACHE_SIZE:
+            self._response_cache.popitem(last=False)
+
+    def _replay_cached_response(self, request: _CommandRequest) -> bool:
+        """Answer from the cache when this key already ran. Returns True on a hit."""
+        key = request.idempotency_key
+        if key is None:
+            return False
+        entry = self._response_cache.get(key)
+        if entry is None:
+            return False
+        if entry.error is not None:
+            request.error = CommandError(
+                code=entry.error.code,
+                message=entry.error.message,
+                hint=entry.error.hint,
+                details={**(entry.error.details or {}), IDEMPOTENT_REPLAY_FIELD: True},
+            )
+        else:
+            request.result = {**(entry.result or {}), IDEMPOTENT_REPLAY_FIELD: True}
+        return True
+
     def _drain_requests(self, budget_s: float | None = None) -> None:
         """Execute queued requests; ``budget_s=None`` drains without a deadline.
 
@@ -252,16 +327,26 @@ def _drain_requests(self, budget_s: float | None = None) -> None:
 
             with request.lock:
                 if request.cancelled:
+                    # Never dispatched, so there is nothing to remember: a retry
+                    # of this key must actually run the command.
                     request.event.set()
                     continue
                 request.executing = True
 
+            if self._replay_cached_response(request):
+                request.event.set()
+                continue
+
             dispatched += 1
             try:
                 request.result = dispatch_command(self._backend, request.name, request.args)
             except Exception as exc:  # noqa: BLE001
                 request.error = exc
             finally:
+                # Recorded even when the client already gave up waiting — that
+                # abandoned response is exactly what a retry needs to replay
+                # instead of applying the command a second time.
+                self._remember_response(request)
                 request.event.set()
 
     def update_display(self) -> None:
diff --git a/remote_script/AbletonCliRemote/server.py b/remote_script/AbletonCliRemote/server.py
index 2ce7df4..af2b9c0 100644
--- a/remote_script/AbletonCliRemote/server.py
+++ b/remote_script/AbletonCliRemote/server.py
@@ -13,6 +13,11 @@
 _REQUEST_KEYS = {"type", "name", "args", "meta", "request_id", "protocol_version"}
 _REQUEST_TYPES = ("command", "subscribe")
 
+#: Mirrors ableton_cli.client.protocol.IDEMPOTENCY_KEY_MAX_LENGTH. The Remote
+#: Script cannot import from the CLI package, so the bound is restated here and
+#: pinned by a test.
+IDEMPOTENCY_KEY_MAX_LENGTH = 128
+
 #: How long a streaming connection waits for the next event before checking
 #: whether the server is shutting down.
 _EVENT_POLL_INTERVAL_S = 0.25
@@ -150,10 +155,30 @@ def _parse_command_request(
             message="meta must be an object",
             hint="Pass a JSON object for meta.",
         )
+    _validate_idempotency_key(meta)
 
     return request_id, str(request_type), name, args, meta
 
 
+def _validate_idempotency_key(meta: dict[str, Any]) -> None:
+    if "idempotency_key" not in meta:
+        return
+    key = meta["idempotency_key"]
+    if not isinstance(key, str) or not key:
+        raise _invalid_request(
+            message="meta.idempotency_key must be a non-empty string",
+            hint="Send a stable opaque token, or omit meta.idempotency_key.",
+        )
+    if len(key) > IDEMPOTENCY_KEY_MAX_LENGTH:
+        raise _invalid_request(
+            message=(
+                f"meta.idempotency_key must be at most {IDEMPOTENCY_KEY_MAX_LENGTH} "
+                f"characters, got {len(key)}"
+            ),
+            hint="Use a short opaque token such as a uuid4 hex.",
+        )
+
+
 class _CommandTCPServer(socketserver.ThreadingTCPServer):
     allow_reuse_address = True
 
diff --git a/skills/ableton-cli/SKILL.md b/skills/ableton-cli/SKILL.md
index d489cf9..6a65ce1 100644
--- a/skills/ableton-cli/SKILL.md
+++ b/skills/ableton-cli/SKILL.md
@@ -409,7 +409,7 @@ uv run ableton-cli effect utility observe 0 0
 uv run ableton-cli config init
 uv run ableton-cli config init --dry-run
 uv run ableton-cli config show
-uv run ableton-cli config set protocol_version 2
+uv run ableton-cli config set protocol_version 3
 uv run ableton-cli completion
 uv run ableton-cli --install-completion
 uv run ableton-cli --show-completion
@@ -423,8 +423,10 @@ uv run ableton-cli --show-completion
 - For low-latency repeated automation operations, prefer `uv run ableton-cli batch stream`.
 - Capability and compatibility checks are explicit through `uv run ableton-cli ping` and `uv run ableton-cli doctor`.
 - Batch steps run once unless the step declares `retry`. `retry.on` defaults to `["REMOTE_BUSY"]`.
-  - Listing `TIMEOUT` in `retry.on` fails with `INVALID_ARGUMENT` unless the step's remote command is idempotent (read commands). A timed-out write may already have been applied, so a retry would double-apply it.
-  - A `TIMEOUT` step failure reports `error.details.may_have_executed` (`true` / `false` / `null` when unknown). Inspect the set with a read command before resending anything other than `false`.
+  - A retrying step carries one `meta.idempotency_key` across every attempt, so the Remote Script replays its stored response instead of applying the command twice. Adding `TIMEOUT` to `retry.on` is therefore safe for write commands too.
+  - A replayed response is marked with `idempotent_replay: true` in `result` (or in `error.details` for a replayed failure).
+  - A `TIMEOUT` step failure that is not retried reports `error.details.may_have_executed` (`true` / `false` / `null` when unknown). Inspect the set with a read command before resending anything other than `false`.
+- Protocol version 3 is required. On `PROTOCOL_VERSION_MISMATCH`, run `uv run ableton-cli install-remote-script --yes`, restart Ableton Live, and re-select `AbletonCliRemote`.
 - Destructive master device deletion requires `--yes`. Live API unsupported operations fail explicitly with `not_supported_by_live_api`.
 - Standard wrapper commands (`synth <type> ...`, `effect <type> ...`) are strict and intentionally fail if required parameter names are missing.
   - If you get `Missing required standard ... keys`, use generic commands instead:
@@ -464,7 +466,7 @@ uv run ableton-cli --output json ping
     "result": {
       "host": "127.0.0.1",
       "port": 8765,
-      "protocol_version": 2,
+      "protocol_version": 3,
       "remote_script_version": "0.5.0",
       "rtt_ms": 2.31
     },
diff --git a/src/ableton_cli/client/_client_core.py b/src/ableton_cli/client/_client_core.py
index a71a1a3..fe7161e 100644
--- a/src/ableton_cli/client/_client_core.py
+++ b/src/ableton_cli/client/_client_core.py
@@ -50,7 +50,13 @@ def _build_backend(
             return RecordingClient(settings, path=record_path)
         return LiveBackendClient(settings)
 
-    def _dispatch(self, name: str, args: dict[str, Any]) -> dict[str, Any]:
+    def _dispatch(
+        self,
+        name: str,
+        args: dict[str, Any],
+        *,
+        idempotency_key: str | None = None,
+    ) -> dict[str, Any]:
         if self.read_only and name not in self._read_only_commands:
             raise AppError(
                 error_code=ErrorCode.READ_ONLY_VIOLATION,
@@ -59,7 +65,7 @@ def _dispatch(self, name: str, args: dict[str, Any]) -> dict[str, Any]:
                 exit_code=ExitCode.EXECUTION_FAILED,
                 details={"command": name},
             )
-        return self._backend.dispatch(name, args)
+        return self._backend.dispatch(name, args, idempotency_key=idempotency_key)
 
     def _call(self, name: str, args: dict[str, Any] | None = None) -> dict[str, Any]:
         payload = {} if args is None else dict(args)
@@ -69,9 +75,11 @@ def execute_remote_command(
         self,
         name: str,
         args: dict[str, Any] | None = None,
+        *,
+        idempotency_key: str | None = None,
     ) -> dict[str, Any]:
         payload = {} if args is None else dict(args)
-        return self._dispatch(name, payload)
+        return self._dispatch(name, payload, idempotency_key=idempotency_key)
 
     @staticmethod
     def _add_if_not_none(args: dict[str, Any], key: str, value: Any) -> None:
diff --git a/src/ableton_cli/client/backends.py b/src/ableton_cli/client/backends.py
index 9f326fa..6fcaf4f 100644
--- a/src/ableton_cli/client/backends.py
+++ b/src/ableton_cli/client/backends.py
@@ -11,7 +11,13 @@
 class ClientBackend(Protocol):
     transport: JsonTransport
 
-    def dispatch(self, name: str, args: dict[str, Any]) -> dict[str, Any]: ...
+    def dispatch(
+        self,
+        name: str,
+        args: dict[str, Any],
+        *,
+        idempotency_key: str | None = None,
+    ) -> dict[str, Any]: ...
 
 
 class _TransportBackend:
@@ -19,7 +25,13 @@ def __init__(self, settings: Settings, transport: JsonTransport) -> None:
         self._settings = settings
         self.transport = transport
 
-    def dispatch(self, name: str, args: dict[str, Any]) -> dict[str, Any]:
+    def dispatch(
+        self,
+        name: str,
+        args: dict[str, Any],
+        *,
+        idempotency_key: str | None = None,
+    ) -> dict[str, Any]:
         meta: dict[str, Any] = {"request_timeout_ms": self._settings.timeout_ms}
         if self._settings.auth_token is not None:
             meta["auth_token"] = self._settings.auth_token
@@ -28,6 +40,7 @@ def dispatch(self, name: str, args: dict[str, Any]) -> dict[str, Any]:
             args=args,
             protocol_version=self._settings.protocol_version,
             meta=meta,
+            idempotency_key=idempotency_key,
         )
         raw_response = self.transport.send(request.to_dict())
         response = parse_response(
diff --git a/src/ableton_cli/client/protocol.py b/src/ableton_cli/client/protocol.py
index 8c93750..e98e6df 100644
--- a/src/ableton_cli/client/protocol.py
+++ b/src/ableton_cli/client/protocol.py
@@ -59,17 +59,55 @@ def _require_response_string(payload: dict[str, Any], key: str) -> str:
     return value
 
 
+#: Upper bound on ``meta.idempotency_key``. Long enough for a uuid4 hex or an
+#: opaque caller-supplied token, short enough that one client cannot grow the
+#: Remote Script's bounded response cache into a memory problem.
+IDEMPOTENCY_KEY_MAX_LENGTH = 128
+
+
+def _validated_idempotency_key(value: str) -> str:
+    if not isinstance(value, str) or not value:
+        raise AppError(
+            error_code=ErrorCode.INVALID_ARGUMENT,
+            message="idempotency_key must be a non-empty string",
+            hint="Pass a stable opaque token, or omit idempotency_key entirely.",
+            exit_code=ExitCode.INVALID_ARGUMENT,
+        )
+    if len(value) > IDEMPOTENCY_KEY_MAX_LENGTH:
+        raise AppError(
+            error_code=ErrorCode.INVALID_ARGUMENT,
+            message=(
+                f"idempotency_key must be at most {IDEMPOTENCY_KEY_MAX_LENGTH} "
+                f"characters, got {len(value)}"
+            ),
+            hint="Use a short opaque token such as a uuid4 hex.",
+            exit_code=ExitCode.INVALID_ARGUMENT,
+        )
+    return value
+
+
 def make_request(
     name: str,
     args: dict[str, Any],
     protocol_version: int,
     meta: dict[str, Any] | None = None,
+    idempotency_key: str | None = None,
 ) -> Request:
+    """Build one request envelope.
+
+    ``request_id`` is fresh on every call, so it can never identify a retry.
+    ``idempotency_key`` is deliberately *not* generated here: the caller that
+    owns the retry loop generates it once and passes the same value to every
+    attempt, which is what makes the Remote Script able to deduplicate them.
+    """
+    request_meta = dict(meta or {})
+    if idempotency_key is not None:
+        request_meta["idempotency_key"] = _validated_idempotency_key(idempotency_key)
     return Request(
         type="command",
         name=name,
         args=args,
-        meta=meta or {},
+        meta=request_meta,
         request_id=uuid.uuid4().hex,
         protocol_version=protocol_version,
     )
diff --git a/src/ableton_cli/commands/batch.py b/src/ableton_cli/commands/batch.py
index 41956fc..5bac433 100644
--- a/src/ableton_cli/commands/batch.py
+++ b/src/ableton_cli/commands/batch.py
@@ -3,45 +3,30 @@
 import json
 import sys
 import time
+import uuid
 from pathlib import Path
 from typing import Annotated, Any
 
 import typer
 
 from ..capabilities import parse_supported_commands, required_remote_commands
-from ..command_specs import remote_command_spec_map
 from ..errors import AppError, ErrorCode, ExitCode
 from ..runtime import execute_command, get_client, get_runtime
 from ._validation import invalid_argument, require_non_empty_string
 
 batch_app = typer.Typer(help="Batch commands", no_args_is_help=True)
 _ASSERT_OPERATORS = frozenset({"eq", "ne", "gt", "gte", "lt", "lte"})
-#: REMOTE_BUSY is the only code that is safe to retry blindly: the Remote Script
-#: raises it from `_execute_command_from_server_thread` on the queue-depth check,
-#: which runs *before* `self._queue.put()`, so a REMOTE_BUSY request provably
-#: never reached Live's main thread and had zero side effects.
-#: TIMEOUT is deliberately absent — a timed-out request may already have been
-#: applied, so retrying it double-applies non-idempotent commands.
+#: REMOTE_BUSY is the only code safe to retry with no further machinery: the
+#: Remote Script raises it from `_execute_command_from_server_thread` on the
+#: queue-depth check, which runs *before* `self._queue.put()`, so the request
+#: provably never reached Live's main thread and had zero side effects.
+#: TIMEOUT stays out of the default so a step must opt in; opting in is safe
+#: because `_execute_step` carries one idempotency key across every attempt and
+#: the Remote Script replays the stored response rather than re-running.
 _DEFAULT_RETRY_CODES = ("REMOTE_BUSY",)
 
 
-def _reject_unsafe_timeout_retry(*, step_index: int, command_name: str) -> None:
-    spec = remote_command_spec_map().get(command_name)
-    if spec is not None and spec.side_effect.idempotent:
-        return
-    known = "" if spec is not None else " with unknown idempotency"
-    raise invalid_argument(
-        message=f"steps[{step_index}].retry.on may not include TIMEOUT for {command_name!r}",
-        hint=(
-            f"TIMEOUT retry is unsafe for non-idempotent command {command_name!r}{known}; "
-            "the command may have already been applied. Remove TIMEOUT from retry.on."
-        ),
-    )
-
-
-def _parse_retry_object(
-    raw_retry: Any, *, step_index: int, command_name: str
-) -> dict[str, Any] | None:
+def _parse_retry_object(raw_retry: Any, *, step_index: int) -> dict[str, Any] | None:
     if raw_retry is None:
         return None
     if not isinstance(raw_retry, dict):
@@ -84,9 +69,6 @@ def _parse_retry_object(
             )
         )
 
-    if "TIMEOUT" in retry_on:
-        _reject_unsafe_timeout_retry(step_index=step_index, command_name=command_name)
-
     return {
         "max_attempts": raw_max_attempts,
         "backoff_ms": raw_backoff_ms,
@@ -274,9 +256,7 @@ def _parse_batch_object(payload: Any, *, source_name: str) -> dict[str, Any]:
         parsed_step = {
             "name": name,
             "args": raw_args,
-            "retry": _parse_retry_object(
-                raw_step.get("retry"), step_index=index, command_name=name
-            ),
+            "retry": _parse_retry_object(raw_step.get("retry"), step_index=index),
             "assert": _parse_assert_object(raw_step.get("assert"), step_index=index),
         }
         steps.append(parsed_step)
@@ -541,19 +521,24 @@ def _execute_step(
     retry_on = set(retry["on"])
     backoff_ms = retry["backoff_ms"]
 
+    # Generated once and reused by every attempt: that is the whole mechanism.
+    # A fresh request_id is minted per send and can never identify a retry, so
+    # this key is what lets the Remote Script recognise a resend and replay the
+    # stored response instead of applying the command a second time.
+    idempotency_key = uuid.uuid4().hex
+
     attempt = 0
     while True:
         attempt += 1
         try:
-            result = client.execute_remote_command(step["name"], step["args"])
+            result = client.execute_remote_command(
+                step["name"],
+                step["args"],
+                idempotency_key=idempotency_key,
+            )
             return result, attempt
         except AppError as exc:
             _annotate_step_timeout(exc, step_index=step_index, name=step["name"])
-            # The Remote Script could not cancel this request before Live began
-            # dispatching it, so a resend would apply the command a second time.
-            # This outranks retry.on: even an opt-in retry must not double-apply.
-            if exc.details.get("may_have_executed") is True:
-                raise
             if exc.error_code not in retry_on:
                 raise
             if attempt >= max_attempts:
diff --git a/src/ableton_cli/config.py b/src/ableton_cli/config.py
index 563cea0..fabec1e 100644
--- a/src/ableton_cli/config.py
+++ b/src/ableton_cli/config.py
@@ -21,7 +21,7 @@ class Settings:
     timeout_ms: int = 15000
     log_level: str = "INFO"
     log_file: str | None = None
-    protocol_version: int = 2
+    protocol_version: int = 3
     config_path: str | None = None
     auth_token: str | None = None
 
@@ -153,7 +153,7 @@ def _settings_from_merged(merged: dict[str, Any], *, resolved_path: Path) -> Set
         raise AppError(
             error_code=ErrorCode.CONFIG_INVALID,
             message=f"protocol_version must be positive: {protocol_version}",
-            hint="Set protocol_version to 2.",
+            hint="Set protocol_version to 3.",
             exit_code=ExitCode.CONFIG_INVALID,
         )
 
diff --git a/src/ableton_cli/contracts/public_snapshot.py b/src/ableton_cli/contracts/public_snapshot.py
index bf5d446..4ecb55e 100644
--- a/src/ableton_cli/contracts/public_snapshot.py
+++ b/src/ableton_cli/contracts/public_snapshot.py
@@ -14,7 +14,7 @@ def build_public_contract_snapshot() -> dict[str, Any]:
     protocol_request = make_request(
         name="example_command",
         args={"example": True},
-        protocol_version=2,
+        protocol_version=3,
         meta={"request_timeout_ms": 15000},
     ).to_dict()
     protocol_request["request_id"] = "<request-id>"
@@ -46,14 +46,14 @@ def build_public_contract_snapshot() -> dict[str, Any]:
             "response_success": {
                 "ok": True,
                 "request_id": "<request-id>",
-                "protocol_version": 2,
+                "protocol_version": 3,
                 "result": {"status": "ok"},
                 "error": None,
             },
             "response_error": {
                 "ok": False,
                 "request_id": "<request-id>",
-                "protocol_version": 2,
+                "protocol_version": 3,
                 "result": None,
                 "error": {
                     "code": ErrorCode.INVALID_ARGUMENT.value,
diff --git a/tests/snapshots/public_contract_snapshot.json b/tests/snapshots/public_contract_snapshot.json
index 3a35036..030e32a 100644
--- a/tests/snapshots/public_contract_snapshot.json
+++ b/tests/snapshots/public_contract_snapshot.json
@@ -20325,7 +20325,7 @@
         "request_timeout_ms": 15000
       },
       "name": "example_command",
-      "protocol_version": 2,
+      "protocol_version": 3,
       "request_id": "<request-id>",
       "type": "command"
     },
@@ -20339,7 +20339,7 @@
         "message": "Example failure"
       },
       "ok": false,
-      "protocol_version": 2,
+      "protocol_version": 3,
       "request_id": "<request-id>",
       "result": null
     },
@@ -20353,7 +20353,7 @@
     "response_success": {
       "error": null,
       "ok": true,
-      "protocol_version": 2,
+      "protocol_version": 3,
       "request_id": "<request-id>",
       "result": {
         "status": "ok"
diff --git a/tests/test_ableton_client.py b/tests/test_ableton_client.py
index 6acaeb1..3e2dda5 100644
--- a/tests/test_ableton_client.py
+++ b/tests/test_ableton_client.py
@@ -17,7 +17,7 @@ def _settings() -> Settings:
         timeout_ms=15000,
         log_level="INFO",
         log_file=None,
-        protocol_version=2,
+        protocol_version=3,
         config_path="/tmp/ableton-cli-test.toml",
     )
 
@@ -780,3 +780,32 @@ def test_client_parameter_commands_share_payload_shape(
         "parameter_ref": {"mode": "index", "index": 2},
         "value": expected_value,
     }
+
+
+def test_retrying_a_command_reuses_the_key_but_not_the_request_id(monkeypatch) -> None:
+    client = AbletonClient(_settings())
+    requests: list[dict[str, Any]] = []
+
+    def _send(request: dict[str, Any]):  # noqa: ANN202
+        requests.append(request)
+        return _ok_response(request, {})
+
+    monkeypatch.setattr(client.transport, "send", _send)
+
+    for _attempt in range(2):
+        client.execute_remote_command("add_notes_to_clip", {}, idempotency_key="step-key")
+
+    assert [request["meta"]["idempotency_key"] for request in requests] == [
+        "step-key",
+        "step-key",
+    ]
+    assert requests[0]["request_id"] != requests[1]["request_id"]
+
+
+def test_command_without_an_idempotency_key_sends_no_key(monkeypatch) -> None:
+    client = AbletonClient(_settings())
+    requests = _capture_requests(monkeypatch, client)
+
+    client.execute_remote_command("add_notes_to_clip", {})
+
+    assert "idempotency_key" not in requests[0]["meta"]
diff --git a/tests/test_batch_retry_safety.py b/tests/test_batch_retry_safety.py
index 6480e72..9abb5c4 100644
--- a/tests/test_batch_retry_safety.py
+++ b/tests/test_batch_retry_safety.py
@@ -24,13 +24,21 @@ def _timeout_error(*, may_have_executed: bool | None = None) -> AppError:
 class _StepClientStub:
     def __init__(self) -> None:
         self.calls: list[tuple[str, dict[str, Any]]] = []
+        self.idempotency_keys: list[str | None] = []
         self._responses: dict[str, list[dict[str, Any] | AppError]] = {}
 
     def set_responses(self, name: str, items: list[dict[str, Any] | AppError]) -> None:
         self._responses[name] = list(items)
 
-    def execute_remote_command(self, name: str, args: dict[str, Any]) -> dict[str, Any]:
+    def execute_remote_command(
+        self,
+        name: str,
+        args: dict[str, Any],
+        *,
+        idempotency_key: str | None = None,
+    ) -> dict[str, Any]:
         self.calls.append((name, args))
+        self.idempotency_keys.append(idempotency_key)
         queue = self._responses.get(name)
         if queue:
             item = queue.pop(0)
@@ -56,9 +64,11 @@ def _run_batch(runner, cli_app, steps: list[dict[str, Any]]):  # noqa: ANN001, A
     )
 
 
-def test_timeout_retry_is_rejected_for_a_non_idempotent_command(
+def test_timeout_retry_of_a_write_reuses_one_idempotency_key(
     runner, cli_app, client: _StepClientStub
 ) -> None:
+    client.set_responses("add_notes_to_clip", [_timeout_error(), {"ok": True}])
+
     result = _run_batch(
         runner,
         cli_app,
@@ -66,17 +76,32 @@ def test_timeout_retry_is_rejected_for_a_non_idempotent_command(
             {
                 "name": "add_notes_to_clip",
                 "args": {},
-                "retry": {"max_attempts": 3, "on": ["TIMEOUT"]},
+                "retry": {"max_attempts": 3, "backoff_ms": 0, "on": ["TIMEOUT"]},
             }
         ],
     )
 
-    assert result.exit_code == 2
-    payload = json.loads(result.stdout)
-    assert payload["ok"] is False
-    assert payload["error"]["code"] == "INVALID_ARGUMENT"
-    assert "add_notes_to_clip" in payload["error"]["hint"]
-    assert client.calls == []
+    assert result.exit_code == 0
+    assert len(client.calls) == 2
+    assert client.idempotency_keys[0] == client.idempotency_keys[1]
+    assert client.idempotency_keys[0] is not None
+
+
+def test_each_step_gets_its_own_idempotency_key(runner, cli_app, client: _StepClientStub) -> None:
+    step = {"name": "add_notes_to_clip", "args": {}, "retry": {"max_attempts": 2}}
+    result = _run_batch(runner, cli_app, [step, dict(step)])
+
+    assert result.exit_code == 0
+    assert len(set(client.idempotency_keys)) == 2
+
+
+def test_step_without_retry_sends_no_idempotency_key(
+    runner, cli_app, client: _StepClientStub
+) -> None:
+    result = _run_batch(runner, cli_app, [{"name": "add_notes_to_clip", "args": {}}])
+
+    assert result.exit_code == 0
+    assert client.idempotency_keys == [None]
 
 
 def test_timeout_retry_is_allowed_for_an_idempotent_command(
@@ -128,12 +153,12 @@ def test_step_without_retry_is_executed_once(runner, cli_app, client: _StepClien
     assert len(client.calls) == 1
 
 
-def test_timeout_that_may_have_executed_is_never_retried(
+def test_timeout_that_may_have_executed_is_retried_under_one_key(
     runner, cli_app, client: _StepClientStub
 ) -> None:
     client.set_responses(
-        "tracks_list",
-        [_timeout_error(may_have_executed=True), {"tracks": []}],
+        "add_notes_to_clip",
+        [_timeout_error(may_have_executed=True), {"ok": True}],
     )
 
     result = _run_batch(
@@ -141,17 +166,18 @@ def test_timeout_that_may_have_executed_is_never_retried(
         cli_app,
         [
             {
-                "name": "tracks_list",
+                "name": "add_notes_to_clip",
                 "args": {},
                 "retry": {"max_attempts": 3, "backoff_ms": 0, "on": ["TIMEOUT"]},
             }
         ],
     )
 
-    assert result.exit_code == 12
-    payload = json.loads(result.stdout)
-    assert payload["error"]["code"] == "TIMEOUT"
-    assert len(client.calls) == 1
+    # The Remote Script recognises the repeated key and replays its stored
+    # response, so resending cannot double-apply the notes.
+    assert result.exit_code == 0
+    assert len(client.calls) == 2
+    assert client.idempotency_keys[0] == client.idempotency_keys[1]
 
 
 def test_step_timeout_exposes_may_have_executed(runner, cli_app, client: _StepClientStub) -> None:
diff --git a/tests/test_cli_json_output.py b/tests/test_cli_json_output.py
index 2d6a0a6..f94a24b 100644
--- a/tests/test_cli_json_output.py
+++ b/tests/test_cli_json_output.py
@@ -201,7 +201,7 @@ def test_ping_includes_capabilities_when_remote_reports_them(runner, cli_app, mo
     class _ClientStub:
         def ping(self):  # noqa: ANN201
             return {
-                "protocol_version": 2,
+                "protocol_version": 3,
                 "remote_script_version": "0.2.0",
                 "supported_commands": ["ping", "song_info"],
                 "command_set_hash": "abc123",
@@ -237,7 +237,7 @@ def test_config_set_updates_key_and_returns_json_envelope(runner, cli_app, tmp_p
                 'host = "127.0.0.1"',
                 "port = 8765",
                 "timeout_ms = 15000",
-                "protocol_version = 2",
+                "protocol_version = 3",
                 "",
             ]
         ),
@@ -386,7 +386,7 @@ def test_auth_token_global_option_is_used_for_dispatch_meta(
     captured: dict[str, object] = {}
     original_dispatch = backends_module._TransportBackend.dispatch
 
-    def _capture_dispatch(self, name, args):  # noqa: ANN001, ANN202
+    def _capture_dispatch(self, name, args, **kwargs):  # noqa: ANN001, ANN202
         request_meta: dict[str, object] = {}
 
         def _fake_send(payload):  # noqa: ANN001, ANN202
@@ -400,7 +400,7 @@ def _fake_send(payload):  # noqa: ANN001, ANN202
             }
 
         monkeypatch.setattr(self.transport, "send", _fake_send)
-        result = original_dispatch(self, name, args)
+        result = original_dispatch(self, name, args, **kwargs)
         captured["meta"] = request_meta
         return result
 
@@ -429,7 +429,7 @@ def _fake_send(payload):  # noqa: ANN001, ANN202
 
 def test_protocol_version_global_option_overrides_config(runner, cli_app, tmp_path) -> None:
     config_path = tmp_path / "config.toml"
-    config_path.write_text("protocol_version = 2\n", encoding="utf-8")
+    config_path.write_text("protocol_version = 3\n", encoding="utf-8")
 
     result = runner.invoke(
         cli_app,
@@ -459,9 +459,9 @@ def test_ping_supports_replay_option_without_network(runner, cli_app, tmp_path:
                 "response": {
                     "ok": True,
                     "request_id": "recorded-request-id",
-                    "protocol_version": 2,
+                    "protocol_version": 3,
                     "result": {
-                        "protocol_version": 2,
+                        "protocol_version": 3,
                         "remote_script_version": "9.9.9",
                         "supported_commands": ["ping"],
                         "command_set_hash": "hash",
@@ -478,7 +478,15 @@ def test_ping_supports_replay_option_without_network(runner, cli_app, tmp_path:
 
     result = runner.invoke(
         cli_app,
-        ["--output", "json", "--replay", str(replay_path), "ping"],
+        [
+            "--config",
+            str(tmp_path / "empty.toml"),
+            "--output",
+            "json",
+            "--replay",
+            str(replay_path),
+            "ping",
+        ],
     )
 
     assert result.exit_code == 0
@@ -519,7 +527,7 @@ def test_read_only_allows_read_commands(runner, cli_app, tmp_path: Path) -> None
                 "response": {
                     "ok": True,
                     "request_id": "recorded-request-id",
-                    "protocol_version": 2,
+                    "protocol_version": 3,
                     "result": {"tempo": 120.0},
                     "error": None,
                 },
@@ -532,7 +540,17 @@ def test_read_only_allows_read_commands(runner, cli_app, tmp_path: Path) -> None
 
     result = runner.invoke(
         cli_app,
-        ["--output", "json", "--read-only", "--replay", str(replay_path), "song", "info"],
+        [
+            "--config",
+            str(tmp_path / "empty.toml"),
+            "--output",
+            "json",
+            "--read-only",
+            "--replay",
+            str(replay_path),
+            "song",
+            "info",
+        ],
     )
 
     assert result.exit_code == 0
@@ -577,7 +595,7 @@ def test_read_only_batch_rejects_write_steps(runner, cli_app, tmp_path: Path) ->
                 "response": {
                     "ok": True,
                     "request_id": "recorded-request-id",
-                    "protocol_version": 2,
+                    "protocol_version": 3,
                     "result": {"tracks": []},
                     "error": None,
                 },
@@ -591,6 +609,8 @@ def test_read_only_batch_rejects_write_steps(runner, cli_app, tmp_path: Path) ->
     result = runner.invoke(
         cli_app,
         [
+            "--config",
+            str(tmp_path / "empty.toml"),
             "--output",
             "json",
             "--read-only",
diff --git a/tests/test_cli_new_commands.py b/tests/test_cli_new_commands.py
index cc1b346..8c1cb96 100644
--- a/tests/test_cli_new_commands.py
+++ b/tests/test_cli_new_commands.py
@@ -3244,7 +3244,13 @@ class _BatchStreamClientStub:
     def __init__(self) -> None:
         self.calls: list[tuple[str, dict[str, object]]] = []
 
-    def execute_remote_command(self, name: str, args: dict[str, object]):  # noqa: ANN201
+    def execute_remote_command(  # noqa: ANN201
+        self,
+        name: str,
+        args: dict[str, object],
+        *,
+        idempotency_key: str | None = None,
+    ):
         self.calls.append((name, args))
         return {"ok": True, "name": name}
 
@@ -3350,7 +3356,13 @@ def set_responses(self, name: str, items: list[dict[str, object] | AppError]) ->
     def ping(self):  # noqa: ANN201
         return dict(self._ping_payload)
 
-    def execute_remote_command(self, name: str, args: dict[str, object]):  # noqa: ANN201
+    def execute_remote_command(  # noqa: ANN201
+        self,
+        name: str,
+        args: dict[str, object],
+        *,
+        idempotency_key: str | None = None,
+    ):
         self.calls.append((name, args))
         queue = self._responses.get(name)
         if queue:
diff --git a/tests/test_client_backends.py b/tests/test_client_backends.py
index 4aeb42d..0ab62f2 100644
--- a/tests/test_client_backends.py
+++ b/tests/test_client_backends.py
@@ -18,7 +18,7 @@ def _settings() -> Settings:
         timeout_ms=15000,
         log_level="INFO",
         log_file=None,
-        protocol_version=2,
+        protocol_version=3,
         config_path="/tmp/ableton-cli-test.toml",
     )
 
@@ -45,9 +45,12 @@ def test_client_dispatch_uses_backend(monkeypatch: pytest.MonkeyPatch) -> None:
     client = AbletonClient(_settings())
     captured: dict[str, Any] = {}
 
-    def _dispatch(name: str, args: dict[str, Any]) -> dict[str, Any]:
+    def _dispatch(
+        name: str, args: dict[str, Any], *, idempotency_key: str | None = None
+    ) -> dict[str, Any]:
         captured["name"] = name
         captured["args"] = args
+        captured["idempotency_key"] = idempotency_key
         return {"tempo": 123.0}
 
     monkeypatch.setattr(client._backend, "dispatch", _dispatch)
@@ -55,13 +58,13 @@ def _dispatch(name: str, args: dict[str, Any]) -> dict[str, Any]:
     result = client.song_info()
 
     assert result == {"tempo": 123.0}
-    assert captured == {"name": "song_info", "args": {}}
+    assert captured == {"name": "song_info", "args": {}, "idempotency_key": None}
 
 
 def test_client_read_only_stops_dispatch_before_backend(monkeypatch: pytest.MonkeyPatch) -> None:
     client = AbletonClient(_settings(), read_only=True)
 
-    def _dispatch(_name: str, _args: dict[str, Any]) -> dict[str, Any]:
+    def _dispatch(_name: str, _args: dict[str, Any], **_kwargs: Any) -> dict[str, Any]:
         raise AssertionError("backend dispatch must not run for blocked write commands")
 
     monkeypatch.setattr(client._backend, "dispatch", _dispatch)
@@ -75,7 +78,7 @@ def _dispatch(_name: str, _args: dict[str, Any]) -> dict[str, Any]:
 def test_client_read_only_blocks_song_undo(monkeypatch: pytest.MonkeyPatch) -> None:
     client = AbletonClient(_settings(), read_only=True)
 
-    def _dispatch(_name: str, _args: dict[str, Any]) -> dict[str, Any]:
+    def _dispatch(_name: str, _args: dict[str, Any], **_kwargs: Any) -> dict[str, Any]:
         raise AssertionError("backend dispatch must not run for blocked write commands")
 
     monkeypatch.setattr(client._backend, "dispatch", _dispatch)
diff --git a/tests/test_config_precedence.py b/tests/test_config_precedence.py
index 2a7ad68..97d0277 100644
--- a/tests/test_config_precedence.py
+++ b/tests/test_config_precedence.py
@@ -14,7 +14,7 @@ def test_config_priority_cli_over_env_over_file_over_default(monkeypatch, tmp_pa
                 "port = 7777",
                 "timeout_ms = 3000",
                 'log_level = "DEBUG"',
-                "protocol_version = 2",
+                "protocol_version = 3",
                 "",
             ]
         ),
@@ -41,7 +41,7 @@ def test_config_defaults_use_protocol_v2_and_longer_timeout(tmp_path: Path) -> N
     settings = resolve_settings(cli_overrides={}, config_path=tmp_path / "missing.toml")
 
     assert settings.timeout_ms == 15000
-    assert settings.protocol_version == 2
+    assert settings.protocol_version == 3
 
 
 def test_config_auth_token_defaults_to_none(tmp_path: Path) -> None:
diff --git a/tests/test_control_surface_timeout.py b/tests/test_control_surface_timeout.py
index 6c510a8..dc4e778 100644
--- a/tests/test_control_surface_timeout.py
+++ b/tests/test_control_surface_timeout.py
@@ -136,3 +136,87 @@ def test_cancelled_request_is_skipped_by_drain_without_executing_flag(
     assert calls == []
     assert request.event.is_set()
     surface.disconnect()
+
+
+def _timing_out_mid_dispatch(
+    monkeypatch: pytest.MonkeyPatch,
+    request: _CommandRequest,
+    calls: list[str],
+) -> None:
+    """Dispatch that has the client's wait expire while Live is applying it.
+
+    This is the only sequence that can double-apply: the request is already
+    ``executing`` when the cancellation lands, so the drain cannot stop it and
+    the client is told TIMEOUT for a command that did in fact run.
+    """
+
+    def _dispatch(_backend: Any, name: str, _args: dict[str, Any]) -> dict[str, Any]:
+        calls.append(name)
+        assert _mark_request_timed_out(request) is True
+        return {"note_count": 1}
+
+    monkeypatch.setattr(control_surface_module, "dispatch_command", _dispatch)
+
+
+def _abandoned_request(
+    surface: control_surface_module.AbletonCliRemoteSurface,
+    *,
+    idempotency_key: str | None,
+) -> _CommandRequest:
+    request = _CommandRequest(
+        name="add_notes_to_clip",
+        args={"notes": []},
+        timeout_ms=1,
+        event=threading.Event(),
+        idempotency_key=idempotency_key,
+    )
+    surface._queue.put(request)
+    return request
+
+
+def test_retry_after_timeout_with_the_same_key_does_not_apply_twice(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    surface = _make_surface(monkeypatch)
+    calls: list[str] = []
+    request = _abandoned_request(surface, idempotency_key="step-key")
+    _timing_out_mid_dispatch(monkeypatch, request, calls)
+
+    try:
+        surface._drain_requests()
+        assert calls == ["add_notes_to_clip"]
+
+        result = surface._execute_command_from_server_thread(
+            "add_notes_to_clip",
+            {"notes": []},
+            {"request_timeout_ms": 1000, "idempotency_key": "step-key"},
+        )
+
+        assert calls == ["add_notes_to_clip"]
+        assert result == {"note_count": 1, "idempotent_replay": True}
+    finally:
+        surface.disconnect()
+
+
+def test_retry_after_timeout_without_a_key_still_applies_twice(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    surface = _make_surface(monkeypatch)
+    calls: list[str] = []
+    request = _abandoned_request(surface, idempotency_key=None)
+    _timing_out_mid_dispatch(monkeypatch, request, calls)
+
+    try:
+        surface._drain_requests()
+
+        surface._execute_command_from_server_thread(
+            "add_notes_to_clip",
+            {"notes": []},
+            {"request_timeout_ms": 1000},
+        )
+
+        # Without a key the Remote Script cannot recognise the resend. This is
+        # why `_execute_step` always carries one across a retry.
+        assert calls == ["add_notes_to_clip", "add_notes_to_clip"]
+    finally:
+        surface.disconnect()
diff --git a/tests/test_live_backend.py b/tests/test_live_backend.py
index 003bac0..28cc0ce 100644
--- a/tests/test_live_backend.py
+++ b/tests/test_live_backend.py
@@ -1076,7 +1076,7 @@ def test_live_backend_ping_info_reports_api_support_matrix() -> None:
 
     result = backend.ping_info()
 
-    assert result["protocol_version"] == 2
+    assert result["protocol_version"] == 3
     assert "remote_script_version" in result
     assert result["api_support"] == {
         "song_new_supported": False,
diff --git a/tests/test_protocol.py b/tests/test_protocol.py
index ae874b8..9a6a04b 100644
--- a/tests/test_protocol.py
+++ b/tests/test_protocol.py
@@ -201,3 +201,45 @@ def test_parse_response_accepts_error_details_object() -> None:
     assert response.ok is False
     assert response.error is not None
     assert response.error["details"] == {"failed_step_index": 1}
+
+
+def test_make_request_omits_idempotency_key_by_default() -> None:
+    request = make_request(name="ping", args={}, protocol_version=3)
+
+    assert "idempotency_key" not in request.meta
+
+
+def test_make_request_never_generates_its_own_idempotency_key() -> None:
+    first = make_request(name="ping", args={}, protocol_version=3, idempotency_key="step-key")
+    second = make_request(name="ping", args={}, protocol_version=3, idempotency_key="step-key")
+
+    # The caller owning the retry loop supplies the key, so it is stable across
+    # attempts while request_id is not.
+    assert first.meta["idempotency_key"] == second.meta["idempotency_key"] == "step-key"
+    assert first.request_id != second.request_id
+
+
+def test_make_request_does_not_mutate_the_caller_meta() -> None:
+    meta: dict[str, object] = {"request_timeout_ms": 15000}
+
+    make_request(name="ping", args={}, protocol_version=3, meta=meta, idempotency_key="step-key")
+
+    assert meta == {"request_timeout_ms": 15000}
+
+
+@pytest.mark.parametrize("value", ["", "x" * 129])
+def test_make_request_rejects_an_invalid_idempotency_key(value: str) -> None:
+    with pytest.raises(AppError) as exc_info:
+        make_request(name="ping", args={}, protocol_version=3, idempotency_key=value)
+
+    assert exc_info.value.error_code == "INVALID_ARGUMENT"
+    assert exc_info.value.exit_code == ExitCode.INVALID_ARGUMENT
+
+
+def test_idempotency_key_bound_matches_the_remote_script() -> None:
+    from ableton_cli.client.protocol import IDEMPOTENCY_KEY_MAX_LENGTH
+    from remote_script.AbletonCliRemote.server import (
+        IDEMPOTENCY_KEY_MAX_LENGTH as REMOTE_MAX_LENGTH,
+    )
+
+    assert IDEMPOTENCY_KEY_MAX_LENGTH == REMOTE_MAX_LENGTH
diff --git a/tests/test_remote_control_surface.py b/tests/test_remote_control_surface.py
index a8769ec..a804128 100644
--- a/tests/test_remote_control_surface.py
+++ b/tests/test_remote_control_surface.py
@@ -401,3 +401,123 @@ def test_disconnect_drains_the_queue_regardless_of_budget(
 
     assert executed == [f"command-{index}" for index in range(5)]
     assert surface._queue.empty()
+
+
+def _enqueue_keyed(
+    surface: control_surface_module.AbletonCliRemoteSurface,
+    name: str,
+    key: str | None,
+) -> control_surface_module._CommandRequest:
+    request = control_surface_module._CommandRequest(
+        name=name,
+        args={},
+        timeout_ms=1000,
+        event=threading.Event(),
+        idempotency_key=key,
+    )
+    surface._queue.put(request)
+    return request
+
+
+def _count_dispatches(monkeypatch: pytest.MonkeyPatch, result: dict[str, Any]) -> list[str]:
+    dispatched: list[str] = []
+
+    def _dispatch(_backend: Any, name: str, _args: dict[str, Any]) -> dict[str, Any]:
+        dispatched.append(name)
+        return dict(result)
+
+    monkeypatch.setattr(control_surface_module, "dispatch_command", _dispatch)
+    return dispatched
+
+
+def test_repeating_an_idempotency_key_replays_instead_of_dispatching(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    surface = _make_surface(monkeypatch)
+    dispatched = _count_dispatches(monkeypatch, {"tempo": 120.0})
+
+    first = _enqueue_keyed(surface, "song_info", "key-1")
+    surface._drain_requests()
+    second = _enqueue_keyed(surface, "song_info", "key-1")
+    surface._drain_requests()
+
+    assert dispatched == ["song_info"]
+    assert first.result == {"tempo": 120.0}
+    assert second.result == {"tempo": 120.0, "idempotent_replay": True}
+
+
+def test_requests_without_a_key_are_always_dispatched(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    surface = _make_surface(monkeypatch)
+    dispatched = _count_dispatches(monkeypatch, {"tempo": 120.0})
+
+    _enqueue_keyed(surface, "song_info", None)
+    _enqueue_keyed(surface, "song_info", None)
+    surface._drain_requests()
+
+    assert dispatched == ["song_info", "song_info"]
+
+
+def test_a_failed_command_is_replayed_as_the_same_failure(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from remote_script.AbletonCliRemote.command_backend import CommandError
+
+    surface = _make_surface(monkeypatch)
+    dispatched: list[str] = []
+
+    def _dispatch(_backend: Any, name: str, _args: dict[str, Any]) -> dict[str, Any]:
+        dispatched.append(name)
+        raise CommandError(code="INVALID_ARGUMENT", message="bad track", hint="fix it")
+
+    monkeypatch.setattr(control_surface_module, "dispatch_command", _dispatch)
+
+    _enqueue_keyed(surface, "track_volume_set", "key-1")
+    surface._drain_requests()
+    second = _enqueue_keyed(surface, "track_volume_set", "key-1")
+    surface._drain_requests()
+
+    assert dispatched == ["track_volume_set"]
+    assert isinstance(second.error, CommandError)
+    assert second.error.code == "INVALID_ARGUMENT"
+    assert second.error.message == "bad track"
+    assert second.error.details == {"idempotent_replay": True}
+
+
+def test_cancelled_requests_leave_nothing_to_replay(monkeypatch: pytest.MonkeyPatch) -> None:
+    surface = _make_surface(monkeypatch)
+    dispatched = _count_dispatches(monkeypatch, {"tempo": 120.0})
+
+    cancelled = _enqueue_keyed(surface, "song_info", "key-1")
+    cancelled.cancelled = True
+    surface._drain_requests()
+    assert dispatched == []
+
+    retry = _enqueue_keyed(surface, "song_info", "key-1")
+    surface._drain_requests()
+
+    assert dispatched == ["song_info"]
+    assert retry.result == {"tempo": 120.0}
+
+
+def test_response_cache_evicts_the_oldest_keys(monkeypatch: pytest.MonkeyPatch) -> None:
+    surface = _make_surface(monkeypatch)
+    monkeypatch.setattr(control_surface_module, "IDEMPOTENCY_CACHE_SIZE", 3)
+    dispatched = _count_dispatches(monkeypatch, {"tempo": 120.0})
+
+    for index in range(4):
+        _enqueue_keyed(surface, "song_info", f"key-{index}")
+    surface._drain_requests()
+    assert len(dispatched) == 4
+    assert list(surface._response_cache) == ["key-1", "key-2", "key-3"]
+
+    # key-0 fell out of the cache, so its retry has to run again.
+    evicted_retry = _enqueue_keyed(surface, "song_info", "key-0")
+    # key-3 is still cached, so its retry is replayed.
+    cached_retry = _enqueue_keyed(surface, "song_info", "key-3")
+    surface._drain_requests()
+
+    assert len(dispatched) == 5
+    assert evicted_retry.result == {"tempo": 120.0}
+    assert cached_retry.result == {"tempo": 120.0, "idempotent_replay": True}
diff --git a/tests/test_remote_events.py b/tests/test_remote_events.py
index 087dd64..9b28e21 100644
--- a/tests/test_remote_events.py
+++ b/tests/test_remote_events.py
@@ -194,7 +194,7 @@ def _subscribe_request(events: list[str]) -> bytes:
                 "args": {"events": events},
                 "meta": {},
                 "request_id": "sub-1",
-                "protocol_version": 2,
+                "protocol_version": 3,
             }
         )
         + "\n"
@@ -224,7 +224,7 @@ def test_server_streams_published_events_to_a_subscriber() -> None:
             line = json.loads(stream.readline().decode("utf-8"))
             assert line == {
                 "type": "event",
-                "protocol_version": 2,
+                "protocol_version": 3,
                 "event": "tempo",
                 "ts": 3.0,
                 "data": {"tempo": 90.0},
diff --git a/tests/test_remote_server.py b/tests/test_remote_server.py
index 0666f92..612ddad 100644
--- a/tests/test_remote_server.py
+++ b/tests/test_remote_server.py
@@ -23,7 +23,7 @@ def test_parse_command_request_accepts_strict_protocol_shape() -> None:
             "args": {},
             "meta": {"request_timeout_ms": 15000},
             "request_id": "request-1",
-            "protocol_version": 2,
+            "protocol_version": 3,
         }
     )
 
@@ -40,7 +40,7 @@ def test_parse_command_request_accepts_strict_protocol_shape() -> None:
             "args": {},
             "meta": {},
             "request_id": "request-1",
-            "protocol_version": 2,
+            "protocol_version": 3,
             "extra": "not allowed",
         },
         {
@@ -49,7 +49,7 @@ def test_parse_command_request_accepts_strict_protocol_shape() -> None:
             "args": {},
             "meta": {},
             "request_id": "request-1",
-            "protocol_version": 2,
+            "protocol_version": 3,
         },
         {
             "type": "command",
@@ -57,7 +57,7 @@ def test_parse_command_request_accepts_strict_protocol_shape() -> None:
             "args": [],
             "meta": {},
             "request_id": "request-1",
-            "protocol_version": 2,
+            "protocol_version": 3,
         },
         {
             "type": "command",
@@ -171,3 +171,28 @@ def _executor(name: str, args: dict[str, Any], meta: dict[str, Any]) -> dict[str
 
     assert first["error"]["code"] == "INVALID_ARGUMENT"
     assert trailing == b""
+
+
+def _command_payload(meta: dict[str, object]) -> dict[str, object]:
+    return {
+        "type": "command",
+        "name": "song_info",
+        "args": {},
+        "meta": meta,
+        "request_id": "request-1",
+        "protocol_version": 3,
+    }
+
+
+def test_parse_command_request_accepts_an_idempotency_key() -> None:
+    request = _parse_command_request(_command_payload({"idempotency_key": "step-key"}))
+
+    assert request[4] == {"idempotency_key": "step-key"}
+
+
+@pytest.mark.parametrize("key", ["", 42, None, "x" * 129])
+def test_parse_command_request_rejects_an_invalid_idempotency_key(key: object) -> None:
+    with pytest.raises(CommandExecutionError) as exc_info:
+        _parse_command_request(_command_payload({"idempotency_key": key}))
+
+    assert exc_info.value.code == "INVALID_ARGUMENT"
diff --git a/tests/test_transport_record_replay.py b/tests/test_transport_record_replay.py
index dc303c6..33829b9 100644
--- a/tests/test_transport_record_replay.py
+++ b/tests/test_transport_record_replay.py
@@ -128,3 +128,37 @@ def send(self, payload: dict[str, Any]) -> dict[str, Any]:
     assert entry["request"] == request_payload
     assert entry["response"] is None
     assert entry["error"]["error_code"] == "TIMEOUT"
+
+
+def test_replay_matching_ignores_the_idempotency_key(tmp_path: Path) -> None:
+    replay_path = tmp_path / "replay.jsonl"
+    _write_jsonl(
+        replay_path,
+        [
+            {
+                "request": {"name": "song_info", "args": {}},
+                "response": {
+                    "ok": True,
+                    "request_id": "recorded-id",
+                    "protocol_version": 3,
+                    "result": {"tempo": 120.0},
+                    "error": None,
+                },
+            }
+        ],
+    )
+    transport = ReplayTransport(path=str(replay_path))
+
+    # Fixtures are keyed on name+args, so a per-attempt idempotency key in meta
+    # must not make a recorded request unmatchable.
+    response = transport.send(
+        {
+            "name": "song_info",
+            "args": {},
+            "meta": {"idempotency_key": "step-key"},
+            "request_id": "runtime-id",
+            "protocol_version": 3,
+        }
+    )
+
+    assert response["result"] == {"tempo": 120.0}