#!/usr/bin/env python3 # -*- coding: utf-8 -*- """`casan init` / `casan verify-harness` (Plan-21 production adoption). Adopt CASAN into an EXISTING project using an explicit runtime contract: managed mode pins the shared Core under $CASAN_HOME; vendored mode installs the same production-only Core under `.casan/runtime/casan-core`. 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) .casan/runtime/casan-core optional self-contained Core (--runtime vendored) `verify` recomputes the resolved harness gate-code hash and compares it to version.lock — the pin+VERIFY half. Drift/tamper of the managed or vendored runtime relative to what the project pinned is caught here (preserves the Plan-16 "gates are trusted code" guarantee). stdlib-only. Resolves the harness via CASAN_HARNESS_ROOT (set by the global launcher) or --harness. """ from __future__ import annotations import argparse from contextlib import contextmanager import hashlib import json import os import re import shutil import stat import subprocess import sys import tempfile 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} SUPPORTED_CLIENTS = ("claude", "codex", "vscode-copilot") VSCODE_EXTENSION_IDS = { "claude": ("anthropic.claude-code",), "codex": ("OpenAI.chatgpt",), "vscode-copilot": ( "GitHub.copilot", "GitHub.copilot-chat", "fpt-casan.casan-governed-chat", ), } PROMPT_MARKER_START = "" PROMPT_MARKER_END = "" CLIENT_LABELS = { "claude": "Claude Code (CLI + VS Code + JetBrains)", "codex": "Codex local (desktop + CLI + IDE)", "vscode-copilot": "VS Code / @casan", } CLIENT_SURFACES = { "claude": { "supported_local": [ "claude-code-cli", "claude-code-vscode", "claude-code-jetbrains", ], "not_covered": ["claude-desktop", "claude-web"], "contract": "shared_claude_code_project_settings", }, "codex": { "supported_local": [ "codex-desktop-local", "codex-cli", "codex-ide-extension-local", ], "not_covered": ["codex-cloud", "codex-web"], "contract": "shared_codex_local_project_hooks", }, "vscode-copilot": { "supported_local": ["vscode-copilot-explicit-at-casan"], "not_covered": ["github-copilot-built-in-chat"], "contract": "casan_owned_explicit_route", }, } CASAN_CODEX_DESCRIPTIONS = { ( "CASAN Plan-20 lifecycle hooks. Review with /hooks; the project " "bootstrap resolves and verifies the pinned global harness." ), ( "CASAN lifecycle hooks. Review with /hooks; the project bootstrap " "resolves and verifies the pinned Core runtime." ), } RUNTIME_MODES = ("managed", "vendored") def _color(code, text): if (not sys.stdout.isatty() or os.environ.get("NO_COLOR") is not None or os.environ.get("TERM") == "dumb"): return text return "\033[%sm%s\033[0m" % (code, text) def _mark(ok): return _color("32" if ok else "31", "✓" if ok else "✗") def _warn_mark(): return _color("33", "!") def _neutral_mark(): return _color("2", "–") def _heading(text): print(_color("1;36", text)) def _details(rows): width = max((len(label) for label, _value in rows), default=0) for label, value in rows: print(" %-*s %s" % (width, label, value)) def _client_names(clients): return ", ".join(CLIENT_LABELS.get(client, client) for client in clients) or "None" def _emit_json_or_human(args, payload, renderer): if getattr(args, "json", False): print(json.dumps(payload, ensure_ascii=False, indent=2)) else: renderer(payload) def _render_init(result): _heading("%s CASAN initialized" % _mark(True)) _details([ ("Project", result["project_id"]), ("Location", result["target"]), ("Level", "%s (%s)" % ( result["target_level_name"].capitalize(), result["target_level"])), ("Runtime", "%s — %s" % ( result["runtime_mode"].capitalize(), result["runtime_path"])), ("Mode", result["enforcement_mode"]), ("Clients", _client_names(result["clients"])), ("Files", "%d created or updated" % len(result["created"])), ("Harness", result["harness_version"]), ]) print() print(_color("1", "Integrations")) for path, status in result["hook_merges"].items(): if status in ("absent", "skipped"): marker = _neutral_mark() display = "not needed" else: marker = _mark(status is not None) display = status or "unavailable" print(" %s %-26s %s" % (marker, path, display)) extension = result["vscode_extension"] if extension.get("status") not in ("skipped",): ok = extension.get("status") == "installed" print(" %s %-26s %s" % ( _mark(ok), "VS Code extension", extension.get("status"))) print() print(_color("1", "Next steps")) print(" 1. Run `casan doctor`") if "codex" in result["clients"]: print(" 2. In a local Codex client, open `/hooks` and trust this project's hook.") def _render_verify(result): ok = result["status"] == "ok" _heading("%s Harness integrity %s" % ( _mark(ok), "verified" if ok else "check failed")) _details([ ("Version", str(result.get("harness_version_now") or "unknown")), ("Location", str(result.get("harness_root") or "unavailable")), ("Expected", str(result.get("expected") or "unavailable")), ("Actual", str(result.get("actual") or "unavailable")), ]) def _render_level(result): _heading("CASAN packaging level") _details([ ("Installed", str(result.get("installed_level") or "unknown")), ("Project", ("%s (%s)" % ( result.get("project_target_level_name"), result.get("project_target_level"))) if result.get("project_target_level") else "not initialized"), ("Runtime", ("%s — %s" % ( result.get("project_runtime_mode"), result.get("project_runtime_path"))) if result.get("project_target_level") else "not initialized"), ("Status", str(result.get("project_level_status") or "unknown")), ]) print() for level, description in result["levels"].items(): print(" %-14s %s" % (level, description)) def _render_doctor(result): ready = result["status"] == "ready" operational = result.get("operational_status") _heading("CASAN Doctor") print(" %s Harness integrity" % _mark(result["integrity"]["ok"])) print(" %s Project bootstrap" % _mark(result["bootstrap"]["ok"])) for client, item in result["client_checks"].items(): label = CLIENT_LABELS.get(client, client) print(" %s %s" % (_mark(item.get("ready", False)), label)) surfaces = item.get("supported_surfaces") or [] if surfaces: print(" Surfaces: %s" % ", ".join(surfaces)) reason = item.get("smoke", {}).get("reason") if reason and not item.get("ready"): print(" %s" % reason) if result["warnings"]: print() print(_color("1;33", "Warnings")) for warning in result["warnings"]: print(" %s %s" % (_warn_mark(), warning)) if result.get("required_actions"): print() print(_color("1;33", "Required actions")) for action in result["required_actions"]: print(" %s %s" % (_warn_mark(), action["message"])) print() if ready and operational == "user_action_required": status = _color("1;33", "CONFIGURED — USER ACTION REQUIRED") elif ready: status = _color("1;32", "READY") else: status = _color("1;31", "NOT READY") print("Status: %s" % status) def _render_uninstall(result): _heading("%s CASAN removed from project" % _mark(True)) _details([ ("Project", result.get("project_id") or "unknown"), ("Location", result["target"]), ("Hooks", "%d integration files cleaned" % len(result["hook_changes"])), ("Files", "%d CASAN-owned files removed" % len(result["removed"])), ("Evidence", "removed" if result["purged"] else "retained"), ("VS Code extension", result["vscode_extension"]["status"]), ]) if result["retained"]: print() print(_color("1", "Retained for safety")) for path in result["retained"]: print(" - %s" % path) @contextmanager def owner_writable(path): """Temporarily make an owner-controlled path writable, then restore modes. Some existing shells intentionally commit `.claude`/`.github` directories as 0555. CASAN may add its config there only when the current user owns the path; it never escalates privileges or leaves permissions broadened. """ changed = [] candidates = [] if os.path.exists(path): candidates.append(path) parent = os.path.dirname(path) while parent and not os.path.exists(parent): parent = os.path.dirname(parent) if parent and parent not in candidates: candidates.append(parent) try: for candidate in candidates: info = os.stat(candidate) mode = stat.S_IMODE(info.st_mode) if mode & stat.S_IWUSR: continue getuid = getattr(os, "getuid", None) if getuid is not None and info.st_uid != getuid(): raise PermissionError( "CASAN cannot write %s: current user does not own %s" % (path, candidate)) os.chmod(candidate, mode | stat.S_IWUSR) changed.append((candidate, mode)) yield finally: for candidate, mode in reversed(changed): try: os.chmod(candidate, mode) except OSError: pass 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 with open(src, "rb") as fh: data = fh.read() with owner_writable(dst): os.makedirs(os.path.dirname(dst), exist_ok=True) 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"), os.environ.get("CASAN_GLOBAL_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): local_root = os.path.abspath(os.path.join(harness, "..", "..")) for p in (os.path.join(local_root, "VERSION"), os.path.join(install_root(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.""" local_root = os.path.abspath(os.path.join(harness, "..", "..")) for recorded in ( os.path.join(local_root, ".harness-hash"), 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): continue return compute_live(harness) def resolve_project_harness(target, explicit=None): lock = _load_json_or( os.path.join(target, ".casan", "version.lock"), {}) runtime_path = lock.get("runtime_path") if lock.get("runtime_mode") == "vendored" and isinstance(runtime_path, str): root = ( None if os.path.isabs(runtime_path) else _safe_project_path(target, runtime_path) ) if not root: return None candidate = os.path.join(root, "packages", "casan-harness") if os.path.isdir(os.path.join(candidate, "scripts", "bash")): return os.path.abspath(candidate) return None return resolve_harness(explicit) def _write(path, text, backups): with owner_writable(path): 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 _safe_project_path(target, relative): """Resolve a manifest-owned relative path without allowing target escape.""" normalized = str(relative).replace("\\", "/").strip("/") if not normalized or os.path.isabs(str(relative)): return None candidate = os.path.abspath(os.path.join(target, *normalized.split("/"))) target_real = os.path.realpath(target) candidate_real = os.path.realpath(candidate) try: if os.path.commonpath((target_real, candidate_real)) != target_real: return None except ValueError: return None return candidate def _file_sha256(path): if not os.path.isfile(path) or os.path.islink(path): return None digest = hashlib.sha256() try: with open(path, "rb") as fh: for chunk in iter(lambda: fh.read(1024 * 1024), b""): digest.update(chunk) except OSError: return None return digest.hexdigest() def _install_vendored_core(target, source_harness): """Atomically install the production Core runtime inside one project.""" source_root = os.path.abspath(os.path.join(source_harness, "..", "..")) copier = os.path.join(source_root, "scripts", "copy-runtime.py") source_bin = os.path.join(source_root, "bin", "casan") source_version = os.path.join(source_root, "VERSION") for required in (source_bin, source_version): if not os.path.isfile(required): raise RuntimeError( "vendored runtime source is incomplete; missing %s" % required) destination = _safe_project_path( target, ".casan/runtime/casan-core") if not destination: raise RuntimeError( "vendored runtime path escapes the project through a symlink") parent = os.path.dirname(destination) os.makedirs(parent, exist_ok=True) staging = tempfile.mkdtemp(prefix=".casan-core-", dir=parent) try: if os.path.isfile(copier): result = subprocess.run( [ sys.executable, copier, "--source-root", source_root, "--destination-root", staging, "--component", "harness", "--clean", ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False, ) if result.returncode != 0: raise RuntimeError( result.stderr.strip() or result.stdout.strip() or "production runtime copy failed") else: shutil.copytree( source_harness, os.path.join(staging, "packages", "casan-harness")) vendored_harness = os.path.join( staging, "packages", "casan-harness") forbidden = [ relative for relative in ( "tests", "level5", "scripts/bash/ci-harness-gate.sh", "scripts/bash/security-gate.sh", "scripts/bash/dashboard-server.py", ) if os.path.exists(os.path.join( vendored_harness, *relative.split("/"))) ] if forbidden: raise RuntimeError( "source is not a production runtime; forbidden paths: %s" % ", ".join(forbidden)) os.makedirs(os.path.join(staging, "bin"), exist_ok=True) shutil.copy2(source_bin, os.path.join(staging, "bin", "casan")) shutil.copy2(source_version, os.path.join(staging, "VERSION")) with open(os.path.join(staging, ".casan-level"), "w", encoding="utf-8") as fh: fh.write("core\n") vendored_hash, hash_source = compute_live(vendored_harness) if hash_source == "error": raise RuntimeError( "cannot verify staged vendored Core: %s" % vendored_hash) with open(os.path.join(staging, ".harness-hash"), "w", encoding="utf-8") as fh: fh.write(vendored_hash + "\n") if os.path.isdir(destination) and not os.path.islink(destination): shutil.rmtree(destination) elif os.path.exists(destination) or os.path.islink(destination): os.unlink(destination) os.replace(staging, destination) staging = None finally: if staging and os.path.isdir(staging): shutil.rmtree(staging) file_count = sum( len(files) for _root, _dirs, files in os.walk(destination)) return destination, file_count def _remove_vendored_core(target): destination = _safe_project_path( target, ".casan/runtime/casan-core") if not destination: return False if not os.path.exists(destination) and not os.path.islink(destination): return False with owner_writable(destination): if os.path.isdir(destination) and not os.path.islink(destination): shutil.rmtree(destination) else: os.unlink(destination) _prune_empty_parents(destination, target) return True def _backup_once(path, backups): if backups is None: return 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 owner_writable(bak): 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 _remove_marker_hooks(doc, marker): """Remove only handlers owned by CASAN, preserving all user hook groups.""" changed = False hooks = doc.get("hooks") if isinstance(doc, dict) else None if not isinstance(hooks, dict): return changed for event in list(hooks): groups = hooks.get(event) if not isinstance(groups, list): continue kept_groups = [] for group in groups: if not isinstance(group, dict): kept_groups.append(group) continue handlers = group.get("hooks") if not isinstance(handlers, list): kept_groups.append(group) continue kept_handlers = [] for handler in handlers: commands = "" if isinstance(handler, dict): commands = "%s %s" % (handler.get("command", ""), handler.get("commandWindows", "")) if marker in commands: changed = True else: kept_handlers.append(handler) if kept_handlers: updated = dict(group) updated["hooks"] = kept_handlers kept_groups.append(updated) elif handlers: changed = True if kept_groups: hooks[event] = kept_groups else: hooks.pop(event, None) return changed def merge_json_hooks( target_file, template_file, marker, backups, owned_scalars=None): """MERGE the template's hook groups into an existing hooks JSON without clobbering the user's own hooks/agents/skills config. Existing CASAN-owned handlers and explicitly listed CASAN-owned scalar metadata are replaced so upgrades cannot leave a stale or client-incompatible value behind. 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 = {} before = json.dumps(doc, ensure_ascii=False, sort_keys=True) for key, owned_values in (owned_scalars or {}).items(): if doc.get(key) in owned_values: doc.pop(key, None) hooks = doc.get("hooks") if not isinstance(hooks, dict): hooks = {} doc["hooks"] = hooks _remove_marker_hooks(doc, marker) for event, groups in (tmpl.get("hooks") or {}).items(): cur = hooks.get(event) if not isinstance(cur, list): cur = [] hooks[event] = cur cur.extend(groups if isinstance(groups, list) else []) 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 = (not existed) or before != json.dumps(doc, ensure_ascii=False, sort_keys=True) if changed: with owner_writable(target_file): _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 remove_json_hooks(target_file, marker, backups): """Disable one CASAN integration without touching non-CASAN hooks.""" if not os.path.exists(target_file): return "absent" doc = _load_json_or(target_file, None) if not isinstance(doc, dict) or not _remove_marker_hooks(doc, marker): return "unchanged" with owner_writable(target_file): _backup_once(target_file, backups) with open(target_file, "w", encoding="utf-8") as fh: fh.write(json.dumps(doc, ensure_ascii=False, indent=2) + "\n") return "removed" def clean_legacy_codex_config(target_file, backups): """Remove only Plan-21's obsolete `[hooks]`/`[casan]` blocks. Current Codex enables hooks by default and rejects/ignores the legacy custom keys. The sentinel comment makes this migration precise; user-authored TOML is otherwise preserved byte-for-byte. """ if not os.path.exists(target_file): return "absent" with open(target_file, "r", encoding="utf-8", errors="replace") as fh: current = fh.read() sentinel = "# CASAN Plan-20 (added by casan init)" created_header = "# CASAN Plan-20 Codex config (created by casan init)" if sentinel not in current and created_header not in current: return "unchanged" lines = current.splitlines() kept = [] skipping = False for line in lines: stripped = line.strip() if stripped == created_header: continue if stripped == sentinel: skipping = True continue if skipping and stripped.startswith("[") and stripped != "[casan]": skipping = False if not skipping: kept.append(line) # Old files created by CASAN put [hooks] immediately before the sentinel. cleaned = "\n".join(kept) cleaned = re.sub( r"(?ms)\n?\[hooks\]\nenabled = true\nproject_hooks = true\n(?=\s*$)", "\n", cleaned) cleaned = cleaned.strip() + ("\n" if cleaned.strip() else "") with owner_writable(target_file): _backup_once(target_file, backups) if cleaned: with open(target_file, "w", encoding="utf-8") as fh: fh.write(cleaned) else: os.unlink(target_file) return "cleaned" def is_casan_source_hub(target): """Distinguish this product's source checkout from an adopted vendored app. Older DevKit adoption copied ``packages/casan-harness`` into an application. That directory alone therefore does not make the application a CASAN source hub. The source checkout also carries the DevKit source, release manifest, and root installer. """ required = ( os.path.join("packages", "casan-harness", "scripts", "bash", "casan-harness.sh"), os.path.join("packages", "casan-devkit", "casan-init.py"), os.path.join("packaging", "levels.json"), "install.sh", ) return all(os.path.isfile(os.path.join(target, item)) for item in required) def _hybrid_prompt_block(clients): enabled = ", ".join(clients) if clients else "none" return """ ## CASAN IDE governance — mandatory - `.casan/config.json` is the source of truth for enabled integrations (currently: %s). - Claude Code and Codex normal prompts are governed only when their integration is enabled and the project hooks are loaded; Codex additionally requires `/hooks` trust review. - GitHub Copilot built-in chat is not globally intercepted. Use the explicit `@casan` participant for a CASAN-owned Copilot turn. - A retained `bin/casan-chat` entrypoint remains a compatible governed route, but it is no longer the only supported prompt boundary. - Do not claim CASAN certification without the matching trace/evidence, and do not bypass a denied or degraded decision. """ % enabled def _replace_managed_prompt_block(path, replacement, backups): if not os.path.isfile(path): return "absent" with open(path, "r", encoding="utf-8", errors="replace") as fh: current = fh.read() pattern = re.compile( re.escape(PROMPT_MARKER_START) + r".*?" + re.escape(PROMPT_MARKER_END), re.DOTALL) if not pattern.search(current): return "unmanaged" updated = pattern.sub(replacement, current) if updated == current: return "unchanged" with owner_writable(path): _backup_once(path, backups) with open(path, "w", encoding="utf-8") as fh: fh.write(updated.rstrip() + "\n") return "migrated" def _legacy_instruction_conflicts(target): """Report unmarked legacy prose that init must not silently rewrite.""" conflicts = [] phrases = ( "The supported prompt boundary is `bin/casan-chat`", "CASAN Core が intercept できない", "Direct Claude, ChatGPT, Codex or Copilot UI prompts are outside", ) for relative in ("AGENTS.md", "CLAUDE.md", os.path.join(".github", "copilot-instructions.md")): path = os.path.join(target, relative) if not os.path.isfile(path): continue with open(path, "r", encoding="utf-8", errors="replace") as fh: current = fh.read() unmanaged = re.sub( re.escape(PROMPT_MARKER_START) + r".*?" + re.escape(PROMPT_MARKER_END), "", current, flags=re.DOTALL) if any(phrase in unmanaged for phrase in phrases): conflicts.append(relative.replace(os.sep, "/")) return conflicts def migrate_vendored_prompt_contract(target, clients, backups): """Migrate only CASAN-owned legacy files; retain all project-owned content.""" replacement = _hybrid_prompt_block(clients) instruction_files = {} for relative in ("AGENTS.md", "CLAUDE.md", os.path.join(".github", "copilot-instructions.md")): instruction_files[relative.replace(os.sep, "/")] = ( _replace_managed_prompt_block( os.path.join(target, relative), replacement, backups)) policy_path = os.path.join(target, ".casan", "prompt-policy.json") policy_status = "absent" if os.path.isfile(policy_path): policy = _load_json_or(policy_path, None) if isinstance(policy, dict): before = json.dumps(policy, ensure_ascii=False, sort_keys=True) policy["direct_external_ui"] = "client_dependent_see_config" policy["agentic_clients"] = clients policy["hybrid_adoption"] = True if before != json.dumps(policy, ensure_ascii=False, sort_keys=True): with owner_writable(policy_path): _backup_once(policy_path, backups) with open(policy_path, "w", encoding="utf-8") as fh: fh.write(json.dumps( policy, ensure_ascii=False, indent=2) + "\n") policy_status = "migrated" else: policy_status = "unchanged" else: policy_status = "unreadable" return { "detected": True, "vendored_harness": "retained_for_compatibility", "managed_instruction_blocks": instruction_files, "prompt_policy": policy_status, "manual_review": _legacy_instruction_conflicts(target), "note": ( "Existing .claude agents/skills/commands, .github content, CI, " "and the vendored harness were preserved."), } def _normalize_runtime_mode(value): aliases = { "1": "managed", "global": "managed", "shared": "managed", "2": "vendored", "local": "vendored", "offline": "vendored", "self-contained": "vendored", "selfcontained": "vendored", } normalized = aliases.get(str(value).strip().lower(), str(value).strip().lower()) if normalized not in RUNTIME_MODES: raise ValueError( "unknown runtime %r (choose managed or vendored)" % value) return normalized def select_runtime_mode(explicit, previous, interactive): """Resolve Core placement without surprising existing projects. An explicit flag always wins. Re-init keeps a valid existing selection. Only a new interactive adoption opens the placement wizard; automation defaults to managed and never waits for input. """ if explicit: return _normalize_runtime_mode(explicit) if previous in RUNTIME_MODES: return previous if not interactive: return "managed" while True: sys.stderr.write( "\nCore runtime placement\n\n" " 1) Managed (Recommended)\n" " Use the shared CASAN installation, pinned by version and hash.\n" " Best for developer workstations and managed CI.\n\n" " 2) Vendored\n" " Copy production-only Core into .casan/runtime/casan-core.\n" " Best for offline, air-gapped, or self-contained repositories.\n\n" "Select runtime [1]: ") sys.stderr.flush() answer = sys.stdin.readline() if not answer or not answer.strip(): return "managed" try: return _normalize_runtime_mode(answer) except ValueError: sys.stderr.write( "Invalid selection. Enter 1 for Managed or 2 for Vendored.\n") def _normalize_clients(values): aliases = { "1": "claude", "2": "codex", "3": "vscode-copilot", "copilot": "vscode-copilot", "vscode": "vscode-copilot", "github-copilot": "vscode-copilot", } selected = [] for value in values: for raw in str(value).split(","): item = aliases.get(raw.strip().lower(), raw.strip().lower()) if item == "all": item_values = list(SUPPORTED_CLIENTS) elif item in ("none", ""): item_values = [] elif item in SUPPORTED_CLIENTS: item_values = [item] else: raise ValueError("unknown --client %r (valid: %s, all, none)" % (raw, ", ".join(SUPPORTED_CLIENTS))) for normalized in item_values: if normalized not in selected: selected.append(normalized) return selected def select_clients(values, interactive): """Normalize repeatable/comma-separated selections. A plain interactive `casan init` presents the requested IDE menu. In non-interactive automation, the historical Claude+Codex default is kept. """ if values: return _normalize_clients(values) if not interactive: return _normalize_clients(["claude,codex"]) while True: sys.stderr.write( "\nClient integrations\n\n" " 1) Claude Code (CLI + VS Code + JetBrains)\n" " 2) Codex local (desktop app + CLI + IDE extension)\n" " 3) GitHub Copilot in VS Code via explicit @casan route\n\n" "Select clients (comma-separated) [1,2]: ") sys.stderr.flush() answer = sys.stdin.readline() values = [answer.strip() or "1,2"] if answer else ["1,2"] try: return _normalize_clients(values) except ValueError: sys.stderr.write( "Invalid selection. Enter 1, 2, 3, a comma-separated list, " "`all`, or `none`.\n") def merge_vscode_recommendations(target_file, clients, backups): existed = os.path.exists(target_file) doc = _load_json_or(target_file, {}) if existed else {} if not isinstance(doc, dict): doc = {} before = json.dumps(doc, ensure_ascii=False, sort_keys=True) recommendations = doc.get("recommendations") if not isinstance(recommendations, list): recommendations = [] managed = {item for values in VSCODE_EXTENSION_IDS.values() for item in values} recommendations = [item for item in recommendations if item not in managed] for client in clients: for extension_id in VSCODE_EXTENSION_IDS.get(client, ()): if extension_id not in recommendations: recommendations.append(extension_id) if recommendations: doc["recommendations"] = recommendations else: doc.pop("recommendations", None) changed = (not existed and bool(doc)) or before != json.dumps( doc, ensure_ascii=False, sort_keys=True) if changed: with owner_writable(target_file): _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 build_vscode_vsix(): builder = os.path.join(devkit_root(), "package-vscode-extension.py") output = os.path.join(devkit_root(), "dist", "casan-governed-chat.vsix") if not os.path.isfile(builder): return None, "builder_missing" result = subprocess.run( [sys.executable, builder, "--output", output], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False) if result.returncode != 0 or not os.path.isfile(output): return None, "build_failed:%s" % result.stderr.strip() return output, "built" def install_vscode_extension(mode): """Install the local CASAN VSIX when requested and `code` is available.""" if mode == "no": return {"status": "skipped", "reason": "disabled"} vsix, status = build_vscode_vsix() if not vsix: return {"status": "failed", "reason": status} code = shutil.which("code") if not code: return { "status": "needs_install", "reason": "code_cli_not_found", "vsix": vsix, "action": "Install this VSIX from VS Code: Extensions: Install from VSIX...", } result = subprocess.run( [code, "--install-extension", vsix, "--force"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False) return { "status": "installed" if result.returncode == 0 else "failed", "extension": "fpt-casan.casan-governed-chat", "vsix": vsix, "command": code, "detail": (result.stdout or result.stderr).strip(), } 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 the CASAN PRODUCT SOURCE HUB into itself. # A vendored application from the previous DevKit model also contains # packages/casan-harness, so that directory by itself is not sufficient to # identify a source hub. Such applications are migrated in place below. hub_marker = os.path.join(target, "packages", "casan-harness", "scripts", "bash", "casan-harness.sh") source_hub = is_casan_source_hub(target) vendored_harness = os.path.isfile(hub_marker) and not source_hub if source_hub and not args.force: sys.stderr.write( "casan init: target is a CASAN SOURCE HUB — refusing to " "adopt CASAN into itself (the Plan-20 hooks would block your own agent). " "Use --force only if you really mean to.\n") 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) previous_config = _load_json_or( os.path.join(target, ".casan", "config.json"), {}) previous_runtime_mode = previous_config.get("runtime_mode") interactive = ( sys.stdin.isatty() and not args.non_interactive and not args.json) try: runtime_mode = select_runtime_mode( args.runtime, previous_runtime_mode, interactive) clients = select_clients(args.client, interactive) except ValueError as error: sys.stderr.write("casan init: %s\n" % error) return 64 runtime_removed = False runtime_files = 0 if runtime_mode == "vendored": try: runtime_root, runtime_files = _install_vendored_core( target, harness) except (OSError, RuntimeError) as error: sys.stderr.write( "casan init: cannot install vendored Core runtime: %s\n" % error) return 1 active_harness = os.path.join( runtime_root, "packages", "casan-harness") runtime_path = os.path.relpath( runtime_root, target).replace(os.sep, "/") else: runtime_removed = _remove_vendored_core(target) active_harness = harness runtime_root = install_root(harness) runtime_path = runtime_root version = harness_version(active_harness) hhash, hsource = compute_harness_hash(active_harness) created = [] backups = [] previous_manifest = _load_json_or( os.path.join(target, ".casan", "init-manifest.json"), {}) owned_files = { str(path).replace("\\", "/") for path in previous_manifest.get("owned_files", []) if isinstance(path, str) } previous_owned_hashes = { str(path).replace("\\", "/"): digest for path, digest in previous_manifest.get( "owned_file_hashes", {}).items() if isinstance(path, str) and isinstance(digest, str) } newly_owned_files = set() level_removed = [] level_retained = [] if not apply_devkit: devkit_owned_files = { path for path in owned_files if (path.startswith(".gitea/workflows/casan-") or path.startswith("apps/")) } level_removed, level_retained = _remove_owned_project_artifacts( target, devkit_owned_files, previous_owned_hashes) for relative in level_removed: owned_files.discard(relative) previous_owned_hashes.pop(relative, None) def created_add(p): created.append(os.path.relpath(p, target)) def mark_owned_if_absent(p): if not os.path.exists(p): relative = os.path.relpath(p, target).replace(os.sep, "/") owned_files.add(relative) newly_owned_files.add(relative) # ── .casan/config.json ── cfg_dir = os.path.join(target, ".casan") cfg = { "schema_version": "21.2", "project_id": project, "created_at": now_iso(), "enforcement_mode": args.mode, "integration_mode": args.integration_mode, "clients": clients, "client_capabilities": { "claude": "project_hook" if "claude" in clients else "disabled", "codex": "project_hook_trust_required" if "codex" in clients else "disabled", "vscode-copilot": "casan_owned_explicit_at_casan" if "vscode-copilot" in clients else "disabled", "vscode-native-copilot": "unsupported_global_interception", }, "client_surfaces": { client: { "enabled": client in clients, **surface, } for client, surface in CLIENT_SURFACES.items() }, "harness_version": version, "adoption_model": ( "managed-global" if runtime_mode == "managed" else "vendored-project"), "runtime_mode": runtime_mode, "runtime_path": runtime_path, "target_level": lvl, "target_level_name": lvl_name, } p = os.path.join(cfg_dir, "config.json") mark_owned_if_absent(p) _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": runtime_root, "runtime_mode": runtime_mode, "runtime_path": runtime_path, "recorded_at": now_iso(), } p = os.path.join(cfg_dir, "version.lock") mark_owned_if_absent(p) _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 flags (reference/export compatibility).", "# .casan/casan-hook.py loads the authoritative values from config.json automatically.", "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") mark_owned_if_absent(p) _write(p, "\n".join(env_lines), backups); created_add(p) # Stable project-local bootstrap. It loads the config above, resolves the # managed/vendored Core and verifies version.lock before dispatching. bootstrap_source = os.path.join(devkit_root(), "templates", "project", "casan-hook.py") bootstrap_target = os.path.join(cfg_dir, "casan-hook.py") if not os.path.isfile(bootstrap_source): sys.stderr.write("casan init: missing project hook bootstrap template\n") return 1 mark_owned_if_absent(bootstrap_target) with open(bootstrap_source, "r", encoding="utf-8") as fh: _write(bootstrap_target, fh.read(), backups) try: os.chmod(bootstrap_target, 0o755) except OSError: pass created_add(bootstrap_target) # ── .specify/ state root marker ── specify = os.path.join(target, ".specify") with owner_writable(os.path.join(specify, "state")): os.makedirs(os.path.join(specify, "state"), exist_ok=True) with owner_writable(os.path.join(specify, "logs")): os.makedirs(os.path.join(specify, "logs"), exist_ok=True) gi = os.path.join(specify, ".gitignore") if not os.path.exists(gi): mark_owned_if_absent(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 = {} claude_dst = os.path.join(target, ".claude", "settings.json") if "claude" in clients: dst = claude_dst mark_owned_if_absent(dst) remove_json_hooks(dst, "claude_hook.py", backups) r = merge_json_hooks(dst, os.path.join(ad, "claude-code", "settings.template.json"), "casan-hook.py", backups) if r and r != "unchanged": created_add(dst) merges[".claude/settings.json"] = r else: legacy = remove_json_hooks(claude_dst, "claude_hook.py", backups) current = remove_json_hooks(claude_dst, "casan-hook.py", backups) merges[".claude/settings.json"] = ( "removed" if "removed" in (legacy, current) else current) codex_hooks_dst = os.path.join(target, ".codex", "hooks.json") if "codex" in clients: dsth = codex_hooks_dst mark_owned_if_absent(dsth) remove_json_hooks(dsth, "codex_hook.py", backups) r = merge_json_hooks(dsth, os.path.join(ad, "codex", "hooks.template.json"), "casan-hook.py", backups, { "description": CASAN_CODEX_DESCRIPTIONS, }) if r and r != "unchanged": created_add(dsth) merges[".codex/hooks.json"] = r else: legacy = remove_json_hooks(codex_hooks_dst, "codex_hook.py", backups) current = remove_json_hooks(codex_hooks_dst, "casan-hook.py", backups) merges[".codex/hooks.json"] = ( "removed" if "removed" in (legacy, current) else current) # Migrate obsolete Plan-21 TOML blocks. Current Codex discovers hooks.json # directly and requires trust through /hooks; no custom [casan] keys. merges[".codex/config.toml"] = clean_legacy_codex_config( os.path.join(target, ".codex", "config.toml"), backups) vscode_file = os.path.join(target, ".vscode", "extensions.json") mark_owned_if_absent(vscode_file) vscode_merge = merge_vscode_recommendations(vscode_file, clients, backups) if vscode_merge not in ("unchanged", "absent"): created_add(vscode_file) merges[".vscode/extensions.json"] = vscode_merge vscode_install = ( install_vscode_extension(args.vscode_install) if "vscode-copilot" in clients else {"status": "skipped", "reason": "vscode-copilot_not_selected"} ) legacy_migration = ( migrate_vendored_prompt_contract(target, clients, backups) if vendored_harness else {"detected": False} ) # ── 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): relative = os.path.relpath( ci_dst, target).replace(os.sep, "/") owned_files.add(relative) newly_owned_files.add(relative) 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") created_before_domain = len(created) n = _copy_tree_missing(dom_src, dom_dst, target, created) if n: domain_files = { path.replace(os.sep, "/") for path in created[created_before_domain:] } owned_files.update(domain_files) newly_owned_files.update(domain_files) level_extras.append("apps/%s/domain (%d files)" % (project, n)) # ── manifest (so uninstall/verify know what init created) ── owned_file_hashes = { path: digest for path, digest in previous_owned_hashes.items() if path in owned_files } for relative in newly_owned_files: owned_path = _safe_project_path(target, relative) digest = _file_sha256(owned_path) if owned_path else None if digest: owned_file_hashes[relative] = digest manifest = { "manifest_version": 2, "created": created, "backups": [os.path.relpath(b, target) for b in backups], "owned_files": sorted(owned_files), "owned_file_hashes": dict(sorted(owned_file_hashes.items())), "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) result = { "status": "initialized", "project_id": project, "target": target, "target_level": lvl, "target_level_name": lvl_name, "harness_version": version, "harness_hash": hhash, "runtime_mode": runtime_mode, "runtime_path": runtime_path, "runtime_files": runtime_files, "runtime_removed": runtime_removed, "enforcement_mode": args.mode, "clients": clients, "created": created, "hook_merges": merges, "vscode_extension": vscode_install, "legacy_migration": legacy_migration, "level_extras": level_extras, "level_removed": level_removed, "level_retained": level_retained, "note": ( "managed runtime is referenced by version/hash lock" if runtime_mode == "managed" else "production-only Core runtime vendored under .casan/runtime/casan-core" ), } _emit_json_or_human(args, result, _render_init) if legacy_migration.get("manual_review"): sys.stderr.write( "casan init: WARNING — legacy CASAN prose outside managed markers " "still needs review in: %s\n" % ", ".join(legacy_migration["manual_review"])) 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): target = os.path.abspath(args.target or os.getcwd()) harness = resolve_project_harness(target, args.harness) if not harness: sys.stderr.write("casan verify-harness: cannot locate the project runtime.\n") return 1 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, } _emit_json_or_human(args, result, _render_verify) if not ok: sys.stderr.write("HARNESS_INTEGRITY_DRIFT — the resolved Core runtime does not match " "the project pin. The runtime 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_runtime_mode": cfg.get("runtime_mode", "managed"), "project_runtime_path": cfg.get("runtime_path"), "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", }, } _emit_json_or_human(args, out, _render_level) return 0 def _commands_in_hooks(path): doc = _load_json_or(path, {}) values = [] for groups in (doc.get("hooks") or {}).values(): if not isinstance(groups, list): continue for group in groups: for handler in (group.get("hooks") or []) if isinstance(group, dict) else []: if isinstance(handler, dict): values.append(str(handler.get("command", ""))) values.append(str(handler.get("commandWindows", ""))) return values def _smoke_bootstrap(target, client): bootstrap = os.path.join(target, ".casan", "casan-hook.py") event = "Begin" if client == "vscode-copilot" else "UserPromptSubmit" payload = { "cwd": target, "project": target, "session_id": "casan-doctor", "turn_id": "casan-doctor-turn", "prompt": "CASAN doctor smoke", } with tempfile.TemporaryDirectory(prefix="casan-doctor-") as tmp: env = dict(os.environ) env["CASAN_APP_ROOT"] = target env["CASAN_STATE_ROOT"] = os.path.join(tmp, ".specify") result = subprocess.run( [sys.executable, bootstrap, "--client", client, "--event", event], input=json.dumps(payload), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, cwd=target, env=env, check=False) try: response = json.loads(result.stdout.strip() or "{}") except ValueError: response = {} return { "ok": result.returncode == 0 and response.get("decision") != "block", "returncode": result.returncode, "decision": response.get("decision"), "trace_id": response.get("trace_id"), "reason": response.get("reason") or result.stderr.strip() or None, } def cmd_doctor(args): target = os.path.abspath(args.target or os.getcwd()) config_path = os.path.join(target, ".casan", "config.json") config = _load_json_or(config_path, {}) if not config: sys.stderr.write("casan doctor: project is not initialized; run `casan init`.\n") return 1 if args.client: try: clients = select_clients(args.client, False) except ValueError as error: sys.stderr.write("casan doctor: %s\n" % error) return 64 else: clients = config.get("clients") or [] harness = resolve_project_harness(target, args.harness) lock = _load_json_or(os.path.join(target, ".casan", "version.lock"), {}) expected = lock.get("harness_hash") actual, source = compute_live(harness) if harness else ("unavailable", "error") integrity_ok = bool(expected and expected == actual) bootstrap_ok = os.path.isfile(os.path.join(target, ".casan", "casan-hook.py")) checks = { "project": target, "project_id": config.get("project_id"), "enforcement_mode": config.get("enforcement_mode"), "integration_mode": config.get("integration_mode"), "clients": clients, "harness_root": harness, "integrity": { "ok": integrity_ok, "expected": expected, "actual": actual, "source": source, }, "bootstrap": {"ok": bootstrap_ok}, "client_checks": {}, "warnings": [], "required_actions": [], } ready = integrity_ok and bootstrap_ok vendored_harness = os.path.isfile(os.path.join( target, "packages", "casan-harness", "scripts", "bash", "casan-harness.sh")) legacy_conflicts = _legacy_instruction_conflicts(target) checks["legacy_migration"] = { "vendored_harness_present": vendored_harness, "vendored_harness_status": ( "retained_for_compatibility" if vendored_harness else "not_present"), "instruction_files_requiring_review": legacy_conflicts, } code = shutil.which("code") installed_extensions = set() if code: listed = subprocess.run( [code, "--list-extensions"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False) if listed.returncode == 0: installed_extensions = { line.strip().lower() for line in listed.stdout.splitlines() if line.strip()} for client in clients: surface = CLIENT_SURFACES.get(client, {}) item = { "configured": True, "supported_surfaces": list(surface.get("supported_local", [])), "not_covered_surfaces": list(surface.get("not_covered", [])), "surface_contract": surface.get("contract"), } if client == "claude": commands = _commands_in_hooks( os.path.join(target, ".claude", "settings.json")) item["hook_present"] = any( "casan-hook.py" in command and "--client claude" in command for command in commands) item["vscode_extension_installed"] = ( "anthropic.claude-code" in installed_extensions if code else None) elif client == "codex": commands = _commands_in_hooks( os.path.join(target, ".codex", "hooks.json")) item["hook_present"] = any( "casan-hook.py" in command and "--client codex" in command for command in commands) item["trust_review_required"] = True item["trust_action"] = "Open /hooks in Codex and trust the current project hook hash." item["vscode_extension_installed"] = ( "openai.chatgpt" in installed_extensions if code else None) elif client == "vscode-copilot": item["hook_present"] = bootstrap_ok item["route"] = "explicit_@casan" item["native_copilot_interception"] = False item["code_cli"] = code item["extension_installed"] = ( "fpt-casan.casan-governed-chat" in installed_extensions if code else False) if not item["extension_installed"]: item["install_action"] = ( "Re-run `casan init --client vscode-copilot " "--vscode-install yes` with the VS Code `code` CLI on PATH.") else: item["hook_present"] = False if bootstrap_ok and item.get("hook_present"): item["smoke"] = _smoke_bootstrap(target, client) else: item["smoke"] = {"ok": False, "reason": "configuration_missing"} client_ready = bool(item.get("hook_present") and item["smoke"].get("ok")) if client == "vscode-copilot": client_ready = client_ready and bool(item.get("extension_installed")) item["ready"] = client_ready ready = ready and client_ready checks["client_checks"][client] = item if "codex" in clients: checks["warnings"].append( "Codex project hooks do not run until their exact hash is reviewed and trusted via /hooks.") checks["required_actions"].append({ "code": "codex_hook_trust", "message": ( "Open /hooks in a local Codex client, review the exact hook hash, " "and trust it before treating Codex as operational."), }) if "vscode-copilot" in clients: checks["warnings"].append( "Only prompts explicitly sent to @casan use the CASAN-owned Copilot route; " "built-in Copilot chat is not globally intercepted.") if vendored_harness: checks["warnings"].append( "A legacy vendored packages/casan-harness is retained. Remove it only " "after CI, bin scripts, and domain smoke tests use the global harness.") if legacy_conflicts: checks["warnings"].append( "Legacy prompt-boundary prose outside CASAN-managed markers conflicts " "with Plan-20 hooks and needs manual review: %s." % ", ".join(legacy_conflicts)) checks["required_actions"].append({ "code": "legacy_prompt_policy_review", "message": ( "Review and migrate conflicting legacy prompt policy: %s." % ", ".join(legacy_conflicts)), }) checks["status"] = "ready" if ready else "not_ready" checks["operational_status"] = ( "not_ready" if not ready else "user_action_required" if checks["required_actions"] else "ready" ) _emit_json_or_human(args, checks, _render_doctor) return 0 if ready else 2 def _clean_integration_json(path, metadata_key, metadata_prefix, remove_if_empty): """Remove CASAN template metadata and delete a CASAN-created empty file.""" if not os.path.isfile(path): return False doc = _load_json_or(path, None) if not isinstance(doc, dict): return False metadata = doc.get(metadata_key) if isinstance(metadata, str) and metadata.startswith(metadata_prefix): doc.pop(metadata_key, None) hooks = doc.get("hooks") if isinstance(hooks, dict) and not hooks: doc.pop("hooks", None) recommendations = doc.get("recommendations") if isinstance(recommendations, list) and not recommendations: doc.pop("recommendations", None) with owner_writable(path): if not doc and remove_if_empty: os.unlink(path) else: with open(path, "w", encoding="utf-8") as fh: fh.write(json.dumps(doc, ensure_ascii=False, indent=2) + "\n") return True def _remove_owned_file(path, removed, target): if not os.path.isfile(path) and not os.path.islink(path): return with owner_writable(path): os.unlink(path) removed.append(os.path.relpath(path, target).replace(os.sep, "/")) def _prune_empty_directory(path): try: os.rmdir(path) except OSError: pass def _prune_empty_parents(path, target): parent = os.path.dirname(path) while parent and parent != target: try: os.rmdir(parent) except OSError: break parent = os.path.dirname(parent) def _remove_owned_project_artifacts(target, owned_files, owned_file_hashes): """Remove safe standalone artifacts while preserving merged/user content.""" removed = [] retained = [] integration_owned_files = { ".claude/settings.json", ".codex/hooks.json", ".codex/config.toml", ".vscode/extensions.json", ".specify/.gitignore", } for relative in sorted( owned_files, key=lambda item: (item.count("/"), item), reverse=True): if (relative in integration_owned_files or relative.startswith(".casan/")): continue path = _safe_project_path(target, relative) if not path: retained.append("%s (invalid ownership path)" % relative) continue if not os.path.isfile(path) and not os.path.islink(path): continue expected_hash = owned_file_hashes.get(relative) current_hash = _file_sha256(path) casan_namespaced_workflow = ( relative.startswith(".gitea/workflows/casan-") and relative.endswith((".yml", ".yaml"))) if casan_namespaced_workflow or ( expected_hash and current_hash == expected_hash): _remove_owned_file(path, removed, target) _prune_empty_parents(path, target) else: retained.append( "%s (modified or legacy file; preserved)" % relative) return removed, retained def _uninstall_vscode_extension(enabled): if not enabled: return { "status": "retained_shared", "reason": "use --remove-vscode-extension to remove it from this machine", } code = shutil.which("code") if not code: return {"status": "not_removed", "reason": "code_cli_not_found"} result = subprocess.run( [code, "--uninstall-extension", "fpt-casan.casan-governed-chat"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False) return { "status": "uninstalled" if result.returncode == 0 else "failed", "command": code, "detail": (result.stdout or result.stderr).strip(), } def cmd_uninstall(args): """Remove one project's CASAN adoption without clobbering user config.""" target = os.path.abspath(args.target or os.getcwd()) config_path = os.path.join(target, ".casan", "config.json") manifest_path = os.path.join(target, ".casan", "init-manifest.json") config = _load_json_or(config_path, {}) manifest = _load_json_or(manifest_path, {}) specify_gitignore = os.path.join(target, ".specify", ".gitignore") try: with open(specify_gitignore, "r", encoding="utf-8") as fh: has_casan_evidence = fh.read().startswith( "# CASAN runtime state — do not commit") except OSError: has_casan_evidence = False if (not config and not manifest and not (args.purge and has_casan_evidence)): sys.stderr.write( "casan uninstall: project is not initialized; nothing to remove.\n") return 1 created = { str(path).replace("\\", "/") for path in manifest.get("created", []) if isinstance(path, str) } owned_files = { str(path).replace("\\", "/") for path in manifest.get("owned_files", []) if isinstance(path, str) } owned_file_hashes = { str(path).replace("\\", "/"): digest for path, digest in manifest.get("owned_file_hashes", {}).items() if isinstance(path, str) and isinstance(digest, str) } hook_changes = {} integration_files = ( (".claude/settings.json", "claude_hook.py", "casan-hook.py"), (".codex/hooks.json", "codex_hook.py", "casan-hook.py"), ) for relative, legacy_marker, current_marker in integration_files: path = os.path.join(target, *relative.split("/")) legacy = remove_json_hooks(path, legacy_marker, None) current = remove_json_hooks(path, current_marker, None) status = "removed" if "removed" in (legacy, current) else current hook_changes[relative] = status codex_config = os.path.join(target, ".codex", "config.toml") hook_changes[".codex/config.toml"] = clean_legacy_codex_config( codex_config, None) vscode_path = os.path.join(target, ".vscode", "extensions.json") hook_changes[".vscode/extensions.json"] = merge_vscode_recommendations( vscode_path, [], None) _clean_integration_json( os.path.join(target, ".claude", "settings.json"), "//", "CASAN Plan-20", ".claude/settings.json" in owned_files) _clean_integration_json( os.path.join(target, ".codex", "hooks.json"), "description", "CASAN Plan-20", ".codex/hooks.json" in owned_files) _clean_integration_json( vscode_path, "_casan_unused", "", ".vscode/extensions.json" in owned_files) removed, retained = _remove_owned_project_artifacts( target, owned_files, owned_file_hashes) if _remove_vendored_core(target): removed.append(".casan/runtime/casan-core") for relative in ( ".casan/config.json", ".casan/version.lock", ".casan/agentic.env", ".casan/casan-hook.py", ".casan/init-manifest.json"): _remove_owned_file( os.path.join(target, *relative.split("/")), removed, target) if args.purge: for relative in (".specify/logs", ".specify/state"): path = os.path.join(target, *relative.split("/")) if os.path.isdir(path): with owner_writable(path): shutil.rmtree(path) removed.append(relative) if has_casan_evidence: _remove_owned_file(specify_gitignore, removed, target) for relative in ( ".casan", ".claude", ".codex", ".vscode", ".specify"): _prune_empty_directory(os.path.join(target, relative)) if not args.purge and os.path.isdir(os.path.join(target, ".specify")): retained.append(".specify/ runtime evidence (use --purge to remove)") backups = manifest.get("backups", []) if backups: retained.append("%d .casan-bak backup(s)" % len(backups)) result = { "status": "uninstalled", "project_id": config.get("project_id") or manifest.get("project_id"), "target": target, "hook_changes": hook_changes, "removed": removed, "purged": bool(args.purge), "vscode_extension": _uninstall_vscode_extension( args.remove_vscode_extension), "retained": retained, } _emit_json_or_human(args, result, _render_uninstall) return 0 def main(argv=None): parser = argparse.ArgumentParser( prog="casan-init", description="Adopt CASAN governance into an existing project") sub = parser.add_subparsers(dest="cmd") pi = sub.add_parser( "init", help="adopt or reconfigure CASAN in the current project", description=( "Adopt CASAN governance into an existing project. New interactive " "projects are guided through runtime and client selection; existing " "projects preserve their current runtime unless --runtime is set."), epilog=( "examples:\n" " casan init\n" " casan init --runtime managed --client claude,codex\n" " casan init --runtime vendored --client codex\n" " casan init --non-interactive --runtime managed --client none"), formatter_class=argparse.RawDescriptionHelpFormatter) 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", action="append", help=("integration to enable; repeat or comma-separate: claude, codex, " "vscode-copilot, all, none. Interactive init shows a menu.")) pi.add_argument( "--non-interactive", action="store_true", help=("do not prompt; new projects default to managed runtime and the " "backward-compatible claude,codex client set")) pi.add_argument( "--vscode-install", choices=["auto", "yes", "no"], default="auto", help=("install the local CASAN @casan VSIX when vscode-copilot is selected " "(default auto: install when `code` is available)")) pi.add_argument("--level", default="core", help="packaging level to adopt: 1|core (default), 2|devkit, 3|platform (preview), 4|enterprise (refused)") pi.add_argument( "--runtime", choices=["managed", "vendored"], help=("Core runtime placement: managed uses the pinned global install " "(default for new adoption); vendored copies a production-only " "Core into the project; re-init preserves the current mode")) pi.add_argument("--mode", choices=["observe", "enforce"], default="enforce", help="agentic policy mode (default: enforce; use observe for a telemetry-only pilot)") 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") pi.add_argument("--json", action="store_true", help="emit the complete machine-readable result") 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") pv.add_argument("--json", action="store_true", help="emit the complete machine-readable result") 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") pl.add_argument("--json", action="store_true", help="emit the complete machine-readable result") pd = sub.add_parser("doctor", help="verify selected client integrations end-to-end") pd.add_argument("--target", help="project root (default: cwd)") pd.add_argument("--client", action="append", help="client(s) to check; default: clients enabled in config") pd.add_argument("--harness", help="override harness root") pd.add_argument("--json", action="store_true", help="emit the complete machine-readable result") pu = sub.add_parser( "uninstall", help="remove CASAN from a project while preserving user configuration") pu.add_argument("--target", help="project root (default: cwd)") pu.add_argument( "--purge", action="store_true", help="also remove .specify runtime logs and state") pu.add_argument( "--remove-vscode-extension", action="store_true", help="also uninstall the shared @casan VS Code extension from this machine") pu.add_argument("--json", action="store_true", help="emit the complete machine-readable result") args = parser.parse_args(argv) if args.cmd == "init": try: return cmd_init(args) except PermissionError as error: sys.stderr.write( "casan init: permission denied while merging project config: " "%s\n" % error) return 77 if args.cmd == "verify": return cmd_verify(args) if args.cmd == "level": return cmd_level(args) if args.cmd == "doctor": return cmd_doctor(args) if args.cmd == "uninstall": return cmd_uninstall(args) parser.print_help() return 64 if __name__ == "__main__": sys.exit(main())