#!/usr/bin/env python3
"""
dsh-double — dsh 自我管理 CLI（按需唤醒的休眠双身）

目标：让 dsh 具备自我管理/自我进化能力，避免"改配置→重启即爆→找外部 agent 救"。
机制：
  - L0 保命外壳（dshd-supervisor，bash）在进程外做探活/拉起/机械回滚重启；
  - 本工具（L1 自我管理）做健康快照、配置预检、蓝绿进化提交、standby 唤醒。
  - standby 平时是磁盘副本（零进程/零端口），只在进化切流与故障修复时被临时唤醒。

用法：
  dsh-double.py snapshot [name]         打健康基线快照并更新 last-known-good
  dsh-double.py check                    对当前配置跑预检（副本 dump-config），输出健康判定
  dsh-double.py preview <变更目录>       把变更仅应用到 standby 做预检+试启动（主零改动）
  dsh-double.py evolve <变更目录>        严格蓝绿提交：预检→主应用→重启主→确认健康→才停 standby；失败回滚
  dsh-double.py wake [--port N]          唤醒 standby 为临时 dsh 实例（默认 3082）
  dsh-double.py stop                     停止已唤醒的 standby
  dsh-double.py rollback [snapshot]      从快照恢复主配置（默认 last-known-good）
  dsh-double.py status                   显示快照/standby/主健康/试验田/最近事件
  dsh-double.py standby-sync             把主配置层同步到 standby

  # 试验田双身（对等双脑）：隔离的完整副 dsh 实例 + 主↔副 agent 通信台账
  dsh-double.py testbed init [--force]      创建试验田（克隆配置层/只读共享/独立数据/基线/令牌）
  dsh-double.py testbed sync                从主重新克隆配置层并刷新基线（会覆盖试验田实验配置）
  dsh-double.py testbed reset               停止并重建试验田（只清 double/testbed/，不碰主数据）
  dsh-double.py testbed up [--port N]       唤醒试验田为完整副 dsh（默认 3081；纯启动，不同步）
  dsh-double.py testbed down                休眠试验田（回到磁盘态，零进程）
  dsh-double.py testbed restart             重启试验田
  dsh-double.py testbed status              试验田状态（磁盘态/运行中/基线/最近通信）
  dsh-double.py testbed diff [--out DIR]    试验田配置 vs 基线 → 生成增量 change_dir（喂给 evolve）
  dsh-double.py testbed baseline            用主当前配置刷新基线（进化成功后重置增量基准）
  dsh-double.py comm [N]                    查看主↔副通信台账最近 N 条（默认 20）

环境变量：DSH_HOME（主 DSH_HOME，默认 ~/.dsh）
选项：
  --home <dir>        主 DSH_HOME（默认 $DSH_HOME 或 ~/.dsh）；演练时指向假主；对试验田操作时指向试验田
  --standby <dir>     standby 目录（默认 <主>/double/standby）
  --port <n>          standby 试启动/唤醒端口（默认 3082，与试验田 3081 分离防撞）
  --no-restart        应用主后不重启主（仅演练/只读场景；默认 false）
  --no-wake           跳过 standby 动态试启动（只做静态预检）
"""

import os
import re
import shutil
import socket
import subprocess
import sys
import time
import urllib.request
from datetime import datetime
from pathlib import Path

HOME = Path(os.path.expanduser("~"))
N22_BIN = HOME / ".nvm/versions/node/v22.23.1/bin"
DSH_WRAPPER = HOME / ".local/bin/dsh"

# 真实 dsh 的 profile node_modules（插件实际安装处）——standby/假主只读共享
REAL_PROFILE_NM = HOME / ".dsh/profiles/web/node_modules"
# 内置 bundle（dsh-base/dsh-web-app/headless 等）在 dsh 全局包自己的依赖里解析
GLOBAL_DSH_NM = (HOME / ".nvm/versions/node/v22.23.1/lib/node_modules"
                 / "@deepseek-ai/dsh/node_modules/@deepseek-ai")

MAIN_PORT = 3080
TESTBED_PORT = 3081
STANDBY_PORT = 3082        # 蓝绿引擎试启动/临时唤醒（与试验田端口分离，防撞）
BOOT_READY_TIMEOUT = 30      # 试启动/重启后等待 boot 完成的最大秒数
HEALTH_TIMEOUT = 40          # 主重启后等待探活健康的最大秒数
LIVE_READY_MARK = "dsh web: http://"

# 探活正则：主 = `dsh web` 结尾（dshd-start 不带 --port 启动）；带 --port 的都是
# 试验田/试启动实例，绝不能算进主健康判定，反之主判定也不能匹配到它们。
MAIN_PAT = r"dsh web$"


def port_pat(port):
    """按端口给探活正则：主端口用无参形态，其余用 --port N 精确匹配。"""
    return MAIN_PAT if port == MAIN_PORT else rf"dsh.*--port {port}\b"

# boot 期致命错误关键字（--dump-config / 启动日志判据）
FATAL_PATTERNS = [
    "plugin tree failed to load",
    "duplicate loader entry id",
    "must be a top-level YAML array",
    "YAMLException",
    "entry .* not found",
    "NO_ADAPTER",
    "failed to apply loader entry",
    "Invalid or unexpected token",
]

# 健康基线文件集（相对 DSH_HOME）——boot 期会改坏的只有这些
BASELINE_FILES = [
    "settings.yaml",
    ".credentials.yaml",
    "profiles/web/package.json",
    "profiles/web/cordis.patch.yml",
    "profiles/web/cordis.yml",
    "profiles/web/pnpm-lock.yaml",
    "profiles/web/pnpm-workspace.yaml",
    "profiles/headless/package.json",
    "profiles/headless/cordis.patch.yml",
    "profiles/headless/cordis.yml",
    "autovision/config.json",
]
BASELINE_DIRS = [
    ".agent-presets",
]

# ---------------------------------------------------------------------------
# 基础设施
# ---------------------------------------------------------------------------

def log(msg, evlog=None):
    line = f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {msg}"
    print(line)
    if evlog:
        with open(evlog, "a") as f:
            f.write(line + "\n")


def die(msg, code=1):
    print(f"ERROR: {msg}", file=sys.stderr)
    sys.exit(code)


def run(cmd, timeout=120, env=None, cwd=None, check=False):
    """运行命令，返回 (returncode, stdout, stderr)。"""
    merged_env = dict(os.environ)
    if env:
        merged_env.update(env)
    try:
        p = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout,
                           env=merged_env, cwd=str(cwd) if cwd else None)
        return p.returncode, p.stdout, p.stderr
    except subprocess.TimeoutExpired:
        return 124, "", f"timeout after {timeout}s"


def dsh_env(home: Path):
    """为指定 DSH_HOME 构造启动环境（强制 nvm node22 + DSH_HOME）。"""
    env = dict(os.environ)
    env["DSH_HOME"] = str(home)
    env["PATH"] = f"{N22_BIN}:" + env.get("PATH", "")
    return env


# ---------------------------------------------------------------------------
# 探活
# ---------------------------------------------------------------------------

def tcp_up(port, host="127.0.0.1", timeout=2):
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return True
    except OSError:
        return False


def http_up(port, timeout=3):
    # 关键：必须绕过系统代理直连本机。urllib.request 默认读 macOS 系统代理
    # 设置，会把 127.0.0.1 的请求转发给代理 → 得到 502 Bad Gateway → 健康 dsh
    # 被误判故障（curl 对 localhost 自动 bypass 所以没暴露）。
    try:
        opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
        with opener.open(f"http://127.0.0.1:{port}/", timeout=timeout) as r:
            return r.status == 200
    except Exception:
        return False


def pgrep_up(pat="dsh web"):
    """按命令行正则匹配进程，返回 (found, pids)。

    关键：不用 `pgrep -f`。macOS 的 pgrep 会**静默排除调用者的父进程链**
    （据实测：由 dsh 自身 spawn 的 CLI 里跑 pgrep -f "dsh web" 匹配不到
    dsh 主进程，只匹配到无关进程），导致插件内探活把健康主误判为故障。
    改为 `ps` 全量扫描 + 只排除自身（os.getpid()），父链中的 dsh 能被正确
    命中；正则语义与 pgrep -f 一致（standby 用 `dsh web.*--port N` 区分端口）。
    """
    rc, out, _ = run(["/bin/ps", "-axo", "pid=,command="])
    if rc != 0:
        return False, []
    self_pid = str(os.getpid())
    pids = []
    for line in out.splitlines():
        line = line.strip()
        if not line:
            continue
        pid, _, cmd = line.partition(" ")
        pid = pid.strip()
        if not pid.isdigit() or pid == self_pid:
            continue
        if re.search(pat, cmd):
            pids.append(pid)
    return bool(pids), pids


def probe(port=MAIN_PORT, pat=None):
    """三合一探活：TCP + HTTP + pgrep。返回 (ok, evidence) 证据链。
    pat 缺省按端口推断（主=无 --port 形态，其余=--port N）。"""
    if pat is None:
        pat = port_pat(port)
    t = tcp_up(port)
    h = http_up(port)
    g, pids = pgrep_up(pat)
    evidence = {
        "tcp_%d" % port: t,
        "http_%d" % port: h,
        "pgrep_pat": pat,
        "pids": pids,
    }
    return (t and h and g), evidence


def wait_probe(port=MAIN_PORT, pat=None, timeout=HEALTH_TIMEOUT, interval=2):
    """等待探活通过，返回 (ok, evidence)。"""
    if pat is None:
        pat = port_pat(port)
    deadline = time.time() + timeout
    last = None
    while time.time() < deadline:
        ok, ev = probe(port, pat)
        last = ev
        if ok:
            return True, ev
        time.sleep(interval)
    return False, last


# ---------------------------------------------------------------------------
# 快照 / standby
# ---------------------------------------------------------------------------

def ensure_dirs(home: Path):
    d = home / "double"
    for sub in ("snapshots", "standby", "evolutions", "tmp"):
        (d / sub).mkdir(parents=True, exist_ok=True)
    return d


def last_known_good(home: Path):
    lkg = home / "double" / "last-known-good"
    return lkg.read_text().strip() if lkg.exists() else None


def set_lkg(home: Path, snapname: str):
    (home / "double" / "last-known-good").write_text(snapname)


def snapshot(home: Path, name=None):
    """复制健康基线到 snapshots/<ts>-<name>，更新 last-known-good。"""
    d = ensure_dirs(home)
    ts = datetime.now().strftime("%Y%m%d-%H%M%S")
    snapname = f"{ts}-{name}" if name else f"{ts}-auto"
    dest = d / "snapshots" / snapname
    if dest.exists():
        shutil.rmtree(dest)
    dest.mkdir(parents=True, exist_ok=True)
    copied = []
    for rel in BASELINE_FILES:
        src = home / rel
        if src.exists():
            tgt = dest / rel
            tgt.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(src, tgt)
            copied.append(rel)
    for rel in BASELINE_DIRS:
        src = home / rel
        if src.is_dir():
            tgt = dest / rel
            tgt.parent.mkdir(parents=True, exist_ok=True)
            shutil.copytree(src, tgt, dirs_exist_ok=True, symlinks=True)
            copied.append(rel)
    # 可选 home 级 patch
    if (home / "cordis.patch.yml").exists():
        shutil.copy2(home / "cordis.patch.yml", dest / "cordis.patch.yml")
        copied.append("cordis.patch.yml")
    set_lkg(home, snapname)
    return snapname, copied


def list_snapshots(home: Path):
    d = home / "double" / "snapshots"
    if not d.exists():
        return []
    return sorted([p.name for p in d.iterdir() if p.is_dir()])


def standby_dir(home: Path, override=None):
    if override:
        return Path(override)
    return home / "double" / "standby"


def sync_standby(home: Path, stb: Path):
    """把主配置层同步到 standby；standby 的 node_modules symlink 到真实 profile。"""
    stb.mkdir(parents=True, exist_ok=True)
    for rel in BASELINE_FILES:
        src = home / rel
        tgt = stb / rel
        if src.is_file():
            tgt.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(src, tgt)
    for rel in BASELINE_DIRS:
        src = home / rel
        if src.is_dir():
            tgt = stb / rel
            tgt.parent.mkdir(parents=True, exist_ok=True)
            if tgt.exists():
                shutil.rmtree(tgt)
            shutil.copytree(src, tgt, symlinks=True)
    if (home / "cordis.patch.yml").exists():
        shutil.copy2(home / "cordis.patch.yml", stb / "cordis.patch.yml")
    # node_modules symlink
    nm_link = stb / "profiles/web/node_modules"
    nm_link.parent.mkdir(parents=True, exist_ok=True)
    if nm_link.is_symlink():
        nm_link.unlink()
    if not nm_link.exists() and REAL_PROFILE_NM.exists():
        nm_link.symlink_to(REAL_PROFILE_NM)
    return stb


def apply_changes(src_dir: Path, target: Path, prefix: Path):
    """把变更目录里的文件按相对路径覆盖到 target。prefix 用于剥掉绝对路径前缀。"""
    applied = []
    for f in src_dir.rglob("*"):
        if f.is_file():
            rel = f.relative_to(src_dir)
            tgt = target / rel
            tgt.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(f, tgt)
            applied.append(str(rel))
    return applied


# ---------------------------------------------------------------------------
# 预检
# ---------------------------------------------------------------------------

def yaml_syntax_ok(path: Path):
    """用 macOS 自带 ruby(psych) 校验 YAML 语法（零第三方依赖）。"""
    rc, _, err = run(["ruby", "-ryaml", "-e", "YAML.load_file(ARGV[0])", str(path)], timeout=30)
    return rc == 0, err.strip()


def json_ok(path: Path):
    try:
        import json
        with open(path) as f:
            json.load(f)
        return True, ""
    except Exception as e:
        return False, str(e)


def check_bundles_present(home: Path, stb: Path):
    """校验 web profile bundles 里的包能否解析到（真实 profile node_modules 或全局）。"""
    pkg = stb / "profiles/web/package.json"
    issues = []
    if not pkg.exists():
        return ["missing profiles/web/package.json"]
    import json
    try:
        data = json.loads(pkg.read_text())
    except Exception as e:
        return [f"profiles/web/package.json invalid JSON: {e}"]
    bundles = (data.get("dsh", {}).get("profile", {}) or {}).get("bundles", [])
    if not isinstance(bundles, list):
        return ["bundles is not a list"]
    nm = stb / "profiles/web/node_modules"
    for b in bundles:
        cands = [nm / b, REAL_PROFILE_NM / b]
        # GLOBAL_DSH_NM 本身已在 @deepseek-ai 目录内：scoped 包（@scope/name）取 name 部分
        if b.startswith("@") and "/" in b:
            global_cand = GLOBAL_DSH_NM / b.split("/", 1)[1]
        else:
            global_cand = GLOBAL_DSH_NM / b
        cands.append(global_cand)
        if not any((c.exists() or c.is_symlink()) for c in cands):
            issues.append(f"bundle '{b}' cannot be resolved (not in node_modules / global dsh)")
    return issues


def check_default_model(home: Path, stb: Path):
    """防 NO_ADAPTER：agent-default-model.provider 必须指向已声明 provider 或真实存在。"""
    issues = []
    settings = stb / "settings.yaml"
    if not settings.exists():
        return []
    import json
    rc, out, err = run(["ruby", "-ryaml", "-rjson", "-e",
                        "s=YAML.load_file(ARGV[0]); print({default_model: s['agent-default-model'], providers: (s.dig('llm-pi-ai','providers')||{}).keys}.to_json)",
                        str(settings)], timeout=30)
    if rc != 0:
        issues.append(f"settings.yaml YAML parse failed: {err.strip()}")
        return issues
    try:
        info = json.loads(out)
    except Exception:
        return issues
    dm = info.get("default_model") or {}
    prov = dm.get("provider")
    if not prov:
        return issues
    known = set(info.get("providers") or [])
    # anliu/opencode-go 是已知 provider；运行期 twin（*-autovision/*-vision）应被质疑
    if prov not in known and (prov.endswith("-autovision") or prov.endswith("-vision")):
        issues.append(f"agent-default-model.provider='{prov}' 不是静态 provider（疑似运行期 twin），重启会 NO_ADAPTER")
    return issues


def dump_config(home: Path, stb: Path, profile="web"):
    """在 standby 上跑 --dump-config，返回 (ok, problems, detail, blob)。
    注意：dump-config 只打印合成 entry 树、不做 EntryGroup duplicate 校验——
    重复 loader entry id 是真实 boot（mountRootInclude）才报，需另做输出解析。"""
    rc, out, err = run([str(DSH_WRAPPER), "--profile", profile, "--dump-config"],
                       timeout=90, env=dsh_env(stb), cwd=str(stb))
    blob = out + "\n" + err
    problems = []
    for pat in FATAL_PATTERNS:
        if re.search(pat, blob, re.IGNORECASE):
            problems.append(pat)
    # 捕获具体错误行
    detail = ""
    for line in blob.splitlines():
        if any(k in line for k in ("duplicate loader entry", "not found", "YAMLException", "must be a top-level", "failed to apply", "NO_ADAPTER")):
            detail = line.strip()
            break
    return (rc == 0 and not problems), problems, detail, blob


def detect_duplicate_entries(blob):
    """从 dump-config 输出提取 loader entry id 统计重复（补 dump-config 不做的
    duplicate 校验：那是真实 boot 才触发的 EntryGroup.update 错误）。"""
    seen = {}
    for m in re.finditer(r"^- id: (\S+)", blob, re.MULTILINE):
        eid = m.group(1)
        seen[eid] = seen.get(eid, 0) + 1
    return [eid for eid, n in seen.items() if n > 1]


def precheck(home: Path, stb: Path, profile="web"):
    """完整静态预检，返回 (issues, detail)。issues 空 = 健康。"""
    issues = []
    detail = ""
    # 1. 关键文件 YAML/JSON 语法
    for rel in ("profiles/web/cordis.patch.yml", "profiles/headless/cordis.patch.yml",
                "settings.yaml", "autovision/config.json", "profiles/web/pnpm-workspace.yaml"):
        p = stb / rel
        if p.exists():
            if p.suffix in (".yml", ".yaml"):
                ok, err = yaml_syntax_ok(p)
                if not ok:
                    issues.append(f"{rel}: YAML 语法错误 {err}")
            elif p.suffix == ".json":
                ok, err = json_ok(p)
                if not ok:
                    issues.append(f"{rel}: JSON 语法错误 {err}")
    # 2. bundles 可解析
    issues += check_bundles_present(home, stb)
    # 3. 默认模型防 NO_ADAPTER
    issues += check_default_model(home, stb)
    # 4. dump-config（静态；不包含 duplicate 校验）+ 输出解析检测重复 entry id
    ok, problems, det, blob = dump_config(home, stb, profile)
    dupes = detect_duplicate_entries(blob)
    if dupes:
        issues.append("dump-config 输出发现重复 loader entry id: " + ", ".join(dupes)
                      + "（真实 boot 会报 duplicate loader entry 崩溃）")
    if problems:
        issues.append("dump-config 发现 boot 错误: " + ", ".join(problems))
        detail = det or detail
    elif not ok:
        issues.append("dump-config 非零退出")
    return issues, (det or detail)


def standby_live_test(stb: Path, port: int, timeout=BOOT_READY_TIMEOUT):
    """在 standby 上动态试启动 web --port <port>，出现 'dsh web: http://' 即 boot 成功。"""
    env = dsh_env(stb)
    outlog = stb / "double-trial.log"
    # 清空旧日志，注意不能覆盖主 web-daemon.log
    outlog.write_text("")
    fl = open(outlog, "a")
    p = subprocess.Popen([str(DSH_WRAPPER), "--profile", "web", "--port", str(port)],
                         env=env, cwd=str(stb), stdout=fl, stderr=fl)
    try:
        deadline = time.time() + timeout
        while time.time() < deadline:
            if p.poll() is not None:
                return False, f"standby 提前退出 rc={p.returncode}", outlog
            if LIVE_READY_MARK in outlog.read_text():
                # 再确认端口在听
                time.sleep(1)
                if tcp_up(port):
                    return True, f"standby boot 成功 @127.0.0.1:{port}", outlog
            time.sleep(0.5)
        return False, f"standby {timeout}s 内未 boot 完成", outlog
    finally:
        try:
            p.terminate()
            p.wait(timeout=10)
        except Exception:
            p.kill()


# ---------------------------------------------------------------------------
# standby 唤醒 / 停止
# ---------------------------------------------------------------------------

def wake_standby(stb: Path, port: int):
    env = dsh_env(stb)
    outlog = stb / "double-wake.log"
    fl = open(outlog, "a")
    p = subprocess.Popen([str(DSH_WRAPPER), "--profile", "web", "--port", str(port)],
                         env=env, cwd=str(stb), stdout=fl, stderr=fl)
    time.sleep(1)
    if p.poll() is not None:
        return False, f"standby 启动即退出 rc={p.returncode}", outlog
    return True, f"standby 已唤醒 @127.0.0.1:{port} pid={p.pid}", outlog


def stop_standby(port=STANDBY_PORT, pat="dsh web"):
    rc, out, _ = run(["pkill", "-f", f"dsh.*--port {port}"])
    return rc == 0, out


def standby_running(port=STANDBY_PORT):
    # 注意：试验田/试启动实例以 `dsh --profile web --port N` 启动，命令行里没有
    # 连续的 "dsh web" 字样，必须用 --port N 匹配（旧 pattern 曾因此恒 False）。
    return probe(port, rf"dsh.*--port {port}\b")


# ---------------------------------------------------------------------------
# 试验田双身（对等双脑）：隔离的完整副 dsh 实例 + 主↔副通信台账
# ---------------------------------------------------------------------------

def testbed_dir(home: Path):
    return home / "double" / "testbed"


def comm_dir(home: Path):
    return home / "double" / "comm"


def comm_file(home: Path):
    return comm_dir(home) / "transcript.jsonl"


def comm_tail(home: Path, n=20):
    f = comm_file(home)
    if not f.exists():
        return []
    lines = f.read_text().splitlines()
    out = []
    for line in lines[-n:]:
        line = line.strip()
        if not line:
            continue
        try:
            import json
            out.append(json.loads(line))
        except Exception:
            out.append({"raw": line[:200]})
    return out


def ensure_twin_token(home: Path):
    """确保 double/config.json 里有 twinToken（主副 /double/twin/* 路由的 Bearer 凭据）。"""
    import json
    cfg_path = home / "double" / "config.json"
    cfg = {}
    if cfg_path.exists():
        try:
            cfg = json.loads(cfg_path.read_text())
        except Exception:
            cfg = {}
    if not cfg.get("twinToken"):
        import secrets
        cfg["twinToken"] = secrets.token_hex(16)
        cfg_path.parent.mkdir(parents=True, exist_ok=True)
        cfg_path.write_text(json.dumps(cfg, indent=2, ensure_ascii=False))
        try:
            cfg_path.chmod(0o600)
        except OSError:
            pass
    return cfg["twinToken"]


def _copy_config_layer(src_home: Path, dst: Path):
    """把 src_home 的健康基线配置层复制到 dst（保留权限）。"""
    for rel in BASELINE_FILES:
        s = src_home / rel
        if s.is_file():
            t = dst / rel
            t.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(s, t)
            if rel == ".credentials.yaml":
                try:
                    t.chmod(0o600)
                except OSError:
                    pass
    for rel in BASELINE_DIRS:
        s = src_home / rel
        if s.is_dir():
            t = dst / rel
            if t.exists():
                shutil.rmtree(t)
            t.parent.mkdir(parents=True, exist_ok=True)
            shutil.copytree(s, t, symlinks=True)
    if (src_home / "cordis.patch.yml").exists():
        shutil.copy2(src_home / "cordis.patch.yml", dst / "cordis.patch.yml")


def _ensure_symlink(link: Path, target: Path):
    link.parent.mkdir(parents=True, exist_ok=True)
    if link.is_symlink():
        link.unlink()
    elif link.exists():
        shutil.rmtree(link)
    link.symlink_to(target)


def testbed_clone(home: Path, tb: Path):
    """克隆主配置层到试验田 + 只读共享 node_modules/attachments 对象库 + 独立数据目录。"""
    _copy_config_layer(home, tb)
    # 只读共享（同机同用户）：profile node_modules（插件解析）、home 根 node_modules（安装缓存）、
    # attachments 对象库（内容寻址不可变）。试验田写附件会经 tmp 走自己的目录，不碰主。
    _ensure_symlink(tb / "profiles/web/node_modules", home / "profiles/web/node_modules")
    if (home / "node_modules").is_dir():
        _ensure_symlink(tb / "node_modules", home / "node_modules")
    if (home / "attachments/v1/objects").is_dir():
        _ensure_symlink(tb / "attachments/v1/objects", home / "attachments/v1/objects")
    # 工作流引导（dsh-agent-instructions 读 $DSH_HOME/AGENTS.md，dsh-skill-filesystem
    # 发现 user-dsh 根 <home>/skills）：把主的全局铁律与 skills 带给试验田 agent，
    # 否则它看不到"改自己先去试验田"的规则（它的 DSH_HOME 是试验田目录）。
    if (home / "AGENTS.md").is_file():
        shutil.copy2(home / "AGENTS.md", tb / "AGENTS.md")
    if (home / "skills").is_dir():
        _ensure_symlink(tb / "skills", home / "skills")
    # 全新独立数据（完全隔离）：会话/存储/工作区由试验田自己长，绝不复制主的历史
    for d in ("sessions", "storages", "attachments/v1/tmp", "work"):
        (tb / d).mkdir(parents=True, exist_ok=True)
    return tb


def testbed_write_baseline(home: Path, tb: Path):
    """用主当前配置层刷新试验田基线（diff 的增量参照；不触碰试验田现有配置）。"""
    bl = tb / "baseline"
    if bl.exists():
        shutil.rmtree(bl)
    bl.mkdir(parents=True)
    _copy_config_layer(home, bl)
    (tb / "baseline.stamp").write_text(datetime.now().isoformat(timespec="seconds"))
    return bl


def testbed_running(port=TESTBED_PORT):
    return probe(port, rf"dsh.*--port {port}\b")


def testbed_up_sync(home: Path, tb: Path, port: int, timeout=BOOT_READY_TIMEOUT + 20):
    """启动试验田为完整副 dsh（detached），等到 boot 就绪。返回 (ok, msg)。"""
    running, ev = testbed_running(port)
    if running:
        return True, f"试验田已在运行 @127.0.0.1:{port} pids={ev.get('pids')}"
    outlog = tb / "double-web.log"
    outlog.parent.mkdir(parents=True, exist_ok=True)
    fl = open(outlog, "ab")
    # start_new_session=True：脱离 CLI 的进程组，CLI 退出后试验田继续常驻
    p = subprocess.Popen([str(DSH_WRAPPER), "--profile", "web", "--port", str(port)],
                         env=dsh_env(tb), cwd=str(tb),
                         stdout=fl, stderr=fl, start_new_session=True)
    deadline = time.time() + timeout
    while time.time() < deadline:
        if p.poll() is not None:
            return False, f"试验田启动即退出 rc={p.returncode}（查 {outlog}）"
        if LIVE_READY_MARK in outlog.read_text(errors="ignore") and tcp_up(port):
            time.sleep(1)
            if tcp_up(port):
                return True, f"试验田已唤醒 @127.0.0.1:{port} pid={p.pid} 日志={outlog}"
        time.sleep(0.5)
    return False, f"试验田 {timeout}s 内未 boot 完成（查 {outlog}）"


def testbed_down_sync(port: int = TESTBED_PORT, timeout=15):
    # 注意：pkill -f 是 POSIX ERE，不支持 \b（PCRE 扩展）——必须用 $ 锚定，
    # 否则 pattern 静默失配、旧进程杀不掉，"重启"变 no-op。
    def _pkill(sig):
        run(["pkill", sig, "-f", rf"dsh.*--port {port}$"])
    _pkill("-TERM")
    deadline = time.time() + timeout
    while time.time() < deadline:
        if not tcp_up(port):
            return True, "试验田已休眠（磁盘态，零进程）"
        time.sleep(1)
    _pkill("-9")
    deadline = time.time() + 8
    while time.time() < deadline:
        if not tcp_up(port):
            return True, "试验田已休眠（SIGKILL 兜底后停止）"
        time.sleep(1)
    return False, f"试验田未能停止（pkill -f 'dsh.*--port {port}$' 无效，请人工检查）"


def testbed_diff_files(home: Path, tb: Path):
    """试验田配置层 vs 基线：返回有差异/新增的相对路径列表。"""
    import filecmp
    bl = tb / "baseline"
    if not bl.is_dir():
        die("试验田没有基线（先 testbed init 或 testbed baseline）")
    rels = set()
    for rel in BASELINE_FILES:
        t, b = tb / rel, bl / rel
        if t.exists() and not b.exists():
            rels.add(rel)
        elif t.exists() and b.exists() and not filecmp.cmp(t, b, shallow=False):
            rels.add(rel)
    for d in BASELINE_DIRS:
        td, bd = tb / d, bl / d
        if not td.is_dir():
            continue
        for f in td.rglob("*"):
            if not f.is_file() or f.is_symlink():
                continue
            rel = str(f.relative_to(tb))
            bf = bl / rel
            if not bf.exists() or not filecmp.cmp(f, bf, shallow=False):
                rels.add(rel)
    return sorted(rels)


# ---------------------------------------------------------------------------
# 命令实现
# ---------------------------------------------------------------------------

def cmd_snapshot(args):
    home = args.home
    snapname, copied = snapshot(home, args.name)
    print(f"快照已建立: {snapname}")
    print("内容: " + ", ".join(copied))
    print(f"last-known-good -> {snapname}")


def cmd_check(args):
    home = args.home
    stb = standby_dir(home, args.standby)
    sync_standby(home, stb)
    print(f"预检目标: standby={stb} (主配置已同步)")
    issues, detail = precheck(home, stb)
    if issues:
        print("健康判定: ❌ 不健康")
        for i in issues:
            print("  - " + i)
        if detail:
            print("  详情: " + detail[:400])
        sys.exit(1)
    else:
        print("健康判定: ✅ 可安全 boot（dump-config 通过，无 boot 致命错误）")
        sys.exit(0)


def cmd_preview(args):
    home = args.home
    stb = standby_dir(home, args.standby)
    src = Path(args.change_dir)
    if not src.is_dir():
        die(f"变更目录不存在: {src}")
    sync_standby(home, stb)
    applied = apply_changes(src, stb, src)
    print(f"变更已应用到 standby（主零改动）: {len(applied)} 个文件")
    for a in applied:
        print("  ~ " + a)
    issues, detail = precheck(home, stb)
    if issues:
        print("preview 判定: ❌ 变更不安全（以下问题）")
        for i in issues:
            print("  - " + i)
        if detail:
            print("  详情: " + detail[:400])
        sys.exit(1)
    if args.no_wake:
        print("preview 判定: ✅ 静态预检通过（跳过动态试启动）")
        sys.exit(0)
    ok, msg, _ = standby_live_test(stb, args.port)
    if not ok:
        print("preview 判定: ❌ 静态通过但试启动失败")
        print("  " + msg)
        sys.exit(1)
    print(f"preview 判定: ✅ 预检 + 试启动均通过（{msg}），可安全应用到主")
    sys.exit(0)


def cmd_evolve(args):
    import shlex
    home = args.home
    stb = standby_dir(home, args.standby)
    src = Path(args.change_dir)
    if not src.is_dir():
        die(f"变更目录不存在: {src}")
    main_port = args.main_port
    supervisor_cmd = shlex.split(args.supervisor_cmd) if args.supervisor_cmd else None
    ev = home / "double" / "evolutions"
    ev.mkdir(parents=True, exist_ok=True)
    evid = datetime.now().strftime("%Y%m%d-%H%M%S")
    evlog = ev / f"{evid}.log"
    log(f"== evolve {evid} 开始（严格蓝绿提交） home={home} main_port={main_port} ==", evlog)

    # 1. pre-evolution 快照
    snapname, _ = snapshot(home, f"pre-evol-{evid}")
    log(f"步骤1 pre-evolution 快照: {snapname}", evlog)

    # 2-3. 应用到 standby + 静态预检
    sync_standby(home, stb)
    applied = apply_changes(src, stb, src)
    log(f"步骤2-3 变更已应用到 standby（主未动）: {len(applied)} 文件", evlog)
    issues, detail = precheck(home, stb)
    if issues:
        for i in issues:
            log(f"  [standby 预检失败] {i}", evlog)
        log("进化中止：主实例零改动，standby 保持待命（未停止，供排查）", evlog)
        print("evolve: ❌ standby 预检失败，主未改动。standby 保持待命（dsh-double stop 可停）。")
        for i in issues:
            print("  - " + i)
        sys.exit(1)

    # 4. standby 动态试启动
    if not args.no_wake:
        ok, msg, _ = standby_live_test(stb, args.port)
        if not ok:
            log(f"步骤4 standby 试启动失败: {msg}", evlog)
            log("进化中止：主实例零改动", evlog)
            print(f"evolve: ❌ standby 试启动失败（{msg}），主未改动。")
            sys.exit(1)
        log(f"步骤4 standby 试启动通过: {msg}", evlog)

    # 5. 应用变更到主
    applied_main = apply_changes(src, home, src)
    log(f"步骤5 变更已应用到主: {len(applied_main)} 文件", evlog)

    # 6-7. 重启主（演练/只读时 --no-restart 跳过）
    if args.no_restart:
        log("步骤6-7 --no-restart：跳过主重启（只读模式）", evlog)
        ok, ev_ = probe(main_port)
        if not ok:
            log("注意：主当前探活不健康（未重启，仅记录）", evlog)
            print("evolve: ⚠️ 变更已应用到主但未重启（--no-restart），主当前探活: " +
                  ("健康" if ok else "不健康"))
            sys.exit(0)
        print("evolve: ✅ 变更已应用到主（--no-restart 未重启）。蓝绿顺序: standby 已停，主保持健康。")
        stop_standby(args.port)
        log("主保持健康，standby 已停止", evlog)
        sys.exit(0)

    restart_cmd = supervisor_cmd or [str(HOME / ".local/bin/dshd-supervisor"), "restart"]
    # 关键：重启进程链（supervisor→dshd-start→dsh web）必须以目标 home 启动。
    # 引擎可能由试验田插件 spawn（其 env 的 DSH_HOME=试验田），不覆盖的话
    # 主会被错误地以试验田 home 拉起（实测踩坑：3080 上跑出 role=testbed）。
    restart_env = {"DSH_HOME": str(home)}
    log("步骤6 重启主…", evlog)
    rc, so, se = run(restart_cmd, timeout=60, env=restart_env)
    log(f"步骤7 重启主完成 rc={rc}", evlog)
    ok, evidence = wait_probe(main_port, timeout=HEALTH_TIMEOUT)
    if ok:
        log("步骤7 主健康确认 ✅  → 现在才停 standby（你的硬约束：主健康前 standby 绝不提前停）", evlog)
        stop_standby(args.port)
        log("步骤8 进化完成，standby 已停止，记录事件", evlog)
        print("evolve: ✅ 主重启并确认健康，standby 随后停止。进化完成。")
        sys.exit(0)
    else:
        log(f"步骤7 主启动失败: {evidence}", evlog)
        log("步骤8 回滚主到 pre-evolution 快照并重启…", evlog)
        # 回滚到本次 pre-evolution 快照
        rollback_to(home, snapname)
        rc, so, se = run(restart_cmd, timeout=60, env=restart_env)
        ok2, ev2 = wait_probe(main_port, timeout=HEALTH_TIMEOUT)
        if ok2:
            log("回滚后主恢复健康 ✅，standby 停止", evlog)
            stop_standby(args.port)
            print("evolve: 🔄 主启动失败，已自动回滚到 pre-evolution 快照并恢复健康。")
            print("  回滚快照: " + snapname)
            sys.exit(0)
        else:
            log(f"!! 回滚后主仍未恢复: {ev2} —— 请人工介入（standby 保持待命 @{args.port}）", evlog)
            print("evolve: ❌ 回滚后主仍未恢复！standby 保持待命（dsh-double wake 可唤起人工窗口）。")
            sys.exit(2)


def rollback_to(home: Path, snapname: str):
    snap = home / "double" / "snapshots" / snapname
    if not snap.is_dir():
        die(f"快照不存在: {snapname}")
    for rel in BASELINE_FILES:
        src = snap / rel
        if src.exists():
            tgt = home / rel
            tgt.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(src, tgt)
    for rel in BASELINE_DIRS:
        src = snap / rel
        if src.is_dir():
            tgt = home / rel
            shutil.rmtree(tgt, ignore_errors=True)
            tgt.parent.mkdir(parents=True, exist_ok=True)
            shutil.copytree(src, tgt, symlinks=True)
    if (snap / "cordis.patch.yml").exists():
        shutil.copy2(snap / "cordis.patch.yml", home / "cordis.patch.yml")
    set_lkg(home, snapname)


def cmd_rollback(args):
    home = args.home
    snapname = args.snapshot or last_known_good(home)
    if not snapname:
        die("没有可用快照（先运行 snapshot）")
    rollback_to(home, snapname)
    print(f"已回滚到快照: {snapname}")
    print("提示: 配置已恢复，需重启 dsh 生效（dshd-supervisor restart）。")


def cmd_wake(args):
    home = args.home
    stb = standby_dir(home, args.standby)
    sync_standby(home, stb)
    ok, msg, _ = wake_standby(stb, args.port)
    if not ok:
        print("wake: ❌ " + msg)
        sys.exit(1)
    print(f"wake: ✅ {msg}")
    print(f"  standby 是独立 DSH_HOME（{stb}），可作备用大脑/人工窗口；完事用 dsh-double stop 退回磁盘态。")


def cmd_stop(args):
    ok, out = stop_standby(args.port)
    print("stop: standby 已停止（回到磁盘休眠态）" if ok else "stop: 无 standby 进程或已停止")


def cmd_status(args):
    home = args.home
    ok, ev = probe(MAIN_PORT)
    print(f"主 dsh @127.0.0.1:{MAIN_PORT}: {'✅ 健康' if ok else '❌ 不可用'}  证据: {ev}")
    lkg = last_known_good(home)
    snaps = list_snapshots(home)
    print(f"last-known-good: {lkg or '(无)'}")
    print(f"快照数: {len(snaps)}")
    for s in snaps[-5:]:
        print("  " + s)
    stb = standby_dir(home, args.standby)
    sok, sev = standby_running(args.port)
    print(f"standby @{args.port}: {'🟢 唤醒中' if sok else '💤 休眠（磁盘副本）'}  证据: {sev}")
    print(f"standby 目录: {stb}")
    tb = testbed_dir(home)
    trunning, tev = testbed_running(TESTBED_PORT)
    print(f"试验田 @{TESTBED_PORT}: {'🟢 运行中' if trunning else ('💤 磁盘态' if tb.is_dir() else '未创建（testbed init）')}  证据: {tev}")
    tail = comm_tail(home, 1)
    print(f"最近双 dsh 通信: {('[%s] %s %s' % (tail[0].get('dir'), tail[0].get('kind'), tail[0].get('status'))) if tail else '(无)'}")
    evd = home / "double" / "evolutions"
    if evd.exists():
        evs = sorted(evd.glob("*.log"))
        if evs:
            print(f"最近进化事件（共 {len(evs)}）:")
            for e in evs[-3:]:
                first = e.read_text().splitlines()
                print(f"  {e.name}: {first[0] if first else ''}")


def cmd_standby_sync(args):
    home = args.home
    stb = standby_dir(home, args.standby)
    sync_standby(home, stb)
    print(f"standby 已同步: {stb}")


# ---------------------------------------------------------------------------
# 试验田 / 通信台账 命令
# ---------------------------------------------------------------------------

def cmd_testbed(args):
    home = args.home
    tb = testbed_dir(home)
    act = args.taction

    if act == "init":
        if tb.exists() and not args.force:
            die(f"试验田已存在: {tb}（--force 重建；sync 只刷新配置层）")
        if args.force and tb.exists():
            testbed_down_sync(args.port)
            shutil.rmtree(tb)
        tb.mkdir(parents=True)
        testbed_clone(home, tb)
        testbed_write_baseline(home, tb)
        ensure_twin_token(home)
        ensure_dirs(tb)  # 试验田自己的 double/（引擎对其操作时的 standby/快照）
        print(f"试验田已创建: {tb}")
        print("  配置层: 从主克隆；node_modules/attachments 对象库只读共享")
        print("  数据: sessions/storages/pet 全新独立（不碰主历史）")
        print(f"  唤醒: dsh-double.py testbed up --port {args.port}")
    elif act == "sync":
        if not tb.is_dir():
            die(f"试验田不存在: {tb}（先 testbed init）")
        testbed_clone(home, tb)
        testbed_write_baseline(home, tb)
        print(f"试验田配置层已从主重新克隆，基线已刷新: {tb}")
        print("  注意：试验田未提交的实验配置被覆盖；数据目录保持独立不受影响")
    elif act == "reset":
        args.force = True
        tb_existing = tb.exists()
        if tb_existing:
            testbed_down_sync(args.port)
            shutil.rmtree(tb)
        tb.mkdir(parents=True)
        testbed_clone(home, tb)
        testbed_write_baseline(home, tb)
        ensure_twin_token(home)
        ensure_dirs(tb)
        print(f"试验田已重建: {tb}（只清 double/testbed/，主数据零影响）")
    elif act == "up":
        if not tb.is_dir():
            die(f"试验田不存在: {tb}（先 testbed init）")
        ok, msg = testbed_up_sync(home, tb, args.port)
        print(("up: ✅ " if ok else "up: ❌ ") + msg)
        sys.exit(0 if ok else 1)
    elif act == "down":
        ok, msg = testbed_down_sync(args.port)
        print(("down: ✅ " if ok else "down: ❌ ") + msg)
        sys.exit(0 if ok else 1)
    elif act == "restart":
        ok1, msg1 = testbed_down_sync(args.port)
        if not ok1:
            print(f"restart: ❌ {msg1}（未执行 up，保持现状）")
            sys.exit(1)
        ok2, msg2 = testbed_up_sync(home, tb, args.port)
        print("restart: " + ("✅" if ok2 else "❌") + f" {msg1} → {msg2}")
        sys.exit(0 if ok2 else 1)
    elif act == "status":
        exists = tb.is_dir()
        running, ev = testbed_running(args.port)
        print(f"试验田目录: {tb} {'存在' if exists else '❌ 不存在（先 init）'}")
        print(f"试验田 @127.0.0.1:{args.port}: {'🟢 运行中' if running else '💤 磁盘态（零进程）'}  证据: {ev}")
        stamp = tb / "baseline.stamp"
        print(f"基线时间: {stamp.read_text().strip() if stamp.exists() else '(无)'}")
        diffs = testbed_diff_files(home, tb) if exists and (tb / 'baseline').is_dir() else []
        print(f"未提交增量（vs 基线）: {len(diffs)} 个文件" + (f" → {', '.join(diffs[:6])}{'…' if len(diffs) > 6 else ''}" if diffs else ""))
        tail = comm_tail(home, 1)
        if tail:
            r = tail[0]
            print(f"最近通信: [{r.get('dir')}] {r.get('kind')} {r.get('status')} @ {r.get('ts')}")
        else:
            print("最近通信: (无台账)")
    elif act == "diff":
        if not tb.is_dir():
            die(f"试验田不存在: {tb}")
        rels = testbed_diff_files(home, tb)
        out = Path(args.out) if args.out else (home / "double" / "tmp" /
                                               f"twin-diff-{datetime.now().strftime('%Y%m%d-%H%M%S')}")
        if out.exists():
            shutil.rmtree(out)
        out.mkdir(parents=True)
        for rel in rels:
            t = out / rel
            t.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(tb / rel, t)
        # 机器可解析行（插件 twinAutoDiff 正则匹配此行）：路径后不要黏任何后缀
        print(f"增量 change_dir: {out}")
        print(f"增量文件数: {len(rels)}（可喂给 evolve 应用到主）")
        for r in rels:
            print("  ~ " + r)
        if not rels:
            print("  （试验田与基线一致，无增量）")
    elif act == "baseline":
        if not tb.is_dir():
            die(f"试验田不存在: {tb}")
        testbed_write_baseline(home, tb)
        print("基线已用主当前配置刷新（增量基准归零；试验田配置未动）")
    else:
        die(f"未知 testbed 动作: {act}")


def cmd_comm(args):
    home = args.home
    recs = comm_tail(home, args.n)
    if not recs:
        print(f"(通信台账为空: {comm_file(home)})")
        return
    for r in recs:
        head = (r.get("text") or r.get("cmd") or r.get("path") or "")
        head = head.replace("\n", " ")[:80]
        reply = (r.get("reply") or "").replace("\n", " ")
        print(f"[{r.get('ts')}] {r.get('dir','?')} {r.get('kind','?')} {r.get('status','?')}")
        print(f"  → {head}")
        if reply:
            print(f"  ← {reply[:120]}{'…' if len(reply) > 120 else ''}")


# ---------------------------------------------------------------------------
# main
# ---------------------------------------------------------------------------

def main():
    import argparse
    ap = argparse.ArgumentParser(prog="dsh-double", description="dsh 自我管理 CLI（按需唤醒的休眠双身）")
    sub = ap.add_subparsers(dest="cmd", required=True)

    def add_common(p):
        p.add_argument("--home", default=None, help="主 DSH_HOME（默认 $DSH_HOME 或 ~/.dsh）")
        p.add_argument("--standby", default=None, help="standby 目录（默认 <主>/double/standby）")
        p.add_argument("--port", type=int, default=STANDBY_PORT, help=f"standby 端口（默认 {STANDBY_PORT}）")
        p.add_argument("--main-port", type=int, default=MAIN_PORT, help=f"主探活端口（默认 {MAIN_PORT}；演练假主可改）")
        p.add_argument("--supervisor-cmd", default=None, help="重启主用的命令字符串（默认 'dshd-supervisor restart'；演练假主可改）")

    p = sub.add_parser("snapshot"); p.add_argument("name", nargs="?", default=None); add_common(p); p.set_defaults(fn=cmd_snapshot)
    p = sub.add_parser("check"); add_common(p); p.set_defaults(fn=cmd_check)
    p = sub.add_parser("preview"); p.add_argument("change_dir"); add_common(p); p.add_argument("--no-wake", action="store_true", help="跳过动态试启动"); p.set_defaults(fn=cmd_preview)
    p = sub.add_parser("evolve"); p.add_argument("change_dir"); add_common(p); p.add_argument("--no-wake", action="store_true"); p.add_argument("--no-restart", action="store_true", help="应用主后不重启（只读/演练）"); p.set_defaults(fn=cmd_evolve)
    p = sub.add_parser("wake"); add_common(p); p.set_defaults(fn=cmd_wake)
    p = sub.add_parser("stop"); add_common(p); p.set_defaults(fn=cmd_stop)
    p = sub.add_parser("rollback"); p.add_argument("snapshot", nargs="?", default=None); add_common(p); p.set_defaults(fn=cmd_rollback)
    p = sub.add_parser("status"); add_common(p); p.set_defaults(fn=cmd_status)
    p = sub.add_parser("standby-sync"); add_common(p); p.set_defaults(fn=cmd_standby_sync)

    tbp = sub.add_parser("testbed", description="试验田双身（隔离的完整副 dsh）")
    tsub = tbp.add_subparsers(dest="taction", required=True)

    def add_tb(p):
        p.add_argument("--home", default=None, help="主 DSH_HOME（默认 $DSH_HOME 或 ~/.dsh）")
        p.add_argument("--port", type=int, default=TESTBED_PORT, help=f"试验田端口（默认 {TESTBED_PORT}）")

    p = tsub.add_parser("init"); p.add_argument("--force", action="store_true", help="已存在时重建"); add_tb(p); p.set_defaults(fn=cmd_testbed)
    p = tsub.add_parser("sync"); add_tb(p); p.set_defaults(fn=cmd_testbed)
    p = tsub.add_parser("reset"); add_tb(p); p.set_defaults(fn=cmd_testbed)
    p = tsub.add_parser("up"); add_tb(p); p.set_defaults(fn=cmd_testbed)
    p = tsub.add_parser("down"); add_tb(p); p.set_defaults(fn=cmd_testbed)
    p = tsub.add_parser("restart"); add_tb(p); p.set_defaults(fn=cmd_testbed)
    p = tsub.add_parser("status"); add_tb(p); p.set_defaults(fn=cmd_testbed)
    p = tsub.add_parser("diff"); p.add_argument("--out", default=None, help="增量输出目录（默认 <主>/double/tmp/twin-diff-<ts>）"); add_tb(p); p.set_defaults(fn=cmd_testbed)
    p = tsub.add_parser("baseline"); add_tb(p); p.set_defaults(fn=cmd_testbed)

    p = sub.add_parser("comm", description="查看主↔副通信台账")
    p.add_argument("n", nargs="?", type=int, default=20, help="最近 N 条（默认 20）")
    p.add_argument("--home", default=None)
    p.set_defaults(fn=cmd_comm)

    args = ap.parse_args()
    if args.home:
        args.home = Path(args.home)
    else:
        args.home = Path(os.environ.get("DSH_HOME", HOME / ".dsh"))
    args.fn(args)


if __name__ == "__main__":
    main()