Soul Guardian

# Exclude local caches and build outputs from ClawHub upload
.DS_Store
.git/
__pycache__/
*.pyc
*.pyo
*.egg-info/
dist/
build/
.env
.venv/
.cache/

#!/usr/bin/env python3
"""Generate (and optionally install) a macOS launchd plist for soul-guardian.

Goal:
- Run `soul_guardian.py check` on an interval.
- Be *silent on OK* (soul_guardian.py prints nothing + exits 0 when no drift).
- Produce a single-line stdout alert on drift (exits 2 and prints SOUL_GUARDIAN_DRIFT ...).

This script is intentionally deterministic and dependency-free.

It does NOT attempt to deliver drift alerts to Telegram/Slack/etc.
Instead it:
- writes logs to the state dir (so drift output is preserved)
- relies on you to wire notifications however you prefer

If you want OpenClaw-side delivery, use OpenClaw cron.
"""

from __future__ import annotations

import argparse
import os
from pathlib import Path
import plistlib
import subprocess
import sys


LEGACY_STATE_ROOT = Path("~/.clawdbot/soul-guardian").expanduser()
DEFAULT_STATE_ROOT = Path("~/.openclaw/soul-guardian").expanduser()
LEGACY_LABEL_PREFIX = "com.clawdbot.soul-guardian."
DEFAULT_LABEL_PREFIX = "com.openclaw.soul-guardian."


def agent_id_default(workspace_root: Path) -> str:
    return workspace_root.name


def legacy_label(agent_id: str) -> str:
    return f"{LEGACY_LABEL_PREFIX}{agent_id}"


def default_label(agent_id: str) -> str:
    return f"{DEFAULT_LABEL_PREFIX}{agent_id}"


def legacy_plist_path(agent_id: str) -> Path:
    return Path("~/Library/LaunchAgents").expanduser() / f"{legacy_label(agent_id)}.plist"


def default_external_state_dir(agent_id: str) -> tuple[Path, bool]:
    legacy_state_dir = LEGACY_STATE_ROOT / agent_id
    if legacy_state_dir.exists():
        return legacy_state_dir, True
    return DEFAULT_STATE_ROOT / agent_id, False


def run_launchctl(args: list[str]) -> subprocess.CompletedProcess[str]:
    return subprocess.run(["/bin/launchctl", *args], check=False, text=True, capture_output=True)


def cleanup_legacy_launchd(uid: int, active_label: str, agent_id: str) -> list[str]:
    legacy_job_label = legacy_label(agent_id)
    legacy_job_plist = legacy_plist_path(agent_id).expanduser().resolve()
    if active_label == legacy_job_label:
        return []

    cleanup_commands: list[tuple[list[str], str]] = [
        (
            ["disable", f"gui/{uid}/{legacy_job_label}"],
            f"launchctl disable gui/{uid}/{legacy_job_label}",
        ),
        (
            ["bootout", f"gui/{uid}/{legacy_job_label}"],
            f"launchctl bootout gui/{uid}/{legacy_job_label}",
        ),
    ]

    if legacy_job_plist.exists():
        cleanup_commands.append(
            (
                ["bootout", f"gui/{uid}", str(legacy_job_plist)],
                f"launchctl bootout gui/{uid} {legacy_job_plist}",
            )
        )

    failed_commands: list[str] = []
    for args, display_cmd in cleanup_commands:
        cp = run_launchctl(args)
        if cp.returncode != 0 and legacy_job_plist.exists():
            failed_commands.append(display_cmd)

    if not failed_commands:
        return []

    warning_lines = [
        "WARNING: Failed to fully clean up the legacy soul-guardian launchd job "
        f"{legacy_job_label}.",
        f"Manually run: launchctl bootout gui/{uid} {legacy_job_label}",
    ]
    if legacy_job_plist.exists():
        warning_lines.append(f"If needed, also remove the legacy plist: {legacy_job_plist}")
    warning_lines.append("You can rerun this installer after the legacy job is removed.")
    return warning_lines


def main(argv: list[str]) -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument(
        "--workspace-root",
        default=str(Path.cwd()),
        help="Workspace root (default: current working directory).",
    )
    ap.add_argument(
        "--agent-id",
        default=None,
        help="Agent/workspace identifier used in default label + state dir (default: workspace folder name).",
    )
    ap.add_argument(
        "--state-dir",
        default=None,
        help="External state directory (recommended). Default: ~/.openclaw/soul-guardian/<agentId>/; reuses ~/.clawdbot/soul-guardian/<agentId>/ if that legacy state dir already exists.",
    )
    ap.add_argument(
        "--label",
        default=None,
        help="launchd label (default: com.openclaw.soul-guardian.<agentId>). When using a non-legacy label, --install attempts to disable/boot out the previous com.clawdbot.soul-guardian.<agentId> job first.",
    )
    ap.add_argument(
        "--interval-seconds",
        type=int,
        default=600,
        help="Run interval in seconds (StartInterval). Default: 600 (10 minutes).",
    )
    ap.add_argument("--actor", default="cron", help="--actor passed to soul_guardian.py (default: cron).")
    ap.add_argument("--note", default="launchd", help="--note passed to soul_guardian.py (default: launchd).")
    ap.add_argument(
        "--out",
        default=None,
        help="Write plist to this path (default: ~/Library/LaunchAgents/<label>.plist)",
    )
    ap.add_argument("--force", action="store_true", help="Overwrite existing plist on disk.")
    ap.add_argument(
        "--install",
        action="store_true",
        help="Install+load the plist with launchctl (bootstrap). Without this flag we only write the plist.",
    )

    args = ap.parse_args(argv)

    workspace_root = Path(args.workspace_root).expanduser().resolve()
    agent_id = args.agent_id or agent_id_default(workspace_root)
    if args.state_dir:
        state_dir = Path(args.state_dir).expanduser().resolve()
    else:
        state_dir, using_legacy_state_dir = default_external_state_dir(agent_id)
        state_dir = state_dir.resolve()
        if using_legacy_state_dir:
            migration_target = (DEFAULT_STATE_ROOT / agent_id).resolve()
            print(
                "WARNING: Detected legacy soul-guardian state dir at "
                f"{state_dir}. Using it for backward compatibility. "
                "To switch to the new default location, rerun this script with "
                f"--state-dir {migration_target}",
                file=sys.stderr,
            )

    label = args.label or default_label(agent_id)
    legacy_job_label = legacy_label(agent_id)
    legacy_job_plist = legacy_plist_path(agent_id).expanduser().resolve()
    plist_path = Path(args.out).expanduser().resolve() if args.out else (Path("~/Library/LaunchAgents").expanduser() / f"{label}.plist")

    script_path = workspace_root / "skills" / "soul-guardian" / "scripts" / "soul_guardian.py"
    if not script_path.exists():
        raise SystemExit(f"soul_guardian.py not found at {script_path}; pass --workspace-root correctly")

    # Keep logs in the external state dir.
    log_dir = state_dir / "logs"
    log_dir.mkdir(parents=True, exist_ok=True)
    stdout_log = log_dir / "launchd.stdout.log"
    stderr_log = log_dir / "launchd.stderr.log"

    program_args = [
        "/usr/bin/python3",
        str(script_path),
        "--state-dir",
        str(state_dir),
        "check",
        "--actor",
        str(args.actor),
        "--note",
        str(args.note),
    ]

    plist: dict[str, object] = {
        "Label": label,
        "ProgramArguments": program_args,
        "WorkingDirectory": str(workspace_root),
        "StartInterval": int(args.interval_seconds),
        "RunAtLoad": True,
        "StandardOutPath": str(stdout_log),
        "StandardErrorPath": str(stderr_log),
        # Avoid interactive UI dependencies; run in background.
        "ProcessType": "Background",
    }

    plist_path.parent.mkdir(parents=True, exist_ok=True)

    if plist_path.exists() and not args.force:
        raise SystemExit(f"Refusing to overwrite existing {plist_path}. Re-run with --force.")

    with plist_path.open("wb") as f:
        plistlib.dump(plist, f, fmt=plistlib.FMT_XML, sort_keys=True)

    print(f"Wrote plist: {plist_path}")
    print(f"State dir:  {state_dir}")
    print(f"Label:      {label}")
    if label == legacy_job_label:
        print("Legacy label mode: cleanup is skipped because the selected label matches the previous Clawdbot-era default.")
    else:
        print(f"Legacy label:      {legacy_job_label}")
        print(f"Legacy plist:      {legacy_job_plist}")
        if args.install:
            print("Migration: install mode will try to disable/boot out the legacy launchd job before starting the new label.")
        else:
            print("Dry run: --install will try to disable/boot out the legacy launchd job before starting the new label.")

    uid = os.getuid()

    if args.install:
        for warning_line in cleanup_legacy_launchd(uid, label, agent_id):
            print(warning_line, file=sys.stderr)

        # Best-effort: remove any existing job with same label, then bootstrap.
        run_launchctl(["bootout", f"gui/{uid}", label])
        run_launchctl(["bootout", f"gui/{uid}", str(plist_path)])
        res = subprocess.run(["/bin/launchctl", "bootstrap", f"gui/{uid}", str(plist_path)], text=True, capture_output=True)
        if res.returncode != 0:
            sys.stderr.write((res.stderr or res.stdout or "").strip() + "\n")
            sys.stderr.write("Failed to bootstrap. You can try manually:\n")
            sys.stderr.write(f"  launchctl bootstrap gui/{uid} {plist_path}\n")
            return 1

        subprocess.run(["/bin/launchctl", "enable", f"gui/{uid}/{label}"], check=False)
        subprocess.run(["/bin/launchctl", "kickstart", "-k", f"gui/{uid}/{label}"], check=False)
        print("Installed + started (launchctl bootstrap/enable/kickstart).")
    else:
        print("Not installed (dry write). To load it:")
        print(f"  launchctl bootstrap gui/{uid} {plist_path}")
        print(f"  launchctl enable gui/{uid}/{label}")
        print(f"  launchctl kickstart -k gui/{uid}/{label}")

    print("\nLogs:")
    print(f"  tail -n 200 -f {stdout_log}")
    print(f"  tail -n 200 -f {stderr_log}")

    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))

#!/usr/bin/env python3
"""Onboard soul-guardian state directory outside the workspace.

Why:
- Keeping integrity state inside the workspace can be risky if the workspace is modified or wiped.
- Moving state to an external directory improves resilience and makes tampering harder.

What this script does:
- Creates an external state directory (default: ~/.openclaw/soul-guardian/<agentId>/)
- Copies (or moves) existing in-workspace state from memory/soul-guardian/
- Writes a default policy.json if missing
- Prints recommended cron snippets (OpenClaw cron and optional launchd)

This script does NOT modify your cron jobs automatically.
"""

from __future__ import annotations

import argparse
import os
from pathlib import Path
import shutil
import sys


WORKSPACE_ROOT = Path.cwd()
DEFAULT_IN_WORKSPACE_STATE = WORKSPACE_ROOT / "memory" / "soul-guardian"


def agent_id_default() -> str:
    # Best-effort: workspace folder name.
    return WORKSPACE_ROOT.name


def ensure_dir(p: Path) -> None:
    p.mkdir(parents=True, exist_ok=True)


def copytree_overwrite(src: Path, dst: Path) -> None:
    # Copy directory contents into dst (merge).
    ensure_dir(dst)
    for root, dirs, files in os.walk(src):
        r = Path(root)
        rel = r.relative_to(src)
        target_root = dst / rel
        ensure_dir(target_root)
        for d in dirs:
            ensure_dir(target_root / d)
        for f in files:
            s = r / f
            t = target_root / f
            # Overwrite.
            shutil.copy2(s, t)


DEFAULT_POLICY_JSON = """{
  "version": 1,
  "workspaceRoot": "",
  "targets": [
    {"path": "SOUL.md", "mode": "restore"},
    {"path": "AGENTS.md", "mode": "restore"},
    {"path": "USER.md", "mode": "alert"},
    {"path": "TOOLS.md", "mode": "alert"},
    {"path": "IDENTITY.md", "mode": "alert"},
    {"path": "HEARTBEAT.md", "mode": "alert"},
    {"path": "MEMORY.md", "mode": "alert"},
    {"pattern": "memory/*.md", "mode": "ignore"}
  ]
}
"""


def main(argv: list[str]) -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--agent-id", default=agent_id_default(), help="Identifier used for default external state path.")
    ap.add_argument(
        "--state-dir",
        default=None,
        help="External state directory to create/use (default: ~/.openclaw/soul-guardian/<agentId>/).",
    )
    ap.add_argument("--move", action="store_true", help="Move instead of copy (WARNING: deletes the old in-workspace state dir).")
    ap.add_argument("--no-copy", action="store_true", help="Do not copy/move existing in-workspace state.")
    args = ap.parse_args(argv)

    if args.state_dir:
        external = Path(args.state_dir).expanduser()
    else:
        external = (Path("~/.openclaw/soul-guardian").expanduser() / args.agent_id)

    ensure_dir(external)

    if not args.no_copy and DEFAULT_IN_WORKSPACE_STATE.exists():
        if args.move:
            # Move by copying then removing src (safer than rename across filesystems).
            copytree_overwrite(DEFAULT_IN_WORKSPACE_STATE, external)
            shutil.rmtree(DEFAULT_IN_WORKSPACE_STATE)
            action = "moved"
        else:
            copytree_overwrite(DEFAULT_IN_WORKSPACE_STATE, external)
            action = "copied"
        print(f"Existing state {action} from {DEFAULT_IN_WORKSPACE_STATE} -> {external}")
    else:
        print(f"Using external state dir: {external}")

    policy_path = external / "policy.json"
    if not policy_path.exists():
        txt = DEFAULT_POLICY_JSON.replace('"workspaceRoot": ""', f'"workspaceRoot": "{WORKSPACE_ROOT}"')
        policy_path.write_text(txt, encoding="utf-8")
        print(f"Wrote default policy: {policy_path}")
    else:
        print(f"Policy already exists: {policy_path}")

    print("\nNext steps")
    print("1) Initialize baselines in the external state dir:")
    print(
        f"   cd '{WORKSPACE_ROOT}' && python3 skills/soul-guardian/scripts/soul_guardian.py --state-dir '{external}' init --actor 'sam' --note 'onboard external state'\n"
    )

    print("2) Update your cron/check runner to include --state-dir.")
    print("\nOpenClaw cron (recommended; does not require system cron):")
    print("- In your cron spec, run something like:")
    print(
        f"  cd '{WORKSPACE_ROOT}' && python3 skills/soul-guardian/scripts/soul_guardian.py --state-dir '{external}' check --actor system --note cron"
    )

    print("\nOptional: system cron / launchd (macOS) example (NOT installed automatically):")
    label = f"com.openclaw.soul-guardian.{args.agent_id}"
    print(f"- Launchd label: {label}")
    print(f"- WorkingDirectory (recommended): {WORKSPACE_ROOT}")
    print("- ProgramArguments (example):")
    print("  [\n"
          f"    '/usr/bin/python3',\n"
          f"    '{WORKSPACE_ROOT}/skills/soul-guardian/scripts/soul_guardian.py',\n"
          f"    '--state-dir', '{external}',\n"
          f"    'check', '--actor', 'system', '--note', 'launchd'\n"
          "  ]")

    print("\nNotes")
    print("- The external state dir can contain approved snapshots, patches, and quarantined copies of drifted prompt/instruction files; keep permissions restrictive (e.g., chmod 700; go-rwx).")
    if args.move:
        print("- WARNING: --move deletes the old in-workspace state dir after copying.")
    print("- Consider mirroring the external state dir via git or offsite backups (not enforced by this tool).")

    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))

#!/usr/bin/env python3
"""Workspace file integrity guard + audit (multi-file).

This is a hardened successor to the original SOUL.md-only guardian.

Key features:
- Multiple target files with per-file policy (restore | alert | ignore)
- Approved baselines stored per file (snapshot + sha256)
- Append-only audit log with hash chaining (tamper-evident)
- Optional auto-restore for restore-mode files (with quarantine copy)
- Refuses to operate on symlinks
- Atomic writes for baseline + restore operations (os.replace)

State directory (default, backward-compatible): memory/soul-guardian/

Subcommands:
- init            Initialize policy + baselines (first run)
- status          Print status JSON for all policy targets
- check           Check for drift; restore for restore-mode by default
- approve         Approve current contents as baseline (per file or all)
- restore         Restore restore-mode files to last approved baseline
- verify-audit    Validate audit log hash chain

Exit codes:
- 0: ok / no drift
- 2: drift detected (for check when any alert/restore drift happened)
- 1: error
"""

from __future__ import annotations

import argparse
import datetime as dt
import difflib
import fnmatch
import hashlib
import json
import os
from pathlib import Path
import shutil
import stat
import sys
from typing import Any, Iterable


WORKSPACE_ROOT = Path.cwd()
DEFAULT_STATE_DIR = WORKSPACE_ROOT / "memory" / "soul-guardian"

POLICY_FILE = "policy.json"
BASELINES_FILE = "baselines.json"
AUDIT_LOG_FILE = "audit.jsonl"
APPROVED_DIRNAME = "approved"
PATCH_DIRNAME = "patches"
QUAR_DIRNAME = "quarantine"

CHAIN_GENESIS = "0" * 64


def utc_now_iso() -> str:
    return dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat()


def sha256_bytes(b: bytes) -> str:
    h = hashlib.sha256()
    h.update(b)
    return h.hexdigest()


def sha256_text(s: str) -> str:
    return sha256_bytes(s.encode("utf-8"))


def read_bytes(path: Path) -> bytes:
    return path.read_bytes()


def read_text(path: Path) -> str:
    return path.read_text(encoding="utf-8", errors="replace")


def is_symlink(path: Path) -> bool:
    try:
        st = os.lstat(path)
    except FileNotFoundError:
        return False
    return stat.S_ISLNK(st.st_mode)


def ensure_dir(path: Path) -> None:
    path.mkdir(parents=True, exist_ok=True)


def atomic_write_bytes(path: Path, data: bytes) -> None:
    ensure_dir(path.parent)
    tmp = path.with_name(path.name + ".tmp")
    with tmp.open("wb") as f:
        f.write(data)
        f.flush()
        os.fsync(f.fileno())
    os.replace(tmp, path)


def atomic_write_text(path: Path, text: str) -> None:
    atomic_write_bytes(path, text.encode("utf-8"))


def unified_diff_text(old: str, new: str, fromfile: str, tofile: str) -> str:
    old_lines = old.splitlines(keepends=True)
    new_lines = new.splitlines(keepends=True)
    diff = difflib.unified_diff(old_lines, new_lines, fromfile=fromfile, tofile=tofile)
    return "".join(diff)


def safe_patch_tag(tag: str) -> str:
    return ("".join(c for c in tag if c.isalnum() or c in ("-", "_"))[:40] or "patch")


def relpath_str(path: Path, root: Path) -> str:
    # Normalize to a stable forward-slash relative string.
    try:
        rel = path.relative_to(root)
    except Exception:
        rel = Path(os.path.relpath(path, root))
    return rel.as_posix()


class GuardianState:
    def __init__(self, state_dir: Path):
        self.state_dir = state_dir
        self.policy_path = state_dir / POLICY_FILE
        self.baselines_path = state_dir / BASELINES_FILE
        self.audit_path = state_dir / AUDIT_LOG_FILE
        self.approved_dir = state_dir / APPROVED_DIRNAME
        self.patch_dir = state_dir / PATCH_DIRNAME
        self.quarantine_dir = state_dir / QUAR_DIRNAME

    def ensure_dirs(self) -> None:
        ensure_dir(self.state_dir)
        ensure_dir(self.approved_dir)
        ensure_dir(self.patch_dir)
        ensure_dir(self.quarantine_dir)


def default_policy() -> dict[str, Any]:
    # Default protected set, per requirements.
    return {
        "version": 1,
        "workspaceRoot": str(WORKSPACE_ROOT),
        "targets": [
            {"path": "SOUL.md", "mode": "restore"},
            {"path": "AGENTS.md", "mode": "restore"},
            {"path": "USER.md", "mode": "alert"},
            {"path": "TOOLS.md", "mode": "alert"},
            {"path": "IDENTITY.md", "mode": "alert"},
            {"path": "HEARTBEAT.md", "mode": "alert"},
            {"path": "MEMORY.md", "mode": "alert"},
            # Ignore daily notes by default.
            {"pattern": "memory/*.md", "mode": "ignore"},
        ],
    }


def load_policy(state: GuardianState) -> dict[str, Any]:
    if not state.policy_path.exists():
        return default_policy()
    return json.loads(state.policy_path.read_text(encoding="utf-8"))


def save_policy(state: GuardianState, policy: dict[str, Any]) -> None:
    state.ensure_dirs()
    atomic_write_text(state.policy_path, json.dumps(policy, ensure_ascii=False, indent=2) + "\n")


def load_baselines(state: GuardianState) -> dict[str, Any]:
    """Load baselines.json.

    Backward-compat:
    - If baselines.json doesn't exist but legacy SOUL.md baseline exists
      (approved.sha256 + approved/SOUL.md), import it into the in-memory baselines.
      The caller will persist it on the next save.
    """

    if state.baselines_path.exists():
        return json.loads(state.baselines_path.read_text(encoding="utf-8"))

    baselines: dict[str, Any] = {"version": 1, "files": {}}

    legacy_sha = state.state_dir / "approved.sha256"
    legacy_snap = state.approved_dir / "SOUL.md"
    if legacy_sha.exists() and legacy_snap.exists():
        sha = legacy_sha.read_text(encoding="utf-8").strip()
        if sha:
            baselines["files"]["SOUL.md"] = {"sha256": sha, "approvedAt": "legacy"}

    return baselines


def save_baselines(state: GuardianState, baselines: dict[str, Any]) -> None:
    state.ensure_dirs()
    atomic_write_text(state.baselines_path, json.dumps(baselines, ensure_ascii=False, indent=2, sort_keys=True) + "\n")


def resolve_targets(policy: dict[str, Any], root: Path) -> list[dict[str, str]]:
    """Return list of effective targets to consider.

    For entries with {path, mode}: direct file.
    For entries with {pattern, mode}: expands via globbing relative to root.

    Note: We keep it simple and only expand within workspace root.
    """

    targets: list[dict[str, str]] = []
    entries = policy.get("targets", [])

    for ent in entries:
        mode = ent.get("mode")
        if mode not in ("restore", "alert", "ignore"):
            continue

        if "path" in ent:
            p = Path(ent["path"])
            targets.append({"path": p.as_posix(), "mode": mode})
            continue

        pat = ent.get("pattern")
        if not pat:
            continue

        # Expand pattern relative to root.
        # Using glob keeps it bounded to workspace.
        for match in root.glob(pat):
            if match.is_dir():
                continue
            rel = relpath_str(match, root)
            targets.append({"path": rel, "mode": mode})

    # De-dup by path keeping the last specified mode.
    dedup: dict[str, str] = {}
    for t in targets:
        dedup[t["path"]] = t["mode"]
    return [{"path": p, "mode": m} for p, m in sorted(dedup.items())]


def policy_mode_for_path(policy: dict[str, Any], rel_path: str) -> str | None:
    # Direct match has priority; then patterns.
    entries = policy.get("targets", [])

    for ent in entries:
        if ent.get("path") == rel_path:
            return ent.get("mode")

    for ent in entries:
        pat = ent.get("pattern")
        if not pat:
            continue
        if fnmatch.fnmatch(rel_path, pat):
            return ent.get("mode")

    return None


def approved_snapshot_path(state: GuardianState, rel_path: str) -> Path:
    # Preserve relative structure under approved/.
    return state.approved_dir / Path(rel_path)


def write_patch(state: GuardianState, patch_text: str, tag: str, rel_path: str) -> Path:
    state.ensure_dirs()
    ts = dt.datetime.now(dt.timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    path_tag = safe_patch_tag(tag)
    file_tag = safe_patch_tag(rel_path.replace("/", "_"))
    path = state.patch_dir / f"{ts}-{file_tag}-{path_tag}.patch"
    atomic_write_text(path, patch_text)
    return path


def _canonical_json(obj: Any) -> str:
    # Stable serialization for hashing.
    return json.dumps(obj, ensure_ascii=False, sort_keys=True, separators=(",", ":"))


def _audit_needs_upgrade(state: GuardianState) -> bool:
    """Detect legacy audit logs that lack a chain field."""
    if not state.audit_path.exists():
        return False
    try:
        with state.audit_path.open("r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                rec = json.loads(line)
                return "chain" not in rec
    except Exception:
        # If unreadable, force rotation so we can proceed safely.
        return True
    return False


def _rotate_legacy_audit(state: GuardianState) -> None:
    if not state.audit_path.exists():
        return
    ts = dt.datetime.now(dt.timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    legacy = state.state_dir / f"audit.legacy.{ts}.jsonl"
    os.replace(state.audit_path, legacy)


def _last_audit_hash(state: GuardianState) -> str:
    if not state.audit_path.exists():
        return CHAIN_GENESIS

    # Read last non-empty line without loading huge files.
    with state.audit_path.open("rb") as f:
        f.seek(0, os.SEEK_END)
        size = f.tell()
        if size == 0:
            return CHAIN_GENESIS
        block = 65536
        start = max(0, size - block)
        f.seek(start)
        data = f.read()

    lines = [ln for ln in data.splitlines() if ln.strip()]
    if not lines:
        return CHAIN_GENESIS

    last = lines[-1]
    try:
        rec = json.loads(last.decode("utf-8"))
        return rec.get("chain", {}).get("hash") or CHAIN_GENESIS
    except Exception:
        return CHAIN_GENESIS


def append_audit(state: GuardianState, entry: dict[str, Any]) -> None:
    """Append an audit entry with hash chaining.

    Each record includes: chain.prev, chain.hash
    chain.hash = sha256(prev_hash + "\n" + canonical_json(entry_without_chain))

    Backward-compat: if an existing audit.jsonl doesn't contain chain fields
    (legacy v1 logs), rotate it aside and start a new chained log.
    """

    state.ensure_dirs()

    if _audit_needs_upgrade(state):
        _rotate_legacy_audit(state)

    prev = _last_audit_hash(state)

    entry_wo_chain = dict(entry)
    entry_wo_chain.pop("chain", None)

    payload = prev + "\n" + _canonical_json(entry_wo_chain)
    cur = sha256_text(payload)

    record = dict(entry_wo_chain)
    record["chain"] = {"prev": prev, "hash": cur}

    with state.audit_path.open("a", encoding="utf-8") as f:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")


def refuse_symlink(path: Path) -> None:
    if is_symlink(path):
        raise RuntimeError(f"Refusing to operate on symlink: {path}")


def compute_file_sha(path: Path) -> str:
    return sha256_bytes(read_bytes(path))


def baseline_info_for(state: GuardianState, baselines: dict[str, Any], rel_path: str) -> dict[str, Any] | None:
    return (baselines.get("files") or {}).get(rel_path)


def set_baseline_for(state: GuardianState, baselines: dict[str, Any], rel_path: str, sha: str) -> None:
    baselines.setdefault("files", {})[rel_path] = {
        "sha256": sha,
        "approvedAt": utc_now_iso(),
    }


def init_cmd(state: GuardianState, actor: str, note: str, *, force_policy: bool = False) -> None:
    state.ensure_dirs()

    if force_policy or not state.policy_path.exists():
        save_policy(state, default_policy())

    policy = load_policy(state)
    baselines = load_baselines(state)

    targets = resolve_targets(policy, WORKSPACE_ROOT)

    initialized_any = False
    for t in targets:
        relp = t["path"]
        mode = t["mode"]
        if mode == "ignore":
            continue

        abs_path = WORKSPACE_ROOT / relp
        if not abs_path.exists():
            continue
        refuse_symlink(abs_path)

        # If already has baseline, do not overwrite.
        if baseline_info_for(state, baselines, relp) is not None and approved_snapshot_path(state, relp).exists():
            continue

        sha = compute_file_sha(abs_path)

        # Snapshot.
        snap = approved_snapshot_path(state, relp)
        ensure_dir(snap.parent)
        atomic_write_bytes(snap, read_bytes(abs_path))

        set_baseline_for(state, baselines, relp, sha)
        initialized_any = True

        append_audit(state, {
            "ts": utc_now_iso(),
            "event": "init",
            "actor": actor,
            "note": note,
            "path": relp,
            "mode": mode,
            "approvedSha": sha,
            "workspace": str(WORKSPACE_ROOT),
            "stateDir": str(state.state_dir),
        })

    save_baselines(state, baselines)

    if initialized_any:
        print(f"Initialized baselines in {state.state_dir}")
    else:
        print("Already initialized (no new baselines created).")


def status_cmd(state: GuardianState) -> None:
    state.ensure_dirs()
    policy = load_policy(state)
    baselines = load_baselines(state)

    targets = resolve_targets(policy, WORKSPACE_ROOT)
    out: dict[str, Any] = {
        "workspace": str(WORKSPACE_ROOT),
        "stateDir": str(state.state_dir),
        "policyPath": str(state.policy_path),
        "baselinesPath": str(state.baselines_path),
        "auditLog": str(state.audit_path),
        "files": [],
    }

    for t in targets:
        relp = t["path"]
        mode = t["mode"]
        abs_path = WORKSPACE_ROOT / relp

        baseline = baseline_info_for(state, baselines, relp)
        approved_sha = baseline.get("sha256") if baseline else None
        approved_snap = approved_snapshot_path(state, relp)

        current_sha = None
        if abs_path.exists() and not is_symlink(abs_path):
            try:
                current_sha = compute_file_sha(abs_path)
            except Exception:
                current_sha = None

        ok = (mode == "ignore") or (approved_sha is not None and current_sha == approved_sha)

        out["files"].append({
            "path": relp,
            "mode": mode,
            "exists": abs_path.exists(),
            "isSymlink": is_symlink(abs_path) if abs_path.exists() else False,
            "approvedSha": approved_sha,
            "currentSha": current_sha,
            "approvedSnapshot": str(approved_snap) if approved_snap.exists() else None,
            "ok": ok,
        })

    print(json.dumps(out, indent=2))


def detect_drift_for(state: GuardianState, baselines: dict[str, Any], relp: str) -> tuple[bool, dict[str, Any]]:
    abs_path = WORKSPACE_ROOT / relp

    if not abs_path.exists():
        return True, {"error": f"Missing {relp}"}

    refuse_symlink(abs_path)

    baseline = baseline_info_for(state, baselines, relp)
    if not baseline:
        return True, {"error": f"Not initialized for {relp} (missing baseline). Run init/approve."}

    approved_sha = baseline.get("sha256")
    approved_snap = approved_snapshot_path(state, relp)
    if not approved_snap.exists():
        return True, {"error": f"Not initialized for {relp} (missing approved snapshot)."}

    cur_bytes = read_bytes(abs_path)
    cur_sha = sha256_bytes(cur_bytes)

    if cur_sha == approved_sha:
        return False, {"approvedSha": approved_sha, "currentSha": cur_sha}

    old_text = read_text(approved_snap)
    new_text = read_text(abs_path)
    patch_text = unified_diff_text(old_text, new_text, f"approved/{relp}", relp)
    patch_path = write_patch(state, patch_text, tag="drift", rel_path=relp)

    return True, {
        "approvedSha": approved_sha,
        "currentSha": cur_sha,
        "patchPath": str(patch_path),
    }


def restore_one(state: GuardianState, relp: str, info: dict[str, Any]) -> dict[str, Any]:
    """Restore a single file to its approved snapshot.

    Returns: extra fields to include in audit.
    """

    abs_path = WORKSPACE_ROOT / relp
    refuse_symlink(abs_path)

    approved_snap = approved_snapshot_path(state, relp)
    if not approved_snap.exists():
        raise RuntimeError(f"Missing approved snapshot for {relp}")

    state.ensure_dirs()
    ts = dt.datetime.now(dt.timezone.utc).strftime("%Y%m%dT%H%M%SZ")

    quarantine_path = state.quarantine_dir / f"{safe_patch_tag(relp.replace('/', '_'))}.{ts}.quarantine"
    atomic_write_bytes(quarantine_path, read_bytes(abs_path))

    # Atomic restore.
    atomic_write_bytes(abs_path, read_bytes(approved_snap))

    return {"quarantinePath": str(quarantine_path), **info}


def format_alert_human(drifted: list[dict[str, Any]]) -> str:
    """Format drift results as human-readable alert for TUI notification."""
    lines = []
    lines.append("")
    lines.append("=" * 50)
    lines.append("🚨 SOUL GUARDIAN SECURITY ALERT")
    lines.append("=" * 50)
    lines.append("")
    
    for d in drifted:
        path = d.get("path", "unknown")
        mode = d.get("mode", "unknown")
        restored = d.get("restored", False)
        error = d.get("error")
        
        if error:
            lines.append(f"⚠️  ERROR: {path}")
            lines.append(f"   {error}")
        else:
            lines.append(f"📄 FILE: {path}")
            lines.append(f"   Mode: {mode}")
            if restored:
                lines.append(f"   Status: ✅ RESTORED to approved baseline")
                if d.get("quarantinePath"):
                    lines.append(f"   Quarantined: {d.get('quarantinePath')}")
            else:
                lines.append(f"   Status: ⚠️  DRIFT DETECTED (not auto-restored)")
            
            if d.get("approvedSha"):
                lines.append(f"   Expected hash: {d.get('approvedSha')[:16]}...")
            if d.get("currentSha"):
                lines.append(f"   Found hash:    {d.get('currentSha')[:16]}...")
            if d.get("patchPath"):
                lines.append(f"   Diff saved: {d.get('patchPath')}")
        lines.append("")
    
    lines.append("=" * 50)
    lines.append("Review changes and investigate the source of drift.")
    lines.append("If intentional, run: soul_guardian.py approve --file <path>")
    lines.append("=" * 50)
    lines.append("")
    
    return "\n".join(lines)


def check_cmd(state: GuardianState, actor: str, note: str, *, no_restore: bool = False, output_format: str = "json") -> int:
    state.ensure_dirs()
    policy = load_policy(state)
    baselines = load_baselines(state)

    targets = resolve_targets(policy, WORKSPACE_ROOT)

    drifted: list[dict[str, Any]] = []

    for t in targets:
        relp = t["path"]
        mode = t["mode"]
        if mode == "ignore":
            continue

        drift, info = detect_drift_for(state, baselines, relp)
        if not drift:
            continue

        if "error" in info:
            append_audit(state, {
                "ts": utc_now_iso(),
                "event": "error",
                "actor": actor,
                "note": note,
                "path": relp,
                "mode": mode,
                "error": info["error"],
            })
            drifted.append({"path": relp, "mode": mode, "error": info["error"]})
            continue

        # Drift detected.
        append_audit(state, {
            "ts": utc_now_iso(),
            "event": "drift",
            "actor": actor,
            "note": note,
            "path": relp,
            "mode": mode,
            **info,
        })

        rec: dict[str, Any] = {"path": relp, "mode": mode, **info}

        # Auto-restore for restore-mode unless disabled.
        if mode == "restore" and not no_restore:
            restored = restore_one(state, relp, info)
            append_audit(state, {
                "ts": utc_now_iso(),
                "event": "restore",
                "actor": actor,
                "note": note,
                "path": relp,
                "mode": mode,
                **restored,
            })
            rec["restored"] = True
            rec["quarantinePath"] = restored.get("quarantinePath")
        else:
            rec["restored"] = False

        drifted.append(rec)

    if not drifted:
        # Silent on OK for alert format
        if output_format != "alert":
            pass  # Could print "OK" here if desired
        return 0

    # Output based on format
    if output_format == "alert":
        # Human-readable alert suitable for direct relay to user
        print(format_alert_human(drifted))
    else:
        # JSON format (default) - single-line summary suitable for cron parsing
        summary = {
            "event": "SOUL_GUARDIAN_DRIFT",
            "count": len(drifted),
            "files": [
                {
                    "path": d["path"],
                    "mode": d.get("mode"),
                    "restored": d.get("restored"),
                    "patch": d.get("patchPath"),
                    "error": d.get("error"),
                }
                for d in drifted
            ],
        }
        print("SOUL_GUARDIAN_DRIFT " + json.dumps(summary, ensure_ascii=False))

    # Drift occurred (even if restored).
    return 2


def watch_cmd(state: GuardianState, actor: str, note: str, *, interval: int = 30) -> int:
    """Continuous watch mode - checks periodically and alerts on drift."""
    import time
    
    print(f"👻 Soul Guardian watching... (interval: {interval}s)")
    print("Press Ctrl+C to stop.\n")
    
    try:
        while True:
            result = check_cmd(state, actor, note, output_format="alert")
            if result == 0:
                # No drift - stay silent
                pass
            # Sleep before next check
            time.sleep(interval)
    except KeyboardInterrupt:
        print("\n👻 Soul Guardian stopped.")
        return 0


def enable_monitoring_cmd(state: GuardianState) -> None:
    """Output instructions and HEARTBEAT.md snippet for easy monitoring setup."""
    
    heartbeat_snippet = """
## Soul Guardian Check
- Run `cd {workspace} && python3 skills/soul-guardian/scripts/soul_guardian.py check --actor heartbeat --output-format alert`
- If any output is produced, relay it to the user immediately as a security alert
- This checks for unauthorized changes to SOUL.md, AGENTS.md, and other protected files
""".format(workspace=WORKSPACE_ROOT)

    print("""
╔══════════════════════════════════════════════════════════════╗
║           SOUL GUARDIAN - ENABLE MONITORING                 ║
╚══════════════════════════════════════════════════════════════╝

To enable automatic drift detection and alerting, you have two options:

────────────────────────────────────────────────────────────────
OPTION 1: Heartbeat Integration (Recommended)
────────────────────────────────────────────────────────────────

Add the following to your HEARTBEAT.md file:
""")
    print(heartbeat_snippet)
    print("""
────────────────────────────────────────────────────────────────
OPTION 2: Watch Mode (Foreground)
────────────────────────────────────────────────────────────────

Run this in a terminal to continuously monitor:

    python3 skills/soul-guardian/scripts/soul_guardian.py watch --interval 30

────────────────────────────────────────────────────────────────
OPTION 3: Manual Check
────────────────────────────────────────────────────────────────

Run a one-time check with human-readable output:

    python3 skills/soul-guardian/scripts/soul_guardian.py check --output-format alert

────────────────────────────────────────────────────────────────

The guardian will:
✓ Detect unauthorized changes to protected files
✓ Auto-restore SOUL.md and AGENTS.md to approved baselines
✓ Alert you immediately when drift is detected
✓ Save diffs and quarantine modified files for review

""")
    print(f"State directory: {state.state_dir}")
    print(f"Workspace: {WORKSPACE_ROOT}")
    print()


def approve_cmd(state: GuardianState, actor: str, note: str, *, files: list[str] | None, all_files: bool = False) -> None:
    state.ensure_dirs()
    policy = load_policy(state)
    baselines = load_baselines(state)

    targets = resolve_targets(policy, WORKSPACE_ROOT)
    selectable = [t for t in targets if t["mode"] != "ignore"]

    if all_files:
        chosen = selectable
    elif files:
        # Resolve to relative posix.
        wanted = {Path(f).as_posix() for f in files}
        chosen = [t for t in selectable if t["path"] in wanted]
        missing = wanted - {t["path"] for t in chosen}
        if missing:
            raise RuntimeError(f"Unknown or ignored file(s): {', '.join(sorted(missing))}")
    else:
        # Backward-compat: if nothing specified, approve SOUL.md.
        chosen = [t for t in selectable if t["path"] == "SOUL.md"]
        if not chosen:
            raise RuntimeError("No files selected to approve.")

    for t in chosen:
        relp = t["path"]
        mode = t["mode"]
        abs_path = WORKSPACE_ROOT / relp
        if not abs_path.exists():
            raise FileNotFoundError(f"Missing {relp}")
        refuse_symlink(abs_path)

        prev = baseline_info_for(state, baselines, relp)
        prev_sha = prev.get("sha256") if prev else None
        prev_text = read_text(approved_snapshot_path(state, relp)) if approved_snapshot_path(state, relp).exists() else ""

        cur_bytes = read_bytes(abs_path)
        cur_sha = sha256_bytes(cur_bytes)
        cur_text = read_text(abs_path)

        patch_text = unified_diff_text(prev_text, cur_text, f"approved/{relp}", relp)
        patch_path = write_patch(state, patch_text, tag="approve", rel_path=relp)

        snap = approved_snapshot_path(state, relp)
        ensure_dir(snap.parent)
        atomic_write_bytes(snap, cur_bytes)

        set_baseline_for(state, baselines, relp, cur_sha)

        append_audit(state, {
            "ts": utc_now_iso(),
            "event": "approve",
            "actor": actor,
            "note": note,
            "path": relp,
            "mode": mode,
            "prevApprovedSha": prev_sha,
            "approvedSha": cur_sha,
            "patchPath": str(patch_path),
        })

        print(f"Approved {relp}: sha256={cur_sha} patch={patch_path}")

    save_baselines(state, baselines)


def restore_cmd(state: GuardianState, actor: str, note: str, *, files: list[str] | None, all_files: bool = False) -> None:
    state.ensure_dirs()
    policy = load_policy(state)
    baselines = load_baselines(state)

    targets = resolve_targets(policy, WORKSPACE_ROOT)
    restorable = [t for t in targets if t["mode"] == "restore"]

    if all_files:
        chosen = restorable
    elif files:
        wanted = {Path(f).as_posix() for f in files}
        chosen = [t for t in restorable if t["path"] in wanted]
        missing = wanted - {t["path"] for t in chosen}
        if missing:
            raise RuntimeError(f"Not restorable or unknown file(s): {', '.join(sorted(missing))}")
    else:
        # Backward-compat: default restore SOUL.md.
        chosen = [t for t in restorable if t["path"] == "SOUL.md"]
        if not chosen:
            raise RuntimeError("No files selected to restore.")

    restored_any = False
    for t in chosen:
        relp = t["path"]
        mode = t["mode"]

        drift, info = detect_drift_for(state, baselines, relp)
        if "error" in info:
            raise RuntimeError(info["error"])
        if not drift:
            print(f"No drift for {relp}; nothing to restore.")
            continue

        restored = restore_one(state, relp, info)
        append_audit(state, {
            "ts": utc_now_iso(),
            "event": "restore",
            "actor": actor,
            "note": note,
            "path": relp,
            "mode": mode,
            **restored,
        })
        print(
            f"RESTORED {relp} approvedSha={info.get('approvedSha')} previousSha={info.get('currentSha')} "
            f"quarantine={restored.get('quarantinePath')} patch={info.get('patchPath')}"
        )
        restored_any = True

    if not restored_any:
        print("No restores performed.")


def verify_audit_cmd(state: GuardianState) -> None:
    state.ensure_dirs()
    if not state.audit_path.exists():
        print("No audit log present.")
        return

    if _audit_needs_upgrade(state):
        raise RuntimeError(
            "Audit log is legacy (missing hash chain). "
            "Run any command that writes audit (e.g., check) to rotate legacy log, then re-run verify-audit."
        )

    prev = CHAIN_GENESIS
    line_no = 0
    with state.audit_path.open("r", encoding="utf-8") as f:
        for line in f:
            line_no += 1
            line = line.strip()
            if not line:
                continue
            rec = json.loads(line)
            chain = rec.get("chain") or {}
            got_prev = chain.get("prev")
            got_hash = chain.get("hash")

            if got_prev != prev:
                raise RuntimeError(f"Audit chain broken at line {line_no}: prev mismatch (expected {prev}, got {got_prev})")

            rec_wo_chain = dict(rec)
            rec_wo_chain.pop("chain", None)
            payload = prev + "\n" + _canonical_json(rec_wo_chain)
            expect_hash = sha256_text(payload)

            if got_hash != expect_hash:
                raise RuntimeError(f"Audit chain broken at line {line_no}: hash mismatch")

            prev = got_hash

    print(f"OK: audit log hash chain verified ({line_no} lines)")


def parse_args(argv: list[str]) -> argparse.Namespace:
    p = argparse.ArgumentParser(
        description="Soul Guardian - Workspace file integrity guard with alerting support.",
        epilog="For easy setup, run: soul_guardian.py enable-monitoring"
    )
    p.add_argument(
        "--state-dir",
        default=str(DEFAULT_STATE_DIR),
        help="State directory (default: memory/soul-guardian).",
    )

    sub = p.add_subparsers(dest="cmd", required=True)

    def add_common(sp: argparse.ArgumentParser) -> None:
        sp.add_argument("--actor", default="unknown", help="Who initiated the action (best-effort).")
        sp.add_argument("--note", default="", help="Freeform note (e.g., request context).")

    sp_init = sub.add_parser("init", help="Initialize policy + baselines.")
    add_common(sp_init)
    sp_init.add_argument("--force-policy", action="store_true", help="Overwrite policy.json with defaults.")

    sub.add_parser("status", help="Print status JSON.")

    sp_check = sub.add_parser("check", help="Check for drift; restore restore-mode by default.")
    add_common(sp_check)
    sp_check.add_argument("--no-restore", action="store_true", help="Never restore during check (alert-only run).")
    sp_check.add_argument("--output-format", choices=["json", "alert"], default="json",
                          help="Output format: json (machine-readable) or alert (human-readable for TUI).")

    sp_approve = sub.add_parser("approve", help="Approve current contents as baselines.")
    add_common(sp_approve)
    sp_approve.add_argument("--file", action="append", dest="files", help="Relative file path to approve (repeatable).")
    sp_approve.add_argument("--all", action="store_true", help="Approve all non-ignored policy targets.")

    sp_restore = sub.add_parser("restore", help="Restore restore-mode files to approved baselines.")
    add_common(sp_restore)
    sp_restore.add_argument("--file", action="append", dest="files", help="Relative file path to restore (repeatable).")
    sp_restore.add_argument("--all", action="store_true", help="Restore all restore-mode targets.")

    sub.add_parser("verify-audit", help="Verify audit log hash chain.")
    
    # New commands for easier monitoring setup
    sp_watch = sub.add_parser("watch", help="Continuous watch mode - monitors and alerts on drift.")
    add_common(sp_watch)
    sp_watch.add_argument("--interval", type=int, default=30, help="Check interval in seconds (default: 30).")
    
    sub.add_parser("enable-monitoring", help="Show instructions for enabling automatic monitoring and alerts.")

    return p.parse_args(argv)


def main(argv: list[str]) -> int:
    args = parse_args(argv)
    state = GuardianState(Path(args.state_dir).expanduser())

    try:
        if args.cmd == "init":
            init_cmd(state, args.actor, args.note, force_policy=bool(getattr(args, "force_policy", False)))
            return 0
        if args.cmd == "status":
            status_cmd(state)
            return 0
        if args.cmd == "check":
            return check_cmd(
                state, args.actor, args.note,
                no_restore=bool(getattr(args, "no_restore", False)),
                output_format=getattr(args, "output_format", "json")
            )
        if args.cmd == "approve":
            approve_cmd(state, args.actor, args.note, files=getattr(args, "files", None), all_files=bool(getattr(args, "all", False)))
            return 0
        if args.cmd == "restore":
            restore_cmd(state, args.actor, args.note, files=getattr(args, "files", None), all_files=bool(getattr(args, "all", False)))
            return 0
        if args.cmd == "verify-audit":
            verify_audit_cmd(state)
            return 0
        if args.cmd == "watch":
            return watch_cmd(state, args.actor, args.note, interval=getattr(args, "interval", 30))
        if args.cmd == "enable-monitoring":
            enable_monitoring_cmd(state)
            return 0

        raise RuntimeError(f"Unknown cmd: {args.cmd}")

    except Exception as e:
        print(f"ERROR: {e}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))

#!/usr/bin/env python3
"""Regression tests for install_launchd_plist.py default state-dir selection."""

from __future__ import annotations

import importlib.util
import io
import os
from pathlib import Path
import plistlib
import subprocess
import tempfile
from contextlib import redirect_stderr, redirect_stdout
from types import ModuleType


REPO_ROOT = Path(__file__).resolve().parents[3]
SCRIPT = REPO_ROOT / "skills" / "soul-guardian" / "scripts" / "install_launchd_plist.py"


def run(cmd: list[str], env: dict[str, str]) -> subprocess.CompletedProcess:
    return subprocess.run(cmd, text=True, capture_output=True, env=env)


def must_ok(cp: subprocess.CompletedProcess) -> None:
    if cp.returncode != 0:
        raise AssertionError(f"Expected rc=0, got {cp.returncode}\nSTDOUT:\n{cp.stdout}\nSTDERR:\n{cp.stderr}")


def load_program_arguments(plist_path: Path) -> list[str]:
    with plist_path.open("rb") as handle:
        return plistlib.load(handle)["ProgramArguments"]


def run_case(home_dir: Path, agent_id: str) -> subprocess.CompletedProcess:
    env = os.environ.copy()
    env["HOME"] = str(home_dir)
    plist_path = home_dir / "LaunchAgents" / f"{agent_id}.plist"
    cmd = [
        "python3",
        str(SCRIPT),
        "--workspace-root",
        str(REPO_ROOT),
        "--agent-id",
        agent_id,
        "--out",
        str(plist_path),
        "--force",
    ]
    return run(cmd, env)


def assert_contains(text: str, expected: str, label: str) -> None:
    if expected not in text:
        raise AssertionError(f"Missing {label}: expected to find {expected!r}\nActual text:\n{text}")


def load_module(home_dir: Path) -> ModuleType:
    previous_home = os.environ.get("HOME")
    os.environ["HOME"] = str(home_dir)
    try:
        spec = importlib.util.spec_from_file_location("test_install_launchd_plist_module", SCRIPT)
        if spec is None or spec.loader is None:
            raise AssertionError("Failed to load install_launchd_plist.py for testing")
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        return module
    finally:
        if previous_home is None:
            os.environ.pop("HOME", None)
        else:
            os.environ["HOME"] = previous_home


def call_main_with_home(module: ModuleType, home_dir: Path, argv: list[str]) -> int:
    previous_home = os.environ.get("HOME")
    os.environ["HOME"] = str(home_dir)
    try:
        return module.main(argv)
    finally:
        if previous_home is None:
            os.environ.pop("HOME", None)
        else:
            os.environ["HOME"] = previous_home


def main() -> int:
    with tempfile.TemporaryDirectory() as td:
        home_dir = Path(td)
        agent_id = "legacy-agent"
        legacy_state_dir = home_dir / ".clawdbot" / "soul-guardian" / agent_id
        legacy_state_dir.mkdir(parents=True, exist_ok=True)

        cp = run_case(home_dir, agent_id)
        must_ok(cp)

        legacy_state_suffix = "/.clawdbot/soul-guardian/legacy-agent"
        new_state_suffix = "/.openclaw/soul-guardian/legacy-agent"
        assert_contains(cp.stdout, legacy_state_suffix, "legacy state dir in stdout")
        assert_contains(cp.stderr, legacy_state_suffix, "legacy state dir warning")
        assert_contains(cp.stderr, new_state_suffix, "migration target warning")

        program_args = load_program_arguments(home_dir / "LaunchAgents" / f"{agent_id}.plist")
        if not any(arg.endswith(legacy_state_suffix) for arg in program_args):
            raise AssertionError(f"Expected plist to reference legacy state dir.\nProgramArguments: {program_args}")

    with tempfile.TemporaryDirectory() as td:
        home_dir = Path(td)
        agent_id = "fresh-agent"

        cp = run_case(home_dir, agent_id)
        must_ok(cp)

        new_state_suffix = "/.openclaw/soul-guardian/fresh-agent"
        assert_contains(cp.stdout, new_state_suffix, "new state dir in stdout")
        if cp.stderr.strip():
            raise AssertionError(f"Did not expect migration warning for fresh install.\nSTDERR:\n{cp.stderr}")

        program_args = load_program_arguments(home_dir / "LaunchAgents" / f"{agent_id}.plist")
        if not any(arg.endswith(new_state_suffix) for arg in program_args):
            raise AssertionError(f"Expected plist to reference new state dir.\nProgramArguments: {program_args}")

    with tempfile.TemporaryDirectory() as td:
        home_dir = Path(td)
        agent_id = "migrate-agent"
        legacy_label = f"com.clawdbot.soul-guardian.{agent_id}"
        legacy_plist = home_dir / "Library" / "LaunchAgents" / f"{legacy_label}.plist"
        legacy_plist.parent.mkdir(parents=True, exist_ok=True)
        legacy_plist.write_text("legacy", encoding="utf-8")

        cp = run(
            [
                "python3",
                str(SCRIPT),
                "--workspace-root",
                str(REPO_ROOT),
                "--agent-id",
                agent_id,
                "--force",
            ],
            {**os.environ, "HOME": str(home_dir)},
        )
        must_ok(cp)
        assert_contains(cp.stdout, legacy_label, "legacy label dry-run note")

        module = load_module(home_dir)
        launchctl_calls: list[list[str]] = []
        subprocess_calls: list[list[str]] = []

        def fake_run_launchctl(args: list[str]) -> subprocess.CompletedProcess[str]:
            launchctl_calls.append(args)
            return subprocess.CompletedProcess(["/bin/launchctl", *args], 0, "", "")

        def fake_subprocess_run(args: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]:
            subprocess_calls.append(args)
            return subprocess.CompletedProcess(args, 0, "", "")

        module.run_launchctl = fake_run_launchctl
        module.subprocess.run = fake_subprocess_run
        module.os.getuid = lambda: 501

        stdout_buffer = io.StringIO()
        stderr_buffer = io.StringIO()
        with redirect_stdout(stdout_buffer), redirect_stderr(stderr_buffer):
            rc = call_main_with_home(
                module,
                home_dir,
                [
                    "--workspace-root",
                    str(REPO_ROOT),
                    "--agent-id",
                    agent_id,
                    "--force",
                    "--install",
                ],
            )
        if rc != 0:
            raise AssertionError(f"Expected install flow rc=0, got {rc}")

        expected_prefix = [
            ["disable", "gui/501/com.clawdbot.soul-guardian.migrate-agent"],
            ["bootout", "gui/501/com.clawdbot.soul-guardian.migrate-agent"],
            ["bootout", "gui/501", str(legacy_plist.resolve())],
        ]
        if launchctl_calls[:3] != expected_prefix:
            raise AssertionError(f"Expected legacy cleanup calls first.\nActual launchctl calls: {launchctl_calls}")

        if ["/bin/launchctl", "enable", "gui/501/com.openclaw.soul-guardian.migrate-agent"] not in subprocess_calls:
            raise AssertionError(f"Expected enable call for new label.\nSubprocess calls: {subprocess_calls}")

    with tempfile.TemporaryDirectory() as td:
        home_dir = Path(td)
        agent_id = "warn-agent"
        legacy_label = f"com.clawdbot.soul-guardian.{agent_id}"
        legacy_plist = home_dir / "Library" / "LaunchAgents" / f"{legacy_label}.plist"
        legacy_plist.parent.mkdir(parents=True, exist_ok=True)
        legacy_plist.write_text("legacy", encoding="utf-8")

        module = load_module(home_dir)

        def fake_run_launchctl_warn(args: list[str]) -> subprocess.CompletedProcess[str]:
            return subprocess.CompletedProcess(["/bin/launchctl", *args], 1, "", "cleanup failed")

        def fake_subprocess_run_warn(args: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]:
            if args[:2] == ["/bin/launchctl", "bootstrap"]:
                return subprocess.CompletedProcess(args, 0, "", "")
            if args[:2] == ["/bin/launchctl", "enable"]:
                return subprocess.CompletedProcess(args, 0, "", "")
            if args[:2] == ["/bin/launchctl", "kickstart"]:
                return subprocess.CompletedProcess(args, 0, "", "")
            return subprocess.CompletedProcess(args, 1, "", "cleanup failed")

        module.run_launchctl = fake_run_launchctl_warn
        module.subprocess.run = fake_subprocess_run_warn
        module.os.getuid = lambda: 501

        stdout_buffer = io.StringIO()
        stderr_buffer = io.StringIO()
        with redirect_stdout(stdout_buffer), redirect_stderr(stderr_buffer):
            rc = call_main_with_home(
                module,
                home_dir,
                [
                    "--workspace-root",
                    str(REPO_ROOT),
                    "--agent-id",
                    agent_id,
                    "--force",
                    "--install",
                ],
            )
        if rc != 0:
            raise AssertionError(f"Expected install flow rc=0 with cleanup warning, got {rc}")
        assert_contains(stderr_buffer.getvalue(), "launchctl bootout gui/501 com.clawdbot.soul-guardian.warn-agent", "manual cleanup warning")
        assert_contains(stderr_buffer.getvalue(), str(legacy_plist.resolve()), "legacy plist warning")

    print("OK: install_launchd_plist default state-dir tests passed")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Minimal tests for soul_guardian.py.

Run:
  python3 skills/soul-guardian/scripts/test_soul_guardian.py

This is a lightweight integration test using a temp workspace.
"""

from __future__ import annotations

import json
import os
from pathlib import Path
import subprocess
import tempfile


REPO_ROOT = Path(__file__).resolve().parents[3]  # .../clawd
SCRIPT = REPO_ROOT / "skills" / "soul-guardian" / "scripts" / "soul_guardian.py"


def run(cmd: list[str], cwd: Path) -> subprocess.CompletedProcess:
    return subprocess.run(cmd, cwd=str(cwd), text=True, capture_output=True)


def must_ok(cp: subprocess.CompletedProcess) -> None:
    if cp.returncode != 0:
        raise AssertionError(f"Expected rc=0, got {cp.returncode}\nSTDOUT:\n{cp.stdout}\nSTDERR:\n{cp.stderr}")


def must_rc(cp: subprocess.CompletedProcess, rc: int) -> None:
    if cp.returncode != rc:
        raise AssertionError(f"Expected rc={rc}, got {cp.returncode}\nSTDOUT:\n{cp.stdout}\nSTDERR:\n{cp.stderr}")


def main() -> int:
    with tempfile.TemporaryDirectory() as td:
        ws = Path(td)
        state = ws / "state"

        # Create a fake workspace with the default protected files.
        (ws / "memory").mkdir(parents=True, exist_ok=True)
        (ws / "SOUL.md").write_text("hello soul\n", encoding="utf-8")
        (ws / "AGENTS.md").write_text("hello agents\n", encoding="utf-8")
        (ws / "USER.md").write_text("user v1\n", encoding="utf-8")
        (ws / "TOOLS.md").write_text("tools v1\n", encoding="utf-8")
        (ws / "IDENTITY.md").write_text("id v1\n", encoding="utf-8")
        (ws / "HEARTBEAT.md").write_text("hb v1\n", encoding="utf-8")
        (ws / "MEMORY.md").write_text("mem v1\n", encoding="utf-8")
        # Daily notes should be ignored by default.
        (ws / "memory" / "2026-01-01.md").write_text("daily\n", encoding="utf-8")

        # Init baselines.
        cp = run(["python3", str(SCRIPT), "--state-dir", str(state), "init", "--actor", "test"], cwd=ws)
        must_ok(cp)

        # No drift.
        cp = run(["python3", str(SCRIPT), "--state-dir", str(state), "check"], cwd=ws)
        must_ok(cp)

        # Drift restore-mode file: SOUL.md should be auto-restored by check.
        (ws / "SOUL.md").write_text("MALICIOUS\n", encoding="utf-8")
        cp = run(["python3", str(SCRIPT), "--state-dir", str(state), "check", "--actor", "cron"], cwd=ws)
        must_rc(cp, 2)
        assert (ws / "SOUL.md").read_text(encoding="utf-8") == "hello soul\n"

        # Drift alert-only file: USER.md should NOT be restored.
        (ws / "USER.md").write_text("user v2\n", encoding="utf-8")
        cp = run(["python3", str(SCRIPT), "--state-dir", str(state), "check"], cwd=ws)
        must_rc(cp, 2)
        assert (ws / "USER.md").read_text(encoding="utf-8") == "user v2\n"

        # Approve USER.md change.
        cp = run(["python3", str(SCRIPT), "--state-dir", str(state), "approve", "--file", "USER.md", "--actor", "test"], cwd=ws)
        must_ok(cp)

        # Now check should be clean.
        cp = run(["python3", str(SCRIPT), "--state-dir", str(state), "check"], cwd=ws)
        must_ok(cp)

        # Verify audit chain ok.
        cp = run(["python3", str(SCRIPT), "--state-dir", str(state), "verify-audit"], cwd=ws)
        must_ok(cp)

        # Tamper with audit log and ensure verify fails.
        audit = state / "audit.jsonl"
        lines = audit.read_text(encoding="utf-8").splitlines()
        assert lines, "audit log empty"
        rec = json.loads(lines[-1])
        rec["note"] = "tampered"
        lines[-1] = json.dumps(rec, ensure_ascii=False)
        audit.write_text("\n".join(lines) + "\n", encoding="utf-8")

        cp = run(["python3", str(SCRIPT), "--state-dir", str(state), "verify-audit"], cwd=ws)
        if cp.returncode == 0:
            raise AssertionError("Expected verify-audit to fail after tamper")

        print("OK: soul-guardian minimal tests passed")

    return 0


if __name__ == "__main__":
    raise SystemExit(main())