技能包:配置与代码分家 —— 每包自建 env/ 放本机专属配置,scripts/ 不再写死本机路径

用户 2026-10-11 原话:
· 19:22「这个环境都是针对一台机器的,同步到别的机器把环境都搞坏了」
· 19:28「不用挪出 用env文件夹区分就行,重点是 脚本是脚本 配置是配置」
· 19:29「每个技能 需要环境配置的 都有自己的 env文件夹」
· 19:30「修改完成后要仔细检查 避免调整后 各技能脚本运行失败」

改动:
1、session-mechanism 新增 env/(roots.env=本机真身不入库 + roots.env.example=样例入库),
   install.py 的 ROOTS_ENV 改指 env/roots.env;全包脚本/hook 的配置定位统一为
   「向上 4 层 × env/roots.env 优先 → 包根 roots.env 兼容 → 技能库那份兜底」。
2、去硬编码:collabctl / keepalive / supervise-launch / board-launch 里的本机路径与本机工作区
   清单全清,改为 DSH_AIROOT(兜底 dirname(DSH_WS_ROOT))+ 扫 <AIROOT>/*/.workbuddy/collab/
   自动发现;保活任务名按 collabd-keepalive-<工作区名> 现推;启动器默认值改 ~/.workbuddy 形态。
3、browser-harness:bh_chrome_keepalive.py 默认值全部改 env 驱动,新增 env/bh.env.example。
4、.gitignore:env/ 下只放行 *.example(三行三种层级)+ 兜住旧位置的包根 roots.env。
5、改包纪律.md 新增「四、包结构:脚本是脚本,配置是配置」,立为全仓通用规矩。
6、文档同步:SKILL.md / deploy.md(新增 §3b 配置层级、§6 换机器必查 6 件事)/ rules.md /
   pitfalls.md / supervise-persistence.md。

验收(本机):py_compile 34 个 py 全过;selftest PASS 109 / FAIL 0;
逐文件实证 8 个 loader 命中 env/roots.env;preflight-lock.sh 同目录实测解出两个根;
install.py --manifest 77 份、语法失败 0。

异机验证(包拷成「新克隆」(只有 *.example)+ 另造假工作区 ws-alpha/ws-beta):
install.py --apply 自生成该机 env/roots.env;collabctl/keepalive 自动发现的确实是
ws-alpha/ws-beta、任务名现推(写死清单确已消除);部署副本本地无 roots.env 时靠
CODEBUDDY_CONFIG_DIR 兜底读到技能库那份;整套 selftest 有宿主库时 PASS 108 / FAIL 1
(唯一 FAIL=该区没有常驻心跳,环境所致;本机常驻在跑时同一套为 PASS 109 / FAIL 0)。
This commit is contained in:
admin committed 2026-10-11 21:49:28 +08:00
1 parent 90cc4315d5
commit d81f75f173
37 files changed
+763 -291

No files matched your search

+52 -6
View File
@@ -11,22 +11,68 @@
⇒ 正解=启动器(本文件):`pythonw.exe board-launch.py`,env 在 Python 进程内设。
⚠️ **看板是全局共用一份**(用户拍板)⇒ 它的配置**固定指向主工作区**(`DSH_WS_ROOT`)那份,
⛔ 不随工作区变。主工作区由 `roots.env` 的 `DSH_WS_ROOT` 决定,兜底写死 ai1net-dsh-server。
⛔ 不随工作区变。主工作区根由**宿主 env → 本包 `env/roots.env`** 决定(2026-10-11 起)。
"""
import os
import sys
# 🔴 主工作区根 —— 看板配置的真源(与 `roots.env` 的 `DSH_WS_ROOT` 同口径)
WS = os.environ.get("DSH_WS_ROOT") or r"E:\ProgramData\AIProject\ai1net-dsh-server"
WS = WS.replace("/", "\\")
import time
HERE = os.path.dirname(os.path.abspath(__file__))
# 🔴 2026-10-11:配置从本包 `env/roots.env` 读(「脚本是脚本 配置是配置」)——
# ⛔ 本文件**不写死**本机工作区 / 盘符;旧版兜底写死 ai1net-dsh-server ⇒ 换机器即错。
def _sm_load_roots():
import os as _os
_here = _os.path.dirname(_os.path.abspath(__file__))
for _up in range(4):
_base = _os.path.normpath(_os.path.join(_here, *([".."] * _up)))
for _rel in (("env", "roots.env"), ("roots.env",)):
_p = _os.path.join(_base, *_rel)
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()
def _fail(msg: str, code: int = 2):
"""⛔ 不静默退出:`pythonw` 下 `stderr` 可能是 `None` ⇒ 先写包内 `logs/`(不入库)留痕,再退出。"""
try:
_d = os.path.join(os.path.dirname(HERE), "logs")
os.makedirs(_d, exist_ok=True)
with open(os.path.join(_d, "board-launch.err.log"), "a", encoding="utf-8", errors="replace") as _f:
_f.write("%s %s\n" % (time.strftime("%Y-%m-%d %H:%M:%S"), msg))
except Exception:
pass
try:
if sys.stderr is not None:
sys.stderr.write(msg + "\n")
except Exception:
pass
raise SystemExit(code)
# 🔴 主工作区根 —— 看板配置的真源(宿主 env → `env/roots.env` 的 `DSH_WS_ROOT`)。
WS = (os.environ.get("DSH_WS_ROOT") or "").strip()
if not WS:
_fail("[board-launch] 未解析到 DSH_WS_ROOT ⇒ 不猜路径、不起看板。"
"要什么:设宿主 env,或在包内 `env/roots.env` 里给出主工作区根。")
WS = WS.replace("/", "\\")
BOARD = os.path.join(HERE, "board.py")
PORT = os.environ.get("DSH_BOARD_PORT") or "20099"
CFG = os.path.join(WS, ".workbuddy", "collab", "collabd.config.json")
os.environ["COLLABD_CONFIG"] = CFG
os.environ.setdefault("CODEBUDDY_CONFIG_DIR", r"E:\ProgramData\.workbuddy")
os.environ.setdefault("CODEBUDDY_CONFIG_DIR", os.path.expanduser("~/.workbuddy"))
os.environ["PYTHONIOENCODING"] = "utf-8"
os.environ["PYTHONUNBUFFERED"] = "1"
+17 -14
View File
@@ -25,24 +25,27 @@
⇒ 现规则:**端口上已有本看板 ⇒ 默认拒绝启动**(提示复用现成那个);要换代码用 `--takeover`。
"""
from __future__ import annotations
# [session-mechanism] roots.env 外置(由 install.py 生成;缺失则回落到按位置推导)
# [session-mechanism] 环境配置外置(由 install.py 生成);2026-10-11 起落在本包 `env/roots.env`
def _sm_load_roots():
import os as _os
_here = _os.path.dirname(_os.path.abspath(__file__))
# 🔴 2026-10-11:「脚本是脚本 配置是配置」⇒ 配置落本包 `env/`。
# 首选 `env/roots.env`;并兼容旧的包根 `roots.env`(拷贝未同步时照旧能跑)。
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
_base = _os.path.normpath(_os.path.join(_here, *([".."] * _up)))
for _rel in (("env", "roots.env"), ("roots.env",)):
_p = _os.path.join(_base, *_rel)
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
+17 -14
View File
@@ -34,24 +34,27 @@ import socket
import subprocess
import time
# [session-mechanism] roots.env 外置(由 install.py 生成;缺失则回落到按位置推导)
# [session-mechanism] 环境配置外置(由 install.py 生成);2026-10-11 起落在本包 `env/roots.env`
def _sm_load_roots():
import os as _os
_here = _os.path.dirname(_os.path.abspath(__file__))
# 🔴 2026-10-11:「脚本是脚本 配置是配置」⇒ 配置落本包 `env/`。
# 首选 `env/roots.env`;并兼容旧的包根 `roots.env`(拷贝未同步时照旧能跑)。
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
_base = _os.path.normpath(_os.path.join(_here, *([".."] * _up)))
for _rel in (("env", "roots.env"), ("roots.env",)):
_p = _os.path.join(_base, *_rel)
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
+81 -15
View File
@@ -42,6 +42,34 @@ import sys
import time
from ctypes import wintypes
# 🔴 2026-10-11:环境配置外置(「脚本是脚本 配置是配置」)⇒ 本包 `env/roots.env`。
# ⛔ 本文件**不写死**任何本机路径 / 本机工作区清单(写死 ⇒ 同步到别的机器会按不存在的路径建计划任务)。
def _sm_load_roots():
import os as _os
_here = _os.path.dirname(_os.path.abspath(__file__))
_cands = []
for _up in range(4):
_base = _os.path.normpath(_os.path.join(_here, *([".."] * _up)))
for _rel in (("env", "roots.env"), ("roots.env",)):
_cands.append(_os.path.join(_base, *_rel))
# 🔴 本机技能库那份兜底:工作区副本(`<ws>/.workbuddy/collab/`)靠它拿到「这台机器」的配置
# ⇒ 全机单一真源(副本目录里⛔ 不放第二份,避免各区分叉)。
_conf = _os.environ.get("CODEBUDDY_CONFIG_DIR") or _os.path.expanduser("~/.workbuddy")
_cands.append(_os.path.join(_conf, "skills", "session-mechanism", "env", "roots.env"))
for _p in _cands:
if _os.path.isfile(_p):
try:
with open(_p, encoding="utf-8") as _f:
for _ln in _f:
_ln = _ln.strip()
if _ln and not _ln.startswith("#") and "=" in _ln:
_k, _v = _ln.split("=", 1)
_os.environ.setdefault(_k.strip(), _v.strip())
except Exception:
pass
return
_sm_load_roots()
# ---------------------------------------------------------------- 常量
# 🔴🔴 2026-10-05 **两条旧结论被实测推翻**(都在本机跑出来的,见 skill 的 pitfalls P0-71):
# ① **`CREATE_BREAKAWAY_FROM_JOB` 在本机被拒** —— 带上它 `subprocess` 直接
@@ -60,18 +88,51 @@ HIDE = (0x08000000 # CREATE_NO_WINDOW(⛔ 没有它 ⇒ 闪黑窗)
# ⇒ **起进程直接失败**(2026-10-05 实测)。⇒ 起法=**先试带、被拒退回不带**;
# **只读查询(`_ps`)一律直接用它**,⛔ 别试 breakaway(试了就是白失败一次)。
HIDE_NO_BREAK = 0x08000000 | 0x00000008 | 0x00000200
AIROOT = "E:/ProgramData/AIProject"
HB_FRESH_S = 90 # 🔴 心跳新鲜阈值(与 `collabd.supervise_alive()` 同口径)
# 🔴 **登记制**:要管哪些工作区,只在**这一处**登记(⛔ 别在各处各写一份路径清单)。
WORKSPACES = [
"E:/ProgramData/AIProject/ai1net-dsh-server",
"E:/ProgramData/AIProject/vibe-product",
"E:/ProgramData/AIProject/contentm_agent",
]
# 看板:全局只一条(⛔ 不是每区一条)
BOARD_PY = "E:/ProgramData/.workbuddy/skills/session-mechanism/scripts/board.py"
def _resolve_airoot() -> str:
"""工作区根所在的**父目录**。**DSH_AIROOT 优先**(`env/roots.env` 注入);
兜底 = `DSH_WS_ROOT` 的父目录。⛔ 不在代码里写死盘符/目录。
🔴 2026-10-11 用户报(原话):「这个环境都是针对一台机器的,同步到别的机器把环境都搞坏了」
—— 旧版这里写死 `E:/ProgramData/AIProject`,同步到别的机器 ⇒ 路径不存在
⇒ `on|ensure` 会**建出一批指向空路径的计划任务**。
"""
cand = (os.environ.get("DSH_AIROOT") or "").strip()
if not cand:
ws = (os.environ.get("DSH_WS_ROOT") or "").strip()
if ws:
cand = os.path.dirname(ws.rstrip("/\\"))
return cand.replace("\\", "/") if cand else ""
def _discover_workspaces(airoot: str) -> list:
"""**自动发现**要管的工作区:扫 `AIROOT` 下装了 `.workbuddy/collab/` 的目录。
⛔ 不维护登记清单(登记=写死 ⇒ 别的机器上要么漏、要么指向空路径;本机也踩过
「本区不在名单里 ⇒ 改名残留的旧任务谁都没扫到」的判断盲区)。
"""
if not airoot or not os.path.isdir(airoot):
return []
out = []
for name in sorted(os.listdir(airoot)):
d = os.path.join(airoot, name)
if os.path.isdir(os.path.join(d, ".workbuddy", "collab")):
out.append(d.replace("\\", "/"))
return out
def _keepalive_tasks(workspaces: list) -> dict:
"""工作区 → 保活任务名。**由目录名推导**:`collabd-keepalive-<basename>`(⛔ 不写死字典)。"""
return {w: "collabd-keepalive-" + os.path.basename(w.rstrip("/\\")) for w in workspaces}
AIROOT = _resolve_airoot()
WORKSPACES = _discover_workspaces(AIROOT)
# 看板:全局只一条(⛔ 不是每区一条);**技能包内相对定位** ⇒ 换机器不用改。
BOARD_PY = os.path.join(os.path.dirname(os.path.abspath(__file__)), "board.py")
BOARD_PORT = 20099
# 🔴🔴 2026-10-05 改(用户定案:「**常驻崩了,拉起来这个事儿,你不能也用计划任务起个程序吗?**」):
@@ -81,11 +142,9 @@ BOARD_PORT = 20099
# ✅ 唯一合格载体=**计划任务**(父链根与会话树无关 —— 实测:父进程已退出、链断在自己身上)。
# ✅ 动作=**`pythonw.exe`**(GUI 子系统 ⇒ 不弹窗;⛔ 不是 `powershell.exe`,那是控制台程序)。
# ⇒ 所以本入口的 `on`/`ensure` 现在**建/启用这两组任务**,`off` **禁用它们+杀进程**。
KEEPALIVE_TASKS = { # 新形态:计划任务 → pythonw collabd.py --supervise
"E:/ProgramData/AIProject/ai1net-dsh-server": "collabd-keepalive-ai1net-dsh-server",
"E:/ProgramData/AIProject/vibe-product": "collabd-keepalive-vibe-product",
"E:/ProgramData/AIProject/contentm_agent": "collabd-keepalive-contentm_agent",
}
# 新形态:计划任务 → pythonw collabd.py --supervise。🔴 2026-10-11:**由工作区名推导**
# (`collabd-keepalive-<basename>`)—— 原为写死字典 ⇒ 换机器/改名即失效。
KEEPALIVE_TASKS = _keepalive_tasks(WORKSPACES)
BOARD_TASK = "dsh-board-keepalive"
# 旧任务:一律**禁用**(⛔ 不删除,可逆)。它们是弹窗事故的来源(动作=powershell.exe)。
@@ -551,7 +610,7 @@ def cmd_ensure():
_p("── ensure:按开关收敛(on⇒补齐缺失的,off⇒全停)──")
on_ws = [w for w in WORKSPACES if switch_of(w) == "on"]
if not on_ws:
_p(" 两区开关都 off ⇒ 走全停")
_p(" 各区开关都 off ⇒ 走全停")
cmd_off()
return
disable_tasks() # 旧形态任务:见一次禁一次(新形态由 start_all 重建)
@@ -634,6 +693,13 @@ def main():
except Exception as e:
print("⛔ 打不开 --out %s:%r" % (out, e))
return 1
# 🔴 2026-10-11 自适配:解析不到工作区根 ⇒ **优雅报错**,⛔ 不猜路径、
# ⛔ 不去建指向空路径的计划任务(那就是用户说的"把环境搞坏了")。
if not AIROOT:
_p("⚠️ 未解析到工作区根 ⇒ 本入口**不猜路径**。")
_p(" 要什么:设 `DSH_AIROOT`(或 `DSH_WS_ROOT`),或先跑 `install.py --apply` 生成 `env/roots.env`。")
if a in ("on", "ensure"):
return 1
try:
if a == "status":
cmd_status()
+17 -14
View File
@@ -66,24 +66,27 @@
(无参数仍=常驻,保留兼容;**定案形态以 `--supervise` 为准**)
"""
from __future__ import annotations
# [session-mechanism] roots.env 外置(由 install.py 生成;缺失则回落到按位置推导)
# [session-mechanism] 环境配置外置(由 install.py 生成);2026-10-11 起落在本包 `env/roots.env`
def _sm_load_roots():
import os as _os
_here = _os.path.dirname(_os.path.abspath(__file__))
# 🔴 2026-10-11:「脚本是脚本 配置是配置」⇒ 配置落本包 `env/`。
# 首选 `env/roots.env`;并兼容旧的包根 `roots.env`(拷贝未同步时照旧能跑)。
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
_base = _os.path.normpath(_os.path.join(_here, *([".."] * _up)))
for _rel in (("env", "roots.env"), ("roots.env",)):
_p = _os.path.join(_base, *_rel)
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
@@ -14,24 +14,27 @@
⛔ 失败一律**静默退出 0** —— 它是旁路,⛔ 不得影响钩子本身、更不得影响 WorkBuddy 运行。
"""
from __future__ import annotations
# [session-mechanism] roots.env 外置(由 install.py 生成;缺失则回落到按位置推导)
# [session-mechanism] 环境配置外置(由 install.py 生成);2026-10-11 起落在本包 `env/roots.env`
def _sm_load_roots():
import os as _os
_here = _os.path.dirname(_os.path.abspath(__file__))
# 🔴 2026-10-11:「脚本是脚本 配置是配置」⇒ 配置落本包 `env/`。
# 首选 `env/roots.env`;并兼容旧的包根 `roots.env`(拷贝未同步时照旧能跑)。
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
_base = _os.path.normpath(_os.path.join(_here, *([".."] * _up)))
for _rel in (("env", "roots.env"), ("roots.env",)):
_p = _os.path.join(_base, *_rel)
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
+31 -14
View File
@@ -63,24 +63,27 @@
⚠️ 任何写动作前**先备份**(`.bak-goalctl-<日期>/`);⛔ 脚本从不删文件。
"""
from __future__ import annotations
# [session-mechanism] roots.env 外置(由 install.py 生成;缺失则回落到按位置推导)
# [session-mechanism] 环境配置外置(由 install.py 生成);2026-10-11 起落在本包 `env/roots.env`
def _sm_load_roots():
import os as _os
_here = _os.path.dirname(_os.path.abspath(__file__))
# 🔴 2026-10-11:「脚本是脚本 配置是配置」⇒ 配置落本包 `env/`。
# 首选 `env/roots.env`;并兼容旧的包根 `roots.env`(拷贝未同步时照旧能跑)。
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
_base = _os.path.normpath(_os.path.join(_here, *([".."] * _up)))
for _rel in (("env", "roots.env"), ("roots.env",)):
_p = _os.path.join(_base, *_rel)
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
@@ -728,6 +731,20 @@ def declare() -> int:
if _is_switch:
try:
GOALS_DIR.mkdir(parents=True, exist_ok=True)
# 🔴 2026-10-11 旧目标收尾:换目标 ⇒ 旧目标整份归档。若它还停在
# 「进行中/等待/阻碍」(= 没做完就被换掉),置为「已暂停」——
# 否则归档件带着「进行中」进 `goals/`,看板照 `lifecycle` 分列,
# 会把这些**已无人推进**的目标一直摆在「执行中」(2026-10-11 实测:
# 看板「执行中」4 张里 3 张都是这类)。⛔ 原本「已完成」的保持不动。
_old_raw = str(g.get("lifecycle") or "").strip()
if not _old_raw.startswith("已完成"):
g["lifecycle"] = "已暂停"
g["lifecycle_at"] = time.strftime("%Y-%m-%dT%H:%M:%S")
g["lifecycle_by"] = "goalctl declare --switch-goal(换目标 ⇒ 旧目标收尾)"
g["lifecycle_why"] = ("换目标 ⇒ 旧目标收尾为「已暂停」(改前=%s);"
"它不再被推进,归档件可回查" % (_old_raw or "未声明"))
print(" 🔴 旧目标已收尾:`lifecycle` %s ⇒ 已暂停(不再推进,看板归「待执行」)"
% (_old_raw or "(未声明)"))
_safe = re.sub(r'[\\/:*?"<>|\r\n\t]+', "-", old_title).strip()[:40] or "untitled"
_dst = GOALS_DIR / ("%s__%s.json" % (_safe, time.strftime("%Y%m%d-%H%M")))
_n = 1
+16 -13
View File
@@ -44,7 +44,7 @@ import time
from pathlib import Path
_STAMP_NAME = "env-stamp.json"
_ROOTS_ENV = "roots.env" # 技能根的权威落点(install.py 生成)
_ROOTS_ENV = "roots.env" # 文件名;落点是技能包 `<包>/env/roots.env`(install.py 生成)
# 🔴🔴 2026-10-04 单包自包含改造:识别特征改成**多个候选**,⛔ 不再只认
# `agent-operating-rules`(那是**另一个**技能 ⇒ 只装本包时会把真技能库判成"不存在")。
# 🔴 旧写法的病:`skills_root()` / `env_check()` 三处都拿`(root)/_SNIFF).is_dir()` 当
@@ -74,21 +74,24 @@ def _norm(p) -> str:
def _load_roots_env() -> dict:
"""读技能包里的 `roots.env`(⛔ 各脚本的 `_sm_load_roots()` 注入 env 前的原始来源)。"""
"""读技能包里的 `env/roots.env`(⛔ 各脚本的 `_sm_load_roots()` 注入 env 前的原始来源)。
查找顺序:向上逐层 `<d>/env/roots.env`(现行落点)→ `<d>/roots.env`(包根兼容)。
"""
out = {}
d = Path(__file__).resolve().parent
for _ in range(5):
f = d / _ROOTS_ENV
if f.is_file():
try:
for ln in f.read_text(encoding="utf-8").splitlines():
ln = ln.strip()
if ln and not ln.startswith("#") and "=" in ln:
k, v = ln.split("=", 1)
out[k.strip()] = v.strip()
return out
except Exception:
return {}
for f in (d / "env" / _ROOTS_ENV, d / _ROOTS_ENV):
if f.is_file():
try:
for ln in f.read_text(encoding="utf-8").splitlines():
ln = ln.strip()
if ln and not ln.startswith("#") and "=" in ln:
k, v = ln.split("=", 1)
out[k.strip()] = v.strip()
return out
except Exception:
return {}
n = d.parent
if n == d:
break
@@ -29,24 +29,27 @@
⛔ **别给本脚本加 `-E`**:`-E` 会屏蔽 `PYTHONUTF8`/`PYTHONIOENCODING` ⇒ stdin 回退 cp936 ⇒
含中文的 payload 解析失败且**静默 fail-open**(同目录 `stop-dialog-guard.py` 已因此"白排查一天")。
"""
# [session-mechanism] roots.env 外置(由 install.py 生成;缺失则回落到按位置推导)
# [session-mechanism] 环境配置外置(由 install.py 生成);2026-10-11 起落在本包 `env/roots.env`
def _sm_load_roots():
import os as _os
_here = _os.path.dirname(_os.path.abspath(__file__))
# 🔴 2026-10-11:「脚本是脚本 配置是配置」⇒ 配置落本包 `env/`。
# 首选 `env/roots.env`;并兼容旧的包根 `roots.env`(拷贝未同步时照旧能跑)。
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
_base = _os.path.normpath(_os.path.join(_here, *([".."] * _up)))
for _rel in (("env", "roots.env"), ("roots.env",)):
_p = _os.path.join(_base, *_rel)
if _os.path.isfile(_p):
try:
with open(_p, encoding="utf-8") as _f:
for _ln in _f:
_ln = _ln.strip()
if _ln and not _ln.startswith("#") and "=" in _ln:
_k, _v = _ln.split("=", 1)
_os.environ.setdefault(_k.strip(), _v.strip())
except Exception:
pass
return
_sm_load_roots()
import io
@@ -32,24 +32,27 @@ lock-guard-hook.py — 「无锁不许改库」的强制钩子(WorkBuddy PreTo
退出码:始终 0(判定通过 JSON 输出表达);脚本自身异常也放行,绝不误伤。
"""
# [session-mechanism] roots.env 外置(由 install.py 生成;缺失则回落到按位置推导)
# [session-mechanism] 环境配置外置(由 install.py 生成);2026-10-11 起落在本包 `env/roots.env`
def _sm_load_roots():
import os as _os
_here = _os.path.dirname(_os.path.abspath(__file__))
# 🔴 2026-10-11:「脚本是脚本 配置是配置」⇒ 配置落本包 `env/`。
# 首选 `env/roots.env`;并兼容旧的包根 `roots.env`(拷贝未同步时照旧能跑)。
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
_base = _os.path.normpath(_os.path.join(_here, *([".."] * _up)))
for _rel in (("env", "roots.env"), ("roots.env",)):
_p = _os.path.join(_base, *_rel)
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()
@@ -69,13 +69,19 @@
(原写「5 条 hook 入口」,2026-10-01 起为 **7 条**:lock-guard ×2 / bash-output-guard / skill-load-guard /
stop-dialog-guard / **session-log-guard ×2**)。
"""
# [session-mechanism] roots.env 外置(由 install.py 生成;缺失则回落到按位置推导)
# [session-mechanism] roots.env 外置(由 install.py 生成 ⇒ `<包>/env/roots.env`)
# 查找顺序:向上 4 层 × (env/roots.env 优先 → 包根 roots.env 兼容) → 技能库那份。
def _sm_load_roots():
import os as _os
_here = _os.path.dirname(_os.path.abspath(__file__))
_cands = []
for _up in range(4):
_p = _os.path.join(_here, *([".."] * _up), "roots.env")
_p = _os.path.normpath(_p)
_base = _os.path.normpath(_os.path.join(_here, *([".."] * _up)))
for _rel in (("env", "roots.env"), ("roots.env",)):
_cands.append(_os.path.join(_base, *_rel))
_conf = _os.environ.get("CODEBUDDY_CONFIG_DIR") or _os.path.expanduser("~/.workbuddy")
_cands.append(_os.path.join(_conf, "skills", "session-mechanism", "env", "roots.env"))
for _p in _cands:
if _os.path.isfile(_p):
try:
with open(_p, encoding="utf-8") as _f:
@@ -41,13 +41,19 @@ AI 仍按常驻判据把两个**非门禁**问题("106 旧控制面要不要
18:38 起本钩子与 `stop-dialog-guard` 就在真实用户发言上**热生效**,未重启宿主;`PreToolUse` 同类)。
⇒ 改完仍应**喂一次模拟载荷端到端自证**,⛔ 别只改配置就宣布"修好了"。
"""
# [session-mechanism] roots.env 外置(由 install.py 生成;缺失则回落到按位置推导)
# [session-mechanism] roots.env 外置(由 install.py 生成 ⇒ `<包>/env/roots.env`)
# 查找顺序:向上 4 层 × (env/roots.env 优先 → 包根 roots.env 兼容) → 技能库那份。
def _sm_load_roots():
import os as _os
_here = _os.path.dirname(_os.path.abspath(__file__))
_cands = []
for _up in range(4):
_p = _os.path.join(_here, *([".."] * _up), "roots.env")
_p = _os.path.normpath(_p)
_base = _os.path.normpath(_os.path.join(_here, *([".."] * _up)))
for _rel in (("env", "roots.env"), ("roots.env",)):
_cands.append(_os.path.join(_base, *_rel))
_conf = _os.environ.get("CODEBUDDY_CONFIG_DIR") or _os.path.expanduser("~/.workbuddy")
_cands.append(_os.path.join(_conf, "skills", "session-mechanism", "env", "roots.env"))
for _p in _cands:
if _os.path.isfile(_p):
try:
with open(_p, encoding="utf-8") as _f:
@@ -32,13 +32,19 @@
含中文的 payload 解码即炸。本脚本现已改为走 `buffer` 显式 UTF-8(读写都加固),但**不要靠加固兜底**,
装的时候也别再引入新雷。(2026-09-15 实测:`-S -E` 曾让本钩子"看起来从未被调用"整整一天。)
"""
# [session-mechanism] roots.env 外置(由 install.py 生成;缺失则回落到按位置推导)
# [session-mechanism] roots.env 外置(由 install.py 生成 ⇒ `<包>/env/roots.env`)
# 查找顺序:向上 4 层 × (env/roots.env 优先 → 包根 roots.env 兼容) → 技能库那份。
def _sm_load_roots():
import os as _os
_here = _os.path.dirname(_os.path.abspath(__file__))
_cands = []
for _up in range(4):
_p = _os.path.join(_here, *([".."] * _up), "roots.env")
_p = _os.path.normpath(_p)
_base = _os.path.normpath(_os.path.join(_here, *([".."] * _up)))
for _rel in (("env", "roots.env"), ("roots.env",)):
_cands.append(_os.path.join(_base, *_rel))
_conf = _os.environ.get("CODEBUDDY_CONFIG_DIR") or _os.path.expanduser("~/.workbuddy")
_cands.append(_os.path.join(_conf, "skills", "session-mechanism", "env", "roots.env"))
for _p in _cands:
if _os.path.isfile(_p):
try:
with open(_p, encoding="utf-8") as _f:
@@ -49,17 +49,21 @@ except Exception:
# ① stdin 的 `payload.cwd`(宿主一定给)
# ② `roots.env` 的 `DSH_WS_ROOT`(副本专属,⛔ 不能只靠宿主 env)
def _sm_load_roots() -> None:
"""把 `roots.env` 以 setdefault 注入 env(宿主 env 优先,⛔ 不覆盖)。"""
p = SKILL / "roots.env"
try:
for line in p.read_text(encoding="utf-8").splitlines():
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
k, v = line.split("=", 1)
os.environ.setdefault(k.strip(), v.strip())
except Exception:
pass
"""把本包 `env/roots.env` 以 setdefault 注入 env(宿主 env 优先,⛔ 不覆盖)。"""
cands = [SKILL / "env" / "roots.env", SKILL / "roots.env"]
_cfg = os.environ.get("CODEBUDDY_CONFIG_DIR") or os.path.expanduser("~/.workbuddy")
cands.append(Path(_cfg) / "skills" / "session-mechanism" / "env" / "roots.env")
for p in cands:
try:
for line in p.read_text(encoding="utf-8").splitlines():
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
k, v = line.split("=", 1)
os.environ.setdefault(k.strip(), v.strip())
return
except Exception:
continue
_sm_load_roots()
+74 -12
View File
@@ -30,7 +30,7 @@
**已在 ⇒ 新起的立刻让位退出**(幂等,⛔ 不会"叠加成两条")。
**一键关掉(新增能力)**:本脚本支持 `off` ⇒ 一个动作做完四件事:
① 关两区开关 ② 禁用所有相关计划任务 ③ 杀掉所有相关进程 ④ 打印复核结果
① 关各区开关 ② 禁用所有相关计划任务 ③ 杀掉所有相关进程 ④ 打印复核结果
⇒ ⛔ 不再需要"禁用任务 + 杀进程"分开记(10-05 就是因为只禁任务没杀进程,以为停了其实没停)。
用法:
@@ -43,16 +43,78 @@ import os
import subprocess
import sys
PYW = r"E:\ProgramData\.workbuddy\binaries\python\versions\3.13.12\pythonw.exe"
AIROOT = r"E:\ProgramData\AIProject"
# 🔴 登记制:工作区 → 保活任务名(⛔ 只在这里写一份)
WS = {
"ai1net-dsh-server": "collabd-keepalive-ai1net-dsh-server",
"vibe-product": "collabd-keepalive-vibe-product",
}
# 看板:全局只一条
BOARD_PY = r"E:\ProgramData\.workbuddy\skills\session-mechanism\scripts\board.py"
# 🔴 2026-10-11:环境配置外置(「脚本是脚本 配置是配置」)⇒ 本包 `env/roots.env`。
# ⛔ 本文件**不写死**任何本机路径 / 本机工作区清单。
def _sm_load_roots():
import os as _os
_here = _os.path.dirname(_os.path.abspath(__file__))
_cands = []
for _up in range(4):
_base = _os.path.normpath(_os.path.join(_here, *([".."] * _up)))
for _rel in (("env", "roots.env"), ("roots.env",)):
_cands.append(_os.path.join(_base, *_rel))
# 🔴 本机技能库那份兜底:工作区副本(`<ws>/.workbuddy/collab/`)靠它拿到「这台机器」的配置
# ⇒ 全机单一真源(副本目录里⛔ 不放第二份,避免各区分叉)。
_conf = _os.environ.get("CODEBUDDY_CONFIG_DIR") or _os.path.expanduser("~/.workbuddy")
_cands.append(_os.path.join(_conf, "skills", "session-mechanism", "env", "roots.env"))
for _p in _cands:
if _os.path.isfile(_p):
try:
with open(_p, encoding="utf-8") as _f:
for _ln in _f:
_ln = _ln.strip()
if _ln and not _ln.startswith("#") and "=" in _ln:
_k, _v = _ln.split("=", 1)
_os.environ.setdefault(_k.strip(), _v.strip())
except Exception:
pass
return
_sm_load_roots()
def _resolve_airoot() -> str:
"""工作区根所在的**父目录**。**DSH_AIROOT 优先**(`env/roots.env` 注入);
兜底 = `DSH_WS_ROOT` 的父目录。⛔ 不在代码里写死盘符/目录。"""
cand = (os.environ.get("DSH_AIROOT") or "").strip()
if not cand:
ws = (os.environ.get("DSH_WS_ROOT") or "").strip()
if ws:
cand = os.path.dirname(ws.rstrip("/\\"))
return os.path.normpath(cand) if cand else ""
def _discover_workspace_names(airoot: str) -> list:
"""**自动发现**要管的工作区名:扫 `AIROOT` 下装了 `.workbuddy/collab/` 的目录。"""
if not airoot or not os.path.isdir(airoot):
return []
return sorted(n for n in os.listdir(airoot)
if os.path.isdir(os.path.join(airoot, n, ".workbuddy", "collab")))
def _resolve_pyw() -> str:
"""`pythonw.exe`:优先**当前解释器**(`sys.executable`);其次宿主 env;
兜底才在同目录**探测**兄弟 `pythonw.exe`。⛔ 不写死版本目录。"""
exe = os.path.abspath(sys.executable or "")
if exe and exe.lower().endswith("pythonw.exe") and os.path.isfile(exe):
return exe
for _k in ("DSH_PYTHONW", "PYTHONW", "DSH_PYTHON"):
_v = (os.environ.get(_k) or "").strip()
if _v and os.path.isfile(_v):
return _v
_sib = os.path.join(os.path.dirname(exe), "pythonw.exe") if exe else ""
if _sib and os.path.isfile(_sib):
return _sib
return exe or "pythonw.exe"
AIROOT = _resolve_airoot()
PYW = _resolve_pyw()
# 🔴 工作区名 → 保活任务名:**由目录名推导**(`collabd-keepalive-<名>`)—— 原为写死字典 ⇒ 换机器/改名即失效。
WS = {n: "collabd-keepalive-" + n for n in _discover_workspace_names(AIROOT)}
# 看板:全局只一条;**技能包内相对定位**(⛔ 不写死绝对路径)
BOARD_PY = os.path.join(os.path.dirname(os.path.abspath(__file__)), "board.py")
BOARD_TASK = "dsh-board-keepalive"
BOARD_PORT = 20099
@@ -245,13 +307,13 @@ def main():
if not APPLY and CMD in ("on", "off"):
print("(干跑)将要执行:%s —— 加 --apply 真做" % CMD)
if CMD == "off":
print(" ① 关两区开关 ② 禁用 %d 个计划任务 ③ 杀相关进程 ④ 复核"
print(" ① 关各区开关 ② 禁用 %d 个计划任务 ③ 杀相关进程 ④ 复核"
% (len(WS) + 1 + len(OLD_TASKS)))
else:
for ws, t in WS.items():
print(" 建任务 %s → 每 %d 分钟 pythonw collabd.py --supervise" % (t, INTERVAL_MIN))
print(" 建任务 %s → pythonw board.py --serve %d" % (BOARD_TASK, BOARD_PORT))
print(" 禁用旧任务 %d 个;两区开关置 on" % len(OLD_TASKS))
print(" 禁用旧任务 %d 个;已发现区的开关置 on" % len(OLD_TASKS))
return 0
if CMD == "status":
cmd_status()
@@ -34,10 +34,15 @@
# 退出码:0 = 放行;1 = 命中硬冲突/硬判定;2 = 环境错误
set -uo pipefail
# [session-mechanism] roots.env 外置(由 install.py 生成;缺失则回落到按位置推导)
# [session-mechanism] 环境配置外置(由 install.py 生成);2026-10-11 起落在本包 `env/roots.env`
# 首选 env/roots.env;并兼容旧的包根 roots.env(拷贝未同步时照旧能跑)。
_SM_HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
_SM_PKG="$(cd "$_SM_HERE/../.." && pwd)"
[ -f "$_SM_PKG/roots.env" ] && . "$_SM_PKG/roots.env"
if [ -f "$_SM_PKG/env/roots.env" ]; then
. "$_SM_PKG/env/roots.env"
elif [ -f "$_SM_PKG/roots.env" ]; then
. "$_SM_PKG/roots.env"
fi
ROOT="${DSH_DOCS_ROOT:-$(cd "$_SM_HERE/.." && pwd)}"
cd "$ROOT" || exit 2
LOCKDIR="$ROOT/05-交接单/.doing-"
@@ -8,13 +8,19 @@
用法:python 07-scripts/handoff-status.py
"""
# [session-mechanism] roots.env 外置(由 install.py 生成;缺失则回落到按位置推导)
# [session-mechanism] roots.env 外置(由 install.py 生成 ⇒ `<包>/env/roots.env`)
# 查找顺序:向上 4 层 × (env/roots.env 优先 → 包根 roots.env 兼容) → 技能库那份。
def _sm_load_roots():
import os as _os
_here = _os.path.dirname(_os.path.abspath(__file__))
_cands = []
for _up in range(4):
_p = _os.path.join(_here, *([".."] * _up), "roots.env")
_p = _os.path.normpath(_p)
_base = _os.path.normpath(_os.path.join(_here, *([".."] * _up)))
for _rel in (("env", "roots.env"), ("roots.env",)):
_cands.append(_os.path.join(_base, *_rel))
_conf = _os.environ.get("CODEBUDDY_CONFIG_DIR") or _os.path.expanduser("~/.workbuddy")
_cands.append(_os.path.join(_conf, "skills", "session-mechanism", "env", "roots.env"))
for _p in _cands:
if _os.path.isfile(_p):
try:
with open(_p, encoding="utf-8") as _f:
@@ -44,9 +44,16 @@ from pathlib import Path
# ── 根目录定位(与包内其它脚本同款:宿主 env > `roots.env` 兜底)────────────
def _sm_load_roots() -> None:
"""把本包 `env/roots.env` 以 setdefault 注入 env(宿主 env 优先,⛔ 不覆盖)。"""
here = os.path.dirname(os.path.abspath(__file__))
cands = []
for up in range(4):
p = os.path.normpath(os.path.join(here, *([".."] * up), "roots.env"))
base = os.path.normpath(os.path.join(here, *([".."] * up)))
for rel in (("env", "roots.env"), ("roots.env",)):
cands.append(os.path.join(base, *rel))
conf = os.environ.get("CODEBUDDY_CONFIG_DIR") or os.path.expanduser("~/.workbuddy")
cands.append(os.path.join(conf, "skills", "session-mechanism", "env", "roots.env"))
for p in cands:
if os.path.isfile(p):
try:
with open(p, encoding="utf-8") as f:
@@ -22,10 +22,11 @@ if [ -z "$ME" ] || [ $# -eq 0 ]; then
exit 2
fi
# [session-mechanism] roots.env 外置(由 install.py 生成;缺失则回落到按位置推导)
# [session-mechanism] roots.env 外置(由 install.py 生成 ⇒ `<包>/env/roots.env`;兼容包根旧位置)
_SM_HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
_SM_PKG="$(cd "$_SM_HERE/../.." && pwd)"
[ -f "$_SM_PKG/roots.env" ] && . "$_SM_PKG/roots.env"
if [ -f "$_SM_PKG/env/roots.env" ]; then . "$_SM_PKG/env/roots.env"
elif [ -f "$_SM_PKG/roots.env" ]; then . "$_SM_PKG/roots.env"; fi
DOCS_ROOT="${DSH_DOCS_ROOT:-$(cd "$_SM_HERE/.." && pwd)}"
CODE_ROOT="${DSH_CODE_REPO:-${DOCS_ROOT%/*}}" # ⛔ 不留盘符:代码库=文档库的上一级(纯 bash 展开,⛔ 不调 dirname)
WS_ROOT="${DSH_WS_ROOT:-$(pwd)}" # ⛔ 不留盘符:默认=当前工作目录(本文件内未被使用)
+35 -29
View File
@@ -13,24 +13,27 @@
python selftest.py -k 投递 # 只跑名字含"投递"的用例
"""
from __future__ import annotations
# [session-mechanism] roots.env 外置(由 install.py 生成;缺失则回落到按位置推导)
# [session-mechanism] 环境配置外置(由 install.py 生成);2026-10-11 起落在本包 `env/roots.env`
def _sm_load_roots():
import os as _os
_here = _os.path.dirname(_os.path.abspath(__file__))
# 🔴 2026-10-11:「脚本是脚本 配置是配置」⇒ 配置落本包 `env/`。
# 首选 `env/roots.env`;并兼容旧的包根 `roots.env`(拷贝未同步时照旧能跑)。
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
_base = _os.path.normpath(_os.path.join(_here, *([".."] * _up)))
for _rel in (("env", "roots.env"), ("roots.env",)):
_p = _os.path.join(_base, *_rel)
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
@@ -5833,7 +5836,7 @@ def t_hook_reads_ws_from_stdin_cwd():
⚖️ 同源判据的既有正确写法(⛔ 必须对齐,不能自创):
`wb-result-hook.py` / `lock-guard-hook.py` 都是**先读 stdin 的
`payload.cwd`**,env / `roots.env` 只作兜底。
`payload.cwd`**,env / `env/roots.env` 只作兜底。
🔴 判据要点:⛔ **不许只查"读没读 cwd"**——还要钉住
① stdin **只读一次**(`sys.stdin.buffer.read()` 是一次性的)
@@ -6454,7 +6457,9 @@ def t_mirror_drift():
(fx_cp / "a.py").write_text("A=1\n", encoding="utf-8")
(fx_cp / "sub" / "b.py").write_text("B=1\n# TAMPER\n", encoding="utf-8") # 待更新
(fx_cp / "r_stale.py").write_text("STALE\n", encoding="utf-8") # 真残留
(fx_cp / "roots.env").write_text("DSH_WS_ROOT=/fixture/ONLY\n", encoding="utf-8") # 白名单
# 🔴 2026-10-11:白名单落点在 `env/roots.env`(配置与代码分家)⇒ 夹具照真实结构摆
(fx_cp / "env").mkdir(parents=True, exist_ok=True)
(fx_cp / "env" / "roots.env").write_text("DSH_WS_ROOT=/fixture/ONLY\n", encoding="utf-8") # 白名单
g, c, only_c, only_g, changed = wm.diff(fx_src, fx_cp)
det = [
@@ -6464,7 +6469,7 @@ def t_mirror_drift():
"c.py" in only_g and len(only_g) == 1),
("抓到**待更新**(两边都有但内容不同)",
"sub/b.py" in changed and "a.py" not in changed),
("**没把白名单当漂移**(roots.env 不该出现在任一差异里)",
("**没把白名单当漂移**(env/roots.env 不该出现在任一差异里)",
not any("roots.env" in x for x in (only_c + only_g + changed))),
]
@@ -6480,10 +6485,10 @@ def t_mirror_drift():
and (fx_cp / "c.py").read_bytes() == (fx_src / "c.py").read_bytes()),
("同步后**篡改已被覆盖回全局版**",
(fx_cp / "sub" / "b.py").read_bytes() == (fx_src / "sub" / "b.py").read_bytes()),
("🔴 **白名单内容原样保住**(roots.env 仍写 /fixture/ONLY,"
("🔴 **白名单内容原样保住**(env/roots.env 仍写 /fixture/ONLY,"
"⛔ 被冲回全局值就会让副本去读写别人的仓库)",
(fx_cp / "roots.env").exists()
and "/fixture/ONLY" in (fx_cp / "roots.env").read_text(encoding="utf-8")),
(fx_cp / "env" / "roots.env").exists()
and "/fixture/ONLY" in (fx_cp / "env" / "roots.env").read_text(encoding="utf-8")),
("**幂等**:再 sync 一次 rc=0", wm.do_sync(fx_src, fx_cp) == 0),
]
@@ -6493,7 +6498,7 @@ def t_mirror_drift():
wsrc = mp.read_text(encoding="utf-8")
det += [
("覆盖循环里有**白名单守卫** `if k in KEEP: continue`"
"(⛔ 拿掉它 ⇒ roots.env 被全局值覆盖 ⇒ 副本去读写别人的仓库)",
"(⛔ 拿掉它 ⇒ env/roots.env 被全局值覆盖 ⇒ 副本去读写别人的仓库)",
re.search(r"for k in sorted\(g\):\s*\n\s*if k in KEEP:\s*\n\s*continue", wsrc) is not None),
("`--sync` 先**备份再删**(⛔ 破坏性动作前必须留退路)",
wsrc.index("copytree") < wsrc.index("unlink()")),
@@ -6543,30 +6548,31 @@ def t_mirror_drift():
]
# ③ 有备份时**能**捞回:造一次"先 sync(留备份)→ 再删 → 再 sync"
(fx_cp2 / "roots.env").write_text("DSH_WS_ROOT=/fixture/BACKUPED\n", encoding="utf-8")
(fx_cp2 / "env").mkdir(parents=True, exist_ok=True)
(fx_cp2 / "env" / "roots.env").write_text("DSH_WS_ROOT=/fixture/BACKUPED\n", encoding="utf-8")
(fx_cp2 / "sub" / "b.py").write_text("B=1\n# DIFF\n", encoding="utf-8") # 制造漂移 ⇒ 触发备份
wm.do_sync(fx_src2, fx_cp2) # ← 这次备份里有 roots.env
(fx_cp2 / "roots.env").unlink()
wm.do_sync(fx_src2, fx_cp2) # ← 这次备份里有 env/roots.env
(fx_cp2 / "env" / "roots.env").unlink()
rc_bk = wm.do_sync(fx_src2, fx_cp2)
det += [
("🔴 **有备份时,白名单被删后 `--sync` 能捞回**"
"(`roots.env` 丢失 ⇒ 副本按位置推导去读写**别的仓库**,静默且难查)",
rc_bk == 0 and (fx_cp2 / "roots.env").exists()
and "/fixture/BACKUPED" in (fx_cp2 / "roots.env").read_text(encoding="utf-8")),
"(`env/roots.env` 丢失 ⇒ 副本按位置推导去读写**别的仓库**,静默且难查)",
rc_bk == 0 and (fx_cp2 / "env" / "roots.env").exists()
and "/fixture/BACKUPED" in (fx_cp2 / "env" / "roots.env").read_text(encoding="utf-8")),
]
shutil.rmtree(str(fx_src2.parent), ignore_errors=True)
shutil.rmtree(str(fx_src.parent), ignore_errors=True)
return det
@case("副本同步:`roots.env` 白名单**只按文件名**判(⛔ 换副本不许靠「我记得跳过哪个」)")
@case("副本同步:`env/roots.env` 白名单**只按路径**判(⛔ 换副本不许靠「我记得跳过哪个」)")
def t_mirror_keep_by_name():
import importlib.util
spec = importlib.util.spec_from_file_location("wm_selftest2", HERE / "workspace_mirror.py")
wm = importlib.util.module_from_spec(spec)
spec.loader.exec_module(wm)
return [
("白名单含 `roots.env`(⛔ 覆盖它 ⇒ 副本读写到别的仓库)", "roots.env" in wm.KEEP),
("白名单含 `env/roots.env`(⛔ 覆盖它 ⇒ 副本读写到别的仓库)", "env/roots.env" in wm.KEEP),
("白名单含 `副本使用说明.md`(副本独有的口径文档)", "副本使用说明.md" in wm.KEEP),
("白名单含 `install.log`(副本自己的安装历史)", "install.log" in wm.KEEP),
("白名单是**元组**(⛔ 散着写容易漏)", isinstance(wm.KEEP, tuple)),
@@ -41,21 +41,25 @@ import sqlite3
import sys
import time
# ── 根目录外置(roots.env 由 install.py 生成;宿主 env 优先,本文件兜底) ──────
# ── 环境配置外置(由 install.py 生成;宿主 env 优先,本文件兜底)
# 🔴 2026-10-11:「脚本是脚本 配置是配置」⇒ 配置落本包 `env/`。
# 首选 `env/roots.env`;并兼容旧的包根 `roots.env`(拷贝未同步时照旧能跑)。 ──────
def _sm_load_roots():
here = os.path.dirname(os.path.abspath(__file__))
for up in range(4):
p = os.path.normpath(os.path.join(here, *([".."] * up), "roots.env"))
if os.path.isfile(p):
try:
for ln in io.open(p, encoding="utf-8"):
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
base = os.path.normpath(os.path.join(here, *([".."] * up)))
for rel in (("env", "roots.env"), ("roots.env",)):
p = os.path.join(base, *rel)
if os.path.isfile(p):
try:
for ln in io.open(p, encoding="utf-8"):
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()
ARGV = sys.argv[1:]
+9 -3
View File
@@ -20,13 +20,19 @@
(快照/代码版本互相"打架")。此处按端口逐个探 `/healthz` 签名,把**所有**实例列出来并停掉。
"""
from __future__ import annotations
# [session-mechanism] roots.env 外置(由 install.py 生成;缺失则回落到按位置推导)
# [session-mechanism] roots.env 外置(由 install.py 生成 ⇒ `<包>/env/roots.env`)
# 查找顺序:向上 4 层 × (env/roots.env 优先 → 包根 roots.env 兼容) → 技能库那份。
def _sm_load_roots():
import os as _os
_here = _os.path.dirname(_os.path.abspath(__file__))
_cands = []
for _up in range(4):
_p = _os.path.join(_here, *([".."] * _up), "roots.env")
_p = _os.path.normpath(_p)
_base = _os.path.normpath(_os.path.join(_here, *([".."] * _up)))
for _rel in (("env", "roots.env"), ("roots.env",)):
_cands.append(_os.path.join(_base, *_rel))
_conf = _os.environ.get("CODEBUDDY_CONFIG_DIR") or _os.path.expanduser("~/.workbuddy")
_cands.append(_os.path.join(_conf, "skills", "session-mechanism", "env", "roots.env"))
for _p in _cands:
if _os.path.isfile(_p):
try:
with open(_p, encoding="utf-8") as _f:
@@ -25,7 +25,9 @@ CB = os.path.join(HERE, "collabd.py")
CFG = os.path.join(HERE, "collabd.config.json")
# 🔴 活动环境目录(宿主库所在)—— 这是**旧看守里有、新形态里被我漏掉**的那一条。
os.environ.setdefault("CODEBUDDY_CONFIG_DIR", r"E:\ProgramData\.workbuddy")
# 🔴 2026-10-11:默认值改成 `~/.workbuddy` 形态(⛔ 不写死本机盘符)——
# 跟着**当前用户**走;宿主 env 里已有的值优先(setdefault)。
os.environ.setdefault("CODEBUDDY_CONFIG_DIR", os.path.expanduser("~/.workbuddy"))
os.environ["COLLABD_CONFIG"] = CFG
os.environ["PYTHONIOENCODING"] = "utf-8"
os.environ["PYTHONUNBUFFERED"] = "1"
@@ -5,7 +5,7 @@
🔴 为什么必须有这个工具(不是"又写个脚本")
· 副本是**整目录拷贝**出去的 ⇒ 全局删掉/新增的文件,副本**不知道**,会静默留着
· 副本的 `roots.env` 是**工作区专属**的 ⇒ ⛔ 不能盲目整目录覆盖(会把工作区指向冲回别人)
· 副本的 `env/roots.env` 是**工作区专属**的 ⇒ ⛔ 不能盲目整目录覆盖(会把工作区指向冲回别人)
· 于是每次同步都变成"人肉算差异+人肉挑例外" ⇒ **迟早漏**(实测已漏过:mut_run.py 差点没补上)
⇒ 正确解法:**让"例外的例外"由工具自己知道**,并且**能自证没漂移**。
@@ -18,7 +18,7 @@
python workspace_mirror.py --list-keep
🔴 三条白名单(**有意保留副本自己的东西,不是残留**)
1. `roots.env` —— 工作区专属键。被覆盖 ⇒ 副本会去读写**另一个仓库**
1. `env/roots.env` —— 工作区专属键(2026-10-11 起落本包 `env/`)。被覆盖 ⇒ 副本会去读写**另一个仓库**
(`setdefault` 兜底:宿主 env 没设时注入本文件的值)
2. `副本使用说明.md` —— 副本独有的口径文档
3. `install.log` —— 副本自己的安装历史(追加式)
@@ -50,7 +50,7 @@ SELF_SUFFIX = (".pyc", ".pyo", ".tmp", ".swp", ".mirtmp")
# 而 SKILL.md 应当**跟着全局走**(它是版本的一部分)⇒ 只需在残留判定里豁免。
IDENT_MARK = "SKILL.md"
NEVER_DELETE = (IDENT_MARK,) # ⛔ 算残留时跳过 ⇒ 删了副本就废了
KEEP = ("roots.env", "副本使用说明.md", "install.log")
KEEP = ("env/roots.env", "副本使用说明.md", "install.log")
# 备份目录名前缀(`_find_in_prior_backups` 靠它认出"这是历史备份" ⇒ ⛔ 改名要同步改那里)
BK_PREFIX = "session-mechanism-副本备份-"
@@ -104,7 +104,7 @@ def _find_in_prior_backups(keep_lost, bk_root: Path):
原实现把**本次刚建的备份**当兜底源 ⇒ 但 `copytree` 发生在**丢失之后**
⇒ 那份备份里**必然也没有**这个文件 ⇒ 兜底永远捞不到。
⇒ 正解=翻**先前**留下的备份,按时间**从新到旧**找第一份命中的。
⛔⛔ **绝不拿全局那份顶替** —— 全局 `roots.env` 指向**别的仓库**,
⛔⛔ **绝不拿全局那份顶替** —— 全局 `env/roots.env` 指向**别的仓库**,
填进去 = 把"副本悄悄指错仓库"变成"副本明确指错仓库",⛔ 更坏。
"""
found = {}
@@ -196,10 +196,10 @@ def do_sync(gpkg: Path, copy_dir: Path) -> int:
print("⛔ %s 不像技能副本(没有 SKILL.md)⇒ 不动" % copy_dir)
return 2
# 🔴 2026-10-04 修:白名单**缺失**也算漂移。
# `diff()` 刻意排除 KEEP(免得每次都报它们)⇒ 副本的 `roots.env` 没了
# `diff()` 刻意排除 KEEP(免得每次都报它们)⇒ 副本的 `env/roots.env` 没了
# ⇒ 差异表里**一个字段都不动** ⇒ 早先直接走"无漂移,幂等返回",
# 白名单兜底层**永远没机会跑**。而这恰恰是最该救的场景:
# `roots.env` 丢了 ⇒ 副本按位置推导 ⇒ 去读写**别的仓库**(静默指错)。
# `env/roots.env` 丢了 ⇒ 副本按位置推导 ⇒ 去读写**别的仓库**(静默指错)。
# ⇒ 这里单独查一遍。
# 🔴🔴 修第二版:「**全局侧有**」⛔ 不等于「副本**该**有」。
# 实测踩到:全新副本本来就没有 `install.log`(从未安装过)⇒ 却被判成
@@ -217,8 +217,8 @@ def do_sync(gpkg: Path, copy_dir: Path) -> int:
_, _, only_c, only_g, changed = diff(gpkg, copy_dir)
if keep_lost:
print("⛔ **白名单缺失**(副本历史里有过,现在没了):%s" % "、".join(keep_lost)
+ ("(⛔ `roots.env` 丢 ⇒ 副本会按位置推导去读写**别的仓库**)"
if "roots.env" in keep_lost else ""))
+ ("(⛔ `env/roots.env` 丢 ⇒ 副本会按位置推导去读写**别的仓库**)"
if "env/roots.env" in keep_lost else ""))
if not (only_c or only_g or changed or keep_lost):
print("✅ 无漂移,⛔ 什么也不做(幂等)")
return 0