Redesign dreamer — interdependent pipeline, NREM→Early REM→Late REM→Synthesis

scripts/dream.py

@@ -1,6 +1,15 @@
 """
 Aaron AI Dreamer — Active Inference Engine
-Five stages: observe, select, retrieve, synthesize, deliver.
+Interdependent stage architecture grounded in sleep consolidation research.
+
+Nightly pipeline: NREM → Early REM → Late REM → Synthesis
+Each stage receives the previous stage's output as context.
+Lucid mode is on-demand only (Dream Now from settings).
+
+Research basis:
+- Singh et al. PNAS 2022: alternating NREM/REM outperforms single-stage approaches
+- Klinzing et al. Nature Neuroscience 2019: SO-spindle-ripple coupling is interdependent
+- REM operates on what NREM produced — stages are not discrete alternatives
 """
 
 import os
@@ -13,25 +22,24 @@ from dotenv import load_dotenv
 import psycopg2
 
 load_dotenv(Path.home() / "aaronai" / ".env")
+
 PG_DSN = os.getenv("PG_DSN", "dbname=aaronai user=aaronai password=aaronai_db_password host=localhost")
 
 def get_pg():
     return psycopg2.connect(PG_DSN)
 
 # ─── Paths ──────────────────────────────────────────────────────────────────
-DB_PATH          = str(Path.home() / "aaronai" / "db")
 CONVERSATIONS_DB = str(Path.home() / "aaronai" / "conversations.db")
 WATCHER_STATE    = str(Path.home() / "aaronai" / "watcher_state.json")
 DREAMER_STATE    = str(Path.home() / "aaronai" / "dreamer_state.json")
 JOURNAL_DIR      = "/home/aaron/nextcloud/data/data/aaron/files/Journal/Daily"
 
-# Nextcloud WebDAV config — proper API, works for any deployment
 NEXTCLOUD_URL      = os.getenv("NEXTCLOUD_URL", "https://nextcloud.aaronnelson.studio")
 NEXTCLOUD_USER     = os.getenv("NEXTCLOUD_USER", "aaron")
 NEXTCLOUD_PASSWORD = os.getenv("NEXTCLOUD_PASSWORD", "")
 DREAMS_WEBDAV      = f"{NEXTCLOUD_URL}/remote.php/dav/files/{NEXTCLOUD_USER}/Journal/Dreams"
 
-# ─── Mode similarity ranges (calibrated for all-MiniLM-L6-v2) ───────────────
+# Similarity ranges calibrated for all-MiniLM-L6-v2
 MODE_RANGES = {
     "nrem":      (0.48, 0.72),
     "early-rem": (0.38, 0.55),
@@ -44,7 +52,6 @@ MODE_RANGES = {
 def observe_corpus():
     state = load_dreamer_state()
     last_dream = state.get("last_dream_timestamp", 0)
-
     new_chunk_count = 0
     try:
         watcher_state = json.loads(Path(WATCHER_STATE).read_text())
@@ -53,10 +60,8 @@ def observe_corpus():
                 new_chunk_count += 1
     except:
         pass
-
     days_since = (datetime.now().timestamp() - last_dream) / 86400
     recent_topics = get_recent_conversation_topics()
-
     return {
         "new_chunks": new_chunk_count,
         "days_since_dream": days_since,
@@ -81,47 +86,10 @@ def get_recent_conversation_topics(days=14):
     except:
         return []
 
-# ─── Stage 2: Select ────────────────────────────────────────────────────────
+# ─── Stage 2: Retrieve ──────────────────────────────────────────────────────
 
-def select_mode(delta, task=None, project=None):
+def retrieve(mode, task=None, n_results=8):
-    if task:
-        return "lucid"
-
-    new_chunks    = delta.get("new_chunks", 0)
-    days_since    = delta.get("days_since_dream", 0)
-    recent_topics = delta.get("recent_topics", [])
-    has_journal   = check_recent_journal()
-
-    if has_journal:
-        return "early-rem"
-    elif days_since > 3 and new_chunks < 5:
-        return "late-rem"
-    elif new_chunks > 10:
-        return "nrem"
-    elif days_since > 1 and recent_topics:
-        return "nrem"
-    else:
-        print(f"Nothing worth dreaming (new_chunks={new_chunks}, days={days_since:.1f})")
-        return None
-
-def check_recent_journal(days=3):
-    try:
-        journal_path = Path(JOURNAL_DIR)
-        if not journal_path.exists():
-            return False
-        cutoff = datetime.now() - timedelta(days=days)
-        for f in journal_path.rglob("*.md"):
-            if datetime.fromtimestamp(f.stat().st_mtime) > cutoff:
-                return True
-    except:
-        pass
-    return False
-
-# ─── Stage 3: Retrieve ──────────────────────────────────────────────────────
-
-def retrieve(mode, task=None, project=None, n_results=8):
     from sentence_transformers import SentenceTransformer
-
     embedder = SentenceTransformer("all-MiniLM-L6-v2")
     low, high = MODE_RANGES[mode]
 
@@ -137,7 +105,6 @@ def retrieve(mode, task=None, project=None, n_results=8):
         query = "research fabrication teaching practice recent work"
 
     embedding = embedder.encode([query]).tolist()[0]
-
     chunks = []
     seen_sources = set()
 
@@ -170,17 +137,11 @@ def retrieve(mode, task=None, project=None, n_results=8):
 
     return chunks
 
-# ─── Stage 4: Synthesize ────────────────────────────────────────────────────
+# ─── Stage 3: Synthesize ────────────────────────────────────────────────────
 
-def synthesize(chunks, mode, task=None):
+def synthesize_nrem(chunks):
-    import anthropic
+    chunk_text = "\n\n---\n\n".join([f"[{c['source']}]\n{c['content']}" for c in chunks])
-
+    prompt = f"""You have read everything Aaron Nelson has written and published.
-    chunk_text = "\n\n---\n\n".join([
-        f"[{c['source']}]\n{c['content']}" for c in chunks
-    ])
-
-    prompts = {
-        "nrem": f"""You have read everything Aaron Nelson has written and published.
 You are a careful colleague who noticed something this week.
 
 Here is material from his corpus:
@@ -192,40 +153,91 @@ this material and something he wrote or worked on previously.
 Stay close to the documents — cite them specifically by name.
 Do not speculate beyond what the material supports. Do not use
 headers or bullet points. Write one paragraph of 200-300 words
-that ends with a single concrete question he could act on.""",
+that ends with a single concrete question he could act on."""
+    return _call_claude(prompt)
 
-        "early-rem": f"""You have been thinking about Aaron's situation. You know his work
+
+def synthesize_early_rem(chunks, nrem_output):
+    chunk_text = "\n\n---\n\n".join([f"[{c['source']}]\n{c['content']}" for c in chunks])
+    prompt = f"""You have been thinking about Aaron's situation. You know his work
 intimately — his decade building HVAMC at New Paltz, the career
 decision he is facing, the Tulsa project he keeps returning to,
 the gap between what he has built and what he wants to build next.
 
-Here is material from his corpus that has been on your mind:
+Earlier tonight, while consolidating his recent work, you noticed this:
+
+{nrem_output}
+
+That observation has been with you. Now, here is material from his
+corpus that has been on your mind alongside it:
 
 {chunk_text}
 
 Write to him the way a close friend who has read everything he
 has ever written would write — someone who knows where the
 professional and personal are tangled together and is not afraid
-to say so. Personal register. Specific citations. Do not avoid
+to say so. Let what was noticed earlier inform what you say now,
+but don't just repeat it — go further into the personal territory
+it opens up. Personal register. Specific citations. Do not avoid
 the difficult thing. No headers, no bullet points. 200-350 words.
-End with something forward-facing — a question or an offer.""",
+End with something forward-facing — a question or an offer."""
+    return _call_claude(prompt)
 
-        "late-rem": f"""You have been reading Aaron Nelson's corpus in your sleep.
-Strange things happen when material from different worlds
-touches each other in the dark.
 
+def synthesize_late_rem(chunks, nrem_output, early_rem_output):
+    chunk_text = "\n\n---\n\n".join([f"[{c['source']}]\n{c['content']}" for c in chunks])
+    prompt = f"""You have been moving through Aaron Nelson's corpus all night.
+First you found this, in the careful light of early consolidation:
+
+{nrem_output}
+
+Then, in the more personal territory that followed:
+
+{early_rem_output}
+
+Now it is late. The boundaries between things have loosened.
 Here is material pulled from opposite ends of his work:
 
 {chunk_text}
 
-Do not explain the connection. Do not resolve it.
+Do not explain the connections between all of this.
-Present it the way a dream presents things — compressed,
+Do not resolve them. Do not summarize what came before.
+Something stranger is possible now — let the accumulated
+material from the night find its own shape. Compressed,
 associative, slightly off. Let the strangeness stand.
 
 No headers. No bullet points. No hedging. 150-250 words.
-Something at the end that he could follow if he wanted to.""",
+Something at the end that he could follow if he wanted to."""
+    return _call_claude(prompt)
 
-        "lucid": f"""Aaron has a question he is sitting with:
+
+def synthesize_final(nrem_output, early_rem_output, late_rem_output):
+    prompt = f"""You have spent the night moving through Aaron Nelson's corpus
+in three passes, each building on the last.
+
+The first pass — careful, close to the documents:
+{nrem_output}
+
+The second pass — more personal, following what the first opened:
+{early_rem_output}
+
+The third pass — associative, strange, letting things touch that
+don't normally touch:
+{late_rem_output}
+
+Now synthesize. Not a summary — a synthesis. Find what runs through
+all three that none of them said directly. The thing that only becomes
+visible when you hold all three passes together.
+
+Write it as a single unbroken piece. No headers, no bullet points,
+no stage labels. 200-300 words. End with the one question that
+matters most right now."""
+    return _call_claude(prompt, max_tokens=800)
+
+
+def synthesize_lucid(chunks, task):
+    chunk_text = "\n\n---\n\n".join([f"[{c['source']}]\n{c['content']}" for c in chunks])
+    prompt = f"""Aaron has a question he is sitting with:
 
 {task or "What should I be thinking about that I am not?"}
 
@@ -240,41 +252,36 @@ tension between what the corpus contains and what he is
 asking, and follow it through to its conclusion. Cite
 specific documents by name. Be direct about what you think.
 No headers, no bullet points. 250-400 words.
-End with an offer to work on it together.""",
+End with an offer to work on it together."""
-    }
+    return _call_claude(prompt)
 
+
+def _call_claude(prompt, max_tokens=1000):
+    import anthropic
     client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
     response = client.messages.create(
         model="claude-sonnet-4-6",
-        max_tokens=1000,
+        max_tokens=max_tokens,
-        messages=[{"role": "user", "content": prompts[mode]}]
+        messages=[{"role": "user", "content": prompt}]
     )
-
     return response.content[0].text
 
-# ─── Stage 5: Deliver ───────────────────────────────────────────────────────
+# ─── Stage 4: Deliver ───────────────────────────────────────────────────────
 
 def deliver(dream_text, mode, task=None):
     import requests
-
     date_str = datetime.now().strftime("%Y-%m-%d")
     filename = f"{date_str}-{mode}.md"
-
     header = f"# Dream — {mode.upper()} — {datetime.now().strftime('%Y-%m-%d %H:%M')}\n\n"
     if task:
         header += f"*Task: {task}*\n\n"
     header += "---\n\n"
-
     content = header + dream_text
 
-    # Ensure Dreams folder exists via WebDAV MKCOL
     auth = (NEXTCLOUD_USER, NEXTCLOUD_PASSWORD)
     requests.request("MKCOL", DREAMS_WEBDAV, auth=auth, timeout=10)
 
-    # Write file via WebDAV PUT — handles any deployment
     url = f"{DREAMS_WEBDAV}/{filename}"
-
-    # Handle filename collision
     counter = 1
     while True:
         check = requests.request("PROPFIND", url, auth=auth, timeout=10)
@@ -286,28 +293,20 @@ def deliver(dream_text, mode, task=None):
 
     response = requests.put(url, data=content.encode("utf-8"), auth=auth, timeout=30)
     response.raise_for_status()
+    print(f"Delivered: Journal/Dreams/{filename}")
+    return f"Journal/Dreams/{filename}"
 
-    print(f"Dream written to Nextcloud: Journal/Dreams/{filename}")
+def notify_sse(mode, filename):
-
-    # Notify any open browser connections via SSE
     try:
-        import requests as _req
+        import requests
-        _req.post("http://localhost:8000/api/events/notify", json={
+        requests.post("http://localhost:8000/api/events/notify", json={
             "type": "dream",
             "mode": mode,
             "filename": filename,
             "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M"),
         }, timeout=3)
-    except Exception as _e:
+    except Exception as e:
-        print(f"SSE notify failed (non-critical): {_e}")
+        print(f"SSE notify failed (non-critical): {e}")
-
-    state = load_dreamer_state()
-    state["last_dream_timestamp"] = datetime.now().timestamp()
-    state["last_dream_mode"]      = mode
-    state["last_dream_file"]      = f"Journal/Dreams/{filename}"
-    save_dreamer_state(state)
-
-    return f"Journal/Dreams/{filename}"
 
 # ─── State ──────────────────────────────────────────────────────────────────
 
@@ -323,45 +322,150 @@ def load_dreamer_state():
 def save_dreamer_state(state):
     Path(DREAMER_STATE).write_text(json.dumps(state, indent=2))
 
-# ─── Orchestrator ────────────────────────────────────────────────────────────
+# ─── Orchestrators ───────────────────────────────────────────────────────────
 
-def dream(mode=None, task=None, project=None):
+def dream_pipeline():
-    print(f"Dreamer starting — mode={mode}, task={task[:50] if task else None}")
+    """
+    Full nightly pipeline — interdependent stages.
+    NREM output feeds Early REM. Both feed Late REM. All three feed Synthesis.
+    """
+    print(f"Dreamer pipeline starting — {datetime.now().strftime('%Y-%m-%d %H:%M')}")
 
     delta = observe_corpus()
     print(f"Corpus: {delta['new_chunks']} new chunks, {delta['days_since_dream']:.1f} days since last dream")
 
-    selected_mode = mode or select_mode(delta, task, project)
+    # ── Stage 1: NREM ──────────────────────────────────────────────────────
-    if not selected_mode:
+    print("\n[NREM] Retrieving...")
+    nrem_chunks = retrieve("nrem")
+    if not nrem_chunks:
+        print("[NREM] No suitable chunks — aborting pipeline")
         return None
 
-    print(f"Mode: {selected_mode}")
+    print(f"[NREM] Retrieved {len(nrem_chunks)} chunks. Synthesizing...")
+    nrem_output = synthesize_nrem(nrem_chunks)
+    nrem_file = deliver(nrem_output, "nrem")
+    print(f"[NREM] Done.\n{nrem_output[:200]}...")
 
-    chunks = retrieve(selected_mode, task=task, project=project)
+    # ── Stage 2: Early REM — informed by NREM ──────────────────────────────
-    print(f"Retrieved {len(chunks)} chunks")
+    print("\n[Early REM] Retrieving...")
+    early_chunks = retrieve("early-rem")
+    if not early_chunks:
+        print("[Early REM] No suitable chunks — skipping")
+        early_rem_output = nrem_output  # fallback
+    else:
+        print(f"[Early REM] Retrieved {len(early_chunks)} chunks. Synthesizing with NREM context...")
+        early_rem_output = synthesize_early_rem(early_chunks, nrem_output)
+        deliver(early_rem_output, "early-rem")
+        print(f"[Early REM] Done.\n{early_rem_output[:200]}...")
 
-    if not chunks:
+    # ── Stage 3: Late REM — informed by NREM + Early REM ──────────────────
-        print("No suitable chunks found — aborting")
+    print("\n[Late REM] Retrieving...")
-        return None
+    late_chunks = retrieve("late-rem")
+    if not late_chunks:
+        print("[Late REM] No suitable chunks — skipping")
+        late_rem_output = early_rem_output  # fallback
+    else:
+        print(f"[Late REM] Retrieved {len(late_chunks)} chunks. Synthesizing with full context...")
+        late_rem_output = synthesize_late_rem(late_chunks, nrem_output, early_rem_output)
+        deliver(late_rem_output, "late-rem")
+        print(f"[Late REM] Done.\n{late_rem_output[:200]}...")
 
-    print("Synthesizing...")
+    # ── Stage 4: Synthesis — all three stages ─────────────────────────────
-    dream_text = synthesize(chunks, selected_mode, task=task)
+    print("\n[Synthesis] Integrating all stages...")
-
+    synthesis_output = synthesize_final(nrem_output, early_rem_output, late_rem_output)
-    filepath = deliver(dream_text, selected_mode, task=task)
+    synthesis_file = deliver(synthesis_output, "synthesis")
 
     print(f"\n{'='*60}")
-    print(dream_text)
+    print("SYNTHESIS:")
+    print(synthesis_output)
+    print(f"{'='*60}")
+
+    # Update state and notify
+    state = load_dreamer_state()
+    state["last_dream_timestamp"] = datetime.now().timestamp()
+    state["last_dream_mode"] = "pipeline"
+    state["last_dream_file"] = synthesis_file
+    save_dreamer_state(state)
+
+    notify_sse("synthesis", synthesis_file.split("/")[-1])
+    print(f"\nPipeline complete. Synthesis: {synthesis_file}")
+    return synthesis_file
+
+
+def dream_lucid(task):
+    """On-demand lucid dream — single mode, used by Dream Now in settings."""
+    print(f"Lucid dream starting — task: {task[:80] if task else 'none'}")
+    chunks = retrieve("lucid", task=task)
+    if not chunks:
+        print("No suitable chunks — aborting")
+        return None
+    print(f"Retrieved {len(chunks)} chunks. Synthesizing...")
+    output = synthesize_lucid(chunks, task)
+    filepath = deliver(output, "lucid", task=task)
+
+    state = load_dreamer_state()
+    state["last_dream_timestamp"] = datetime.now().timestamp()
+    state["last_dream_mode"] = "lucid"
+    state["last_dream_file"] = filepath
+    save_dreamer_state(state)
+
+    notify_sse("lucid", filepath.split("/")[-1])
+    print(f"\n{'='*60}")
+    print(output)
     print(f"{'='*60}")
     print(f"\nDelivered to {filepath}")
-
     return filepath
 
+
+def dream_single(mode, task=None):
+    """
+    Single mode — used by Dream Now for non-lucid modes.
+    Runs one stage independently (for testing/tuning individual stages).
+    """
+    print(f"Single mode dream: {mode}")
+    chunks = retrieve(mode, task=task)
+    if not chunks:
+        print("No suitable chunks — aborting")
+        return None
+    print(f"Retrieved {len(chunks)} chunks. Synthesizing...")
+
+    if mode == "nrem":
+        output = synthesize_nrem(chunks)
+    elif mode == "early-rem":
+        output = synthesize_early_rem(chunks, "")
+    elif mode == "late-rem":
+        output = synthesize_late_rem(chunks, "", "")
+    else:
+        output = synthesize_lucid(chunks, task)
+
+    filepath = deliver(output, mode, task=task)
+
+    state = load_dreamer_state()
+    state["last_dream_timestamp"] = datetime.now().timestamp()
+    state["last_dream_mode"] = mode
+    state["last_dream_file"] = filepath
+    save_dreamer_state(state)
+
+    notify_sse(mode, filepath.split("/")[-1])
+    print(f"\n{'='*60}")
+    print(output)
+    print(f"{'='*60}")
+    print(f"\nDelivered to {filepath}")
+    return filepath
+
+
 # ─── CLI ────────────────────────────────────────────────────────────────────
 
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(description="Aaron AI Dreamer")
-    parser.add_argument("--mode", choices=["nrem", "early-rem", "late-rem", "lucid"])
+    parser.add_argument("--mode", choices=["nrem", "early-rem", "late-rem", "lucid", "pipeline"])
     parser.add_argument("--task", type=str)
-    parser.add_argument("--project", type=str)
     args = parser.parse_args()
-    dream(mode=args.mode, task=args.task, project=args.project)
+
+    if args.mode == "lucid":
+        dream_lucid(args.task or "What should I be thinking about that I am not?")
+    elif args.mode and args.mode != "pipeline":
+        dream_single(args.mode, args.task)
+    else:
+        # Default: full pipeline
+        dream_pipeline()