feat: era-rollup + narrative engine (consolidation steps 3-4)

Complete the consolidation pipeline: summaries -> profile + eras -> narrative.

- memory: eras table (per-month digests) + Era, summaries_by_month, store_era,
  list_eras, recall_eras; narrative table + set/get_narrative
- lyra/era.py (lyra-era): groups session gists by the month the session occurred
  (real timestamps) and map-reduces each month into a "what was happening" digest
- lyra/narrative.py (lyra-narrative): distills profile + recent eras into the
  current arc/trends/callbacks ("remember when…", "you're trending toward…")
- chat.build_messages injects the narrative alongside the profile

Verified on the real corpus: 17 monthly eras (Dec 2024-Jun 2026) + a narrative
that surfaces specific callbacks (the $573 Hollywood session, 4 years sober).

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

.env.example

@@ -2,6 +2,10 @@
 LOCAL_BASE_URL=http://localhost:11434
 LOCAL_MODEL=qwen2.5:7b-instruct
 
+# MI50 backend — OpenAI-compatible llama.cpp server on the home-lab GPU box (CT202).
+MI50_BASE_URL=http://10.0.0.42:8080/v1
+MI50_MODEL=local-gpu
+
 # Cloud backend (OpenAI) — higher quality, costs money.
 OPENAI_API_KEY=
 CLOUD_MODEL=gpt-4o-mini

lyra/chat.py

@@ -39,6 +39,21 @@ def build_messages(session_id: str, user_msg: str) -> list[Message]:
     """Assemble the full, tiered message list for one turn."""
     messages: list[Message] = [{"role": "system", "content": persona.system_prompt()}]
 
+    # Semantic memory: the distilled profile (who Brian is) — answers identity
+    # questions that raw recall can't. Always in context when it exists.
+    profile = memory.get_profile()
+    if profile:
+        messages.append(
+            {"role": "system", "content": "What you know about Brian:\n" + profile}
+        )
+
+    # Time-aware memory: the current narrative (recent arc, trends, callbacks).
+    narrative = memory.get_narrative()
+    if narrative:
+        messages.append(
+            {"role": "system", "content": "What's going on with Brian lately:\n" + narrative}
+        )
+
     recent = memory.recent(session_id, n=RECENT_N)
     recent_ids = {ex.id for ex in recent}
 
@@ -73,7 +88,9 @@ def build_messages(session_id: str, user_msg: str) -> list[Message]:
 def respond(session_id: str, user_msg: str, backend: Backend = "cloud") -> str:
     """Produce Lyra's reply to a single user message and persist the exchange."""
     cfg = config.load()
-    model = cfg.local_model if backend == "local" else cfg.cloud_model
+    model = {"local": cfg.local_model, "cloud": cfg.cloud_model, "mi50": cfg.mi50_model}.get(
+        backend, backend
+    )
     logbus.log(
         "info", "chat request", session=session_id, backend=backend,
         model=model, embed=cfg.embed_backend,

lyra/config.py

@@ -14,6 +14,8 @@ load_dotenv()
 class Config:
     local_base_url: str
     local_model: str
+    mi50_base_url: str  # OpenAI-compatible llama.cpp server on the MI50 box
+    mi50_model: str
     openai_api_key: str
     cloud_model: str
     embed_backend: str  # "cloud" (OpenAI) or "local" (Ollama)
@@ -27,6 +29,8 @@ def load() -> Config:
     return Config(
         local_base_url=os.getenv("LOCAL_BASE_URL", "http://localhost:11434"),
         local_model=os.getenv("LOCAL_MODEL", "qwen2.5:7b-instruct"),
+        mi50_base_url=os.getenv("MI50_BASE_URL", "http://10.0.0.42:8080/v1"),
+        mi50_model=os.getenv("MI50_MODEL", "local-gpu"),
         openai_api_key=os.getenv("OPENAI_API_KEY", ""),
         cloud_model=os.getenv("CLOUD_MODEL", "gpt-4o-mini"),
         embed_backend=os.getenv("EMBED_BACKEND", "cloud").lower(),

lyra/era.py

@@ -0,0 +1,83 @@
+"""Era rollups: per-month "what was happening" digests (consolidation step 3).
+
+Groups session gists by the calendar month the session occurred (from real
+exchange timestamps) and map-reduces each month into one digest. These are the
+temporal memory tier — they answer "what was going on last December" and feed
+the narrative engine. Runs on the consolidation backend (MI50 in steady state).
+"""
+from __future__ import annotations
+
+
+from lyra import config, llm, logbus, memory
+from lyra.llm import Backend, Message
+
+BATCH_CHARS = 18000
+
+_PROMPT = """You are writing a monthly memory digest about Brian from the session \
+summaries below (all from the same month). Capture: what he was focused on (poker \
+and otherwise), notable events/results/decisions, recurring themes, and his mood \
+and arc across the month. Third person, referring to him as "Brian". 5-10 \
+sentences. This is a memory record, not a reply. No preamble."""
+
+_MERGE_PROMPT = """Merge these partial monthly digests (same month) into one \
+coherent digest about Brian for that month. Keep it tight, 5-10 sentences, no \
+repetition. Third person."""
+
+
+def _batch_texts(texts: list[str], budget: int) -> list[str]:
+    blocks, buf, size = [], [], 0
+    for t in texts:
+        if size + len(t) > budget and buf:
+            blocks.append("\n\n".join(buf))
+            buf, size = [], 0
+        buf.append(t)
+        size += len(t)
+    if buf:
+        blocks.append("\n\n".join(buf))
+    return blocks
+
+
+def _call(prompt: str, body: str, backend: Backend) -> str:
+    messages: list[Message] = [
+        {"role": "system", "content": prompt},
+        {"role": "user", "content": body},
+    ]
+    return llm.complete(messages, backend=backend)
+
+
+def _digest_month(gists: list[str], backend: Backend) -> str:
+    """Map-reduce a month's session gists into one digest."""
+    blocks = _batch_texts(gists, BATCH_CHARS)
+    partials = [_call(_PROMPT, b, backend) for b in blocks]
+    while len(partials) > 1:
+        partials = [_call(_MERGE_PROMPT, g, backend) for g in _batch_texts(partials, BATCH_CHARS)]
+    return partials[0]
+
+
+def rebuild_eras(backend: Backend | None = None) -> dict:
+    """(Re)build a digest for every month that has session gists."""
+    backend = backend or config.load().summary_backend
+    by_month = memory.summaries_by_month()
+    months = 0
+    for month in sorted(by_month):
+        digest = _digest_month(by_month[month], backend)
+        memory.store_era(month, digest, len(by_month[month]))
+        months += 1
+        logbus.log("info", "era built", month=month, sessions=len(by_month[month]))
+    report = {"months": months}
+    logbus.log("info", "eras complete", **report)
+    return report
+
+
+def main() -> int:
+    report = rebuild_eras()
+    if not report["months"]:
+        print("No summaries yet — run lyra-summarize first.")
+        return 1
+    for era in memory.list_eras():
+        print(f"\n## {era.month}  ({era.session_count} sessions)\n{era.content}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

lyra/llm.py

@@ -14,7 +14,7 @@ class Message(TypedDict):
     content: str
 
 
-Backend = Literal["local", "cloud"]
+Backend = Literal["local", "cloud", "mi50"]
 
 
 def complete(messages: list[Message], backend: Backend = "local") -> str:
@@ -26,6 +26,12 @@ def complete(messages: list[Message], backend: Backend = "local") -> str:
         resp = client.chat.completions.create(model=cfg.cloud_model, messages=messages)
         return resp.choices[0].message.content or ""
 
+    if backend == "mi50":
+        # MI50 box runs an OpenAI-compatible llama.cpp server; key is unused.
+        client = OpenAI(api_key="not-needed", base_url=cfg.mi50_base_url)
+        resp = client.chat.completions.create(model=cfg.mi50_model, messages=messages)
+        return resp.choices[0].message.content or ""
+
     resp = httpx.post(
         f"{cfg.local_base_url}/api/chat",
         json={"model": cfg.local_model, "messages": messages, "stream": False},

lyra/memory.py

@@ -43,6 +43,33 @@ CREATE TABLE IF NOT EXISTS summaries (
     last_exchange_id INTEGER NOT NULL,
     created_at TEXT NOT NULL
 );
+
+-- Derived semantic memory: standing facts about the user, distilled from the
+-- session gists by the consolidation pass. Single row (id='self').
+CREATE TABLE IF NOT EXISTS profile (
+    id TEXT PRIMARY KEY,
+    content TEXT NOT NULL,
+    sessions_covered INTEGER NOT NULL,
+    updated_at TEXT NOT NULL
+);
+
+-- Temporal memory: one "what was happening" digest per calendar month, rolled
+-- up from that month's session gists. month is "YYYY-MM".
+CREATE TABLE IF NOT EXISTS eras (
+    month TEXT PRIMARY KEY,
+    content TEXT NOT NULL,
+    embedding BLOB NOT NULL,
+    session_count INTEGER NOT NULL,
+    created_at TEXT NOT NULL
+);
+
+-- The current narrative: time-aware arc/trends/callbacks (vs the timeless
+-- profile). Distilled from profile + recent eras. Single row (id='current').
+CREATE TABLE IF NOT EXISTS narrative (
+    id TEXT PRIMARY KEY,
+    content TEXT NOT NULL,
+    updated_at TEXT NOT NULL
+);
 """
 
 _conn: sqlite3.Connection | None = None
@@ -86,6 +113,15 @@ class Summary:
     score: float | None = None
 
 
+@dataclass
+class Era:
+    month: str  # "YYYY-MM"
+    content: str
+    session_count: int
+    created_at: str
+    score: float | None = None
+
+
 def _to_blob(vec: list[float]) -> bytes:
     return np.asarray(vec, dtype=np.float32).tobytes()
 
@@ -290,6 +326,136 @@ def unsummarized_count(session_id: str) -> int:
     return int(r["n"])
 
 
+def list_summaries() -> list[Summary]:
+    """Every session gist (for the profile/era consolidation passes)."""
+    conn = _connection()
+    rows = conn.execute(
+        "SELECT session_id, content, last_exchange_id, created_at FROM summaries "
+        "ORDER BY created_at ASC"
+    ).fetchall()
+    return [
+        Summary(
+            session_id=r["session_id"],
+            content=r["content"],
+            last_exchange_id=r["last_exchange_id"],
+            created_at=r["created_at"],
+        )
+        for r in rows
+    ]
+
+
+def set_profile(content: str, sessions_covered: int, profile_id: str = "self") -> None:
+    """Store/replace the derived semantic profile."""
+    now = datetime.now(timezone.utc).isoformat()
+    conn = _connection()
+    with conn:
+        conn.execute(
+            "INSERT INTO profile (id, content, sessions_covered, updated_at) "
+            "VALUES (?, ?, ?, ?) "
+            "ON CONFLICT(id) DO UPDATE SET content=excluded.content, "
+            "sessions_covered=excluded.sessions_covered, updated_at=excluded.updated_at",
+            (profile_id, content, sessions_covered, now),
+        )
+
+
+def get_profile(profile_id: str = "self") -> str | None:
+    conn = _connection()
+    r = conn.execute("SELECT content FROM profile WHERE id = ?", (profile_id,)).fetchone()
+    return r["content"] if r else None
+
+
+# --- Era tier (per-month temporal rollups) ---
+
+
+def summaries_by_month() -> dict[str, list[str]]:
+    """Map "YYYY-MM" -> list of session gists for sessions that occurred that month.
+
+    A session's month comes from its earliest exchange timestamp (real ChatGPT
+    dates for imported sessions), not when it was summarized.
+    """
+    conn = _connection()
+    rows = conn.execute(
+        """
+        SELECT substr(MIN(e.created_at), 1, 7) AS month, s.content AS content
+        FROM summaries s JOIN exchanges e ON e.session_id = s.session_id
+        GROUP BY s.session_id
+        """
+    ).fetchall()
+    out: dict[str, list[str]] = {}
+    for r in rows:
+        out.setdefault(r["month"], []).append(r["content"])
+    return out
+
+
+def store_era(month: str, content: str, session_count: int) -> None:
+    """Embed and persist a month's digest, replacing any prior one."""
+    [embedding] = llm.embed([content])
+    now = datetime.now(timezone.utc).isoformat()
+    conn = _connection()
+    with conn:
+        conn.execute(
+            "INSERT INTO eras (month, content, embedding, session_count, created_at) "
+            "VALUES (?, ?, ?, ?, ?) "
+            "ON CONFLICT(month) DO UPDATE SET content=excluded.content, "
+            "embedding=excluded.embedding, session_count=excluded.session_count, "
+            "created_at=excluded.created_at",
+            (month, content, _to_blob(embedding), session_count, now),
+        )
+
+
+def list_eras() -> list[Era]:
+    """All month digests, chronological."""
+    conn = _connection()
+    rows = conn.execute(
+        "SELECT month, content, session_count, created_at FROM eras ORDER BY month ASC"
+    ).fetchall()
+    return [
+        Era(month=r["month"], content=r["content"],
+            session_count=r["session_count"], created_at=r["created_at"])
+        for r in rows
+    ]
+
+
+def set_narrative(content: str, narrative_id: str = "current") -> None:
+    """Store/replace the current narrative."""
+    now = datetime.now(timezone.utc).isoformat()
+    conn = _connection()
+    with conn:
+        conn.execute(
+            "INSERT INTO narrative (id, content, updated_at) VALUES (?, ?, ?) "
+            "ON CONFLICT(id) DO UPDATE SET content=excluded.content, updated_at=excluded.updated_at",
+            (narrative_id, content, now),
+        )
+
+
+def get_narrative(narrative_id: str = "current") -> str | None:
+    conn = _connection()
+    r = conn.execute("SELECT content FROM narrative WHERE id = ?", (narrative_id,)).fetchone()
+    return r["content"] if r else None
+
+
+def recall_eras(query: str, k: int = 2) -> list[Era]:
+    """Top-k month digests most similar to `query` (time-based context)."""
+    [q_vec] = llm.embed([query])
+    q = np.asarray(q_vec, dtype=np.float32)
+    conn = _connection()
+    rows = conn.execute(
+        "SELECT month, content, embedding, session_count, created_at FROM eras"
+    ).fetchall()
+    if not rows:
+        return []
+    matrix = np.stack([_from_blob(r["embedding"]) for r in rows])
+    norms = np.linalg.norm(matrix, axis=1)
+    scores = (matrix @ q) / (norms * np.linalg.norm(q) + 1e-9)
+    top_idx = np.argsort(scores)[::-1][:k]
+    return [
+        Era(month=rows[i]["month"], content=rows[i]["content"],
+            session_count=rows[i]["session_count"], created_at=rows[i]["created_at"],
+            score=float(scores[i]))
+        for i in top_idx
+    ]
+
+
 def recall_summaries(query: str, k: int = 3, exclude_session: str | None = None) -> list[Summary]:
     """Top-k session summaries most similar to `query` (the long-term gist tier)."""
     [q_vec] = llm.embed([query])

lyra/narrative.py

@@ -0,0 +1,66 @@
+"""Narrative engine (consolidation step 4): the current arc, trends, callbacks.
+
+Where the profile is timeless ("who Brian is"), the narrative is time-aware
+("what's going on lately, where things are trending"). It distills the profile
+plus the most recent monthly era digests into the current story — recent focus,
+notable trends or changes, mood/arc, and a few specific callbacks worth
+referencing. Injected into chat so Lyra follows along like a friend who's been
+paying attention. Runs on the consolidation backend (MI50 in steady state).
+"""
+from __future__ import annotations
+
+
+from lyra import config, llm, logbus, memory
+from lyra.llm import Backend, Message
+
+RECENT_ERAS = 4
+
+_PROMPT = """You are distilling the CURRENT narrative about Brian — what a close \
+friend who has been following along would keep in mind right now. From his profile \
+and recent monthly digests below, write: what he's been focused on lately, any \
+notable trends or changes (improving, slipping, new patterns), his current arc and \
+mood, and 2-4 specific things worth referencing back to him ("remember when…"). \
+Third person, referring to him as "Brian". 6-10 sentences. This is a memory note, \
+not a reply. No preamble."""
+
+
+def rebuild_narrative(backend: Backend | None = None) -> str | None:
+    """(Re)derive the current narrative from the profile + recent era digests."""
+    backend = backend or config.load().summary_backend
+    profile = memory.get_profile()
+    eras = memory.list_eras()
+    if not profile and not eras:
+        return None
+
+    parts = []
+    if profile:
+        parts.append("PROFILE (timeless):\n" + profile)
+    recent = eras[-RECENT_ERAS:]
+    if recent:
+        parts.append(
+            "RECENT MONTHS (oldest first):\n"
+            + "\n\n".join(f"[{e.month}]\n{e.content}" for e in recent)
+        )
+    body = "\n\n".join(parts)
+
+    messages: list[Message] = [
+        {"role": "system", "content": _PROMPT},
+        {"role": "user", "content": body},
+    ]
+    narrative = llm.complete(messages, backend=backend)
+    memory.set_narrative(narrative)
+    logbus.log("info", "narrative rebuilt", chars=len(narrative), eras=len(recent))
+    return narrative
+
+
+def main() -> int:
+    narrative = rebuild_narrative()
+    if narrative is None:
+        print("Need a profile and/or eras first — run lyra-profile and lyra-era.")
+        return 1
+    print(narrative)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

lyra/profile.py

@@ -0,0 +1,84 @@
+"""Profile derivation: distill standing facts about the user (semantic memory).
+
+This is consolidation step 2. It reads every session gist and map-reduces them
+into one profile document — who Brian is as a player and person — which is then
+injected into every prompt. This is what answers identity/abstract questions
+("what kind of player am I", "what are my leaks") that raw recall handles badly,
+because those are patterns across many sessions, not facts in any single message.
+"""
+from __future__ import annotations
+
+
+from lyra import config, llm, logbus, memory
+from lyra.llm import Backend, Message
+
+BATCH_CHARS = 18000
+
+_MAP_PROMPT = """From these session summaries, extract durable facts about Brian \
+— things that are stably true, not one-off events. Cover, where present: poker \
+games/formats/stakes he plays, his playing style and strengths, recurring leaks \
+and tendencies, mental-game patterns (tilt triggers, scared money, fatigue), \
+relevant personal context, and how he likes to be coached. Terse bullet points. \
+Omit anything not supported by the summaries."""
+
+_REDUCE_PROMPT = """Merge these fact lists into one deduplicated profile of Brian. \
+Organize under these headings: Poker Style, Leaks & Tendencies, Mental Game, \
+Personal Context, Working With Brian. Keep it tight — bullets, no fluff, no \
+repetition. Resolve contradictions toward the more recent/frequent signal."""
+
+
+def _batch_texts(texts: list[str], budget: int) -> list[str]:
+    """Group texts into joined blocks under `budget` chars."""
+    blocks, buf, size = [], [], 0
+    for t in texts:
+        if size + len(t) > budget and buf:
+            blocks.append("\n\n".join(buf))
+            buf, size = [], 0
+        buf.append(t)
+        size += len(t)
+    if buf:
+        blocks.append("\n\n".join(buf))
+    return blocks
+
+
+def _call(prompt: str, body: str, backend: Backend) -> str:
+    messages: list[Message] = [
+        {"role": "system", "content": prompt},
+        {"role": "user", "content": body},
+    ]
+    return llm.complete(messages, backend=backend)
+
+
+def rebuild_profile(backend: Backend | None = None) -> str | None:
+    """Re-derive the profile from all current session gists and store it."""
+    backend = backend or config.load().summary_backend
+    summaries = memory.list_summaries()
+    if not summaries:
+        return None
+
+    # MAP: extract facts from batches of gists.
+    blocks = _batch_texts([s.content for s in summaries], BATCH_CHARS)
+    partials = [_call(_MAP_PROMPT, b, backend) for b in blocks]
+    logbus.log("info", "profile map done", batches=len(partials), sessions=len(summaries))
+
+    # REDUCE: fold partials together until one remains.
+    while len(partials) > 1:
+        partials = [_call(_REDUCE_PROMPT, g, backend) for g in _batch_texts(partials, BATCH_CHARS)]
+    profile = partials[0]
+
+    memory.set_profile(profile, len(summaries))
+    logbus.log("info", "profile rebuilt", sessions=len(summaries), chars=len(profile))
+    return profile
+
+
+def main() -> int:
+    profile = rebuild_profile()
+    if profile is None:
+        print("No summaries yet — run lyra-summarize first.")
+        return 1
+    print(profile)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

lyra/summary.py

@@ -1,17 +1,27 @@
 """Session summarization: compact a session's raw exchanges into a stored gist.
 
-This is the compaction half of the tiered memory. Raw exchanges stay for detail
+This is the first consolidation stage. Raw exchanges stay for detail recall; the
-recall; the summary is what surfaces when an *older* session is recalled later —
+summary is what surfaces when an *older* session is recalled, and it's the input
-"a month ago is a general idea," per the design.
+to the profile (semantic memory) and era-rollup tiers.
+
+Long sessions are summarized in chunks, then the partial gists are merged, so a
+big imported conversation doesn't blow the local model's context window.
 """
 from __future__ import annotations
 
-from lyra import config, llm, logbus, memory
+import sys
-from lyra.llm import Backend
+import time
+from concurrent.futures import ThreadPoolExecutor, as_completed
 
-# Re-summarize a session once it has accumulated this many new raw exchanges
+from lyra import config, llm, logbus, memory
-# beyond what its current summary covers.
+from lyra.llm import Backend, Message
+
+_RETRIES = 4
+
+# Re-summarize a session once it has accumulated this many new raw exchanges.
 SUMMARIZE_AFTER = 20
+# Transcript budget per LLM call; longer sessions are chunked + merged.
+MAX_TRANSCRIPT_CHARS = 24000
 
 _PROMPT = """You are compacting a conversation into a long-term memory record \
 (not replying to anyone). Write a concise gist of the session below: what was \
@@ -24,29 +34,54 @@ def _transcript(exchanges: list[memory.Exchange]) -> str:
     return "\n".join(f"{ex.role}: {ex.content}" for ex in exchanges)
 
 
-def summarize_session(session_id: str, backend: Backend | None = None) -> str | None:
+def _chunk(text: str, budget: int) -> list[str]:
-    """(Re)generate and store the gist for a session. Returns the summary text.
+    """Split on line boundaries into pieces under `budget` chars."""
+    chunks, buf, size = [], [], 0
+    for line in text.splitlines(keepends=True):
+        if size + len(line) > budget and buf:
+            chunks.append("".join(buf))
+            buf, size = [], 0
+        buf.append(line)
+        size += len(line)
+    if buf:
+        chunks.append("".join(buf))
+    return chunks
 
-    Returns None if the session has no exchanges. The summarizer defaults to the
+
-    local backend so routine compaction stays free.
+def _summarize_text(text: str, backend: Backend) -> str:
-    """
+    messages: list[Message] = [
+        {"role": "system", "content": _PROMPT},
+        {"role": "user", "content": text},
+    ]
+    # Retry transient backend errors (e.g. the GPU server restarting) with backoff.
+    for attempt in range(_RETRIES):
+        try:
+            return llm.complete(messages, backend=backend)
+        except Exception as exc:
+            if attempt == _RETRIES - 1:
+                raise
+            logbus.log("debug", "summary retry", attempt=attempt + 1, error=str(exc)[:80])
+            time.sleep(5 * (attempt + 1))
+    raise RuntimeError("unreachable")
+
+
+def _summarize_transcript(transcript: str, backend: Backend) -> str:
+    """Transcript -> gist (LLM only, no DB). Chunks + merges if oversized."""
+    if len(transcript) <= MAX_TRANSCRIPT_CHARS:
+        return _summarize_text(transcript, backend)
+    partials = [_summarize_text(c, backend) for c in _chunk(transcript, MAX_TRANSCRIPT_CHARS)]
+    return _summarize_text("Partial summaries to merge:\n\n" + "\n\n".join(partials), backend)
+
+
+def summarize_session(session_id: str, backend: Backend | None = None) -> str | None:
+    """(Re)generate and store the gist for a session. Returns the summary text."""
     exchanges = memory.history(session_id)
     if not exchanges:
         return None
-
     backend = backend or config.load().summary_backend
-    messages = [
+    gist = _summarize_transcript(_transcript(exchanges), backend)
-        {"role": "system", "content": _PROMPT},
+    memory.store_summary(session_id, gist, exchanges[-1].id)
-        {"role": "user", "content": _transcript(exchanges)},
+    logbus.log("info", "summarized session", session=session_id, exchanges=len(exchanges))
-    ]
-    gist = llm.complete(messages, backend=backend)
-
-    last_id = exchanges[-1].id
-    memory.store_summary(session_id, gist, last_id)
-    logbus.log(
-        "info", "summarized session", session=session_id,
-        exchanges=len(exchanges), backend=backend,
-    )
     return gist
 
 
@@ -54,3 +89,64 @@ def maybe_summarize(session_id: str, backend: Backend | None = None) -> None:
     """Summarize the session if enough new turns have accumulated since last time."""
     if memory.unsummarized_count(session_id) >= SUMMARIZE_AFTER:
         summarize_session(session_id, backend=backend)
+
+
+def summarize_all(
+    backend: Backend | None = None, limit: int | None = None, workers: int = 8
+) -> dict:
+    """Summarize every session that needs it. Idempotent and resumable.
+
+    LLM summarization runs concurrently across `workers` threads (great for a
+    cloud backend). DB reads (loading transcripts) and writes (store_summary,
+    which also embeds) happen on the main thread, so the single SQLite
+    connection is never touched from multiple threads.
+    """
+    backend = backend or config.load().summary_backend
+
+    # Main thread: collect the work (transcripts) for sessions needing a summary.
+    todo: list[tuple[str, str, int]] = []
+    for s in memory.list_sessions():
+        sid = s["id"]
+        if memory.get_summary(sid) and memory.unsummarized_count(sid) == 0:
+            continue
+        exchanges = memory.history(sid)
+        if not exchanges:
+            continue
+        todo.append((sid, _transcript(exchanges), exchanges[-1].id))
+        if limit is not None and len(todo) >= limit:
+            break
+
+    done, failed = 0, 0
+    logbus.log("info", "summarize-all starting", todo=len(todo), backend=backend, workers=workers)
+
+    def work(item: tuple[str, str, int]) -> tuple[str, str, int]:
+        sid, transcript, last_id = item
+        return sid, _summarize_transcript(transcript, backend), last_id
+
+    with ThreadPoolExecutor(max_workers=workers) as pool:
+        futures = {pool.submit(work, item): item for item in todo}
+        for fut in as_completed(futures):
+            sid = futures[fut][0]
+            try:
+                _, gist, last_id = fut.result()
+                memory.store_summary(sid, gist, last_id)  # main thread: embed + write
+                done += 1
+            except Exception as exc:
+                failed += 1
+                logbus.log("error", "summarize failed", session=sid, error=str(exc)[:120])
+            if (done + failed) % 25 == 0:
+                logbus.log("info", "summarize-all progress", done=done, failed=failed, total=len(todo))
+
+    report = {"summarized": done, "failed": failed, "total": len(todo)}
+    logbus.log("info", "summarize-all complete", **report)
+    return report
+
+
+def main() -> int:
+    limit = int(sys.argv[1]) if len(sys.argv) > 1 else None
+    print(summarize_all(limit=limit))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

lyra/web/server.py

@@ -32,7 +32,10 @@ _CLOUD = {"OPENAI", "cloud", "custom"}
 
 
 def _backend_for(label: str | None) -> Backend:
-    if label and label.upper() in {"PRIMARY", "SECONDARY", "FALLBACK", "LOCAL"}:
+    key = (label or "").lower()
+    if key == "mi50":
+        return "mi50"
+    if key in {"local", "primary", "secondary", "fallback"}:
         return "local"
     return "cloud"
 

lyra/web/static/index.html

@@ -123,6 +123,11 @@
               <span>Local — Ollama</span>
               <small>Free, private, runs on your home lab (LOCAL_MODEL)</small>
             </label>
+            <label class="radio-label">
+              <input type="radio" name="backend" value="mi50">
+              <span>MI50 — local GPU</span>
+              <small>Free, llama.cpp on the MI50 box (MI50_BASE_URL)</small>
+            </label>
             <label class="radio-label">
               <input type="radio" name="backend" value="cloud">
               <span>Cloud — OpenAI</span>

pyproject.toml

@@ -17,6 +17,10 @@ dependencies = [
 lyra = "lyra.__main__:main"
 lyra-web = "lyra.web.server:serve"
 lyra-import = "lyra.ingest:main"
+lyra-summarize = "lyra.summary:main"
+lyra-profile = "lyra.profile:main"
+lyra-era = "lyra.era:main"
+lyra-narrative = "lyra.narrative:main"
 
 [dependency-groups]
 dev = [