#!/usr/bin/env python3 # -*- coding: utf-8 -*- """dsh —— 本项目工作流「统一入口」（第 1 版，2026-10-01） 存在理由（实测）：规则写成「多步祈使句」⇒ 每步 = 1 次工具调用； 每棒固定开销 12~15 次。本入口把多步合进 1 次调用。 用法： dsh open <会话名> [目标文件...] 开工：状态 → 判锁范围 → 抢锁（原 3 步） dsh close <会话名> [单号...] 收尾：反序释放 + 清单提醒（原 2~5 步） dsh stat [--full] 中途查：锁 + git + 本会话日志水位（原 3~4 步） dsh log [会话id] 看本会话诊断日志字节数 + 档位判断（原 1~2 步） dsh probe <探针名> [参数...] 跑已存探针（替代「写→跑→清」3 步） dsh collect 收尾四件套（原 4 步） ⛔ 只读生产日志（仅 stat 取大小）；⛔ 不常驻；⛔ 不写任何日志文件。 """ import os import shutil import subprocess import sys import time PY = r"E:/ProgramData/.workbuddy/binaries/python/versions/3.13.12/python.exe" WS = r"E:/ProgramData/AIProject/ai1net-dsh-server" DOC = r"D:/github/dsh_shenxian/dsh-server-docs" CFG = r"E:/ProgramData/.workbuddy" GUARD = DOC + "/07-scripts/handoff-guard.sh" PREFLIGHT = DOC + "/07-scripts/preflight-lock.sh" DOMAIN = "ai1net-dsh-server/" SOFT_LOG = 5 * 1024 * 1024 HARD_LOG = 8 * 1024 * 1024 COLLECT = [ ("docs-audit", PY, DOC + "/07-scripts/docs-audit.py"), ("docs-index-stats", PY, DOC + "/07-scripts/docs-index-stats.py", "--write"), ("docs-manifest", PY, DOC + "/07-scripts/docs-manifest.py"), ("docs-sync-check", "bash", DOC + "/07-scripts/docs-sync-check.sh"), ] # ⛔ 2026-10-01 实测：这些底层脚本**不认 `--help`**（会当成路径/参数照跑；`docs-manifest.py --help` # 甚至去写 `--help\docs-manifest.json`）⇒ 入口自己拦，否则 `dsh collect --help` 会**静默跑四件套** USAGE = { "open": 'dsh open <会话名> [目标文件...] # ⚠️ 必须带目标文件，否则跳过 preflight', "close": "dsh close <会话名> [单号...] # ⛔ 单号可省：省了只释放 OWNER==本会话名的", "stat": "dsh stat [--full] # --full 再跑一遍 state.py（默认不跑，只给锁/git/日志三项）", "log": "dsh log [会话id]", "probe": "dsh probe <探针名> [参数...]", "collect": "dsh collect # 收尾四件套（⚠️ 会写 INDEX.md，需先持文档库锁）", } def say(text): sys.stdout.buffer.write((text + "\n").encode("utf-8", "replace")) sys.stdout.buffer.flush() def run(cmd, cwd=WS): say("$ " + " ".join(str(c) for c in cmd)) try: p = subprocess.run(cmd, cwd=cwd, capture_output=True, timeout=600) except Exception as exc: say("!! 执行失败: %r" % (exc,)) return 127, "" out = (p.stdout or b"") + (p.stderr or b"") txt = out.decode("utf-8", "replace") lines = txt.splitlines() if len(lines) > 40: lines = lines[:40] + ["…（共 %d 行，已截断）" % len(txt.splitlines())] say("\n".join(lines).rstrip()) say("rc=%d" % p.returncode) return p.returncode, txt def bash_path(): return shutil.which("bash") or r"E:/ProgramData/.workbuddy/binaries/PortableGit/versions/1.2.0/bin/bash.exe" def newest_log(): d = os.path.join(CFG, "logs", time.strftime("%Y-%m-%d"), "sdk", "conversations") if not os.path.isdir(d): return None best, mt = None, -1 for n in os.listdir(d): if not n.endswith(".log"): continue f = os.path.join(d, n) st = os.stat(f).st_mtime if st > mt: best, mt = f, st return best def find_session_log(sid=None): """按**会话 id** 精确定位本会话日志；命中不了才退回 `newest_log()`。 ⚠️ 2026-10-01 实测隐患：`newest_log()` 只按 mtime 取 ⇒ **可能取到别的会话的日志**， 于是软/硬档位算到别人头上（虚高虚低都可能）。本函数优先精确匹配， ⛔ 且**不改变** `newest_log()` 原语义（它仍是兜底）。 返回 `(精确命中的路径 or None, 实际采用的路径 or None)`。 """ sid = (sid or os.environ.get("CODEBUDDY_SESSION_ID") or "").strip() if not sid: return None, newest_log() d = os.path.join(CFG, "logs", time.strftime("%Y-%m-%d"), "sdk", "conversations") if os.path.isdir(d): for n in sorted(os.listdir(d)): if n.endswith(".log") and sid in n: p = os.path.join(d, n) return p, p return None, newest_log() def log_size(f): """字节数 + 档位（软 5 MiB / 硬 8 MiB，与 §G 一致）。""" n = os.path.getsize(f) tier = "🟢 低于软档" if n >= HARD_LOG: tier = "🔴 硬档：立刻停手 + 建接续会话" elif n >= SOFT_LOG: tier = "🟡 软档：开始写交接材料" return n, tier def cmd_stat(a): """一次取齐「锁 / git / 本会话日志水位」= 原 3~4 次工具调用。 ⛔ 只读：不抢锁、不碰文件、不写日志。`--full` 才追加跑 state.py。 """ say("== 锁（只读查询，⛔ 不抢）==") run([bash_path(), GUARD, "--status"]) say("== git ==") run(["git", "-C", WS, "log", "-1", "--format=%h %ad %s", "--date=format:%m-%d %H:%M"]) rc, txt = run(["git", "-C", WS, "status", "--porcelain=v1", "--branch"]) if rc == 0: dirty = [x for x in txt.splitlines() if x.strip() and not x.startswith("##")] say("（本工作区改动 %d 项）" % len(dirty)) say("== 本会话日志水位 ==") sid = os.environ.get("CODEBUDDY_SESSION_ID") or "" exact, f = find_session_log(sid) if not f or not os.path.exists(f): say("找不到会话日志（logs/<今天>/sdk/conversations/ 下无 .log）") else: n, tier = log_size(f) say("会话 id=%s 精确命中=%s" % (sid or "（环境里没有 id）", "是" if exact else "否（回退今日最新 mtime）")) say("%s\n %d B (%.2f MiB) Δ到硬档 %d KB\n %s" % ( f, n, n / 1048576.0, max(HARD_LOG - n, 0) // 1024, tier)) if "--full" in a: say("== state.py（--full）==") run([PY, os.path.join(WS, "state.py")]) return 0 def cmd_open(a): if not a: say("用法: dsh open <会话名> [目标文件...]") return 2 name, targets = a[0], a[1:] say("== 1/3 跑状态 ==") run([PY, os.path.join(WS, "state.py")]) say("== 2/3 判可锁定范围 ==") if targets: rc, _ = run([bash_path(), PREFLIGHT, name] + list(targets)) if rc != 0: say("⛔ preflight rc=%d ⇒ 拒绝开工（未归类或机制层非空）" % rc) return rc else: say("（未给目标文件 ⇒ 跳过 preflight）") say("== 3/3 抢锁 ==") rc, _ = run([bash_path(), GUARD, "--claim-exec", name, "--domains", DOMAIN]) if rc != 0: say("⛔ 抢不到锁 ⇒ 停手报告（R9）") return rc say("✅ 开工就绪。收尾务必跑: dsh close %s" % name) return 0 def my_doing_locks(name): """扫 05-交接单/.doing-*，只挑 OWNER 第 1 行 == 本会话名的单号锁。 ⚠️ 2026-10-01 实测：底层 `--release <单号>` 是**裸 `rm -rf`，不做任何归属校验** （`--release-skeleton` / `--release-publish` 2026-09-25 都补了校验，**这个漏了**） ⇒ 归属必须由**本入口自己卡**，否则 `dsh close` 会变成绕过 R9 的后门。 """ d = os.path.join(DOC, "05-交接单") out = [] if not os.path.isdir(d): return out for n in sorted(os.listdir(d)): if not n.startswith(".doing-"): continue try: with open(os.path.join(d, n, "OWNER"), "r", encoding="utf-8", errors="replace") as fh: first = fh.readline().strip() except OSError: continue if first == name: # OWNER 第 1 行 = 会话名（claim 时写入） out.append(n[len(".doing-"):]) return out def cmd_close(a): if not a: say("用法: dsh close <会话名> [单号...]") return 2 name, tasks = a[0], a[1:] say("== 反序释放 ==") # 【1】单号锁：显式单号优先；没给就只释放「OWNER == 本会话名」的那几个 if not tasks: tasks = my_doing_locks(name) if tasks: for t in tasks: run([bash_path(), GUARD, "--release", t]) say("（本会话的单号锁 %d 个已释放）" % len(tasks)) else: say("（无需释放的单号锁 —— 本会话没有 .doing-* 占用）") # 【2】全局执行锁（含域锁） rc, _ = run([bash_path(), GUARD, "--release-exec", name]) say("== 收尾清单（逐条自问，本入口不代做）==") for line in ( "1. 记忆/台账写了吗（.workbuddy/memory/<日期>.md 追加）", "2. 提交边界：未明确要求 ⇒ 不 commit/push", "3. 清 tmp/ 本棒残留", "4. 结束语写明『锁已释放』或点名锁在谁手上", ): say(" - " + line) return 0 def cmd_log(a): """⚠️ 语义保留：`dsh log <已存在路径>` 仍按路径走；给 id 则精确匹配；不给则回退今日本会话。""" want = a[0] if a else "" if want and os.path.exists(want): f = want else: _exact, f = find_session_log(want) if not f or not os.path.exists(f): say("找不到会话日志%s" % ("（指定: %s）" % want if want else "")) return 1 n, tier = log_size(f) say("%s\n %d B (%.2f MiB) Δ到硬档 %d KB\n %s" % ( f, n, n / 1048576.0, max(HARD_LOG - n, 0) // 1024, tier)) return 0 def cmd_probe(a): if not a: d = os.path.join(WS, "tmp") names = sorted(n for n in os.listdir(d)) if os.path.isdir(d) else [] say("已存探针（tmp/）: " + (", ".join(names) or "（空）")) return 0 p = os.path.join(WS, "tmp", a[0]) if not p.endswith(".py"): p += ".py" if not os.path.exists(p): say("⛔ 探针不存在: %s（复用已有的，不要每次新写）" % p) return 1 return run([PY, p] + list(a[1:]))[0] def cmd_collect(_a): bad = 0 for item in COLLECT: say("== %s ==" % item[0]) rc, _ = run(list(item[1:])) bad += (rc != 0) say("四件套完成：%d 项非零" % bad) return 1 if bad else 0 def main(): argv = sys.argv[1:] if not argv or argv[0] in ("-h", "--help", "help"): say(__doc__) return 0 table = {"open": cmd_open, "close": cmd_close, "stat": cmd_stat, "log": cmd_log, "probe": cmd_probe, "collect": cmd_collect} fn = table.get(argv[0]) if not fn: say("未知子命令: %s（可用: %s）" % (argv[0], " / ".join(table))) return 2 if any(x in ("-h", "--help") for x in argv[1:]): say(USAGE.get(argv[0], "")) return 0 return fn(argv[1:]) if __name__ == "__main__": sys.exit(main())