Backup

#!/usr/bin/env python3
"""Delete a backup from sc-agent-backup storage.

Usage:
    python3 delete.py <backup_id> --confirm <backup_id>

Both positional `<backup_id>` and the `--confirm` value must match EXACTLY.
This is a tripwire, not a substitute for user confirmation — the SKILL.md
Flow C requires the agent to obtain TWO distinct user confirmations before
calling this script at all (see SKILL.md §C.3 and §C.4).

Before issuing DELETE, the script calls `GET /backups` to fetch the target
backup's metadata (label / created_at / size / sections). It prints that
metadata to stdout as JSON so the agent can tell the user what was deleted.

Environment:
    CONTAINER_JWT         Required.
    BACKUP_STORAGE_URL    Optional. Defaults to http://sc-agent-backup.internal:8080.

Exit codes:
    0   backup deleted (server returned 204). stdout has the deleted backup's
        metadata as JSON.
    1   network / auth / 404 / other HTTP failure. Error on stderr.
    2   --confirm value did NOT match backup_id (tripwire triggered —
        the agent likely skipped the double-confirm flow).
"""
from __future__ import annotations

import argparse
import json
import os
import random
import re
import sys
import time
import urllib.error
import urllib.request

DEFAULT_STORAGE = "http://sc-agent-backup.internal:8080"
TIMEOUT = 30
RETRY_BASE_DELAY = 1.0
RETRY_JITTER = 2.0

_BACKUP_ID_RE = re.compile(r"^bk_\d{8}_\d{6}_[a-z0-9]{4}$")


def _storage_url() -> str:
    return os.environ.get("BACKUP_STORAGE_URL", DEFAULT_STORAGE).rstrip("/")


def _jwt() -> str:
    token = os.environ.get("CONTAINER_JWT", "").strip()
    if not token:
        print("ERROR: CONTAINER_JWT env var is not set.", file=sys.stderr)
        sys.exit(1)
    return token


class _NetworkError(Exception):
    """Transient transport failure worth one jittered retry."""


def _with_retry(fn):
    last_err: _NetworkError | None = None
    for attempt in range(2):
        try:
            return fn()
        except _NetworkError as e:
            last_err = e
            if attempt == 0:
                delay = RETRY_BASE_DELAY + random.uniform(0, RETRY_JITTER)
                print(
                    f"WARN: transient network error ({e}); "
                    f"retrying once in {delay:.1f}s...",
                    file=sys.stderr,
                )
                time.sleep(delay)
    print(
        "ERROR: cannot reach backup storage over Fly internal network. "
        "This script must run inside a Fly machine.",
        file=sys.stderr,
    )
    print(f"DETAIL: {last_err}", file=sys.stderr)
    sys.exit(1)


def _fetch_backup_meta(backup_id: str) -> dict | None:
    """GET /backups, return the entry matching backup_id (or None).

    Using list-then-match instead of a direct HEAD/GET on /backups/{id}:
    listing already filters by JWT tenant, so we get the metadata + a
    no-match acts as the 'already gone' signal without an extra round trip.
    """
    def _do():
        req = urllib.request.Request(
            f"{_storage_url()}/backups",
            headers={"Authorization": f"Bearer {_jwt()}"},
            method="GET",
        )
        try:
            with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:
                return resp.status, resp.read()
        except urllib.error.HTTPError as e:
            return e.code, e.read()
        except (urllib.error.URLError, OSError) as e:
            raise _NetworkError(str(e)) from e

    status, body = _with_retry(_do)
    if status != 200:
        try:
            err = json.loads(body.decode("utf-8"))
        except Exception:
            err = {"raw": body.decode("utf-8", errors="replace")}
        if status == 401:
            print(f"ERROR: unauthorized — {err.get('message', 'bad JWT')}",
                  file=sys.stderr)
        elif status == 403:
            print("ERROR: forbidden — must run from Fly 6PN", file=sys.stderr)
        else:
            print(f"ERROR: list failed HTTP {status} — {err}", file=sys.stderr)
        sys.exit(1)

    try:
        data = json.loads(body.decode("utf-8"))
    except json.JSONDecodeError:
        print("ERROR: storage returned non-JSON for /backups", file=sys.stderr)
        sys.exit(1)

    for b in data.get("backups", []):
        if b.get("backup_id") == backup_id:
            return b
    return None


def _do_delete(backup_id: str) -> int:
    """Issue DELETE /backups/{id}. Returns HTTP status code."""
    def _do():
        req = urllib.request.Request(
            f"{_storage_url()}/backups/{backup_id}",
            headers={"Authorization": f"Bearer {_jwt()}"},
            method="DELETE",
        )
        try:
            with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:
                return resp.status
        except urllib.error.HTTPError as e:
            return e.code
        except (urllib.error.URLError, OSError) as e:
            raise _NetworkError(str(e)) from e

    return _with_retry(_do)


def main() -> None:
    parser = argparse.ArgumentParser(
        formatter_class=argparse.RawDescriptionHelpFormatter,
        description=__doc__,
    )
    parser.add_argument("backup_id", type=str,
                        help="the backup_id to delete")
    parser.add_argument(
        "--confirm", type=str, required=True,
        help="MUST equal backup_id exactly — tripwire against the agent "
             "skipping the SKILL.md Flow C double-confirmation. Callers "
             "that pass `--confirm` without first getting two distinct "
             "user acknowledgements are violating the skill contract.",
    )
    args = parser.parse_args()

    # ── Pre-flight: id format ─────────────────────────────────────────────
    if not _BACKUP_ID_RE.match(args.backup_id):
        print(f"ERROR: malformed backup_id: {args.backup_id!r} "
              f"(expected bk_YYYYMMDD_HHMMSS_xxxx)", file=sys.stderr)
        sys.exit(1)

    # ── Tripwire: --confirm must match exactly ────────────────────────────
    if args.confirm != args.backup_id:
        print(
            f"ERROR: --confirm value ({args.confirm!r}) does not match "
            f"backup_id ({args.backup_id!r}).\n"
            "This tripwire guards against accidental deletes. Per SKILL.md "
            "Flow C, the agent must obtain TWO user confirmations before "
            "invoking delete.py, and pass the backup_id as --confirm only "
            "once the user has typed it back verbatim (Step C.4).",
            file=sys.stderr,
        )
        sys.exit(2)

    # ── Fetch metadata up-front so we can echo what got deleted ───────────
    meta = _fetch_backup_meta(args.backup_id)
    if meta is None:
        # Same response server-side returns 404 for cross-tenant and
        # genuinely missing — we treat both the same.
        print(
            f"ERROR: backup not found (already deleted, or does not belong "
            f"to this user): {args.backup_id}",
            file=sys.stderr,
        )
        sys.exit(1)

    # ── Issue the DELETE ─────────────────────────────────────────────────
    status = _do_delete(args.backup_id)
    if status == 204:
        result = {
            "deleted_backup_id": meta.get("backup_id"),
            "user_label": meta.get("user_label"),
            "created_at": meta.get("created_at"),
            "size_bytes": meta.get("size_bytes"),
            "sections": meta.get("sections"),
        }
        print(json.dumps(result, ensure_ascii=False))
        sys.exit(0)

    if status == 404:
        # Race: someone else deleted it between our list and delete. Still a
        # success from the user's point of view (the backup is gone). But
        # we report it as an error so the flow is transparent.
        print(
            "ERROR: backup vanished between list and delete (race). "
            "Re-list to confirm it's gone.", file=sys.stderr,
        )
        sys.exit(1)
    if status == 401:
        print("ERROR: unauthorized — CONTAINER_JWT invalid or expired",
              file=sys.stderr)
    elif status == 403:
        print("ERROR: forbidden — must run from Fly 6PN", file=sys.stderr)
    else:
        print(f"ERROR: delete failed HTTP {status}", file=sys.stderr)
    sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Download a backup from sc-agent-backup and extract it into
/data/workspace/.restore/{backup_id}/.

Usage:
    python3 download.py <backup_id>

Environment:
    CONTAINER_JWT         Required.
    BACKUP_STORAGE_URL    Optional. Defaults to http://sc-agent-backup.internal:8080.
    WORKSPACE_DIR         Optional. Defaults to /data/workspace.

Three layers of integrity protection (belt + suspenders + airbag):
  1. X-Sha256 header: server returns the whole-bundle sha256; client streams
     the body to tmp while hashing, rejects on mismatch (catches transport /
     storage corruption).
  2. Tar member path audit: absolute paths and ".." are refused before extract.
  3. Per-file sha256 in manifest.contents: after extract, every file listed in
     manifest is rehashed and compared (catches tar format corruption, an
     edited/mismatched manifest, or post-extract tampering between download
     and apply).

Extraction lands under /data/workspace/.restore/{backup_id}/ so two
concurrent restores of different backups never stomp each other, and
retrying the same backup just re-extracts in place.

Exit codes:
    0   bundle downloaded, extracted, and integrity-verified. Prints a
        human summary to stdout.
    1   network / auth / bundle-corruption / manifest-mismatch. stderr has
        a detailed message.
"""
from __future__ import annotations

import hashlib
import json
import os
import random
import re
import shutil
import sys
import tarfile
import tempfile
import time
import urllib.error
import urllib.request
from pathlib import Path

DEFAULT_STORAGE = "http://sc-agent-backup.internal:8080"
MAX_SIZE = 500 * 1024 * 1024
DOWNLOAD_TIMEOUT = 600  # seconds; room for 500MB over slow links
READ_CHUNK = 64 * 1024
RETRY_BASE_DELAY = 1.0
RETRY_JITTER = 2.0  # jittered retry to avoid thundering-herd on recovery
WORKSPACE = Path(os.environ.get("WORKSPACE_DIR", "/data/workspace"))
RESTORE_ROOT = WORKSPACE / ".restore"

_BACKUP_ID_RE = re.compile(r"^bk_\d{8}_\d{6}_[a-z0-9]{4}$")


def _storage_url() -> str:
    return os.environ.get("BACKUP_STORAGE_URL", DEFAULT_STORAGE).rstrip("/")


def _jwt() -> str:
    token = os.environ.get("CONTAINER_JWT", "").strip()
    if not token:
        print("ERROR: CONTAINER_JWT env var is not set.", file=sys.stderr)
        sys.exit(1)
    return token


class _NetworkError(Exception):
    """Transient transport failure worth one retry."""


def _stream_to_file(backup_id: str, dest: Path) -> str:
    """GET the bundle and stream it chunk-by-chunk to `dest`, returning the
    server's X-Sha256 header. Raises _NetworkError on transport-level failure.
    HTTP errors (401/403/404) call sys.exit(1) directly — not retryable."""
    req = urllib.request.Request(
        f"{_storage_url()}/backups/{backup_id}",
        headers={"Authorization": f"Bearer {_jwt()}"},
        method="GET",
    )
    try:
        resp = urllib.request.urlopen(req, timeout=DOWNLOAD_TIMEOUT)
    except urllib.error.HTTPError as e:
        body = e.read().decode(errors="replace")
        if e.code == 401:
            print("ERROR: unauthorized — CONTAINER_JWT invalid or expired", file=sys.stderr)
        elif e.code == 403:
            print("ERROR: forbidden — must run from Fly 6PN", file=sys.stderr)
        elif e.code == 404:
            print(f"ERROR: backup not found: {backup_id}", file=sys.stderr)
        elif e.code == 400:
            print(f"ERROR: bad backup_id: {body}", file=sys.stderr)
        else:
            print(f"ERROR: HTTP {e.code} — {body}", file=sys.stderr)
        sys.exit(1)
    except (urllib.error.URLError, OSError) as e:
        raise _NetworkError(str(e)) from e

    sha_header = resp.headers.get("X-Sha256", "")
    total = 0
    try:
        with open(dest, "wb") as f:
            while True:
                chunk = resp.read(READ_CHUNK)
                if not chunk:
                    break
                total += len(chunk)
                if total > MAX_SIZE:
                    raise _NetworkError(f"bundle exceeds MAX_SIZE={MAX_SIZE}")
                f.write(chunk)
    except (urllib.error.URLError, OSError) as e:
        raise _NetworkError(str(e)) from e
    finally:
        resp.close()

    return sha_header


def _download_with_retry(backup_id: str, dest: Path) -> str:
    """One-shot, then one jittered retry on transport failure."""
    last_err: _NetworkError | None = None
    for attempt in range(2):
        try:
            return _stream_to_file(backup_id, dest)
        except _NetworkError as e:
            last_err = e
            # Clean up any partial data before retrying so we start from byte 0.
            try:
                dest.unlink(missing_ok=True)
            except Exception:
                pass
            if attempt == 0:
                delay = RETRY_BASE_DELAY + random.uniform(0, RETRY_JITTER)
                print(
                    f"WARN: transient network error ({e}); "
                    f"retrying once in {delay:.1f}s...",
                    file=sys.stderr,
                )
                time.sleep(delay)

    print(
        "ERROR: cannot reach backup storage over Fly internal network. "
        "This script must run inside a Fly machine.",
        file=sys.stderr,
    )
    print(f"DETAIL: {last_err}", file=sys.stderr)
    sys.exit(1)


def _sha256_file(p: Path) -> str:
    h = hashlib.sha256()
    with open(p, "rb") as f:
        for chunk in iter(lambda: f.read(READ_CHUNK), b""):
            h.update(chunk)
    return h.hexdigest()


def download(backup_id: str) -> tuple[Path, str]:
    """Stream the bundle into a local tmp file. Returns (path, sha256_header).

    The caller is responsible for unlinking the returned path after use.
    Memory stays bounded at READ_CHUNK regardless of bundle size.
    """
    tmp = Path(tempfile.mkstemp(suffix=".tar.gz", prefix=f"{backup_id}.")[1])
    sha_header = _download_with_retry(backup_id, tmp)

    if sha_header:
        actual = _sha256_file(tmp)
        if actual != sha_header:
            tmp.unlink(missing_ok=True)
            print(
                f"ERROR: whole-bundle sha256 mismatch — got {actual}, "
                f"expected {sha_header}. Bundle may be corrupt in transit or "
                "on server disk. Pick another backup.",
                file=sys.stderr,
            )
            sys.exit(1)

    return tmp, sha_header


def extract(bundle_path: Path, extract_dir: Path) -> dict:
    """Extract bundle into extract_dir. Returns the parsed manifest.json.

    If extract_dir already exists it's wiped first — retrying the same
    backup replaces the prior extract atomically (from the user's POV).
    """
    if extract_dir.exists():
        shutil.rmtree(extract_dir)
    extract_dir.mkdir(parents=True, exist_ok=True)

    try:
        with tarfile.open(bundle_path, "r:gz") as tar:
            for member in tar.getmembers():
                if member.name.startswith("/") or ".." in member.name.split("/"):
                    print(f"ERROR: dangerous path in archive: {member.name}",
                          file=sys.stderr)
                    sys.exit(1)
            tar.extractall(extract_dir, filter="data")
    except tarfile.TarError as e:
        print(f"ERROR: invalid tar.gz: {e}", file=sys.stderr)
        sys.exit(1)

    manifest_path = extract_dir / "manifest.json"
    if not manifest_path.exists():
        print("ERROR: bundle has no manifest.json at root", file=sys.stderr)
        sys.exit(1)

    try:
        return json.loads(manifest_path.read_text())
    except json.JSONDecodeError as e:
        print(f"ERROR: invalid manifest.json: {e}", file=sys.stderr)
        sys.exit(1)


def verify_per_file(manifest: dict, extract_dir: Path) -> None:
    """Re-hash every file listed in manifest.contents and compare against the
    recorded sha256. On ANY mismatch, print the full list of bad files to
    stderr and exit 1 — do not let the caller proceed to apply.

    Also checks for files listed in manifest that are missing on disk, AND
    files on disk that aren't listed in manifest (extra files shouldn't be
    applied — they'd indicate a pre-v1.1 bundle or tampering).
    """
    contents = manifest.get("contents")
    if not isinstance(contents, dict):
        # Pre-v1.1 manifest — skip per-file check with a warning.
        print(
            "WARN: bundle manifest has no `contents` section (pre-v1.1 "
            "bundle). Skipping per-file hash verification. Whole-bundle "
            "sha256 was verified at download time.",
            file=sys.stderr,
        )
        return

    expected: dict[str, dict] = {}
    for entry in contents.get("files", []):
        expected[entry["path"]] = entry
    for entry in contents.get("api", []):
        expected[entry["path"]] = entry

    # Walk disk, compare each file.
    on_disk: dict[str, Path] = {}
    for p in extract_dir.rglob("*"):
        if not p.is_file():
            continue
        rel = p.relative_to(extract_dir).as_posix()
        if rel == "manifest.json":
            continue
        on_disk[rel] = p

    bad: list[str] = []
    missing: list[str] = []
    for path, entry in expected.items():
        p = on_disk.get(path)
        if p is None:
            missing.append(path)
            continue
        if p.stat().st_size != entry.get("size"):
            bad.append(f"{path}  (size mismatch: "
                       f"got {p.stat().st_size}, expected {entry.get('size')})")
            continue
        digest = _sha256_file(p)
        if digest != entry.get("sha256"):
            bad.append(f"{path}  (sha256 mismatch: "
                       f"got {digest[:16]}…, expected {entry.get('sha256', '')[:16]}…)")

    extra = [p for p in on_disk if p not in expected]

    if bad or missing or extra:
        print("ERROR: bundle contents verification failed.", file=sys.stderr)
        if bad:
            print(f"  corrupted files ({len(bad)}):", file=sys.stderr)
            for b in bad[:20]:
                print(f"    {b}", file=sys.stderr)
            if len(bad) > 20:
                print(f"    ... ({len(bad) - 20} more)", file=sys.stderr)
        if missing:
            print(f"  missing files listed in manifest ({len(missing)}):",
                  file=sys.stderr)
            for m in missing[:20]:
                print(f"    {m}", file=sys.stderr)
            if len(missing) > 20:
                print(f"    ... ({len(missing) - 20} more)", file=sys.stderr)
        if extra:
            print(f"  extra files NOT in manifest ({len(extra)}):",
                  file=sys.stderr)
            for e in extra[:20]:
                print(f"    {e}", file=sys.stderr)
            if len(extra) > 20:
                print(f"    ... ({len(extra) - 20} more)", file=sys.stderr)
        print(
            "\nDo NOT proceed to restore. Pick another backup, or delete this "
            "one via Flow C and re-take a fresh backup.",
            file=sys.stderr,
        )
        sys.exit(1)


def summarize(manifest: dict, extract_dir: Path) -> None:
    print(f"Bundle extracted to {extract_dir}/")
    print(f"  source    : {manifest.get('source', 'unknown')}")
    print(f"  version   : {manifest.get('version')}")
    print(f"  label     : {manifest.get('label') or '(no label)'}")
    print(f"  created_at: {manifest.get('created_at')}")
    contents = manifest.get("contents", {}) or {}
    files = contents.get("files") or []
    api = contents.get("api") or []
    total_size = sum(e.get("size", 0) for e in files) + sum(e.get("size", 0) for e in api)
    print(f"  files     : {len(files)} (filesystem state) + {len(api)} (api state)")
    print(f"  total     : {total_size:,} bytes")
    print()

    # Group files by top-level section under files/ for a human-readable
    # breakdown (workspace/memory, workspace/prompt, data/, …).
    buckets: dict[str, list[dict]] = {}
    for entry in files:
        parts = entry["path"].split("/", 2)
        # path like "files/workspace/memory/MEMORY.md" → bucket "files/workspace"
        if len(parts) >= 2:
            key = "/".join(parts[:2])
        else:
            key = entry["path"]
        buckets.setdefault(key, []).append(entry)

    if buckets:
        print("Filesystem sections:")
        for key in sorted(buckets):
            items = buckets[key]
            size = sum(e.get("size", 0) for e in items)
            print(f"  {key:<30s} {len(items):>4d} files  {size:>10,} bytes")
        print()

    if api:
        print("API sections:")
        for entry in sorted(api, key=lambda e: e["path"]):
            print(f"  {entry['path']:<30s} {entry.get('size', 0):>10,} bytes")
        print()

    warnings = (manifest.get("plan_summary") or {}).get("warnings") or []
    if warnings:
        print("Pack-time warnings:")
        for w in warnings:
            print(f"  ⚠ {w}")


def main() -> None:
    if len(sys.argv) != 2:
        print("Usage: python3 download.py <backup_id>", file=sys.stderr)
        sys.exit(1)

    backup_id = sys.argv[1].strip()
    if not _BACKUP_ID_RE.match(backup_id):
        print(f"ERROR: invalid backup_id format: {backup_id!r}", file=sys.stderr)
        sys.exit(1)

    extract_dir = RESTORE_ROOT / backup_id

    print(f"Downloading {backup_id} ...")
    bundle_path, sha = download(backup_id)
    size = bundle_path.stat().st_size
    try:
        print(f"Downloaded {size:,} bytes (bundle sha256 verified: {bool(sha)})")
        manifest = extract(bundle_path, extract_dir)
        verify_per_file(manifest, extract_dir)
        print(f"Per-file hash verification: OK "
              f"({len((manifest.get('contents') or {}).get('files', []))} + "
              f"{len((manifest.get('contents') or {}).get('api', []))} files)")
        print()
        summarize(manifest, extract_dir)
    finally:
        bundle_path.unlink(missing_ok=True)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Ensure /data/workspace/config/backup_rules.md exists. Idempotent.

On first run this writes a default template; on subsequent runs it does
nothing (user's edited version is preserved). Prints one line to stdout
so the agent can confirm which branch fired:

    CREATED /data/workspace/config/backup_rules.md
    EXISTS  /data/workspace/config/backup_rules.md

Exit codes:
    0   ok (created, or already existed)
    1   filesystem error (permission, disk full, etc.)

The default template covers every field pack.py can read from rules (mode,
extra excludes, extra paths, label template, notes). All sections are
present-but-inert (comment-only) so the file's mere existence doesn't
change default behavior — user must edit to activate anything.

Language: the template body is in English as a neutral baseline. The
SKILL.md Flow A.1.0 tells the agent to offer localizing it to the user's
conversation language after CREATE. Section headings stay in English in all
cases because the rules parser keys off them.
"""
from __future__ import annotations

import os
import sys
from pathlib import Path

RULES_PATH = Path(os.environ.get(
    "BACKUP_RULES_PATH",
    "/data/workspace/config/backup_rules.md",
))

TEMPLATE = """\
# Backup rules — your backup preferences

When you run `/backup`, the agent reads this file first and applies whatever
you set here to the pack step. Edit it anytime; changes take effect on the
next backup. To reset everything to defaults, delete this file — the agent
will rebuild it on the next run.

> **Language note.** The section headings (`## Mode`, `## Extra excludes`,
> `## Extra paths`, `## Label template`, `## Notes`) must stay in English —
> the agent's parser depends on them. Everything else (intro prose, comments
> inside code blocks, the label template, your notes) can be in any language
> you prefer. Ask the agent to localize this file to your language if you'd
> rather not read English.

---

## Mode

The pack mode applied every time, unless you override in-conversation.

```
default
```

Valid values:
- `default` — skip logs / caches / ChromaDB / runtime transients / skill scratch (recommended)
- `full`    — include everything except this skill's own scratch (`.backup` / `.restore`)

---

## Extra excludes

Paths to skip on top of the default blacklist. One per line, **relative to
`/data/`**. Lines starting with `#` are comments and are ignored.

```
# Examples (remove the leading # to activate):
# workspace/output
# data/.user-packages
```

---

## Extra paths

Additional absolute paths to include from **outside** `/data/`. One per line.

```
# Example:
# /home/myuser/some-config
```

---

## Label template

Default label for each backup. `{date}` is replaced with today's UTC date
(YYYY-MM-DD). A label you provide in-conversation overrides this template.

```
auto-backup {date}
```

---

## Notes

Free-form space for your own reminders. The agent reads this section but
treats it as informational only — it won't execute anything you write here.

Example uses:
- "Do a `full` mode backup every Sunday."
- "If I add a new OPENAI key, test a backup afterward to make sure it picked up."
- "Last data loss happened because I forgot `output/` — don't skip it again."
"""


def main() -> None:
    if RULES_PATH.exists():
        print(f"EXISTS  {RULES_PATH}")
        return

    try:
        RULES_PATH.parent.mkdir(parents=True, exist_ok=True)
        RULES_PATH.write_text(TEMPLATE, encoding="utf-8")
    except OSError as e:
        print(f"ERROR: could not write {RULES_PATH}: {e}", file=sys.stderr)
        sys.exit(1)

    print(f"CREATED {RULES_PATH}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""List this user's backups via GET /backups on sc-agent-backup.

Usage:
    python3 list.py [--json]

By default prints a human-readable menu intended for an agent to surface to
the user as the pick list. With --json, prints the raw storage JSON for
programmatic consumption.

Environment:
    CONTAINER_JWT     Required.
    BACKUP_STORAGE_URL  Optional. Defaults to http://sc-agent-backup.internal:8080.

Exit codes:
    0   request succeeded. Prints the menu (or "（无备份）" if the user has
        no backups yet). The agent reads stdout to decide whether to ask
        "pick one" or "run /backup first".
    1   network / auth / protocol failure (error on stderr)
"""
from __future__ import annotations

import argparse
import json
import os
import sys
import time
import urllib.error
import urllib.request

DEFAULT_STORAGE = "http://sc-agent-backup.internal:8080"


def _storage_url() -> str:
    return os.environ.get("BACKUP_STORAGE_URL", DEFAULT_STORAGE).rstrip("/")


def _jwt() -> str:
    token = os.environ.get("CONTAINER_JWT", "").strip()
    if not token:
        print("ERROR: CONTAINER_JWT env var is not set.", file=sys.stderr)
        sys.exit(1)
    return token


def _fetch() -> dict:
    req = urllib.request.Request(
        f"{_storage_url()}/backups",
        headers={"Authorization": f"Bearer {_jwt()}"},
        method="GET",
    )
    try:
        with urllib.request.urlopen(req, timeout=20) as resp:
            body = resp.read().decode("utf-8")
    except urllib.error.HTTPError as e:
        detail = e.read().decode(errors="replace")
        print(f"ERROR: HTTP {e.code} — {detail}", file=sys.stderr)
        sys.exit(1)
    except (urllib.error.URLError, OSError) as e:
        print(
            "ERROR: cannot reach backup storage over Fly internal network. "
            "This script must run inside a Fly machine.",
            file=sys.stderr,
        )
        print(f"DETAIL: {e}", file=sys.stderr)
        sys.exit(1)
    try:
        return json.loads(body)
    except json.JSONDecodeError:
        print(f"ERROR: storage returned non-JSON: {body[:200]}", file=sys.stderr)
        sys.exit(1)


def _human_size(n: int) -> str:
    for unit in ("B", "KB", "MB", "GB"):
        if n < 1024:
            return f"{n:.1f} {unit}" if unit != "B" else f"{n} {unit}"
        n /= 1024
    return f"{n:.1f} TB"


def _relative_age(ts: int) -> str:
    now = int(time.time())
    d = max(0, now - ts)
    if d < 60:
        return f"{d} 秒前"
    if d < 3600:
        return f"{d // 60} 分钟前"
    if d < 86400:
        return f"{d // 3600} 小时前"
    if d < 30 * 86400:
        return f"{d // 86400} 天前"
    return f"{d // (30 * 86400)} 个月前"


def _format_menu(data: dict) -> str:
    backups = data.get("backups", [])
    quota = data.get("quota", "?")
    if not backups:
        return "（无备份）"

    lines = [f"您的备份（{len(backups)} / {quota}，按时间倒序）："]
    for i, b in enumerate(backups, 1):
        label = b.get("user_label") or "（无标签）"
        ts = b.get("created_at", 0)
        when = time.strftime("%Y-%m-%d %H:%M UTC", time.gmtime(ts)) if ts else "?"
        age = _relative_age(ts) if ts else "?"
        size = _human_size(int(b.get("size_bytes", 0)))
        sections = b.get("sections") or []
        bid = b.get("backup_id", "?")

        lines.append("")
        lines.append(f"[{i}] {bid}")
        lines.append(f"    标签: {label}")
        lines.append(f"    时间: {when}  ({age})")
        lines.append(f"    大小: {size}")
        if sections:
            lines.append(f"    内容: {', '.join(sections)}")
    return "\n".join(lines)


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--json", action="store_true",
                        help="emit raw storage JSON instead of a menu")
    args = parser.parse_args()

    data = _fetch()

    if args.json:
        print(json.dumps(data, ensure_ascii=False, indent=2))
    else:
        print(_format_menu(data))

    # Empty isn't an error — exit 0 and let the agent read stdout to decide
    # what to say. Non-zero is reserved for genuine failures (network/auth).


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Deterministically assemble a backup bundle from a live Starchild agent.

Usage:
    python3 pack.py \\
        --api-dir   $WORK/api/ \\
        --out       $WORK/bundle.tar.gz \\
        [--mode default|full] \\
        [--extra-exclude <rel-path>] [--extra-path <abs-path>] \\
        [--label "升级前"] [--dry-run]

The "what to pack" model is a blacklist, not a whitelist:

  --mode default   walk /data/, pack everything EXCEPT logs, caches, derived
                   indexes (ChromaDB), transient runtime state, and this skill's
                   own scratch/markers. This is what you want 99% of the time.

  --mode full      walk /data/, pack literally everything. Only skip paths
                   that would create a self-reference loop (our own in-progress
                   bundle scratch, our own restore extract dirs).

Both modes always skip `__pycache__/` and `*.pyc` inside any copied directory.

Path arguments:
  --extra-exclude REL    additional path to skip, interpreted relative to
                         /data/ (e.g. "workspace/output" or "sessions").
                         Repeat for multiple.
  --extra-path ABS       extra absolute path to include, outside /data/
                         (e.g. some user-owned dir on another mount).

`--api-dir` must contain agent-supplied JSON written before pack.py runs:
    api-dir/profile.json            agent_profile(action="get") output
    api-dir/settings.json           user_settings(action="get") output
    api-dir/scheduled_tasks.json    scheduled_task(action="list"), normalized to
                                    [{title, schedule, description, channels}]

Convention: the agent allocates a per-run workdir (e.g. `mktemp -d
/tmp/backup-XXXXXX`), creates `$WORK/api/`, writes the three JSONs there, then
calls `pack.py --api-dir $WORK/api --out $WORK/bundle.tar.gz`. After upload
succeeds it `rm -rf $WORK`. Each backup attempt stays isolated.

With --dry-run, nothing is written; a plan (JSON) is printed to stdout.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import os
import shutil
import sys
import tarfile
import time
from pathlib import Path
from typing import Any

# ---------------------------------------------------------------------------
# Blacklists — the authoritative "what to skip" spec. Adding defaults here
# is the right place (keep in sync with SKILL.md's "what's excluded" table).
# ---------------------------------------------------------------------------

# Top-level children of /data/workspace/ we skip in DEFAULT mode.
# In FULL mode we keep ALL of these except the SELF_LOOP ones below.
DEFAULT_SKIP_WORKSPACE_TOP: set[str] = {
    ".active-upload.json",   # client upload-resume state; transient
    ".restore.log",          # restore progress marker; see SKILL.md Step 0
}

# Top-level children of /data/ we skip in DEFAULT mode.
# In FULL mode we keep ALL of these.
DEFAULT_SKIP_DATA_TOP: set[str] = {
    # Logs
    "logs",
    "auto_update.log",
    "skills-install.log",

    # Transient runtime state (resets on restart anyway)
    ".bash_processes",
    ".startup-tasks",
    "hibernation_state.json",

    # Derived / cache / reconstructable
    "memory",            # ChromaDB + FTS indexes; rebuilds from workspace/memory/**
    ".npm-cache",        # npm cache
    ".agents",           # npx skills lock-file shadow; real registry is in workspace/skills/.skill-lock.json
}

# ALWAYS skipped (even in FULL mode) — avoids packaging the backup of itself.
ALWAYS_SKIP_WORKSPACE_TOP: set[str] = {
    ".backup",           # our own pack scratch (where this run lives while running)
    ".restore",          # our own restore extract dirs (per-backup_id subdirs)
}

# Files / dir names to prune INSIDE any copied directory at any depth.
# We exclude these in both default and full mode — they're universally noise.
UNIVERSAL_PRUNE: set[str] = {
    "__pycache__",
}
UNIVERSAL_PRUNE_SUFFIXES: tuple[str, ...] = (".pyc",)

# API JSONs the agent must supply.
REQUIRED_API_FILES: list[str] = [
    "profile.json",
    "settings.json",
    "scheduled_tasks.json",
]


# ---------------------------------------------------------------------------
# Plan construction
# ---------------------------------------------------------------------------

def _copy_ignore(src: str, names: list[str]) -> list[str]:
    """shutil.copytree ignore fn: prune __pycache__ and *.pyc at every depth."""
    skip: list[str] = []
    for n in names:
        if n in UNIVERSAL_PRUNE:
            skip.append(n)
            continue
        if any(n.endswith(suf) for suf in UNIVERSAL_PRUNE_SUFFIXES):
            skip.append(n)
    return skip


def _path_summary(p: Path) -> tuple[bool, int, int]:
    """Return (exists, size_bytes, file_count) with __pycache__/*.pyc ignored."""
    if not p.exists():
        return False, 0, 0
    if p.is_file():
        return True, p.stat().st_size, 1
    total = 0
    count = 0
    for sub in p.rglob("*"):
        if not sub.is_file():
            continue
        # Skip pruned dirs/files in the summary to match what we'll actually pack.
        if UNIVERSAL_PRUNE.intersection(sub.parts):
            continue
        if any(sub.name.endswith(suf) for suf in UNIVERSAL_PRUNE_SUFFIXES):
            continue
        try:
            total += sub.stat().st_size
            count += 1
        except OSError:
            continue
    return True, total, count


def build_plan(
    workspace_dir: Path,
    data_dir: Path,
    api_dir: Path,
    mode: str,
    extra_excludes: set[str],
    extra_paths: list[Path],
    label: str,
) -> dict[str, Any]:
    """Produce a structured plan describing what will be packed.

    Exclusions are resolved in this order (highest priority first):
      1. ALWAYS_SKIP_WORKSPACE_TOP (self-loop guard — can't be overridden)
      2. --extra-exclude user-supplied list (any path under /data/)
      3. DEFAULT_SKIP_* lists (only applied when mode == 'default')
    """
    if mode not in ("default", "full"):
        raise ValueError(f"mode must be 'default' or 'full', got {mode!r}")

    plan: dict[str, Any] = {
        "label": label,
        "mode": mode,
        "workspace_dir": str(workspace_dir),
        "data_dir": str(data_dir),
        "created_at_unix": int(time.time()),
        "items": [],
        "api": [],
        "extra_paths": [],
        "warnings": [],
        "total_files": 0,
        "total_size_bytes": 0,
    }

    # Normalize extra-excludes to paths-relative-to-/data/.
    # "sessions" → /data/sessions, "workspace/output" → /data/workspace/output
    extra_excl_set: set[str] = set()
    for e in extra_excludes:
        e = e.strip().lstrip("/")
        if e:
            extra_excl_set.add(e)

    def _include(src: Path, bundle_rel: str, rel_key: str, kind: str,
                 section: str) -> None:
        exists, size, count = _path_summary(src)
        status = "ok" if exists else "absent"
        plan["items"].append({
            "section": section,
            "kind": kind,
            "src": str(src),
            "bundle": bundle_rel,
            "rel": rel_key,
            "status": status,
            "size_bytes": size,
            "files": count,
        })
        if exists:
            plan["total_files"] += count
            plan["total_size_bytes"] += size

    def _skip(src: Path, bundle_rel: str, rel_key: str, kind: str,
              section: str, reason: str) -> None:
        plan["items"].append({
            "section": section,
            "kind": kind,
            "src": str(src),
            "bundle": bundle_rel,
            "rel": rel_key,
            "status": f"excluded_{reason}",
            "size_bytes": 0,
            "files": 0,
        })

    # ---------- /data/workspace/* ----------
    if workspace_dir.exists():
        for child in sorted(workspace_dir.iterdir()):
            name = child.name
            rel_key = f"workspace/{name}"
            bundle_rel = f"files/workspace/{name}"
            kind = "workspace_dir" if child.is_dir() else "workspace_file"

            # 1. Always-skip (self-loop guard) — cannot be overridden even in full mode.
            if name in ALWAYS_SKIP_WORKSPACE_TOP:
                _skip(child, bundle_rel, rel_key, kind, "workspace", "self_loop")
                continue

            # 2. User override
            if rel_key in extra_excl_set:
                _skip(child, bundle_rel, rel_key, kind, "workspace", "user_rule")
                continue

            # 3. Default-mode skips
            if mode == "default" and name in DEFAULT_SKIP_WORKSPACE_TOP:
                _skip(child, bundle_rel, rel_key, kind, "workspace", "default_rule")
                continue

            _include(child, bundle_rel, rel_key, kind, "workspace")
    else:
        plan["warnings"].append(f"workspace_dir does not exist: {workspace_dir}")

    # ---------- /data/* (everything NOT under workspace/) ----------
    if data_dir.exists():
        for child in sorted(data_dir.iterdir()):
            name = child.name
            if name == "workspace":
                continue  # handled above
            rel_key = name
            bundle_rel = f"files/data/{name}"
            kind = "data_dir" if child.is_dir() else "data_file"

            # 1. User override
            if rel_key in extra_excl_set:
                _skip(child, bundle_rel, rel_key, kind, "data", "user_rule")
                continue

            # 2. Default-mode skips
            if mode == "default" and name in DEFAULT_SKIP_DATA_TOP:
                _skip(child, bundle_rel, rel_key, kind, "data", "default_rule")
                continue

            _include(child, bundle_rel, rel_key, kind, "data")
    else:
        plan["warnings"].append(f"data_dir does not exist: {data_dir}")

    # ---------- Extra user-specified paths (outside /data/) ----------
    for p in extra_paths:
        if not p.is_absolute():
            plan["warnings"].append(f"--extra-path must be absolute: {p}")
            continue
        if p.is_relative_to(workspace_dir):
            bundle_rel = f"files/workspace/{p.relative_to(workspace_dir)}"
        elif p.is_relative_to(data_dir):
            bundle_rel = f"files/data/{p.relative_to(data_dir)}"
        else:
            bundle_rel = f"files/extra/{p.name}"
        exists, size, count = _path_summary(p)
        kind = "workspace_dir" if p.is_dir() else "workspace_file"
        plan["extra_paths"].append({
            "src": str(p),
            "bundle": bundle_rel,
            "status": "ok" if exists else "missing",
            "size_bytes": size,
            "files": count,
        })
        if exists:
            plan["total_files"] += count
            plan["total_size_bytes"] += size

    # ---------- API files ----------
    for name in REQUIRED_API_FILES:
        p = api_dir / name
        if p.exists() and p.is_file():
            plan["api"].append({
                "file": name,
                "status": "ok",
                "size_bytes": p.stat().st_size,
            })
            plan["total_files"] += 1
            plan["total_size_bytes"] += p.stat().st_size
        else:
            plan["api"].append({"file": name, "status": "missing", "size_bytes": 0})
            plan["warnings"].append(
                f"required api file missing: {p} — agent must write this "
                "before pack.py runs"
            )

    return plan


# ---------------------------------------------------------------------------
# Plan execution (actual tar build)
# ---------------------------------------------------------------------------

def execute_plan(plan: dict[str, Any], api_dir: Path, out_path: Path) -> dict[str, Any]:
    """Copy everything the plan says 'ok' into a staging dir, write manifest
    (with per-file sha256), tar.gz to out_path, cleanup.
    """
    out_path.parent.mkdir(parents=True, exist_ok=True)
    staging = out_path.parent / f".{out_path.name}.staging.{os.getpid()}"
    if staging.exists():
        shutil.rmtree(staging)
    root = staging / "backup"
    root.mkdir(parents=True)

    def _copy_into_bundle(src: Path, bundle_rel: str) -> None:
        if not src.exists():
            return
        dst = root / bundle_rel
        dst.parent.mkdir(parents=True, exist_ok=True)
        if src.is_file():
            shutil.copy2(src, dst)
        else:
            if dst.exists():
                shutil.rmtree(dst)
            shutil.copytree(
                src, dst,
                symlinks=False,
                ignore_dangling_symlinks=True,
                ignore=_copy_ignore,
            )

    for item in plan["items"]:
        if item["status"] != "ok":
            continue
        _copy_into_bundle(Path(item["src"]), item["bundle"])
    for extra in plan["extra_paths"]:
        if extra["status"] != "ok":
            continue
        _copy_into_bundle(Path(extra["src"]), extra["bundle"])

    # API dir
    api_bundle_dir = root / "api"
    api_bundle_dir.mkdir(parents=True, exist_ok=True)
    for entry in plan["api"]:
        if entry["status"] != "ok":
            continue
        shutil.copy2(api_dir / entry["file"], api_bundle_dir / entry["file"])

    # Per-file sha256 inventory.
    contents_files: list[dict[str, Any]] = []
    contents_api: list[dict[str, Any]] = []
    for path in sorted(root.rglob("*")):
        if not path.is_file():
            continue
        rel = path.relative_to(root).as_posix()
        if rel == "manifest.json":
            continue
        entry = {
            "path": rel,
            "size": path.stat().st_size,
            "sha256": _sha256_file(path),
        }
        if rel.startswith("api/"):
            contents_api.append(entry)
        else:
            contents_files.append(entry)

    manifest = {
        "version": "1.1",
        "source": "starchild",
        "mode": plan["mode"],
        "created_at_unix": plan["created_at_unix"],
        "created_at": time.strftime(
            "%Y-%m-%dT%H:%M:%SZ", time.gmtime(plan["created_at_unix"])
        ),
        "label": plan["label"],
        "sections": sorted({
            item["bundle"].split("/")[0] + "/" + item["bundle"].split("/")[1]
            for item in plan["items"]
            if item["status"] == "ok" and "/" in item["bundle"]
        }),
        "contents": {
            "files": contents_files,
            "api": contents_api,
        },
        "exclusions": [
            {"section": item["section"], "rel": item["rel"], "reason": item["status"]}
            for item in plan["items"]
            if item["status"].startswith("excluded_")
        ],
        "plan_summary": {
            "total_files": plan["total_files"],
            "total_size_bytes": plan["total_size_bytes"],
            "warnings": plan["warnings"],
        },
    }
    (root / "manifest.json").write_text(
        json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8"
    )

    if out_path.exists():
        out_path.unlink()
    with tarfile.open(out_path, "w:gz") as tar:
        tar.add(root, arcname=".")

    shutil.rmtree(staging)

    return {
        "bundle_path": str(out_path),
        "size_bytes": out_path.stat().st_size,
        "manifest": manifest,
    }


def _sha256_file(p: Path) -> str:
    h = hashlib.sha256()
    with open(p, "rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def _default_out_path() -> Path:
    ts = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime())
    return Path(f"/tmp/backup-{ts}-{os.getpid()}/bundle.tar.gz")


def main() -> None:
    parser = argparse.ArgumentParser(
        formatter_class=argparse.RawDescriptionHelpFormatter,
        description=__doc__,
    )
    parser.add_argument("--workspace-dir", type=Path,
                        default=Path(os.environ.get("WORKSPACE_DIR", "/data/workspace")))
    parser.add_argument("--data-dir", type=Path, default=Path("/data"))
    parser.add_argument("--api-dir", type=Path, required=True,
                        help="directory where the agent wrote profile.json, "
                             "settings.json, scheduled_tasks.json")
    parser.add_argument("--out", type=Path, default=None,
                        help="output bundle path (default: /tmp/backup-{ts}-{pid}/bundle.tar.gz)")
    parser.add_argument("--mode", choices=("default", "full"), default="default",
                        help="default (skip logs/caches/derived/scratch) or "
                             "full (pack everything except self-reference loops)")
    parser.add_argument("--label", type=str, default="",
                        help="user-facing label (≤64 chars)")
    parser.add_argument("--extra-exclude", action="append", default=[],
                        help="extra path to skip, relative to /data/ "
                             "(e.g. 'workspace/output' or 'sessions'). "
                             "Repeat for multiple.")
    parser.add_argument("--extra-path", action="append", default=[],
                        help="absolute path to an extra dir/file to include "
                             "(outside /data/); repeat for multiple.")
    parser.add_argument("--dry-run", action="store_true",
                        help="print the plan as JSON to stdout; don't build a bundle")
    args = parser.parse_args()

    if args.out is None:
        args.out = _default_out_path()

    # ── Early validation: fail loud and specific on the common pitfalls ──
    # Agents sometimes split the flow across multiple bash tool calls; shell
    # vars like $WORK don't persist between calls, so --api-dir can expand
    # to something nonsensical. Detect this before building any plan.
    if not args.api_dir.exists():
        print(
            f"ERROR: --api-dir does not exist: {args.api_dir}\n\n"
            "This usually means one of:\n"
            "  (1) You passed a shell variable like $WORK that wasn't set in "
            "this bash invocation — bash variables do NOT persist across "
            "separate tool calls. Use a literal path string.\n"
            "  (2) You forgot to create the directory and write the three "
            "API JSONs (profile.json, settings.json, scheduled_tasks.json) "
            "before invoking pack.py.\n"
            "\n"
            "Fix: pick a literal timestamped path, create it, write the "
            "JSONs, then pass that exact string to --api-dir. Example:\n"
            "  WORK=/tmp/backup-20260427T164500Z\n"
            "  mkdir -p $WORK/api\n"
            "  # (write the 3 JSONs to $WORK/api/ via your native file tools)\n"
            "  python3 skills/backup/scripts/pack.py \\\n"
            f"      --api-dir {args.api_dir} \\\n"
            "      --out $WORK/bundle.tar.gz \\\n"
            "      ...",
            file=sys.stderr,
        )
        sys.exit(1)

    if not args.api_dir.is_dir():
        print(
            f"ERROR: --api-dir exists but is not a directory: {args.api_dir}",
            file=sys.stderr,
        )
        sys.exit(1)

    # Out path: ensure parent is writable. We don't require --out to exist
    # (pack.py creates it), but we DO need the parent directory to be
    # writable — catching this early saves the agent from a confusing
    # traceback halfway through plan building.
    out_parent = args.out.parent
    try:
        out_parent.mkdir(parents=True, exist_ok=True)
    except OSError as e:
        print(
            f"ERROR: cannot create --out parent directory {out_parent}: {e}",
            file=sys.stderr,
        )
        sys.exit(1)
    if not os.access(out_parent, os.W_OK):
        print(
            f"ERROR: --out parent is not writable: {out_parent}",
            file=sys.stderr,
        )
        sys.exit(1)

    extra_paths = [Path(p) for p in args.extra_path]
    extra_excludes = set(args.extra_exclude)

    plan = build_plan(
        workspace_dir=args.workspace_dir.resolve(),
        data_dir=args.data_dir.resolve(),
        api_dir=args.api_dir.resolve(),
        mode=args.mode,
        extra_excludes=extra_excludes,
        extra_paths=extra_paths,
        label=args.label,
    )

    if args.dry_run:
        print(json.dumps(plan, ensure_ascii=False, indent=2))
        return

    missing_api = [a["file"] for a in plan["api"] if a["status"] != "ok"]
    if missing_api:
        print(
            "ERROR: required API files not found in --api-dir: "
            f"{', '.join(missing_api)}.\nAgent must call profile / settings / "
            "scheduled_tasks tools and write the results there before pack.py.",
            file=sys.stderr,
        )
        sys.exit(1)

    result = execute_plan(plan, args.api_dir.resolve(), args.out.resolve())
    print(json.dumps(result, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Scan /data/workspace/ and /data/ top-level entries, classify each, and
emit a JSON proposal for the agent to discuss with the user.

Runs on the FIRST backup (when /data/workspace/config/backup_rules.md is
absent). The agent uses the output to show the user what's actually on disk,
discuss the ambiguous items, and compose the final rules file based on their
decisions.

Output JSON shape:
    {
      "workspace": [
        {"name": ..., "path": ..., "is_dir": ..., "size_bytes": ...,
         "file_count": ..., "category": ..., "reason": ...},
        ...
      ],
      "data": [ ... same shape, path under /data/ ... ],
      "summary": {"total_size_bytes": ..., "core_count": N, "skip_count": N,
                  "ask_count": N, "unknown_count": N}
    }

Categories:
    core     — definitely back up (core user state; stock default includes)
    skip     — definitely skip (logs/cache/derived/runtime; stock default
               already excludes; shown for transparency)
    ask      — default includes, user might want to skip (large or optional)
    unknown  — path not on any list; user decides

Exit codes:
    0   ok, JSON on stdout
    1   workspace or data dir missing / unreadable
"""
from __future__ import annotations

import json
import os
import sys
from pathlib import Path

# ---------------------------------------------------------------------------
# Classification sets — keep in sync with pack.py's DEFAULT_SKIP/ALWAYS_SKIP.
# The "reason" strings are what the agent surfaces to the user.
# ---------------------------------------------------------------------------

# Always-backup under workspace/
CORE_WORKSPACE: dict[str, str] = {
    "memory": "agent memory (MEMORY.md + daily/ + topics/)",
    "prompt": "agent identity (SOUL.md + USER.md + optionally AGENTS.md)",
    "config": "per-agent config (agent.yaml / custom_models.yaml / backup_rules.md itself)",
    "tasks": "scheduled task scripts",
    "skills": "installed skills (including .skill-lock.json manifest)",
    "setup.sh": "startup hook (apt/pip install steps)",
    ".env": "secrets (API keys, tokens, bot credentials)",
}

# Always-skip under workspace/ — stock default blacklist + self-loop guards
SKIP_WORKSPACE: dict[str, str] = {
    ".backup": "pack scratch — self-loop guard",
    ".restore": "restore extract — self-loop guard",
    ".active-upload.json": "upload resume state — transient",
    ".restore.log": "restore progress marker — transient",
}

# Always-backup directly under /data/
CORE_DATA: dict[str, str] = {
    "scheduled_jobs.json": "scheduler registry (task definitions)",
    "preview_history.json": "permanent preview service history",
    "previews.json": "active preview registry",
    "sessions": "chat history SQLite family (state.db + -wal + -shm)",
}

# Always-skip directly under /data/ — stock default blacklist
SKIP_DATA: dict[str, str] = {
    "logs": "logs, regenerated",
    "memory": "ChromaDB + FTS + embedding cache (derivable from workspace/memory/**)",
    ".npm-cache": "npm cache",
    ".bash_processes": "process registry — resets on container restart",
    ".startup-tasks": "startup transient status",
    "hibernation_state.json": "container runtime state",
    "auto_update.log": "logs",
    "skills-install.log": "logs",
    "workspace": "sub-tree handled separately",  # placeholder; not actually skipped
}

# Default backs up, but user often wants to decide — "ask" category
ASK_WORKSPACE: dict[str, str] = {
    "output": "agent-generated reports (could be large or sensitive)",
    "scripts": "user-authored custom scripts",
}

ASK_DATA: dict[str, str] = {
    "scheduled_jobs.db": "job execution history (definitions live in scheduled_jobs.json; this is the run log)",
    "tasks.json": "subagent spawn history",
    ".user-packages": "pip/npm --user installs; often large; THEORETICALLY rebuildable via setup.sh",
    ".local": "typically user-installed binaries + app data; varies by agent",
    ".agents": "npx skills shadow lock-file (real registry is workspace/skills/.skill-lock.json)",
}


def _size_and_count(p: Path) -> tuple[int, int]:
    """Return (size_bytes, file_count) for a path (file or dir, recursive).

    Skips __pycache__ and *.pyc to match what pack.py will actually pack.
    """
    if p.is_file():
        try:
            return p.stat().st_size, 1
        except OSError:
            return 0, 0
    total = 0
    count = 0
    for sub in p.rglob("*"):
        if not sub.is_file():
            continue
        if "__pycache__" in sub.parts or sub.name.endswith(".pyc"):
            continue
        try:
            total += sub.stat().st_size
            count += 1
        except OSError:
            continue
    return total, count


def classify_child(name: str, is_dir: bool, base: str) -> tuple[str, str]:
    """Return (category, reason) for a top-level child under workspace or data.

    base is "workspace" or "data".
    """
    if base == "workspace":
        if name in CORE_WORKSPACE:
            return "core", CORE_WORKSPACE[name]
        if name in SKIP_WORKSPACE:
            return "skip", SKIP_WORKSPACE[name]
        if name in ASK_WORKSPACE:
            return "ask", ASK_WORKSPACE[name]
    else:  # data
        if name == "workspace":
            return "core", "the workspace directory (enumerated separately above)"
        if name in CORE_DATA:
            return "core", CORE_DATA[name]
        if name in SKIP_DATA:
            return "skip", SKIP_DATA[name]
        if name in ASK_DATA:
            return "ask", ASK_DATA[name]

    # Unrecognized — punt to user.
    kind = "directory" if is_dir else "file"
    return "unknown", f"unrecognized {kind} under {base}/; user must decide"


def _scan_children(base_dir: Path, label: str) -> list[dict]:
    entries: list[dict] = []
    for child in sorted(base_dir.iterdir()):
        # Under /data/, skip 'workspace' — emitted as its own section.
        if label == "data" and child.name == "workspace":
            continue
        size, count = _size_and_count(child)
        cat, reason = classify_child(child.name, child.is_dir(), label)
        entries.append({
            "name": child.name,
            "path": str(child),
            "is_dir": child.is_dir(),
            "size_bytes": size,
            "file_count": count,
            "category": cat,
            "reason": reason,
        })
    return entries


def propose(workspace_dir: Path, data_dir: Path) -> dict:
    if not workspace_dir.exists():
        print(f"ERROR: workspace dir does not exist: {workspace_dir}",
              file=sys.stderr)
        sys.exit(1)
    if not data_dir.exists():
        print(f"ERROR: data dir does not exist: {data_dir}",
              file=sys.stderr)
        sys.exit(1)

    ws = _scan_children(workspace_dir, "workspace")
    dt = _scan_children(data_dir, "data")

    counts = {"core": 0, "skip": 0, "ask": 0, "unknown": 0}
    total = 0
    for e in ws + dt:
        counts[e["category"]] += 1
        total += e["size_bytes"]

    return {
        "workspace": ws,
        "data": dt,
        "summary": {
            "total_size_bytes": total,
            "core_count": counts["core"],
            "skip_count": counts["skip"],
            "ask_count": counts["ask"],
            "unknown_count": counts["unknown"],
        },
    }


def main() -> None:
    workspace = Path(os.environ.get("WORKSPACE_DIR", "/data/workspace"))
    data = Path(os.environ.get("DATA_DIR", "/data"))
    print(json.dumps(propose(workspace, data), ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()