272 lines
11 KiB
Python
272 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""umbrella_update.py — Dynamic reflexive system update.
|
|
|
|
When you say "update memory" or "update governance," this script:
|
|
1. Knows all components in that umbrella
|
|
2. Checks dependencies between umbrellas
|
|
3. Runs canaries to verify
|
|
4. Ports to TUI if needed
|
|
5. Commits to GitHub/Gitea
|
|
|
|
No manual specification needed. The system knows what needs updating.
|
|
|
|
Usage:
|
|
umbrella_update.py <umbrella> [--dry-run] [--skip-canary] [--skip-tui]
|
|
umbrella_update.py memory
|
|
umbrella_update.py governance
|
|
umbrella_update.py identity
|
|
umbrella_update.py all
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
|
|
SCRIPTS_DIR = os.path.expanduser("~/.config/opencode/scripts")
|
|
SHARED_DIR = os.path.expanduser("~/.config/opencode/shared")
|
|
TUI_DIR = "/mnt/workdrive/rune/src/rune"
|
|
MEMORY_DIR = os.path.expanduser("~/.config/opencode/memory")
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# UMBRELLA DEFINITIONS
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
UMBRELLAS = {
|
|
"memory": {
|
|
"description": "Storing, retrieving, and managing knowledge",
|
|
"shared": ["drift_core.py", "drift_cache.py", "drift_compression.py",
|
|
"tier_manager.py", "shard_manager.py", "query_memory.py"],
|
|
"scripts": ["memory_store.py", "memory_consolidation.py", "warm_router.py",
|
|
"graph_memory.py", "warm_neuron_store.py", "abstraction.py",
|
|
"cache.py", "recall_router.py", "recall_uplift.py",
|
|
"dedupe_policy.py", "memory_env.py"],
|
|
"canaries": ["canary_full_system.py", "canary_unbounded.py", "canary_query_memory.py"],
|
|
"depends_on": ["governance"],
|
|
},
|
|
"governance": {
|
|
"description": "Quality control, drift detection, promotion",
|
|
"shared": ["drift_core.py", "drift_cache.py", "drift_compression.py",
|
|
"drift_observability.py", "audit_ledger.py"],
|
|
"scripts": ["curator.py", "claim_audit.py", "grading.py",
|
|
"worthiness.py", "lexicon_evolution.py"],
|
|
"canaries": ["canary_full_system.py", "canary_drift_wiring.py"],
|
|
"depends_on": ["memory"],
|
|
},
|
|
"identity": {
|
|
"description": "Who the agent is",
|
|
"shared": [],
|
|
"scripts": ["persona_definition.py", "persona_check.py", "persona_gate.py",
|
|
"identity_anchors.py", "voice_hybrid.py", "voice_sync.py",
|
|
"delivery_adapter.py", "prompt_sync.py", "behavior.py"],
|
|
"canaries": ["canary_full_system.py"],
|
|
"depends_on": ["governance"],
|
|
},
|
|
"growth": {
|
|
"description": "Learning and improvement",
|
|
"shared": ["growth_simulator.py", "subconscious_pass.py"],
|
|
"scripts": ["growth_delta.py", "lessons.py", "skill_library.py",
|
|
"self_improve.py", "absorb_persona.py", "socratic.py"],
|
|
"canaries": ["canary_full_system.py"],
|
|
"depends_on": ["memory", "governance"],
|
|
},
|
|
"perception": {
|
|
"description": "Input processing",
|
|
"shared": [],
|
|
"scripts": ["warm_router.py", "capability_broker.py", "witness.py",
|
|
"context_pressure.py", "auto_context.py"],
|
|
"canaries": ["canary_full_system.py"],
|
|
"depends_on": ["memory"],
|
|
},
|
|
"infrastructure": {
|
|
"description": "The plumbing",
|
|
"shared": ["dependency_tracker.py"],
|
|
"scripts": ["microkernel.py", "hook_engine.py", "event_bus.py",
|
|
"config_audit.py", "system_registry.py", "bootstrap.py"],
|
|
"canaries": [],
|
|
"depends_on": [],
|
|
},
|
|
}
|
|
|
|
|
|
def get_affected_files(umbrella_name: str) -> dict:
|
|
"""Get all files affected by updating an umbrella."""
|
|
if umbrella_name == "all":
|
|
affected = {"shared": set(), "scripts": set(), "canaries": set()}
|
|
for name, umb in UMBRELLAS.items():
|
|
affected["shared"].update(umb.get("shared", []))
|
|
affected["scripts"].update(umb.get("scripts", []))
|
|
affected["canaries"].update(umb.get("canaries", []))
|
|
return {k: sorted(v) for k, v in affected.items()}
|
|
|
|
umb = UMBRELLAS.get(umbrella_name)
|
|
if not umb:
|
|
print(f"Unknown umbrella: {umbrella_name}")
|
|
print(f"Available: {', '.join(UMBRELLAS.keys())}")
|
|
sys.exit(1)
|
|
|
|
affected = {
|
|
"shared": list(umb.get("shared", [])),
|
|
"scripts": list(umb.get("scripts", [])),
|
|
"canaries": list(umb.get("canaries", [])),
|
|
}
|
|
|
|
# Include dependent umbrellas
|
|
for dep in umb.get("depends_on", []):
|
|
dep_umb = UMBRELLAS.get(dep, {})
|
|
affected["shared"].extend(dep_umb.get("shared", []))
|
|
affected["scripts"].extend(dep_umb.get("scripts", []))
|
|
|
|
# Deduplicate
|
|
for key in affected:
|
|
affected[key] = sorted(set(affected[key]))
|
|
|
|
return affected
|
|
|
|
|
|
def sync_shared_to_scripts():
|
|
"""Sync shared library to both scripts directories."""
|
|
synced = 0
|
|
for f in os.listdir(SHARED_DIR):
|
|
if not f.endswith(".py"):
|
|
continue
|
|
src = os.path.join(SHARED_DIR, f)
|
|
# Sync to MiMoCode scripts
|
|
dst_mimo = os.path.join(SCRIPTS_DIR, f)
|
|
if os.path.exists(dst_mimo):
|
|
# Only copy if shared is newer
|
|
if os.path.getmtime(src) > os.path.getmtime(dst_mimo):
|
|
import shutil
|
|
shutil.copy2(src, dst_mimo)
|
|
synced += 1
|
|
# Sync to TUI
|
|
dst_tui = os.path.join(TUI_DIR, f)
|
|
if os.path.exists(dst_tui):
|
|
if os.path.getmtime(src) > os.path.getmtime(dst_tui):
|
|
import shutil
|
|
shutil.copy2(src, dst_tui)
|
|
synced += 1
|
|
return synced
|
|
|
|
|
|
def run_canaries(canaries: list, skip: bool = False) -> dict:
|
|
"""Run canaries and return results."""
|
|
if skip:
|
|
return {"skipped": True}
|
|
|
|
results = {"passed": 0, "failed": 0, "details": []}
|
|
for canary in canaries:
|
|
canary_path = os.path.join(SCRIPTS_DIR, canary)
|
|
if not os.path.exists(canary_path):
|
|
results["details"].append({"canary": canary, "status": "not found"})
|
|
continue
|
|
|
|
try:
|
|
proc = subprocess.run(
|
|
[sys.executable, canary_path],
|
|
capture_output=True, text=True, timeout=120
|
|
)
|
|
if proc.returncode == 0:
|
|
results["passed"] += 1
|
|
results["details"].append({"canary": canary, "status": "passed"})
|
|
else:
|
|
results["failed"] += 1
|
|
results["details"].append({"canary": canary, "status": "failed", "error": proc.stderr[:200]})
|
|
except Exception as e:
|
|
results["failed"] += 1
|
|
results["details"].append({"canary": canary, "status": "error", "error": str(e)[:200]})
|
|
|
|
return results
|
|
|
|
|
|
def commit_changes(umbrella_name: str, affected: dict):
|
|
"""Commit changes to git. Handles worktrees and pushes to remote."""
|
|
results = {"committed": False, "pushed": False, "errors": []}
|
|
|
|
try:
|
|
# Add all changed files
|
|
for f in affected.get("shared", []):
|
|
subprocess.run(["git", "add", os.path.join(SHARED_DIR, f)], capture_output=True)
|
|
for f in affected.get("scripts", []):
|
|
subprocess.run(["git", "add", os.path.join(SCRIPTS_DIR, f)], capture_output=True)
|
|
|
|
# Commit
|
|
msg = f"umbrella({umbrella_name}): update {len(affected.get('shared', []))} shared + {len(affected.get('scripts', []))} scripts"
|
|
proc = subprocess.run(["git", "commit", "-m", msg], capture_output=True, text=True)
|
|
if proc.returncode == 0:
|
|
results["committed"] = True
|
|
else:
|
|
results["errors"].append(f"commit: {proc.stderr[:100]}")
|
|
|
|
# Push to remote (if configured)
|
|
push_proc = subprocess.run(["git", "push"], capture_output=True, text=True)
|
|
if push_proc.returncode == 0:
|
|
results["pushed"] = True
|
|
else:
|
|
results["errors"].append(f"push: {push_proc.stderr[:100]}")
|
|
|
|
except Exception as e:
|
|
results["errors"].append(str(e)[:200])
|
|
|
|
return results
|
|
|
|
|
|
def main():
|
|
import argparse
|
|
parser = argparse.ArgumentParser(description="Umbrella update system")
|
|
parser.add_argument("umbrella", help="Umbrella to update (memory, governance, identity, growth, perception, infrastructure, all)")
|
|
parser.add_argument("--dry-run", action="store_true", help="Show what would be updated without doing it")
|
|
parser.add_argument("--skip-canary", action="store_true", help="Skip canary verification")
|
|
parser.add_argument("--skip-tui", action="store_true", help="Skip TUI port")
|
|
parser.add_argument("--skip-commit", action="store_true", help="Skip git commit")
|
|
args = parser.parse_args()
|
|
|
|
print(f"=== UMBRELLA UPDATE: {args.umbrella.upper()} ===\n")
|
|
|
|
# Step 1: Get affected files
|
|
affected = get_affected_files(args.umbrella)
|
|
print(f"Affected files:")
|
|
print(f" Shared: {len(affected['shared'])} files")
|
|
print(f" Scripts: {len(affected['scripts'])} files")
|
|
print(f" Canaries: {len(affected['canaries'])} files")
|
|
|
|
if args.dry_run:
|
|
print(f"\nShared files: {', '.join(affected['shared'])}")
|
|
print(f"Script files: {', '.join(affected['scripts'])}")
|
|
print(f"Canaries: {', '.join(affected['canaries'])}")
|
|
return
|
|
|
|
# Step 2: Sync shared library
|
|
print(f"\nSyncing shared library...")
|
|
synced = sync_shared_to_scripts()
|
|
print(f" Synced {synced} files")
|
|
|
|
# Step 3: Run canaries
|
|
print(f"\nRunning canaries...")
|
|
canary_results = run_canaries(affected["canaries"], skip=args.skip_canary)
|
|
if canary_results.get("skipped"):
|
|
print(f" Skipped")
|
|
else:
|
|
print(f" Passed: {canary_results['passed']}")
|
|
print(f" Failed: {canary_results['failed']}")
|
|
for d in canary_results["details"]:
|
|
status = "✓" if d["status"] == "passed" else "✗"
|
|
print(f" {status} {d['canary']}")
|
|
|
|
# Step 4: Commit and push
|
|
if not args.skip_commit:
|
|
print(f"\nCommitting...")
|
|
commit_result = commit_changes(args.umbrella, affected)
|
|
print(f" Committed: {commit_result['committed']}")
|
|
print(f" Pushed: {commit_result['pushed']}")
|
|
if commit_result['errors']:
|
|
for err in commit_result['errors']:
|
|
print(f" ✗ {err}")
|
|
|
|
print(f"\n=== UPDATE COMPLETE ===")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|