Add tool label registry and SSE v2 tool event shaping

Pure functions build tool_start/tool_end payloads with run_id, friendly
label, truncated input (config dropped), status, duration and artifact ref.
Labels cover every agent tool, both old and #55 YNAB names, and the
homebox_/bookstack_ MCP prefixes, falling back to "Using <tool>…".

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

blueprints/conversation/tool_events.py

@@ -0,0 +1,77 @@
+"""Pure shaping of LangChain tool events into SSE `tool_start` / `tool_end` payloads."""
+
+import json
+import re
+
+from .tool_labels import tool_label
+
+INPUT_VALUE_MAX_CHARS = 200
+# Injected by LangChain rather than chosen by the model; never shown to the client.
+_HIDDEN_INPUT_KEYS = frozenset({"config", "callbacks", "run_manager"})
+
+_TOOL_HEADER_RE = re.compile(r"^\[(\S+) result\]")
+
+
+def _truncate(text: str) -> str:
+    if len(text) > INPUT_VALUE_MAX_CHARS:
+        return text[: INPUT_VALUE_MAX_CHARS - 1] + "…"
+    return text
+
+
+def summarize_tool_input(raw) -> dict[str, str | int | float | bool]:
+    """Reduce a tool's input to flat, display-safe scalars."""
+    if raw is None:
+        return {}
+    if not isinstance(raw, dict):
+        return {"input": _truncate(str(raw))}
+    summary: dict[str, str | int | float | bool] = {}
+    for key, value in raw.items():
+        if key in _HIDDEN_INPUT_KEYS or value is None:
+            continue
+        if isinstance(value, (bool, int, float)):
+            summary[key] = value
+        elif isinstance(value, str):
+            summary[key] = _truncate(value)
+        else:
+            try:
+                summary[key] = _truncate(json.dumps(value, default=str))
+            except (TypeError, ValueError):
+                summary[key] = _truncate(str(value))
+    return summary
+
+
+def tool_start_payload(event: dict) -> dict:
+    name = event["name"]
+    return {
+        "type": "tool_start",
+        "run_id": event["run_id"],
+        "tool": name,
+        "label": tool_label(name),
+        "input": summarize_tool_input(event.get("data", {}).get("input")),
+    }
+
+
+def tool_end_payload(
+    event: dict,
+    started_at: float | None,
+    artifact_ref: dict | None,
+    *,
+    now: float,
+    status: str = "ok",
+) -> dict:
+    name = event["name"]
+    return {
+        "type": "tool_end",
+        "run_id": event["run_id"],
+        "tool": name,
+        "label": tool_label(name),
+        "status": status,
+        "ms": round((now - started_at) * 1000) if started_at is not None else 0,
+        "artifact": artifact_ref,
+    }
+
+
+def tool_name_from_message(text: str) -> str:
+    """Recover the tool name from a persisted `[tool_name result]` header."""
+    match = _TOOL_HEADER_RE.match(text)
+    return match.group(1) if match else "tool"

blueprints/conversation/tool_labels.py

@@ -0,0 +1,64 @@
+"""Friendly, user-facing labels for agent tool calls (shown in the activity timeline)."""
+
+TOOL_LABELS: dict[str, str] = {
+    "get_current_date": "Checking the date…",
+    "web_search": "Searching the web…",
+    "simba_search": "Searching household documents…",
+    "save_user_memory": "Saving to memory…",
+    # YNAB (current names)
+    "ynab_budget_summary": "Checking budget…",
+    "ynab_search_transactions": "Searching transactions…",
+    "ynab_category_spending": "Reviewing category spending…",
+    "ynab_category_allocations": "Reviewing category budgets…",
+    "ynab_insights": "Analyzing spending trends…",
+    # YNAB (names after the #55 mirror)
+    "ynab_account_balances": "Checking account balances…",
+    "ynab_spending_report": "Building spending report…",
+    "ynab_category_budget": "Reviewing category budgets…",
+    "ynab_recent_changes": "Checking recent budget changes…",
+    "ynab_sql": "Querying budget data…",
+    # Mealie
+    "mealie_todays_meals": "Checking today's meals…",
+    "mealie_meal_plans": "Checking the meal plan…",
+    "mealie_get_recipe": "Fetching recipe…",
+    "mealie_shopping_lists": "Checking shopping lists…",
+    "mealie_create_meal_plan": "Adding to the meal plan…",
+    # Obsidian / journal
+    "obsidian_search_notes": "Searching notes…",
+    "obsidian_read_note": "Reading note…",
+    "obsidian_create_note": "Creating note…",
+    "obsidian_create_task": "Creating task…",
+    "journal_get_today": "Reading today's journal…",
+    "journal_get_tasks": "Checking tasks…",
+    "journal_add_task": "Adding task…",
+    "journal_complete_task": "Completing task…",
+    # Google
+    "get_calendar_events": "Checking calendar…",
+    "create_calendar_event": "Creating calendar event…",
+    "gmail_search_emails": "Searching email…",
+    "gmail_get_email": "Reading email…",
+    # SimbaKit
+    "simbakit_device_status": "Checking Simba's devices…",
+    "simbakit_weight_history": "Checking Simba's weight…",
+    "simbakit_litter_activity": "Checking litter box activity…",
+    "simbakit_drinking_activity": "Checking water fountain activity…",
+    # Beli
+    "beli_search_restaurants": "Searching restaurants…",
+}
+
+# MCP tools are discovered at runtime, so they match by name prefix.
+TOOL_PREFIX_LABELS: tuple[tuple[str, str], ...] = (
+    ("homebox_", "Checking home inventory…"),
+    ("bookstack_", "Checking the wiki…"),
+)
+
+
+def tool_label(tool_name: str) -> str:
+    """Return the friendly label for a tool, falling back to "Using <tool>…"."""
+    label = TOOL_LABELS.get(tool_name)
+    if label:
+        return label
+    for prefix, prefix_label in TOOL_PREFIX_LABELS:
+        if tool_name.startswith(prefix):
+            return prefix_label
+    return f"Using {tool_name}…"

tests/unit/test_tool_events.py

@@ -0,0 +1,110 @@
+"""Tests for SSE tool-event shaping and the tool label registry."""
+
+from langchain_core.tools import BaseTool
+
+from blueprints.conversation import agents
+from blueprints.conversation.tool_events import (
+    INPUT_VALUE_MAX_CHARS,
+    summarize_tool_input,
+    tool_end_payload,
+    tool_name_from_message,
+    tool_start_payload,
+)
+from blueprints.conversation.tool_labels import TOOL_LABELS, tool_label
+
+
+def _event(name="web_search", run_id="run-1", input=None):
+    return {"name": name, "run_id": run_id, "data": {"input": input or {}}}
+
+
+class TestToolLabel:
+    def test_known_tool(self):
+        assert tool_label("web_search") == "Searching the web…"
+
+    def test_fallback(self):
+        assert tool_label("mystery_tool") == "Using mystery_tool…"
+
+    def test_mcp_prefixes(self):
+        assert tool_label("homebox_search_items") == "Checking home inventory…"
+        assert tool_label("bookstack_get_page") == "Checking the wiki…"
+
+    def test_old_and_new_ynab_names(self):
+        for name in (
+            "ynab_category_spending",
+            "ynab_insights",
+            "ynab_account_balances",
+            "ynab_spending_report",
+            "ynab_category_budget",
+            "ynab_recent_changes",
+            "ynab_sql",
+        ):
+            assert name in TOOL_LABELS
+
+    def test_every_agent_tool_has_a_label(self):
+        names = [v.name for v in vars(agents).values() if isinstance(v, BaseTool)]
+        assert len(names) > 20
+        assert [n for n in names if n not in TOOL_LABELS] == []
+
+
+class TestSummarizeToolInput:
+    def test_drops_config_and_none(self):
+        raw = {"content": "hi", "config": {"configurable": {}}, "date": None}
+        assert summarize_tool_input(raw) == {"content": "hi"}
+
+    def test_truncates_long_strings(self):
+        out = summarize_tool_input({"query": "x" * 500})
+        assert len(out["query"]) == INPUT_VALUE_MAX_CHARS
+        assert out["query"].endswith("…")
+
+    def test_keeps_scalars(self):
+        assert summarize_tool_input({"days": 7, "flag": True, "f": 1.5}) == {
+            "days": 7,
+            "flag": True,
+            "f": 1.5,
+        }
+
+    def test_serialises_nested_values(self):
+        assert summarize_tool_input({"tags": ["a", "b"]}) == {"tags": '["a", "b"]'}
+
+    def test_non_dict_input(self):
+        assert summarize_tool_input("raw") == {"input": "raw"}
+        assert summarize_tool_input(None) == {}
+
+
+class TestPayloads:
+    def test_tool_start(self):
+        payload = tool_start_payload(_event(input={"query": "cats"}))
+        assert payload == {
+            "type": "tool_start",
+            "run_id": "run-1",
+            "tool": "web_search",
+            "label": "Searching the web…",
+            "input": {"query": "cats"},
+        }
+
+    def test_tool_end_ok(self):
+        ref = {"id": "a1", "kind": "mealie.recipe", "title": "Soup"}
+        payload = tool_end_payload(_event(), 10.0, ref, now=10.25)
+        assert payload == {
+            "type": "tool_end",
+            "run_id": "run-1",
+            "tool": "web_search",
+            "label": "Searching the web…",
+            "status": "ok",
+            "ms": 250,
+            "artifact": ref,
+        }
+
+    def test_tool_end_error_without_start(self):
+        payload = tool_end_payload(_event(), None, None, now=5.0, status="error")
+        assert payload["status"] == "error"
+        assert payload["ms"] == 0
+        assert payload["artifact"] is None
+
+
+class TestToolNameFromMessage:
+    def test_parses_header(self):
+        assert tool_name_from_message("[web_search result]\nbody") == "web_search"
+
+    def test_unknown_header(self):
+        assert tool_name_from_message("something else") == "tool"