rune/shared/dream_cycle.py

#!/usr/bin/env python3
"""dream_cycle.py — Background processing for Rune.

Rune doesn't stop thinking when you stop talking. The dream cycle
runs between interactions, processing memories, consolidating knowledge,
and generating insights.

This is what makes Rune alive — she thinks even when you're not there.

Usage:
    dream_cycle.py run                 # Run a dream cycle
    dream_cycle.py status              # Check dream status
    dream_cycle.py schedule            # Show scheduled dream times
"""

import json
import os
import sys
import time
from datetime import datetime, timezone

SCRIPTS_DIR = os.path.expanduser("~/.config/opencode/scripts")
SHARED_DIR = os.path.expanduser("~/.config/opencode/shared")
DREAM_STATE_FILE = os.path.expanduser("~/.config/opencode/memory/dream_state.json")

sys.path.insert(0, SCRIPTS_DIR)
sys.path.insert(0, SHARED_DIR)


class DreamCycle:
    """Background processing for Rune."""

    def __init__(self):
        self.state = self._load_state()

    def _load_state(self) -> dict:
        """Load dream state from disk."""
        if os.path.exists(DREAM_STATE_FILE):
            try:
                with open(DREAM_STATE_FILE) as f:
                    return json.load(f)
            except json.JSONDecodeError:
                return self._default_state()
        return self._default_state()

    def _default_state(self) -> dict:
        return {
            "last_dream": None,
            "dream_count": 0,
            "insights": [],
            "consolidations": 0,
            "probes": 0,
        }

    def _save_state(self):
        """Save dream state to disk."""
        os.makedirs(os.path.dirname(DREAM_STATE_FILE), exist_ok=True)
        with open(DREAM_STATE_FILE, "w") as f:
            json.dump(self.state, f, indent=2)

    def dream(self) -> dict:
        """Run a dream cycle.

        What happens when Rune dreams:
        1. Consolidate all memory sources
        2. Run curiosity probe on dead weight
        3. Check drift direction
        4. Generate insights from patterns
        5. Update dream state
        """
        results = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "actions": [],
        }

        # 1. Consolidate all memory sources
        try:
            import memory_consolidation
            mc = memory_consolidation.MemoryConsolidation()
            counts = mc.consolidate_all(full=True)
            results["actions"].append({
                "type": "consolidation",
                "neurons": counts.get("neurons", 0),
                "graph": counts.get("graph", 0),
                "abstraction": counts.get("abstraction", 0),
                "memory": counts.get("memory", 0),
            })
            self.state["consolidations"] += sum(counts.values())
        except Exception as e:
            results["actions"].append({"type": "consolidation_error", "error": str(e)[:100]})

        # 2. Run curiosity probe
        try:
            import memory_store
            db = memory_store.connect()
            probed = memory_store.curiosity_probe(db, limit=20)
            found = sum(1 for p in probed if p.get("found"))
            results["actions"].append({
                "type": "curiosity_probe",
                "probed": len(probed),
                "found": found,
            })
            self.state["probes"] += len(probed)
            db.close()
        except Exception as e:
            results["actions"].append({"type": "probe_error", "error": str(e)[:100]})

        # 3. Check drift direction
        try:
            import drift_core
            import memory_store
            db = memory_store.connect()
            direction = drift_core.compute_drift_direction(db)
            results["actions"].append({
                "type": "drift_check",
                "direction": direction["direction"],
                "stability": direction["stability_ratio"],
            })
            db.close()
        except Exception as e:
            results["actions"].append({"type": "drift_error", "error": str(e)[:100]})

        # 4. Run tier demotion sweep
        try:
            import memory_store
            db = memory_store.connect()
            demoted = memory_store.tier_demotion_sweep(db)
            results["actions"].append({
                "type": "tier_sweep",
                "hot_to_warm": demoted.get("hot_to_warm", 0),
                "warm_to_cold": demoted.get("warm_to_cold", 0),
            })
            db.close()
        except Exception as e:
            results["actions"].append({"type": "tier_error", "error": str(e)[:100]})

        # Update state
        self.state["last_dream"] = results["timestamp"]
        self.state["dream_count"] += 1
        self._save_state()

        return results

    def get_status(self) -> dict:
        """Get dream cycle status."""
        return {
            "last_dream": self.state.get("last_dream"),
            "dream_count": self.state.get("dream_count", 0),
            "consolidations": self.state.get("consolidations", 0),
            "probes": self.state.get("probes", 0),
            "insights": len(self.state.get("insights", [])),
        }


# ═══════════════════════════════════════════════════════════════════════════════
# CLI
# ═══════════════════════════════════════════════════════════════════════════════

def main():
    import argparse
    parser = argparse.ArgumentParser(description="Dream cycle for Rune")
    sub = parser.add_subparsers(dest="cmd")

    sub.add_parser("run", help="Run a dream cycle")
    sub.add_parser("status", help="Check dream status")

    args = parser.parse_args()

    if not args.cmd:
        parser.print_help()
        sys.exit(1)

    cycle = DreamCycle()

    if args.cmd == "run":
        results = cycle.dream()
        print("=== Dream Cycle ===")
        print(f"  Timestamp: {results['timestamp']}")
        print(f"  Actions: {len(results['actions'])}")
        for a in results["actions"]:
            print(f"    {a['type']}: {a}")

    elif args.cmd == "status":
        status = cycle.get_status()
        print("=== Dream Cycle Status ===")
        print(f"  Last dream: {status['last_dream'] or 'never'}")
        print(f"  Dream count: {status['dream_count']}")
        print(f"  Consolidations: {status['consolidations']}")
        print(f"  Probes: {status['probes']}")
        print(f"  Insights: {status['insights']}")


if __name__ == "__main__":
    main()