#!/usr/bin/env python """🔴 **新建协作工作区**(技能内脚本,用户 2026-10-03 要求"每次使用任务会话机制都可以有对应的目标文件夹")。 用法: python init_workspace.py <工作区绝对路径> [--title 目标标题] [--short 目标简称] [--port 单例端口] [--no-main-session] 做四件事(⛔ 顺序固定,缺一件机制就跑不起来): ① 建工作区骨架(`.workbuddy/collab/`、`tmp/supervise-inbox/`、`交付物/`) ② 写**部署配置** `collabd.config.json`(`COLLABD_CONFIG` 指向它 ⇒ `collabd.py` 才认这个工作区) ③ 写**目标** `goal.json`(title/short/topics/lifecycle=进行中/taskgraph) ③b 写**最小任务图**(⛔ 漏了它 ⇒ 常驻每轮报「读任务图失败」,且目标判据缺一条腿) ④ 跑 `collabd.py --ensure-goal-dir` ⇒ **建目标文件夹 + 落《目标执行状态.md》骨架**(幂等) ⑤ 建一条**主会话排期**(`automation_update` 由调用方做 ⇒ 本脚本只**打印**它该用的参数, ⛔ 不自己写宿主库 —— 创建新会话**只能走自动化入口**,这是宿主规矩) 🔴 关键约束(都是踩过的): · `workspace` 必须**正斜杠**(⛔ 反斜杠会让 `cwds` 裂组) · `singleton_port` **每个工作区一个**(⛔ 复用 ⇒ 常驻单例互抢、静默只有一个活着) · `board_ext` 指 `.RETIRED-*` ⇒ 新工作区**不带**使用方扩展(技能侧零项目串) · 排期名必须是两级前缀 `[主]-<类别>-<具体>`(⛔ 光杆名 ⇒ 角色判不出,看板认不出) """ import argparse import json import os import subprocess import sys from pathlib import Path sys.stdout.reconfigure(encoding="utf-8", errors="replace") SK = Path(__file__).resolve().parent DEFAULT_TITLE = "检查会话协作是否运行正常 + 协作机制问题排查与修复" def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("workspace") ap.add_argument("--title", default=DEFAULT_TITLE) ap.add_argument("--short", default="") ap.add_argument("--topics", default="会话协作自检,机制排查与修复") ap.add_argument("--port", type=int, default=0) ap.add_argument("--python", default=sys.executable) # ⛔ 2026-10-06 删掉 `--no-task` / `--with-keeper` 两个开关: # 它们的存在前提是"本脚本会铺 ps1 并登记计划任务",而那条形态**已整套废弃**(P0-74)。 # ⇒ 常驻的开/关/自愈统一走 `collabctl.py on|off|ensure`,⛔ 本脚本不再碰任务。 a = ap.parse_args() ws = Path(a.workspace).resolve() ws_s = str(ws).replace("\\", "/") # 🔴 正斜杠(红线) short = a.short or ws.name topics = [t.strip() for t in a.topics.split(",") if t.strip()] # 🔴🔴 端口**跨进程稳定**(2026-10-04 修):原来用 `hash(ws_s)`,而 # **Python 3.11+ 字符串 hash 每进程随机化**(PYTHONHASHSEED)⇒ # **同一工作区每跑一次就换一个端口** ⇒ 旧守卫/防火墙/别的配置里的端口全部对不上。 # ⇒ 改成 `zlib.crc32`(确定性、与版本无关);已有配置里写了端口就沿用它。 import zlib derived = 20000 + (zlib.crc32(ws_s.encode("utf-8")) % 5000) # 按**工作区路径**算,天然各区不同 port = a.port or derived _port_from = "按路径算出" if not a.port else "命令行指定" print("=== 新建协作工作区:%s ===" % ws_s) # ⚠️ 端口目录靠后生成(要读旧配置才能定),但**冲突自检**要提前做 —— 见下方 `_port_clash`。 _port_clash: list[str] = [] def _scan_peer_ports() -> dict: """扫**本机其它工作区**的 `singleton_port`(只读)。 🔴🔴 2026-10-05 加(用户报「各工作区环境是否该独立」时坐实): 实测三个区(`ai1net` / `vibe-product` / `agent-product`)的 `collabd.config.json` **`singleton_port` 全是 20099**(= `collabd.py` 的**默认值**)。 真因不是算法 —— 算法按路径算,三区分别该是 `23924` / `24942` / `21826`; 真因是**前两个区的 config 是早期手工建的,一直抄着默认值**,而 `init_workspace` 的「沿用既有端口」**无条件信任**旧值 ⇒ 抄错的值**永远修不回来**。 ⚠️ 后果(`init_workspace.py` 文件头早就写了): 「`singleton_port` 每个工作区一个(⛔ 复用 ⇒ 常驻单例互抢、**静默只有一个活着**)」 ⇒ 这是**静默故障**:两个区都以为自己起了常驻,实际只有一个绑上端口。 做法:扫 `AIProject` 下各兄弟工作区的配置(本区**不算**),拿端口 ⇒ 撞了就报警并改用算出来的值。 """ out: dict = {} try: parent = ws.parent # 通常= …/AIProject if not parent.is_dir(): return out for sib in parent.iterdir(): if not sib.is_dir() or sib.resolve() == ws: continue cp = sib / ".workbuddy" / "collab" / "collabd.config.json" if not cp.is_file(): continue try: pv = int((json.loads(cp.read_text(encoding="utf-8")) or {}).get( "singleton_port") or 0) except Exception: continue if pv: out.setdefault(pv, []).append(sib.name) except Exception: pass return out # ① 骨架 for d in (".workbuddy/collab", "tmp/supervise-inbox", "交付物", ".workbuddy/collab/logs"): (ws / d).mkdir(parents=True, exist_ok=True) print("① 骨架已建(.workbuddy/collab / tmp/supervise-inbox / 交付物 / logs)") cfgp = ws / ".workbuddy/collab/collabd.config.json" gp = ws / "tmp/supervise-inbox/goal.json" # ══════════════════════════════════════════════════════════════════ # 🔴🔴 2026-10-04 改为**幂等自配置**(用户:「我要求在任何工作区使用都能 # **自己检查环境配置好环境**」)。原来本脚本**无条件重写** `collabd.config.json` # 与 `goal.json` ⇒ 对**已部署且有真实内容**的工作区是**破坏性**的 # (实测 `vibe-product` 的 `targets` 里登记着 5 个真实目标 ⇒ 会被冲掉)。 # ⇒ 现在:**先读现状 → 只补缺的 → 已有的一律不动**。 _old_cfg = {} if cfgp.exists(): try: _old_cfg = json.loads(cfgp.read_text(encoding="utf-8")) except Exception: _old_cfg = {} # 🔴🔴 端口冲突自检 + 纠偏(2026-10-05 加)—— # 「沿用既有」**不能无条件信任**:实测三区 config 里 `singleton_port` 全是 20099 # (早期手工建配置时抄了 `collabd.py` 的默认值),而**谁都没报错** # ⇒ 常驻单例互抢、静默只有一个活着。⇒ 沿用前**先比对**: # · 旧值 == 按路径算出的值 ⇒ 正常沿用; # · 旧值 ≠ 算出值,且**别区占了它** ⇒ 🔴 报警 + **改用算出值**(抄错的值该修); # · 旧值 ≠ 算出值,但没有别区占用 ⇒ **沿用**(可能是刻意指定的,⛔ 不擅改)。 _bad_port = False if _old_cfg.get("singleton_port"): _old_port = int(_old_cfg["singleton_port"]) _peers = _scan_peer_ports() if _old_port == derived: port, _port_from = _old_port, "沿用既有(与算出值一致)" else: _who = _scan_peer_ports().get(_old_port) or [] if _who: _bad_port = True print(" 🔴 **端口冲突**:本区既有 `singleton_port=%d` 已被 **%s** 占用。" % (_old_port, "、".join(_who))) print(" 🔴 后果:两区常驻**互抢单例** ⇒ ⛔ 静默只有一个活着(文件头 §关键约束 早写明)。") print(" 🔴 处置:改用按本区路径算出的 **%d**(=%s)。" % (derived, ws_s)) port, _port_from = derived, "纠偏(原值与他区冲突)" _port_clash.append("%d ← %s" % (_old_port, "、".join(_who))) else: port, _port_from = _old_port, "沿用既有" _old_goal = {} if gp.exists(): try: _old_goal = json.loads(gp.read_text(encoding="utf-8")) except Exception: _old_goal = {} _reuse = bool(_old_cfg) or bool(_old_goal) if _reuse: print(" ♻️ **检测到既有部署**(配置 %s / 目标 %s)⇒ **只补缺,不覆盖已有**" % ("有" if _old_cfg else "⛔无", "有" if _old_goal else "⛔无")) # ② 部署配置(**补缺式**:以旧配置为底,只加现行键;⛔ 不动用户改过的值) cfg = { "_说明": "本文件=**本工作区**的协作机制部署配置(⛔ 不属于技能)。由 `init_workspace.py` 生成;" "`COLLABD_CONFIG` 必须指向它,否则 `collabd.py` 不认这个工作区。", "workspace": ws_s, "inbox": "tmp/supervise-inbox", "live": "交付物/本机协作-实时状态.md", "taskgraph": "交付物/任务图-会话协作自检.json", "targets": {}, "singleton_port": port, "interval": 10, "idle_min": 12, "stuck_min": 30, "vacuum_min": 5, "health_every": 300, "supervise_interval": 10, "wake_enable": False, # 🔴 唤醒/投递整套已退役 "board_ext": ".workbuddy/collab/board_ext.RETIRED.py", # ⛔ 不带使用方扩展 "log": ".workbuddy/collab/logs/_collabd.log", } if _old_cfg: _merged = dict(cfg) _merged.update(_old_cfg) # ⛔ 旧的**优先**(含 targets / 自定义路径) cfg = _merged _added = [k for k in cfg if k not in _old_cfg] cfg["workspace"] = ws_s # 这两项必须以本次为准(工作区可能搬过家) cfg["singleton_port"] = port # ⚠️ 纠偏时**必须以本次为准**(旧值可能在 `_old_cfg` 里,会被 update 盖回去) cfgp.write_text(json.dumps(cfg, ensure_ascii=False, indent=1), encoding="utf-8", newline="") print("② 部署配置:%s%s(端口 %d,%s)" % (cfgp.relative_to(ws), (" 补了 %s" % _added) if _added else " 已最新", port, _port_from)) # ③ 目标(**补缺式**:已有 goal.json ⛔ 一个字都不改) goal = { "title": a.title, "short": short, "topics": topics, "lifecycle": "进行中", "lifecycle_at": "", "lifecycle_by": "", "lifecycle_why": "init_workspace.py 新建", "taskgraph": "交付物/任务图-会话协作自检.json", "taskgraph_set_at": "", "taskgraph_set_by": "", "declared_at": "", "declared_by": "", "acceptance_state": {}, "_唯一来源": "本文件 title/short/topics 是真源;`execution_doc` 指向目标文件夹里的《目标执行状态.md》。", } if _old_goal: print("③ 目标文件:**已存在,⛔ 不覆盖**(title=%r)" % str(_old_goal.get("title"))[:40]) else: gp.write_text(json.dumps(goal, ensure_ascii=False, indent=1), encoding="utf-8", newline="") print("③ 目标已写:title=%r short=%r lifecycle=进行中" % (a.title, short)) # ③b 任务图(🔴 2026-10-03 14:35x 补):配置里 `taskgraph` 指向它, # 但初始化时**没建这个文件** ⇒ 常驻每 10 秒报「goal_state 读任务图失败」 # ⇒ 更要命的是**判据第二路(任务图 nodes)永远取不到** ⇒ 目标状态判据缺一条腿。 # ⇒ 写一份最小任务图:节点全 `pending`,由任务会话逐个转 `done`。 # 🔴 2026-10-04 **同样改成补缺式**(⛔ 别覆盖既有任务图 —— 那是任务会话逐个转 done 的**进度真源**, # 冲掉 = 全部工作量白费;实测 `vibe-product` 的 `taskgraph` 指向 `交付物/任务图.json`, # 与本脚本的默认路径**不同名** ⇒ 更不能碰)。 tg_rel = (_old_cfg.get("taskgraph") or "交付物/任务图-会话协作自检.json") tgp = ws / tg_rel if tgp.exists(): try: _n = len((json.loads(tgp.read_text(encoding="utf-8")) or {}).get("nodes") or []) except Exception: _n = "?" print("③b 任务图:**已存在,⛔ 不覆盖**(%s,%s 个节点)" % (tg_rel, _n)) else: # 🔴🔴 2026-10-04 修:旧配置**可能没有 `taskgraph` 键**(实测 `vibe-product` 就没有) # ⇒ 落回默认名"任务图-会话协作自检.json",而用户真实的图叫**别的名字**(如 `任务图.json`) # ⇒ **新建了一份空图,真实的图被"忽略"** ⇒ 判据第二路(任务图 nodes)仍取不到, # 而用户以为"初始化过了"。⇒ 先在 `交付物/` 里找一张**看起来是任务图的 json** 再决定。 _cand = [] try: for f in sorted((ws / "交付物").glob("*.json")): try: j = json.loads(f.read_text(encoding="utf-8")) except Exception: continue if isinstance(j, dict) and isinstance(j.get("nodes"), list): _cand.append(f) except Exception: pass if _cand: # ⚠️ 有多张 ⇒ **不猜**,只报告并**不新建**(⛔ 猜错等于改用户的目标进度) if len(_cand) == 1: print("③b 任务图:配置里没登记 `taskgraph`,但找到既有任务图 **%s** ⇒ ⛔ 不新建、不改它" % _cand[0].name) tg_rel = None # 🔴🔴 关键:**置 None 才不会落到下面的写入分支** else: print("③b ⚠️ 配置里没登记 `taskgraph`,且 `交付物/` 里有 **%d 张**像任务图的 json(%s)" " ⇒ ⛔ 不猜、**不新建**。请在配置里登记 `taskgraph` 指明哪一张。" % (len(_cand), "、".join(f.name for f in _cand[:3]))) tg_rel = None if tg_rel: tg = {"title": a.title, "short": short, "topics": topics, "nodes": [{"id": "N1", "title": "核对目标并完成首轮协作", "status": "pending", "line": topics[0] if topics else "", "note": "init_workspace.py 生成的最小任务图"}]} tgp.parent.mkdir(parents=True, exist_ok=True) tgp.write_text(json.dumps(tg, ensure_ascii=False, indent=1), encoding="utf-8", newline="") print("③b 任务图已写:%s(1 个节点,pending)" % tg_rel) # ④ 目标文件夹(走产品入口,⛔ 不自己 mkdir 文档) # 🔴🔴 2026-10-03 **改用「本工作区的副本」,⛔ 不再用技能目录那份**: # 定案口径=**各工作区的程序各自独立**(技能目录=源,`.workbuddy/collab/`=各区副本, # `deploy_code.py` 分发)⇒ 提示用户去跑技能目录那份,等于**又拉起第二个常驻**, # 它会写**技能目录的 tmp/**(⛔ 那不是任何工作区的 inbox)⇒ 状态写到别处, # 看板读不到、故障排查也被带偏(本轮 `SESSION_END`-style 误判的同族)。 # ⚠️ 本副本若还不存在(首次初始化、还没分发)⇒ 回落用技能目录那份, # 但**打印时必须说清"这次是回落"**(⛔ 静默回落=用户以为跑的是副本)。 _own = ws / ".workbuddy" / "collab" / "collabd.py" _fallback = not _own.is_file() _cd = SK / "collabd.py" if _fallback else _own if _fallback: print(" ⚠️ 本工作区还没有 collabd.py 副本 ⇒ 本次**回落用技能目录那份**;" "请随后跑 deploy_code.py 分发,否则各区程序不独立。") env = dict(os.environ) env["COLLABD_CONFIG"] = str(cfgp).replace("\\", "/") env["COLLABD_NO_ENSURE"] = "1" r = subprocess.run([a.python, str(_cd), "--ensure-goal-dir"], env=env, cwd=str(ws), capture_output=True, timeout=120) out = r.stdout.decode("utf-8", "replace").strip() print("④ --ensure-goal-dir rc=%s" % r.returncode) for ln in out.splitlines(): print(" " + ln) gd = [ln for ln in out.splitlines() if ln.startswith("目标目录:")] gdir = gd[0].split(":", 1)[1].strip() if gd else "" # ⑤ 主会话排期参数(⛔ 本脚本不建 —— 创建新会话只能走自动化入口) print() print("⑤ 主会话排期(请由调用方用 automation_update 建):") print(" name = [主]-%s-主会话" % short) print(" cwds = %s" % ws_s) print(" 提示:主会话标题必须是**两级前缀**(`[主]-<类别>-<具体>`),⛔ 光杆名看板认不出。") print() print("✅ 完成。后续协作(⛔ **都跑本工作区的副本**,别跑技能目录那份 —— " "各区程序各自独立,跑了目录那份会写错地方):") print(" export COLLABD_CONFIG=%s" % str(cfgp).replace("\\", "/")) _cd_show = _cd print(" python %s --supervise" % _cd_show) # 🔴🔴 2026-10-05 改口径(用户当轮原话:「看板共用一份 各个工作区不是都应该有自己独立的看板」—— # 这是在**问**,答案是否定的,且**早有定案**:SKILL.md:809 记的 2026-10-03 16:3x 用户拍板 # 「⛔ 不再为每个工作区各起一个看板 —— **看板只保留一份**(就是主工作区这一个), # 其它工作区靠 `peer_workspaces` 并列查看」。 # ⇒ 老输出写的是「python board.py --serve 8789 --takeover」(**暗示各区各起一份**)—— # 与定案冲突,且 8789 这个端口本身也是随便举的 ⇒ 会把新区的常驻引导到**错形态**上。 # ⇒ 现在**只说本区要做什么**(起常驻),看板那件事改为一句**指向主视图**的说明。 # ⚠️ 判据:本区**只起常驻**,⛔ 不起看板;要并看 ⇒ 把本区路径加进**主工作区**配置的 # `peer_workspaces`(只读),重启主看板即可。 print(" 📺 看板:**全平台只保留一份**(主工作区那份)⇒ 本区⛔ 不起看板。") print(" 要并看本区 ⇒ 把 `%s` 加进**主工作区**配置的 `peer_workspaces`(只读),重启主看板。" % ws_s) if _fallback: print(" ⚠️ 上面是**技能目录那份**(副本还没分发)⇒ 先跑 deploy_code.py --ws %s" % ws_s) # ══ ⑥ 🔴🔴 **常驻载体**(2026-10-04:用户「任何工作区都要能自己配好」) # 🔴🔴 2026-10-06 收敛(依据:pitfalls §P0-74 的定案三条 + SKILL.md「两条起法必须收敛」): # ⛔ **不再铺 `start-supervise.ps1`**、⛔ **不再做 ps1 语法校验**、 # ⛔ **不再依赖 `assets/start-supervise.ps1.tpl`**。 # 旧形态=铺 ps1 + 任务动作 `powershell.exe -File start-supervise.ps1` # ⇒ PowerShell 是**控制台程序** ⇒ 每次触发分配 `conhost.exe` ⇒ **闪黑窗** # (用户 2026-10-05「又弹了窗口」;实测抓到 `powershell.exe` + `conhost.exe` 这对进程)。 # 新形态=**计划任务 → `pythonw.exe` → `supervise-launch.py`**(GUI 子系统 ⇒ 零控制台)。 # ⚠️ env(`CODEBUDDY_CONFIG_DIR` / `COLLABD_CONFIG`)由**启动器在进程内设** —— # 这正是 `supervise-launch.py` 存在的全部理由(旧注释"直接跑解释器没法设 env"已被它推翻)。 # ⛔ **建任务这件事只许有一处实现** = `collabctl.py`(常驻唯一管理入口,受 `supervise.switch` # 总电闸管辖)。本脚本**不建任务** —— 自己建就绕过了开关,是"第二条起法"。 _cdir = ws / ".workbuddy" / "collab" _launcher = _cdir / "supervise-launch.py" if _launcher.is_file(): print("⑥ 常驻载体:✅ 启动器在位(`%s`)" % _launcher.relative_to(ws)) else: print("⑥ 常驻载体:⛔ 缺启动器 `supervise-launch.py`") print(" 📌 先跑 `deploy_code.py --ws %s` 分发副本。" % ws_s) print(" 📌 常驻的开/关/自愈**一律走唯一入口**:`collabctl.py on|off|ensure`") print(" (⛔ 本脚本不建计划任务 —— 那会绕过 `supervise.switch` 总电闸,是「第二条起法」)") # ⑤ 自检收口放在⑥ 之后(⛔ 顺序:先铺载体再体检,否则体检会漏掉"载体有没有") print() print("⑤ 自检(问 `_env.env_check`,⛔ 不自己另算一套):") try: import importlib.util as _iu _sp = _iu.spec_from_file_location("_env_chk", SK / "hooks" / "_env.py") _e = _iu.module_from_spec(_sp) _sp.loader.exec_module(_e) _rep = _e.env_check(ws=ws_s, scope="workspace") _dp = _rep.get("deploy") or {} print(" 技能根 = %s" % _rep.get("skills_root")) print(" 已部署 = %s" % _dp.get("deployed")) for _m in (_dp.get("missing") or []): print(" ⛔ %s" % _m) for _m in (_dp.get("stale") or []): print(" ⚠️ %s" % _m) # 🔴 真的按程序判一次「能不能认这个工作区」——⛔ 不用"文件在不在"当通过 _r2 = subprocess.run([a.python, str(_cd), "--where"], env=env, cwd=str(ws), capture_output=True, timeout=120) _o2 = _r2.stdout.decode("utf-8", "replace") _ws_ok = ("COLLABD_CONFIG" in _o2) or (ws_s in _o2) print(" `collabd --where` 认这个工作区 = %s(rc=%s)" % ("✅" if _ws_ok else "⛔", _r2.returncode)) # 🔴🔴 2026-10-06 收敛(P0-74):载体从 `start-supervise.ps1` 换成 **`supervise-launch.py`**。 # ⇒ 自检一并改:**不再验 ps1、不再验 UTF-8 BOM、不再验模板变量填完没** —— # 那三个判据的**被测对象已经不存在了**(留着就是过期的探照灯)。 # 判据换成**行为级可核的一条**:启动器在位。 # ⚠️ 任务本身**不在这里验** —— 建任务归 `collabctl.py`(唯一实现), # 本脚本自己建/自己验就绕过了 `supervise.switch` 总电闸。 _lc = ws / ".workbuddy" / "collab" / "supervise-launch.py" if _lc.is_file(): print(" 常驻启动器 = ✅ 在位(`supervise-launch.py`)") else: print(" 常驻启动器 = ⛔ **缺**(先跑 `deploy_code.py --ws %s` 分发副本)" % ws_s) print(" 📌 常驻开/关/自愈走唯一入口:`collabctl.py on|off|ensure`" "(本区任务名 `collabd-keepalive-%s`)" % (ws.name or "ws")) if _dp.get("deployed") and _ws_ok and not (_dp.get("missing") or []) : print(" ✅ **本工作区可以直接用**(配置/目标/任务图齐了,程序也认)") else: print(" ⛔ **还没到位**,按上面 ⛔/⚠️ 逐条补完再跑一次本脚本(幂等,可反复跑)") except Exception as _ex: print(" ⚠️ 自检跑不了:%r(⛔ 不当通过)" % (_ex,)) return 0 if __name__ == "__main__": sys.exit(main())