288 lines
11 KiB
Python
288 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())
|