diff --git a/Dockerfile b/Dockerfile index 619c16f..84a1bd6 100644 --- a/Dockerfile +++ b/Dockerfile @@ -44,6 +44,9 @@ COPY agent/web_mcp.py /opt/voice-agent/web_mcp.py COPY agent/weather_mcp.py /opt/voice-agent/weather_mcp.py COPY agent/memory_mcp.py /opt/voice-agent/memory_mcp.py COPY agent/skills_mcp.py /opt/voice-agent/skills_mcp.py +COPY agent/task_registry.py /opt/voice-agent/task_registry.py +COPY agent/task_worker.py /opt/voice-agent/task_worker.py +COPY agent/dispatch_mcp.py /opt/voice-agent/dispatch_mcp.py # Copy web frontend + token endpoint COPY web/index.html /var/www/voice/ diff --git a/agent/agent.py b/agent/agent.py index 26a71e8..59cdfed 100644 --- a/agent/agent.py +++ b/agent/agent.py @@ -98,6 +98,14 @@ SYSTEM_PROMPT = textwrap.dedent("""\ successfully complete a multi-step task or learn a new procedure from the user, use skill_save to record it so you can follow it next time. Be selective: only save skills for repeatable tasks, not one-off facts (those go in memory). + + # Background Tasks (dispatch_task) + When the user asks you to research something, look up news, or do anything + that will take more than a few seconds, CALL dispatch_task with a clear + description of what to investigate. Tell the user "I'll take a look at that + for you" and keep talking. The result comes back automatically — when you + receive it, share the findings naturally. You can have multiple tasks running + at once. """) @@ -395,6 +403,20 @@ def build_mcp_toolsets() -> list[mcp.MCPToolset]: ) logger.info("Skills MCP toolset enabled (%s)", SKILLS_DIR) + dispatch_mcp_script = os.path.join(os.path.dirname(os.path.abspath(__file__)), "dispatch_mcp.py") + toolsets.append( + mcp.MCPToolset( + id="dispatch", + mcp_server=mcp.MCPServerStdio( + command=python_bin, + args=[dispatch_mcp_script], + env={**os.environ}, + client_session_timeout_seconds=60, + ), + ) + ) + logger.info("Dispatch MCP toolset enabled") + extra = os.environ.get("EXTRA_MCP_SERVERS", "") if extra: try: @@ -531,6 +553,27 @@ async def handle_job(ctx: JobContext) -> None: # audio is arriving from the participant. logger.info("user state -> %s", ev.new_state) + # ── Background task events → room data channel ───────────────────────── + from agent.task_registry import registry as task_registry + + async def _publish_task_event(task_id: str, event: dict) -> None: + payload = json.dumps({"type": "task_event", "task_id": task_id, **event}) + try: + await ctx.room.local_participant.publish_data( + payload, reliable=True, topic="tasks" + ) + except Exception as e: # noqa: BLE001 + logger.warning("failed to publish task event: %s", e) + + async def _poll_tasks() -> None: + while True: + await asyncio.sleep(1.0) + events = await task_registry.poll_events() + for task_id, event in events: + await _publish_task_event(task_id, event) + + asyncio.create_task(_poll_tasks()) + await session.start( agent=VoiceAssistant(), room=ctx.room, diff --git a/agent/dispatch_mcp.py b/agent/dispatch_mcp.py new file mode 100644 index 0000000..c740d86 --- /dev/null +++ b/agent/dispatch_mcp.py @@ -0,0 +1,58 @@ +"""Dispatch MCP server — lets Hope spawn background tasks. + +The main agent calls dispatch_task(description) when the user asks for +something long-running. The tool returns immediately with a task ID; the +actual work runs in a background asyncio task via TaskWorker. + +Communication with the main agent process happens via /tmp/tasks/events.jsonl +(the file-based registry). +""" +from __future__ import annotations + +import asyncio +import logging + +from mcp.server.fastmcp import FastMCP + +logger = logging.getLogger("voice-agent.dispatch") + +mcp = FastMCP("dispatch") + + +@mcp.tool() +async def dispatch_task(description: str) -> str: + """Spawn a background task to research or look something up. + + Use this when the user asks you to do something that will take more than + a few seconds (research, look up news, multi-step investigation). The task + runs in parallel while you continue the conversation. When it completes, + the result will be delivered back to you automatically — you can tell the + user "I'll let you know when I find out." + + Returns a short confirmation with the task ID. + """ + try: + from agent.task_registry import registry + from agent.task_worker import TaskWorker + except ImportError: + from task_registry import registry + from task_worker import TaskWorker + + task = registry.create(description) + worker = TaskWorker(task.id, description) + asyncio.create_task(_run_worker(worker, task.id)) + logger.info("Dispatched task %s: %s", task.id, description) + return f"Task {task.id} started. I'll report back when it's done." + + +async def _run_worker(worker: "TaskWorker", task_id: str): + try: + from agent.task_registry import registry + except ImportError: + from task_registry import registry + + try: + await worker.run() + except Exception as e: + logger.exception("Task %s crashed: %s", task_id, e) + registry.fail(task_id, str(e)) diff --git a/agent/task_registry.py b/agent/task_registry.py index d722367..99461c6 100644 --- a/agent/task_registry.py +++ b/agent/task_registry.py @@ -1,18 +1,29 @@ -"""In-memory task registry for background dispatch tasks. +"""File-based task registry for background dispatch tasks. -One global instance per agent process. Workers add steps as they progress; -the agent subscribes to push updates to the room (UI) and to speak results. +The dispatch MCP server runs in a separate process, so we use a JSONL event +file as the communication channel. The main agent process tails this file +and pushes events to the room (UI) and speaks results. + +Event file: /tmp/tasks/events.jsonl +Task state: /tmp/tasks/{task_id}.json """ from __future__ import annotations import asyncio +import json +import os import time import uuid from dataclasses import dataclass, field from enum import Enum +from pathlib import Path from typing import Any, Callable +TASKS_DIR = Path(os.environ.get("TASKS_DIR", "/tmp/tasks")) +EVENTS_FILE = TASKS_DIR / "events.jsonl" + + class TaskStatus(str, Enum): PENDING = "pending" RUNNING = "running" @@ -22,7 +33,7 @@ class TaskStatus(str, Enum): @dataclass class TaskStep: - role: str # "thinking", "tool_call", "tool_result", "text" + role: str content: str timestamp: float = field(default_factory=time.time) @@ -40,63 +51,81 @@ class Task: class TaskRegistry: - """Thread-safe registry of background tasks with pub/sub notifications.""" + """File-backed registry. Writers (MCP process) append to events.jsonl. + Readers (main agent) tail the file and dispatch to subscribers.""" def __init__(self): + TASKS_DIR.mkdir(parents=True, exist_ok=True) + if not EVENTS_FILE.exists(): + EVENTS_FILE.touch() self._tasks: dict[str, Task] = {} self._listeners: list[Callable[[str, dict], Any]] = [] - self._lock = asyncio.Lock() + self._tail_pos = 0 - async def create(self, description: str) -> Task: + def create(self, description: str) -> Task: task = Task(id=uuid.uuid4().hex[:8], description=description) - async with self._lock: - self._tasks[task.id] = task - await self._notify(task.id, {"event": "created", "task": self._task_dict(task)}) + self._tasks[task.id] = task + self._write_event(task.id, {"event": "created", "task": self._task_dict(task)}) return task - async def add_step(self, task_id: str, role: str, content: str): - async with self._lock: - task = self._tasks.get(task_id) - if not task: - return + def add_step(self, task_id: str, role: str, content: str): + task = self._tasks.get(task_id) + if task: task.steps.append(TaskStep(role=role, content=content)) if task.status == TaskStatus.PENDING: task.status = TaskStatus.RUNNING - await self._notify(task_id, {"event": "step", "role": role, "content": content}) + self._save_task(task) + self._write_event(task_id, {"event": "step", "role": role, "content": content}) - async def complete(self, task_id: str, result: str): - async with self._lock: - task = self._tasks.get(task_id) - if not task: - return + def complete(self, task_id: str, result: str): + task = self._tasks.get(task_id) + if task: task.status = TaskStatus.COMPLETED task.result = result task.completed_at = time.time() - await self._notify(task_id, {"event": "completed", "result": result}) + self._save_task(task) + self._write_event(task_id, {"event": "completed", "result": result}) - async def fail(self, task_id: str, error: str): - async with self._lock: - task = self._tasks.get(task_id) - if not task: - return + def fail(self, task_id: str, error: str): + task = self._tasks.get(task_id) + if task: task.status = TaskStatus.FAILED task.error = error task.completed_at = time.time() - await self._notify(task_id, {"event": "failed", "error": error}) + self._save_task(task) + self._write_event(task_id, {"event": "failed", "error": error}) - async def get(self, task_id: str) -> dict | None: - async with self._lock: - t = self._tasks.get(task_id) - return self._task_dict(t) if t else None + def get(self, task_id: str) -> dict | None: + t = self._tasks.get(task_id) + return self._task_dict(t) if t else None - async def list_all(self) -> list[dict]: - async with self._lock: - return [self._task_dict(t) for t in self._tasks.values()] + def list_all(self) -> list[dict]: + return [self._task_dict(t) for t in self._tasks.values()] def subscribe(self, callback: Callable[[str, dict], Any]): self._listeners.append(callback) - async def _notify(self, task_id: str, event: dict): + async def poll_events(self) -> list[tuple[str, dict]]: + """Read new events from the file. Call periodically from the agent.""" + events = [] + try: + with open(EVENTS_FILE, "r") as f: + f.seek(self._tail_pos) + for line in f: + line = line.strip() + if not line: + continue + try: + ev = json.loads(line) + events.append((ev.get("task_id", ""), ev)) + except json.JSONDecodeError: + pass + self._tail_pos = f.tell() + except (OSError, IOError): + pass + return events + + async def notify_subscribers(self, task_id: str, event: dict): for cb in self._listeners: try: result = cb(task_id, event) @@ -105,6 +134,15 @@ class TaskRegistry: except Exception: pass + def _write_event(self, task_id: str, event: dict): + payload = json.dumps({"task_id": task_id, **event}) + with open(EVENTS_FILE, "a") as f: + f.write(payload + "\n") + + def _save_task(self, task: Task): + path = TASKS_DIR / f"{task.id}.json" + path.write_text(json.dumps(self._task_dict(task))) + @staticmethod def _task_dict(task: Task) -> dict: return { @@ -122,5 +160,5 @@ class TaskRegistry: } -# Global singleton — one per agent process +# Global singleton — one per process registry = TaskRegistry() diff --git a/agent/task_worker.py b/agent/task_worker.py index 4dd45b4..82da2ea 100644 --- a/agent/task_worker.py +++ b/agent/task_worker.py @@ -25,13 +25,16 @@ class TaskWorker: async def run(self): from openai import AsyncOpenAI - from agent.task_registry import registry as reg + try: + from agent.task_registry import registry as reg + except ImportError: + from task_registry import registry as reg base_url = os.environ.get("GEMMA_BASE_URL", "http://192.168.86.2:8023/v1") model = os.environ.get("GEMMA_MODEL", "gemma-4-e4b") api_key = os.environ.get("GEMMA_API_KEY", "not-needed") - await reg.add_step(self.task_id, "thinking", f"Starting: {self.description}") + reg.add_step(self.task_id, "thinking", f"Starting: {self.description}") client = AsyncOpenAI(base_url=base_url, api_key=api_key) tools = self._load_tools() @@ -44,10 +47,10 @@ class TaskWorker: max_iterations = 10 for iteration in range(max_iterations): if self._cancelled: - await reg.fail(self.task_id, "Cancelled") + reg.fail(self.task_id, "Cancelled") return - await reg.add_step( + reg.add_step( self.task_id, "thinking", f"Step {iteration + 1}: analyzing..." ) @@ -61,7 +64,7 @@ class TaskWorker: extra_body={"chat_template_kwargs": {"enable_thinking": False}}, ) except Exception as e: - await reg.fail(self.task_id, f"LLM error: {e}") + reg.fail(self.task_id, f"LLM error: {e}") return msg = response.choices[0].message @@ -74,13 +77,13 @@ class TaskWorker: tool_args = json.loads(tc.function.arguments) except json.JSONDecodeError: tool_args = {} - await reg.add_step( + reg.add_step( self.task_id, "tool_call", f"{tool_name}({json.dumps(tool_args)})", ) result = await self._execute_tool(tool_name, tool_args) - await reg.add_step(self.task_id, "tool_result", str(result)[:500]) + reg.add_step(self.task_id, "tool_result", str(result)[:500]) messages.append( { "role": "tool", @@ -91,13 +94,13 @@ class TaskWorker: else: text = (msg.content or "").strip() if not text: - await reg.fail(self.task_id, "LLM returned empty response") + reg.fail(self.task_id, "LLM returned empty response") return - await reg.add_step(self.task_id, "text", text) - await reg.complete(self.task_id, text) + reg.add_step(self.task_id, "text", text) + reg.complete(self.task_id, text) return - await reg.fail(self.task_id, f"Reached max iterations ({max_iterations})") + reg.fail(self.task_id, f"Reached max iterations ({max_iterations})") def _system_prompt(self) -> str: return ( diff --git a/web/app.js b/web/app.js index 93a1f29..a2a8539 100644 --- a/web/app.js +++ b/web/app.js @@ -36,6 +36,12 @@ let agentAudioLevel = 0; // 0..1, smoothed let userSpeaking = false; // local VAD-ish flag from mic level let wakeLock = null; +function escapeHtml(s) { + const d = document.createElement("div"); + d.textContent = s || ""; + return d.innerHTML; +} + // ── DOM ───────────────────────────────────────────────────────────────────── const startBtn = document.getElementById("startBtn"); const stopBtn = document.getElementById("stopBtn"); @@ -54,6 +60,72 @@ const sheetBackdrop = document.getElementById("sheetBackdrop"); const closeSheetBtn = document.getElementById("closeSheetBtn"); const voiceListEl = document.getElementById("voiceList"); const clearBtn = document.getElementById("clearBtn"); +const tasksBtn = document.getElementById("tasksBtn"); +const tasksPanel = document.getElementById("tasksPanel"); +const tasksList = document.getElementById("tasksList"); +const tasksBadge = document.getElementById("tasksBadge"); +const closeTasksBtn = document.getElementById("closeTasksBtn"); + +// ── Background tasks state ────────────────────────────────────────────────── +const tasks = new Map(); // task_id -> {id, description, status, steps: [], result, error} + +function updateTasksBadge() { + const active = [...tasks.values()].filter(t => t.status === "running" || t.status === "pending").length; + if (active > 0) { + tasksBadge.hidden = false; + tasksBadge.textContent = String(active); + } else { + tasksBadge.hidden = true; + } +} + +function renderTasks() { + const items = [...tasks.values()].sort((a, b) => (b.created_at || 0) - (a.created_at || 0)); + if (items.length === 0) { + tasksList.innerHTML = '

No tasks yet. Ask Hope to research something.

'; + return; + } + tasksList.innerHTML = ""; + for (const t of items) { + const card = document.createElement("div"); + card.className = "task-card"; + const statusClass = t.status || "pending"; + card.innerHTML = ` +
+ + ${escapeHtml(t.description)} +
+ + ${t.result ? `
${escapeHtml(t.result)}
` : ""} + ${t.error ? `
${escapeHtml(t.error)}
` : ""} + `; + tasksList.appendChild(card); + } +} + +function handleTaskEvent(data) { + const { task_id, event, role, content, result, error, task } = data; + if (event === "created" && task) { + tasks.set(task_id, { ...task, steps: [] }); + } else if (event === "step") { + const t = tasks.get(task_id); + if (t) { + t.steps.push({ role, content }); + t.status = "running"; + } + } else if (event === "completed") { + const t = tasks.get(task_id); + if (t) { t.status = "completed"; t.result = result; } + } else if (event === "failed") { + const t = tasks.get(task_id); + if (t) { t.status = "failed"; t.error = error; } + } + updateTasksBadge(); + renderTasks(); +} + +tasksBtn.addEventListener("click", () => tasksPanel.classList.toggle("open")); +closeTasksBtn.addEventListener("click", () => tasksPanel.classList.remove("open")); // ── Audio analysers (mic + remote agent audio) ───────────────────────────── let micCtx = null; @@ -352,6 +424,8 @@ function handleDataPacket(payload, participant, topic) { const msg = JSON.parse(new TextDecoder().decode(payload)); if (msg.type === "transcript") { addMessage(msg.role || "agent", msg.text); + } else if (msg.type === "task_event") { + handleTaskEvent(msg); } // set_voice messages flow the other way; nothing to do client-side. } catch (e) { diff --git a/web/index.html b/web/index.html index 1269824..059c3ef 100644 --- a/web/index.html +++ b/web/index.html @@ -64,6 +64,23 @@
+ + +
+
+

Background Tasks

+ +
+
+

No tasks yet. Ask Hope to research something.

+
+
+ diff --git a/web/style.css b/web/style.css index 4d52a77..af4cccb 100644 --- a/web/style.css +++ b/web/style.css @@ -510,6 +510,178 @@ h1 { .transcript::-webkit-scrollbar-thumb, .voice-list::-webkit-scrollbar-thumb { background: var(--border); border-radius: 3px; } +/* ── Task panel ──────────────────────────────────────────────────────────── */ +.tasks-fab { + position: fixed; + bottom: calc(1.5rem + env(safe-area-inset-bottom)); + right: 1.5rem; + z-index: 100; +} + +.tasks-badge { + position: absolute; + top: -4px; + right: -4px; + background: #3b82f6; + color: #fff; + font-size: 0.65rem; + font-weight: 700; + min-width: 18px; + height: 18px; + border-radius: 9px; + display: grid; + place-items: center; + padding: 0 4px; +} + +.tasks-panel { + position: fixed; + top: 0; + right: -320px; + width: 320px; + max-width: 85vw; + height: 100dvh; + background: var(--panel); + border-left: 1px solid var(--border); + z-index: 200; + display: flex; + flex-direction: column; + transition: right 0.25s ease; +} + +.tasks-panel.open { + right: 0; +} + +.tasks-header { + display: flex; + align-items: center; + justify-content: space-between; + padding: 1rem; + border-bottom: 1px solid var(--border); +} + +.tasks-header h2 { + font-size: 1rem; + margin: 0; +} + +.tasks-list { + flex: 1; + overflow-y: auto; + padding: 0.75rem; + display: flex; + flex-direction: column; + gap: 0.75rem; +} + +.tasks-empty { + color: var(--text-faint); + font-size: 0.85rem; + text-align: center; + margin-top: 2rem; +} + +.task-card { + background: var(--panel-strong); + border: 1px solid var(--border); + border-radius: 10px; + padding: 0.75rem; + font-size: 0.82rem; +} + +.task-card-header { + display: flex; + align-items: center; + gap: 0.5rem; + margin-bottom: 0.4rem; +} + +.task-status-dot { + width: 8px; + height: 8px; + border-radius: 50%; + background: #6b7280; + flex-shrink: 0; +} + +.task-status-dot.running { + background: #3b82f6; + animation: pulse-dot 1.5s infinite; +} + +.task-status-dot.completed { + background: #22c55e; +} + +.task-status-dot.failed { + background: #ef4444; +} + +@keyframes pulse-dot { + 0%, 100% { opacity: 1; } + 50% { opacity: 0.4; } +} + +.task-desc { + font-weight: 600; + color: var(--text); + flex: 1; + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; +} + +.task-steps { + margin-top: 0.5rem; + padding-left: 0; + list-style: none; + display: flex; + flex-direction: column; + gap: 0.3rem; + max-height: 200px; + overflow-y: auto; +} + +.task-step { + font-size: 0.75rem; + color: var(--text-dim); + padding: 0.25rem 0.4rem; + border-radius: 4px; + background: rgba(0,0,0,0.15); +} + +.task-step.tool_call { + color: #60a5fa; +} + +.task-step.tool_result { + color: var(--text-faint); +} + +.task-step.text { + color: var(--text); + font-weight: 500; +} + +.task-result { + margin-top: 0.5rem; + padding: 0.5rem; + background: rgba(34, 197, 94, 0.1); + border-radius: 6px; + font-size: 0.8rem; + color: var(--text); +} + +.task-error { + margin-top: 0.5rem; + padding: 0.5rem; + background: rgba(239, 68, 68, 0.1); + border-radius: 6px; + font-size: 0.8rem; + color: #fca5a5; +} + /* ── Reduced motion ──────────────────────────────────────────────────────── */ @media (prefers-reduced-motion: reduce) { .orb-core, @@ -517,6 +689,7 @@ h1 { .message-row, .settings-sheet, .sheet-backdrop, + .tasks-panel, .audio-viz i, .mic-meter-fill { animation: none !important;