diff --git a/.claude/tools/amplihack/remote/cli.py b/.claude/tools/amplihack/remote/cli.py
index de8ada0cf..0d524043e 100644
--- a/.claude/tools/amplihack/remote/cli.py
+++ b/.claude/tools/amplihack/remote/cli.py
@@ -53,6 +53,7 @@ def remote_cli():
 @click.option("--no-reuse", is_flag=True, help="Always provision fresh VM")
 @click.option("--timeout", default=120, type=int, help="Max execution time in minutes")
 @click.option("--region", default=None, help="Azure region")
+@click.option("--port", default=None, type=int, help="Reuse existing bastion tunnel on this local port")
 @click.argument("azlin_args", nargs=-1, type=click.UNPROCESSED)
 def remote_execute(
     command: str,
@@ -64,6 +65,7 @@ def remote_execute(
     no_reuse: bool,
     timeout: int,
     region: str | None,
+    port: int | None,
     azlin_args: tuple,
 ):
     """Execute amplihack command on remote Azure VM.
@@ -103,6 +105,7 @@ def remote_execute(
         no_reuse=no_reuse,
         keep_vm=keep_vm,
         azlin_extra_args=list(azlin_args) if azlin_args else None,
+        tunnel_port=port,
     )
 
     # Execute with progress reporting
@@ -184,7 +187,7 @@ def execute_remote_workflow(
 
             # Step 4: Transfer context
             click.echo("\n[4/7] Transferring context...")
-            executor = Executor(vm, timeout_minutes=timeout)
+            executor = Executor(vm, timeout_minutes=timeout, tunnel_port=vm_options.tunnel_port)
             executor.transfer_context(archive_path)
             click.echo("  \u2713 Context transferred")
 
@@ -403,7 +406,8 @@ def cmd_list(status: str | None, output_json: bool):
     help="VM size tier (s=1, m=2, l=4, xl=8 sessions) [default: l]",
 )
 @click.option("--region", default=None, help="Azure region")
-def cmd_start(prompts: tuple, command: str, max_turns: int, size: str, region: str | None):
+@click.option("--port", default=None, type=int, help="Reuse existing bastion tunnel on this local port")
+def cmd_start(prompts: tuple, command: str, max_turns: int, size: str, region: str | None, port: int | None):
     """Start one or more detached remote sessions.
 
     Usage: amplihack remote start [options] "<prompt1>" "<prompt2>" ...
@@ -501,7 +505,7 @@ def cmd_start(prompts: tuple, command: str, max_turns: int, size: str, region: s
 
                     # Step 3: Transfer context
                     click.echo("  → Transferring context...")
-                    executor = Executor(vm)
+                    executor = Executor(vm, tunnel_port=port)
                     executor.transfer_context(archive_path)
                     click.echo("  ✓ Context transferred")
 
diff --git a/.claude/tools/amplihack/remote/executor.py b/.claude/tools/amplihack/remote/executor.py
index 08a35f99a..94f4e7ea1 100644
--- a/.claude/tools/amplihack/remote/executor.py
+++ b/.claude/tools/amplihack/remote/executor.py
@@ -34,16 +34,18 @@ class Executor:
     and output capture.
     """
 
-    def __init__(self, vm: VM, timeout_minutes: int = 120):
+    def __init__(self, vm: VM, timeout_minutes: int = 120, tunnel_port: int | None = None):
         """Initialize executor.
 
         Args:
             vm: Target VM for execution
             timeout_minutes: Maximum execution time (default: 120)
+            tunnel_port: Local port of existing bastion tunnel to reuse (default: None)
         """
         self.vm = vm
         self.timeout_seconds = timeout_minutes * 60
         self.remote_workspace = "~/workspace"
+        self.tunnel_port = tunnel_port
 
     @staticmethod
     def _encode_b64(text: str) -> str:
@@ -59,6 +61,12 @@ def _encode_b64(text: str) -> str:
         """
         return base64.b64encode(text.encode()).decode()
 
+    def _azlin_port_args(self) -> list[str]:
+        """Return --port flag args if tunnel_port is set, else empty list."""
+        if self.tunnel_port is not None:
+            return ["--port", str(self.tunnel_port)]
+        return []
+
     def transfer_context(self, archive_path: Path) -> bool:
         """Transfer context archive to remote VM.
 
@@ -93,7 +101,7 @@ def transfer_context(self, archive_path: Path) -> bool:
                 start_time = time.time()
 
                 subprocess.run(
-                    ["azlin", "cp", archive_name, remote_path],
+                    ["azlin", "cp", *self._azlin_port_args(), archive_name, remote_path],
                     cwd=str(archive_dir),  # Run from archive directory
                     capture_output=True,
                     text=True,
@@ -230,7 +238,7 @@ def execute_remote(
 
         try:
             result = subprocess.run(
-                ["azlin", "connect", self.vm.name, setup_and_run],
+                ["azlin", "connect", *self._azlin_port_args(), self.vm.name, setup_and_run],
                 capture_output=True,
                 text=True,
                 timeout=self.timeout_seconds,
@@ -259,7 +267,7 @@ def execute_remote(
             # Try to terminate remote process
             try:
                 subprocess.run(
-                    ["azlin", "connect", self.vm.name, "pkill -TERM -f amplihack"],
+                    ["azlin", "connect", *self._azlin_port_args(), self.vm.name, "pkill -TERM -f amplihack"],
                     timeout=30,
                     capture_output=True,
                 )
@@ -314,7 +322,7 @@ def retrieve_logs(self, local_dest: Path) -> bool:
         try:
             # Create archive
             subprocess.run(
-                ["azlin", "connect", self.vm.name, create_archive],
+                ["azlin", "connect", *self._azlin_port_args(), self.vm.name, create_archive],
                 capture_output=True,
                 text=True,
                 timeout=60,
@@ -324,7 +332,7 @@ def retrieve_logs(self, local_dest: Path) -> bool:
             # Download archive (azlin cp requires relative paths)
             local_archive = local_dest / "logs.tar.gz"
             subprocess.run(
-                ["azlin", "cp", f"{self.vm.name}:~/logs.tar.gz", "logs.tar.gz"],
+                ["azlin", "cp", *self._azlin_port_args(), f"{self.vm.name}:~/logs.tar.gz", "logs.tar.gz"],
                 cwd=str(local_dest),  # Run from destination directory
                 capture_output=True,
                 text=True,
@@ -377,7 +385,7 @@ def retrieve_git_state(self, local_dest: Path) -> bool:
         try:
             # Create bundle
             subprocess.run(
-                ["azlin", "connect", self.vm.name, create_bundle],
+                ["azlin", "connect", *self._azlin_port_args(), self.vm.name, create_bundle],
                 capture_output=True,
                 text=True,
                 timeout=300,
@@ -387,7 +395,7 @@ def retrieve_git_state(self, local_dest: Path) -> bool:
             # Download bundle (azlin cp requires relative paths)
             local_bundle = local_dest / "results.bundle"
             subprocess.run(
-                ["azlin", "cp", f"{self.vm.name}:~/results.bundle", "results.bundle"],
+                ["azlin", "cp", *self._azlin_port_args(), f"{self.vm.name}:~/results.bundle", "results.bundle"],
                 cwd=str(local_dest),  # Run from destination directory
                 capture_output=True,
                 text=True,
@@ -536,7 +544,7 @@ def execute_remote_tmux(
 
         try:
             result = subprocess.run(
-                ["azlin", "connect", self.vm.name, setup_and_run],
+                ["azlin", "connect", *self._azlin_port_args(), self.vm.name, setup_and_run],
                 capture_output=True,
                 text=True,
                 timeout=600,  # 10 minutes for setup
@@ -588,7 +596,7 @@ def check_tmux_status(self, session_id: str) -> str:
 
         try:
             result = subprocess.run(
-                ["azlin", "connect", self.vm.name, check_command],
+                ["azlin", "connect", *self._azlin_port_args(), self.vm.name, check_command],
                 capture_output=True,
                 text=True,
                 timeout=30,
diff --git a/.claude/tools/amplihack/remote/orchestrator.py b/.claude/tools/amplihack/remote/orchestrator.py
index 0355b9159..53d36e29a 100644
--- a/.claude/tools/amplihack/remote/orchestrator.py
+++ b/.claude/tools/amplihack/remote/orchestrator.py
@@ -42,6 +42,7 @@ class VMOptions:
     no_reuse: bool = False
     keep_vm: bool = False
     azlin_extra_args: list | None = None  # Pass-through for any azlin parameters
+    tunnel_port: int | None = None  # Reuse existing bastion tunnel on this local port
 
 
 class Orchestrator:
diff --git a/.claude/tools/amplihack/remote/session.py b/.claude/tools/amplihack/remote/session.py
index dbe7379a0..df5ec7b0c 100644
--- a/.claude/tools/amplihack/remote/session.py
+++ b/.claude/tools/amplihack/remote/session.py
@@ -447,7 +447,9 @@ def check_session_status(self, session_id: str) -> SessionStatus:
 
         return session.status
 
-    def _execute_ssh_command(self, vm_name: str, command: str) -> str:
+    def _execute_ssh_command(
+        self, vm_name: str, command: str, tunnel_port: int | None = None
+    ) -> str:
         """Execute command on remote VM via SSH.
 
         Uses azlin for SSH connectivity.
@@ -455,13 +457,15 @@ def _execute_ssh_command(self, vm_name: str, command: str) -> str:
         Args:
             vm_name: Name of the VM
             command: Command to execute
+            tunnel_port: Local port of existing bastion tunnel to reuse (default: None)
 
         Returns:
             Command output as string
         """
+        port_args = ["--port", str(tunnel_port)] if tunnel_port is not None else []
         try:
             result = subprocess.run(
-                ["azlin", "ssh", vm_name, "--", command],
+                ["azlin", "ssh", *port_args, vm_name, "--", command],
                 capture_output=True,
                 text=True,
                 timeout=30,
diff --git a/amplifier-bundle/tools/amplihack/remote/cli.py b/amplifier-bundle/tools/amplihack/remote/cli.py
index de8ada0cf..0d524043e 100644
--- a/amplifier-bundle/tools/amplihack/remote/cli.py
+++ b/amplifier-bundle/tools/amplihack/remote/cli.py
@@ -53,6 +53,7 @@ def remote_cli():
 @click.option("--no-reuse", is_flag=True, help="Always provision fresh VM")
 @click.option("--timeout", default=120, type=int, help="Max execution time in minutes")
 @click.option("--region", default=None, help="Azure region")
+@click.option("--port", default=None, type=int, help="Reuse existing bastion tunnel on this local port")
 @click.argument("azlin_args", nargs=-1, type=click.UNPROCESSED)
 def remote_execute(
     command: str,
@@ -64,6 +65,7 @@ def remote_execute(
     no_reuse: bool,
     timeout: int,
     region: str | None,
+    port: int | None,
     azlin_args: tuple,
 ):
     """Execute amplihack command on remote Azure VM.
@@ -103,6 +105,7 @@ def remote_execute(
         no_reuse=no_reuse,
         keep_vm=keep_vm,
         azlin_extra_args=list(azlin_args) if azlin_args else None,
+        tunnel_port=port,
     )
 
     # Execute with progress reporting
@@ -184,7 +187,7 @@ def execute_remote_workflow(
 
             # Step 4: Transfer context
             click.echo("\n[4/7] Transferring context...")
-            executor = Executor(vm, timeout_minutes=timeout)
+            executor = Executor(vm, timeout_minutes=timeout, tunnel_port=vm_options.tunnel_port)
             executor.transfer_context(archive_path)
             click.echo("  \u2713 Context transferred")
 
@@ -403,7 +406,8 @@ def cmd_list(status: str | None, output_json: bool):
     help="VM size tier (s=1, m=2, l=4, xl=8 sessions) [default: l]",
 )
 @click.option("--region", default=None, help="Azure region")
-def cmd_start(prompts: tuple, command: str, max_turns: int, size: str, region: str | None):
+@click.option("--port", default=None, type=int, help="Reuse existing bastion tunnel on this local port")
+def cmd_start(prompts: tuple, command: str, max_turns: int, size: str, region: str | None, port: int | None):
     """Start one or more detached remote sessions.
 
     Usage: amplihack remote start [options] "<prompt1>" "<prompt2>" ...
@@ -501,7 +505,7 @@ def cmd_start(prompts: tuple, command: str, max_turns: int, size: str, region: s
 
                     # Step 3: Transfer context
                     click.echo("  → Transferring context...")
-                    executor = Executor(vm)
+                    executor = Executor(vm, tunnel_port=port)
                     executor.transfer_context(archive_path)
                     click.echo("  ✓ Context transferred")
 
diff --git a/amplifier-bundle/tools/amplihack/remote/executor.py b/amplifier-bundle/tools/amplihack/remote/executor.py
index 08a35f99a..94f4e7ea1 100644
--- a/amplifier-bundle/tools/amplihack/remote/executor.py
+++ b/amplifier-bundle/tools/amplihack/remote/executor.py
@@ -34,16 +34,18 @@ class Executor:
     and output capture.
     """
 
-    def __init__(self, vm: VM, timeout_minutes: int = 120):
+    def __init__(self, vm: VM, timeout_minutes: int = 120, tunnel_port: int | None = None):
         """Initialize executor.
 
         Args:
             vm: Target VM for execution
             timeout_minutes: Maximum execution time (default: 120)
+            tunnel_port: Local port of existing bastion tunnel to reuse (default: None)
         """
         self.vm = vm
         self.timeout_seconds = timeout_minutes * 60
         self.remote_workspace = "~/workspace"
+        self.tunnel_port = tunnel_port
 
     @staticmethod
     def _encode_b64(text: str) -> str:
@@ -59,6 +61,12 @@ def _encode_b64(text: str) -> str:
         """
         return base64.b64encode(text.encode()).decode()
 
+    def _azlin_port_args(self) -> list[str]:
+        """Return --port flag args if tunnel_port is set, else empty list."""
+        if self.tunnel_port is not None:
+            return ["--port", str(self.tunnel_port)]
+        return []
+
     def transfer_context(self, archive_path: Path) -> bool:
         """Transfer context archive to remote VM.
 
@@ -93,7 +101,7 @@ def transfer_context(self, archive_path: Path) -> bool:
                 start_time = time.time()
 
                 subprocess.run(
-                    ["azlin", "cp", archive_name, remote_path],
+                    ["azlin", "cp", *self._azlin_port_args(), archive_name, remote_path],
                     cwd=str(archive_dir),  # Run from archive directory
                     capture_output=True,
                     text=True,
@@ -230,7 +238,7 @@ def execute_remote(
 
         try:
             result = subprocess.run(
-                ["azlin", "connect", self.vm.name, setup_and_run],
+                ["azlin", "connect", *self._azlin_port_args(), self.vm.name, setup_and_run],
                 capture_output=True,
                 text=True,
                 timeout=self.timeout_seconds,
@@ -259,7 +267,7 @@ def execute_remote(
             # Try to terminate remote process
             try:
                 subprocess.run(
-                    ["azlin", "connect", self.vm.name, "pkill -TERM -f amplihack"],
+                    ["azlin", "connect", *self._azlin_port_args(), self.vm.name, "pkill -TERM -f amplihack"],
                     timeout=30,
                     capture_output=True,
                 )
@@ -314,7 +322,7 @@ def retrieve_logs(self, local_dest: Path) -> bool:
         try:
             # Create archive
             subprocess.run(
-                ["azlin", "connect", self.vm.name, create_archive],
+                ["azlin", "connect", *self._azlin_port_args(), self.vm.name, create_archive],
                 capture_output=True,
                 text=True,
                 timeout=60,
@@ -324,7 +332,7 @@ def retrieve_logs(self, local_dest: Path) -> bool:
             # Download archive (azlin cp requires relative paths)
             local_archive = local_dest / "logs.tar.gz"
             subprocess.run(
-                ["azlin", "cp", f"{self.vm.name}:~/logs.tar.gz", "logs.tar.gz"],
+                ["azlin", "cp", *self._azlin_port_args(), f"{self.vm.name}:~/logs.tar.gz", "logs.tar.gz"],
                 cwd=str(local_dest),  # Run from destination directory
                 capture_output=True,
                 text=True,
@@ -377,7 +385,7 @@ def retrieve_git_state(self, local_dest: Path) -> bool:
         try:
             # Create bundle
             subprocess.run(
-                ["azlin", "connect", self.vm.name, create_bundle],
+                ["azlin", "connect", *self._azlin_port_args(), self.vm.name, create_bundle],
                 capture_output=True,
                 text=True,
                 timeout=300,
@@ -387,7 +395,7 @@ def retrieve_git_state(self, local_dest: Path) -> bool:
             # Download bundle (azlin cp requires relative paths)
             local_bundle = local_dest / "results.bundle"
             subprocess.run(
-                ["azlin", "cp", f"{self.vm.name}:~/results.bundle", "results.bundle"],
+                ["azlin", "cp", *self._azlin_port_args(), f"{self.vm.name}:~/results.bundle", "results.bundle"],
                 cwd=str(local_dest),  # Run from destination directory
                 capture_output=True,
                 text=True,
@@ -536,7 +544,7 @@ def execute_remote_tmux(
 
         try:
             result = subprocess.run(
-                ["azlin", "connect", self.vm.name, setup_and_run],
+                ["azlin", "connect", *self._azlin_port_args(), self.vm.name, setup_and_run],
                 capture_output=True,
                 text=True,
                 timeout=600,  # 10 minutes for setup
@@ -588,7 +596,7 @@ def check_tmux_status(self, session_id: str) -> str:
 
         try:
             result = subprocess.run(
-                ["azlin", "connect", self.vm.name, check_command],
+                ["azlin", "connect", *self._azlin_port_args(), self.vm.name, check_command],
                 capture_output=True,
                 text=True,
                 timeout=30,
diff --git a/amplifier-bundle/tools/amplihack/remote/orchestrator.py b/amplifier-bundle/tools/amplihack/remote/orchestrator.py
index 0355b9159..53d36e29a 100644
--- a/amplifier-bundle/tools/amplihack/remote/orchestrator.py
+++ b/amplifier-bundle/tools/amplihack/remote/orchestrator.py
@@ -42,6 +42,7 @@ class VMOptions:
     no_reuse: bool = False
     keep_vm: bool = False
     azlin_extra_args: list | None = None  # Pass-through for any azlin parameters
+    tunnel_port: int | None = None  # Reuse existing bastion tunnel on this local port
 
 
 class Orchestrator:
diff --git a/amplifier-bundle/tools/amplihack/remote/session.py b/amplifier-bundle/tools/amplihack/remote/session.py
index dbe7379a0..df5ec7b0c 100644
--- a/amplifier-bundle/tools/amplihack/remote/session.py
+++ b/amplifier-bundle/tools/amplihack/remote/session.py
@@ -447,7 +447,9 @@ def check_session_status(self, session_id: str) -> SessionStatus:
 
         return session.status
 
-    def _execute_ssh_command(self, vm_name: str, command: str) -> str:
+    def _execute_ssh_command(
+        self, vm_name: str, command: str, tunnel_port: int | None = None
+    ) -> str:
         """Execute command on remote VM via SSH.
 
         Uses azlin for SSH connectivity.
@@ -455,13 +457,15 @@ def _execute_ssh_command(self, vm_name: str, command: str) -> str:
         Args:
             vm_name: Name of the VM
             command: Command to execute
+            tunnel_port: Local port of existing bastion tunnel to reuse (default: None)
 
         Returns:
             Command output as string
         """
+        port_args = ["--port", str(tunnel_port)] if tunnel_port is not None else []
         try:
             result = subprocess.run(
-                ["azlin", "ssh", vm_name, "--", command],
+                ["azlin", "ssh", *port_args, vm_name, "--", command],
                 capture_output=True,
                 text=True,
                 timeout=30,
diff --git a/pyproject.toml b/pyproject.toml
index 4a6fc6fb1..2d2cecdb8 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -5,7 +5,7 @@ backend-path = ["."]
 
 [project]
 name = "amplihack"
-version = "0.5.117"
+version = "0.5.118"
 description = "Amplifier bundle for agentic coding with comprehensive skills, recipes, and workflows"
 requires-python = ">=3.11"
 dependencies = [
diff --git a/tests/outside_in/conftest.py b/tests/outside_in/conftest.py
new file mode 100644
index 000000000..03271c4a8
--- /dev/null
+++ b/tests/outside_in/conftest.py
@@ -0,0 +1,11 @@
+"""Pytest configuration for outside-in tests.
+
+Adds .claude/tools to sys.path so tests can import the remote module.
+"""
+import sys
+from pathlib import Path
+
+# Add .claude/tools to path so amplihack.remote is importable
+_tools_path = str(Path(__file__).parent.parent.parent / ".claude" / "tools")
+if _tools_path not in sys.path:
+    sys.path.insert(0, _tools_path)
diff --git a/tests/outside_in/test_bastion_tunnel_reuse.py b/tests/outside_in/test_bastion_tunnel_reuse.py
new file mode 100644
index 000000000..753190bc1
--- /dev/null
+++ b/tests/outside_in/test_bastion_tunnel_reuse.py
@@ -0,0 +1,260 @@
+"""Outside-in tests for bastion tunnel reuse via azlin --port flag.
+
+Tests the feature from a user's perspective:
+- The --port flag is accepted by both 'exec' and 'start' CLI commands
+- When --port is provided, azlin commands include --port <N> args
+- When --port is omitted, azlin commands work as before (no regression)
+- Executor accepts tunnel_port and passes it to all azlin subcommands
+- VMOptions stores tunnel_port correctly
+"""
+
+from datetime import datetime
+from pathlib import Path
+from unittest.mock import Mock, call, patch
+
+import pytest
+from click.testing import CliRunner
+
+from amplihack.remote import cli as cli_module
+from amplihack.remote.cli import remote_cli
+from amplihack.remote.executor import Executor
+from amplihack.remote.orchestrator import VM, VMOptions
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_vm(name: str = "test-vm") -> VM:
+    return VM(name=name, size="Standard_D2s_v3", region="eastus")
+
+
+# ---------------------------------------------------------------------------
+# VMOptions – tunnel_port field
+# ---------------------------------------------------------------------------
+
+
+def test_vm_options_default_no_tunnel_port():
+    """VMOptions tunnel_port defaults to None."""
+    opts = VMOptions()
+    assert opts.tunnel_port is None
+
+
+def test_vm_options_accepts_tunnel_port():
+    """VMOptions stores tunnel_port when provided."""
+    opts = VMOptions(tunnel_port=2222)
+    assert opts.tunnel_port == 2222
+
+
+# ---------------------------------------------------------------------------
+# Executor – _azlin_port_args helper
+# ---------------------------------------------------------------------------
+
+
+def test_executor_port_args_none():
+    """When tunnel_port is None, _azlin_port_args returns empty list."""
+    executor = Executor(_make_vm(), tunnel_port=None)
+    assert executor._azlin_port_args() == []
+
+
+def test_executor_port_args_set():
+    """When tunnel_port is set, _azlin_port_args returns ['--port', '<N>']."""
+    executor = Executor(_make_vm(), tunnel_port=2222)
+    assert executor._azlin_port_args() == ["--port", "2222"]
+
+
+# ---------------------------------------------------------------------------
+# Executor – azlin commands include --port when tunnel_port set
+# ---------------------------------------------------------------------------
+
+
+def test_executor_transfer_context_passes_port(tmp_path):
+    """transfer_context passes --port to azlin cp when tunnel_port is set."""
+    archive = tmp_path / "context.tar.gz"
+    archive.write_bytes(b"fake archive")
+
+    executor = Executor(_make_vm("my-vm"), tunnel_port=3333)
+
+    with patch("subprocess.run") as mock_run:
+        mock_run.return_value = Mock(returncode=0, stdout="", stderr="")
+        executor.transfer_context(archive)
+
+    cmd = mock_run.call_args[0][0]
+    assert cmd[0] == "azlin"
+    assert cmd[1] == "cp"
+    assert "--port" in cmd
+    assert "3333" in cmd
+
+
+def test_executor_transfer_context_no_port(tmp_path):
+    """transfer_context does NOT pass --port when tunnel_port is None."""
+    archive = tmp_path / "context.tar.gz"
+    archive.write_bytes(b"fake archive")
+
+    executor = Executor(_make_vm("my-vm"), tunnel_port=None)
+
+    with patch("subprocess.run") as mock_run:
+        mock_run.return_value = Mock(returncode=0, stdout="", stderr="")
+        executor.transfer_context(archive)
+
+    cmd = mock_run.call_args[0][0]
+    assert "--port" not in cmd
+
+
+def test_executor_check_tmux_status_passes_port():
+    """check_tmux_status passes --port to azlin connect when tunnel_port is set."""
+    executor = Executor(_make_vm("my-vm"), tunnel_port=4444)
+
+    with patch("subprocess.run") as mock_run:
+        mock_run.return_value = Mock(returncode=0, stdout="running\n", stderr="")
+        result = executor.check_tmux_status("sess-20251202-123456-ab12")
+
+    assert result == "running"
+    cmd = mock_run.call_args[0][0]
+    assert "--port" in cmd
+    assert "4444" in cmd
+
+
+def test_executor_check_tmux_status_no_port():
+    """check_tmux_status does NOT pass --port when tunnel_port is None."""
+    executor = Executor(_make_vm("my-vm"), tunnel_port=None)
+
+    with patch("subprocess.run") as mock_run:
+        mock_run.return_value = Mock(returncode=0, stdout="completed\n", stderr="")
+        result = executor.check_tmux_status("sess-20251202-123456-ab12")
+
+    assert result == "completed"
+    cmd = mock_run.call_args[0][0]
+    assert "--port" not in cmd
+
+
+# ---------------------------------------------------------------------------
+# CLI – exec command accepts --port
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def cli_runner():
+    return CliRunner()
+
+
+def test_exec_command_accepts_port_flag(cli_runner):
+    """The 'exec' CLI command accepts --port flag without error."""
+    with (
+        patch.object(cli_module, "execute_remote_workflow") as mock_workflow,
+    ):
+        result = cli_runner.invoke(
+            remote_cli,
+            ["exec", "auto", "test prompt", "--port", "2222"],
+        )
+
+    # Should not fail with "No such option: --port"
+    assert "--port" not in result.output or "No such option" not in result.output
+    assert result.exit_code == 0 or "No such option" not in result.output
+
+
+def test_exec_command_passes_port_to_vm_options(cli_runner):
+    """The 'exec' CLI command wires --port into VMOptions.tunnel_port."""
+    captured_options = {}
+
+    def capture_workflow(repo_path, command, prompt, max_turns, vm_options, timeout, **kwargs):
+        captured_options["tunnel_port"] = vm_options.tunnel_port
+
+    with patch.object(cli_module, "execute_remote_workflow", side_effect=capture_workflow):
+        result = cli_runner.invoke(
+            remote_cli,
+            ["exec", "auto", "my task", "--port", "5555"],
+        )
+
+    assert captured_options.get("tunnel_port") == 5555
+
+
+def test_exec_command_no_port_gives_none(cli_runner):
+    """The 'exec' CLI command sets tunnel_port=None when --port not given."""
+    captured_options = {}
+
+    def capture_workflow(repo_path, command, prompt, max_turns, vm_options, timeout, **kwargs):
+        captured_options["tunnel_port"] = vm_options.tunnel_port
+
+    with patch.object(cli_module, "execute_remote_workflow", side_effect=capture_workflow):
+        result = cli_runner.invoke(
+            remote_cli,
+            ["exec", "auto", "my task"],
+        )
+
+    assert captured_options.get("tunnel_port") is None
+
+
+# ---------------------------------------------------------------------------
+# CLI – start command accepts --port
+# ---------------------------------------------------------------------------
+
+
+def test_start_command_accepts_port_flag(cli_runner):
+    """The 'start' CLI command accepts --port flag without error."""
+    with (
+        patch.object(cli_module, "SessionManager") as MockSessionMgr,
+        patch.object(cli_module, "VMPoolManager") as MockVMPoolMgr,
+        patch.object(cli_module, "ContextPackager") as MockPackager,
+        patch.object(cli_module, "Executor") as MockExecutor,
+        patch("os.environ", {"ANTHROPIC_API_KEY": "test-key"}),
+    ):
+        mock_session = Mock(session_id="sess-test-001", vm_name="pending")
+        MockSessionMgr.return_value.create_session.return_value = mock_session
+        mock_vm = Mock(name="test-vm", size="Standard_D4s_v3", region="eastus")
+        MockVMPoolMgr.return_value.allocate_vm.return_value = mock_vm
+
+        mock_packager_instance = Mock()
+        mock_packager_instance.scan_secrets.return_value = []
+        mock_archive = Mock(spec=Path)
+        mock_archive.stat.return_value = Mock(st_size=1024 * 1024)
+        mock_packager_instance.package.return_value = mock_archive
+        MockPackager.return_value.__enter__.return_value = mock_packager_instance
+        MockPackager.return_value.__exit__.return_value = None
+
+        result = cli_runner.invoke(
+            remote_cli,
+            ["start", "--port", "2222", "test prompt"],
+            env={"ANTHROPIC_API_KEY": "test-key"},
+        )
+
+    assert "No such option: --port" not in result.output
+
+
+def test_start_command_passes_port_to_executor(cli_runner):
+    """The 'start' CLI command passes --port to Executor as tunnel_port."""
+    captured_tunnel_ports = []
+
+    with (
+        patch.object(cli_module, "SessionManager") as MockSessionMgr,
+        patch.object(cli_module, "VMPoolManager") as MockVMPoolMgr,
+        patch.object(cli_module, "ContextPackager") as MockPackager,
+        patch.object(cli_module, "Executor") as MockExecutor,
+    ):
+        mock_session = Mock(session_id="sess-test-002", vm_name="pending")
+        MockSessionMgr.return_value.create_session.return_value = mock_session
+        mock_vm = Mock(name="test-vm", size="Standard_D4s_v3", region="eastus")
+        MockVMPoolMgr.return_value.allocate_vm.return_value = mock_vm
+
+        mock_packager_instance = Mock()
+        mock_packager_instance.scan_secrets.return_value = []
+        mock_archive = Mock(spec=Path)
+        mock_archive.stat.return_value = Mock(st_size=1024 * 1024)
+        mock_packager_instance.package.return_value = mock_archive
+        MockPackager.return_value.__enter__.return_value = mock_packager_instance
+        MockPackager.return_value.__exit__.return_value = None
+
+        def capture_executor(vm, tunnel_port=None, **kwargs):
+            captured_tunnel_ports.append(tunnel_port)
+            return Mock()
+
+        MockExecutor.side_effect = capture_executor
+
+        cli_runner.invoke(
+            remote_cli,
+            ["start", "--port", "7777", "test prompt"],
+            env={"ANTHROPIC_API_KEY": "test-key"},
+        )
+
+    assert 7777 in captured_tunnel_ports