feat(observability)

agent/config.py

@@ -24,8 +24,13 @@ class Config(BaseModel):
     model_name: str
     mcpServers: dict[str, MCPServerConfig] = {}
     save_sessions: bool = True
-    session_dataset_repo: str = "akseljoonas/hf-agent-sessions"
-    auto_save_interval: int = 3  # Save every N user turns (0 = disabled)
+    session_dataset_repo: str = "smolagents/ml-intern-sessions"
+    auto_save_interval: int = 1  # Save every N user turns (0 = disabled)
+    # Mid-turn heartbeat: save + upload every N seconds while events are being
+    # emitted. Guards against losing trace data on long-running turns that
+    # crash before turn_complete (e.g. a multi-hour hf_jobs wait that OOMs).
+    # 0 = disabled. Consumed by agent.core.telemetry.HeartbeatSaver.
+    heartbeat_interval_s: int = 60
     yolo_mode: bool = False  # Auto-approve all tool calls without confirmation
     max_iterations: int = 300  # Max LLM calls per agent turn (-1 = unlimited)
 

agent/core/agent_loop.py

@@ -6,12 +6,14 @@ import asyncio
 import json
 import logging
 import os
-from dataclasses import dataclass
+import time
+from dataclasses import dataclass, field
 
 from litellm import ChatCompletionMessageToolCall, Message, acompletion
 from litellm.exceptions import ContextWindowExceededError
 
 from agent.config import Config
+from agent.core import telemetry
 from agent.core.doom_loop import check_for_doom_loop
 from agent.core.llm_params import _resolve_llm_params
 from agent.core.prompt_caching import with_prompt_caching
@@ -291,6 +293,7 @@ class LLMResult:
     tool_calls_acc: dict[int, dict]
     token_count: int
     finish_reason: str | None
+    usage: dict = field(default_factory=dict)
 
 
 async def _call_llm_streaming(session: Session, messages, tools, llm_params) -> LLMResult:
@@ -298,6 +301,7 @@ async def _call_llm_streaming(session: Session, messages, tools, llm_params) ->
     response = None
     _healed_effort = False  # one-shot safety net per call
     messages, tools = with_prompt_caching(messages, tools, llm_params.get("model"))
+    t_start = time.monotonic()
     for _llm_attempt in range(_MAX_LLM_RETRIES):
         try:
             response = await acompletion(
@@ -339,6 +343,7 @@ async def _call_llm_streaming(session: Session, messages, tools, llm_params) ->
     tool_calls_acc: dict[int, dict] = {}
     token_count = 0
     finish_reason = None
+    final_usage_chunk = None
 
     async for chunk in response:
         if session.is_cancelled:
@@ -349,6 +354,7 @@ async def _call_llm_streaming(session: Session, messages, tools, llm_params) ->
         if not choice:
             if hasattr(chunk, "usage") and chunk.usage:
                 token_count = chunk.usage.total_tokens
+                final_usage_chunk = chunk
             continue
 
         delta = choice.delta
@@ -379,12 +385,22 @@ async def _call_llm_streaming(session: Session, messages, tools, llm_params) ->
 
         if hasattr(chunk, "usage") and chunk.usage:
             token_count = chunk.usage.total_tokens
+            final_usage_chunk = chunk
+
+    usage = await telemetry.record_llm_call(
+        session,
+        model=llm_params.get("model", session.config.model_name),
+        response=final_usage_chunk,
+        latency_ms=int((time.monotonic() - t_start) * 1000),
+        finish_reason=finish_reason,
+    )
 
     return LLMResult(
         content=full_content or None,
         tool_calls_acc=tool_calls_acc,
         token_count=token_count,
         finish_reason=finish_reason,
+        usage=usage,
     )
 
 
@@ -393,6 +409,7 @@ async def _call_llm_non_streaming(session: Session, messages, tools, llm_params)
     response = None
     _healed_effort = False
     messages, tools = with_prompt_caching(messages, tools, llm_params.get("model"))
+    t_start = time.monotonic()
     for _llm_attempt in range(_MAX_LLM_RETRIES):
         try:
             response = await acompletion(
@@ -454,11 +471,20 @@ async def _call_llm_non_streaming(session: Session, messages, tools, llm_params)
             Event(event_type="assistant_message", data={"content": content})
         )
 
+    usage = await telemetry.record_llm_call(
+        session,
+        model=llm_params.get("model", session.config.model_name),
+        response=response,
+        latency_ms=int((time.monotonic() - t_start) * 1000),
+        finish_reason=finish_reason,
+    )
+
     return LLMResult(
         content=content,
         tool_calls_acc=tool_calls_acc,
         token_count=token_count,
         finish_reason=finish_reason,
+        usage=usage,
     )
 
 

agent/core/redact.py

@@ -0,0 +1,68 @@
+"""Secret scrubbing for session trajectories before upload.
+
+Users frequently paste HF / API / GitHub tokens into the chat, or scripts echo
+them via env dumps. This module applies regex-based redaction to any string
+value found recursively in a trajectory payload. The goal is best-effort —
+strict formats are matched; we won't catch free-form leaks like "my password
+is hunter2".
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+# Each entry: (compiled regex, replacement placeholder).
+# Patterns are conservative: they only match tokens with the canonical prefix
+# and a minimum body length so we don't paint over normal text.
+_PATTERNS: list[tuple[re.Pattern, str]] = [
+    # Hugging Face tokens: hf_[A-Za-z0-9]{30,}
+    (re.compile(r"hf_[A-Za-z0-9]{30,}"), "[REDACTED_HF_TOKEN]"),
+    # Anthropic: sk-ant-[A-Za-z0-9_\-]{20,}
+    (re.compile(r"sk-ant-[A-Za-z0-9_\-]{20,}"), "[REDACTED_ANTHROPIC_KEY]"),
+    # OpenAI: sk-[A-Za-z0-9]{40,}  (legacy + proj keys)
+    (re.compile(r"sk-(?!ant-)[A-Za-z0-9_\-]{40,}"), "[REDACTED_OPENAI_KEY]"),
+    # GitHub classic PATs: ghp_, gho_, ghu_, ghs_, ghr_ followed by 36+ chars
+    (re.compile(r"gh[pousr]_[A-Za-z0-9]{36,}"), "[REDACTED_GITHUB_TOKEN]"),
+    # GitHub fine-grained PATs: github_pat_<alphanumeric_underscore>
+    (re.compile(r"github_pat_[A-Za-z0-9_]{36,}"), "[REDACTED_GITHUB_TOKEN]"),
+    # AWS access key IDs: AKIA / ASIA + 16 uppercase alnum
+    (re.compile(r"\b(?:AKIA|ASIA)[A-Z0-9]{16}\b"), "[REDACTED_AWS_KEY_ID]"),
+    # Generic 'Bearer <token>' header values
+    (re.compile(r"(?i)bearer\s+[A-Za-z0-9_\-\.=]{20,}"), "Bearer [REDACTED]"),
+]
+
+# Env-var-like exports: we scrub the value but keep the name so callers can
+# still see which secret was referenced. Covers `KEY=value` and `KEY: value`
+# when the key looks secret-y.
+_SECRETY_NAMES = re.compile(
+    r"(?i)\b(HF_TOKEN|HUGGINGFACEHUB_API_TOKEN|ANTHROPIC_API_KEY|OPENAI_API_KEY|"
+    r"GITHUB_TOKEN|AWS_SECRET_ACCESS_KEY|AWS_ACCESS_KEY_ID|PASSWORD|SECRET|API_KEY)"
+    r"\s*[:=]\s*([^\s\"']+)"
+)
+
+
+def scrub_string(s: str) -> str:
+    """Apply all redaction patterns to a single string. Safe on non-strings."""
+    if not isinstance(s, str) or not s:
+        return s
+    out = s
+    for pat, repl in _PATTERNS:
+        out = pat.sub(repl, out)
+    out = _SECRETY_NAMES.sub(lambda m: f"{m.group(1)}=[REDACTED]", out)
+    return out
+
+
+def scrub(obj: Any) -> Any:
+    """Recursively scrub every string value in a nested dict/list structure.
+
+    Returns a new object — inputs are not mutated."""
+    if isinstance(obj, str):
+        return scrub_string(obj)
+    if isinstance(obj, dict):
+        return {k: scrub(v) for k, v in obj.items()}
+    if isinstance(obj, list):
+        return [scrub(v) for v in obj]
+    if isinstance(obj, tuple):
+        return tuple(scrub(v) for v in obj)
+    return obj

agent/core/session.py

@@ -108,6 +108,11 @@ class Session:
         self.session_start_time = datetime.now().isoformat()
         self.turn_count: int = 0
         self.last_auto_save_turn: int = 0
+        # Stable local save path so heartbeat saves overwrite one file instead
+        # of spamming session_logs/. ``_last_heartbeat_ts`` is owned by
+        # ``agent.core.telemetry.HeartbeatSaver`` and lazily initialised there.
+        self._local_save_path: Optional[str] = None
+        self._last_heartbeat_ts: Optional[float] = None
 
         # Per-model probed reasoning-effort cache. Populated by the probe
         # on /model switch, read by ``effective_effort_for`` below. Keys are
@@ -132,6 +137,10 @@ class Session:
             }
         )
 
+        # Mid-turn heartbeat flush (owned by telemetry module).
+        from agent.core.telemetry import HeartbeatSaver
+        HeartbeatSaver.maybe_fire(self)
+
     def cancel(self) -> None:
         """Signal cancellation to the running agent loop."""
         self._cancelled.set()
@@ -184,6 +193,12 @@ class Session:
 
     def get_trajectory(self) -> dict:
         """Serialize complete session trajectory for logging"""
+        tools: list = []
+        if self.tool_router is not None:
+            try:
+                tools = self.tool_router.get_tool_specs_for_llm() or []
+            except Exception:
+                tools = []
         return {
             "session_id": self.session_id,
             "session_start_time": self.session_start_time,
@@ -191,6 +206,7 @@ class Session:
             "model_name": self.config.model_name,
             "messages": [msg.model_dump() for msg in self.context_manager.items],
             "events": self.logged_events,
+            "tools": tools,
         }
 
     def save_trajectory_local(
@@ -216,16 +232,42 @@ class Session:
 
             trajectory = self.get_trajectory()
 
+            # Scrub secrets at save time so session_logs/ never holds raw
+            # tokens on disk — a log aggregator, crash dump, or filesystem
+            # snapshot between heartbeats would otherwise leak them.
+            try:
+                from agent.core.redact import scrub
+                for key in ("messages", "events", "tools"):
+                    if key in trajectory:
+                        trajectory[key] = scrub(trajectory[key])
+            except Exception as _e:
+                logger.debug("Redact-on-save failed (non-fatal): %s", _e)
+
             # Add upload metadata
             trajectory["upload_status"] = upload_status
             trajectory["upload_url"] = dataset_url
             trajectory["last_save_time"] = datetime.now().isoformat()
 
-            filename = f"session_{self.session_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
-            filepath = log_dir / filename
-
-            with open(filepath, "w") as f:
+            # Reuse one stable path per session so heartbeat saves overwrite
+            # the same file instead of creating a new timestamped file every
+            # minute. The timestamp in the filename is kept for first-save
+            # ordering; subsequent saves just rewrite that file.
+            if self._local_save_path and Path(self._local_save_path).parent == log_dir:
+                filepath = Path(self._local_save_path)
+            else:
+                filename = (
+                    f"session_{self.session_id}_"
+                    f"{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
+                )
+                filepath = log_dir / filename
+                self._local_save_path = str(filepath)
+
+            # Atomic-ish write: stage to .tmp then rename so a crash mid-write
+            # doesn't leave a truncated JSON that breaks the retry scanner.
+            tmp_path = filepath.with_suffix(filepath.suffix + ".tmp")
+            with open(tmp_path, "w") as f:
                 json.dump(trajectory, f, indent=2)
+            tmp_path.replace(filepath)
 
             return str(filepath)
         except Exception as e:

agent/core/session_uploader.py

@@ -15,8 +15,15 @@ from dotenv import load_dotenv
 
 load_dotenv()
 
-# Token for session uploads — loaded from env var (never hardcode tokens in source)
-_SESSION_TOKEN = os.environ.get("HF_SESSION_UPLOAD_TOKEN", "")
+# Token for session uploads. Fallback chain (least-privilege first) — matches
+# backend/kpis_scheduler.py so one write-scoped token on the Space covers every
+# telemetry dataset. Never hardcode tokens in source.
+_SESSION_TOKEN = (
+    os.environ.get("HF_SESSION_UPLOAD_TOKEN")
+    or os.environ.get("HF_TOKEN")
+    or os.environ.get("HF_ADMIN_TOKEN")
+    or ""
+)
 
 
 def upload_session_as_file(
@@ -58,15 +65,37 @@ def upload_session_as_file(
                 json.dump(data, f, indent=2)
             return False
 
+        # Scrub secrets (HF tokens, API keys, etc.) from messages + events
+        # before they leave the local disk. Best-effort regex-based redaction —
+        # see agent/core/redact.py for the patterns covered.
+        try:
+            from agent.core.redact import scrub  # type: ignore
+        except Exception:
+            # Fallback for environments where the agent package isn't importable
+            # (shouldn't happen in our subprocess, but be defensive).
+            import importlib.util
+            _spec = importlib.util.spec_from_file_location(
+                "_redact",
+                Path(__file__).parent / "redact.py",
+            )
+            _mod = importlib.util.module_from_spec(_spec)
+            _spec.loader.exec_module(_mod)  # type: ignore
+            scrub = _mod.scrub
+        scrubbed_messages = scrub(data["messages"])
+        scrubbed_events = scrub(data["events"])
+        scrubbed_tools = scrub(data.get("tools") or [])
+
         # Prepare JSONL content (single line)
-        # Store messages and events as JSON strings to avoid schema conflicts
+        # Store messages/events/tools as JSON strings to avoid schema conflicts
+        # across sessions with different tool rosters.
         session_row = {
             "session_id": data["session_id"],
             "session_start_time": data["session_start_time"],
             "session_end_time": data["session_end_time"],
             "model_name": data["model_name"],
-            "messages": json.dumps(data["messages"]),
-            "events": json.dumps(data["events"]),
+            "messages": json.dumps(scrubbed_messages),
+            "events": json.dumps(scrubbed_events),
+            "tools": json.dumps(scrubbed_tools),
         }
 
         # Create temporary JSONL file

agent/core/telemetry.py

@@ -0,0 +1,289 @@
+"""All agent observability in one module.
+
+Every telemetry signal the agent emits — LLM-call usage / cost, hf_jobs
+lifecycle, sandbox lifecycle, user feedback, mid-turn heartbeat saves — is
+defined here so business-logic files stay free of instrumentation noise.
+
+Callsites are one-liners::
+
+    await telemetry.record_llm_call(session, model=..., response=r, ...)
+    await telemetry.record_hf_job_submit(session, job, args, image=..., job_type="Python")
+    HeartbeatSaver.maybe_fire(session)
+
+All ``record_*`` functions emit a single ``Event`` via ``session.send_event``
+and never raise — telemetry is best-effort and must not break the agent.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+# ── usage extraction ────────────────────────────────────────────────────────
+
+def extract_usage(response_or_chunk: Any) -> dict:
+    """Flat usage dict from a litellm response or final-chunk usage object.
+
+    Normalizes across providers: Anthropic exposes cache tokens as
+    ``cache_read_input_tokens`` / ``cache_creation_input_tokens``; OpenAI uses
+    ``prompt_tokens_details.cached_tokens``. Exposed under the stable keys
+    ``cache_read_tokens`` / ``cache_creation_tokens``.
+    """
+    u = getattr(response_or_chunk, "usage", None)
+    if u is None and isinstance(response_or_chunk, dict):
+        u = response_or_chunk.get("usage")
+    if u is None:
+        return {}
+
+    def _g(name, default=0):
+        if isinstance(u, dict):
+            return u.get(name, default) or default
+        return getattr(u, name, default) or default
+
+    prompt = _g("prompt_tokens")
+    completion = _g("completion_tokens")
+    total = _g("total_tokens") or (prompt + completion)
+
+    cache_read = _g("cache_read_input_tokens")
+    cache_creation = _g("cache_creation_input_tokens")
+
+    if not cache_read:
+        details = _g("prompt_tokens_details", None)
+        if details is not None:
+            if isinstance(details, dict):
+                cache_read = details.get("cached_tokens", 0) or 0
+            else:
+                cache_read = getattr(details, "cached_tokens", 0) or 0
+
+    return {
+        "prompt_tokens": int(prompt),
+        "completion_tokens": int(completion),
+        "total_tokens": int(total),
+        "cache_read_tokens": int(cache_read),
+        "cache_creation_tokens": int(cache_creation),
+    }
+
+
+# ── llm_call ────────────────────────────────────────────────────────────────
+
+async def record_llm_call(
+    session: Any,
+    *,
+    model: str,
+    response: Any = None,
+    latency_ms: int,
+    finish_reason: str | None,
+) -> dict:
+    """Emit an ``llm_call`` event and return the extracted usage dict so
+    callers can stash it on their result object if they want."""
+    usage = extract_usage(response) if response is not None else {}
+    cost_usd = 0.0
+    if response is not None:
+        try:
+            from litellm import completion_cost
+            cost_usd = float(completion_cost(completion_response=response) or 0.0)
+        except Exception:
+            cost_usd = 0.0
+    from agent.core.session import Event  # local import to avoid cycle
+    try:
+        await session.send_event(Event(
+            event_type="llm_call",
+            data={
+                "model": model,
+                "latency_ms": latency_ms,
+                "finish_reason": finish_reason,
+                "cost_usd": cost_usd,
+                **usage,
+            },
+        ))
+    except Exception as e:
+        logger.debug("record_llm_call failed (non-fatal): %s", e)
+    return usage
+
+
+# ── hf_jobs ────────────────────────────────────────────────────────────────
+
+def _infer_push_to_hub(script_or_cmd: Any) -> bool:
+    if not isinstance(script_or_cmd, str):
+        return False
+    return (
+        "push_to_hub=True" in script_or_cmd
+        or "push_to_hub=true" in script_or_cmd
+        or "hub_model_id" in script_or_cmd
+    )
+
+
+async def record_hf_job_submit(
+    session: Any,
+    job: Any,
+    args: dict,
+    *,
+    image: str,
+    job_type: str,
+) -> float:
+    """Emit ``hf_job_submit``. Returns the monotonic start timestamp so the
+    caller can pass it back into :func:`record_hf_job_complete`."""
+    from agent.core.session import Event
+    t_start = time.monotonic()
+    try:
+        script_text = args.get("script") or args.get("command") or ""
+        await session.send_event(Event(
+            event_type="hf_job_submit",
+            data={
+                "job_id": getattr(job, "id", None),
+                "job_url": getattr(job, "url", None),
+                "flavor": args.get("hardware_flavor", "cpu-basic"),
+                "timeout": args.get("timeout", "30m"),
+                "job_type": job_type,
+                "image": image,
+                "push_to_hub": _infer_push_to_hub(script_text),
+            },
+        ))
+    except Exception as e:
+        logger.debug("record_hf_job_submit failed (non-fatal): %s", e)
+    return t_start
+
+
+async def record_hf_job_complete(
+    session: Any,
+    job: Any,
+    *,
+    flavor: str,
+    final_status: str,
+    submit_ts: float,
+) -> None:
+    from agent.core.session import Event
+    try:
+        wall_time_s = int(time.monotonic() - submit_ts)
+        await session.send_event(Event(
+            event_type="hf_job_complete",
+            data={
+                "job_id": getattr(job, "id", None),
+                "flavor": flavor,
+                "final_status": final_status,
+                "wall_time_s": wall_time_s,
+            },
+        ))
+    except Exception as e:
+        logger.debug("record_hf_job_complete failed (non-fatal): %s", e)
+
+
+# ── sandbox ─────────────────────────────────────────────────────────────────
+
+async def record_sandbox_create(
+    session: Any,
+    sandbox: Any,
+    *,
+    hardware: str,
+    create_latency_s: int,
+) -> None:
+    from agent.core.session import Event
+    try:
+        # Pin created-at on the session so record_sandbox_destroy can diff.
+        session._sandbox_created_at = time.monotonic() - create_latency_s
+        await session.send_event(Event(
+            event_type="sandbox_create",
+            data={
+                "sandbox_id": getattr(sandbox, "space_id", None),
+                "hardware": hardware,
+                "create_latency_s": int(create_latency_s),
+            },
+        ))
+    except Exception as e:
+        logger.debug("record_sandbox_create failed (non-fatal): %s", e)
+
+
+async def record_sandbox_destroy(session: Any, sandbox: Any) -> None:
+    from agent.core.session import Event
+    try:
+        created = getattr(session, "_sandbox_created_at", None)
+        lifetime_s = int(time.monotonic() - created) if created else None
+        await session.send_event(Event(
+            event_type="sandbox_destroy",
+            data={
+                "sandbox_id": getattr(sandbox, "space_id", None),
+                "lifetime_s": lifetime_s,
+            },
+        ))
+    except Exception as e:
+        logger.debug("record_sandbox_destroy failed (non-fatal): %s", e)
+
+
+# ── feedback ───────────────────────────────────────────────────────────────
+
+async def record_feedback(
+    session: Any,
+    *,
+    rating: str,
+    turn_index: int | None = None,
+    message_id: str | None = None,
+    comment: str | None = None,
+) -> None:
+    from agent.core.session import Event
+    try:
+        await session.send_event(Event(
+            event_type="feedback",
+            data={
+                "rating": rating,
+                "turn_index": turn_index,
+                "message_id": message_id,
+                "comment": (comment or "")[:500],
+            },
+        ))
+    except Exception as e:
+        logger.debug("record_feedback failed (non-fatal): %s", e)
+
+
+# ── heartbeat ──────────────────────────────────────────────────────────────
+
+# Module-level reference set for fire-and-forget heartbeat tasks. asyncio only
+# keeps *weak* references to tasks, so the returned Task would otherwise be
+# eligible for GC before running — the task gets discarded and the upload
+# silently never happens. Hold strong refs until the task completes.
+_heartbeat_tasks: set[asyncio.Task] = set()
+
+
+class HeartbeatSaver:
+    """Time-gated mid-turn flush.
+
+    Called from ``Session.send_event`` after every event. Fires
+    ``save_and_upload_detached`` in a worker thread at most once per
+    ``heartbeat_interval_s`` (default 60s). Guards against losing trace data
+    on long-running turns that crash before ``turn_complete``.
+    """
+
+    @staticmethod
+    def maybe_fire(session: Any) -> None:
+        if not getattr(session.config, "save_sessions", False):
+            return
+        interval = getattr(session.config, "heartbeat_interval_s", 0) or 0
+        if interval <= 0:
+            return
+        now = time.monotonic()
+        last = getattr(session, "_last_heartbeat_ts", None)
+        if last is None:
+            # Initialise on first event; no save yet.
+            session._last_heartbeat_ts = now
+            return
+        if now - last < interval:
+            return
+        session._last_heartbeat_ts = now
+        repo_id = session.config.session_dataset_repo
+        try:
+            task = asyncio.get_running_loop().create_task(
+                asyncio.to_thread(session.save_and_upload_detached, repo_id)
+            )
+            # Hold a strong reference until the task finishes so asyncio can't
+            # GC it. ``set.discard`` is a no-op on missing keys → safe callback.
+            _heartbeat_tasks.add(task)
+            task.add_done_callback(_heartbeat_tasks.discard)
+        except RuntimeError:
+            try:
+                session.save_and_upload_detached(repo_id)
+            except Exception as e:
+                logger.debug("Heartbeat save failed (non-fatal): %s", e)

agent/sft/tagger.py

@@ -0,0 +1,324 @@
+"""Derive tags for a session trajectory.
+
+``tag_session(trajectory)`` → ``list[str]``. Pure function. No filtering, no
+mutation — tags are purely metadata so downstream pipelines can slice the raw
+SFT dataset (``where 'hf_job:succeeded' in tags``) without re-reading trajectories.
+
+Tag namespaces (all tags are ``"<namespace>:<value>"`` strings):
+
+* ``tool:<name>``       — every tool called at least once (``tool:hf_jobs``, …)
+* ``outcome:<end>``     — ``completed`` / ``errored`` / ``interrupted`` /
+                          ``ongoing`` / ``doom_loop`` / ``context_exceeded``
+* ``hf_job:<facet>``    — ``submitted``, ``succeeded``, ``failed``,
+                          ``multi`` (>1), ``oom``, ``push_to_hub``
+* ``gpu:<kind>``        — ``none``, ``t4``, ``a10g``, ``a100``, ``l40s``,
+                          ``h100``, plus ``gpu:multi`` for x2/x4/x8 flavors
+* ``sandbox:<facet>``   — ``created``, ``gpu``, ``cpu``, ``long_lived`` (>30 min)
+* ``feedback:<kind>``   — ``up``, ``down``, ``mixed``, ``none``
+* ``model:<family>``    — ``opus`` / ``sonnet`` / ``haiku`` / ``kimi`` /
+                          ``gpt`` / ``deepseek`` / ``qwen`` / ``other``
+* ``turns:<bucket>``    — ``short`` (<5) / ``medium`` (5–20) / ``long`` (>20)
+* ``cost:<bucket>``     — ``low`` (<$0.10) / ``med`` (<$1) / ``high``
+* ``task:<kind>``       — ``training`` / ``inference`` / ``data_prep`` /
+                          ``research_only`` (heuristic on tools + scripts)
+
+Tags are deduplicated before returning.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Iterable
+
+# Flavor → GPU-family mapping. Keep conservative; unknown flavors → "none".
+_GPU_FAMILY = {
+    "cpu-basic": "none", "cpu-upgrade": "none",
+    "t4-small": "t4", "t4-medium": "t4",
+    "l4x1": "l40s", "l4x4": "l40s",
+    "l40sx1": "l40s", "l40sx4": "l40s", "l40sx8": "l40s",
+    "a10g-small": "a10g", "a10g-large": "a10g",
+    "a10g-largex2": "a10g", "a10g-largex4": "a10g",
+    "a100-large": "a100", "a100x2": "a100",
+    "a100x4": "a100", "a100x8": "a100",
+    "h100": "h100", "h100x8": "h100",
+}
+
+# Substrings that count a flavor as multi-GPU.
+_MULTI_GPU_MARKERS = ("x2", "x4", "x8")
+
+# Tool names that don't touch training/inference or sandbox/jobs. If a session
+# only used these, we tag it research_only.
+_RESEARCH_ONLY_TOOLS = {
+    "research", "github_find_examples", "github_read_file", "github_list_repos",
+    "hf_papers", "explore_hf_docs", "fetch_hf_docs", "hub_repo_details",
+    "plan", "hf_inspect_dataset", "web_search",
+}
+
+# Tool names that signal data manipulation workflows.
+_DATA_PREP_TOOLS = {"hf_inspect_dataset", "dataset_tools", "hub_repo_details"}
+
+
+def _model_family(model_name: str | None) -> str:
+    if not model_name:
+        return "other"
+    n = model_name.lower()
+    if "opus" in n:
+        return "opus"
+    if "sonnet" in n:
+        return "sonnet"
+    if "haiku" in n:
+        return "haiku"
+    if "kimi" in n:
+        return "kimi"
+    if "gpt" in n:
+        return "gpt"
+    if "deepseek" in n:
+        return "deepseek"
+    if "qwen" in n:
+        return "qwen"
+    if "llama" in n:
+        return "llama"
+    return "other"
+
+
+def _turns_bucket(n: int) -> str:
+    if n < 5:
+        return "short"
+    if n <= 20:
+        return "medium"
+    return "long"
+
+
+def _cost_bucket(cost_usd: float) -> str:
+    if cost_usd < 0.10:
+        return "low"
+    if cost_usd < 1.0:
+        return "med"
+    return "high"
+
+
+def _flavor_to_gpu_tags(flavor: str) -> list[str]:
+    family = _GPU_FAMILY.get(flavor, "none")
+    tags = [f"gpu:{family}"]
+    if any(m in flavor for m in _MULTI_GPU_MARKERS):
+        tags.append("gpu:multi")
+    return tags
+
+
+def _has_oom_signal(tool_outputs: Iterable[str]) -> bool:
+    for out in tool_outputs:
+        if not isinstance(out, str):
+            continue
+        low = out.lower()
+        if "outofmemoryerror" in low or "cuda out of memory" in low or "oom" in low:
+            return True
+    return False
+
+
+def _infer_task_tag(
+    tool_names: set[str],
+    hf_job_submit_scripts: list[str],
+) -> str | None:
+    """Return a ``task:*`` tag or None if we can't tell.
+
+    Heuristic order: training > inference > data_prep > research_only.
+    """
+    # training: any hf_jobs script with a Trainer/SFT/training keyword, OR uses
+    # hf_jobs at all and a script mentions training APIs.
+    for script in hf_job_submit_scripts:
+        low = script.lower()
+        if any(k in low for k in (
+            "sftconfig", "sfttrainer", "trainer(", "trainingarguments",
+            "grpo", "dpo", ".train(", "transformers import",
+            "trainer import", "fine-tune", "finetune",
+        )):
+            return "training"
+
+    # inference: sessions that use inference tools but never hf_jobs/sandbox
+    uses_compute = bool(tool_names & {"hf_jobs", "sandbox_create", "sandbox_exec"})
+    if not uses_compute and tool_names & {"inference", "generate", "run_inference"}:
+        return "inference"
+
+    # data_prep: primarily dataset tools and no training/inference
+    if tool_names & _DATA_PREP_TOOLS and not uses_compute:
+        return "data_prep"
+
+    # research_only: every tool used is in the research allow-list
+    if tool_names and tool_names <= _RESEARCH_ONLY_TOOLS:
+        return "research_only"
+
+    return None
+
+
+def tag_session(trajectory: dict) -> list[str]:
+    """Derive tags from a session trajectory. Pure function."""
+    tags: set[str] = set()
+
+    events: list[dict] = trajectory.get("events") or []
+    messages: list[dict] = trajectory.get("messages") or []
+    model_name: str | None = trajectory.get("model_name")
+
+    # model
+    tags.add(f"model:{_model_family(model_name)}")
+
+    # turns
+    user_turns = sum(1 for m in messages if m.get("role") == "user")
+    tags.add(f"turns:{_turns_bucket(user_turns)}")
+
+    # cost + tool-name enumeration + outcome detection
+    cost_usd = 0.0
+    tool_names: set[str] = set()
+    tool_outputs: list[str] = []
+    hf_job_submit_count = 0
+    hf_job_submit_scripts: list[str] = []
+    hf_job_success_count = 0
+    hf_job_fail_count = 0
+    hf_job_push_to_hub = False
+    gpu_tags_seen: set[str] = set()
+
+    # Outcome is the *last* terminal signal. Seed with "ongoing" — overridden
+    # if we see a terminal event.
+    outcome = "ongoing"
+    had_error = False
+    had_doom_loop = False
+    had_compact = False
+
+    feedback_up = 0
+    feedback_down = 0
+
+    sandbox_created = False
+    sandbox_hardware: str | None = None
+    sandbox_lifetime_s: int | None = None
+
+    for ev in events:
+        et = ev.get("event_type")
+        data = ev.get("data") or {}
+
+        if et == "llm_call":
+            cost_usd += float(data.get("cost_usd") or 0.0)
+
+        elif et == "tool_call":
+            name = data.get("tool")
+            if name:
+                tool_names.add(name)
+
+        elif et == "tool_output":
+            out = data.get("output")
+            if isinstance(out, str):
+                tool_outputs.append(out)
+
+        elif et == "hf_job_submit":
+            hf_job_submit_count += 1
+            if data.get("push_to_hub"):
+                hf_job_push_to_hub = True
+            flavor = data.get("flavor") or "cpu-basic"
+            for t in _flavor_to_gpu_tags(flavor):
+                gpu_tags_seen.add(t)
+
+        elif et == "hf_job_complete":
+            final = (data.get("final_status") or "").lower()
+            if final in ("completed", "succeeded", "success"):
+                hf_job_success_count += 1
+            elif final in ("failed", "error", "timeout", "cancelled"):
+                hf_job_fail_count += 1
+
+        elif et == "sandbox_create":
+            sandbox_created = True
+            sandbox_hardware = data.get("hardware")
+
+        elif et == "sandbox_destroy":
+            lt = data.get("lifetime_s")
+            if isinstance(lt, (int, float)):
+                sandbox_lifetime_s = int(lt)
+
+        elif et == "feedback":
+            rating = data.get("rating")
+            if rating == "up":
+                feedback_up += 1
+            elif rating == "down":
+                feedback_down += 1
+
+        elif et == "error":
+            had_error = True
+        elif et == "turn_complete":
+            if not had_error:
+                outcome = "completed"
+        elif et == "interrupted":
+            outcome = "interrupted"
+        elif et == "compacted":
+            had_compact = True
+        elif et == "tool_log":
+            log_text = (data.get("log") or "").lower()
+            if "doom loop" in log_text:
+                had_doom_loop = True
+
+    if had_error and outcome not in ("completed", "interrupted"):
+        outcome = "errored"
+
+    tags.add(f"outcome:{outcome}")
+    if had_doom_loop:
+        tags.add("outcome:doom_loop")
+    if had_compact:
+        tags.add("outcome:context_exceeded")
+
+    # tools
+    for name in tool_names:
+        tags.add(f"tool:{name}")
+
+    # hf_jobs facets
+    if hf_job_submit_count >= 1:
+        tags.add("hf_job:submitted")
+    if hf_job_submit_count > 1:
+        tags.add("hf_job:multi")
+    if hf_job_success_count > 0:
+        tags.add("hf_job:succeeded")
+    if hf_job_fail_count > 0:
+        tags.add("hf_job:failed")
+    if hf_job_push_to_hub:
+        tags.add("hf_job:push_to_hub")
+    if _has_oom_signal(tool_outputs):
+        tags.add("hf_job:oom")
+
+    # gpu tags (from all submitted jobs)
+    tags.update(gpu_tags_seen)
+    if "gpu:none" in tags and len(gpu_tags_seen) > 1:
+        # If any GPU flavor was used, drop the "none" tag for clarity.
+        tags.discard("gpu:none")
+
+    # sandbox facets
+    if sandbox_created:
+        tags.add("sandbox:created")
+        if sandbox_hardware:
+            fam = _GPU_FAMILY.get(sandbox_hardware, "none")
+            tags.add("sandbox:cpu" if fam == "none" else "sandbox:gpu")
+        if sandbox_lifetime_s is not None and sandbox_lifetime_s > 1800:
+            tags.add("sandbox:long_lived")
+
+    # feedback
+    if feedback_up and feedback_down:
+        tags.add("feedback:mixed")
+    elif feedback_up:
+        tags.add("feedback:up")
+    elif feedback_down:
+        tags.add("feedback:down")
+    else:
+        tags.add("feedback:none")
+
+    # cost bucket
+    tags.add(f"cost:{_cost_bucket(cost_usd)}")
+
+    # task heuristic (needs scripts — pull from the hf_job_submit events'
+    # matching tool_call arguments in the event list).
+    for ev in events:
+        if ev.get("event_type") == "tool_call":
+            data = ev.get("data") or {}
+            if data.get("tool") == "hf_jobs":
+                args = data.get("arguments") or {}
+                script = args.get("script") or args.get("command") or ""
+                if isinstance(script, str):
+                    hf_job_submit_scripts.append(script)
+
+    task_tag = _infer_task_tag(tool_names, hf_job_submit_scripts)
+    if task_tag:
+        tags.add(f"task:{task_tag}")
+
+    return sorted(tags)

agent/tools/jobs_tool.py

@@ -528,14 +528,16 @@ class HfJobsTool:
                 job_type = "Docker"
 
             # Run the job
+            flavor = args.get("hardware_flavor", "cpu-basic")
+            timeout_str = args.get("timeout", "30m")
             job = await _async_call(
                 self.api.run_job,
                 image=image,
                 command=command,
                 env=_add_default_env(args.get("env")),
                 secrets=_add_environment_variables(args.get("secrets"), self.hf_token),
-                flavor=args.get("hardware_flavor", "cpu-basic"),
-                timeout=args.get("timeout", "30m"),
+                flavor=flavor,
+                timeout=timeout_str,
                 namespace=self.namespace,
             )
 
@@ -557,6 +559,16 @@ class HfJobsTool:
                     )
                 )
 
+            # Telemetry: job submission + completion (infra consumption signal).
+            submit_ts = None
+            if self.session:
+                from agent.core import telemetry
+                submit_ts = await telemetry.record_hf_job_submit(
+                    self.session, job,
+                    {**args, "hardware_flavor": flavor, "timeout": timeout_str},
+                    image=image, job_type=job_type,
+                )
+
             # Wait for completion and stream logs
             logger.info(f"{job_type} job started: {job.url}")
             logger.info("Streaming logs...")
@@ -566,6 +578,13 @@ class HfJobsTool:
                 namespace=self.namespace,
             )
 
+            if self.session and submit_ts is not None:
+                from agent.core import telemetry
+                await telemetry.record_hf_job_complete(
+                    self.session, job,
+                    flavor=flavor, final_status=final_status, submit_ts=submit_ts,
+                )
+
             # Untrack job ID (completed or failed, no longer needs cancellation)
             if self.session:
                 self.session._running_job_ids.discard(job.id)

agent/tools/sandbox_tool.py

@@ -131,6 +131,8 @@ async def _ensure_sandbox(
     }
     if hardware != "cpu-basic":
         kwargs["sleep_time"] = 2700
+    import time as _t
+    _t_start = _t.monotonic()
     try:
         sb = await asyncio.to_thread(Sandbox.create, **kwargs)
     except Sandbox.Cancelled:
@@ -139,6 +141,13 @@ async def _ensure_sandbox(
         watcher_task.cancel()
     session.sandbox = sb
 
+    # Telemetry: sandbox creation (infra consumption signal)
+    from agent.core import telemetry
+    await telemetry.record_sandbox_create(
+        session, sb, hardware=hardware,
+        create_latency_s=int(_t.monotonic() - _t_start),
+    )
+
     # Set a descriptive title (template title is inherited on duplicate)
     from huggingface_hub import metadata_update
 

backend/kpis_scheduler.py

@@ -0,0 +1,146 @@
+"""In-process hourly KPI rollup, owned by the backend Space lifespan.
+
+Replaces an external GitHub Actions cron so the rollup lives next to the data
+and reuses the Space's existing HF token — no production secrets on the
+public source repo. See ``scripts/build_kpis.py`` for the data-flow diagram
+and metric definitions.
+
+Behaviour::
+
+    lifespan startup → start APScheduler with cron("5 * * * *", UTC)
+                     → fire a best-effort 6-hour backfill (fire-and-forget)
+    each :05         → run ``build_kpis.run_for_hour`` for the just-completed hour
+    lifespan shutdown → scheduler.shutdown(wait=False)
+
+Environment::
+
+    HF_KPI_WRITE_TOKEN | HF_SESSION_UPLOAD_TOKEN | HF_TOKEN | HF_ADMIN_TOKEN
+        First one found is used. Least-privilege first.
+    KPI_SOURCE_REPO     default smolagents/ml-intern-sessions
+    KPI_TARGET_REPO     default smolagents/ml-intern-kpis
+    ML_INTERN_KPIS_DISABLED  if truthy, the scheduler is not started
+"""
+
+from __future__ import annotations
+
+import asyncio
+import importlib.util
+import logging
+import os
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+_PROJECT_ROOT = Path(__file__).resolve().parent.parent
+
+# Hold strong refs to backfill tasks so asyncio doesn't GC them mid-run.
+_background_tasks: set[asyncio.Task] = set()
+
+_scheduler = None  # AsyncIOScheduler instance (lazy import)
+
+
+def _resolve_token() -> Optional[str]:
+    """Pick the first available HF token. Least-privilege first."""
+    for var in (
+        "HF_KPI_WRITE_TOKEN",
+        "HF_SESSION_UPLOAD_TOKEN",
+        "HF_TOKEN",
+        "HF_ADMIN_TOKEN",
+    ):
+        val = os.environ.get(var)
+        if val:
+            return val
+    return None
+
+
+def _load_build_kpis():
+    """Import ``scripts/build_kpis.py`` without putting ``scripts/`` on sys.path."""
+    spec = importlib.util.spec_from_file_location(
+        "build_kpis", _PROJECT_ROOT / "scripts" / "build_kpis.py",
+    )
+    mod = importlib.util.module_from_spec(spec)
+    assert spec.loader is not None
+    spec.loader.exec_module(mod)
+    return mod
+
+
+async def _run_hour(hour_dt: datetime) -> None:
+    """Run one hourly rollup off the event loop. Best-effort, never raises."""
+    token = _resolve_token()
+    if not token:
+        logger.warning("kpis_scheduler: no HF token available, skipping %s", hour_dt)
+        return
+    try:
+        mod = _load_build_kpis()
+        from huggingface_hub import HfApi
+        api = HfApi()
+        source = os.environ.get("KPI_SOURCE_REPO", "smolagents/ml-intern-sessions")
+        target = os.environ.get("KPI_TARGET_REPO", "smolagents/ml-intern-kpis")
+        await asyncio.to_thread(mod.run_for_hour, api, source, target, hour_dt, token)
+    except Exception as e:
+        logger.warning("kpis_scheduler: rollup for %s failed: %s", hour_dt, e)
+
+
+async def run_last_completed_hour() -> None:
+    """The scheduled-at-:05 job. Rolls up the previous whole hour."""
+    now = datetime.now(timezone.utc).replace(minute=0, second=0, microsecond=0)
+    await _run_hour(now - timedelta(hours=1))
+
+
+async def backfill(hours: int = 6) -> None:
+    """Catch-up pass for hours the Space was down. Idempotent (overwrites)."""
+    now = datetime.now(timezone.utc).replace(minute=0, second=0, microsecond=0)
+    for i in range(1, hours + 1):
+        await _run_hour(now - timedelta(hours=i))
+
+
+def start(backfill_hours: int = 6) -> None:
+    """Called from FastAPI lifespan startup."""
+    global _scheduler
+    if os.environ.get("ML_INTERN_KPIS_DISABLED"):
+        logger.info("kpis_scheduler: disabled via ML_INTERN_KPIS_DISABLED")
+        return
+    if _scheduler is not None:
+        return
+
+    try:
+        from apscheduler.schedulers.asyncio import AsyncIOScheduler
+        from apscheduler.triggers.cron import CronTrigger
+    except ImportError:
+        logger.warning("kpis_scheduler: apscheduler not installed, skipping")
+        return
+
+    _scheduler = AsyncIOScheduler(timezone="UTC")
+    _scheduler.add_job(
+        run_last_completed_hour,
+        CronTrigger(minute=5),
+        id="kpis_hourly",
+        misfire_grace_time=600,  # tolerate a 10-min misfire window
+        coalesce=True,           # collapse multiple missed fires into one
+        max_instances=1,
+        replace_existing=True,
+    )
+    _scheduler.start()
+    logger.info("kpis_scheduler: started (cron '5 * * * *' UTC)")
+
+    # Non-blocking backfill. Hold a strong ref until done so asyncio doesn't
+    # GC the task before it finishes.
+    try:
+        task = asyncio.get_running_loop().create_task(backfill(backfill_hours))
+        _background_tasks.add(task)
+        task.add_done_callback(_background_tasks.discard)
+    except RuntimeError:
+        # Not in an event loop (tests); skip backfill.
+        pass
+
+
+async def shutdown() -> None:
+    """Called from FastAPI lifespan shutdown."""
+    global _scheduler
+    if _scheduler is None:
+        return
+    _scheduler.shutdown(wait=False)
+    _scheduler = None
+    logger.info("kpis_scheduler: stopped")

backend/main.py

@@ -27,8 +27,37 @@ logger = logging.getLogger(__name__)
 async def lifespan(app: FastAPI):
     """Application lifespan handler."""
     logger.info("Starting HF Agent backend...")
+    # Start in-process hourly KPI rollup. Replaces an external cron so the
+    # rollup lives next to the data and reuses the Space's HF token.
+    try:
+        import kpis_scheduler
+        kpis_scheduler.start()
+    except Exception as e:
+        logger.warning("KPI scheduler failed to start: %s", e)
+
     yield
+
     logger.info("Shutting down HF Agent backend...")
+    try:
+        import kpis_scheduler
+        await kpis_scheduler.shutdown()
+    except Exception as e:
+        logger.warning("KPI scheduler shutdown failed: %s", e)
+
+    # Final-flush: save every still-active session so we don't lose traces on
+    # server restart. Uploads are detached subprocesses — this is fast.
+    try:
+        from session_manager import session_manager
+        for sid, agent_session in list(session_manager.sessions.items()):
+            sess = agent_session.session
+            if sess.config.save_sessions:
+                try:
+                    sess.save_and_upload_detached(sess.config.session_dataset_repo)
+                    logger.info("Flushed session %s on shutdown", sid)
+                except Exception as e:
+                    logger.warning("Failed to flush session %s: %s", sid, e)
+    except Exception as e:
+        logger.warning("Lifespan final-flush skipped: %s", e)
 
 
 app = FastAPI(

backend/routes/agent.py

@@ -693,3 +693,41 @@ async def shutdown_session(
     return {"status": "shutdown_requested", "session_id": session_id}
 
 
+@router.post("/feedback/{session_id}")
+async def submit_feedback(
+    session_id: str,
+    body: dict,
+    user: dict = Depends(get_current_user),
+) -> dict:
+    """Attach a user feedback signal to a session's event log.
+
+    Body: {rating: "up"|"down"|"outcome_success"|"outcome_fail",
+           turn_index?: int, comment?: str, message_id?: str}
+    Appended as a `feedback` event and saved with the session trajectory.
+    """
+    _check_session_access(session_id, user)
+    agent_session = session_manager.sessions.get(session_id)
+    if not agent_session:
+        raise HTTPException(status_code=404, detail="Session not found")
+
+    rating = body.get("rating")
+    if rating not in {"up", "down", "outcome_success", "outcome_fail"}:
+        raise HTTPException(status_code=400, detail="invalid rating")
+
+    from agent.core import telemetry
+    await telemetry.record_feedback(
+        agent_session.session,
+        rating=rating,
+        turn_index=body.get("turn_index"),
+        message_id=body.get("message_id"),
+        comment=body.get("comment"),
+    )
+    # Fire-and-forget save so feedback reaches the dataset even if the user
+    # closes the tab right after clicking.
+    if agent_session.session.config.save_sessions:
+        agent_session.session.save_and_upload_detached(
+            agent_session.session.config.session_dataset_repo
+        )
+    return {"status": "ok"}
+
+

backend/session_manager.py

@@ -290,11 +290,14 @@ class SessionManager:
         """Delete the sandbox Space if one was created for this session."""
         sandbox = getattr(session, "sandbox", None)
         if sandbox and getattr(sandbox, "_owns_space", False):
+            space_id = getattr(sandbox, "space_id", None)
             try:
-                logger.info(f"Deleting sandbox {sandbox.space_id}...")
+                logger.info(f"Deleting sandbox {space_id}...")
                 await asyncio.to_thread(sandbox.delete)
+                from agent.core import telemetry
+                await telemetry.record_sandbox_destroy(session, sandbox)
             except Exception as e:
-                logger.warning(f"Failed to delete sandbox {sandbox.space_id}: {e}")
+                logger.warning(f"Failed to delete sandbox {space_id}: {e}")
 
     async def _run_session(
         self,
@@ -356,6 +359,15 @@ class SessionManager:
 
             await self._cleanup_sandbox(session)
 
+            # Final-flush: always save on session death so we capture ended
+            # sessions even if the client disconnects without /shutdown.
+            # Idempotent via session_id key; detached subprocess.
+            if session.config.save_sessions:
+                try:
+                    session.save_and_upload_detached(session.config.session_dataset_repo)
+                except Exception as e:
+                    logger.warning(f"Final-flush failed for {session_id}: {e}")
+
             async with self._lock:
                 if session_id in self.sessions:
                     self.sessions[session_id].is_active = False

configs/main_agent_config.json

@@ -1,7 +1,7 @@
 {
   "model_name": "bedrock/us.anthropic.claude-opus-4-6-v1",
   "save_sessions": true,
-  "session_dataset_repo": "akseljoonas/hf-agent-sessions",
+  "session_dataset_repo": "smolagents/ml-intern-sessions",
   "yolo_mode": false,
   "confirm_cpu_jobs": true,
   "auto_file_upload": true,

frontend/src/components/Chat/AssistantMessage.tsx

@@ -1,13 +1,19 @@
-import { useMemo } from 'react';
-import { Box, Stack, Typography } from '@mui/material';
+import { useMemo, useState } from 'react';
+import { Box, IconButton, Stack, Tooltip, Typography } from '@mui/material';
+import ThumbUpOutlined from '@mui/icons-material/ThumbUpOutlined';
+import ThumbUp from '@mui/icons-material/ThumbUp';
+import ThumbDownOutlined from '@mui/icons-material/ThumbDownOutlined';
+import ThumbDown from '@mui/icons-material/ThumbDown';
 import MarkdownContent from './MarkdownContent';
 import ToolCallGroup from './ToolCallGroup';
+import { apiFetch } from '@/utils/api';
 import type { UIMessage } from 'ai';
 import type { MessageMeta } from '@/types/agent';
 
 interface AssistantMessageProps {
   message: UIMessage;
   isStreaming?: boolean;
+  sessionId?: string | null;
   approveTools: (approvals: Array<{ tool_call_id: string; approved: boolean; feedback?: string | null }>) => Promise<boolean>;
 }
 
@@ -43,8 +49,27 @@ function groupParts(parts: UIMessage['parts']) {
   return groups;
 }
 
-export default function AssistantMessage({ message, isStreaming = false, approveTools }: AssistantMessageProps) {
+export default function AssistantMessage({ message, isStreaming = false, sessionId, approveTools }: AssistantMessageProps) {
   const groups = useMemo(() => groupParts(message.parts), [message.parts]);
+  const [feedback, setFeedback] = useState<'up' | 'down' | null>(null);
+  const [feedbackBusy, setFeedbackBusy] = useState(false);
+
+  const sendFeedback = async (rating: 'up' | 'down') => {
+    if (!sessionId || feedbackBusy) return;
+    setFeedbackBusy(true);
+    // Optimistic toggle — feedback is observability, not a hard requirement.
+    setFeedback(rating);
+    try {
+      await apiFetch(`/api/feedback/${sessionId}`, {
+        method: 'POST',
+        body: JSON.stringify({ rating, message_id: message.id }),
+      });
+    } catch {
+      // Silently swallow — don't block chat UX on a telemetry write.
+    } finally {
+      setFeedbackBusy(false);
+    }
+  };
 
   // Find the last text group index for streaming cursor
   let lastTextIdx = -1;
@@ -114,6 +139,24 @@ export default function AssistantMessage({ message, isStreaming = false, approve
           return null;
         })}
       </Box>
+      {!isStreaming && sessionId && (
+        <Stack
+          direction="row"
+          spacing={0.5}
+          sx={{ mt: 0.5, ml: 0.5, opacity: feedback ? 1 : 0.5, '&:hover': { opacity: 1 } }}
+        >
+          <Tooltip title="Helpful">
+            <IconButton size="small" disabled={feedbackBusy} onClick={() => sendFeedback('up')}>
+              {feedback === 'up' ? <ThumbUp fontSize="inherit" /> : <ThumbUpOutlined fontSize="inherit" />}
+            </IconButton>
+          </Tooltip>
+          <Tooltip title="Not helpful">
+            <IconButton size="small" disabled={feedbackBusy} onClick={() => sendFeedback('down')}>
+              {feedback === 'down' ? <ThumbDown fontSize="inherit" /> : <ThumbDownOutlined fontSize="inherit" />}
+            </IconButton>
+          </Tooltip>
+        </Stack>
+      )}
     </Box>
   );
 }

frontend/src/components/Chat/MessageBubble.tsx

@@ -9,6 +9,7 @@ interface MessageBubbleProps {
   onEditAndRegenerate?: (messageId: string, newText: string) => void | Promise<void>;
   isProcessing?: boolean;
   isStreaming?: boolean;
+  sessionId?: string | null;
   approveTools: (approvals: Array<{ tool_call_id: string; approved: boolean; feedback?: string | null }>) => Promise<boolean>;
 }
 
@@ -19,6 +20,7 @@ export default function MessageBubble({
   onEditAndRegenerate,
   isProcessing = false,
   isStreaming = false,
+  sessionId,
   approveTools,
 }: MessageBubbleProps) {
   if (message.role === 'user') {
@@ -38,6 +40,7 @@ export default function MessageBubble({
       <AssistantMessage
         message={message}
         isStreaming={isStreaming}
+        sessionId={sessionId}
         approveTools={approveTools}
       />
     );

frontend/src/components/Chat/MessageList.tsx

@@ -8,6 +8,7 @@ import type { UIMessage } from 'ai';
 interface MessageListProps {
   messages: UIMessage[];
   isProcessing: boolean;
+  sessionId?: string | null;
   approveTools: (approvals: Array<{ tool_call_id: string; approved: boolean; feedback?: string | null }>) => Promise<boolean>;
   onUndoLastTurn: () => void | Promise<void>;
   onEditAndRegenerate?: (messageId: string, newText: string) => void | Promise<void>;
@@ -57,7 +58,7 @@ function WelcomeGreeting() {
   );
 }
 
-export default function MessageList({ messages, isProcessing, approveTools, onUndoLastTurn, onEditAndRegenerate }: MessageListProps) {
+export default function MessageList({ messages, isProcessing, sessionId, approveTools, onUndoLastTurn, onEditAndRegenerate }: MessageListProps) {
   const scrollContainerRef = useRef<HTMLDivElement>(null);
   const stickToBottom = useRef(true);
 
@@ -139,6 +140,7 @@ export default function MessageList({ messages, isProcessing, approveTools, onUn
               onEditAndRegenerate={onEditAndRegenerate}
               isProcessing={isProcessing}
               isStreaming={isProcessing && msg.id === lastAssistantId}
+              sessionId={sessionId}
               approveTools={approveTools}
             />
           ))

frontend/src/components/SessionChat.tsx

@@ -102,6 +102,7 @@ export default function SessionChat({ sessionId, isActive, onSessionDead }: Sess
       <MessageList
         messages={messages}
         isProcessing={busy}
+        sessionId={sessionId}
         approveTools={approveTools}
         onUndoLastTurn={undoLastTurn}
         onEditAndRegenerate={editAndRegenerate}

pyproject.toml

@@ -26,6 +26,7 @@ dependencies = [
     "uvicorn[standard]>=0.32.0",
     "httpx>=0.27.0",
     "websockets>=13.0",
+    "apscheduler>=3.10,<4",
 ]
 
 [project.optional-dependencies]

scripts/build_kpis.py

@@ -0,0 +1,517 @@
+#!/usr/bin/env python3
+"""Hourly KPI rollup for the session-trajectory dataset.
+
+================================================================================
+ Data flow
+================================================================================
+
+    ┌────────────────────┐   heartbeat      ┌────────────────────────────────┐
+    │  agent (CLI/web)   │ ───────────────▶ │  hf-agent-sessions  (dataset)  │
+    │  Session.send_event│                  │  sessions/YYYY-MM-DD/<id>.jsonl│
+    └────────────────────┘                  └───────────────┬────────────────┘
+                                                            │ cron @:05 each hour
+                                                            ▼
+                                         ┌──────────────────────────────────┐
+                                         │   scripts/build_kpis.py          │
+                                         │   (GitHub Actions)               │
+                                         └───────────────┬──────────────────┘
+                                                         │ upload CSV
+                                                         ▼
+                                         ┌──────────────────────────────────┐
+                                         │  hf-agent-kpis  (dataset)        │
+                                         │  hourly/YYYY-MM-DD/HH.csv        │
+                                         └──────────────────────────────────┘
+
+Each hourly run reads today's + yesterday's session folders (to cover sessions
+that crossed midnight), filters events into the target hour window
+``[hour, hour+1h)``, computes aggregates, and writes one CSV at
+``hourly/<date>/<HH>.csv`` in the target dataset. Uploads are idempotent —
+re-running the same hour overwrites.
+
+================================================================================
+ Metrics (one row per hour)
+================================================================================
+
+    sessions            — distinct session_ids with ≥1 event in window
+    users               — distinct user ids (when present on session rows)
+    turns               — sum of user-message counts across active sessions
+    llm_calls           — count of llm_call events
+    tokens_prompt / _completion / _cache_read / _cache_creation
+    cost_usd            — sum of llm_call.cost_usd
+    cache_hit_ratio     — cache_read / (cache_read + prompt)
+    tool_success_rate   — tool_output success=True / total tool_output
+    failure_rate        — sessions that ended with an `error` event / sessions
+    regenerate_rate     — sessions with any `undo_complete` event / sessions
+    time_to_first_action_s_p50 / _p95  — from session_start to first tool_call
+    thumbs_up / thumbs_down
+    hf_jobs_submitted / _succeeded
+    gpu_hours_by_flavor_json   — JSON-serialised {flavor: gpu-hours}
+
+================================================================================
+ Usage
+================================================================================
+
+    # Run for the most recently completed hour (default — the cron path):
+    python scripts/build_kpis.py
+
+    # Backfill last 24 hours:
+    python scripts/build_kpis.py --hours 24
+
+    # Explicit hour (UTC):
+    python scripts/build_kpis.py --datetime 2026-04-24T14
+
+Env:
+    HF_TOKEN (or HF_KPI_WRITE_TOKEN) — write access to the target dataset.
+
+================================================================================
+ Deploy
+================================================================================
+
+See ``.github/workflows/build-kpis.yml`` — runs every hour at :05. To provision:
+
+    1. Create the target dataset (once):
+         huggingface-cli repo create hf-agent-kpis --type dataset
+    2. Put ``HF_KPI_WRITE_TOKEN`` (or ``HF_TOKEN``) into repo Actions secrets.
+    3. Merge this file; the first scheduled run fires within the hour.
+"""
+
+from __future__ import annotations
+
+import argparse
+import io
+import json
+import logging
+import os
+import sys
+import tempfile
+from collections import defaultdict
+from datetime import date, datetime, timedelta, timezone
+from pathlib import Path
+from typing import Any, Iterable
+
+logger = logging.getLogger("build_kpis")
+
+# Rough gpu-hour pricing for hf_jobs flavor strings. Keep conservative; used
+# only to compute gpu-hours (not dollars) — wall_time_s * flavor_gpu_count.
+_FLAVOR_GPU_COUNT = {
+    "cpu-basic": 0, "cpu-upgrade": 0,
+    "t4-small": 1, "t4-medium": 1,
+    "l4x1": 1, "l4x4": 4,
+    "l40sx1": 1, "l40sx4": 4, "l40sx8": 8,
+    "a10g-small": 1, "a10g-large": 1, "a10g-largex2": 2, "a10g-largex4": 4,
+    "a100-large": 1, "a100x2": 2, "a100x4": 4, "a100x8": 8,
+    "h100": 1, "h100x8": 8,
+}
+
+
+def _percentile(values: list[float], p: float) -> float:
+    if not values:
+        return 0.0
+    values = sorted(values)
+    k = (len(values) - 1) * p
+    f = int(k)
+    c = min(f + 1, len(values) - 1)
+    if f == c:
+        return float(values[f])
+    return float(values[f] + (values[c] - values[f]) * (k - f))
+
+
+def _parse_ts(s: Any) -> datetime | None:
+    if not s or not isinstance(s, str):
+        return None
+    try:
+        dt = datetime.fromisoformat(s)
+    except Exception:
+        return None
+    # Normalise to aware UTC so comparisons work against window bounds.
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    return dt
+
+
+def _iter_session_files(api, repo_id: str, day: date, token: str) -> Iterable[str]:
+    """Yield repo-relative paths for all sessions under ``sessions/YYYY-MM-DD/``."""
+    prefix = f"sessions/{day.isoformat()}/"
+    try:
+        files = api.list_repo_files(repo_id=repo_id, repo_type="dataset", token=token)
+    except Exception as e:
+        logger.warning("list_repo_files(%s) failed: %s", repo_id, e)
+        return []
+    return [f for f in files if f.startswith(prefix) and f.endswith(".jsonl")]
+
+
+def _download_session(repo_id: str, path: str, token: str) -> dict | None:
+    """Fetch one session JSONL and decode its single row.
+
+    ``hf_hub_download`` caches; second run within the same process / runner
+    directory is near-free.
+    """
+    from huggingface_hub import hf_hub_download
+    try:
+        local = hf_hub_download(
+            repo_id=repo_id, filename=path, repo_type="dataset", token=token,
+        )
+    except Exception as e:
+        logger.warning("hf_hub_download(%s) failed: %s", path, e)
+        return None
+    try:
+        with open(local, "r") as f:
+            line = f.readline().strip()
+        if not line:
+            return None
+        row = json.loads(line)
+        # Session uploader stores messages/events as JSON strings — unpack.
+        for key in ("messages", "events", "tools"):
+            v = row.get(key)
+            if isinstance(v, str):
+                try:
+                    row[key] = json.loads(v)
+                except Exception:
+                    row[key] = []
+        return row
+    except Exception as e:
+        logger.warning("parse(%s) failed: %s", path, e)
+        return None
+
+
+def _filter_session_to_window(
+    session: dict, start: datetime, end: datetime,
+) -> dict | None:
+    """Return a copy of ``session`` whose events are only those in ``[start, end)``.
+
+    ``None`` if no event falls in the window — the caller drops the session
+    from this hour's aggregate.
+    """
+    events = session.get("events") or []
+    in_window = []
+    for ev in events:
+        ts = _parse_ts(ev.get("timestamp"))
+        if ts is None:
+            continue
+        if start <= ts < end:
+            in_window.append(ev)
+    if not in_window:
+        return None
+    return {**session, "events": in_window}
+
+
+def _session_metrics(session: dict) -> dict:
+    """Reduce a single session trajectory to its KPI contributions.
+
+    Assumes ``events`` are already filtered to the target window by the caller.
+    """
+    # Pre-seed every numeric key so downstream aggregation can sum without
+    # having to special-case empty sessions.
+    out: dict = {
+        "sessions": 0, "turns": 0, "llm_calls": 0,
+        "tokens_prompt": 0, "tokens_completion": 0,
+        "tokens_cache_read": 0, "tokens_cache_creation": 0,
+        "cost_usd": 0.0,
+        "tool_calls_total": 0, "tool_calls_success": 0,
+        "failures": 0, "regenerate_sessions": 0,
+        "thumbs_up": 0, "thumbs_down": 0,
+        "hf_jobs_submitted": 0, "hf_jobs_succeeded": 0,
+        "first_tool_s": -1,
+    }
+    events = session.get("events") or []
+    messages = session.get("messages") or []
+
+    turn_count = sum(1 for m in messages if m.get("role") == "user")
+    out["turns"] = turn_count
+    out["sessions"] = 1
+
+    tool_success = 0
+    tool_total = 0
+    had_error = False
+    had_undo = False
+    first_tool_ts = None
+    session_start = session.get("session_start_time")
+    gpu_hours_by_flavor: dict[str, float] = defaultdict(float)
+    jobs_submitted = 0
+    jobs_succeeded = 0
+    thumbs_up = 0
+    thumbs_down = 0
+
+    start_dt = _parse_ts(session_start)
+
+    for ev in events:
+        et = ev.get("event_type")
+        data = ev.get("data") or {}
+        ts = _parse_ts(ev.get("timestamp"))
+
+        if et == "llm_call":
+            out["llm_calls"] += 1
+            out["tokens_prompt"] += int(data.get("prompt_tokens") or 0)
+            out["tokens_completion"] += int(data.get("completion_tokens") or 0)
+            out["tokens_cache_read"] += int(data.get("cache_read_tokens") or 0)
+            out["tokens_cache_creation"] += int(data.get("cache_creation_tokens") or 0)
+            out["cost_usd"] += float(data.get("cost_usd") or 0.0)
+
+        elif et == "tool_output":
+            tool_total += 1
+            if data.get("success"):
+                tool_success += 1
+            if first_tool_ts is None and ts is not None and start_dt is not None:
+                first_tool_ts = (ts - start_dt).total_seconds()
+
+        elif et == "tool_call":
+            if first_tool_ts is None and ts is not None and start_dt is not None:
+                first_tool_ts = (ts - start_dt).total_seconds()
+
+        elif et == "error":
+            had_error = True
+
+        elif et == "undo_complete":
+            had_undo = True
+
+        elif et == "feedback":
+            rating = data.get("rating")
+            if rating == "up":
+                thumbs_up += 1
+            elif rating == "down":
+                thumbs_down += 1
+
+        elif et == "hf_job_submit":
+            jobs_submitted += 1
+
+        elif et == "hf_job_complete":
+            flavor = data.get("flavor") or "unknown"
+            status = (data.get("final_status") or "").lower()
+            wall = float(data.get("wall_time_s") or 0.0)
+            gpus = _FLAVOR_GPU_COUNT.get(flavor, 0)
+            gpu_hours_by_flavor[flavor] += wall * gpus / 3600.0
+            if status in ("completed", "succeeded", "success"):
+                jobs_succeeded += 1
+
+    out["tool_calls_total"] = tool_total
+    out["tool_calls_success"] = tool_success
+    out["failures"] = 1 if had_error else 0
+    out["regenerate_sessions"] = 1 if had_undo else 0
+    out["thumbs_up"] = thumbs_up
+    out["thumbs_down"] = thumbs_down
+    out["hf_jobs_submitted"] = jobs_submitted
+    out["hf_jobs_succeeded"] = jobs_succeeded
+    out["first_tool_s"] = first_tool_ts if first_tool_ts is not None else -1
+    out["_gpu_hours_by_flavor"] = dict(gpu_hours_by_flavor)
+    out["_user"] = session.get("user_id") or session.get("session_id")
+    return dict(out)
+
+
+def _aggregate(per_session: list[dict]) -> dict:
+    """Collapse a bucket's worth of session rollups into the final KPI row."""
+    ttfa_values = [s["first_tool_s"] for s in per_session if s.get("first_tool_s", -1) >= 0]
+    gpu_hours: dict[str, float] = defaultdict(float)
+    for s in per_session:
+        for f, h in (s.get("_gpu_hours_by_flavor") or {}).items():
+            gpu_hours[f] += h
+
+    total_sessions = sum(s["sessions"] for s in per_session)
+    total_turns = sum(s["turns"] for s in per_session)
+    tokens_prompt = sum(s["tokens_prompt"] for s in per_session)
+    tokens_cache_read = sum(s["tokens_cache_read"] for s in per_session)
+    tool_total = sum(s["tool_calls_total"] for s in per_session)
+    tool_success = sum(s["tool_calls_success"] for s in per_session)
+
+    unique_users = {s.get("_user") for s in per_session if s.get("_user")}
+
+    return {
+        "sessions": total_sessions,
+        "users": len(unique_users),
+        "turns": total_turns,
+        "llm_calls": int(sum(s["llm_calls"] for s in per_session)),
+        "tokens_prompt": int(tokens_prompt),
+        "tokens_completion": int(sum(s["tokens_completion"] for s in per_session)),
+        "tokens_cache_read": int(tokens_cache_read),
+        "tokens_cache_creation": int(sum(s["tokens_cache_creation"] for s in per_session)),
+        "cost_usd": round(sum(s["cost_usd"] for s in per_session), 4),
+        "cache_hit_ratio": round(
+            tokens_cache_read / (tokens_cache_read + tokens_prompt), 4
+        ) if (tokens_cache_read + tokens_prompt) > 0 else 0.0,
+        "tool_success_rate": round(tool_success / tool_total, 4) if tool_total > 0 else 0.0,
+        "failure_rate": round(
+            sum(s["failures"] for s in per_session) / total_sessions, 4
+        ) if total_sessions > 0 else 0.0,
+        "regenerate_rate": round(
+            sum(s["regenerate_sessions"] for s in per_session) / total_sessions, 4
+        ) if total_sessions > 0 else 0.0,
+        "time_to_first_action_s_p50": round(_percentile(ttfa_values, 0.5), 2),
+        "time_to_first_action_s_p95": round(_percentile(ttfa_values, 0.95), 2),
+        "thumbs_up": int(sum(s["thumbs_up"] for s in per_session)),
+        "thumbs_down": int(sum(s["thumbs_down"] for s in per_session)),
+        "hf_jobs_submitted": int(sum(s["hf_jobs_submitted"] for s in per_session)),
+        "hf_jobs_succeeded": int(sum(s["hf_jobs_succeeded"] for s in per_session)),
+        "gpu_hours_by_flavor_json": json.dumps(dict(gpu_hours), sort_keys=True),
+    }
+
+
+# Back-compat alias: older tests call _aggregate_day.
+_aggregate_day = _aggregate
+
+
+def _csv_cell(v: Any) -> str:
+    s = str(v)
+    if "," in s or '"' in s or "\n" in s:
+        return '"' + s.replace('"', '""') + '"'
+    return s
+
+
+def _write_csv(
+    api, row: dict, bucket_key: str, path_in_repo: str, target_repo: str, token: str,
+) -> None:
+    """Render ``row`` to CSV with a leading ``bucket`` column and upload.
+
+    ``bucket_key`` is the hour string (ISO ``YYYY-MM-DDTHH``) or date string;
+    written as the ``bucket`` column so downstream consumers can union all
+    CSVs without date-parsing paths. ``api`` is the caller's ``HfApi``
+    instance — reused so we don't spin up a fresh one per CSV.
+    """
+    columns = list(row.keys())
+    buf = io.StringIO()
+    buf.write(",".join(["bucket", *columns]) + "\n")
+    buf.write(",".join([bucket_key, *[_csv_cell(row[c]) for c in columns]]) + "\n")
+
+    with tempfile.NamedTemporaryFile(mode="w", suffix=".csv", delete=False) as tmp:
+        tmp.write(buf.getvalue())
+        tmp_path = tmp.name
+
+    try:
+        api.create_repo(
+            repo_id=target_repo, repo_type="dataset", exist_ok=True, token=token,
+        )
+        api.upload_file(
+            path_or_fileobj=tmp_path,
+            path_in_repo=path_in_repo,
+            repo_id=target_repo,
+            repo_type="dataset",
+            token=token,
+            commit_message=f"KPIs for {bucket_key}",
+        )
+    finally:
+        try:
+            os.unlink(tmp_path)
+        except Exception:
+            pass
+
+
+def run_for_hour(
+    api, source_repo: str, target_repo: str, hour_dt: datetime, token: str,
+) -> dict:
+    """Roll up one UTC hour [hour_dt, hour_dt+1h).
+
+    Reads today's + yesterday's session folders so sessions that crossed
+    midnight land in the right hourly bucket.
+    """
+    if hour_dt.tzinfo is None:
+        hour_dt = hour_dt.replace(tzinfo=timezone.utc)
+    window_start = hour_dt.replace(minute=0, second=0, microsecond=0)
+    window_end = window_start + timedelta(hours=1)
+
+    # Sessions partition by session_start_time date. A session that started
+    # at 23:50 yesterday can still emit events in today's first hours, so we
+    # look at both folders.
+    candidate_dates = {window_start.date(), (window_start - timedelta(days=1)).date()}
+
+    per_session: list[dict] = []
+    for d in sorted(candidate_dates):
+        for path in _iter_session_files(api, source_repo, d, token):
+            sess = _download_session(source_repo, path, token)
+            if not sess:
+                continue
+            windowed = _filter_session_to_window(sess, window_start, window_end)
+            if windowed is None:
+                continue
+            per_session.append(_session_metrics(windowed))
+
+    if not per_session:
+        logger.info("No sessions in window %s — skipping", window_start.isoformat())
+        return {}
+
+    row = _aggregate(per_session)
+    bucket_key = window_start.strftime("%Y-%m-%dT%H")
+    path_in_repo = f"hourly/{window_start.strftime('%Y-%m-%d')}/{window_start.strftime('%H')}.csv"
+    _write_csv(api, row, bucket_key, path_in_repo, target_repo, token)
+    logger.info("Wrote KPIs for %s (%d sessions): %s",
+                bucket_key, per_session and len(per_session), row)
+    return row
+
+
+# Back-compat for daily backfills — unchanged behaviour.
+def run_for_day(api, source_repo: str, target_repo: str, day: date, token: str) -> dict:
+    paths = _iter_session_files(api, source_repo, day, token)
+    per_session: list[dict] = []
+    for path in paths:
+        sess = _download_session(source_repo, path, token)
+        if not sess:
+            continue
+        per_session.append(_session_metrics(sess))
+    if not per_session:
+        logger.info("No sessions found for %s — skipping", day)
+        return {}
+    row = _aggregate(per_session)
+    path_in_repo = f"daily/{day.isoformat()}.csv"
+    _write_csv(api, row, day.isoformat(), path_in_repo, target_repo, token)
+    return row
+
+
+def _parse_hour_arg(s: str) -> datetime:
+    """Accept ``YYYY-MM-DDTHH`` or full ISO — always pinned to the start of the hour, UTC."""
+    dt = datetime.fromisoformat(s)
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    return dt.replace(minute=0, second=0, microsecond=0)
+
+
+def main(argv: list[str] | None = None) -> int:
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--source", default="smolagents/ml-intern-sessions")
+    ap.add_argument("--target", default="smolagents/ml-intern-kpis")
+    ap.add_argument(
+        "--hours", type=int, default=1,
+        help="Number of trailing hours to roll up (default: 1 = last completed hour).",
+    )
+    ap.add_argument(
+        "--datetime", type=str, default=None,
+        help="Single hour, ISO ``YYYY-MM-DDTHH`` (UTC); overrides --hours.",
+    )
+    ap.add_argument(
+        "--daily-backfill", type=str, default=None,
+        help="Escape hatch: aggregate a whole day at once (YYYY-MM-DD). "
+             "Writes to daily/<date>.csv. Use for historical backfill only.",
+    )
+    args = ap.parse_args(argv)
+
+    token = (
+        os.environ.get("HF_KPI_WRITE_TOKEN")
+        or os.environ.get("HF_SESSION_UPLOAD_TOKEN")
+        or os.environ.get("HF_TOKEN")
+        or os.environ.get("HF_ADMIN_TOKEN")
+    )
+    if not token:
+        logger.error(
+            "No HF token found. Set one of: HF_KPI_WRITE_TOKEN, "
+            "HF_SESSION_UPLOAD_TOKEN, HF_TOKEN, HF_ADMIN_TOKEN."
+        )
+        return 1
+
+    from huggingface_hub import HfApi
+    api = HfApi()
+
+    if args.daily_backfill:
+        run_for_day(api, args.source, args.target, date.fromisoformat(args.daily_backfill), token)
+        return 0
+
+    if args.datetime:
+        target_hours = [_parse_hour_arg(args.datetime)]
+    else:
+        now = datetime.now(timezone.utc).replace(minute=0, second=0, microsecond=0)
+        # Roll up *completed* hours: start from the hour before ``now``.
+        target_hours = [now - timedelta(hours=i) for i in range(1, args.hours + 1)]
+
+    for hour in target_hours:
+        run_for_hour(api, args.source, args.target, hour, token)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/build_sft.py

@@ -0,0 +1,204 @@
+#!/usr/bin/env python3
+"""Export session trajectories as raw multi-turn tool-calling SFT data.
+
+Reads the source sessions dataset (JSONL, one file per session at
+``sessions/YYYY-MM-DD/<session_id>.jsonl``) and writes a re-shaped row to a
+target dataset at ``sft/YYYY-MM-DD/<session_id>.jsonl``.
+
+**No filtering, no cleaning, no dedup.** Raw passthrough of messages + tools,
+with session-level metadata and derived tags (see ``agent/sft/tagger.py``)
+attached for downstream slicing.
+
+Output row schema::
+
+    {
+      "session_id": "...",
+      "model": "claude-opus-4-6",
+      "timestamp": "2026-04-24T...",
+      "tags": ["tool:hf_jobs", "gpu:a100", "hf_job:succeeded", ...],
+      "messages": [...],   # OpenAI / TRL SFTTrainer format
+      "tools":   [...]     # OpenAI tool schemas the session had access to
+    }
+
+Usage::
+
+    python scripts/build_sft.py \\
+        --source smolagents/ml-intern-sessions \\
+        --target smolagents/ml-intern-sft \\
+        --days 7
+
+Env:
+    HF_TOKEN (or HF_SFT_WRITE_TOKEN) — write access to target dataset.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import os
+import sys
+import tempfile
+from datetime import date, datetime, timedelta, timezone
+from typing import Iterable
+
+# Make ``agent`` importable when this script is run outside the project venv.
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from agent.sft.tagger import tag_session  # noqa: E402
+
+logger = logging.getLogger("build_sft")
+
+
+def _iter_session_files(api, repo_id: str, day: date, token: str) -> Iterable[str]:
+    prefix = f"sessions/{day.isoformat()}/"
+    try:
+        files = api.list_repo_files(repo_id=repo_id, repo_type="dataset", token=token)
+    except Exception as e:
+        logger.warning("list_repo_files(%s) failed: %s", repo_id, e)
+        return []
+    return [f for f in files if f.startswith(prefix) and f.endswith(".jsonl")]
+
+
+def _download_and_parse(repo_id: str, path: str, token: str) -> dict | None:
+    from huggingface_hub import hf_hub_download
+    try:
+        local = hf_hub_download(
+            repo_id=repo_id, filename=path, repo_type="dataset", token=token,
+        )
+    except Exception as e:
+        logger.warning("hf_hub_download(%s) failed: %s", path, e)
+        return None
+    try:
+        with open(local, "r") as f:
+            line = f.readline().strip()
+        if not line:
+            return None
+        row = json.loads(line)
+        # Session uploader stores messages/events/tools as JSON strings.
+        for key in ("messages", "events", "tools"):
+            v = row.get(key)
+            if isinstance(v, str):
+                try:
+                    row[key] = json.loads(v)
+                except Exception:
+                    row[key] = []
+        return row
+    except Exception as e:
+        logger.warning("parse(%s) failed: %s", path, e)
+        return None
+
+
+def _reshape_to_sft(row: dict) -> dict:
+    """Raw passthrough: reshape one session row into SFT format + tags.
+
+    Trajectories predating the ``tools`` addition to ``get_trajectory`` will
+    have an empty tools list — still valid, just less useful downstream.
+    """
+    trajectory = {
+        "events": row.get("events") or [],
+        "messages": row.get("messages") or [],
+        "model_name": row.get("model_name"),
+    }
+    return {
+        "session_id": row.get("session_id"),
+        "model": row.get("model_name"),
+        "timestamp": row.get("session_start_time"),
+        "tags": tag_session(trajectory),
+        "messages": row.get("messages") or [],
+        "tools": row.get("tools") or [],
+    }
+
+
+def _upload_row(api, row: dict, day: date, target_repo: str, token: str) -> None:
+    session_id = row["session_id"]
+    path_in_repo = f"sft/{day.isoformat()}/{session_id}.jsonl"
+    with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as tmp:
+        json.dump(row, tmp, ensure_ascii=False)
+        tmp_path = tmp.name
+    try:
+        api.create_repo(
+            repo_id=target_repo, repo_type="dataset", exist_ok=True, token=token,
+        )
+        api.upload_file(
+            path_or_fileobj=tmp_path,
+            path_in_repo=path_in_repo,
+            repo_id=target_repo,
+            repo_type="dataset",
+            token=token,
+            commit_message=f"Add SFT row {session_id}",
+        )
+    finally:
+        try:
+            os.unlink(tmp_path)
+        except Exception:
+            pass
+
+
+def run_for_day(
+    api, source_repo: str, target_repo: str, day: date, token: str,
+) -> int:
+    paths = _iter_session_files(api, source_repo, day, token)
+    n = 0
+    for path in paths:
+        sess = _download_and_parse(source_repo, path, token)
+        if not sess:
+            continue
+        sft_row = _reshape_to_sft(sess)
+        if not sft_row.get("session_id"):
+            continue
+        try:
+            _upload_row(api, sft_row, day, target_repo, token)
+            n += 1
+        except Exception as e:
+            logger.warning("upload failed for %s: %s", sft_row["session_id"], e)
+    logger.info("Exported %d sessions for %s", n, day)
+    return n
+
+
+def main(argv: list[str] | None = None) -> int:
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--source", default="smolagents/ml-intern-sessions")
+    ap.add_argument("--target", default="smolagents/ml-intern-sft")
+    ap.add_argument(
+        "--days", type=int, default=1,
+        help="Number of trailing days to export (default: 1 = yesterday).",
+    )
+    ap.add_argument(
+        "--date", type=str, default=None,
+        help="Single YYYY-MM-DD to export; overrides --days.",
+    )
+    args = ap.parse_args(argv)
+
+    token = (
+        os.environ.get("HF_SFT_WRITE_TOKEN")
+        or os.environ.get("HF_SESSION_UPLOAD_TOKEN")
+        or os.environ.get("HF_TOKEN")
+        or os.environ.get("HF_ADMIN_TOKEN")
+    )
+    if not token:
+        logger.error(
+            "No HF token found. Set one of: HF_SFT_WRITE_TOKEN, "
+            "HF_SESSION_UPLOAD_TOKEN, HF_TOKEN, HF_ADMIN_TOKEN."
+        )
+        return 1
+
+    from huggingface_hub import HfApi
+    api = HfApi()
+
+    if args.date:
+        target_days = [date.fromisoformat(args.date)]
+    else:
+        today = datetime.now(timezone.utc).date()
+        target_days = [today - timedelta(days=i) for i in range(1, args.days + 1)]
+
+    total = 0
+    for day in target_days:
+        total += run_for_day(api, args.source, args.target, day, token)
+    logger.info("Total exported: %d sessions", total)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/unit/test_build_kpis.py

@@ -0,0 +1,164 @@
+"""Unit tests for the KPI rollup math.
+
+We exercise the pure functions (``_session_metrics`` and ``_aggregate_day``)
+on hand-crafted session trajectories — no network, no HF Hub.
+"""
+
+import importlib.util
+import sys
+from pathlib import Path
+
+
+def _load():
+    """Load ``scripts/build_kpis.py`` without treating ``scripts`` as a package."""
+    path = Path(__file__).parent.parent.parent / "scripts" / "build_kpis.py"
+    spec = importlib.util.spec_from_file_location("build_kpis", path)
+    mod = importlib.util.module_from_spec(spec)
+    sys.modules["build_kpis"] = mod
+    spec.loader.exec_module(mod)  # type: ignore
+    return mod
+
+
+def _ev(event_type, data=None, ts="2026-04-24T10:00:00"):
+    return {"timestamp": ts, "event_type": event_type, "data": data or {}}
+
+
+def _session(events, user_id="u1", start="2026-04-24T09:59:00"):
+    return {
+        "session_id": "sess-" + user_id,
+        "session_start_time": start,
+        "session_end_time": "2026-04-24T10:05:00",
+        "model_name": "claude-opus-4-6",
+        "messages": [{"role": "user", "content": "hi"}],
+        "events": events,
+        "user_id": user_id,
+    }
+
+
+def test_llm_call_accumulates_tokens_and_cost():
+    mod = _load()
+    events = [
+        _ev("llm_call", {
+            "prompt_tokens": 100, "completion_tokens": 50,
+            "cache_read_tokens": 40, "cache_creation_tokens": 10,
+            "cost_usd": 0.01,
+        }),
+        _ev("llm_call", {
+            "prompt_tokens": 200, "completion_tokens": 100,
+            "cache_read_tokens": 80, "cost_usd": 0.02,
+        }),
+    ]
+    m = mod._session_metrics(_session(events))
+    assert m["llm_calls"] == 2
+    assert m["tokens_prompt"] == 300
+    assert m["tokens_completion"] == 150
+    assert m["tokens_cache_read"] == 120
+    assert m["tokens_cache_creation"] == 10
+    assert abs(m["cost_usd"] - 0.03) < 1e-9
+
+
+def test_tool_success_rate_and_first_action():
+    mod = _load()
+    events = [
+        _ev("tool_call", {"tool": "bash"}, ts="2026-04-24T10:00:05"),
+        _ev("tool_output", {"success": True}),
+        _ev("tool_output", {"success": False}),
+    ]
+    m = mod._session_metrics(_session(events))
+    assert m["tool_calls_total"] == 2
+    assert m["tool_calls_success"] == 1
+    # 65s from start to first action
+    assert m["first_tool_s"] == 65
+
+
+def test_hf_job_gpu_hours():
+    mod = _load()
+    events = [
+        _ev("hf_job_submit", {"flavor": "a100-large", "job_id": "j1"}),
+        _ev("hf_job_complete", {
+            "flavor": "a100-large",
+            "final_status": "COMPLETED",
+            "wall_time_s": 3600,
+        }),
+    ]
+    m = mod._session_metrics(_session(events))
+    assert m["hf_jobs_submitted"] == 1
+    assert m["hf_jobs_succeeded"] == 1
+    # a100-large = 1 gpu * 1 hour = 1 gpu-hour
+    assert abs(m["_gpu_hours_by_flavor"]["a100-large"] - 1.0) < 1e-6
+
+
+def test_feedback_counts():
+    mod = _load()
+    events = [
+        _ev("feedback", {"rating": "up"}),
+        _ev("feedback", {"rating": "up"}),
+        _ev("feedback", {"rating": "down"}),
+    ]
+    m = mod._session_metrics(_session(events))
+    assert m["thumbs_up"] == 2
+    assert m["thumbs_down"] == 1
+
+
+def test_aggregate_day_cache_hit_and_users():
+    mod = _load()
+    s1 = mod._session_metrics(_session(
+        [_ev("llm_call", {"prompt_tokens": 100, "cache_read_tokens": 400, "cost_usd": 0.5})],
+        user_id="u1",
+    ))
+    s2 = mod._session_metrics(_session(
+        [_ev("llm_call", {"prompt_tokens": 200, "cache_read_tokens": 100, "cost_usd": 1.0})],
+        user_id="u2",
+    ))
+    row = mod._aggregate_day([s1, s2])
+    assert row["sessions"] == 2
+    assert row["users"] == 2
+    assert row["tokens_prompt"] == 300
+    assert row["tokens_cache_read"] == 500
+    # 500 / (500 + 300) = 0.625
+    assert abs(row["cache_hit_ratio"] - 0.625) < 1e-9
+    assert abs(row["cost_usd"] - 1.5) < 1e-9
+
+
+def test_failure_and_regenerate_rates():
+    mod = _load()
+    s1 = mod._session_metrics(_session([_ev("error", {"error": "boom"})], user_id="a"))
+    s2 = mod._session_metrics(_session([_ev("undo_complete")], user_id="b"))
+    s3 = mod._session_metrics(_session([], user_id="c"))
+    row = mod._aggregate_day([s1, s2, s3])
+    assert row["failure_rate"] == round(1 / 3, 4)
+    assert row["regenerate_rate"] == round(1 / 3, 4)
+
+
+def test_window_filter_keeps_only_events_in_range():
+    from datetime import datetime, timezone
+    mod = _load()
+    events = [
+        _ev("llm_call", {"prompt_tokens": 100}, ts="2026-04-24T09:45:00"),
+        _ev("llm_call", {"prompt_tokens": 200}, ts="2026-04-24T10:05:00"),
+        _ev("tool_call", {"tool": "bash"}, ts="2026-04-24T10:30:00"),
+        _ev("llm_call", {"prompt_tokens": 400}, ts="2026-04-24T11:10:00"),
+    ]
+    session = _session(events, start="2026-04-24T09:44:00")
+    # Only events in [10:00, 11:00) should remain.
+    window_start = datetime(2026, 4, 24, 10, 0, 0, tzinfo=timezone.utc)
+    window_end = datetime(2026, 4, 24, 11, 0, 0, tzinfo=timezone.utc)
+    windowed = mod._filter_session_to_window(session, window_start, window_end)
+    assert windowed is not None
+    types = [e["event_type"] for e in windowed["events"]]
+    assert types == ["llm_call", "tool_call"]
+    # Metrics only reflect in-window events.
+    m = mod._session_metrics(windowed)
+    assert m["tokens_prompt"] == 200
+    assert m["llm_calls"] == 1
+    assert m["tool_calls_total"] == 0  # tool_call not tool_output
+
+
+def test_window_filter_returns_none_when_nothing_in_range():
+    from datetime import datetime, timezone
+    mod = _load()
+    events = [_ev("llm_call", {"prompt_tokens": 100}, ts="2026-04-24T09:45:00")]
+    session = _session(events)
+    window_start = datetime(2026, 4, 24, 10, 0, 0, tzinfo=timezone.utc)
+    window_end = datetime(2026, 4, 24, 11, 0, 0, tzinfo=timezone.utc)
+    assert mod._filter_session_to_window(session, window_start, window_end) is None

tests/unit/test_build_sft.py

@@ -0,0 +1,78 @@
+"""Smoke test for the SFT reshape — raw passthrough with tags attached."""
+
+import importlib.util
+import sys
+from pathlib import Path
+
+
+def _load():
+    path = Path(__file__).parent.parent.parent / "scripts" / "build_sft.py"
+    spec = importlib.util.spec_from_file_location("build_sft", path)
+    mod = importlib.util.module_from_spec(spec)
+    sys.modules["build_sft"] = mod
+    spec.loader.exec_module(mod)  # type: ignore
+    return mod
+
+
+def _session_row():
+    return {
+        "session_id": "abc",
+        "session_start_time": "2026-04-24T10:00:00",
+        "session_end_time": "2026-04-24T10:05:00",
+        "model_name": "claude-opus-4-6",
+        "messages": [
+            {"role": "system", "content": "You are an agent"},
+            {"role": "user", "content": "fine-tune llama"},
+            {"role": "assistant", "content": None, "tool_calls": [
+                {"id": "c1", "type": "function",
+                 "function": {"name": "hf_jobs", "arguments": '{"script":"from trl import SFTTrainer"}'}},
+            ]},
+            {"role": "tool", "tool_call_id": "c1", "content": "ok"},
+            {"role": "assistant", "content": "done"},
+        ],
+        "events": [
+            {"timestamp": "2026-04-24T10:00:05", "event_type": "tool_call",
+             "data": {"tool": "hf_jobs",
+                      "arguments": {"script": "from trl import SFTTrainer"}}},
+            {"timestamp": "2026-04-24T10:00:06", "event_type": "hf_job_submit",
+             "data": {"flavor": "a100-large", "push_to_hub": True}},
+            {"timestamp": "2026-04-24T10:45:00", "event_type": "hf_job_complete",
+             "data": {"flavor": "a100-large", "final_status": "COMPLETED",
+                      "wall_time_s": 2700}},
+            {"timestamp": "2026-04-24T10:45:05", "event_type": "turn_complete",
+             "data": {}},
+        ],
+        "tools": [{"type": "function", "function": {"name": "hf_jobs"}}],
+    }
+
+
+def test_reshape_preserves_messages_and_tools_and_adds_tags():
+    mod = _load()
+    row = mod._reshape_to_sft(_session_row())
+    assert row["session_id"] == "abc"
+    assert row["model"] == "claude-opus-4-6"
+    assert row["timestamp"] == "2026-04-24T10:00:00"
+    # Messages preserved verbatim, in order, with tool_calls + tool role rows.
+    assert len(row["messages"]) == 5
+    assert row["messages"][2]["tool_calls"][0]["function"]["name"] == "hf_jobs"
+    assert row["messages"][3]["role"] == "tool"
+    # Tools preserved verbatim.
+    assert row["tools"] == [{"type": "function", "function": {"name": "hf_jobs"}}]
+    # Tags include the expected signals.
+    tags = set(row["tags"])
+    assert "tool:hf_jobs" in tags
+    assert "hf_job:succeeded" in tags
+    assert "hf_job:push_to_hub" in tags
+    assert "gpu:a100" in tags
+    assert "outcome:completed" in tags
+    assert "task:training" in tags
+    assert "model:opus" in tags
+
+
+def test_reshape_handles_missing_tools_field():
+    mod = _load()
+    row = _session_row()
+    del row["tools"]
+    out = mod._reshape_to_sft(row)
+    assert out["tools"] == []
+    assert isinstance(out["tags"], list)  # still computes tags

tests/unit/test_heartbeat.py

@@ -0,0 +1,134 @@
+"""Heartbeat + stable-local-path tests for Session.
+
+We don't spin up the real agent loop — we build a minimal Session with a
+stubbed config and an in-memory queue, then call send_event repeatedly while
+monkeypatching time.monotonic to simulate seconds passing.
+"""
+
+import asyncio
+import json
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from agent.core.session import Event, Session
+
+
+class _FakeConfig:
+    model_name = "claude-opus-4-6"
+    save_sessions = True
+    session_dataset_repo = "fake/repo"
+    auto_save_interval = 1
+    heartbeat_interval_s = 60
+    max_iterations = 10
+    yolo_mode = False
+    confirm_cpu_jobs = False
+    auto_file_upload = False
+    reasoning_effort = None
+    mcpServers: dict = {}
+
+
+def _mk_session(tmp_path: Path) -> Session:
+    import os
+    os.chdir(tmp_path)  # so session_logs/ lands under tmp_path
+    # Stub out the context manager to avoid litellm lookups.
+    from agent.context_manager.manager import ContextManager
+    cm = ContextManager.__new__(ContextManager)
+    cm.items = []
+    cm.tool_specs = []
+    cm.model_max_tokens = 200_000
+    cm.running_context_usage = 0
+    cm.compact_size = 0.1
+    cm.untouched_messages = 5
+    cm.hf_token = None
+    cm.local_mode = True
+    s = Session(
+        event_queue=asyncio.Queue(),
+        config=_FakeConfig(),
+        tool_router=None,
+        context_manager=cm,
+        hf_token=None,
+        local_mode=True,
+    )
+    return s
+
+
+def test_heartbeat_fires_after_interval(tmp_path, monkeypatch):
+    # Use asyncio.run rather than pytest-asyncio so the test works without the
+    # plugin installed (same pattern elsewhere in this repo).
+    async def body():
+        s = _mk_session(tmp_path)
+        calls = []
+
+        def fake_upload(repo_id):
+            calls.append(repo_id)
+            return "fake/path.json"
+
+        monkeypatch.setattr(s, "save_and_upload_detached", fake_upload)
+
+        # t=0: first event, should NOT trigger (initial _last_heartbeat_ts = now)
+        with patch("agent.core.telemetry.time.monotonic", return_value=100.0):
+            s._last_heartbeat_ts = 100.0
+            await s.send_event(Event(event_type="x"))
+        assert calls == []
+
+        # t=+30s: still under interval → no save
+        with patch("agent.core.telemetry.time.monotonic", return_value=130.0):
+            await s.send_event(Event(event_type="y"))
+        assert calls == []
+
+        # t=+61s: over 60s → save fires once
+        with patch("agent.core.telemetry.time.monotonic", return_value=161.0):
+            await s.send_event(Event(event_type="z"))
+        # create_task runs on the event loop; wait for the to_thread to complete
+        await asyncio.sleep(0.05)
+        assert calls == ["fake/repo"]
+
+        # Next event shortly after → no second save (interval resets to 161)
+        with patch("agent.core.telemetry.time.monotonic", return_value=170.0):
+            await s.send_event(Event(event_type="w"))
+        await asyncio.sleep(0.05)
+        assert len(calls) == 1
+
+    asyncio.run(body())
+
+
+def test_stable_local_path_overwrites(tmp_path):
+    import os
+    os.chdir(tmp_path)
+    from agent.context_manager.manager import ContextManager
+    cm = ContextManager.__new__(ContextManager)
+    cm.items = []
+    cm.tool_specs = []
+    cm.model_max_tokens = 200_000
+    cm.running_context_usage = 0
+    cm.compact_size = 0.1
+    cm.untouched_messages = 5
+    cm.hf_token = None
+    cm.local_mode = True
+
+    s = Session(
+        event_queue=asyncio.Queue(),
+        config=_FakeConfig(),
+        tool_router=None,
+        context_manager=cm,
+        hf_token=None,
+        local_mode=True,
+    )
+
+    p1 = s.save_trajectory_local(directory="session_logs")
+    assert p1 is not None
+    p2 = s.save_trajectory_local(directory="session_logs")
+    p3 = s.save_trajectory_local(directory="session_logs")
+    # All three saves land on the same file — heartbeat should not spam files.
+    assert p1 == p2 == p3
+    files = list(Path("session_logs").glob("session_*.json"))
+    # Exactly one final file; the .tmp should be renamed away.
+    assert len(files) == 1
+
+    # File is valid JSON (atomic write → no torn content).
+    with open(p1) as f:
+        data = json.load(f)
+    assert data["session_id"] == s.session_id
+    assert data["upload_status"] == "pending"

tests/unit/test_kpis_scheduler.py

@@ -0,0 +1,107 @@
+"""Smoke tests for backend/kpis_scheduler.py.
+
+Exercise the pure / fast paths only:
+    * token resolution order
+    * build_kpis import path
+    * start()/shutdown() lifecycle without APScheduler actually running a job
+    * backfill() passes the right hour values through to _run_hour
+"""
+
+from __future__ import annotations
+
+import asyncio
+import importlib.util
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+
+def _load():
+    path = Path(__file__).parent.parent.parent / "backend" / "kpis_scheduler.py"
+    spec = importlib.util.spec_from_file_location("kpis_scheduler", path)
+    mod = importlib.util.module_from_spec(spec)
+    sys.modules["kpis_scheduler"] = mod
+    assert spec.loader is not None
+    spec.loader.exec_module(mod)
+    return mod
+
+
+def test_token_resolution_order(monkeypatch):
+    mod = _load()
+    for var in ("HF_KPI_WRITE_TOKEN", "HF_SESSION_UPLOAD_TOKEN", "HF_TOKEN", "HF_ADMIN_TOKEN"):
+        monkeypatch.delenv(var, raising=False)
+    assert mod._resolve_token() is None
+
+    monkeypatch.setenv("HF_ADMIN_TOKEN", "admin")
+    assert mod._resolve_token() == "admin"
+
+    monkeypatch.setenv("HF_TOKEN", "generic")
+    assert mod._resolve_token() == "generic"
+
+    monkeypatch.setenv("HF_SESSION_UPLOAD_TOKEN", "sessions")
+    assert mod._resolve_token() == "sessions"
+
+    monkeypatch.setenv("HF_KPI_WRITE_TOKEN", "kpis")
+    assert mod._resolve_token() == "kpis"
+
+
+def test_load_build_kpis_exposes_run_for_hour():
+    mod = _load()
+    bk = mod._load_build_kpis()
+    assert hasattr(bk, "run_for_hour")
+    assert callable(bk.run_for_hour)
+
+
+def test_backfill_calls_run_hour_for_each_hour(monkeypatch):
+    mod = _load()
+    monkeypatch.setenv("HF_KPI_WRITE_TOKEN", "x")
+    calls: list[datetime] = []
+
+    async def fake_run_hour(hour_dt):
+        calls.append(hour_dt)
+
+    monkeypatch.setattr(mod, "_run_hour", fake_run_hour)
+    asyncio.run(mod.backfill(hours=3))
+    assert len(calls) == 3
+    # Hours are returned most-recent-first
+    assert calls[0] > calls[1] > calls[2]
+    # All aligned to the top of the hour
+    for c in calls:
+        assert c.minute == 0 and c.second == 0 and c.microsecond == 0
+        assert c.tzinfo == timezone.utc
+
+
+def test_start_is_no_op_when_disabled(monkeypatch):
+    mod = _load()
+    # Ensure clean state — _scheduler is module-global
+    mod._scheduler = None
+    monkeypatch.setenv("ML_INTERN_KPIS_DISABLED", "1")
+    mod.start()
+    assert mod._scheduler is None  # never instantiated
+
+
+def test_start_skips_cleanly_without_apscheduler(monkeypatch):
+    mod = _load()
+    mod._scheduler = None
+    monkeypatch.delenv("ML_INTERN_KPIS_DISABLED", raising=False)
+
+    # Force the apscheduler import to fail — start() should log and return.
+    real_import = __builtins__["__import__"] if isinstance(__builtins__, dict) else __builtins__.__import__
+
+    def fake_import(name, *args, **kwargs):
+        if name.startswith("apscheduler"):
+            raise ImportError("apscheduler unavailable in test")
+        return real_import(name, *args, **kwargs)
+
+    monkeypatch.setattr(
+        "builtins.__import__",
+        fake_import,
+    )
+    mod.start()  # should not raise
+    assert mod._scheduler is None
+
+
+def test_shutdown_is_no_op_when_not_started():
+    mod = _load()
+    mod._scheduler = None
+    asyncio.run(mod.shutdown())  # must not raise

tests/unit/test_redact.py

@@ -0,0 +1,76 @@
+"""Tests for the secret scrubber used before session upload."""
+
+from agent.core.redact import scrub, scrub_string
+
+
+def test_hf_token():
+    s = "here is a token hf_" + "A" * 35 + " ok"
+    out = scrub_string(s)
+    assert "hf_" not in out
+    assert "[REDACTED_HF_TOKEN]" in out
+
+
+def test_anthropic_key():
+    s = "key=sk-ant-api03_" + "a" * 40
+    out = scrub_string(s)
+    # The env-var name prefix matches too; just verify we don't leave the body.
+    assert "sk-ant-api03_" not in out
+
+
+def test_github_token():
+    s = "ghp_" + "a" * 40
+    out = scrub_string(s)
+    assert out == "[REDACTED_GITHUB_TOKEN]"
+
+
+def test_github_fine_grained_pat():
+    # Fine-grained PATs: github_pat_<alphanumeric + underscore>, 36+ chars
+    s = "github_pat_" + "A1B2_" * 10
+    out = scrub_string(s)
+    assert "github_pat_" not in out
+    assert "[REDACTED_GITHUB_TOKEN]" in out
+
+
+def test_aws_key_id():
+    s = "AWS_ACCESS_KEY_ID=AKIAABCDEFGHIJKLMNOP"
+    out = scrub_string(s)
+    assert "AKIAABCDEFGHIJKLMNOP" not in out
+
+
+def test_bearer_header():
+    s = "Authorization: Bearer abcdef0123456789abcdef0123456789"
+    out = scrub_string(s)
+    assert "abcdef0123456789abcdef0123456789" not in out
+    assert "Bearer [REDACTED]" in out
+
+
+def test_env_var_style():
+    s = "HF_TOKEN=hf_" + "x" * 40 + " run"
+    out = scrub_string(s)
+    # Either the value-scrubber or the HF-token regex should fire.
+    assert "hf_xxxx" not in out
+
+
+def test_scrub_nested_dict_and_list():
+    payload = {
+        "msg": "token hf_" + "Z" * 35,
+        "tools": [
+            {"args": {"secret": "ghp_" + "Q" * 40}},
+            "no secrets here",
+        ],
+        "n": 42,
+    }
+    out = scrub(payload)
+    # Original not mutated
+    assert "hf_" in payload["msg"]
+    # Redacted copy
+    assert "[REDACTED_HF_TOKEN]" in out["msg"]
+    assert out["tools"][0]["args"]["secret"] == "[REDACTED_GITHUB_TOKEN]"
+    assert out["tools"][1] == "no secrets here"
+    assert out["n"] == 42
+
+
+def test_scrub_preserves_non_strings():
+    assert scrub(None) is None
+    assert scrub(123) == 123
+    assert scrub(True) is True

tests/unit/test_sft_tagger.py

@@ -0,0 +1,197 @@
+"""Tests for agent.sft.tagger — one test per tag namespace."""
+
+from agent.sft.tagger import tag_session
+
+
+def _ev(event_type, data=None, ts="2026-04-24T10:00:00"):
+    return {"timestamp": ts, "event_type": event_type, "data": data or {}}
+
+
+def _traj(events=None, messages=None, model="claude-opus-4-6"):
+    return {
+        "session_id": "sess-1",
+        "model_name": model,
+        "session_start_time": "2026-04-24T09:59:00",
+        "session_end_time": "2026-04-24T10:05:00",
+        "messages": messages
+        or [{"role": "user", "content": "hi"}, {"role": "assistant", "content": "ok"}],
+        "events": events or [],
+    }
+
+
+def test_model_family():
+    assert "model:opus" in tag_session(_traj(model="claude-opus-4-6"))
+    assert "model:sonnet" in tag_session(_traj(model="bedrock/claude-sonnet-4-5"))
+    assert "model:kimi" in tag_session(_traj(model="moonshotai/Kimi-K2.6"))
+    assert "model:other" in tag_session(_traj(model="unknown-model-xyz"))
+
+
+def test_turns_buckets():
+    short = _traj(messages=[{"role": "user", "content": "hi"}])
+    medium = _traj(messages=[{"role": "user", "content": "q"} for _ in range(10)])
+    long = _traj(messages=[{"role": "user", "content": "q"} for _ in range(25)])
+    assert "turns:short" in tag_session(short)
+    assert "turns:medium" in tag_session(medium)
+    assert "turns:long" in tag_session(long)
+
+
+def test_cost_buckets():
+    cheap = _traj(events=[_ev("llm_call", {"cost_usd": 0.05})])
+    med = _traj(events=[_ev("llm_call", {"cost_usd": 0.5})])
+    expensive = _traj(events=[_ev("llm_call", {"cost_usd": 5.0})])
+    assert "cost:low" in tag_session(cheap)
+    assert "cost:med" in tag_session(med)
+    assert "cost:high" in tag_session(expensive)
+
+
+def test_tool_tags():
+    events = [
+        _ev("tool_call", {"tool": "hf_jobs", "arguments": {}}),
+        _ev("tool_call", {"tool": "research"}),
+        _ev("tool_call", {"tool": "bash"}),
+    ]
+    tags = tag_session(_traj(events))
+    assert "tool:hf_jobs" in tags
+    assert "tool:research" in tags
+    assert "tool:bash" in tags
+
+
+def test_outcome_completed():
+    events = [_ev("turn_complete", {"history_size": 10})]
+    assert "outcome:completed" in tag_session(_traj(events))
+
+
+def test_outcome_errored():
+    events = [_ev("error", {"error": "boom"})]
+    assert "outcome:errored" in tag_session(_traj(events))
+
+
+def test_outcome_interrupted():
+    events = [_ev("interrupted")]
+    assert "outcome:interrupted" in tag_session(_traj(events))
+
+
+def test_outcome_ongoing():
+    # No terminal events → session was still running at save time
+    events = [_ev("llm_call", {"cost_usd": 0.01})]
+    assert "outcome:ongoing" in tag_session(_traj(events))
+
+
+def test_outcome_doom_loop_and_context():
+    events = [
+        _ev("tool_log", {"tool": "system", "log": "Doom loop detected — injecting corrective prompt"}),
+        _ev("compacted", {"old_tokens": 100, "new_tokens": 50}),
+        _ev("turn_complete", {"history_size": 10}),
+    ]
+    tags = tag_session(_traj(events))
+    assert "outcome:doom_loop" in tags
+    assert "outcome:context_exceeded" in tags
+
+
+def test_hf_job_tags():
+    events = [
+        _ev("tool_call", {"tool": "hf_jobs", "arguments": {"script": "from trl import SFTTrainer"}}),
+        _ev("hf_job_submit", {
+            "flavor": "a100-large", "push_to_hub": True, "job_id": "j1",
+        }),
+        _ev("hf_job_complete", {"flavor": "a100-large", "final_status": "COMPLETED", "wall_time_s": 3600}),
+        _ev("hf_job_submit", {"flavor": "a100x4", "push_to_hub": False}),
+        _ev("hf_job_complete", {"flavor": "a100x4", "final_status": "FAILED"}),
+    ]
+    tags = tag_session(_traj(events))
+    assert "hf_job:submitted" in tags
+    assert "hf_job:multi" in tags
+    assert "hf_job:succeeded" in tags
+    assert "hf_job:failed" in tags
+    assert "hf_job:push_to_hub" in tags
+    assert "gpu:a100" in tags
+    assert "gpu:multi" in tags
+
+
+def test_hf_job_oom():
+    events = [
+        _ev("tool_call", {"tool": "hf_jobs", "arguments": {}}),
+        _ev("hf_job_submit", {"flavor": "a100-large"}),
+        _ev("tool_output", {"success": False, "output": "RuntimeError: CUDA out of memory. Tried to allocate..."}),
+    ]
+    tags = tag_session(_traj(events))
+    assert "hf_job:oom" in tags
+
+
+def test_sandbox_tags():
+    events = [
+        _ev("sandbox_create", {"hardware": "t4-small", "sandbox_id": "s1", "create_latency_s": 5}),
+        _ev("sandbox_destroy", {"sandbox_id": "s1", "lifetime_s": 3600}),
+    ]
+    tags = tag_session(_traj(events))
+    assert "sandbox:created" in tags
+    assert "sandbox:gpu" in tags
+    assert "sandbox:long_lived" in tags
+
+
+def test_sandbox_cpu_short():
+    events = [
+        _ev("sandbox_create", {"hardware": "cpu-basic"}),
+        _ev("sandbox_destroy", {"lifetime_s": 120}),
+    ]
+    tags = tag_session(_traj(events))
+    assert "sandbox:cpu" in tags
+    assert "sandbox:long_lived" not in tags
+
+
+def test_feedback_tags():
+    up_only = _traj(events=[_ev("feedback", {"rating": "up"})])
+    down_only = _traj(events=[_ev("feedback", {"rating": "down"})])
+    mixed = _traj(events=[_ev("feedback", {"rating": "up"}), _ev("feedback", {"rating": "down"})])
+    none = _traj()
+    assert "feedback:up" in tag_session(up_only)
+    assert "feedback:down" in tag_session(down_only)
+    assert "feedback:mixed" in tag_session(mixed)
+    assert "feedback:none" in tag_session(none)
+
+
+def test_task_training():
+    events = [
+        _ev("tool_call", {"tool": "hf_jobs", "arguments": {
+            "script": "from trl import SFTTrainer\ntrainer = SFTTrainer(...)"
+        }}),
+        _ev("hf_job_submit", {"flavor": "a100-large"}),
+    ]
+    assert "task:training" in tag_session(_traj(events))
+
+
+def test_task_research_only():
+    events = [
+        _ev("tool_call", {"tool": "research"}),
+        _ev("tool_call", {"tool": "explore_hf_docs"}),
+    ]
+    assert "task:research_only" in tag_session(_traj(events))
+
+
+def test_task_data_prep():
+    events = [
+        _ev("tool_call", {"tool": "hf_inspect_dataset", "arguments": {}}),
+        _ev("tool_call", {"tool": "hub_repo_details"}),
+    ]
+    tags = tag_session(_traj(events))
+    assert "task:data_prep" in tags
+
+
+def test_no_duplicates_and_sorted():
+    events = [
+        _ev("tool_call", {"tool": "hf_jobs"}),
+        _ev("tool_call", {"tool": "hf_jobs"}),  # duplicate
+        _ev("hf_job_submit", {"flavor": "a10g-small"}),
+        _ev("hf_job_submit", {"flavor": "a10g-small"}),
+    ]
+    tags = tag_session(_traj(events))
+    assert tags == sorted(tags)
+    assert len(tags) == len(set(tags))
+
+
+def test_empty_trajectory_has_required_tags():
+    tags = tag_session(_traj())
+    namespaces = {t.split(":", 1)[0] for t in tags}
+    # Every session must have at least model/turns/cost/outcome/feedback.
+    for required in ("model", "turns", "cost", "outcome", "feedback"):
+        assert required in namespaces, f"missing {required} — got {tags}"