1958 lines
77 KiB
Python
Executable File
1958 lines
77 KiB
Python
Executable File
#!/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 = "<!-- CASAN_PROMPT_ENFORCEMENT_START -->"
|
||
PROMPT_MARKER_END = "<!-- CASAN_PROMPT_ENFORCEMENT_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",
|
||
},
|
||
}
|
||
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"
|
||
_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))
|
||
print()
|
||
status = _color("1;32", "READY") if ready else _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):
|
||
"""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 are replaced so upgrades cannot leave a stale command 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)
|
||
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_PROMPT_ENFORCEMENT_START -->
|
||
## 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.
|
||
<!-- CASAN_PROMPT_ENFORCEMENT_END -->""" % 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)
|
||
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": [],
|
||
}
|
||
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.")
|
||
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["status"] = "ready" if ready else "not_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())
|