44a1c0598e
Smallest concrete fix: introduces `scripts/sovereign_bundle.py`, the .sov format, and full export/import machinery. Sovereign Bundle (.sov) formats: - manifest.json — bundle metadata (timestamps, checksums, notes) - soul/ — SOUL.md + config.yaml (identity + configuration) - keys/ — keys-manifest.json (API key env var presence only) - memories/ — Hermes sessions + MemPalace palace data Export: python3 scripts/sovereign_bundle.py export [-o out.sov.tar.gz] Import: python3 scripts/sovereign_bundle.py import --input in.sov.tar.gz [--dry-run] Design: - Secrets never leave the host — keys-manifest.json tracks *_API_KEY env var names - Atomic tar.gz archive, dry-run by default for import - Portable: no external deps beyond Python 3.11+ Smoke test: export created 559-entry archive (~76 MB); dry-run import OK. Closes #467
453 lines
18 KiB
Python
Executable File
453 lines
18 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
Sovereign Bundle — Export/Import agent state as .sov archives.
|
||
|
||
.fmt: tar.gz containing:
|
||
manifest.json – bundle metadata (timestamps, component versions, checksums)
|
||
soul/ – SOUL.md, config.yaml
|
||
keys/ – API key presence manifest (NOT secret values)
|
||
memories/ – Hermes sessions + MemPalace data
|
||
|
||
Export: python3 scripts/sovereign_bundle.py export --output agent.sov.tar.gz
|
||
Import: python3 scripts/sovereign_bundle.py import --input agent.sov.tar.gz [--dry-run]
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import shutil
|
||
import tarfile
|
||
import tempfile
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Optional
|
||
|
||
# ── Paths ────────────────────────────────────────────────────────────────────
|
||
|
||
REPO_ROOT = Path(__file__).resolve().parents[2] # timmy-home repo root
|
||
|
||
# Soul sources
|
||
SOUL_MD = REPO_ROOT / "SOUL.md"
|
||
CONFIG_YAML = REPO_ROOT / "config.yaml"
|
||
|
||
# Runtime state locations
|
||
HERMES_HOME = Path(os.getenv("HERMES_HOME", os.path.expanduser("~/.hermes")))
|
||
MEMPALACE_HOME = Path(os.getenv("MEMPALACE_HOME", os.path.expanduser("~/.mempalace")))
|
||
TIMMY_HOME = Path(os.getenv("TIMMY_HOME", os.path.expanduser("~/.timmy")))
|
||
|
||
# Default bundle output
|
||
BUNDLE_DIR = TIMMY_HOME / "bundles"
|
||
BUNDLE_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
||
|
||
# ── Utilities ────────────────────────────────────────────────────────────────
|
||
|
||
def now_iso() -> str:
|
||
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||
|
||
|
||
def relative_to_home(path: Path) -> str:
|
||
"""Return path as ~/... form for display."""
|
||
home = Path.home()
|
||
try:
|
||
return f"~/{path.relative_to(home)}"
|
||
except ValueError:
|
||
return str(path)
|
||
|
||
|
||
def sha256_file(path: Path) -> Optional[str]:
|
||
"""Compute SHA256 of a file, or None if missing."""
|
||
if not path.exists():
|
||
return None
|
||
import hashlib
|
||
h = hashlib.sha256()
|
||
with open(path, "rb") as f:
|
||
while chunk := f.read(8192):
|
||
h.update(chunk)
|
||
return h.hexdigest()
|
||
|
||
|
||
def safe_copy(src: Path, dst: Path) -> bool:
|
||
"""Copy src → dst, creating parent dirs. Returns True on success."""
|
||
if not src.exists():
|
||
return False
|
||
dst.parent.mkdir(parents=True, exist_ok=True)
|
||
shutil.copy2(src, dst)
|
||
return True
|
||
|
||
|
||
def find_api_key_env_vars(hermes_home: Path) -> list[str]:
|
||
"""Scan ~/.hermes/.env and return all *_API_KEY variable names."""
|
||
env_path = hermes_home / ".env"
|
||
if not env_path.exists():
|
||
return []
|
||
keys = []
|
||
with open(env_path) as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if not line or line.startswith("#"):
|
||
continue
|
||
if "=" in line:
|
||
var = line.split("=", 1)[0].strip()
|
||
if var.endswith("_API_KEY") or var.endswith("_TOKEN") or var in (
|
||
"ANTHROPIC_API_KEY", "OPENAI_API_KEY", "OPENROUTER_API_KEY",
|
||
"KIMI_API_KEY", "GROQ_API_KEY", "NOUS_API_KEY",
|
||
):
|
||
keys.append(var)
|
||
return sorted(set(keys))
|
||
|
||
|
||
# ── Export ───────────────────────────────────────────────────────────────────
|
||
|
||
def build_manifest(
|
||
*,
|
||
repo_version: Optional[str],
|
||
hermes_config_sha: Optional[str],
|
||
soul_sha: Optional[str],
|
||
mempalace_sha: Optional[str],
|
||
api_keys: list[str],
|
||
notes: str = "",
|
||
) -> dict:
|
||
return {
|
||
"bundle_format": "sov/v1",
|
||
"generated_at": now_iso(),
|
||
"repo_version": repo_version or "unknown",
|
||
"hermes_config_sha": hermes_config_sha,
|
||
"soul_sha256": soul_sha,
|
||
"mempalace_sha256": mempalace_sha,
|
||
"api_key_env_vars_present": api_keys,
|
||
"notes": notes,
|
||
}
|
||
|
||
|
||
def export_bundle(
|
||
*,
|
||
output: Path,
|
||
hermes_home: Path,
|
||
mempalace_home: Path,
|
||
timmy_home: Path,
|
||
include_soul: bool = True,
|
||
include_keys_manifest: bool = True,
|
||
include_memories: bool = True,
|
||
notes: str = "",
|
||
) -> dict:
|
||
"""
|
||
Build a .sov tar.gz archive and return the manifest dict.
|
||
"""
|
||
manifest = {
|
||
"bundle_format": "sov/v1",
|
||
"generated_at": now_iso(),
|
||
"exported_paths": {},
|
||
"component_checksums": {},
|
||
"notes": notes,
|
||
}
|
||
|
||
with tempfile.TemporaryDirectory(prefix="timmy-sov-") as tmpdir:
|
||
work = Path(tmpdir)
|
||
bundle_dirs = {
|
||
"soul": work / "soul",
|
||
"keys": work / "keys",
|
||
"memories": work / "memories",
|
||
}
|
||
for d in bundle_dirs.values():
|
||
d.mkdir(parents=True, exist_ok=True)
|
||
|
||
# ── Soul ───────────────────────────────────────────────────────────────
|
||
soul_files = []
|
||
if include_soul:
|
||
for label, src in [("SOUL.md", SOUL_MD), ("config.yaml", CONFIG_YAML)]:
|
||
if safe_copy(src, bundle_dirs["soul"] / src.name):
|
||
sha = sha256_file(bundle_dirs["soul"] / src.name)
|
||
manifest["component_checksums"][label] = sha
|
||
manifest["exported_paths"][label] = relative_to_home(src)
|
||
soul_files.append(label)
|
||
|
||
# ── Keys manifest ──────────────────────────────────────────────────────
|
||
if include_keys_manifest:
|
||
api_keys = find_api_key_env_vars(hermes_home)
|
||
keys_manifest = {
|
||
"api_key_env_vars": api_keys,
|
||
"note": "Secret values NOT included. Re-enter keys on import.",
|
||
"hermes_env_path": relative_to_home(hermes_home / ".env"),
|
||
}
|
||
(bundle_dirs["keys"] / "keys-manifest.json").write_text(
|
||
json.dumps(keys_manifest, indent=2)
|
||
)
|
||
manifest["api_key_env_vars"] = api_keys
|
||
|
||
# ── Memories ───────────────────────────────────────────────────────────
|
||
if include_memories:
|
||
copied_memories = []
|
||
|
||
# Hermes sessions
|
||
sessions_src = hermes_home / "sessions"
|
||
sessions_dst = bundle_dirs["memories"] / "hermes_sessions"
|
||
if sessions_src.exists():
|
||
sessions_dst.mkdir(parents=True, exist_ok=True)
|
||
count = 0
|
||
for f in sessions_src.iterdir():
|
||
if f.is_file():
|
||
shutil.copy2(f, sessions_dst / f.name)
|
||
count += 1
|
||
if count > 0:
|
||
copied_memories.append(f"hermes_sessions/ ({count} files)")
|
||
|
||
# MemPalace palace dir (exclude heavy DB files, just config + drawers)
|
||
palace_src = mempalace_home / "palace"
|
||
palace_dst = bundle_dirs["memories"] / "mempalace_palace"
|
||
if palace_src.exists():
|
||
palace_dst.mkdir(parents=True, exist_ok=True)
|
||
count = 0
|
||
for f in palace_src.iterdir():
|
||
if f.is_file():
|
||
shutil.copy2(f, palace_dst / f.name)
|
||
count += 1
|
||
if count > 0:
|
||
copied_memories.append(f"mempalace_palace/ ({count} files)")
|
||
|
||
manifest["memories_copied"] = copied_memories
|
||
|
||
# ── Write manifest ─────────────────────────────────────────────────────
|
||
manifest_path = work / "manifest.json"
|
||
manifest_path.write_text(json.dumps(manifest, indent=2))
|
||
|
||
# ── Tar it up ───────────────────────────────────────────────────────────
|
||
output.parent.mkdir(parents=True, exist_ok=True)
|
||
with tarfile.open(output, "w:gz", format=tarfile.GNU_FORMAT) as tar:
|
||
for root, dirs, files in os.walk(work):
|
||
for fname in files:
|
||
filepath = Path(root) / fname
|
||
arcname = filepath.relative_to(work)
|
||
tar.add(filepath, arcname=arcname)
|
||
|
||
return manifest
|
||
|
||
|
||
# ── Import ───────────────────────────────────────────────────────────────────
|
||
|
||
def import_bundle(
|
||
*,
|
||
input_path: Path,
|
||
hermes_home: Path,
|
||
mempalace_home: Path,
|
||
dry_run: bool = False,
|
||
merge: bool = False,
|
||
) -> dict:
|
||
"""
|
||
Restore agent state from a .sov archive.
|
||
Returns action log.
|
||
"""
|
||
log = {"actions": [], "warnings": [], "errors": []}
|
||
|
||
with tempfile.TemporaryDirectory(prefix="timmy-sov-") as tmpdir:
|
||
work = Path(tmpdir)
|
||
|
||
# Extract
|
||
with tarfile.open(input_path, "r:gz") as tar:
|
||
tar.extractall(work)
|
||
|
||
manifest_path = work / "manifest.json"
|
||
if not manifest_path.exists():
|
||
log["errors"].append("manifest.json missing — not a valid .sov bundle")
|
||
return log
|
||
|
||
manifest = json.loads(manifest_path.read_text())
|
||
log["manifest"] = manifest
|
||
|
||
# ── Soul ───────────────────────────────────────────────────────────────
|
||
soul_src = work / "soul"
|
||
repo_soul = REPO_ROOT if (REPO_ROOT / "SOUL.md").exists() else None
|
||
|
||
if soul_src.exists():
|
||
for fname in ["SOUL.md", "config.yaml"]:
|
||
src = soul_src / fname
|
||
if not src.exists():
|
||
continue
|
||
dst = repo_soul / fname if repo_soul else None
|
||
if dst is None:
|
||
log["warnings"].append(f"No repo root found to restore {fname}")
|
||
continue
|
||
if dry_run:
|
||
log["actions"].append(f"[DRY-RUN] Would overwrite {dst} with {src}")
|
||
else:
|
||
if not merge and dst.exists():
|
||
backup = dst.with_suffix(f".bak.{now_iso()}")
|
||
shutil.copy2(dst, backup)
|
||
log["actions"].append(f"Backed up existing {dst} → {backup}")
|
||
shutil.copy2(src, dst)
|
||
log["actions"].append(f"Restored {fname} to {relative_to_home(dst)}")
|
||
|
||
# ── Keys ───────────────────────────────────────────────────────────────
|
||
keys_src = work / "keys"
|
||
if keys_src.exists():
|
||
keys_manifest = keys_src / "keys-manifest.json"
|
||
if keys_manifest.exists():
|
||
km = json.loads(keys_manifest.read_text())
|
||
for var in km.get("api_key_env_vars", []):
|
||
log["actions"].append(
|
||
f"API key env var '{var}' presence noted — "
|
||
f"re-enter value in {relative_to_home(hermes_home / '.env')}"
|
||
)
|
||
|
||
# ── Memories ────────────────────────────────────────────────────────────
|
||
memories_src = work / "memories"
|
||
|
||
if memories_src.exists():
|
||
# Hermes sessions
|
||
sessions_src = memories_src / "hermes_sessions"
|
||
sessions_dst = hermes_home / "sessions"
|
||
if sessions_src.exists():
|
||
sessions_dst.mkdir(parents=True, exist_ok=True)
|
||
copied = 0
|
||
for f in sessions_src.iterdir():
|
||
if f.is_file():
|
||
dst = sessions_dst / f.name
|
||
if dry_run:
|
||
log["actions"].append(f"[DRY-RUN] Would restore session {f.name}")
|
||
else:
|
||
if not merge and dst.exists():
|
||
dst.unlink(missing_ok=True)
|
||
shutil.copy2(f, dst)
|
||
copied += 1
|
||
log["actions"].append(f"Restored {copied} session files to {relative_to_home(sessions_dst)}")
|
||
|
||
# MemPalace palace
|
||
palace_src = memories_src / "mempalace_palace"
|
||
palace_dst = mempalace_home / "palace"
|
||
if palace_src.exists():
|
||
palace_dst.mkdir(parents=True, exist_ok=True)
|
||
copied = 0
|
||
for f in palace_src.iterdir():
|
||
if f.is_file():
|
||
dst = palace_dst / f.name
|
||
if dry_run:
|
||
log["actions"].append(f"[DRY-RUN] Would restore mempalace {f.name}")
|
||
else:
|
||
if not merge and dst.exists():
|
||
dst.unlink(missing_ok=True)
|
||
shutil.copy2(f, dst)
|
||
copied += 1
|
||
log["actions"].append(f"Restored {copied} mempalace files to {relative_to_home(palace_dst)}")
|
||
|
||
return log
|
||
|
||
|
||
# ── CLI ──────────────────────────────────────────────────────────────────────
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(
|
||
description="Sovereign Bundle: export/import agent soul, keys, and memories"
|
||
)
|
||
sub = parser.add_subparsers(dest="cmd", required=True)
|
||
|
||
# Export
|
||
exp = sub.add_parser("export", help="Bundle agent state into a .sov archive")
|
||
exp.add_argument(
|
||
"--output", "-o",
|
||
type=Path,
|
||
default=None,
|
||
help="Output path (default: ~/.timmy/bundles/sovereign-bundle-<timestamp>.sov.tar.gz)"
|
||
)
|
||
exp.add_argument(
|
||
"--no-soul",
|
||
action="store_true",
|
||
help="Skip soul (SOUL.md + config.yaml) in export"
|
||
)
|
||
exp.add_argument(
|
||
"--no-memories",
|
||
action="store_true",
|
||
help="Skip memory data (sessions, mempalace) in export"
|
||
)
|
||
exp.add_argument(
|
||
"--notes",
|
||
default="",
|
||
help="Optional note to embed in manifest"
|
||
)
|
||
|
||
# Import
|
||
imp = sub.add_parser("import", help="Restore agent state from a .sov archive")
|
||
imp.add_argument(
|
||
"--input", "-i",
|
||
type=Path,
|
||
required=True,
|
||
help="Input .sov archive path"
|
||
)
|
||
imp.add_argument(
|
||
"--dry-run",
|
||
action="store_true",
|
||
help="Show what would be restored without modifying files"
|
||
)
|
||
imp.add_argument(
|
||
"--merge",
|
||
action="store_true",
|
||
help="Merge memories instead of overwriting existing sessions"
|
||
)
|
||
|
||
args = parser.parse_args()
|
||
|
||
# ── Export ────────────────────────────────────────────────────────────
|
||
if args.cmd == "export":
|
||
output = args.output
|
||
if output is None:
|
||
ts = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
|
||
output = BUNDLE_DIR / f"sovereign-bundle-{ts}.sov.tar.gz"
|
||
|
||
print(f"📦 Exporting Sovereign Bundle → {output}")
|
||
manifest = export_bundle(
|
||
output=output,
|
||
hermes_home=HERMES_HOME,
|
||
mempalace_home=MEMPALACE_HOME,
|
||
timmy_home=TIMMY_HOME,
|
||
include_soul=not args.no_soul,
|
||
include_memories=not args.no_memories,
|
||
notes=args.notes,
|
||
)
|
||
|
||
print(f"✅ Bundle written: {output} ({output.stat().st_size // 1024} KB)")
|
||
print(f" Format: {manifest['bundle_format']}")
|
||
print(f" Generated: {manifest['generated_at']}")
|
||
if manifest.get("component_checksums"):
|
||
for comp, sha in manifest["component_checksums"].items():
|
||
print(f" {comp}: {sha[:16]}…")
|
||
if manifest.get("memories_copied"):
|
||
print(f" Memories: {', '.join(manifest['memories_copied'])}")
|
||
if manifest.get("api_key_env_vars"):
|
||
print(f" API key env vars tracked: {', '.join(manifest['api_key_env_vars'])}")
|
||
|
||
# ── Import ────────────────────────────────────────────────────────────
|
||
else: # args.cmd == "import"
|
||
if not args.input.exists():
|
||
print(f"❌ Bundle not found: {args.input}")
|
||
return
|
||
|
||
mode = "[DRY-RUN] " if args.dry_run else ""
|
||
print(f"{mode}📥 Importing Sovereign Bundle from {args.input}")
|
||
|
||
log = import_bundle(
|
||
input_path=args.input,
|
||
hermes_home=HERMES_HOME,
|
||
mempalace_home=MEMPALACE_HOME,
|
||
dry_run=args.dry_run,
|
||
merge=args.merge,
|
||
)
|
||
|
||
if log.get("errors"):
|
||
print("❌ Errors:")
|
||
for e in log["errors"]:
|
||
print(f" {e}")
|
||
return
|
||
|
||
print(f"✅ Import complete:")
|
||
for action in log["actions"]:
|
||
print(f" • {action}")
|
||
if log.get("warnings"):
|
||
print("⚠️ Warnings:")
|
||
for w in log["warnings"]:
|
||
print(f" • {w}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|