Deploy 928494e archive cache and packaging fix

Source commit: 928494e329ae12b309700a3b05a7037f15eb0037

research/__init__.py

@@ -0,0 +1 @@
+"""Research agent fast-agent home support package."""

research/app_jobs.py

@@ -0,0 +1,423 @@
+"""In-memory research job lifecycle with bounded, caller-scoped retention."""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+from collections.abc import Callable, Coroutine
+from contextvars import ContextVar
+from dataclasses import dataclass, field
+from time import time
+from typing import Any
+from uuid import uuid4
+
+from fast_agent import AgentAuth
+
+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:
+    """Return a non-secret identity suitable for authorizing app backend calls."""
+    if auth is not None:
+        if auth.subject:
+            return f"{auth.provider or 'unknown'}:{auth.subject}"
+        if auth.token:
+            digest = hashlib.sha256(auth.token.encode()).hexdigest()
+            return f"token:{digest}"
+    return f"session:{session_id or 'unknown'}"
+
+
+def format_elapsed(seconds: float) -> str:
+    total = max(0, int(seconds))
+    minutes, secs = divmod(total, 60)
+    hours, minutes = divmod(minutes, 60)
+    if hours:
+        return f"{hours:d}:{minutes:02d}:{secs:02d}"
+    return f"{minutes:02d}:{secs:02d}"
+
+
+def display_activity_source(source: str | None) -> str:
+    return (source or "research/agent_loop").replace("/", " / ")
+
+
+@dataclass(slots=True)
+class ResearchJob:
+    id: str
+    topic: str
+    owner_id: str
+    headline: str = "Briefing the researcher"
+    workspace_id: str | None = None
+    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
+    archive_space_url: str | None = None
+    archive_app_url: str | None = None
+    archive_template_version: 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 = "Briefing the researcher"
+    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
+
+    @property
+    def artifact_id(self) -> str:
+        return self.workspace_id or self.id
+
+    @property
+    def harness_session_id(self) -> str:
+        """Keep model persistence separate from job and artifact identities."""
+        return f"{self.id}-research"
+
+    def add_event(
+        self,
+        message: str,
+        *,
+        kind: str = "status",
+        progress: float | None = None,
+        total: float | None = None,
+        now: float | None = None,
+    ) -> None:
+        self.updated_at = time() if now is None else now
+        self.events.append(
+            {
+                "ts": self.updated_at,
+                "kind": kind,
+                "message": message,
+                "progress": progress,
+                "total": total,
+            }
+        )
+        self.event_count_total += 1
+        del self.events[:-100]
+
+    def snapshot(self, *, now: float | None = None) -> dict[str, Any]:
+        current = time() if now is None else now
+        done = self.status in TERMINAL_STATUSES
+        elapsed_seconds = (self.updated_at if done else current) - self.created_at
+        events = [
+            {
+                **event,
+                "elapsed": format_elapsed(
+                    float(event.get("ts") or self.created_at) - self.created_at
+                ),
+            }
+            for event in self.events
+        ]
+        summaries = [
+            {
+                **summary,
+                "source_label": display_activity_source(summary.get("source")),
+                "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,
+            "headline": self.headline,
+            "workspace_id": self.workspace_id,
+            "status": self.status,
+            "phase": self.phase,
+            "events": events,
+            "timeline_events": events[-12:],
+            "recent_events": events[-2:],
+            "activity_roll": list(
+                reversed(
+                    [event for event in events if event["kind"] == "Activity"][-6:]
+                )
+            ),
+            "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,
+            "activity_source_label": display_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,
+            "archive_space_url": self.archive_space_url,
+            "archive_app_url": self.archive_app_url,
+            "archive_template_version": self.archive_template_version,
+            "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,
+                "source": self.activity_source,
+            }
+        )
+        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:
+    job: ResearchJob
+    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,
+        *,
+        completed_ttl: float = 24 * 60 * 60,
+        queued_ttl: float = 60 * 60,
+        max_jobs: int = 500,
+        clock: Callable[[], float] = time,
+    ) -> None:
+        self._jobs: dict[str, ResearchJob] = {}
+        self._lock = asyncio.Lock()
+        self._completed_ttl = completed_ttl
+        self._queued_ttl = queued_ttl
+        self._max_jobs = max_jobs
+        self._clock = clock
+
+    async def create(
+        self,
+        topic: str,
+        owner: str,
+    ) -> ResearchJob:
+        async with self._lock:
+            now = self._clock()
+            self._prune(now)
+            job = ResearchJob(
+                id=f"research-{uuid4().hex[:12]}",
+                topic=topic,
+                owner_id=owner,
+                created_at=now,
+                updated_at=now,
+            )
+            job.add_event("Your research request is ready.", kind="Setup", now=now)
+            self._jobs[job.id] = job
+            self._enforce_limit()
+            return job
+
+    async def begin(self, job_id: str, owner: str) -> BeginResult | None:
+        """Atomically claim a queued 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":
+                return BeginResult(job=job, started=False)
+            job.status = "running"
+            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:
+        async with self._lock:
+            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
+
+    def _prune(self, now: float) -> None:
+        expired = [
+            job_id
+            for job_id, job in self._jobs.items()
+            if (
+                job.status in TERMINAL_STATUSES
+                and now - job.updated_at >= self._completed_ttl
+            )
+            or (job.status == "queued" and now - job.updated_at >= self._queued_ttl)
+        ]
+        for job_id in expired:
+            self._jobs.pop(job_id)
+
+    def _enforce_limit(self) -> None:
+        excess = len(self._jobs) - self._max_jobs
+        if excess <= 0:
+            return
+        evictable = sorted(
+            (
+                job
+                for job in self._jobs.values()
+                if job.status == "queued" or job.status in TERMINAL_STATUSES
+            ),
+            key=lambda job: job.updated_at,
+        )
+        for job in evictable[:excess]:
+            self._jobs.pop(job.id, None)
+
+
+def unavailable_snapshot(job_id: str) -> dict[str, Any]:
+    """Safe response for expired, restarted, or unauthorized historical apps."""
+    return {
+        "job_id": job_id,
+        "topic": "",
+        "headline": "Research unavailable",
+        "workspace_id": None,
+        "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",
+        "activity_source_label": "research / agent_loop",
+        "turn_count": 0,
+        "result": None,
+        "markdown_report": None,
+        "markdown_report_uri": None,
+        "markdown_report_error": None,
+        "archive_space_url": None,
+        "archive_app_url": None,
+        "archive_template_version": 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

@@ -0,0 +1,381 @@
+"""Optional timeline and trace-export hooks for research jobs."""
+
+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):
+    """Project fast-agent tool events into the app's timeline."""
+
+    def __init__(self, job: ResearchJob) -> None:
+        self.job = job
+        self._activities: dict[str, tuple[str, str, str, str]] = {}
+
+    async def on_tool_start(
+        self,
+        tool_name: str,
+        server_name: str,
+        arguments: dict | None,
+        tool_use_id: str | None = None,
+    ) -> str:
+        tool_call_id = tool_use_id or f"{server_name}/{tool_name}/{uuid4().hex[:8]}"
+        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(
+        self,
+        tool_call_id: str,
+        progress: float,
+        total: float | None,
+        message: str | None,
+    ) -> None:
+        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,
+        tool_call_id: str,
+        success: bool,
+        content: list[Any] | None,
+        error: str | None,
+    ) -> None:
+        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,
+        tool_name: str,
+        server_name: str,
+        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(
+            _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._activities else None
+
+    async def ensure_tool_call_exists(
+        self,
+        tool_use_id: str,
+        tool_name: str,
+        server_name: str,
+        arguments: dict | None = None,
+    ) -> str:
+        if tool_use_id in self._activities:
+            return tool_use_id
+        return await self.on_tool_start(
+            tool_name,
+            server_name,
+            arguments,
+            tool_use_id,
+        )
+
+
+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.archive_space_url = workspace.archive_space_url
+    job.archive_app_url = workspace.archive_app_url
+    job.archive_template_version = workspace.archive_installed_version
+    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
+        / "sessions"
+        / "research-traces"
+        / job.id
+        / f"{job.id}__research__codex.jsonl"
+    )
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    exporter = SessionTraceExporter(
+        session_manager=SessionManager(home_override=home),
+        progress_callback=lambda message: job.add_event(message, kind="trace"),
+    )
+    result = exporter.export(
+        ExportRequest(
+            target=job.harness_session_id,
+            agent_name="research",
+            output_path=output_path,
+        )
+    )
+    job.trace_path = str(result.output_path)
+    job.add_event(
+        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.harness_session_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:
+    try:
+        await asyncio.to_thread(export_trace, job, home)
+    except Exception as exc:
+        job.trace_error = str(exc)
+        job.add_event(f"Trace export failed: {exc}", kind="trace")

research/archive_provisioning.py

@@ -0,0 +1,241 @@
+"""Provision a private, versioned report-browser Space for one user."""
+
+from __future__ import annotations
+
+import json
+import os
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Literal
+
+from huggingface_hub import HfApi, Volume
+from huggingface_hub.errors import RepositoryNotFoundError
+
+ARCHIVE_TEMPLATE_SPACE = os.getenv(
+    "RESEARCH_ARCHIVE_TEMPLATE_SPACE",
+    "evalstate/research-archive-template",
+)
+ARCHIVE_TEMPLATE_VERSION = "1.2.1"
+ARCHIVE_MARKER_PATH = "archive-template.json"
+ARCHIVE_SPACE_NAME = "research-agent"
+ARCHIVE_MOUNT_PATH = "/research"
+
+
+class ArchiveProvisioningError(RuntimeError):
+    """Archive Space could not be safely provisioned."""
+
+
+class ArchiveSpaceCollisionError(ArchiveProvisioningError):
+    """The desired Space name exists but is not managed by this application."""
+
+
+@dataclass(frozen=True, slots=True)
+class ArchiveProvisioning:
+    space_id: str
+    space_url: str
+    app_url: str
+    template_space: str
+    template_version: str
+    installed_version: str
+    status: Literal["ready", "provisioning", "version_mismatch"]
+    created: bool
+    volume_updated: bool
+
+
+def ensure_archive_space(
+    *,
+    username: str,
+    bucket_id: str,
+    token: str | None,
+    api: HfApi | None = None,
+    template_space: str = ARCHIVE_TEMPLATE_SPACE,
+) -> ArchiveProvisioning:
+    """Create or verify a user's managed archive Space and bucket mount."""
+    expected_bucket = f"{username}/research-agent"
+    if bucket_id != expected_bucket:
+        raise ArchiveProvisioningError(
+            f"Refusing to mount unexpected bucket {bucket_id!r}; "
+            f"expected {expected_bucket!r}."
+        )
+    if not token:
+        raise ArchiveProvisioningError(
+            "A caller token is required to provision the archive Space."
+        )
+
+    api = api or HfApi()
+    space_id = f"{username}/{ARCHIVE_SPACE_NAME}"
+    desired_volume = Volume(
+        type="bucket",
+        source=bucket_id,
+        mount_path=ARCHIVE_MOUNT_PATH,
+        read_only=False,
+    )
+    created = False
+    try:
+        info = api.space_info(space_id, token=token)
+    except RepositoryNotFoundError:
+        template = _read_marker(api, template_space, token)
+        _validate_template_marker(template, template_space)
+        api.duplicate_repo(
+            from_id=template_space,
+            to_id=space_id,
+            repo_type="space",
+            private=True,
+            exist_ok=True,
+            space_hardware="cpu-basic",
+            space_volumes=[desired_volume],
+            space_variables=[
+                {
+                    "key": "RESEARCH_ARCHIVE_MANAGED",
+                    "value": "true",
+                    "description": "Managed by the Research Agent provisioner.",
+                },
+                {
+                    "key": "RESEARCH_ARCHIVE_TEMPLATE_VERSION",
+                    "value": ARCHIVE_TEMPLATE_VERSION,
+                    "description": "Installed archive template version.",
+                },
+                {
+                    "key": "RESEARCH_ARCHIVE_BUCKET",
+                    "value": bucket_id,
+                    "description": "Mounted Research Agent bucket.",
+                },
+            ],
+            token=token,
+        )
+        created = True
+        info = api.space_info(space_id, token=token)
+
+    installed = _read_marker(api, space_id, token)
+    _validate_managed_marker(installed, space_id)
+    installed_version = str(installed["template_version"])
+
+    volume_updated = not _has_expected_volume(info, bucket_id)
+    if volume_updated:
+        api.set_space_volumes(
+            space_id,
+            [desired_volume],
+            token=token,
+        )
+
+    _ensure_variable(
+        api,
+        space_id,
+        "RESEARCH_ARCHIVE_TEMPLATE_VERSION",
+        installed_version,
+        "Installed archive template version.",
+        token,
+    )
+    _ensure_variable(
+        api,
+        space_id,
+        "RESEARCH_ARCHIVE_BUCKET",
+        bucket_id,
+        "Mounted Research Agent bucket.",
+        token,
+    )
+
+    status: Literal["ready", "provisioning", "version_mismatch"]
+    if installed_version != ARCHIVE_TEMPLATE_VERSION:
+        status = "version_mismatch"
+    elif created or volume_updated:
+        status = "provisioning"
+    else:
+        status = "ready"
+
+    return ArchiveProvisioning(
+        space_id=space_id,
+        space_url=f"https://huggingface.co/spaces/{space_id}",
+        app_url=f"https://{username}-{ARCHIVE_SPACE_NAME}.hf.space",
+        template_space=template_space,
+        template_version=ARCHIVE_TEMPLATE_VERSION,
+        installed_version=installed_version,
+        status=status,
+        created=created,
+        volume_updated=volume_updated,
+    )
+
+
+def _read_marker(api: HfApi, repo_id: str, token: str) -> dict[str, Any]:
+    try:
+        path = api.hf_hub_download(
+            repo_id,
+            ARCHIVE_MARKER_PATH,
+            repo_type="space",
+            token=token,
+            force_download=True,
+        )
+        marker = json.loads(Path(path).read_text())
+    except Exception as exc:
+        raise ArchiveProvisioningError(
+            f"Space {repo_id!r} has no readable {ARCHIVE_MARKER_PATH}."
+        ) from exc
+    if not isinstance(marker, dict):
+        raise ArchiveProvisioningError(
+            f"Space {repo_id!r} has an invalid {ARCHIVE_MARKER_PATH}."
+        )
+    return marker
+
+
+def _validate_template_marker(marker: dict[str, Any], template_space: str) -> None:
+    _validate_managed_marker(marker, template_space)
+    version = str(marker.get("template_version", ""))
+    if version != ARCHIVE_TEMPLATE_VERSION:
+        raise ArchiveProvisioningError(
+            f"Template {template_space!r} is version {version!r}; "
+            f"the provisioner expects {ARCHIVE_TEMPLATE_VERSION!r}."
+        )
+
+
+def _validate_managed_marker(marker: dict[str, Any], space_id: str) -> None:
+    if (
+        marker.get("schema_version") != 1
+        or marker.get("managed_by") != "research-agent"
+        or marker.get("template") != "research-archive"
+        or not marker.get("template_version")
+    ):
+        raise ArchiveSpaceCollisionError(
+            f"Space {space_id!r} exists but is not a managed Research Archive."
+        )
+
+
+def _has_expected_volume(info: Any, bucket_id: str) -> bool:
+    runtime = getattr(info, "runtime", None)
+    raw = getattr(runtime, "raw", None)
+    volumes = raw.get("volumes", []) if isinstance(raw, dict) else []
+    return any(
+        volume.get("type") == "bucket"
+        and volume.get("source") == bucket_id
+        and volume.get("mountPath") == ARCHIVE_MOUNT_PATH
+        and not volume.get("readOnly", False)
+        for volume in volumes
+        if isinstance(volume, dict)
+    )
+
+
+def _ensure_variable(
+    api: HfApi,
+    space_id: str,
+    key: str,
+    value: str,
+    description: str,
+    token: str,
+) -> None:
+    variables = api.get_space_variables(space_id, token=token)
+    current = variables.get(key)
+    current_value = (
+        current.value
+        if hasattr(current, "value")
+        else current.get("value")
+        if isinstance(current, dict)
+        else None
+    )
+    if current_value == value:
+        return
+    api.add_space_variable(
+        space_id,
+        key,
+        value,
+        description=description,
+        token=token,
+    )

research/research_workspace.py

@@ -0,0 +1,221 @@
+"""Resolve and prepare per-user Hugging Face bucket workspaces."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import re
+from contextvars import ContextVar
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from typing import Any, Mapping
+from uuid import uuid4
+
+from huggingface_hub import HfApi, get_token
+from huggingface_hub.errors import BucketNotFoundError
+
+from fast_agent import AgentAuth
+from fast_agent.mcp.server.common import normalize_serve_oauth_provider
+
+
+_SAFE_SEGMENT = re.compile(r"[^A-Za-z0-9._-]+")
+
+
+@dataclass(frozen=True, slots=True)
+class ResearchWorkspace:
+    username: str
+    session_id: str
+    bucket_id: str
+    root: str
+    scratch: str
+    output: str
+    bucket_created: bool
+    marker_paths: tuple[str, ...]
+    bearer_token: str | None
+    archive_space_id: str | None = None
+    archive_space_url: str | None = None
+    archive_app_url: str | None = None
+    archive_status: str | None = None
+    archive_template_version: str | None = None
+    archive_installed_version: str | None = None
+    archive_error: str | None = None
+
+
+current_research_workspace: ContextVar[ResearchWorkspace | None] = ContextVar(
+    "current_research_workspace",
+    default=None,
+)
+
+
+def ensure_workspace(
+    *,
+    auth: AgentAuth | None,
+    request_metadata: Mapping[str, Any],
+    open_metadata: Mapping[str, object],
+    create_bucket: bool = True,
+    write_markers: bool = True,
+    api: HfApi | None = None,
+) -> ResearchWorkspace:
+    """Resolve identity/session, ensure the bucket exists, and write markers."""
+    token = _token(auth)
+    whoami = _whoami(auth, token)
+    username = _username(whoami)
+    session_id = _session_id(request_metadata, open_metadata)
+    bucket_id = f"{username}/research-agent"
+    root = f"hf://buckets/{bucket_id}/{session_id}/"
+
+    api = api or HfApi()
+    bucket_created = False
+    try:
+        api.bucket_info(bucket_id, token=token)
+    except BucketNotFoundError as exc:
+        if not create_bucket:
+            raise RuntimeError(
+                f"Bucket {bucket_id!r} is not accessible: {exc}"
+            ) from exc
+        try:
+            api.create_bucket(bucket_id, private=True, exist_ok=True, token=token)
+            bucket_created = True
+        except Exception as create_exc:
+            raise RuntimeError(
+                f"Could not create/access bucket {bucket_id!r}: {create_exc}"
+            ) from create_exc
+
+    marker_paths: tuple[str, ...] = ()
+    if write_markers:
+        marker = {
+            "server": "research-agent",
+            "username": username,
+            "session_id": session_id,
+            "bucket_id": bucket_id,
+            "checked_at": datetime.now(UTC).isoformat(),
+        }
+        try:
+            api.batch_bucket_files(
+                bucket_id,
+                add=[
+                    (
+                        json.dumps(marker, indent=2).encode("utf-8"),
+                        f"{session_id}/scratch/.workspace.json",
+                    ),
+                    (b"", f"{session_id}/output/.keep"),
+                ],
+                token=token,
+            )
+        except Exception as exc:
+            raise RuntimeError(
+                f"Bucket {bucket_id!r} is accessible but marker write failed: {exc}"
+            ) from exc
+        marker_paths = (
+            f"{root}scratch/.workspace.json",
+            f"{root}output/.keep",
+        )
+
+    return ResearchWorkspace(
+        username=username,
+        session_id=session_id,
+        bucket_id=bucket_id,
+        root=root,
+        scratch=f"{root}scratch/",
+        output=f"{root}output/",
+        bucket_created=bucket_created,
+        marker_paths=marker_paths,
+        bearer_token=token,
+    )
+
+
+def _token(auth: AgentAuth | None) -> str | None:
+    if auth is not None and auth.token:
+        return auth.token
+    oauth_provider = normalize_serve_oauth_provider(os.getenv("FAST_AGENT_SERVE_OAUTH"))
+    if oauth_provider == "huggingface":
+        raise RuntimeError(
+            "Hugging Face OAuth is enabled, but this request has no caller token."
+        )
+    env_token = os.getenv("HF_TOKEN")
+    if env_token:
+        return env_token
+    return get_token()
+
+
+def _whoami(auth: AgentAuth | None, token: str | bool | None) -> Mapping[str, Any]:
+    """Return the authoritative Hugging Face whoami payload for this caller."""
+    claims = dict(auth.claims) if auth is not None else {}
+    whoami = claims.get("huggingface_whoami")
+    if isinstance(whoami, dict) and whoami:
+        return whoami
+
+    try:
+        return HfApi().whoami(token=token)
+    except Exception as exc:
+        raise RuntimeError(
+            "Could not determine the Hugging Face user. Provide a bearer token, "
+            "enable Hugging Face OAuth, set HF_TOKEN, or run `hf auth login`."
+        ) from exc
+
+
+def _username(whoami: Mapping[str, Any]) -> str:
+    username = safe_segment(whoami.get("name"))
+    if username:
+        return username
+    raise RuntimeError(
+        f"Hugging Face whoami response did not include a usable name: {dict(whoami)!r}."
+    )
+
+
+def _session_id(
+    request_metadata: Mapping[str, Any],
+    open_metadata: Mapping[str, object],
+) -> str:
+    candidates = [
+        request_metadata.get("research_workspace_id"),
+        open_metadata.get("research_workspace_id"),
+        request_metadata.get("request_session_id"),
+        request_metadata.get("harness_session_id"),
+        request_metadata.get("requested_session_id"),
+        request_metadata.get("mcp_session_id"),
+        open_metadata.get("harness_session_id"),
+        open_metadata.get("requested_session_id"),
+        open_metadata.get("mcp_session_id"),
+    ]
+    for candidate in candidates:
+        value = _safe_session_segment(candidate)
+        if value:
+            return value
+    # No usable session identity was supplied. Never fall back to a shared
+    # constant ("default") — concurrent runs would collide on one bucket path
+    # and leak one run's report into another's UI. Mint a unique id instead.
+    return f"session-{uuid4().hex}"
+
+
+def _safe_session_segment(value: object) -> str | None:
+    """Sanitize a session id, keeping distinct inputs on distinct segments.
+
+    ``safe_segment`` truncates to 96 chars and maps disallowed characters to
+    ``-``, so two different client-supplied ids can collapse to the same
+    segment. When sanitization loses information, append a short stable hash of
+    the original so the mapping stays collision-resistant (and deterministic, so
+    the same input still resolves to the same workspace across requests).
+    """
+    if value is None:
+        return None
+    raw = str(value).strip().strip("/")
+    if not raw:
+        return None
+    safe = safe_segment(raw)
+    if safe is None:
+        return None
+    if safe != raw:
+        digest = hashlib.sha256(raw.encode("utf-8")).hexdigest()[:8]
+        safe = f"{safe[:87].rstrip('.-_')}-{digest}"
+    return safe
+
+
+def safe_segment(value: object) -> str | None:
+    if value is None:
+        return None
+    text = str(value).strip().strip("/")
+    if not text:
+        return None
+    return _SAFE_SEGMENT.sub("-", text)[:96].strip(".-_") or None