weichen
c170939bb7
Why: approved design (2026-10-12) - persistent manifest becomes the startup authority, closing the param-rebuild drift, non-atomic per-key writes, and missing-config-creates-empty-system recovery gaps (plan r2 U1 / KTD1-KTD2). Evidence: SELF - python3 tests/test_svc_identity.py 21 tests OK (1 skip = real-PG identity NOT_RUN by design); make test-governance 169 OK incl. existing svc db-binding and autorecover suites; receipts appended via make evidence. Impact: scripts/svc.sh gains manifest-aware set/del/record_state/autostart plus lock-held migrate, pre-launch identity verify and an init action; new scripts/svc_manifest.py (atomic temp-fsync-rename, identity anchors, derived svc.env); unmigrated instances keep the exact legacy path (regressions green). Refs: docs/recovery/u1-identity-manifest.md; plan r2 section 7 U1
322 lines
12 KiB
Python
322 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""svc 实例身份 manifest（U1）：manifest.json 为启动权威，svc.env 为派生只读视图。
|
||
|
||
设计合同：docs/recovery/u1-identity-manifest.md（已批准 2026-10-12）。
|
||
落地说明：v1.0 以「identity 锚点（数据根/数据库身份，冻结）+ kv 权威键值映射 +
|
||
desired 期望态」承载 KTD1/KTD2 的全部不变量；完整 typed 分区随 U2/U6 演进。
|
||
|
||
用法: svc_manifest.py ENV_DIR <migrate|init|status|set|del|bulk-set|set-desired|verify> [参数]
|
||
|
||
不变量（对应设计 §1）：
|
||
I3 原子性 —— 整份 temp→fsync→rename→dirsync，任意时刻只有完整旧版或新版。
|
||
I4 阻塞 —— manifest 存在时，数据 marker 缺失/身份不符即拒绝，不新建空系统。
|
||
I5 机密 —— 本文件不新增任何机密；kv 沿袭 svc.env 既有内容（DSN 迁出属 U6）。
|
||
"""
|
||
import json
|
||
import os
|
||
import secrets
|
||
import sys
|
||
import time
|
||
|
||
SCHEMA = 1
|
||
MANIFEST = "manifest.json"
|
||
DERIVED = "svc.env"
|
||
MARKER_REL = os.path.join(".agentx", "data-marker.json")
|
||
PROTECTED = ("db", "dbref", "dsn", "conffile") # 显式持久化写入可重锚；启动期漂移由 verify 拦截
|
||
|
||
|
||
def die(msg, code=1):
|
||
print("[manifest] %s" % msg, file=sys.stderr)
|
||
sys.exit(code)
|
||
|
||
|
||
def now_iso():
|
||
return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
|
||
|
||
|
||
def atomic_write(path, text, mode=0o600):
|
||
d = os.path.dirname(path) or "."
|
||
os.makedirs(d, exist_ok=True)
|
||
tmp = path + ".new"
|
||
with open(tmp, "w", encoding="utf-8") as f:
|
||
f.write(text)
|
||
f.flush()
|
||
os.fsync(f.fileno())
|
||
os.chmod(tmp, mode)
|
||
os.replace(tmp, path)
|
||
dfd = os.open(d, os.O_DIRECTORY)
|
||
try:
|
||
os.fsync(dfd)
|
||
finally:
|
||
os.close(dfd)
|
||
|
||
|
||
def read_kv(env_dir):
|
||
kv = {}
|
||
path = os.path.join(env_dir, DERIVED)
|
||
if os.path.exists(path):
|
||
with open(path, encoding="utf-8") as f:
|
||
for line in f:
|
||
line = line.rstrip("\n")
|
||
if line and "=" in line and not line.startswith("#"):
|
||
k, v = line.split("=", 1)
|
||
kv[k] = v
|
||
return kv
|
||
|
||
|
||
def load(env_dir):
|
||
path = os.path.join(env_dir, MANIFEST)
|
||
if not os.path.exists(path):
|
||
return None
|
||
try:
|
||
with open(path, encoding="utf-8") as f:
|
||
m = json.load(f)
|
||
except Exception as e: # noqa: BLE001 —— 损坏必须阻塞，不许猜测重建
|
||
die("manifest 损坏（拒绝猜测，人工处置）：%s: %s" % (path, e), 2)
|
||
if m.get("schema") != SCHEMA:
|
||
die("manifest schema 不受支持：%s（当前 %s）" % (m.get("schema"), SCHEMA), 2)
|
||
for field in ("revision", "instance_id", "identity", "kv", "desired"):
|
||
if field not in m:
|
||
die("manifest 缺少必需字段 %s（拒绝猜测）" % field, 2)
|
||
return m
|
||
|
||
|
||
def derive_text(m):
|
||
kv = dict(m["kv"])
|
||
if m["desired"] == "running":
|
||
kv["autostart"] = "1"
|
||
else:
|
||
kv.pop("autostart", None)
|
||
return "".join("%s=%s\n" % (k, v) for k, v in kv.items())
|
||
|
||
|
||
def save(env_dir, m):
|
||
atomic_write(os.path.join(env_dir, MANIFEST),
|
||
json.dumps(m, ensure_ascii=False, indent=2, sort_keys=True) + "\n")
|
||
atomic_write(os.path.join(env_dir, DERIVED), derive_text(m))
|
||
|
||
|
||
def data_root_of(kv, env_dir):
|
||
kind = kv.get("kind", "agent")
|
||
if kind == "static":
|
||
return os.path.abspath(env_dir) # static 无数据目录语义，锚定实例目录本身
|
||
parts = (kv.get("exec") or "").split()
|
||
for i, w in enumerate(parts):
|
||
if w == "--data" and i + 1 < len(parts):
|
||
return os.path.abspath(parts[i + 1])
|
||
return os.path.abspath(os.path.join(env_dir, "pgdata" if kind == "postgres" else "data"))
|
||
|
||
|
||
def db_anchor_of(kv):
|
||
mode = kv.get("db", "")
|
||
if mode == "pg":
|
||
return "pg", kv.get("dbref", "")
|
||
if mode == "sqlite":
|
||
return "sqlite", kv.get("dsn", "") or "default"
|
||
return "", ""
|
||
|
||
|
||
def marker_path(data_root):
|
||
return os.path.join(data_root, MARKER_REL)
|
||
|
||
|
||
def write_marker(data_root, instance_id):
|
||
atomic_write(marker_path(data_root), json.dumps({
|
||
"schema": 1,
|
||
"instance_id": instance_id,
|
||
"data_root": data_root,
|
||
"created_at": now_iso(),
|
||
}, ensure_ascii=False, indent=2) + "\n")
|
||
|
||
|
||
def read_marker(data_root):
|
||
path = marker_path(data_root)
|
||
if not os.path.exists(path):
|
||
return None
|
||
try:
|
||
with open(path, encoding="utf-8") as f:
|
||
return json.load(f)
|
||
except Exception: # noqa: BLE001 —— marker 损坏等同身份不可证
|
||
return {"schema": -1}
|
||
|
||
|
||
def reanchor(env_dir, m):
|
||
"""按显式持久化写入重算身份锚点；数据根变化时在新根补写 marker（目录存在才写）。"""
|
||
ident = dict(m["identity"])
|
||
ident["data_root"] = data_root_of(m["kv"], env_dir)
|
||
ident["db_mode"], ident["db_ref"] = db_anchor_of(m["kv"])
|
||
if ident != m["identity"]:
|
||
if os.path.isdir(ident["data_root"]):
|
||
write_marker(ident["data_root"], m["instance_id"])
|
||
m["identity"] = ident
|
||
return m
|
||
|
||
|
||
def build(env_dir, kv, allow_create):
|
||
kind = kv.get("kind", "agent")
|
||
name = kv.get("env") or os.path.basename(os.path.abspath(env_dir))
|
||
data_root = data_root_of(kv, env_dir)
|
||
if not os.path.isdir(data_root):
|
||
if not allow_create:
|
||
die("数据根不存在且非显式 init，拒绝建立身份（防新建空系统掩盖丢数据）：%s" % data_root, 3)
|
||
os.makedirs(data_root, exist_ok=True)
|
||
existing = read_marker(data_root)
|
||
if existing is not None:
|
||
if existing.get("instance_id") != kv.get("instance_id"):
|
||
die("数据根已携带其他实例 marker（instance_id=%s）：数据归属冲突，拒绝迁移/初始化，人工处置"
|
||
% existing.get("instance_id"), 4)
|
||
instance_id = existing["instance_id"]
|
||
else:
|
||
instance_id = kv.get("instance_id") or "svc-%s-%s" % (name, secrets.token_hex(6))
|
||
write_marker(data_root, instance_id)
|
||
db_mode, db_ref = db_anchor_of(kv)
|
||
m = {
|
||
"schema": SCHEMA,
|
||
"revision": 1,
|
||
"instance_id": instance_id,
|
||
"service_group": name,
|
||
"created_at": now_iso(),
|
||
"desired": "running" if kv.get("autostart") == "1" else "stopped",
|
||
"identity": {"data_root": data_root, "db_mode": db_mode, "db_ref": db_ref},
|
||
"kv": {k: v for k, v in kv.items() if k != "autostart"},
|
||
}
|
||
save(env_dir, m)
|
||
return m
|
||
|
||
|
||
def cmd_migrate(env_dir):
|
||
if load(env_dir) is not None:
|
||
print("manifest 已存在，幂等跳过")
|
||
return 0
|
||
if not os.path.exists(os.path.join(env_dir, DERIVED)):
|
||
die("svc.env 不存在：先完成首次登记（start）或使用 init", 3)
|
||
m = build(env_dir, read_kv(env_dir), allow_create=False)
|
||
print("已迁移 rev=1 instance_id=%s data=%s" % (m["instance_id"], m["identity"]["data_root"]))
|
||
return 0
|
||
|
||
|
||
def cmd_init(env_dir):
|
||
if load(env_dir) is not None:
|
||
die("manifest 已存在（init 仅用于显式建立；重建需人工处置）", 3)
|
||
kv = read_kv(env_dir)
|
||
m = build(env_dir, kv, allow_create=True)
|
||
print("已建立 rev=1 instance_id=%s data=%s" % (m["instance_id"], m["identity"]["data_root"]))
|
||
return 0
|
||
|
||
|
||
def cmd_status(env_dir):
|
||
m = load(env_dir)
|
||
if m is None:
|
||
print(json.dumps({"exists": False}, ensure_ascii=False))
|
||
return 0
|
||
print(json.dumps({
|
||
"exists": True, "revision": m["revision"], "instance_id": m["instance_id"],
|
||
"desired": m["desired"], "identity": m["identity"],
|
||
}, ensure_ascii=False))
|
||
return 0
|
||
|
||
|
||
def mutate(env_dir, fn):
|
||
m = load(env_dir)
|
||
if m is None:
|
||
die("manifest 不存在：先 migrate/init", 3)
|
||
before = json.dumps(m["identity"], sort_keys=True)
|
||
fn(m)
|
||
m["revision"] = int(m["revision"]) + 1
|
||
m = reanchor(env_dir, m)
|
||
save(env_dir, m)
|
||
if before != json.dumps(m["identity"], sort_keys=True):
|
||
print("[manifest] 身份锚点已随显式持久化写入更新（data=%s db=%s/%s）"
|
||
% (m["identity"]["data_root"], m["identity"]["db_mode"], m["identity"]["db_ref"]))
|
||
return 0
|
||
|
||
|
||
def cmd_set(env_dir, key, value):
|
||
return mutate(env_dir, lambda m: m["kv"].__setitem__(key, value))
|
||
|
||
|
||
def cmd_del(env_dir, key):
|
||
def drop(m):
|
||
m["kv"].pop(key, None)
|
||
return mutate(env_dir, drop)
|
||
|
||
|
||
def cmd_bulk_set(env_dir):
|
||
kv = {}
|
||
for line in sys.stdin.read().splitlines():
|
||
if line and "=" in line and not line.startswith("#"):
|
||
k, v = line.split("=", 1)
|
||
kv[k] = v
|
||
kv.pop("autostart", None) # desired 独占期望态
|
||
|
||
def replace(m):
|
||
m["kv"] = kv
|
||
return mutate(env_dir, replace)
|
||
|
||
|
||
def cmd_set_desired(env_dir, state):
|
||
if state not in ("running", "stopped"):
|
||
die("desired 取值仅 running|stopped", 2)
|
||
return mutate(env_dir, lambda m: m.__setitem__("desired", state))
|
||
|
||
|
||
def cmd_verify(env_dir, data_root=None, db_mode=None, db_ref=None):
|
||
m = load(env_dir)
|
||
if m is None:
|
||
die("manifest 缺失：本实例已进入 manifest 治理，缺失即事件（先 init/人工处置，禁止静默重建）", 4)
|
||
ident = m["identity"]
|
||
marker = read_marker(ident["data_root"])
|
||
if marker is None or marker.get("schema") == -1:
|
||
die("数据 marker 缺失/损坏：data=%s（数据归属不可证即拒绝启动，防新建空库掩盖）"
|
||
% marker_path(ident["data_root"]), 5)
|
||
if marker.get("instance_id") != m["instance_id"]:
|
||
die("数据 marker 与 manifest 身份不符：marker=%s manifest=%s（数据被换过或身份被改，人工处置）"
|
||
% (marker.get("instance_id"), m["instance_id"]), 5)
|
||
if data_root is not None and os.path.abspath(data_root) != ident["data_root"]:
|
||
die("数据根漂移：生效 %s ≠ 锚定 %s；数据根是冻结身份，换根须显式迁移（docs/recovery/u1-identity-manifest.md §5）"
|
||
% (os.path.abspath(data_root), ident["data_root"]), 6)
|
||
if db_mode and ident["db_mode"] and db_mode != ident["db_mode"]:
|
||
die("数据库模式漂移：生效 %s ≠ 锚定 %s" % (db_mode, ident["db_mode"]), 6)
|
||
if db_mode and ident["db_mode"] and db_ref is not None and (db_ref or "") != ident["db_ref"]:
|
||
die("数据库绑定漂移：生效 %s ≠ 锚定 %s（重绑须经显式登记命令）" % (db_ref or "<空>", ident["db_ref"]), 6)
|
||
print("身份校验通过 instance_id=%s data=%s db=%s/%s"
|
||
% (m["instance_id"], ident["data_root"], ident["db_mode"] or "-", ident["db_ref"] or "-"))
|
||
return 0
|
||
|
||
|
||
def main(argv):
|
||
if len(argv) < 3:
|
||
die("用法: svc_manifest.py ENV_DIR <migrate|init|status|set|del|bulk-set|set-desired|verify> [参数]", 2)
|
||
env_dir, cmd = argv[1], argv[2]
|
||
args = argv[3:]
|
||
if cmd == "migrate":
|
||
return cmd_migrate(env_dir)
|
||
if cmd == "init":
|
||
return cmd_init(env_dir)
|
||
if cmd == "status":
|
||
return cmd_status(env_dir)
|
||
if cmd == "set" and len(args) == 2:
|
||
return cmd_set(env_dir, args[0], args[1])
|
||
if cmd == "del" and len(args) == 1:
|
||
return cmd_del(env_dir, args[0])
|
||
if cmd == "bulk-set":
|
||
return cmd_bulk_set(env_dir)
|
||
if cmd == "set-desired" and len(args) == 1:
|
||
return cmd_set_desired(env_dir, args[0])
|
||
if cmd == "verify":
|
||
data_root = db_mode = db_ref = None
|
||
it = iter(args)
|
||
for a in it:
|
||
if a == "--data":
|
||
data_root = next(it, None)
|
||
elif a == "--db-mode":
|
||
db_mode = next(it, None)
|
||
elif a == "--db-ref":
|
||
db_ref = next(it, None)
|
||
return cmd_verify(env_dir, data_root, db_mode, db_ref)
|
||
die("未知命令或参数不全：%s %s" % (cmd, " ".join(args)), 2)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main(sys.argv))
|