feat(shared-metrics): emit tool_search, tool_batch and connector_chain rows

tool_search: one row per search with the catalogs read, a hits bucket and the
outcome. tool_batch: one row per model reply that calls tools, counting each
tool_call entry. connector_chain: one row per executed connector entry with the
searches since the last execute and whether the tool was described first.

hermes_cli/observability/shared_metrics_events.py

@@ -51,6 +51,12 @@ def record_slash_command(*, command: str, surface: str) -> None:
     _emit(contract.SLASH_COMMAND_MARK, fields_.slash_command_fields, command=command, surface=surface)
 
 
+def record_tool_search(*, scope: str, hits: int, outcome: str) -> None:
+    from .shared_metrics_connectors import tool_search_fields
+
+    _emit(contract.TOOL_SEARCH_MARK, tool_search_fields, scope=scope, hits=hits, outcome=outcome)
+
+
 def record_compression(
     *, trigger: str, outcome: str, tokens_before: int | None, context_length: int | None,
     failure_class: str | None = None,

hermes_cli/observability/shared_metrics_model.py

@@ -166,9 +166,13 @@ def record_tool_call_quality(agent: Any, tool_calls: Iterable[Any], repaired_ids
         if not tool_calls or not enabled():
             return
         from .shared_metrics_signals import tool_unavailable_fields
+        from .shared_metrics_connectors import tool_batch_fields
         route = model_route(getattr(agent, "provider", None), getattr(agent, "model", None))
         valid_names = getattr(agent, "valid_tool_names", None) or frozenset()
         required = _required_params(getattr(agent, "tools", None))
+        tool_calls = list(tool_calls)
+        if (batch := tool_batch_fields(route, tool_calls)) is not None:
+            record_process_mark(contract.TOOL_BATCH_MARK, batch)
         for tc in tool_calls:
             fn = tc.function
             issue = tool_call_issue(

tools/tool_search.py

@@ -434,6 +434,7 @@ def dispatch_tool_search(args: dict[str, Any], *, current_tool_defs: list[dict[s
     queries, err = _string_list_arg(args, "queries", dedupe=False, max_items=_MAX_QUERIES_PER_CALL,
                                     retry_hint="Retry with fewer, more targeted queries.")
     if err:
+        _record_search_metric(current_tool_defs, hits=0, outcome="error")
         return err
     raw_limit = args.get("limit")
     limit = (config.search_default_limit if raw_limit is None
@@ -467,9 +468,22 @@ def dispatch_tool_search(args: dict[str, Any], *, current_tool_defs: list[dict[s
                                "results": results, "tools": tools_map}
     if hosted_failure:
         payload["connectors"] = connectors_unavailable(hosted_failure, verb="searched")
+    _record_search_metric(current_tool_defs, catalog=catalog, hits=len(tools_map),
+                          outcome="connector_unavailable" if hosted_failure else "hits" if tools_map else "no_hits")
     return json.dumps(payload, ensure_ascii=False)
 
 
+def _record_search_metric(current_tool_defs: list[dict[str, Any]], *, hits: int, outcome: str,
+                          catalog: Optional[list[CatalogEntry]] = None) -> None:
+    """One ``hermes.tool_search.count`` row: which corpora the search read, never the query."""
+    from hermes_cli.observability.shared_metrics_events import record_tool_search
+
+    local = bool(catalog if catalog is not None else _deferrable_in(current_tool_defs))
+    hosted = connections_in_scope(current_tool_defs)
+    scope = "both" if local and hosted else "connector" if hosted else "local"
+    record_tool_search(scope=scope, hits=hits, outcome=outcome)
+
+
 def dispatch_tool_describe(args: dict[str, Any], *, current_tool_defs: list[dict[str, Any]],
                            config: Optional[ToolSearchConfig] = None,
                            connector_describe: Optional[Any] = None, call: Optional[Any] = None) -> str: