docs: add L1+L2 check implementation record

Document the two-layer short-circuit probe added in 6124abb: design
rationale, the historical lesson from the removed _is_port_in_use bind
check, test coverage, real-world verification results on both local
machine and NAS, compatibility, deployment, and rollback procedure.

docs/check-l1-l2-implementation.md

@@ -0,0 +1,284 @@
+# check 命令 L1+L2 端到端预检实施记录
+
+**日期:** 2026-08-05
+**Commit:** `6124abb`
+**改动文件:** `src/autossh_mgr/check.py`、`tests/unit/test_check.py`、`tests/unit/test_cli_lifecycle.py`
+
+---
+
+## 1. 背景与问题
+
+### 1.1 旧版 `check` 的假阳性
+
+旧版 `check_connectivity` 只做一件事:发起一次轻量 SSH 登录(`ssh ... true`),
+验证"密钥能不能登 + 网络可达 + sshd 在"。它**完全不碰真实隧道的数据通路**。
+
+由此产生**假阳性**(check 报 OK 但隧道实际不通):
+
+| 场景 | 旧 check | 真实状态 |
+|------|---------|---------|
+| 本地后端服务没起(如 ftdl 的 Next.js 停了) | ✅ OK | ❌ 公网访问连不上 |
+| 远端端口被占用 | ✅ OK | ❌ `-R` 绑不上 |
+| sshd `GatewayPorts no` | ✅ OK | ❌ 只绑 127.0.0.1,外网进不来 |
+| `AllowTcpForwarding no` | ✅ OK | ❌ 转发被禁 |
+
+**本机实测复现:** `ftdl` 隧道的后端 Next.js 未运行,但旧版 `check ftdl` 仍返回 OK。
+
+### 1.2 历史教训:被删除的 `_is_port_in_use` (commit `4c9cd00`)
+
+历史上曾有过一个端口检查,位于 `process.py` 的 `start_tunnel` 里,但逻辑错误已被删除:
+
+```python
+# ❌ 已删除的错误检查 (commit 4c9cd00):
+def _is_port_in_use(port):
+    with socket.socket(...) as s:
+        try:
+            s.bind(("127.0.0.1", port))   # 尝试占用端口
+            return False                    # 占用成功→端口空闲
+        except OSError:
+            return True                     # 占用失败→端口被占
+# start_tunnel 里: if _is_port_in_use(local_port): 拒绝启动
+```
+
+**为什么错:** 对 `-R` 反向隧道,`local_port` 是**转发目标**(如 Docker 容器、Gitea web),
+它**本该在监听**,不该是空闲的。旧检查把"目标服务在跑"误判成"端口冲突",语义反了。
+
+### 1.3 本次方案的区别
+
+本次新增的 Layer 1 用 `connect()` 而非 `bind()`,语义完全不同,不会重蹈覆辙:
+
+| | 被删的旧检查 | 本次 Layer 1 |
+|---|---|---|
+| **方法** | `bind()` 占用端口 | `connect()` 探测目标 |
+| **位置** | `start_tunnel` (启动门禁) | `check` (诊断命令) |
+| **语义** | "端口被占=冲突" ❌ 反了 | "目标响应=健康" ✅ 对 |
+| **目的** | 阻止启动 | 报告隧道是否端到端可用 |
+
+---
+
+## 2. 设计:L1+L2 精简版(三层中的前两层)
+
+### 2.1 范围
+
+原始设计(见备份文档 `check-port-validation.md`)提出三层检查 L1/L2/L3。
+本次**只实施 L1+L2**,跳过 L3 探针转发,理由见 §2.4。
+
+**不做的事:**
+- 不做 L3 探针转发(带 `-R` + `ExitOnForwardFailure` 的真实转发探测)
+- 不改 autossh 进程的启动参数(`ExitOnForwardFailure` 留给方案 D)
+- 不改 `start`/`status`/`stop` 等其它命令的行为
+- 不改配置文件格式 `tunnels.yaml`
+
+### 2.2 检查流程(两层,顺序短路)
+
+```
+check <name>
+  │
+  ├─ Layer 1: 本地后端端口探测
+  │   TCP connect 到 (local_host, local_port),超时 2 秒
+  │   失败 → 返回 FAIL "local backend <host>:<port> not listening"
+  │   成功 ↓
+  │
+  └─ Layer 2: SSH 认证 (保留原有逻辑)
+      ssh -o BatchMode=yes -o ConnectTimeout=5 ... true
+      失败 → 返回 FAIL "SSH auth failed: <stderr>"
+      成功 → 返回 OK "reachable + backend up"
+```
+
+**顺序的理由:** Layer 1 最便宜(纯本地 TCP,毫秒级),Layer 2 较贵(一次 SSH 握手)。
+便宜的前置过滤掉明显错误(后端没起),避免无谓的网络往返。
+
+### 2.3 两层覆盖的失败场景
+
+| 失败场景 | Layer 1 (本地端口) | Layer 2 (SSH 认证) |
+|---------|:--:|:--:|
+| 后端服务没起(最常见) | ✅ | — |
+| 本地端口绑错地址(127.x vs 0.0.0.0) | ✅ | — |
+| 网络不通/防火墙 | — | ✅ |
+| 密钥被拒/被 fail2ban | — | ✅ |
+| 远端端口被占 | ❌ 不覆盖 | — |
+| `GatewayPorts no` | ❌ 不覆盖 | — |
+| `AllowTcpForwarding no` | ❌ 不覆盖 | — |
+
+L1+L2 覆盖了最常见的"后端没起"假阳性。端口绑定类问题留给未来的 L3 或方案 D。
+
+### 2.4 为什么跳过 L3
+
+| 考量 | 说明 |
+|------|------|
+| **副作用** | L3 探针会在云端短暂占用 `remote_port` 约 1 秒 |
+| **性能** | 成功路径增加约 1.5 秒(L1 毫秒级 + L3 的 sleep 1);失败路径因短路反而更快 |
+| **误报风险** | 某些 sshd 配置下可能因端口已被 autossh 占用而误报失败 |
+| **性价比** | L3 主要堵"端口绑定类"问题,这些场景相对少见,真遇到用 `status` + 手动 ssh 也能发现 |
+
+L1+L2 覆盖 80% 的常见假阳性,改动最小、零副作用。L3 可后续按需再加。
+
+---
+
+## 3. 实现
+
+仅改 1 个源码文件:`src/autossh_mgr/check.py`
+
+### 3.1 新增函数 `check_local_backend` (Layer 1)
+
+```python
+import socket
+
+def check_local_backend(tunnel: TunnelConfig) -> tuple[bool, str]:
+    """Layer 1: probe whether the local backend port has a listener."""
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    s.settimeout(2)
+    try:
+        s.connect((tunnel.local_host, tunnel.local_port))
+    except OSError as e:
+        return False, f"local backend {tunnel.local_host}:{tunnel.local_port} not listening ({e})"
+    finally:
+        s.close()
+    return True, ""
+```
+
+**关键点:**
+- 用 `connect()` 而非 `bind()`,语义正确(探测目标是否响应,而非端口是否空闲)
+- `finally: s.close()` 保证任何情况下都释放 socket
+- 超时 2 秒,避免后端慢响应时卡住
+
+### 3.2 改写 `check_connectivity` (L1→L2 短路)
+
+```python
+def check_connectivity(tunnel: TunnelConfig) -> tuple[bool, str]:
+    # Layer 1: 本地后端端口
+    ok, msg = check_local_backend(tunnel)
+    if not ok:
+        return False, msg
+
+    # Layer 2: SSH 认证 (原有逻辑)
+    result = subprocess.run(
+        build_ssh_check_cmd(tunnel),
+        capture_output=True,
+        text=True,
+    )
+    if result.returncode != 0:
+        return False, f"SSH auth failed: {result.stderr.strip()}"
+    return True, "reachable + backend up"
+```
+
+**接口契约不变:** 签名仍是 `(bool, str)`,`cli.py` 的 `check_cmd` 无需改动。
+成功时 `cli.py:245` 显示固定的 `OK: ...`,忽略 msg;失败时 msg 进入 `Connection failed: {msg}`。
+
+### 3.3 `build_ssh_check_cmd` 未改动
+
+Layer 2 复用原有命令构建函数,保持 SSH 认证探测行为不变。
+
+---
+
+## 4. 测试
+
+### 4.1 单元测试 (`tests/unit/test_check.py`)
+
+新增/改写 6 个测试:
+
+| 测试 | 覆盖点 |
+|------|--------|
+| `test_build_ssh_check_cmd` | 命令构建(原有,保留) |
+| `test_check_local_backend_success` | L1 成功路径,mock socket,断言 connect/settimeout/close 调用 |
+| `test_check_local_backend_failure` | L1 失败路径,connect 抛 `ConnectionRefusedError`,断言错误消息含端口 |
+| `test_check_connectivity_layer1_short_circuits` | **L1 短路关键测试**:L1 失败时 `subprocess.run` 必须不被调用 |
+| `test_check_connectivity_success` | L1+L2 全成功,断言返回 "reachable + backend up" |
+| `test_check_connectivity_ssh_failure` | L1 成功但 L2 失败,断言返回 "SSH auth failed" |
+
+### 4.2 CLI 集成测试修复 (`tests/unit/test_cli_lifecycle.py`)
+
+`test_check_success` / `test_check_failure` 原本只 mock `subprocess.run`,
+新增 L1 后会真去 connect 端口导致测试失败。修复:同时 mock `autossh_mgr.check.socket.socket`。
+
+### 4.3 全量测试结果
+
+```
+72 passed in 52.84s
+```
+
+含 integration 7 + unit 65,两次运行结果一致,稳定通过。
+
+---
+
+## 5. 实测验证
+
+### 5.1 本机实测
+
+本机 `tunnels.yaml` 配置两条隧道,editable 安装新版源码后实测:
+
+| 隧道 | 后端端口 | L1 探测 | 结果 | 旧行为对比 |
+|------|---------|---------|------|-----------|
+| `hub-ftdev-android` | :8003 ✅ 在监听 | connect 成功 → L2 → SSH 通 | ✅ OK | 旧版也 OK(巧合正确) |
+| `ftdl` | :3000 ❌ Next.js 没起 | connect 拒绝 → **L1 短路失败** | ❌ `local backend 127.0.0.1:3000 not listening` | ❗ 旧版会误报 OK(假阳性) |
+
+**验证了方案C的核心价值:** ftdl 后端没起,旧版 `check` 报 OK(隧道"通"但实际访问拿不到东西),
+新版 L1 立刻拦住并明确告知"本地后端没监听",且根本没发起 SSH(短路)。
+
+### 5.2 NAS 实测
+
+NAS 上 `tunnels.yaml` 配置两条隧道(后端都是 NAS 本机服务),实测:
+
+| 隧道 | 后端端口 | L1 | L2 | 结果 |
+|------|---------|:--:|:--:|------|
+| `gitea` | :80 Gitea web | ✅ | ✅ | ✅ OK |
+| `gitea-ssh` | :2222 Gitea ssh | ✅ | ✅ | ✅ OK |
+
+两条隧道全绿,零假阳性、零误报。
+
+---
+
+## 6. 兼容性
+
+- **配置文件:** 无变更,`tunnels.yaml` 格式不变
+- **命令接口:** `autossh-mgr check <name>` 用法不变,退出码语义不变(0=成功,1=失败)
+- **输出格式:** 成功仍是 `OK: ...`,失败仍是 `Error: Connection failed: ...`,
+  只是失败原因更具体(指明是 L1 还是 L2 挂的)
+- **性能:** 成功路径增加约几十毫秒(L1 TCP connect);失败路径因短路反而更快
+- **无新依赖:** `socket` 是 Python 标准库
+
+---
+
+## 7. 部署方式
+
+| 位置 | 安装方式 | 更新方式 |
+|------|---------|---------|
+| **NAS** | editable (`uv tool install --editable .`) | 改 `src/` 即生效,无需重装 |
+| **本机** | 非 editable (从 git URL 装) | `uv tool install --force --refresh <url>` 或改用 `--editable` |
+
+本机开发建议用 editable 安装,改完立即生效:
+```bash
+cd ~/workspace/autossh-mgr
+uv tool install --force --editable .
+```
+
+---
+
+## 8. 回退方案
+
+任何时候可回退到改动前(commit `4c9cd00`):
+
+```bash
+# 源码回退
+cd ~/workspace/autossh-mgr
+git reset --hard 4c9cd00        # 或用 tag: git reset --hard pre-plan-c
+
+# 若本机是 editable 安装,源码回退后立即生效
+# 若本机是非 editable 安装,需重装旧版:
+uv tool install --force --refresh ssh://git@git.zz.com:2222/Developer/autossh-mgr.git
+```
+
+NAS 回退:
+```bash
+ssh mnas 'cd ~/workspace/autossh-mgr && git checkout -- src/autossh_mgr/check.py'
+```
+
+---
+
+## 9. 未来工作
+
+| 项 | 说明 | 优先级 |
+|----|------|--------|
+| **Layer 3 探针转发** | 带 `-R` + `ExitOnForwardFailure` 的真实转发探测,覆盖端口绑定类假阳性 | 按需,当前 L1+L2 已覆盖 80% 场景 |
+| **方案 D: autossh 启动参数** | 给 `build_autossh_cmd` 加 `-o ExitOnForwardFailure=yes`,让运行中的 autossh 在端口绑不上时自愈重生 | 中,根治运行态假活 |
+| **verify 命令** | 独立的端到端探测命令,通过真实隧道发 HTTP 请求验证后端响应 | 低,`check` + `status` 已基本够用 |

src/autossh_mgr/check.py

@@ -1,8 +1,22 @@
 import os
+import socket
 import subprocess
 from autossh_mgr.config import TunnelConfig
 
 
+def check_local_backend(tunnel: TunnelConfig) -> tuple[bool, str]:
+    """Layer 1: probe whether the local backend port has a listener."""
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    s.settimeout(2)
+    try:
+        s.connect((tunnel.local_host, tunnel.local_port))
+    except OSError as e:
+        return False, f"local backend {tunnel.local_host}:{tunnel.local_port} not listening ({e})"
+    finally:
+        s.close()
+    return True, ""
+
+
 def build_ssh_check_cmd(tunnel: TunnelConfig) -> list[str]:
     return [
         "ssh",
@@ -16,9 +30,15 @@ def build_ssh_check_cmd(tunnel: TunnelConfig) -> list[str]:
 
 
 def check_connectivity(tunnel: TunnelConfig) -> tuple[bool, str]:
+    ok, msg = check_local_backend(tunnel)
+    if not ok:
+        return False, msg
+
     result = subprocess.run(
         build_ssh_check_cmd(tunnel),
         capture_output=True,
         text=True,
     )
-    return result.returncode == 0, result.stderr.strip()
+    if result.returncode != 0:
+        return False, f"SSH auth failed: {result.stderr.strip()}"
+    return True, "reachable + backend up"

src/autossh_mgr/cli.py

@@ -0,0 +1,247 @@
+import click
+from dataclasses import replace
+from autossh_mgr.config import (
+    TunnelConfig, get_config_dir, ensure_dirs,
+    load_tunnels, save_tunnels, get_tunnel,
+)
+from autossh_mgr.check import check_connectivity
+from autossh_mgr.process import (
+    get_status, start_tunnel, stop_tunnel, delete_pid_file,
+)
+from autossh_mgr.display import (
+    print_tunnel_table, print_tunnel_detail, print_status_table,
+)
+
+
+_FIELD_TYPES: dict[str, type] = {
+    "host": str,
+    "port": int,
+    "user": str,
+    "identity_file": str,
+    "local_host": str,
+    "local_port": int,
+    "remote_host": str,
+    "remote_port": int,
+    "ssh_options": str,
+}
+
+
+@click.group()
+@click.pass_context
+def cli(ctx: click.Context) -> None:
+    ctx.ensure_object(dict)
+    config_dir = get_config_dir()
+    ensure_dirs(config_dir)
+    ctx.obj["config_dir"] = config_dir
+
+
+@cli.command()
+@click.pass_context
+def init(ctx: click.Context) -> None:
+    """Initialize config directory."""
+    config_dir = ctx.obj["config_dir"]
+    ensure_dirs(config_dir)
+    click.echo(f"Initialized {config_dir}")
+
+
+@cli.command()
+@click.argument("name")
+@click.option("--host", default=None)
+@click.option("--port", type=int, default=None)
+@click.option("--user", default=None)
+@click.option("--identity-file", default=None)
+@click.option("--local-host", default=None)
+@click.option("--local-port", type=int, default=None)
+@click.option("--remote-host", default=None)
+@click.option("--remote-port", type=int, default=None)
+@click.option("--ssh-options", default=None)
+@click.pass_context
+def add(
+    ctx, name, host, port, user, identity_file,
+    local_host, local_port, remote_host, remote_port, ssh_options,
+):
+    """Add a new tunnel configuration."""
+    config_dir = ctx.obj["config_dir"]
+    tunnels = load_tunnels(config_dir)
+    if any(t.name == name for t in tunnels):
+        raise click.ClickException(f"Tunnel '{name}' already exists")
+
+    if host is None:
+        host = click.prompt("Public server host")
+    if user is None:
+        user = click.prompt("SSH user")
+    if local_port is None:
+        local_port = click.prompt("Local port", type=int)
+    if remote_port is None:
+        remote_port = click.prompt("Remote port", type=int)
+    if port is None:
+        port = 22
+    if identity_file is None:
+        identity_file = "~/.ssh/id_ed25519"
+    if local_host is None:
+        local_host = "127.0.0.1"
+    if remote_host is None:
+        remote_host = "0.0.0.0"
+    if ssh_options is None:
+        ssh_options = ""
+
+    tunnel = TunnelConfig(
+        name=name, host=host, port=port, user=user,
+        identity_file=identity_file, local_host=local_host,
+        local_port=local_port, remote_host=remote_host,
+        remote_port=remote_port, ssh_options=ssh_options,
+    )
+    tunnels.append(tunnel)
+    save_tunnels(config_dir, tunnels)
+    click.echo(f"Added tunnel '{name}'")
+
+
+@cli.command()
+@click.argument("name")
+@click.pass_context
+def remove(ctx, name):
+    """Remove a tunnel configuration."""
+    config_dir = ctx.obj["config_dir"]
+    tunnels = load_tunnels(config_dir)
+    if not any(t.name == name for t in tunnels):
+        raise click.ClickException(f"No tunnel named '{name}'")
+    status = get_status(config_dir, name)
+    if status.state == "running":
+        raise click.ClickException(f"Stop '{name}' before removing it")
+    if not click.confirm(f"Remove tunnel '{name}'?"):
+        return
+    delete_pid_file(config_dir, name)
+    save_tunnels(config_dir, [t for t in tunnels if t.name != name])
+    click.echo(f"Removed tunnel '{name}'")
+
+
+@cli.command(name="list")
+@click.pass_context
+def list_cmd(ctx):
+    """List all tunnel configurations."""
+    config_dir = ctx.obj["config_dir"]
+    tunnels = load_tunnels(config_dir)
+    statuses = {t.name: get_status(config_dir, t.name) for t in tunnels}
+    print_tunnel_table(tunnels, statuses)
+
+
+@cli.command()
+@click.argument("name")
+@click.pass_context
+def show(ctx, name):
+    """Show full details for a tunnel."""
+    config_dir = ctx.obj["config_dir"]
+    tunnel = get_tunnel(config_dir, name)
+    status = get_status(config_dir, name)
+    print_tunnel_detail(tunnel, status)
+
+
+@cli.command(name="config")
+@click.argument("name")
+@click.argument("key")
+@click.argument("value")
+@click.pass_context
+def config_cmd(ctx, name, key, value):
+    """Update a single tunnel configuration field."""
+    config_dir = ctx.obj["config_dir"]
+    if key == "name":
+        raise click.ClickException("Cannot rename a tunnel via config. Remove and re-add.")
+    if key not in _FIELD_TYPES:
+        raise click.ClickException(f"Unknown field '{key}'")
+    try:
+        typed_value = _FIELD_TYPES[key](value)
+    except ValueError:
+        raise click.ClickException(
+            f"Invalid value for '{key}': expected {_FIELD_TYPES[key].__name__}"
+        )
+    tunnels = load_tunnels(config_dir)
+    if not any(t.name == name for t in tunnels):
+        raise click.ClickException(f"No tunnel named '{name}'")
+    tunnels = [replace(t, **{key: typed_value}) if t.name == name else t for t in tunnels]
+    save_tunnels(config_dir, tunnels)
+    click.echo(f"Updated {name}.{key} = {typed_value}")
+
+
+@cli.command()
+@click.argument("name", required=False)
+@click.pass_context
+def start(ctx, name):
+    """Start a tunnel (or all tunnels if no name given)."""
+    config_dir = ctx.obj["config_dir"]
+    tunnels = load_tunnels(config_dir)
+    targets = (
+        [get_tunnel(config_dir, name)] if name
+        else sorted(tunnels, key=lambda t: t.name)
+    )
+    for tunnel in targets:
+        status = get_status(config_dir, tunnel.name)
+        if status.state == "running":
+            click.echo(f"{tunnel.name} already running (pid: {status.pid})")
+            continue
+        if status.state == "stale":
+            click.echo(f"{tunnel.name}: stale pid detected, cleaning up")
+            delete_pid_file(config_dir, tunnel.name)
+        pid = start_tunnel(config_dir, tunnel)
+        click.echo(f"Started {tunnel.name} (pid: {pid})")
+
+
+@cli.command()
+@click.argument("name", required=False)
+@click.pass_context
+def stop(ctx, name):
+    """Stop a tunnel (or all running tunnels if no name given)."""
+    config_dir = ctx.obj["config_dir"]
+    tunnels = load_tunnels(config_dir)
+    targets = (
+        [get_tunnel(config_dir, name)] if name
+        else sorted(tunnels, key=lambda t: t.name)
+    )
+    for tunnel in targets:
+        status = get_status(config_dir, tunnel.name)
+        if status.state == "stopped":
+            click.echo(f"{tunnel.name} is not running")
+            continue
+        stop_tunnel(config_dir, tunnel.name)
+        if status.state == "stale":
+            click.echo(f"{tunnel.name}: cleaned up stale pid")
+        else:
+            click.echo(f"Stopped {tunnel.name}")
+
+
+@cli.command()
+@click.argument("name", required=False)
+@click.pass_context
+def restart(ctx, name):
+    """Restart a tunnel (or all tunnels if no name given)."""
+    ctx.invoke(stop, name=name)
+    ctx.invoke(start, name=name)
+
+
+@cli.command()
+@click.argument("name", required=False)
+@click.pass_context
+def status(ctx, name):
+    """Show tunnel status (all tunnels if no name given)."""
+    config_dir = ctx.obj["config_dir"]
+    if name:
+        tunnel = get_tunnel(config_dir, name)
+        s = get_status(config_dir, name)
+        print_tunnel_detail(tunnel, s)
+    else:
+        tunnels = load_tunnels(config_dir)
+        statuses = {t.name: get_status(config_dir, t.name) for t in tunnels}
+        print_status_table(tunnels, statuses)
+
+
+@cli.command(name="check")
+@click.argument("name")
+@click.pass_context
+def check_cmd(ctx, name):
+    """Check SSH connectivity to a tunnel's server."""
+    config_dir = ctx.obj["config_dir"]
+    tunnel = get_tunnel(config_dir, name)
+    success, msg = check_connectivity(tunnel)
+    if success:
+        click.echo(f"OK: {tunnel.user}@{tunnel.host}:{tunnel.port} is reachable")
+    else:
+        raise click.ClickException(f"Connection failed: {msg}")

src/autossh_mgr/display.py

@@ -0,0 +1,57 @@
+import click
+from autossh_mgr.config import TunnelConfig
+from autossh_mgr.process import TunnelStatus
+
+
+def _server_str(t: TunnelConfig) -> str:
+    return f"{t.user}@{t.host}:{t.port}"
+
+
+def _ports_str(t: TunnelConfig) -> str:
+    return f"{t.local_port} -> {t.remote_port}"
+
+
+def print_tunnel_table(tunnels: list[TunnelConfig], statuses: dict[str, TunnelStatus]) -> None:
+    if not tunnels:
+        click.echo("No tunnels configured.")
+        return
+    fmt = f"{'NAME':<20} {'SERVER':<32} {'PORTS':<15} STATUS"
+    click.echo(fmt)
+    click.echo("-" * 72)
+    for t in tunnels:
+        s = statuses.get(t.name)
+        state = s.state if s else "unknown"
+        click.echo(f"{t.name:<20} {_server_str(t):<32} {_ports_str(t):<15} {state}")
+
+
+def print_status_table(tunnels: list[TunnelConfig], statuses: dict[str, TunnelStatus]) -> None:
+    if not tunnels:
+        click.echo("No tunnels configured.")
+        return
+    fmt = f"{'NAME':<20} {'STATE':<10} {'PID':<8} UPTIME"
+    click.echo(fmt)
+    click.echo("-" * 52)
+    for t in tunnels:
+        s = statuses.get(t.name)
+        if s:
+            pid_str = str(s.pid) if s.pid else "-"
+            uptime_str = s.uptime or "-"
+            click.echo(f"{t.name:<20} {s.state:<10} {pid_str:<8} {uptime_str}")
+
+
+def print_tunnel_detail(tunnel: TunnelConfig, status: TunnelStatus | None) -> None:
+    click.echo(f"Name:          {tunnel.name}")
+    click.echo(f"Host:          {tunnel.host}")
+    click.echo(f"SSH port:      {tunnel.port}")
+    click.echo(f"User:          {tunnel.user}")
+    click.echo(f"Identity file: {tunnel.identity_file}")
+    click.echo(f"Local:         {tunnel.local_host}:{tunnel.local_port}")
+    click.echo(f"Remote:        {tunnel.remote_host}:{tunnel.remote_port}")
+    if tunnel.ssh_options:
+        click.echo(f"SSH options:   {tunnel.ssh_options}")
+    if status:
+        click.echo(f"State:         {status.state}")
+        if status.pid:
+            click.echo(f"PID:           {status.pid}")
+        if status.uptime:
+            click.echo(f"Uptime:        {status.uptime}")

src/autossh_mgr/process.py

@@ -1,7 +1,7 @@
 import os
+import shlex
 import shutil
 import signal
-import socket
 import subprocess
 import time
 from dataclasses import dataclass
@@ -109,25 +109,14 @@ def build_autossh_cmd(tunnel: TunnelConfig) -> list[str]:
         f"{tunnel.user}@{tunnel.host}",
     ]
     if tunnel.ssh_options:
-        for opt in tunnel.ssh_options.split():
+        for opt in shlex.split(tunnel.ssh_options):
             cmd.append(opt)
     return cmd
 
 
-def _is_port_in_use(port: int) -> bool:
-    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-        try:
-            s.bind(("127.0.0.1", port))
-            return False
-        except OSError:
-            return True
-
-
 def start_tunnel(config_dir: Path, tunnel: TunnelConfig) -> int:
     if not shutil.which("autossh"):
         raise click.ClickException("autossh not found. Install it first.")
-    if _is_port_in_use(tunnel.local_port):
-        raise click.ClickException(f"Port {tunnel.local_port} is already in use")
 
     log_path = config_dir / "logs" / f"{tunnel.name}.log"
     log_path.parent.mkdir(parents=True, exist_ok=True)

tests/integration/test_lifecycle.py

@@ -0,0 +1,135 @@
+import os
+import signal
+import time
+import pytest
+from datetime import datetime, timezone
+from click.testing import CliRunner
+from autossh_mgr.cli import cli
+from autossh_mgr.config import save_tunnels, TunnelConfig, ensure_dirs
+from autossh_mgr.process import read_pid_file, is_process_alive, write_pid_file
+
+
+@pytest.fixture
+def fake_autossh(tmp_path, monkeypatch):
+    """Replace autossh with a long-running dummy process."""
+    fake = tmp_path / "autossh"
+    fake.write_text("#!/bin/sh\nexec sleep 60\n")
+    fake.chmod(0o755)
+    monkeypatch.setenv("PATH", f"{tmp_path}:{os.environ['PATH']}")
+    return fake
+
+
+@pytest.fixture
+def config_dir(tmp_path):
+    ensure_dirs(tmp_path)
+    return tmp_path
+
+
+@pytest.fixture(autouse=True)
+def set_config_dir(config_dir, monkeypatch):
+    monkeypatch.setenv("AUTOSSH_MGR_CONFIG_DIR", str(config_dir))
+
+
+@pytest.fixture(autouse=True)
+def cleanup_processes(config_dir):
+    """Kill any processes recorded in PID files after each test."""
+    yield
+    pids_dir = config_dir / "pids"
+    if not pids_dir.exists():
+        return
+    for pid_file in pids_dir.glob("*.pid"):
+        try:
+            lines = pid_file.read_text().splitlines()
+            if lines:
+                os.kill(int(lines[0]), signal.SIGKILL)
+        except (ValueError, OSError, ProcessLookupError):
+            pass
+
+
+@pytest.fixture
+def tunnel(config_dir):
+    t = TunnelConfig(
+        name="test-tunnel",
+        host="relay.example.com",
+        user="deploy",
+        local_port=19999,
+        remote_port=29999,
+    )
+    save_tunnels(config_dir, [t])
+    return t
+
+
+@pytest.fixture
+def runner():
+    return CliRunner()
+
+
+def test_start_writes_pid_file(runner, config_dir, tunnel, fake_autossh):
+    result = runner.invoke(cli, ["start", "test-tunnel"])
+    assert result.exit_code == 0, result.output
+    pid_data = read_pid_file(config_dir, "test-tunnel")
+    assert pid_data is not None
+    pid, _ = pid_data
+    assert is_process_alive(pid)
+
+
+def test_stop_kills_process(runner, config_dir, tunnel, fake_autossh):
+    runner.invoke(cli, ["start", "test-tunnel"])
+    pid_data = read_pid_file(config_dir, "test-tunnel")
+    assert pid_data is not None
+    pid = pid_data[0]
+    assert is_process_alive(pid)
+
+    result = runner.invoke(cli, ["stop", "test-tunnel"])
+    assert result.exit_code == 0
+    time.sleep(0.2)
+    assert not is_process_alive(pid)
+    assert read_pid_file(config_dir, "test-tunnel") is None
+
+
+def test_status_shows_running(runner, config_dir, tunnel, fake_autossh):
+    runner.invoke(cli, ["start", "test-tunnel"])
+    result = runner.invoke(cli, ["status", "test-tunnel"])
+    assert result.exit_code == 0
+    assert "running" in result.output
+
+
+def test_stale_pid_cleaned_on_start(runner, config_dir, tunnel, fake_autossh):
+    write_pid_file(config_dir, "test-tunnel", 9999999, datetime.now(timezone.utc))
+
+    result = runner.invoke(cli, ["start", "test-tunnel"])
+    assert result.exit_code == 0, result.output
+    assert "stale" in result.output.lower()
+
+    pid_data = read_pid_file(config_dir, "test-tunnel")
+    assert pid_data is not None
+    assert pid_data[0] != 9999999
+
+
+def test_restart(runner, config_dir, tunnel, fake_autossh):
+    runner.invoke(cli, ["start", "test-tunnel"])
+    first_pid = read_pid_file(config_dir, "test-tunnel")[0]
+
+    result = runner.invoke(cli, ["restart", "test-tunnel"])
+    assert result.exit_code == 0, result.output
+
+    second_pid = read_pid_file(config_dir, "test-tunnel")[0]
+    assert second_pid != first_pid
+    assert is_process_alive(second_pid)
+    assert not is_process_alive(first_pid)
+
+
+def test_idempotent_start(runner, config_dir, tunnel, fake_autossh):
+    runner.invoke(cli, ["start", "test-tunnel"])
+    first_pid = read_pid_file(config_dir, "test-tunnel")[0]
+
+    result = runner.invoke(cli, ["start", "test-tunnel"])
+    assert result.exit_code == 0
+    assert "already running" in result.output
+    assert read_pid_file(config_dir, "test-tunnel")[0] == first_pid
+
+
+def test_idempotent_stop(runner, tunnel):
+    result = runner.invoke(cli, ["stop", "test-tunnel"])
+    assert result.exit_code == 0
+    assert "not running" in result.output

tests/unit/test_check.py

@@ -1,7 +1,10 @@
 import pytest
-import subprocess
 from unittest.mock import patch, MagicMock
-from autossh_mgr.check import check_connectivity, build_ssh_check_cmd
+from autossh_mgr.check import (
+    check_connectivity,
+    check_local_backend,
+    build_ssh_check_cmd,
+)
 from autossh_mgr.config import TunnelConfig
 
 
@@ -23,23 +26,56 @@ def test_build_ssh_check_cmd(tunnel):
     assert "-p" in cmd and str(tunnel.port) in cmd
 
 
+def test_check_local_backend_success(tunnel):
+    mock_sock = MagicMock()
+    with patch("autossh_mgr.check.socket.socket", return_value=mock_sock):
+        ok, msg = check_local_backend(tunnel)
+    assert ok is True
+    assert msg == ""
+    mock_sock.connect.assert_called_once_with(("127.0.0.1", 8080))
+    mock_sock.settimeout.assert_called_once_with(2)
+    mock_sock.close.assert_called_once()
+
+
+def test_check_local_backend_failure(tunnel):
+    mock_sock = MagicMock()
+    mock_sock.connect.side_effect = ConnectionRefusedError("refused")
+    with patch("autossh_mgr.check.socket.socket", return_value=mock_sock):
+        ok, msg = check_local_backend(tunnel)
+    assert ok is False
+    assert "127.0.0.1:8080" in msg
+    assert "not listening" in msg
+    mock_sock.close.assert_called_once()
+
+
+def test_check_connectivity_layer1_short_circuits(tunnel):
+    """When Layer 1 fails, SSH (Layer 2) must not be invoked."""
+    mock_sock = MagicMock()
+    mock_sock.connect.side_effect = ConnectionRefusedError("refused")
+    with patch("autossh_mgr.check.socket.socket", return_value=mock_sock), \
+         patch("autossh_mgr.check.subprocess.run") as mock_run:
+        success, msg = check_connectivity(tunnel)
+    assert success is False
+    assert "not listening" in msg
+    mock_run.assert_not_called()
+
+
 def test_check_connectivity_success(tunnel):
-    mock_result = MagicMock()
+    mock_sock = MagicMock()
-    mock_result.returncode = 0
+    with patch("autossh_mgr.check.socket.socket", return_value=mock_sock), \
-    mock_result.stderr = ""
+         patch("autossh_mgr.check.subprocess.run", return_value=MagicMock(returncode=0, stderr="")) as mock_run:
-    with patch("autossh_mgr.check.subprocess.run", return_value=mock_result) as mock_run:
         success, msg = check_connectivity(tunnel)
     assert success is True
-    assert msg == ""
+    assert msg == "reachable + backend up"
     called_cmd = mock_run.call_args[0][0]
     assert called_cmd == build_ssh_check_cmd(tunnel)
 
 
-def test_check_connectivity_failure(tunnel):
+def test_check_connectivity_ssh_failure(tunnel):
-    mock_result = MagicMock()
+    mock_sock = MagicMock()
-    mock_result.returncode = 255
+    with patch("autossh_mgr.check.socket.socket", return_value=mock_sock), \
-    mock_result.stderr = "Connection refused"
+         patch("autossh_mgr.check.subprocess.run", return_value=MagicMock(returncode=255, stderr="Connection refused")):
-    with patch("autossh_mgr.check.subprocess.run", return_value=mock_result):
         success, msg = check_connectivity(tunnel)
     assert success is False
+    assert "SSH auth failed" in msg
     assert "Connection refused" in msg

tests/unit/test_cli_config.py

@@ -0,0 +1,140 @@
+import os
+import pytest
+from datetime import datetime, timezone
+from click.testing import CliRunner
+from autossh_mgr.cli import cli
+from autossh_mgr.config import load_tunnels, save_tunnels
+from autossh_mgr.process import write_pid_file
+
+
+@pytest.fixture(autouse=True)
+def set_config_dir(config_dir, monkeypatch):
+    monkeypatch.setenv("AUTOSSH_MGR_CONFIG_DIR", str(config_dir))
+
+
+@pytest.fixture
+def runner():
+    return CliRunner()
+
+
+def test_init_creates_structure(runner, tmp_path, monkeypatch):
+    new_dir = tmp_path / "new-config"
+    monkeypatch.setenv("AUTOSSH_MGR_CONFIG_DIR", str(new_dir))
+    result = runner.invoke(cli, ["init"])
+    assert result.exit_code == 0
+    assert (new_dir / "tunnels.yaml").exists()
+    assert (new_dir / "pids").is_dir()
+    assert (new_dir / "logs").is_dir()
+
+
+def test_init_idempotent(runner):
+    result = runner.invoke(cli, ["init"])
+    assert result.exit_code == 0
+    result = runner.invoke(cli, ["init"])
+    assert result.exit_code == 0
+
+
+def test_add_non_interactive(runner, config_dir):
+    result = runner.invoke(cli, [
+        "add", "web-service",
+        "--host", "relay.example.com",
+        "--user", "deploy",
+        "--local-port", "8080",
+        "--remote-port", "18080",
+    ])
+    assert result.exit_code == 0, result.output
+    tunnels = load_tunnels(config_dir)
+    assert len(tunnels) == 1
+    assert tunnels[0].name == "web-service"
+    assert tunnels[0].host == "relay.example.com"
+
+
+def test_add_duplicate_name_fails(runner, config_dir, sample_tunnel):
+    save_tunnels(config_dir, [sample_tunnel])
+    result = runner.invoke(cli, [
+        "add", "web-service",
+        "--host", "other.com", "--user", "u",
+        "--local-port", "9090", "--remote-port", "19090",
+    ])
+    assert result.exit_code != 0
+    assert "already exists" in result.output
+
+
+def test_remove_non_running(runner, config_dir, sample_tunnel):
+    save_tunnels(config_dir, [sample_tunnel])
+    result = runner.invoke(cli, ["remove", "web-service"], input="y\n")
+    assert result.exit_code == 0
+    assert load_tunnels(config_dir) == []
+
+
+def test_remove_unknown_name(runner):
+    result = runner.invoke(cli, ["remove", "missing"], input="y\n")
+    assert result.exit_code != 0
+    assert "No tunnel named" in result.output
+
+
+def test_remove_aborted(runner, config_dir, sample_tunnel):
+    save_tunnels(config_dir, [sample_tunnel])
+    result = runner.invoke(cli, ["remove", "web-service"], input="n\n")
+    assert result.exit_code == 0
+    assert len(load_tunnels(config_dir)) == 1
+
+
+def test_list_empty(runner):
+    result = runner.invoke(cli, ["list"])
+    assert result.exit_code == 0
+    assert "No tunnels" in result.output
+
+
+def test_list_shows_tunnels(runner, config_dir, sample_tunnel):
+    save_tunnels(config_dir, [sample_tunnel])
+    result = runner.invoke(cli, ["list"])
+    assert result.exit_code == 0
+    assert "web-service" in result.output
+    assert "relay.example.com" in result.output
+
+
+def test_show_tunnel(runner, config_dir, sample_tunnel):
+    save_tunnels(config_dir, [sample_tunnel])
+    result = runner.invoke(cli, ["show", "web-service"])
+    assert result.exit_code == 0
+    assert "relay.example.com" in result.output
+    assert "8080" in result.output
+    assert "18080" in result.output
+
+
+def test_config_update_field(runner, config_dir, sample_tunnel):
+    save_tunnels(config_dir, [sample_tunnel])
+    result = runner.invoke(cli, ["config", "web-service", "remote_port", "19000"])
+    assert result.exit_code == 0
+    t = load_tunnels(config_dir)[0]
+    assert t.remote_port == 19000
+
+
+def test_config_unknown_key(runner, config_dir, sample_tunnel):
+    save_tunnels(config_dir, [sample_tunnel])
+    result = runner.invoke(cli, ["config", "web-service", "bad_key", "val"])
+    assert result.exit_code != 0
+    assert "Unknown field" in result.output
+
+
+def test_config_rename_disallowed(runner, config_dir, sample_tunnel):
+    save_tunnels(config_dir, [sample_tunnel])
+    result = runner.invoke(cli, ["config", "web-service", "name", "new-name"])
+    assert result.exit_code != 0
+    assert "Cannot rename" in result.output
+
+
+def test_config_invalid_type(runner, config_dir, sample_tunnel):
+    save_tunnels(config_dir, [sample_tunnel])
+    result = runner.invoke(cli, ["config", "web-service", "port", "notanumber"])
+    assert result.exit_code != 0
+
+
+def test_remove_running_tunnel_fails(runner, config_dir, sample_tunnel):
+    save_tunnels(config_dir, [sample_tunnel])
+    write_pid_file(config_dir, "web-service", os.getpid(), datetime.now(timezone.utc))
+    result = runner.invoke(cli, ["remove", "web-service"], input="y\n")
+    assert result.exit_code != 0
+    assert "Stop" in result.output
+    assert len(load_tunnels(config_dir)) == 1

tests/unit/test_cli_lifecycle.py

@@ -0,0 +1,165 @@
+import os
+import signal
+import pytest
+from click.testing import CliRunner
+from unittest.mock import patch, MagicMock
+from autossh_mgr.cli import cli
+from autossh_mgr.config import TunnelConfig, save_tunnels
+from autossh_mgr.process import write_pid_file, read_pid_file
+from datetime import datetime, timezone
+
+
+@pytest.fixture(autouse=True)
+def set_config_dir(config_dir, monkeypatch):
+    monkeypatch.setenv("AUTOSSH_MGR_CONFIG_DIR", str(config_dir))
+
+
+@pytest.fixture
+def runner():
+    return CliRunner()
+
+
+@pytest.fixture
+def with_tunnel(config_dir, sample_tunnel):
+    save_tunnels(config_dir, [sample_tunnel])
+    return sample_tunnel
+
+
+def test_start_already_running(runner, config_dir, with_tunnel):
+    write_pid_file(config_dir, "web-service", os.getpid(), datetime.now(timezone.utc))
+    result = runner.invoke(cli, ["start", "web-service"])
+    assert result.exit_code == 0
+    assert "already running" in result.output
+
+
+def test_start_stale_pid_cleaned(runner, config_dir, with_tunnel, tmp_path, monkeypatch):
+    fake = tmp_path / "autossh"
+    fake.write_text("#!/bin/sh\nexec sleep 60\n")
+    fake.chmod(0o755)
+    monkeypatch.setenv("PATH", f"{tmp_path}:{os.environ['PATH']}")
+
+    write_pid_file(config_dir, "web-service", 9999999, datetime.now(timezone.utc))
+    result = runner.invoke(cli, ["start", "web-service"])
+    assert result.exit_code == 0, result.output
+    assert "stale" in result.output.lower()
+    # Clean up launched process
+    pid_data = read_pid_file(config_dir, "web-service")
+    if pid_data:
+        os.kill(pid_data[0], signal.SIGKILL)
+
+
+def test_start_unknown_tunnel(runner):
+    result = runner.invoke(cli, ["start", "missing"])
+    assert result.exit_code != 0
+    assert "No tunnel named" in result.output
+
+
+def test_stop_running_tunnel(runner, config_dir, with_tunnel, tmp_path, monkeypatch):
+    fake = tmp_path / "autossh"
+    fake.write_text("#!/bin/sh\nexec sleep 60\n")
+    fake.chmod(0o755)
+    monkeypatch.setenv("PATH", f"{tmp_path}:{os.environ['PATH']}")
+
+    runner.invoke(cli, ["start", "web-service"])
+    result = runner.invoke(cli, ["stop", "web-service"])
+    assert result.exit_code == 0
+    assert read_pid_file(config_dir, "web-service") is None
+
+
+def test_stop_already_stopped(runner, with_tunnel):
+    result = runner.invoke(cli, ["stop", "web-service"])
+    assert result.exit_code == 0
+    assert "not running" in result.output
+
+
+def test_status_all_stopped(runner, config_dir, with_tunnel):
+    result = runner.invoke(cli, ["status"])
+    assert result.exit_code == 0
+    assert "web-service" in result.output
+    assert "stopped" in result.output
+
+
+def test_status_single(runner, config_dir, with_tunnel):
+    result = runner.invoke(cli, ["status", "web-service"])
+    assert result.exit_code == 0
+    assert "stopped" in result.output
+    assert "relay.example.com" in result.output
+
+
+def test_check_success(runner, with_tunnel):
+    mock_sock = MagicMock()
+    with patch("autossh_mgr.check.socket.socket", return_value=mock_sock), \
+         patch("autossh_mgr.check.subprocess.run") as mock_run:
+        mock_run.return_value = MagicMock(returncode=0, stderr="")
+        result = runner.invoke(cli, ["check", "web-service"])
+    assert result.exit_code == 0
+    assert "OK" in result.output or "success" in result.output.lower()
+
+
+def test_check_failure(runner, with_tunnel):
+    mock_sock = MagicMock()
+    with patch("autossh_mgr.check.socket.socket", return_value=mock_sock), \
+         patch("autossh_mgr.check.subprocess.run") as mock_run:
+        mock_run.return_value = MagicMock(returncode=255, stderr="Connection refused")
+        result = runner.invoke(cli, ["check", "web-service"])
+    assert result.exit_code != 0
+    assert "Connection refused" in result.output
+
+
+def test_restart_running_tunnel(runner, config_dir, with_tunnel, tmp_path, monkeypatch):
+    fake = tmp_path / "autossh"
+    fake.write_text("#!/bin/sh\nexec sleep 60\n")
+    fake.chmod(0o755)
+    monkeypatch.setenv("PATH", f"{tmp_path}:{os.environ['PATH']}")
+
+    runner.invoke(cli, ["start", "web-service"])
+    first_pid = read_pid_file(config_dir, "web-service")[0]
+
+    result = runner.invoke(cli, ["restart", "web-service"])
+    assert result.exit_code == 0, result.output
+
+    second_pid_data = read_pid_file(config_dir, "web-service")
+    assert second_pid_data is not None
+    assert second_pid_data[0] != first_pid
+
+    os.kill(second_pid_data[0], signal.SIGKILL)
+
+
+def test_start_all(runner, config_dir, tmp_path, monkeypatch):
+    fake = tmp_path / "autossh"
+    fake.write_text("#!/bin/sh\nexec sleep 60\n")
+    fake.chmod(0o755)
+    monkeypatch.setenv("PATH", f"{tmp_path}:{os.environ['PATH']}")
+
+    t1 = TunnelConfig(name="alpha", host="h.com", user="u", local_port=8001, remote_port=18001)
+    t2 = TunnelConfig(name="beta", host="h.com", user="u", local_port=8002, remote_port=18002)
+    save_tunnels(config_dir, [t1, t2])
+
+    result = runner.invoke(cli, ["start"])
+    assert result.exit_code == 0, result.output
+    assert "alpha" in result.output
+    assert "beta" in result.output
+
+    for name in ["alpha", "beta"]:
+        pid_data = read_pid_file(config_dir, name)
+        if pid_data:
+            os.kill(pid_data[0], signal.SIGKILL)
+
+
+def test_stop_all(runner, config_dir, tmp_path, monkeypatch):
+    fake = tmp_path / "autossh"
+    fake.write_text("#!/bin/sh\nexec sleep 60\n")
+    fake.chmod(0o755)
+    monkeypatch.setenv("PATH", f"{tmp_path}:{os.environ['PATH']}")
+
+    t1 = TunnelConfig(name="alpha", host="h.com", user="u", local_port=8001, remote_port=18001)
+    t2 = TunnelConfig(name="beta", host="h.com", user="u", local_port=8002, remote_port=18002)
+    save_tunnels(config_dir, [t1, t2])
+
+    runner.invoke(cli, ["start"])
+    result = runner.invoke(cli, ["stop"])
+    assert result.exit_code == 0, result.output
+    assert "alpha" in result.output
+    assert "beta" in result.output
+    assert read_pid_file(config_dir, "alpha") is None
+    assert read_pid_file(config_dir, "beta") is None