diff --git a/.gitignore b/.gitignore index 0957dbf..1f2cb3b 100644 --- a/.gitignore +++ b/.gitignore @@ -7,6 +7,13 @@ __pycache__/ *.bak-* skills-backup/ +# skill 扫盘注册表(本机运行数据:名字 -> 位置,含绝对路径,不入库) +skill.json +skill.json.tmp + +# 单测临时输出 +_*.txt + # 编辑器与系统 .vscode/ .idea/ diff --git a/README.md b/README.md index 0e6532b..68e6d04 100644 --- a/README.md +++ b/README.md @@ -219,11 +219,38 @@ UNC 目录扫描(那会触发冷启动卡死),直接报错返回。 是指向 `.cc-switch/skills/...` 的 junction,选它推送时自动推 cc-switch 真身。 解不出真身时才落回原诊断(「自动切换也没找到真身,无法推送」)。 +**skill.json 注册表:推拉遇空壳/链接的分层扫盘落库(2026-09-25 Round 10)**—— +不一上来就全盘忙扫,按三层走,且**推拉永远优先用户选定的目录**(选定目录有效就直接用, +注册表只在空壳/链接时兜底解析真身): + +| 层 | 时机 | 做什么 | +| --- | --- | --- | +| ① 内存注册表 | 启动即读 | `oc.py` / `ocskill.py` 启动时把 `skill.json` 读进内存(`load_registry()`),后续查询零扫盘 | +| ② 内嵌常见位置快扫 | 空壳/链接、内存没命中 | `builtin_skill_roots()`:oc(`.config/opencode/skills`)、wb(`.workbuddy/skills`)、**cc-switch**、claude code + `agents.md` 终端表 + `OC_SKILL_ALT_ROOTS` + WSL 两根(VM 在跑才带,绝不为快扫冷启动),秒级、命中即落库 | +| ③ 后台扫盘 | 上面都没中 | `start_background_scan()`:daemon 线程扫内嵌根 + 扩展浅层根(home / `.config` / `.local/share` / cwd 下一级的 `skills/`,不递归不全盘),**不阻塞当前推拉**;单飞(运行中不叠加线程);UNC 根 6s 限时防半死;扫完落库 | + +落库结构就是「名字 -> 位置」的一堆键值(`skills. = [路径…]`,只增不删, +查询端做存在性 + `SKILL.md` 校验,过期条目只是查不到、不误导): + +```json +{ "version": 1, "updated": "2026-09-25 21:10:00", + "skills": { "java-expert": ["C:\\Users\\…\\.workbuddy\\skills\\java-expert", "…"] } } +``` + +- 文件位置:本目录 `skill.json`(**本机运行数据,已 gitignore,不入库**); +- 推送(菜单 `_flow_push` + CLI `cmd_push`)、拉取(`cmd_pull`)**两端对称**接入: + 源是空壳/链接就走上面三层自动切真身并打印「原 / 新」,没找到真身则启动后台扫盘 + 并提示「稍后重试即可自动切」;zip / SKILL.md / base64 输入不参与切换; +- 扫描结果落库后,下次同名直接走 ① 内存命中(实测首次 0.49s、重启后 0.02s)。 + **回归单测**:`python _test_diag.py`(空壳 / 正常 skill / junction 链接 / 断链 / 大小写 / WSL UNC 真链接 / 跨根兜底 / real_src 自动切换等 20+ 项断言,2 个文件 symlink 场景在拦截 symlink 的 机器上自动 SKIP;含 WSL 场景,首次跑会拉起 VM); `python _test_mcpctl.py`(热重载前置探活 + 探活措辞 10 项:未运行 skip 不探测、 -失败仍报错、host 复用、10061/10060 措辞、skip 整句不含「连不上」等)。 +失败仍报错、host 复用、10061/10060 措辞、skip 整句不含「连不上」等); +`python _test_registry.py`(skill.json 注册表 34 项:内存加载 / 落库原子写 / 损坏文件 +不炸 / 内嵌 oc·wb·cc-switch·claude 根 / 快扫命中即落库 / 分层查找顺序 / 后台扫盘单飞 +与完成落库 / 点前缀与无 SKILL.md 不登记 / 用户选定目录优先 / 扩展根浅层扫描)。 --- diff --git a/_test_diag.py b/_test_diag.py index dd29863..62459fe 100644 --- a/_test_diag.py +++ b/_test_diag.py @@ -11,6 +11,7 @@ from ocskill import ( # noqa: E402 diag_skill_md, skill_brief, probe_push, find_skill_md, _wsl_running, _alt_skill_md, _alt_skill_dirs, real_src, ) +import ocskill # noqa: E402 FAIL: list[str] = [] @@ -24,6 +25,12 @@ def check(name: str, cond: bool, extra: str = "") -> None: tmp = Path(tempfile.mkdtemp(prefix="ocskill_diag_")).resolve() # resolve 掉 8.3 短路径,否则 symlink 视角解析失败 print("tmp =", tmp) +# 注册表隔离(2026-09-25 Round 10):skill.json 指到本轮临时文件并清空内存, +# 否则上一轮测试落库的临时真身会被 real_src 的 tier① 内存命中,跨轮串味假失败。 +_orig_registry = (ocskill.REGISTRY_PATH, ocskill._REG) +ocskill.REGISTRY_PATH = tmp / "skill.json" +ocskill._REG = None + # ---------- 1. 空壳目录 ---------- empty = tmp / "empty-skill" empty.mkdir() diff --git a/_test_registry.py b/_test_registry.py new file mode 100644 index 0000000..27a0829 --- /dev/null +++ b/_test_registry.py @@ -0,0 +1,223 @@ +# -*- coding: utf-8 -*- +"""skill.json 注册表单测:内存加载 / 落库 / 内嵌常见根 / 分层查找 / 后台扫盘。 + +覆盖 2026-09-25 需求:推拉遇软链接空壳 -> 先查内存注册表 -> 再快扫内嵌常见 +终端位置(oc/wb/cc-switch…)-> 都没中才后台扫盘(不阻塞),扫完落库 skill.json +(名字 -> 位置 的一堆键值);用户选定的有效目录永远优先(real_src 返回 None)。 +""" +import json +import os +import sys +import tempfile +import time +from pathlib import Path + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import ocskill # noqa: E402 + +FAIL: list[str] = [] + + +def check(name: str, cond: bool, extra: str = "") -> None: + print(("PASS " if cond else "FAIL ") + name + (" | " + extra if extra else "")) + if not cond: + FAIL.append(name) + + +tmp = Path(tempfile.mkdtemp(prefix="ocskill_reg_")).resolve() +print("tmp =", tmp) + +_orig = { # 测完还原,别污染同进程后续用例 + "REGISTRY_PATH": ocskill.REGISTRY_PATH, + "_REG": ocskill._REG, + "env_alt": os.environ.get("OC_SKILL_ALT_ROOTS"), + "builtin": ocskill.builtin_skill_roots, + "deep": ocskill.deep_scan_roots, + "bg": ocskill.start_background_scan, +} + + +def reset_registry(path: Path) -> None: + ocskill.REGISTRY_PATH = path + ocskill._REG = None # 强制下次 load_registry 重读 + + +def mk_skill(folder: Path, name: str) -> Path: + folder.mkdir(parents=True, exist_ok=True) + (folder / "SKILL.md").write_text( + f"---\nname: {name}\ndescription: 测试技能\n---\n# {name}\n", + encoding="utf-8") + return folder + + +try: + # ---------- 1. 空表起步 + remember + 落库 + 启动重读入内存 ---------- + reset_registry(tmp / "skill.json") + reg = ocskill.load_registry() + check("1 空表起步 skills={}", reg.get("skills") == {}, str(reg)) + real_a = mk_skill(tmp / "roots" / "A" / "foo-skill", "foo-skill") + ocskill.remember_skills({"foo-skill": [str(real_a)]}) + check("1 remember 进内存", ocskill.registry_locations("foo-skill") == [str(real_a)], + str(ocskill.registry_locations("foo-skill"))) + check("1 remember 落库文件存在", (tmp / "skill.json").is_file()) + data = json.loads((tmp / "skill.json").read_text(encoding="utf-8")) + check("1 落库结构 = version/updated/skills", + {"version", "updated", "skills"} <= set(data.keys()) + and data["skills"].get("foo-skill") == [str(real_a)], str(data)[:200]) + ocskill._REG = None # 模拟下次启动 + reg2 = ocskill.load_registry() + check("1 启动重读入内存(skill.json -> 内存)", + reg2.get("skills", {}).get("foo-skill") == [str(real_a)], str(reg2)[:200]) + # 重复 remember 不产生重复条目 + ocskill.remember_skills({"foo-skill": [str(real_a)]}) + check("1 remember 幂等(不重复)", + ocskill.registry_locations("foo-skill") == [str(real_a)]) + # 损坏文件不炸:从空表起步 + (tmp / "skill.json").write_text("{broken json", encoding="utf-8") + ocskill._REG = None + reg3 = ocskill.load_registry() + check("1 损坏 skill.json -> 空表起步不炸", reg3.get("skills") == {}, str(reg3)) + + # ---------- 2. 内嵌常见终端根 ---------- + roots = ocskill.builtin_skill_roots() + home = Path.home() + for tag, expect in [ + ("oc", home / ".config" / "opencode" / "skills"), + ("wb", home / ".workbuddy" / "skills"), + ("cc-switch", home / ".cc-switch" / "skills"), + ("claude", home / ".claude" / "skills"), + ]: + key = os.path.normcase(os.path.abspath(str(expect))) + check(f"2 内嵌 {tag} 位置", key in [os.path.normcase(os.path.abspath(r)) for r in roots], + str(expect)) + agent_dirs = [os.path.normcase(os.path.abspath(t["dir"])) + for t in ocskill.load_agent_table() if t.get("dir")] + got = [os.path.normcase(os.path.abspath(r)) for r in roots] + check("2 agents.md 终端目录纳入", all(d in got for d in agent_dirs), + f"agents={agent_dirs}") + check("2 内嵌根无重复", len(got) == len(set(got)), str(len(got))) + check("2 env 追加根", True) # 用例 3 实际验证 + + # ---------- 3. tier1 快扫 scan_roots_for_name + env 追加 ---------- + root_b = tmp / "roots" / "B" + target = mk_skill(root_b / "quick-skill", "quick-skill") + found = ocskill.scan_roots_for_name("quick-skill", roots=[str(root_b)]) + check("3 快扫按名字命中真身", found == [str(target)], str(found)) + check("3 快扫找不到时返回空", ocskill.scan_roots_for_name("nope-xyz", roots=[str(root_b)]) == []) + # env 追加的根也进 builtin + os.environ["OC_SKILL_ALT_ROOTS"] = str(root_b) + roots_env = ocskill.builtin_skill_roots() + check("2 env 追加根进 builtin", + os.path.normcase(os.path.abspath(str(root_b))) in + [os.path.normcase(os.path.abspath(r)) for r in roots_env], str(roots_env)[:200]) + + # ---------- 4. real_src 分层:① 内存注册表命中(零扫盘) ---------- + reset_registry(tmp / "skill2.json") + ocskill.remember_skills({"foo-skill": [str(real_a)]}) + shell1 = tmp / "shell" / "foo-skill" + shell1.mkdir(parents=True) # 空壳:目录在、没有 SKILL.md + bg_called: list[str] = [] + ocskill.start_background_scan = lambda name="": bg_called.append(name) or True + real = ocskill.real_src(str(shell1)) + check("4 空壳 -> 内存注册表一步命中", real == str(real_a), f"real={real}") + check("4 注册表命中不触发后台扫盘", bg_called == [], str(bg_called)) + + # ---------- 5. real_src 分层:② tier1 快扫命中并落库 ---------- + shell2 = tmp / "shell" / "quick-skill" + shell2.mkdir(parents=True) + ocskill._REG = None # 空注册表,逼走 tier1 + real2 = ocskill.real_src(str(shell2)) + check("5 空壳 -> 快扫内嵌常见位置命中", real2 == str(target), f"real={real2}") + check("5 快扫命中即落库", + "quick-skill" in ocskill.load_registry().get("skills", {}), + str(ocskill.load_registry().get("skills", {}))[:200]) + check("5 快扫命中不触发后台扫盘", bg_called == [], str(bg_called)) + + # ---------- 6. real_src 分层:③ 都没中 -> 启动后台扫盘(不阻塞返回 None) ---------- + shell3 = tmp / "shell" / "ghost-skill" + shell3.mkdir(parents=True) + ocskill._REG = None + os.environ["OC_SKILL_ALT_ROOTS"] = str(tmp / "nowhere") # 指到不存在的根 + real3 = ocskill.real_src(str(shell3)) + check("6 没真身 -> 返回 None(走原诊断路径)", real3 is None, str(real3)) + check("6 没真身 -> 启动后台扫盘", bg_called == ["ghost-skill"], str(bg_called)) + + # ---------- 7. 后台扫盘:单飞 + 扫完落库 skill.json ---------- + ocskill.start_background_scan = _orig["bg"] # 还原真身 + root_c = tmp / "roots" / "C" + mk_skill(root_c / "bg-skill-1", "bg-skill-1") + mk_skill(root_c / "bg-skill-2", "bg-skill-2") + (root_c / ".hidden-skill").mkdir() # 点前缀跳过 + (root_c / "no-md-skill").mkdir() # 无 SKILL.md 跳过 + reset_registry(tmp / "skill3.json") + ocskill.builtin_skill_roots = lambda distro=ocskill.DEFAULT_DISTRO: [str(root_c)] + ocskill.deep_scan_roots = lambda: [] + # 放慢扫描以便验证单飞 + _real_scan = ocskill._scan_root_skills + + def _slow_scan(root): + time.sleep(0.4) + return _real_scan(root) + + ocskill._scan_root_skills = _slow_scan + try: + started = ocskill.start_background_scan("x") + check("7 后台扫盘启动成功", started is True, str(started)) + again = ocskill.start_background_scan("y") + check("7 后台扫盘单飞(运行中不叠加线程)", again is False, str(again)) + deadline = time.time() + 10 + while ocskill._BG_SCAN_RUNNING and time.time() < deadline: + time.sleep(0.05) + check("7 后台扫盘完成", not ocskill._BG_SCAN_RUNNING) + finally: + ocskill._scan_root_skills = _real_scan + skills = ocskill.load_registry().get("skills", {}) + check("7 扫盘结果进内存注册表", + "bg-skill-1" in skills and "bg-skill-2" in skills, str(list(skills))) + check("7 点前缀 / 无 SKILL.md 不登记", + "no-md-skill" not in skills and ".hidden-skill" not in skills, str(list(skills))) + data3 = json.loads((tmp / "skill3.json").read_text(encoding="utf-8")) + check("7 扫盘结果落库 skill.json", "bg-skill-1" in data3.get("skills", {}), + str(data3)[:200]) + + # ---------- 8. 用户选定的有效目录优先(real_src 返回 None 不动) ---------- + valid = mk_skill(tmp / "valid-skill", "valid-skill") + ocskill._REG = None + check("8 有效目录不动(选定优先)", ocskill.real_src(str(valid)) is None, str(valid)) + # zip / SKILL.md 文件输入也不动 + mdfile = valid / "SKILL.md" + check("8 SKILL.md 文件不动", ocskill.real_src(str(mdfile)) is None, str(mdfile)) + + # ---------- 9. deep_scan_roots 浅层扫描 ---------- + old_cwd = os.getcwd() + work = tmp / "cwd" + (work / "someapp" / "skills").mkdir(parents=True) + try: + os.chdir(work) + deep = _orig["deep"]() # 用原函数(case7 把它补丁成了 lambda: []) + finally: + os.chdir(old_cwd) + key = os.path.normcase(os.path.abspath(str(work / "someapp" / "skills"))) + check("9 扩展根扫到 /someapp/skills", + key in [os.path.normcase(os.path.abspath(d)) for d in deep], str(deep)) + check("9 扩展根无重复", + len([os.path.normcase(os.path.abspath(d)) for d in deep]) == + len(set(os.path.normcase(os.path.abspath(d)) for d in deep)), str(len(deep))) + +finally: + # ---------- 还原现场 ---------- + ocskill.REGISTRY_PATH = _orig["REGISTRY_PATH"] + ocskill._REG = _orig["_REG"] + ocskill.builtin_skill_roots = _orig["builtin"] + ocskill.deep_scan_roots = _orig["deep"] + ocskill.start_background_scan = _orig["bg"] + if _orig["env_alt"] is None: + os.environ.pop("OC_SKILL_ALT_ROOTS", None) + else: + os.environ["OC_SKILL_ALT_ROOTS"] = _orig["env_alt"] + +print() +if FAIL: + print(f"{len(FAIL)} FAILED: {FAIL}") + sys.exit(1) +print("ALL PASS") diff --git a/oc.py b/oc.py index 10960d2..5add1e0 100644 --- a/oc.py +++ b/oc.py @@ -667,6 +667,10 @@ def dispatch(argv: list[str]) -> int: def main() -> int: fix_console() + try: + ocskill.load_registry() # 启动时把 skill.json 读进内存(空壳真身查询零扫盘) + except Exception: # noqa: BLE001 —— 注册表读失败绝不挡住菜单 + pass if len(sys.argv) > 1: return dispatch(sys.argv[1:]) menu_top() diff --git a/ocskill.py b/ocskill.py index 5ae664f..15a754b 100644 --- a/ocskill.py +++ b/ocskill.py @@ -25,6 +25,8 @@ ocskill —— 打通本机与 WSL OpenCode 的 skill 互通 import argparse import base64 import binascii +import glob +import json import os import queue import re @@ -673,6 +675,293 @@ def find_skill_md(folder: Path, depth: int = 0) -> Path | None: return None +# ---------------------------------------------------------------- skill 注册表(skill.json,2026-09-25) +# 用户要求的推拉兜底流程: +# 遇软链接/空壳 -> 不一上来就全盘忙扫:① 先查内存注册表(启动时已把 skill.json +# 读进内存)-> ② 再快扫内嵌常见终端的 skill 位置(oc / wb / cc-switch / claude …) +# -> ③ 都没命中才后台扫盘(不阻塞当前推拉),扫完把结果落库 skill.json。 +# skill.json 按「名字 -> 位置」的一堆键值存储;推拉永远优先用户选定的目录, +# 注册表只在空壳/链接兜底(真身解析)。 +REGISTRY_PATH = Path(__file__).resolve().parent / "skill.json" +_REG: dict | None = None +_REG_LOCK = threading.Lock() +_BG_SCAN_RUNNING = False + +# 内嵌常见终端 skill 位置(tier1 快扫;相对 home / cwd 解析,路径不存在自动跳过) +_BUILTIN_HOME_ROOTS = ( + ".config/opencode/skills", # opencode(oc) + ".workbuddy/skills", # workbuddy(wb) + ".cc-switch/skills", # cc-switch + ".claude/skills", # claude code +) +_BUILTIN_CWD_ROOTS = ( + ".workbuddy/skills", # 项目内 wb skill + ".opencode/skills", # 项目内 oc skill + ".claude/skills", # 项目内 claude skill +) + + +def builtin_skill_roots(distro: str = DEFAULT_DISTRO) -> list[str]: + """内嵌常见终端的 skill 根(tier1 快扫用)—— 先扫这几个,不一上来全盘忙扫。 + + 组成:① 内嵌 home/cwd 常见终端位置;② agents.md 终端表里配置的目录 + (用户新增终端自动纳入);③ 环境变量 OC_SKILL_ALT_ROOTS 追加; + ④ WSL 侧两根(仅 VM 已在跑才带 —— 快扫绝不唤醒 WSL,避免冷启动拖慢推拉)。 + 全程只拼路径 + 一次 wsl 列表查询;目录不存在由扫描端跳过。 + """ + home = Path.home() + roots: list[Path] = [home / p for p in _BUILTIN_HOME_ROOTS] + try: + roots += [Path.cwd() / p for p in _BUILTIN_CWD_ROOTS] + except OSError: + pass + for t in load_agent_table(): + if t.get("dir"): + roots.append(Path(t["dir"])) + extra = os.environ.get("OC_SKILL_ALT_ROOTS", "") + roots += [Path(p) for p in extra.split(os.pathsep) if p.strip()] + if os.name == "nt": + try: + wsl_on = _wsl_running(distro) + except Exception: + wsl_on = False + if wsl_on: + unc = Path("\\\\wsl.localhost\\%s" % distro) + roots += [unc / "root" / ".config" / "opencode" / "skills", + unc / "root" / ".workbuddy" / "skills"] + out: list[str] = [] + seen: set[str] = set() + for r in roots: + key = os.path.normcase(os.path.abspath(str(r))) + if key in seen: + continue + seen.add(key) + out.append(str(r)) + return out + + +def deep_scan_roots() -> list[str]: + """后台扫盘的扩展根(tier2):常见终端之外的兜底位置。 + + 只扫浅层、只认名为 skills 的目录(不递归、不全盘): + home / home/.config / home/.local/share / cwd 下一级子目录里的 skills/。 + —— 市面终端(cursor / cline / roo …)把 skill 放 `~/<终端>/skills` 的, + 这层就能兜住;再深的全盘扫既慢又容易扫进无关目录,不做。 + """ + out: list[str] = [] + try: + bases = [Path.home(), Path.home() / ".config", + Path.home() / ".local" / "share", Path.cwd()] + except OSError: + return out + for base in bases: + try: + for child in sorted(base.iterdir()): + if not child.is_dir() or child.name.startswith("."): + continue + s = child / "skills" + try: + if s.is_dir(): + out.append(str(s)) + except OSError: + continue + except OSError: + continue + dedup: list[str] = [] + seen: set[str] = set() + for r in out: + key = os.path.normcase(os.path.abspath(r)) + if key not in seen: + seen.add(key) + dedup.append(r) + return dedup + + +def load_registry() -> dict: + """启动时把 skill.json 读到内存(只读一次,之后全走内存缓存 _REG)。""" + global _REG + with _REG_LOCK: + if _REG is None: + reg: dict = {"version": 1, "updated": "", "skills": {}} + try: + if REGISTRY_PATH.is_file(): + data = json.loads(REGISTRY_PATH.read_text(encoding="utf-8")) + if isinstance(data, dict) and isinstance(data.get("skills"), dict): + reg = data + reg.setdefault("version", 1) + reg.setdefault("updated", "") + except (OSError, ValueError): + pass # 文件不存在 / 损坏:从空注册表起步,不炸推拉 + _REG = reg + return _REG + + +def save_registry() -> None: + """把内存注册表落库 skill.json(临时文件 + os.replace 原子写,避免半截文件)。""" + with _REG_LOCK: + reg = _REG or {"version": 1, "updated": "", "skills": {}} + try: + import datetime as _dt + reg["updated"] = _dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S") + tmp = REGISTRY_PATH.with_suffix(".json.tmp") + tmp.write_text(json.dumps(reg, ensure_ascii=False, indent=2), + encoding="utf-8") + os.replace(str(tmp), str(REGISTRY_PATH)) + except OSError as exc: + print(" (skill.json 落库失败:%s)" % exc) + + +def remember_skills(found: dict) -> None: + """把扫到的 {名字: [位置]} 合并进注册表并落库(只增不删、位置去重)。 + + 真身被删/移动的旧位置不主动清(合并语义)—— 查询端会做存在性校验, + 过期条目只是查不到,不会误导。 + """ + if not found: + return + reg = load_registry() + skills = reg.setdefault("skills", {}) + changed = False + for name, paths in found.items(): + bucket = skills.setdefault(name, []) + for p in paths: + if p not in bucket: + bucket.append(p) + changed = True + if changed: + save_registry() + + +def registry_locations(name: str) -> list[str]: + """内存注册表里该名字的位置列表(存在性 / SKILL.md 校验由调用方做)。""" + reg = load_registry() + return list(reg.get("skills", {}).get(name) or []) + + +def scan_roots_for_name(name: str, roots: list[str] | None = None, + distro: str = DEFAULT_DISTRO) -> list[str]: + """tier1 快扫:在给定根里按名字找真身目录(<根>/<名> 含 SKILL.md 才算)。 + + roots 缺省 = builtin_skill_roots()(内嵌常见终端位置,秒级、不唤醒 WSL)。 + 只读、任何失败跳过该根;返回真身目录字符串列表(去重)。 + """ + out: list[str] = [] + seen: set[str] = set() + for r in (roots if roots is not None else builtin_skill_roots(distro)): + cand = Path(r) / name + try: + if find_skill_md(cand) is None: + continue + except OSError: + continue + key = os.path.normcase(os.path.abspath(str(cand))) + if key not in seen: + seen.add(key) + out.append(str(cand)) + return out + + +def _scan_root_skills(root: str) -> dict: + """扫一个根下所有合格 skill -> {名字: [目录]}(有 SKILL.md 才登记)。""" + found: dict = {} + base = Path(root) + try: + children = sorted(base.iterdir()) + except OSError: + return found + for child in children: + if not child.is_dir() or child.name.startswith("."): + continue + if not NAME_RE.match(child.name): + continue + try: + if find_skill_md(child) is None: + continue + except OSError: + continue + found.setdefault(child.name, []).append(str(child)) + return found + + +def start_background_scan(name: str = "") -> bool: + """tier3 后台扫盘(不阻塞当前推拉):扫完把最终结果落库 skill.json。 + + 已有扫描在跑则直接返回 False(单飞,不叠加线程)。扫描范围 = + 内嵌常见终端根 + 扩展浅层根;UNC 根按 6 秒限时(防 9p 半死拖住线程)。 + 扫完打印一行进度(后台输出,不参与交互)。 + """ + global _BG_SCAN_RUNNING + with _REG_LOCK: + if _BG_SCAN_RUNNING: + return False + _BG_SCAN_RUNNING = True + + def _run() -> None: + global _BG_SCAN_RUNNING + try: + roots = list(dict.fromkeys( + builtin_skill_roots() + deep_scan_roots())) + merged: dict = {} + for r in roots: + if not _is_unc_root(r): # 本机根直扫 + one = _scan_root_skills(r) + else: # UNC 根限时(半死 VM 不拖线程) + ok, res = _guard(lambda r=r: _scan_root_skills(r), 6.0) + one = res if ok and isinstance(res, dict) else {} + for n, paths in one.items(): + merged.setdefault(n, []).extend(paths) + remember_skills(merged) + total = len(load_registry().get("skills", {})) + print("\n [skill.json] 后台扫盘完成:本次登记 %d 个 skill," + "库内共 %d 个名字(已落库 %s)" + % (len(merged), total, REGISTRY_PATH)) + if name and name not in merged: + print(" 提示:后台扫盘也没找到「%s」的真身," + "可能它只存在于未纳入扫描的位置。" % name) + finally: + _BG_SCAN_RUNNING = False + + threading.Thread(target=_run, daemon=True, name="skill-registry-scan").start() + return True + + +def real_src_locations(src: str) -> list[str]: + """空壳/链接源的真身位置分层查找(real_src 用;返回目录列表,可能为空)。 + + ① 内存注册表(启动已读 skill.json)→ ② 内嵌常见终端位置快扫(命中即落库) + → ③ 现有跨根兜底 _alt_skill_dirs(.trash / env 根)→ 都没中调用方再后台扫盘。 + """ + try: + p = Path(src) + name = p.name + except (OSError, ValueError): + return [] + + def _same(a, b) -> bool: + return os.path.normcase(os.path.abspath(str(a))) == \ + os.path.normcase(os.path.abspath(str(b))) + + def _valid(d) -> bool: + try: + return (not _same(d, p)) and Path(d).is_dir() and \ + find_skill_md(Path(d)) is not None + except OSError: + return False + + # ① 内存注册表:零扫盘直接命中(skill.json 启动已进内存) + hits = [d for d in registry_locations(name) if _valid(d)] + if hits: + return hits + # ② tier1 快扫:内嵌常见终端位置(秒级,不唤醒 WSL) + quick = scan_roots_for_name(name) + if quick: + remember_skills({name: quick}) # 扫到即落库,下次走 ① + return quick + # ③ 现有跨根兜底(含 .trash 回收站、OC_SKILL_ALT_ROOTS) + return [d for d in _alt_skill_dirs(p) if _valid(d)] + + + # ---------------------------------------------------------------- 软链接诊断(SKILL.md 缺失时) def _is_link(path) -> bool: """是否是链接 —— Windows 上 symlink 与 junction 都要认。 @@ -835,8 +1124,11 @@ def real_src(src: str) -> str | None: 选了 oc 里的空壳/链接 skill,推送时直接推真身(如 wb),不再提醒手动换来源)。 - 链接(symlink / junction)-> 解析链接目标,目标里有 SKILL.md 才切; - - 目录在但没有 SKILL.md(空壳)-> 跨根找同名 skill 真身目录; - - 有效目录 / zip / SKILL.md 文件 -> 返回 None(保持原路径不动)。 + - 目录在但没有 SKILL.md(空壳)-> 分层找真身:内存注册表(skill.json,启动 + 已读入)-> 内嵌常见终端位置快扫(oc/wb/cc-switch…,命中即落库)-> 跨根兜底 + _alt_skill_dirs(.trash / env 根);都没命中再启动后台扫盘(不阻塞本次推拉, + 扫完落库,下次直接走注册表)。 + - 有效目录 / zip / SKILL.md 文件 -> 返回 None(用户选定的目录优先,保持原路径不动)。 全程只读;任何一步失败返回 None,由调用方走原诊断路径。 """ try: @@ -859,12 +1151,15 @@ def real_src(src: str) -> str | None: return str(real) except OSError: pass - # 2) 空壳 / 断链:跨根找真身目录 + # 2) 空壳 / 断链:注册表 -> 快扫 -> 跨根兜底的分层查找 try: if find_skill_md(p) is None: - for d in _alt_skill_dirs(p): + for d in real_src_locations(str(p)): if not _same(d, p): return d + # 都没命中:后台扫盘兜底(tier3,daemon 线程,不阻塞当前推拉; + # 扫完落库 skill.json,下次同名走内存注册表一步命中) + start_background_scan(p.name) except OSError: pass return None @@ -1275,6 +1570,20 @@ def cmd_push(args) -> int: with tempfile.TemporaryDirectory(prefix="ocskill_") as tmp: work = Path(tmp) + # 源是软链接 / 空壳目录 -> 与菜单同一条分层链自动切真身(2026-09-25 CLI 对称; + # 只对目录生效 —— zip / SKILL.md / base64 走原路径,不动) + try: + src_path = clean_input(args.source) + if Path(src_path).is_dir(): + real = real_src(src_path) + if real and os.path.normcase(os.path.abspath(real)) != \ + os.path.normcase(os.path.abspath(src_path)): + print("源是软链接或空壳,已自动切换到真实位置推送:") + print(f" 原:{src_path}") + print(f" 新:{real}") + args.source = real + except OSError: + pass candidate = stage_candidate(args.source, work, args.base64) ok, err, meta = validate_skill(candidate) @@ -1346,6 +1655,25 @@ def cmd_pull(args) -> int: raise SystemExit(probe["err"]) src_dir = Path(probe["src"]) + # 来源是软链接 / 空壳 -> 与推送同一条分层链自动切真身(2026-09-25 推拉对称: + # 注册表 -> 内嵌常见终端快扫 -> 跨根兜底;没命中再后台扫盘,不阻塞本次拉取) + try: + need_real = find_skill_md(src_dir) is None + except OSError: + need_real = True + if need_real: + real = real_src(str(src_dir)) + if real and os.path.normcase(os.path.abspath(real)) != \ + os.path.normcase(os.path.abspath(str(src_dir))): + print(f"来源是软链接或空壳,已自动切换到真实位置拉取:") + print(f" 原:{src_dir}") + print(f" 新:{real}") + src_dir = Path(real) + else: + start_background_scan(name) + print(f"⚠ 来源目录缺 SKILL.md(空壳或链接断了):{src_dir}") + print(" 已在后台扫盘找真身(结果落库 skill.json,稍后重试即可自动切)。") + # 输出目录不由用户决定:默认直接落本机 OC 的 skill 目录(写完即被 watch 热加载) out_dir = Path(args.out) if args.out else Path(WIN_LOCAL_ROOT) out_dir.mkdir(parents=True, exist_ok=True) @@ -1571,6 +1899,7 @@ def main(argv: list[str] | None = None) -> int: if not getattr(args, "cmd", None): parser.print_help() return 0 + load_registry() # 启动时把 skill.json 读进内存(读失败走空表,不影响执行) try: return args.func(args) except KeyboardInterrupt: