feat: background task dispatch system with live UI panel
- agent/task_registry.py: file-based JSONL event registry (cross-process) - agent/task_worker.py: autonomous LLM loop with weather/time/memory/web tools - agent/dispatch_mcp.py: MCP tool exposing dispatch_task to the main agent - agent/agent.py: registers dispatch toolset, polls task events → room data - web: slide-out task panel (FAB button + badge), live step streaming via data channel topic 'tasks', status dots (running/completed/failed) - Dockerfile: copies new task_*.py and dispatch_mcp.py files The dispatch MCP runs in its own process; events flow through /tmp/tasks/events.jsonl which the main agent tails every second and forwards to the browser. Tasks run up to 10 LLM iterations with tool calls.
This commit is contained in:
@@ -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,
|
||||
|
||||
@@ -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))
|
||||
+75
-37
@@ -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()
|
||||
|
||||
+14
-11
@@ -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 (
|
||||
|
||||
Reference in New Issue
Block a user