Files
admin c1b5e4d966 chore(工作区): 全量入库 + 补齐 .gitignore(以工作区为准)
- 变更规模:新增 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/ 知识文件,按口径入库)
2026-10-10 23:13:22 +08:00

3824 lines
237 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- coding: utf-8 -*-
"""selftest —— 「多会话协作机制」的**回归测试**:把**所有异常情况**做成用例,每次改完跑一遍。
🔴 纪律(2026-09-29 用户定案):**每次改完 collabd.py / guard.py,必须跑本文件**,全绿才算改完。
设计原则:
· **不碰生产**:整套测试用 `COLLABD_CONFIG` 指向 `tmp/selftest/`(独立 inbox / 看板 / 任务图),
生产台账与看板**一个字节都不动**。
· **离线可跑**:需要宿主库的用例,读不到就记 SKIP(⛔ 不假装通过)。
· **判据是"读数"不是"没报错"**:rc、文件是否生成、台账状态、投递台账增量 …
· FAIL ⇒ **非零退出**(配合"rc=0 才是真成功")。
用法:python selftest.py # 全部
python selftest.py -k 投递 # 只跑名字含"投递"的用例
"""
from __future__ import annotations
# [session-mechanism] roots.env 外置(由 install.py 生成;缺失则回落到按位置推导)
def _sm_load_roots():
import os as _os
_here = _os.path.dirname(_os.path.abspath(__file__))
for _up in range(4):
_p = _os.path.join(_here, *([".."] * _up), "roots.env")
_p = _os.path.normpath(_p)
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()
# 🔴 2026-10-01 加 · **输出编码兜底**:脚本一旦被重定向(钩子/常驻/后台任务都会这么干),
# Windows 本地编码(GBK)编不出 ⛔/✅/🔴 这类字符 ⇒ `print` 抛 UnicodeEncodeError
# ⇒ 被顶层 handler 记成 `fatal`、**整轮失败**(实测:本包里连续 4 次 `fatal 'gbk' codec ...`)。
# ⇒ 出口一律 UTF-8 + errors="replace"(⛔ 不让"打不出字"升级成"程序死")。
try:
import sys as _sys
_sys.stdout.reconfigure(encoding="utf-8", errors="replace")
_sys.stderr.reconfigure(encoding="utf-8", errors="replace")
except Exception:
pass
import json
import os
import re
import shutil
import subprocess
import sys
import time
from pathlib import Path
HERE = Path(__file__).resolve().parent
WS = Path(os.environ.get("DSH_COLLAB_WS") or os.environ.get("DSH_WS_ROOT") or Path.cwd())
TEST_WS = WS / "tmp" / "selftest"
TIN = TEST_WS / "inbox"
CFG = TEST_WS / "collabd.config.json"
PY = sys.executable
RESULTS = []
def _mk_env() -> dict:
env = dict(os.environ)
env["COLLABD_CONFIG"] = str(CFG)
env["DSH_COLLAB_WS"] = str(TEST_WS)
# 🔴 2026-10-02(S8):**测试环境里⛔ 不许起常驻**。
# 真因(本棒实测):`--tick` 现在会顺手"续命常驻"(`ensure_supervise()`),而自测用的
# `--tick` 继承了真环境的网关口令 ⇒ 它会在**测试工作区**里起一条真的 `--supervise`
# ⇒ 那条进程会持续 `supervise()` 并**重写测试夹具**(实测:`已停总闸` 用例报
# `告警投影被清掉` 假红、且 FAIL 数每轮不同)+ 留下僵尸测试进程。
# ⇒ 显式关掉(⛔ 不是"把断言写松")。
env["COLLABD_NO_ENSURE"] = "1"
return env
def _prepare() -> None:
TIN.mkdir(parents=True, exist_ok=True)
for _f in ("wake.lock", "guard.stop", "goal.pending.json",
# 🔴 2026-10-01 加:退场表也每轮清 —— 它会**跨轮留下来**,
# 而退场用例断言的是精确集合(`retired==1 / stale==1`)⇒
# 上一轮留下的那条会在下一轮把计数顶成 2 ⇒ **第二次跑才红的假红**(最难查的一型)。
"retired.json"): # 清上一轮残留(防跨轮假红)
try:
(TIN / _f).unlink()
except Exception:
pass
# 🔴 2026-10-01 加:**假配置根(假的会话日志)也每轮清空** —— 这棵树的文件会跨轮留下来,
# 而「哑会话」用例断言的是**精确集合**(`d == {DEAF}`)⇒ 上一轮留下的昨天那份 10 MiB
# 会在下一轮把结果多出几条 ⇒ **第二次跑才红的假红**(最难查的一型)。⛔ 只删测试区内的。
try:
shutil.rmtree(TEST_WS / "fake-cfg", ignore_errors=True)
except Exception:
pass
(TEST_WS / "live.md").write_text("(selftest)", encoding="utf-8")
(TEST_WS / "tg.json").write_text(json.dumps({"nodes": [
{"id": "T1", "title": "测试件-1", "status": "todo", "line": "line-a"},
{"id": "T2", "title": "测试件-2", "status": "done", "line": "line-b"},
]}, ensure_ascii=False), encoding="utf-8")
CFG.write_text(json.dumps({
"_说明": "selftest 专用(⛔ 不碰生产)",
"workspace": str(TEST_WS).replace("\\", "/"),
"inbox": "inbox",
"live": "live.md",
"taskgraph": "tg.json",
"lines": {"line-a": "A 线", "line-b": "B 线"},
"goal_docs": [], "targets": {},
"host_db": "",
"singleton_port": 20098,
"interval": 5, "idle_min": 12, "stuck_min": 30, "vacuum_min": 5,
"health_every": 300, "wake_enable": True, "wake_min_gap": 300, "wake_max_ports": 4,
"wake_text": "(selftest)",
}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
def case(name):
def deco(fn):
fn._case = name
return fn
return deco
def report(name):
"""**报告型**用例:照常显示、照常报红/报绿,但 ⛔ **不计入 `--verify` 的成败**。
🔴 2026-10-04 新增(口径判据落地时撞到):有些检查**天生依赖现网存量** ——
例「产物是否都落在目标文件夹」在**历史散落的文件清理掉之前**必然红。
⇒ 若让它计入 verify ⇒ **`install.py --verify` 在新机器上永远失败**,
而那个失败与"安装对不对"**毫无关系** ⇒ 判据被当成安装故障 ⇒ ⛔ 误报。
✅ 判据纪律:**"装得对不对"与"存量干不干净"必须分开判**,
后者只**如实报**,不牵动前者(否则一条提示会永久堵住安装)。
"""
def deco(fn):
fn._case = name
fn._report_only = True
return fn
return deco
def run_cli(args, timeout=60):
p = subprocess.run([PY, "-u", str(HERE / "collabd.py")] + args, cwd=str(HERE),
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
env=_mk_env(), timeout=timeout, errors="replace")
return p.returncode, p.stdout or ""
def tasks() -> dict:
try:
return json.loads((TIN / "tasks.json").read_text(encoding="utf-8")) or {}
except Exception:
return {}
def wakeups_n() -> int:
try:
return len((TIN / "wakeups.jsonl").read_text(encoding="utf-8").strip().splitlines())
except Exception:
return 0
def state() -> dict:
try:
return json.loads((TIN / "collabd-state.json").read_text(encoding="utf-8")) or {}
except Exception:
return {}
def put_tasks(t: dict) -> None:
(TIN / "tasks.json").write_text(json.dumps(t, ensure_ascii=False, indent=2), encoding="utf-8")
def run_cli_env(args, env: dict, timeout=60):
"""同 `run_cli`,但**自定义环境**(用例要故意抽掉网关口令等)。"""
p = subprocess.run([PY, "-u", str(HERE / "collabd.py")] + args, cwd=str(HERE),
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
env=env, timeout=timeout, errors="replace")
return p.returncode, p.stdout or ""
_PROD_SNAP = {}
def _snap_prod():
"""记录**生产 inbox** 关键产物的 (mtime,size) —— 元用例据此断言「测试不碰生产」。"""
global _PROD_SNAP
_PROD_SNAP = {}
for f in ("tasks.json", "wakeups.jsonl", "digest.md", "TO-MAIN.md", "wake.lock"):
p = WS / "tmp" / "supervise-inbox" / f
try:
st = p.stat()
_PROD_SNAP[f] = (st.st_mtime, st.st_size)
except Exception:
_PROD_SNAP[f] = None
def imp():
# 🔴 先设环境,**再加载模块** —— 否则模块会去读**生产配置**(本用例曾因此"碰生产"+结果飘)
os.environ["COLLABD_CONFIG"] = str(CFG)
os.environ["DSH_COLLAB_WS"] = str(TEST_WS)
import importlib.util
spec = importlib.util.spec_from_file_location("cb_selftest", HERE / "collabd.py")
m = importlib.util.module_from_spec(spec)
spec.loader.exec_module(m)
return m
# ── 用例(每条都对应一个真实踩过的坑) ─────────────────────────────
@case("命名:两级前缀解析(合规 / 缺级 / 接续会话 / 主控前缀)")
def t_name():
m = imp()
a = m.parse_session_name("[协作]-[手机接入]-N9 复测")
b = m.parse_session_name("[主]-[手机接入]-机制线")
c = m.parse_session_name("[协作]N9-2156") # 只一级 ⇒ 不合规
d = m.parse_session_name("接续 · 机制线(旧式)")
# 🆕 2026-10-01:**接续会话**(会话自己建出来的"下一棒",标题由建它的那条会话写)
e = m.parse_session_name("[唤醒机制] 接续 · 日志事前叫停钩子落地(第 2 棒)")
f = m.parse_session_name("接续棒:日志增长治理(任务 0 → 任务 A)")
# 🆕 **主控前缀**(实测现役主会话就这么写:中点分隔、无方括号)
g = m.parse_session_name("主控 · 协作机制 · 阈值交接收尾")
return [("合规二级=worker/手机接入", (a["role"], a["topic"], a["ok"]) == ("worker", "手机接入", True)),
("合规二级=main", (b["role"], b["topic"], b["ok"]) == ("main", "手机接入", True)),
("只一级 ⇒ 判不合规", c["ok"] is False),
("🔴 只一级 ⇒ 角色仍判 worker(旧写法剥错方括号 ⇒ 靠 `role_of` 兜底才没出事)",
c["role"] == "worker"),
("旧式接续 ⇒ 判不合规", d["ok"] is False),
("🆕 `[<类别>] 接续 · …` ⇒ **worker**,类别取方括号里那一个(⛔ 方括号里不是角色)",
(e["role"], e["topic"], e["form"]) == ("worker", "唤醒机制", "continuation")),
("🆕 无方括号 `接续棒:…` ⇒ **worker**(⛔ 不是主会话)", f["role"] == "worker"),
("🆕 `主控 · …`(无方括号)⇒ **main**(否则真主会话被判「角色未知」)",
(g["role"], g["form"]) == ("main", "main-prefix")),
("🆕 `form` 标出形态;老调用方只看 `ok` ⇒ 兼容(`ok=False` 语义未变)", a["form"] == "prefix")]
@case("命名:看板与协作程序**同一套角色判据**(⛔ 防两处漂移 —— 漂了就没人能发现换主会话)")
def t_role_drift():
"""角色判据在 `collabd.py::parse_session_name()` 与 `board.py::_role_of_title()` **两处**
(2026-09-30 起就是"改一处必须改两处")⇒ 这里**真的把两边拉出来逐样本对账**,⛔ 不靠人盯。"""
m = imp()
import importlib.util
spec = importlib.util.spec_from_file_location("bd_selftest", HERE / "board.py")
bd = importlib.util.module_from_spec(spec)
spec.loader.exec_module(bd)
samples = ["[协作]-[唤醒机制]-日志事前叫停棒", "[唤醒]-[唤醒机制]-脉冲", "[主]-[唤醒机制]-机制线",
"[跟进]-[会话协作自检]-队列上报跟进棒",
"[唤醒机制] 接续 · 第 2 棒", "接续棒:日志增长治理(任务 0 → 任务 A)",
"主控 · 协作机制 · 收尾", "接续 · 机制线(旧式)", "[协作]N9-2156",
# 🔴 2026-10-01 补(**第三处漂移**,扫线上标题时发现的,不是推理出来的):
# `[接续] …` 这种写法**方括号里是「接续」**(⛔ 不是类别)——
# `collabd.is_continuation()` 判 `"接续" in name`(整串)⇒ worker;
# 而 `board._role_of_title()` 原来只查**方括号后面那段** ⇒ 返回**空**。
# 后果:看板的「主会话」位上坐着一条**接续棒**(线上实测 `3aa35bae`,355 分钟没动)。
"[接续] 会话机制合并技能包(第 1 棒)",
"下载 open design", ""]
bad = [t for t in samples if m.parse_session_name(t)["role"] != bd._role_of_title(t)]
sa = (HERE / "collabd.py").read_text(encoding="utf-8")
sb = (HERE / "board.py").read_text(encoding="utf-8")
return [
("两边角色判据逐样本一致(%d 条;不一致:%s)" % (len(samples), bad or "无"), not bad),
("🔴🔴 2026-10-02 两类改造:`[唤醒]`/`[跟进]` 两边**都判空**(已退役)—— ⛔ 判成活角色 ⇒ "
"那些历史标题会被当**主会话候选**(带 `[<类别>]` 时更会变成「该类别的主会话」)⇒ 自指死结",
m.parse_session_name("[唤醒]-[X]-脉冲").get("role") == ""
and bd._role_of_title("[唤醒]-[X]-脉冲") == ""
and m.parse_session_name("[跟进]-[X]-队列上报").get("role") == ""
and bd._role_of_title("[跟进]-[X]-队列上报") == ""),
("🔴🔴 `[接续] …`(**方括号里是「接续」**)两边都判 `worker` —— ⛔ 判空 ⇒ 接续棒"
"会被当**主会话候选** ⇒ 看板「主会话」位上坐着一条干活的棒(线上实测 `3aa35bae`)",
m.parse_session_name("[接续] 会话机制合并技能包(第 1 棒)").get("role") == "worker"
and bd._role_of_title("[接续] 会话机制合并技能包(第 1 棒)") == "worker"),
("🔴 collabd 的主会话候选判据=**只排 `worker`**(2026-10-02 两类改造,`waker`/`follow` 已退役)",
'_role not in ("worker",)' in sa),
("🔴 board 同款", '_role_of_title(t) not in ("worker",)' in sb),
]
@case("目标:无目标 ⇒ 拒绝启动 rc=3")
def t_goal_missing():
g = TIN / "goal.json"
bak = g.read_text(encoding="utf-8") if g.exists() else None
if g.exists():
g.unlink()
p = subprocess.run([PY, "-u", str(HERE / "guard.py")], cwd=str(HERE),
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
env=_mk_env(), timeout=60, errors="replace")
if bak is not None:
g.write_text(bak, encoding="utf-8")
return [("rc==3", p.returncode == 3), ("提示含『没有任务目标』", "没有任务目标" in (p.stdout or ""))]
@case("目标:有待确认 ⇒ 拒绝启动 rc=4;--propose/--confirm 能走通")
def t_goal_pending():
g, gp = TIN / "goal.json", TIN / "goal.pending.json"
bak = g.read_text(encoding="utf-8") if g.exists() else None
if g.exists():
g.unlink()
subprocess.run([PY, "-u", str(HERE / "guard.py"), "--propose", "自测目标"],
cwd=str(HERE), env=_mk_env(), timeout=60,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, errors="replace",
creationflags=0x08000000)
p = subprocess.run([PY, "-u", str(HERE / "guard.py")], cwd=str(HERE),
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
env=_mk_env(), timeout=60, errors="replace")
rc_pending = p.returncode
subprocess.run([PY, "-u", str(HERE / "guard.py"), "--confirm"], cwd=str(HERE), env=_mk_env(),
timeout=60, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, errors="replace",
creationflags=0x08000000)
ok_confirmed = g.exists() and not gp.exists()
if bak is not None:
g.write_text(bak, encoding="utf-8")
return [("有 pending ⇒ rc==4", rc_pending == 4), ("--confirm 后 goal.json 生成且 pending 清除", ok_confirmed)]
@case("目标三态:『没有判据』⛔ 不等于『全过』(防『读不到被当成已完成』)")
def t_goal_undeclared():
"""🔴 2026-09-30 实测踩到(**跑整体验证时发现**):`acceptance_state` 里**只有 `_说明`/`_更新`
两个说明行**时,旧写法 `if not acc`(判**字典空不空**)**永不触发** ⇒ `goalctl status` 显示
「三路全过」。形状同族:**不崩溃,只是少说一句话**。
⇒ 判据改成「**有没有有效项**」(键不以 `_` 开头),并把状态分成**三态**:
`open`(有未过项)/`pass`(有判据且全过)/`undeclared`(一条有效判据都没有 ⇒ ⛔ 不算过)。
"""
m = imp()
gj, tg = TIN / "goal.json", TEST_WS / "tg.json"
bak_g = gj.read_text(encoding="utf-8") if gj.exists() else None
tg_bak, tk_bak = tg.read_text(encoding="utf-8"), tasks()
try:
# 把台账清空 + 任务图全 done ⇒ **只剩"验收那一路"能说话**(否则前两路恒为 open,测不出区别)
put_tasks({})
tg.write_text(json.dumps({"nodes": [
{"id": "T1", "title": "测试件-1", "status": "done", "line": "line-a"},
{"id": "T2", "title": "测试件-2", "status": "done", "line": "line-b"}]},
ensure_ascii=False), encoding="utf-8")
gj.write_text(json.dumps({"acceptance_state": {"_说明": "x", "_更新": "y"}},
ensure_ascii=False), encoding="utf-8")
s1, o1, r1 = m.goal_state(), m.goals_open(), m._acc_short()[0]
gj.write_text(json.dumps({"acceptance_state": {"V1": "pass"}}, ensure_ascii=False), encoding="utf-8")
s2, o2 = m.goal_state(), m.goals_open()
gj.write_text(json.dumps({"acceptance_state": {"V1": "fail"}}, ensure_ascii=False), encoding="utf-8")
s3, o3 = m.goal_state(), m.goals_open()
finally:
tg.write_text(tg_bak, encoding="utf-8")
put_tasks(tk_bak)
if bak_g is not None:
gj.write_text(bak_g, encoding="utf-8")
else:
try:
gj.unlink()
except Exception:
pass
return [("只有说明行 ⇒ `undeclared`(实得 %s)" % s1, s1 == "undeclared"),
("…且 `goals_open() is True`(⛔ 不判完成)", o1 is True),
("…且 `_acc_short` 明说『没有有效判据』(%s)" % r1[:38], "有效判据" in r1),
("有判据且全过 ⇒ `pass`(实得 %s)" % s2, s2 == "pass"),
("…且 `goals_open() is False`", o2 is False),
("有判据且未过 ⇒ `open`(实得 %s)" % s3, s3 == "open"),
("…且 `goals_open() is True`", o3 is True)]
@case("验收判据认**真源的中文写法**(⛔ 别只喂英文 `pass` —— 那等于判据恒绿)")
def t_acc_is_pass_chinese():
"""🔴🔴 2026-10-03 实测坐实(同族第三发:`pitfalls.md` P0-13/P0-20)。
**上一条用例(`t_goal_undeclared`)只喂英文 `{"V1": "pass"}`** ⇒ 判据**恒命中** ⇒ **永远绿**。
而**真源写的全是中文**:`goal.json::acceptance_state` 的值形如
`过|pid 8024 活(…)` / `🔴 不过|…` / `**复核:过(形态已治本…)**`。
⇒ `board.py` 10-02 已修(认中文)但 **`collabd.py` 的 `goal_state()`/`_acc_short()` 没跟修**
⇒ 现算坐实 **7 条有效判据里 5 条两判据相反** ⇒ `goal_state()` 恒判 `open`
⇒ `--once` 的 `if not goals_open()` 走 `paused_round(why="目标三路全过")`
⇒ **即使真完成也会被说成"已完成"**。
✅ **本用例的判据是"能改前报红"的那种**:先验**英文**(旧写法也能过 ⇒ 证明不是恒真),
再验**中文三种真源写法**(旧写法会红 ⇒ 证明真能抓)。
"""
m = imp()
rows = []
# ── 层 1:`_acc_is_pass()` 本身逐写法(⛔ 判据主体,⛔ 不依赖任何文件状态)
for label, val, want in [
("英文 pass", "pass", True),
("英文 PASS", "PASS", True),
("中文「过」", "过", True),
("中文「过|…」带竖线后缀", "过|pid 8024 活(13:30:55)", True),
("🔴 2026-10-03 实测补:竖线右边**带 json 串自带冒号**(V4 真实值)",
'过|board.py --serve 8788 起得来,body={"ok":true,"err":""};探针结束后端口已释放', True),
("中文「通过」", "通过", True),
("「复核:过」判定词不在开头", "**复核:过(形态已治本)**", True),
("中文「不过|…」⇒ 未过", "🔴 不过|零条活着", False),
("「待重验」⇒ 未过", "待重验(判据随口径变更作废)", False),
("英文 fail", "fail", False),
("空值 ⇒ 判不出来 ⇒ 未过", "", False),
]:
got = m._acc_is_pass(val)
rows.append(("_acc_is_pass(%s) = %s(应 %s)" % (label, got, want), got is want))
# ── 层 2:端到端(喂中文进真源 ⇒ `goal_state()` 必须是 `pass`,⛔ 不是 `open`)
gj, tg = TIN / "goal.json", TEST_WS / "tg.json"
bak_g = gj.read_text(encoding="utf-8") if gj.exists() else None
tg_bak, tk_bak = tg.read_text(encoding="utf-8"), tasks()
try:
put_tasks({}) # 前两路清空 ⇒ 只剩验收那一路能说话
tg.write_text(json.dumps({"nodes": [
{"id": "T1", "title": "测试件-1", "status": "done", "line": "line-a"}]},
ensure_ascii=False), encoding="utf-8")
gj.write_text(json.dumps({"acceptance_state": {
"V1-中文写法": "过|pid 1234 活,心跳 8 秒前",
"V2-带括注": "过(复核:已实测)",
"_说明": "这行不参与判定"}}, ensure_ascii=False), encoding="utf-8")
s_cn, short_cn = m.goal_state(), m._acc_short()
gj.write_text(json.dumps({"acceptance_state": {
"V1-真不过": "🔴 不过|零条活着"}}, ensure_ascii=False), encoding="utf-8")
s_bad = m.goal_state()
finally:
tg.write_text(tg_bak, encoding="utf-8")
put_tasks(tk_bak)
if bak_g is not None:
gj.write_text(bak_g, encoding="utf-8")
else:
try:
gj.unlink()
except Exception:
pass
rows.append(("真源全中文且全过 ⇒ `goal_state()` = pass(实得 %s)" % s_cn, s_cn == "pass"))
rows.append(("…且 `_acc_short` 报『全部 pass』(%s)" % short_cn[0][:34],
"全部 pass" in short_cn[0]))
rows.append(("真源有一条中文「不过」⇒ `open`(实得 %s)" % s_bad, s_bad == "open"))
return rows
@case("两类检查会话是**两件事**:prompt 必须不同 + 闸④「本项目无待执行排期」")
def t_two_kinds_of_check_agent():
"""🔴🔴 2026-10-03 08:0x 用户明令:「**这里面是两件事 提示词应该不一样**」。
用户原话两条:
1、会话全结束(**还要加上 定时任务中没有本项目待执行的任务**)且**队列不为空** ⇒
**结果检查会话**去跟进协作会话执行结果,判断是否创建对应协作会话继续完成目标;
2、会话全结束(同上)且**队列为空 >20 分钟** ⇒ **目标检查会话**去核对目标完成状态,
判断是建协作会话还是**修改目标状态**,并执行。
⇒ 改前**两套 prompt 完全共用一个模板** ⇒ 结果检查会话拿到了"改目标状态"的出口(越权),
目标检查会话拿到的是"看队列 head"的开场(串味)。本用例钉死"两套必须不同"。
✅ 判据**能改前报红**:把 `_check_prompt` 改成无视 `reason` 恒返回同一份 ⇒ 本条必红。
"""
import re as _re
m = imp()
rows = []
pr = m._check_prompt(1, "自检目标", "sessions-ended")
pg = m._check_prompt(1, "自检目标", "queue-empty")
rows.append(("两套 prompt 渲染都不炸(各 %d/%d 字符)" % (len(pr), len(pg)), True))
rows.append(("⛔ 两套 prompt **不是同一份**(实得 %s)" % (pr == pg), pr != pg))
rows.append(("结果检查:标题是「结果检查会话」(%s)", "【结果检查会话" in pr))
rows.append(("目标检查:标题是「目标检查会话」(%s)", "【目标检查会话" in pg))
# 职责边界:结果检查 ⛔ 不许有改目标状态的出口
rows.append(("结果检查 ⛔ 不含「改目标状态」的出口", "没有「改目标状态」" in pr))
rows.append(("⛔ 结果检查里不该出现 `--set-life`", "--set-life" not in pr))
# 目标检查必须有改状态的出口
rows.append(("目标检查含 `--set-life 已完成` 出口", "--set-life 已完成" in pg))
rows.append(("目标检查讲了三路完成度(含 `acceptance_state`)", "acceptance_state" in pg))
rows.append(("目标检查有「不确定就不改」的 fail-safe",
"不确定就不改" in pg))
# 排期名分类
rows.append(("排期名分类:结果检查 → [%s]" % m.CHECK_KINDS["sessions-ended"][1],
m.CHECK_KINDS["sessions-ended"][1] == "结果检查"))
rows.append(("排期名分类:目标检查 → [%s]" % m.CHECK_KINDS["queue-empty"][1],
m.CHECK_KINDS["queue-empty"][1] == "目标检查"))
# 闸④
rows.append(("闸④ 函数存在:`ws_pending_schedules()`", callable(m.ws_pending_schedules)))
ps = m.ws_pending_schedules(exclude="__不存在的排期名__")
rows.append(("闸④ 现网能读出本工作区待执行排期(%d 条,测试工作区应为 0)" % len(ps),
isinstance(ps, list)))
# ⚠️ 判据必须读**函数体**而不是 docstring —— docstring 里写"fail-safe"只是作者的说明,
# ⛔ 拿它当判据=判"作者记得写这四个字"(同 P0-36 的教训:别拿字面量当事实)。
# 真判据=**读不到库时返回非空** ⇒ 只能靠"把 RO 连接打坏"来验,见下面那条。
rows.append(("闸④ 判据注释里写明 fail-safe 方向=不建",
"fail-safe" in (m.ws_pending_schedules.__doc__ or "")))
# 真判据:临时把只读连接打坏 ⇒ 必须返回**非空**(当"有排期"⇒ 闸拦 ⇒ 不建)
_orig = m._ro_conn
try:
def _boom():
raise RuntimeError("自测:故意让读库失败")
m._ro_conn = _boom
broken = m.ws_pending_schedules()
finally:
m._ro_conn = _orig
rows.append(("🔴 闸④ 读库失败时返回**非空**(fail-safe=当有排期⇒不建),实得 %r"
% (broken[:1],), bool(broken)))
rows.append(("🔴 闸④ 读库失败时**不**返回空列表(否则闸会放行)", broken != []))
# 非法 reason 必须被拒
rows.append(("非法 reason 不在 CHECK_KINDS(%s)", "瞎写的" not in m.CHECK_KINDS))
# 旧名必须已废(⛔ 静默变成"什么都给"更坏)
try:
m.CHECK_PROMPT(1, "x")
rows.append(("旧名 `CHECK_PROMPT` 已废(会抛)", False))
except RuntimeError:
rows.append(("旧名 `CHECK_PROMPT` 已废(会抛)", True))
return rows
@case("检查会话的 prompt:**引用的命令必须真存在**(⛔ 引用不存在的命令 ⇒ 静默当成功)")
def t_check_prompt_cmds_exist():
"""🔴🔴 2026-10-03 实测坐实:`CHECK_PROMPT` 让检查会话用
`goalctl.py set-life已完成` 改生命周期 —— ⛔ **该命令在 `goalctl.py` 里根本不存在**。
实跑后果:只打印用法、**rc=0 静默放行** ⇒ 检查会话**以为改成功了**,实际一个字没改。
✅ 真源=`collabd.py --set-life`(`L4200` 的 CLI 分支 ⇒ `goal_life_set()`)。
✅ 本用例判据**能改前报红**:占位符个数对不上会当场 `TypeError`;
prompt 里仍写 `goalctl.py set-life` 则本条**必红**。
"""
import re as _re
m = imp()
rows = []
# 🔴 2026-10-03 08:0x:`_check_prompt` 已拆两套(`reason` 必传)⇒ **两套都要验**,
# ⛔ 只验默认那套=漏掉"结果检查里不该有 --set-life"这类越权。
for reason, want_life in (("queue-empty", True), ("sessions-ended", False)):
tag = "目标检查" if want_life else "结果检查"
try:
p = m._check_prompt(1, "自检目标", reason)
ok = True
except TypeError as e:
p, ok = "", False
rows.append(("%s:`_check_prompt()` 渲染不炸(实得 %s)"
% (tag, e.__class__.__name__), False))
if not ok:
continue
rows.append(("%s:渲染成功,无残留占位符(实得 %s)"
% (tag, _re.findall(r"%[sd]", p)), not _re.findall(r"%[sd]", p)))
rows.append(("%s:坏引用 `用...goalctl.py set-life` 不在正文里" % tag,
not _re.search(r"用`[^`]*goalctl\.py set-life", p)))
if want_life:
rows.append(("目标检查:指向 `collabd.py --set-life`(⛔ 不是不存在的命令)",
"--set-life" in p and "collabd.py" in p))
rows.append(("目标检查:`--set-life` 与值之间有空格",
"--set-life 已完成" in p))
else:
rows.append(("结果检查:⛔ 不含 `--set-life`(那是目标检查的出口)",
"--set-life" not in p))
return rows
@case("台账四态:running → blocked(带原因) → 解除后原因清掉")
def t_states():
run_cli(["--report", "C1", "--state", "running", "--by", "t", "--line", "line-a"])
ok1 = (tasks().get("C1") or {}).get("state") == "running"
run_cli(["--report", "C1", "--state", "blocked", "--by", "t", "--reason", "缺一份对接单"])
v = tasks().get("C1") or {}
ok2 = v.get("state") == "blocked" and "对接单" in str(v.get("block_reason"))
run_cli(["--report", "C1", "--state", "pending", "--by", "t"])
v = tasks().get("C1") or {}
ok3 = v.get("state") == "pending" and not v.get("block_reason")
t = tasks(); t.pop("C1", None); (TIN / "tasks.json").write_text(json.dumps(t, ensure_ascii=False, indent=2), encoding="utf-8")
return [("上报 running 生效", ok1), ("blocked 带原因", ok2), ("解除后原因被清掉", ok3)]
@case("已停总闸:goal.run=paused ⇒ --once 只写一行、--tick ⛔ 不投递(防『人为停掉被读成机制坏了』)")
def t_goal_paused():
"""🔴 2026-10-01 体检修实测:**暂停中的需求被当成"机制故障"** 报给用户。
真因:`goal.json.run=paused` 时 `collabd.py` **零处**引用 `run` ⇒ `--once` 照跑、照写
`NEXT.md`/`STALL.md`/`NEED-USER.md`/`digest.md`,而 `digest.md` 经钩子**注入每个会话每一轮**。
⇒ 本用例钉死两件事:① 已停时**只留一行"已停"** 并把告警投影清掉 ② 已停时 `--tick` ⛔ 不投递。
"""
gp = TIN / "goal.json"
_had = gp.exists()
_old = gp.read_text(encoding="utf-8") if _had else ""
try:
# 埋脏投影:已停时它们**必须被清掉**(⛔ 留着就被钩子继续注入)
for _n in ("NEXT.md", "STALL.md", "NEED-USER.md", "queue.md"):
(TIN / _n).write_text("(selftest 脏投影)", encoding="utf-8")
gp.write_text(json.dumps({"run": "paused", "acceptance_state": {}}, ensure_ascii=False),
encoding="utf-8")
w0 = wakeups_n()
rc1, out1 = run_cli(["--once"])
_txt = ""
try:
_txt = (TIN / "digest.md").read_text(encoding="utf-8")
except Exception:
pass
ok_line = ("已停" in _txt) and len([_l for _l in _txt.splitlines() if _l.strip()]) <= 6
ok_clear = all(not (TIN / _n).exists() for _n in ("NEXT.md", "STALL.md", "NEED-USER.md", "queue.md"))
rc2, out2 = run_cli(["--tick"])
ok_no_wake = wakeups_n() == w0
ok_tick = ("paused" in out2)
# 恢复 active ⇒ 必须**退出已停态**(`paused` 标记被清),⛔ 不许一次暂停就永久停摆
gp.write_text(json.dumps({"run": "active", "acceptance_state": {}}, ensure_ascii=False),
encoding="utf-8")
rc3, _ = run_cli(["--once"])
ok_resume = (not state().get("paused")) and rc3 == 0
return [("已停时 --once rc==0", rc1 == 0),
("digest 只有一行『已停』(⛔ 不产告警)", ok_line),
("告警投影被清掉(NEXT/STALL/NEED-USER/queue)", ok_clear),
("已停时 --tick ⛔ 不投递(%d→%d)" % (w0, wakeups_n()), ok_no_wake and rc2 == 0),
("--tick 明确印 paused(⛔ 不静默)", ok_tick),
("恢复 active ⇒ 退出已停态", ok_resume)]
finally:
if _had:
gp.write_text(_old, encoding="utf-8")
else:
try:
gp.unlink()
except Exception:
pass
@case("投递:协作程序(--once / --report)**零投递**")
def t_no_deliver():
w0 = wakeups_n()
run_cli(["--once"])
run_cli(["--report", "C2", "--state", "running", "--by", "t"])
w1 = wakeups_n()
t = tasks(); t.pop("C2", None); (TIN / "tasks.json").write_text(json.dumps(t, ensure_ascii=False, indent=2), encoding="utf-8")
return [("--once 与 --report 合计零投递(%d→%d)" % (w0, w1), w1 == w0)]
@case("投递:目标会话(**跟进会话**)在跑 ⇒ 必须延后(target-busy)")
def t_target_busy():
"""🔴🔴 2026-10-02 23:4x **本用例已随「跟进会话整套退役」失效**(⛔ 保留代码,⛔ 删了=没有用例=假绿)。
🔴 **为什么整组退役**(不是改断言、不是修代码):用户逐字「按照新的逻辑整体修改」/
「按照之前的讨论 唤醒会话 跟进会话 和 上报程序 都去掉才对」/「上报机制也不需要了」。
⇒ `follow_for_topic()` 已**短路**返回 `why="follow-retired"` ⇒ 这条路径**根本走不到**:
`target-busy`/`target-deaf`/「真投出」/「台账增条」这四组断言**永久拿不到**那个状态。
⚠️ ⛔ **不许把这四条改成"预期 follow-retired"** —— 那等于**把断言改恒真**(P0-13 同族):
改完之后任何实现都能过,等于没有守卫。
✅ **正确做法**:整组标记退役,⛔ 不动断言(将来口径变回时可直接复活)。
⚠️ 退役判据(机读,⛔ 不看代码注释):`follow_for_topic()` **真跑一次**返回 `why == "follow-retired"`。
—— 本用例**后半段(真调用)已改判这条**;前半段的旧断言**刻意留着不改**。
"""
m = imp()
# ✅ 退役判据(真调用,⛔ 不是 grep 源码)
r = m.follow_for_topic({}, "任意类别", [])
if r.get("why") != "follow-retired" or str(r.get("sid") or ""):
return [("🔴 两类改造:`follow_for_topic()` 必须短路成 `follow-retired` 且 `sid` 为空"
"(现 = %s / sid=%r)" % (r.get("why"), r.get("sid")), False)]
return [("✅ 两类改造:`follow_for_topic()` 短路成 `follow-retired`(唤醒/跟进整套退役)", True),
("⏸ 下面 2 条为**已退役**的旧断言(target-busy 的 ① 恒真 / ② 才有分量),"
"⛔ 刻意不改 —— 路径已死,改成新预期=断言恒真。判据见下方源码注释", True)]
# ═════════════ 以下=旧实现,⛔ 刻意保留勿删(口径若变回,直接复活这段即可) ═════════════
FAKE = "f0117000-0000-4000-8000-00000000000f"
_old_scan, _old_st, _old_busy = m._scan_follows, m._session_status, m._session_busy
try:
m._scan_follows = lambda topics=None: {"by_topic": {}, "sids": [FAKE], "cand": [],
"topics": list(topics or []), "err": ""}
m._session_status = lambda sid: ("working" if str(sid) == FAKE else _old_st(sid))
m._session_busy = lambda sid: ("busy" if str(sid) == FAKE else _old_busy(sid))
r1 = m._deliver_str("自测:不该被投出", "selftest-busy", {"roles": {}})
m._session_busy = lambda sid: ("idle" if str(sid) == FAKE else _old_busy(sid))
r2 = m._deliver_str("自测:不该被投出", "selftest-busy", {"roles": {}})
finally:
m._scan_follows, m._session_status, m._session_busy = _old_scan, _old_st, _old_busy
return [("① 目标 `busy` ⇒ skipped==target-busy(现 = %s)" % (r1.get("skipped") or "空"),
r1.get("skipped") == "target-busy"),
("② 目标 `idle` ⇒ ⛔ 不延后,往下走到 live 判(现 = %s)" % (r2.get("skipped") or "空"),
r2.get("skipped") != "target-busy")]
@case("投递:错名修复 —— `why` 透传(有候选但不活 ⇒ follow-not-live,⛔ 不再记 no-follow-session)")
def t_deliver_why_passthrough():
"""🔴 2026-10-02 修(投递链路真因):`_deliver_str` 旧实现只按 `_want==""` 分档 ⇒ 把
**(a) 真·一条候选都没有** 与 **(b) 有候选、但都不在活会话里** **合并记成同一个 `no-follow-session`**
—— 字面意思与事实**相反**(现场实况:名册里 **6 条 `[跟进]` 都在、一条都不活**),
⇒ **把两天的排查引向"再多建一条"**,而多建**不管用**。
⇒ 本用例钉住:`skipped` 必须**透传 `follow_for_topic()` 已算准的 `why`**,两档分得开。
⚠️ 用**假的 `follow_for_topic`** 造前提(真实路径要求"网关探到活会话、且候选不在其中",
本机不可控 ⇒ 会变成天天 SKIP 的摆设用例);只替 `follow_for_topic` 与 `need_user`,
并临时补上网关口令过掉 `no-token` 闸,`finally` 里**逐项还原**。
"""
m = imp()
FAKE = "f0f0f0f0-0000-4000-8000-00000000000e"
_old_rft, _old_nu = m.follow_for_topic, m.need_user
_old_tok = os.environ.get(m.GW_ENV)
_seen = []
os.environ[m.GW_ENV] = "selftest-token" # ⛔ 不设 ⇒ 早退成 `no-token`,走不到本分支
try:
m.need_user = lambda t: _seen.append(t)
m.follow_for_topic = lambda st, topic="", live_sids=None: {
"sid": "", "source": "", "topic": topic, "why": "follow-not-live",
"cand": [FAKE], "all": [FAKE]}
r1 = m._deliver_str("自测:有候选但不活", "selftest-why-live", {"roles": {}})
m.follow_for_topic = lambda st, topic="", live_sids=None: {
"sid": "", "source": "", "topic": topic, "why": "no-follow-session",
"cand": [], "all": []}
r2 = m._deliver_str("自测:一条候选都没有", "selftest-why-none", {"roles": {}})
finally:
m.follow_for_topic, m.need_user = _old_rft, _old_nu
if _old_tok is None:
os.environ.pop(m.GW_ENV, None)
else:
os.environ[m.GW_ENV] = _old_tok
return [
("① 有候选·都不活 ⇒ skipped==follow-not-live(现 = %s)" % (r1.get("skipped") or "空"),
r1.get("skipped") == "follow-not-live"),
("② 真·无候选 ⇒ skipped==no-follow-session(现 = %s)" % (r2.get("skipped") or "空"),
r2.get("skipped") == "no-follow-session"),
("③ 两档确实分开(⛔ 不再合并成一个名字)",
(r1.get("skipped") or "") != (r2.get("skipped") or "")),
("④ 两档各喊一次用户(⛔ 不静默丢件)", len(_seen) == 2),
]
@case("投递:粗判第一条候选已哑、同类别另有活候选 ⇒ ⛔ 不许终局判 target-deaf(修『走不到精确重算』)")
def t_coarse_deaf_not_terminal():
# 🔴🔴 2026-10-02 23:4x **本用例随「跟进会话整套退役」失效**(⛔ 保留代码,⛔ 删了=没有用例=假绿)。
# `follow_for_topic()` 已短路成 `follow-retired` ⇒ "粗判 vs 精确"那条判据链**永久走不到**。
# ⚠️ ⛔ 不许把断言改成"预期 follow-retired"(=断言恒真,P0-13 同族);判据见 `t_target_busy`。
return [("⏸ 已退役(2026-10-02 两类改造):`follow_for_topic()` 短路 ⇒ 粗判/精确重算那条链走不到", True),
("✅ 退役判据:真调用返回 `follow-retired`",
imp().follow_for_topic({}, "X", []).get("why") == "follow-retired")]
# ═════════════ 以下=旧实现,⛔ 刻意保留勿删 ═════════════
"""🔴 2026-10-02(S5 · 验收 V2)钉死**两条判据打架**:粗判(`live_sids=[]`)与精确
(拿锁后用活网关列表重算)对**同一条**会话给出相反结论。
真因两处,**同族**、必须同批修(本用例同时钉住):
① **终局短路错位**:旧实现拿**粗判的 `_tgt`** 去 `_deaf_sids()` 判定并**直接返回**
`target-deaf`(还在取锁之前)⇒ 只要"最近活动的那条候选"恰好哑了,**每轮都短路**,
永远走不到拿锁后的精确重算 ⇒ "候选里还有活的"这个事实**一辈子没机会被发现**。
② **候选收窄过早**:`follow_for_topic()` 旧实现先把候选收窄成**单条**(`by_topic` 首见
=最近活动),再交给 `_pick_live()` ⇒ "最近活动那条死了" ≡ "这个类别投不出去",
同类别里活着的候选**轮不到**。
⚠️ 造前提的手法(⛔ 不依赖现场恰好有活会话 —— 那会让用例天天 SKIP=没测):
伪造 `_scan_follows`(两条同类候选:`DEAF` 最近活动、`LIVE` 活着)+假日志根把 `DEAF` 判哑
+伪造 `discover_gateways()`/`_gw_http()`。**真实 `follow_for_topic()` / `_pick_live()` /
`_deliver_str()` 全程在位**——被替掉的只有"外部世界",判据链一条不换。
🎯 **回退反证成立**:把 `follow_for_topic()` 的 `_lst` 换回 `[_lst[0]]`(或把粗判恢复成
`_tgt in _deaf_sids()`)⇒ 下面①~⑤ 立即变红。
"""
m = imp()
TP = "机制排查与修复"
DEAF = "deadbeef-0000-4000-8000-0000000000d1" # 最近活动 ⇒ 粗判的 `c[0]`
LIVE = "deadbeef-0000-4000-8000-0000000000d2" # 同类别、**活着** ⇒ 本该投它
# ⚠️ 假日志根**用完必须清**(`_fake_cfg` 是**追加**写入同一棵树):本用例的 DEAF 那份 10 MiB
# 会被后面的 `t_deaf_detect`(断言"结果恰好只有那一条哑的")读到 ⇒ **假红**(本轮实测踩到:
# 47 PASS / 1 FAIL,红的正是 `t_deaf_detect`)。与 `t_deaf_latest_day_only` 同一手法。
shutil.rmtree(TEST_WS / "fake-cfg", ignore_errors=True)
cfg = _fake_cfg({DEAF: DEAF_MIB + 100_000}) # 只把 DEAF 判哑(LIVE 无日志 ⇒ 不哑)
restore = _with_fake_cfg(cfg)
_old = (m._scan_follows, m.discover_gateways, m._gw_http, m.need_user)
_old_tok = os.environ.get(m.GW_ENV)
_seen, _st_in = [], {"roles": {}}
try:
# 清掉 wake:⛔ 否则 `same-item`/`too-soon` 会先短路,走不到精确阶段(假红)
(TIN / "collabd-state.json").write_text(
json.dumps({"wake": {"hash": "", "ts": 0}}), encoding="utf-8")
(TIN / "wake.lock").write_text("free", encoding="utf-8")
(TIN / "wakeups.jsonl").unlink(missing_ok=True)
os.environ[m.GW_ENV] = "selftest-token" # ⛔ 不给 ⇒ 早退成 `no-token`,走不到投递
m._scan_follows = lambda topics=None: {
"by_topic": {TP: DEAF}, "by_topic_all": {TP: [DEAF, LIVE]},
"sids": [DEAF, LIVE], "cand": [], "topics": [TP], "err": ""}
m.discover_gateways = lambda *a, **k: [{"port": 1, "sessionId": LIVE}]
m._gw_http = lambda *a, **k: (200, "{}")
m.need_user = lambda t: _seen.append(t)
r = m._deliver_str("自测:粗判第一条哑", "selftest-coarse-deaf", _st_in, topic=TP)
_w = (r.get("delivered") or {})
_sid_in_wake = str((_st_in.get("wake") or {}).get("sessionId") or "")
_n = wakeups_n()
finally:
m._scan_follows, m.discover_gateways, m._gw_http, m.need_user = _old
if _old_tok is None:
os.environ.pop(m.GW_ENV, None)
else:
os.environ[m.GW_ENV] = _old_tok
restore()
shutil.rmtree(TEST_WS / "fake-cfg", ignore_errors=True) # ⛔ 不留残渣给后面用例
return [
("① ⛔ 不许短路成 target-deaf(现 = %s)" % (r.get("skipped") or "空(↓已投出)"),
r.get("skipped") != "target-deaf"),
("② 走到精确阶段并**真投出**(http=%r)" % (_w.get("http"),), _w.get("http") == 200),
("③ 投给**活着的那条** `%s…`(⛔ 不是粗判的第一条 `%s…`)" % (LIVE[:8], DEAF[:8]),
_sid_in_wake == LIVE),
("④ 投递台账**真增 1 条**(现 %d)" % _n, _n == 1),
("⑤ ⛔ 没对用户喊话(粗判假警报已消除;现 %d 次)" % len(_seen), len(_seen) == 0),
]
@case("投递:跨进程互斥锁存在且会被释放")
def t_lock_released():
m = imp()
m._deliver_str("自测", "selftest", {"roles": {}})
_p = TIN / "wake.lock"
try:
_txt = _p.read_text(encoding="utf-8", errors="replace").strip()
except Exception:
_txt = ""
try:
import inspect as _ins
_src = _ins.getsource(m._deliver_str)
except Exception:
_src = ""
return [
# 🔴 2026-10-01 **契约变更**:旧契约="释放就是把文件删掉";新契约=
# "**文件永久存在,释放 = 原地把内容写成 `free`**"。为什么必须变:
# `_deliver_str` 每轮取锁后都在 `finally` 里 `unlink(wake.lock)`,而宿主有
# **SafeDelete 批量删除护栏**(按"本轮删除次数"计数,达 50 即要求确认并**终止进程**)
# ⇒ 常驻 ≈8 分钟(10 s/轮 × 50)就被杀(现场 `_supervise-bg.out.log` 原文
# `{"count":50,"threshold":50,"targetCount":1,"targets":[…\wake.lock]}`)。
# ⚠️ 断言随之改判**语义**而非存在性:`不存在`(从未取锁)**或** `内容=free` 都算"已释放"。
("锁已释放(不存在 或 内容=free;实得 %r)" % _txt, (not _p.exists()) or _txt == "free"),
# 🔴 反向那条**非有不可**:只判"已释放"的话,"每轮把文件 unlink 掉"这种旧写法**照样绿** ——
# 而它正是护栏杀进程的来源。⚠️ 且必须要求**源码真读到**(`_src` 为空=没验,⛔ 不许恒真)。
("锁释放 ⛔ 不用 unlink(源码读到 %d 字符,%s `_lk.unlink`)"
% (len(_src), "无" if "_lk.unlink" not in _src else "仍有"),
bool(_src) and "_lk.unlink" not in _src),
]
@case("投递:锁被**别的进程**持有时 ⇒ ⛔ 不投(`locked`)")
def t_lock_held():
# 🔴 为什么非有此例:上面那条只验"**能释放**",⛔ 验不到"**真能互斥**"。
# 而"跨进程互斥"是本机制**避免同一内容成对重复投递**的唯一屏障(实测事故:22:24:04/30 等
# 四组成对投递把主会话压成 `parkInQueue`「只进不出」⇒ 用户看到"卡死")。
# ⚠️ 判据必须是**真的把锁占住**再调 —— 伪造"别的进程正持有"(内容=假 pid + mtime=现在 ⇒ 不陈旧)。
m = imp()
_p = TIN / "wake.lock"
_p.parent.mkdir(parents=True, exist_ok=True)
try:
m._deliver_str("自测", "selftest", {"roles": {}}) # 先走一遍:把锁建起来、并释放成 free
_p.write_text("999999", encoding="utf-8") # 伪造"别的进程正持有"(⛔ 不陈旧 ⇒ 不许抢占)
r = m._deliver_str("自测-锁被别人拿着", "selftest", {"roles": {}})
finally:
try:
_p.write_text("free", encoding="utf-8") # 收尾:还原成未持有(⛔ 不 unlink)
except Exception:
pass
return [("skipped==locked(现 = %s)" % (r.get("skipped") or "空"), r.get("skipped") == "locked")]
# ── 🔴 2026-10-01 加:**会话哑掉(诊断日志撞 ~10 MiB 上限)** 的回归面 ──────────────
# 真因(本机实测):宿主**单会话诊断日志有 10 MiB 硬上限**,撞上即**拒写**(EPERM)
# ⇒ 该会话**界面不再刷新**(=哑);而投递只认 `http=200` ⇒ 向已哑会话投递**仍记
# `ok:true` =假绿**,通知全落空机制却毫无察觉(实测主会话 f8a792ab 停在 10,485,715 B,
# 其后累计 45,422 行被丢)。
# ⇒ 修法:`collabd.py` 加 `_deaf_sids()`(≥9.5 MiB 判哑)+ 投递**两道闸**
# (粗判按登记目标 · **出手前再按网关实际目标**)⇒ 记 `skipped=target-deaf` + 喊用户。
# ⚠️ 阈值分工(⛔ 别混):**9.5 MiB** =「已哑、不许再投」(`_deaf_sids` 的拦截线);
# **5 MiB** =「该交接了」(会话主动换,见 CODEBUDDY.md §1.5 G)。
DEAF_MIB = 9_500_000 # ⚠️ 与 `collabd._deaf_sids()` 的阈值**逐字同源**(那边改了这里必须跟)
def _fake_cfg(sizes: dict, day_offset: int = 0) -> Path:
"""在**测试区内**造一份假 `logs/<日期>/sdk/conversations/`(⛔ 生产日志目录一个字节不动)。
`_deaf_sids()` 只读 `CODEBUDDY_CONFIG_DIR/logs/...` ⇒ 换掉配置根即可**完全离线**复现本坑。
用 `truncate` 建**稀疏文件**:秒级拿到 10 MiB 的**体积读数**(`getsize`),⛔ 不真写 10 MiB。
`day_offset`:0=今天(默认)、1=昨天 —— 🔴 **按天分目录**正是"永久判哑"那个坑的土壤。
"""
_d = time.strftime("%Y-%m-%d", time.localtime(time.time() - 86400 * int(day_offset)))
root = TEST_WS / "fake-cfg" / "logs" / _d / "sdk" / "conversations"
root.mkdir(parents=True, exist_ok=True)
for _sid, _n in (sizes or {}).items():
with open(root / (_sid + ".log"), "wb") as f:
f.truncate(int(_n))
return TEST_WS / "fake-cfg"
def _with_fake_cfg(cfg: Path):
"""把 `CODEBUDDY_CONFIG_DIR` 临时指到假配置根,返回**还原函数**(⛔ 必须放在 finally 里调)。
为什么非还原不可:`_db()` 也读同一个变量 —— 不还原会把**后面所有用例**(主会话解析等)
一起带偏(那是"改一个用例弄红一片"的经典假红)。
"""
old = os.environ.get("CODEBUDDY_CONFIG_DIR")
def _restore():
if old is None:
os.environ.pop("CODEBUDDY_CONFIG_DIR", None)
else:
os.environ["CODEBUDDY_CONFIG_DIR"] = old
os.environ["CODEBUDDY_CONFIG_DIR"] = str(cfg)
return _restore
@case("哑会话:诊断日志 ≥ 9.5 MiB ⇒ _deaf_sids() 判哑(只读 · 判据=读数、含边界)")
def t_deaf_detect():
"""⚠️ 本用例走 **Python 直调**(不起子进程):`_deaf_sids()` 每次调用**现读**环境变量
⇒ 换 `CODEBUDDY_CONFIG_DIR` 即生效,无需重启进程。"""
DEAF = "deadbeef-0000-4000-8000-000000000001"
NEAR = "deadbeef-0000-4000-8000-000000000002" # 差 1 字节 ⇒ 验**边界不含**
OKAY = "cafe0000-1111-4222-8333-444444444444"
cfg = _fake_cfg({DEAF: DEAF_MIB + 100_000, NEAR: DEAF_MIB - 1, OKAY: 4096})
restore = _with_fake_cfg(cfg)
try:
d = imp()._deaf_sids()
finally:
restore()
return [("≥9.5 MiB ⇒ 判哑(%s…)" % DEAF[:8], DEAF in d),
("9,499,999 B(差 1 字节)⇒ ⛔ 不判哑(边界不含)", NEAR not in d),
("KB 级日志 ⇒ ⛔ 不判哑", OKAY not in d),
("⛔ 不误伤:结果里恰好只有那一条哑的", d == {DEAF})]
@case("哑会话:投递命中已哑目标 ⇒ skipped=target-deaf(⛔ 不投 + 喊用户;⛔ 不记假绿)")
def t_deaf_skip():
# 🔴🔴 2026-10-02 23:4x **本用例随「跟进会话整套退役」失效**(⛔ 保留代码,⛔ 删了=没有用例=假绿)。
# 投给「已哑跟进会话」这条路已不存在(`follow-retired`)⇒ `skipped==target-deaf` 拿不到。
# ⚠️ ⛔ 不许改成"预期 follow-retired"(=断言恒真);判据见 `t_target_busy`。
# ✅ 但**「不记假绿」这条原则仍活着**,改由退役判据承担:真调用拿 `sid=""` ⇒ 投不出去 ⇒ 台账不增。
m0 = imp()
_r = m0.follow_for_topic({}, "X", [])
return [("⏸ 已退役(2026-10-02 两类改造):投给已哑跟进会话这条路不存在了", True),
("✅ 退役判据:`sid` 为空 ⇒ **投不出去**(⛔ 记 `ok:true` 就是假绿,退役后天然满足)",
(not str(_r.get("sid") or "")) and _r.get("why") == "follow-retired")]
# ═════════════ 以下=旧实现,⛔ 刻意保留勿删 ═════════════
"""🔴 这条钉住的正是**假绿**:旧实现向已哑会话投递**仍记 `ok:true · http:200`**。
判据三件:① 判读是 `target-deaf` ② 投递台账**零增量** ③ 已把"该点哪一下"落到 `NEED-USER.md`。"""
DEAF = "deadbeef-0000-4000-8000-000000000001"
cfg = _fake_cfg({DEAF: DEAF_MIB + 100_000})
try:
(TIN / "NEED-USER.md").unlink() # ⚠️ 先清:否则上一轮留下的那份会让"喊过用户"**恒真**
except Exception:
pass
w0 = wakeups_n()
restore = _with_fake_cfg(cfg)
try:
m = imp()
# 🔴 2026-10-01 改:登记成 `main` 已经没用(投递**只看跟进会话**,用户口径见 `_deliver_str`)
# ⇒ 改成**伪造一条跟进会话**,让这道闸真的被走到(⛔ 否则只会落 `no-follow-session`)。
_old_scan = m._scan_follows
m._scan_follows = lambda topics=None: {"by_topic": {}, "sids": [DEAF], "cand": [],
"topics": list(topics or []), "err": ""}
try:
r = m._deliver_str("自测:哑会话不该被投", "selftest-deaf", {"roles": {}})
finally:
m._scan_follows = _old_scan
_sk = str(r.get("skipped") or "")
_nu = (TIN / "NEED-USER.md").exists()
_w1 = wakeups_n()
finally:
restore()
return [("skipped==target-deaf(现 = %s)" % (_sk or "空"), _sk == "target-deaf"),
("⛔ 零投递(wakeups %d→%d)" % (w0, _w1), _w1 == w0),
("已喊用户(NEED-USER.md 落盘)", _nu)]
@case("哑会话:**只认最新一天** —— 昨天撞过上限、今天已正常轮转 ⇒ ⛔ 不判哑(修『永久判哑』)")
def t_deaf_latest_day_only():
"""🔴 本用例钉住的正是**方向相反的新假绿**:旧实现扫「今天+昨天」**任一**命中即判哑,
而宿主会话日志**按天分目录**、昨天那份**永不再增长** ⇒ **昨天撞过 10 MiB、今天已轮转**的
会话被**永久判哑** ⇒ 本该投出去的唤醒被静默拦下。
实证 `a202550c`:昨天 10.0 MB、**今天 4.2 MB 且宿主仍在写它**。
⇒ 判据改为「**每个会话取它最新一天那份日志**」。⛔ 今天没有日志 ⇒ 仍退到昨天(保守兜底,不许丢)。
"""
ROTATED = "a202550c-0000-4000-8000-00000000000a" # 昨 10 MiB / 今 4.2 MB ⇒ 已轮转 ⇒ ⛔ 不判哑
OLDONLY = "0badf00d-0000-4000-8000-00000000000b" # 只有昨天、且撞上限 ⇒ 仍判哑(⛔ 兜底不许丢)
NEWDEAF = "feedface-0000-4000-8000-00000000000c" # 今天才撞上限 ⇒ 判哑
HEALTHY = "5ca1ab1e-0000-4000-8000-00000000000d" # 今 4 KB ⇒ ⛔ 不判哑
shutil.rmtree(TEST_WS / "fake-cfg", ignore_errors=True) # ⛔ 先清空:前面两条用例也在同一棵树下造过文件
_fake_cfg({ROTATED: DEAF_MIB + 500_000, OLDONLY: DEAF_MIB + 500_000}, day_offset=1) # 昨天那份
cfg = _fake_cfg({ROTATED: 4_200_000, NEWDEAF: DEAF_MIB + 500_000, HEALTHY: 4096}) # 今天那份
restore = _with_fake_cfg(cfg)
try:
d = imp()._deaf_sids()
finally:
restore()
shutil.rmtree(TEST_WS / "fake-cfg", ignore_errors=True)
return [("昨 10 MB / 今 4.2 MB(已轮转)⇒ ⛔ 不判哑", ROTATED not in d),
("今天才撞上限 ⇒ 判哑", NEWDEAF in d),
("只有昨天那份、且撞上限 ⇒ 仍判哑(兜底不丢)", OLDONLY in d),
("今天 KB 级 ⇒ ⛔ 不判哑", HEALTHY not in d),
("⛔ 不误伤:结果恰好={今天哑的, 只有昨天的}(现 %d 条)" % len(d),
d == {NEWDEAF, OLDONLY})]
# ── 🔴 2026-09-30 加:**投递=宿主钩子唤起(`--tick`)** 的回归面 ────────────────
# 背景(用户口径「又给我整到自动任务去了」):投递改由**宿主钩子**唤起 `collabd.py --tick`
# (钩子是宿主子进程 ⇒ 自带网关口令)。
# ⚠️ **2026-09-30 同日用户改口**(取代上面那句里的"⛔ 不靠自动化排期"):
# 「自动任务(符合条件时)是通过 协作程序去唤醒 主会话,这样流程统一」
# ⇒ **允许自动任务**当"闹钟"(`goalctl.py wake` ⇒ `--tick`);⛔ 但判断与投递**仍只在本程序一处**。
# 🔴 下面两条约束**与"谁拨这一下"无关**,**仍然有效**(改触发源⛔ 不免除它们):
# ① `--once`(协作程序/投影轮)**⛔ 不许消费队列** —— 它在 UserPromptSubmit 上先跑;
# ② `--tick` **投不出去就不许消费队列** —— 旧实现无条件 `notified[kid]=stt` ⇒ **静默丢件**。
@case("投递(新):纯投影 --once ⛔ 不消费队列")
def t_projection_no_consume():
put_tasks({"P1": {"state": "running", "by": "t", "line": "line-a"}})
(TIN / "collabd-state.json").write_text(json.dumps(
{"notified": {}, "notify_pending": ["P1=running"]}, ensure_ascii=False), encoding="utf-8")
rc, _ = run_cli(["--once"])
s = state()
kept = "P1=running" in list(s.get("notify_pending") or [])
no_aw = not s.get("notify_awaiting")
put_tasks({})
return [("rc==0", rc == 0),
("--once 后 notify_pending 原样保留(现 = %s)" % (s.get("notify_pending"),), kept),
("--once 未设 notify_awaiting", no_aw)]
@case("投递(新):投不出去(no-token)⇒ ⛔ 不消费队列(防静默丢件)")
def t_no_consume_when_skipped():
put_tasks({"Q1": {"state": "running", "by": "t", "line": "line-a"}})
(TIN / "collabd-state.json").write_text(json.dumps(
{"notified": {}, "notify_pending": ["Q1=running"]}, ensure_ascii=False), encoding="utf-8")
env = _mk_env()
env.pop("CODEBUDDY_GATEWAY_PASSWORD", None) # 抽掉口令 ⇒ 投递必被挡(no-token)
rc, out = run_cli_env(["--tick"], env)
s = state()
pend = list(s.get("notify_pending") or [])
notified = dict(s.get("notified") or {})
put_tasks({})
return [("rc==0", rc == 0),
("队列项仍在(现 = %s)" % (pend,), "Q1=running" in pend),
("⛔ 未被标记 notified", notified.get("Q1") != "running"),
("⛔ 未设 notify_awaiting", not s.get("notify_awaiting"))]
@case("宿主卡住探针:只读尾部、测试环境零命中、⛔ 不误报")
def t_park_probe():
# 2026-09-30 立(用户给的窗口 22:55–23:20 实测定型):指纹 = parkInQueue + hasWaiter=false。
# ⚠️ 测试工作区名是 `selftest` ⇒ 候选日志目录里没有它的日志 ⇒ 必须**零命中且不报错**(⛔ 不碰生产日志)。
m = imp()
r = m.probe_host_park()
real = ('[2026/9/29 22:53:00.208] [Info] [pid=32356] [AcpView] [AcpView][PromptIterator] '
'received prompt session=fe146dd9-… route=parkInQueue runState=idle queueLen=0 hasWaiter=false')
echo = ('[2026/9/30 00:36:15.002] [Info] [pid=50716] [SandboxShell] ProcessOutput | processId=pipe-18 | '
'content=Info] [pid=32356] [AcpView] route=parkInQueue queueLen=1 hasWaiter=false | stderr(0)=""')
ok_line = ('[2026/9/29 23:12:16.901] [Info] [pid=55508] [AcpView] [AcpView][PromptIterator] '
'received prompt session=fe146dd9-… route=resolveWaiter runState=idle queueLen=0 hasWaiter=true')
return [("返回 dict 且含 n/last/file", all(k in r for k in ("n", "last", "file"))),
("测试环境零命中(不读生产日志)", r["n"] == 0),
("只读尾部(常量 ≤ 512KB)", 0 < m.PARK_TAIL <= 512 * 1024),
("⛔ 不命中「取证命令的回声行」(防假阳性)", m._is_park_line(echo) is False),
("✅ 命中真正的 parkInQueue 记录行", m._is_park_line(real) is True),
("✅ 不误判正常行(resolveWaiter/hasWaiter=true)", m._is_park_line(ok_line) is False)]
@case("投递消费回查:投出去 ≠ 跑起来了 ⇒ 未消费必须升级为 NEED-USER")
def t_consume_check():
m = imp()
nu = TIN / "NEED-USER.md"
if nu.exists():
nu.unlink()
# ① 刚投出去(宽限期内)⇒ waiting,⛔ 不报警
st1 = {"wake": {"expect": {"sid": "0" * 36, "at": time.time()}}}
r1 = m.check_delivery_consumed(st1)
quiet = not nu.exists() # ⚠️ 必须在跑 ② 之前取快照(② 会写这个文件 ⇒ 否则本项必假红)
# ② 投出很久且目标会话零活动 ⇒ unconsumed,且必须落 NEED-USER.md(喊人)
st2 = {"wake": {"expect": {"sid": "0" * 36, "at": time.time() - 9999}}}
r2 = m.check_delivery_consumed(st2)
shouted = nu.exists() and "切走再切回" in nu.read_text(encoding="utf-8", errors="replace")
return [("宽限期内 = waiting 且不报警", r1.get("waiting") is True and quiet),
("超期未动 = unconsumed", r2.get("unconsumed") is True),
("落 NEED-USER 且写明处置动作(切走再切回)", shouted),
("回查后清掉 expect(⛔ 不重复报)", "expect" not in (st2.get("wake") or {}))]
@case("need_user:同句节流(⛔ 不每轮刷时间戳)+ 人写的「## 解除条件」两路都必须原样保留")
def t_need_user_throttle_keep():
"""🔴 2026-10-01 立(治两处现场事故,均已在生产实测):
· **噪音**:常驻每 ~10 s 用同一句话重写 NEED-USER.md ⇒ 时间戳一直变、人看不出「这条是新的还是旧的」
(现场:22:06–22:12 之间被刷了 20+ 次)。
· **抹掉人的回信口**:人手工写了「## 解除条件」,下一轮被机器**静默抹掉** ⇒ 人就再也不写了
(现场:21:03 手写的解除条件当场被覆盖)。
⚠️ 判据必须能「改前报红」(P0-13):把节流去掉 ⇒ ③ 报红;把保留 `_keep` 去掉 ⇒ ④ 报红。"""
import inspect as _ins
m = imp()
nu = TIN / "NEED-USER.md"
human = "\n## 解除条件\n- 用户手写:这是人唯一的回信口,⛔ 不许抹\n"
try:
if nu.exists():
nu.unlink()
# ① 第一次喊 ⇒ 必须落文件(⛔ 不许"喊了但没落")
m.need_user("自测原因-A")
got1 = nu.exists() and ("自测原因-A" in nu.read_text(encoding="utf-8", errors="replace"))
# ② 模拟人回信:追加「## 解除条件」
nu.write_text(nu.read_text(encoding="utf-8", errors="replace").rstrip() + human,
encoding="utf-8")
mt2 = nu.stat().st_mtime
# ③ 同一句话、节流窗口内再喊 ⇒ ⛔ 不重写(mtime 不变)+ 人的话必须还在
m.need_user("自测原因-A")
t3 = nu.read_text(encoding="utf-8", errors="replace")
quiet = (nu.stat().st_mtime == mt2) and ("用户手写" in t3)
# ④ 换一个原因 ⇒ 允许重写,但**人的话仍要原样保留**
m.need_user("自测原因-B")
t4 = nu.read_text(encoding="utf-8", errors="replace")
keep = ("自测原因-B" in t4) and ("## 解除条件" in t4) and ("用户手写" in t4)
# ⑤ 源码级:need_user 段内零 unlink(删文件 = 撞宿主 SafeDelete 批量删除护栏,P0-6)
try:
_src = _ins.getsource(m.need_user)
except Exception:
_src = ""
no_unlink = bool(_src) and (".unlink" not in _src)
return [("① 首次喊 ⇒ NEED-USER.md 落盘且含原因", got1),
("③ 同原因在窗口内 ⇒ ⛔ 不重写(mtime 未变)且人的话还在", quiet),
("④ 换原因才重写,且人的「## 解除条件」原样保留", keep),
("⑤ need_user 段内零 unlink", no_unlink)]
finally:
if nu.exists():
nu.unlink()
@case("投递忙判据:读宿主状态机日志的 busy=(⛔ 不再用 `status=='working'` 那档恒真的判据)")
def t_srsm_busy():
"""🔴 2026-10-01 立(治「状态窗口死结」,现场已在生产实测)。
病根:`sessions.status` 取值域只有 `working/completed/error/archived` —— **没有"空闲"这一档**,
活着的会话跑完一轮、停在等下一轮时 status **仍是 `working`** ⇒ 拿它判"忙"**恒真**
⇒ 投递永远等不到空闲(实测 `e2ccdea3` 连打 6 分钟 `延后投递…等它空闲`+`投递未成(target-busy)`)。
正解:宿主自己的 `[SessionRunStateMachine]` 日志(⛔ 不进数据库,查库查不到)。
⚠️ 判据必须能「改前报红」(P0-13):把"新鲜度"那半去掉 ⇒ ③ 报红;把「先判终结态」去掉 ⇒ ⑤ 报红。
"""
m = imp()
tmpd = TEST_WS / "logs"
tmpd.mkdir(parents=True, exist_ok=True)
f = tmpd / "fakews__abcd.log"
SID = "11111111-2222-3333-4444-555555555555"
OTHER = "99999999-2222-3333-4444-555555555555"
def line(ts, sid, busy):
return ("[%s] [Info] [pid=1] [SessionRunStateMachine] transition | sessionId=%s | event=X | "
"from=a | to=b | lifecycle=running | busy=%s | queueBusy=true" % (ts, sid, busy))
now = time.strftime("%Y/%m/%d %H:%M:%S")
old = time.strftime("%Y/%m/%d %H:%M:%S", time.localtime(time.time() - 7200))
orig_dirs, orig_status = m._log_dirs, m._session_status
try:
m._log_dirs = lambda: [f]
# ① 末条 `busy=true` 且新鲜 ⇒ busy(正对照)
f.write_text(line(now, SID, "true") + "\n", encoding="utf-8")
c1 = m._session_busy(SID) == "busy"
# ② 末条 `busy=false` ⇒ idle(⛔ 不能被前一条 busy=true 带偏 ⇒ 必须取**最后一条**)
f.write_text(line(now, SID, "true") + "\n" + line(now, SID, "false") + "\n", encoding="utf-8")
c2 = m._session_busy(SID) == "idle"
# ③ 末条 `busy=true` 但已是 2 小时前 ⇒ idle —— 这正是"会话早停了、只是没写 busy=false"那一类
f.write_text(line(old, SID, "true") + "\n", encoding="utf-8")
c3 = m._session_busy(SID) == "idle"
# ④ 日志里**没有**该 sid ⇒ ''(⛔ 不猜、⛔ 不回落成 busy 或 idle)
f.write_text(line(now, OTHER, "true") + "\n", encoding="utf-8")
c4 = m._session_busy(SID) == ""
# ⑤ 库里已**终结** ⇒ 即便日志末条还是 `busy=true` 也**不判忙**
# (实测:自动化拉起的会话跑完**不写 `busy=false`**,直接在库里变 completed)
f.write_text(line(now, SID, "true") + "\n", encoding="utf-8")
m._session_status = lambda s: "completed"
c5 = (m._target_busy(SID) is False)
finally:
m._log_dirs, m._session_status = orig_dirs, orig_status
try:
f.unlink()
except Exception:
pass
# ⑥ 源码级**反回归**:正则必须用**命名组**。
# 本轮真踩过:写成 `(?:\.(\d+))?` 时内层 `(\d+)` **仍是捕获组** ⇒ 后面 `m.group(8)` 全错位一位
# ⇒ `busy` 恒读成 False ⇒ **恒判"空闲"**(方向相反:会往正在跑的会话里插话)。
# ⛔ 这条断言不是"看代码像不像" —— 它是**唯一**能在这类错位再发生时立刻报红的东西。
# ⚠️ 判据范围必须**从正则定义处起**(`getsource(_session_busy)` 只给函数体 ⇒ 看不见正则本身);
# 末端取 `def _target_busy` ⇒ 恰好覆盖「正则 + `_session_busy`」这一段。
_all = (HERE / "collabd.py").read_text(encoding="utf-8", errors="replace")
_i = _all.find("_SRSM_RE = re.compile(")
_j = _all.find("def _target_busy", _i if _i >= 0 else 0)
seg = _all[_i:_j] if (_i >= 0 and _j > _i) else ""
named = bool(seg) and ("?P<busy>" in seg) and not re.search(r"group\(\s*[\"']?[1-9]", seg)
return [("① 末条 busy=true 且新鲜 ⇒ `busy`", c1),
("② 末条 busy=false ⇒ `idle`(取的是**最后一条**)", c2),
("③ 末条 busy=true 但已 2 小时 ⇒ `idle`(新鲜度真在起作用)", c3),
("④ 日志无该 sid ⇒ `''`(⛔ 不猜)", c4),
("⑤ 库里已终结 ⇒ 日志仍 busy=true 也**不判忙**", c5),
("⑥ `_session_busy` 用**命名组**、零位置组引用(防'数括号'错位)", named)]
@case("投递(新):--tick 在位、可跑、且**投递段已真删**")
def t_tick_wired():
m = imp()
known = "--tick" in {"--where", "--once", "--supervise", "--tick", "--report", "--reqs"}
rc, out = run_cli(["--tick"])
src = (HERE / "collabd.py").read_text(encoding="utf-8", errors="replace")
# 🔴 2026-10-01 改:原先这里写**绝对路径**指到某个具体工作区的 `tools/wb-result-hook.py`
# ⇒ 两处都不对:① 违反「技能里只用相对路径」(包去依赖"使用方的"项目目录);
# ② 宿主接线自 2026-10-01 起已改指**包内**那份 ⇒ 老写法在别的机器上恒走"文件不在,跳过",
# 看着绿、其实**什么都没验**(静默假绿)。现在按包内相对路径取自己那份钩子。
hook = HERE / "hooks" / "wb-result-hook.py"
hook_ok, hook_note = True, "(钩子文件不在,跳过)"
if hook.exists():
hs = hook.read_text(encoding="utf-8", errors="replace")
hook_ok = ("--tick" in hs) and ("maybe_run_supervisor_tick" in hs)
hook_note = "钩子里有 `--tick` 调用"
# 🔴🔴 2026-10-03 **投递段退役**:本用例原判据是「`--tick` 用 `mutate=True` 调 `supervise()`
# ⇒ 它是唯一投递方」——**这条判据已随投递退役而失效**(没有"投递方"这个角色了)。
# ⚠️ ⛔ **不许把它改成"`--tick` 用 `deliver=False` 调 ⇒ 它是唯一不投递方"** ——
# 那是**换个说法说同一件事**,仍对着源码字面匹配,一重构就假红(P0-13 同族)。
# ✅ **正解=退役它,换成"投递段真删"的判据**,且判据必须**可证伪**:
# ① **真调用** `supervise({}, deliver=True, mutate=True)`(⛔ 传最"能投"的那组参数)
# ⇒ `deliver.skipped` 必须为 `retired-20261003`。
# ⚠️⛔ **不许去读生产 state 取证**(本文件 L169 的元用例明令「自测不碰生产」,
# 且那样判据依赖"生产已被新代码跑过"⇒ 换台机器必假红)。
# ② **`supervise()` 函数体内零个调用**——判据走 **AST**(⛔ 不是字面 grep:
# docstring 里**必然**提到 `_deliver_str`(要说明退役了什么)⇒ 字面判据永久假红)。
try:
_r = m.supervise({}, deliver=True, mutate=True)
except Exception as _e:
_r = {"deliver": {"skipped": "raised:%r" % (_e,)}}
_sk = str((_r.get("deliver") or {}).get("skipped") or "")
calls = _calls_in(src, "supervise")
r_nocall = ("_deliver_str" not in calls) and ("follow_for_topic" not in calls)
return [("`--tick` 属已知参数", known),
("`--tick` 可跑 rc==0", rc == 0),
("🔴 2026-10-03 退役判据①:真调用 `supervise(deliver=True, mutate=True)` ⇒ `deliver.skipped` 必为 "
"`retired-20261003`(实测 = %r)" % (_sk,), _sk == "retired-20261003"),
("🔴 2026-10-03 退役判据②:`supervise()` 体内**零个** `_deliver_str`/`follow_for_topic` **调用**"
"(AST 判据,⛔ 不看 docstring;实测调用集 = %s)" % (sorted(calls) or "空",), r_nocall),
(hook_note, hook_ok)]
def _calls_in(src: str, fname: str) -> set:
"""🔴 **函数 `fname` 体内真正被调用的名字集合**(走 AST,⛔ 不做字面 grep)。
🔴🔴 **为什么必须走 AST**(2026-10-03 实测踩到):退役说明的 docstring 里**必然**提到
被删的函数名(否则读者不知道删了什么)⇒ 任何 `grep "_deliver_str(" in <函数体切片>`
都**永久假红** ⇒ 只能判"**调用**",⛔ 不能判"**出现**"。
⚠️ 附带好处:AST 只认 `Name`/`Attribute` 的 `func` 位 ⇒ 注释、字符串、docstring 一律不算。
"""
import ast as _ast
try:
tree = _ast.parse(src)
except Exception:
return {"<parse-error>"}
for node in _ast.walk(tree):
if isinstance(node, _ast.FunctionDef) and node.name == fname:
out = set()
for sub in _ast.walk(node):
if isinstance(sub, _ast.Call):
f = sub.func
if isinstance(f, _ast.Name):
out.add(f.id)
elif isinstance(f, _ast.Attribute):
out.add(f.attr)
return out
return {"<no-such-function>"}
@case("对账:僵尸(台账 running 但没人跑)⇒ 标 blocked")
def t_reconcile_zombie():
# 🔴 2026-09-30 修(**本用例曾假红**):僵尸的判据是「**除观察者外无人跑** ∧ 无 IN_PROGRESS」——
# 而测试机上**天天有别的工作会话在跑** ⇒ 前提不成立 ⇒ 不能判它红(⛔ 也不许假装绿)。
# 正确姿势:**前提不成立就 SKIP**(读不出宿主库、或确实有别的会话在跑)。
m = imp()
put_tasks({"Z1": {"state": "running", "by": "已经死掉的棒", "t": time.time()}})
r = m.reconcile({})
if r.get("skip"):
put_tasks({})
return [("SKIP:读不到宿主库(%s)" % r["skip"], True)]
# 🔴 2026-10-01 加:僵尸分支的**完整前提** =「**除观察者外**无在跑会话」∧「无 IN_PROGRESS 运行」。
# 旧用例只靠 `adopted` 反推**前一半** ⇒ 后半没覆盖 ⇒ **自动化会话跑本自测时**,
# 它**自己**的 automation run 必然还是 `IN_PROGRESS`、且它自己就是唯一在跑的会话
# ⇒ 前提不成立却没 SKIP ⇒ **必然假红**(2026-10-01 实测:整套只有本条红,
# 而 `reconcile` 返回 `adopted=[] zombie=[]`)。⚠️ 同一类假红的另一半,照既有姿势补 SKIP。
try:
_me = (os.environ.get("CODEBUDDY_SESSION_ID") or "")[:8]
_con = m._db()
_inprog = int(_con.execute("select count(*) from automation_runs "
"where status='IN_PROGRESS'").fetchone()[0])
_others = [s for (s,) in _con.execute("select id from sessions where status='working'")
if not (_me and str(s)[:8] == _me)]
_con.close()
except Exception:
_inprog, _others = 0, []
if _others or _inprog:
put_tasks({})
return [("SKIP:前提不成立(另有 %d 个会话在跑 · %d 个自动化运行 IN_PROGRESS)"
% (len(_others), _inprog), True)]
if r.get("adopted"):
put_tasks({})
return [("SKIP:当前确有 %d 个别的会话在跑 ⇒ 『除观察者外无人跑』前提不成立"
% len(r["adopted"]), True)]
v = m._load_tasks().get("Z1") or {}
ok = v.get("state") == "blocked" and "执行者已消失" in str(v.get("block_reason"))
put_tasks({})
return [("Z1 被标 blocked 且原因是『执行者已消失』", ok)]
@case("对账:幂等(连跑两次结论一致)")
def t_reconcile_idempotent():
# 🔴 2026-09-30 修(**同一类假红**):第 1 次会**收编**当前正在跑的别的工作会话(环境相关),
# 第 2 次才进入"没有新东西"的稳态 ⇒ 幂等性只能比较 **第 2 次 vs 第 3 次**。
run_cli(["--reconcile"])
_, o1 = run_cli(["--reconcile"])
_, o2 = run_cli(["--reconcile"])
put_tasks({})
return [("第 2 次 = 第 3 次输出一致", o1.strip().splitlines()[:1] == o2.strip().splitlines()[:1]),
("第 2 次已无新收编(『收编 0 条』)", "收编 0 条" in o1)]
@case("CLI:未知参数 ⇒ rc=2 且**不落常驻**")
def t_unknown_arg():
rc, out = run_cli(["--nonsense"], timeout=20)
return [("rc==2", rc == 2), ("提示未知参数", "未知参数" in out)]
@case("CLI:--where / --reqs 可跑且 rc=0")
def t_basic_cli():
rc1, o1 = run_cli(["--where"])
rc2, _ = run_cli(["--reqs"])
return [("--where rc==0 且打印路径", rc1 == 0 and "workspace" in o1), ("--reqs rc==0", rc2 == 0)]
@case("结构:关键名字/设施在位(防『误删常量』重演)")
def t_symbols():
src = (HERE / "collabd.py").read_text(encoding="utf-8")
need = ["CLAIMS", "SELF_SID", "STALE", "DOING_TTL", "def reconcile(", "def supervise(",
"def _deliver_str(", "def _guard_says_stop(", "def parse_session_name(",
"def topic_of(", "def load_goal(", "GOAL_F"]
miss = [n for n in need if n not in src]
return [("关键符号齐全(缺:%s)" % (miss or "无"), not miss)]
@case("结构:所有 subprocess 调用都带 creationflags(不显窗)")
def t_nowindow():
bad = []
for fn in ("collabd.py", "guard.py"):
for i, l in enumerate((HERE / fn).read_text(encoding="utf-8").splitlines(), 1):
if "subprocess.run(" in l or "subprocess.Popen(" in l:
seg = " ".join((HERE / fn).read_text(encoding="utf-8").splitlines()[i - 1:i + 4])
if "creationflags" not in seg:
bad.append("%s:%d" % (fn, i))
return [("无『没带 creationflags』的子进程调用(%s)" % (bad or "无"), not bad)]
@case("结构:fatal 不再 exit 0(防『假绿』)")
def t_failopen():
src = (HERE / "collabd.py").read_text(encoding="utf-8")
i = src.find('log("fatal %s" % e)')
seg = src[i:i + 200] if i >= 0 else ""
return [("fatal 分支是非零退出", "sys.exit(1)" in seg), ("⛔ 不是 exit(0)", "sys.exit(0)" not in seg)]
@case("结构:投递段**已真删**(⛔ 旧「投递方唯一」判据已于 2026-10-03 退役)")
def t_deliver_switch():
src = (HERE / "collabd.py").read_text(encoding="utf-8")
m = imp()
# 🔴🔴 2026-10-03:本用例原有一条判据「`--tick` 是唯一投递方(`supervise(st, deliver=True)`)」
# ——**投递退役后这个角色已不存在**,那条判据永久拿不到状态。
# ⚠️ ⛔ **不许把它改成 `supervise(st, deliver=False)`**(=换个说法说同一件事,仍是字面匹配脆断)。
# ✅ 退役 + 换可证伪判据:`supervise()` 体内**零个** `_deliver_str`/`follow_for_topic` **调用**
# (AST 判据 —— ⛔ 字面 grep 会被 docstring 里的退役说明永久假红,见 `_calls_in` 的注释)。
calls = _calls_in(src, "supervise")
r = m.supervise({}, deliver=True, mutate=True)
sk = str((r.get("deliver") or {}).get("skipped") or "")
return [("supervise 有 deliver 参数", re.search(r"def supervise\(st: dict, deliver: bool", src) is not None),
("supervise 有 mutate 参数(签名兼容保留,⛔ 体内已不读)",
re.search(r"def supervise\(st: dict, deliver: bool = True, mutate: bool = True\)", src) is not None),
("one_round 走**纯投影**(deliver=False, mutate=False)",
re.search(r"supervise\(st,\s*deliver=False,\s*mutate=False\)", src) is not None),
("🔴 2026-10-03 退役判据①:`supervise()` 体内**零个** `_deliver_str`/`follow_for_topic` **调用**"
"(AST;实测调用集 = %s)" % (sorted(calls) or "空",),
("_deliver_str" not in calls) and ("follow_for_topic" not in calls)),
("🔴 2026-10-03 退役判据②:真调用 `supervise(deliver=True, mutate=True)` ⇒ "
"`deliver.skipped` 必为 `retired-20261003`(实测 = %r)" % (sk,),
sk == "retired-20261003"),
("wake_round 已标注停用", "已停用" in src)]
@case("主会话:**解析出来的**(不只是读登记)+ ⛔ 拒绝盲投(防『投错窗口』回潮)")
def t_main_resolution():
"""用户 2026-09-30:「协作机制能发现 主会话 换了吗」——旧答案是**不能**(登记静态+路径依赖,
且看板与投递共用同一份登记 ⇒ 一起钉在旧 id 上)。本用例把修法钉住。"""
src = (HERE / "collabd.py").read_text(encoding="utf-8")
m = imp() # 🔴 2026-10-01:本用例起要**真跑函数**(见下面那条"能吃类别集合")
return [
("有 resolve_main(登记失效时按当前事实解析)", "def resolve_main(" in src),
("登记**有效才认**(`reg in live`)", "or reg in live" in src),
("按**工作区**解析(用户定则:**一个工作区一条主会话**)", "def _same_ws(" in src),
("工作区比较**同时**小写+统一斜杠(⛔ 防 `E:/x` 与 `E:\\x` 判成两个)",
'replace("\\\\", "/")' in src),
("🔴 按工作区解析时**排除「协作侧」命名**(2026-10-02 两类改造:唤醒/跟进已退役 ⇒ 只排 `worker`)"
"—— 接续会话是干活的棒,⛔ 不是主会话候选(否则它冒充主会话 ⇒ 通知投给它自己)",
'_role not in ("worker",)' in src),
("⛔ **不得按 `status='working'` 筛主会话**(主会话两轮之间是空闲 ⇒ 加了必然漏掉它)",
"status='working' \"\n \"order by last_activity_at" not in src),
("⛔ **不得按 `is_background_automation` 排主会话**(接续会话本身就是自动化起的 —— 实测主会话也是 1)",
"coalesce(is_background_automation, 0) = 0" not in src),
("🔴 **认「主控」前缀**(用户 2026-09-30 定名:主会话的显式标记)",
'startswith("主控")' in src and '"prefix:主控"' in src),
("解析不出 ⇒ 明确报 `no-main-session`", '"no-main-session"' in src),
("目标不在活会话里 ⇒ 明确报 `main-not-live`", '"main-not-live"' in src),
("🔴 **盲选回落已删除**(⛔ 不再随手投给任意一条会话)",
re.search(r'g = next\(\(x for x in gws if x\["sessionId"\]\), None\)', src) is None),
# ── 🔴 2026-09-30 新增:**同一工作区 · 多任务类别**(用户要求"同一个工作区 多会话协作")───
("🆕 有 `resolve_mains()`(按**任务类别**各解析一条主会话)", "def resolve_mains(" in src),
("🆕 有 `_scan_mains()`(与 `resolve_main` **共用同一遍扫描** ⇒ **同一份 live**,⛔ 不各扫各的=防两边漂移)",
"def _scan_mains(" in src and src.count("_scan_mains(st, live)") >= 2),
# 🔴 2026-10-01 体检修发现的**投递断点**:解析出一条**已经结束**的「主控」会话当投递目标
# ⇒ 投递一路 `main-not-live`(程序在跑,通知永远发不出去)。⇒ **候选只认活着的**。
("🆕 候选**只认活着的**:`_pick_live` 在位且两处解析都用它",
"def _pick_live(" in src and src.count("_pick_live(named, live)") >= 2
and src.count("_pick_live(cand, live)") >= 2),
# 🔴🔴 2026-10-01 **两条旧断言被本轮口径推翻 ⇒ 改判据**(⛔ 不是删掉 —— 性质还得有人守):
# 用户逐字:「**换新会话 唤醒的是 跟进会话,主会话只能是用户触发**」
# ⇒ `_deliver_str` 的投递目标**从"该类别的主会话"改成"该类别的跟进会话"**
# (`_rms = resolve_mains(st, …)` + `main_for_topic(_rms, topic)` **故意**从源码里消失了)。
# ⚠️ 那两条旧断言查的正是这两个字面量 ⇒ 留着就是**恒红**(不是代码坏了,是**判据过时**)。
# ⚠️ 但它守的**性质**("按类别选、⛔ 不投错窗口")必须继续守 ⇒ 换成查新函数,并补两条。
("🆕 投递目标经 `follow_for_topic()` **按类别**取(⛔ 不再是主会话、也不再是「唯一那一条」)",
re.search(r"_rft = follow_for_topic\(st, topic,", src) is not None),
("🆕 `follow_for_topic` 对「已登记类别」**严格取该类别**(⛔ 不回落别类别=防投错窗口)",
re.search(r"if tp and tp in tps:", src) is not None),
("🔴 `follow_for_topic` **绝不返回主会话**(用户口径:主会话只由用户触发)"
"—— 2026-10-02 两类改造后**真调用**恒返回 `sid=\"\"`(短路退役)⇒ 更不可能返回主会话",
(not str(imp().follow_for_topic({}, "X", []).get("sid") or ""))),
("🔴 解析不出跟进会话 ⇒ 记 `no-follow-session` + **喊用户**(⛔ 不再降级投主会话)",
'"no-follow-session"' in src and 'follow-not-live' in src
and '"no-main-session"' not in src.split("def _deliver_str(")[1].split("\ndef ")[0]),
("🆕 `main_for_topic` 对「已登记类别」**严格取该类别**(⛔ 不回落别类别=防投错窗口)",
'if tp and tp in (rms.get("topics") or [])' in src),
("🆕 `_deliver_str` 支持 `topic=` 参数(默认空 ⇒ 走 default,向后兼容)",
re.search(r'def _deliver_str\(text: str, key: str, st: dict, topic: str = ""\)', src) is not None),
# 🔴🔴 2026-10-03 退役:本条原判据 `'topic=str((v or {}).get("line") or "")' in src`
# 指的是 `supervise()` 里那行 `_deliver_str(..., topic=…)`——**投递段已真删 ⇒ 该行不存在**。
# ⚠️ ⛔ 不许改成 `topic="" in src` 之类(=断言恒真,P0-13 同族)。
# ✅ 退役 + 换可证伪判据:`_deliver_str()` 本体**仍支持 `topic=` 参数**
# (它是"将来重建收件人"的实现,签名不许退化)⇒ 改成断言**函数签名里有 `topic: str = ""`**,
# 那是真判据(谁把参数删了就会红),且**不是**靠一个字面串匹配调用点。
("🆕 `supervise()` 曾按台账条目的 `line` 传 `topic=`(⛔ 2026-10-03 随投递退役)",
"topic=str((v or {}).get(\"line\") or \"\")" not in src),
("🆕 退役后仍保留:`_deliver_str()` 本体**签名仍带 `topic=`**(重建收件人时的实现)",
re.search(r'def _deliver_str\(text: str, key: str, st: dict, topic: str = ""\)', src) is not None),
# 🔴 2026-10-01 改:原来这条是**纯源码匹配** `"for _t in _as_list(short):" in src` ——
# 我今天给 `_in_project` 加了第三条判据、顺手把这一行提成 `_tps = _as_list(short)`,
# 行为**一模一样**,它却报红了(**脆断**:对着源码字面写断言,一重构就假红)。
# ⇒ 改成**行为断言**(真喂一个 list 进去看结果),源码那半只留"函数还在"。
("🆕 归属判据能吃**类别集合**(`_in_project` 的 main_sid / short 传 list 都对)",
"def _as_list(" in src
and m._in_project("", "[协作]-[B类]-x", ["M1"], ["A类", "B类"])
and m._in_project("M1", "随便什么标题", ["M1", "M2"], ["A类"])),
("🆕 收尾确认**排除全部主会话**(⛔ 否则别类别的主会话被算成「在跑」⇒ 永远拦住)",
'mains = list(_rms0.get("all_sids") or [])' in src and "if sid in mains:" in src),
("🆕 两处 `_goal_topics()` 同款(技能里 goal.topics 是**唯一**类别源,缺省回落 `short`)",
src.count("def _goal_topics(") == 1 and '"topics"' in src),
("换了目标 ⇒ 跟随 + 显式告警(⛔ 不静默)",
"def _follow_main(" in src and "主会话变更:" in src and "need_user(" in src),
]
@case("主会话:看板与投递**同一套判据**(⛔ 防两处漂移 —— 漂了就没人能发现换主会话)")
def t_main_same_rule():
a = (HERE / "collabd.py").read_text(encoding="utf-8")
b = (HERE / "board.py").read_text(encoding="utf-8")
pat = r'def _main_sid\(st: dict\) -> str:'
return [
("两处都定义了 _main_sid", re.search(pat, a) is not None and re.search(pat, b) is not None),
("两处都按 roles -> wake 的顺序取(逐字同款)",
'st.get("roles")' in a and 'st.get("roles")' in b
and 'st.get("wake")' in a and 'st.get("wake")' in b),
]
@case("主会话:登记为 main 但**已哑** ⇒ ⛔ 不认它 + `--declare main` 是**换**不是「加」")
def t_main_deaf_and_declare():
"""🔴🔴 2026-10-01 修 **V2「主会话可响应」的硬阻塞**(两个缺陷叠在一起,缺一条都治不好)。
**缺陷①**:`resolve_main` 的第①条快路原为 `if reg and (not live or reg in live)` ——
**只查"活着",⛔ 不查"哑"**。而"哑"(会话诊断日志撞 ~10 MiB ⇒ 宿主拒写 ⇒
**界面永不刷新**)**照样在 `live` 里** ⇒ 一条哑掉的主会话来一次认一次,一路投到 `target-deaf`。
⚠️ 而 ② 走的 `_pick_live()` **本来就排哑**(2026-10-01 专为"别再投给哑会话"加的)——
**① 却从没接上它** ⇒ 同一条哑会话「② 排得掉、① 排不掉」(判据打架,当日第三次同族)。
**缺陷②**:`--declare --role main` 是**加**不是**换**(P0-11 的正解①就是它,号称"零代码改动立刻可用")
—— 而 `_main_sid()` 取 `roles` 里**第一条** `main`,Python dict 对已存在的键赋值**不改位置**
⇒ 老那条**永远排前面** ⇒ 新登记的主会话**永远轮不到** ⇒ 逃生口形同虚设。
🔴 而且它掩盖得更深:`_deliver_str` 的**粗判不传活会话**(`resolve_mains(st, [])`)
⇒ `_pick_live` 在 `live` 为空时**回退 `c[0]`** ⇒ 每轮先认下那条哑的、**直接返回**,
**走不到后面"拿锁后用活网关列表重算"那一步**。
"""
m = imp()
A = "aaaa1111-0000-0000-0000-000000000001"
B = "bbbb2222-0000-0000-0000-000000000002"
# ── 缺陷①:把 A 伪造成"哑",A 登记在 B 前面 ──
_orig = m._deaf_sids
try:
m._deaf_sids = lambda: {A}
r_live = m.resolve_main({"roles": {A: "main", B: "main"}}, [A, B])
# 粗判那条路(不传活会话)⇒ 只能拿第一条 —— 这正是"必须让登记权威化"的理由
r_coarse = m.resolve_main({"roles": {A: "main", B: "main"}}, [])
finally:
m._deaf_sids = _orig
# ── 缺陷②:`--declare --role main` 必须把别的 main 降级 ──
# 在**测试区**里造一份 state(⛔ 不碰生产),用完还原
SP = TIN / "collabd-state.json"
bak = SP.read_text(encoding="utf-8") if SP.exists() else None
SP.write_text(json.dumps({"roles": {A: "main", B: "worker"}}, ensure_ascii=False), encoding="utf-8")
_env = _mk_env()
_env["CODEBUDDY_SESSION_ID"] = B
p = subprocess.run([PY, "-u", str(HERE / "collabd.py"), "--declare", "--role", "main"],
cwd=str(HERE), stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
text=True, env=_env, timeout=60, errors="replace")
_st = json.loads(SP.read_text(encoding="utf-8"))
_out = p.stdout or ""
if bak is None:
SP.unlink(missing_ok=True)
else:
SP.write_text(bak, encoding="utf-8")
_roles = _st.get("roles") or {}
S = (HERE / "collabd.py").read_text(encoding="utf-8")
return [
("🔴 ① 排哑后才认下**活着的**那条登记(拿到 %s)" % (r_live["sid"][:8] or "(空)"),
r_live["sid"] == B and r_live["source"] == "roles:main"),
("🔴 ① 换了登记的那条 ⇒ **报 `switched_from`**(调用方才会降级+告警,⛔ 不静默换)",
r_live["switched_from"] == A),
("🔴 ① 走的是**同一个** `_pick_live`(⛔ 不再有两台判据)",
"_reg = _pick_live(_rm, live)" in S),
("⚠️ 粗判(**不传活会话**)退回第一条 ⇒ 这就是「必须让登记权威化」的理由(坑已固化)",
r_coarse["sid"] == A),
("🔴 ② `--declare --role main` 把原 main **降级为 worker**(换,⛔ 不是加)",
_roles.get(B) == "main" and _roles.get(A) == "worker"),
("🔴 ② 且**说出口**(⛔ 不静默改别人的角色)", "已降级为 worker" in _out),
]
@case("结构:守护生命周期(优雅退出设施在位)")
def t_graceful():
src = (HERE / "guard.py").read_text(encoding="utf-8")
return [("有 _shutdown(优雅收尾)", "def _shutdown(" in src),
("子进程打 DSH_GUARDED 标", "DSH_GUARDED" in src),
("子程序侧有 _guard_says_stop", "_guard_says_stop" in (HERE / "collabd.py").read_text(encoding="utf-8"))]
@case("归属判据:看板 ≡ 收尾确认(⛔ 防两处漂移;⛔ 不按 cwd 推断)")
def t_scope_parity():
"""🔴 用户 2026-09-30 两条明示:「**要明确哪些会话是属于某个需求目标项目的**」+
「**这个看板上应该明确显示是哪个需求目标项目**」。
`collabd._in_project()`(`--ready-next` 用)与 `board.in_project()`(看板用)是**重复实现**
⇒ 一旦漂移就会出这种怪事:**看板说"本项目没人在跑",而 `--ready-next` 被别的项目的会话拦死**。
本用例把两者摆到同一张用例表上逐例比对(含"AI变现日报"这类**别人家的会话**必须判否)。
"""
code = r'''
import importlib.util, inspect, json, sys
sys.dont_write_bytecode = True # ⛔ 别让测试自己往技能目录里落 __pycache__
def load(name, path):
spec = importlib.util.spec_from_file_location(name, path)
m = importlib.util.module_from_spec(spec); spec.loader.exec_module(m); return m
H = sys.argv[1]
cb = load("cb", H + "/collabd.py")
bd = load("bd", H + "/board.py")
cases = [
("main1", "任意标题(主会话)", "main1", "手机接入", True),
("aaa", "[协作]-[手机接入]-N10 复测", "main1", "手机接入", True),
("bbb", "AI变现日报 · 今日生成", "main1", "手机接入", False),
("ccc", "随便一个标题", "main1", "手机接入", False),
("ddd", "[其他主题]-x", "main1", "手机接入", False),
("eee", "x", "main1", "", False), # short 空 ⇒ 只剩主会话能命中
("ggg", "x", "ggg", "", True), # 主会话判据 ⛔ 不依赖 short
("hhh", "[其他主题][手机接入]", "main1", "手机接入", True),
("fff", "[手机接入] 加个 [手机接入x] 干扰", "main1", "手机接入", True),
]
bad = []
for sid, ti, msid, short, exp in cases:
a = bool(cb._in_project(sid, ti, msid, short))
b = bool(bd.in_project(sid, ti, {"main_sid": msid, "short": short}))
if a != b:
bad.append("%s 判据漂移(collabd=%s / board=%s)" % (sid, a, b))
if a != exp:
bad.append("%s 实得 %s / 期望 %s" % (sid, a, exp))
# 🆕 2026-09-30:**同一个工作区 · 多任务类别**(用户要求「通过协作会话名称前缀区分具体任务会话」)——
# 两处实现都必须认"**任一**已登记类别",且⛔ **不得**把未登记的类别算进来。
mt = [
# sid, 标题, 默认主会话, 类别清单, 期望
("m1", "主控 · 机制线", "m1", ["手机接入", "机制线"], True), # 默认主会话
("w1", "[协作]-[机制线]-唤醒A", "m1", ["手机接入", "机制线"], True), # 第 2 类别
("w2", "[协作]-[手机接入]-N9", "m1", ["手机接入", "机制线"], True), # 第 1 类别
("w3", "[协作]-[别的类别]-x", "m1", ["手机接入", "机制线"], False), # 未登记 ⇒ 不算
("w4", "唤醒机制线专项(无方括号)", "m1", ["手机接入", "机制线"], False), # 归属判据要求 `[类别]`
("w5", "[协作]-[机制线]-x", "m1", ["手机接入"], False), # 只登记一类 ⇒ 另一类不算
("w6", "x", "m1", [], False), # 无类别 ⇒ 只剩主会话判据
("w7", "x", "w7", ["手机接入", "机制线"], True), # 主会话判据与类别无关
]
for sid, ti, msid, tps, exp in mt:
a = bool(cb._in_project(sid, ti, msid, tps))
b = bool(bd.in_project(sid, ti, {"main_sid": msid, "main_sids": [msid] if msid else [],
"topics": tps, "short": (tps[0] if tps else "")}))
if a != b:
bad.append("%s 多类别判据漂移(collabd=%s / board=%s)" % (sid, a, b))
if a != exp:
bad.append("%s 多类别实得 %s / 期望 %s" % (sid, a, exp))
# 🆕 结构对齐:两处 `_scan_*` 的 `by_topic` 值**必须是 dict**({sid,source,explicit})——
# ⛔ 实测踩到:board 曾存裸 sid ⇒ 调用方把整个 dict 当 sid 塞进 `all_sids`(静默错值)。
_cbs = inspect.getsource(cb)
_bds = inspect.getsource(bd)
for _nm, _s in (("collabd", _cbs), ("board", _bds)):
if '"explicit"' not in _s or '"by_topic"' not in _s:
bad.append("%s 的 by_topic 结构不合规(须存 {sid,source,explicit})" % _nm)
sig = str(inspect.signature(bd.in_project))
if "cwd" in sig:
bad.append("board.in_project 仍带 cwd 参数:%s" % sig)
src = open(H + "/board.py", encoding="utf-8").read()
if "project_scopes" in src:
bad.append("board.py 仍残留 cwd 版的 project_scopes()")
print(json.dumps({"bad": bad, "sig": sig}, ensure_ascii=False))
'''
p = subprocess.run([PY, "-c", code, str(HERE)], stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
text=True, env=_mk_env(), timeout=30, errors="replace")
try:
r = json.loads((p.stdout or "").strip().splitlines()[-1])
except Exception:
return [("子进程跑不起来(rc=%s):%s" % (p.returncode, (p.stdout or "")[-300:]), False)]
return [("15 例判据一致且与期望相符(7 基础 + 8 多类别;%s)" % ("、".join(r["bad"]) or "全对"), not r["bad"]),
("board.in_project 不带 cwd 参数(%s)" % r["sig"], "cwd" not in r["sig"])]
@case("看板:风格系统(三风格令牌在位 + 组件里零硬编码色)")
def t_board_styles():
"""🔴 用户 2026-09-30:「画架构图参考 archify 的样式,**右上角加个风格切换(保留当前风格)**」。
两件必须守住:
① **`base`(保留的当前风格)必须在**,且**跟随系统深浅色** —— ⛔ 删了就等于改了用户的原风格。
② **组件里不许出现硬编码颜色** —— 颜色只许出现在令牌块(`--color-*` 等)里。
⚠️ 一旦有人在 `.card{}`/SVG 里写死 `#fff`,换风格就会**花**(这是最容易犯的)。
"""
p = HERE.parent / "assets" / "board.html"
src = p.read_text(encoding="utf-8")
need = ['[data-style="archify-dark"]', '[data-style="archify-light"]',
':root:not([data-style]), :root[data-style="base"]', # base 跟随系统深色
'id="stylesw"', 'function applyStyle(', "dsh-board-style"] # 切换件 + 持久化
misc = [n for n in need if n not in src]
bad = []
for i, l in enumerate(src.splitlines(), 1):
s = l.strip()
if s.startswith(("--color-", "--font", "--dot", "--node", "--radius", "--shadow")):
continue # 令牌块本身当然有颜色值
if re.search(r"#[0-9a-fA-F]{3,8}\b", l) or re.search(r"rgba?\(\s*\d", l):
bad.append("board.html:%d" % i)
return [("三风格令牌 + 切换件 + 持久化齐全(缺:%s)" % (misc or "无"), not misc),
("组件里零硬编码颜色(越界:%s)" % (bad[:4] or "无"), not bad)]
@case("看板:验收摘要行 ⛔ 不许把『一条判据都没有』说成『无』(同族红线)")
def t_board_acc_summary():
"""🔴 2026-09-30 跑整体验证时抓到:`board.py` 的命令行摘要写的是
`[k for k,v in acc.items() if v!='pass'] or "无"` —— `acc` 已滤掉 `_` 说明行,
所以**一条有效判据都没有**时列表为空 ⇒ 打印「验收非 pass:**无**」,
读起来就是「全部 pass」。**形状同族**:不崩溃,只是少说一句话。
⇒ 统一到 `_acc_summary()`:空 ⇒ 显式「判不出来,⛔ 不因此判完成」。
"""
os.environ["COLLABD_CONFIG"] = str(CFG)
os.environ["DSH_COLLAB_WS"] = str(TEST_WS)
import importlib.util
spec = importlib.util.spec_from_file_location("bd_selftest", HERE / "board.py")
b = importlib.util.module_from_spec(spec)
spec.loader.exec_module(b)
s_empty = b._acc_summary({})
s_pass = b._acc_summary({"V1": "pass", "V2": "pass"})
s_open = b._acc_summary({"V1": "pass", "V2": "fail"})
return [("空 ⇒ 显式『判不出来』(%s)" % s_empty[:26], "判不出来" in s_empty and "不因此判完成" in s_empty),
("⛔ 空时不得只说『无』(%s)" % s_empty[:12], s_empty.strip() != "无"),
("全 pass ⇒ 说清条数(%s)" % s_pass, "pass" in s_pass and "2" in s_pass),
("有未过 ⇒ 点名(%s)" % s_open, "V2" in s_open)]
@case("看板:『未归类』不许被静默丢掉(同族红线:读到了就要说)")
def t_board_orphan_visible():
"""🔴 2026-09-30 跑整体验证时在**真实工作区**抓到:类别清单已迁到「任务类别」,
而台账里还躺着**跨工作区时代的旧线名**(各 2 件且都已完成)。
旧版:这些行也当"分工位"进 `labor`,架构图又只画前 4 格 ⇒ **那几格连同里面的件整块消失**
(读者只看到 4 个"件 0 · 完成 0")。**形状=不崩溃,只是少说一句话** —— 最难发现的那类。
现在:`kind='legacy'` ⇒ **不算类别**(⛔ 不让工作区名冒充类别),但**折叠成一格画出来**
+ **图外点名线名与件数** ⇒ 两头都堵住。
"""
os.environ["COLLABD_CONFIG"] = str(CFG)
os.environ["DSH_COLLAB_WS"] = str(TEST_WS)
import importlib.util
spec = importlib.util.spec_from_file_location("bd_orphan", HERE / "board.py")
b = importlib.util.module_from_spec(spec)
spec.loader.exec_module(b)
# ① `_orphan` 只收 legacy 行,且件数**求和**(⛔ 不是只数条数)
o0 = b._orphan([{"line": "唤醒机制", "kind": "topic", "total": 3, "done": 3}])
o1 = b._orphan([{"line": "唤醒机制", "kind": "topic", "total": 3, "done": 3},
{"line": "old-ws", "kind": "legacy", "total": 2, "done": 2, "open": 0},
{"line": "old-ws2", "kind": "legacy", "total": 1, "done": 0, "open": 1}])
# ② 结构:看板真的会把它画出来 + 真的会点名(⛔ 只改数据不改渲染 = 没修)
H = (HERE.parent / "assets" / "board.html").read_text(encoding="utf-8")
need = ["kind==='legacy'", "未归类", "不属任何当前任务类别",
"不在当前任务类别清单里", # 图外提示要点名
"旧值 · 不在当前任务类别清单"] # 台账表要标注
miss = [n for n in need if n not in H]
return [("全 topic ⇒ 无未归类(n=%d)" % o0["n"], o0["n"] == 0),
("只收 legacy 行(n=%d)" % o1["n"], o1["n"] == 2),
("件数求和(总 %d / 完成 %d / 未完 %d)" % (o1["total"], o1["done"], o1["open"]),
(o1["total"], o1["done"], o1["open"]) == (3, 2, 1)),
("看板渲染里四处齐全(缺:%s)" % (miss or "无"), not miss)]
@case("看板:**同工作区但没归入本项目**的会话不许静默丢(⛔ 也不许把它算进本项目)")
def t_board_unrecognized_visible():
"""🔴 2026-10-01 用户报:「**为什么创建了协作会话 但是 看板中没有展示**」。
真事:M5 那根棒被命名成 `[协作]-本机协作-M5 …` —— **第 2 级没带方括号**、而且用的是
`goal.short`(本机协作)而非**任务类别**(`topics`=唤醒机制)。
⇒ `parse_session_name()` 判 `ok=False`、`_in_project()` 也找不到 `[唤醒机制]`
⇒ 它**整条从看板上消失**(不进协作会话层、也不算"本项目会话")。
**形状=不崩溃,只是少说一句话** —— 正是最难发现的那类(同族红线)。
⇒ 两头都要堵:
① **说出口**:`_sessions()` 交出 `unrecognized`,看板在图外说明里点名;
② **⛔ 不许改归属**:它们**不进 `mine`**、**不算「本项目会话共 N 个」** ——
归属判据仍是 §2.3 三级(主会话 sid / 标题含 `[<类别>]` / `--declare`),
⚠️ `cwd` **只用来决定"要不要提醒"**,⛔ 绝不用来决定"归不归本项目"。
"""
os.environ["COLLABD_CONFIG"] = str(CFG)
os.environ["DSH_COLLAB_WS"] = str(TEST_WS)
import importlib.util
spec = importlib.util.spec_from_file_location("bd_unrec", HERE / "board.py")
b = importlib.util.module_from_spec(spec)
spec.loader.exec_module(b)
H = (HERE.parent / "assets" / "board.html").read_text(encoding="utf-8")
S = (HERE / "board.py").read_text(encoding="utf-8")
need = [ # 看板真会画(⛔ 只算不画 = 没修)
"sessions_unrecognized", "未归入本项目", "会被漏管",
"值取 `goal.json` 的 `topics`(任务类别),⛔ 不是 `short` 简称",
]
miss = [n for n in need if n not in H]
# ① 结构:快照真带这个键(连同"读不到库"的降级分支也带 ⇒ ⛔ 别只在成功路上加)
has_key = S.count('"sessions_unrecognized"') >= 1 and S.count('"unrecognized": unrec') >= 1
# ② ⛔ 它**不许**混进 `mine`:cwd 只出现在告警分支里,且该分支显式 continue
guarded = "_ct == _wstail" in S and "⛔ 绝不用它决定" in S
# ③ 命名规范那句话必须说对(⛔ 别再说"主题取自 short" —— 那正是本 bug 的源头)
C = (HERE / "collabd.py").read_text(encoding="utf-8")
spec_ok = ("`[协作]-[<类别>]-<具体>`" in C) and ("主题取自 `goal.json` 的 `short`" not in C)
return [("快照真带 `sessions_unrecognized`(含降级分支)", has_key),
("cwd 只用于告警、⛔ 不用于定归属(判据未松)", guarded),
("看板渲染里三处齐全(缺:%s)" % (miss or "无"), not miss),
("派活命名规范已纠偏(⛔ 不再说\"主题取自 short\")", spec_ok)]
@case("归属:两处 `in_project` 判据**逐样本对账**(⛔ 漂了就静默漏管)")
def t_in_project_parity():
"""🔴🔴 2026-10-01 用户报:「**看板没有正常显示协作会话**」。
真因(**扫出来的,不是推理出来的**):两处归属判据原来**都只认带方括号**的写法
(`"[%s]" % 类别 in 标题`),而**实际派活出来的棒名是** `[协作]-机制排查与修复-常驻投递容器`
—— **第 2 级没带方括号** ⇒ 判 `False` ⇒ 那条会话(**当时正在 `working`**)落到
`sessions_unrecognized` ⇒ `board.html` 第三层画**空框「暂无协作会话」**。
⚠️ 更糟的是它和 `_topic_in_title()` **自相矛盾** —— 后者的注释白纸黑字写着
「判据是**子串**(不要求 `[...]` 包裹)…**两种写法都要认**」⇒ **同一个标题两处结论相反**。
⇒ 本用例把**两台判据喂同一批标题、逐条比布尔**(⛔ 不靠人盯、⛔ 不只查源码字符串),
并同时验**精度没丢**:正文里恰好提到类别名的**无关会话**仍不许被收进来
(靠"角色前缀闸"——`_role_of_title()` 解析不出角色的标题一律不算)。
"""
os.environ["COLLABD_CONFIG"] = str(CFG)
os.environ["DSH_COLLAB_WS"] = str(TEST_WS)
import importlib.util
spec = importlib.util.spec_from_file_location("bd_inproj", HERE / "board.py")
bd = importlib.util.module_from_spec(spec)
spec.loader.exec_module(bd)
m = imp()
_tops = ["会话协作自检", "机制排查与修复"]
sc = {"main_sid": "", "main_sids": [], "topics": _tops, "short": "会话协作自检",
"workspace": str(TEST_WS)}
# (标题, 期望) —— 期望是**这一版**的口径;⛔ 别把期望写成"跟旧版一样"
cases = [
("[协作]-机制排查与修复-常驻投递容器", True), # ← 本 bug 的正主:二级**无方括号**
# 🔴 2026-10-02 两类改造:唤醒/跟进**已退役** ⇒ 角色判空 ⇒ 判据②那条(「角色可解析+
# 标题含类别」)**不再命中** ⇒ 期望由 True 变 **False**。
# ⛔ 这不是"判据坏了",是**退役的正确后果**:那两类不再是活类别。
# ⛔ 顺带消掉了一个自指隐患:若它们还能被判成"本项目的会话",就会被当**主会话候选**。
("[唤醒]-会话协作自检-脉冲", False),
("[跟进]-会话协作自检-队列上报", False),
("[协作]-[机制排查与修复]-合规写法", True), # 老写法(带方括号)向后兼容
("主控 · 会话协作自检 · 收尾", True), # 口语式主会话(无方括号)
("接续 · 会话协作自检 · 第 2 棒", True), # 接续棒**带类别** ⇒ 认得出(新约定的目的)
("接续 · 会话机制合并包 · 任务4b", False), # 接续棒**不带类别** ⇒ 认不出(未改)
("[协作]-本机协作-M5 棒", False), # 用 `short`/昵称 ⇒ **仍认不出**(规范照旧)
("复盘:今天在会话协作自检里犯的错", False), # ← **精度闸**:无关会话恰提到类别名
("下载 open design", False),
("", False),
]
rows = []
for t, _ in cases:
a = bool(bd.in_project("", t, sc))
b = bool(m._in_project("", t, {"main_sid": "", "main_sids": []}, _tops))
rows.append((t, a, b))
_mismatch = [t for t, a, b in rows if a != b]
# ⚠️ 2026-10-01 **我自己写错又改对的一处**:原来写成
# `for (t, _), (_, a, _b) in zip(...)` —— 两个 `_` **互相覆盖**,
# 结果拿去跟 `a` 比的是**标题**不是**期望值** ⇒ 全表恒红。
# ⇒ 元教训:**对账类断言别用 `_` 当占位名**(一个函数里两个 `_` 必然屏蔽)。
_wrong = [(t, a, exp) for (t, exp), (_t, a, _b) in zip(cases, rows) if a != exp]
H = (HERE.parent / "assets" / "board.html").read_text(encoding="utf-8")
S = (HERE / "board.py").read_text(encoding="utf-8")
return [
("两处 `in_project` 逐样本一致(%d 条;不一致:%s)" % (len(cases), _mismatch or "无"),
not _mismatch),
("🆕 **二级无方括号也算本项目的会话**(`[协作]-机制排查与修复-…` ⇒ True;"
"⛔ 旧判据判 False ⇒ 看板第三层画空框而它正在跑)", bd.in_project("", "[协作]-机制排查与修复-常驻投递容器", sc)),
("🆕 精度未丢:无关会话恰好提到类别名 ⇒ **仍不算**(角色前缀闸)",
bd.in_project("", "复盘:今天在会话协作自检里犯的错", sc) is False),
("期望值逐条成立(不符:%s)" % (_wrong or "无"), not _wrong),
("🔴 判据用**角色+类别子串**两条并列(⛔ 别退回「只认方括号」)",
"_topic_in_title(ti, _tps) and _role_of_title(ti)" in S),
("🔴 看板的退场说明在渲染里真的存在(⛔ 只改数据不改渲染 = 没修)",
"sessions_retired" in H and "已收起" in H),
]
@case("会话退场:过期会话**收起但不删除**,且 ⛔ 不吃掉在跑的棒与主会话")
def t_session_retire():
"""🔴 2026-10-01 用户:「**通过接续会话的时候,如何处理过期的会话,避免越堆越多**」。
**为什么必须有**:接续=会话自己建下一棒,前棒**不会自动消失** —— 它在宿主库里永远是
一条 `completed` 记录、标题同族。实测本工作区已堆 5 条「接续 · 会话机制合并…」⇒
看板/状态稿每轮都为旧棒占位,真在跑的棒被埋在里面。
⇒ 两条退场路径:① **显式** `--retire self`(接续时由前棒自己登记,准确、即时)
② **窗口兜底**(`completed` 且超 `session_live_min` 分钟 ⇒ 前棒中途死掉也收得掉)
⛔ 两条闸门**都绕开 `working`**(在跑的一定要摊)+ **主会话是版面锚点**(⛔ 不抽)。
⛔ **收起 ≠ 删除**(宿主库原样在、`retired.json` 全留痕)⇒ 必须**报数**,⛔ 不许静默。
"""
os.environ["COLLABD_CONFIG"] = str(CFG)
os.environ["DSH_COLLAB_WS"] = str(TEST_WS)
import importlib.util
spec = importlib.util.spec_from_file_location("bd_retire", HERE / "board.py")
b = importlib.util.module_from_spec(spec)
spec.loader.exec_module(b)
m = imp()
now = time.time()
def _ms(minutes_ago):
return int((now - minutes_ago * 60) * 1000)
def _row(sid, title, status, minutes_ago):
return {"id": sid, "title": title, "custom_title": title, "status": status,
"cwd": str(TEST_WS), "updated_at": _ms(minutes_ago),
"last_activity_at": _ms(minutes_ago)}
_MAIN, _RUN, _NEW, _OLD, _RUNRET, _RET = ("aaaa0001", "bbbb0002", "cccc0003",
"dddd0004", "eeee0005", "ffff0006")
rows = [
_row(_MAIN, "主控 · 会话协作自检 · 主会话", "completed", 300), # 主会话:再旧也不许抽掉
_row(_RUN, "[协作]-会话协作自检-在跑的棒", "working", 300), # 在跑:再旧也不许收起
_row(_NEW, "[协作]-会话协作自检-刚完的", "completed", 10),
_row(_OLD, "[协作]-会话协作自检-很旧的", "completed", 300), # 窗口兜底收
_row(_RUNRET, "[协作]-会话协作自检-已退场但仍在跑", "working", 5),
_row(_RET, "[协作]-会话协作自检-已退场", "completed", 5), # 显式退场收
]
sc = {"main_sid": _MAIN, "main_sids": [_MAIN], "topics": ["会话协作自检"],
"short": "会话协作自检", "workspace": str(TEST_WS)}
# ① 显式退场表(写进**测试区**的 inbox,⛔ 不碰生产)
(TIN / "retired.json").unlink(missing_ok=True)
rc1, out1 = 0, ""
_env = _mk_env()
_env["CODEBUDDY_SESSION_ID"] = _RET
p = subprocess.run([PY, "-u", str(HERE / "collabd.py"), "--retire", "self",
"--why", "自测"],
cwd=str(HERE), stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
text=True, env=_env, timeout=60, errors="replace")
rc1, out1 = p.returncode, p.stdout or ""
_tbl = json.loads((TIN / "retired.json").read_text(encoding="utf-8")) if (TIN / "retired.json").exists() else {}
r1 = b._sessions(limit=30, sc=sc, rows=rows)
ids = [x["id8"] for x in (r1.get("mine") or [])]
hid = r1.get("hidden") or {}
hid_ids = [x["id8"] for x in (hid.get("recent") or [])]
# ② 退场表自身也要有过期清理(⛔ 否则这张表自己会越堆越多)
pr = m._prune_retired({_RET: {"at": now}, "zzzz9999": {"at": now - 8 * 86400}})
# ③ 短 id 也要能命中(本程序各处只存 8 位)
hit8 = m.is_retired(_RET[:8], m._retired())
(TIN / "retired.json").unlink(missing_ok=True)
return [
("`--retire self` rc=0 且落盘(%s)" % (out1.strip().splitlines() or ["(空输出)"])[0][:60],
rc1 == 0 and _RET in _tbl),
("退场表按**完整会话 id** 存(供 8 位短 id 前缀比对)", _RET in _tbl),
("🆕 显式退场的那条**被收起**(不在 mine 里)", _RET[:8] not in ids),
("🆕 超窗口(>%d 分钟)的 completed 也**被收起**" % int(b.LIVE_MIN), _OLD[:8] not in ids),
("🔴 **在跑的棒永不收起**(即使它已 300 分钟没动)", _RUN[:8] in ids),
("🔴 **主会话是版面锚点**:再旧也不抽掉", _MAIN[:8] in ids),
("🔴 显式退场 ⛔ 也不吃掉 **working** 的会话(它还在干活)", _RUNRET[:8] in ids),
("窗口内刚完成的留在版面上", _NEW[:8] in ids),
("报数:显式退场 %d / 超窗口 %d(应为 1 / 1)" % (hid.get("retired", 0), hid.get("stale", 0)),
(hid.get("retired") == 1 and hid.get("stale") == 1)),
("收起明细点名到会话(%s)" % (hid_ids or "空"), _RET[:8] in hid_ids and _OLD[:8] in hid_ids),
("⛔ 收起 ≠ 删除:宿主库行**一个字没动**(`mine` 里少了、明细里还在)",
len(rows) == 6 and len(hid.get("recent") or []) == 2),
("退场表自清理:超 7 天的记录被丢掉(剩 %d 条)" % len(pr), _RET in pr and "zzzz9999" not in pr),
("短 id(8 位前缀)也能命中退场表", hit8),
]
@case("看板:任务类别**来源**必须说得出口(declared / fallback)")
def t_board_topics_source():
"""🔴 2026-10-01 用户订正:「**目标是 通过对话在调用 会话协作skill时说明的,不是固定的**」。
`topics` 缺省回落 `short` 是**兼容**(单类别部署照旧),但看板**不许把"没说明过"
显示成"就是这么定的"** —— 那还是同族红线:**没有却说有**。
⇒ `_goal_topics_source()` 必须给出三态,且**看板真会把它渲染出来**(⛔ 只算不画=没修)。
"""
os.environ["COLLABD_CONFIG"] = str(CFG)
os.environ["DSH_COLLAB_WS"] = str(TEST_WS)
import importlib.util
spec = importlib.util.spec_from_file_location("bd_tsrc", HERE / "board.py")
b = importlib.util.module_from_spec(spec)
spec.loader.exec_module(b)
s_dec = b._goal_topics_source({"topics": ["甲", "乙"], "topics_declared_at": "2026-10-01T00:10",
"topics_declared_by": "对话", "short": "简"})
s_fb = b._goal_topics_source({"short": "简"})
s_no = b._goal_topics_source({})
s_cand = b._goal_topics_source({"_topics候选": {"候选": ["丙"]}, "short": "简"})
H = (HERE.parent / "assets" / "board.html").read_text(encoding="utf-8")
need = ["topics_source", "未声明", "调用时在对话里说明"]
miss = [n for n in need if n not in H]
return [("有 topics ⇒ declared(实得 %s)" % s_dec["kind"], s_dec["kind"] == "declared"),
("…且带说明时间/人(%s / %s)" % (s_dec["at"], s_dec["by"]),
bool(s_dec["at"]) and bool(s_dec["by"])),
("只有 short ⇒ fallback(实得 %s)" % s_fb["kind"], s_fb["kind"] == "fallback"),
("都没有 ⇒ none(实得 %s)" % s_no["kind"], s_no["kind"] == "none"),
("候选清单能读出来(%s)" % s_cand["candidates"], s_cand["candidates"] == ["丙"]),
("看板真会渲染来源(缺:%s)" % (miss or "无"), not miss)]
@case("看板:多目标 tab(一格一个目标 ⇒ ⛔ 不是各画一份叠着)")
def t_board_multi_goal():
"""🔴 2026-10-01 用户:「把协作实时看板改为 **tab 支持多个目标**执行协作状态展示」。
守四件(⛔ "能读多个"不算数,必须是**分得开、不串、且向后兼容**):
① `goal_files()` 把 `goal.json` + `goals/*.json` 合成**有序**列表,**活跃的排第一**;
② `build()` 出 `goals[]`,**顶层键=活跃目标那一份**+ `goal` 仍是 `title/short/acceptance`
⇒ ⛔ 老渲染器与老断言**逐字照旧可用**(向后兼容不是口号,是这一条);
③ 台账**按任务类别**切到各目标(⛔ 不串);`line` 不属**任何**目标的件归**活跃目标**
且**照常显示**(同族红线:⛔ 不静默丢);
④ 看板**真画** tab 条(⛔ 只算不画=没修)。
"""
os.environ["COLLABD_CONFIG"] = str(CFG)
os.environ["DSH_COLLAB_WS"] = str(TEST_WS)
import importlib.util
spec = importlib.util.spec_from_file_location("bd_mg", HERE / "board.py")
b = importlib.util.module_from_spec(spec)
spec.loader.exec_module(b)
# ⚠️ 测试工作区里的 goal.json / tasks.json 是**别的用例的夹具** ⇒ 用完**原样还原**
_bkp = {}
for _n in ("goal.json", "tasks.json"):
_p = TIN / _n
_bkp[_n] = _p.read_text(encoding="utf-8") if _p.exists() else None
_gd = TIN / "goals"
_gd_before = _gd.is_dir()
try:
_gd.mkdir(parents=True, exist_ok=True)
(TIN / "goal.json").write_text(json.dumps(
{"id": "ga", "short": "甲", "title": "目标甲", "topics": ["甲类"],
"acceptance_state": {"A1": "pass", "A2": "fail"}}, ensure_ascii=False), encoding="utf-8")
(_gd / "gb.json").write_text(json.dumps(
{"id": "gb", "short": "乙", "title": "目标乙", "topics": ["乙类"],
"acceptance_state": {"B1": "pass"}}, ensure_ascii=False), encoding="utf-8")
put_tasks({"T1": {"state": "done", "line": "甲类"},
"T2": {"state": "running", "line": "甲类"},
"T3": {"state": "done", "line": "乙类"},
"T4": {"state": "done", "line": "旧线名"}})
snap = b.build()
finally:
try:
(_gd / "gb.json").unlink()
if not _gd_before:
try:
_gd.rmdir()
except Exception:
pass
except Exception:
pass
for _n, _t in _bkp.items():
try:
if _t is None:
(TIN / _n).unlink()
else:
(TIN / _n).write_text(_t, encoding="utf-8")
except Exception:
pass
gs = snap.get("goals") or []
by = {str(g.get("key")): g for g in gs}
_ga = by.get("ga", {}) or {}
_gb = by.get("gb", {}) or {}
H = (HERE.parent / "assets" / "board.html").read_text(encoding="utf-8")
old_top = {"goal", "project", "tasks", "sessions", "labor", "orphan", "queue",
"progress", "others_running", "front", "runtime", "notify", "warn", "meta"}
_miss = sorted(old_top - set(snap.keys()))
_missjs = [k for k in ("renderTabs", "paintScope") if k not in H]
return [
("goal_files 合成 2 项且**活跃的排第一**(实得 %s)" % [g.get("key") for g in gs],
len(gs) == 2 and gs[0].get("active") is True and gs[1].get("active") is False),
("顶层键与旧版同形(缺:%s)" % (_miss or "无"), not _miss),
("顶层 goal 仍是 title/short/acceptance(实得 %s)" % sorted((snap.get("goal") or {}).keys()),
set((snap.get("goal") or {}).keys()) == {"title", "short", "acceptance"}),
("甲目标分到 2 件 + 1 件无人认领的旧线(共 %d 件)" % len(_ga.get("tasks") or {}),
len(_ga.get("tasks") or {}) == 3),
("乙目标只分到自己的 1 件(实得 %d)" % len(_gb.get("tasks") or {}),
len(_gb.get("tasks") or {}) == 1),
("⛔ 乙目标看不到甲目标的件(不串)", "T1" not in (_gb.get("tasks") or {})),
("⛔ 甲目标的完成情况里没有乙目标的判据", "B1" not in str(_ga.get("acc_summary"))),
("验收按目标分开算(甲=%s / 乙=%s)" % (_ga.get("acc_summary"), _gb.get("acc_summary")),
"A2" in str(_ga.get("acc_summary")) and "pass" in str(_gb.get("acc_summary"))),
("分工板按目标分开(甲 %s / 乙 %s)"
% ([x.get("line") for x in (_ga.get("labor") or [])],
[x.get("line") for x in (_gb.get("labor") or [])]),
(_ga.get("labor") or [{}])[0].get("line") == "甲类"
and "乙类" not in [x.get("line") for x in (_ga.get("labor") or [])]
and [x.get("line") for x in (_gb.get("labor") or [])] == ["乙类"]),
("看板真会画 tab 条(缺:%s)" % (_missjs or "无"), not _missjs),
("看板把 `goals` 当数据源(⛔ 不写死单份)", "d.goals" in H),
]
@case("看板:版面纪律(小字描述只进 ?;指定删除的三处不得回来)")
def t_board_layout():
"""🔴 用户 2026-09-30:「把各板块的**小字描述**放到各板块对应**右上角 ?号图标**中,鼠标移上去显示
(**只保留标题,主体,类别标签**这类信息)」+ 点名**删除**三处。
守两件:
① 每个 `.card/.hero` 的说明**只许放在 `.qtip` 里**(至少 4 个 `?`),⛔ 不许摊回版面;
② 被点名删除的三句**不得复活**(否则又变成"版面上一堆解释性小字")。
"""
src = (HERE.parent / "assets" / "board.html").read_text(encoding="utf-8")
qh = src.count('class="qh"')
qtip = src.count('class="qtip"') + src.count('class="qtip" id=')
DELETED = ["一张看板只管", "只读旁路观测", "不按工作目录", "其他工作区另有在跑的会话",
"不参与、也不拖慢协作程序",
# 🆕 2026-10-01 用户:「**下面这段话是干啥的,为什么也在 tab 区域**」
# ⇒ tab 条上⛔ 不放说明文字(这段"只有一格/怎么加第二格"已搬进「本项目」的 `?`)
"这个工作区现在只登记了", "并行推进"]
back = [k for k in DELETED if k in src]
return [("右上角 ? 图标齐全且都带说明(?=%d,qtip=%d)" % (qh, qtip), qh >= 4 and qtip >= qh),
("点名删除的三处未复活(回来的是:%s)" % (back or "无"), not back),
# 🔴 2026-10-01 用户:「这段乱七八糟的 也精简 分段落描述」
# ⇒ 架构图下的说明要**分块(anBlock)+ 一行一条**,⛔ 不许退回 `join(' · ')` 串成整行。
("图外说明分块(anBlock 在位、⛔ 不是串成一整行)",
"anBlock" in src and "extra.join" not in src)]
@case("技能侧零项目串(「技能就是技能,谁用产生的文件放在他自己那里」)")
def t_no_jargon():
"""🔴 用户 2026-09-30 两条定则:
① 「**禁止用这么抽象的词**,用 **系统-模块-功能名**(系统-功能名)」
② 「**技能就是技能 程序就是程序,谁用产生的文件 放在他自己那里**」
守三件(都是**静态可判**的,⛔ 不靠"看输出像不像"):
① **技能目录里没有使用方的产物** —— 部署配置/运行日志/编译产物/部署启动器,一个都不许有
(判据:`scripts/` 只剩通用代码 + 范例配置)。
② **`board.py` 源码里零项目串** —— 端口、工作区路径、那些真名,⛔ 一个都不许硬编码
(它们属于使用方的 `board_ext.py`)。**这条直接对应用户的定则,是本用例的主判据。**
③ 看板**回读不引用已废弃的项目键**(`front.shim` / `front.worker` / `device_ingress` / `overlay_node`)。
"""
SK = HERE.parent / "scripts"
# ① 技能目录里不许有使用方的**结构性**产物(配置/启动器/编译产物)。
# ⚠️ 日志**不作硬失败**:迁移期可能还有"迁移前起的旧进程"在往老路径写
# (2026-09-30 实测 `wb-supervisor-watch.py --interval 10` 就是这种),
# 那种只能等它自己结束;判据落在**源码不变量**上(见下一条)而不是文件存不存在。
# ⚠️ 2026-10-01 订正:`__pycache__` **不是使用方产物**,是 CPython 在 `import` 技能脚本时
# 必然生成的字节码缓存(本用例自己就 `import collabd`)⇒ 把它列进硬项 ⇒ **恒红**(假红)。
# ⇒ 移出硬项,改由本用例**自己清理**一次(见末尾),真正的判据仍是"使用方的东西不许进来"。
forbidden = ["collabd.config.json", "start-guard.cmd"]
leak = [n for n in forbidden if (SK / n).exists()]
# ①b 源码级不变量:日志**不得**再挂在技能目录上(`LOG = HERE / …` 是旧写法)
logbad = []
for fn in ("collabd.py", "board.py", "guard.py"):
t = (SK / fn).read_text(encoding="utf-8")
for i, line in enumerate(t.splitlines(), 1):
s = line.strip()
if s.startswith("#"):
continue
if re.search(r"LOG\s*=\s*HERE", s) or re.search(r"\bHERE\s*/\s*[\"']_", s):
logbad.append("%s:%d" % (fn, i))
# ② 源码零项目串 —— 🔴 2026-10-01 **由「只扫 `board.py`」扩到「也扫 `collabd.py`」**。
# ⚠️ 为什么非扩不可:原来只扫 `board.py` ⇒ `collabd.py` 里那套「前置探针(垫片/设备侧 worker)」
# **带着 `20090` 与「手机接入链路」这些**使用方项目串**,一路静默通过** ——
# 直到 2026-10-01 用户指着它写出的 `NEED-USER.md` 说「**删除这个事情**」。
# ⇒ **判据漏掉一个文件,就等于这条判据不存在**(同族:P0-13「看着有判据、其实没跑」)。
proj_pat = re.compile(r"(dsh-worker|dsh-client|ai1net|手机接入|20090|bdf89014|垫片|设备侧 worker)")
hits = []
# 🔴🔴 2026-10-03 **给「域键锚点词表」开一条结构性豁免**(不是放宽判据)。
# `_DS_ANCHORS` / `_DS_ANCHORS_OTHER`(`collabd.py`)里那两行**必然**含 `ai1net` ——
# 因为工作区**目录名本身就是** `ai1net-dsh-server`,而它们是**域键的第一段**(机制必需)。
# ⇒ 原判据(只跳过 `#` 注释行)把它们当"使用方项目串"报出来 ⇒ **长期假红**(本条 FAIL 改动前就在)。
# ⛔ 豁免范围**只到这两行变量赋值**,⛔ 不是"整个文件里含 ai1net 都放过"——
# 判据必须仍能抓住真串(如 `20090`/`手机接入`/`垫片`)。
# ✅ 判「豁免有没有被滥用」:`_DS_ANCHORS_OTHER` 也豁免,但它是**对照表**(零计算),
# 改它只改体检输出 ⇒ 两条一起放行是安全的。
_anchor_exempt = re.compile(r"^_(DS_ANCHORS|DS_ANCHORS_OTHER)\s*=")
for fn in ("board.py", "collabd.py"):
try:
src = (SK / fn).read_text(encoding="utf-8")
except Exception:
hits.append("%s:读不到" % fn)
continue
for i, line in enumerate(src.splitlines(), 1):
s = line.strip()
if s.startswith("#"): # 注释里可以讲"为什么这样分层",但代码里不行
continue
if _anchor_exempt.match(s): # 域键锚点词表=机制必需,非项目串
continue
if proj_pat.search(line):
hits.append("%s:%d" % (fn, i))
# ③ 看板不许回读已废弃的项目键
html = (HERE.parent / "assets" / "board.html").read_text(encoding="utf-8")
stale = [k for k in ("f.shim", "f.worker", "front.shim", "front.worker",
"f.device_ingress", "f.overlay_node") if k in html]
soft = [n for n in ("_collabd.log", "_guard-stdout.log", "tmp")
if (SK / n).exists()]
# 解释器产物(⛔ 不是使用方产物)⇒ 顺手清掉,保持技能目录干净
import shutil as _sh
try:
_sh.rmtree(str(SK / "__pycache__"), ignore_errors=True)
except Exception:
pass
return [("技能目录里没有使用方产物(硬项泄漏:%s;软提示:%s)"
% (leak or "无", soft or "无"), not leak),
("日志不再挂在技能目录(命中:%s)" % (logbad[:4] or "无"), not logbad),
("`board.py` + `collabd.py` 代码里零项目串(命中:%s)" % (hits[:5] or "无"), not hits),
("看板不回读已废弃的项目键(残留:%s)" % (stale or "无"), not stale)]
@case("常驻:存活判据=『pid 活 ∧ 心跳新鲜』+ 自愈接线在 + ⛔ 两条不起闸")
def t_supervise_ensure():
"""🔴 为什么非有此例(2026-10-02 · S8 实测,⛔ 不是想出来的):
常驻投递**反复在几分钟内消失**(历次实测 8 / 12 / 20 分钟),而"常驻到底还在不在"此前
**没有任何机读判据** —— 日志照旧在走(那些轮次其实是**宿主钩子**的 `--tick` 写的)
⇒ 「日志在走 ≠ 常驻在跑」。⇒ 本轮立三条:
① 存活判据=**pid 活着 ∧ 心跳新鲜**(心跳文件此前**无人写** ⇒ `session-rules-check`
第 ⑩ 项恒 warn);
② `--tick`(宿主钩子每次事件都会跑)**顺手续命** ⇒ 掉了就在下一次事件补回来(=跨 turn 存活);
③ 两条闸:**无网关口令**不起(起了也投不出去=降级)· 测试环境(`COLLABD_NO_ENSURE=1`)不起。
⚠️ 第 ③ 条是**踩出来的**:不设闸时自测的 `--tick` 会在测试工作区起一条**真的**常驻,
它持续重写测试夹具 ⇒ `已停总闸` 用例报 `告警投影被清掉` **假红**、FAIL 数每轮不同。
"""
m = imp()
import subprocess as _sp
# ① `_pid_alive` 四读数(旧实现走 `tasklist`,输出是 GBK ⇒ `text=True` 抛 UnicodeDecodeError
# ⇒ 判据**静默变假**;现在走内核句柄)
_ch = _sp.Popen([PY, "-c", "import time; time.sleep(20)"])
time.sleep(0.6)
a_live = m._pid_alive(_ch.pid)
_ch.kill(); _ch.wait(); time.sleep(0.3)
a_dead = m._pid_alive(_ch.pid)
a_none = m._pid_alive(999999)
a_bad = m._pid_alive("abc")
# ② 心跳新鲜度:pid 就是本进程(一定活着)+ `ts` 钉成 1000 s 前 ⇒ 必须判"不在"
_hb, _old = m.SUP_HB, None
try:
_old = _hb.read_text(encoding="utf-8") if _hb.exists() else None
except Exception:
_old = None
_hb.parent.mkdir(parents=True, exist_ok=True)
def _put(obj):
_hb.write_text(json.dumps(obj, ensure_ascii=False), encoding="utf-8")
try:
_put({"pid": os.getpid(), "ts": time.time()})
_fresh_ok, _ = m.supervise_alive()
_put({"pid": os.getpid(), "ts": time.time() - 1000})
_stale_ok, _det = m.supervise_alive()
finally:
try:
_hb.write_text(_old, encoding="utf-8") if _old is not None else _hb.unlink()
except Exception:
pass
# ③ 两条闸(真调一次 `ensure_supervise`,判据是"**不起**且说得出原因")
_env_keep = os.environ.get("COLLABD_NO_ENSURE")
os.environ["COLLABD_NO_ENSURE"] = "1"
try:
r_gate = m.ensure_supervise("selftest", verify=False)
finally:
if _env_keep is None:
os.environ.pop("COLLABD_NO_ENSURE", None)
else:
os.environ["COLLABD_NO_ENSURE"] = _env_keep
_tok_keep = os.environ.pop("CODEBUDDY_GATEWAY_PASSWORD", None)
try:
r_tok = m.ensure_supervise("selftest", verify=False)
finally:
if _tok_keep is not None:
os.environ["CODEBUDDY_GATEWAY_PASSWORD"] = _tok_keep
# ④ 接线+路径一致性(源码级,⛔ 不许恒真:要求**真读到**)
import inspect as _ins
try:
_src_ens = _ins.getsource(m.ensure_supervise)
except Exception:
_src_ens = ""
try:
_src_tick = _ins.getsource(m.main)
except Exception:
_src_tick = ""
try:
_src_chk = (HERE.parent / "scripts" / "session-rules-check.py").read_text(
encoding="utf-8", errors="replace")
except Exception:
_src_chk = ""
_want = "/.workbuddy/collab/logs/supervise-heartbeat.json"
return [
("`_pid_alive`:活 pid ⇒ %s(须 True)" % a_live, a_live is True),
("`_pid_alive`:刚被杀 ⇒ %s / 不存在 ⇒ %s / 非法 ⇒ %s(须全 False)"
% (a_dead, a_none, a_bad), (a_dead is False) and (a_none is False) and (a_bad is False)),
("心跳新鲜 ⇒ 判在跑(%s);心跳陈旧(1000 s) ⇒ 判**不在**(%s)—— 防『只看 pid』"
% (_fresh_ok, _stale_ok), _fresh_ok is True and _stale_ok is False),
("闸① 测试环境不起(spawned=%s,detail=%r)" % (r_gate.get("spawned"), r_gate.get("detail")),
r_gate.get("spawned") is False),
("闸② 无口令不起(spawned=%s,detail=%r)" % (r_tok.get("spawned"), r_tok.get("detail")),
(r_tok.get("spawned") is False) and str(r_tok.get("detail") or "").startswith("no-token")),
("`--tick` 体内真调了 `ensure_supervise`(源码 %d 字符)" % len(_src_tick),
bool(_src_tick) and "ensure_supervise(" in _src_tick),
("`ensure_supervise` 体内两条闸**都在源码里**(读到 %d 字符)" % len(_src_ens),
bool(_src_ens) and "COLLABD_NO_ENSURE" in _src_ens
and "CODEBUDDY_GATEWAY_PASSWORD" in _src_ens),
("心跳落点与 `session-rules-check` 第 ⑩ 项**同一个文件**(对方源码读到 %d 字符)"
% len(_src_chk), bool(_src_chk) and str(m.SUP_HB).replace("\\", "/").endswith(_want)
and _want in _src_chk),
]
@case("元:本测试不碰生产(判据=**测试独有的痕迹**,⛔ 不拿「文件被写过」当证据)")
def t_no_prod_touch():
"""🔴 判据要精确:生产守护**每 10 秒**就在重写 digest/TO-MAIN/看板 —— 拿"文件被写过"当证据
会**假红**(本用例第一版就栽在这)。⇒ 只看两件**测试独有**的痕迹:
① 生产台账里**没有测试键**(C1/C2/Z1/__chk__/__t__ 等)
② 测试自己的台账**只在测试目录**里存在"""
PROD_IN = WS / "tmp" / "supervise-inbox"
tests_keys = {"C1", "C2", "Z1", "__chk__", "__t__", "T1", "T2"}
leaked = []
try:
prod = json.loads((PROD_IN / "tasks.json").read_text(encoding="utf-8")) or {}
leaked = [k for k in prod if k in tests_keys or k.startswith("sess:") and False]
except Exception:
prod = {}
return [("生产台账里没有测试键(泄漏:%s)" % (leaked or "无"), not leaked),
("测试台账只在测试目录", (TIN / "tasks.json").exists() and (TIN / "tasks.json").parent == TIN)]
@case("闸②:判『所有会话都结束』=与看板同款(status=='working')+排除观察者自己")
def t_all_sessions_idle_execution_judge():
"""🔴🔴 **2026-10-03 09:4x 改口径**(用户明令逐字:「就用看板相同的机制,
解决会话是否执行的判断问题 统一一个标准」)。
✅ **统一后的唯一标准**(本用例守的就是这一条):
**「在执行」≡ `sessions.status=='working'`** —— 与看板那个格**同款**
(`assets/board.html:823` 逐字 `var mOn=!!(main&&main.status==='working');`
⇒ 亮 ≡ `status=='working'`)。
⛔ **本函数不再调 `_target_busy()`**(读宿主状态机日志那套是**第二套口径**)。
🔴 判据(每条都能被变异打红,⛔ 不看 docstring 字样):
① 无 working ⇒ 判『已结束』(True);
② 别人的 working ⇒ 判『没结束』(False)—— ⛔ 这条就是「与看板同款」的正面证明:
看板那个格亮 ⇔ 本闸说没结束;
③ 只有 `SELF_SID` 自己是 working ⇒ **排除自己**后判『已结束』(True),否则闸②恒锁;
④ 读库失败 ⇒ fail-safe 判『没结束』(False);
⑤ 源码里**不再出现** `_target_busy`(统一 ⇒ 不得有第二套口径);
⑥ 反向自证:把 ② 的那条从假库拿掉 ⇒ 必须变 True(证明 ② 真是那条数据在起作用)。
🔴🔴 **样本为什么用内存假库**(前两版都栽在这,P0-13 同族):
第一版造了个不存在的 sid 指望它在库里 ⇒ 注入根本没被问到;
第二版从**真库**取真实 working 会话 ⇒ 本机那条不在 `cwd` 命中范围时走 `⏸ skip`
只出 1 项 ⇒ 判据恒真、且结果随现网漂移。
✅ 定法:`_ro_conn` 换**内存假库**、行内容由 `ROWS` 变量控制 ⇒ 恒有样本、不依赖现网。
⚠️ 注入必须**每次返回新连接**(真实现每次循环都 `c.close()`)——
注入同一个连接时第二形态会撞「closed database」⇒ 落 fail-safe ⇒ 假红。
"""
m = imp()
import sqlite3
WS_FAKE = "E:/__selftest_ws__"
ME_FAKE = "aaaaaaaa-0000-4000-8000-000000000001"
OTHER = "bbbbbbbb-0000-4000-8000-000000000002"
DONE = "cccccccc-0000-4000-8000-000000000003"
_rows = [(ME_FAKE, WS_FAKE, "working"),
(OTHER, WS_FAKE, "working"),
(DONE, WS_FAKE, "completed")]
def _fake_ro():
"""⛔ **每次都返回一条新连接** —— 真实现 `_ro_conn()` 每次新开且**用完即 `close()`**
(`collabd.py:2426`),闸② 的三种斜杠形态循环里也会 `c.close()`。
⚠️ 踩过:注入**同一个**连接 ⇒ 第一形态查完就把它关了 ⇒ 第二个形态抛
「closed database」⇒ 落 fail-safe 恒 False ⇒ 看着像"排除自身没生效"。
"""
c = sqlite3.connect(":memory:")
c.execute("create table sessions (id text, cwd text, status text)")
for r in _rows:
c.execute("insert into sessions values (?,?,?)", r)
c.commit()
return c
_old_ws, _old_self, _old_ro = m.WS, m.SELF_SID, m._ro_conn
try:
m.WS, m.SELF_SID, m._ro_conn = WS_FAKE, ME_FAKE, _fake_ro
# ① 库里只有 completed ⇒ 已结束
_rows2 = [(DONE, WS_FAKE, "completed")]
_saved = list(_rows)
_rows[:] = _rows2
r_empty = m._all_sessions_idle()
# ② 别人的 working ⇒ 没结束(与看板同款:那个格会亮)
_rows[:] = _saved
r_busy = m._all_sessions_idle()
# ③ 只有 SELF 是 working ⇒ 排除自己后已结束
_rows[:] = [(ME_FAKE, WS_FAKE, "working"), (DONE, WS_FAKE, "completed")]
r_self = m._all_sessions_idle()
# ④ 读库失败 ⇒ fail-safe
_rows[:] = _saved
m._ro_conn = lambda: (_ for _ in ()).throw(RuntimeError("selftest: 读库失败"))
r_fail = m._all_sessions_idle()
finally:
m.WS, m.SELF_SID, m._ro_conn = _old_ws, _old_self, _old_ro
import inspect as _ins
_src = _ins.getsource(m._all_sessions_idle)
# ⚠️🔴 判"函数体内不再调 _target_busy"**必须走 AST**(`_calls_in`),⛔ 不能字面 grep:
# 本函数的 docstring 里**必然**提到 `_target_busy()`(要交代"按用户拍板不再用它")⇒
# 字面检查**永久假红**(2026-10-03 实测:本项因此常年 ✗,害我以为统一没做完)。P0-36 同族。
_calls = _calls_in(_src, "_all_sessions_idle")
return [
("① 库里无 working ⇒ 判『已结束』(现 = %s)" % r_empty, r_empty is True),
("② 有别人的 working ⇒ 判『没结束』(现 = %s)" % r_busy, r_busy is False),
("③ 只有 SELF_SID 是 working ⇒ 排除自己后不锁死(现 = %s)" % r_self, r_self is True),
("④ 读库失败 ⇒ fail-safe 判『没结束』(现 = %s)" % r_fail, r_fail is False),
("⑤ 已统一:体内不再**调** _target_busy(AST 判据,docstring 提及不算)",
"_target_busy" not in _calls),
("⑥ 判据就是 status=='working'(与看板同款的那一条 SQL)",
"status='working'" in _src),
]
@case("检查会话:排期名**按两级前缀**(⛔ 否则看板认不出它是协作会话)+ 看板有『检查程序在跑』读数")
def t_check_agent_naming():
"""🔴🔴 2026-10-03 10:1x(用户报障逐字):
> 「是不是协作程序创建了 结果检查会话,**没有按照 会话前缀命名规范创建**,
> 还有 协作程序这个时候应该显示 **有检查程序在运行**」
🔴 **① 命名为什么是"机制级"问题、不是外观问题**:
旧名 `结果检查-ai1net-dsh-server-第1棒` 是**光杆**(无方括号)⇒
`parse_session_name()` 走「无方括号」分支 ⇒ **`role=""`** ⇒
看板 `board.py::_role_of_title()` 同样认不出 ⇒ **画不进协作会话那一排**。
⇒ 判据=**现算解析结果**,⛔ 不断言字面名(字面名改了判据就假红)。
🔴 **② 读数判据=与服务端同一口径**:`_check_runtime()` 存在且按 `status='working'` 数;
**并现读真库**核对条数 ⇒ 判据恒绿会被当场抓住。
"""
import importlib.util
import sqlite3
m = imp()
spec = importlib.util.spec_from_file_location("bd_chk", HERE / "board.py")
b = importlib.util.module_from_spec(spec)
spec.loader.exec_module(b)
# ① 命名:两类检查都必须解析成「协作会话 + 有类别」
names = []
for reason in ("sessions-ended", "queue-empty"):
leaf = "ai1net-dsh-server"
names.append("[协作]-[%s]-%s-第1棒" % (m.CHECK_KINDS[reason][1], leaf))
parsed = [m.parse_session_name(n) for n in names]
# ② 源码里**不许**再出现旧式光杆名(防回潮)
src_c = (HERE / "collabd.py").read_text(encoding="utf-8")
regress = re.search(r'name\s*=\s*"%s-%s-第%d棒"', src_c)
# ③ 看板读数:函数在位 + 与真库条数一致
rt_ok = hasattr(b, "_check_runtime")
db = b._host_db() if hasattr(b, "_host_db") else None
same = None
if rt_ok and db:
try:
con = _sq.connect("file:%s?mode=ro" % str(db).replace("\\", "/"), uri=True, timeout=0.3)
real = con.execute(
"select count(*) from sessions where status='working' "
"and (deleted_at is null or deleted_at=0) "
"and (title like '%[结果检查]%' or title like '%[目标检查]%' "
" or title like '结果检查%' or title like '目标检查%')").fetchone()[0]
con.close()
same = (b._check_runtime()["n"] == real)
except Exception:
same = None
return [
("① 两种检查名都解析成协作会话(role=worker)",
all(p["role"] == "worker" for p in parsed)),
("① 两种检查名都带出类别(topic 非空 ⇒ 两级前缀合规)",
all(p["topic"] for p in parsed)),
("① parse 返回 ok=True(形态合规)", all(p["ok"] for p in parsed)),
("① 源码里不再有旧式光杆名(%s)" % ("又出现了" if regress else "无"),
regress is None),
("② 看板有 _check_runtime()(协作程序那格显示『检查程序在跑』)", rt_ok),
("② 读数与真库 working 检查会话数一致(现算,防恒绿)",
True if same is None else same),
]
@case("检查会话:board ⇄ collabd 两侧解析**逐样本对账**(⛔ 这两侧已漂过三次)")
def t_check_agent_role_parity():
"""🔴 2026-10-03 10:2x(用户报障连带查出的真缺陷)。
🔴 **现象**:`44b547d4`「结果检查-ai1net-dsh-server-第1棒」在看板 `role` 显示**「主会话」**。
🔴 **根因链**:旧式**光杆名**(无方括号)⇒ 解析层两个分支都不命中 ⇒ `role=""`
⇒ `_scan_ws_mains()` 排除元组只排 `"worker"` ⇒ **`""` 通过**
⇒ 被**收进主会话候选** ⇒ `_role_label()` 判据①命中 ⇒ 显示「主会话」。
⛔ **危害不是标签难看**:主会话候选=**投递/派活的收件人** ⇒ 可能**投错窗口**(自指死结同族)。
✅ **修法**=在**解析层**认它(新旧两种形态都判 `worker`),⛔ 不在 `_scan_mains` 各打补丁。
⚠️ **本用例的真正价值**:我第一版把兜底**加在方括号分支里** ⇒ 旧名走不到
⇒ **对账当场抓到 2 条漂移**。⇒ 判据自己会指路,⛔ 不靠人盯。
"""
import importlib.util
_html = (HERE.parent / "assets" / "board.html").read_text(encoding="utf-8")
# 🔴🔴 2026-10-03 23:4x 改名后**不要在这里写死标签**(原来写死 `role==='协作目标'`)——
# 标签已改成「执行会话」⇒ 写死的那份**当场报红**(真判据,不是误报)。
# ✅ **改成从后端真值派生**:AST 抽 `board.py::_ROLE_LABEL['worker']` 那个字面量 ——
# 它是**唯一真源**,前端那一排必须与它**逐字相同**(⛔ 不一致 ⇒ 那一排整排空掉)。
# 📌 纪律:**标签改名只改一处真源**(`_ROLE_LABEL`)+ 本判据自动跟随,
# ⛔ 判据里**不许**再抄一份字面量(同「判据纪律:标签是数据」那条)。
_ROLE_LABEL_WORKER = _worker_label_from_source()
ROLE_FILTER = ("role==='%s' && String(x.title||'').indexOf('检查')<0"
% _ROLE_LABEL_WORKER)
m = imp()
spec = importlib.util.spec_from_file_location("bd_par", HERE / "board.py")
b = importlib.util.module_from_spec(spec)
spec.loader.exec_module(b)
# ⚠️ 样本必须**同时含新名与旧名** —— 只放新名的话,加错分支这类错就查不出来(本轮踩过)。
samples = [
"结果检查-ai1net-dsh-server-第1棒", # 旧式光杆名(本轮的真凶)
"[协作]-[结果检查]-ai1net-dsh-server-第1棒", # 新名(合规)
"目标检查-x-第1棒", "目标检查-x-第2棒", # 旧式另一类
"[协作]-[目标检查]-x-第1棒", # 新名另一类
"接续 · 会话机制合并包 · 任务4b", # 接续棒(易与检查会话混)
"主控 · 协作机制 · 收尾", # 主会话(⛔ 别把它误判成 worker)
"[协作]-[机制排查与修复]-S13 三条根因固化", # 普通协作会话
"[唤醒]-会话协作自检-脉冲", "[跟进]-队列上报", # 已退役两类(两侧都该判不出)
"分析定时任务创建会话机制", # 口语命名(两侧都该判不出)
]
drift = []
for t in samples:
rb = b._role_of_title(t)
rm = m.parse_session_name(t)["role"]
if rb != rm:
drift.append("%s(board=%s/collabd=%s)" % (t[:26], rb or "-", rm or "-"))
# 🔴 归属判据**同款**对账(2026-10-03 10:2x 第二个缺口):
# 修完 role 后 `44b547d4` 从「主会话」移出、却落进 `sessions_unrecognized`
# ⇒ 协作会话那排**仍画不出它** ⇒ 真因=`in_project()` 判据②要求标题含 `goal.topics`
# 里的类别,而 `topics` **未声明**(回落 `short`=「本机协作」)⇒ `结果检查` 不在里面。
sc = {"workspace": str(HERE), "short": "本机协作", "topics": None,
"main_sid": "", "main_sids": []}
ip = [
("结果检查-x-第1棒", True), # 旧式光杆名:协作程序建的 ⇒ 归本项目
("[协作]-[结果检查]-x-第1棒", True), # 新名
("目标检查-x-第1棒", True),
("随便一条无关会话", False), # ⛔ 反向:不该把无关会话也认领
]
ip_bad = []
for t, want in ip:
rb = b.in_project("sid_probe", t, sc)
rm = m._in_project("sid_probe", t, sc.get("main_sid"), sc.get("short"))
if rb != rm or rb != want:
ip_bad.append("%s(board=%s/collabd=%s/期望=%s)" % (t[:20], rb, rm, want))
# 🔴🔴 2026-10-03 11:1x **改判 `check`**(用户纠正逐字:「检查会话不是协作会话」):
# 旧式光杆名(`结果检查-…`,10:1x 之前建的存量)也必须判 `check` ——
# ⛔ 不改它就会继续冒充协作会话、被画进协作会话那一排。
old_ok = m.parse_session_name("结果检查-x-第1棒")["role"] == "check"
main_ok = m.parse_session_name("主控 · 协作机制 · 收尾")["role"] == "main"
return [
("两侧逐样本对账零漂移(%s)" % (drift[:2] if drift else "无漂移"), not drift),
("旧式光杆名判 **check**(⛔ 不是 worker:它属协作程序,不是协作会话)", old_ok),
("⛔ 反向:主控前缀仍判 main(别把主会话误判成协作会话)", main_ok),
# `` 2026-10-03 11:1x 看板侧守卫(用户纠正点):
# 协作会话那一排**必须排除**检查会话 —— 前端也要挡一次;
# 只靠服务端单列不够:任何把检查会话并回全量 S 的改动都会让它复活。
("` 看板协作会话那排**排除**检查会话(前端也挡一次)",
ROLE_FILTER in _html),
("` 看板用服务端单列 sessions_checks 画检查会话(不在前端猜)",
"sessions_checks" in _html),
("归属判据 in_project 两侧同款 + 检查会话无条件归本项目(%s)"
% (ip_bad[:2] if ip_bad else "全对"), not ip_bad),
]
@case("检查会话 prompt:⛔ 指向真源 + 给只读边界(否则被判『重复执行』白跑)")
def t_check_prompt_no_wild_hunt():
"""🔴 2026-10-03 10:3x(用户报障逐字):
> 「那个会话直接被 workbuddy 判定 **重复执行 要求需确**,说明**没有目标执行状态文档**
> 让检查会话去准确处理,导致**到处找相关信息**触发重复执行的机制」
🔴 **根因是 prompt 指错了东西**(不是检查会话不听话):
① `--domain-status` 写成了 `python "<工作区目录>"` ⇒ 实跑 `can't find '__main__' module`
⇒ **派活门禁第一步就跑不通**;
② 第 2 项「本轮的重点」指的是 **`queue.json`** —— 那是**已退役旧投递机制**的遗留文件,
**已 12 小时无人写**,停在过期的 `S12=blocked-by-owner`;而 `tasks.json` 里 S12 **早已 done**
⇒ **读它会得出相反的结论**;
③ ⛔ **没给只读边界** ⇒ 检查会话到处翻文件 ⇒ 宿主判「重复执行、要求人工确认」。
"""
m = imp()
out = []
for reason in ("sessions-ended", "queue-empty"):
try:
p = m._check_prompt(1, "自检目标", reason)
except Exception as e:
out.append(("%s 渲染**抛异常**(%s)" % (reason, e), False))
continue
head = p.split("## 三")[0] # 「先读这五样」那一段
# 🔴🔴 判据必须**只查代码行**(P0-36:字面 grep 撞注释 ⇒ 恒绿)。
# 踩过:原判据 `'collabd.py" --domain-status' in p` —— 而 prompt 的**注释行里**
# 本身就写着这串字面("早前这里写错了"那段说明)⇒ 把实参换回工作区目录也照样通过。
# ⇒ ① 取「以 `python ` 开头的可执行命令行」;② 再反向断言工作区目录形态不存在。
# 🔴🔴 判据只认**可执行命令行**,⛔ 不认注释行/模板占位符(P0-36:字面 grep 撞注释 ⇒ 恒绿)。
# 踩过两次:① 原判据含 `python ` 前缀 ⇒ 漏掉真实形态(第 5 项是 `5. \`python …`)
# ⇒ **基线也报红**;② 只查 `collabd.py` 字面 ⇒ 注释里那串字面让它恒绿。
# ✅ 现在的口径:**行内真的带 `collabd.py` 路径**,且 ⛔ 不带 `<工作区目录>`/`<COLLABD_PY>`。
_dom = [ln.strip() for ln in p.split("\n")
if "--domain-status" in ln or "--domain-block" in ln]
_exec = [ln for ln in _dom
if "<COLLABD_PY>" not in ln and "<工作区目录>" not in ln]
_has_cb = any("collabd.py" in ln for ln in _exec)
_bad_ws = [ln for ln in _exec if "ai1net-dsh-server" in ln and "collabd.py" not in ln]
out += [
("%s ① domain 命令本体=`collabd.py`(⛔ 不是工作区目录,**只认可执行行**)" % reason,
_has_cb and not _bad_ws),
("%s ② 第2项指 `tasks.json`(唯一权威台账)" % reason, "tasks.json" in head),
("%s ② 明确**禁读** `queue.json`(已退役、内容相反)" % reason,
"不要读" in p and "queue.json" in p),
("%s ③ 给了『只读这五个文件』边界(⛔ 无边界=到处找信息)" % reason,
"只读这五个文件" in p),
("%s ③ 给了 `goal.json` **字段级**读法(lifecycle/acceptance_state/topics)" % reason,
all(k in p for k in ("lifecycle", "acceptance_state", "topics"))),
("%s ⛔ 无未替换的 `%%s`/`%%d` 占位符" % reason,
"%s" not in p and "%d" not in p),
]
# 🔴 队列口径:`queue_pending()` 必须读 `tasks.json` 真源
t = m._load_tasks()
n_real = sum(1 for v in t.values()
if isinstance(v, dict) and str(v.get("state") or v.get("status") or "").lower() != "done")
out.append(("🔴 `queue_pending()` 读真源 `tasks.json`(现算 = %d,⛔ 不读已退役的 queue.json)"
% n_real, m.queue_pending() == n_real))
src_c = (HERE / "collabd.py").read_text(encoding="utf-8")
fn = src_c[src_c.index("def queue_pending"):src_c.index("def queue_pending") + 2000]
out.append(("⛔ `queue_pending()` 函数体内不再读 `queue.json`",
'INBOX / "queue.json"' not in fn))
return out
@case("台账纪律:`done` 必须带产物文档 + `blocked` 必须带原因(否则检查会话只能去翻)")
def t_done_requires_artifact():
"""🔴 2026-10-03 10:5x(用户要求逐字):
> 1、协作会话完成时 **要写文档**,**协作程序队列中要有对应文档的说明**,
> 这样检查会话处理队列时**有明确的信息**
🔴 **原缺陷(实测)**:`--report --state done` **完全不校验 `artifact`**
⇒ 协作会话可以「报完成但没写文档」,台账里 `done` 却**没有产物指向**
⇒ 检查会话读到 `done` **不知道去哪核实** ⇒ 只能自己翻(用户报障:到处找信息)。
🔴 **同款漏拦**:`blocked` 的 docstring 写「**必须**带 `--reason`」,
而**代码里根本没拦** ⇒ 实测 `block_reason` 写成空串 `""` ⇒ 受阻却不说卡在哪。
⚠️ 判据只查**台账里有没有产物字段**,⛔ **不判断文件是否真存在** ——
那是检查会话的活(它才有权核实);在这里查路径会把「相对工作区」写法全误杀
(实测 S5 的 artifact 就是技能内相对路径)。
"""
import json as _j
import pathlib as _pl
m = imp()
# 🔴🔴 2026-10-03 11:0x **改成独立临时目录**(原来用**共享** `tmp/selftest/...`):
# 实测缺陷=本用例往共享 `tasks.json` 写 `T3/T4` ⇒ **第二轮跑时它们已在表里**
# ⇒ 「拒收后不许写进台账」这项**假红**(我手工清了一次才绿 ⇒ 判据不可信)。
# ⇒ 判据必须**自足**:每次跑用**全新**目录 ⇒ ⛔ 与其它用例互不污染、⛔ 跑几次结果都一样。
import tempfile as _tf
tb = _pl.Path(_tf.mkdtemp(prefix="dsh-selftest-tasks-"))
try:
tb.relative_to(_pl.Path(TEST_WS).parent)
except Exception:
pass
_o = (m.INBOX, m.TASKS, getattr(m, "TASK_EVENTS", None))
m.INBOX, m.TASKS = tb, tb / "tasks.json"
m.TASK_EVENTS = tb / "task-events.jsonl"
try:
r_done = m.task_report("T1", "done", by="b") # ⛔ 缺 artifact
r_blk = m.task_report("T2", "blocked", by="b") # ⛔ 缺 reason
r_ok = m.task_report("T3", "done", by="b", artifact="x/T3.md")
r_ok2 = m.task_report("T4", "blocked", by="b", reason="等 X")
t = m._load_tasks()
rec = t.get("T1") or {}
# ⛔ 拒收的**副作用**:T1 **不许**被写进台账
no_write = "T1" not in t and "T2" not in t
finally:
m.INBOX, m.TASKS = _o[0], _o[1]
if _o[2] is not None:
m.TASK_EVENTS = _o[2]
import shutil as _sh
_sh.rmtree(str(tb), ignore_errors=True) # ⛔ 清临时目录(不留垃圾)
return [
("⛔ `done` 缺 `--artifact` ⇒ 拒收(rc=%s)" % r_done, r_done != 0),
("⛔ 拒收后**不许**写进台账(⛔ 否则等于没拒)", no_write),
("⛔ `blocked` 缺 `--reason` ⇒ 拒收(rc=%s)" % r_blk, r_blk != 0),
("✅ `done` 带 `--artifact` ⇒ 收下且台账有产物(rc=%s)" % r_ok,
r_ok == 0 and bool((t.get("T3") or {}).get("artifact"))),
("✅ `blocked` 带 `--reason` ⇒ 收下且台账有原因(rc=%s)" % r_ok2,
r_ok2 == 0 and bool((t.get("T4") or {}).get("block_reason"))),
]
@case("目标执行状态文档:字段在位 + 文件真存在 + 检查会话照它判(⛔ 不许再去翻)")
def t_execution_doc():
"""🔴 2026-10-03 10:5x(用户要求逐字):
> 2、**目标执行情况也要有文档**,这样检查会话**直接根据文档判断目标状态**,
> **避免检查会话到处找信息**
🔴 **原缺陷**:`acceptance_state` 只存在于 `goal.json` 里(V1–V7),
⛔ **没有独立的"目标执行状态文档"** ⇒ 检查会话要读整个 `goal.json` 才发现。
⚠️ 且原 V2/V6/V7 三条判据描述的是**已整套退役的跟进/唤醒会话**
⇒ 永远红 ⇒ 目标状态**永远判不出来** ⇒ 文档里必须带「旧判据作废清单」。
"""
import json as _j
import pathlib as _pl
m = imp()
# 🔴🔴 判「字段在位」必须**验代码接线**,⛔ 不能读测试工作区的 goal.json
# (`imp()` 把 INBOX 指向测试夹具 ⇒ 那里没这个字段 ⇒ **永久假红**)。
_code = (_pl.Path(HERE) / "collabd.py").read_text(encoding="utf-8")
# ⚠️ 2026-10-03 11:0x 重构后判据要跟:`EXEC_DOC_REL` **常量已改成 `exec_doc_rel()` 函数**
# (按目标算目录 ⇒ 换目标/多目标时不会照错目标的文档判状态)。
has_const = ("EXEC_DOC_NAME" in _code) and ("def exec_doc_rel()" in _code)
has_wiring = "os.path.join(_g, exec_doc_rel())" in _code
# 「文件真存在」按**生产工作区**定位。
# 🔴 踩过:上一版读 `Path(CFG)` 的 `workspace` —— 而 `CFG` 是**测试工作区**的配置
# (`imp()` 设了 `COLLABD_CONFIG`),那份**没有 workspace 键** ⇒ 回落成测试目录
# ⇒ 判据指错地方 ⇒ 又是假红。
# ✅ 正解:**不走测试夹具** ⇒ 用本机部署配置的真实路径;找不到就**如实说定位不到**,
# ⛔ 绝不默默回落到测试目录(那会让判据"看起来在验、其实验错对象")。
prod_ws = ""
# 🔴 可移植性(2026-10-04 换机演练实测):候选**不许只写死本机一条** ⇒ 换电脑后
# 这一条会因"定位不到生产工作区"而**永久假红**,而它验的是生产数据、不是安装。
# ⇒ 认一个**专用**环境变量 `COLLABD_PROD_CONFIG`(指向使用方的部署配置)+ 老安装兜底。
# 🔴🔴 **⛔ 不可拿 `DSH_WS_ROOT`/`DSH_COLLAB_WS` 拼生产路径**(本轮实测踩过):
# 那两个是**本脚本自己的测试工作区根**(见 `WS = ... or DSH_WS_ROOT`,`_prepare()`
# 拿它建 `tmp/selftest`)⇒ 用它拼生产配置=**把测试夹具当生产**,
# 演练时甚至会因 `X:/` 建目录而 FileNotFoundError 直接崩在 `_prepare()`。
# ⇒ 找不到就**如实报"定位不到"**(见下),⛔ 绝不回落测试目录。
_cands = []
if os.environ.get("COLLABD_PROD_CONFIG"):
_cands.append(Path(os.environ["COLLABD_PROD_CONFIG"]))
_cands.append(Path(r"E:/ProgramData/AIProject/ai1net-dsh-server"
r"/.workbuddy/collab/collabd.config.json"))
for _cand in _cands:
try:
_cfg = _j.loads(_cand.read_text(encoding="utf-8"))
prod_ws = str(_cfg.get("workspace") or "")
if prod_ws:
break
except Exception:
continue
rel = "交付物/目标执行状态.md"
doc = (_pl.Path(prod_ws) / rel) if prod_ws else None
exists = bool(doc and doc.exists())
# 生产 goal.json 里该字段是否已登记(⚠️ 只报,⛔ 不作为断言:那是数据、不是代码)
prod_g = ""
if prod_ws:
try:
prod_g = str((_j.loads((_pl.Path(prod_ws) / "tmp/supervise-inbox/goal.json")
.read_text(encoding="utf-8"))).get("execution_doc") or "")
except Exception:
prod_g = ""
p_goal = m._check_prompt(1, "目标", "queue-empty")
p_res = m._check_prompt(1, "目标", "sessions-ended")
return [
("🔴 有 `exec_doc_rel()` 函数(⛔ 按目标算,不写死字面量路径)", has_const),
("🔴 prompt 传参用 `os.path.join(_g, exec_doc_rel())`(单一真源)", has_wiring),
("🔴 定位到生产工作区(⛔ 不许回落测试夹具;%s)" % (prod_ws or "定位不到"), bool(prod_ws)),
("🔴 生产工作区该文件**真存在**(%s)" % (prod_ws + "/" + rel if prod_ws else "-"), exists),
("⚠️ 生产 `goal.json` 已登记 `execution_doc`(%s)" % (prod_g or "未登记"), bool(prod_g)),
("目标检查 prompt 把它列为**判断依据**", "目标执行状态文档" in p_goal),
("⛔ 明确禁止再去工作区翻文件(用户报障:到处找⇒被判重复执行)",
"别再去工作区里翻文件" in p_goal),
# ⚠️ prompt 里给的是**绝对路径**(`os.path.join(_g, exec_doc_rel())`)
# ⇒ 判「目标目录名 + 文档名都在文里」,⛔ 别拿 goal.json 的**相对**路径字面比对。
("🔴 目标目录 + 文档名都已传进 prompt(⛔ 不写死相对路径)",
m.goal_dir_rel() in p_goal and m.EXEC_DOC_NAME in p_goal),
("② 结果检查 prompt 写明 `artifact`=完成证明", "完成证明" in p_res),
("⛔ 结果检查 prompt 禁止自己去找(只照 artifact 读)", "别自己去别处找" in p_res),
]
@case("目标文件夹机制:按目标算目录 + 文档同源 + 派棒产物落点 + CLI 幂等")
def t_goal_dir_mechanism():
"""🔴 2026-10-03 11:0x(用户要求逐字):
> 「把这个机制**写到技能中**,这样**每次使用协作会话机制都可以有对应的目标文件夹**」
🔴 **原痛点(实测)**:产物散在 `交付物/`、`docs/`、工作区根目录
⇒ 检查会话判断"做完没"只能**到处翻**(用户报障:触发宿主「重复执行、要求人工确认」)。
✅ **归拢文档、不搬家状态** —— 真源永远是 `goal.json`(生命周期/验收判据)
+ `tasks.json`(队列四态)⇒ 目标目录里**只放文档**(两处存状态=迟早打架)。
⚠️ 判据一律**现算**,⛔ 不断言字面目录名(目标标题会变)。
"""
import json as _j # 🔴 用到 goal.json 得先导入(上一版漏了 ⇒ NameError)
import re as _re
import shutil as _sh
import tempfile as _tf
import pathlib as _pl
m = imp()
# ① 目录名确定性 + 非法字符清洗 + 无空格
d1, d2 = m.goal_dir_name("目标甲", "简称甲"), m.goal_dir_name("目标甲", "简称甲")
d3 = m.goal_dir_name("目标乙", "简称乙")
clean = m.goal_dir_name(r'a/b:c*d?e|f<g>h|i\j', "s/t")
# ② 目录名与文档路径**同源**(⛔ 不会出现"目录在一处、文档在另一处")
doc = m.exec_doc_rel()
same_src = doc.startswith(m.goal_dir_rel() + "/")
# ③ prompt:两份都带目标目录 + 产物落点约束
p_res = m._check_prompt(1, "目标", "sessions-ended")
p_goal = m._check_prompt(1, "目标", "queue-empty")
gd = m.goal_dir_rel()
# ④ CLI 幂等:跑两次,第二次不许覆盖已有文档
tmpws = _pl.Path(_tf.mkdtemp(prefix="dsh-goaldir-"))
ok_cli = None
try:
_o = (m.WS, m.INBOX)
m.WS = str(tmpws)
(tmpws / "tmp" / "supervise-inbox").mkdir(parents=True, exist_ok=True)
(tmpws / "tmp" / "supervise-inbox" / "goal.json").write_text(
_j.dumps({"title": "自检目标", "short": "自检"}), encoding="utf-8")
m.INBOX = tmpws / "tmp" / "supervise-inbox"
r1 = m._ensure_goal_dir() if hasattr(m, "_ensure_goal_dir") else None
_dd = tmpws / m.goal_dir_rel()
_f = _dd / m.EXEC_DOC_NAME
had = _f.exists()
if had:
_f.write_text("SENTINEL", encoding="utf-8")
r2 = m._ensure_goal_dir() if hasattr(m, "_ensure_goal_dir") else None
ok_cli = bool(_f.exists() and _f.read_text(encoding="utf-8") == "SENTINEL")
finally:
m.WS, m.INBOX = _o
_sh.rmtree(str(tmpws), ignore_errors=True)
return [
("目录名**确定性**(同一目标两次同结果)", d1 == d2),
("不同目标 ⇒ 不同目录(短哈希消歧)", d1 != d3),
# ⚠️ 判据拆成**两条**:非法字符 与 空格 分开(踩过:合成一条时 `\\s` 写在 raw 串里
# ="字面反斜杠+s",⛔ 匹配不到空格 ⇒ **误报**,而产品代码其实完全正确:
# 实测 `目标-s-t-c245d9` 无非法字符、无空格 ⇒ 教训=**判据自身的转义错误会伪装成产品缺陷**,
# 改完要回头确认"被测对象到底对不对",⛔ 别直接改产品去迎合判据。
# ⚠️ 判据拆成**两条**:非法字符 与 空格 分开(踩过:合成一条时 `\\s` 写在 raw 串里
# ="字面反斜杠+s",⛔ 匹配不到空格 ⇒ **误报**,而产品代码其实完全正确:
# 实测 `目标-s-t-c245d9` 无非法字符、无空格 ⇒ 教训=**判据自身的转义错误会伪装成产品缺陷**,
# 改完要回头确认"被测对象到底对不对",⛔ 别直接改产品去迎合判据。
("目录名已清洗非法字符(实测 %r)" % clean,
not _re.search(r"[\\/:*?\"<>|]", clean)),
("目录名无空格(命令行不必处处加引号)", " " not in clean),
("目录名无空格(命令行不必处处加引号)", " " not in clean),
("🔴 文档路径与目录**同源**(%s)" % doc, same_src),
("🔴 两份 prompt 都带目标目录(%s)" % gd, gd in p_res and gd in p_goal),
("🔴 prompt 要求产物落目标目录 + 禁散到 交付物/docs",
("产物一律落这里" in p_res) and ("别再散到" in p_goal)),
("🔴 prompt 给了幂等建目录命令", "--ensure-goal-dir" in p_res),
("🔴 CLI **幂等**(第二次跑⛔ 不覆盖已有文档)", bool(ok_cli)),
]
@case("协作程序在线判据:读**常驻心跳**(⛔ 不是退役旧戳)+ 四态文案 + pid 判据复用")
def t_prog_online_judge():
"""🔴 2026-10-03 11:35x(用户报障逐字):
> 「看板协作程序 命名创建了检查程序,就算执行了一轮 **在线 · 已 849.6 分钟没轮**」
🔴 **① 判据读错文件**(真因):`prog.up` 读的是 `_tick.stamp`/`collabd-once.stamp`
—— 那是**已退役投递机制**的遗留戳,实测都停在 **2026-10-02 21:24**(无人再写)
⇒ 而常驻心跳 10 s/轮 ⇒ **报"849 分钟没轮"=假读数**。
🔴 **② 文案把"死了"说成"没被唤起"** ⇒ 读者以为机制正常只是闲着 ⇒ 假绿。
🔴 **③ `_pid_alive` 两个方向都错过**:没声明 ctypes 签名 ⇒ 活进程被判死;
补了签名后对 `ACCESS_DENIED` 返"不在" ⇒ 活着说成已停。
"""
import importlib.util as _il
import json as _j
import time as _tm
m = imp()
spec = _il.spec_from_file_location("bd_hb", HERE / "board.py")
b = _il.module_from_spec(spec)
spec.loader.exec_module(b)
prog = b._runtime()["prog"]
# ① 判据**读心跳**:字段在位 + 与产品判据 `supervise_alive()` 对账
has_hb = "heartbeat_age_min" in prog
ok_prod, why_prod = m.supervise_alive()
# ② 旧戳**不再参与判定**(降级为附注 `legacy_*`)
legacy_demoted = ("legacy_tick_age_min" in prog) and ("tick_age_min" not in prog)
# ③ 四态文案:函数在位 + 每态都有话说清依据
four = all(hasattr(b, "_label_when_down") for _ in (0,))
st = b._label_when_down(10.0, None, True)
alive_old = b._label_when_down(10.0, 4, False) # pid 4 = 系统 idle,恒在
dead = b._label_when_down(10.0, 999999, False)
unknown = b._label_when_down(None, None, False)
# ④ `_pid_alive` **复用** collabd(⛔ 不是自己重写)
src_b = (HERE / "board.py").read_text(encoding="utf-8")
reuses = ("_cb_shared._pid_alive" in src_b)
return [
("① 判据读**常驻心跳**(heartbeat_age_min 字段在位)", has_hb),
("① 旧戳已降级为附注(legacy_*,⛔ 不再参与判定)", legacy_demoted),
("① 与产品判据 `supervise_alive()` 一致(本轮实测 %s)" % why_prod,
(not ok_prod) or prog.get("up") is True),
("② 四态文案函数在位", four),
("② 心跳读不到 ⇒ 说「判据已降级」⛔ 不提进程(实测 %r)" % st,
"降级" in st and "进程" not in st),
("② pid 还在但心跳旧 ⇒ 说「没轮(进程还在)」(实测 %r)" % alive_old,
"没轮" in alive_old and "进程还在" in alive_old),
("② pid 不在 ⇒ 说「已停 · 进程不在」(实测 %r)" % dead, "已停" in dead),
("② 什么都读不到 ⇒ 说「状态未知」(实测 %r)" % unknown, "未知" in unknown),
("③ `_pid_alive` **复用 collabd.py**(⛔ 两份同名实现=活例)", reuses),
]
@case("闸④:排除**已过期的一次性**排期(⛔ 否则它永远卡住闸=自己锁死自己)")
def t_gate4_stale_once():
"""🔴 2026-10-03 11:5x 体检发现(**同型第三次**死锁)。
🔴 **现象**:检查排期 `[检查]-…第3棒` `fire_at` 已过 15 分钟、`last_run_at` 仍 `None`
⇒ `ws_pending_schedules()` 判「本工作区还有待执行排期」⇒ **闸④ 永远拦**
⇒ 机制再也建不出下一棒检查会话(**自己把自己锁死**)。
前两次同型:改名前的旧名排期、「清理残留排期」那条排期把自己困在闸④。
🔴 **根因**=**宿主 `last_run_at` 根本不写**(2026-10-01 实测坐实:明明在跑,值仍是 `None`)
⇒ ⛔ 靠 `last_run_at` 收口**永远收不了**。
✅ **正解**=排除「**已过期的一次性**排期」,宽限 **15 分钟**
(宿主扫描周期 ≤30 s、实测到点延迟 20~35 s ⇒ 15 min 是安全余量)。
⚠️ 四条边界(缺一条就误杀或漏杀):
① 刚过期 ⇒ **仍算待执行**(马上要开 ⇒ 闸④ 该拦)
② 过期 >15 min ⇒ **排除**
③ `next_run_at` 为 NULL ⇒ **仍算待执行**(⛔ 别误杀)
④ `recurring` 过期 ⇒ **仍算待执行**(它会再触发)
"""
import re as _re
import time as _tm
m = imp()
_c = (HERE / "collabd.py").read_text(encoding="utf-8")
# ① 查得到两列
two_cols = ("schedule_type" in _c and "next_run_at" in _c)
# ② 宽限是 15 分钟(⛔ 写成 0 ⇒ 刚过期的也被排除 ⇒ 闸④ 放行"马上要开"的排期 ⇒ 同一件事派两遍)
# ⚠️⚠️ 判据自身的坑(踩了两次):正则的**捕获组只是因子 `15`**,⛔ **不是** `15*60*1000`。
# 我曾拿它当毫秒去比 `900000 <= x` ⇒ 永远 False ⇒ **误判产品坏了**。
# ⇒ 判据直接按**分钟**比:`limit_min` 就是那个因子本身。
mlim = _re.search(r"_STALE_MS\s*=\s*(\d+)\s*\*\s*60\s*\*\s*1000", _c)
limit_min = int(mlim.group(1)) if mlim else -1
# ⚠️ 旧算法残留已删(它引用已被改名的 `limit_ms` ⇒ NameError ⇒ 整条用例假红)。
# ⛔ 教训:改判据里的变量名,⛔ 一定要 grep 一遍**所有**引用点(同 P0-3 的"改一处漏一处")。
# ③ 只排 `once`(⛔ recurring 永远不该被"过期排除")
only_once = 'str(_st or "").lower() == "once"' in _c
# ④ 🔴 2026-10-03 12:0x **语义已改**:`next_run_at` 为 NULL 的 `once` 排期
# **也要排除** —— 实测宿主在排期**到点触发后会把 `next_run_at` 清成 NULL**,
# 而 `status` 仍 `ACTIVE`、`last_run_at` 仍 `None`
# ⇒ 我上一版写"NULL ⇒ 仍算待执行(别误杀)"**是错的** ⇒ 那样它照样永��卡闸④。
# (第五种情形:(a) 已过宽限 / (b) 已被消费 ⇒ 两者都排除。)
null_excluded = "if str(_st or \"\").lower() == \"once\":" in _c and "_gone = (not _na)" in _c
# ⑤ 真库现读:造一条过期 once 也不能把**别人**的排期算进来(⛔ 只验判据形状,不动生产)
live = m.ws_pending_schedules()
return [
("① 闸④ 查询**取到** `schedule_type` 与 `next_run_at` 两列", two_cols),
("② 过期宽限 = 15 分钟(实测 %s 分钟;⛔ 写 0 会放行『马上要开』的排期)"
% limit_min, 10 <= limit_min <= 30),
("③ 只排 `once`(⛔ `recurring` 会再触发,永不按过期排除)", only_once),
("④ `once` 且 `next_run_at` 为 NULL(已被宿主消费)⇒ **也排除**", null_excluded),
("⑤ 判据现读不抛(ws_pending_schedules() = %s)" % (live if live else "[]"), True),
]
@case("看板 tab:**跨工作区**并列查看(⛔ 原本只认一个 INBOX,天生看不见别的目录)")
def t_tab_peer_workspace():
"""🔴 2026-10-03 15:44x(用户报障逐字):
> 「为什么 会话协作看板 tab 选项不能切换看另外两个工作区的目标」
🔴 **根因**:`goal_files()` 扫的是 `INBOX/goal.json` + `INBOX/goals/*.json`
—— **两者都在同一个 INBOX 下**,而 `INBOX` 由部署配置的 `workspace` 决定
⇒ 一个 `board.py --serve` 实例**天生只看见一个工作区**(注释里也明写"同一个工作区")。
✅ 修法:配置新增 `peer_workspaces`(要并列查看的其它工作区)⇒ 读它们的 `goal.json` 进 tab。
🔴 **严格只读**这条边界必须钉住:⛔ 不写对方文件、⛔ 不起对方进程、⛔ 不改对方状态。
"""
import inspect as _ins
import re as _re
m = imp()
spec = _ins.getsource # 占位(下面直接读文本)
src_b = (HERE / "board.py").read_text(encoding="utf-8")
src_h = (HERE.parent / "assets" / "board.html").read_text(encoding="utf-8")
# ① 函数在位
has_fn = "_peer_workspaces" in src_b
# ② 排除本工作区(⛔ 否则活跃目标出现两格)
excludes_self = 's == str(WS or "").replace(' in src_b
# ③ 去重键带工作区前缀(⛔ 否则同名目标互相顶掉)
key_prefixed = 'k = "%s/%s" % (ws_name, k)' in src_b
# ④ _goal_block 透传 peer(⛔ 上游加了下游不带 ⇒ 前端永远看不到)
block_carries = '"peer": peer' in src_b
call_passes = 'it.get("peer") or ""' in src_b
# ⑤ 形参必须存在(⛔ 少了 ⇒ TypeError ⇒ 快照生不出 ⇒ 看板一直 warming —— 本轮实测踩到)
# 🔴 判据=**只扫 `_goal_block()` 的签名**(⛔ 别用裸子串:命中注释/docstring 就变假绿)
_sig_m = _re.search(r"def\s+_goal_block\s*\(([^)]*)\)", src_b, _re.S)
_sig = " ".join((_sig_m.group(1) if _sig_m else "").split())
sig_ok = ("peer" in _sig and "ws_root" in _sig)
sig_txt = "_goal_block(%s)" % (_sig[:60] or "!! 找不到该函数")
# ⑥ 前端显示 peer 标记
# ⚠️ 2026-10-03 20:30x 判据跟着改版走:peer 标记 20:2x 从**第一行徽章**挪到了**第二行**
# (用户要的两行式:第一行=目标名+目标状态,第二行=工作区名+判据状态)⇒ 字面量从 `g.peer?` 变 `g.peer||`。
# 🔴 **剥掉注释再查**(20:2x 踩过:注释里留着旧写法 ⇒ 判据被注释骗、变异不报红)。
_h_nc = re.sub(r"/\*.*?\*/", "", src_h, flags=re.S)
_h_nc = re.sub(r"//[^\n]*", "", _h_nc)
front_shows = ("g.peer||" in _h_nc) or ("g.peer?" in _h_nc)
# ⑦ 只读取向:源码里 ⛔ 不能有写对方文件的调用
no_write = not _re.search(r"_root\s*/\s*\"tmp\".*\.write_text", src_b)
# ⑧~⑩ **现网真读数** —— 🔴 **⛔ 不能沿用 `imp()` 注入的测试环境**:
# `imp()` 把 `COLLABD_CONFIG`/`DSH_COLLAB_WS` 指到 `tmp/selftest/` ⇒ board 只看见测试那 1 格
# ⇒ 断言写什么都只是**假象**(2026-10-03 实测:命令行直接加载=3 格,用例内=1 格)。
# ⇒ 正确姿势:**真磁盘上造两个工作区** + **干净环境**加载 board 现读(P0-13:判据要能证伪)。
import tempfile
def _load_board(cfg_p, ws_env, tag):
"""在**指定配置**下加载 `board.py`(⛔ 用完还原 env,防污染后续用例)。"""
_old = {k: os.environ.get(k) for k in ("COLLABD_CONFIG", "DSH_COLLAB_WS")}
try:
os.environ["COLLABD_CONFIG"] = str(cfg_p)
if ws_env is None:
os.environ.pop("DSH_COLLAB_WS", None)
else:
os.environ["DSH_COLLAB_WS"] = str(ws_env)
_u = __import__("importlib.util", fromlist=["util"])
_sp = _u.spec_from_file_location(tag, HERE / "board.py")
_mm = _u.module_from_spec(_sp)
_sp.loader.exec_module(_mm)
return _mm
finally:
for _k, _v in _old.items():
if _v is None:
os.environ.pop(_k, None)
else:
os.environ[_k] = _v
def _dump(p, obj):
Path(p).write_text(json.dumps(obj, ensure_ascii=False), encoding="utf-8")
def _mk_ws(root, gid, title):
"""真磁盘上造一个**最小工作区**(只含 `tmp/supervise-inbox/goal.json`)。"""
_r = Path(root) / gid
(_r / "tmp" / "supervise-inbox").mkdir(parents=True, exist_ok=True)
_dump(_r / "tmp" / "supervise-inbox" / "goal.json",
{"id": gid, "title": title, "short": gid})
return _r
_td = tempfile.mkdtemp(prefix="selftest-peer-") # ⛔ 不用 TEST_WS(会被别的用例清掉)
try:
_a = _mk_ws(_td, "wsA", "A 区目标")
_b = _mk_ws(_td, "wsB", "B 区目标")
_c_on = _a / "tmp" / "peer.on.json"
_dump(_c_on, {"workspace": str(_a).replace("\\", "/"),
"inbox": "tmp/supervise-inbox",
"peer_workspaces": [str(_b).replace("\\", "/")]})
gs1 = _load_board(_c_on, _a, "bd_peer_on").goal_files()
p1 = [g.get("peer") for g in gs1 if g.get("peer")]
a1 = [g for g in gs1 if g.get("active")]
live_ok = (len(gs1) == 2 and p1 == ["wsB"] and len(a1) == 1)
live_txt = "登记 1 个 peer ⇒ goal_files()=%d 格(peer=%s,active %d 个)" % (
len(gs1), p1, len(a1))
# ⑨ **变异对照**(🔴 判据必须可证伪:去掉 `peer_workspaces` ⇒ 必须回落到 1 格)
_c_off = _a / "tmp" / "peer.off.json"
_dump(_c_off, {"workspace": str(_a).replace("\\", "/"),
"inbox": "tmp/supervise-inbox"})
gs0 = _load_board(_c_off, _a, "bd_peer_off").goal_files()
p0 = [g.get("peer") for g in gs0 if g.get("peer")]
mut_ok = (len(gs0) == 1 and not p0)
mut_txt = "去掉 peer_workspaces ⇒ %d 格(peer %d 个)" % (len(gs0), len(p0))
finally:
shutil.rmtree(_td, ignore_errors=True)
# ⑩ 生产侧:⛔ **不硬编码路径** —— 指不到真工作区 ⇒ SKIP(⛔ 不假装通过)
_pcfg = WS / ".workbuddy" / "collab" / "collabd.config.json"
if not _pcfg.is_file():
prod_ok, prod_txt = True, "SKIP:读不到生产配置(%s)" % _pcfg
else:
try:
_n = len((json.loads(_pcfg.read_text(encoding="utf-8")) or {}).get(
"peer_workspaces") or [])
gp = _load_board(_pcfg, None, "bd_peer_prod").goal_files()
pp = [g.get("peer") for g in gp if g.get("peer")]
ap = [g for g in gp if g.get("active")]
prod_ok = (_n >= 1 and len(pp) == _n and len(ap) == 1)
prod_txt = "生产配置登记 %d 个 peer ⇒ %d 格(peer=%s,active %d)" % (
_n, len(gp), pp, len(ap))
except Exception as e: # pragma: no cover
prod_ok, prod_txt = False, "抛异常:%s" % e
return [
("① 有 `_peer_workspaces()`(配置驱动的并列查看)", has_fn),
("② **排除本工作区**(⛔ 否则活跃目标出现两格)", excludes_self),
("③ 去重键带**工作区前缀**(⛔ 同名目标互相顶掉)", key_prefixed),
("④ `_goal_block()` **透传** peer(⛔ 上游加了下游不带=前端看不到)", block_carries),
("④ 调用点按位置传入 peer", call_passes),
("⑤ `_goal_block` 形参含 peer/ws_root(⛔ 少了 ⇒ TypeError ⇒ 看板 warming):%s" % sig_txt,
sig_ok),
("⑥ 前端 tab 显示 peer 标记(⛔ 否则三格标题一样,读者分不清)", front_shows),
("⑦ **严格只读**(⛔ 源码里不得有写对方文件的调用)", no_write),
("⑧ **现网真读数**(真磁盘两工作区+干净环境):%s" % live_txt, live_ok),
("⑨ **配置缺省 ⇒ 回落 1 格**(向后兼容;⛔ 它只守「配置可关」这一侧,抓不住代码坏 ⇒ 证伪看 ⑧):%s"
% mut_txt, mut_ok),
("⑩ 生产侧按生产配置现读:%s" % prod_txt, prod_ok),
]
@case("跨工作区 peer 格:⛔ **不得吃本工作区数据**(吃了=把本区执行情况挂到别人名下=说假话)")
def t_peer_block_no_self_data():
"""🔴 2026-10-03 16:0x 用户报障,逐字:「**选择另一个工作区目标 tab 下面没有显示对应工作区目标和执行情况**」
🔴🔴 **实测真因(比"没显示"更糟 —— 看板在说假话)**:`build()` 里的 `tasks`/`srows`/`st`
**只取一次、所有格共用** ⇒ 三格的 `labor`/`sessions`/`progress` **md5 完全相同**
⇒ 切到 peer 那格,看到的其实是**本工作区**的执行情况,只是标题挂着对方工作区的名字。
✅ 修法:peer 格只喂 `_peer_tasks()`/`_peer_srows()`/`_peer_state()` —— **只读对方目录下的文件**;
读不到 ⇒ **空 + 界面如实说明**,⛔ 绝不拿本工作区的数据补位。
🔴 **判据可证伪**:把 `_peer_tasks(_wr)` 换回本区的 `tasks` ⇒ ⑤ **必然报红**(两侧 tasks 变相同)。
"""
import tempfile
src = (HERE / "board.py").read_text(encoding="utf-8")
# ① peer 分支喂的是对方数据源
feeds_peer = ("_peer_tasks(_wr)" in src and "_peer_state(_wr)" in src
and "_peer_srows(srows, _wr)" in src)
# ② 补捞(⛔ 没有 ⇒ 全库最近 50 条里捞不到对方会话 ⇒ 显示 0 条=假象)
has_backfill = "_peer_session_rows" in src
# ③ peer 格的作用域要换到对方工作区
scope_swaps = 'sc["workspace"] = ws_root' in src
# ④ 本区切分基准只按本区目标算
own_only = "if not it.get(\"peer\")" in src
# ⑤ 真读数(本区有台账 / 对方无 ⇒ peer 格必须为空,且 ⛔ 不得等于本区的)
_td = tempfile.mkdtemp(prefix="selftest-peerblk-")
live_ok, live_txt = True, "SKIP:未跑"
try:
A = Path(_td) / "wsA"
B = Path(_td) / "wsB"
for _r, _gid in ((A, "wsA"), (B, "wsB")):
(_r / "tmp" / "supervise-inbox").mkdir(parents=True, exist_ok=True)
(_r / "tmp" / "supervise-inbox" / "goal.json").write_text(
json.dumps({"id": _gid, "title": _gid + " 目标", "short": _gid, "topics": [_gid]},
ensure_ascii=False), encoding="utf-8")
(A / "tmp" / "supervise-inbox" / "tasks.json").write_text(
json.dumps({"S1": {"state": "done", "line": "wsA"}}, ensure_ascii=False),
encoding="utf-8")
cfg = A / "tmp" / "cfg.json"
cfg.write_text(json.dumps({"workspace": str(A).replace("\\", "/"),
"inbox": "tmp/supervise-inbox",
"peer_workspaces": [str(B).replace("\\", "/")]},
ensure_ascii=False), encoding="utf-8")
_old = {k: os.environ.get(k) for k in ("COLLABD_CONFIG", "DSH_COLLAB_WS")}
try:
os.environ["COLLABD_CONFIG"] = str(cfg)
os.environ["DSH_COLLAB_WS"] = str(A)
_u = __import__("importlib.util", fromlist=["util"])
_sp = _u.spec_from_file_location("bd_peerblk", HERE / "board.py")
bd = _u.module_from_spec(_sp)
_sp.loader.exec_module(bd)
try:
gs = (bd.build() or {}).get("goals") or []
except Exception as e: # pragma: no cover
gs, live_txt = [], "SKIP:build() 抛异常(%r)" % e
if gs:
own = [g for g in gs if not g.get("peer")]
per = [g for g in gs if g.get("peer")]
ta = (own[0].get("tasks") if own else None) or {}
tb = (per[0].get("tasks") if per else None) or {}
same = (json.dumps(ta, sort_keys=True, ensure_ascii=False)
== json.dumps(tb, sort_keys=True, ensure_ascii=False))
live_ok = bool(ta) and (not tb) and (not same)
live_txt = "本区 tasks=%d 件 / peer tasks=%d 件 / 两者相同?%s" % (
len(ta), len(tb), same)
finally:
for _k, _v in _old.items():
if _v is None:
os.environ.pop(_k, None)
else:
os.environ[_k] = _v
finally:
shutil.rmtree(_td, ignore_errors=True)
return [
("① peer 分支喂的是**对方**数据源 `_peer_tasks/_peer_srows/_peer_state`(⛔ 不是本区 tasks/srows/st)",
feeds_peer),
("② 有 `_peer_session_rows()` **补捞**(⛔ 没有 ⇒ 全库最近 50 条捞不到对方会话 ⇒ 显示 0 条=假象)",
has_backfill),
("③ peer 格把作用域换到**对方工作区**(⛔ 不换 ⇒ `_sessions()` 的 cwd 硬过滤把对方会话全排除)",
scope_swaps),
("④ 本区切分基准**只按本区目标**算(⛔ 混入 peer topics ⇒ 台账件两边都不进 ⇒ 静默丢失)", own_only),
("⑤ **真读数**(本区有台账/对方无 ⇒ peer 格必须为空且 ≠ 本区):%s" % live_txt, live_ok),
]
@case("跨工作区 tab:**有差异的字段必须逐个被视图取到**(⛔ 漏一个 ⇒ 静默沿用本区 ⇒ 几个 tab 看起来一样)")
def t_view_takes_all_scoped_fields():
"""🔴 2026-10-03 20:2x 用户报障(逐字):「**为什么不同目标,协作程序框图里面的内容是一样的**」
🔴🔴 **真因(实测,不是猜)**:服务端**数据是对的**(本区块内 `sessions_checks=4` 条、
两个 peer 块各 0 条),但前端 `scopeView()` 合并视图时**只显式取了 7 个字段**
⇒ 其余字段**原封不动沿用顶层 `d`**(=**本工作区**那份)
⇒ 切 tab 时「协作程序状态/检查状态/检查会话/完成情况」**全都显示同一个工作区的**。
🔴 **这是我今天第三次踩「加字段必须顺着数据流查到消费点」**(前两次:加了没渲染 /
判据写死字面量)⇒ **光靠"记得改"已经不管用了**。
✅ **本条判据是数据驱动的**:跑一次真 `build()` ⇒ 逐字段比较两块 ⇒
**凡"两块有差异"而"视图没取"的字段一律报红** ⇒ 以后新增字段漏了会被自动抓住。
"""
import re as _re2
_root = os.environ.get("DSH_COLLAB_WS") or os.environ.get("DSH_WS_ROOT") or ""
_pcfg = (Path(_root) / ".workbuddy" / "collab" / "collabd.config.json") if _root else None
if not _pcfg or not _pcfg.is_file():
return [("SKIP:没指到真工作区(跑的时候设 `DSH_COLLAB_WS=<工作区>` 才会真跑)", True)]
_old = {k: os.environ.get(k) for k in ("COLLABD_CONFIG", "DSH_COLLAB_WS")}
try:
os.environ["COLLABD_CONFIG"] = str(_pcfg)
os.environ.pop("DSH_COLLAB_WS", None)
_u = __import__("importlib.util", fromlist=["util"])
_sp = _u.spec_from_file_location("bd_viewchk", HERE / "board.py")
bd = _u.module_from_spec(_sp)
_sp.loader.exec_module(bd)
gs = (bd.build() or {}).get("goals") or []
if len(gs) < 2:
return [("SKIP:本工作区只登记了 1 个目标(⛔ 比不出差异)", True)]
a, b2 = gs[0], gs[1]
_IDENT = {"key", "peer", "ws_root", "active", "life", "life_at", "life_by", "peer_scope"}
_diff = [k for k in sorted(set(a.keys()) | set(b2.keys())) if k not in _IDENT
and json.dumps(a.get(k), sort_keys=True, ensure_ascii=False)
!= json.dumps(b2.get(k), sort_keys=True, ensure_ascii=False)]
# 🔴 提取赋值清单前**先剥掉注释** —— ⛔ 否则"把某行注释掉"这种变异**照样匹配得到**
# (本轮实测踩到:注释掉 `v.sessions_checks=` 之后判据仍报绿 ⇒ 判据被注释骗了)。
_html = (HERE.parent / "assets" / "board.html").read_text(encoding="utf-8")
_html = _re2.sub(r"/\*.*?\*/", "", _html, flags=_re2.S)
_html = _re2.sub(r"//[^\n]*", "", _html)
_fn = _re2.search(r"function scopeView\(d,g\)\{.*?\n\}", _html, _re2.S)
_taken = set(_re2.findall(r"v\.(\w+)\s*=", _fn.group(0))) if _fn else set()
_missing = [k for k in _diff if k not in _taken]
return [("两块有差异的字段 %d 个,**视图必须逐个取到**(⛔ 漏了会静默沿用本区):%s"
% (len(_diff), ("全部已取 ✔" if not _missing else "⛔ 漏了 %s" % _missing)),
not _missing)]
except Exception as _e: # pragma: no cover
return [("用例异常:%r" % _e, False)]
finally:
for _k, _v in _old.items():
if _v is None:
os.environ.pop(_k, None)
else:
os.environ[_k] = _v
@case("每个 hook 都能真正跑起来(⛔ 不许一导入就崩)")
def t_hooks_actually_runnable():
"""🔴🔴 2026-10-03 **实测抓到的真缺陷**(用户报障「协作程序相关的连线 有用的保留 没用的
可以去掉」,我逐条去核实"这三条线还活着吗"时撞出来的):
⚠️ **事故链**:用户 10-02 报「后台一闪一闪」⇒ 为根治闪屏,给 `wb-result-hook.py` 加了
`_win_pythonw()`,写成 `PYW = _win_pythonw()`,而**函数体内部引用 `PYW`**
⇒ 调用时 `PYW` 尚未赋值 ⇒ **模块导入即 `NameError`**
⇒ 该 hook 挂的**三个事件(PreToolUse/UserPromptSubmit/SessionEnd)全部静默废掉**
⇒ 图上「钩子 → 协作程序 --once/--tick」两条线**其实早就断了,只是没人发现**
(`hook.log` 停在 10-02 21:24、三个 stamp 也停在同一分钟 —— 三处独立证据互相印证)。
🔴 **为什么这条判据必须存在**:⛔ 语法检查(`ast.parse`)**照样通过** ——
`NameError` 是**运行期**错误,语法完全合法。⇒ 「能解析」≠「能跑」,
光靠语法自检永远抓不到这类缺陷。而它造成的后果是**整条链路静默失效**,
看板上的连线会一直画着、看着一切正常。
✅ **本条判据=真的把每个 hook 起一次**(喂最小合法 stdin),断言 `rc=0` 且无 traceback。
⚠️ 起法要点:⛔ 必须用 `sys.executable`(当前解释器)而**不是** `python`(PATH 可能指向别处);
喂完**必须等它自己跑完**(hook 会 spawn 后台子进程,不能立刻判)。
"""
import subprocess as _sp2
import tempfile as _tf2
# 🔴 **只测真 hook**:目录里的 `*.bak-*` 是历史备份,⛔ 不是在跑的钩子
# (它们恰好也都能起来 ⇒ 混进来只是给输出添噪音,且会让人误以为有 12 个钩子)。
hooks = sorted(p for p in (HERE / "hooks").glob("*.py") if ".bak-" not in p.name)
out = []
for hk in hooks:
payload = json.dumps({"session_id": "selftest-probe",
"cwd": str(TEST_WS),
"hook_event_name": "UserPromptSubmit",
"prompt": "selftest probe"})
try:
r = _sp2.run([sys.executable, str(hk)], input=payload.encode("utf-8"),
capture_output=True, timeout=60, cwd=str(TEST_WS))
except Exception as exc:
out.append(("%s 起不来:%r" % (hk.name, exc), False))
continue
err = (r.stderr or b"").decode("utf-8", "replace")
# 🔴 两个判据缺一不可:退出码 0 **且** stderr 里没有 traceback
# (有的脚本出错仍返回 0,只看 rc 会漏)。
out.append(("%s 起得来(rc=%d%s)" % (
hk.name, r.returncode,
",stderr 有 traceback:%s" % err.strip().splitlines()[-1] if "Traceback" in err else ""),
r.returncode == 0 and "Traceback" not in err))
return out or [("⛔ hooks 目录里一个 hook 都没有 ⇒ 判据本身失效", False)]
@case("架构图:每条线的**两端都落在真实框上**(⛔ 不许线头悬空)")
def t_board_edges_land_on_boxes():
"""🔴🔴 2026-10-03 用户红框圈出一条线问「**这跟红框标注的线到底有没有用
有用就连对目标,没用就删除了**」—— 取证发现它**既没用不了、也没连对**:
🔴 **病根=把「与框宽有关的 x」写死**:④ 那条线写死 `x=340`,而它成立的前提是
当时 `PX=280`、宽 580。后来框**收窄到 470** 且改成**跟随画布中心** ⇒ 框变 405..875
⇒ 340 掉到框外**左侧 65px** ⇒ 箭头悬空。
🔴 **同一个病 09-30 修过一次**(当时原话「WorkBuddy 左边那根线是要连到哪里?」)
—— 当时**只改了线的位置、没把它与框宽绑死** ⇒ 框一变就复发。
⇒ 教训:**⛔ 凡是"与可变量(框宽/框位)有关"的坐标,一律现算,别写死。**
✅ **本条判据=把源码里那几个常量取出来重算一遍**,逐条断言:
① ④ 出口 `RX` **必须落在协作程序框内**(`PX < RX < PX+PW`)—— 这条就是"连对目标";
② `RX` **必须小于 `hkX`** —— 否则会**穿过** Hook 框("连过去却穿框"同样是错);
③ `--once`/`--tick` 两个出口**必须落在 Hook 框顶边之内**;
④ 四个出口**两两不重叠**(重叠=两条线并成一条、看不出谁是谁)。
⚠️ 取常量方式:**从源码正则抽**(⛔ 不另写一份副本)—— 副本会与源码漂移,
那样这条判据就变成了"检查我自己写的数"(同族陷阱:判据必须扎在真源上)。
⚠️ 已知同源缺陷(本条**抓不到**,⛔ 别以为它是万能的):源码里若有人把 `hkX=430` 又写回
`hkX=400`,而 `RX` 仍由 `PX+24` 现算 ⇒ 两者同时满足本条 ⇒ 但**观感上 ④ 仍穿框**。
⇒ 那属于"把条件写复杂了",根治办法是**保持 `RX < hkX` 恒成立**(别靠加更多分支)。
"""
import re as _re3
src = (HERE.parent / "assets" / "board.html").read_text(encoding="utf-8")
# 🔴🔴 2026-10-03 **又踩一次「朴素扫描被注释骗」**(同族坑,见 P0-36):
# ④ 那段注释里**逐字写着** `RXV=PX-40`/`RXH=PX+28` ⇒ 正则先命中注释
# ⇒ 抽出来是一整段散文 ⇒ `eval` 失败 ⇒ 判据整条失效。
# ✅ **正解=先剥注释再抽**(顺序不能反):块注释 → 行注释 → 再找赋值。
_code = _re3.sub(r"/\*.*?\*/", "", src, flags=_re3.S)
_code = _re3.sub(r"(?m)^\s*//.*$", "", _code) # 只吃**整行**注释(行尾注释常跟在代码后)
_code = _re3.sub(r"//[^\n]*", "", _code)
def _n(pat, dflt=None, hay=None):
m = _re3.search(pat, hay if hay is not None else _code)
return float(m.group(1)) if m else dflt
CX = _n(r"var CX=(\d+)")
PW = _n(r"var PW=(\d+)")
# 🔴 `PX` 源码里是**表达式** `CX-PW/2`(与 PW 同一行声明)⇒ ⛔ 不能按纯数字抽。
PX = (CX - PW / 2) if (CX is not None and PW is not None) else None
hkX = _n(r"var hkX=(\d+)")
hkW = _n(r", hkW=(\d+)")
if None in (CX, PW, PX, hkX, hkW):
return [("⛔ 抽不出几何常量(CX/PW/hkX/hkW 实得 %s)⇒ 判据本身失效,⛔ 别当通过用"
% (CX, PW, hkX, hkW), False)]
# 🔴🔴 2026-10-03 **自查抓到的一个真漏洞(本条第一版栽在这)**:
# 第一版 RX 在判据里**自己重算** `PX+24` ⇒ 它检查的是「我算的数」,
# **不是「源码里那个数」** ⇒ 把源码改成 `var RX=340;`(原事故)它照样报绿。
# 🔴 这就是判据的经典自欺:**判据与被检对象各算一遍,必然有一份是假的**。
# ✅ **正解=判据只认源码里的真实写法**:抽 `var ...=` 后面到底是**表达式**还是**字面量**,
# 若是字面量**直接报红**(那正是"写死与框宽有关的坐标"这个病根本身)。
# 🔴🔴 2026-10-03 **三改=④ 变折线后本条同步**(用户原话:「**稍微向左边折一下再连过去**」
# +「**还把 hook 图框挤到旁边 多难看**」):
# ④ 不再是"一条竖直线",而是 `RXV`(框外左侧竖直段)→ `RYY`(折点)→ `RXH`(**框内**终点)。
# ⇒ **竖直段看 `RXV`、终点看 `RXH`**,两个都要查;
# ⚠️ 且新增一条**反向判据**:`hkX` 必须是**它自己的版面值**(⛔ 不许被线牵着走)——
# 上一版我为腾位置把它 400→430,用户明确指正 ⇒ 这条写进判据防复发。
res = []
def _expr(name, env=None):
# 🔴 `var RXV=…, RXH=…, RYY=…` 三个变量**在同一行逗号声明** ⇒
# ① 只有**第一个**变量带 `var`,后两个没有 ⇒ 正则要写成「`var NAME=`」**或**「裸 `NAME=`」
# (⚠️ 只写 `var %s=` 会漏掉 `RXH`/`RYY`,实测踩到);
# ② 取值必须止于**下一个逗号或分号**,⛔ 否则 `var RXV=PX-40, RXH=…` 会把后面一起吞进来。
m = _re3.search(r"\b(?:var\s+)?%s=([^,;]+)" % _re3.escape(name), _code)
if not m:
res.append(("⛔ 源码里抽不到 `var %s=` ⇒ 判据失效,⛔ 别当通过用" % name, False))
return None, None
e = m.group(1).strip()
lit = _re3.fullmatch(r"(\d+(?:\.\d+)?)", e)
if lit:
res.append(("⛔ 源码 `var %s=%s` 是**写死字面量**(病根:框一改它就落到框外)"
" ⇒ 必须由 PX 现算" % (name, e), False))
try:
# 🔴 源码里 `R4.y` / `R4.h` 是**属性访问**写法 ⇒ eval 环境不能只给 dict,
# 否则 `R4.y+(R4.h/2)` 会 `AttributeError`(实测踩到 ⇒ 整条判据失效)。
# ✅ 两种都支持:dict 下标(`R4["y"]`)与属性(`R4.y`)——用一个既支持点又支持下标的类型。
class _Box(dict):
__getattr__ = dict.__getitem__
v = float(eval(e, {"__builtins__": {}}, # noqa: S307
{"PX": PX, "PW": PW, "CX": CX,
"R4": _Box({"y": R4Y, "h": R4H})}))
except Exception:
res.append(("⛔ `var %s=%s` 求值不了 ⇒ 判据失效,⛔ 别当通过用" % (name, e), False))
return None, e
return v, e
R4Y, R4H = _n(r"R4=\{y:(\d+)"), _n(r"h:(\d+)\}, R5")
if R4Y is None or R4H is None:
return res + [("⛔ 抽不到协作程序行几何 `R4={y:...,h:...}` ⇒ 判据失效", False)]
RXV, _ = _expr("RXV")
RXH, _ = _expr("RXH")
RYY, _ = _expr("RYY")
once_x = hkX + 40
tick_x = PX + PW / 2 + 120
if RXV is None or RXH is None or RYY is None:
return res
res += [
# 🔴🔴 2026-10-03 **判据跟着改口径**(用户原话:「**线连到左边不行 非的连到下边
# 是咋想的**」):④ 终点现在**落在框的左边框上**(`RXH==PX`),
# ⛔ 不再是"框内某点"(旧判据 `PX<RXH<PX+PW` 已不适用 ⇒ 会被误判为红)。
# ✅ 正确口径=**终点必须落在"某个边框上"**(左边/下边/上边/右边皆可),
# 唯独⛔ 不能"绕到左边却从下边进"(那是**无意义的折**)。
("④ 折线**终点 RXH=%.0f 落在框边框上**(左边框 x=%.0f)——⛔ 不是绕左却从下边进"
% (RXH, PX), abs(RXH - PX) < 0.5),
# ✅ 折点 y 必须是**框的垂直中点** ⇒ 两段折线(竖 + 横)一眼看懂。
("④ 折点 RYY=%.0f 落在框的垂直中点 %.0f(⛔ 不贴边、⛔ 不写死)"
% (RYY, R4Y + R4H / 2), abs(RYY - (R4Y + R4H / 2)) < 0.5),
("④ **竖直段 RXV=%.0f 在 Hook 框左边 %.0f 之外**(⛔ 不穿 Hook 框)" % (RXV, hkX),
RXV < hkX),
# 🔴 新增:**⛔ 不许有多余折段**(每段都得有理由;绕左边再从下边进=白绕)。
("④ 折线**只有两段**(⛔ 绕左边却从下边钻进来=那几段横移毫无意义)",
src.count("L'+RXH+' '+(R4.y+R4.h)") == 0),
# 🔴🔴 新增**反向判据**(防"为凑一条线去挪别人的框"):hkX 必须仍是 400。
# 上一版我把它 400→430 来腾位置,用户明确指正「把 hook 图框挤到旁边 多难看」。
("Hook 框左边界 hkX=%.0f **未被连线牵动**(⛔ 框的位置是它自己的版面决策,"
"不许为凑一条线挪框)" % hkX, hkX == 400),
("--once 出口 %.0f 落在 Hook 框顶边 %.0f..%.0f 内" % (once_x, hkX, hkX + hkW),
hkX <= once_x <= hkX + hkW),
("--tick 出口 %.0f 落在 Hook 框顶边 %.0f..%.0f 内" % (tick_x, hkX, hkX + hkW),
hkX <= tick_x <= hkX + hkW),
]
# ④ 两两不重叠:⛔ 重叠=两条线并一条,读者分不出谁是谁(P0-25 同族)
outs = [("④", RXH), ("--once", once_x), ("--tick", tick_x)]
for i in range(len(outs)):
for j in range(i + 1, len(outs)):
gap = abs(outs[i][1] - outs[j][1])
res.append(("%s 与 %s 间隔 %.0fpx(⛔ 不许并成一条)" % (outs[i][0], outs[j][0], gap),
gap >= 30))
return res
@case("目标完成 ⇒ 真的收工(⛔ 不许只改状态不关进程);阻碍 ⇒ 不关")
def t_goal_done_stops_supervise():
"""🔴🔴 2026-10-03 **接上用户口径第①条**(逐字):
「**目标完成后需要把目标状态改为 完成,关闭后台任务 和 协作程序
(有新目标或之前的目标有调整 在打开)**」;
第②条:「**目标 遇到阻碍** …(**不用关闭后台任务和协作程序**)」。
🔴 **改前取证**:`goal_life_set()` 原来**只写 `goal.json` 四个字段就 `return`**,
全文无 `stop`/`kill` 调用 ⇒ "完成即收工"**在代码里根本没实现**。
✅ 本条判据查三件事(**源码级**,⛔ 不靠"跑一遍看看"):
① `supervise_stop()` **存在**(有停的手段);
② `goal_life_set()` 在 `life == GOAL_LIFE_DONE` 分支**真的调它**;
③ `ensure_supervise()` **会清 `guard.stop`** —— 🔴 这条最容易漏:
停止标志是常驻的退出条件,不清就**永远起不来**("完成"变成不可逆,与口径相反)。
⚠️ 判据一律**剥注释后**匹配(注释里写着"关闭后台任务"等字样,别被骗)。
"""
import re as _re4
src = (HERE / "collabd.py").read_text(encoding="utf-8")
code = _re4.sub(r"/\*.*?\*/", "", src, flags=_re4.S)
code = _re4.sub(r"(?m)^\s*#.*$", "", code)
code = _re4.sub(r"#[^\n]*", "", code)
def _fn(name):
m = _re4.search(r"\ndef %s\(.*?(?=\ndef )" % _re4.escape(name), code, _re4.S)
return m.group(0) if m else ""
fn_stop, fn_life, fn_ens = _fn("supervise_stop"), _fn("goal_life_set"), _fn("ensure_supervise")
res = [
("`supervise_stop()` 存在(没有它就无从关闭协作程序)", bool(fn_stop)),
("`supervise_stop()` 真的能停:写了停止标志 `guard.stop`", "guard.stop" in fn_stop),
("`supervise_stop()` 有强停兜底(⛔ 只写标志可能停不掉)",
"taskkill" in fn_stop or "SIGKILL" in fn_stop),
("`goal_life_set()` 在**已完成**分支调用 `supervise_stop()`(完成 ⇒ 收工)",
"GOAL_LIFE_DONE" in fn_life and "supervise_stop" in fn_life),
("`goal_life_set()` 对**阻碍/已暂停**明确不调停(用户口径②:阻碍不用关)",
("GOAL_LIFE_BLOCK" in fn_life or "保持运行" in fn_life)
and _re4.search(r"GOAL_LIFE_DONE.*?supervise_stop", fn_life, _re4.S) is not None),
("`ensure_supervise()` **会清 `guard.stop`**(⛔ 不清 ⇒ 完成后再也起不来)",
"guard.stop" in fn_ens and "unlink" in fn_ens),
]
return res
@case("跨区自愈:**只补进行中的区**(⛔ 已完成/阻碍/已收工的一律不拉)")
def t_peer_supervise_sweep():
"""🔴 2026-10-03 补的跨区自愈(实测坐实缺口:本区 `ensure_supervise()` 只看本区心跳,
本区日志搜别区名**零命中** ⇒ 别区常驻被外部收走后**没人知道、没人补**)。
✅ 本条守三条**越界红线**(少一条就可能去动别的区)——源码级断言:
① 目标 `lifecycle` **必须是「进行中」**才拉(用户口径:完成=已收工,阻碍=不用开);
② **⛔ 只读别区心跳**(不得写别区的 `goal.json`/队列 —— 那是别的区主会话的领地);
③ 有 `guard.stop` 的区**⛔ 不拉**(刚被收工的区,拉起来=对抗用户的停)。
⚠️ 另守两条工程约束:**接在主循环里**(⛔ 没接=写了函数也不生效)
+ **有节流**(⛔ 每轮都去 stat 别区是白耗 IO)。
"""
import re as _re5
src = (HERE / "collabd.py").read_text(encoding="utf-8")
code = _re5.sub(r"/\*.*?\*/", "", src, flags=_re5.S)
code = _re5.sub(r"(?m)^\s*#.*$", "", code)
code = _re5.sub(r"#[^\n]*", "", code)
m = _re5.search(r"\ndef peer_supervise_sweep\(.*?(?=\ndef )", code, _re5.S)
fn = m.group(0) if m else ""
if not fn:
return [("⛔ 源码里没有 `peer_supervise_sweep()` ⇒ 跨区自愈不存在", False)]
return [
("跨区自愈函数存在(⛔ 缺它=别区停了没人管)", bool(fn)),
("闸①:目标 lifecycle **必须是「进行中」**才拉(已完成/阻碍 ⛔ 不拉)",
"GOAL_LIFE_RUN" in fn and "lifecycle" in fn),
("闸②:⛔ **只读**别区心跳(只 read_text 心跳,⛔ 不写别区 goal.json)",
"supervise-heartbeat.json" in fn and fn.count("write_text") == 0),
("闸③:有 `guard.stop` 的区 ⛔ 不拉(别对抗用户的收工)", "guard.stop" in fn),
("起别区时用**它自己的副本+它自己的 COLLABD_CONFIG**(⛔ 指向本区就串了)",
"COLLABD_CONFIG" in fn and ".workbuddy" in fn),
("有节流 `PEER_SWEEP_EVERY`(⛔ 每轮都碰别区=白耗 IO)", "PEER_SWEEP_EVERY" in src),
("**接在主循环里**(⛔ 写了函数不接=永不生效)", "peer_supervise_sweep()" in code),
("Popen 带 `creationflags`(⛔ 漏了会闪黑窗,仓内另有判据守这条)",
"creationflags" in fn),
]
@case("初始化脚本:**指向本区副本**(⛔ 不许再让用户跑技能目录那份)")
def t_init_points_to_own_copy():
"""🔴 2026-10-03 修的误导:`init_workspace.py` 原先 `--ensure-goal-dir` 与末尾
「后续协作」提示**全部指向技能目录的 `collabd.py`/`board.py`**。
🔴 **为什么有害**(不是洁癖):定案口径是**各工作区程序各自独立**(技能目录=源,
`.workbuddy/collab/`=各区副本,`deploy_code.py` 分发)⇒ 让用户跑目录那份
**会拉起第二个常驻、写到技能目录的 `tmp/`**(⛔ 那不是任何工作区的 inbox)
⇒ 状态写到别处:看板读不到、故障排查被带偏。
✅ 判据:`--ensure-goal-dir` 的实参**优先取本区副本**,末尾提示同理;
⚠️ 副本不存在时**允许回落**,但⛔ 必须**打印说明**(静默回落=用户以为在跑副本)。
"""
src = (HERE / "init_workspace.py").read_text(encoding="utf-8")
code = re.sub(r"\"\"\".*?\"\"\"", "", src, flags=re.S)
code = re.sub(r"(?m)^\s*#.*$", "", code)
res = [
("有「取本区副本」这一步(`_own = ws/.workbuddy/collab/collabd.py`)",
".workbuddy" in code and "collabd.py" in code),
("`--ensure-goal-dir` 用的是**本区副本优先**的路径(⛔ 不是硬编码 SK)",
code.count("SK / \"collabd.py\"") <= 2 and "_cd" in code),
("副本缺失时**回落**用技能目录那份(⛔ 硬失败会挡住首次初始化)",
"_fallback" in code),
("回落时**打印说明**(⛔ 静默回落=用户以为在跑副本)",
code.count("print") >= 2 and "回落" in code),
("末尾「后续协作」提示也指向副本/带说明(⛔ 别再让人跑目录那份)",
"_cd_show" in code and "_bd_show" in code),
]
return res
@case("自指型判据:**必须被排除**(⛔ 否则 `goal_state()` 只能自己满足自己,永远锁死)")
def t_acc_selfref_excluded():
"""🔴🔴 2026-10-04 用户点破:「**A6 判据自指 ⇒ `goal_state()` 逻辑死锁**」。
🔴 **死锁形状**(不是时序问题、⛔ 也等主会话解决不了):
`goal_state()` 的**路3 要读这条判据**,而这条判据判的内容**就是**「三路并集的结论」
⇒ 它**只能由"自己写自己过"满足** ⇒ 固定点之外永不到达。
🔴 **实测坐实**:把 A6 临时写成「过|…」⇒ `goal_state()` 立刻 pass;还原 ⇒ 立刻回 open。
✅ 修法=**路3 排除自指型判据**,判据查四件事(源码级 + 行为级各一半):
① `_acc_is_selfref()` 存在,且**登记制**(`_ACC_SELFREF_KEYS`)与**文本兜底**(`_ACC_SELFREF_TOKENS`)**两条都在**;
② `goal_state()` 路3 与 `_acc_short()` **两处都剔**(⛔ 只剔一处 ⇒ 唤醒正文仍会念「还差 A6」,
造出「`goal_state()` 已判 pass、这里还在报」的新自相矛盾);
③ **fail-closed**:剔完一条不剩 ⇒ 判「未声明」,⛔ 不算过;
④ **兜底必须窄**(`_ACC_SELFREF_WIN` 有上限)—— 扫整段会**误伤 A2**(它正文里顺口提了一句
`goal_state()` 只读文本)⇒ 悄悄放掉一条真门禁,**比原 bug 更坏**(原 bug 是吵,这个是静默失守)。
⚠️ **源码级那两条必须按 AST 精确定位,⛔ 不可用「函数体里含某串」**(2026-10-04 变异验证实测栽过):
- 🔴 `_acc_is_selfref` 在 `goal_state()` 里出现**两次**(`_real` 的推导式 + `_dropped` 的推导式)
⇒ 判据写成 `"_acc_is_selfref" in fn_gs` 时,**只把 `_real` 那处改掉照样 PASS**(漏网)。
- 🔴 `undeclared` 在 `goal_state()` 尾部 `return "pass" if eff else "undeclared"` 里也有
⇒ 判据写成 `re.search(r"if not _real.*?undeclared", ...)` 时,**把中间那个 `return` 拆掉照样 PASS**(漏网)。
⇒ 一律改成**在 AST 上找指定 Assign / 指定 If 的 body**,别用"整段文本里有没有"。
"""
import ast as _ast5
import re as _re5
src = (HERE / "collabd.py").read_text(encoding="utf-8")
code = _re5.sub(r"\"\"\".*?\"\"\"", "", src, flags=_re5.S)
code = _re5.sub(r"(?m)^\s*#.*$", "", code)
def _fn_src(name):
m = _re5.search(r"\ndef %s\(.*?(?=\ndef )" % _re5.escape(name), code, _re5.S)
return m.group(0) if m else ""
def _fn_node(name):
"""AST 上**唯一定位**某个顶层函数(同名多个 ⇒ 返回 None,⛔ 不许取第一个)。"""
hits = [n for n in _ast5.walk(_ast5.parse(src))
if isinstance(n, _ast5.FunctionDef) and n.name == name
and isinstance(getattr(n, "col_offset", 0), int)]
tops = [n for n in hits if n.col_offset == 0]
return tops[0] if len(tops) == 1 else None
def _unparse(nodes):
return "".join(_ast5.unparse(n) for n in nodes)
fn_sr, fn_gs, fn_as = _fn_src("_acc_is_selfref"), _fn_src("goal_state"), _fn_src("_acc_short")
nd_gs, nd_as = _fn_node("goal_state"), _fn_node("_acc_short")
# --- AST:`_real = [...]` 这个**赋值**的推导式里到底有没有剔 ---
# ⚠️⚠️ 判据写法上的两个坑(2026-10-04 变异验证实测各栽一次):
# ① `_real` 是 **Assign 左侧的名字**,⛔ **推导式的 target 里没有它**
# (`goal_state` 的是 `(_k, _v)`、`_acc_short` 的是 `k`)⇒ 只能在 Assign 上找。
# ② 过滤条件在**生成器的 ifs 子句**里,⛔ 不在 `elt`;且 `elt` 是**单个节点**
# (`goal_state` 那个还是 Tuple)⇒ `list(elt)` 会 TypeError。
def _real_filter_excludes_selfref(node):
if node is None:
return False
for n in _ast5.walk(node):
if not (isinstance(n, _ast5.Assign) and len(n.targets) == 1
and isinstance(n.targets[0], _ast5.Name)
and n.targets[0].id == "_real"):
continue
if not isinstance(n.value, _ast5.ListComp):
continue
scan = list(_ast5.walk(n.value.elt)) + [x for g in n.value.generators
for i in g.ifs for x in _ast5.walk(i)]
if any(isinstance(x, _ast5.Call) and getattr(x.func, "id", "") == "_acc_is_selfref"
for x in scan):
return True
return False
# --- AST:`if not _real:` 这个 If 的 body 里到底有没有 `return "undeclared"` ---
def _not_real_returns_undeclared(node):
if node is None:
return False
for n in _ast5.walk(node):
if not isinstance(n, _ast5.If):
continue
t = n.test
is_not_real = (isinstance(t, _ast5.UnaryOp) and isinstance(t.op, _ast5.Not)
and isinstance(t.operand, _ast5.Name) and t.operand.id == "_real")
if not is_not_real:
continue
for stmt in n.body:
if isinstance(stmt, _ast5.Return) and isinstance(stmt.value, _ast5.Constant) \
and stmt.value.value == "undeclared":
return True
return False
res = [
("`_acc_is_selfref()` 存在(没有它就无从识别自指)", bool(fn_sr)),
("**登记制**在(`_ACC_SELFREF_KEYS`):⛔ 不依赖判据文本,是唯一权威路径",
"_ACC_SELFREF_KEYS" in code),
("**文本兜底**在(`_ACC_SELFREF_TOKENS` + 有窗口上限):补登记漏网的",
"_ACC_SELFREF_TOKENS" in code and "_ACC_SELFREF_WIN" in code),
("窗口**有上限且不为 0**(⛔ 扫整段会误伤 A2 那类真判据 ⇒ 静默失守)",
bool(_re5.search(r"_ACC_SELFREF_WIN\s*=\s*([1-9]\d*)", code))),
("AST:`goal_state()` 里 `_real` 的推导式**真的剔**(⛔ 不看整段文本,见 docstring 变异坑)",
_real_filter_excludes_selfref(nd_gs)),
("AST:`_acc_short()` 里也剔(⛔ 只剔一处 ⇒ 唤醒正文仍念「还差 A6」,造出新自相矛盾)",
_real_filter_excludes_selfref(nd_as)),
("AST:`if not _real:` 的 body 里**真的 `return \"undeclared\"``(fail-closed)",
_not_real_returns_undeclared(nd_gs)),
("⛔ **没放松实质门禁**:`for _k, _v in _real:` 仍逐条判 pass(其余判据照判)",
"for _k, _v in _real:" in fn_gs or "for _k, _v in _real" in fn_gs),
]
# ---- 行为级:真起子进程跑真函数,用真数据量(⛔ 源码含关键字不算数) ----
try:
import subprocess as _sp
probe = r'''
import importlib.util, json, os, sys, tempfile
from pathlib import Path
spec = importlib.util.spec_from_file_location("cd_probe", sys.argv[1])
m = importlib.util.module_from_spec(spec)
sys.modules["cd_probe"] = m
spec.loader.exec_module(m)
sr = m._acc_is_selfref
res = {}
# ① 真自指(测试3 的 A6 原句形状):必须判 True
A6 = "\U0001f534 不过|goal_state() 返回 pass 且 goals_open() 为空(三路全过)"
res["A6 真自指判 True"] = sr("a6 目标状态", A6) is True
# ② 真门禁 A2:正文**深处**顺口提一句 goal_state() ⇒ 必须判 False(⛔ 误伤=静默放掉真门禁)
A2 = ("\U0001f534 不过|" + "A2 快照陷阱说明。" * 40 + " 另注:goal_state() 只读文本、不重验事实。")
res["A2 长正文顺口提及判 False"] = sr("a2 快照", A2) is False
# ③ 登记制:文本完全看不出自指,只要登记过就排除
res["登记制压过文本"] = sr("A6 目标状态", "过|已确认") is True
# ④ 普通真判据:一律不排除
res["普通判据判 False"] = sr("a1 产物", "\U0001f534 不过|产出三份文档") is False
# ⑤🔴 端到端:真造 goal.json 跑**真 goal_state()**(⛔ 这才是死锁的硬证据)
# 🔴🔴 **隔离必须用 `COLLABD_CONFIG`,⛔ 不可用 `DSH_COLLAB_WS`/`COLLABD_WORKSPACE`**:
# 真源 `load_cfg()` 的候选顺序是 ① `COLLABD_CONFIG` → ② `<ws>/.workbuddy/collab/collabd.config.json`
# ⇒ **只设 env 不给 config,会回落读到真工作区的配置**(本判据曾因此把真 `goal.json` 覆盖掉,
# 2026-10-04 已从 `bak-goalctl-20261002/` 恢复)。⇒ 这里**显式写一份临时 config**。
# ⚠️ `TG`(任务图)也由这份 config 的 `taskgraph` 决定 ⇒ 一并指到夹具,⛔ 不读真任务图。
tmp = tempfile.mkdtemp(prefix="gs_")
_cfg = Path(tmp) / "collabd.config.json"
_cfg.write_text(json.dumps({
"workspace": tmp.replace("\\", "/"),
"inbox": "inbox",
"taskgraph": "inbox/taskgraph.json",
"live": "inbox/realtime.md",
}, ensure_ascii=False), encoding="utf-8")
os.environ["COLLABD_CONFIG"] = str(_cfg)
spec2 = importlib.util.spec_from_file_location("cd_probe2", sys.argv[1])
m2 = importlib.util.module_from_spec(spec2)
sys.modules["cd_probe2"] = m2
spec2.loader.exec_module(m2)
# 🔴 自检:**INBOX 必须真落在临时目录**,否则说明隔离没生效 ⇒ 宁可报红也别再写真文件
if str(m2.INBOX).replace("\\", "/").find(tmp.replace("\\", "/")) < 0:
print(json.dumps({"__隔离失效: INBOX=%s 不在临时目录" % m2.INBOX: False}, ensure_ascii=False))
raise SystemExit(0)
inbox = Path(m2.INBOX); inbox.mkdir(parents=True, exist_ok=True)
(inbox / "tasks.json").write_text(json.dumps({"t": {"state": "done"}}), encoding="utf-8")
(inbox / "taskgraph.json").write_text(json.dumps({"nodes": [{"id": "N1", "status": "done"}]}),
encoding="utf-8")
# ⚠️ 判据文本必须用**真源真会写出来的形状**(`过|…`,⛔ 不带 `🔴` 前缀):
# `_acc_is_pass` 取竖线左边那段判,`🔴 过` 的 head 仍带 `🔴` ⇒ 会被判「未过」(假红)。
def _write(acc):
(inbox / "goal.json").write_text(json.dumps(
{"nodes": [{"id": "N1", "status": "done"}], "acceptance_state": acc}),
encoding="utf-8")
# 场景 X:**只有自指判据、且它没过** ⇒ 正确行为=`undeclared`(⛔ 不是 pass)
_write({"a6 目标状态": "\U0001f534 不过|goal_state() 返回 pass 且 goals_open() 为空(三路全过)"})
res["隔离生效: INBOX 落在临时目录"] = True
res["端到端X: 全自指且未过 ⇒ undeclared(不是pass)"] = (m2.goal_state() == "undeclared")
# 场景 Y:自指判据已写「过」 + **一条真判据也过了** ⇒ 正确行为=`pass`
#(⛔ 这是修复前会卡死的那个形状:路3 要读 A6,A6 又判"三路全过")
_write({"a6 目标状态": "过|goal_state() 返回 pass", "a1 产物": "过|三份文档已交"})
res["端到端Y: 自指+真判据全过 ⇒ pass(不再死锁)"] = (m2.goal_state() == "pass")
# 场景 Z:自指 + **一条真判据没过** ⇒ 必须 `open`(⛔ 不许因为剔了自指就把真门禁也放过)
_write({"a6 目标状态": "\U0001f534 不过|goal_state() 返回 pass", "a1 产物": "不过|只交了两份"})
res["端到端Z: 真判据没过 ⇒ open(没放松门禁)"] = (m2.goal_state() == "open")
# 场景 W:自指**已写「过」** + 真判据**没过** ⇒ 仍必须 `open`
#(⛔ 这条专抓"因为剔了自指就整路放行"的写法:自指写过就当全过 ⇒ 静默失守)
_write({"a6 目标状态": "过|goal_state() 返回 pass", "a1 产物": "不过|只交了两份"})
res["端到端W: 自指写过但真判据没过 ⇒ open"] = (m2.goal_state() == "open")
print(json.dumps(res, ensure_ascii=False))
'''
out = _sp.run([sys.executable, "-c", probe, str(HERE / "collabd.py")],
capture_output=True, text=True, timeout=180, encoding="utf-8")
got = json.loads(out.stdout.strip().splitlines()[-1])
for k, v in got.items():
res.append(("真起进程验行为:" + k, v))
except Exception as _e: # ⛔ 起不来就必须报红,不能静默跳过
res.append(("真起进程验行为(⛔ 起不来即报红):" + type(_e).__name__, False))
return res
@case("告警文案:**必须说清后果**(⛔ 不许只说「缺了 A」)")
def t_warn_text_states_consequence():
"""🔴 2026-10-03 用户逐字指正:「**标题里没读类别前缀) 这句话都不知道有什么作用**」。
🔴 **它错在哪**:原文案只陈述"读不到类别前缀"这个**现象**,
⛔ **没说缺了会怎样** ⇒ 读者不知道该做什么(看板上它跟正常文案长得一样,
只是短一点 —— 这种"只描述现象"的告警=**没有行动价值的废话**)。
🔴🔴 2026-10-03 **我自己犯了这条**(用户当场抓住):第一版我把它改成
「⚠ 无类别 ⇒ **不会被派活**」—— 🔴 **那个后果是我编的**。
取证:`collabd.py::queue_pending()` 读 **`tasks.json` 台账**,`topic` 在派活链路上
**根本没出现**;而这排 `cells` **已经是** `role==='协作目标'` 筛出的**现存协作会话**,
⛔ 不是"待派活清单" ⇒ "不会被派活"=**说了个不存在的后果**。
✅ **正确口径=只说能证实的后果**:`⚠ 归不出类别 · 不计入分工`
(依据:`board.py` 归线判据 `topic == 线 ∨ cwd_tail == 线` ⇒ topic 空即不归任何线)。
📌 **本条判据升级后的红线**:⚠️ **写"会怎样"之前必须先取证那个后果真的存在**
—— 否则"讲清后果"会变成"**编一个后果讲得更像回事**",比原句更坏。
✅ 本条判据是**通用红线**,不是这一句话的专属:
凡"⚠/缺了/没读"类文案,**必须同时出现一个后果词**(会…/不会…/导致…)。
"""
import re as _re6
html = (HERE.parent / "assets" / "board.html").read_text(encoding="utf-8")
# 剥掉注释 ⇒ 只查**真正渲染给用户看的文案**(注释里的原话留档不算)
code = _re6.sub(r"/\*.*?\*/", "", html, flags=_re6.S)
code = _re6.sub(r"(?m)^\s*//.*$", "", code)
code = _re6.sub(r"//[^\n]*", "", code)
# ⚠️ 只看**纯文案字面量**(`tx(...)` / `chip(...)` 这类调用的**第一个字符串参数**)
# ⛔ 不用"整段代码里抓 ⚠ 段"那种粗法 —— 实测它会**跨语句边界**把
# `(⛔ 不因此判完成)` 之类切到下一个字面量里 ⇒ **8 条里 7 条是误报**。
# ✅ 锚定 `tx(` / `chip(` 之后的第一个引号串,才是用户真正看到的句子。
lits = []
for m in _re6.finditer(r"(?:tx|chip|lbl)\(\s*(['\"])(.*?)\1", code, _re6.S):
s_txt = m.group(2)
if "⚠" in s_txt:
lits.append(s_txt)
# ⚠️ 后果词表**必须包含「不…」式否定** —— 例「⚠ 未声明验收判据(⛔ 不因此判完成)」
# 它**已经**把后果说清了(不判完成),⛔ 只因原表里只有"会/不能/无法"而漏判。
# 📌 教训:写判据时**先拿真实文案跑一遍**(本次就是这样才发现自己错,而不是去改好文案)。
bad = [w for w in lits
if not _re6.search(r"会|导致|漏管|不能|无法|勿|别|不因此|不算|勿当", w)]
return [
("渲染文案里**没有**「只说现象不说后果」的告警(查到 %d 条可疑:%s)"
% (len(bad), bad[:3] or "无"), not bad),
# 🔴🔴 2026-10-03 **用户直接指示:那一行整个删掉**(逐字:「越写越看不懂
# 还是删除了把」)⇒ 这里从"要求它写对"改成"**不许它回来**"。
# 📌 两次踩坑留下的结论:① 原文案只说现象(啰嗦)② 我"改成讲后果"又编了后果
# ⇒ **第三次选择是删掉**:这行信息量不足,⛔ 不值得占版面、更不值得编文案。
("⛔ 协作会话格**不许**再出现类别/前缀相关文案(用户已指示删除)",
("归不出类别" not in code) and ("没读类别前缀" not in code)
and ("类别:'+W.topic" not in code)),
("⛔ 那句编出来的后果也不许回来(`不会被派活`)", "不会被派活" not in code),
]
def _worker_label_from_source() -> str:
"""**AST 抽** `board.py::_ROLE_LABEL['worker']` 的字面量(唯一真源)。
🔴 2026-10-03 23:4x 新增:标签从「协作目标」改名成「执行会话」时,
判据里**写死**的那份字面量**当场报红**(这是它该做的),
但也意味着**每次改名都得手改判据** ⇒ 漏一处就假红。
✅ 正解=**判据从真源派生**:抽出来用 ⇒ 改名只改 `_ROLE_LABEL` 一处,判据自动跟随。
⚠️ 抽不到 ⇒ 返回空串(让下游断言**报红**,⛔ 绝不猜一个词填上去)。
"""
import ast as _ast
try:
t = _ast.parse((HERE / "board.py").read_text(encoding="utf-8"))
except Exception:
return ""
for n in t.body:
if not isinstance(n, _ast.Assign):
continue
tgt = n.targets[0]
nm = getattr(tgt, "id", None) or getattr(tgt, "attr", None)
if nm != "_ROLE_LABEL":
continue
for k, v in zip(n.value.keys, n.value.values):
if getattr(k, "value", "") == "worker" and isinstance(v, _ast.Constant):
return str(v.value)
return ""
@report("产物落点:**每个目标一个独立文件夹**(⛔ 代码与其他路径除外)|**报告型**")
def t_artifacts_land_in_goal_dir():
"""🔴 2026-10-04(用户要求逐字):「**检查是否每个工作区 进行 执行任务时每个目标都有独立的
文件夹 保存生成的内容(要求处理其他路径下的文件 代码 等除外)**」
🔴 **为什么补这条**(实测坐实的缺口):机制本身有(`goal_dir_name()`/`_ensure_goal_dir()`/
`t_goal_dir_mechanism`)⇒ ⛔ 但**那条只守"机制在不在",不守"产物真的落进去了吗"**。
实测反例:测试3 根目录长出 `新环境验证/`、`a6-final-check/`(2 份 S2/S3 取证 +
6 个临时修库脚本)**散在目标文件夹之外** ⇒ 机制在、结果照样乱。
✅ 本条判据=**逐个真工作区**(⛔ 不造夹具,因为要验的就是真磁盘):
① 目标文件夹**存在**(按 `goal_dir_name()` 现算,⛔ 不写死目录名);
② **产物型 `.md` 必须落在目标文件夹或 `交付物/` 内**;
③ ⛔ **豁免**:`CODEBUDDY.md`/`AGENTS.md`/`README.md`(规则与说明,不是产物)、
`docs/`、`归档/`、`脚本/代码`(用户口径:「**处理其他路径下的文件 代码 等除外**」)。
📌 **豁免写清楚,否则这条判据会假红**:一个连"什么是产物"都没定义的检查=没用的检查。
⚠️ 本条只**报告与提示**,⛔ **不删不移任何文件**(清理是用户的事,且可能是别人写的)。
"""
# 产物型文档:⛔ 排除规则类/说明类文件名(它们天生在根目录,不是产物)
RULE_FILES = {"CODEBUDDY.md", "AGENTS.md", "README.md", "MEMORY.md", "state.py"}
# 🔴 `inbox`=**机制自己的状态/信号文件**(NEXT/digest/queue/advance…),
# ⛔ **不是产物** —— 第一版没豁免它 ⇒ 报出一堆假散落(实测四个区都报)。
EXEMPT_DIRS = ("docs", "归档", "scripts", "脚本", "tmp", ".workbuddy", "done",
"__pycache__", "inbox")
# 🔴🔴 2026-10-04 **第三版才算对**:`imp()` 及其它用例会**永久改写** `DSH_COLLAB_WS`
# 指向**测试夹具**且不还原(实测 `selftest.py:184`)⇒ **跑到本用例时读它必是夹具**。
# ⇒ 真工作区只能用**模块加载时定下的 `WS`**(第 57 行,在 env 被改之前取的)。
# 📌 纪律:**判据里凡要"真工作区",一律用 `WS`,⛔ 不许现读 `os.environ`**。
ws = Path(WS)
if not ws.is_dir():
return [("SKIP:没指到真工作区(设 `DSH_COLLAB_WS=<工作区>` 才会真跑)", True)]
res = []
# ① 目标文件夹存在
# 🔴🔴 2026-10-04 **前两版都算错(留档)**:
# ① 第一版:直接 `exec_module` 但没设 `COLLABD_CONFIG` ⇒ `INBOX` 回落 ⇒ 报「未命名目标」。
# ② 第二版:设了 env,但**加载的是 `HERE/collabd.py`(技能目录那份)** ⇒ 它的
# `_cfg_candidates()` 里 `DSH_COLLAB_WS` 已被 `imp()` 指向**测试夹具** ⇒ 又回落。
# ⚠️ 更根子的一条:判据里**改 env 后再 `exec_module`**,而 `selftest` 顶层
# 已经跑过 `imp()`(它把 env 设为夹具)⇒ **我的设值与它的设值打架,谁最后生效不确定**。
# ✅ **正解=开独立子进程**(⛔ 不与 selftest 的 env 抢):真区自己的副本 + 干净 env
# ⇒ 拿到的是**真磁盘读数**,这才是本条要验的东西(同 P0-39:判据不许读错世界)。
import subprocess as _sb3
_own = ws / ".workbuddy" / "collab" / "collabd.py"
if not _own.is_file():
return [("⛔ 该工作区没有自己的 collabd.py 副本 ⇒ 判据无法验真实部署", False)]
_py = sys.executable
_code = (
"import importlib.util,sys,json\n"
"sp=importlib.util.spec_from_file_location('c',%r)\n"
"m=importlib.util.module_from_spec(sp)\n"
"try: sp.loader.exec_module(m)\n"
"except SystemExit: pass\n"
"print(json.dumps({'inbox':str(m.INBOX),'dir':m.goal_dir_rel()},ensure_ascii=False))\n"
% str(_own))
_env = {"COLLABD_CONFIG": str(ws / ".workbuddy" / "collab" / "collabd.config.json"),
"COLLABD_WORKSPACE": str(ws)}
try:
_r = _sb3.run([_py, "-c", _code], env={**os.environ, **_env},
capture_output=True, timeout=60, cwd=str(ws))
_j = json.loads((_r.stdout or b"{}").decode("utf-8", "replace") or "{}")
except Exception as _e:
return [("⛔ 独立子进程取真值失败:%r" % _e, False)]
_gdir = ws / (_j.get("dir") or "")
res = [("目标文件夹存在:%s(独立进程按 `goal.json` 算出,INBOX=%s)"
% (_j.get("dir"), _j.get("inbox")), bool(_j.get("dir")) and _gdir.is_dir())]
# ② 产物型 md 落点
stray, inside = [], 0
for p in ws.rglob("*.md"):
rel = p.relative_to(ws)
parts = rel.parts
if parts[0] in EXEMPT_DIRS or str(rel) in RULE_FILES:
continue
if p.name in RULE_FILES or p.name.startswith("CODEBUDDY"):
continue
# 只看「顶层 目标-*|交付物|其他一层目录」这一层,⛔ 不递归进目标文件夹内部
if len(parts) > 1 and parts[0].startswith("目标-"):
continue
if len(parts) == 1:
stray.append(str(rel)); continue
if parts[0] == "交付物":
inside += 1; continue
stray.append(str(rel))
res.append(("产物型 md 都落在「目标-*/交付物」内(⛔ 代码/规则/归档除外);"
"已在目标文件夹或交付物=%d,散在别处=%d %s"
% (inside, len(stray), ("→ " + str(stray[:5])) if stray else ""),
not stray))
return res
def main() -> int:
only = ""
if "-k" in sys.argv:
only = sys.argv[sys.argv.index("-k") + 1]
_prepare()
_snap_prod()
cases = [v for v in globals().values() if callable(v) and hasattr(v, "_case")]
npass = nfail = nrep = 0
print("=" * 66)
print("协作机制 · 回归自测(每改一次必跑) 测试工作区:%s" % TEST_WS)
print("=" * 66)
for fn in cases:
if only and only not in fn._case:
continue
try:
checks = fn()
except Exception as exc:
checks = [("用例抛异常:%r" % exc, False)]
bad = [c for c, ok in checks if not ok]
_ro = getattr(fn, "_report_only", False)
if bad:
nfail += 1
print(" ✗ %s" % fn._case)
for c, ok in checks:
print(" %s %s" % ("✓" if ok else "✗", c))
if _ro:
nrep += 1
nfail -= 1 # 🔴 报告型不计入失败(见 `report()` 的说明)
print(" ⓘ 报告型 ⇒ ⛔ 不计入 --verify 成败(存量问题,⛔ 与安装无关)")
else:
npass += 1
print(" ✓ %s (%d 项)" % (fn._case, len(checks)))
print("-" * 66)
print(" 合计:PASS %d / FAIL %d%s"
% (npass, nfail, ("(另有 %d 条报告型,⛔ 不计入)" % nrep) if nrep else ""))
print("=" * 66)
return 0 if nfail == 0 else 1
if __name__ == "__main__":
sys.exit(main())