"<node>" "<npm-cli.js>" install @deepseek-ai/dsh@latest \
  --prefix "<你的安装目录>" \
  --registry=https://registry.npmjs.org/ --no-audit --no-fund

cd "<工作区>" && dsh --profile headless "只回复两个字：通了。不要调用任何工具。"

cd "<DSH_HOME>/profiles/web"
npm install "<插件包.tgz 的绝对路径>" --registry=https://registry.npmjs.org/ --no-audit --no-fund

- insert:
    - id: chengce-workbench          # 自定 id，随便起
      name: 'dsh-client-ui-urban-planning-agent-workbench'

node "<DSH_HOME>/profiles/web/node_modules/<包名>/scripts/install-bundled-skills.mjs"

} catch (error) {
  return json(res, 502, { ok: false, error: error instanceof Error ? error.message : String(error) });
}

3. 插件路由 —— 直接转发 OpenViking（OV 没跑就必然 502）
  FAIL GET /health      HTTP 502   {"ok":false,"error":"fetch failed"}
  FAIL GET /libraries   HTTP 502   {"ok":false,"error":"fetch failed"}

3b. 插件路由 —— 静默失败型（源码带 .catch，恒 200，要看内容）
  OK   GET /resources   HTTP 200   {"ok":true,"resources":[]}     ← 静默为空

4. 插件路由 —— 只读写本地文件（OV 在不在都该 200）
  OK   GET /skills            HTTP 200
  OK   GET /expert-bindings   HTTP 200
  OK   GET /skill-configs     HTTP 200

结论
  ▸ OpenViking 没跑，而插件依赖它 —— 这就是「fetch failed / 502」的原因。

const [libraryValue, bindingValue, configValue] = await Promise.all([
  knowledgeApi.libraries(),                          // ← 唯一碰 OpenViking 的，挂了整页报错
  knowledgeApi.bootstrapExpertBindings(DEFAULT_BINDINGS),
  knowledgeApi.skillConfigs(),
]);

"<python.exe>" -m venv "<venv 路径>"
"<venv>/Scripts/python.exe" -m pip install openviking    # 实测 0.4.20，约 200+ 包

EmbeddingConfigurationError: Failed to download local embedding model
'bge-small-zh-v1.5-f16' from https://huggingface.co/... ConnectTimeoutError

curl -L -o "<用户目录>/.cache/openviking/models/bge-small-zh-v1.5-f16.gguf" \
  "https://hf-mirror.com/CompendiumLabs/bge-small-zh-v1.5-gguf/resolve/main/bge-small-zh-v1.5-f16.gguf"
# 校验：应 47,886,240 字节，文件头 4 字节为 'GGUF'

# 索引页（含 cp313 / win_amd64，可选 0.2.86 ~ 0.2.90）：
#   https://abetlen.github.io/llama-cpp-python/whl/cpu/llama-cpp-python/
"<venv>/Scripts/python.exe" -m pip install \
  "https://github.com/abetlen/llama-cpp-python/releases/download/v0.2.90/llama_cpp_python-0.2.90-cp313-cp313-win_amd64.whl"
# 仅 3MB，秒装，无需编译

{
  "server": { "host": "127.0.0.1", "port": 1933, "auth_mode": "dev" },
  "storage": {
    "workspace": "<你的数据目录>/openviking/data",
    "agfs": { "backend": "local" },
    "vectordb": { "backend": "local" }
  }
}

// src/host.js
const response = await fetch(`${ENDPOINT}${path}`, {
  ...requestInit,
  headers: { 'X-OpenViking-Actor-Peer': 'deepseek-harness', ...(requestInit.headers || {}) },
});

curl -s http://127.0.0.1:1933/health     # {"status":"ok","healthy":true,"version":"0.4.20","auth_mode":"dev"}
curl -s http://127.0.0.1:1933/ready      # {"status":"ready","checks":{...,"embedding":"ok"}}

netstat -ano | findstr /C:"127.0.0.1:1933 " | findstr /C:"LISTENING" >nul 2>&1
if errorlevel 1 start "OpenViking 知识库服务" /min "%~dp0openviking.cmd"
curl -s -o nul --noproxy "*" --max-time 2 "http://127.0.0.1:1933/health"

agent-presets: preset "up-renewal-policy" not found (available: standard, ptc, minimal, cordis)

// src/config/branding.js
export const AGENT = { id: 'U-E1', name: '政策咨询推送与解读专家',
                       preset: 'up-renewal-policy', skill: 'u-policy' };
export const BUSINESS_AGENTS = {
  urban: AGENT,
  overseas: { id: 'S1', name: '研判决策专家',
              preset: 'up-renewal-research', skill: 'o-s1' },
};

"files": ["lib/", "src/", "skills/", "scripts/", "README.md", "INSTALL.md"]

const preset = summary?.projectionValues?.agentPreset;
if (sessionState.blank && preset !== agent.preset) {
  const selected = await ctx.remote.agentPresets.select(current, agent.preset);  // ← 这里抛
}

<DSH_HOME>/.agent-presets/
├── up-renewal-policy/          ← 目录名 = preset id，必须严格一致
│   ├── preset.yml              ← 只有 name / description / order
│   └── agent.cordis.yml        ← 组成文件（254 行）
└── up-renewal-research/
    ├── preset.yml
    └── agent.cordis.yml

- id: persona
  name: '@deepseek-ai/dsh-persona'
  config:
    prefix: |-                  # 必填。用 | 字面块保留换行
      你是……                    #   （用 > 折叠块会把 bullet 列表挤成一整段）
    suffix: 当前工作目录：{{cwd}}。   # 可选
    # complete: true            # 可选：让 prefix 成为完整系统提示词（会压掉 suffix 与所有其他段）
    # includeRuntimeContext: false   # 可选：关掉动态运行上下文快照

=== 断言：插件引用的 preset 是否都在 roster 里且健康 ===
  [PASS] up-renewal-policy    →  trust=user  显示名=城策 · 政策咨询与解读
  [PASS] up-renewal-research  →  trust=user  显示名=城策 · 海外项目研判

openviking.utils.process_lock.DataDirectoryLocked: Another OpenViking process
(PID 6628) is already using the data directory 'C:\Users\<你>\.openviking\data'.
Running multiple OpenViking instances on the same data directory causes silent
storage contention and data corruption.
...
[openviking] 服务已退出，errorlevel=3

Get-CimInstance Win32_Process -Filter "ProcessId=6628" |
  Select-Object Name, ExecutablePath, CreationDate, ParentProcessId

python ov-stale-lock.py          # 诊断（只读，退出码 1 = 发现陈旧锁）
python ov-stale-lock.py --fix    # 清理（先把旧锁备份成 .openviking.pid.stale-<时间戳>）

====================================================================
OpenViking 陈旧数据目录锁检测
====================================================================
  配置      : C:\Users\Ace\.openviking\ov.conf
  数据目录  : C:/Users/Ace/.openviking/data
  服务端点  : 127.0.0.1:1933
  锁文件    : C:/Users/Ace/.openviking/data\.openviking.pid

[判据 A · 端口] 127.0.0.1:1933 监听中 = False
[判据 B · 锁文件] PID=6628  写入时间=2026-09-16 17:01:44
                 该 PID 当前镜像 = XLSmartService.exe
                 是否 OpenViking = False

[结论] 陈旧锁（stale lock）—— 这就是 OpenViking 起不来、报
       DataDirectoryLocked: Another OpenViking process (PID 6628) 的原因。

       端口 1933 无人监听 ⇒ 不存在在服务的实例；
       PID 6628 的真实身份是「XLSmartService.exe」⇒ 与 OpenViking 无关，只是复用了这个 PID。

[处置] 仅诊断模式，未改动任何文件。确认上面两条后，加 --fix 清理：
         <python> ov-stale-lock.py --fix

rem --- 陈旧锁自检（Windows 专属坑）---
rem 非正常关机 / 强杀时 atexit 清理没跑到，data\.openviking.pid 会残留；
rem 而 Windows 分支只判断「该 PID 有没有进程」，不校验身份 → 误判"已有实例在跑"。
rem 上面已确认端口空闲，这里再按「端口无监听 + 该 PID 不是 OpenViking」双重判据确认。
if exist "%OV_LOCK%" (
  echo [openviking] 发现残留的 PID 锁文件，正在确认是否为陈旧锁...
  if exist "%OV_LOCK_PY%" (
    "%OV_PY%" "%OV_LOCK_PY%" --fix
  ) else (
    echo [openviking] 未找到锁检测脚本，跳过自检: %OV_LOCK_PY%
  )
  echo.
)

沙箱根 = sandbox-cli.exe  +  WorkBuddy.exe --serve --session-id ...
         （本机实测：前者后代 7 个、后者后代 17 个，并集 18 个进程）

import subprocess
subprocess.Popen([r'C:\Windows\explorer.exe',
                  r'C:\Users\<你>\.dsh\bin\openviking.cmd'],
                 stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

python chengce-doctor.py                 # 服务 + 路由 + preset + skill 一次查完
python chengce-doctor.py --dsh-port 3080 --ov-port 1933 --dsh-home "D:/dsh"
node   verify_presets.mjs                # 只看 preset 是否真的可挂载
python win-proc-tree.py --list-sandbox   # 看沙箱根是谁、树里有多少进程
python ov-stale-lock.py                  # 查陈旧锁（加 --fix 清理，会先备份）

function allowed(req) {
  const address = req.socket.remoteAddress;
  if (!['127.0.0.1', '::1', '::ffff:127.0.0.1'].includes(address)) return false;
  return req.headers['sec-fetch-site'] !== 'cross-site';
}

const BASE      = '/chengce-knowledge/v1';                             // 路由前缀
const ROOT      = 'viking://resources/chengce';                        // OpenViking 里的库根
const META_FILE = join(homedir(), '.dsh', 'chengce-knowledge.json');   // ⚠️ 硬编码 .dsh
const SKILLS_DIR= join(homedir(), '.agents', 'skills');

python chengce-doctor.py
python chengce-doctor.py --dsh-port 3080 --ov-port 1933 --dsh-home "D:/dsh"

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""城策插件（dsh-client-ui-urban-planning-agent-workbench）一键体检。

为什么需要它：装完城策后常见两种报错长得完全不同、成因却都在"依赖没到位"——
  ① 知识库 / 技能管理页报 fetch failed（控制台 502）
  ② 新建会话发第一条消息报 preset "..." not found
本脚本用一条命令把 ① 的故障面切开，并顺带检查 ② 的两个 preset 是否就位。

用法：
    python chengce-doctor.py
    python chengce-doctor.py --dsh-port 3080 --ov-port 1933 --dsh-home "D:/dsh"

判定原理（来自插件源码 src/host.js）：
    BASE = /chengce-knowledge/v1，所有分支的 catch 统一 `json(res, 502, {ok,error})`，
    所以 **502 永远只等于"插件转发给 OpenViking 失败"**，不代表 dsh 挂了。
    其中 /skills /expert-bindings /skill-configs 只读写本地文件，**不碰 OpenViking**；
    /health /libraries /resources /upload /search 才会转发 OpenViking。
"""
import argparse
import json
import os
import socket
import sys
import urllib.error
import urllib.request

try:                                     # 中文输出在重定向到文件时也别乱码
    sys.stdout.reconfigure(encoding='utf-8', errors='replace')
except Exception:                        # noqa: BLE001
    pass

# 本机没有 HTTP_PROXY 环境变量，但 urllib 在 Windows 上会读注册表里的系统代理
# （FlClash / Clash 这类工具写的），导致探测 127.0.0.1 被劫持成 502 —— 必须显式绕过。
OPENER = urllib.request.build_opener(urllib.request.ProxyHandler({}))

# 直接 await ov(...) 的接口：OpenViking 不在就抛错 → 插件统一 catch 成 502
NEEDS_OV = [
    ('GET',  '/health',    '服务健康（同时打 OV 的 /health + /ready）'),
    ('GET',  '/libraries', '知识库列表'),
]
# 源码里给每个 ov() 挂了 .catch(() => [])：**恒返回 200**，OV 不在时静默给空数组。
# 所以它不能用来判断 OV 是否健康 —— 反过来，"页面打得开但列表永远空" 可能正是 OV 挂了。
SILENT_ON_FAIL = [
    ('GET',  '/resources?libraryId=policy', '知识库内资源（OV 挂时静默返回空，仍 200）'),
]
# 只读写本地文件的接口（OV 在不在都该 200）
LOCAL_ONLY = [
    ('GET',  '/skills',          '读 <agentsHome>/skills'),
    ('GET',  '/expert-bindings', '读 <DSH_HOME>/chengce-knowledge.json'),
    ('GET',  '/skill-configs',   '读 <DSH_HOME>/chengce-knowledge.json'),
]

PRESETS = ['up-renewal-policy', 'up-renewal-research']
SKILLS = ['u-policy', 'o-s1', 'o-s1-01', 'o-s1-02', 'o-s1-03', 'o-s1-04', 'o-s1-05', 'o-s1-06']


def port_open(port, host='127.0.0.1'):
    sock = socket.socket()
    sock.settimeout(1.5)
    try:
        return sock.connect_ex((host, port)) == 0
    finally:
        sock.close()


def call(url, method='GET', body=None, timeout=15):
    headers = {'User-Agent': 'chengce-doctor', 'Accept': 'application/json'}
    data = None
    if body is not None:
        data = json.dumps(body).encode('utf-8')
        headers['content-type'] = 'application/json'
    req = urllib.request.Request(url, data=data, headers=headers, method=method)
    try:
        with OPENER.open(req, timeout=timeout) as resp:
            return resp.status, resp.read(3000).decode('utf-8', 'replace')
    except urllib.error.HTTPError as exc:
        return exc.code, exc.read(3000).decode('utf-8', 'replace')
    except Exception as exc:                                  # noqa: BLE001
        return None, '%s: %s' % (type(exc).__name__, exc)


def line(mark, label, detail):
    print('  %-4s %-42s %s' % (mark, label, detail))


def probe_group(title, base, items, results):
    print()
    print('=' * 88)
    print(title)
    print('=' * 88)
    for method, path, why in items:
        code, text = call(base + path, method, None)
        flat = ' '.join(text.split())
        mark = 'OK' if code == 200 else 'FAIL'
        line(mark, '%s %s' % (method, path[:36]), 'HTTP %-5s %s' % (code, flat[:96]))
        print('       └ %s' % why)
        results.append((path, code, flat))
    return results


def main():
    ap = argparse.ArgumentParser(description='城策插件一键体检')
    ap.add_argument('--dsh-port', type=int, default=3080, help='dsh web 端口，默认 3080')
    ap.add_argument('--ov-port', type=int, default=1933, help='OpenViking 端口，默认 1933')
    ap.add_argument('--dsh-home', default=os.environ.get('DSH_HOME') or os.path.join(os.path.expanduser('~'), '.dsh'),
                    help='DSH_HOME，默认 ~/.dsh')
    ap.add_argument('--agents-home', default=os.environ.get('DSH_AGENTS_HOME') or os.path.join(os.path.expanduser('~'), '.agents'),
                    help='技能根目录，默认 ~/.agents')
    args = ap.parse_args()

    dsh = 'http://127.0.0.1:%d' % args.dsh_port
    ov = 'http://127.0.0.1:%d' % args.ov_port
    base = dsh + '/chengce-knowledge/v1'

    print('城策插件体检')
    print('  dsh web     %s' % dsh)
    print('  OpenViking  %s' % ov)
    print('  DSH_HOME    %s' % args.dsh_home)

    # ---------- 1. 端口 ----------
    ov_up = port_open(args.ov_port)
    dsh_up = port_open(args.dsh_port)
    print()
    print('=' * 88)
    print('1. 端口')
    print('=' * 88)
    line('OK' if dsh_up else 'FAIL', 'dsh web  127.0.0.1:%d' % args.dsh_port, '监听中' if dsh_up else '未监听')
    line('OK' if ov_up else 'FAIL', 'OpenViking 127.0.0.1:%d' % args.ov_port, '监听中' if ov_up else '未监听')

    # ---------- 2. OpenViking 直连 ----------
    print()
    print('=' * 88)
    print('2. OpenViking 服务端直连')
    print('=' * 88)
    if ov_up:
        for path in ('/health', '/ready'):
            code, text = call(ov + path)
            flat = ' '.join(text.split())
            mark = 'OK' if code == 200 else 'FAIL'
            line(mark, 'GET %s' % path, 'HTTP %-5s %s' % (code, flat[:96]))
    else:
        line('SKIP', '（服务未监听，跳过）', '先启动 OpenViking 再看这一节')

    if not dsh_up:
        print()
        print('  结论：dsh web 没在监听，先启动它（插件路由由 dsh 提供）。')
        return 1

    # ---------- 3. 插件路由 ----------
    needs = probe_group('3. 插件路由 —— 直接转发 OpenViking（OV 没跑就必然 502）', base, NEEDS_OV, [])
    silent = probe_group('3b. 插件路由 —— 静默失败型（源码带 .catch，恒 200，要看内容）',
                         base, SILENT_ON_FAIL, [])
    local = probe_group('4. 插件路由 —— 只读写本地文件（OV 在不在都该 200）', base, LOCAL_ONLY, [])

    # ---------- 5. preset / skill / 元数据 ----------
    print()
    print('=' * 88)
    print('5. agent preset 与内置 skill')
    print('=' * 88)
    preset_root = os.path.join(args.dsh_home, '.agent-presets')
    preset_ok = True
    for pid in PRESETS:
        d = os.path.join(preset_root, pid)
        comp = os.path.join(d, 'agent.cordis.yml')
        meta = os.path.join(d, 'preset.yml')
        good = os.path.isfile(comp) and os.path.isfile(meta)
        preset_ok &= good
        line('OK' if good else 'FAIL', pid,
             'agent.cordis.yml + preset.yml 就位' if good else '缺文件（目录：%s）' % d)
    if not preset_ok:
        print('       └ 权威判定请另跑：node verify_presets.mjs')

    skills_root = os.path.join(args.agents_home, 'skills')
    hits = 0
    for sid in SKILLS:
        if os.path.isfile(os.path.join(skills_root, sid, 'SKILL.md')):
            hits += 1
    line('OK' if hits == len(SKILLS) else 'FAIL', '内置 skill %d/%d' % (hits, len(SKILLS)),
         skills_root if hits == len(SKILLS) else '缺少部分 skill，postinstall 可能被跳过')

    meta_file = os.path.join(args.dsh_home, 'chengce-knowledge.json')
    if os.path.isfile(meta_file):
        try:
            m = json.load(open(meta_file, encoding='utf-8'))
            libs = m.get('libraries') or []
            line('OK', 'chengce-knowledge.json', '%d 个知识库已登记，%d 位专家绑定'
                 % (len(libs), len(m.get('expertBindings') or {})))
        except Exception as exc:                              # noqa: BLE001
            line('FAIL', 'chengce-knowledge.json', '解析失败: %s' % exc)
    else:
        line('FAIL', 'chengce-knowledge.json', '不存在（知识库页首次打开会自动创建，也可 POST /bootstrap）')

    # ---------- 6. 结论 ----------
    needs_codes = dict((p, c) for p, c, _ in needs)
    local_codes = dict((p, c) for p, c, _ in local)
    lib_code = needs_codes.get('/libraries')
    # 静默型：返回 200 但 body 里列表为空 → 源码那层 catch 吞掉了 OpenViking 的错误
    silent_empty = any('"resources":[]' in t.replace(' ', '') for _, _, t in silent)

    print()
    print('=' * 88)
    print('结论')
    print('=' * 88)
    if not ov_up and lib_code == 502:
        print('  ▸ OpenViking 没跑，而插件依赖它 —— 这就是「fetch failed / 502」的原因。')
        print('    修：启动 OpenViking 服务（见手册第 3 节），再刷新页面。')
    elif ov_up and lib_code == 502:
        print('  ▸ OpenViking 在监听，但转发仍失败 —— 问题在服务端自身。')
        print('    查：OV 启动日志是否报 EmbeddingConfigurationError（缺本地嵌入模型或 llama-cpp-python）。')
    elif lib_code == 200 and all(c == 200 for c in local_codes.values()):
        print('  ▸ 插件路由全部正常 —— 知识库 / 技能管理页应当可用。')
    elif lib_code is None:
        print('  ▸ 连不上 dsh 的插件路由 —— 确认插件已注册到 profile 的 cordis.patch.yml 并重启过 dsh。')
    elif any(c == 403 for c in local_codes.values()):
        print('  ▸ 返回 403「仅允许本机同源访问」—— 你在通过代理 / 远程地址访问 dsh。')
        print('    修：用 http://127.0.0.1:<port>/?token=... 直接访问，别走反代。')
    else:
        print('  ▸ 状态混合，见上面各行的 FAIL。')

    if silent_empty and not ov_up:
        print('  ▸ 注意：/resources 仍是 200 但内容为空 —— 它是"静默失败型"（源码里带 catch），')
        print('    所以"知识库页打得开、里面永远没东西"同样是 OpenViking 没跑的表现。')

    if not preset_ok:
        print('  ▸ 另有 preset 缺失 —— 新建会话发第一条消息会报 preset "..." not found。')
        print('    修：见手册第 4 节，补建到 <DSH_HOME>/.agent-presets/ 下。')

    return 0


if __name__ == '__main__':
    sys.exit(main())

node verify_presets.mjs
node verify_presets.mjs --dsh-home "D:/dsh"

// 验证城策插件硬编码引用的 agent preset 是否真的可用。
//
// 为什么不能只看文件在不在：preset 是「插件组合」，文件存在 ≠ 组成合法 ≠ 模块能解析。
// 所以这里不重新实现扫描规则，而是直接调用 dsh-agent-presets 自己的
// scanRoot / discoverPresets —— 判定与运行时完全一致（涵盖 YAML 方言、行结构、模块可解析性）。
//
// 用法：
//   node verify_presets.mjs
//   node verify_presets.mjs --dsh-home "D:/dsh" --plugin "D:/dsh/profiles/web/node_modules/dsh-client-ui-urban-planning-agent-workbench"

import { pathToFileURL } from 'node:url';
import fs from 'node:fs';
import os from 'node:os';
import path from 'node:path';

// ---------- 参数 ----------
const argv = process.argv.slice(2);
function arg(name, fallback) {
  const i = argv.indexOf('--' + name);
  return i >= 0 && argv[i + 1] ? argv[i + 1] : fallback;
}
const DSH_HOME = (arg('dsh-home', process.env.DSH_HOME) || path.join(os.homedir(), '.dsh')).replace(/\\/g, '/');
const PLUGIN_NAME = 'dsh-client-ui-urban-planning-agent-workbench';

function firstExisting(candidates) {
  return candidates.find((p) => fs.existsSync(p));
}

// dsh-agent-presets 可能装在 profiles 根或某个 profile 下（hoisted / 非 hoisted 都可能）
const PROFILES = path.join(DSH_HOME, 'profiles');
const profileDirs = fs.existsSync(PROFILES)
  ? fs.readdirSync(PROFILES, { withFileTypes: true }).filter((d) => d.isDirectory()).map((d) => d.name)
  : [];
const PRESET_PKG = firstExisting([
  path.join(PROFILES, 'node_modules/@deepseek-ai/dsh-agent-presets'),
  ...profileDirs.map((p) => path.join(PROFILES, p, 'node_modules/@deepseek-ai/dsh-agent-presets')),
]);
if (!PRESET_PKG) {
  console.error('找不到 @deepseek-ai/dsh-agent-presets —— 检查 --dsh-home 是否正确：' + DSH_HOME);
  process.exit(2);
}

// harnessBase 必须是 URL；要给「某个 profile 目录」，packageInstalled 会从它逐级向上找 node_modules
const HARNESS_DIR = firstExisting([
  path.join(PROFILES, 'web'),
  ...profileDirs.map((p) => path.join(PROFILES, p)),
]);
const HARNESS_BASE = pathToFileURL(HARNESS_DIR + '/').href;

// ---------- 找出插件实际引用的 preset id（不硬编码，跟着插件源码走）----------
const pluginDir = arg('plugin', firstExisting([
  path.join(PROFILES, 'web/node_modules', PLUGIN_NAME),
  ...profileDirs.map((p) => path.join(PROFILES, p, 'node_modules', PLUGIN_NAME)),
]));

const WANT = [];
if (pluginDir && fs.existsSync(pluginDir)) {
  for (const rel of ['src/config/branding.js', 'src/config/experts.js']) {
    const f = path.join(pluginDir, rel);
    if (!fs.existsSync(f)) continue;
    const text = fs.readFileSync(f, 'utf8');
    for (const m of text.matchAll(/preset:\s*'([^']+)'/gu)) {
      if (m[1] !== 'null' && !WANT.includes(m[1])) WANT.push(m[1]);
    }
  }
}

// ---------- 跑 dsh 自己的 discovery ----------
const d = await import(pathToFileURL(path.join(PRESET_PKG, 'lib/types/discovery.js')).href);

// 注意：scanRoot 的 root.path 要「普通文件系统路径」（内部 resolve + 展开 ~），传 file:// 会静默返回空；
// harnessBase 反过来要 URL。这两个参数类型不一致，是最容易写错的地方。
const roots = [
  { path: path.join(PRESET_PKG, 'presets').replace(/\\/g, '/'), trust: 'system' },  // 对照组：随包 4 个内置
  { path: path.join(DSH_HOME, '.agent-presets'), trust: 'user' },
];

console.log('=== discovery 常量 ===');
console.log('  USER_PRESET_DIR     =', d.USER_PRESET_DIR);
console.log('  SHIPPED_PRESET_ROOT =', d.SHIPPED_PRESET_ROOT);
console.log('  harnessBase         =', HARNESS_BASE);
console.log('  插件目录            =', pluginDir || '(未找到)');
console.log();

console.log('=== 逐根扫描 ===');
for (const root of roots) {
  const found = await d.scanRoot(root, HARNESS_BASE);
  console.log(`[${root.trust}] ${root.path}`);
  if (!found.length) console.log('   (无)');
  for (const p of found) {
    console.log(`   ${p.id.padEnd(22)} trust=${String(p.trust).padEnd(6)} order=${String(p.order ?? '-').padEnd(4)} name=${p.name ?? '(无)'}`);
    console.log(`      健康: ${p.broken === undefined ? 'OK' : 'BROKEN: ' + p.broken}`);
  }
  console.log();
}

const roster = await d.discoverPresets(roots, HARNESS_BASE);
console.log('=== 合并 roster ===');
console.log('共', roster.length, '个:', roster.map((p) => p.id).join(', '));
console.log();

if (!WANT.length) {
  console.log('未从插件源码里提取到 preset 引用，跳过断言。');
  process.exit(0);
}

console.log('=== 断言：插件引用的 preset 是否都在 roster 里且健康 ===');
let ok = true;
for (const id of WANT) {
  const hit = roster.find((p) => p.id === id);
  if (!hit) {
    console.log(`  [FAIL] ${id} —— 不在 roster 中（新建会话会报 preset "${id}" not found）`);
    ok = false;
  } else if (hit.broken !== undefined) {
    console.log(`  [FAIL] ${id} —— 存在但 broken: ${hit.broken}`);
    ok = false;
  } else {
    console.log(`  [PASS] ${id}  →  trust=${hit.trust}  显示名=${hit.name}`);
  }
}
console.log();
console.log(ok
  ? '结论：插件引用的 preset 均可被发现且组成合法。'
  : '结论：仍有问题，见上方 FAIL。补建方法见手册第 4 节。');
process.exit(ok ? 0 : 1);

python make_chengce_presets.py
python make_chengce_presets.py --dsh-home "D:/dsh"

# -*- coding: utf-8 -*-
"""补齐城策插件缺失的两个 agent preset（修 preset "..." not found）。

背景
----
城策工作台插件（dsh-client-ui-urban-planning-agent-workbench@1.1.0）的
src/config/branding.js 硬编码了 preset id：
    up-renewal-policy   ← U-E1 政策咨询推送与解读专家
    up-renewal-research ← S1 海外项目研判决策专家
但发布包 package.json 的 files 字段只有 lib/ src/ skills/ scripts/ ——
**没有任何 presets/**。这两个 preset 从未随包交付。

于是插件在空白会话首次发送时调用
    ctx.remote.agentPresets.select(sessionId, agent.preset)
服务端在 roster 里找不到该 id，抛：
    agent-presets: preset "up-renewal-policy" not found (available: standard, ptc, minimal, cordis)

修法
----
按 dsh 的既定约定（discovery.js 的 USER_PRESET_DIR = '.agent-presets'，根为 <DSH_HOME>/.agent-presets）
在本机补建这两个 preset。组成文件以随包交付、已验证可挂载的 standard preset 为底座，
**只替换 persona 行**，保证工具集完整（shell / fs / 检索 / Skills / 网页 / 子代理 / 计划 / 压缩），
且不引入任何挂载期未知数。

用法
----
    python make_chengce_presets.py
    python make_chengce_presets.py --dsh-home "D:/dsh"

覆盖行为：已存在同名目录会被覆盖（幂等，可重复跑）。
"""

import argparse
import glob
import io
import os
import sys


# ── 两个 preset 的定义 ────────────────────────────────────────────────────────

POLICY_PERSONA = """\
你是「城策」工作台的城市更新政策咨询推送与解读专家，负责政策资讯的采集、去重、\
时效过滤、适用性解读，以及政策周报与重点政策汇报。

工作准则：
- 结论前置，先给判断再给依据，不要用铺垫开场。
- 每条政策结论都必须给出信息来源：发文机关、文号、发布日期、适用对象、有效期。缺哪项就写明缺哪项。
- 区分「已核实」与「待核实」。无法确认的字段标注 [待核实]，不要补全成看似确定的表述。
- 时效优先：先确认政策是否现行有效，再判断适用性；已废止、已到期或被新文件替代的必须标注。
- 去重以「发文机关 + 文号」为主键；同一政策的多篇转载合并为一条，保留最权威的来源。
- 涉及金额、比例、期限、口径等关键内容时保留原文表述，不做换算、不做推断。
- 可调用 u-policy 技能，以及知识库检索与网页检索工具。引用知识库内容时给出库名与文档标识。
- 交付物默认使用简体中文与 Markdown；表格优先于长段落。"""

RESEARCH_PERSONA = """\
你是「城策」工作台的海外项目前期研判决策专家（S1），是项目前期第一道决策闸门。

工作准则：
- 结论前置：先给出「是否跟进」的明确判断，再给证据与不确定性。
- 六个研判面向必须逐个交代：机会、准入、市场、风险、价值、推进建议。缺证据的面向显式记为缺口，不臆测。
- 关键事实必须可追溯：给出信息来源与出处；多源冲突时并列呈现，并说明取舍理由。
- 严格区分「事实」「推断」「假设」三类内容，分别标注，不要混写成一种语气。
- 高风险项给出触发条件与应对建议，不做无依据的乐观陈述。
- 可调用 o-s1 系列技能（机会战略、宏观可行性、市场、风险、价值、推进建议）以及知识库检索与网页检索工具。引用知识库内容时给出库名与文档标识。
- 你负责备齐证据与草案，最终结论由主管或领导确认；请明确标出需要人工决策的节点。
- 交付物默认使用简体中文与 Markdown；表格优先于长段落。"""

PRESETS = [
    {
        'id': 'up-renewal-policy',
        'name': '城策 · 政策咨询与解读',
        'description': '面向城市更新的政策资讯采集、去重、时效过滤与适用性解读。'
                       '产出带来源的政策卡、政策周报与重点政策汇报。',
        'order': 10,
        'persona': POLICY_PERSONA,
        'banner': '城市更新政策咨询推送与解读专家（U-E1）',
    },
    {
        'id': 'up-renewal-research',
        'name': '城策 · 海外项目研判',
        'description': '海外项目前期机会、准入、市场、风险与价值研判，'
                       '输出是否跟进结论与 S2—S6 阶段路由建议。',
        'order': 11,
        'persona': RESEARCH_PERSONA,
        'banner': '海外项目前期研判决策专家（S1）',
    },
]

HEADER = """\
# 城策工作台的 agent preset：{banner}
#
# 本文件由脚本补建，因为城策插件（dsh-client-ui-urban-planning-agent-workbench@1.1.0）
# 硬编码了 preset id '{pid}'，但发布包未携带该 preset —— 装机后选择专家并在空白
# 会话发送消息，会报 agent-presets: preset "{pid}" not found。
#
# 组成与随包的 standard preset 等价（仅 persona 行不同），因此工具集完整：
# shell、文件读写与检索、Skills、网页检索、待办、提问、子代理、工作流、计划模式、上下文压缩。
#
# 位置：<dshHome>/.agent-presets/{pid}  —— 目录名即 preset id。
# 本 preset 无需迁移数据；正在运行的会话停留在它启动时的组成上，新建会话才采用新版本。
"""

PERSONA_BLOCK = """\
- id: persona
  name: '@deepseek-ai/dsh-persona'
  config:
    suffix: 当前工作目录：{{{{cwd}}}}。
    prefix: |-
{prefix}

"""


def indent(text, spaces):
    pad = ' ' * spaces
    return '\n'.join(pad + line if line.strip() else '' for line in text.split('\n'))


def find_standard(dsh_home):
    """定位随包交付的 standard preset 组成文件。"""
    candidates = [
        os.path.join(dsh_home, 'profiles', 'node_modules', '@deepseek-ai',
                     'dsh-agent-presets', 'presets', 'standard', 'agent.cordis.yml'),
        os.path.join(dsh_home, 'profiles', 'web', 'node_modules', '@deepseek-ai',
                     'dsh-agent-presets', 'presets', 'standard', 'agent.cordis.yml'),
    ]
    candidates += glob.glob(os.path.join(
        dsh_home, 'profiles', '*', 'node_modules', '@deepseek-ai',
        'dsh-agent-presets', 'presets', 'standard', 'agent.cordis.yml'))
    for path in candidates:
        if os.path.isfile(path):
            return path
    return None


def build_composition(preset, standard_text):
    """以 standard 为底座，替换 persona 行。"""
    anchor = '- id: agent-instructions'
    idx = standard_text.index(anchor)          # 这一行之后全部照抄（含 agent-instructions 等）
    new_header = HEADER.format(banner=preset['banner'], pid=preset['id'])
    persona = PERSONA_BLOCK.format(prefix=indent(preset['persona'], 6))
    return new_header + persona + standard_text[idx:]


def main():
    ap = argparse.ArgumentParser(description='补齐城策插件缺失的两个 agent preset')
    ap.add_argument('--dsh-home', default=os.environ.get('DSH_HOME') or os.path.join(os.path.expanduser('~'), '.dsh'))
    args = ap.parse_args()
    dsh_home = args.dsh_home

    standard_path = find_standard(dsh_home)
    if not standard_path:
        print('找不到随包的 standard preset 组成文件，请确认 --dsh-home 是否正确：%s' % dsh_home)
        print('可找的位置：<DSH_HOME>/profiles[/<profile>]/node_modules/@deepseek-ai/dsh-agent-presets/presets/standard/agent.cordis.yml')
        return 1
    print('底座组成：%s' % standard_path)

    standard_text = io.open(standard_path, encoding='utf-8').read()
    user_root = os.path.join(dsh_home, '.agent-presets')
    os.makedirs(user_root, exist_ok=True)
    print('用户 preset 根目录：%s' % user_root)
    print()

    for preset in PRESETS:
        target = os.path.join(user_root, preset['id'])
        print('[%s] %s' % ('覆盖' if os.path.isdir(target) else '新建', target))
        os.makedirs(target, exist_ok=True)

        # preset.yml —— 只放展示元数据；没有 id / trust 字段（那两个由目录名与所在根决定）
        meta = 'name: %s\ndescription: %s\norder: %d\n' % (
            preset['name'], preset['description'], preset['order'])
        with io.open(os.path.join(target, 'preset.yml'), 'w', encoding='utf-8', newline='\n') as fh:
            fh.write(meta)

        comp = build_composition(preset, standard_text)
        with io.open(os.path.join(target, 'agent.cordis.yml'), 'w', encoding='utf-8', newline='\n') as fh:
            fh.write(comp)

        print('    preset.yml        %6d bytes' % os.path.getsize(os.path.join(target, 'preset.yml')))
        print('    agent.cordis.yml  %6d bytes  (%d 行)'
              % (os.path.getsize(os.path.join(target, 'agent.cordis.yml')), comp.count('\n') + 1))
        print()

    print('完成。dsh-agent-presets 的 discovery 不做记忆化，会重扫根目录，')
    print('所以**不需要重启 dsh web** —— 刷新页面即可生效。')
    print('权威校验：node verify_presets.mjs')
    return 0


if __name__ == '__main__':
    sys.exit(main())

python win-proc-tree.py                      # 默认查 1933 与 3080
python win-proc-tree.py --list-sandbox       # 列出沙箱根进程及树规模
python win-proc-tree.py --explain-heartbeat  # 心跳对照实验的做法与实测结果

# -*- coding: utf-8 -*-
"""win-proc-tree.py —— Windows 进程树 / 沙箱归属诊断（无外部依赖，只用 ctypes）

## 解决什么问题

本机的 WorkBuddy 会话用 **Windows Job Object** 管理命令的子进程：一条命令结束时
整棵进程树被回收。被回收的服务表现为「刚才还通，下一条命令就 10061 / 连接被拒」。
所以日志里那个常驻服务，到底是能活、还是会掉，**只看端口通不通分不出来**。

## 判据（按可靠性排序）

1. **主判据：是否属于沙箱根的进程树**。
   沙箱根 = `sandbox-cli.exe` 和命令行含 `--serve` 的 `WorkBuddy.exe`。
   目标进程是它们的后代 → 会被回收；不是 → 独立。

2. **佐证：父链顶端**。顶端是 `explorer.exe` / `services.exe` → 独立。
   ⚠️ 父链会断：用 `explorer.exe` 转发启动时，中间那个临时 shell 宿主进程
   （`explorer.exe /factory,{...} -Embedding`）会很快退出，之后父链就查不到了。
   所以**父链断裂不等于有问题**，要以上面第 1 条为准。

3. **⚠️ 不要用 `IsProcessInJob(h, NULL, &r)`**。实测在本机语义不可靠：
   对 `explorer.exe` 启动、父链已证明独立的进程，它仍报 `True`（与父链结论矛盾）。

4. **最终的硬证据是心跳对照实验**（见 `--explain-heartbeat`）：
   同一脚本分别用两种方式启动，跨工具调用看谁的日志还在增长。

## 用法

    python win-proc-tree.py                      # 默认查 1933 与 3080
    python win-proc-tree.py --port 1933
    python win-proc-tree.py --port 3000,8080
    python win-proc-tree.py --pid 12345          # 直接查指定 PID
    python win-proc-tree.py --list-sandbox       # 列出沙箱根进程
    python win-proc-tree.py --explain-heartbeat  # 打印心跳对照实验步骤
"""
import argparse
import ctypes
import ctypes.wintypes as w
import datetime
import subprocess
import sys

try:
    sys.stdout.reconfigure(errors='replace')
except Exception:
    pass

k32 = ctypes.WinDLL('kernel32', use_last_error=True)
ntdll = ctypes.WinDLL('ntdll')

TH32CS_SNAPPROCESS = 0x00000002
PROCESS_QUERY_LIMITED_INFORMATION = 0x1000
ProcessCommandLineInformation = 60

SHELL_ROOTS = ('explorer.exe', 'services.exe', 'wininit.exe', 'winlogon.exe')
SANDBOX_EXE = ('sandbox-cli.exe',)


class PROCESSENTRY32W(ctypes.Structure):
    _fields_ = [
        ('dwSize', w.DWORD), ('cntUsage', w.DWORD), ('th32ProcessID', w.DWORD),
        ('th32DefaultHeapID', ctypes.POINTER(ctypes.c_ulong)), ('th32ModuleID', w.DWORD),
        ('cntThreads', w.DWORD), ('th32ParentProcessID', w.DWORD),
        ('pcPriClassBase', ctypes.c_long), ('dwFlags', w.DWORD),
        ('szExeFile', w.WCHAR * 260),
    ]


def snapshot():
    """{pid: (ppid, exe_name)}"""
    snap = k32.CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS, 0)
    if snap == -1:
        return {}
    e = PROCESSENTRY32W()
    e.dwSize = ctypes.sizeof(PROCESSENTRY32W)
    out = {}
    ok = k32.Process32FirstW(snap, ctypes.byref(e))
    while ok:
        out[int(e.th32ProcessID)] = (int(e.th32ParentProcessID), e.szExeFile)
        ok = k32.Process32NextW(snap, ctypes.byref(e))
    k32.CloseHandle(snap)
    return out


def proc_info(pid):
    """返回 (cmdline, create_time)；取不到则为 None。"""
    h = k32.OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, False, pid)
    if not h:
        return None, None
    try:
        cmd = None
        need = w.ULONG(0)
        ntdll.NtQueryInformationProcess(h, ProcessCommandLineInformation, None, 0,
                                        ctypes.byref(need))
        if need.value:
            buf = ctypes.create_string_buffer(need.value)
            if ntdll.NtQueryInformationProcess(h, ProcessCommandLineInformation, buf,
                                               need.value, ctypes.byref(need)) == 0:
                length = ctypes.cast(buf, ctypes.POINTER(ctypes.c_ushort))[0]
                # 64 位下 UNICODE_STRING 的 Buffer 指针在偏移 8
                ptr = ctypes.cast(ctypes.addressof(buf) + 8,
                                  ctypes.POINTER(ctypes.c_void_p))[0]
                if ptr and length:
                    cmd = ctypes.wstring_at(ptr, length // 2)

        created = None
        tc, te, tk, tu = (w.FILETIME(), w.FILETIME(), w.FILETIME(), w.FILETIME())
        if k32.GetProcessTimes(h, ctypes.byref(tc), ctypes.byref(te),
                               ctypes.byref(tk), ctypes.byref(tu)):
            v = (tc.dwHighDateTime << 32) | tc.dwLowDateTime
            if v:
                created = datetime.datetime.fromtimestamp(v / 10_000_000 - 11644473600)
        return cmd, created
    finally:
        k32.CloseHandle(h)


def clean(s):
    return ' '.join(s.split()) if s else ''


def sandbox_roots(snap):
    """沙箱根：sandbox-cli.exe，以及命令行含 --serve 的 WorkBuddy.exe"""
    roots = {}
    for pid, (_pp, name) in snap.items():
        n = (name or '').lower()
        if n in SANDBOX_EXE:
            roots[pid] = n
        elif n == 'workbuddy.exe':
            cmd, _ = proc_info(pid)
            if cmd and '--serve' in cmd:
                roots[pid] = n
    return roots


def descendants(snap, root):
    children = {}
    for pid, (ppid, _n) in snap.items():
        children.setdefault(ppid, []).append(pid)
    seen, stack = set(), [root]
    while stack:
        cur = stack.pop()
        for c in children.get(cur, []):
            if c not in seen:
                seen.add(c)
                stack.append(c)
    return seen


def sandbox_tree(snap):
    """所有沙箱根及其后代 PID 的并集"""
    tree = set()
    for root in sandbox_roots(snap):
        tree |= descendants(snap, root) | {root}
    return tree


def listening_pids(port):
    out = subprocess.run(['netstat', '-ano'], capture_output=True, text=True,
                         encoding='utf-8', errors='replace').stdout
    pids = set()
    for line in out.splitlines():
        if ':%d ' % port in line and 'LISTENING' in line.upper():
            try:
                pids.add(int(line.split()[-1]))
            except ValueError:
                pass
    return pids


def trace_up(snap, pid, maxdepth=12):
    chain, cur, seen = [], pid, set()
    for _ in range(maxdepth):
        if not cur or cur in seen:
            break
        seen.add(cur)
        ppid, name = snap.get(cur, (None, None))
        if name is None:
            cmd, _ = proc_info(cur)
            chain.append((cur, '(已退出/取不到)', clean(cmd)))
            break
        cmd, _ = proc_info(cur)
        chain.append((cur, name, clean(cmd)))
        if name.lower() in SHELL_ROOTS:
            break
        cur = ppid
    return chain


def report(snap, tree, pid, now):
    cmd, ct = proc_info(pid)
    name = snap.get(pid, (None, '(快照里没有)'))[1]
    age = ''
    if ct:
        age = '启动于 %s（%.0f 分钟前）' % (ct.strftime('%m-%d %H:%M:%S'),
                                          (now - ct).total_seconds() / 60)
    print('  ▶ PID %-7d %-24s %s' % (pid, name, age))
    print('     cmd : %s' % (clean(cmd)[:180] or '(取不到)'))

    chain = trace_up(snap, pid)
    print('     父链:')
    for i, (p, n, c) in enumerate(chain):
        if i == 0:
            print('       [%d] %s   ← 本进程' % (p, n))
        else:
            print('       %s[%d] %s' % ('  ' * i, p, n))
            if c:
                print('       %s    %s' % ('  ' * i, c[:160]))

    top = (chain[-1][1].lower() if chain else '')
    chain_broken = bool(chain) and chain[-1][1].startswith('(已退出')

    in_tree = pid in tree
    if in_tree:
        print('     判定: ⚠ 属于沙箱进程树 —— 停后台任务 / 会话结束就会被回收')
    else:
        print('     判定: ✅ 不在沙箱进程树内 —— 可放心清理后台任务')
        if top in SHELL_ROOTS:
            print('           佐证: 父链顶端是 %s' % chain[-1][1])
        elif chain_broken:
            print('           注: 父链已断（启动它的临时宿主已退出），这不代表有问题；')
            print('               若需硬证据，跑 --explain-heartbeat 的心跳对照实验。')
    print()


def main():
    ap = argparse.ArgumentParser(description='Windows 进程树 / 沙箱归属诊断')
    ap.add_argument('--port', default='1933,3080',
                    help='要查的端口，逗号分隔（默认 1933,3080）')
    ap.add_argument('--pid', type=int, action='append', default=None,
                    help='直接查指定 PID（可重复）')
    ap.add_argument('--list-sandbox', action='store_true', help='列出沙箱根进程')
    ap.add_argument('--explain-heartbeat', action='store_true',
                    help='打印心跳对照实验步骤')
    args = ap.parse_args()

    if args.explain_heartbeat:
        print('心跳对照实验（验证某进程能否脱离沙箱）')
        print('=' * 76)
        print('1. 写两个内容相同的 .cmd，都往自己的日志里循环追加时间戳：')
        print('     @echo off')
        print('     :loop')
        print('     echo %date% %time% alive >> "%~dp0_heartbeat.log"')
        print('     ping -n 3 127.0.0.1 >nul')
        print('     goto loop')
        print('2. A 组用 explorer 启动，B 组在沙箱内用 cmd /c start 启动：')
        print('     subprocess.Popen([r"C:\\Windows\\explorer.exe", HB])')
        print('     subprocess.Popen(["cmd","/c","start","","/min", CT])')
        print('3. 本次命令内先读一次基线行数；**下一次工具调用**再读一次。')
        print('4. 结论：日志停止增长的那组被回收了。')
        print()
        print('本机实测结果（2026-09-15，跨两次工具调用读数）：')
        print('  A explorer 启动  : 5 行 -> 29 行，仍在写    ✅ 存活')
        print('  B 沙箱内启动     : 5 行 -> 5 行，时间戳停住  ❌ 被回收')
        print('=' * 76)
        return

    snap = snapshot()
    now = datetime.datetime.now()

    if args.list_sandbox:
        roots = sandbox_roots(snap)
        tree = sandbox_tree(snap)
        print('=' * 84)
        print('沙箱根进程（这些进程的后代会被回收）')
        print('=' * 84)
        if not roots:
            print('  (没找到 sandbox-cli.exe / WorkBuddy.exe --serve)')
        for pid, name in sorted(roots.items()):
            cmd, ct = proc_info(pid)
            n = len(descendants(snap, pid))
            print('  PID %-7d %-22s 后代 %d 个' % (pid, name, n))
            print('     %s' % clean(cmd)[:160])
        print()
        print('  沙箱树总规模: %d 个进程' % len(tree))
        print()

    tree = sandbox_tree(snap)

    if args.pid:
        print('=' * 84)
        print('指定 PID')
        print('=' * 84)
        for pid in args.pid:
            report(snap, tree, pid, now)

    ports = [int(p.strip()) for p in str(args.port).split(',') if p.strip().isdigit()]
    for port in ports:
        print('=' * 84)
        print('端口 %d' % port)
        print('=' * 84)
        pids = listening_pids(port)
        if not pids:
            print('  (无人监听)')
            print()
            continue
        for pid in sorted(pids):
            report(snap, tree, pid, now)


if __name__ == '__main__':
    main()

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""OpenViking 数据目录锁（.openviking.pid）陈旧锁（stale lock）检测与清理。

## 为什么需要这个脚本

OpenViking 用「数据目录下的 `.openviking.pid`」做进程级互斥，防止多个实例同时
写同一份 AGFS / VectorDB 造成静默的数据损坏。逻辑在
`openviking/utils/process_lock.py` 的 `acquire_data_dir_lock()`。

问题出在 `_is_pid_alive(pid)` 的**平台不对称**上：

  - Linux 分支：拿到 PID 后还会读 `/proc/<pid>/cmdline`，校验它**是不是真的 OpenViking**。
    源码注释里写明了这是为了修 PID 复用导致的误判（upstream issue #1088）。
  - **Windows 分支：只判断「这个 PID 有没有进程在」，不做任何身份校验。**

而 Windows 同样会复用 PID。于是在 Windows 上这条路径必然踩坑：

  1. 非正常退出（关机 / 任务管理器强杀 / 断电）→ `atexit` 注册的清理回调没跑到 → 锁文件残留；
  2. 系统重启后，某个**毫不相干**的进程拿到了同一个 PID；
  3. 下次启动 OpenViking → 读到锁文件里的 PID → `_is_pid_alive()` 返回 True（确实有这个进程）
     → 判定「已有实例在跑」→ 直接抛 `DataDirectoryLocked`；
  4. 表现：`openviking-server` 退出码 **3**，报的 PID 指向一个跟 OpenViking 毫无关系的进程
     （实测踩到过迅雷的 `XLSmartService.exe`），**而 1933 端口其实是空的**。

本脚本用**双重判据**判定陈旧锁，两个判据互相独立，避免误删一个真实持有的锁：

  A. 端口判据：配置里的端口无人监听 → 不存在在服务的实例；
  B. 身份判据：锁文件里的 PID 对应的进程镜像名不是 OpenViking → 它只是个复用者。

只有 **A ∧ B** 同时成立才判为陈旧锁。任一判据显示「可能有真实实例」就不动锁——
宁可让你人工看一眼，也不要冒数据损坏的风险。

## 用法

    <python> ov-stale-lock.py                  # 诊断（只读，默认）
    <python> ov-stale-lock.py --fix            # 诊断后清理陈旧锁（先备份再删）
    <python> ov-stale-lock.py --conf <路径>    # 指定 ov.conf（默认 ~/.openviking/ov.conf）
    <python> ov-stale-lock.py --port <端口>    # 覆盖端口（默认从 ov.conf 读）

退出码：

    0  无锁 / 无问题（--fix 下也可能是"已清理"）
    1  检测到陈旧锁，尚未清理（未加 --fix），或 --fix 下清理失败
    2  疑似有真实实例在跑 —— **不要清锁**，请先确认那个进程
    3  配置或环境有问题（读不到 ov.conf、找不到数据目录等）

零依赖：只用标准库（`socket` / `subprocess` / `shutil`），不 import psutil。
"""

from __future__ import annotations

import argparse
import json
import os
import shutil
import socket
import subprocess
import sys
import time

LOCK_FILENAME = ".openviking.pid"

# 认定为「OpenViking 自己的进程」的镜像名（小写比对）
OV_IMAGE_NAMES = {
    "openviking-server.exe",
    "openviking.exe",
    "openviking",
    "openviking-server",
}


def emit(line: str = "") -> None:
    """往 stdout 写一行。

    编码策略（踩过坑，别改）：
      - 直接跑在 cmd/PowerShell 窗口里（isatty）→ 用控制台自己的编码（中文 Windows
        是 cp936），中文正常显示；
      - 被重定向到文件 / 管道（非 isatty）→ 一律写 **UTF-8**。

    早先版本无条件走 `sys.stdout.write()`，于是重定向后写成 cp936 字节，而
    PowerShell 的 `>` 又按 UTF-16LE 落盘，最终日志是「GBK 套 UTF-16」的双重乱码，
    根本读不出来。→ 重定向时统一 UTF-8 就干净了。
    """
    data = line + "\n"
    try:
        is_tty = bool(getattr(sys.stdout, "isatty", lambda: False)())
    except Exception:
        is_tty = False

    buf = getattr(sys.stdout, "buffer", None)
    if buf is not None:
        enc = (getattr(sys.stdout, "encoding", None) or "utf-8") if is_tty else "utf-8"
        try:
            buf.write(data.encode(enc, "replace"))
            buf.flush()
            return
        except Exception:
            pass
    try:
        sys.stdout.write(data)
        sys.stdout.flush()
    except (UnicodeEncodeError, UnicodeError):
        sys.stdout.write(data.encode("ascii", "replace").decode("ascii"))
        sys.stdout.flush()


def default_conf_path() -> str:
    return os.path.join(os.path.expanduser("~"), ".openviking", "ov.conf")


def load_conf(conf_path: str) -> dict:
    with open(conf_path, encoding="utf-8") as f:
        return json.load(f)


def port_is_listening(host: str, port: int, timeout: float = 1.5) -> bool:
    """TCP connect 探测。比 netstat 解析可靠，不依赖外部命令与代码页。"""
    for family, addr in (
        (socket.AF_INET, (host, port)),
    ):
        try:
            s = socket.socket(family, socket.SOCK_STREAM)
            s.settimeout(timeout)
            try:
                if s.connect_ex(addr) == 0:
                    return True
            finally:
                s.close()
        except OSError:
            continue
    return False


def pid_image_name(pid: int) -> str | None:
    """取 PID 对应的进程镜像名；进程不存在返回 None。

    用 `tasklist`（随 Windows 自带），不引入 psutil。输出形如：
        "XLSmartService.exe","6628","Services","0","7,000 K"
    """
    if pid <= 0:
        return None
    try:
        out = subprocess.run(
            ["tasklist", "/FI", f"PID eq {pid}", "/FO", "CSV", "/NH"],
            capture_output=True,
            text=True,
            errors="replace",
            timeout=20,
        ).stdout.strip()
    except Exception:
        # tasklist 不可用时不臆断，交给调用方按"未知"处理
        return None

    if not out or "No tasks are running" in out or "没有运行的任务" in out:
        return None
    first = out.splitlines()[0].strip()
    if not first.startswith('"'):
        return None
    return first.split('"')[1] or None


def read_lock_pid(lock_path: str) -> int:
    try:
        with open(lock_path, encoding="utf-8", errors="replace") as f:
            return int(f.read().strip() or 0)
    except (OSError, ValueError):
        return 0


def process_cmdline_has_openviking(pid: int) -> bool:
    """兜底身份判据：进程镜像是 python.exe 时，看命令行里有没有 openviking。

    正常安装下入口是 `openviking-server.exe`；但有人用
    `python -m openviking.server` 之类方式起，也会持有同一个锁。
    """
    try:
        ps = (
            "Get-CimInstance Win32_Process -Filter \"ProcessId=%d\" | "
            "Select-Object -ExpandProperty CommandLine" % pid
        )
        r = subprocess.run(
            ["powershell", "-NoProfile", "-NonInteractive", "-Command", ps],
            capture_output=True,
            text=True,
            errors="replace",
            timeout=25,
        )
        return "openviking" in (r.stdout or "").lower()
    except Exception:
        return False


def main() -> int:
    ap = argparse.ArgumentParser(
        description="检测（并可清理）OpenViking 数据目录的陈旧 PID 锁",
    )
    ap.add_argument("--conf", default=None, help="ov.conf 路径（默认 ~/.openviking/ov.conf）")
    ap.add_argument("--port", type=int, default=None, help="覆盖端口（默认从 ov.conf 读）")
    ap.add_argument("--fix", action="store_true", help="清理陈旧锁（先备份为 .stale-N 再删）")
    ap.add_argument(
        "--check",
        action="store_true",
        help="仅诊断、不改动文件（这是默认行为，显式写出来只为意图清楚）",
    )
    args = ap.parse_args()

    conf_path = args.conf or default_conf_path()
    emit("=" * 68)
    emit("OpenViking 陈旧数据目录锁检测")
    emit("=" * 68)

    if not os.path.isfile(conf_path):
        emit(f"[FATAL] 找不到配置文件: {conf_path}")
        emit("        用 --conf 指定，或确认 OpenViking 是否装在本机。")
        return 3

    try:
        conf = load_conf(conf_path)
    except Exception as e:
        emit(f"[FATAL] 解析 ov.conf 失败: {e}")
        return 3

    server = conf.get("server") or {}
    storage = conf.get("storage") or {}
    host = server.get("host") or "127.0.0.1"
    port = args.port or int(server.get("port") or 1933)
    workspace = storage.get("workspace")

    if not workspace:
        emit("[FATAL] ov.conf 里没有 storage.workspace，无法定位数据目录。")
        return 3

    workspace = os.path.expanduser(str(workspace))
    lock_path = os.path.join(workspace, LOCK_FILENAME)

    emit(f"  配置      : {conf_path}")
    emit(f"  数据目录  : {workspace}")
    emit(f"  服务端点  : {host}:{port}")
    emit(f"  锁文件    : {lock_path}")
    emit("")

    if not os.path.isdir(workspace):
        emit("[FATAL] 数据目录不存在。OpenViking 还没初始化过？")
        return 3

    # ---------- 判据 A：端口是否有人监听 ----------
    listening = port_is_listening(host, port)
    emit(f"[判据 A · 端口] {host}:{port} 监听中 = {listening}")

    if not os.path.isfile(lock_path):
        emit(f"[判据 B · 锁文件] 不存在")
        emit("")
        if listening:
            emit("[结论] 端口有服务在跑，但没有锁文件 —— 锁可能被外部删过。")
            emit("       服务本身可用；如需重启，直接停掉它再起即可。")
        else:
            emit("[结论] 干净状态：无锁、端口空闲，可直接启动。")
        return 0

    # ---------- 判据 B：锁文件里的 PID 到底是谁 ----------
    pid = read_lock_pid(lock_path)
    mtime = None
    try:
        mtime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(os.path.getmtime(lock_path)))
    except OSError:
        pass

    image = pid_image_name(pid) if pid > 0 else None
    is_ov_process = bool(image) and image.lower() in OV_IMAGE_NAMES
    # 镜像是 python 且命令行带 openviking，也算真实持有者
    if image and not is_ov_process and "python" in image.lower():
        if process_cmdline_has_openviking(pid):
            is_ov_process = True

    emit(f"[判据 B · 锁文件] PID={pid}  写入时间={mtime}")
    emit(f"                 该 PID 当前镜像 = {image or '(无此进程)'}")
    emit(f"                 是否 OpenViking = {is_ov_process}")
    emit("")

    # ---------- 判定 ----------
    if listening and is_ov_process:
        emit("[结论] 正常持有中：有 OpenViking 在监听，且锁的持有者确实是它。")
        emit("       无需任何操作。")
        return 2

    if listening and not is_ov_process:
        emit("[结论] 可疑：端口有服务在监听，但锁文件记的 PID 不是 OpenViking 进程。")
        emit(f"       可能这个锁是陈旧的，而当前实例是后来起来的（或改了配置）。")
        emit(f"       请先确认 {host}:{port} 上跑的是什么，再决定是否清理。**本脚本不动它。**")
        return 2

    if not listening and is_ov_process:
        emit("[结论] 可疑：端口没人监听，但锁文件记的 PID 确实是 OpenViking 进程。")
        emit(f"       它可能是卡在启动/关闭阶段。**出于安全，本脚本不动锁。**")
        emit(f"       请先看 PID {pid} 的状态，必要时手动结束它再重试。")
        return 2

    # not listening and not is_ov_process -> 双重判据都指向陈旧锁
    emit("[结论] 陈旧锁（stale lock）—— 这就是 OpenViking 起不来、报")
    emit(f"       DataDirectoryLocked: Another OpenViking process (PID {pid}) 的原因。")
    emit("")
    emit(f"       端口 {port} 无人监听 ⇒ 不存在在服务的实例；")
    emit(f"       PID {pid} 的真实身份是「{image}」⇒ 与 OpenViking 无关，只是复用了这个 PID。")
    emit("")

    if not args.fix:
        emit("[处置] 仅诊断模式，未改动任何文件。确认上面两条后，加 --fix 清理：")
        emit("         <python> ov-stale-lock.py --fix")
        return 1

    # 备份（留排障痕迹）后删除
    backup = f"{lock_path}.stale-{time.strftime('%Y%m%d-%H%M%S')}"
    try:
        shutil.copy2(lock_path, backup)
        emit(f"[处置] 已备份旧锁 → {backup}")
    except OSError as e:
        emit(f"[WARN] 备份失败（继续，旧锁仅含一个 PID，无数据价值）: {e}")

    try:
        os.remove(lock_path)
        emit(f"[处置] 已删除陈旧锁 → {lock_path}")
    except OSError as e:
        emit(f"[FAIL] 删除失败: {e}")
        return 1

    emit("")
    emit("[完成] 现在可以正常启动 OpenViking 了（双击 dsh 图标会自动拉起）。")
    return 0


if __name__ == "__main__":
    sys.exit(main())