diff --git a/lyra/chat.py b/lyra/chat.py index 6b9a0ff..67d876b 100644 --- a/lyra/chat.py +++ b/lyra/chat.py @@ -6,7 +6,7 @@ session, then asks the model for a reply and persists both sides. """ from __future__ import annotations -from lyra import llm, memory, persona +from lyra import config, llm, logbus, memory, persona from lyra.llm import Backend, Message RECALL_K = 5 @@ -34,6 +34,7 @@ def build_messages(session_id: str, user_msg: str) -> list[Message]: recalled = [ ex for ex in memory.recall(user_msg, k=RECALL_K) if ex.id not in recent_ids ] + logbus.log("debug", "context built", recent=len(recent), recalled=len(recalled)) if recalled: messages.append(_memory_note(recalled)) @@ -46,8 +47,16 @@ 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 + logbus.log( + "info", "chat request", session=session_id, backend=backend, + model=model, embed=cfg.embed_backend, + ) + messages = build_messages(session_id, user_msg) reply = llm.complete(messages, backend=backend) + logbus.log("info", "reply", session=session_id, chars=len(reply)) memory.remember(session_id, "user", user_msg) memory.remember(session_id, "assistant", reply) diff --git a/lyra/logbus.py b/lyra/logbus.py new file mode 100644 index 0000000..08aefd4 --- /dev/null +++ b/lyra/logbus.py @@ -0,0 +1,31 @@ +"""In-memory live log bus. + +A thread-safe ring buffer that any part of Lyra can publish to and the web +server streams to the browser over SSE. Deliberately process-local and +ephemeral — it's an activity feed, not durable logging. +""" +from __future__ import annotations + +import threading +import time +from collections import deque + +_LOCK = threading.Lock() +_EVENTS: deque[dict] = deque(maxlen=500) +_SEQ = 0 + + +def log(level: str, msg: str, **fields) -> None: + """Publish an event. `level` is info/debug/error/system; fields are extras.""" + global _SEQ + with _LOCK: + _SEQ += 1 + _EVENTS.append( + {"seq": _SEQ, "ts": time.time(), "level": level, "msg": msg, "fields": fields} + ) + + +def since(seq: int) -> list[dict]: + """All buffered events with seq greater than `seq` (for SSE catch-up/polling).""" + with _LOCK: + return [e for e in _EVENTS if e["seq"] > seq] diff --git a/lyra/web/server.py b/lyra/web/server.py index fe61c7d..2688613 100644 --- a/lyra/web/server.py +++ b/lyra/web/server.py @@ -10,15 +10,21 @@ re-stored). from __future__ import annotations import asyncio +import json +import time from pathlib import Path from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse from fastapi.staticfiles import StaticFiles -from lyra import chat, memory +from lyra import chat, logbus, memory from lyra.llm import Backend + +def _sse(event: dict) -> str: + return f"data: {json.dumps(event)}\n\n" + _STATIC = Path(__file__).parent / "static" # UI backend labels -> our two backends. Cloud is the default. @@ -79,7 +85,11 @@ def create_app() -> FastAPI: user_msg = _last_user_message(body.get("messages", [])) memory.ensure_session(session_id) - reply = await asyncio.to_thread(chat.respond, session_id, user_msg, backend) + try: + reply = await asyncio.to_thread(chat.respond, session_id, user_msg, backend) + except Exception as exc: + logbus.log("error", "chat failed", session=session_id, error=str(exc)) + reply = f"[error] {exc}" return { "object": "chat.completion", @@ -92,14 +102,25 @@ def create_app() -> FastAPI: ], } - @app.get("/stream/thinking/{session_id}") - async def thinking_stream(session_id: str) -> StreamingResponse: - # Inert until cognitive layers exist: open the stream, emit keep-alives only. + @app.get("/stream/logs") + async def stream_logs(request: Request) -> StreamingResponse: + """Live activity feed: replay the recent buffer, then stream new events.""" async def gen(): - yield ": connected\n\n" + backlog = logbus.since(0) + last = backlog[-1]["seq"] if backlog else 0 + for e in backlog: + yield _sse(e) + yield _sse( + {"seq": last, "ts": time.time(), "level": "system", + "msg": "live log connected", "fields": {}} + ) while True: - await asyncio.sleep(25) - yield ": keep-alive\n\n" + if await request.is_disconnected(): + break + for e in logbus.since(last): + last = e["seq"] + yield _sse(e) + await asyncio.sleep(0.5) return StreamingResponse(gen(), media_type="text/event-stream") diff --git a/lyra/web/static/index.html b/lyra/web/static/index.html index f668797..3cbb822 100644 --- a/lyra/web/static/index.html +++ b/lyra/web/static/index.html @@ -35,7 +35,7 @@

Actions

- + @@ -68,7 +68,7 @@ - +
@@ -80,10 +80,10 @@
- +