From ec7a11a1eb09032e0e4f2b27c69c28dceeb6df06 Mon Sep 17 00:00:00 2001
From: Xiaoxukuaipao <2812156857@qq.com>
Date: Tue, 14 Jul 2026 18:14:01 -0700
Subject: [PATCH] Fix detached job lifecycle management

---
 CHANGELOG.md                                  |  13 +
 README.md                                     |  34 +-
 .../autodl-remote/.codex-plugin/plugin.json   |   2 +-
 plugins/autodl-remote/README.md               |  23 +-
 plugins/autodl-remote/bin/autodl-remote       | 509 +++++++++++++--
 .../skills/autodl-remote/SKILL.md             |   8 +-
 tests/test-autodl-remote.sh                   | 595 ++++++++++++++++++
 7 files changed, 1109 insertions(+), 75 deletions(-)
 create mode 100755 tests/test-autodl-remote.sh

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 47687a3..00da8bb 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,18 @@
 # Changelog
 
+## 0.8.1
+
+- Added `job stop <name>` and `run stop <name>` for detached-job lifecycle control. They terminate the isolated runner process group with `TERM` then `KILL`, and record `status=cancelled`.
+- Reject active detached-job name collisions by default. `exec --replace --name <name>` now stops the recorded job before launching its replacement.
+- Fixed detached direct-command argv quoting so literal arguments (including `<|endoftext|>`) are preserved instead of being re-parsed by a shell.
+- Mark a recorded `running` job as `lost` when its runner PID is gone, with a recovery hint; `run list` uses the same status normalization.
+- Made `job tail <name> --lines N` and `run tail <name> --lines N` equivalent to the existing option-first form, and reject unknown trailing arguments.
+- Fixed `job tail --lines N`, `run tail`, and dashboard log previews for tqdm-style carriage-return progress by normalizing logical records and removing ANSI CSI sequences before selecting the final lines.
+- Limited non-following normalized log output to 64 KiB by default so one oversized progress record cannot flood a tool result.
+- Serialized only SSH ControlMaster creation per account/target, added stale socket and lock recovery, and retried a failed initial master connection once.
+- Added explicit SSH connection, monitoring, and keychain/Expect timeout bounds while keeping long-running non-monitoring sessions unbounded.
+- Added deterministic CLI regressions for logical log tails, output bounds, concurrent cold starts, stale connection state, retry behavior, and timeout cleanup.
+
 ## 0.8.0
 
 - Added optional tmux backend for long-running remote jobs with `exec --tmux --name <run> -- <command>`.
diff --git a/README.md b/README.md
index 5ec4bd3..085017f 100644
--- a/README.md
+++ b/README.md
@@ -38,7 +38,7 @@
 | ⬆️ 显式上传 | 用 `put` 或 `sync-up` 把本地改过的文件送到远端 |
 | ⬇️ 显式下载 | 用 `get` 或 `sync-down` 只拉回需要查看或提交的结果 |
 | 📜 日志查看 | 用 `tail` 查看远端训练日志，不必下载大文件 |
-| 🧾 Job 状态 | 用 `job status` 和 `job tail` 跟踪 `exec --detach --name` 创建的长任务 |
+| 🧾 Job 生命周期 | 用 `job status`、`job tail`、`job stop` 跟踪并停止 `exec --detach --name` 创建的长任务 |
 | 🧵 tmux 后端 | 可选用 `exec --tmux` 让长任务保留远端终端 pane，方便 dashboard 展示现场输出 |
 | 📊 Dashboard | 用只读页面查看多台远端设备、run 状态、最近日志或 tmux pane 输出 |
 | 🧠 脚本执行 | 用 `exec --script` 或 `exec --stdin` 避免复杂 quoting |
@@ -46,6 +46,12 @@
 | 📝 项目约定 | 首次使用自动生成 `.autodl-remote/CONVENTIONS.md` 记录远端约定 |
 | 🧩 Codex 友好 | 插件不替你判断同步策略，把决策权留给用户和 Codex |
 
+## 🆕 0.8.1 修复
+
+- 新增 `job stop` / `run stop`，停止 detach 作业的整个进程组，并保留已完成作业的原始退出状态。
+- 防止活跃同名 job 覆盖；需要替换时使用显式 `--replace`。
+- 修复 detach 命令参数引用、尾随 `job tail --lines` 解析，以及死 runner 的 `lost` 状态归一化。
+
 ## 🆕 0.8.0 更新
 
 &emsp;&emsp;这版重点补上了多机训练时最需要的“现场感”：
@@ -273,6 +279,7 @@ autodl-remote dashboard --fleet llm-exp --open --watch 5 --lines 120
 | `autodl-remote tmux capture train --lines 200` | 抓取 tmux pane 最近输出 |
 | `autodl-remote job list` | 列出本地记录的远端长任务 |
 | `autodl-remote job status train` | 查看长任务 PID、日志、远端状态和退出码 |
+| `autodl-remote job stop train` | 终止 detach 作业的整个进程组，并记录 cancelled |
 | `autodl-remote job tail train` | 查看命名长任务日志 |
 | `autodl-remote fleet status exp` | 查看多台远端设备状态 |
 | `autodl-remote dashboard --fleet exp --watch 5 --lines 120` | 打开只读 dashboard，持续展示远端状态和日志 |
@@ -320,10 +327,35 @@ autodl-remote put-run train.py -- python train.py
 autodl-remote exec --detach --name train -- python train.py
 autodl-remote job status train
 autodl-remote job tail train
+# 主动停止时会先向整个 detached 进程组发送 TERM，必要时再发送 KILL。
+autodl-remote job stop train
 autodl-remote cat -- outputs/metrics.json
 autodl-remote get outputs/metrics.json outputs/metrics.json
 ```
 
+非 `--follow` 的 `job tail` / `run tail` 会先把 tqdm 使用的回车更新规范化为逻辑日志行、移除 ANSI 控制序列，再取最后 N 行。单次输出默认最多 64 KiB。
+
+同名的运行中 detach job 默认会拒绝提交，避免覆盖元数据。只有明确指定 `--replace` 才会先停止旧作业再启动新作业：
+
+```bash
+autodl-remote exec --detach --replace --name train -- python train.py --resume
+```
+
+`job tail train --lines 20` 与 `job tail --lines 20 train` 等价；若 status 文件写着 `running` 但 runner PID 已不存在，`job status` / `run list` 会显示 `lost` 并提示使用 `job stop` 清理可能仍存活的同进程组子进程。
+
+SSH 连接与监控边界可通过环境变量调整：
+
+```bash
+AUTODL_REMOTE_TAIL_MAX_BYTES=65536
+AUTODL_REMOTE_CONNECT_TIMEOUT=10
+AUTODL_REMOTE_CONNECTION_ATTEMPTS=2
+AUTODL_REMOTE_MONITOR_TIMEOUT=30
+# 可选；未设置时按首连预算自动计算，且不少于 60 秒
+# AUTODL_REMOTE_MASTER_LOCK_TIMEOUT=60
+```
+
+这些值都必须是正整数；单位除字节数和重试次数外均为秒。默认 master 锁等待为 `max(60, 2 × ConnectTimeout × ConnectionAttempts + 5)`，因此提高建连预算时，并发等待者不会比创建者更早放弃。
+
 需要看远端终端现场时，使用可选 tmux 后端：
 
 ```bash
diff --git a/plugins/autodl-remote/.codex-plugin/plugin.json b/plugins/autodl-remote/.codex-plugin/plugin.json
index 431d08a..fcabca5 100644
--- a/plugins/autodl-remote/.codex-plugin/plugin.json
+++ b/plugins/autodl-remote/.codex-plugin/plugin.json
@@ -1,6 +1,6 @@
 {
   "name": "autodl-remote",
-  "version": "0.8.0",
+  "version": "0.8.1",
   "description": "Bash-only SSH account, fleet, command, tmux, upload, download, job, dashboard, and shutdown tool layer for AutoDL.",
   "author": {
     "name": "haibarazz",
diff --git a/plugins/autodl-remote/README.md b/plugins/autodl-remote/README.md
index 6726300..c9a1d5a 100644
--- a/plugins/autodl-remote/README.md
+++ b/plugins/autodl-remote/README.md
@@ -75,6 +75,7 @@ autodl-remote exec --tmux --name train-live --model baseline --stage training --
 autodl-remote tmux capture train-live --lines 200
 autodl-remote run status train
 autodl-remote run tail train
+autodl-remote run stop train
 autodl-remote dashboard
 ```
 
@@ -250,7 +251,7 @@ autodl-remote tmux capture [--account name] [--remote path] [--lines 120] <sessi
 autodl-remote tmux attach-cmd [--account name] [--remote path] <session>
 autodl-remote tmux kill [--account name] [--remote path] <session>
 
-autodl-remote exec [--account name] [--remote path] [--detach|--tmux] [--cwd path] [--name name] [--model name] [--tag k=v] [--stage stage] [--purpose text] [--script local_script] [--stdin] -- <command>
+autodl-remote exec [--account name] [--remote path] [--detach|--tmux] [--cwd path] [--name name] [--replace] [--model name] [--tag k=v] [--stage stage] [--purpose text] [--script local_script] [--stdin] -- <command>
 autodl-remote put-run <local-path>... -- <command>
 autodl-remote put [--account name] [--remote path] <local-path> [remote-path]
 autodl-remote get [--account name] [--remote path] <remote-path> [local-path]
@@ -262,10 +263,12 @@ autodl-remote cat [--account name] [--remote path] [--lines 200] -- <remote-path
 autodl-remote tail [--account name] [--remote path] [--lines 120] [--follow] -- <remote-path>
 autodl-remote job list
 autodl-remote job status <name>
+autodl-remote job stop <name>
 autodl-remote job tail [--lines 120] [--follow] <name>
 autodl-remote run submit [metadata/options] -- <command>
 autodl-remote run list
 autodl-remote run status <name>
+autodl-remote run stop <name>
 autodl-remote run tail [--lines 120] [--follow] <name>
 autodl-remote run note <name> [--model name] [--tag k=v] [--stage stage] [--purpose text] [--output path]
 autodl-remote shutdown [--wait seconds]
@@ -283,9 +286,25 @@ autodl-remote history
 - Use `put/get/sync-up/sync-down` explicitly.
 - Keep large models, datasets, checkpoints, and training outputs on AutoDL unless the user explicitly pulls them.
 - Use `exec --script` or `exec --stdin` for multi-line shell/Python snippets.
-- Use `job status` and `job tail` for detached training jobs created with `exec --detach --name`.
+- Use `job status`, `job tail`, and `job stop` for detached training jobs created with `exec --detach --name`. Active names are rejected unless `exec --replace` explicitly stops the old job first.
+- Non-following `job tail` / `run tail` normalizes carriage-return progress records, strips ANSI CSI sequences, and caps output at 64 KiB before returning the last requested logical lines.
 - Use `exec --tmux --name` when live pane output matters, especially for long tasks, parallel runs, or multiple hosts.
 - Use `fleet` and `dashboard` for multi-device visibility; the dashboard is display-only.
 - Use `dashboard --watch <seconds> --lines <n>` when the user wants to watch remote logs without spending chat tokens.
 - Use run metadata flags (`--model`, `--tag`, `--stage`, `--purpose`) when Codex launches experiments so later dashboards are understandable.
 - Use `shutdown` instead of guessing provider-specific poweroff commands.
+
+## Runtime Bounds
+
+The following positive-integer environment variables override connection and log-output defaults:
+
+```bash
+AUTODL_REMOTE_TAIL_MAX_BYTES=65536
+AUTODL_REMOTE_CONNECT_TIMEOUT=10
+AUTODL_REMOTE_CONNECTION_ATTEMPTS=2
+AUTODL_REMOTE_MONITOR_TIMEOUT=30
+# Optional; when unset, it follows the full master-start budget and is at least 60s.
+# AUTODL_REMOTE_MASTER_LOCK_TIMEOUT=60
+```
+
+Timeout values are seconds. The automatic master-lock timeout is `max(60, 2 * ConnectTimeout * ConnectionAttempts + 5)`. The monitor timeout applies to short captured calls such as status checks and non-following log snapshots; long-running foreground commands and `--follow` remain streaming operations.
diff --git a/plugins/autodl-remote/bin/autodl-remote b/plugins/autodl-remote/bin/autodl-remote
index 66ab2e0..0383eae 100755
--- a/plugins/autodl-remote/bin/autodl-remote
+++ b/plugins/autodl-remote/bin/autodl-remote
@@ -2,13 +2,18 @@
 set -u
 
 PROGRAM="autodl-remote"
-PROGRAM_VERSION="0.8.0"
+PROGRAM_VERSION="0.8.1"
 APP_DIR="$HOME/.autodl-remote"
 ACCOUNTS_DIR="$APP_DIR/accounts"
 DEFAULT_ACCOUNT_FILE="$APP_DIR/default-account"
 CONFIG_FILE=".autodl-remote.conf"
 HISTORY_DIR=".autodl-remote"
 KEYCHAIN_SERVICE="autodl-remote"
+TAIL_MAX_BYTES="${AUTODL_REMOTE_TAIL_MAX_BYTES:-65536}"
+SSH_CONNECT_TIMEOUT="${AUTODL_REMOTE_CONNECT_TIMEOUT:-10}"
+SSH_CONNECTION_ATTEMPTS="${AUTODL_REMOTE_CONNECTION_ATTEMPTS:-2}"
+SSH_MASTER_LOCK_TIMEOUT="${AUTODL_REMOTE_MASTER_LOCK_TIMEOUT:-}"
+SSH_MONITOR_TIMEOUT="${AUTODL_REMOTE_MONITOR_TIMEOUT:-30}"
 
 die() {
   printf '%s: %s\n' "$PROGRAM" "$*" >&2
@@ -53,7 +58,7 @@ project:
   tmux kill [--account name] [--remote path] <session>
 
 remote:
-  exec [--account name] [--remote path] [--detach|--tmux] [--cwd path] [--name name] [--model name] [--tag k=v] [--stage stage] [--purpose text] [--script local_script] [--stdin] [--allow-dangerous] -- <command>
+  exec [--account name] [--remote path] [--detach|--tmux] [--cwd path] [--name name] [--replace] [--model name] [--tag k=v] [--stage stage] [--purpose text] [--script local_script] [--stdin] [--allow-dangerous] -- <command>
   put-run <local-path>... -- <command>
   put [--account name] [--remote path] <local-path> [remote-path]
   get [--account name] [--remote path] <remote-path> [local-path]
@@ -65,10 +70,12 @@ remote:
   tail [--account name] [--remote path] [--lines 120] [--follow] -- <remote-path>
   job list
   job status <name>
+  job stop <name>
   job tail [--lines 120] [--follow] <name>
   run submit [metadata/options] -- <command>
   run list
   run status <name>
+  run stop <name>
   run tail [--lines 120] [--follow] <name>
   run note <name> [--model name] [--tag k=v] [--stage stage] [--purpose text] [--output path]
   shutdown [--wait seconds]
@@ -86,6 +93,14 @@ shell_quote() {
   printf '%q' "$1"
 }
 
+shell_quote_argv() {
+  local arg output=""
+  for arg in "$@"; do
+    output="${output:+$output }$(shell_quote "$arg")"
+  done
+  printf '%s' "$output"
+}
+
 strip_quotes() {
   local value="$1"
   value="${value#"${value%%[![:space:]]*}"}"
@@ -257,7 +272,7 @@ job_path() {
 }
 
 write_job() {
-  local name="$1" pid="$2" log="$3" status_path="$4" cwd="$5" command="$6" model="${7:-}" tags="${8:-}" purpose="${9:-}" stage="${10:-}" outputs="${11:-}" backend="${12:-detach}" tmux_session="${13:-}" tmux_socket="${14:-}" file
+  local name="$1" pid="$2" log="$3" status_path="$4" cwd="$5" command="$6" model="${7:-}" tags="${8:-}" purpose="${9:-}" stage="${10:-}" outputs="${11:-}" backend="${12:-detach}" tmux_session="${13:-}" tmux_socket="${14:-}" process_group="${15:-false}" file
   mkdir -p "$(jobs_dir)"
   file="$(job_path "$name")"
   cat > "$file" <<EOF
@@ -278,6 +293,7 @@ OUTPUTS="$(config_quote "$outputs")"
 BACKEND="$(config_quote "$backend")"
 TMUX_SESSION="$(config_quote "$tmux_session")"
 TMUX_SOCKET="$(config_quote "$tmux_socket")"
+PROCESS_GROUP="$(config_quote "$process_group")"
 EOF
 }
 
@@ -353,26 +369,38 @@ apply_job_context() {
   apply_context_override "$account" "$remote"
 }
 
-job_remote_status_summary() {
-  local file="$1" pid status_path output rc status alive
-  apply_job_context "$file"
-  pid="$(job_get "$file" PID)"
-  status_path="$(job_get "$file" STATUS_PATH)"
-  output="$(ssh_capture "if [ -f $(shell_quote "$status_path") ]; then cat $(shell_quote "$status_path"); elif kill -0 $(shell_quote "$pid") 2>/dev/null; then printf 'status=running\npid=%s\n' $(shell_quote "$pid"); else printf 'status=unknown\npid=%s\n' $(shell_quote "$pid"); fi; if kill -0 $(shell_quote "$pid") 2>/dev/null; then echo alive=yes; else echo alive=no; fi" 2>/dev/null)"
-  rc=$?
-  if [ "$rc" -ne 0 ]; then
-    printf 'unreachable\n'
-    return 0
-  fi
+job_status_payload() {
+  local pid="$1" status_path="$2"
+  ssh_capture "if [ -f $(shell_quote "$status_path") ]; then cat $(shell_quote "$status_path"); elif kill -0 $(shell_quote "$pid") 2>/dev/null; then printf 'status=running\npid=%s\n' $(shell_quote "$pid"); else printf 'status=unknown\npid=%s\n' $(shell_quote "$pid"); fi; if kill -0 $(shell_quote "$pid") 2>/dev/null; then echo alive=yes; else echo alive=no; fi"
+}
+
+job_status_from_payload() {
+  local output="$1" status alive
   status="$(printf '%s\n' "$output" | awk -F= '$1=="status"{print $2; exit}')"
   alive="$(printf '%s\n' "$output" | awk -F= '$1=="alive"{print $2; exit}')"
   [ -n "$status" ] || status="unknown"
   if [ "$alive" = "yes" ] && [ "$status" = "unknown" ]; then
     status="running"
+  elif [ "$alive" = "no" ] && [ "$status" = "running" ]; then
+    status="lost"
   fi
   printf '%s\n' "$status"
 }
 
+job_remote_status_summary() {
+  local file="$1" pid status_path output rc
+  apply_job_context "$file"
+  pid="$(job_get "$file" PID)"
+  status_path="$(job_get "$file" STATUS_PATH)"
+  output="$(job_status_payload "$pid" "$status_path" 2>/dev/null)"
+  rc=$?
+  if [ "$rc" -ne 0 ]; then
+    printf 'unreachable\n'
+    return 0
+  fi
+  job_status_from_payload "$output"
+}
+
 remote_probe() {
   local output rc
   output="$(ssh_capture "root=$(shell_quote "$REMOTE_ROOT"); printf 'online=yes\n'; printf 'host='; hostname 2>/dev/null || echo unknown; printf 'remote_root=%s\n' \"\$root\"; printf 'disk='; df -h \"\$root\" 2>/dev/null | awk 'NR==2{print \$3 \"/\" \$2 \" \" \$5}' || echo unknown; if command -v nvidia-smi >/dev/null 2>&1; then printf 'gpu='; nvidia-smi --query-gpu=name,memory.used,memory.total,utilization.gpu --format=csv,noheader,nounits 2>/dev/null | head -n 1 | awk -F, '{gsub(/^ +| +$/, \"\", \$1); gsub(/^ +| +$/, \"\", \$2); gsub(/^ +| +$/, \"\", \$3); gsub(/^ +| +$/, \"\", \$4); print \$1 \" \" \$2 \"/\" \$3 \" MiB \" \$4 \"%\"}'; else echo 'gpu=unavailable'; fi; printf 'processes='; pgrep -af 'python|torchrun|accelerate' 2>/dev/null | head -n 3 | tr '\n' '|' | sed 's/|$//'; printf '\n'" 2>&1)"
@@ -457,11 +485,13 @@ load_project() {
 }
 
 ssh_common_args() {
-  local control_path
+  local control_master="${1:-no}" control_path
   control_path="$(ssh_control_path)"
-  SSH_ARGS=(-p "$ACCOUNT_PORT" -o ServerAliveInterval=30 -o ServerAliveCountMax=3 -o StrictHostKeyChecking=accept-new -o ControlMaster=auto -o ControlPersist=600 -o "ControlPath=$control_path")
+  validate_ssh_timeouts
+  SSH_ARGS=(-p "$ACCOUNT_PORT" -o "ConnectTimeout=$SSH_CONNECT_TIMEOUT" -o "ConnectionAttempts=$SSH_CONNECTION_ATTEMPTS" -o ServerAliveInterval=30 -o ServerAliveCountMax=3 -o StrictHostKeyChecking=accept-new -o "ControlMaster=$control_master" -o "ControlPath=$control_path")
+  [ "$control_master" != "yes" ] || SSH_ARGS+=(-o ControlPersist=600)
   [ -n "$ACCOUNT_KEY" ] && SSH_ARGS+=(-i "$ACCOUNT_KEY")
-  if [ "$ACCOUNT_AUTH" != "keychain" ] && [ "$ACCOUNT_AUTH" != "prompt" ]; then
+  if [ "$control_master" != "yes" ] || { [ "$ACCOUNT_AUTH" != "keychain" ] && [ "$ACCOUNT_AUTH" != "prompt" ]; }; then
     SSH_ARGS+=(-o BatchMode=yes)
   fi
 }
@@ -469,21 +499,19 @@ ssh_common_args() {
 scp_common_args() {
   local control_path
   control_path="$(ssh_control_path)"
-  SCP_ARGS=(-P "$ACCOUNT_PORT" -o ServerAliveInterval=30 -o ServerAliveCountMax=3 -o StrictHostKeyChecking=accept-new -o ControlMaster=auto -o ControlPersist=600 -o "ControlPath=$control_path")
+  validate_ssh_timeouts
+  SCP_ARGS=(-P "$ACCOUNT_PORT" -o "ConnectTimeout=$SSH_CONNECT_TIMEOUT" -o "ConnectionAttempts=$SSH_CONNECTION_ATTEMPTS" -o ServerAliveInterval=30 -o ServerAliveCountMax=3 -o StrictHostKeyChecking=accept-new -o ControlMaster=no -o "ControlPath=$control_path")
   [ -n "$ACCOUNT_KEY" ] && SCP_ARGS+=(-i "$ACCOUNT_KEY")
-  if [ "$ACCOUNT_AUTH" != "keychain" ] && [ "$ACCOUNT_AUTH" != "prompt" ]; then
-    SCP_ARGS+=(-o BatchMode=yes)
-  fi
+  SCP_ARGS+=(-o BatchMode=yes)
 }
 
 rsync_ssh_command() {
   local cmd control_path
   control_path="$(ssh_control_path)"
-  cmd="ssh -p $(shell_quote "$ACCOUNT_PORT") -o ServerAliveInterval=30 -o ServerAliveCountMax=3 -o StrictHostKeyChecking=accept-new -o ControlMaster=auto -o ControlPersist=600 -o ControlPath=$(shell_quote "$control_path")"
+  validate_ssh_timeouts
+  cmd="ssh -p $(shell_quote "$ACCOUNT_PORT") -o ConnectTimeout=$(shell_quote "$SSH_CONNECT_TIMEOUT") -o ConnectionAttempts=$(shell_quote "$SSH_CONNECTION_ATTEMPTS") -o ServerAliveInterval=30 -o ServerAliveCountMax=3 -o StrictHostKeyChecking=accept-new -o ControlMaster=no -o ControlPath=$(shell_quote "$control_path")"
   [ -n "$ACCOUNT_KEY" ] && cmd="$cmd -i $(shell_quote "$ACCOUNT_KEY")"
-  if [ "$ACCOUNT_AUTH" != "keychain" ] && [ "$ACCOUNT_AUTH" != "prompt" ]; then
-    cmd="$cmd -o BatchMode=yes"
-  fi
+  cmd="$cmd -o BatchMode=yes"
   printf '%s\n' "$cmd"
 }
 
@@ -496,9 +524,116 @@ ssh_control_dir() {
 }
 
 ssh_control_path() {
-  printf '%s/ssh-%%C\n' "$(ssh_control_dir)"
+  local account_id connection_id
+  account_id="$(printf '%s\n' "$ACCOUNT_NAME" | cksum | awk '{print $1}')"
+  connection_id="$(printf '%s\n' "$ACCOUNT_TARGET:$ACCOUNT_PORT" | cksum | awk '{print $1}')"
+  printf '%s/ssh-%s-%s\n' "$(ssh_control_dir)" "$account_id" "$connection_id"
 }
 
+validate_ssh_timeouts() {
+  case "$SSH_CONNECT_TIMEOUT" in
+    ""|0|*[!0-9]*) die "AUTODL_REMOTE_CONNECT_TIMEOUT must be a positive integer" ;;
+  esac
+  case "$SSH_CONNECTION_ATTEMPTS" in
+    ""|0|*[!0-9]*) die "AUTODL_REMOTE_CONNECTION_ATTEMPTS must be a positive integer" ;;
+  esac
+  case "$SSH_MASTER_LOCK_TIMEOUT" in
+    "") ;;
+    0|*[!0-9]*) die "AUTODL_REMOTE_MASTER_LOCK_TIMEOUT must be a positive integer" ;;
+  esac
+  case "$SSH_MONITOR_TIMEOUT" in
+    ""|0|*[!0-9]*) die "AUTODL_REMOTE_MONITOR_TIMEOUT must be a positive integer" ;;
+  esac
+}
+
+ssh_master_lock_timeout() {
+  local minimum
+  if [ -n "$SSH_MASTER_LOCK_TIMEOUT" ]; then
+    printf '%s\n' "$SSH_MASTER_LOCK_TIMEOUT"
+    return 0
+  fi
+  minimum=$((SSH_CONNECT_TIMEOUT * SSH_CONNECTION_ATTEMPTS * 2 + 5))
+  [ "$minimum" -ge 60 ] || minimum=60
+  printf '%s\n' "$minimum"
+}
+
+path_mtime() {
+  stat -f '%m' "$1" 2>/dev/null || stat -c '%Y' "$1" 2>/dev/null
+}
+
+ssh_master_check() {
+  local control_path
+  control_path="$(ssh_control_path)"
+  ssh -p "$ACCOUNT_PORT" -o "ConnectTimeout=$SSH_CONNECT_TIMEOUT" -o ControlMaster=no -o "ControlPath=$control_path" -O check "$ACCOUNT_TARGET" >/dev/null 2>&1
+}
+
+acquire_ssh_master_lock() {
+  local lock_dir="$1" started now owner owner_command lock_mtime lock_timeout
+  started="$(date +%s)"
+  lock_timeout="$(ssh_master_lock_timeout)"
+  while ! mkdir "$lock_dir" 2>/dev/null; do
+    if [ -f "$lock_dir/pid" ]; then
+      owner="$(sed -n '1p' "$lock_dir/pid" 2>/dev/null || true)"
+      case "$owner" in
+        ""|*[!0-9]*) owner="" ;;
+        *)
+          owner_command="$(ps -p "$owner" -o command= 2>/dev/null || true)"
+          if ! kill -0 "$owner" 2>/dev/null || [ -z "$owner_command" ] || [ "${owner_command#*autodl-remote}" = "$owner_command" ]; then
+            rm -rf "$lock_dir"
+            continue
+          fi
+          ;;
+      esac
+    else
+      owner=""
+    fi
+    if [ -z "$owner" ]; then
+      now="$(date +%s)"
+      lock_mtime="$(path_mtime "$lock_dir" || printf '%s\n' "$now")"
+      if [ $((now - lock_mtime)) -ge "$lock_timeout" ]; then
+        rm -rf "$lock_dir"
+        continue
+      fi
+    fi
+    now="$(date +%s)"
+    if [ $((now - started)) -ge "$lock_timeout" ]; then
+      printf '%s: timed out waiting for SSH master lock: %s\n' "$PROGRAM" "$lock_dir" >&2
+      return 1
+    fi
+    sleep 0.1
+  done
+  printf '%s\n' "$$" > "$lock_dir/pid.tmp"
+  mv "$lock_dir/pid.tmp" "$lock_dir/pid"
+}
+
+ensure_ssh_master() (
+  local control_path lock_dir attempt rc=1
+  validate_ssh_timeouts
+  ssh_master_check && return 0
+  control_path="$(ssh_control_path)"
+  lock_dir="$control_path.lock"
+  acquire_ssh_master_lock "$lock_dir" || return 1
+  trap 'rm -rf "$lock_dir"' EXIT
+  trap 'exit 129' HUP
+  trap 'exit 130' INT
+  trap 'exit 143' TERM
+  ssh_master_check && return 0
+
+  for attempt in 1 2; do
+    rm -f "$control_path"
+    ssh_common_args yes
+    if run_transport_timed "$SSH_CONNECT_TIMEOUT" ssh -MNf "${SSH_ARGS[@]}" "$ACCOUNT_TARGET"; then
+      ssh_master_check && return 0
+    else
+      rc=$?
+    fi
+    ssh_master_check && return 0
+    sleep 0.2
+  done
+  printf '%s: failed to establish shared SSH master for %s\n' "$PROGRAM" "$ACCOUNT_NAME" >&2
+  return "$rc"
+)
+
 keychain_password() {
   command -v security >/dev/null 2>&1 || die "macOS security command is not available."
   security find-generic-password -a "$ACCOUNT_NAME" -s "$KEYCHAIN_SERVICE" -w 2>/dev/null
@@ -506,10 +641,10 @@ keychain_password() {
 
 expect_with_password() {
   command -v expect >/dev/null 2>&1 || die "Account uses keychain auth, but expect is not available."
-  local password="$1"
-  shift
-  AUTODL_REMOTE_PASSWORD="$password" expect -f /dev/stdin "$@" <<'EXPECT'
-    set timeout -1
+  local password="$1" timeout_seconds="$2"
+  shift 2
+  AUTODL_REMOTE_PASSWORD="$password" AUTODL_REMOTE_EXPECT_TIMEOUT="$timeout_seconds" expect -f /dev/stdin "$@" <<'EXPECT'
+    set timeout $env(AUTODL_REMOTE_EXPECT_TIMEOUT)
     set password $env(AUTODL_REMOTE_PASSWORD)
     log_user 0
     spawn {*}$argv
@@ -517,15 +652,23 @@ expect_with_password() {
     expect {
       -nocase -re "are you sure.*continue connecting" {
         send "yes\r"
-        exp_continue
+        exp_continue -continue_timer
       }
       -nocase -re "yes/no" {
         send "yes\r"
-        exp_continue
+        exp_continue -continue_timer
       }
       -nocase -re "password:" {
         send "$password\r"
-        exp_continue
+        exp_continue -continue_timer
+      }
+      timeout {
+        puts stderr "autodl-remote: SSH transport timed out"
+        set child_pid [exp_pid]
+        catch {close}
+        catch {exec kill -TERM $child_pid}
+        catch {wait}
+        exit 124
       }
       eof
     }
@@ -534,11 +677,108 @@ expect_with_password() {
 EXPECT
 }
 
-run_transport() {
+terminate_descendants() {
+  local parent="$1" signal="$2" child children
+  command -v pgrep >/dev/null 2>&1 || return 0
+  children="$(pgrep -P "$parent" 2>/dev/null || true)"
+  for child in $children; do
+    terminate_descendants "$child" "$signal"
+  done
+  for child in $children; do
+    kill "-$signal" "$child" 2>/dev/null || true
+  done
+}
+
+terminate_process_tree() {
+  local pid="$1" signal="$2" process_group="$3"
+  kill -0 "$pid" 2>/dev/null || return 0
+  if [ "$process_group" = "true" ]; then
+    kill "-$signal" -- "-$pid" 2>/dev/null || true
+  else
+    terminate_descendants "$pid" "$signal"
+    kill "-$signal" "$pid" 2>/dev/null || true
+  fi
+}
+
+run_command_with_timeout() (
+  local timeout_seconds="$1" pid="" watchdog="" sleeper="" marker="" timer_pid_file="" rc process_group=false
+  shift
+  cleanup_timeout_runner() {
+    if [ -n "$pid" ]; then
+      terminate_process_tree "$pid" TERM "$process_group"
+      sleep 0.1
+      terminate_process_tree "$pid" KILL "$process_group"
+    fi
+    [ -z "$sleeper" ] || kill "$sleeper" 2>/dev/null || true
+    [ -z "$watchdog" ] || kill "$watchdog" 2>/dev/null || true
+    [ -z "$marker" ] || rm -f "$marker"
+    [ -z "$timer_pid_file" ] || rm -f "$timer_pid_file"
+  }
+  trap cleanup_timeout_runner EXIT
+  trap 'exit 129' HUP
+  trap 'exit 130' INT
+  trap 'exit 143' TERM
+
+  marker="$(mktemp "${TMPDIR:-/tmp}/autodl-remote-timeout.XXXXXX")" || die "mktemp failed"
+  timer_pid_file="$(mktemp "${TMPDIR:-/tmp}/autodl-remote-timer.XXXXXX")" || { rm -f "$marker"; die "mktemp failed"; }
+  rm -f "$marker"
+  rm -f "$timer_pid_file"
+  if command -v setsid >/dev/null 2>&1; then
+    setsid "$@" &
+    process_group=true
+  else
+    "$@" &
+  fi
+  pid=$!
+  (
+    sleep "$timeout_seconds" &
+    sleeper=$!
+    printf '%s\n' "$sleeper" > "$timer_pid_file"
+    wait "$sleeper" || exit 0
+    if kill -0 "$pid" 2>/dev/null; then
+      : > "$marker"
+      terminate_process_tree "$pid" TERM "$process_group"
+      sleep 1
+      terminate_process_tree "$pid" KILL "$process_group"
+    fi
+  ) </dev/null >/dev/null 2>&1 &
+  watchdog=$!
+  wait "$pid"
+  rc=$?
+  pid=""
+  for _ in 1 2 3 4 5 6 7 8 9 10; do
+    [ -f "$timer_pid_file" ] && break
+    kill -0 "$watchdog" 2>/dev/null || break
+    sleep 0.01
+  done
+  if [ -f "$timer_pid_file" ]; then
+    sleeper="$(sed -n '1p' "$timer_pid_file" 2>/dev/null || true)"
+    case "$sleeper" in
+      ""|*[!0-9]*) ;;
+      *) kill "$sleeper" 2>/dev/null || true; sleeper="" ;;
+    esac
+  fi
+  kill "$watchdog" 2>/dev/null || true
+  wait "$watchdog" 2>/dev/null || true
+  watchdog=""
+  if [ -f "$marker" ]; then
+    printf '%s: transport timed out after %ss\n' "$PROGRAM" "$timeout_seconds" >&2
+    return 124
+  fi
+  return "$rc"
+)
+
+run_transport_timed() {
+  local timeout_seconds="$1" rc
+  shift
   if [ "$ACCOUNT_AUTH" = "keychain" ]; then
     local password
-    password="$(keychain_password)" || die "No Keychain password saved for account: $ACCOUNT_NAME"
-    expect_with_password "$password" "$@"
+    password="$(run_command_with_timeout "$timeout_seconds" keychain_password)"
+    rc=$?
+    [ "$rc" -eq 0 ] || { [ "$rc" -eq 124 ] && return 124; die "No Keychain password saved for account: $ACCOUNT_NAME"; }
+    expect_with_password "$password" "$timeout_seconds" "$@"
+  elif [ "$ACCOUNT_AUTH" = "ssh-key" ]; then
+    run_command_with_timeout "$timeout_seconds" "$@"
   else
     "$@"
   fi
@@ -546,15 +786,40 @@ run_transport() {
 
 ssh_run() {
   local command="$1"
+  ensure_ssh_master || return $?
   ssh_common_args
-  run_transport ssh "${SSH_ARGS[@]}" "$ACCOUNT_TARGET" "$command"
+  ssh "${SSH_ARGS[@]}" "$ACCOUNT_TARGET" "$command"
 }
 
 ssh_capture() {
   local command="$1"
+  ensure_ssh_master || return $?
   ssh_common_args
-  run_transport ssh "${SSH_ARGS[@]}" "$ACCOUNT_TARGET" "$command" | tr -d '\r'
-  return "${PIPESTATUS[0]}"
+  run_command_with_timeout "$SSH_MONITOR_TIMEOUT" ssh "${SSH_ARGS[@]}" "$ACCOUNT_TARGET" "$command"
+}
+
+validate_tail_settings() {
+  local lines="$1"
+  case "$lines" in
+    ""|*[!0-9]*) die "--lines must be a non-negative integer" ;;
+  esac
+  case "$TAIL_MAX_BYTES" in
+    ""|0|*[!0-9]*) die "AUTODL_REMOTE_TAIL_MAX_BYTES must be a positive integer" ;;
+  esac
+}
+
+normalized_tail_filter_command() {
+  local lines="$1" awk_program
+  awk_program='{ sub(/\r$/, ""); gsub(/\r/, "\n"); gsub(sprintf("%c", 27) "\\[[0-?]*[ -/]*[@-~]", ""); print }'
+  printf 'LC_ALL=C awk %s | tail -n %s | tail -c %s' \
+    "$(shell_quote "$awk_program")" "$lines" "$TAIL_MAX_BYTES"
+}
+
+normalized_log_tail_command() {
+  local log="$1" lines="$2" input_bytes
+  input_bytes=$((TAIL_MAX_BYTES * 4))
+  printf 'tail -c %s %s | %s' \
+    "$input_bytes" "$(shell_quote "$log")" "$(normalized_tail_filter_command "$lines")"
 }
 
 remote_join() {
@@ -614,15 +879,17 @@ upload_local_file() {
   local local_path="$1" remote_path="$2"
   [ -f "$local_path" ] || die "Local script does not exist or is not a file: $local_path"
   ensure_remote_dir "$(dirname "$remote_path")"
+  ensure_ssh_master || return $?
   scp_common_args
-  run_transport scp "${SCP_ARGS[@]}" "$local_path" "$ACCOUNT_TARGET:$remote_path"
+  scp "${SCP_ARGS[@]}" "$local_path" "$ACCOUNT_TARGET:$remote_path"
 }
 
 upload_stdin_to_remote() {
   local remote_path="$1"
   ensure_remote_dir "$(dirname "$remote_path")"
+  ensure_ssh_master || return $?
   ssh_common_args
-  run_transport ssh "${SSH_ARGS[@]}" "$ACCOUNT_TARGET" "cat > $(shell_quote "$remote_path")"
+  ssh "${SSH_ARGS[@]}" "$ACCOUNT_TARGET" "cat > $(shell_quote "$remote_path")"
 }
 
 upload_file_via_ssh() {
@@ -643,7 +910,7 @@ remote_tmux_capture() {
   local session="$1" socket="$2" lines="$3" output rc
   [ -n "$session" ] || { printf '[tmux session missing]\n'; return 1; }
   [ -n "$socket" ] || socket="$(tmux_socket_path)"
-  output="$(ssh_capture "if ! command -v tmux >/dev/null 2>&1; then printf '%s\n' $(shell_quote "$(tmux_missing_text)"); exit 127; fi; if tmux -S $(shell_quote "$socket") has-session -t $(shell_quote "$session") 2>/dev/null; then tmux -S $(shell_quote "$socket") capture-pane -p -J -t $(shell_quote "$session:0.0") -S -$lines; else printf '[tmux session missing] %s\n' $(shell_quote "$session"); exit 42; fi" 2>&1)"
+  output="$(ssh_capture "if ! command -v tmux >/dev/null 2>&1; then printf '%s\n' $(shell_quote "$(tmux_missing_text)"); exit 127; fi; if tmux -S $(shell_quote "$socket") has-session -t $(shell_quote "$session") 2>/dev/null; then tmux -S $(shell_quote "$socket") capture-pane -p -J -t $(shell_quote "$session:0.0") -S -$lines | $(normalized_tail_filter_command "$lines"); else printf '[tmux session missing] %s\n' $(shell_quote "$session"); exit 42; fi" 2>&1)"
   rc=$?
   printf '%s\n' "$output"
   return "$rc"
@@ -979,7 +1246,7 @@ cmd_doctor() {
 
 cmd_exec() {
   load_project
-  local detach=false tmux_mode=false cwd="" cwd_arg="" name="" allow_dangerous=false script="" stdin_mode=false remote_script=""
+  local detach=false tmux_mode=false cwd="" cwd_arg="" name="" replace=false allow_dangerous=false script="" stdin_mode=false remote_script=""
   local account_override="" remote_override="" model="" tags="" purpose="" stage="" outputs=""
   while [ $# -gt 0 ]; do
     case "$1" in
@@ -989,6 +1256,7 @@ cmd_exec() {
       --tmux) tmux_mode=true; detach=true; shift ;;
       --cwd) cwd_arg="$2"; shift 2 ;;
       --name) name="$2"; shift 2 ;;
+      --replace) replace=true; shift ;;
       --model) model="$2"; shift 2 ;;
       --tag) tags="$(append_csv "$tags" "$2")"; shift 2 ;;
       --purpose) purpose="$2"; shift 2 ;;
@@ -1018,7 +1286,7 @@ cmd_exec() {
       upload_file_via_ssh "$local_script" "$remote_script" || return 1
     fi
     if [ $# -gt 0 ]; then
-      cmd="$* $(shell_quote "$remote_script")"
+      cmd="$(shell_quote_argv "$@") $(shell_quote "$remote_script")"
     else
       cmd="bash $(shell_quote "$remote_script")"
     fi
@@ -1035,10 +1303,10 @@ cmd_exec() {
     else
       upload_stdin_to_remote "$remote_script" || return 1
     fi
-    cmd="$* < $(shell_quote "$remote_script")"
+    cmd="$(shell_quote_argv "$@") < $(shell_quote "$remote_script")"
   else
     [ $# -gt 0 ] || die "exec requires a command"
-    cmd="$*"
+    cmd="$(shell_quote_argv "$@")"
   fi
 
   if [ "$allow_dangerous" != "true" ]; then
@@ -1046,6 +1314,27 @@ cmd_exec() {
   fi
   [ -n "$name" ] || name="exec-$(date +%Y%m%d-%H%M%S)"
   name="$(sanitize_name "$name")"
+  local existing_file existing_status
+  existing_file="$(job_path "$name")"
+  if [ -f "$existing_file" ]; then
+    existing_status="$(job_remote_status_summary "$existing_file")"
+    case "$existing_status" in
+      running|lost)
+        if [ "$replace" != "true" ]; then
+          die "Job name is already active: $name (status=$existing_status). Use --replace to stop it before submitting a replacement."
+        fi
+        info "Replacing existing job: $name (status=$existing_status)"
+        cmd_job_stop "$name" || return $?
+        apply_context_override "$account_override" "$remote_override"
+        ;;
+      unreachable)
+        die "Cannot safely replace job $name because its remote status is unreachable."
+        ;;
+      *)
+        [ "$replace" != "true" ] || info "Replacing inactive job: $name (status=$existing_status)"
+        ;;
+    esac
+  fi
   local log="$REMOTE_ROOT/.autodl-remote/logs/$name.log"
   status_path="$REMOTE_ROOT/.autodl-remote/jobs/$name.status"
   info "log=$log"
@@ -1059,6 +1348,12 @@ set +e
 started_at=\$(date +%Y-%m-%dT%H:%M:%S%z)
 pid=\$BASHPID
 printf 'status=running\npid=%s\nstarted_at=%s\nlog=%s\ncwd=%s\n' "\$pid" "\$started_at" $(shell_quote "$log") $(shell_quote "$cwd") > $(shell_quote "$status_path")
+cancelled() {
+  finished_at=\$(date +%Y-%m-%dT%H:%M:%S%z)
+  printf 'status=cancelled\npid=%s\nstarted_at=%s\nfinished_at=%s\nexit_code=143\nlog=%s\ncwd=%s\n' "\$pid" "\$started_at" "\$finished_at" $(shell_quote "$log") $(shell_quote "$cwd") > $(shell_quote "$status_path")
+  exit 143
+}
+trap cancelled INT TERM
 if [ "\${AUTODL_REMOTE_TMUX:-}" = "1" ]; then
   cd $(shell_quote "$cwd") && bash -lc $(shell_quote "$cmd") 2>&1 | tee $(shell_quote "$log")
   rc=\${PIPESTATUS[0]}
@@ -1130,7 +1425,7 @@ EOF
       rm -f "$launcher_local"
       remote_cmd="bash $(shell_quote "$launcher")"
     else
-      remote_cmd="mkdir -p $(shell_quote "$(dirname "$log")") $(shell_quote "$(dirname "$status_path")") && chmod +x $(shell_quote "$runner") && if command -v setsid >/dev/null 2>&1; then setsid bash $(shell_quote "$runner") < /dev/null > /dev/null 2>&1 & else nohup bash $(shell_quote "$runner") < /dev/null > /dev/null 2>&1 & fi; printf 'pid=%s\nlog=%s\nstatus=%s\n' \"\$!\" $(shell_quote "$log") $(shell_quote "$status_path")"
+      remote_cmd="mkdir -p $(shell_quote "$(dirname "$log")") $(shell_quote "$(dirname "$status_path")") && chmod +x $(shell_quote "$runner") && if command -v setsid >/dev/null 2>&1; then setsid bash $(shell_quote "$runner") < /dev/null > /dev/null 2>&1 & process_group=yes; else nohup bash $(shell_quote "$runner") < /dev/null > /dev/null 2>&1 & process_group=no; fi; printf 'pid=%s\nlog=%s\nstatus=%s\nprocess_group=%s\n' \"\$!\" $(shell_quote "$log") $(shell_quote "$status_path") \"\$process_group\""
     fi
     result="$(ssh_capture "bash -lc $(shell_quote "$remote_cmd")" 2>&1)"
     detach_rc=$?
@@ -1139,8 +1434,11 @@ EOF
     pid="$(printf '%s\n' "$result" | awk -F= '$1=="pid"{print $2; exit}')"
     tmux_session="$(printf '%s\n' "$result" | awk -F= '$1=="tmux_session"{print $2; exit}')"
     tmux_socket="$(printf '%s\n' "$result" | awk -F= '$1=="tmux_socket"{print $2; exit}')"
+    local process_group
+    process_group="$(printf '%s\n' "$result" | awk -F= '$1=="process_group"{print $2; exit}')"
+    [ "$backend" != "tmux" ] || process_group=false
     [ -n "$pid" ] || die "Remote detach did not return a pid."
-    write_job "$name" "$pid" "$log" "$status_path" "$cwd" "$cmd" "$model" "$tags" "$purpose" "$stage" "$outputs" "$backend" "$tmux_session" "$tmux_socket"
+    write_job "$name" "$pid" "$log" "$status_path" "$cwd" "$cmd" "$model" "$tags" "$purpose" "$stage" "$outputs" "$backend" "$tmux_session" "$tmux_socket" "${process_group:-false}"
     write_history "exec-$backend" "$cmd log=$log"
   else
     local inner rc
@@ -1180,17 +1478,17 @@ cmd_put() {
     [ "$dry_run" = "true" ] && args+=(--dry-run)
     [ "$delete" = "true" ] && args+=(--delete)
     if [ -d "$local_path" ]; then
-      run_transport rsync "${args[@]}" "$local_path/" "$ACCOUNT_TARGET:$remote_path/"
+      rsync "${args[@]}" "$local_path/" "$ACCOUNT_TARGET:$remote_path/"
     else
-      run_transport rsync "${args[@]}" "$local_path" "$ACCOUNT_TARGET:$remote_path"
+      rsync "${args[@]}" "$local_path" "$ACCOUNT_TARGET:$remote_path"
     fi
   else
     [ "$dry_run" != "true" ] || die "dry-run requires rsync."
     scp_common_args
     if [ -d "$local_path" ]; then
-      run_transport scp -r "${SCP_ARGS[@]}" "$local_path" "$ACCOUNT_TARGET:$remote_path"
+      scp -r "${SCP_ARGS[@]}" "$local_path" "$ACCOUNT_TARGET:$remote_path"
     else
-      run_transport scp "${SCP_ARGS[@]}" "$local_path" "$ACCOUNT_TARGET:$remote_path"
+      scp "${SCP_ARGS[@]}" "$local_path" "$ACCOUNT_TARGET:$remote_path"
     fi
   fi
   write_history put "$local_arg -> $remote_path"
@@ -1227,17 +1525,17 @@ cmd_get() {
     [ "$dry_run" = "true" ] && args+=(--dry-run)
     if [ "$kind" = "dir" ]; then
       mkdir -p "$local_path"
-      run_transport rsync "${args[@]}" "$ACCOUNT_TARGET:$remote_path/" "$local_path/"
+      rsync "${args[@]}" "$ACCOUNT_TARGET:$remote_path/" "$local_path/"
     else
-      run_transport rsync "${args[@]}" "$ACCOUNT_TARGET:$remote_path" "$local_path"
+      rsync "${args[@]}" "$ACCOUNT_TARGET:$remote_path" "$local_path"
     fi
   else
     [ "$dry_run" != "true" ] || die "dry-run requires rsync."
     scp_common_args
     if [ "$kind" = "dir" ]; then
-      run_transport scp -r "${SCP_ARGS[@]}" "$ACCOUNT_TARGET:$remote_path" "$local_path"
+      scp -r "${SCP_ARGS[@]}" "$ACCOUNT_TARGET:$remote_path" "$local_path"
     else
-      run_transport scp "${SCP_ARGS[@]}" "$ACCOUNT_TARGET:$remote_path" "$local_path"
+      scp "${SCP_ARGS[@]}" "$ACCOUNT_TARGET:$remote_path" "$local_path"
     fi
   fi
   write_history get "$remote_path -> $local_arg"
@@ -1341,7 +1639,7 @@ cmd_job_list() {
 
 cmd_job_status() {
   load_project
-  [ $# -gt 0 ] || die "job status requires a name"
+  [ $# -eq 1 ] || die "job status requires exactly one name"
   local name file pid log status_path cwd command backend tmux_session tmux_socket remote_status rc
   name="$(sanitize_name "$1")"
   file="$(job_path "$name")"
@@ -1368,10 +1666,15 @@ cmd_job_status() {
   info "log=$log"
   info "status_path=$status_path"
   info "command=$command"
-  remote_status="$(ssh_capture "if [ -f $(shell_quote "$status_path") ]; then cat $(shell_quote "$status_path"); elif kill -0 $(shell_quote "$pid") 2>/dev/null; then printf 'status=running\npid=%s\n' $(shell_quote "$pid"); else printf 'status=unknown\npid=%s\n' $(shell_quote "$pid"); fi; if kill -0 $(shell_quote "$pid") 2>/dev/null; then echo alive=yes; else echo alive=no; fi" 2>&1)"
+  remote_status="$(job_status_payload "$pid" "$status_path" 2>&1)"
   rc=$?
   if [ "$rc" -eq 0 ]; then
-    printf '%s\n' "$remote_status"
+    if [ "$(job_status_from_payload "$remote_status")" = "lost" ]; then
+      printf '%s\n' "$remote_status" | awk -F= '$1 == "status" { print "status=lost"; next } { print }'
+      info "hint=runner pid is not alive; run job stop $name to terminate any surviving process-group children, then submit a replacement if needed"
+    else
+      printf '%s\n' "$remote_status"
+    fi
   else
     info "remote_status=unreachable"
     printf '%s\n' "$remote_status" >&2
@@ -1379,20 +1682,81 @@ cmd_job_status() {
   fi
 }
 
+cmd_job_stop() {
+  load_project
+  [ $# -eq 1 ] || die "job stop requires exactly one name"
+  local name file pid log status_path cwd backend tmux_session tmux_socket process_group current_status group_probe remote_cmd output rc
+  name="$(sanitize_name "$1")"
+  file="$(job_path "$name")"
+  [ -f "$file" ] || die "Job not found: $name"
+  pid="$(job_get "$file" PID)"
+  log="$(job_get "$file" LOG_PATH)"
+  status_path="$(job_get "$file" STATUS_PATH)"
+  cwd="$(job_get "$file" CWD)"
+  backend="$(job_get "$file" BACKEND)"
+  tmux_session="$(job_get "$file" TMUX_SESSION)"
+  tmux_socket="$(job_get "$file" TMUX_SOCKET)"
+  process_group="$(job_get "$file" PROCESS_GROUP)"
+  [ -n "$pid" ] || die "Job has no recorded pid: $name"
+  apply_job_context "$file"
+  current_status="$(job_remote_status_summary "$file")"
+  case "$current_status" in
+    exited|cancelled)
+      info "status=$current_status"
+      info "Job is already complete; its recorded status was not changed."
+      return 0
+      ;;
+    unreachable)
+      die "Cannot safely stop job $name because its remote status is unreachable."
+      ;;
+  esac
+
+  if [ "${backend:-detach}" = "tmux" ]; then
+    [ -n "$tmux_session" ] && [ -n "$tmux_socket" ] || die "tmux job has incomplete session metadata: $name"
+    remote_cmd="tmux -S $(shell_quote "$tmux_socket") kill-session -t $(shell_quote "$tmux_session") 2>/dev/null || true; finished_at=\$(date +%Y-%m-%dT%H:%M:%S%z); printf 'status=cancelled\npid=%s\nfinished_at=%s\nexit_code=143\nlog=%s\ncwd=%s\n' $(shell_quote "$pid") \"\$finished_at\" $(shell_quote "$log") $(shell_quote "$cwd") > $(shell_quote "$status_path"); printf 'status=cancelled\npid=%s\nbackend=tmux\n' $(shell_quote "$pid")"
+  else
+    if [ -z "$process_group" ]; then
+      group_probe="$(ssh_capture "pid=$(shell_quote "$pid"); pgid=\$(ps -o pgid= -p \"\$pid\" 2>/dev/null | tr -d '[:space:]'); if [ -n \"\$pgid\" ]; then [ \"\$pgid\" = \"\$pid\" ] && echo yes || echo no; elif kill -0 -- -\"\$pid\" 2>/dev/null; then echo yes; else echo no; fi" 2>/dev/null)" || return $?
+      [ "$group_probe" = "yes" ] || die "Legacy job $name is not an isolated process group, so it cannot be stopped safely."
+      process_group=yes
+    fi
+    [ "$process_group" = "yes" ] || die "Job $name was not launched in an isolated process group, so it cannot be stopped safely."
+    remote_cmd="pid=$(shell_quote "$pid"); status_path=$(shell_quote "$status_path"); log=$(shell_quote "$log"); cwd=$(shell_quote "$cwd"); group_alive() { if command -v ps >/dev/null 2>&1; then ps -eo pgid=,stat= 2>/dev/null | awk -v pgid=\"\$pid\" '\$1 == pgid && \$2 !~ /^Z/ { found=1 } END { exit !found }'; else kill -0 -- -\"\$pid\" 2>/dev/null; fi; }; if group_alive; then kill -TERM -- -\"\$pid\" || exit 1; for _ in 1 2 3 4 5; do group_alive || break; sleep 1; done; if group_alive; then kill -KILL -- -\"\$pid\" || exit 1; for _ in 1 2 3 4 5; do group_alive || break; sleep 1; done; fi; group_alive && { echo 'autodl-remote: process group remained alive after TERM and KILL' >&2; exit 1; }; fi; finished_at=\$(date +%Y-%m-%dT%H:%M:%S%z); printf 'status=cancelled\npid=%s\nfinished_at=%s\nexit_code=143\nlog=%s\ncwd=%s\n' \"\$pid\" \"\$finished_at\" \"\$log\" \"\$cwd\" > \"\$status_path\"; printf 'status=cancelled\npid=%s\nbackend=detach\n' \"\$pid\""
+  fi
+  output="$(ssh_capture "bash -lc $(shell_quote "$remote_cmd")" 2>&1)"
+  rc=$?
+  if [ "$rc" -eq 0 ]; then
+    printf '%s\n' "$output"
+  else
+    printf '%s\n' "$output" >&2
+    return "$rc"
+  fi
+}
+
 cmd_job_tail() {
   load_project
-  local lines=120 follow=false
+  local lines=120 follow=false name=""
   while [ $# -gt 0 ]; do
     case "$1" in
       --lines) lines="$2"; shift 2 ;;
       --follow|-f) follow=true; shift ;;
-      --) shift; break ;;
-      *) break ;;
+      --)
+        shift
+        [ -z "$name" ] && [ $# -gt 0 ] && { name="$1"; shift; }
+        [ $# -eq 0 ] || die "unknown job tail argument: $1"
+        break
+        ;;
+      *)
+        [ -z "$name" ] || die "unknown job tail argument: $1"
+        name="$1"
+        shift
+        ;;
     esac
   done
-  [ $# -gt 0 ] || die "job tail requires a name"
-  local name file log backend tmux_session tmux_socket flag=""
-  name="$(sanitize_name "$1")"
+  validate_tail_settings "$lines"
+  [ -n "$name" ] || die "job tail requires a name"
+  local file log backend tmux_session tmux_socket flag=""
+  name="$(sanitize_name "$name")"
   file="$(job_path "$name")"
   [ -f "$file" ] || die "Job not found: $name"
   log="$(job_get "$file" LOG_PATH)"
@@ -1405,7 +1769,11 @@ cmd_job_tail() {
     info "tmux capture failed; falling back to log tail."
   fi
   [ "$follow" = "true" ] && flag="-f"
-  ssh_run "tail -n $lines $flag $(shell_quote "$log")"
+  if [ "$follow" = "true" ]; then
+    ssh_run "tail -n $lines $flag $(shell_quote "$log")"
+  else
+    ssh_run "$(normalized_log_tail_command "$log" "$lines")"
+  fi
 }
 
 cmd_job() {
@@ -1415,6 +1783,7 @@ cmd_job() {
   case "$sub" in
     list|ls) cmd_job_list "$@" ;;
     status) cmd_job_status "$@" ;;
+    stop) cmd_job_stop "$@" ;;
     tail) cmd_job_tail "$@" ;;
     *) die "unknown job command: $sub" ;;
   esac
@@ -1518,6 +1887,7 @@ cmd_run() {
     submit) cmd_run_submit "$@" ;;
     list|ls) cmd_run_list "$@" ;;
     status) cmd_job_status "$@" ;;
+    stop) cmd_job_stop "$@" ;;
     tail) cmd_job_tail "$@" ;;
     note) cmd_run_note "$@" ;;
     *) die "unknown run command: $sub" ;;
@@ -1666,7 +2036,7 @@ remote_log_tail() {
   load_account "$account"
   REMOTE_ROOT="${remote%/}"
   validate_remote_root "$REMOTE_ROOT"
-  output="$(ssh_capture "if [ -f $(shell_quote "$log") ]; then tail -n $lines $(shell_quote "$log"); else printf '[log missing] %s\n' $(shell_quote "$log"); fi" 2>&1)"
+  output="$(ssh_capture "if [ -f $(shell_quote "$log") ]; then $(normalized_log_tail_command "$log" "$lines"); else printf '[log missing] %s\n' $(shell_quote "$log"); fi" 2>&1)"
   rc=$?
   if [ "$rc" -eq 0 ]; then
     printf '%s\n' "$output"
@@ -2145,6 +2515,7 @@ cmd_dashboard() {
   case "$lines" in
     ""|*[!0-9]*) die "--lines must be a non-negative integer" ;;
   esac
+  validate_tail_settings "$lines"
   if [ -n "$watch_seconds" ]; then
     case "$watch_seconds" in
       ""|0|*[!0-9]*) die "--watch must be a positive integer number of seconds" ;;
@@ -2233,6 +2604,7 @@ cmd_tmux_capture() {
   case "$lines" in
     ""|*[!0-9]*) die "--lines must be a non-negative integer" ;;
   esac
+  validate_tail_settings "$lines"
   [ $# -gt 0 ] || die "tmux capture requires a session name"
   session="$(sanitize_name "$1")"
   apply_context_override "$account_override" "$remote_override"
@@ -2364,8 +2736,9 @@ cmd_put_run() {
 
 cmd_shell() {
   load_project
+  ensure_ssh_master || return $?
   ssh_common_args
-  run_transport ssh -t "${SSH_ARGS[@]}" "$ACCOUNT_TARGET" "cd $(shell_quote "$REMOTE_ROOT") && exec bash -l"
+  ssh -t "${SSH_ARGS[@]}" "$ACCOUNT_TARGET" "cd $(shell_quote "$REMOTE_ROOT") && exec bash -l"
 }
 
 cmd_history() {
diff --git a/plugins/autodl-remote/skills/autodl-remote/SKILL.md b/plugins/autodl-remote/skills/autodl-remote/SKILL.md
index 2626ec1..0188c09 100644
--- a/plugins/autodl-remote/skills/autodl-remote/SKILL.md
+++ b/plugins/autodl-remote/skills/autodl-remote/SKILL.md
@@ -80,8 +80,10 @@ autodl-remote exec --script scripts/remote_check.sh
 cat scripts/remote_check.sh | autodl-remote exec --stdin -- bash
 autodl-remote job list
 autodl-remote job status train
-autodl-remote job tail train
+autodl-remote job tail train --lines 120
+autodl-remote job stop train
 autodl-remote run list
+autodl-remote run stop train
 autodl-remote run note train --stage evaluating
 autodl-remote fleet create rag-exp
 autodl-remote fleet add rag-exp nmb1 --account autodl-nmb1 --remote /root/autodl-tmp/LLM-RAG --tags a800,rag
@@ -102,9 +104,9 @@ autodl-remote shutdown
 - If both local and remote have similar files, do not assume one should overwrite the other. Compare or inspect first, then choose `put` or `get`.
 - Use `exec --detach` for long training jobs and then `tail` the log path.
 - Use `exec --tmux --name <run>` when live pane visibility matters, when several experiments run in parallel, or when multiple hosts need clearer terminal-state monitoring.
-- Prefer `job status <name>` and `job tail <name>` for detached jobs created with `--name`.
+- Prefer `job status <name>`, `job tail <name>`, and `job stop <name>` for detached jobs created with `--name`. Do not reuse an active name unless `exec --replace` explicitly stops the recorded job first.
 - Prefer `tmux capture <name>` for tmux-backed jobs when the user asks what the remote terminal currently shows.
-- Prefer `run list`, `run status`, `run tail`, and `run note` when the user is thinking in terms of experiments/models rather than raw jobs.
+- Prefer `run list`, `run status`, `run tail`, `run stop`, and `run note` when the user is thinking in terms of experiments/models rather than raw jobs.
 - For multiple machines, use `fleet add/list/status` by default. Use `dashboard --fleet <name>` only after the user asks for a dashboard, overview, or live monitoring.
 - When inspecting a non-default fleet device, pass `--account` and `--remote` to `tree`, `ls`, `cat`, `tail`, `put`, `get`, and `exec` instead of rebinding the project.
 - The dashboard is read-only and can be used for one device or many devices. Do not add operational assumptions just because the dashboard exists.
diff --git a/tests/test-autodl-remote.sh b/tests/test-autodl-remote.sh
new file mode 100755
index 0000000..acf7b49
--- /dev/null
+++ b/tests/test-autodl-remote.sh
@@ -0,0 +1,595 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+CLI="${AUTODL_REMOTE_TEST_CLI:-$ROOT_DIR/plugins/autodl-remote/bin/autodl-remote}"
+TMP_ROOT="$(mktemp -d "${TMPDIR:-/tmp}/autodl-remote-test.XXXXXX")"
+cleanup() {
+  if [ "${KEEP_TEST_TMP:-}" = "1" ]; then
+    printf 'Kept test directory: %s\n' "$TMP_ROOT" >&2
+  else
+    rm -rf "$TMP_ROOT"
+  fi
+}
+trap cleanup EXIT
+
+HOME_DIR="$TMP_ROOT/home"
+PROJECT_DIR="$TMP_ROOT/project"
+FAKE_BIN="$TMP_ROOT/bin"
+FAKE_REMOTE_BIN="$TMP_ROOT/remote-bin"
+REMOTE_DIR="$TMP_ROOT/remote"
+mkdir -p "$HOME_DIR/.autodl-remote/accounts" "$PROJECT_DIR/.autodl-remote/jobs" "$FAKE_BIN" "$FAKE_REMOTE_BIN" "$REMOTE_DIR"
+
+cat > "$HOME_DIR/.autodl-remote/accounts/test.conf" <<EOF
+TARGET="test@example.invalid"
+PORT="22"
+KEY=""
+AUTH="prompt"
+DEFAULT_REMOTE="$REMOTE_DIR"
+EOF
+
+cat > "$PROJECT_DIR/.autodl-remote.conf" <<EOF
+ACCOUNT="test"
+REMOTE_ROOT="$REMOTE_DIR"
+EOF
+
+cat > "$FAKE_BIN/ssh" <<'EOF'
+#!/usr/bin/env bash
+set -euo pipefail
+if [ -n "${FAKE_SSH_ARGS_LOG:-}" ]; then
+  printf '%s\n' "$*" >> "$FAKE_SSH_ARGS_LOG"
+fi
+if [ -n "${FAKE_SSH_OUTPUT_FILE:-}" ]; then
+  exec /bin/cat "$FAKE_SSH_OUTPUT_FILE"
+fi
+if [ -n "${FAKE_SSH_STATE_DIR:-}" ]; then
+  state_dir="${FAKE_SSH_STATE_DIR:?}"
+  mkdir -p "$state_dir"
+  control_master="auto"
+  operation=""
+  previous=""
+  master_start=false
+  for argument in "$@"; do
+    if [ "$previous" = "-O" ]; then
+      operation="$argument"
+    elif [ "$previous" = "-o" ]; then
+      case "$argument" in
+        ControlMaster=*) control_master="${argument#ControlMaster=}" ;;
+      esac
+    fi
+    case "$argument" in
+      -*M*N*f*|-*M*f*N*|-*N*M*f*|-*N*f*M*|-*f*M*N*|-*f*N*M*) master_start=true ;;
+    esac
+    previous="$argument"
+  done
+
+  if [ "$operation" = "check" ]; then
+    [ -f "$state_dir/master" ] && exit 0
+    exit 255
+  fi
+
+  create_master() {
+    if [ "${FAKE_SSH_FAIL_MASTER_ONCE:-}" = "1" ] && [ ! -f "$state_dir/failed-once" ]; then
+      : > "$state_dir/failed-once"
+      return 255
+    fi
+    if mkdir "$state_dir/creating" 2>/dev/null; then
+      if [ "${FAKE_SSH_CONTROL_RACE:-}" = "1" ]; then
+        sleep 0.2
+      fi
+      : > "$state_dir/master"
+      rmdir "$state_dir/creating"
+      return 0
+    fi
+    sleep 5
+    return 255
+  }
+
+  if [ "$master_start" = "true" ] || [ "$control_master" = "yes" ]; then
+    if [ "${FAKE_SSH_HANG_MASTER:-}" = "1" ]; then
+      exec sleep 5
+    fi
+    create_master
+    exit $?
+  fi
+  if [ "$control_master" = "auto" ] && [ ! -f "$state_dir/master" ]; then
+    create_master || exit $?
+  fi
+fi
+if [ "${FAKE_SSH_HANG_SESSION:-}" = "1" ]; then
+  exec sleep 5
+fi
+if [ "${FAKE_SSH_HANG_SESSION:-}" = "child" ]; then
+  (
+    sleep 5 &
+    grandchild=$!
+    while [ ! -f "${FAKE_SSH_CHILD_PID_FILE:?}" ]; do sleep 0.01; done
+    printf '%s\n' "$grandchild" >> "${FAKE_SSH_CHILD_PID_FILE:?}"
+    wait "$grandchild"
+  ) &
+  child=$!
+  printf '%s\n' "$child" > "${FAKE_SSH_CHILD_PID_FILE:?}"
+  wait "$child"
+  exit $?
+fi
+command="${!#}"
+if [ -n "${FAKE_SSH_REMOTE_BIN:-}" ]; then
+  PATH="$FAKE_SSH_REMOTE_BIN:$PATH"
+  export PATH
+fi
+exec bash -c "$command"
+EOF
+chmod +x "$FAKE_BIN/ssh"
+
+cat > "$FAKE_BIN/scp" <<'EOF'
+#!/usr/bin/env bash
+set -euo pipefail
+source_path="${@: -2:1}"
+target_path="${!#}"
+target_path="${target_path#*:}"
+mkdir -p "$(dirname "$target_path")"
+cp "$source_path" "$target_path"
+EOF
+chmod +x "$FAKE_BIN/scp"
+
+cat > "$FAKE_REMOTE_BIN/setsid" <<'EOF'
+#!/usr/bin/env python3
+import os
+import sys
+
+os.setsid()
+os.execvp(sys.argv[1], sys.argv[1:])
+EOF
+chmod +x "$FAKE_REMOTE_BIN/setsid"
+
+cat > "$FAKE_BIN/security" <<'EOF'
+#!/usr/bin/env bash
+if [ "${FAKE_SECURITY_HANG:-}" = "1" ]; then
+  exec sleep 5
+fi
+printf 'test-password\n'
+EOF
+chmod +x "$FAKE_BIN/security"
+
+cat > "$FAKE_BIN/tmux" <<'EOF'
+#!/usr/bin/env bash
+set -euo pipefail
+case " $* " in
+  *" has-session "*) exit 0 ;;
+  *" capture-pane "*) exec /bin/cat "${FAKE_TMUX_OUTPUT_FILE:?}" ;;
+  *) exit 2 ;;
+esac
+EOF
+chmod +x "$FAKE_BIN/tmux"
+
+write_job() {
+  local name="$1" log="$2" backend="${3:-detach}"
+  cat > "$PROJECT_DIR/.autodl-remote/jobs/$name.conf" <<EOF
+NAME="$name"
+PID="1234"
+ACCOUNT="test"
+REMOTE_ROOT="$REMOTE_DIR"
+LOG_PATH="$log"
+STATUS_PATH="$REMOTE_DIR/$name.status"
+CWD="$REMOTE_DIR"
+STARTED_AT="2026-07-13T00:00:00-0700"
+COMMAND="python train.py"
+BACKEND="$backend"
+TMUX_SESSION="${name}-session"
+TMUX_SOCKET="$REMOTE_DIR/tmux.sock"
+EOF
+}
+
+run_cli() {
+  (
+    cd "$PROJECT_DIR"
+    HOME="$HOME_DIR" PATH="$FAKE_BIN:$PATH" AUTODL_REMOTE_CONTROL_DIR="$TMP_ROOT/control" FAKE_SSH_REMOTE_BIN="$FAKE_REMOTE_BIN" FAKE_SSH_STATE_DIR="${FAKE_SSH_STATE_DIR:-$TMP_ROOT/default-ssh-state}" "$CLI" "$@"
+  )
+}
+
+assert_eq() {
+  local expected="$1" actual="$2" label="$3"
+  if [ "$actual" != "$expected" ]; then
+    printf 'FAIL: %s\nexpected:\n%s\nactual:\n%s\n' "$label" "$expected" "$actual" >&2
+    return 1
+  fi
+}
+
+assert_le() {
+  local actual="$1" limit="$2" label="$3"
+  if [ "$actual" -gt "$limit" ]; then
+    printf 'FAIL: %s (actual=%s limit=%s)\n' "$label" "$actual" "$limit" >&2
+    return 1
+  fi
+}
+
+assert_contains() {
+  local needle="$1" haystack="$2" label="$3"
+  if ! printf '%s' "$haystack" | grep -Fq -- "$needle"; then
+    printf 'FAIL: %s\nmissing: %s\nactual:\n%s\n' "$label" "$needle" "$haystack" >&2
+    return 1
+  fi
+}
+
+wait_for_file_content() {
+  local file="$1" label="$2"
+  for _ in $(seq 1 100); do
+    [ -s "$file" ] && return 0
+    sleep 0.05
+  done
+  printf 'FAIL: timed out waiting for %s (%s)\n' "$label" "$file" >&2
+  return 1
+}
+
+wait_for_matching_file_content() {
+  local file="$1" pattern="$2" label="$3"
+  for _ in $(seq 1 100); do
+    grep -Fq -- "$pattern" "$file" 2>/dev/null && return 0
+    sleep 0.05
+  done
+  printf 'FAIL: timed out waiting for %s (%s)\n' "$label" "$file" >&2
+  return 1
+}
+
+test_job_tail_counts_logical_lines() {
+  local log="$REMOTE_DIR/tqdm.log" output
+  printf '\033[31mepoch 1\033[0m\repoch 2\r\nepoch 3\n\033[32mepoch 4\033[0m\repoch 5' > "$log"
+  write_job train "$log"
+
+  output="$(run_cli job tail --lines 3 train)"
+  assert_eq $'epoch 3\nepoch 4\nepoch 5' "$output" "job tail selects the last N normalized logical lines"
+}
+
+test_job_tail_caps_long_output() {
+  local log="$REMOTE_DIR/long.log" output bytes
+  awk 'BEGIN { printf "\033[35m"; for (i = 0; i < 70000; i++) printf "x"; printf "END\033[0m" }' > "$log"
+  write_job long "$log"
+
+  output="$(run_cli job tail --lines 1 long)"
+  bytes="$(LC_ALL=C printf '%s' "$output" | wc -c | tr -d ' ')"
+  assert_le "$bytes" 65536 "job tail output has a hard byte ceiling"
+  assert_eq "END" "${output: -3}" "job tail keeps the newest bytes"
+  if printf '%s' "$output" | LC_ALL=C grep -q $'\033'; then
+    printf 'FAIL: job tail strips ANSI control sequences\n' >&2
+    return 1
+  fi
+}
+
+test_tmux_job_tail_uses_the_same_output_policy() {
+  local fixture="$REMOTE_DIR/tmux-pane.log" output bytes
+  awk 'BEGIN { printf "\033[36m"; for (i = 0; i < 70000; i++) printf "y"; printf "END\033[0m" }' > "$fixture"
+  write_job tmux-run "$REMOTE_DIR/tmux-run.log" tmux
+
+  output="$(FAKE_TMUX_OUTPUT_FILE="$fixture" run_cli job tail --lines 1 tmux-run)"
+  bytes="$(LC_ALL=C printf '%s' "$output" | wc -c | tr -d ' ')"
+  assert_le "$bytes" 65536 "tmux job tail output has the same byte ceiling"
+  assert_eq "END" "${output: -3}" "tmux job tail keeps the newest bytes"
+  if printf '%s' "$output" | LC_ALL=C grep -q $'\033'; then
+    printf 'FAIL: tmux job tail strips ANSI control sequences\n' >&2
+    return 1
+  fi
+}
+
+test_invalid_tail_limit_fails_in_the_parent_process() {
+  local output rc
+  set +e
+  output="$(AUTODL_REMOTE_TAIL_MAX_BYTES=invalid run_cli job tail --lines 3 train 2>&1)"
+  rc=$?
+  set -e
+  assert_eq "2" "$rc" "invalid tail byte limit is rejected"
+  if ! printf '%s' "$output" | grep -q 'AUTODL_REMOTE_TAIL_MAX_BYTES must be a positive integer'; then
+    printf 'FAIL: invalid tail byte limit reports the expected diagnostic\n%s\n' "$output" >&2
+    return 1
+  fi
+}
+
+test_job_tail_accepts_options_after_name() {
+  local log="$REMOTE_DIR/tail-position.log" output rc
+  printf 'one\ntwo\nthree\n' > "$log"
+  write_job tail-position "$log"
+
+  output="$(run_cli job tail tail-position --lines 2)"
+  assert_eq $'two\nthree' "$output" "job tail accepts --lines after the job name"
+
+  set +e
+  output="$(run_cli job tail tail-position --unexpected 2>&1)"
+  rc=$?
+  set -e
+  assert_eq "2" "$rc" "job tail rejects unknown trailing arguments"
+  assert_contains 'unknown job tail argument: --unexpected' "$output" "job tail explains an unknown trailing argument"
+}
+
+test_detached_direct_argv_preserves_shell_metacharacters() {
+  local name="quoted-argv" log output
+  log="$REMOTE_DIR/.autodl-remote/logs/$name.log"
+  output="$(run_cli exec --detach --name "$name" -- printf '%s\n' '<|endoftext|>')"
+  wait_for_file_content "$log" "quoted detached command output"
+  assert_eq '<|endoftext|>' "$(tr -d '\n' < "$log")" "detached direct argv preserves literal shell metacharacters"
+  assert_contains 'exit_code=0' "$(run_cli job status "$name")" "quoted detached command exits successfully"
+}
+
+test_detached_stdin_preserves_literal_shell_metacharacters() {
+  local name="quoted-stdin" log output
+  log="$REMOTE_DIR/.autodl-remote/logs/$name.log"
+  output="$(printf "%s\\n" "printf '%s\\n' '<|endoftext|>'" | run_cli exec --detach --stdin --name "$name" -- bash)"
+  wait_for_file_content "$log" "quoted stdin detached command output"
+  assert_eq '<|endoftext|>' "$(tr -d '\n' < "$log")" "detached stdin preserves literal shell metacharacters"
+  assert_contains 'exit_code=0' "$(run_cli job status "$name")" "quoted stdin detached command exits successfully"
+}
+
+test_job_stop_terminates_the_whole_detached_process_group() {
+  local name="stop-group" log child_pid output
+  log="$REMOTE_DIR/.autodl-remote/logs/$name.log"
+  output="$(run_cli exec --detach --name "$name" -- bash -c 'sleep 30 & child=$!; echo "$child"; wait "$child"')"
+  wait_for_file_content "$log" "detached child pid"
+  child_pid="$(head -n 1 "$log")"
+  case "$child_pid" in
+    ''|*[!0-9]*) printf 'FAIL: detached child pid is invalid: %s\n%s\n' "$child_pid" "$output" >&2; return 1 ;;
+  esac
+
+  output="$(run_cli job stop "$name")"
+  assert_contains 'status=cancelled' "$output" "job stop reports cancellation"
+  for _ in $(seq 1 40); do
+    kill -0 "$child_pid" 2>/dev/null || break
+    sleep 0.05
+  done
+  if kill -0 "$child_pid" 2>/dev/null; then
+    printf 'FAIL: job stop left the detached child running (pid=%s)\n' "$child_pid" >&2
+    return 1
+  fi
+  assert_contains 'status=cancelled' "$(run_cli job status "$name")" "runner records cancelled after job stop"
+}
+
+test_job_stop_preserves_completed_job_status() {
+  local name="already-exited" status output
+  status="$REMOTE_DIR/.autodl-remote/jobs/$name.status"
+  run_cli exec --detach --name "$name" -- printf 'done\n' >/dev/null
+  wait_for_matching_file_content "$status" 'status=exited' "completed detached job status"
+
+  output="$(run_cli job stop "$name")"
+  assert_contains 'status=exited' "$output" "stopping a completed job is a non-mutating no-op"
+  output="$(run_cli job status "$name")"
+  assert_contains 'status=exited' "$output" "completed job remains exited after stop"
+  assert_contains 'exit_code=0' "$output" "completed job preserves its exit code after stop"
+}
+
+test_legacy_detached_job_stop_detects_its_process_group() {
+  local name="legacy-stop" file output
+  file="$PROJECT_DIR/.autodl-remote/jobs/$name.conf"
+  run_cli exec --detach --name "$name" -- sleep 30 >/dev/null
+  sed '/^PROCESS_GROUP=/d' "$file" > "$file.tmp"
+  mv "$file.tmp" "$file"
+
+  output="$(run_cli job stop "$name")"
+  assert_contains 'status=cancelled' "$output" "legacy detach job is stopped after PGID detection"
+}
+
+test_active_name_collision_requires_replace_and_stops_the_old_group() {
+  local name="collision" file first_pid second_pid output rc
+  file="$PROJECT_DIR/.autodl-remote/jobs/$name.conf"
+  run_cli exec --detach --name "$name" -- sleep 30 >/dev/null
+  first_pid="$(awk -F= '$1=="PID" { gsub(/"/, "", $2); print $2; exit }' "$file")"
+
+  set +e
+  output="$(run_cli exec --detach --name "$name" -- sleep 30 2>&1)"
+  rc=$?
+  set -e
+  assert_eq "2" "$rc" "active job name collision is rejected"
+  assert_contains 'already active' "$output" "active job name collision explains the rejection"
+  assert_eq "$first_pid" "$(awk -F= '$1=="PID" { gsub(/"/, "", $2); print $2; exit }' "$file")" "collision does not overwrite the existing metadata"
+
+  run_cli exec --detach --replace --name "$name" -- sleep 30 >/dev/null
+  second_pid="$(awk -F= '$1=="PID" { gsub(/"/, "", $2); print $2; exit }' "$file")"
+  if [ "$first_pid" = "$second_pid" ]; then
+    printf 'FAIL: --replace did not record a new runner pid\n' >&2
+    return 1
+  fi
+  for _ in $(seq 1 40); do
+    kill -0 -- "-$first_pid" 2>/dev/null || break
+    sleep 0.05
+  done
+  if kill -0 -- "-$first_pid" 2>/dev/null; then
+    printf 'FAIL: --replace left the old process group running (pgid=%s)\n' "$first_pid" >&2
+    return 1
+  fi
+  run_cli job stop "$name" >/dev/null
+}
+
+test_running_status_with_dead_runner_is_lost() {
+  local name="lost-runner" file status output
+  file="$PROJECT_DIR/.autodl-remote/jobs/$name.conf"
+  status="$REMOTE_DIR/$name.status"
+  write_job "$name" "$REMOTE_DIR/lost.log"
+  sed 's/^PID="1234"/PID="999999"/' "$file" > "$file.tmp"
+  mv "$file.tmp" "$file"
+  printf 'status=running\npid=999999\n' > "$status"
+
+  output="$(run_cli job status "$name")"
+  assert_contains 'status=lost' "$output" "job status marks a dead runner as lost"
+  assert_contains 'hint=runner pid is not alive' "$output" "job status includes a recovery hint for a lost job"
+  assert_contains 'lost' "$(run_cli run list)" "run list uses the same lost-state normalization"
+  assert_contains 'status=lost' "$(run_cli run status "$name")" "run status uses the same lost-state normalization"
+}
+
+test_captured_ssh_output_preserves_carriage_returns() {
+  local fixture="$TMP_ROOT/ssh-output.bin" output
+  printf 'alpha\rbeta\n' > "$fixture"
+  output="$(FAKE_SSH_OUTPUT_FILE="$fixture" run_cli account test test)"
+  assert_eq $'Testing test -> test@example.invalid:22\nalpha\rbeta' "$output" "generic captured SSH output round-trips carriage returns"
+}
+
+test_parallel_cold_start_finishes_before_deadline() {
+  local state_dir="$TMP_ROOT/ssh-state" output_dir="$TMP_ROOT/concurrent-output" pid alive rc
+  local pids=()
+  mkdir -p "$state_dir" "$output_dir"
+  rm -f "$state_dir/master"
+
+  for index in 1 2 3 4 5 6 7 8; do
+    (
+      FAKE_SSH_CONTROL_RACE=1 FAKE_SSH_STATE_DIR="$state_dir" run_cli account test test
+    ) > "$output_dir/$index" 2>&1 &
+    pids+=("$!")
+  done
+
+  for _ in $(seq 1 60); do
+    if [ -n "$(jobs -pr)" ]; then alive=true; else alive=false; fi
+    [ "$alive" = "true" ] || break
+    sleep 0.05
+  done
+
+  if [ "$alive" = "true" ]; then
+    kill "${pids[@]}" 2>/dev/null || true
+    wait "${pids[@]}" 2>/dev/null || true
+    printf 'FAIL: parallel cold-start calls exceeded the three-second deadline\n' >&2
+    return 1
+  fi
+
+  rc=0
+  for pid in "${pids[@]}"; do
+    wait "$pid" || rc=$?
+  done
+  assert_eq "0" "$rc" "parallel cold-start calls all succeed"
+}
+
+test_monitor_call_has_total_timeout() {
+  local account_file="$HOME_DIR/.autodl-remote/accounts/test.conf" child_pid_file="$TMP_ROOT/ssh-child.pid" child_pid output rc
+  sed 's/^AUTH=.*/AUTH="ssh-key"/; s#^KEY=.*#KEY="/tmp/fake-key"#' "$account_file" > "$account_file.tmp"
+  mv "$account_file.tmp" "$account_file"
+
+  set +e
+  output="$(FAKE_SSH_HANG_SESSION=child FAKE_SSH_CHILD_PID_FILE="$child_pid_file" AUTODL_REMOTE_MONITOR_TIMEOUT=1 run_cli account test test 2>&1)"
+  rc=$?
+  set -e
+
+  sed 's/^AUTH=.*/AUTH="prompt"/; s#^KEY=.*#KEY=""#' "$account_file" > "$account_file.tmp"
+  mv "$account_file.tmp" "$account_file"
+  assert_eq "124" "$rc" "read-only monitoring call returns the timeout status"
+  if ! printf '%s' "$output" | grep -q 'transport timed out after 1s'; then
+    printf 'FAIL: monitor timeout reports a clear diagnostic\n%s\n' "$output" >&2
+    return 1
+  fi
+  while IFS= read -r child_pid; do
+    if kill -0 "$child_pid" 2>/dev/null; then
+      printf 'FAIL: monitor timeout left a descendant process running (pid=%s)\n' "$child_pid" >&2
+      return 1
+    fi
+  done < "$child_pid_file"
+}
+
+test_keychain_master_start_times_out() {
+  local account_file="$HOME_DIR/.autodl-remote/accounts/test.conf" state_dir="$TMP_ROOT/keychain-state" output rc
+  mkdir -p "$state_dir"
+  sed 's/^AUTH=.*/AUTH="keychain"/' "$account_file" > "$account_file.tmp"
+  mv "$account_file.tmp" "$account_file"
+
+  set +e
+  output="$(FAKE_SSH_HANG_MASTER=1 FAKE_SSH_STATE_DIR="$state_dir" AUTODL_REMOTE_CONNECT_TIMEOUT=1 run_cli account test test 2>&1)"
+  rc=$?
+  set -e
+
+  sed 's/^AUTH=.*/AUTH="prompt"/' "$account_file" > "$account_file.tmp"
+  mv "$account_file.tmp" "$account_file"
+  assert_eq "124" "$rc" "keychain master startup returns the timeout status"
+  if ! printf '%s' "$output" | grep -q 'SSH transport timed out'; then
+    printf 'FAIL: keychain timeout reports a clear diagnostic\n%s\n' "$output" >&2
+    return 1
+  fi
+}
+
+test_keychain_lookup_is_inside_the_timeout() {
+  local account_file="$HOME_DIR/.autodl-remote/accounts/test.conf" state_dir="$TMP_ROOT/keychain-lookup-state" output rc
+  mkdir -p "$state_dir"
+  sed 's/^AUTH=.*/AUTH="keychain"/' "$account_file" > "$account_file.tmp"
+  mv "$account_file.tmp" "$account_file"
+
+  set +e
+  output="$(FAKE_SECURITY_HANG=1 FAKE_SSH_STATE_DIR="$state_dir" AUTODL_REMOTE_CONNECT_TIMEOUT=1 run_cli account test test 2>&1)"
+  rc=$?
+  set -e
+
+  sed 's/^AUTH=.*/AUTH="prompt"/' "$account_file" > "$account_file.tmp"
+  mv "$account_file.tmp" "$account_file"
+  assert_eq "124" "$rc" "keychain lookup returns the timeout status"
+  if ! printf '%s' "$output" | grep -q 'transport timed out after 1s'; then
+    printf 'FAIL: keychain lookup timeout reports a clear diagnostic\n%s\n' "$output" >&2
+    return 1
+  fi
+}
+
+test_connection_options_and_master_retry() {
+  local state_dir="$TMP_ROOT/retry-state" args_log="$TMP_ROOT/ssh-args.log" output master_line
+  mkdir -p "$state_dir"
+  output="$(FAKE_SSH_FAIL_MASTER_ONCE=1 FAKE_SSH_STATE_DIR="$state_dir" FAKE_SSH_ARGS_LOG="$args_log" AUTODL_REMOTE_CONNECT_TIMEOUT=7 AUTODL_REMOTE_CONNECTION_ATTEMPTS=4 run_cli account test test)"
+  if ! printf '%s' "$output" | grep -q '^host='; then
+    printf 'FAIL: command did not recover after the first master start failure\n' >&2
+    return 1
+  fi
+  if ! grep -q 'ConnectTimeout=7' "$args_log" || ! grep -q 'ConnectionAttempts=4' "$args_log"; then
+    printf 'FAIL: explicit SSH connection bounds were not forwarded\n' >&2
+    return 1
+  fi
+  master_line="$(grep -- '-MNf' "$args_log" | tail -n 1)"
+  if ! printf '%s' "$master_line" | grep -q 'ControlMaster=yes' || printf '%s' "$master_line" | grep -q 'ControlMaster=no'; then
+    printf 'FAIL: master startup must contain only ControlMaster=yes\n%s\n' "$master_line" >&2
+    return 1
+  fi
+}
+
+test_stale_control_state_is_recovered() {
+  local state_dir account_id connection_id control_path output kind
+  mkdir -p "$TMP_ROOT/control"
+  account_id="$(printf '%s\n' 'test' | cksum | awk '{print $1}')"
+  connection_id="$(printf '%s\n' 'test@example.invalid:22' | cksum | awk '{print $1}')"
+  control_path="$TMP_ROOT/control/ssh-$account_id-$connection_id"
+  for kind in dead-pid missing-pid malformed-pid; do
+    state_dir="$TMP_ROOT/stale-$kind"
+    mkdir -p "$state_dir"
+    : > "$control_path"
+    mkdir "$control_path.lock"
+    case "$kind" in
+      dead-pid) printf '999999\n' > "$control_path.lock/pid" ;;
+      malformed-pid) printf 'not-a-pid\n' > "$control_path.lock/pid" ;;
+    esac
+
+    if [ "$kind" = "dead-pid" ]; then
+      output="$(FAKE_SSH_STATE_DIR="$state_dir" run_cli account test test)"
+    else
+      output="$(FAKE_SSH_STATE_DIR="$state_dir" AUTODL_REMOTE_MASTER_LOCK_TIMEOUT=1 run_cli account test test)"
+    fi
+    if ! printf '%s' "$output" | grep -q '^host='; then
+      printf 'FAIL: stale control state was not recovered (%s)\n' "$kind" >&2
+      return 1
+    fi
+    if [ -e "$control_path" ] || [ -e "$control_path.lock" ]; then
+      printf 'FAIL: stale socket or lock remains after recovery (%s)\n' "$kind" >&2
+      return 1
+    fi
+  done
+}
+
+run_test() {
+  local name="$1"
+  [ -z "${AUTODL_REMOTE_TEST_FILTER:-}" ] || [ "$AUTODL_REMOTE_TEST_FILTER" = "$name" ] || return 0
+  "$name"
+  printf 'PASS: %s\n' "$name"
+}
+
+run_test test_job_tail_counts_logical_lines
+run_test test_job_tail_caps_long_output
+run_test test_tmux_job_tail_uses_the_same_output_policy
+run_test test_invalid_tail_limit_fails_in_the_parent_process
+run_test test_job_tail_accepts_options_after_name
+run_test test_detached_direct_argv_preserves_shell_metacharacters
+run_test test_detached_stdin_preserves_literal_shell_metacharacters
+run_test test_job_stop_terminates_the_whole_detached_process_group
+run_test test_job_stop_preserves_completed_job_status
+run_test test_legacy_detached_job_stop_detects_its_process_group
+run_test test_active_name_collision_requires_replace_and_stops_the_old_group
+run_test test_running_status_with_dead_runner_is_lost
+run_test test_captured_ssh_output_preserves_carriage_returns
+run_test test_parallel_cold_start_finishes_before_deadline
+run_test test_monitor_call_has_total_timeout
+run_test test_keychain_master_start_times_out
+run_test test_keychain_lookup_is_inside_the_timeout
+run_test test_connection_options_and_master_retry
+run_test test_stale_control_state_is_recovered