diff --git a/agent/agent.py b/agent/agent.py index 8a8573c..8c04bdf 100644 --- a/agent/agent.py +++ b/agent/agent.py @@ -202,7 +202,11 @@ SYSTEM_PROMPT = textwrap.dedent("""\ After calling dispatch_task, tell the user "I'll get on that for you" without asking for more permission or details. The result comes back automatically — when you receive it, share the findings naturally in your conversational style. - You can have multiple tasks running at once. + You can have multiple tasks running at once. If the user asks about a + task's progress or you want to check one yourself, call task_status with + its id (it reports pending/running/completed/failed and whether it looks + stalled); task_status can also be called to check up on a task. Use + list_tasks to see every task you have dispatched. # Music (media player in the web UI) You can play music on the user's web player. For any music request, call diff --git a/agent/dispatch_mcp.py b/agent/dispatch_mcp.py index 4687378..2b06377 100644 --- a/agent/dispatch_mcp.py +++ b/agent/dispatch_mcp.py @@ -12,6 +12,7 @@ from __future__ import annotations import asyncio import logging import os +import time from mcp.server.fastmcp import FastMCP @@ -64,5 +65,38 @@ async def _run_worker(worker: "TaskWorker", task_id: str): registry.fail(task_id, str(e)) +def _status_summary(task_id: str, stalled_s: float = 45.0) -> str: + """Human-readable live status for one task, plus a stall hint.""" + try: + from agent.task_registry import status_summary + except ImportError: + from task_registry import status_summary + return status_summary(task_id, stalled_s) + + +@mcp.tool() +async def task_status(task_id: str) -> str: + """Check the live status of a background task you dispatched earlier, + including whether it is pending, running, completed, failed, or looking + stalled (no activity for a while). Pass the task id from dispatch_task + (or from list_tasks).""" + return _status_summary(task_id) + + +@mcp.tool() +async def list_tasks() -> str: + """List every background task Hope has dispatched in this conversation, + with their live status so you can report progress to the user.""" + try: + from agent.task_registry import registry + except ImportError: + from task_registry import registry + + tasks = registry.list_all() + if not tasks: + return "No background tasks have been dispatched yet." + return "\n".join(_status_summary(t["id"]) for t in tasks) + + if __name__ == "__main__": mcp.run(transport="stdio") diff --git a/agent/task_registry.py b/agent/task_registry.py index db8211a..438c9c7 100644 --- a/agent/task_registry.py +++ b/agent/task_registry.py @@ -48,6 +48,7 @@ class Task: error: str | None = None created_at: float = field(default_factory=time.time) completed_at: float | None = None + last_activity: float = field(default_factory=time.time) class TaskRegistry: @@ -78,6 +79,7 @@ class TaskRegistry: task = self._tasks.get(task_id) if task: task.steps.append(TaskStep(role=role, content=content)) + task.last_activity = time.time() if task.status == TaskStatus.PENDING: task.status = TaskStatus.RUNNING self._save_task(task) @@ -163,8 +165,37 @@ class TaskRegistry: "error": task.error, "created_at": task.created_at, "completed_at": task.completed_at, + "last_activity": task.last_activity, } # Global singleton — one per process registry = TaskRegistry() + + +def status_summary(task_id: str, stalled_s: float = 45.0) -> str: + """Human-readable live status for one task, plus a stall hint. + + Kept on the registry module (no MCP dependency) so it can be reused and + unit-tested; the dispatch MCP's task_status/list_tasks tools call this. + """ + t = registry.get(task_id) + if not t: + return f"Task {task_id} was not found." + status = t.get("status", "unknown") + steps = t.get("steps") or [] + created = t.get("created_at") or 0 + last = t.get("last_activity") or created + elapsed = time.time() - created + idle = time.time() - last + parts = [f"Task {task_id} ({t.get('description', '')[:60]}): {status.upper()}"] + parts.append(f"elapsed {elapsed:,.0f}s, {len(steps)} step(s)") + if idle > 0.5: + parts.append(f"last activity {idle:,.0f}s ago") + if status in ("pending", "running") and idle > stalled_s: + parts.append(f"STALLED (no activity for {idle:,.0f}s)") + if t.get("error"): + parts.append(f"error: {t['error']}") + if t.get("result"): + parts.append(f"result: {t['result'][:240]}") + return ". ".join(parts) diff --git a/agent/task_worker.py b/agent/task_worker.py index 3c3f377..c592edb 100644 --- a/agent/task_worker.py +++ b/agent/task_worker.py @@ -16,6 +16,35 @@ from datetime import datetime, timezone logger = logging.getLogger("voice-agent.tasks") +def _html_to_text(html: str) -> str: + """Strip a raw HTML page to readable plain text (no external deps).""" + if not html: + return "" + # Drop non-content blocks first. + html = re.sub( + r"(?is)<(script|style|noscript|head|title|svg|nav|footer|form|iframe)[^>]*>.*?", + " ", + html, + ) + # Turn block-level tags into paragraph breaks. + html = re.sub( + r"(?i)]*>", + "\n", + html, + ) + text = re.sub(r"<[^>]+>", " ", html) + for ent, ch in ( + (" ", " "), ("&", "&"), ("<", "<"), (">", ">"), + (""", '"'), ("'", "'"), ("'", "'"), ("'", "'"), + ("…", "..."), ("—", "-"), ("–", "-"), + ): + text = text.replace(ent, ch) + text = re.sub(r"[ \t]+", " ", text) + text = re.sub(r"\n\s*\n+", "\n", text) + return text.strip() + + class TaskWorker: """Runs a background research/lookup task using the LLM + tools.""" @@ -289,6 +318,36 @@ class TaskWorker: }, ] + async def _fetch_page_text(self, url: str, timeout: float = 20.0) -> str: + """Fetch a page directly from the container and reduce it to plain text. + + Used as a fallback when Firecrawl's scrape times out or returns nothing. + The container has outbound internet (the self-hosted Firecrawl does not + reliably, and its 30s default scrape limit is too short for news sites). + """ + if not url: + return "No URL provided." + import httpx + + try: + async with httpx.AsyncClient( + timeout=timeout, + follow_redirects=True, + headers={ + "User-Agent": ( + "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/124.0 Safari/537.36" + ) + }, + ) as client: + resp = await client.get(url) + if resp.status_code >= 400: + return f"Fetch failed (HTTP {resp.status_code})." + text = _html_to_text(resp.text) + return text[:8000] + except httpx.HTTPError as exc: + return f"Fetch failed: {exc}" + async def _execute_tool(self, name: str, args: dict) -> str: import httpx @@ -351,19 +410,24 @@ class TaskWorker: f"{firecrawl_base}/v1/search", json={"query": query, "limit": 5}, ) - if resp.status_code == 200: - data = resp.json() - items = data.get("data", []) - lines = [] - for item in items[:5]: - title = (item.get("title") or "Untitled").strip() - description = (item.get("description") or "").strip() - url = (item.get("url") or "").strip() - lines.append(f"{title}: {description}") - if url: - lines.append(f"URL: {url}") - return "\n".join(lines) or "No results found." - return f"Search failed (HTTP {resp.status_code})." + if resp.status_code != 200: + return f"Search failed (HTTP {resp.status_code})." + data = resp.json() + items = data.get("data", []) + if not items: + return "No results found for that query." + # Search results only carry titles/descriptions. Fetch the + # top pages ourselves (container has outbound internet) so the + # model gets real content instead of just snippets — small + # models won't reliably chain a second scrape call. + blocks = [] + for item in items[:3]: + title = (item.get("title") or "Untitled").strip() + url = (item.get("url") or "").strip() + desc = (item.get("description") or "").strip() + body = await self._fetch_page_text(url) if url else (desc or "No URL.") + blocks.append(f"{title}\n{url}\n{body[:1800]}") + return "\n\n---\n\n".join(blocks) or "No results found." elif name == "web_scrape": url = args.get("url", "") @@ -372,22 +436,29 @@ class TaskWorker: firecrawl_base = os.environ.get("FIRECRAWL_BASE", "") if not firecrawl_base: return "Web scraping is not configured." - async with httpx.AsyncClient(timeout=90) as client: - resp = await client.post( - f"{firecrawl_base}/v1/scrape", - json={"url": url, "formats": ["markdown"], "timeout": 60000}, - ) - if resp.status_code != 200: - return f"Scrape failed (HTTP {resp.status_code})." - data = resp.json() - if not data.get("success"): - return f"Scrape failed: {data.get('error', 'unknown error')}" - result = data.get("data", {}) - markdown = (result.get("markdown") or "").strip() - if not markdown: - return "No page content was extracted." - title = (result.get("metadata", {}).get("title") or "").strip() - return f"Page: {title}\n\n{markdown[:8000]}" + # Firecrawl renders JS-heavy pages with a real browser, but its + # default scrape timeout (~30s) is too short for big news sites. + # Ask for 90s and give the HTTP client the same headroom; if it + # returns nothing, fall back to a direct fetch from the container. + try: + async with httpx.AsyncClient(timeout=120) as client: + resp = await client.post( + f"{firecrawl_base}/v1/scrape", + json={"url": url, "formats": ["markdown"], "timeout": 90000}, + ) + if resp.status_code == 200: + data = resp.json() + if data.get("success"): + result = data.get("data", {}) + markdown = (result.get("markdown") or "").strip() + if markdown: + title = (result.get("metadata", {}).get("title") or url).strip() + return f"Page: {title}\n\n{markdown[:8000]}" + # fall through to direct fetch on empty/failed scrape + except (httpx.HTTPError, Exception): + pass + body = await self._fetch_page_text(url) + return body or f"Could not extract content from {url}." else: return f"Unknown tool: {name}" diff --git a/tests/test_background_tasks.py b/tests/test_background_tasks.py index b790d6a..dae1514 100644 --- a/tests/test_background_tasks.py +++ b/tests/test_background_tasks.py @@ -88,6 +88,53 @@ def test_task_event_directories_are_isolated(tmp_path): assert second_events == [] +def test_task_summary_reports_live_status(tmp_path): + """task_status/list_tasks summarise live state including a stall hint.""" + import time as _time + + sys.path.insert(0, str(Path(__file__).parents[1] / "agent")) + from task_registry import status_summary # noqa: E402 + import task_registry as tr_mod # noqa: E402 + + # Isolate: status_summary reads the module-level registry. + held = tr_mod.registry + iso = tr_mod.TaskRegistry(tmp_path) + tr_mod.registry = iso + try: + task = iso.create("test task") + task_id = task.id + iso.add_step(task_id, "thinking", "computing") + summary = status_summary(task_id, stalled_s=45.0) + assert "RUNNING" in summary, summary + assert "step(s)" in summary, summary + + # Stalled: force last activity far in the past. + task.last_activity = _time.time() - 100 + iso._save_task(task) + assert "STALLED" in status_summary(task_id, stalled_s=45.0) + + # Completed with result. + iso.complete(task_id, "the answer") + done = status_summary(task_id, stalled_s=45.0) + assert "COMPLETED" in done, done + assert "the answer" in done, done + + # Unknown id. + assert "was not found" in status_summary("nope", stalled_s=45.0) + finally: + tr_mod.registry = held + + +def test_registry_tracks_last_activity(tmp_path): + reg = TaskRegistry(tmp_path) + task = reg.create("tracked") + before = task.last_activity + task.last_activity = before - 10 + reg.add_step(task.id, "thinking", "step") + data = reg.get(task.id) + assert data["last_activity"] > before - 10, data["last_activity"] + + @pytest.mark.slow def test_dispatch_mcp_task_reaches_terminal_event(): """The MCP request returns immediately while its child task stays alive."""