diff --git a/configs/glados_webapp_config.yaml b/configs/glados_webapp_config.yaml new file mode 100644 index 00000000..81d3e8ea --- /dev/null +++ b/configs/glados_webapp_config.yaml @@ -0,0 +1,65 @@ +# GLaDOS configuration with the webapp observability console enabled. +# +# Start it with: `uv run glados webapp --config configs/glados_webapp_config.yaml` +# Then open http://127.0.0.1:8050/ in a browser. +# +# The console is an in-process HTTP server (stdlib only) that streams the +# engine's live state - minds/slots/subagents/MCP/emotion/audio/lanes and the +# ObservabilityBus event log - over Server-Sent-Events plus a JSON snapshot API. +Glados: + llm_model: "llama3.2" + completion_url: "http://localhost:11434/api/chat" + api_key: null # Add your API key here if needed! + interruptible: true + audio_io: "sounddevice" # local hardware. For a browser-mic setup use "websocket". + input_mode: "audio" # audio, text, or both + tts_enabled: true + asr_muted: false + tui_theme: "aperture" + asr_engine: "tdt" + llm_headers: null # Optional extra headers (e.g., OpenRouter HTTP-Referer, X-Title) + wake_word: null + voice: "glados" + announcement: "All neural network modules are now loaded. System Operational." + + # --- webapp observability console --------------------------------------- + # Default OFF. Enable here, or set GLADOS_WEBAPP_ENABLED=1 / _PORT / _HOST + # environment variables to switch it on without editing YAML. The demo is + # intentionally loopback-only because it has no authentication boundary. + webapp: + enabled: true + host: "127.0.0.1" # Listen address + port: 8050 # Listen port; open http://127.0.0.1:8050/ + + autonomy: + enabled: true + tick_interval_s: 10 + cooldown_s: 20 + autonomy_parallel_calls: 2 # parallel autonomy LLM workers -> the console's "Autonomy" lane + autonomy_queue_max: null + coalesce_ticks: true + jobs: + enabled: true + poll_interval_s: 1 + hacker_news: + enabled: false + interval_s: 1800 + top_n: 5 + min_score: 200 + weather: + enabled: false + interval_s: 3600 + latitude: null + longitude: null + timezone: "auto" + temp_change_c: 4 + wind_alert_kmh: 40 + + mcp_servers: + - name: "system_info" + transport: "stdio" + command: "python" + args: ["-m", "glados.mcp.system_info_server"] + + personality_preprompt: + - system: "You are GLaDOS, a sarcastic AI assistant." diff --git a/docs/webapp.md b/docs/webapp.md new file mode 100644 index 00000000..23ceb99d --- /dev/null +++ b/docs/webapp.md @@ -0,0 +1,145 @@ +# Webapp Console + +An in-process "mission control" for the running GLaDOS engine. It streams the +live parallel state — the two autonomy lanes, subagent contexts, tool state, +PAD emotion, audio/MCP health — to a browser over HTTP + Server-Sent Events. +No separate service to run; no new runtime dependencies. + +## Decoupled launcher (key design) + +The webapp console is **not** part of the core engine. It is started by a +dedicated CLI command — `glados webapp` — that mirrors how `glados tui` works: +it loads the config, builds a `Glados` engine, starts the in-process +`WebappServer` on its own port, then runs the engine loop and shuts the server +down when the loop exits. The engine itself holds no webapp knowledge. + +The observable state (`ObservabilityBus`, `MindRegistry`, `TaskSlotStore`, +subagent memory, interaction/emotion state) only exists inside the running +`Glados` object, so the console server runs in the same process and reads those +objects directly — exactly the pattern the WebSocket audio backend uses. This +keeps it dependency-light and side-effect free for every engine entry point +(`start`, `tui`, `say`). + +The webapp and the TUI are **mutually exclusive UI options** — you run either +`glados webapp` or `glados tui`, never both; the core engine stays agnostic to +which one is attached. + +## Running it + +The console is **off by default**. Enable it, then start it with the webapp +launcher: + +1) YAML config: + + ```yaml + webapp: + enabled: true + host: 127.0.0.1 + port: 8050 + ``` + +2) Environment variables (no config edit): + + ```bash + GLADOS_WEBAPP_ENABLED=1 GLADOS_WEBAPP_PORT=8050 uv run glados webapp + ``` + +Both can be combined with `--config`, `--input-mode`, `--tts-enabled`/ +`--tts-disabled`, and `--asr-muted`/`--asr-unmuted`: + +```bash +uv run glados webapp --config ./configs/glados_webapp_config.yaml +``` + +Then open `http://127.0.0.1:8050/`. + +> **Dummy fallback.** If the page is opened without a live engine — e.g. +> `examples/webapp/index.html` or a directly-opened static file — the page +> detects the missing API and falls back to simulated data so it can be +> previewed and styled. + +## Endpoints + +| Method | Path | Purpose | +| ------ | --------------------- | ------- | +| GET | `/` | Static console (`static/index.html`). | +| GET | `/api/snapshot` | Aggregate JSON snapshot (minds, agents, slots, lanes, audio, emotion, MCP, interaction, vision). | +| GET | `/api/state` | Lightweight state JSON for the live gauges. | +| GET | `/api/stream` | SSE stream (see contract below). | +| GET | `/api/minds` | Registered mind statuses. | +| GET | `/api/minds/{id}` | Single mind status. | +| GET | `/api/minds/{id}/memory` | That agent's private jsonlines memory entries. | +| GET | `/api/slots` | Task slots (summary fields). | +| GET | `/api/slots/{id}` | Full slot including the on-demand report. | +| GET | `/api/agents` | Subagent statuses (`agent_id, title, running, tick_count, last_tick`). | + +The API is deliberately read-only and loopback-only. It does not enable CORS, +rejects cross-origin browser requests, and accepts only `127.0.0.1` or +`localhost` as its configured host. Remote control needs a separately designed +authentication and authorization boundary; this demo does not pretend to +provide one. + +## SSE contract — `/api/stream` + +Each connection first replays the last 100 events from the bus, then receives: + +- `obs` events — the same shape the TUI `ObsScreen` renders: + + ```json + {"timestamp": 1750000000.0, "source": "autonomy", "kind": "slot.update", + "level": "info", "message": "weather brief -> done", "meta": {"slot": "s_weather"}} + ``` + + Real `source`/`kind` combos include `llm.request`, `llm.queue`, + `llm.tool_calls`, `autonomy.dispatch`, `autonomy.slot.update`, + `subagent.start/stop`, `tool.start/finish/error/timeout`, `tts.*`, + `mcp.*`, `vision.update`, `text.user_input`. + +- `state` events — every ~0.5 s, mirroring `/api/state`, so gauges, clock, and + lane chips stay live without re-sending the whole snapshot. + +### Multi-consumer (the important bit) + +The TUI's `ObservabilityScreen` consumes the bus via `drain()`, which is a +*single-consumer* FIFO. The webapp never calls `drain()`. Instead every SSE +connection registers its own private `subscribe()` queue on +`ObservabilityBus`, so multiple browsers each get their own copy and never +steal events from the TUI or from each other. This is fully backward-compatible: +`drain()` and `snapshot()` keep their existing behavior. + +## Failure behavior + +Because the webapp is the point of the `glados webapp` launcher, a disabled +console or a bind failure (port in use) is **fatal for that command**: the +launcher logs an error and exits rather than running the engine without the +console. This does not affect other entry points — `tui`, `start`, and `say` +are completely independent of the webapp and never bind a port. + +- Request exceptions return JSON error bodies and never crash a worker thread. +- Client disconnects free their subscription. + +## Lifecycle + +The `glados webapp` command orchestrates the whole lifecycle: + +1. Load `GladosConfig.from_yaml` and apply CLI overrides. +2. Refuse to start if `webapp.enabled` is false. +3. Build `Glados.from_config`, then `WebappServer(engine, host, port).start()`. +4. Run `engine.run()`; on any exit (`try/finally`), shut the server down. + +The core engine stays decoupled and side-effect free. This mirrors the TUI's +launcher pattern: the same `GLADOS_WEBAPP_*` environment variables live on +`GladosConfig.webapp`, so the launcher reads one merged config. +`GladosConfig.webapp` stays a field on the shared config model, but the engine +never reads it at runtime — the decoupling is at the launcher boundary. + +## Development + +- Console page: `src/glados/webapp/static/index.html` — single self-contained + file (Aperture/GLaDOS theme, no build step). `examples/webapp/index.html` is + the standalone mockup it derives from. +- Server: `src/glados/webapp/server.py`; serializers: `serializers.py`; + config: `config.py`. +- Tests: `tests/test_webapp.py` — bus fan-out, serializers, and an in-process + HTTP smoke test against a stub engine. +- Example config: `configs/glados_webapp_config.yaml`. diff --git a/examples/webapp/index.html b/examples/webapp/index.html new file mode 100644 index 00000000..277d9a4a --- /dev/null +++ b/examples/webapp/index.html @@ -0,0 +1,719 @@ + + + + + +GLaDOS Core Console — Mockup + + + + + +
+
+
+
GLaDOS
CORE CONSOLE · OBSERVATION DECK
+
+ UP 03:12:48 + --:--:-- + SYSTEMS NOMINAL +
+ + + +
+ + +
+
+

Mission Control

+ watch the brain think, in parallel. +
autonomy pool 2/4in-flight 0
+
+
+
0
Slots tracked
+
0
Active minds
+
0
Inferences in-flight
+
+
+
+

Dual Lane Orchestration

+
+
+
+ +
Priority Lane
direct user agent · 1 lane
+
0 in-flight
+
+
+
+
+
+ +
Autonomy Lane
pooled brains · 2 workers
+
0 in-flight
+
+
+
+
+
+
+

Brainstem PAD + interaction

+
+
Pleasure
0.0
+
Arousal
0.0
+
Dominance
0.0
+
+
+
●Last user ----
+
●Last spoke ----
+
+
+
+
+

The Wire · live stream

+
+
+
+ + +
+

Neural Core

all parallel inferences, in-flight and queued.
--
+
+
+
Priority Lane
the direct user agent · always first
--
+
+
+
+
Autonomy Lane
pool of Minds · up to 12 workers
--
+
+
+
+
+

Worker pool autonomy_parallel_calls = 2

+
+
+
+ + +
+

Minds

independent subagents, each with a private context.
+
+
+ + +
+

Slots

shared state the main agent consumes.
shared
+
+
+ + +
+

The Wire

observability bus stream.
+
+
+
+ + +
+

Vitals

everything else the TUI tracks.
+
+

Audio

+

Interaction

+

Vision

+

MCP servers

+
+
+
+
+ + + + + + diff --git a/src/glados/cli.py b/src/glados/cli.py index 5d142754..8975c781 100644 --- a/src/glados/cli.py +++ b/src/glados/cli.py @@ -13,6 +13,7 @@ from .TTS import tts_glados from .utils import spoken_text_converter as stc from .utils.resources import resource_path +from .webapp import WebappServer # Type aliases for clarity type FileHash = str @@ -272,6 +273,54 @@ def tui( sys.exit() +def run_webapp( + config_path: str | Path | list[str] | list[Path] = "glados_config.yaml", + input_mode: str | None = None, + tts_enabled: bool | None = None, + asr_muted: bool | None = None, +) -> None: + """Run GLaDOS with its browser-based observability console.""" + from loguru import logger + + glados_config = GladosConfig.from_yaml(config_path) + updates: dict[str, object] = {} + if input_mode: + updates["input_mode"] = input_mode + if tts_enabled is not None: + updates["tts_enabled"] = tts_enabled + if asr_muted is not None: + updates["asr_muted"] = asr_muted + if updates: + glados_config = glados_config.model_copy(update=updates) + + webapp_config = glados_config.webapp + if webapp_config is None or not webapp_config.enabled: + logger.error( + "The webapp console is disabled. Enable webapp.enabled or set " + "GLADOS_WEBAPP_ENABLED=1." + ) + raise SystemExit(1) + + glados = Glados.from_config(glados_config) + server = WebappServer(glados, host=webapp_config.host, port=webapp_config.port) + server.start() + if not server.is_running: + logger.error( + "Webapp console could not bind {}:{} - aborting.", + webapp_config.host, + webapp_config.port, + ) + glados._graceful_shutdown() + raise SystemExit(1) + + try: + if glados.announcement: + glados.play_announcement() + glados.run() + finally: + server.shutdown() + + def parser_add_config(parser: argparse.ArgumentParser) -> None: """ Add the '--config' argument to the given parser. @@ -389,6 +438,11 @@ def main() -> int: help="Override TUI theme (aperture, ice, matrix, mono, ember)", ) + webapp_parser = subparsers.add_parser( + "webapp", help="Start GLaDOS with the browser observability console" + ) + parser_add_common_tui_cli_args(webapp_parser) + # Say command say_parser = subparsers.add_parser("say", help="Make GLaDOS speak text") say_parser.add_argument("text", type=str, help="Text for GLaDOS to speak") @@ -419,6 +473,13 @@ def main() -> int: asr_muted=args.asr_muted, theme=args.theme, ) + elif args.command == "webapp": + run_webapp( + args.config, + input_mode=args.input_mode, + tts_enabled=args.tts_enabled, + asr_muted=args.asr_muted, + ) else: # Default to start if no command specified start(DEFAULT_CONFIG) diff --git a/src/glados/core/engine.py b/src/glados/core/engine.py index 5aa82884..07d896de 100644 --- a/src/glados/core/engine.py +++ b/src/glados/core/engine.py @@ -31,6 +31,7 @@ from ..autonomy.summarization import estimate_tokens from ..mcp import MCPManager, MCPServerConfig from ..observability import MindRegistry, ObservabilityBus, trim_message +from ..webapp.config import WebappConfig from ..vision import VisionConfig, VisionState from ..vision.constants import SYSTEM_PROMPT_VISION_HANDLING from .audio_data import AudioMessage @@ -124,6 +125,7 @@ class GladosConfig(BaseModel): vision: VisionConfig | None = None autonomy: AutonomyConfig | None = None mcp_servers: list[MCPServerConfig] | None = None + webapp: WebappConfig | None = None @model_validator(mode="after") def _resolve_api_key_from_env(self) -> "GladosConfig": @@ -134,6 +136,25 @@ def _resolve_api_key_from_env(self) -> "GladosConfig": self.api_key = env_key return self + @model_validator(mode="after") + def _apply_webapp_env(self) -> "GladosConfig": + """Apply optional ``GLADOS_WEBAPP_*`` environment overrides.""" + flag = os.environ.get("GLADOS_WEBAPP_ENABLED") + host = os.environ.get("GLADOS_WEBAPP_HOST") + port = os.environ.get("GLADOS_WEBAPP_PORT") + if flag is None and host is None and port is None: + return self + + webapp = self.webapp or WebappConfig() + self.webapp = WebappConfig.model_validate( + { + "enabled": webapp.enabled if flag is None else flag, + "host": host or webapp.host, + "port": port or webapp.port, + } + ) + return self + @classmethod def from_yaml(cls, paths: str | Path | list[str] | list[Path], key_to_config: tuple[str, ...] = ("Glados",)) -> "GladosConfig": """ diff --git a/src/glados/observability/bus.py b/src/glados/observability/bus.py index 4d3f5096..da6d8def 100644 --- a/src/glados/observability/bus.py +++ b/src/glados/observability/bus.py @@ -1,3 +1,5 @@ +"""Thread-safe observability event distribution.""" + from __future__ import annotations from collections import deque @@ -10,10 +12,15 @@ class ObservabilityBus: - def __init__(self, max_history: int = 500) -> None: + """Thread-safe event bus with bounded history and independent consumers.""" + + def __init__(self, max_history: int = 500, subscriber_max: int = 100) -> None: + """Initialize bounded history and per-consumer queue capacity.""" self._queue: queue.Queue[ObservabilityEvent] = queue.Queue() self._lock = threading.Lock() self._history: deque[ObservabilityEvent] = deque(maxlen=max_history) + self._subscriber_max = max(1, subscriber_max) + self._subscribers: list[queue.Queue[ObservabilityEvent]] = [] def emit( self, @@ -23,6 +30,7 @@ def emit( level: str = "info", meta: dict[str, Any] | None = None, ) -> ObservabilityEvent: + """Construct, publish, and return an observability event.""" event = ObservabilityEvent( timestamp=time.time(), source=source, @@ -35,11 +43,39 @@ def emit( return event def publish(self, event: ObservabilityEvent) -> None: + """Publish an event to history, the legacy queue, and all subscribers.""" with self._lock: self._history.append(event) - self._queue.put(event) + for subscriber in self._subscribers: + try: + subscriber.put_nowait(event) + except queue.Full: + # A slow subscriber must not block producers. Keep its + # newest events by evicting one oldest item. + try: + subscriber.get_nowait() + subscriber.put_nowait(event) + except (queue.Empty, queue.Full): + pass + self._queue.put(event) + + def subscribe(self) -> queue.Queue[ObservabilityEvent]: + """Return a bounded queue which receives every future event.""" + subscriber: queue.Queue[ObservabilityEvent] = queue.Queue(maxsize=self._subscriber_max) + with self._lock: + self._subscribers.append(subscriber) + return subscriber + + def unsubscribe(self, subscriber: queue.Queue[ObservabilityEvent]) -> None: + """Remove a queue previously returned by :meth:`subscribe`.""" + with self._lock: + try: + self._subscribers.remove(subscriber) + except ValueError: + pass def drain(self, max_items: int = 100) -> list[ObservabilityEvent]: + """Remove up to ``max_items`` from the legacy single-consumer queue.""" events: list[ObservabilityEvent] = [] for _ in range(max_items): try: @@ -49,6 +85,7 @@ def drain(self, max_items: int = 100) -> list[ObservabilityEvent]: return events def snapshot(self, limit: int | None = None) -> list[ObservabilityEvent]: + """Return a stable copy of recent event history.""" with self._lock: events = list(self._history) if limit is None or limit <= 0: @@ -56,6 +93,7 @@ def snapshot(self, limit: int | None = None) -> list[ObservabilityEvent]: return events[-limit:] def clear(self) -> None: + """Clear history and the legacy queue.""" with self._lock: self._history.clear() try: diff --git a/src/glados/webapp/__init__.py b/src/glados/webapp/__init__.py new file mode 100644 index 00000000..c9c8401b --- /dev/null +++ b/src/glados/webapp/__init__.py @@ -0,0 +1,14 @@ +"""GLaDOS webapp observability console. + +An in-process web server (stdlib only) that exposes the engine's live state to a +browser: a snapshot REST API plus a Server-Sent-Events stream built on the +thread-safe ``ObservabilityBus``. Enable it via ``webapp: {enabled: true}`` in +the config (or ``GLADOS_WEBAPP_ENABLED=1``) and open http://127.0.0.1:8050/. +""" + +from __future__ import annotations + +from .config import WebappConfig +from .server import WebappServer + +__all__ = ["WebappConfig", "WebappServer"] diff --git a/src/glados/webapp/config.py b/src/glados/webapp/config.py new file mode 100644 index 00000000..c7a9419a --- /dev/null +++ b/src/glados/webapp/config.py @@ -0,0 +1,21 @@ +"""Configuration for the Glados webapp observability console.""" + +from __future__ import annotations + +from typing import Literal + +from pydantic import BaseModel, Field + + +class WebappConfig(BaseModel): + """Webapp observability console server configuration.""" + + enabled: bool = Field( + default=False, + description="Serve the webapp observability console (default off).", + ) + host: Literal["127.0.0.1", "localhost"] = Field( + default="127.0.0.1", + description="Loopback listen address; remote access is intentionally disabled.", + ) + port: int = Field(default=8050, ge=0, le=65535, description="Listen port.") diff --git a/src/glados/webapp/serializers.py b/src/glados/webapp/serializers.py new file mode 100644 index 00000000..b6844fa7 --- /dev/null +++ b/src/glados/webapp/serializers.py @@ -0,0 +1,249 @@ +"""Map live engine state objects to plain JSON-serializable dicts. + +These helpers mirror the TUI's panels: they read the exact same thread-safe +accessors (MindRegistry, TaskSlotStore, SubagentManager, MCPManager, AudioState, +InteractionState, queue sizes, ...) and turn them into plain dicts for the +webapp. No state is duplicated - the console just reads what the engine tracks. +""" +# The console intentionally adapts several optional engine components through a +# runtime-shaped boundary. Concrete protocols for every combination would add +# coupling without improving safety at this read-only telemetry edge. +# ruff: noqa: ANN401 + +from __future__ import annotations + +from dataclasses import asdict +import json +import time +from typing import Any, cast + +try: + import numpy as np + + _HAS_NUMPY = True +except Exception: # pragma: no cover + np = None # type: ignore[assignment] + _HAS_NUMPY = False + + +def _plain(value: Any) -> Any: + """Recursively coerce numpy scalars/arrays and dataclasses to JSON-safe values.""" + if _HAS_NUMPY and isinstance(value, np.generic): + return value.item() + if _HAS_NUMPY and isinstance(value, np.ndarray): + return value.tolist() + if hasattr(value, "to_dict") and callable(value.to_dict): + return _plain(value.to_dict()) + if hasattr(value, "__dataclass_fields__"): + return _plain(asdict(value)) + if isinstance(value, dict): + return {str(k): _plain(v) for k, v in value.items()} + if isinstance(value, list | tuple): + return [_plain(v) for v in value] + return value + + +def to_jsonable(obj: Any) -> Any: + """Return a recursively JSON-compatible representation of ``obj``.""" + return _plain(obj) + + +def dumps(obj: Any) -> str: + """Serialize to JSON without choking on dataclasses, numpy, or timestamps.""" + return json.dumps(obj, default=_json_default) + + +def _json_default(value: Any) -> Any: + """Coerce otherwise unsupported values for :func:`json.dumps`.""" + try: + plain = _plain(value) + return str(value) if plain is value else plain + except Exception: # pragma: no cover + return str(value) + + +def serialize_event(event: Any) -> dict[str, Any]: + """ObservabilityEvent -> {timestamp, source, kind, level, message, meta}.""" + return { + "timestamp": getattr(event, "timestamp", time.time()), + "source": getattr(event, "source", ""), + "kind": getattr(event, "kind", ""), + "level": getattr(event, "level", "info"), + "message": getattr(event, "message", ""), + "meta": to_jsonable(getattr(event, "meta", {})), + } + + +def serialize_mind(mind: Any) -> dict[str, Any]: + """Serialize a registered mind's public status.""" + return { + "mind_id": mind.mind_id, + "title": mind.title, + "status": mind.status, + "summary": mind.summary, + "role": mind.role, + "updated_at": mind.updated_at, + } + + +def serialize_slot(slot: Any) -> dict[str, Any]: + """Serialize task-slot summary fields.""" + return { + "slot_id": slot.slot_id, + "title": slot.title, + "status": slot.status, + "summary": slot.summary, + "notify_user": slot.notify_user, + "importance": slot.importance, + "confidence": slot.confidence, + "next_run": slot.next_run, + "updated_at": slot.updated_at, + "has_report": bool(slot.report), + } + + +def serialize_slot_full(slot: Any) -> dict[str, Any]: + """Serialize a task slot including its report.""" + data = serialize_slot(slot) + data["report"] = slot.report + return data + + +def serialize_memory_entry(entry: Any) -> dict[str, Any]: + """Serialize a subagent memory entry.""" + return { + "key": entry.key, + "value": to_jsonable(entry.value), + "created_at": entry.created_at, + "shown_at": entry.shown_at, + } + + +def serialize_agent(agent_status: Any) -> dict[str, Any]: + """Serialize a subagent's runtime status.""" + return { + "agent_id": agent_status.agent_id, + "title": agent_status.title, + "running": agent_status.running, + "tick_count": agent_status.tick_count, + "last_tick": agent_status.last_tick, + } + + +def _audio_state(engine: Any) -> dict[str, Any]: + """Read the current audio meter state from the engine.""" + audio_state = getattr(engine, "audio_state", None) + if audio_state is None: + return {"rms": 0.0, "vad_active": False} + snap = audio_state.snapshot() + return {"rms": float(snap.rms), "vad_active": bool(snap.vad_active)} + + +def _emotion_state(engine: Any) -> dict[str, Any] | None: + """Return the optional emotion-agent state.""" + agent = getattr(engine, "_emotion_agent", None) + if agent is None: + return None + try: + return cast(dict[str, Any], to_jsonable(agent.state)) + except Exception: # pragma: no cover + return None + + +def _mcp_summary(engine: Any) -> dict[str, Any]: + """Return a failure-tolerant MCP status summary.""" + manager = getattr(engine, "mcp_manager", None) + if manager is None: + return {"enabled": False, "servers": []} + try: + servers = manager.status_snapshot() + except Exception: # pragma: no cover + servers = [] + return {"enabled": True, "servers": servers} + + +def build_lanes(engine: Any) -> dict[str, Any]: + """Summarize priority and autonomy inference lanes.""" + return { + "enabled": bool(getattr(engine, "autonomy_config", None) and engine.autonomy_config.enabled), + "priority": {"queue": int(engine.llm_queue_priority.qsize())}, + "autonomy": { + "queue": int(engine.llm_queue_autonomy.qsize()), + "inflight": int(engine._autonomy_inflight.value()), + "workers": len(getattr(engine, "autonomy_llm_processors", ())), + }, + } + + +def build_state(engine: Any) -> dict[str, Any]: + """Lightweight payload streamed periodically to keep gauges/clock live.""" + return { + "t": time.time(), + "interaction": { + "seconds_since_user": engine.interaction_state.seconds_since_user(), + "seconds_since_assistant": engine.interaction_state.seconds_since_assistant(), + }, + "lanes": build_lanes(engine), + "audio": _audio_state(engine), + "emotion": _emotion_state(engine), + "mcp": _mcp_summary(engine), + "speaking": bool(engine.currently_speaking_event.is_set()), + } + + +def build_snapshot(engine: Any) -> dict[str, Any]: + """Aggregate snapshot for the console's initial paint (GET /api/snapshot).""" + vision_state = getattr(engine, "vision_state", None) + snapshot: dict[str, Any] = { + "t": time.time(), + "version": "0.1", + "autonomy_enabled": bool(getattr(engine, "autonomy_config", None) and engine.autonomy_config.enabled), + "lanes": build_lanes(engine), + "audio": _audio_state(engine), + "emotion": _emotion_state(engine), + "mcp": _mcp_summary(engine), + "interaction": { + "seconds_since_user": engine.interaction_state.seconds_since_user(), + "seconds_since_assistant": engine.interaction_state.seconds_since_assistant(), + }, + "speaking": bool(engine.currently_speaking_event.is_set()), + "minds": [_safe(serialize_mind, m) for m in engine.mind_registry.snapshot()], + "slots": ( + [_safe(serialize_slot, s) for s in engine.autonomy_slots.list_slots()] + if getattr(engine, "autonomy_slots", None) + else [] + ), + "agents": (_safe_agents(engine) if getattr(engine, "subagent_manager", None) else []), + "vision": vision_state.snapshot() if vision_state else None, + } + return snapshot + + +def _safe(serializer: Any, item: Any) -> dict[str, Any]: + """Serialize one item without allowing telemetry to break snapshots.""" + try: + return cast(dict[str, Any], serializer(item)) + except Exception: # pragma: no cover + return {} + + +def _safe_agents(engine: Any) -> list[dict[str, Any]]: + """Return all subagent statuses, or an empty list if unavailable.""" + try: + return [_safe(serialize_agent, a) for a in engine.subagent_manager.list_agents()] + except Exception: # pragma: no cover + return [] + + +__all__ = [ + "build_snapshot", + "build_state", + "dumps", + "serialize_agent", + "serialize_event", + "serialize_memory_entry", + "serialize_mind", + "serialize_slot", + "serialize_slot_full", + "to_jsonable", +] diff --git a/src/glados/webapp/server.py b/src/glados/webapp/server.py new file mode 100644 index 00000000..75f108ee --- /dev/null +++ b/src/glados/webapp/server.py @@ -0,0 +1,393 @@ +"""In-process webapp observability console server. + +Runs inside the Glados engine process (mirroring the websocket-audio server +pattern) so it can read live thread-safe state directly. A single stdlib +:class:`http.server.ThreadingHTTPServer` serves the static console, a JSON +snapshot API, and a Server-Sent-Events (SSE) stream that pushes observability +events plus periodic state pings to every connected browser. + +Endpoints +--------- + GET / static console (``static/index.html``) + GET /api/snapshot aggregate JSON snapshot + GET /api/state lightweight state JSON + GET /api/stream SSE: "obs" events + "state" pings + GET /api/minds registered mind statuses + GET /api/minds/{id} single mind status + GET /api/minds/{id}/memory that agent's jsonlines memory entries + GET /api/slots task slots (summary fields) + GET /api/slots/{id} full slot incl. on-demand report + GET /api/agents registered subagent statuses +""" +# The HTTP boundary intentionally accepts the live engine and its optional +# components structurally; serializers isolate those dynamic reads. +# ruff: noqa: ANN401 + +from __future__ import annotations + +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +import threading +from typing import Any, cast +from urllib.parse import unquote, urlparse + +from loguru import logger + +from .serializers import ( + build_snapshot, + build_state, + dumps, + serialize_agent, + serialize_event, + serialize_memory_entry, + serialize_mind, + serialize_slot, + serialize_slot_full, +) + +STATIC_DIR = Path(__file__).resolve().parent / "static" + + +def _serialize_or(serializer: Any, item: Any) -> dict[str, Any]: + """Serialize one API item, failing closed to an empty object.""" + try: + return cast(dict[str, Any], serializer(item)) + except Exception: # pragma: no cover + return {} + + +# --------------------------------------------------------------------------- helpers + + +def _content_type(name: str) -> str: + """Map a static asset suffix to its HTTP content type.""" + return { + ".html": "text/html; charset=utf-8", + ".js": "application/javascript; charset=utf-8", + ".css": "text/css; charset=utf-8", + ".json": "application/json; charset=utf-8", + ".svg": "image/svg+xml", + ".png": "image/png", + ".ico": "image/x-icon", + ".wasm": "application/wasm", + }.get(Path(name).suffix.lower(), "application/octet-stream") + + +def _find_mind(engine: Any, mind_id: str) -> Any | None: + """Find a mind status by identifier in a registry snapshot.""" + for mind in engine.mind_registry.snapshot(): + if mind.mind_id == mind_id: + return mind + return None + + +def _find_store(engine: Any) -> Any | None: + """Return the optional task-slot store.""" + return getattr(engine, "autonomy_slots", None) + + +def _agent_manager(engine: Any) -> Any | None: + """Return the optional subagent manager.""" + return getattr(engine, "subagent_manager", None) + + +class _EngineHTTPServer(ThreadingHTTPServer): + """Threading server carrying the live engine reference to handlers.""" + + daemon_threads = True + + def __init__(self, address: tuple[str, int], engine: Any) -> None: + """Attach the live engine to a standard threaded HTTP server.""" + self.engine = engine + super().__init__(address, _Handler) + + +class _Handler(BaseHTTPRequestHandler): + """Stateless handler; reads the engine reference from ``self.server.engine``.""" + + protocol_version = "HTTP/1.1" + + # ------------------------------------------------------------ utilities + def _path(self) -> str: + """Return the decoded request path without its query string.""" + return urlparse(self.path).path + + def _json(self, code: int, payload: Any) -> None: + """Send one length-delimited JSON response.""" + body = dumps(payload).encode("utf-8") + self.send_response(code) + self.send_header("Content-Type", "application/json; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def _text(self, code: int, text: str) -> None: + """Send one UTF-8 plain-text response.""" + body = text.encode("utf-8") + self.send_response(code) + self.send_header("Content-Type", "text/plain; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def _file(self, name: str) -> None: + """Serve a file confined to the packaged static directory.""" + target = (STATIC_DIR / name).resolve() + if not target.is_relative_to(STATIC_DIR.resolve()) or not target.is_file(): + return self._text(404, "Not found") + body = target.read_bytes() + self.send_response(200) + self.send_header("Content-Type", _content_type(name)) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, fmt: str, *args: Any) -> None: + """Route request logging through Loguru.""" + logger.debug("[webapp] {} {}", self.address_string(), fmt % args) + + # --------------------------------------------------------------- do_GET + def do_GET(self) -> None: # noqa: N802 - required by BaseHTTPRequestHandler + """Serve the console, its assets, or a read-only API endpoint.""" + if not self._same_origin(): + return self._json(403, {"error": "cross-origin requests are not allowed"}) + path = self._path() + if path in ("/", "/index.html"): + return self._file("index.html") + if path.startswith("/static/"): + return self._file(path[len("/static/") :]) + if path.startswith("/api/"): + try: + return self._route_api(path) + except (BrokenPipeError, ConnectionResetError): # pragma: no cover + return + except OSError: # pragma: no cover + return + except Exception: # pragma: no cover - keep request failures isolated + logger.exception("webapp API request failed") + return self._json(500, {"error": "internal server error"}) + return self._text(404, "Not found") + + def _same_origin(self) -> bool: + """Allow direct clients and same-origin browsers, never cross-origin pages.""" + request_host = self.headers.get("Host", "") + hostname = urlparse(f"//{request_host}").hostname + if hostname not in {"127.0.0.1", "localhost"}: + return False + origin = self.headers.get("Origin") + if origin is None: + return True + parsed = urlparse(origin) + return parsed.scheme in {"http", "https"} and parsed.netloc.lower() == request_host.lower() + + # -------------------------------------------------------------- routing + def _route_api(self, path: str) -> None: + """Dispatch a read-only API path.""" + engine = cast(_EngineHTTPServer, self.server).engine + if path == "/api/stream": + return self._stream(engine) + if path == "/api/snapshot": + return self._json(200, build_snapshot(engine)) + if path == "/api/state": + return self._json(200, build_state(engine)) + if path == "/api/minds": + minds = [_serialize_or(serialize_mind, m) for m in engine.mind_registry.snapshot()] + return self._json(200, {"minds": minds}) + if path == "/api/agents": + return self._json(200, {"agents": self._agent_list(engine)}) + if path == "/api/slots": + slots = [_serialize_or(serialize_slot, s) for s in self._slots(engine)] + return self._json(200, {"slots": slots}) + + mind_rest = _sub_path(path, "/api/minds/") + if mind_rest is not None: + parts = mind_rest.split("/", 1) + mind_id = unquote(parts[0]) + sub = parts[1] if len(parts) > 1 else "" + if sub == "memory": + entries = self._memory(engine, mind_id) + if entries is None: + return self._json(404, {"error": "No subagent memory for that mind"}) + return self._json(200, {"agent_id": mind_id, "memory": entries}) + if sub: + return self._json(404, {"error": "Unknown sub-path"}) + mind = _find_mind(engine, mind_id) + if mind is None: + return self._json(404, {"error": "mind not found"}) + return self._json(200, serialize_mind(mind)) + + slot_id = _sub_path(path, "/api/slots/") + if slot_id is not None: + store = _find_store(engine) + slot = store.get_slot(unquote(slot_id)) if store is not None else None + if slot is None: + return self._json(404, {"error": "slot not found"}) + return self._json(200, serialize_slot_full(slot)) + + self._json(404, {"error": "not found"}) + + # --------------------------------------------------------------- SSE + def _stream(self, engine: Any) -> None: + """SSE stream: replay history then push live events + periodic state. + + Uses a private :meth:`ObservabilityBus.subscribe` queue so each browser + gets its own copy of the stream instead of competing over the TUI's + single-consumer ``drain()`` queue. + """ + import queue as _queue + + bus = engine.observability_bus + self.send_response(200) + self.send_header("Content-Type", "text/event-stream; charset=utf-8") + self.send_header("Cache-Control", "no-cache") + self.send_header("Connection", "keep-alive") + self.end_headers() + + sub = bus.subscribe() + try: + # Subscribe before taking history so an event cannot land in the + # gap between replay and live delivery. Skip the same event object + # if it appears in both history and the private queue. + history = bus.snapshot(limit=100)[-100:] + replayed = {id(event) for event in history} + for event in history: + self._send_sse("obs", serialize_event(event)) + import time as _time + last_state_time = _time.time() + while not engine.shutdown_event.is_set(): + try: + event = sub.get(timeout=0.5) + except _queue.Empty: + self._send_sse("state", build_state(engine)) + last_state_time = _time.time() + continue + if id(event) in replayed: + replayed.remove(id(event)) + continue + self._send_sse("obs", serialize_event(event)) + # Emit state frame if at least 0.5 seconds have elapsed + now = _time.time() + if now - last_state_time >= 0.5: + self._send_sse("state", build_state(engine)) + last_state_time = now + except (BrokenPipeError, ConnectionResetError, OSError): # pragma: no cover + pass + except Exception: # pragma: no cover - stream already has HTTP headers + logger.exception("webapp event stream failed") + finally: + bus.unsubscribe(sub) + + def _send_sse(self, event_type: str, payload: Any) -> None: + """Write and flush one Server-Sent Event frame.""" + data = dumps(payload) + frame = (f"event: {event_type}\ndata: {data}\n\n").encode() + self.wfile.write(frame) + self.wfile.flush() + + # -------------------------------------------------------------- helpers + def _slots(self, engine: Any) -> list[Any]: + """Return task slots without propagating telemetry read failures.""" + store = _find_store(engine) + if store is None: + return [] + try: + return cast(list[Any], store.list_slots()) + except Exception: # pragma: no cover + return [] + + def _agent_list(self, engine: Any) -> list[dict[str, Any]]: + """Return serialized subagent statuses.""" + manager = _agent_manager(engine) + if manager is None: + return [] + try: + return [_serialize_or(serialize_agent, a) for a in manager.list_agents()] + except Exception: # pragma: no cover + return [] + + def _memory(self, engine: Any, agent_id: str) -> list[dict[str, Any]] | None: + """Return serialized private memory for a known subagent.""" + manager = _agent_manager(engine) + if manager is None: + return None + try: + subagent = manager.get(agent_id) + except Exception: # pragma: no cover + return None + if subagent is None: + return None + try: + entries = subagent.memory.list_all() + except Exception: # pragma: no cover + return None + return [_serialize_or(serialize_memory_entry, e) for e in entries] + + +def _sub_path(path: str, prefix: str) -> str | None: + """Extract a non-empty, non-directory suffix beneath an API prefix.""" + if not path.startswith(prefix): + return None + rest = path[len(prefix) :] + if not rest or rest.endswith("/"): + return None + return rest + + +class WebappServer: + """Lifecycle wrapper: start/stop the console server on a background thread.""" + + def __init__(self, engine: Any, host: str = "127.0.0.1", port: int = 8050) -> None: + """Configure a loopback-only server without binding it yet.""" + if host not in {"127.0.0.1", "localhost"}: + raise ValueError("The unauthenticated webapp console is loopback-only") + self.engine = engine + self.host = host + self.port = port + self._server: _EngineHTTPServer | None = None + self._thread: threading.Thread | None = None + self.bound_port: int | None = None + + def start(self) -> None: + """Bind the socket and start serving on a daemon thread.""" + if self._server is not None: + return + try: + self._server = _EngineHTTPServer((self.host, self.port), self.engine) + self.bound_port = self._server.server_address[1] + except OSError as exc: + logger.error("webapp: failed to bind {}:{} - {}", self.host, self.port, exc) + self._server = None + return + self._thread = threading.Thread( + target=self._server.serve_forever, + name="GladosWebappServer", + daemon=True, + ) + self._thread.start() + logger.success("Webapp console live: http://{}:{}/", self.host, self.bound_port) + + def shutdown(self) -> None: + """Stop serving, close the socket, and join the server thread.""" + server = self._server + if server is not None: + server.shutdown() + server.server_close() + self._server = None + if self._thread is not None and self._thread.is_alive(): + self._thread.join(timeout=2.0) + self._thread = None + self.bound_port = None + + @property + def url(self) -> str: + """Return the effective console URL, including an ephemeral port.""" + port = self.bound_port or self.port + return f"http://{self.host}:{port}/" + + @property + def is_running(self) -> bool: + """Report whether a server socket is currently active.""" + return self._server is not None + + +__all__ = ["WebappServer"] diff --git a/src/glados/webapp/static/index.html b/src/glados/webapp/static/index.html new file mode 100644 index 00000000..041ec7ae --- /dev/null +++ b/src/glados/webapp/static/index.html @@ -0,0 +1,706 @@ + + + + + +GLaDOS Core Console + + + + + +
+
+
+
GLaDOS
CORE CONSOLE · OBSERVATION DECK
+
+ UP 03:12:48 + --:--:-- + SYSTEMS NOMINAL +
+ + + +
+ + +
+
+

Mission Control

+ watch the brain think, in parallel. +
autonomy pool 2/4in-flight 0
+
+
+
0
Slots tracked
+
0
Active minds
+
0
Inferences in-flight
+
+
+
+

Dual Lane Orchestration

+
+
+
+ +
Priority Lane
direct user agent · 1 lane
+
0 queued
+
+
+
+
+
+ +
Autonomy Lane
pooled brains · 2 workers
+
0 in-flight
+
+
+
+
+
+
+

Brainstem PAD + interaction

+
+
Pleasure
0.0
+
Arousal
0.0
+
Dominance
0.0
+
+
+
●Last user ----
+
●Last spoke ----
+
+
+
+
+

The Wire · live stream

+
+
+
+ + +
+

Neural Core

all parallel inferences, in-flight and queued.
--
+
+
+
Priority Lane
the direct user agent · always first
--
+
+
+
+
Autonomy Lane
pool of Minds · up to 12 workers
--
+
+
+
+
+

Worker pool autonomy_parallel_calls = 2

+
+
+
+ + +
+

Minds

independent subagents, each with a private context.
+
+
+ + +
+

Slots

shared state the main agent consumes.
shared
+
+
+ + +
+

The Wire

observability bus stream.
+
+
+
+ + +
+

Vitals

everything else the TUI tracks.
+
+

Audio

+

Interaction

+

Vision

+

MCP servers

+
+
+
+
+ + + + + + diff --git a/tests/test_webapp.py b/tests/test_webapp.py new file mode 100644 index 00000000..50f25eb9 --- /dev/null +++ b/tests/test_webapp.py @@ -0,0 +1,462 @@ +"""Tests for the in-process webapp observability console.""" + +import http.client +import json +from pathlib import Path +import time +from types import SimpleNamespace + +import pytest + +from glados.observability import ObservabilityBus, ObservabilityEvent +from glados.webapp.serializers import ( + build_snapshot, + build_state, + serialize_event, +) +from glados.webapp.server import WebappServer + +# --------------------------------------------------------------------------- bus fan-out + + +class TestObservabilityBusFanOut: + def test_subscribe_receives_every_published_event(self) -> None: + bus = ObservabilityBus() + sub = bus.subscribe() + try: + for i in range(5): + bus.emit("autonomy", "tick", f"event {i}", meta={"i": i}) + got = [sub.get_nowait().message for _ in range(5)] + assert got == [f"event {i}" for i in range(5)] + finally: + bus.unsubscribe(sub) + + def test_drain_still_works_independently_of_subscribers(self) -> None: + bus = ObservabilityBus() + sub = bus.subscribe() + try: + bus.emit("llm", "queue", "to drain caller", level="debg") + # The single-consumer drain() queue must still see the event + # (mirrors the TUI ObservabilityScreen). + drained = bus.drain(max_items=10) + assert [e.message for e in drained] == ["to drain caller"] + # And the subscriber got its own independent copy. + assert sub.get_nowait().message == "to drain caller" + finally: + bus.unsubscribe(sub) + + def test_unsubscribe_stops_delivery(self) -> None: + bus = ObservabilityBus() + sub = bus.subscribe() + bus.emit("engine", "start", "a") + bus.unsubscribe(sub) + bus.emit("engine", "start", "b") + # subscriber keeps only the first event + assert sub.get_nowait().message == "a" + assert sub.empty() + + def test_slow_subscriber_keeps_newest_events(self) -> None: + bus = ObservabilityBus(subscriber_max=2) + sub = bus.subscribe() + try: + for i in range(4): + bus.emit("engine", "tick", str(i)) + assert [sub.get_nowait().message for _ in range(2)] == ["2", "3"] + finally: + bus.unsubscribe(sub) + + +# --------------------------------------------------------------------------- serializers +# Stub engine carrying just the accessors the serializers touch. + + +class _FakeInteraction: + def seconds_since_user(self) -> int: + return 12 + + def seconds_since_assistant(self) -> int: + return 3 + + +class _FakeAudioState: + def snapshot(self) -> SimpleNamespace: + return SimpleNamespace(rms=-30.5, vad_active=True) + + +class _FakeMindRegistry: + def snapshot(self) -> list: + return [ + SimpleNamespace( + mind_id="m1", + title="Forecast Mind", + status="running", + summary="watching the sky", + role="weather-summarizer", + updated_at=time.time(), + ) + ] + + +class _FakeEngine: + def __init__(self) -> None: + self.autonomy_config = SimpleNamespace(enabled=True) + self.llm_queue_priority = _FakeQueue(1) + self.llm_queue_autonomy = _FakeQueue(2) + self._autonomy_inflight = SimpleNamespace(value=lambda: 3) + self.autonomy_llm_processors = [None, None] + self.audio_state = _FakeAudioState() + self._emotion_agent = None + self.mcp_manager = None + self.autonomy_slots = None + self.subagent_manager = None + self.vision_state = None + self.mind_registry = _FakeMindRegistry() + self.interaction_state = _FakeInteraction() + self.currently_speaking_event = SimpleNamespace(is_set=lambda: False) + self.shutdown_event = SimpleNamespace(is_set=lambda: False) + self.observability_bus = ObservabilityBus() + self._command_order = ["/mcp", "/tts"] + + +class _FakeQueue: + def __init__(self, size: int) -> None: + self._size = size + + def qsize(self) -> int: + return self._size + + +def test_serialize_event() -> None: + ev = type( + "E", + (), + {"timestamp": 1, "source": "llm", "kind": "queue", "level": "info", "message": "hi", "meta": {"slot": "s1"}}, + )() + out = serialize_event(ev) + assert out["source"] == "llm" + assert out["meta"]["slot"] == "s1" + + +def test_build_snapshot_and_state() -> None: + engine = _FakeEngine() + snap = build_snapshot(engine) + assert snap["lanes"]["priority"]["queue"] == 1 + assert snap["lanes"]["autonomy"]["inflight"] == 3 + assert snap["minds"][0]["mind_id"] == "m1" + assert snap["speaking"] is False + + state = build_state(engine) + assert state["lanes"]["autonomy"]["inflight"] == 3 + assert state["interaction"]["seconds_since_user"] == 12 + + +# --------------------------------------------------------------------------- HTTP server + + +def test_webapp_server_serves_snapshot_and_stream() -> None: + engine = _FakeEngine() + engine.observability_bus.emit("autonomy", "tick", "hello stream", meta={"x": 1}) + server = WebappServer(engine, host="127.0.0.1", port=0) + server.start() + try: + assert server.is_running + port = server.bound_port + assert port + + conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5) + conn.request("GET", "/api/snapshot") + resp = conn.getresponse() + assert resp.status == 200 + payload = json.loads(resp.read()) + assert "minds" in payload + conn.close() + + # static console + conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5) + conn.request("GET", "/") + resp = conn.getresponse() + assert resp.status == 200 + body = resp.read().decode("utf-8") + assert "GLaDOS Core Console" in body + conn.close() + + # SSE stream replays history then pushes state pings + conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5) + conn.request("GET", "/api/stream") + resp = conn.getresponse() + assert resp.status == 200 + assert resp.getheader("Content-Type", "").startswith("text/event-stream") + head = "" + for _ in range(10): + head += resp.readline().decode("utf-8", "replace") + if "hello stream" in head: + break + assert "hello stream" in head + assert "event:" in head + conn.close() + finally: + server.shutdown() + + +class _SnapshotPublishingBus(ObservabilityBus): + """Publish once immediately after copying history to expose replay gaps.""" + + def __init__(self) -> None: + super().__init__() + self._published_during_snapshot = False + + def snapshot(self, limit: int | None = None) -> list[ObservabilityEvent]: + events = super().snapshot(limit) + if not self._published_during_snapshot: + self._published_during_snapshot = True + self.emit("engine", "race", "between replay and live") + return events + + +def test_stream_does_not_lose_event_at_replay_boundary() -> None: + engine = _FakeEngine() + engine.observability_bus = _SnapshotPublishingBus() + server = WebappServer(engine, host="127.0.0.1", port=0) + server.start() + try: + conn = http.client.HTTPConnection("127.0.0.1", server.bound_port, timeout=5) + conn.request("GET", "/api/stream") + resp = conn.getresponse() + body = "" + for _ in range(10): + body += resp.readline().decode("utf-8", "replace") + if "between replay and live" in body: + break + assert "between replay and live" in body + conn.close() + finally: + server.shutdown() + + +def test_server_rejects_cross_origin_browser_request() -> None: + engine = _FakeEngine() + server = WebappServer(engine, host="127.0.0.1", port=0) + server.start() + try: + conn = http.client.HTTPConnection("127.0.0.1", server.bound_port, timeout=5) + conn.request("GET", "/api/snapshot", headers={"Origin": "https://example.test"}) + resp = conn.getresponse() + assert resp.status == 403 + assert resp.getheader("Access-Control-Allow-Origin") is None + conn.close() + + # A forged host without Origin models a DNS-rebinding request. + conn = http.client.HTTPConnection("127.0.0.1", server.bound_port, timeout=5) + conn.request("GET", "/api/snapshot", headers={"Host": "example.test"}) + resp = conn.getresponse() + assert resp.status == 403 + conn.close() + finally: + server.shutdown() + + +def test_server_refuses_non_loopback_bind() -> None: + with pytest.raises(ValueError, match="loopback-only"): + WebappServer(_FakeEngine(), host="0.0.0.0", port=8050) # noqa: S104 + + +# --------------------------------------------------------------------------- env override + + +def test_webapp_env_override_enables_disabled_config(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + """GLADOS_WEBAPP_* env vars enable the console even when YAML disables it.""" + from glados.core.engine import GladosConfig + + cfg = tmp_path / "wp_disabled.yaml" + cfg.write_text( + "Glados:\n" + " llm_model: llama\n" + " completion_url: http://localhost:11434/api/chat\n" + " api_key: null\n" + " interruptible: true\n" + " audio_io: sounddevice\n" + " input_mode: text\n" + " asr_engine: tdt\n" + " wake_word: null\n" + " voice: glados\n" + " announcement: null\n" + " webapp:\n" + " enabled: false\n" + " host: 127.0.0.1\n" + " port: 8050\n" + " personality_preprompt:\n" + " - system: you are a test\n" + ) + monkeypatch.setenv("GLADOS_WEBAPP_ENABLED", "1") + monkeypatch.setenv("GLADOS_WEBAPP_PORT", "8085") + + resolved = GladosConfig.from_yaml(cfg) + assert resolved.webapp is not None + assert resolved.webapp.enabled is True + assert resolved.webapp.port == 8085 + + +def test_webapp_env_absent_keeps_yaml_disabled(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + """With no env flag, a disabled-by-YAML webapp stays disabled.""" + from glados.core.engine import GladosConfig + + cfg = tmp_path / "wp_disabled.yaml" + cfg.write_text( + "Glados:\n" + " llm_model: llama\n" + " completion_url: http://localhost:11434/api/chat\n" + " api_key: null\n" + " interruptible: true\n" + " audio_io: sounddevice\n" + " input_mode: text\n" + " asr_engine: tdt\n" + " wake_word: null\n" + " voice: glados\n" + " announcement: null\n" + " webapp:\n" + " enabled: false\n" + " personality_preprompt: []\n" + ) + for name in ("GLADOS_WEBAPP_ENABLED", "GLADOS_WEBAPP_HOST", "GLADOS_WEBAPP_PORT"): + monkeypatch.delenv(name, raising=False) + resolved = GladosConfig.from_yaml(cfg) + assert resolved.webapp is None or resolved.webapp.enabled is False + + +def test_webapp_port_env_preserves_yaml_enabled_state(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + """A port-only override must not unexpectedly enable the console.""" + from glados.core.engine import GladosConfig + + cfg = tmp_path / "wp_port.yaml" + cfg.write_text( + "Glados:\n" + " llm_model: llama\n" + " completion_url: http://localhost:11434/api/chat\n" + " api_key: null\n" + " interruptible: true\n" + " audio_io: sounddevice\n" + " input_mode: text\n" + " asr_engine: tdt\n" + " wake_word: null\n" + " voice: glados\n" + " announcement: null\n" + " webapp:\n" + " enabled: false\n" + " personality_preprompt: []\n" + ) + monkeypatch.delenv("GLADOS_WEBAPP_ENABLED", raising=False) + monkeypatch.setenv("GLADOS_WEBAPP_PORT", "8099") + resolved = GladosConfig.from_yaml(cfg) + assert resolved.webapp is not None + assert resolved.webapp.enabled is False + assert resolved.webapp.port == 8099 + + +# --------------------------------------------------------------------------- webapp launcher + + +class _LauncherEngine: + """Minimal stand-in for the real engine in launcher tests.""" + + def __init__(self) -> None: + self.announcement = None + self.ran = False + self.fail_announcement = False + + def play_announcement(self) -> None: + """Play the fake announcement or raise for lifecycle testing.""" + if self.fail_announcement: + raise RuntimeError("announcement failed") + + def run(self) -> None: + self.ran = True + + +class _FakeServer: + """Stub WebappServer that records start/stop without binding a socket.""" + + def __init__(self, engine: object, host: str, port: int) -> None: + self.engine = engine + self.host = host + self.port = port + self.is_running = False + self.shutdown_called = False + + def start(self) -> None: + self.is_running = True + + def shutdown(self) -> None: + self.shutdown_called = True + self.is_running = False + + +def _fake_glados_config(enabled: bool) -> SimpleNamespace: + from glados.webapp import WebappConfig + + return SimpleNamespace(webapp=WebappConfig(enabled=enabled, host="127.0.0.1", port=0)) + + +def test_run_webapp_refuses_when_disabled(monkeypatch: pytest.MonkeyPatch) -> None: + """A disabled webapp aborts the launcher without building the engine.""" + from glados import cli + + monkeypatch.setattr(cli.GladosConfig, "from_yaml", lambda *a, **k: _fake_glados_config(enabled=False)) + monkeypatch.setattr( + cli.Glados, + "from_config", + lambda c: (_ for _ in ()).throw(AssertionError("engine must not be built")), + ) + + with pytest.raises(SystemExit) as exc: + cli.run_webapp("x.yaml") + assert exc.value.code == 1 + + +def test_run_webapp_starts_server_then_shuts_down(monkeypatch: pytest.MonkeyPatch) -> None: + """The launcher builds the engine, starts its server, runs, then shuts down.""" + from glados import cli + + engine = _LauncherEngine() + created: list[_FakeServer] = [] + + def _server_factory(engine: object, host: str, port: int) -> _FakeServer: + server = _FakeServer(engine, host, port) + created.append(server) + return server + + monkeypatch.setattr(cli.GladosConfig, "from_yaml", lambda *a, **k: _fake_glados_config(enabled=True)) + monkeypatch.setattr(cli.Glados, "from_config", lambda c: engine) + monkeypatch.setattr(cli, "WebappServer", _server_factory) + + cli.run_webapp("x.yaml") + + assert engine.ran is True + assert len(created) == 1 + server = created[0] + assert server.port == 0 + assert server.shutdown_called is True + assert server.is_running is False + + +def test_run_webapp_shuts_down_when_announcement_fails(monkeypatch: pytest.MonkeyPatch) -> None: + """The HTTP socket is closed even if startup audio raises.""" + from glados import cli + + engine = _LauncherEngine() + engine.announcement = "hello" + engine.fail_announcement = True + created: list[_FakeServer] = [] + + def _server_factory(engine: object, host: str, port: int) -> _FakeServer: + server = _FakeServer(engine, host, port) + created.append(server) + return server + + monkeypatch.setattr(cli.GladosConfig, "from_yaml", lambda *a, **k: _fake_glados_config(enabled=True)) + monkeypatch.setattr(cli.Glados, "from_config", lambda config: engine) + monkeypatch.setattr(cli, "WebappServer", _server_factory) + + with pytest.raises(RuntimeError, match="announcement failed"): + cli.run_webapp("x.yaml") + + assert created[0].shutdown_called is True