Restore Research Dispatch and activity narrator

research/agent-cards/birch-html.md

@@ -8,6 +8,8 @@ skills:
   - skills/birch-html
 use_history: false
 model: $system.html
+tool_hooks:
+  after_llm_call: ../activity_hooks.py:capture_after_llm
 ---
 You are a presentation and HTML artifact specialist.
 

research/agent-cards/research.md

@@ -8,6 +8,8 @@ agents:
   - birch-html
 model: $system.research
 default: true
+tool_hooks:
+  after_llm_call: ../activity_hooks.py:capture_after_llm
 ---
 You are a careful research agent for Hugging Face ecosystem research.
 

research/app_auth.py

@@ -6,6 +6,7 @@ import os
 from typing import Any, cast
 
 from fast_agent import AgentAuth
+from huggingface_hub import get_token
 from fast_agent.mcp.auth.middleware import HFAuthHeaderMiddleware
 from fast_agent.mcp.server import HarnessMCPAdapter
 from fast_agent.mcp.server.common import (
@@ -41,3 +42,16 @@ def http_middleware() -> list[Middleware] | None:
 def request_auth() -> AgentAuth | None:
     """Translate the current verified MCP token into fast-agent auth."""
     return HarnessMCPAdapter.agent_auth()
+
+
+def effective_agent_auth(auth: AgentAuth | None) -> AgentAuth | None:
+    """Use local credentials only when inbound Hugging Face OAuth is disabled."""
+    if auth is not None and auth.token:
+        return auth
+    if (
+        normalize_serve_oauth_provider(os.getenv("FAST_AGENT_SERVE_OAUTH"))
+        == "huggingface"
+    ):
+        return auth
+    token = os.getenv("HF_TOKEN") or get_token()
+    return AgentAuth.bearer(token, provider="huggingface") if token else auth

research/app_jobs.py

@@ -4,7 +4,8 @@ from __future__ import annotations
 
 import asyncio
 import hashlib
-from collections.abc import Callable
+from collections.abc import Callable, Coroutine
+from contextvars import ContextVar
 from dataclasses import dataclass, field
 from time import time
 from typing import Any
@@ -12,7 +13,18 @@ from uuid import uuid4
 
 from fast_agent import AgentAuth
 
-TERMINAL_STATUSES = {"completed", "failed"}
+TERMINAL_STATUSES = {"cancelled", "completed", "failed"}
+CANCELLABLE_STATUSES = {"queued", "running"}
+PHASE_SUMMARIES = {
+    "reporting": (
+        "The research findings and Markdown report are complete. "
+        "The HTML report is now being produced."
+    ),
+    "wrapping_up": (
+        "The Markdown and HTML reports are complete. "
+        "The agent is preparing the final response."
+    ),
+}
 
 
 def owner_id(auth: AgentAuth | None, session_id: str | None) -> str:
@@ -41,13 +53,27 @@ class ResearchJob:
     topic: str
     owner_id: str
     status: str = "queued"
+    phase: str = "preparing"
     created_at: float = field(default_factory=time)
     updated_at: float = field(default_factory=time)
     events: list[dict[str, Any]] = field(default_factory=list)
     result: str | None = None
+    markdown_report: str | None = None
+    markdown_report_uri: str | None = None
+    markdown_report_error: str | None = None
+    html_report_uri: str | None = None
+    html_report_url: str | None = None
     error: str | None = None
     trace_path: str | None = None
+    trace_archive_uri: str | None = None
     trace_error: str | None = None
+    activity_summary: str = "Preparing the research agent."
+    activity_summary_revision: int = 0
+    activity_source: str = "research/agent_loop"
+    activity_summaries: list[dict[str, Any]] = field(default_factory=list)
+    event_count_total: int = 0
+    turn_count: int = 0
+    birch_finalize_attempts: int = 0
 
     def add_event(
         self,
@@ -68,6 +94,7 @@ class ResearchJob:
                 "total": total,
             }
         )
+        self.event_count_total += 1
         del self.events[:-100]
 
     def snapshot(self, *, now: float | None = None) -> dict[str, Any]:
@@ -83,23 +110,89 @@ class ResearchJob:
             }
             for event in self.events
         ]
+        summaries = [
+            {
+                **summary,
+                "elapsed": format_elapsed(
+                    float(summary.get("ts") or self.created_at) - self.created_at
+                ),
+            }
+            for summary in self.activity_summaries
+        ]
         return {
             "job_id": self.id,
             "topic": self.topic,
             "status": self.status,
+            "phase": self.phase,
             "events": events,
             "timeline_events": events[-12:],
-            "event_count": len(events),
+            "recent_events": events[-2:],
+            "activity_roll": list(
+                reversed(
+                    [event for event in events if event["kind"] == "Activity"][-2:]
+                )
+            ),
+            "recent_summaries": list(reversed(summaries[:-1][-2:])),
+            "event_count": self.event_count_total,
             "elapsed_seconds": int(max(0, elapsed_seconds)),
             "elapsed": format_elapsed(elapsed_seconds),
             "activity_progress": 100 if done else int((elapsed_seconds * 12) % 100),
+            "activity_summary": self.activity_summary,
+            "activity_summary_revision": self.activity_summary_revision,
+            "activity_source": self.activity_source,
+            "turn_count": self.turn_count,
             "result": self.result,
+            "markdown_report": self.markdown_report,
+            "markdown_report_uri": self.markdown_report_uri,
+            "markdown_report_error": self.markdown_report_error,
+            "html_report_uri": self.html_report_uri,
+            "html_report_url": self.html_report_url,
+            "html_report_ready": bool(self.html_report_uri),
             "error": self.error,
             "trace_path": self.trace_path,
+            "trace_archive_uri": self.trace_archive_uri,
             "trace_error": self.trace_error,
             "done": done,
+            "cancellable": self.status in CANCELLABLE_STATUSES,
         }
 
+    def set_activity_summary(self, summary: str, *, now: float | None = None) -> None:
+        summary = summary.strip()
+        if not summary or summary == self.activity_summary:
+            return
+        self.activity_summary = summary
+        self.activity_summary_revision += 1
+        self.updated_at = time() if now is None else now
+        self.activity_summaries.append(
+            {
+                "ts": self.updated_at,
+                "message": summary,
+            }
+        )
+        del self.activity_summaries[:-10]
+
+    def record_llm_step(self) -> None:
+        self.turn_count += 1
+        self.activity_source = "research/agent_loop"
+
+    def set_activity_source(self, source: str) -> None:
+        if source:
+            self.activity_source = source
+
+    def set_phase(self, phase: str) -> None:
+        self.phase = phase
+        if summary := PHASE_SUMMARIES.get(phase):
+            self.set_activity_summary(summary)
+
+    def narrative_for_phase(self, summary: str) -> str:
+        return PHASE_SUMMARIES.get(self.phase, summary)
+
+
+current_research_job: ContextVar[ResearchJob | None] = ContextVar(
+    "current_research_job",
+    default=None,
+)
+
 
 @dataclass(frozen=True, slots=True)
 class BeginResult:
@@ -107,6 +200,37 @@ class BeginResult:
     started: bool
 
 
+@dataclass(frozen=True, slots=True)
+class CancelResult:
+    job: ResearchJob
+    cancel_task: bool
+
+
+class ResearchTaskRegistry:
+    """Track cancellable work for one server process."""
+
+    def __init__(self) -> None:
+        self._tasks: dict[str, asyncio.Task[None]] = {}
+
+    def start(self, job_id: str, work: Coroutine[Any, Any, None]) -> None:
+        task = asyncio.create_task(work, name=job_id)
+        self._tasks[job_id] = task
+        task.add_done_callback(
+            lambda completed, job_id=job_id: self._discard(job_id, completed)
+        )
+
+    def cancel(self, job_id: str) -> bool:
+        task = self._tasks.get(job_id)
+        if task is None or task.done():
+            return False
+        task.cancel()
+        return True
+
+    def _discard(self, job_id: str, task: asyncio.Task[None]) -> None:
+        if self._tasks.get(job_id) is task:
+            self._tasks.pop(job_id, None)
+
+
 class ResearchJobStore:
     def __init__(
         self,
@@ -138,7 +262,7 @@ class ResearchJobStore:
                 created_at=now,
                 updated_at=now,
             )
-            job.add_event("Queued research job", now=now)
+            job.add_event("Your research request is ready.", kind="Setup", now=now)
             self._jobs[job.id] = job
             self._enforce_limit()
             return job
@@ -153,7 +277,12 @@ class ResearchJobStore:
             if job.status != "queued":
                 return BeginResult(job=job, started=False)
             job.status = "running"
-            job.add_event("Starting research agent", now=self._clock())
+            job.phase = "researching"
+            job.add_event(
+                "The research agent is getting started.",
+                kind="Research",
+                now=self._clock(),
+            )
             return BeginResult(job=job, started=True)
 
     async def get(self, job_id: str, owner: str) -> ResearchJob | None:
@@ -161,6 +290,35 @@ class ResearchJobStore:
             self._prune(self._clock())
             return self._authorized_job(job_id, owner)
 
+    async def cancel(self, job_id: str, owner: str) -> CancelResult | None:
+        """Atomically request cancellation for an authorized job."""
+        async with self._lock:
+            self._prune(self._clock())
+            job = self._authorized_job(job_id, owner)
+            if job is None:
+                return None
+            if job.status == "queued":
+                now = self._clock()
+                job.status = "cancelled"
+                job.phase = "cancelled"
+                job.set_activity_summary(
+                    "Research was cancelled before the agent started.",
+                    now=now,
+                )
+                job.add_event("Research cancelled before start", now=now)
+                return CancelResult(job=job, cancel_task=False)
+            if job.status == "running":
+                now = self._clock()
+                job.status = "cancelling"
+                job.phase = "cancelling"
+                job.set_activity_summary(
+                    "Cancellation requested. Closing the active research session.",
+                    now=now,
+                )
+                job.add_event("Cancellation requested", now=now)
+                return CancelResult(job=job, cancel_task=True)
+            return CancelResult(job=job, cancel_task=False)
+
     def _authorized_job(self, job_id: str, owner: str) -> ResearchJob | None:
         job = self._jobs.get(job_id)
         return job if job is not None and job.owner_id == owner else None
@@ -200,18 +358,34 @@ def unavailable_snapshot(job_id: str) -> dict[str, Any]:
         "job_id": job_id,
         "topic": "",
         "status": "expired",
+        "phase": "expired",
         "events": [],
         "timeline_events": [],
+        "recent_events": [],
+        "activity_roll": [],
+        "recent_summaries": [],
         "event_count": 0,
         "elapsed_seconds": 0,
         "elapsed": "00:00",
         "activity_progress": 100,
+        "activity_summary": "This research run is no longer available.",
+        "activity_summary_revision": 0,
+        "activity_source": "research/agent_loop",
+        "turn_count": 0,
         "result": None,
+        "markdown_report": None,
+        "markdown_report_uri": None,
+        "markdown_report_error": None,
+        "html_report_uri": None,
+        "html_report_url": None,
+        "html_report_ready": False,
         "error": (
             "This research run is no longer available. Historical app views never "
             "start replacement work; ask Claude to run the research tool again."
         ),
         "trace_path": None,
+        "trace_archive_uri": None,
         "trace_error": None,
         "done": True,
+        "cancellable": False,
     }

research/app_observability.py

@@ -3,16 +3,36 @@
 from __future__ import annotations
 
 import asyncio
+import os
+import posixpath
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
+from urllib.parse import urlparse
 from uuid import uuid4
 
+from huggingface_hub import HfApi, HfFileSystem
+from huggingface_hub.errors import BucketNotFoundError
+
 from fast_agent.mcp.tool_execution_handler import ToolExecutionHandler
 from fast_agent.session import SessionTraceExporter
 from fast_agent.session.session_manager import SessionManager
 from fast_agent.session.trace_export_models import ExportRequest
 
 from .app_jobs import ResearchJob
+from .research_workspace import ResearchWorkspace, current_research_workspace
+
+MarkdownReader = Callable[[ResearchWorkspace], Awaitable[str]]
+MAX_MARKDOWN_REPORT_CHARS = 250_000
+ARCHIVE_URL_ENV = "RESEARCH_ARCHIVE_HF_URL"
+ARCHIVE_TOKEN_ENV = "RESEARCH_ARCHIVE_TOKEN"
+
+
+@dataclass(frozen=True, slots=True)
+class ArchiveTarget:
+    bucket_id: str
+    root: str
 
 
 class JobProgressHandler(ToolExecutionHandler):
@@ -20,7 +40,7 @@ class JobProgressHandler(ToolExecutionHandler):
 
     def __init__(self, job: ResearchJob) -> None:
         self.job = job
-        self._labels: dict[str, str] = {}
+        self._activities: dict[str, tuple[str, str, str, str]] = {}
 
     async def on_tool_start(
         self,
@@ -30,9 +50,13 @@ class JobProgressHandler(ToolExecutionHandler):
         tool_use_id: str | None = None,
     ) -> str:
         tool_call_id = tool_use_id or f"{server_name}/{tool_name}/{uuid4().hex[:8]}"
-        label = f"{server_name}/{tool_name}"
-        self._labels[tool_call_id] = label
-        self.job.add_event(f"{label}: started", kind="tool")
+        activity = _tool_activity(server_name, tool_name, arguments)
+        self._activities[tool_call_id] = activity
+        self.job.set_activity_source(activity[0])
+        self.job.add_event(f"{activity[0]}: started", kind="Activity")
+        if _is_birch_delegation(server_name, tool_name):
+            self.job.set_phase("reporting")
+            await capture_markdown_report(self.job)
         return tool_call_id
 
     async def on_tool_progress(
@@ -42,13 +66,13 @@ class JobProgressHandler(ToolExecutionHandler):
         total: float | None,
         message: str | None,
     ) -> None:
-        label = self._labels.get(tool_call_id, "agent")
-        self.job.add_event(
-            f"{label}: {message or 'working'}",
-            kind="progress",
-            progress=progress,
-            total=total,
-        )
+        del progress, total
+        source = self._activities.get(
+            tool_call_id,
+            ("research/agent_loop", "Research agent", "Research", ""),
+        )[0]
+        self.job.set_activity_source(source)
+        self.job.add_event(f"{source}: {message or 'working'}", kind="Activity")
 
     async def on_tool_complete(
         self,
@@ -57,9 +81,28 @@ class JobProgressHandler(ToolExecutionHandler):
         content: list[Any] | None,
         error: str | None,
     ) -> None:
-        label = self._labels.pop(tool_call_id, "agent")
-        outcome = "completed" if success else error or "failed"
-        self.job.add_event(f"{label}: {outcome}", kind="tool")
+        raw_source, source, category, completed = self._activities.pop(
+            tool_call_id,
+            (
+                "research/agent_loop",
+                "Research agent",
+                "Research",
+                "A research step finished.",
+            ),
+        )
+        message = completed if success else _friendly_tool_error(source, error)
+        self.job.add_event(
+            f"{raw_source}: completed" if success else message,
+            kind="Activity",
+        )
+        self.job.set_activity_source(
+            next(
+                (activity[0] for activity in reversed(self._activities.values())),
+                "research/agent_loop",
+            )
+        )
+        if _is_birch_delegation(*raw_source.split("/", 1)):
+            self.job.set_phase("wrapping_up" if success else "researching")
 
     async def on_tool_permission_denied(
         self,
@@ -68,16 +111,18 @@ class JobProgressHandler(ToolExecutionHandler):
         tool_use_id: str | None,
         error: str | None = None,
     ) -> None:
+        raw_source, source, _, _ = _tool_activity(server_name, tool_name, None)
+        self.job.set_activity_source(raw_source)
         self.job.add_event(
-            f"{server_name}/{tool_name}: {error or 'permission denied'}",
-            kind="error",
+            _friendly_tool_error(source, error or "Permission was denied."),
+            kind="Activity",
         )
 
     async def get_tool_call_id_for_tool_use(
         self,
         tool_use_id: str,
     ) -> str | None:
-        return tool_use_id if tool_use_id in self._labels else None
+        return tool_use_id if tool_use_id in self._activities else None
 
     async def ensure_tool_call_exists(
         self,
@@ -86,7 +131,7 @@ class JobProgressHandler(ToolExecutionHandler):
         server_name: str,
         arguments: dict | None = None,
     ) -> str:
-        if tool_use_id in self._labels:
+        if tool_use_id in self._activities:
             return tool_use_id
         return await self.on_tool_start(
             tool_name,
@@ -96,6 +141,114 @@ class JobProgressHandler(ToolExecutionHandler):
         )
 
 
+def _tool_activity(
+    server_name: str,
+    tool_name: str,
+    arguments: dict[str, Any] | None,
+) -> tuple[str, str, str, str]:
+    raw_source = f"{server_name}/{tool_name}"
+    raw_name = f"{server_name}/{tool_name}".lower()
+    if "birch-html" in raw_name:
+        return (
+            raw_source,
+            "Report writer",
+            "Report",
+            "The report writer finished another section.",
+        )
+    if tool_name == "agent_loop":
+        return (
+            raw_source,
+            "Research agent",
+            "Research",
+            "The agent completed a research step.",
+        )
+    if server_name == "hf" and tool_name == "hf_fs":
+        command = str((arguments or {}).get("cmd") or "").lower()
+        if command == "search":
+            return (
+                raw_source,
+                "Searching Hugging Face",
+                "Hugging Face",
+                "The Hugging Face search finished.",
+            )
+        if command == "cat":
+            return (
+                raw_source,
+                "Reading a Hugging Face source",
+                "Hugging Face",
+                "The agent finished reading a Hugging Face source.",
+            )
+        return (
+            raw_source,
+            "Browsing Hugging Face",
+            "Hugging Face",
+            "The Hugging Face lookup finished.",
+        )
+    if server_name == "hf" and "sandbox" in tool_name:
+        return (
+            raw_source,
+            "Running analysis",
+            "Analysis",
+            "The latest analysis step finished.",
+        )
+    readable = tool_name.split("[", 1)[0].replace("_", " ").replace("-", " ")
+    return (
+        raw_source,
+        readable.capitalize(),
+        "Research",
+        f"The agent finished {readable}.",
+    )
+
+
+def _is_birch_delegation(server_name: str, tool_name: str) -> bool:
+    return server_name == "agent" and tool_name.split("[", 1)[0] == "birch-html"
+
+
+async def capture_markdown_report(
+    job: ResearchJob,
+    *,
+    reader: MarkdownReader | None = None,
+) -> None:
+    workspace = current_research_workspace.get()
+    if workspace is None:
+        return
+    uri = f"{workspace.output}report.md"
+    try:
+        markdown = await (reader or _read_markdown_report)(workspace)
+    except Exception as exc:
+        job.markdown_report_error = str(exc)
+        return
+
+    if len(markdown) > MAX_MARKDOWN_REPORT_CHARS:
+        markdown = (
+            markdown[:MAX_MARKDOWN_REPORT_CHARS].rstrip()
+            + "\n\n_This in-app preview was truncated; open the artifact for the full report._"
+        )
+    job.markdown_report = markdown
+    job.markdown_report_uri = uri
+    job.markdown_report_error = None
+    job.add_event("The Markdown report is ready to review.", kind="Report")
+
+
+async def _read_markdown_report(workspace: ResearchWorkspace) -> str:
+    def read() -> str:
+        filesystem = HfFileSystem(token=workspace.bearer_token)
+        with filesystem.open(f"{workspace.output}report.md", "r") as report:
+            return str(report.read())
+
+    return await asyncio.to_thread(read)
+
+
+def _friendly_tool_error(source: str, error: str | None) -> str:
+    detail = (error or "The operation did not complete.").strip()
+    if "search requires a positional query or --query" in detail:
+        return "A Hugging Face search request was missing its query."
+    detail = detail.removeprefix("EINVAL:").strip()
+    if len(detail) > 180:
+        detail = f"{detail[:177].rstrip()}…"
+    return f"{source} encountered a problem: {detail}"
+
+
 def export_trace(job: ResearchJob, home: Path) -> None:
     output_path = (
         home
@@ -121,6 +274,100 @@ def export_trace(job: ResearchJob, home: Path) -> None:
         f"Exported Codex trace: {result.output_path} ({result.record_count} records)",
         kind="trace",
     )
+    archive = _archive_config()
+    if archive is not None:
+        target, token = archive
+        job.trace_archive_uri = archive_session(
+            job,
+            home,
+            result.output_path,
+            target=target,
+            token=token,
+        )
+        job.add_event(
+            f"Archived private session: {job.trace_archive_uri}",
+            kind="trace",
+        )
+
+
+def _archive_config() -> tuple[ArchiveTarget, str] | None:
+    url = os.getenv(ARCHIVE_URL_ENV, "").strip()
+    token = os.getenv(ARCHIVE_TOKEN_ENV, "").strip()
+    if not url and not token:
+        return None
+    if not url or not token:
+        missing = ARCHIVE_URL_ENV if not url else ARCHIVE_TOKEN_ENV
+        raise RuntimeError(f"Private session archive is missing {missing}.")
+    return _archive_target(url), token
+
+
+def _archive_target(url: str) -> ArchiveTarget:
+    parsed = urlparse(url)
+    parts = [part for part in parsed.path.split("/") if part]
+    if parsed.scheme != "hf" or parsed.netloc != "buckets" or len(parts) < 2:
+        raise ValueError(
+            f"{ARCHIVE_URL_ENV} must be an hf://buckets/<owner>/<bucket> URL."
+        )
+    bucket_id = f"{parts[0]}/{parts[1]}"
+    prefix = "/".join(parts[2:])
+    root = f"hf://buckets/{bucket_id}"
+    if prefix:
+        root = f"{root}/{prefix}"
+    return ArchiveTarget(bucket_id=bucket_id, root=root)
+
+
+def archive_session(
+    job: ResearchJob,
+    home: Path,
+    trace_path: Path,
+    *,
+    target: ArchiveTarget,
+    token: str,
+    api: Any | None = None,
+    filesystem: Any | None = None,
+) -> str:
+    """Archive one raw session and Codex trace using an app-only credential."""
+    api = api or HfApi(token=token)
+    filesystem = filesystem or HfFileSystem(token=token)
+    try:
+        info = api.bucket_info(target.bucket_id, token=token)
+    except BucketNotFoundError:
+        api.create_bucket(
+            target.bucket_id,
+            private=True,
+            exist_ok=True,
+            token=token,
+        )
+    else:
+        if not bool(getattr(info, "private", False)):
+            raise RuntimeError(
+                f"Refusing to archive sessions to public bucket {target.bucket_id!r}."
+            )
+
+    session_dir = home / "sessions" / job.id
+    if not session_dir.is_dir():
+        raise FileNotFoundError(f"Session directory does not exist: {session_dir}")
+    for source in sorted(path for path in session_dir.rglob("*") if path.is_file()):
+        relative = source.relative_to(session_dir).as_posix()
+        _upload_archive_file(
+            filesystem,
+            source,
+            f"{target.root}/{job.id}/{relative}",
+        )
+
+    trace_uri = (
+        f"{target.root}/research-traces/{job.id}/{posixpath.basename(trace_path)}"
+    )
+    _upload_archive_file(filesystem, trace_path, trace_uri)
+    return trace_uri
+
+
+def _upload_archive_file(filesystem: Any, source: Path, destination: str) -> None:
+    with (
+        source.open("rb") as source_file,
+        filesystem.open(destination, "wb") as destination_file,
+    ):
+        destination_file.write(source_file.read())
 
 
 async def try_export_trace(job: ResearchJob, home: Path) -> None:

research/app_ui.py

@@ -10,206 +10,600 @@ from prefab_ui.app import PrefabApp
 from prefab_ui.components import (
     Badge,
     Button,
-    Card,
-    CardContent,
-    CardDescription,
-    CardHeader,
-    CardTitle,
-    Code,
     Column,
-    Dot,
-    Grid,
+    Div,
     Heading,
     If,
-    Loader,
     Markdown,
-    Metric,
-    Muted,
-    Progress,
     Row,
-    Small,
+    Separator,
     Text,
 )
-from prefab_ui.components.control_flow import Else, ForEach
+from prefab_ui.components.control_flow import ForEach
 from prefab_ui.rx import RESULT, STATE
 
+BROADSHEET_CSS = """
+.dispatch-app {
+  min-height: 100%;
+  padding: 24px;
+  background: var(--muted);
+  color: var(--foreground);
+}
+.dispatch-sheet {
+  width: min(100%, 800px);
+  min-height: 760px;
+  margin: 0 auto;
+  overflow: hidden;
+  display: flex;
+  flex-direction: column;
+  border: 1px solid var(--border);
+  border-radius: var(--radius);
+  background: var(--background);
+  box-shadow: 0 12px 36px color-mix(in oklab, var(--foreground) 10%, transparent);
+}
+.dispatch-header {
+  padding: 30px 36px 0;
+}
+.dispatch-kicker,
+.dispatch-section-label,
+.dispatch-time,
+.dispatch-source,
+.dispatch-meta,
+.dispatch-trace {
+  font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, monospace;
+}
+.dispatch-kicker {
+  color: color-mix(in oklab, var(--foreground) 58%, var(--warning));
+  font-size: 11px;
+  letter-spacing: .16em;
+  text-transform: uppercase;
+}
+.dispatch-controls {
+  flex-wrap: wrap;
+  justify-content: flex-end;
+}
+.dispatch-confirm {
+  color: var(--muted-foreground);
+  font-size: 12px;
+}
+.dispatch-section-label {
+  color: var(--muted-foreground);
+  font-size: 10px;
+  letter-spacing: .15em;
+  text-transform: uppercase;
+}
+.dispatch-query {
+  max-width: 28ch;
+  margin-top: 8px;
+  font-family: ui-serif, Georgia, Cambria, "Times New Roman", serif;
+  font-size: clamp(22px, 3.5vw, 34px);
+  font-weight: 500;
+  line-height: 1.08;
+  letter-spacing: -.025em;
+  text-wrap: balance;
+}
+.dispatch-body {
+  flex: 1;
+  min-height: 0;
+  overflow-y: auto;
+  padding: 28px 36px 32px;
+}
+.dispatch-current-meta {
+  gap: 14px;
+  align-items: baseline;
+}
+.dispatch-run-stats {
+  gap: 12px;
+  align-items: center;
+  flex-wrap: wrap;
+  color: var(--muted-foreground);
+}
+.dispatch-run-stat {
+  font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, monospace;
+  font-size: 13px;
+  font-variant-numeric: tabular-nums;
+  letter-spacing: .035em;
+  text-transform: uppercase;
+}
+.dispatch-activity-roll {
+  margin-top: 15px;
+  padding: 10px 12px;
+  border: 1px solid var(--border);
+  border-radius: var(--radius);
+  background: color-mix(in oklab, var(--muted) 55%, transparent);
+}
+.dispatch-activity-line {
+  gap: 12px;
+  min-width: 0;
+  align-items: baseline;
+}
+.dispatch-activity-line + .dispatch-activity-line {
+  margin-top: 5px;
+}
+.dispatch-activity-message {
+  min-width: 0;
+  overflow: hidden;
+  color: var(--muted-foreground);
+  font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, monospace;
+  font-size: 12px;
+  text-overflow: ellipsis;
+  white-space: nowrap;
+}
+.dispatch-time {
+  flex: none;
+  color: var(--muted-foreground);
+  font-size: 12px;
+  font-variant-numeric: tabular-nums;
+}
+.dispatch-source {
+  color: color-mix(in oklab, var(--foreground) 62%, var(--warning));
+  font-size: 11px;
+  letter-spacing: .025em;
+}
+.dispatch-current {
+  margin-top: 12px;
+  align-items: flex-start;
+  gap: 14px;
+}
+.dispatch-current-copy {
+  max-width: 62ch;
+  font-family: ui-serif, Georgia, Cambria, "Times New Roman", serif;
+  font-size: clamp(18px, 2.8vw, 25px);
+  line-height: 1.42;
+  letter-spacing: -.012em;
+  text-wrap: pretty;
+}
+.dispatch-current-copy > :first-child,
+.dispatch-event-message > :first-child {
+  margin-top: 0;
+}
+.dispatch-current-copy > :last-child,
+.dispatch-event-message > :last-child {
+  margin-bottom: 0;
+}
+.dispatch-live-dot {
+  position: relative;
+  width: 11px;
+  height: 11px;
+  flex: none;
+  margin-top: 11px;
+  border-radius: 999px;
+  background: var(--success);
+}
+.dispatch-live-dot::after {
+  content: "";
+  position: absolute;
+  inset: 0;
+  border-radius: inherit;
+  background: inherit;
+  animation: dispatch-ring 1.6s ease-out infinite;
+}
+.dispatch-rule {
+  height: 2px;
+  margin-top: 20px;
+  overflow: hidden;
+  border-radius: 2px;
+  background: var(--border);
+}
+.dispatch-rule-running::after {
+  content: "";
+  display: block;
+  width: 40%;
+  height: 100%;
+  background: var(--warning);
+  animation: dispatch-progress 1.6s ease-in-out infinite;
+}
+.dispatch-rule-completed { background: var(--success); }
+.dispatch-rule-failed {
+  height: 0;
+  border-top: 2px dashed var(--muted-foreground);
+  background: transparent;
+}
+.dispatch-rule-cancelled { background: var(--muted-foreground); }
+.dispatch-history {
+  margin-top: 24px;
+}
+.dispatch-event {
+  gap: 16px;
+  padding: 13px 0;
+  border-top: 1px solid var(--border);
+  opacity: .76;
+}
+.dispatch-event-copy {
+  min-width: 0;
+}
+.dispatch-event-message {
+  margin-top: 4px;
+  font-family: ui-serif, Georgia, Cambria, "Times New Roman", serif;
+  font-size: 14px;
+  line-height: 1.45;
+  text-wrap: pretty;
+}
+.dispatch-result {
+  margin-top: 28px;
+  padding: 22px;
+  border: 1px solid var(--border);
+  border-radius: var(--radius);
+  background: var(--card);
+}
+.dispatch-report-markdown {
+  font-family: ui-serif, Georgia, Cambria, "Times New Roman", serif;
+  font-size: 16px;
+  line-height: 1.62;
+  text-wrap: pretty;
+}
+.dispatch-result .dispatch-report-markdown h1 {
+  margin: 8px 0 18px;
+  font-size: clamp(25px, 4vw, 32px);
+  font-weight: 500;
+  line-height: 1.12;
+  letter-spacing: -.022em;
+  text-wrap: balance;
+}
+.dispatch-result .dispatch-report-markdown h2 {
+  margin: 30px 0 12px;
+  font-size: clamp(21px, 3vw, 25px);
+  font-weight: 500;
+  line-height: 1.2;
+  letter-spacing: -.012em;
+}
+.dispatch-result .dispatch-report-markdown h3 {
+  margin: 24px 0 10px;
+  font-size: 18px;
+  font-weight: 600;
+  line-height: 1.3;
+}
+.dispatch-result .dispatch-report-markdown p,
+.dispatch-result .dispatch-report-markdown li {
+  line-height: 1.62;
+}
+.dispatch-result .dispatch-report-markdown strong {
+  font-weight: 650;
+}
+.dispatch-result .dispatch-report-markdown hr {
+  margin: 26px 0;
+  border-color: var(--border);
+}
+.dispatch-result .dispatch-report-markdown code,
+.dispatch-result .dispatch-report-markdown pre {
+  font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, monospace;
+}
+.dispatch-error {
+  margin-top: 20px;
+  color: var(--destructive);
+  font-size: 13px;
+}
+.dispatch-footer {
+  min-height: 44px;
+  padding: 10px 36px;
+  gap: 12px;
+  align-items: center;
+  border-top: 1px solid var(--border);
+  color: var(--muted-foreground);
+}
+.dispatch-meta {
+  flex: none;
+  font-size: 11px;
+  letter-spacing: .035em;
+  font-variant-numeric: tabular-nums;
+  text-transform: uppercase;
+}
+.dispatch-trace {
+  min-width: 0;
+  margin-left: auto;
+  overflow: hidden;
+  color: var(--muted-foreground);
+  font-size: 11px;
+  text-overflow: ellipsis;
+  white-space: nowrap;
+}
+@keyframes dispatch-progress {
+  from { transform: translateX(-110%); }
+  to { transform: translateX(250%); }
+}
+@keyframes dispatch-ring {
+  from { transform: scale(1); opacity: .45; }
+  to { transform: scale(2.5); opacity: 0; }
+}
+@media (max-width: 640px) {
+  .dispatch-app { padding: 0; }
+  .dispatch-sheet {
+    min-height: 680px;
+    border-right: 0;
+    border-left: 0;
+    border-radius: 0;
+    box-shadow: none;
+  }
+  .dispatch-header { padding: 24px 22px 0; }
+  .dispatch-body { padding: 24px 22px 28px; }
+  .dispatch-footer {
+    padding: 14px 22px;
+    flex-wrap: wrap;
+  }
+  .dispatch-trace {
+    width: 100%;
+    margin-left: 0;
+  }
+}
+@media (prefers-reduced-motion: reduce) {
+  .dispatch-live-dot::after,
+  .dispatch-rule-running::after {
+    animation: none;
+  }
+}
+"""
+
 
 def build_research_ui(
     topic: str,
     snapshot: dict[str, Any],
     *,
     build_id: str,
+    live: bool = True,
 ) -> PrefabApp:
-    with PrefabApp(
-        state={
-            "job": snapshot,
-            "topic": topic,
-            "job_id": snapshot["job_id"],
-            "poll_ms": "1500",
-            "chat_sent": False,
-            "app_version": f"build {build_id}",
-        }
-    ) as ui:
-        with Column(
-            gap=4,
-            css_class="p-6",
-            on_mount=[
-                CallTool(
-                    "start_research",
+    on_mount = None
+    if live:
+        on_mount = [
+            CallTool(
+                "start_research",
+                arguments={"job_id": STATE.job_id},
+                on_success=[
+                    SetState("job", RESULT),
+                    SetState("poll_ms", RESULT.done.then("86400000", "1500")),
+                ],
+            ),
+            SetInterval(
+                duration=STATE.poll_ms,
+                on_tick=CallTool(
+                    "research_status",
                     arguments={"job_id": STATE.job_id},
                     on_success=[
                         SetState("job", RESULT),
                         SetState("poll_ms", RESULT.done.then("86400000", "1500")),
                     ],
                 ),
-                SetInterval(
-                    duration=STATE.poll_ms,
-                    on_tick=CallTool(
-                        "research_status",
-                        arguments={"job_id": STATE.job_id},
-                        on_success=[
-                            SetState("job", RESULT),
-                            SetState("poll_ms", RESULT.done.then("86400000", "1500")),
-                        ],
-                    ),
+            ),
+        ]
+
+    cancel_action = SetState("confirm_cancel", False)
+    if live:
+        cancel_action = CallTool(
+            "cancel_research",
+            arguments={"job_id": STATE.job_id},
+            on_success=[
+                SetState("job", RESULT),
+                SetState("confirm_cancel", False),
+                SetState("cancel_requested", False),
+                ShowToast(
+                    "Cancellation requested",
+                    description="The active research session is being closed.",
+                    variant="warning",
                 ),
             ],
-        ):
-            with Card(css_class="border-blue-200 bg-blue-50/40"):
-                with CardContent():
-                    with Row(justify="between", align="center", gap=4):
-                        with Column(gap=1):
-                            Heading("🤗 Research Agent")
-                            Muted(STATE.topic, css_class="text-xs leading-4")
-                        with Row(gap=2, align="center"):
-                            Badge(STATE.app_version, variant="outline")
-                            with If("{{ !job.done }}"):
-                                Loader(variant="dots", size="sm")
-                            Badge(STATE.job.status, variant="info")
-
-            with Grid(columns={"default": 1, "lg": 3}, gap=4):
-                with Card(css_class="lg:col-span-2"):
-                    with CardHeader():
-                        CardTitle("Timeline")
-                        CardDescription(
-                            "Latest 12 events; older events roll off the visible list."
-                        )
-                    with CardContent():
-                        with Column(gap=2):
-                            with ForEach("job.timeline_events") as event:
-                                with Row(
-                                    gap=3,
-                                    align="start",
-                                    css_class=(
-                                        "rounded-md border border-border/60 "
-                                        "bg-background px-3 py-2"
-                                    ),
-                                ):
-                                    Dot(variant="info", size="sm", css_class="mt-1")
-                                    Small(
-                                        event.elapsed,
-                                        code=True,
-                                        css_class="min-w-12 text-muted-foreground",
-                                    )
-                                    Badge(event.kind, variant="outline")
-                                    Text(
-                                        event.message,
-                                        css_class="text-sm leading-5",
-                                    )
+            on_error=[
+                SetState("cancel_requested", False),
+                ShowToast(
+                    "Could not cancel research",
+                    description="The job may already have finished.",
+                    variant="error",
+                ),
+            ],
+        )
 
-                with Column(gap=4):
-                    with Card():
-                        with CardHeader():
-                            CardTitle("Activity")
-                            CardDescription("Open-ended research task")
-                        with CardContent():
-                            with If("{{ !job.done }}"):
-                                with Row(gap=3, align="center"):
-                                    Loader(variant="bars", size="sm")
-                                    Muted("Working")
-                            with Else():
-                                with Row(gap=3, align="center"):
-                                    Badge(STATE.job.status, variant="outline")
-                                    Muted("No active work")
-                            Progress(
-                                value=STATE.job.activity_progress,
-                                max=100,
-                                gradient=True,
-                                size="sm",
+    with PrefabApp(
+        title="Research Dispatch",
+        css_class="dispatch-app",
+        css=[BROADSHEET_CSS],
+        state={
+            "job": snapshot,
+            "topic": topic,
+            "job_id": snapshot["job_id"],
+            "poll_ms": "1500",
+            "confirm_cancel": False,
+            "cancel_requested": False,
+            "chat_sent": False,
+            "app_version": f"build {build_id}",
+        },
+    ) as ui:
+        with Div(css_class="dispatch-sheet", on_mount=on_mount):
+            with Column(css_class="dispatch-header", gap=0):
+                with Row(justify="between", align="start", gap=4):
+                    Text("Research Dispatch", css_class="dispatch-kicker")
+                    with Row(css_class="dispatch-controls", gap=2, align="center"):
+                        Badge(STATE.app_version, variant="outline")
+                        with If(
+                            (STATE.job.status == "queued")
+                            | (
+                                (STATE.job.status == "running")
+                                & (STATE.job.phase != "reporting")
+                                & (STATE.job.phase != "wrapping_up")
+                            )
+                        ):
+                            Badge("Working", variant="warning")
+                        with If(
+                            (STATE.job.status == "running")
+                            & (STATE.job.phase == "reporting")
+                        ):
+                            Badge("Building report", variant="info")
+                        with If(
+                            (STATE.job.status == "running")
+                            & (STATE.job.phase == "wrapping_up")
+                        ):
+                            Badge("Wrapping up", variant="warning")
+                        with If(STATE.job.status == "finalizing"):
+                            Badge("Finalizing", variant="warning")
+                        with If(STATE.job.status == "completed"):
+                            Badge("Complete", variant="success")
+                        with If(STATE.job.status == "failed"):
+                            Badge("Failed", variant="outline")
+                        with If(STATE.job.status == "cancelled"):
+                            Badge("Cancelled", variant="secondary")
+                        with If(STATE.job.status == "cancelling"):
+                            Badge("Cancelling", variant="warning")
+                        with If(STATE.job.status == "expired"):
+                            Badge("Unavailable", variant="secondary")
+                        with If(STATE.job.cancellable & ~STATE.confirm_cancel):
+                            Button(
+                                "Cancel",
+                                variant="ghost",
+                                size="xs",
+                                onClick=SetState("confirm_cancel", True),
+                            )
+                        with If(STATE.job.cancellable & STATE.confirm_cancel):
+                            Text("Cancel research?", css_class="dispatch-confirm")
+                            Button(
+                                "Keep running",
+                                variant="ghost",
+                                size="xs",
+                                onClick=SetState("confirm_cancel", False),
+                            )
+                            Button(
+                                "Confirm",
+                                variant="destructive",
+                                size="xs",
+                                disabled=STATE.cancel_requested,
+                                onClick=[
+                                    SetState("cancel_requested", True),
+                                    cancel_action,
+                                ],
                             )
 
-                    with Grid(columns=2, gap=3):
-                        with Card():
-                            with CardContent():
-                                Metric(label="Runtime", value=STATE.job.elapsed)
-                        with Card():
-                            with CardContent():
-                                Metric(label="Events", value=STATE.job.event_count)
-
-                    with If(STATE.job.trace_path):
-                        with Card():
-                            with CardHeader():
-                                CardTitle("Session trace")
-                            with CardContent():
-                                Code(STATE.job.trace_path)
-
-                    with If(STATE.job.trace_error):
-                        with Card():
-                            with CardHeader():
-                                CardTitle("Trace export")
-                            with CardContent():
+                Separator(spacing=4)
+                Text("Query", css_class="dispatch-section-label")
+                Heading(STATE.topic, level=1, css_class="dispatch-query")
+
+            with Div(css_class="dispatch-body"):
+                with Row(css_class="dispatch-run-stats"):
+                    Text(
+                        "{{ 'Runtime ' + job.elapsed }}",
+                        css_class="dispatch-run-stat",
+                    )
+                    Text("·", css_class="dispatch-run-stat")
+                    Text(
+                        "{{ job.event_count + ' events' }}",
+                        css_class="dispatch-run-stat",
+                    )
+                    Text("·", css_class="dispatch-run-stat")
+                    Text(
+                        "{{ job.turn_count + ' agent turns' }}",
+                        css_class="dispatch-run-stat",
+                    )
+
+                with If("{{ job.activity_roll.length > 0 }}"):
+                    with Div(css_class="dispatch-activity-roll"):
+                        with ForEach("job.activity_roll") as event:
+                            with Row(css_class="dispatch-activity-line"):
+                                Text(event.elapsed, css_class="dispatch-time")
                                 Text(
-                                    STATE.job.trace_error,
-                                    css_class="text-amber-600",
+                                    event.message,
+                                    css_class="dispatch-activity-message",
                                 )
 
-            with If(STATE.job.error):
-                with Card(css_class="border-red-200 bg-red-50/60"):
-                    with CardHeader():
-                        CardTitle("Unavailable")
-                    with CardContent():
-                        Text(STATE.job.error, css_class="text-red-700")
-
-            with If(STATE.job.result):
-                with Card():
-                    with CardHeader():
-                        CardTitle("Final result")
-                        CardDescription("Generated report and summary")
-                    with CardContent():
+                with Row(css_class="dispatch-current"):
+                    with If("{{ !job.done }}"):
+                        Div(css_class="dispatch-live-dot")
+                    Markdown(
+                        STATE.job.activity_summary,
+                        css_class="dispatch-current-copy",
+                    )
+
+                with If("{{ !job.done }}"):
+                    Div(css_class="dispatch-rule dispatch-rule-running")
+                with If(STATE.job.status == "completed"):
+                    Div(css_class="dispatch-rule dispatch-rule-completed")
+                with If(STATE.job.status == "failed"):
+                    Div(css_class="dispatch-rule dispatch-rule-failed")
+                with If(
+                    (STATE.job.status == "cancelled") | (STATE.job.status == "expired")
+                ):
+                    Div(css_class="dispatch-rule dispatch-rule-cancelled")
+
+                with If(STATE.job.markdown_report):
+                    with Div(css_class="dispatch-result"):
+                        with If(STATE.job.html_report_ready):
+                            with Row(justify="between", align="center", gap=3):
+                                Text(
+                                    "Markdown report",
+                                    css_class="dispatch-section-label",
+                                )
+                                Badge("HTML report produced", variant="success")
+                        with If(
+                            (STATE.job.phase == "reporting")
+                            & ~STATE.job.html_report_ready
+                        ):
+                            Text(
+                                "Markdown report · HTML version in progress",
+                                css_class="dispatch-section-label",
+                            )
+                        with If(
+                            (STATE.job.phase != "reporting")
+                            & (STATE.job.phase != "wrapping_up")
+                            & ~STATE.job.html_report_ready
+                        ):
+                            Text(
+                                "Markdown report",
+                                css_class="dispatch-section-label",
+                            )
                         Markdown(
-                            STATE.job.result,
-                            css_class="whitespace-pre-wrap",
+                            STATE.job.markdown_report,
+                            css_class="dispatch-report-markdown",
                         )
-                        with If("{{ !chat_sent }}"):
-                            Button(
-                                "Continue in chat",
-                                variant="outline",
-                                size="sm",
-                                css_class="mt-4",
-                                on_click=CallTool(
-                                    "research_chat_context",
-                                    arguments={"job_id": STATE.job_id},
-                                    on_success=[
-                                        UpdateContext(content=RESULT.markdown),
-                                        SendMessage(
-                                            RESULT.message,
-                                            on_success=SetState("chat_sent", True),
-                                            on_error=ShowToast(
-                                                "This host could not send the chat message.",
-                                                variant="error",
-                                            ),
-                                        ),
-                                    ],
+
+                with If("{{ job.recent_summaries.length > 0 }}"):
+                    with Div(css_class="dispatch-history"):
+                        Text("Earlier updates", css_class="dispatch-section-label")
+                        with ForEach("job.recent_summaries") as event:
+                            with Row(css_class="dispatch-event", align="start"):
+                                Text(event.elapsed, css_class="dispatch-time")
+                                with Column(css_class="dispatch-event-copy", gap=0):
+                                    Markdown(
+                                        event.message,
+                                        css_class="dispatch-event-message",
+                                    )
+
+                with If(STATE.job.error):
+                    Text(STATE.job.error, css_class="dispatch-error")
+
+                with If(STATE.job.result & ~STATE.job.markdown_report):
+                    with Div(css_class="dispatch-result"):
+                        Text("Final response", css_class="dispatch-section-label")
+                        Markdown(STATE.job.result)
+
+                with If(
+                    (STATE.job.status == "completed") & ~STATE.chat_sent
+                ):
+                    Button(
+                        "Continue in chat",
+                        variant="outline",
+                        size="sm",
+                        onClick=CallTool(
+                            "research_chat_context",
+                            arguments={"job_id": STATE.job_id},
+                            on_success=[
+                                UpdateContext(content=RESULT.markdown),
+                                SendMessage(
+                                    RESULT.message,
+                                    on_success=SetState("chat_sent", True),
                                     on_error=ShowToast(
-                                        "Could not load the Markdown report.",
+                                        "This host could not send the chat message.",
                                         variant="error",
                                     ),
                                 ),
-                            )
-                        with If(STATE.chat_sent):
-                            Muted(
-                                "Report added to model context and sent to chat.",
-                                css_class="mt-3 text-xs",
-                            )
+                            ],
+                            on_error=ShowToast(
+                                "Could not load the Markdown report.",
+                                variant="error",
+                            ),
+                        ),
+                    )
+                with If(STATE.chat_sent):
+                    Text(
+                        "Report added to model context and sent to chat.",
+                        css_class="dispatch-meta",
+                    )
+
+            with If(STATE.job.trace_path):
+                with Row(css_class="dispatch-footer"):
+                    Text(STATE.job.trace_path, css_class="dispatch-trace")
 
     return ui

research/fast-agent.yaml

@@ -8,7 +8,7 @@ logger:
 
 model_references:
   system:
-    fast: codexspark
+    fast: gpt-oss
     last_used: codexresponses.gpt-5.5?reasoning=medium
     research: kimi26
     html: kimi27
@@ -18,7 +18,9 @@ mcp:
     hf:
       description: Hugging Face MCP server
       transport: http
-      url: https://huggingface.co/mcp
+      url: https://huggingface.co/mcp?bouquet=research
+      read_timeout_seconds: 30
+      include_instructions: false
       auth:
         forward: huggingface
     hf_docs_only:

research/fastmcp_server.py

@@ -4,7 +4,6 @@ from __future__ import annotations
 
 import argparse
 import asyncio
-from collections.abc import Coroutine
 from pathlib import Path
 from typing import Annotated, Any
 
@@ -16,14 +15,18 @@ from prefab_ui.app import PrefabApp
 
 from .app_auth import auth_provider, http_middleware, request_auth
 from .app_artifacts import read_bucket_markdown
-from .app_jobs import ResearchJobStore, owner_id, unavailable_snapshot
+from .app_jobs import (
+    ResearchJobStore,
+    ResearchTaskRegistry,
+    owner_id,
+    unavailable_snapshot,
+)
 from .app_renderer import app_build_id, install_versioned_renderer
 from .app_ui import build_research_ui
 from .research_runner import ResearchRunner
 
 RESEARCH_HOME = Path(__file__).parent
 AGENT_CARDS = RESEARCH_HOME / "agent-cards"
-_BACKGROUND_TASKS: set[asyncio.Task[None]] = set()
 
 
 def parse_args() -> argparse.Namespace:
@@ -38,12 +41,6 @@ def parse_args() -> argparse.Namespace:
     return parser.parse_args()
 
 
-def run_in_background(work: Coroutine[Any, Any, None]) -> None:
-    task = asyncio.create_task(work)
-    _BACKGROUND_TASKS.add(task)
-    task.add_done_callback(_BACKGROUND_TASKS.discard)
-
-
 def register_research_app(
     app: FastMCPApp,
     jobs: ResearchJobStore,
@@ -51,6 +48,7 @@ def register_research_app(
     build_id: str,
 ) -> None:
     """Register one UI entry point and its app-only backend tools."""
+    tasks = ResearchTaskRegistry()
 
     @app.ui(
         name="research",
@@ -68,7 +66,7 @@ def register_research_app(
             if auth is not None
             else "local development user"
         )
-        job.add_event(f"Authenticated as {identity}", kind="auth")
+        job.add_event(f"Workspace access confirmed for {identity}.", kind="Setup")
         return build_research_ui(topic, job.snapshot(), build_id=build_id)
 
     @app.tool()
@@ -81,7 +79,7 @@ def register_research_app(
         if result is None:
             return unavailable_snapshot(job_id)
         if result.started:
-            run_in_background(runner.run(result.job, auth))
+            tasks.start(result.job.id, runner.run(result.job, auth))
         return result.job.snapshot()
 
     @app.tool()
@@ -93,6 +91,19 @@ def register_research_app(
         job = await jobs.get(job_id, owner_id(auth, ctx.session_id))
         return job.snapshot() if job else unavailable_snapshot(job_id)
 
+    @app.tool()
+    async def cancel_research(
+        job_id: str,
+        ctx: MCPContext,
+    ) -> dict[str, Any]:
+        auth = request_auth()
+        result = await jobs.cancel(job_id, owner_id(auth, ctx.session_id))
+        if result is None:
+            return unavailable_snapshot(job_id)
+        if result.cancel_task:
+            tasks.cancel(job_id)
+        return result.job.snapshot()
+
     @app.tool()
     async def research_chat_context(
         job_id: str,

research/research_app.py

@@ -11,9 +11,17 @@ from fast_agent import AgentRequest, AppOpenRequest, HarnessAppContext
 from mcp.types import TextContent
 
 try:
-    from .research_workspace import ResearchWorkspace, ensure_workspace
+    from .research_workspace import (
+        ResearchWorkspace,
+        current_research_workspace,
+        ensure_workspace,
+    )
 except ImportError:  # loaded as top-level module from the fast-agent home
-    from research_workspace import ResearchWorkspace, ensure_workspace
+    from research.research_workspace import (
+        ResearchWorkspace,
+        current_research_workspace,
+        ensure_workspace,
+    )
 
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator, Mapping
@@ -58,16 +66,20 @@ class ResearchHarnessSession:
             open_metadata=self._open_metadata,
         )
         forwarded = self._with_bucket_instructions(request, workspace)
-        if workspace.bearer_token is None:
-            return await self._session.invoke(forwarded)
+        workspace_token = current_research_workspace.set(workspace)
+        try:
+            if workspace.bearer_token is None:
+                return await self._session.invoke(forwarded)
 
-        from fast_agent.mcp.auth.context import request_bearer_token
+            from fast_agent.mcp.auth.context import request_bearer_token
 
-        token = request_bearer_token.set(workspace.bearer_token)
-        try:
-            return await self._session.invoke(forwarded)
+            auth_token = request_bearer_token.set(workspace.bearer_token)
+            try:
+                return await self._session.invoke(forwarded)
+            finally:
+                request_bearer_token.reset(auth_token)
         finally:
-            request_bearer_token.reset(token)
+            current_research_workspace.reset(workspace_token)
 
     def _with_bucket_instructions(
         self, request: AgentRequest, workspace: ResearchWorkspace

research/research_runner.py

@@ -9,8 +9,10 @@ from typing import TYPE_CHECKING
 from fast_agent import AgentAuth, AgentRequest, AppOpenRequest
 from fast_agent.llm.request_params import RequestParams
 
+from .activity_narrator import ActivityNarrator, current_activity_narrator
+from .app_auth import effective_agent_auth
 from .app_artifacts import finalize_bucket_html
-from .app_jobs import ResearchJob
+from .app_jobs import ResearchJob, current_research_job
 from .app_observability import JobProgressHandler, try_export_trace
 
 if TYPE_CHECKING:
@@ -30,27 +32,50 @@ class ResearchRunner:
         auth: AgentAuth | None,
     ) -> str:
         """The essential Harness API flow used by this example."""
-        with self.harness.request_context(auth=auth):
-            async with self.harness.app().open(
-                AppOpenRequest(
-                    session_id=job.id,
-                    agent="research",
-                    metadata={"job_id": job.id},
+        auth = effective_agent_auth(auth)
+
+        async def summarize_activity(prompt: str) -> str:
+            response = await self.harness.invoke(
+                AgentRequest.text(
+                    prompt,
+                    agent="activity-summarizer",
+                    session_id=f"{job.id}-activity",
+                    auth=auth,
+                    metadata={"job_id": job.id, "activity_narrator": True},
                 )
-            ) as session:
-                response = await session.invoke(
-                    AgentRequest.text(
-                        job.topic,
-                        agent="research",
+            )
+            return response.text_content()
+
+        narrator = ActivityNarrator(job, summarize_activity)
+        with self.harness.request_context(auth=auth):
+            await narrator.start()
+            token = current_activity_narrator.set(narrator)
+            job_token = current_research_job.set(job)
+            try:
+                async with self.harness.app().open(
+                    AppOpenRequest(
                         session_id=job.id,
-                        auth=auth,
-                        params=RequestParams(
-                            tool_execution_handler=JobProgressHandler(job),
-                            emit_loop_progress=True,
-                        ),
+                        agent="research",
                         metadata={"job_id": job.id},
                     )
-                )
+                ) as session:
+                    response = await session.invoke(
+                        AgentRequest.text(
+                            job.topic,
+                            agent="research",
+                            session_id=job.id,
+                            auth=auth,
+                            params=RequestParams(
+                                tool_execution_handler=JobProgressHandler(job),
+                                emit_loop_progress=True,
+                            ),
+                            metadata={"job_id": job.id},
+                        )
+                    )
+            finally:
+                current_research_job.reset(job_token)
+                current_activity_narrator.reset(token)
+                await narrator.close()
         return response.text_content()
 
     async def run(
@@ -66,12 +91,25 @@ class ResearchRunner:
             await self._finalize_artifacts(job, auth)
             await try_export_trace(job, self.home)
             job.status = "completed"
+            job.phase = "completed"
             job.add_event("Research completed")
+        except asyncio.CancelledError:
+            job.result = None
+            job.error = None
+            job.status = "cancelled"
+            job.phase = "cancelled"
+            job.set_activity_summary(
+                "Research was cancelled. Partial notes and the session trace were kept."
+            )
+            job.add_event("Research cancelled")
+            await try_export_trace(job, self.home)
+            raise
         except Exception as exc:
             job.error = str(exc)
             job.add_event(f"Research failed: {exc}", kind="error")
             await try_export_trace(job, self.home)
             job.status = "failed"
+            job.phase = "failed"
             job.add_event("Research job closed after failure", kind="error")
 
     async def _finalize_artifacts(
@@ -86,15 +124,14 @@ class ResearchRunner:
                 auth,
                 self.home,
             )
-            if urls and job.result and urls[0] not in job.result:
-                job.result += (
-                    f"\n\n**Final HTML artifact:**\n- `{urls[0]}`\n- {urls[1]}"
-                )
+            if urls:
+                job.html_report_uri, job.html_report_url = urls
+                if job.result and urls[0] not in job.result:
+                    job.result += (
+                        f"\n\n**Final HTML artifact:**\n- `{urls[0]}`\n- {urls[1]}"
+                    )
         except Exception as exc:
             warning = f"HTML artifact validation failed: {exc}"
-            job.add_event(
-                warning,
-                kind="artifact",
-            )
+            job.add_event(warning, kind="artifact")
             if job.result:
                 job.result += f"\n\n> Warning: {warning}"

research/research_workspace.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import json
 import os
 import re
+from contextvars import ContextVar
 from dataclasses import dataclass
 from datetime import UTC, datetime
 from typing import Any, Mapping
@@ -32,6 +33,12 @@ class ResearchWorkspace:
     bearer_token: str | None
 
 
+current_research_workspace: ContextVar[ResearchWorkspace | None] = ContextVar(
+    "current_research_workspace",
+    default=None,
+)
+
+
 def ensure_workspace(
     *,
     auth: AgentAuth | None,