admin
c1b5e4d966
- 变更规模：新增 514 / 修改 62 / 重命名 155 / 删除 4（归档重组与文档轮次） - .gitignore 修：`归档/**/db-cwd归一-备份-*/` —— 原规则写绝对层级（归档/db-cwd归一-…）， 目录搬进 归档/配置与备份/ 后**静默失效**，43 MB 的 DB 备份又变成未跟踪 - .gitignore 补：嵌套 git 内部数据（归档/内嵌git-20261008/、归档/skills-git-旧线-20261007/dotgit-原样移出/） - .gitignore 补：运行态与部署副本（.workbuddy/collab/、.workbuddy/tools/、.workbuddy/.load-pending、.workbuddy/tmp-*） - .gitignore 补：备份件（*.bak-*） - 未跟踪文件从 2190 降到 890（其余为 归档/ 归档件与 .workbuddy/memory/ 知识文件，按口径入库）
289 lines
11 KiB
Python
289 lines
11 KiB
Python
#!/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())
|