coder_skills.py

#!/usr/bin/env python3
"""coder-skills — carry personal Claude config into Coder workspaces.

Engineers run this once on their laptop to:
  1. Separate cred files from skill code in ~/.claude/
  2. Push code to a per-engineer private GitHub repo
     (`lumalabs/dotfiles-<gh-username>`)
  3. Sync cred files into the per-engineer S3 prefix in the Coder
     workspace-auth bucket (the same bucket Coder workspaces already use
     for ~/.claude/.credentials.json)

After migrate, every Coder workspace claim auto-loads both layers:
  - git pulls the dotfiles repo into ~/.claude/
  - workspace's existing 30s S3 sync daemon brings down the cred files

Edits flow back via standard `git push` (code) or
`lumalabs-skills rotate` (creds).

Subcommands:
  migrate                One-time setup. Interactive.
  push                   Force one-shot git push + S3 sync (laptop → cloud)
  pull                   Force one-shot git pull + S3 sync (cloud → laptop)
  rotate <cred-name>     Update a credential file (clipboard-based, no
                         secret values in chat history)
  status                 Show current state of code + creds layers

Manifest at ~/.claude/.skills-vault.json maps file paths to S3 blob names
+ permissions. Committed to git (no secret values).
"""

from __future__ import annotations

import argparse
import getpass
import json
import os
import re
import shutil
import subprocess
import sys
from pathlib import Path

# ── Defaults — override via env or flags ──────────────────────────────
HOME = Path(os.environ.get("HOME", os.path.expanduser("~")))
CLAUDE_DIR = HOME / ".claude"
AUTH_DIR = HOME / ".coder-auth"  # mirrors the workspace's auth-bucket sync dir
MANIFEST_FILE = CLAUDE_DIR / ".skills-vault.json"

# S3 auth-bucket where Coder workspaces sync per-engineer creds. Same
# bucket the workspace's existing auth-sync daemon uses.
AUTH_BUCKET = os.environ.get("CODER_AUTH_BUCKET", "lumalabs-dev-oregon-coder-workspace-auth")
AUTH_BUCKET_REGION = os.environ.get("CODER_AUTH_BUCKET_REGION", "us-west-2")
AUTH_BUCKET_PROFILE = os.environ.get("CODER_AUTH_BUCKET_PROFILE", "luma-dev")

# Per-engineer dotfiles repo naming: lumalabs/dotfiles-<gh-username>.
# Override via env if your team uses a different convention.
DOTFILES_REPO_OWNER = os.environ.get("CODER_DOTFILES_OWNER", "lumalabs")
DOTFILES_REPO_PREFIX = os.environ.get("CODER_DOTFILES_PREFIX", "dotfiles-")

# Heuristic: which files are credentials? Used during `migrate` to flag
# candidates. Engineer confirms each one — these are SUGGESTIONS, not
# automatic moves.
CRED_PATTERNS = [
    re.compile(r".*api[_-]?key$", re.I),
    re.compile(r".*token.*", re.I),
    re.compile(r".*\.pem$", re.I),
    re.compile(r".*credential.*\.json$", re.I),
    re.compile(r".*oauth.*\.json$", re.I),
    re.compile(r".*client_secret.*\.json$", re.I),
    re.compile(r".*private_key.*", re.I),
]

# Files that should NEVER be touched by this tool — caches, scratch, npm
# stuff. Explicitly excluded from both git tracking and cred consideration.
ALWAYS_EXCLUDE = {
    "node_modules",
    "cache",
    "paste-cache",
    "shell-snapshots",
    "history.jsonl",
    "projects",
    "tasks",
    "todos",
    "sessions",
    "session-env",
    "stats-cache.json",
    "telemetry",
    "ide",
    "downloads",
    "debug",
    "file-history",
    "backups",
    ".credentials.json",  # already synced by Coder's existing daemon
    "plugins",  # local plugin caches
}


# ── Helpers ───────────────────────────────────────────────────────────


def _run(cmd: list[str], **kwargs) -> subprocess.CompletedProcess:
    """Run a command, fail loud."""
    return subprocess.run(cmd, check=True, **kwargs)


def _run_quiet(cmd: list[str]) -> int:
    """Run a command, swallow output, return exit code."""
    return subprocess.call(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)


def _load_manifest() -> dict:
    if not MANIFEST_FILE.exists():
        return {"version": 1, "credentials": [], "dotfiles_repo": None}
    return json.loads(MANIFEST_FILE.read_text())


def _save_manifest(manifest: dict) -> None:
    CLAUDE_DIR.mkdir(parents=True, exist_ok=True)
    MANIFEST_FILE.write_text(json.dumps(manifest, indent=2) + "\n")


def _aws_s3_sync(src: str, dest: str) -> None:
    """Wrap aws s3 sync with our profile/region defaults."""
    _run(
        [
            "aws",
            "s3",
            "sync",
            "--profile",
            AUTH_BUCKET_PROFILE,
            "--region",
            AUTH_BUCKET_REGION,
            "--no-progress",
            src,
            dest,
        ]
    )


def _gh_username() -> str:
    """Whoami via the gh CLI."""
    result = _run(["gh", "api", "user", "--jq", ".login"], capture_output=True, text=True)
    return result.stdout.strip()


def _coder_user_id() -> str | None:
    """Engineer's Coder user-id, if they're already in a Coder workspace.

    Used as the S3 prefix key (matches the existing auth-sync daemon's
    expected layout: `s3://<bucket>/<coder-user-id>/...`). On laptop,
    we don't have CODER_USER_ID set, so we fall back to asking the
    engineer once + caching in the manifest.
    """
    return os.environ.get("CODER_USER_ID")


def _get_or_prompt_user_id(manifest: dict) -> str:
    """Same key as the workspace uses. Cached in the manifest."""
    if uid := manifest.get("coder_user_id"):
        return uid
    if uid := _coder_user_id():
        manifest["coder_user_id"] = uid
        return uid
    print("Coder user-id is the S3 prefix where your auth blobs live.")
    print("Get it from https://coder.oregon.dev.lumalabs.ai/settings/account")
    print("(it's the UUID under 'Username' in the URL of your account page).")
    print()
    uid = input("Coder user-id (UUID): ").strip()
    if not re.match(r"^[0-9a-f-]{36}$", uid):
        print(f"error: '{uid}' doesn't look like a UUID", file=sys.stderr)
        sys.exit(1)
    manifest["coder_user_id"] = uid
    return uid


def _looks_like_cred(path: Path) -> bool:
    name = path.name
    if name in ALWAYS_EXCLUDE:
        return False
    return any(p.match(name) for p in CRED_PATTERNS)


def _yn(prompt: str, default: bool = True) -> bool:
    suffix = " [Y/n] " if default else " [y/N] "
    while True:
        resp = input(prompt + suffix).strip().lower()
        if not resp:
            return default
        if resp in ("y", "yes"):
            return True
        if resp in ("n", "no"):
            return False


# ── Subcommand: migrate ───────────────────────────────────────────────


def cmd_migrate(_args: argparse.Namespace) -> int:
    """One-time setup. Interactive — never silently does anything destructive."""
    if not CLAUDE_DIR.exists():
        print(f"error: {CLAUDE_DIR} doesn't exist", file=sys.stderr)
        return 1

    manifest = _load_manifest()
    if manifest.get("credentials") or manifest.get("dotfiles_repo"):
        print(f"warning: {MANIFEST_FILE} already exists with content.")
        if not _yn("Continue (will re-prompt for each cred)?", default=False):
            return 0

    print("=" * 72)
    print("coder-skills migrate")
    print("=" * 72)
    print()
    print("This tool sets up a per-engineer flow for your ~/.claude/ config:")
    print()
    print("  • Skill code, helpers, CLAUDE.md, commands  → private GitHub repo")
    print("  • Credential files (API keys, OAuth, etc)   → S3 (Coder auth-bucket)")
    print()
    print("Both layers auto-load in every Coder workspace you spin up.")
    print()
    if not _yn("Proceed?", default=True):
        return 0

    # Step 1: prereqs
    print()
    print("[1/5] Checking prereqs...")
    for tool in ("git", "gh", "aws"):
        if not shutil.which(tool):
            print(f"error: {tool} not in PATH", file=sys.stderr)
            return 1
    if _run_quiet(["gh", "auth", "status"]) != 0:
        print("error: gh CLI not authenticated. Run: gh auth login", file=sys.stderr)
        return 1
    if _run_quiet(["aws", "sts", "get-caller-identity", "--profile", AUTH_BUCKET_PROFILE]) != 0:
        print(
            f"error: aws profile '{AUTH_BUCKET_PROFILE}' not authenticated. "
            f"Run: aws sso login --profile {AUTH_BUCKET_PROFILE}",
            file=sys.stderr,
        )
        return 1
    user_id = _get_or_prompt_user_id(manifest)
    print(f"  ✓ git/gh/aws available; coder user-id = {user_id}")

    # Step 2: scan + interactive cred selection
    print()
    print("[2/5] Scanning ~/.claude/ for credential candidates...")
    print()
    print("    For each file, you'll be asked: is this a credential?")
    print("    Credentials get moved to ~/.coder-auth/ and synced to S3.")
    print("    Everything else stays in ~/.claude/ and gets committed to git.")
    print()
    candidates: list[Path] = []
    for entry in sorted(CLAUDE_DIR.iterdir()):
        if entry.is_dir():
            continue
        if entry.name.startswith("."):
            # .credentials.json already handled, .skills-vault.json is ours
            continue
        if entry.name in ALWAYS_EXCLUDE:
            continue
        if _looks_like_cred(entry):
            candidates.append(entry)

    AUTH_DIR.mkdir(mode=0o700, parents=True, exist_ok=True)
    creds: list[dict] = []
    for c in candidates:
        size = c.stat().st_size
        print(f"    {c.name}  ({size}B)")
        if not _yn("      → credential?", default=True):
            continue
        # Default the auth_blob name to a kebab-cased version of the
        # original. Engineers rarely need to change this.
        suggested = "claude-" + c.name.lower().replace("_", "-").replace(".", "-")
        # Move to auth dir
        dest = AUTH_DIR / suggested
        if dest.exists():
            print(f"      ! {dest} already exists, skipping move (assumed already migrated)")
        else:
            shutil.move(str(c), str(dest))
            dest.chmod(0o600)
        creds.append({"file": str(c.relative_to(HOME)), "auth_blob": suggested, "mode": "0600"})
    manifest["credentials"] = creds
    _save_manifest(manifest)
    print(f"  ✓ {len(creds)} credentials moved to {AUTH_DIR}")

    # Step 3: initialize git working tree at ~/.claude/
    print()
    print("[3/5] Initializing ~/.claude/ as a git working tree...")
    if not (CLAUDE_DIR / ".git").exists():
        _run(["git", "-C", str(CLAUDE_DIR), "init", "-b", "main"])
    # .gitignore — never commit creds, caches, etc.
    gitignore_lines = sorted(
        list(ALWAYS_EXCLUDE)
        + [c["file"].replace(".claude/", "", 1) for c in creds]
        + [
            "# patterns",
            "*.pem",
            "*api_key",
            "*token",
            "*token.json",
            "*.credentials.json",
        ]
    )
    (CLAUDE_DIR / ".gitignore").write_text("\n".join(gitignore_lines) + "\n")
    print(f"  ✓ git init + .gitignore ({len(gitignore_lines)} lines)")

    # Step 4: create + push private GitHub repo
    print()
    print("[4/5] Creating per-engineer dotfiles repo on GitHub...")
    user = _gh_username()
    repo = f"{DOTFILES_REPO_OWNER}/{DOTFILES_REPO_PREFIX}{user}"
    print(f"    repo: {repo} (private)")
    if not _yn("    Create now?", default=True):
        print("    Skipping. You'll need to set this up manually later.")
        return 0
    if _run_quiet(["gh", "repo", "view", repo]) != 0:
        _run(
            [
                "gh",
                "repo",
                "create",
                repo,
                "--private",
                "--description",
                f"Personal Claude config for {user}",
            ]
        )
        print(f"  ✓ created {repo}")
    else:
        print(f"  ✓ repo {repo} already exists, will push to it")
    # Make sure remote is wired
    if _run_quiet(["git", "-C", str(CLAUDE_DIR), "remote", "get-url", "origin"]) != 0:
        _run(
            [
                "git",
                "-C",
                str(CLAUDE_DIR),
                "remote",
                "add",
                "origin",
                f"https://github.com/{repo}.git",
            ]
        )
    manifest["dotfiles_repo"] = repo
    _save_manifest(manifest)
    # Initial commit + push
    _run(
        ["git", "-C", str(CLAUDE_DIR), "add", "-A"],
    )
    if _run_quiet(["git", "-C", str(CLAUDE_DIR), "diff", "--cached", "--quiet"]) != 0:
        _run(
            [
                "git",
                "-C",
                str(CLAUDE_DIR),
                "commit",
                "-m",
                "Initial dotfiles via coder-skills migrate",
            ]
        )
        _run(["git", "-C", str(CLAUDE_DIR), "push", "-u", "origin", "main"])
        print("  ✓ initial commit + push")
    else:
        print("  ✓ no changes to commit (idempotent re-run)")

    # Step 5: push creds to S3
    print()
    print("[5/5] Pushing credential blobs to S3...")
    if creds:
        s3_dest = f"s3://{AUTH_BUCKET}/{user_id}/"
        _aws_s3_sync(str(AUTH_DIR), s3_dest)
        print(f"  ✓ {len(creds)} blobs pushed to {s3_dest}")
    else:
        print("  ✓ no creds to push")

    print()
    print("=" * 72)
    print("Migration complete.")
    print("=" * 72)
    print()
    print(f"Code:  {repo}  (push/pull via standard git)")
    print(f"Creds: s3://{AUTH_BUCKET}/{user_id}/  (use `lumalabs-skills rotate`)")
    print()
    print("Next workspace you spin up will auto-clone both layers.")
    return 0


# ── Subcommand: push ──────────────────────────────────────────────────


def cmd_push(_args: argparse.Namespace) -> int:
    """One-shot git push + S3 sync laptop→cloud."""
    manifest = _load_manifest()
    if not manifest.get("dotfiles_repo"):
        print("error: not migrated yet. Run `lumalabs-skills migrate`.", file=sys.stderr)
        return 1
    user_id = _get_or_prompt_user_id(manifest)

    print(f"git push (~/.claude → {manifest['dotfiles_repo']})...")
    _run(["git", "-C", str(CLAUDE_DIR), "push"])

    print(f"aws s3 sync (~/.coder-auth → s3://{AUTH_BUCKET}/{user_id}/)...")
    _aws_s3_sync(str(AUTH_DIR), f"s3://{AUTH_BUCKET}/{user_id}/")
    print("done")
    return 0


# ── Subcommand: pull ──────────────────────────────────────────────────


def cmd_pull(_args: argparse.Namespace) -> int:
    """One-shot git pull + S3 sync cloud→laptop."""
    manifest = _load_manifest()
    if not manifest.get("dotfiles_repo"):
        print("error: not migrated yet. Run `lumalabs-skills migrate`.", file=sys.stderr)
        return 1
    user_id = _get_or_prompt_user_id(manifest)

    print(f"git pull (~/.claude ← {manifest['dotfiles_repo']})...")
    _run(["git", "-C", str(CLAUDE_DIR), "pull"])

    AUTH_DIR.mkdir(mode=0o700, parents=True, exist_ok=True)
    print(f"aws s3 sync (~/.coder-auth ← s3://{AUTH_BUCKET}/{user_id}/)...")
    _aws_s3_sync(f"s3://{AUTH_BUCKET}/{user_id}/", str(AUTH_DIR))
    # Tighten perms (s3 sync downloads at 0644)
    for f in AUTH_DIR.rglob("*"):
        if f.is_file():
            f.chmod(0o600)
    print("done")
    return 0


# ── Subcommand: rotate ────────────────────────────────────────────────


def cmd_rotate(args: argparse.Namespace) -> int:
    """Update a credential file. Reads the new value via getpass (no echo)."""
    manifest = _load_manifest()
    if not manifest.get("dotfiles_repo"):
        print("error: not migrated yet. Run `lumalabs-skills migrate`.", file=sys.stderr)
        return 1
    target = args.cred
    matches = [c for c in manifest["credentials"] if c["file"].endswith(target) or c["auth_blob"] == target]
    if not matches:
        print(f"error: no credential matches '{target}'.", file=sys.stderr)
        print("Known credentials:")
        for c in manifest["credentials"]:
            print(f"  {c['file']}  →  {c['auth_blob']}")
        return 1
    if len(matches) > 1:
        print(f"error: '{target}' is ambiguous; matches:", file=sys.stderr)
        for c in matches:
            print(f"  {c['file']}  →  {c['auth_blob']}")
        return 1
    cred = matches[0]
    print(f"Rotating {cred['file']} (S3 blob: {cred['auth_blob']})")
    new_value = getpass.getpass("New value (input hidden): ")
    if not new_value:
        print("error: empty value, aborting.", file=sys.stderr)
        return 1
    AUTH_DIR.mkdir(mode=0o700, parents=True, exist_ok=True)
    blob_path = AUTH_DIR / cred["auth_blob"]
    blob_path.write_text(new_value)
    blob_path.chmod(int(cred.get("mode", "0600"), 8))
    print(f"  wrote {blob_path}")
    user_id = _get_or_prompt_user_id(manifest)
    _aws_s3_sync(str(AUTH_DIR), f"s3://{AUTH_BUCKET}/{user_id}/")
    print("  pushed to S3")
    return 0


# ── Subcommand: status ────────────────────────────────────────────────


def cmd_status(_args: argparse.Namespace) -> int:
    """Show current state of code + creds layers."""
    manifest = _load_manifest()
    if not manifest.get("dotfiles_repo"):
        print("not migrated. Run `lumalabs-skills migrate`.")
        return 1
    user_id = manifest.get("coder_user_id", "(not set — pulls will prompt)")
    print(f"Dotfiles repo: {manifest['dotfiles_repo']}")
    print(f"Coder user-id: {user_id}")
    print(f"Auth bucket:   s3://{AUTH_BUCKET}/")
    print()
    print(f"Credentials ({len(manifest['credentials'])}):")
    for c in manifest["credentials"]:
        target = HOME / c["file"]
        present = "✓" if target.exists() or (AUTH_DIR / c["auth_blob"]).exists() else "✗"
        print(f"  {present}  {c['file']}  →  {c['auth_blob']}")
    print()
    print("Git status:")
    _run(["git", "-C", str(CLAUDE_DIR), "status", "--short"])
    return 0


# ── Entrypoint ────────────────────────────────────────────────────────


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser(prog="lumalabs-skills", description=__doc__.strip().split("\n")[0])
    sub = p.add_subparsers(dest="cmd", required=True)
    sub.add_parser("migrate", help="One-time setup")
    sub.add_parser("push", help="git push + s3 sync (laptop → cloud)")
    sub.add_parser("pull", help="git pull + s3 sync (cloud → laptop)")
    rot = sub.add_parser("rotate", help="Rotate a credential")
    rot.add_argument("cred", help="Credential filename or auth-blob name (e.g., linear_api_key)")
    sub.add_parser("status", help="Show state")

    args = p.parse_args(argv)
    handler = {
        "migrate": cmd_migrate,
        "push": cmd_push,
        "pull": cmd_pull,
        "rotate": cmd_rotate,
        "status": cmd_status,
    }[args.cmd]
    try:
        return handler(args)
    except subprocess.CalledProcessError as e:
        print(f"error: command failed: {' '.join(e.cmd)}", file=sys.stderr)
        return e.returncode
    except KeyboardInterrupt:
        print("\naborted")
        return 130


if __name__ == "__main__":
    sys.exit(main())