Agent-System-Dev/scripts/svc_manifest.py

#!/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,
            "db_marker": "fresh",
        },
        "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))