Files
CASAN/packages/casan-devkit/casan-init.py
T
thanhnvandClaude Opus 4.8 ff4e9d5a53 feat(install): level-aware casan init + merge-safe adoption + hub guardrail
Answers the 3 adoption questions (Plan-21 follow-up):

1) LEVEL SELECTION (4 packaging levels, packaging/levels.json):
   - install.sh --level core|devkit; platform refused (preview service),
     enterprise refused (future). Level recorded in .casan-level.
   - casan init --level 1..4: L1=gate+Plan-20 hooks only; L2=+CI+domain-pack;
     L3=L2 base+preview note; L4=refused. New `casan level show|set`.
   - levels.json core now includes adapters/ + schemas/ + install scripts.

2) EXISTING SHELLS (agents/skills): init MERGES Plan-20 hooks into an existing
   .claude/settings.json and .codex/{hooks.json,config.toml} idempotently
   instead of clobbering — preserves the project's own hooks/agents/skills and
   unrelated keys. Re-running never duplicates the CASAN hook.

3) NO RE-INDEX / NO SHELL REWRITE: init only adds config; it does not parse or
   index code and does not rewrite the project shell.

Safety fixes after a test accidentally ran init in the real repo:
   - launcher shim now SELF-LOCATES its install from its own path (no ambient
     CASAN_HOME cross-talk).
   - casan init REFUSES to adopt a CASAN source hub into itself (--force to
     override), so the Plan-20 hooks can't block the developing agent.
   - test always runs init inside throwaway dirs; +source-hub guard test.

hybrid-install-tests.sh: 41/41 PASS.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-23 22:00:46 +07:00

531 lines
21 KiB
Python
Executable File

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""`casan init` / `casan verify-harness` (Plan-21 hybrid adoption).
Adopt CASAN into an EXISTING project by writing only per-project config — the
shared harness stays under $CASAN_HOME and is NOT copied into the repo. This is
the codegraph-style flow: global install once, then `casan init` per project.
What init writes into the target repo:
.casan/config.json project id, enforcement/integration mode, clients
.casan/version.lock pinned harness version + gate-code integrity hash
.casan/agentic.env Plan-20 bridge feature flags
.specify/ runtime state root marker (logs/traces/admissions)
.claude/settings.json Plan-20 Claude Code hooks (--client claude|all)
.codex/hooks.json+config Plan-20 Codex hooks (--client codex|all)
`verify` recomputes the resolved harness gate-code hash and compares it to
version.lock — the pin+VERIFY half. Drift/tamper of the global harness relative
to what the project pinned is caught here (preserves the Plan-16 "gates are
trusted code" guarantee even though the harness lives outside the repo).
stdlib-only. Resolves the harness via CASAN_HARNESS_ROOT (set by the global
launcher) or --harness.
"""
from __future__ import annotations
import argparse
import json
import os
import re
import sys
import time
PROJECT_RE = re.compile(r"^[a-z][a-z0-9-]{1,62}$")
# Packaging levels (docs/packaging/CASAN_PACKAGING_PLAN.md, packaging/levels.json).
# Cumulative: devkit⊃core, platform⊃devkit, enterprise⊃platform.
LEVEL_NAME = {"1": "core", "2": "devkit", "3": "platform", "4": "enterprise",
"core": "core", "devkit": "devkit", "platform": "platform", "enterprise": "enterprise"}
LEVEL_NUM = {"core": 1, "devkit": 2, "platform": 3, "enterprise": 4}
def now_iso():
return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
def devkit_root():
return os.path.dirname(os.path.abspath(__file__))
def _copy_if_absent(src, dst, created_rel, target, created):
if not os.path.exists(src) or os.path.exists(dst):
return False
os.makedirs(os.path.dirname(dst), exist_ok=True)
with open(src, "rb") as fh:
data = fh.read()
with open(dst, "wb") as fh:
fh.write(data)
created.append(os.path.relpath(dst, target))
return True
def _copy_tree_missing(src_dir, dst_dir, target, created):
if not os.path.isdir(src_dir):
return 0
n = 0
for dp, _dn, fns in os.walk(src_dir):
rel = os.path.relpath(dp, src_dir)
for fn in fns:
s = os.path.join(dp, fn)
d = os.path.join(dst_dir, rel, fn) if rel != "." else os.path.join(dst_dir, fn)
if _copy_if_absent(s, d, None, target, created):
n += 1
return n
def resolve_harness(explicit):
for cand in (explicit, os.environ.get("CASAN_HARNESS_ROOT")):
if cand and os.path.isdir(os.path.join(cand, "scripts", "bash")):
return os.path.abspath(cand)
install = os.environ.get("CASAN_INSTALL_ROOT")
if install:
c = os.path.join(install, "packages", "casan-harness")
if os.path.isdir(c):
return os.path.abspath(c)
# Fallback: this file lives at packages/casan-devkit/casan-init.py.
c = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "casan-harness"))
return c if os.path.isdir(c) else None
def install_root(harness):
if os.environ.get("CASAN_INSTALL_ROOT"):
return os.path.abspath(os.environ["CASAN_INSTALL_ROOT"])
return os.path.abspath(os.path.join(harness, "..", ".."))
def harness_version(harness):
for p in (os.path.join(install_root(harness), "VERSION"),
os.path.join(harness, "..", "..", "VERSION")):
try:
with open(p, "r", encoding="utf-8") as fh:
v = fh.read().strip()
if v:
return v
except (OSError, IOError):
continue
return "0.0.0"
def compute_live(harness):
"""ALWAYS recompute the gate-code hash from the actual files on disk. Used by
verify so a tampered harness cannot hide behind a stale recorded hash."""
sys.path.insert(0, os.path.join(harness, "scripts", "python"))
try:
import harness_hash # noqa: E402
return harness_hash.compute(harness), "computed"
except Exception as exc: # noqa: BLE001
return "unavailable:%s" % exc, "error"
def compute_harness_hash(harness):
"""For PINNING at init: use the value recorded at install time if present
(it equals a live compute of the same files), else compute live. Verify must
NOT use this — it must call compute_live() to detect drift."""
recorded = os.path.join(install_root(harness), ".harness-hash")
try:
with open(recorded, "r", encoding="utf-8") as fh:
v = fh.read().strip()
if v:
return v, "recorded"
except (OSError, IOError):
pass
return compute_live(harness)
def _write(path, text, backups):
if os.path.exists(path):
bak = path + ".casan-bak"
if not os.path.exists(bak):
with open(path, "r", encoding="utf-8", errors="replace") as fh:
old = fh.read()
with open(bak, "w", encoding="utf-8") as fh:
fh.write(old)
backups.append(bak)
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w", encoding="utf-8") as fh:
fh.write(text)
def _copy_template(src, dst, backups):
try:
with open(src, "r", encoding="utf-8") as fh:
_write(dst, fh.read(), backups)
return True
except (OSError, IOError):
return False
def _load_json_or(path, default):
try:
with open(path, "r", encoding="utf-8") as fh:
return json.load(fh)
except (OSError, ValueError):
return default
def _backup_once(path, backups):
bak = path + ".casan-bak"
if os.path.exists(path) and not os.path.exists(bak):
with open(path, "r", encoding="utf-8", errors="replace") as fh:
old = fh.read()
with open(bak, "w", encoding="utf-8") as fh:
fh.write(old)
backups.append(bak)
def _has_marker_hook(groups, marker):
if not isinstance(groups, list):
return False
for g in groups:
if not isinstance(g, dict):
continue
for h in g.get("hooks", []) or []:
if isinstance(h, dict) and marker in str(h.get("command", "")):
return True
return False
def merge_json_hooks(target_file, template_file, marker, backups):
"""MERGE the template's hook groups into an existing hooks JSON without
clobbering the user's own hooks/agents/skills config. Idempotent by `marker`:
re-running never duplicates the CASAN hook. Returns created|merged|unchanged|None."""
tmpl = _load_json_or(template_file, None)
if not isinstance(tmpl, dict):
return None
existed = os.path.exists(target_file)
doc = _load_json_or(target_file, {}) if existed else {}
if not isinstance(doc, dict):
doc = {}
hooks = doc.get("hooks")
if not isinstance(hooks, dict):
hooks = {}
doc["hooks"] = hooks
changed = not existed
for event, groups in (tmpl.get("hooks") or {}).items():
cur = hooks.get(event)
if not isinstance(cur, list):
cur = []
hooks[event] = cur
if _has_marker_hook(cur, marker):
continue
cur.extend(groups if isinstance(groups, list) else [])
changed = True
for k, v in tmpl.items(): # carry template scalars (e.g. "version") if absent
if k != "hooks" and k not in doc:
doc[k] = v
changed = True
if changed:
_backup_once(target_file, backups)
os.makedirs(os.path.dirname(target_file), exist_ok=True)
with open(target_file, "w", encoding="utf-8") as fh:
fh.write(json.dumps(doc, ensure_ascii=False, indent=2) + "\n")
return ("created" if not existed else "merged") if changed else "unchanged"
def merge_codex_config(target_file, mode, integration_mode, backups):
"""Append a CASAN [casan] section (and, only if absent, a [hooks] enabler) to
an existing .codex/config.toml without duplicating tables. stdlib-only; no TOML
writer needed. Idempotent by the '[casan]' marker."""
casan_block = (
"\n# CASAN Plan-20 (added by casan init)\n"
"[casan]\n"
'enforcement_mode = "%s"\n'
'integration_mode = "%s"\n' % (mode, integration_mode)
)
hooks_block = "\n[hooks]\nenabled = true\nproject_hooks = true\n"
if not os.path.exists(target_file):
os.makedirs(os.path.dirname(target_file), exist_ok=True)
with open(target_file, "w", encoding="utf-8") as fh:
fh.write("# CASAN Plan-20 Codex config (created by casan init)\n"
+ hooks_block + casan_block)
return "created"
with open(target_file, "r", encoding="utf-8", errors="replace") as fh:
cur = fh.read()
if "[casan]" in cur:
return "unchanged"
_backup_once(target_file, backups)
add = ("" if "[hooks]" in cur else hooks_block) + casan_block
with open(target_file, "w", encoding="utf-8") as fh:
fh.write(cur.rstrip() + "\n" + add)
return "merged"
def cmd_init(args):
harness = resolve_harness(args.harness)
if not harness:
sys.stderr.write("casan init: cannot locate the harness. Install CASAN first "
"(install.sh) or set CASAN_HARNESS_ROOT.\n")
return 1
target = os.path.abspath(args.target or os.getcwd())
if not os.path.isdir(target):
sys.stderr.write("casan init: target is not a directory: %s\n" % target)
return 66
# Guardrail: refuse to adopt a CASAN SOURCE HUB into itself. Installing the
# Plan-20 PreToolUse hook into the CASAN repo would block the very agent
# developing CASAN (no admission => deny). Detect the hub by the harness
# source living inside the target. Override with --force for the rare
# intentional case.
hub_marker = os.path.join(target, "packages", "casan-harness", "scripts", "bash", "casan-harness.sh")
if os.path.exists(hub_marker) and not args.force:
sys.stderr.write(
"casan init: target looks like a CASAN SOURCE HUB (%s exists) — refusing to "
"adopt CASAN into itself (the Plan-20 hooks would block your own agent). "
"Use --force only if you really mean to.\n" % os.path.relpath(hub_marker, target))
return 65
project = args.project or re.sub(r"[^a-z0-9-]", "-", os.path.basename(target).lower()).strip("-")
if not PROJECT_RE.match(project):
sys.stderr.write("casan init: --project must match ^[a-z][a-z0-9-]{1,62}$ (got %r)\n" % project)
return 64
# Packaging level to adopt into the project.
lvl_name = LEVEL_NAME.get(str(args.level).lower())
if not lvl_name:
sys.stderr.write("casan init: --level must be 1..4 or core|devkit|platform|enterprise\n")
return 64
lvl = LEVEL_NUM[lvl_name]
if lvl == 4:
sys.stderr.write("casan init: Level 4 (enterprise) is FUTURE / not shipped — refusing "
"(no fake-complete adoption). See packaging/levels.json.\n")
return 3
preview = (lvl == 3) # platform is a separate preview SERVICE; init applies the L2 base
apply_devkit = (lvl >= 2)
clients = ["claude", "codex"] if args.client == "all" else [args.client]
version = harness_version(harness)
hhash, hsource = compute_harness_hash(harness)
created = []
backups = []
def created_add(p):
created.append(os.path.relpath(p, target))
# ── .casan/config.json ──
cfg_dir = os.path.join(target, ".casan")
cfg = {
"schema_version": "21.1",
"project_id": project,
"created_at": now_iso(),
"enforcement_mode": args.mode,
"integration_mode": args.integration_mode,
"clients": clients,
"harness_version": version,
"adoption_model": "hybrid-global",
"target_level": lvl,
"target_level_name": lvl_name,
}
p = os.path.join(cfg_dir, "config.json")
_write(p, json.dumps(cfg, ensure_ascii=False, indent=2) + "\n", backups); created_add(p)
# ── .casan/version.lock (pin) ──
lock = {
"harness_version": version,
"harness_hash": hhash,
"hash_algo": "sha256",
"hash_source": hsource,
"install_root": install_root(harness),
"recorded_at": now_iso(),
}
p = os.path.join(cfg_dir, "version.lock")
_write(p, json.dumps(lock, ensure_ascii=False, indent=2) + "\n", backups); created_add(p)
# ── .casan/agentic.env (Plan-20 flags) ──
env_lines = [
"# CASAN Plan-20 agentic bridge flags. Source before starting the client.",
"CASAN_AGENTIC_BRIDGE_ENABLED=1",
"CASAN_AGENTIC_ENFORCEMENT_MODE=%s" % args.mode,
"CASAN_AGENTIC_INTEGRATION_MODE=%s" % args.integration_mode,
"",
]
p = os.path.join(cfg_dir, "agentic.env")
_write(p, "\n".join(env_lines), backups); created_add(p)
# ── .specify/ state root marker ──
specify = os.path.join(target, ".specify")
os.makedirs(os.path.join(specify, "state"), exist_ok=True)
os.makedirs(os.path.join(specify, "logs"), exist_ok=True)
gi = os.path.join(specify, ".gitignore")
if not os.path.exists(gi):
_write(gi, "# CASAN runtime state — do not commit\nlogs/\nstate/\n", backups); created_add(gi)
# ── Plan-20 client hooks — MERGED into any existing config, never clobbered.
# A project that already has a shell (.claude/agents, skills, its own hooks)
# keeps everything; CASAN hooks are added idempotently.
ad = os.path.join(harness, "adapters")
merges = {}
if "claude" in clients:
dst = os.path.join(target, ".claude", "settings.json")
r = merge_json_hooks(dst, os.path.join(ad, "claude-code", "settings.template.json"),
"claude_hook.py", backups)
if r and r != "unchanged":
created_add(dst)
merges[".claude/settings.json"] = r
if "codex" in clients:
dsth = os.path.join(target, ".codex", "hooks.json")
r = merge_json_hooks(dsth, os.path.join(ad, "codex", "hooks.template.json"),
"codex_hook.py", backups)
if r and r != "unchanged":
created_add(dsth)
merges[".codex/hooks.json"] = r
dstc = os.path.join(target, ".codex", "config.toml")
r = merge_codex_config(dstc, args.mode, args.integration_mode, backups)
if r and r != "unchanged":
created_add(dstc)
merges[".codex/config.toml"] = r
# ── Level 2 (devkit) adoption extras: CI workflow + domain-pack scaffold.
# These are the real difference between L1 (gate/hooks only) and L2 (full
# adoption). Copied only if absent — never clobber the project's own files.
level_extras = []
if apply_devkit:
dk = devkit_root()
ci_src = os.path.join(dk, "templates", "gitea-workflow", "ci.yml")
ci_dst = os.path.join(target, ".gitea", "workflows", "casan-ci.yml")
if _copy_if_absent(ci_src, ci_dst, None, target, created):
level_extras.append(".gitea/workflows/casan-ci.yml")
dom_src = os.path.join(dk, "templates", "domain-pack")
dom_dst = os.path.join(target, "apps", project, "domain")
n = _copy_tree_missing(dom_src, dom_dst, target, created)
if n:
level_extras.append("apps/%s/domain (%d files)" % (project, n))
# ── manifest (so uninstall/verify know what init created) ──
manifest = {
"created": created,
"backups": [os.path.relpath(b, target) for b in backups],
"project_id": project,
"at": now_iso(),
}
p = os.path.join(cfg_dir, "init-manifest.json")
_write(p, json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", backups)
print(json.dumps({
"status": "initialized",
"project_id": project,
"target": target,
"target_level": lvl,
"target_level_name": lvl_name,
"harness_version": version,
"harness_hash": hhash,
"enforcement_mode": args.mode,
"clients": clients,
"created": created,
"hook_merges": merges,
"level_extras": level_extras,
"note": ("harness NOT copied into repo (hybrid model); client hooks MERGED "
"(existing config preserved); run `casan verify-harness` to check the pin"),
}, ensure_ascii=False, indent=2))
if preview:
sys.stderr.write("casan init: NOTE — Level 3 (platform) is a PREVIEW SERVICE (Control "
"Panel/Dashboard), adopted by DEPLOYING it, not by repo config. "
"Applied the Level 2 base here; deploy platform separately.\n")
if hsource == "error":
sys.stderr.write("casan init: WARNING — could not compute harness hash; "
"pin verification will be unavailable.\n")
return 0
def cmd_verify(args):
harness = resolve_harness(args.harness)
if not harness:
sys.stderr.write("casan verify-harness: cannot locate the harness.\n")
return 1
target = os.path.abspath(args.target or os.getcwd())
lock_path = os.path.join(target, ".casan", "version.lock")
if not os.path.exists(lock_path):
sys.stderr.write("casan verify-harness: no .casan/version.lock (run `casan init` first).\n")
return 1
with open(lock_path, "r", encoding="utf-8") as fh:
lock = json.load(fh)
expected = lock.get("harness_hash")
actual, _src = compute_live(harness) # live recompute — never the cached hash
ok = (expected == actual) and expected and not str(expected).startswith("unavailable")
result = {
"status": "ok" if ok else "drift",
"expected": expected,
"actual": actual,
"harness_version_lock": lock.get("harness_version"),
"harness_version_now": harness_version(harness),
"harness_root": harness,
}
print(json.dumps(result, ensure_ascii=False, indent=2))
if not ok:
sys.stderr.write("HARNESS_INTEGRITY_DRIFT — the resolved harness does not match the "
"project pin. The global harness changed or was tampered.\n")
return 3
return 0
def cmd_level(args):
"""Show the installed packaging level and the project's target level."""
harness = resolve_harness(args.harness)
installed = None
if harness:
try:
with open(os.path.join(install_root(harness), ".casan-level"), "r", encoding="utf-8") as fh:
installed = fh.read().strip()
except (OSError, IOError):
installed = "unknown"
target = os.path.abspath(args.target or os.getcwd())
cfg = _load_json_or(os.path.join(target, ".casan", "config.json"), {})
status_map = {1: "implemented", 2: "implemented", 3: "preview", 4: "future"}
tl = cfg.get("target_level")
out = {
"installed_level": installed,
"project_target_level": tl,
"project_target_level_name": cfg.get("target_level_name"),
"project_level_status": status_map.get(tl, "unknown") if tl else None,
"levels": {
"1 core": "implemented — harness + gates + CLI",
"2 devkit": "implemented — + adoption tooling (casan init, CI, domain-pack)",
"3 platform": "preview — Control Panel/Dashboard SERVICE (deploy separately)",
"4 enterprise": "future — not shipped",
},
}
print(json.dumps(out, ensure_ascii=False, indent=2))
return 0
def main(argv=None):
parser = argparse.ArgumentParser(prog="casan-init", description="CASAN hybrid adoption")
sub = parser.add_subparsers(dest="cmd")
pi = sub.add_parser("init", help="adopt CASAN into the current project (config only)")
pi.add_argument("--target", help="project root (default: cwd)")
pi.add_argument("--project", help="project id (^[a-z][a-z0-9-]{1,62}$; default: dir name)")
pi.add_argument("--client", choices=["claude", "codex", "all"], default="all")
pi.add_argument("--level", default="devkit",
help="packaging level to adopt: 1|core, 2|devkit (default), 3|platform (preview), 4|enterprise (refused)")
pi.add_argument("--mode", choices=["observe", "enforce"], default="observe")
pi.add_argument("--integration-mode", dest="integration_mode",
choices=["project_hook", "managed_hook", "casan_owned"], default="project_hook")
pi.add_argument("--force", action="store_true",
help="bypass the source-hub safety guard (adopt CASAN into a CASAN checkout)")
pi.add_argument("--harness", help="override harness root")
pv = sub.add_parser("verify", help="verify the resolved harness matches the project pin")
pv.add_argument("--target", help="project root (default: cwd)")
pv.add_argument("--harness", help="override harness root")
pl = sub.add_parser("level", help="show installed + project packaging level")
pl.add_argument("--show", action="store_true", help="(default) show levels")
pl.add_argument("--target", help="project root (default: cwd)")
pl.add_argument("--harness", help="override harness root")
args = parser.parse_args(argv)
if args.cmd == "init":
return cmd_init(args)
if args.cmd == "verify":
return cmd_verify(args)
if args.cmd == "level":
return cmd_level(args)
parser.print_help()
return 64
if __name__ == "__main__":
sys.exit(main())