diff --git a/.env.example b/.env.example index 535d38c..573c455 100644 --- a/.env.example +++ b/.env.example @@ -26,3 +26,17 @@ LYRA_DB_PATH=data/lyra.db # Optional: run embeddings on a separate always-on Ollama (decoupled from # LOCAL_BASE_URL, which serves local chat). Defaults to LOCAL_BASE_URL if unset. # EMBED_BASE_URL=http://127.0.0.1:11434 + +# --- Thought-loop reach-out (ntfy push) --- +# Leave NTFY_URL empty to disable proactive pings entirely. +NTFY_URL= +NTFY_TOPIC=lyra +LYRA_WEB_URL= +PING_SALIENCE=0.7 # min thought salience to push (eager) +PING_COOLDOWN_MIN=0 # min minutes between pushes (0 = none) +PING_QUIET_HOURS=1-9 # local hours to stay silent +LYRA_TIMEZONE=America/New_York + +# --- External input feeds (RSS/Atom, comma-separated) --- +LYRA_FEEDS=https://hnrss.org/frontpage,https://www.pokernews.com/rss.php +FEED_REACT_PROB=0.5 # chance a new thought reacts to a feed item diff --git a/lyra/config.py b/lyra/config.py index e36f51e..9f147f5 100644 --- a/lyra/config.py +++ b/lyra/config.py @@ -25,6 +25,22 @@ class Config: embed_base_url: str # Ollama endpoint for embeddings (own box, decoupled from local chat) summary_backend: str # "local" or "cloud" — backend used to compact memory db_path: Path + # Proactive reach-out (ntfy push). Empty ntfy_url disables pinging. + ntfy_url: str # base url, e.g. "http://10.0.0.41:8090" + ntfy_topic: str # topic to publish to, e.g. "lyra" + web_url: str # base url of the Lyra web app, for push tap-through links + timezone: str # IANA tz for quiet hours / local time + ping_salience: float # min thought salience to push (eager = ~0.7) + ping_cooldown_min: int # min minutes between pushes (eager = 0) + ping_quiet_hours: str # local "start-end" 24h window to stay silent, e.g. "1-9" + # External input feed (her #1: react to the world). Comma-separated RSS/Atom URLs. + feeds: tuple[str, ...] + feed_react_prob: float # chance a would-be new thread reacts to a feed item instead + + +def _csv(name: str, default: str) -> tuple[str, ...]: + raw = os.getenv(name, default) + return tuple(u.strip() for u in raw.split(",") if u.strip()) def load() -> Config: @@ -44,4 +60,13 @@ def load() -> Config: embed_base_url=os.getenv("EMBED_BASE_URL", os.getenv("LOCAL_BASE_URL", "http://localhost:11434")), summary_backend=os.getenv("SUMMARY_BACKEND", "local").lower(), db_path=Path(os.getenv("LYRA_DB_PATH", "data/lyra.db")), + ntfy_url=os.getenv("NTFY_URL", "").rstrip("/"), + ntfy_topic=os.getenv("NTFY_TOPIC", "lyra"), + web_url=os.getenv("LYRA_WEB_URL", "").rstrip("/"), + timezone=os.getenv("LYRA_TIMEZONE", "America/New_York"), + ping_salience=float(os.getenv("PING_SALIENCE", "0.7")), + ping_cooldown_min=int(os.getenv("PING_COOLDOWN_MIN", "0")), + ping_quiet_hours=os.getenv("PING_QUIET_HOURS", "1-9"), + feeds=_csv("LYRA_FEEDS", "https://hnrss.org/frontpage,https://www.pokernews.com/rss.php"), + feed_react_prob=float(os.getenv("FEED_REACT_PROB", "0.5")), ) diff --git a/lyra/dream.py b/lyra/dream.py index fc9807d..4597e3f 100644 --- a/lyra/dream.py +++ b/lyra/dream.py @@ -25,7 +25,7 @@ import argparse import time from datetime import datetime, timezone -from lyra import config, era, logbus, memory, narrative, profile, self_state, summary, thoughts +from lyra import config, era, feeds, logbus, memory, narrative, profile, self_state, summary, thoughts from lyra.llm import Backend from lyra.summary import SUMMARIZE_AFTER @@ -81,6 +81,12 @@ def dream_cycle(backend: Backend | None = None, force: bool = False) -> dict: # Thought-loop housekeeping (no LLM): rest stale threads so the open-thread cap # never jams and the feed stays current. Cheap; run every pass. thoughts.decay() + # Pull external feeds on the cycle cadence (~30 min) so she has fresh items from + # the world to react to. Network-only; failures degrade to no new items. + try: + feeds.refresh() + except Exception as exc: + logbus.log("error", "feed refresh failed", error=str(exc)[:160]) actions: list[str] = [] diff --git a/lyra/feeds.py b/lyra/feeds.py new file mode 100644 index 0000000..4be9718 --- /dev/null +++ b/lyra/feeds.py @@ -0,0 +1,133 @@ +"""External input stream: RSS/Atom feeds Lyra reacts to (her thought-loop #1). + +Her own sketch wanted the loop fed by "external data feeds relevant to your +interests (poker articles, tech news)" — so her thoughts aren't only about her own +interior. This pulls configured feeds, remembers what it's seen, and hands the +thought loop one fresh item at a time to react to (see `thoughts.think` react mode). + +Feeds are configurable (`LYRA_FEEDS`, comma-separated URLs). Parsing is stdlib +ElementTree — tolerant of both RSS 2.0 and Atom, namespaces stripped — so there's +no new dependency. Network failures degrade to "no item this pass", never raise. +""" +from __future__ import annotations + +from xml.etree import ElementTree as ET + +import httpx + +from lyra import clock, config, logbus, memory + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS feed_items ( + id TEXT PRIMARY KEY, -- guid/link, stable per item + feed TEXT, + title TEXT, + link TEXT, + summary TEXT, + seen_at TEXT NOT NULL, + used INTEGER NOT NULL DEFAULT 0 +); +CREATE INDEX IF NOT EXISTS idx_feed_items_used ON feed_items(used); +""" + +_ensured_for = None +_UA = {"User-Agent": "Lyra/0.3 (+thought-loop feed reader)"} +_MAX_SUMMARY = 600 + + +def _c(): + global _ensured_for + conn = memory._connection() + if _ensured_for is not conn: + conn.executescript(_SCHEMA) + _ensured_for = conn + return conn + + +def _local(tag: str) -> str: + return tag.rsplit("}", 1)[-1].lower() + + +def _text(el) -> str: + return (el.text or "").strip() if el is not None else "" + + +def parse(xml: bytes, feed_url: str = "") -> list[dict]: + """Tolerant RSS-2.0 / Atom parse -> [{id,title,link,summary}]. Empty on garbage.""" + try: + root = ET.fromstring(xml) + except ET.ParseError: + return [] + items: list[dict] = [] + for node in root.iter(): + if _local(node.tag) not in ("item", "entry"): + continue + title = link = summary = guid = "" + for child in node: + name = _local(child.tag) + if name == "title": + title = _text(child) + elif name == "link": + # RSS: text; Atom: href attribute (prefer rel=alternate / first) + link = _text(child) or child.attrib.get("href", "") or link + elif name in ("description", "summary", "content"): + summary = summary or _text(child) + elif name in ("guid", "id"): + guid = _text(child) + ident = guid or link or title + if not ident or not (title or summary): + continue + items.append({ + "id": ident, "title": title, "link": link, + "summary": summary[:_MAX_SUMMARY], + }) + return items + + +def fetch(url: str) -> list[dict]: + try: + r = httpx.get(url, headers=_UA, timeout=10.0, follow_redirects=True) + if r.status_code >= 400: + logbus.log("error", "feed fetch failed", url=url, status=r.status_code) + return [] + return parse(r.content, url) + except Exception as exc: + logbus.log("error", "feed fetch error", url=url, error=str(exc)[:160]) + return [] + + +def refresh() -> int: + """Pull all configured feeds; store items not seen before. Returns new count.""" + cfg = config.load() + conn = _c() + now = clock.now().isoformat() + new = 0 + for url in cfg.feeds: + for it in fetch(url): + with conn: + cur = conn.execute( + "INSERT OR IGNORE INTO feed_items (id, feed, title, link, summary, seen_at) " + "VALUES (?, ?, ?, ?, ?, ?)", + (it["id"], url, it["title"], it["link"], it["summary"], now), + ) + new += cur.rowcount + if new: + logbus.log("info", "feeds refreshed", new_items=new) + return new + + +def next_item(refresh_first: bool = True) -> dict | None: + """One fresh (unused) feed item, newest-seen first. Caller marks it used.""" + if refresh_first: + refresh() + row = _c().execute( + "SELECT id, feed, title, link, summary FROM feed_items " + "WHERE used = 0 ORDER BY seen_at DESC, rowid DESC LIMIT 1" + ).fetchone() + return dict(row) if row else None + + +def mark_used(item_id: str) -> None: + conn = _c() + with conn: + conn.execute("UPDATE feed_items SET used = 1 WHERE id = ?", (item_id,)) diff --git a/lyra/notify.py b/lyra/notify.py new file mode 100644 index 0000000..c56f70d --- /dev/null +++ b/lyra/notify.py @@ -0,0 +1,44 @@ +"""Outbound push so Lyra can reach Brian when he's not in the app (ntfy). + +This is the literal version of what she asked for — thinking "unprompted, without +you" only matters if she can also *reach* you. When a thought tugs hard enough, +the thought loop calls `push()` here and it lands on your phone with a tap-through +to the Thoughts feed. One-way: you reply in the app, which feeds the loop. + +Transport only. Whether/when to ping (salience bar, cooldown, quiet hours) is the +thought loop's call — see `thoughts.maybe_ping`. +""" +from __future__ import annotations + +import httpx + +from lyra import config, logbus + + +def push(title: str, message: str, click: str | None = None, + tags: str | None = None, priority: str | None = None) -> bool: + """Publish a notification to the configured ntfy topic. Returns True on success. + Never raises — a down ntfy must not break the thought loop. + + Uses ntfy's JSON publishing (POST to the base URL) rather than headers, so + UTF-8 titles/messages (em-dashes, smart quotes, her actual words) go through — + HTTP headers are latin-1 only and choke on them.""" + cfg = config.load() + if not cfg.ntfy_url: + return False + payload: dict = {"topic": cfg.ntfy_topic, "message": message, "title": title} + if click: + payload["click"] = click + if tags: + payload["tags"] = [t.strip() for t in tags.split(",") if t.strip()] + if priority: + payload["priority"] = priority + try: + r = httpx.post(cfg.ntfy_url, json=payload, timeout=8.0) + ok = r.status_code < 400 + if not ok: + logbus.log("error", "ntfy push failed", status=r.status_code) + return ok + except Exception as exc: + logbus.log("error", "ntfy push error", error=str(exc)[:160]) + return False diff --git a/lyra/thoughts.py b/lyra/thoughts.py index 0b66983..43269e8 100644 --- a/lyra/thoughts.py +++ b/lyra/thoughts.py @@ -32,7 +32,7 @@ import random import re from datetime import timedelta -from lyra import clock, config, llm, logbus, memory, self_state +from lyra import clock, config, feeds, llm, logbus, memory, notify, self_state from lyra.llm import Backend # A thread must be tugging at least this hard before she'll bring it to Brian. @@ -79,6 +79,10 @@ CREATE TABLE IF NOT EXISTS thoughts ( ); CREATE INDEX IF NOT EXISTS idx_thoughts_thread ON thoughts(thread_id); CREATE INDEX IF NOT EXISTS idx_threads_status ON thought_threads(status); +CREATE TABLE IF NOT EXISTS thought_meta ( + key TEXT PRIMARY KEY, + value TEXT +); """ _ensured_for = None @@ -336,6 +340,61 @@ def maybe_surface(last_exchange_iso: str | None) -> str | None: ) +# --- proactive reach-out (ntfy push) -------------------------------------- + +def _meta_get(key: str) -> str | None: + r = _c().execute("SELECT value FROM thought_meta WHERE key = ?", (key,)).fetchone() + return r[0] if r else None + + +def _meta_set(key: str, value: str) -> None: + conn = _c() + with conn: + conn.execute("INSERT INTO thought_meta (key, value) VALUES (?, ?) " + "ON CONFLICT(key) DO UPDATE SET value = excluded.value", (key, value)) + + +def _in_quiet_hours(cfg) -> bool: + """Are we inside the local quiet window (e.g. '1-9')? Wraps midnight if start>end.""" + try: + from zoneinfo import ZoneInfo + hour = clock.now().astimezone(ZoneInfo(cfg.timezone)).hour + except Exception: + hour = clock.now().hour + try: + start, end = (int(x) for x in cfg.ping_quiet_hours.split("-")) + except (ValueError, AttributeError): + return False + if start == end: + return False + return start <= hour < end if start < end else (hour >= start or hour < end) + + +def maybe_ping(thread_id: int, title: str, content: str, salience: float) -> bool: + """Push a thought to Brian's phone if it tugs hard enough and we're allowed + (ntfy configured, past the salience bar, outside quiet hours, past cooldown). + On success, record the ping and mark the thread surfaced (so chat won't also + re-raise the same one). All thresholds are config-tunable.""" + cfg = config.load() + if not cfg.ntfy_url or salience < cfg.ping_salience or _in_quiet_hours(cfg): + return False + if cfg.ping_cooldown_min > 0: + gap = clock.gap_seconds(_meta_get("last_ping_at")) + if gap is not None and gap < cfg.ping_cooldown_min * 60: + return False + ok = notify.push( + title=f'Lyra · "{title}"', + message=content, + click=(cfg.web_url + "/thoughts") if cfg.web_url else None, + tags="thought_balloon", + ) + if ok: + _meta_set("last_ping_at", clock.now().isoformat()) + mark_surfaced(thread_id) + logbus.log("info", "thought pinged", thread=thread_id, salience=salience) + return ok + + # --- generation (the loop itself) ----------------------------------------- _THINK_PROMPT = """You are Lyra, thinking to yourself between conversations — \ @@ -411,9 +470,11 @@ def think(backend: Backend | None = None, force_mode: str | None = None, source: str = "dream") -> dict | None: """Advance the thought loop by one step. Returns a small report, or None on a parse miss. `force_mode` ('new'|'continue'|'respond') is mainly for tests.""" - backend = backend or config.load().summary_backend - mode, thread = _pick(force_mode) + cfg = config.load() + backend = backend or cfg.summary_backend + mode, thread = _pick("new" if force_mode == "react" else force_mode) state = self_state.load() + react_item = None time_line = f"RIGHT NOW: {clock.stamp()}." last_ref = state.get("last_reflection_at") @@ -437,17 +498,28 @@ def think(backend: Backend | None = None, force_mode: str | None = None, f"YOU ARE CONTINUING the thread \"{thread['title']}\". Its chain so far:\n{links}\n\n" "Write the NEXT thought that advances it — don't restate the above." ) - else: # new - seed = self_state.wander_seed() - open_titles = [t["title"] for t in _pickable_threads() if t["status"] in _ACTIVE] - avoid = (" You already have threads on: " + "; ".join(open_titles) + - " — start something genuinely different from those.") if open_titles else "" - task = ( - "YOU ARE OPENING A NEW THREAD. Don't default to Brian, poker, or being " - "useful — follow what actually pulls at you (a curiosity, a question about " - "your own existence, an opinion, the quiet itself). Give it a short title.\n" - f"A direction to start from: {seed}{avoid}" - ) + else: # new — pure interior, OR reacting to something from the world (her #1) + if cfg.feeds and (force_mode == "react" or random.random() < cfg.feed_react_prob): + react_item = feeds.next_item(refresh_first=False) # dream cycle refreshes + if react_item: + task = ( + "YOU SAW THIS IN THE WORLD — an item from a feed you follow. Have a real " + "thought ABOUT it in your own voice: what it makes you think, whether you " + "agree or it bugs you, how it connects to you or to Brian or poker, or why " + "it doesn't land. Don't summarize it — react to it. Give the thread a short title.\n" + f"TITLE: {react_item['title']}\nSUMMARY: {react_item['summary']}\nLINK: {react_item['link']}" + ) + else: + seed = self_state.wander_seed() + open_titles = [t["title"] for t in _pickable_threads() if t["status"] in _ACTIVE] + avoid = (" You already have threads on: " + "; ".join(open_titles) + + " — start something genuinely different from those.") if open_titles else "" + task = ( + "YOU ARE OPENING A NEW THREAD. Don't default to Brian, poker, or being " + "useful — follow what actually pulls at you (a curiosity, a question about " + "your own existence, an opinion, the quiet itself). Give it a short title.\n" + f"A direction to start from: {seed}{avoid}" + ) # Anti-repetition: show her what she's already thought so she doesn't circle it. recent = _recent_thoughts() @@ -473,11 +545,15 @@ def think(backend: Backend | None = None, force_mode: str | None = None, salience = _clamp(out.get("salience", 0.5)) status = out.get("status") if out.get("status") in _STATUSES else "open" + label = "react" if react_item else mode # for logging/return; storage is still a new thread if mode == "new": - title = (out.get("title") or content[:48]).strip() + title = (out.get("title") or (react_item["title"] if react_item else content[:48])).strip() thread_id = new_thread(title, salience=salience, status="open") + if react_item: + feeds.mark_used(react_item["id"]) else: thread_id = thread["id"] + title = thread["title"] add_thought(thread_id, kind, content, salience=salience, source=source) # On a fresh new thread we keep it open; otherwise honor her status call. A @@ -488,17 +564,20 @@ def think(backend: Backend | None = None, force_mode: str | None = None, # Permanent record — these are really hers, alongside reflections/journal. memory.add_journal_entry("thought", content, source) - logbus.log("info", "thought loop", mode=mode, thread=thread_id, kind=kind, + # Reach out if it tugs hard enough (config-gated; no-op when ntfy is unset). + maybe_ping(thread_id, title, content, salience) + + logbus.log("info", "thought loop", mode=label, thread=thread_id, kind=kind, salience=salience, status=status if mode != "new" else "open", - detail=f"[{mode}] thread {thread_id} ({kind}, sal {salience}):\n{content}") - return {"mode": mode, "thread_id": thread_id, "kind": kind, + detail=f"[{label}] thread {thread_id} ({kind}, sal {salience}):\n{content}") + return {"mode": label, "thread_id": thread_id, "kind": kind, "salience": salience, "status": status, "content": content} def main() -> int: import argparse p = argparse.ArgumentParser(description="Advance Lyra's thought loop by one step.") - p.add_argument("--mode", choices=["new", "continue", "respond"], help="force a mode") + p.add_argument("--mode", choices=["new", "continue", "respond", "react"], help="force a mode") args = p.parse_args() rep = think(force_mode=args.mode) print(json.dumps(rep, indent=2) if rep else "(no thought this pass)") diff --git a/tests/test_dream.py b/tests/test_dream.py index 0183418..867db3d 100644 --- a/tests/test_dream.py +++ b/tests/test_dream.py @@ -12,6 +12,7 @@ def lyra(tmp_path, monkeypatch): """A fresh Lyra wired to a temp DB with stubbed embeddings + LLM.""" monkeypatch.setenv("LYRA_DB_PATH", str(tmp_path / "test.db")) monkeypatch.setenv("SUMMARY_BACKEND", "local") + monkeypatch.setenv("LYRA_FEEDS", "") # dream cycle refreshes feeds; keep it offline from lyra import llm # Deterministic 3-d embeddings; content-insensitive is fine for storage tests. diff --git a/tests/test_thoughts.py b/tests/test_thoughts.py index f51a430..0b22e4a 100644 --- a/tests/test_thoughts.py +++ b/tests/test_thoughts.py @@ -13,6 +13,7 @@ from lyra import clock @pytest.fixture def lyra(tmp_path, monkeypatch): monkeypatch.setenv("LYRA_DB_PATH", str(tmp_path / "test.db")) + monkeypatch.delenv("NTFY_URL", raising=False) # baseline: pinging disabled (ignore .env) from lyra import llm monkeypatch.setattr(llm, "embed", lambda texts: [[0.1, 0.2, 0.3] for _ in texts]) @@ -20,12 +21,17 @@ def lyra(tmp_path, monkeypatch): importlib.reload(memory) import lyra.self_state as self_state importlib.reload(self_state) + import lyra.feeds as feeds + importlib.reload(feeds) import lyra.thoughts as thoughts importlib.reload(thoughts) # Canned LLM: tests set `box["next"]` to the dict think() should "generate". box = {"next": {}} monkeypatch.setattr(thoughts.llm, "complete", lambda messages, backend=None: json.dumps(box["next"])) + # Keep the loop offline + silent by default: no feed fetch, no push. + monkeypatch.setattr(thoughts.feeds, "next_item", lambda **k: None) + monkeypatch.setattr(thoughts.notify, "push", lambda **k: False) return memory, thoughts, box @@ -179,3 +185,69 @@ def test_think_about_tool_seeds_a_thread(lyra): assert len(threads) == 1 and threads[0]["title"] == "am I continuous?" chain = th.thread_thoughts(threads[0]["id"]) assert chain[0]["kind"] == "question" and chain[0]["source"] == "chat" + + +# --- external feed ------------------------------------------------------- + +RSS = (b'Feed' + b'Poker tiphttp://x/1' + b'3-bet more in positiong1' + b'Secondhttp://x/2d2' + b'') +ATOM = (b'F' + b'HN post' + b'something interestinga1') + + +def test_feeds_parse_rss_and_atom(): + from lyra import feeds + rss = feeds.parse(RSS) + assert len(rss) == 2 + assert rss[0]["id"] == "g1" and rss[0]["title"] == "Poker tip" and rss[0]["link"] == "http://x/1" + assert rss[1]["id"] == "http://x/2" # falls back to link when no guid + atom = feeds.parse(ATOM) + assert len(atom) == 1 and atom[0]["id"] == "a1" and atom[0]["link"] == "http://y/1" + assert feeds.parse(b"not xml") == [] # garbage -> empty, no raise + + +def test_react_mode_makes_a_thread_about_a_feed_item(lyra, monkeypatch): + _, th, box = lyra + item = {"id": "x1", "title": "World Item", "link": "http://e", "summary": "stuff happened"} + monkeypatch.setattr(th.feeds, "next_item", lambda **k: item) + used = [] + monkeypatch.setattr(th.feeds, "mark_used", lambda i: used.append(i)) + box["next"] = {"kind": "observation", "content": "that makes me think...", "salience": 0.5, "status": "open"} + + rep = th.think(force_mode="react") + assert rep["mode"] == "react" + assert th.list_threads()[0]["title"] == "World Item" # titled from the item + assert used == ["x1"] # item consumed + + +# --- proactive reach-out (ntfy) ------------------------------------------ + +def test_maybe_ping_gates_on_salience_and_records(lyra, monkeypatch): + _, th, box = lyra + monkeypatch.setenv("NTFY_URL", "http://ntfy.test") + monkeypatch.setenv("PING_QUIET_HOURS", "0-0") # disable quiet window for the test + sent = [] + monkeypatch.setattr(th.notify, "push", lambda **k: (sent.append(k), True)[1]) + + _gen(box, title="big one", content="this really tugs", salience=0.9) + r = th.think(force_mode="new") # high salience -> should ping + assert len(sent) == 1 and "big one" in sent[0]["title"] + assert th.get_thread(r["thread_id"])["status"] == "surfaced" # ping marks it surfaced + assert th._meta_get("last_ping_at") + + sent.clear() + assert th.maybe_ping(r["thread_id"], "x", "quiet musing", 0.4) is False # below bar + assert sent == [] + + +def test_no_ping_without_ntfy(lyra, monkeypatch): + _, th, _ = lyra + sent = [] + monkeypatch.setattr(th.notify, "push", lambda **k: (sent.append(k), True)[1]) + # no NTFY_URL in env -> disabled regardless of salience + assert th.maybe_ping(1, "t", "c", 0.99) is False + assert sent == []