feat: dream cycle — drives-driven unattended consolidation + reflection

Lyra's inner loop for when no one's talking to her. Each pass senses her own
backlog/novelty, lets four drives build from real signals, and acts on those
past threshold:
- continuity -> summarize sessions with new turns
- coherence  -> rebuild profile/eras/narrative (stale once new gists land)
- curiosity  -> reflect() and evolve the self-state
- stability  -> readout of how caught-up she ended up

Drives are rendered into chat context so she can feel them. Causal chain:
consolidation creates gists -> coherence rises -> integration fires next.

- lyra/dream.py: dream_cycle() + lyra-dream CLI (--force, --loop SECONDS)
- memory: backlog_stats(), profile_sessions_covered(), WAL + busy_timeout
  so a separate dream process coexists with the web server
- self_state: DEFAULT_DRIVES baseline + drives in render_for_context
- tests/test_dream.py: backlog sensing + a full forced pass (LLM stubbed)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

lyra/memory.py

@@ -98,6 +98,10 @@ def _connection() -> sqlite3.Connection:
         # the one that created it. Safe here under single-user, low-concurrency use.
         _conn = sqlite3.connect(cfg.db_path, check_same_thread=False)
         _conn.row_factory = sqlite3.Row
+        # WAL + a busy timeout so a separate dream-cycle process can read/write
+        # alongside the web server without tripping "database is locked".
+        _conn.execute("PRAGMA busy_timeout=5000")
+        _conn.execute("PRAGMA journal_mode=WAL")
         _conn.executescript(SCHEMA)
         _conn_path = cfg.db_path
     return _conn
@@ -373,6 +377,54 @@ def get_profile(profile_id: str = "self") -> str | None:
     return r["content"] if r else None
 
 
+def profile_sessions_covered(profile_id: str = "self") -> int:
+    """How many session gists the current profile was built from (0 if none)."""
+    conn = _connection()
+    r = conn.execute(
+        "SELECT sessions_covered FROM profile WHERE id = ?", (profile_id,)
+    ).fetchone()
+    return int(r["sessions_covered"]) if r else 0
+
+
+def backlog_stats(ripe_threshold: int = 20) -> dict:
+    """Snapshot of the consolidation backlog, for the dream cycle to sense.
+
+    Returns, in one pass over the exchanges: how many sessions have any
+    unsummarized turns ("dirty"), how many are "ripe" (never summarized, or
+    >= `ripe_threshold` new turns since their last summary), the total
+    unsummarized exchanges, and the high-water exchange id (to detect new
+    activity since the previous cycle).
+    """
+    conn = _connection()
+    rows = conn.execute(
+        """
+        SELECT
+            SUM(CASE WHEN e.id > COALESCE(su.last_exchange_id, 0) THEN 1 ELSE 0 END)
+                AS unsummarized,
+            (su.session_id IS NULL) AS no_summary
+        FROM exchanges e
+        LEFT JOIN summaries su ON su.session_id = e.session_id
+        GROUP BY e.session_id
+        """
+    ).fetchall()
+    dirty = ripe = unsummarized_total = 0
+    for r in rows:
+        u = int(r["unsummarized"] or 0)
+        unsummarized_total += u
+        if u > 0:
+            dirty += 1
+            if r["no_summary"] or u >= ripe_threshold:
+                ripe += 1
+    mx = conn.execute("SELECT COALESCE(MAX(id), 0) AS m FROM exchanges").fetchone()["m"]
+    return {
+        "sessions": len(rows),
+        "dirty": dirty,
+        "ripe": ripe,
+        "unsummarized_total": unsummarized_total,
+        "max_exchange_id": int(mx),
+    }
+
+
 # --- Era tier (per-month temporal rollups) ---