仓级：按新规则重新规范化行尾符（--renormalize）
上一步立了 .gitattributes（text eol=lf），这一步把已入库的文件按新规则 统一成 LF。245 个文件，其中 product-planning 239 个、session-mechanism 6 个。 这批改动**只有行尾符**：`git diff --ignore-cr-at-eol` 对全部 245 个文件 零内容差异，未触碰任何二进制与资源文件。 单独一笔的理由：把"改规则"与"按规则刷数据"分开，将来若有谁的改动撞上， 可以直接对着这一笔回退，而不牵连 .gitattributes 本身。
This commit is contained in:
admin committed
1 parent
cf0ec41245
commit
9e779c4889
245 files changed
+44412
-44412
No files matched your search
@@ -1,43 +1,43 @@
|
||||
{
|
||||
"_说明": "复制为 collabd.config.json（放在 collabd.py 同目录）并改成你的项目路径；也可用环境变量 COLLABD_CONFIG 指向别处的配置文件。",
|
||||
"workspace": "E:/ProgramData/AIProject/<你的工作区>",
|
||||
"inbox": "tmp/supervise-inbox",
|
||||
"live": "交付物/协作实时状态.md",
|
||||
"taskgraph": "交付物/任务图.json",
|
||||
"lines": {
|
||||
"line-a": "A 线",
|
||||
"line-b": "B 线"
|
||||
},
|
||||
"goal_docs": [
|
||||
[
|
||||
"验收判据 / 目标",
|
||||
"交付物/目标与验收判据.md"
|
||||
],
|
||||
[
|
||||
"进度看板",
|
||||
"交付物/进度看板.md"
|
||||
]
|
||||
],
|
||||
"targets": {
|
||||
"line-a": [
|
||||
[
|
||||
"棒1 执行单",
|
||||
"docs/执行单_*.md"
|
||||
]
|
||||
],
|
||||
"line-b": [
|
||||
[
|
||||
"对接单产出",
|
||||
"docs/*对接单*.md"
|
||||
]
|
||||
]
|
||||
},
|
||||
"host_db": "",
|
||||
"singleton_port": 20099,
|
||||
"interval": 10,
|
||||
"idle_min": 12,
|
||||
"stuck_min": 30,
|
||||
"vacuum_min": 5,
|
||||
"health_every": 300,
|
||||
"_边界": "⛔ 本机制（协作）不含任何业务专属项（如业务端口探测、业务客户端拉起）——那些属各业务线自己的件。"
|
||||
}
|
||||
{
|
||||
"_说明": "复制为 collabd.config.json（放在 collabd.py 同目录）并改成你的项目路径；也可用环境变量 COLLABD_CONFIG 指向别处的配置文件。",
|
||||
"workspace": "E:/ProgramData/AIProject/<你的工作区>",
|
||||
"inbox": "tmp/supervise-inbox",
|
||||
"live": "交付物/协作实时状态.md",
|
||||
"taskgraph": "交付物/任务图.json",
|
||||
"lines": {
|
||||
"line-a": "A 线",
|
||||
"line-b": "B 线"
|
||||
},
|
||||
"goal_docs": [
|
||||
[
|
||||
"验收判据 / 目标",
|
||||
"交付物/目标与验收判据.md"
|
||||
],
|
||||
[
|
||||
"进度看板",
|
||||
"交付物/进度看板.md"
|
||||
]
|
||||
],
|
||||
"targets": {
|
||||
"line-a": [
|
||||
[
|
||||
"棒1 执行单",
|
||||
"docs/执行单_*.md"
|
||||
]
|
||||
],
|
||||
"line-b": [
|
||||
[
|
||||
"对接单产出",
|
||||
"docs/*对接单*.md"
|
||||
]
|
||||
]
|
||||
},
|
||||
"host_db": "",
|
||||
"singleton_port": 20099,
|
||||
"interval": 10,
|
||||
"idle_min": 12,
|
||||
"stuck_min": 30,
|
||||
"vacuum_min": 5,
|
||||
"health_every": 300,
|
||||
"_边界": "⛔ 本机制（协作）不含任何业务专属项（如业务端口探测、业务客户端拉起）——那些属各业务线自己的件。"
|
||||
}
|
||||
@@ -1,302 +1,302 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""supervise-ensure-hook.py —— UserPromptSubmit 钩子：**开工那一刻自动把常驻补上**。
|
||||
|
||||
🔴🔴 为什么必须有这个脚本（2026-10-04 用户两次纠正 + 一次「我要说几遍」）
|
||||
用户原话：「**调用执行任务完成目标时，判断是否已经开启，如果没开启就要开启**」
|
||||
⇒ 这句话是**技能必须自己做的动作**，⛔ 不是「写进文档等我照着做」。
|
||||
|
||||
🔴 实测的病根（`vibe-product` 区）
|
||||
· 该区常驻**停 16.4 小时**，而我在那边「一步一步按文档走」——第一步就没人做
|
||||
· `ensure_supervise()` **早就存在**、能自己起常驻，
|
||||
⛔ 但它**只被文档和自测引用**，**没有任何地方在开工时调它**
|
||||
⇒ 能力齐了，**缺一根接线**。本脚本就是那根接线。
|
||||
|
||||
⚖️ 三条边界（⚠️ 越界就是又一次「凭空造需求」）
|
||||
1. ⛔ **不碰「没启用」的工作区** —— 用户 2026-10-04 口径：
|
||||
「**没有目标就说明 当前用户是用的是基础会话方式**」⇒ 那是**正常态**，
|
||||
**不是故障**，**不需要"开启"**。⇒ 判据＝`goal.json` 有**且** `lifecycle` 有效。
|
||||
2. ⛔ **不代用户开目标** —— 想让它有 ⇒ **用户开**（`goal_life_set` 是用户动作）。
|
||||
3. ⛔ **只做「常驻在不在」这一件事** —— ⛔ 不建任务会话、⛔ 不建检查会话
|
||||
（那两件是**主会话**的判断，见 SKILL.md T 表）。
|
||||
⇒ 越权等级：`本脚本 ≤ 第 3 条`；`第 2 条` 用户专属；`建会话` 主会话专属。
|
||||
|
||||
输出：`hookSpecificOutput.additionalContext` 注入一句**短**提醒（⛔ 不长篇大论）。
|
||||
退出码恒 0（⛔ 钩子失败 ⇔ 用户的话被吞 ⇒ 最坏的结果）。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
HERE = Path(__file__).resolve().parent
|
||||
SKILL = HERE.parent.parent
|
||||
try:
|
||||
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# 🔴🔴 2026-10-04 用户问「为什么 vibe 会话说技能没有调整」时挖出的真缺陷：
|
||||
# **本钩子在两个工作区全是空转，一次都没生效过。**
|
||||
# 病根＝我写它时只读 `DSH_WS_ROOT`，而 `settings.json` 的 hook 条目
|
||||
# **没有 env 字段** ⇒ 宿主 env 里 `DSH_WS_ROOT` 恒空
|
||||
# （实测 Process/User/Machine 三级全空）⇒ 第①步就 `_noop()` 返回。
|
||||
# ⇒ 「能力齐了、缺一根接线」我以为接上了，**其实那根线接在墙上没插进插头**。
|
||||
# ✅ 正确取法＝**照抄 `wb-result-hook.py` / `lock-guard-hook.py`**：
|
||||
# ① stdin 的 `payload.cwd`（宿主一定给）
|
||||
# ② `roots.env` 的 `DSH_WS_ROOT`（副本专属，⛔ 不能只靠宿主 env）
|
||||
def _sm_load_roots() -> None:
|
||||
"""把本包 `env/roots.env` 以 setdefault 注入 env（宿主 env 优先，⛔ 不覆盖）。"""
|
||||
cands = [SKILL / "env" / "roots.env", SKILL / "roots.env"]
|
||||
_cfg = os.environ.get("CODEBUDDY_CONFIG_DIR") or os.path.expanduser("~/.workbuddy")
|
||||
cands.append(Path(_cfg) / "skills" / "session-mechanism" / "env" / "roots.env")
|
||||
for p in cands:
|
||||
try:
|
||||
for line in p.read_text(encoding="utf-8").splitlines():
|
||||
line = line.strip()
|
||||
if not line or line.startswith("#") or "=" not in line:
|
||||
continue
|
||||
k, v = line.split("=", 1)
|
||||
os.environ.setdefault(k.strip(), v.strip())
|
||||
return
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
|
||||
_sm_load_roots()
|
||||
|
||||
COLLABD = SKILL / "scripts" / "collabd.py"
|
||||
PY = sys.executable
|
||||
|
||||
|
||||
def _norm(p: str) -> str:
|
||||
"""`E:\\x\\y` / `E:/x/y` ⇒ 统一正斜杠 + 去尾斜杠（⛔ 与 `tasks.json` 去重键同形）。"""
|
||||
return (p or "").replace("\\", "/").rstrip("/")
|
||||
|
||||
|
||||
def _stdin_payload() -> dict:
|
||||
"""读 hook stdin（⛔ 最多 1 MiB，⛔ 失败当没有）。
|
||||
|
||||
🔴🔴 **必须缓存**（2026-10-04 实测栽到）：
|
||||
`sys.stdin.buffer.read()` 是**一次性**的 —— 读第一次之后再调
|
||||
`_ws_root()` 只会拿到空串 ⇒ 表现为「注入零输出」，
|
||||
**和「没触发」一模一样** ⇒ 又一次假绿。
|
||||
⇒ 只读一次、存进 `_PAYLOAD`，后续一律走缓存。
|
||||
"""
|
||||
global _PAYLOAD
|
||||
if _PAYLOAD is None:
|
||||
try:
|
||||
raw = sys.stdin.buffer.read(1024 * 1024)
|
||||
_PAYLOAD = json.loads(raw.decode("utf-8", "replace")) if raw else {}
|
||||
if not isinstance(_PAYLOAD, dict):
|
||||
_PAYLOAD = {}
|
||||
except Exception:
|
||||
_PAYLOAD = {}
|
||||
return _PAYLOAD
|
||||
|
||||
|
||||
_PAYLOAD = None
|
||||
|
||||
|
||||
def _ws_root() -> str:
|
||||
"""🔴 工作区真源＝**stdin 的 `payload.cwd`**（与 `wb-result-hook` 同一口径）。
|
||||
|
||||
⚠️ 为什么不能只信 env：`settings.json` 的 hook 条目**没有 env 字段**，
|
||||
宿主 env 里 `DSH_WS_ROOT` 实测恒空（10-04 三级全空）⇒ 只读 env 的钩子
|
||||
**一次都不会触发**（这正是本脚本上一版的病）。
|
||||
⚠️ 也不直接用 `Path.cwd()`：钩子的 cwd 是**宿主进程目录**，⛔ 不是会话工作区。
|
||||
"""
|
||||
return _norm(str(_stdin_payload().get("cwd") or ""))
|
||||
|
||||
|
||||
def _out(obj) -> None:
|
||||
data = json.dumps(obj, ensure_ascii=False).encode("utf-8")
|
||||
try:
|
||||
sys.stdout.buffer.write(data)
|
||||
sys.stdout.buffer.flush()
|
||||
except Exception:
|
||||
sys.stdout.write(data.decode("utf-8", "replace"))
|
||||
sys.stdout.flush()
|
||||
|
||||
|
||||
def _noop() -> None:
|
||||
"""⛔ 什么都不注入 —— 绝大多数情况（⛔ 别每次提交都吵用户）。"""
|
||||
_out({"hookSpecificOutput": {"hookEventName": "UserPromptSubmit",
|
||||
"additionalContext": ""}})
|
||||
|
||||
|
||||
def _cfg(ws: Path):
|
||||
"""读该工作区的部署配置（⛔ 读不到 ⇒ 视为**未启用**，⛔ 不碰）。"""
|
||||
p = ws / ".workbuddy" / "collab" / "collabd.config.json"
|
||||
try:
|
||||
return json.loads(p.read_text(encoding="utf-8"))
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def _goal(ws: Path):
|
||||
"""读 `goal.json`（⛔ 没有 ⇒ **基础会话方式**，⛔ 不碰）。"""
|
||||
for rel in (("tmp", "supervise-inbox", "goal.json"),):
|
||||
p = ws.joinpath(*rel)
|
||||
try:
|
||||
if p.is_file():
|
||||
return json.loads(p.read_text(encoding="utf-8"))
|
||||
except Exception:
|
||||
return None
|
||||
return None
|
||||
|
||||
|
||||
def main() -> int:
|
||||
# ① 拿工作区（🔴 真源＝stdin 的 `payload.cwd`；env / `roots.env` 只兜底）
|
||||
ws = _ws_root() or (os.environ.get("DSH_WS_ROOT") or "").strip()
|
||||
if not ws:
|
||||
_noop()
|
||||
return 0
|
||||
ws = Path(ws)
|
||||
if not ws.is_dir():
|
||||
_noop()
|
||||
return 0
|
||||
|
||||
# ② ⚖️ 边界①：**没启用 ⇒ 完全不碰**（不是故障，是用户在用基础会话方式）
|
||||
cfg = _cfg(ws)
|
||||
if not cfg:
|
||||
_noop()
|
||||
return 0
|
||||
goal = _goal(ws)
|
||||
if not goal:
|
||||
# 📌 无 `goal.json` ⇒ 基础会话方式 ⇒ ⛔ 不起常驻、⛔ 不注入任何东西
|
||||
_noop()
|
||||
return 0
|
||||
|
||||
# 🔴🔴 2026-10-04 用户拍板（原话：「**当然需要判断如果没有起就要起**」）：
|
||||
# ⇒ 判据＝「**有 `goal.json` 就归我管**」，⛔ **不看 `lifecycle` 的值**。
|
||||
# ⚠️ 为什么**故意不看** lifecycle（这是口径，⛔ 不是漏写）：
|
||||
# 用户说「继续完成目标」这个动作本身**就是"有活"的证据** ——
|
||||
# 目标处于「已完成／等待」时用户仍然开口 ⇒ 说明还有事要办。
|
||||
# 若这里加一道「lifecycle 必须进行中」的闸，就会出现
|
||||
# **用户说了继续、却什么都没发生** ⇒ 又回到"感觉没改到位"。
|
||||
# ⚠️ 已知副作用（**接受**，⛔ 不是缺陷）：`已完成` 的区里常驻会被拉起，
|
||||
# 但 `maybe_spawn_check_agent()` 五道闸① 仍要求「进行中」⇒ 暂时不建检查会话。
|
||||
# ⇒ 那份「只起常驻、不建检查会话」的中间态是**刻意的**：
|
||||
# 先保证后台在，检查会话等目标状态被改回「进行中」再说。
|
||||
# ⛔ 变异对照见 `selftest.py::t_ensure_hook_fires_on_any_lifecycle()`。
|
||||
|
||||
# ③ 问程序自己「常驻在不在」（⛔ 别手拼路径、⛔ 别读快照）
|
||||
# 🔴🔴 2026-10-04 实测栽到：`env = dict(os.environ)` 之后再往 `env` 里塞
|
||||
# `COLLABD_CONFIG` 是**无效**的 —— `exec_module()` 期间 `collabd.load_cfg()`
|
||||
# 读的是 **`os.environ`**（真进程环境），不是那个局部字典。
|
||||
# ⇒ 实测 `CFG_USED` 仍指向 **别的区**，把「A 区常驻在位」当成本区结论
|
||||
# ⇒ 又一次**假绿**。正解＝**直接改 `os.environ`**（并记住原值，出栈还原）。
|
||||
cfg_env_key = "COLLABD_CONFIG"
|
||||
_cfg_path = str(ws / ".workbuddy" / "collab" / "collabd.config.json").replace("\\", "/")
|
||||
_saved_env = os.environ.get(cfg_env_key)
|
||||
os.environ[cfg_env_key] = _cfg_path
|
||||
try:
|
||||
import importlib.util
|
||||
_sp = importlib.util.spec_from_file_location("_cd_hook", COLLABD)
|
||||
m = importlib.util.module_from_spec(_sp)
|
||||
_sp.loader.exec_module(m)
|
||||
# 📌 自检：加载完必须确认它**真的在看本区**（⛔ 不确认＝又一次假绿）
|
||||
_used = str(getattr(m, "CFG_USED", "") or "").replace("\\", "/")
|
||||
if Path(_used).resolve() != Path(_cfg_path).resolve():
|
||||
_noop()
|
||||
return 0
|
||||
alive, det = m.supervise_alive()
|
||||
except Exception:
|
||||
_noop()
|
||||
return 0
|
||||
finally:
|
||||
if _saved_env is None:
|
||||
os.environ.pop(cfg_env_key, None)
|
||||
else:
|
||||
os.environ[cfg_env_key] = _saved_env
|
||||
|
||||
# ④ 在位 ⇒ 零输出（⛔ 别每次提交都吵）
|
||||
if alive:
|
||||
_noop()
|
||||
return 0
|
||||
|
||||
# ⑤ 🔴 不在位 ⇒ 自己起来（这就是用户那句「没开启就要开启」）
|
||||
# 🔴 复查用的 `m` 是**上面已加载的本区模块**；此时 `os.environ` 已还原
|
||||
# ⇒ `supervise_alive()` 若重新读 env 会验到别的区 ⇒ 必须把 env 再钉回去。
|
||||
#
|
||||
# 🔴🔴 2026-10-04 **时间预算（必须守住，⛔ 超时＝用户那句话被吞＝最坏结果）**：
|
||||
# 钩子注册里 `timeout=30`（`settings.json` 的 `UserPromptSubmit` 第 6 条）
|
||||
# ⇒ 本函数总耗时必须**显著小于 30 s**，否则宿主掐掉钩子 ⇒ 输入丢失。
|
||||
# 实测正常路径 3.5 s，但**起不来**那条路径原实现最坏 =
|
||||
# `subprocess timeout 40` ＋ 回验循环 `10 × 0.9 ≈ 9` ⇒ **理论 49 s > 30 s**。
|
||||
# ✅ 收口（2026-10-04 用户批准「可以」）：
|
||||
# · 子进程超时 `40 → 15`（`--ensure` 本身是"起一条即返回"，15 s 足够）
|
||||
# · 回验循环 `10 × 0.9 ≈ 9 → 5 × 0.8 = 4`（心跳节拍实测 ~1 s 内就出来）
|
||||
# ⇒ 最坏 ≈ 15 + 4 + 1 = **20 s**，留 10 s 余量给解释器启动与 import。
|
||||
_ENSURE_SUBPROC_TIMEOUT = 15
|
||||
_VERIFY_TRIES, _VERIFY_GAP = 5, 0.8
|
||||
try:
|
||||
_restore_env = os.environ.get(cfg_env_key)
|
||||
os.environ[cfg_env_key] = _cfg_path
|
||||
env = dict(os.environ)
|
||||
# 🔴 参数名是 **`--ensure`**（⛔ **没有** `--ensure-supervise`）
|
||||
# 🔴🔴 **⛔ 不看它的返回码**（2026-10-04 实测坐实）：它会报
|
||||
# `{"ok": false, "spawned": true}` ⇒ **`spawned` 看着像成功、其实秒退**。
|
||||
# 那个区因此「起了三次、活零次」，而我照着返回码报了"起好了"。
|
||||
# ⇒ 唯一可信判据＝**等它自己活下来，再问 `supervise_alive()`**。
|
||||
subprocess.run([PY, str(COLLABD), "--ensure"],
|
||||
env=env, cwd=str(ws), capture_output=True,
|
||||
timeout=_ENSURE_SUBPROC_TIMEOUT)
|
||||
ok = False
|
||||
for _ in range(_VERIFY_TRIES):
|
||||
time.sleep(_VERIFY_GAP)
|
||||
try:
|
||||
if m.supervise_alive()[0]:
|
||||
ok = True
|
||||
break
|
||||
except Exception:
|
||||
pass
|
||||
except Exception:
|
||||
ok = False
|
||||
finally:
|
||||
if _restore_env is None:
|
||||
os.environ.pop(cfg_env_key, None)
|
||||
else:
|
||||
os.environ[cfg_env_key] = _restore_env
|
||||
if not ok:
|
||||
# ⛔ 起不来 ⇒ **只报事实**，⛔ 不许编原因（今天栽过 5 次）
|
||||
# 🔴 但「起不来」本身就是用户要的那条红线（10-04 原话「起不来就先报错」）
|
||||
# ⇒ 这里必须**报出来**，⛔ 不许再静默零输出。
|
||||
_out({"hookSpecificOutput": {"hookEventName": "UserPromptSubmit",
|
||||
"additionalContext": (
|
||||
"🔴 **会话机制 · 常驻没起来**（本工作区有目标在册，但常驻不在位，"
|
||||
"自动补起失败）——⛔ 先修好这个再往下做，别当它已经好了。")}})
|
||||
return 0
|
||||
|
||||
# ⑥ 起好了 ⇒ 验一次，⛔ 打印「已启动」不算（照 supervise-persistence 的验收口径）
|
||||
# 🔴 同样要把 env 钉回本区，否则复查读的是别的区的心跳（⛔ 验错对象＝假绿）
|
||||
_r2 = os.environ.get(cfg_env_key)
|
||||
os.environ[cfg_env_key] = _cfg_path
|
||||
try:
|
||||
time.sleep(1.0)
|
||||
alive2, det2 = m.supervise_alive()
|
||||
except Exception:
|
||||
alive2, det2 = False, "复查失败"
|
||||
finally:
|
||||
if _r2 is None:
|
||||
os.environ.pop(cfg_env_key, None)
|
||||
else:
|
||||
os.environ[cfg_env_key] = _r2
|
||||
msg = ("🔧 **会话机制 · 常驻已自动补起**（本工作区 goal 在册但常驻不在位）"
|
||||
if alive2 else
|
||||
"🔴 **会话机制 · 常驻起后仍未活**（%s）——⛔ 这是事实，先修好再往下做。" % det2)
|
||||
_out({"hookSpecificOutput": {"hookEventName": "UserPromptSubmit",
|
||||
"additionalContext": msg}})
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
sys.exit(main())
|
||||
except Exception:
|
||||
# 🔴 铁律：钩子**绝不能**让用户的话被吞 ⇒ 任何异常都静默退出 0
|
||||
sys.exit(0)
|
||||
# -*- coding: utf-8 -*-
|
||||
"""supervise-ensure-hook.py —— UserPromptSubmit 钩子：**开工那一刻自动把常驻补上**。
|
||||
|
||||
🔴🔴 为什么必须有这个脚本（2026-10-04 用户两次纠正 + 一次「我要说几遍」）
|
||||
用户原话：「**调用执行任务完成目标时，判断是否已经开启，如果没开启就要开启**」
|
||||
⇒ 这句话是**技能必须自己做的动作**，⛔ 不是「写进文档等我照着做」。
|
||||
|
||||
🔴 实测的病根（`vibe-product` 区）
|
||||
· 该区常驻**停 16.4 小时**，而我在那边「一步一步按文档走」——第一步就没人做
|
||||
· `ensure_supervise()` **早就存在**、能自己起常驻，
|
||||
⛔ 但它**只被文档和自测引用**，**没有任何地方在开工时调它**
|
||||
⇒ 能力齐了，**缺一根接线**。本脚本就是那根接线。
|
||||
|
||||
⚖️ 三条边界（⚠️ 越界就是又一次「凭空造需求」）
|
||||
1. ⛔ **不碰「没启用」的工作区** —— 用户 2026-10-04 口径：
|
||||
「**没有目标就说明 当前用户是用的是基础会话方式**」⇒ 那是**正常态**，
|
||||
**不是故障**，**不需要"开启"**。⇒ 判据＝`goal.json` 有**且** `lifecycle` 有效。
|
||||
2. ⛔ **不代用户开目标** —— 想让它有 ⇒ **用户开**（`goal_life_set` 是用户动作）。
|
||||
3. ⛔ **只做「常驻在不在」这一件事** —— ⛔ 不建任务会话、⛔ 不建检查会话
|
||||
（那两件是**主会话**的判断，见 SKILL.md T 表）。
|
||||
⇒ 越权等级：`本脚本 ≤ 第 3 条`；`第 2 条` 用户专属；`建会话` 主会话专属。
|
||||
|
||||
输出：`hookSpecificOutput.additionalContext` 注入一句**短**提醒（⛔ 不长篇大论）。
|
||||
退出码恒 0（⛔ 钩子失败 ⇔ 用户的话被吞 ⇒ 最坏的结果）。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
HERE = Path(__file__).resolve().parent
|
||||
SKILL = HERE.parent.parent
|
||||
try:
|
||||
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# 🔴🔴 2026-10-04 用户问「为什么 vibe 会话说技能没有调整」时挖出的真缺陷：
|
||||
# **本钩子在两个工作区全是空转，一次都没生效过。**
|
||||
# 病根＝我写它时只读 `DSH_WS_ROOT`，而 `settings.json` 的 hook 条目
|
||||
# **没有 env 字段** ⇒ 宿主 env 里 `DSH_WS_ROOT` 恒空
|
||||
# （实测 Process/User/Machine 三级全空）⇒ 第①步就 `_noop()` 返回。
|
||||
# ⇒ 「能力齐了、缺一根接线」我以为接上了，**其实那根线接在墙上没插进插头**。
|
||||
# ✅ 正确取法＝**照抄 `wb-result-hook.py` / `lock-guard-hook.py`**：
|
||||
# ① stdin 的 `payload.cwd`（宿主一定给）
|
||||
# ② `roots.env` 的 `DSH_WS_ROOT`（副本专属，⛔ 不能只靠宿主 env）
|
||||
def _sm_load_roots() -> None:
|
||||
"""把本包 `env/roots.env` 以 setdefault 注入 env（宿主 env 优先，⛔ 不覆盖）。"""
|
||||
cands = [SKILL / "env" / "roots.env", SKILL / "roots.env"]
|
||||
_cfg = os.environ.get("CODEBUDDY_CONFIG_DIR") or os.path.expanduser("~/.workbuddy")
|
||||
cands.append(Path(_cfg) / "skills" / "session-mechanism" / "env" / "roots.env")
|
||||
for p in cands:
|
||||
try:
|
||||
for line in p.read_text(encoding="utf-8").splitlines():
|
||||
line = line.strip()
|
||||
if not line or line.startswith("#") or "=" not in line:
|
||||
continue
|
||||
k, v = line.split("=", 1)
|
||||
os.environ.setdefault(k.strip(), v.strip())
|
||||
return
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
|
||||
_sm_load_roots()
|
||||
|
||||
COLLABD = SKILL / "scripts" / "collabd.py"
|
||||
PY = sys.executable
|
||||
|
||||
|
||||
def _norm(p: str) -> str:
|
||||
"""`E:\\x\\y` / `E:/x/y` ⇒ 统一正斜杠 + 去尾斜杠（⛔ 与 `tasks.json` 去重键同形）。"""
|
||||
return (p or "").replace("\\", "/").rstrip("/")
|
||||
|
||||
|
||||
def _stdin_payload() -> dict:
|
||||
"""读 hook stdin（⛔ 最多 1 MiB，⛔ 失败当没有）。
|
||||
|
||||
🔴🔴 **必须缓存**（2026-10-04 实测栽到）：
|
||||
`sys.stdin.buffer.read()` 是**一次性**的 —— 读第一次之后再调
|
||||
`_ws_root()` 只会拿到空串 ⇒ 表现为「注入零输出」，
|
||||
**和「没触发」一模一样** ⇒ 又一次假绿。
|
||||
⇒ 只读一次、存进 `_PAYLOAD`，后续一律走缓存。
|
||||
"""
|
||||
global _PAYLOAD
|
||||
if _PAYLOAD is None:
|
||||
try:
|
||||
raw = sys.stdin.buffer.read(1024 * 1024)
|
||||
_PAYLOAD = json.loads(raw.decode("utf-8", "replace")) if raw else {}
|
||||
if not isinstance(_PAYLOAD, dict):
|
||||
_PAYLOAD = {}
|
||||
except Exception:
|
||||
_PAYLOAD = {}
|
||||
return _PAYLOAD
|
||||
|
||||
|
||||
_PAYLOAD = None
|
||||
|
||||
|
||||
def _ws_root() -> str:
|
||||
"""🔴 工作区真源＝**stdin 的 `payload.cwd`**（与 `wb-result-hook` 同一口径）。
|
||||
|
||||
⚠️ 为什么不能只信 env：`settings.json` 的 hook 条目**没有 env 字段**，
|
||||
宿主 env 里 `DSH_WS_ROOT` 实测恒空（10-04 三级全空）⇒ 只读 env 的钩子
|
||||
**一次都不会触发**（这正是本脚本上一版的病）。
|
||||
⚠️ 也不直接用 `Path.cwd()`：钩子的 cwd 是**宿主进程目录**，⛔ 不是会话工作区。
|
||||
"""
|
||||
return _norm(str(_stdin_payload().get("cwd") or ""))
|
||||
|
||||
|
||||
def _out(obj) -> None:
|
||||
data = json.dumps(obj, ensure_ascii=False).encode("utf-8")
|
||||
try:
|
||||
sys.stdout.buffer.write(data)
|
||||
sys.stdout.buffer.flush()
|
||||
except Exception:
|
||||
sys.stdout.write(data.decode("utf-8", "replace"))
|
||||
sys.stdout.flush()
|
||||
|
||||
|
||||
def _noop() -> None:
|
||||
"""⛔ 什么都不注入 —— 绝大多数情况（⛔ 别每次提交都吵用户）。"""
|
||||
_out({"hookSpecificOutput": {"hookEventName": "UserPromptSubmit",
|
||||
"additionalContext": ""}})
|
||||
|
||||
|
||||
def _cfg(ws: Path):
|
||||
"""读该工作区的部署配置（⛔ 读不到 ⇒ 视为**未启用**，⛔ 不碰）。"""
|
||||
p = ws / ".workbuddy" / "collab" / "collabd.config.json"
|
||||
try:
|
||||
return json.loads(p.read_text(encoding="utf-8"))
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def _goal(ws: Path):
|
||||
"""读 `goal.json`（⛔ 没有 ⇒ **基础会话方式**，⛔ 不碰）。"""
|
||||
for rel in (("tmp", "supervise-inbox", "goal.json"),):
|
||||
p = ws.joinpath(*rel)
|
||||
try:
|
||||
if p.is_file():
|
||||
return json.loads(p.read_text(encoding="utf-8"))
|
||||
except Exception:
|
||||
return None
|
||||
return None
|
||||
|
||||
|
||||
def main() -> int:
|
||||
# ① 拿工作区（🔴 真源＝stdin 的 `payload.cwd`；env / `roots.env` 只兜底）
|
||||
ws = _ws_root() or (os.environ.get("DSH_WS_ROOT") or "").strip()
|
||||
if not ws:
|
||||
_noop()
|
||||
return 0
|
||||
ws = Path(ws)
|
||||
if not ws.is_dir():
|
||||
_noop()
|
||||
return 0
|
||||
|
||||
# ② ⚖️ 边界①：**没启用 ⇒ 完全不碰**（不是故障，是用户在用基础会话方式）
|
||||
cfg = _cfg(ws)
|
||||
if not cfg:
|
||||
_noop()
|
||||
return 0
|
||||
goal = _goal(ws)
|
||||
if not goal:
|
||||
# 📌 无 `goal.json` ⇒ 基础会话方式 ⇒ ⛔ 不起常驻、⛔ 不注入任何东西
|
||||
_noop()
|
||||
return 0
|
||||
|
||||
# 🔴🔴 2026-10-04 用户拍板（原话：「**当然需要判断如果没有起就要起**」）：
|
||||
# ⇒ 判据＝「**有 `goal.json` 就归我管**」，⛔ **不看 `lifecycle` 的值**。
|
||||
# ⚠️ 为什么**故意不看** lifecycle（这是口径，⛔ 不是漏写）：
|
||||
# 用户说「继续完成目标」这个动作本身**就是"有活"的证据** ——
|
||||
# 目标处于「已完成／等待」时用户仍然开口 ⇒ 说明还有事要办。
|
||||
# 若这里加一道「lifecycle 必须进行中」的闸，就会出现
|
||||
# **用户说了继续、却什么都没发生** ⇒ 又回到"感觉没改到位"。
|
||||
# ⚠️ 已知副作用（**接受**，⛔ 不是缺陷）：`已完成` 的区里常驻会被拉起，
|
||||
# 但 `maybe_spawn_check_agent()` 五道闸① 仍要求「进行中」⇒ 暂时不建检查会话。
|
||||
# ⇒ 那份「只起常驻、不建检查会话」的中间态是**刻意的**：
|
||||
# 先保证后台在，检查会话等目标状态被改回「进行中」再说。
|
||||
# ⛔ 变异对照见 `selftest.py::t_ensure_hook_fires_on_any_lifecycle()`。
|
||||
|
||||
# ③ 问程序自己「常驻在不在」（⛔ 别手拼路径、⛔ 别读快照）
|
||||
# 🔴🔴 2026-10-04 实测栽到：`env = dict(os.environ)` 之后再往 `env` 里塞
|
||||
# `COLLABD_CONFIG` 是**无效**的 —— `exec_module()` 期间 `collabd.load_cfg()`
|
||||
# 读的是 **`os.environ`**（真进程环境），不是那个局部字典。
|
||||
# ⇒ 实测 `CFG_USED` 仍指向 **别的区**，把「A 区常驻在位」当成本区结论
|
||||
# ⇒ 又一次**假绿**。正解＝**直接改 `os.environ`**（并记住原值，出栈还原）。
|
||||
cfg_env_key = "COLLABD_CONFIG"
|
||||
_cfg_path = str(ws / ".workbuddy" / "collab" / "collabd.config.json").replace("\\", "/")
|
||||
_saved_env = os.environ.get(cfg_env_key)
|
||||
os.environ[cfg_env_key] = _cfg_path
|
||||
try:
|
||||
import importlib.util
|
||||
_sp = importlib.util.spec_from_file_location("_cd_hook", COLLABD)
|
||||
m = importlib.util.module_from_spec(_sp)
|
||||
_sp.loader.exec_module(m)
|
||||
# 📌 自检：加载完必须确认它**真的在看本区**（⛔ 不确认＝又一次假绿）
|
||||
_used = str(getattr(m, "CFG_USED", "") or "").replace("\\", "/")
|
||||
if Path(_used).resolve() != Path(_cfg_path).resolve():
|
||||
_noop()
|
||||
return 0
|
||||
alive, det = m.supervise_alive()
|
||||
except Exception:
|
||||
_noop()
|
||||
return 0
|
||||
finally:
|
||||
if _saved_env is None:
|
||||
os.environ.pop(cfg_env_key, None)
|
||||
else:
|
||||
os.environ[cfg_env_key] = _saved_env
|
||||
|
||||
# ④ 在位 ⇒ 零输出（⛔ 别每次提交都吵）
|
||||
if alive:
|
||||
_noop()
|
||||
return 0
|
||||
|
||||
# ⑤ 🔴 不在位 ⇒ 自己起来（这就是用户那句「没开启就要开启」）
|
||||
# 🔴 复查用的 `m` 是**上面已加载的本区模块**；此时 `os.environ` 已还原
|
||||
# ⇒ `supervise_alive()` 若重新读 env 会验到别的区 ⇒ 必须把 env 再钉回去。
|
||||
#
|
||||
# 🔴🔴 2026-10-04 **时间预算（必须守住，⛔ 超时＝用户那句话被吞＝最坏结果）**：
|
||||
# 钩子注册里 `timeout=30`（`settings.json` 的 `UserPromptSubmit` 第 6 条）
|
||||
# ⇒ 本函数总耗时必须**显著小于 30 s**，否则宿主掐掉钩子 ⇒ 输入丢失。
|
||||
# 实测正常路径 3.5 s，但**起不来**那条路径原实现最坏 =
|
||||
# `subprocess timeout 40` ＋ 回验循环 `10 × 0.9 ≈ 9` ⇒ **理论 49 s > 30 s**。
|
||||
# ✅ 收口（2026-10-04 用户批准「可以」）：
|
||||
# · 子进程超时 `40 → 15`（`--ensure` 本身是"起一条即返回"，15 s 足够）
|
||||
# · 回验循环 `10 × 0.9 ≈ 9 → 5 × 0.8 = 4`（心跳节拍实测 ~1 s 内就出来）
|
||||
# ⇒ 最坏 ≈ 15 + 4 + 1 = **20 s**，留 10 s 余量给解释器启动与 import。
|
||||
_ENSURE_SUBPROC_TIMEOUT = 15
|
||||
_VERIFY_TRIES, _VERIFY_GAP = 5, 0.8
|
||||
try:
|
||||
_restore_env = os.environ.get(cfg_env_key)
|
||||
os.environ[cfg_env_key] = _cfg_path
|
||||
env = dict(os.environ)
|
||||
# 🔴 参数名是 **`--ensure`**（⛔ **没有** `--ensure-supervise`）
|
||||
# 🔴🔴 **⛔ 不看它的返回码**（2026-10-04 实测坐实）：它会报
|
||||
# `{"ok": false, "spawned": true}` ⇒ **`spawned` 看着像成功、其实秒退**。
|
||||
# 那个区因此「起了三次、活零次」，而我照着返回码报了"起好了"。
|
||||
# ⇒ 唯一可信判据＝**等它自己活下来，再问 `supervise_alive()`**。
|
||||
subprocess.run([PY, str(COLLABD), "--ensure"],
|
||||
env=env, cwd=str(ws), capture_output=True,
|
||||
timeout=_ENSURE_SUBPROC_TIMEOUT)
|
||||
ok = False
|
||||
for _ in range(_VERIFY_TRIES):
|
||||
time.sleep(_VERIFY_GAP)
|
||||
try:
|
||||
if m.supervise_alive()[0]:
|
||||
ok = True
|
||||
break
|
||||
except Exception:
|
||||
pass
|
||||
except Exception:
|
||||
ok = False
|
||||
finally:
|
||||
if _restore_env is None:
|
||||
os.environ.pop(cfg_env_key, None)
|
||||
else:
|
||||
os.environ[cfg_env_key] = _restore_env
|
||||
if not ok:
|
||||
# ⛔ 起不来 ⇒ **只报事实**，⛔ 不许编原因（今天栽过 5 次）
|
||||
# 🔴 但「起不来」本身就是用户要的那条红线（10-04 原话「起不来就先报错」）
|
||||
# ⇒ 这里必须**报出来**，⛔ 不许再静默零输出。
|
||||
_out({"hookSpecificOutput": {"hookEventName": "UserPromptSubmit",
|
||||
"additionalContext": (
|
||||
"🔴 **会话机制 · 常驻没起来**（本工作区有目标在册，但常驻不在位，"
|
||||
"自动补起失败）——⛔ 先修好这个再往下做，别当它已经好了。")}})
|
||||
return 0
|
||||
|
||||
# ⑥ 起好了 ⇒ 验一次，⛔ 打印「已启动」不算（照 supervise-persistence 的验收口径）
|
||||
# 🔴 同样要把 env 钉回本区，否则复查读的是别的区的心跳（⛔ 验错对象＝假绿）
|
||||
_r2 = os.environ.get(cfg_env_key)
|
||||
os.environ[cfg_env_key] = _cfg_path
|
||||
try:
|
||||
time.sleep(1.0)
|
||||
alive2, det2 = m.supervise_alive()
|
||||
except Exception:
|
||||
alive2, det2 = False, "复查失败"
|
||||
finally:
|
||||
if _r2 is None:
|
||||
os.environ.pop(cfg_env_key, None)
|
||||
else:
|
||||
os.environ[cfg_env_key] = _r2
|
||||
msg = ("🔧 **会话机制 · 常驻已自动补起**（本工作区 goal 在册但常驻不在位）"
|
||||
if alive2 else
|
||||
"🔴 **会话机制 · 常驻起后仍未活**（%s）——⛔ 这是事实，先修好再往下做。" % det2)
|
||||
_out({"hookSpecificOutput": {"hookEventName": "UserPromptSubmit",
|
||||
"additionalContext": msg}})
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
sys.exit(main())
|
||||
except Exception:
|
||||
# 🔴 铁律：钩子**绝不能**让用户的话被吞 ⇒ 任何异常都静默退出 0
|
||||
sys.exit(0)
|
||||
@@ -1,94 +1,94 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""handoff-status.py — 一条命令看清 05-交接单 里哪些单子【待执行】/【已完成】/【无状态】。
|
||||
|
||||
为什么需要它：`05-交接单/交接单-已完成/` 那层目录摊平后，**"目录位置"不再表达状态**。
|
||||
状态改由**机读字段**承载（与布局解耦，摊平/搬家都不失效）：
|
||||
① 单子头部 `- 状态：…` 行（模板已有）→ ② 回退取 `05-交接单/README.md` 台账表的「状态」列
|
||||
|
||||
用法：python 07-scripts/handoff-status.py
|
||||
"""
|
||||
# [session-mechanism] roots.env 外置（由 install.py 生成 ⇒ `<包>/env/roots.env`）
|
||||
# 查找顺序：向上 4 层 × (env/roots.env 优先 → 包根 roots.env 兼容) → 技能库那份。
|
||||
def _sm_load_roots():
|
||||
import os as _os
|
||||
_here = _os.path.dirname(_os.path.abspath(__file__))
|
||||
_cands = []
|
||||
for _up in range(4):
|
||||
_base = _os.path.normpath(_os.path.join(_here, *([".."] * _up)))
|
||||
for _rel in (("env", "roots.env"), ("roots.env",)):
|
||||
_cands.append(_os.path.join(_base, *_rel))
|
||||
_conf = _os.environ.get("CODEBUDDY_CONFIG_DIR") or _os.path.expanduser("~/.workbuddy")
|
||||
_cands.append(_os.path.join(_conf, "skills", "session-mechanism", "env", "roots.env"))
|
||||
for _p in _cands:
|
||||
if _os.path.isfile(_p):
|
||||
try:
|
||||
with open(_p, encoding="utf-8") as _f:
|
||||
for _ln in _f:
|
||||
_ln = _ln.strip()
|
||||
if _ln and not _ln.startswith("#") and "=" in _ln:
|
||||
_k, _v = _ln.split("=", 1)
|
||||
_os.environ.setdefault(_k.strip(), _v.strip())
|
||||
except Exception:
|
||||
pass
|
||||
return
|
||||
_sm_load_roots()
|
||||
|
||||
import os
|
||||
import re
|
||||
|
||||
ROOT = os.environ.get("DSH_DOCS_ROOT") or os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
D = os.path.join(ROOT, "05-交接单")
|
||||
README = os.path.join(D, "README.md")
|
||||
|
||||
HEAD_STATUS = re.compile(r"^[\s>*\-|#]*\*{0,2}状态\*{0,2}\s*[:：]\s*(.+?)\s*$", re.M)
|
||||
DONE_KW = ("已完成", "已落地", "已归档", "已收口", "已闭合", "已实施", "已上线",
|
||||
"已修复", "已释放", "已结束", "已通过", "全部完成", "全绿", "✅")
|
||||
WAIT_KW = ("待执行", "待开工", "进行中", "在途", "待验证", "待跑", "规划中", "未开工",
|
||||
"部分执行", "受阻", "未做", "设计定稿", "🔄", "⏳", "⏸", "🟡")
|
||||
|
||||
table = {}
|
||||
if os.path.exists(README):
|
||||
with open(README, encoding="utf-8") as fh:
|
||||
for line in fh:
|
||||
if not line.startswith("|"):
|
||||
continue
|
||||
cells = [c.strip() for c in line.strip().strip("|").split("|")]
|
||||
if len(cells) < 2 or ".md" not in cells[0]:
|
||||
continue
|
||||
table[cells[0].strip("`").split("/")[-1].strip()] = cells[1]
|
||||
|
||||
groups = {"待执行": [], "已完成": [], "无状态": []}
|
||||
targets = []
|
||||
for root2, dirs2, _f in os.walk(D):
|
||||
dirs2[:] = [x for x in dirs2 if not x.startswith(".")]
|
||||
for f2 in _f:
|
||||
if f2.endswith(".md") and f2 != "README.md" and not f2.startswith("."):
|
||||
rel = os.path.relpath(os.path.join(root2, f2), D).replace("\\", "/")
|
||||
targets.append((rel, os.path.join(root2, f2)))
|
||||
for n, path in sorted(targets):
|
||||
base = n.split("/")[-1]
|
||||
with open(path, encoding="utf-8", errors="ignore") as fh:
|
||||
head = "".join(fh.readlines()[:30])
|
||||
m = HEAD_STATUS.search(head)
|
||||
if m:
|
||||
s, src = m.group(1)[:44], "头部"
|
||||
elif base in table:
|
||||
s, src = table[base][:44], "台账表"
|
||||
else:
|
||||
s, src = "", ""
|
||||
if not s:
|
||||
groups["无状态"].append((n, "—"))
|
||||
elif any(k in s for k in DONE_KW):
|
||||
groups["已完成"].append((n, "%s ·%s" % (s, src)))
|
||||
elif any(k in s for k in WAIT_KW):
|
||||
groups["待执行"].append((n, "%s ·%s" % (s, src)))
|
||||
else:
|
||||
groups["无状态"].append((n, "%s ·%s未识别" % (s, src)))
|
||||
|
||||
for g in ("待执行", "已完成", "无状态"):
|
||||
print("【%s】%d 个" % (g, len(groups[g])))
|
||||
for n, s in groups[g]:
|
||||
print(" %-44s %s" % (n[:44], s))
|
||||
print()
|
||||
print("合计 %d 个单子（不含 README.md）" % sum(len(v) for v in groups.values()))
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""handoff-status.py — 一条命令看清 05-交接单 里哪些单子【待执行】/【已完成】/【无状态】。
|
||||
|
||||
为什么需要它：`05-交接单/交接单-已完成/` 那层目录摊平后，**"目录位置"不再表达状态**。
|
||||
状态改由**机读字段**承载（与布局解耦，摊平/搬家都不失效）：
|
||||
① 单子头部 `- 状态：…` 行（模板已有）→ ② 回退取 `05-交接单/README.md` 台账表的「状态」列
|
||||
|
||||
用法：python 07-scripts/handoff-status.py
|
||||
"""
|
||||
# [session-mechanism] roots.env 外置（由 install.py 生成 ⇒ `<包>/env/roots.env`）
|
||||
# 查找顺序：向上 4 层 × (env/roots.env 优先 → 包根 roots.env 兼容) → 技能库那份。
|
||||
def _sm_load_roots():
|
||||
import os as _os
|
||||
_here = _os.path.dirname(_os.path.abspath(__file__))
|
||||
_cands = []
|
||||
for _up in range(4):
|
||||
_base = _os.path.normpath(_os.path.join(_here, *([".."] * _up)))
|
||||
for _rel in (("env", "roots.env"), ("roots.env",)):
|
||||
_cands.append(_os.path.join(_base, *_rel))
|
||||
_conf = _os.environ.get("CODEBUDDY_CONFIG_DIR") or _os.path.expanduser("~/.workbuddy")
|
||||
_cands.append(_os.path.join(_conf, "skills", "session-mechanism", "env", "roots.env"))
|
||||
for _p in _cands:
|
||||
if _os.path.isfile(_p):
|
||||
try:
|
||||
with open(_p, encoding="utf-8") as _f:
|
||||
for _ln in _f:
|
||||
_ln = _ln.strip()
|
||||
if _ln and not _ln.startswith("#") and "=" in _ln:
|
||||
_k, _v = _ln.split("=", 1)
|
||||
_os.environ.setdefault(_k.strip(), _v.strip())
|
||||
except Exception:
|
||||
pass
|
||||
return
|
||||
_sm_load_roots()
|
||||
|
||||
import os
|
||||
import re
|
||||
|
||||
ROOT = os.environ.get("DSH_DOCS_ROOT") or os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
D = os.path.join(ROOT, "05-交接单")
|
||||
README = os.path.join(D, "README.md")
|
||||
|
||||
HEAD_STATUS = re.compile(r"^[\s>*\-|#]*\*{0,2}状态\*{0,2}\s*[:：]\s*(.+?)\s*$", re.M)
|
||||
DONE_KW = ("已完成", "已落地", "已归档", "已收口", "已闭合", "已实施", "已上线",
|
||||
"已修复", "已释放", "已结束", "已通过", "全部完成", "全绿", "✅")
|
||||
WAIT_KW = ("待执行", "待开工", "进行中", "在途", "待验证", "待跑", "规划中", "未开工",
|
||||
"部分执行", "受阻", "未做", "设计定稿", "🔄", "⏳", "⏸", "🟡")
|
||||
|
||||
table = {}
|
||||
if os.path.exists(README):
|
||||
with open(README, encoding="utf-8") as fh:
|
||||
for line in fh:
|
||||
if not line.startswith("|"):
|
||||
continue
|
||||
cells = [c.strip() for c in line.strip().strip("|").split("|")]
|
||||
if len(cells) < 2 or ".md" not in cells[0]:
|
||||
continue
|
||||
table[cells[0].strip("`").split("/")[-1].strip()] = cells[1]
|
||||
|
||||
groups = {"待执行": [], "已完成": [], "无状态": []}
|
||||
targets = []
|
||||
for root2, dirs2, _f in os.walk(D):
|
||||
dirs2[:] = [x for x in dirs2 if not x.startswith(".")]
|
||||
for f2 in _f:
|
||||
if f2.endswith(".md") and f2 != "README.md" and not f2.startswith("."):
|
||||
rel = os.path.relpath(os.path.join(root2, f2), D).replace("\\", "/")
|
||||
targets.append((rel, os.path.join(root2, f2)))
|
||||
for n, path in sorted(targets):
|
||||
base = n.split("/")[-1]
|
||||
with open(path, encoding="utf-8", errors="ignore") as fh:
|
||||
head = "".join(fh.readlines()[:30])
|
||||
m = HEAD_STATUS.search(head)
|
||||
if m:
|
||||
s, src = m.group(1)[:44], "头部"
|
||||
elif base in table:
|
||||
s, src = table[base][:44], "台账表"
|
||||
else:
|
||||
s, src = "", ""
|
||||
if not s:
|
||||
groups["无状态"].append((n, "—"))
|
||||
elif any(k in s for k in DONE_KW):
|
||||
groups["已完成"].append((n, "%s ·%s" % (s, src)))
|
||||
elif any(k in s for k in WAIT_KW):
|
||||
groups["待执行"].append((n, "%s ·%s" % (s, src)))
|
||||
else:
|
||||
groups["无状态"].append((n, "%s ·%s未识别" % (s, src)))
|
||||
|
||||
for g in ("待执行", "已完成", "无状态"):
|
||||
print("【%s】%d 个" % (g, len(groups[g])))
|
||||
for n, s in groups[g]:
|
||||
print(" %-44s %s" % (n[:44], s))
|
||||
print()
|
||||
print("合计 %d 个单子（不含 README.md）" % sum(len(v) for v in groups.values()))
|
||||
@@ -1,156 +1,156 @@
|
||||
#!/usr/bin/env bash
|
||||
# preflight-lock.sh — 【开工门禁】「先判定可锁定范围，确认锁定后再处理」
|
||||
#
|
||||
# 用户口径（2026-09-22）：
|
||||
# 「会话执行任务前先判断，当前任务涉及范围是否都可锁定，确认锁定后开始处理」
|
||||
#
|
||||
# 与 handoff-guard.sh 的关系：**门禁（本脚本，先判定）→ 抢锁（handoff-guard.sh，后动手）**。
|
||||
# ⛔ 本脚本**不建任何锁**，只做判定与提示；它**不替换** handoff-guard.sh。
|
||||
#
|
||||
# 在 CODEBUDDY.md §6（并发纪律）里是**开工第 0 步**：判定全部可锁 ⇒ 才去 --claim-exec。
|
||||
#
|
||||
# 用法：
|
||||
# bash preflight-lock.sh "<会话名>" <目标文件1> [目标文件2 ...]
|
||||
#
|
||||
# 退出码：0 = 全部可锁且已锁（可以开工）｜1 = 存在不可锁项（停手）｜2 = 用法错误
|
||||
#
|
||||
set -uo pipefail
|
||||
|
||||
ME="${1:-}"; shift || true
|
||||
if [ -z "$ME" ] || [ $# -eq 0 ]; then
|
||||
echo "用法：bash preflight-lock.sh \"<会话名>\" <目标文件...>" >&2
|
||||
exit 2
|
||||
fi
|
||||
|
||||
# [session-mechanism] roots.env 外置（由 install.py 生成 ⇒ `<包>/env/roots.env`；兼容包根旧位置）
|
||||
_SM_HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
_SM_PKG="$(cd "$_SM_HERE/../.." && pwd)"
|
||||
if [ -f "$_SM_PKG/env/roots.env" ]; then . "$_SM_PKG/env/roots.env"
|
||||
elif [ -f "$_SM_PKG/roots.env" ]; then . "$_SM_PKG/roots.env"; fi
|
||||
DOCS_ROOT="${DSH_DOCS_ROOT:-$(cd "$_SM_HERE/.." && pwd)}"
|
||||
CODE_ROOT="${DSH_CODE_REPO:-${DOCS_ROOT%/*}}" # ⛔ 不留盘符：代码库＝文档库的上一级（纯 bash 展开，⛔ 不调 dirname）
|
||||
WS_ROOT="${DSH_WS_ROOT:-$(pwd)}" # ⛔ 不留盘符：默认＝当前工作目录（本文件内未被使用）
|
||||
LOCKROOT="$DOCS_ROOT/05-交接单/.locks"
|
||||
|
||||
# ── 机制层（⛔ 不受 architecture.md 的 src/** 分层管辖，但**全平台共用**）────────
|
||||
# 判据：这些文件任一改动都会改变**所有会话**的行为 ⇒ 与层④同性质（全局影响）。
|
||||
# 2026-09-22 实测踩到：首版门禁把本任务自己的 4 个目标文件全判进【D】未归类 ⇒ 误报。
|
||||
declare -a MECHANISM_RE=(
|
||||
'^07-scripts/handoff-guard\.sh$' # 全局锁本体
|
||||
'^07-scripts/preflight-lock\.sh$' # 本门禁自身（同属"全平台共用"，2026-10-01 补齐——
|
||||
# ⛔ 漏登时它会把自己判成【D】未归类 ⇒ 任何人改本文件都开不了工）
|
||||
'^07-scripts/lock-guard-hook\.py$' # 无锁拒写钩子
|
||||
'^07-scripts/op-lock\.sh$' # 服务器侧操作锁
|
||||
'^07-scripts/(skill-load-guard|stop-dialog-guard|session-log-guard|bash-output-guard)\.py$'
|
||||
# ↑ 本行＝**全部宿主钩子脚本**。🔴 **新增钩子必须加进本行**，否则本门禁把它判成
|
||||
# 【D】未归类 ⇒ **直接拒开工**（2026-10-01 实测：`session-log-guard.py` 与
|
||||
# **既有**的 `bash-output-guard.py` 都踩过；后者自注册起一直缺登）。
|
||||
# 同步面还有**两处**：宿主 settings.json hooks 段 ＋ 库内 CODEBUDDY.md §A 那行清单。
|
||||
'^07-scripts/docs-sync-check\.sh$' # 推送前对账（所有会话共用）
|
||||
'^07-scripts/check-layering\.mjs$' # 分层 ratchet
|
||||
'^07-scripts/layering-baseline\.json$'
|
||||
'^state\.py$' # 所有会话的开机第 0 步
|
||||
'^CODEBUDDY\.md$' # 每会话自动加载的规则
|
||||
'\.workbuddy/skills/' # 🔴 **技能包全体**（2026-10-01 补登）。判据：任何技能都被
|
||||
# **任意会话**加载 ⇒ 改它＝改全平台行为，与上面几条同性质。
|
||||
# ⛔ 漏登时会发生什么（本轮实测）：技能包**不在**
|
||||
# `ai1net-dsh-server/`／`dsh-server-docs/` 等工作区根下 ⇒
|
||||
# 下面的"剥根前缀"全剥不掉 ⇒ 落【D】未归类 ⇒ 门禁报
|
||||
# 「先定域（R2）」—— **提示词本身是错的**（技能包根本不可域锁，
|
||||
# 正解是"独占"）。⇒ 换个会话改本技能包，一律开不了工。
|
||||
)
|
||||
|
||||
# ── 不可锁资源清单（命中 ⇒ 必须走"秒级独占"或串行排队）──────────────
|
||||
# 判据来源：architecture.md R1–R4 层号 ＋ docs/tsconfig.json outDir
|
||||
declare -a NONLOCKABLE_RE=(
|
||||
'^src/(config|crypto|isolation|index)\.ts$' # 层④ 零业务依赖 · 全层依赖
|
||||
'^src/db/schema\.ts$' # 迁移号 = 全局单调序列
|
||||
'^src/web/server\.ts$' # 唯一挂载注册点
|
||||
'^package\.json$' # 依赖账本单一
|
||||
'^07-scripts/layering-baseline\.json$' # 分层 ratchet 单一
|
||||
)
|
||||
|
||||
declare -a SHARED_RE=(
|
||||
'^lib/' # 构建产物（outDir=lib）· 全局独占
|
||||
'^node_modules/'
|
||||
)
|
||||
|
||||
warn_shared=(); need_skeleton=(); ok_files=(); unknown=(); mechanism=()
|
||||
|
||||
for f in "$@"; do
|
||||
rel="${f//\\//}"
|
||||
# 剥掉已知根前缀（顺序要紧：先长后短，否则残留中间段 ⇒ 正则永不匹配）
|
||||
rel="${rel#*ai1net-dsh-server/}"
|
||||
rel="${rel#*aliyun-dsh-server/}"
|
||||
rel="${rel#*dsh-server-docs/}"
|
||||
rel="${rel#*dsh_shenxian/}"
|
||||
rel="${rel#./}"
|
||||
matched=0
|
||||
for re in "${MECHANISM_RE[@]}"; do
|
||||
if printf '%s' "$rel" | grep -qE "$re"; then mechanism+=("$rel"); matched=1; break; fi
|
||||
done
|
||||
[ "$matched" = 1 ] && continue
|
||||
for re in "${NONLOCKABLE_RE[@]}"; do
|
||||
if printf '%s' "$rel" | grep -qE "$re"; then need_skeleton+=("$rel"); matched=1; break; fi
|
||||
done
|
||||
[ "$matched" = 1 ] && continue
|
||||
for re in "${SHARED_RE[@]}"; do
|
||||
if printf '%s' "$rel" | grep -qE "$re"; then warn_shared+=("$rel"); matched=1; break; fi
|
||||
done
|
||||
[ "$matched" = 1 ] && continue
|
||||
if printf '%s' "$rel" | grep -qE '^(src|poc|test|web|docs|scripts|交接单|skills)/'; then
|
||||
ok_files+=("$rel")
|
||||
else
|
||||
unknown+=("$rel")
|
||||
fi
|
||||
done
|
||||
|
||||
echo "══════════ 开工门禁 · 可锁定范围判定 ══════════"
|
||||
echo "会话：$ME"
|
||||
echo
|
||||
|
||||
echo "【A】可独立锁定（域锁，全程持有）：${#ok_files[@]} 个"
|
||||
for f in "${ok_files[@]:0:8}"; do echo " ✓ $f"; done
|
||||
[ ${#ok_files[@]} -gt 8 ] && echo " …还有 $(( ${#ok_files[@]} - 8 )) 个"
|
||||
|
||||
echo
|
||||
echo "【B】不可全程锁定（层④／迁移号／唯一注册点）→ 改走**秒级独占**：${#need_skeleton[@]} 个"
|
||||
for f in "${need_skeleton[@]}"; do echo " ⚠ $f"; done
|
||||
[ ${#need_skeleton[@]} -eq 0 ] && echo " （无）"
|
||||
|
||||
echo
|
||||
echo "【E】机制层（⛔ 不受 src 分层管辖，但**全平台共用** ⇒ 与层④同性质）：${#mechanism[@]} 个"
|
||||
for f in "${mechanism[@]}"; do echo " ◆ $f"; done
|
||||
[ ${#mechanism[@]} -eq 0 ] && echo " （无）"
|
||||
[ ${#mechanism[@]} -gt 0 ] && echo " ⚠ 改这些 = 改所有会话的行为 ⇒ 必须: ①独占②兼容旧行为③改后跑自证"
|
||||
|
||||
echo
|
||||
echo "【C】共享资源（构建产物 / 依赖树）→ 纳入发布锁：${#warn_shared[@]} 个"
|
||||
for f in "${warn_shared[@]:0:4}"; do echo " ⚠ $f"; done
|
||||
[ ${#warn_shared[@]} -eq 0 ] && echo " （无）"
|
||||
|
||||
if [ ${#unknown[@]} -gt 0 ]; then
|
||||
echo
|
||||
echo "【D】未归类路径（⛔ 必须先想清"属哪个域"，R2）：${#unknown[@]} 个"
|
||||
for f in "${unknown[@]:0:6}"; do echo " ? $f"; done
|
||||
fi
|
||||
|
||||
echo
|
||||
echo "════════════════════════════════════════════"
|
||||
rc=0
|
||||
[ ${#unknown[@]} -gt 0 ] && rc=1
|
||||
[ ${#mechanism[@]} -gt 0 ] && rc=1
|
||||
if [ "$rc" = 1 ]; then
|
||||
echo "结论：**暂不可并行开工** ——"
|
||||
[ ${#unknown[@]} -gt 0 ] && echo " · 【D】${#unknown[@]} 个未归类路径：先定域（R2）"
|
||||
if [ ${#mechanism[@]} -gt 0 ]; then
|
||||
echo " · 【E】${#mechanism[@]} 个机制层文件：**全平台共用** ⇒ ⛔ 不得与其他会话并行"
|
||||
echo " ⇒ 仅当**确认无其他会话在跑**时，才允许独占开工；否则停手等窗口"
|
||||
fi
|
||||
exit 1
|
||||
fi
|
||||
echo "结论：**可以锁定** ——"
|
||||
echo " · ${#ok_files[@]} 个走域锁｜${#need_skeleton[@]} 个走秒级独占｜${#warn_shared[@]} 个并发布锁"
|
||||
echo " · 下一步：bash handoff-guard.sh --claim-exec \"$ME\" --domains <由 A 组反推的域>"
|
||||
echo " ⛔ 本脚本是**提示性门禁**：hook 只拦 Write/Edit，Bash 写入不受限（有意安全阀）"
|
||||
exit 0
|
||||
#!/usr/bin/env bash
|
||||
# preflight-lock.sh — 【开工门禁】「先判定可锁定范围，确认锁定后再处理」
|
||||
#
|
||||
# 用户口径（2026-09-22）：
|
||||
# 「会话执行任务前先判断，当前任务涉及范围是否都可锁定，确认锁定后开始处理」
|
||||
#
|
||||
# 与 handoff-guard.sh 的关系：**门禁（本脚本，先判定）→ 抢锁（handoff-guard.sh，后动手）**。
|
||||
# ⛔ 本脚本**不建任何锁**，只做判定与提示；它**不替换** handoff-guard.sh。
|
||||
#
|
||||
# 在 CODEBUDDY.md §6（并发纪律）里是**开工第 0 步**：判定全部可锁 ⇒ 才去 --claim-exec。
|
||||
#
|
||||
# 用法：
|
||||
# bash preflight-lock.sh "<会话名>" <目标文件1> [目标文件2 ...]
|
||||
#
|
||||
# 退出码：0 = 全部可锁且已锁（可以开工）｜1 = 存在不可锁项（停手）｜2 = 用法错误
|
||||
#
|
||||
set -uo pipefail
|
||||
|
||||
ME="${1:-}"; shift || true
|
||||
if [ -z "$ME" ] || [ $# -eq 0 ]; then
|
||||
echo "用法：bash preflight-lock.sh \"<会话名>\" <目标文件...>" >&2
|
||||
exit 2
|
||||
fi
|
||||
|
||||
# [session-mechanism] roots.env 外置（由 install.py 生成 ⇒ `<包>/env/roots.env`；兼容包根旧位置）
|
||||
_SM_HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
_SM_PKG="$(cd "$_SM_HERE/../.." && pwd)"
|
||||
if [ -f "$_SM_PKG/env/roots.env" ]; then . "$_SM_PKG/env/roots.env"
|
||||
elif [ -f "$_SM_PKG/roots.env" ]; then . "$_SM_PKG/roots.env"; fi
|
||||
DOCS_ROOT="${DSH_DOCS_ROOT:-$(cd "$_SM_HERE/.." && pwd)}"
|
||||
CODE_ROOT="${DSH_CODE_REPO:-${DOCS_ROOT%/*}}" # ⛔ 不留盘符：代码库＝文档库的上一级（纯 bash 展开，⛔ 不调 dirname）
|
||||
WS_ROOT="${DSH_WS_ROOT:-$(pwd)}" # ⛔ 不留盘符：默认＝当前工作目录（本文件内未被使用）
|
||||
LOCKROOT="$DOCS_ROOT/05-交接单/.locks"
|
||||
|
||||
# ── 机制层（⛔ 不受 architecture.md 的 src/** 分层管辖，但**全平台共用**）────────
|
||||
# 判据：这些文件任一改动都会改变**所有会话**的行为 ⇒ 与层④同性质（全局影响）。
|
||||
# 2026-09-22 实测踩到：首版门禁把本任务自己的 4 个目标文件全判进【D】未归类 ⇒ 误报。
|
||||
declare -a MECHANISM_RE=(
|
||||
'^07-scripts/handoff-guard\.sh$' # 全局锁本体
|
||||
'^07-scripts/preflight-lock\.sh$' # 本门禁自身（同属"全平台共用"，2026-10-01 补齐——
|
||||
# ⛔ 漏登时它会把自己判成【D】未归类 ⇒ 任何人改本文件都开不了工）
|
||||
'^07-scripts/lock-guard-hook\.py$' # 无锁拒写钩子
|
||||
'^07-scripts/op-lock\.sh$' # 服务器侧操作锁
|
||||
'^07-scripts/(skill-load-guard|stop-dialog-guard|session-log-guard|bash-output-guard)\.py$'
|
||||
# ↑ 本行＝**全部宿主钩子脚本**。🔴 **新增钩子必须加进本行**，否则本门禁把它判成
|
||||
# 【D】未归类 ⇒ **直接拒开工**（2026-10-01 实测：`session-log-guard.py` 与
|
||||
# **既有**的 `bash-output-guard.py` 都踩过；后者自注册起一直缺登）。
|
||||
# 同步面还有**两处**：宿主 settings.json hooks 段 ＋ 库内 CODEBUDDY.md §A 那行清单。
|
||||
'^07-scripts/docs-sync-check\.sh$' # 推送前对账（所有会话共用）
|
||||
'^07-scripts/check-layering\.mjs$' # 分层 ratchet
|
||||
'^07-scripts/layering-baseline\.json$'
|
||||
'^state\.py$' # 所有会话的开机第 0 步
|
||||
'^CODEBUDDY\.md$' # 每会话自动加载的规则
|
||||
'\.workbuddy/skills/' # 🔴 **技能包全体**（2026-10-01 补登）。判据：任何技能都被
|
||||
# **任意会话**加载 ⇒ 改它＝改全平台行为，与上面几条同性质。
|
||||
# ⛔ 漏登时会发生什么（本轮实测）：技能包**不在**
|
||||
# `ai1net-dsh-server/`／`dsh-server-docs/` 等工作区根下 ⇒
|
||||
# 下面的"剥根前缀"全剥不掉 ⇒ 落【D】未归类 ⇒ 门禁报
|
||||
# 「先定域（R2）」—— **提示词本身是错的**（技能包根本不可域锁，
|
||||
# 正解是"独占"）。⇒ 换个会话改本技能包，一律开不了工。
|
||||
)
|
||||
|
||||
# ── 不可锁资源清单（命中 ⇒ 必须走"秒级独占"或串行排队）──────────────
|
||||
# 判据来源：architecture.md R1–R4 层号 ＋ docs/tsconfig.json outDir
|
||||
declare -a NONLOCKABLE_RE=(
|
||||
'^src/(config|crypto|isolation|index)\.ts$' # 层④ 零业务依赖 · 全层依赖
|
||||
'^src/db/schema\.ts$' # 迁移号 = 全局单调序列
|
||||
'^src/web/server\.ts$' # 唯一挂载注册点
|
||||
'^package\.json$' # 依赖账本单一
|
||||
'^07-scripts/layering-baseline\.json$' # 分层 ratchet 单一
|
||||
)
|
||||
|
||||
declare -a SHARED_RE=(
|
||||
'^lib/' # 构建产物（outDir=lib）· 全局独占
|
||||
'^node_modules/'
|
||||
)
|
||||
|
||||
warn_shared=(); need_skeleton=(); ok_files=(); unknown=(); mechanism=()
|
||||
|
||||
for f in "$@"; do
|
||||
rel="${f//\\//}"
|
||||
# 剥掉已知根前缀（顺序要紧：先长后短，否则残留中间段 ⇒ 正则永不匹配）
|
||||
rel="${rel#*ai1net-dsh-server/}"
|
||||
rel="${rel#*aliyun-dsh-server/}"
|
||||
rel="${rel#*dsh-server-docs/}"
|
||||
rel="${rel#*dsh_shenxian/}"
|
||||
rel="${rel#./}"
|
||||
matched=0
|
||||
for re in "${MECHANISM_RE[@]}"; do
|
||||
if printf '%s' "$rel" | grep -qE "$re"; then mechanism+=("$rel"); matched=1; break; fi
|
||||
done
|
||||
[ "$matched" = 1 ] && continue
|
||||
for re in "${NONLOCKABLE_RE[@]}"; do
|
||||
if printf '%s' "$rel" | grep -qE "$re"; then need_skeleton+=("$rel"); matched=1; break; fi
|
||||
done
|
||||
[ "$matched" = 1 ] && continue
|
||||
for re in "${SHARED_RE[@]}"; do
|
||||
if printf '%s' "$rel" | grep -qE "$re"; then warn_shared+=("$rel"); matched=1; break; fi
|
||||
done
|
||||
[ "$matched" = 1 ] && continue
|
||||
if printf '%s' "$rel" | grep -qE '^(src|poc|test|web|docs|scripts|交接单|skills)/'; then
|
||||
ok_files+=("$rel")
|
||||
else
|
||||
unknown+=("$rel")
|
||||
fi
|
||||
done
|
||||
|
||||
echo "══════════ 开工门禁 · 可锁定范围判定 ══════════"
|
||||
echo "会话：$ME"
|
||||
echo
|
||||
|
||||
echo "【A】可独立锁定（域锁，全程持有）：${#ok_files[@]} 个"
|
||||
for f in "${ok_files[@]:0:8}"; do echo " ✓ $f"; done
|
||||
[ ${#ok_files[@]} -gt 8 ] && echo " …还有 $(( ${#ok_files[@]} - 8 )) 个"
|
||||
|
||||
echo
|
||||
echo "【B】不可全程锁定（层④／迁移号／唯一注册点）→ 改走**秒级独占**：${#need_skeleton[@]} 个"
|
||||
for f in "${need_skeleton[@]}"; do echo " ⚠ $f"; done
|
||||
[ ${#need_skeleton[@]} -eq 0 ] && echo " （无）"
|
||||
|
||||
echo
|
||||
echo "【E】机制层（⛔ 不受 src 分层管辖，但**全平台共用** ⇒ 与层④同性质）：${#mechanism[@]} 个"
|
||||
for f in "${mechanism[@]}"; do echo " ◆ $f"; done
|
||||
[ ${#mechanism[@]} -eq 0 ] && echo " （无）"
|
||||
[ ${#mechanism[@]} -gt 0 ] && echo " ⚠ 改这些 = 改所有会话的行为 ⇒ 必须: ①独占②兼容旧行为③改后跑自证"
|
||||
|
||||
echo
|
||||
echo "【C】共享资源（构建产物 / 依赖树）→ 纳入发布锁：${#warn_shared[@]} 个"
|
||||
for f in "${warn_shared[@]:0:4}"; do echo " ⚠ $f"; done
|
||||
[ ${#warn_shared[@]} -eq 0 ] && echo " （无）"
|
||||
|
||||
if [ ${#unknown[@]} -gt 0 ]; then
|
||||
echo
|
||||
echo "【D】未归类路径（⛔ 必须先想清"属哪个域"，R2）：${#unknown[@]} 个"
|
||||
for f in "${unknown[@]:0:6}"; do echo " ? $f"; done
|
||||
fi
|
||||
|
||||
echo
|
||||
echo "════════════════════════════════════════════"
|
||||
rc=0
|
||||
[ ${#unknown[@]} -gt 0 ] && rc=1
|
||||
[ ${#mechanism[@]} -gt 0 ] && rc=1
|
||||
if [ "$rc" = 1 ]; then
|
||||
echo "结论：**暂不可并行开工** ——"
|
||||
[ ${#unknown[@]} -gt 0 ] && echo " · 【D】${#unknown[@]} 个未归类路径：先定域（R2）"
|
||||
if [ ${#mechanism[@]} -gt 0 ]; then
|
||||
echo " · 【E】${#mechanism[@]} 个机制层文件：**全平台共用** ⇒ ⛔ 不得与其他会话并行"
|
||||
echo " ⇒ 仅当**确认无其他会话在跑**时，才允许独占开工；否则停手等窗口"
|
||||
fi
|
||||
exit 1
|
||||
fi
|
||||
echo "结论：**可以锁定** ——"
|
||||
echo " · ${#ok_files[@]} 个走域锁｜${#need_skeleton[@]} 个走秒级独占｜${#warn_shared[@]} 个并发布锁"
|
||||
echo " · 下一步：bash handoff-guard.sh --claim-exec \"$ME\" --domains <由 A 组反推的域>"
|
||||
echo " ⛔ 本脚本是**提示性门禁**：hook 只拦 Write/Edit，Bash 写入不受限（有意安全阀）"
|
||||
exit 0
|
||||
@@ -1,261 +1,261 @@
|
||||
# 🔴 变异验证 · 2026-10-04 · 会话级忙判据两个新缺陷
|
||||
#
|
||||
# ⚠️ 纪律（来自 references/00-动手前必过.md ③④）：
|
||||
# ① **先验基线绿**：基线 PASS 17/FAIL 0（忙判据那条用例）。
|
||||
# ② **锚在唯一位置**：`def _session_busy_conv` / `def _conv_ts` 唯一定位，⛔ 不锚"文本出现过"。
|
||||
# ③ **每个变异先验「被检对象真被改了」**：比对长度+md5，⛔ 没真改就算**失败**（不许跳过分母）。
|
||||
# ④ ⛔ **跳过的变异一律计入失败**。
|
||||
#
|
||||
# 用法：python mut_run.py
|
||||
import hashlib
|
||||
import importlib.util
|
||||
import io
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
COLLABD = os.path.join(HERE, "collabd.py")
|
||||
SELFTEST = os.path.join(HERE, "selftest.py")
|
||||
PY = sys.executable
|
||||
|
||||
|
||||
def md5(p):
|
||||
return hashlib.md5(open(p, "rb").read()).hexdigest()
|
||||
|
||||
|
||||
# ── 变异体定义 ────────────────────────────────────────────────────────────
|
||||
# (编号, 目标判据, 人类可读描述, 变异函数)
|
||||
def mut_order_revert(src):
|
||||
"""M1 针对 ⑭：**丢掉 key 的时间戳分量** = 退回"纯遍历顺序"⇒ 昨天盖掉今天。
|
||||
|
||||
⚠️ 踩坑记录（这条本身就是变异验证的教训）：第一版写成
|
||||
`key[1] >= last_t[1]`（只把比较从元组换成 seq）⇒ **等价变异、恒绿**。
|
||||
原因：样本里每个文件内部时间戳本来就升序，**任何"取最后一条"的写法都等价**。
|
||||
⇒ 变异必须动**语义**（时间戳参与/不参与），⛔ 不能只动写法。
|
||||
"""
|
||||
old = " key = (ts if ts is not None else 0.0, seq)"
|
||||
new = " key = (0.0, seq) # MUTANT: 丢掉时间戳 = 退回纯遍历顺序"
|
||||
assert old in src, "锚点1 未命中"
|
||||
return src.replace(old, new, 1)
|
||||
|
||||
|
||||
def mut_freshness_off(src):
|
||||
"""M2 针对 ⑮：把新鲜度闸阈值放大到"永不过期"（缺陷原样）。"""
|
||||
old = " if last_t[0] > 0 and (time.time() - last_t[0]) > SRSM_FRESH:"
|
||||
new = " if False: # MUTANT: 关掉新鲜度闸"
|
||||
assert old in src, "锚点2 未命中"
|
||||
return src.replace(old, new, 1)
|
||||
|
||||
|
||||
def mut_tz_local(src):
|
||||
"""M3 针对 ⑰：时间戳改按**本地时区**解析（mktime）⇒ 东八区差 8h ⇒ 恒判过期。"""
|
||||
old = " t = float(calendar.timegm((int(m.group(\"Y\")), int(m.group(\"M\")), int(m.group(\"D\")),\n int(m.group(\"h\")), int(m.group(\"m\")), int(m.group(\"s\")), 0, 0, 0)))"
|
||||
new = (" import calendar as _c # MUTANT: 按本地时区解析\n"
|
||||
" _st = (int(m.group(\"Y\")), int(m.group(\"M\")), int(m.group(\"D\")),\n"
|
||||
" int(m.group(\"h\")), int(m.group(\"m\")), int(m.group(\"s\")), 0, 0, -1)\n"
|
||||
" t = float(time.mktime(_st))")
|
||||
assert old in src, "锚点3 未命中"
|
||||
return src.replace(old, new, 1)
|
||||
|
||||
|
||||
def mut_tz_double_offset(src):
|
||||
"""M9 针对 ⑰：额外+8h 偏移（另一种时区错法）⇒ 「2 分钟前」变「8 小时后」⇒ 恒不过期。"""
|
||||
old = " ms = m.group(\"ms\")\n return t + (float(\"0.\" + ms) if ms else 0.0)"
|
||||
new = (" ms = m.group(\"ms\")\n"
|
||||
" return t + 28800.0 + (float(\"0.\" + ms) if ms else 0.0) # MUTANT: 多加 8h")
|
||||
assert old in src, "锚点9 未命中"
|
||||
return src.replace(old, new, 1)
|
||||
|
||||
|
||||
def mut_freshness_too_hard(src):
|
||||
"""M4 针对 ⑯：新鲜度闸写太狠（窗口=0）⇒ 连正在跑的也判 idle（反向回归）。"""
|
||||
old = "(time.time() - last_t[0]) > SRSM_FRESH:"
|
||||
new = "(time.time() - last_t[0]) > 0: # MUTANT: 闸写太狠"
|
||||
assert old in src, "锚点4 未命中"
|
||||
return src.replace(old, new, 1)
|
||||
|
||||
|
||||
def mut_ts_none(src):
|
||||
"""M5 针对 ⑰：时间戳一律返回 None（解析失败）⇒ 保守当 busy ⇒ ⑯ 报红。"""
|
||||
old = " m = _CONV_TS.match(ln)"
|
||||
new = " return None # MUTANT: 一律解析失败\n m = _CONV_TS.match(ln)"
|
||||
assert old in src, "锚点5 未命中"
|
||||
return src.replace(old, new, 1)
|
||||
|
||||
|
||||
def mut_busy_always(src):
|
||||
"""M8 针对 ⑯：`idle` 也被当成 busy（陈旧闸写太狠的另一种写法）⇒ ⑯/⑭ 报红。"""
|
||||
old = " if last_to in (\"working\", \"planning\"):"
|
||||
new = " if True: # MUTANT: 不分状态一律当在跑"
|
||||
assert old in src, "锚点8 未命中"
|
||||
return src.replace(old, new, 1)
|
||||
|
||||
|
||||
def mut_order_reverse_files(src):
|
||||
"""M6 针对 ⑭：把 `_conv_log` 的返回顺序反过来（[昨天, 今天]）⇒ 顺序依赖立刻现形。"""
|
||||
old = " for cand in (\"%s.log\" % sid, \"%s.log\" % str(sid).replace(\"-\", \"\")):"
|
||||
new = " for cand in (\"%s.log\" % sid,): # MUTANT: 去掉无连字符形态"
|
||||
assert old in src, "锚点6 未命中"
|
||||
s = src.replace(old, new, 1)
|
||||
return s.replace(" for d in (time.strftime(\"%Y-%m-%d\"),\n time.strftime(\"%Y-%m-%d\", time.localtime(time.time() - 86400))):",
|
||||
" for d in (time.strftime(\"%Y-%m-%d\", time.localtime(time.time() - 86400)),\n time.strftime(\"%Y-%m-%d\")):", 1)
|
||||
|
||||
|
||||
def mut_freshness_only_first_file(src):
|
||||
"""M7 针对 ⑮：陈旧判定只在**第一个文件**生效 ⇒ 单文件陈旧样本仍判 busy（⑮ 报红）。"""
|
||||
old = " if last_t[0] > 0 and (time.time() - last_t[0]) > SRSM_FRESH:"
|
||||
new = " if last_t[0] > 0 and (time.time() - last_t[0]) > SRSM_FRESH and len(_conv_log(sid)) > 1: # MUTANT: 陈旧判定只在多文件时生效"
|
||||
assert old in src, "锚点7 未命中"
|
||||
return src.replace(old, new, 1)
|
||||
|
||||
|
||||
MUTS = [
|
||||
("M1", "⑭跨日取最新", mut_order_revert),
|
||||
("M2", "⑮新鲜度闸", mut_freshness_off),
|
||||
("M3", "⑰UTC时区", mut_tz_local),
|
||||
("M4", "⑯反向回归", mut_freshness_too_hard),
|
||||
("M5", "⑰时间戳解析", mut_ts_none),
|
||||
("M6", "⑭跨日取最新", mut_order_reverse_files),
|
||||
("M7", "⑮新鲜度闸", mut_freshness_only_first_file),
|
||||
("M8", "⑯反向回归", mut_busy_always),
|
||||
("M9", "⑰UTC时区", mut_tz_double_offset),
|
||||
]
|
||||
|
||||
|
||||
def run_selftest(workdir):
|
||||
"""⚠️ **必须用工作目录里的那份 `selftest.py`**，⛔ 不能用技能目录的原版绝对路径。
|
||||
|
||||
🔴🔴 2026-10-04 当场栽过（写在这里当护栏，别删）：`selftest.py` 里
|
||||
`HERE = Path(__file__).resolve().parent`，而 `imp()` 加载的是 **`HERE / "collabd.py"`**
|
||||
⇒ 只改 `cwd` 没用、只传原版绝对路径也没用 ⇒ **测的一直是原版 ⇒ 恒绿 ⇒ 7/7 全是假象**。
|
||||
⇒ 判据：`HERE` 指向 `workdir` 时才算数（下面 `assert` 硬查）。
|
||||
"""
|
||||
st = os.path.join(workdir, "selftest.py")
|
||||
assert os.path.abspath(os.path.dirname(st)) == os.path.abspath(workdir), "测的不是工作目录里的 selftest"
|
||||
p = subprocess.run([PY, st, "-k", "忙判据"],
|
||||
capture_output=True, text=True, encoding="utf-8",
|
||||
errors="replace", cwd=workdir, timeout=600)
|
||||
out = p.stdout + p.stderr
|
||||
# 🔴 护栏：确认真加载的是 workdir 里那份（防止以后再有人改回绝对路径）
|
||||
assert workdir.replace("\\", "/") in out.replace("\\", "/"), "输出里看不到工作目录 ⇒ 加载的不是变异体"
|
||||
return out
|
||||
|
||||
|
||||
def verdict(out):
|
||||
"""返回 (是否报红, 是否真找到用例)。⛔ 没找到用例 = 验证作废，不算绿。"""
|
||||
ran = ("忙判据" in out) and ("投递忙判据" in out)
|
||||
if not ran:
|
||||
return None, False
|
||||
# 该用例整体 FAIL ⇔ 汇总行 PASS 0/ FAIL >=1，或用例行前是 ✗
|
||||
m = re.search(r"合计：PASS (\d+) / FAIL (\d+)", out)
|
||||
if not m:
|
||||
return None, True
|
||||
return (int(m.group(2)) >= 1), True
|
||||
|
||||
|
||||
def main():
|
||||
print("=" * 74)
|
||||
print("变异验证 · 会话级忙判据（每个新判据 ≥2 个变异体，⛔ 跳过计入失败）")
|
||||
print("=" * 74)
|
||||
orig_src = open(COLLABD, "rb").read().decode("utf-8")
|
||||
orig_md5 = md5(COLLABD)
|
||||
work = tempfile.mkdtemp(prefix="mutrun-")
|
||||
shutil.copy2(COLLABD, os.path.join(work, "collabd.py"))
|
||||
shutil.copy2(SELFTEST, os.path.join(work, "selftest.py"))
|
||||
|
||||
# 替身：把 collabd 换成变异体
|
||||
with open(os.path.join(work, "collabd.py"), "w", encoding="utf-8") as f:
|
||||
f.write(orig_src)
|
||||
|
||||
base_out = run_selftest(work)
|
||||
base_red, base_ran = verdict(base_out)
|
||||
print("\n【基线】mutant 未植入时的原版：")
|
||||
print(" 找到用例=%s 报红=%s => %s"
|
||||
% (base_ran, base_red, "✅ 绿（可谈抓得住）" if (base_ran and not base_red) else "❌ 基线不绿，验证作废"))
|
||||
if not (base_ran and not base_red):
|
||||
print("\n❌ 基线不绿 ⇒ 停在此处（恒红比漏网更坏：会让所有变异都'红'，验证作废）。")
|
||||
return 2
|
||||
|
||||
results = []
|
||||
for mid, target, fn in MUTS:
|
||||
# 还原原版再植入
|
||||
with open(os.path.join(work, "collabd.py"), "w", encoding="utf-8") as f:
|
||||
f.write(orig_src)
|
||||
try:
|
||||
mutated = fn(orig_src)
|
||||
except AssertionError as e:
|
||||
results.append((mid, target, "锚点未命中:%s" % e, False))
|
||||
print(" ⛔ %-3s %-12s 锚点未命中 ⇒ 计入失败" % (mid, target))
|
||||
continue
|
||||
# ③ 验「被检对象真被改了」：长度差 + md5 + 语法
|
||||
if mutated == orig_src:
|
||||
results.append((mid, target, "变异体与原版相同（replace 没生效）", False))
|
||||
print(" ⛔ %-3s %-12s 真没被改 ⇒ 计入失败" % (mid, target))
|
||||
continue
|
||||
with open(os.path.join(work, "collabd.py"), "w", encoding="utf-8") as f:
|
||||
f.write(mutated)
|
||||
dlen = len(mutated) - len(orig_src)
|
||||
mm = md5(os.path.join(work, "collabd.py"))
|
||||
cp = subprocess.run([PY, "-m", "py_compile", os.path.join(work, "collabd.py")],
|
||||
capture_output=True, text=True, encoding="utf-8")
|
||||
changed = (mm != orig_md5) and (dlen != 0)
|
||||
if not changed:
|
||||
results.append((mid, target, "被检对象未真改", False))
|
||||
print(" ⛔ %-3s %-12s md5/长度未变 ⇒ 计入失败" % (mid, target))
|
||||
continue
|
||||
if cp.returncode != 0:
|
||||
results.append((mid, target, "变异体语法错（测的是崩溃不是判据）", False))
|
||||
print(" ⛔ %-3s %-12s 语法错 ⇒ 计入失败" % (mid, target))
|
||||
continue
|
||||
out = run_selftest(work)
|
||||
# 🔴🔴 最关键的一道：**确认被检代码里真的有变异标记**。
|
||||
# 上一轮 7/7 全"假绿"就是栽在这：变异体压根没被加载，测的是原版。
|
||||
# ⛔ 只看"跑绿/跑红"不够 —— 必须证明**跑的那份**含变异。
|
||||
marker = "# MUTANT:"
|
||||
if marker not in mutated:
|
||||
results.append((mid, target, "变异体缺标记（无法证明被加载）", False))
|
||||
print(" ⛔ %-3s %-12s 变异体无标记 ⇒ 计入失败" % (mid, target))
|
||||
continue
|
||||
if marker not in open(os.path.join(work, "collabd.py"), encoding="utf-8").read():
|
||||
results.append((mid, target, "工作目录里没有变异体", False))
|
||||
print(" ⛔ %-3s %-12s 工作目录无变异体 ⇒ 计入失败" % (mid, target))
|
||||
continue
|
||||
red, ran = verdict(out)
|
||||
if red is None:
|
||||
results.append((mid, target, "没跑到用例（验证作废）", False))
|
||||
print(" ⛔ %-3s %-12s 没跑到用例 ⇒ 计入失败" % (mid, target))
|
||||
continue
|
||||
ok = bool(red)
|
||||
results.append((mid, target, "报红" if ok else "❌未报红(假绿)", ok))
|
||||
print(" %s %-3s %-12s 真改(len%+d) ⇒ %s"
|
||||
% ("✅" if ok else "❌", mid, target, dlen, "报红 ✅" if ok else "未报红 ❌ 假绿"))
|
||||
|
||||
# 还原
|
||||
with open(os.path.join(work, "collabd.py"), "w", encoding="utf-8") as f:
|
||||
f.write(orig_src)
|
||||
|
||||
print("\n" + "-" * 74)
|
||||
npass = sum(1 for r in results if r[3])
|
||||
ntot = len(results)
|
||||
print("变异汇总：%d/%d 报红" % (npass, ntot))
|
||||
# 每个新判据的覆盖
|
||||
cov = {}
|
||||
for mid, target, msg, ok in results:
|
||||
cov.setdefault(target, []).append((mid, ok))
|
||||
print("\n每个新判据的变异体覆盖（≥2 个）：")
|
||||
for target, lst in cov.items():
|
||||
okc = sum(1 for _, o in lst if o)
|
||||
flag = "✅" if len(lst) >= 2 and okc == len(lst) else "❌"
|
||||
print(" %s %-12s 变异体 %d 个，报红 %d 个 %s"
|
||||
% (flag, target, len(lst), okc, [m for m, _ in lst]))
|
||||
shutil.rmtree(work, ignore_errors=True)
|
||||
print("\n" + ("✅ 全部变异体按预期报红" if npass == ntot and ntot >= 2 else "❌ 有变异体未报红/被跳过"))
|
||||
return 0 if npass == ntot else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# 🔴 变异验证 · 2026-10-04 · 会话级忙判据两个新缺陷
|
||||
#
|
||||
# ⚠️ 纪律（来自 references/00-动手前必过.md ③④）：
|
||||
# ① **先验基线绿**：基线 PASS 17/FAIL 0（忙判据那条用例）。
|
||||
# ② **锚在唯一位置**：`def _session_busy_conv` / `def _conv_ts` 唯一定位，⛔ 不锚"文本出现过"。
|
||||
# ③ **每个变异先验「被检对象真被改了」**：比对长度+md5，⛔ 没真改就算**失败**（不许跳过分母）。
|
||||
# ④ ⛔ **跳过的变异一律计入失败**。
|
||||
#
|
||||
# 用法：python mut_run.py
|
||||
import hashlib
|
||||
import importlib.util
|
||||
import io
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
COLLABD = os.path.join(HERE, "collabd.py")
|
||||
SELFTEST = os.path.join(HERE, "selftest.py")
|
||||
PY = sys.executable
|
||||
|
||||
|
||||
def md5(p):
|
||||
return hashlib.md5(open(p, "rb").read()).hexdigest()
|
||||
|
||||
|
||||
# ── 变异体定义 ────────────────────────────────────────────────────────────
|
||||
# (编号, 目标判据, 人类可读描述, 变异函数)
|
||||
def mut_order_revert(src):
|
||||
"""M1 针对 ⑭：**丢掉 key 的时间戳分量** = 退回"纯遍历顺序"⇒ 昨天盖掉今天。
|
||||
|
||||
⚠️ 踩坑记录（这条本身就是变异验证的教训）：第一版写成
|
||||
`key[1] >= last_t[1]`（只把比较从元组换成 seq）⇒ **等价变异、恒绿**。
|
||||
原因：样本里每个文件内部时间戳本来就升序，**任何"取最后一条"的写法都等价**。
|
||||
⇒ 变异必须动**语义**（时间戳参与/不参与），⛔ 不能只动写法。
|
||||
"""
|
||||
old = " key = (ts if ts is not None else 0.0, seq)"
|
||||
new = " key = (0.0, seq) # MUTANT: 丢掉时间戳 = 退回纯遍历顺序"
|
||||
assert old in src, "锚点1 未命中"
|
||||
return src.replace(old, new, 1)
|
||||
|
||||
|
||||
def mut_freshness_off(src):
|
||||
"""M2 针对 ⑮：把新鲜度闸阈值放大到"永不过期"（缺陷原样）。"""
|
||||
old = " if last_t[0] > 0 and (time.time() - last_t[0]) > SRSM_FRESH:"
|
||||
new = " if False: # MUTANT: 关掉新鲜度闸"
|
||||
assert old in src, "锚点2 未命中"
|
||||
return src.replace(old, new, 1)
|
||||
|
||||
|
||||
def mut_tz_local(src):
|
||||
"""M3 针对 ⑰：时间戳改按**本地时区**解析（mktime）⇒ 东八区差 8h ⇒ 恒判过期。"""
|
||||
old = " t = float(calendar.timegm((int(m.group(\"Y\")), int(m.group(\"M\")), int(m.group(\"D\")),\n int(m.group(\"h\")), int(m.group(\"m\")), int(m.group(\"s\")), 0, 0, 0)))"
|
||||
new = (" import calendar as _c # MUTANT: 按本地时区解析\n"
|
||||
" _st = (int(m.group(\"Y\")), int(m.group(\"M\")), int(m.group(\"D\")),\n"
|
||||
" int(m.group(\"h\")), int(m.group(\"m\")), int(m.group(\"s\")), 0, 0, -1)\n"
|
||||
" t = float(time.mktime(_st))")
|
||||
assert old in src, "锚点3 未命中"
|
||||
return src.replace(old, new, 1)
|
||||
|
||||
|
||||
def mut_tz_double_offset(src):
|
||||
"""M9 针对 ⑰：额外+8h 偏移（另一种时区错法）⇒ 「2 分钟前」变「8 小时后」⇒ 恒不过期。"""
|
||||
old = " ms = m.group(\"ms\")\n return t + (float(\"0.\" + ms) if ms else 0.0)"
|
||||
new = (" ms = m.group(\"ms\")\n"
|
||||
" return t + 28800.0 + (float(\"0.\" + ms) if ms else 0.0) # MUTANT: 多加 8h")
|
||||
assert old in src, "锚点9 未命中"
|
||||
return src.replace(old, new, 1)
|
||||
|
||||
|
||||
def mut_freshness_too_hard(src):
|
||||
"""M4 针对 ⑯：新鲜度闸写太狠（窗口=0）⇒ 连正在跑的也判 idle（反向回归）。"""
|
||||
old = "(time.time() - last_t[0]) > SRSM_FRESH:"
|
||||
new = "(time.time() - last_t[0]) > 0: # MUTANT: 闸写太狠"
|
||||
assert old in src, "锚点4 未命中"
|
||||
return src.replace(old, new, 1)
|
||||
|
||||
|
||||
def mut_ts_none(src):
|
||||
"""M5 针对 ⑰：时间戳一律返回 None（解析失败）⇒ 保守当 busy ⇒ ⑯ 报红。"""
|
||||
old = " m = _CONV_TS.match(ln)"
|
||||
new = " return None # MUTANT: 一律解析失败\n m = _CONV_TS.match(ln)"
|
||||
assert old in src, "锚点5 未命中"
|
||||
return src.replace(old, new, 1)
|
||||
|
||||
|
||||
def mut_busy_always(src):
|
||||
"""M8 针对 ⑯：`idle` 也被当成 busy（陈旧闸写太狠的另一种写法）⇒ ⑯/⑭ 报红。"""
|
||||
old = " if last_to in (\"working\", \"planning\"):"
|
||||
new = " if True: # MUTANT: 不分状态一律当在跑"
|
||||
assert old in src, "锚点8 未命中"
|
||||
return src.replace(old, new, 1)
|
||||
|
||||
|
||||
def mut_order_reverse_files(src):
|
||||
"""M6 针对 ⑭：把 `_conv_log` 的返回顺序反过来（[昨天, 今天]）⇒ 顺序依赖立刻现形。"""
|
||||
old = " for cand in (\"%s.log\" % sid, \"%s.log\" % str(sid).replace(\"-\", \"\")):"
|
||||
new = " for cand in (\"%s.log\" % sid,): # MUTANT: 去掉无连字符形态"
|
||||
assert old in src, "锚点6 未命中"
|
||||
s = src.replace(old, new, 1)
|
||||
return s.replace(" for d in (time.strftime(\"%Y-%m-%d\"),\n time.strftime(\"%Y-%m-%d\", time.localtime(time.time() - 86400))):",
|
||||
" for d in (time.strftime(\"%Y-%m-%d\", time.localtime(time.time() - 86400)),\n time.strftime(\"%Y-%m-%d\")):", 1)
|
||||
|
||||
|
||||
def mut_freshness_only_first_file(src):
|
||||
"""M7 针对 ⑮：陈旧判定只在**第一个文件**生效 ⇒ 单文件陈旧样本仍判 busy（⑮ 报红）。"""
|
||||
old = " if last_t[0] > 0 and (time.time() - last_t[0]) > SRSM_FRESH:"
|
||||
new = " if last_t[0] > 0 and (time.time() - last_t[0]) > SRSM_FRESH and len(_conv_log(sid)) > 1: # MUTANT: 陈旧判定只在多文件时生效"
|
||||
assert old in src, "锚点7 未命中"
|
||||
return src.replace(old, new, 1)
|
||||
|
||||
|
||||
MUTS = [
|
||||
("M1", "⑭跨日取最新", mut_order_revert),
|
||||
("M2", "⑮新鲜度闸", mut_freshness_off),
|
||||
("M3", "⑰UTC时区", mut_tz_local),
|
||||
("M4", "⑯反向回归", mut_freshness_too_hard),
|
||||
("M5", "⑰时间戳解析", mut_ts_none),
|
||||
("M6", "⑭跨日取最新", mut_order_reverse_files),
|
||||
("M7", "⑮新鲜度闸", mut_freshness_only_first_file),
|
||||
("M8", "⑯反向回归", mut_busy_always),
|
||||
("M9", "⑰UTC时区", mut_tz_double_offset),
|
||||
]
|
||||
|
||||
|
||||
def run_selftest(workdir):
|
||||
"""⚠️ **必须用工作目录里的那份 `selftest.py`**，⛔ 不能用技能目录的原版绝对路径。
|
||||
|
||||
🔴🔴 2026-10-04 当场栽过（写在这里当护栏，别删）：`selftest.py` 里
|
||||
`HERE = Path(__file__).resolve().parent`，而 `imp()` 加载的是 **`HERE / "collabd.py"`**
|
||||
⇒ 只改 `cwd` 没用、只传原版绝对路径也没用 ⇒ **测的一直是原版 ⇒ 恒绿 ⇒ 7/7 全是假象**。
|
||||
⇒ 判据：`HERE` 指向 `workdir` 时才算数（下面 `assert` 硬查）。
|
||||
"""
|
||||
st = os.path.join(workdir, "selftest.py")
|
||||
assert os.path.abspath(os.path.dirname(st)) == os.path.abspath(workdir), "测的不是工作目录里的 selftest"
|
||||
p = subprocess.run([PY, st, "-k", "忙判据"],
|
||||
capture_output=True, text=True, encoding="utf-8",
|
||||
errors="replace", cwd=workdir, timeout=600)
|
||||
out = p.stdout + p.stderr
|
||||
# 🔴 护栏：确认真加载的是 workdir 里那份（防止以后再有人改回绝对路径）
|
||||
assert workdir.replace("\\", "/") in out.replace("\\", "/"), "输出里看不到工作目录 ⇒ 加载的不是变异体"
|
||||
return out
|
||||
|
||||
|
||||
def verdict(out):
|
||||
"""返回 (是否报红, 是否真找到用例)。⛔ 没找到用例 = 验证作废，不算绿。"""
|
||||
ran = ("忙判据" in out) and ("投递忙判据" in out)
|
||||
if not ran:
|
||||
return None, False
|
||||
# 该用例整体 FAIL ⇔ 汇总行 PASS 0/ FAIL >=1，或用例行前是 ✗
|
||||
m = re.search(r"合计：PASS (\d+) / FAIL (\d+)", out)
|
||||
if not m:
|
||||
return None, True
|
||||
return (int(m.group(2)) >= 1), True
|
||||
|
||||
|
||||
def main():
|
||||
print("=" * 74)
|
||||
print("变异验证 · 会话级忙判据（每个新判据 ≥2 个变异体，⛔ 跳过计入失败）")
|
||||
print("=" * 74)
|
||||
orig_src = open(COLLABD, "rb").read().decode("utf-8")
|
||||
orig_md5 = md5(COLLABD)
|
||||
work = tempfile.mkdtemp(prefix="mutrun-")
|
||||
shutil.copy2(COLLABD, os.path.join(work, "collabd.py"))
|
||||
shutil.copy2(SELFTEST, os.path.join(work, "selftest.py"))
|
||||
|
||||
# 替身：把 collabd 换成变异体
|
||||
with open(os.path.join(work, "collabd.py"), "w", encoding="utf-8") as f:
|
||||
f.write(orig_src)
|
||||
|
||||
base_out = run_selftest(work)
|
||||
base_red, base_ran = verdict(base_out)
|
||||
print("\n【基线】mutant 未植入时的原版：")
|
||||
print(" 找到用例=%s 报红=%s => %s"
|
||||
% (base_ran, base_red, "✅ 绿（可谈抓得住）" if (base_ran and not base_red) else "❌ 基线不绿，验证作废"))
|
||||
if not (base_ran and not base_red):
|
||||
print("\n❌ 基线不绿 ⇒ 停在此处（恒红比漏网更坏：会让所有变异都'红'，验证作废）。")
|
||||
return 2
|
||||
|
||||
results = []
|
||||
for mid, target, fn in MUTS:
|
||||
# 还原原版再植入
|
||||
with open(os.path.join(work, "collabd.py"), "w", encoding="utf-8") as f:
|
||||
f.write(orig_src)
|
||||
try:
|
||||
mutated = fn(orig_src)
|
||||
except AssertionError as e:
|
||||
results.append((mid, target, "锚点未命中:%s" % e, False))
|
||||
print(" ⛔ %-3s %-12s 锚点未命中 ⇒ 计入失败" % (mid, target))
|
||||
continue
|
||||
# ③ 验「被检对象真被改了」：长度差 + md5 + 语法
|
||||
if mutated == orig_src:
|
||||
results.append((mid, target, "变异体与原版相同（replace 没生效）", False))
|
||||
print(" ⛔ %-3s %-12s 真没被改 ⇒ 计入失败" % (mid, target))
|
||||
continue
|
||||
with open(os.path.join(work, "collabd.py"), "w", encoding="utf-8") as f:
|
||||
f.write(mutated)
|
||||
dlen = len(mutated) - len(orig_src)
|
||||
mm = md5(os.path.join(work, "collabd.py"))
|
||||
cp = subprocess.run([PY, "-m", "py_compile", os.path.join(work, "collabd.py")],
|
||||
capture_output=True, text=True, encoding="utf-8")
|
||||
changed = (mm != orig_md5) and (dlen != 0)
|
||||
if not changed:
|
||||
results.append((mid, target, "被检对象未真改", False))
|
||||
print(" ⛔ %-3s %-12s md5/长度未变 ⇒ 计入失败" % (mid, target))
|
||||
continue
|
||||
if cp.returncode != 0:
|
||||
results.append((mid, target, "变异体语法错（测的是崩溃不是判据）", False))
|
||||
print(" ⛔ %-3s %-12s 语法错 ⇒ 计入失败" % (mid, target))
|
||||
continue
|
||||
out = run_selftest(work)
|
||||
# 🔴🔴 最关键的一道：**确认被检代码里真的有变异标记**。
|
||||
# 上一轮 7/7 全"假绿"就是栽在这：变异体压根没被加载，测的是原版。
|
||||
# ⛔ 只看"跑绿/跑红"不够 —— 必须证明**跑的那份**含变异。
|
||||
marker = "# MUTANT:"
|
||||
if marker not in mutated:
|
||||
results.append((mid, target, "变异体缺标记（无法证明被加载）", False))
|
||||
print(" ⛔ %-3s %-12s 变异体无标记 ⇒ 计入失败" % (mid, target))
|
||||
continue
|
||||
if marker not in open(os.path.join(work, "collabd.py"), encoding="utf-8").read():
|
||||
results.append((mid, target, "工作目录里没有变异体", False))
|
||||
print(" ⛔ %-3s %-12s 工作目录无变异体 ⇒ 计入失败" % (mid, target))
|
||||
continue
|
||||
red, ran = verdict(out)
|
||||
if red is None:
|
||||
results.append((mid, target, "没跑到用例（验证作废）", False))
|
||||
print(" ⛔ %-3s %-12s 没跑到用例 ⇒ 计入失败" % (mid, target))
|
||||
continue
|
||||
ok = bool(red)
|
||||
results.append((mid, target, "报红" if ok else "❌未报红(假绿)", ok))
|
||||
print(" %s %-3s %-12s 真改(len%+d) ⇒ %s"
|
||||
% ("✅" if ok else "❌", mid, target, dlen, "报红 ✅" if ok else "未报红 ❌ 假绿"))
|
||||
|
||||
# 还原
|
||||
with open(os.path.join(work, "collabd.py"), "w", encoding="utf-8") as f:
|
||||
f.write(orig_src)
|
||||
|
||||
print("\n" + "-" * 74)
|
||||
npass = sum(1 for r in results if r[3])
|
||||
ntot = len(results)
|
||||
print("变异汇总：%d/%d 报红" % (npass, ntot))
|
||||
# 每个新判据的覆盖
|
||||
cov = {}
|
||||
for mid, target, msg, ok in results:
|
||||
cov.setdefault(target, []).append((mid, ok))
|
||||
print("\n每个新判据的变异体覆盖（≥2 个）：")
|
||||
for target, lst in cov.items():
|
||||
okc = sum(1 for _, o in lst if o)
|
||||
flag = "✅" if len(lst) >= 2 and okc == len(lst) else "❌"
|
||||
print(" %s %-12s 变异体 %d 个，报红 %d 个 %s"
|
||||
% (flag, target, len(lst), okc, [m for m, _ in lst]))
|
||||
shutil.rmtree(work, ignore_errors=True)
|
||||
print("\n" + ("✅ 全部变异体按预期报红" if npass == ntot and ntot >= 2 else "❌ 有变异体未报红/被跳过"))
|
||||
return 0 if npass == ntot else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in new issue
Block a user