feat(python): usage-estimation bindings in ratel-ai

src/sdk/python/CHANGELOG.md

@@ -6,6 +6,10 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/) and
 
 ## [Unreleased]
 
+### Added
+
+- **Usage estimation** bound from `ratel-ai-core`: `estimate_tokens` and `estimate_cost_usd`, plus `ToolCatalog(observe=True)` which records the full-catalog-vs-selected-top-K saving on each search into `last_savings` (in-memory, best-effort, never emitted). No wire format — usage/cost telemetry rides the `ratel-ai-cloud` ADR-0013 event. See [ADR-0015](../../../docs/adr/0015-usage-estimation-in-core.md).
+
 ## [0.2.0] - 2026-06-16
 
 ### Changed

src/sdk/python/native/src/lib.rs

@@ -13,6 +13,20 @@ use ratel_ai_core as core;
 use ratel_ai_core::{JsonlSink, MemorySink, NoopSink, Origin, TraceEvent};
 use serde_json::Value;
 
+/// Estimate the token footprint of a string (`len / 4` heuristic). Module-level
+/// mirror of `ratel_ai_core::estimate_tokens`, exposed so the Python SDK reads
+/// token maths from the core instead of re-deriving it.
+#[pyfunction]
+fn estimate_tokens(text: &str) -> u64 {
+    core::estimate_tokens(text)
+}
+
+/// Estimate the USD cost of a generation from its model and token counts.
+#[pyfunction]
+fn estimate_cost_usd(model: &str, input_tokens: u64, output_tokens: u64) -> f64 {
+    core::estimate_cost_usd(model, input_tokens, output_tokens)
+}
+
 /// A single search result: the matched tool id and its BM25 score. Mirrors the
 /// TS SDK's `SearchHit` (camelCase `toolId` there → snake_case `tool_id` here).
 #[pyclass(frozen)]
@@ -93,6 +107,16 @@ impl ToolRegistry {
         Ok(())
     }
 
+    /// Total context-token footprint of the full registered catalog.
+    fn catalog_tokens(&self) -> u64 {
+        self.inner.catalog_tokens()
+    }
+
+    /// Footprint of the tools with the given ids (e.g. a search's hits).
+    fn tokens_for(&self, ids: Vec<String>) -> u64 {
+        self.inner.tokens_for(&ids)
+    }
+
     fn search(&self, query: String, top_k: u32) -> Vec<SearchHit> {
         self.inner
             .search(&query, top_k as usize)
@@ -224,6 +248,16 @@ impl SkillRegistry {
         });
     }
 
+    /// Total context-token footprint of the full registered skill corpus.
+    fn catalog_tokens(&self) -> u64 {
+        self.inner.catalog_tokens()
+    }
+
+    /// Footprint of the skills with the given ids (e.g. a search's hits).
+    fn tokens_for(&self, ids: Vec<String>) -> u64 {
+        self.inner.tokens_for(&ids)
+    }
+
     fn search(&self, query: String, top_k: u32) -> Vec<SkillHit> {
         self.inner
             .search(&query, top_k as usize)
@@ -316,5 +350,7 @@ fn _native(m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_class::<SearchHit>()?;
     m.add_class::<SkillRegistry>()?;
     m.add_class::<SkillHit>()?;
+    m.add_function(pyo3::wrap_pyfunction!(estimate_tokens, m)?)?;
+    m.add_function(pyo3::wrap_pyfunction!(estimate_cost_usd, m)?)?;
     Ok(())
 }

src/sdk/python/ratel_ai/__init__.py

@@ -10,7 +10,14 @@
 - `register_mcp_server` — ingest an upstream MCP server's tools (extra: mcp).
 """
 
-from ._native import SearchHit, SkillHit, SkillRegistry, ToolRegistry
+from ._native import (
+    SearchHit,
+    SkillHit,
+    SkillRegistry,
+    ToolRegistry,
+    estimate_cost_usd,
+    estimate_tokens,
+)
 from .catalog import (
     ExecutableTool,
     Executor,
@@ -56,6 +63,8 @@
     "ToolRegistry",
     "TraceSinkConfig",
     "UpstreamServerInfo",
+    "estimate_cost_usd",
+    "estimate_tokens",
     "format_upstream_line",
     "get_skill_content_tool",
     "invoke_tool_tool",

src/sdk/python/ratel_ai/_native.pyi

@@ -29,6 +29,8 @@ class ToolRegistry:
     ) -> None: ...
     def search(self, query: str, top_k: int) -> list[SearchHit]: ...
     def search_with_origin(self, query: str, top_k: int, origin: str) -> list[SearchHit]: ...
+    def catalog_tokens(self) -> int: ...
+    def tokens_for(self, ids: list[str]) -> int: ...
     def record_event(self, event: dict[str, Any]) -> None: ...
     def set_trace_sink(
         self,
@@ -38,6 +40,9 @@ class ToolRegistry:
     ) -> None: ...
     def drain_trace_events(self) -> list[dict[str, Any]]: ...
 
+def estimate_tokens(text: str) -> int: ...
+def estimate_cost_usd(model: str, input_tokens: int, output_tokens: int) -> float: ...
+
 class SkillHit:
     """A single skill BM25 search result."""
 
@@ -62,6 +67,8 @@ class SkillRegistry:
     ) -> None: ...
     def search(self, query: str, top_k: int) -> list[SkillHit]: ...
     def search_with_origin(self, query: str, top_k: int, origin: str) -> list[SkillHit]: ...
+    def catalog_tokens(self) -> int: ...
+    def tokens_for(self, ids: list[str]) -> int: ...
     def record_event(self, event: dict[str, Any]) -> None: ...
     def set_trace_sink(
         self,

src/sdk/python/ratel_ai/catalog.py

@@ -55,14 +55,27 @@ class TraceSinkConfig:
 
 
 class ToolCatalog:
-    """Registry + executors. Register tools once, then search and invoke by id."""
+    """Registry + executors. Register tools once, then search and invoke by id.
 
-    def __init__(self, trace: TraceSinkConfig | None = None) -> None:
+    Pass `observe=True` to also record Ratel's tokens-saved metric on every search
+    — the full registered catalog vs the selected top-K, computed natively in
+    `ratel-ai-core` — into `last_savings` (in-memory only; nothing is emitted to
+    the trace stream or the network). Omit it and behavior is unchanged.
+    """
+
+    def __init__(
+        self,
+        trace: TraceSinkConfig | None = None,
+        observe: bool = False,
+    ) -> None:
         self._registry = ToolRegistry()
         self._executors: dict[str, Executor] = {}
         self._tools: dict[str, Tool] = {}
         if trace is not None:
             self._registry.set_trace_sink(trace.kind, trace.session_id, trace.path)
+        self._observe = bool(observe)
+        # The most recent search's savings (full vs selected tokens), or None.
+        self.last_savings: dict[str, int] | None = None
 
     def register(self, tool: ExecutableTool) -> None:
         if tool.execute is None:
@@ -83,8 +96,33 @@ def register(self, tool: ExecutableTool) -> None:
             output_schema=tool.output_schema,
         )
 
-    def search(self, query: str, top_k: int, origin: SearchOrigin = "direct") -> list[SearchHit]:
-        return self._registry.search_with_origin(query, top_k, origin)
+    def search(
+        self, query: str, top_k: int, origin: SearchOrigin = "direct"
+    ) -> list[SearchHit]:
+        hits = self._registry.search_with_origin(query, top_k, origin)
+        if self._observe:
+            self._record_savings(hits, top_k)
+        return hits
+
+    def _record_savings(self, hits: list[SearchHit], top_k: int) -> None:
+        """Record the full-catalog-vs-top-K token saving into `last_savings`.
+
+        Best-effort: never raises. The footprint maths run in the core
+        (`catalog_tokens` / `tokens_for`); Python only records the result, in
+        memory — nothing is emitted to the trace stream or the network.
+        """
+        try:
+            full = int(self._registry.catalog_tokens())
+            selected = int(self._registry.tokens_for([hit.tool_id for hit in hits]))
+            saved = max(0, full - selected)
+            self.last_savings = {
+                "full_catalog_tokens": full,
+                "selected_tokens": selected,
+                "tokens_saved": saved,
+                "top_k": top_k,
+            }
+        except Exception:
+            pass
 
     def has(self, tool_id: str) -> bool:
         return tool_id in self._executors

src/sdk/python/tests/test_savings.py

@@ -0,0 +1,91 @@
+"""Tool-selection savings + native usage maths (ADR-0015).
+
+The token/cost/savings maths live in `ratel-ai-core`; these tests pin the native
+surface and the `ToolCatalog(observe=True)` recording that builds on it.
+"""
+
+from __future__ import annotations
+
+from ratel_ai import ExecutableTool, ToolCatalog, TraceSinkConfig, _native
+
+
+def _exec_tool(tool_id: str, description: str) -> ExecutableTool:
+    return ExecutableTool(
+        id=tool_id,
+        name=tool_id,
+        description=description,
+        input_schema={"properties": {"x": {"type": "string"}}},
+        output_schema={},
+        execute=lambda args: {},
+    )
+
+
+# -- native token maths -----------------------------------------------------
+
+
+def test_estimate_tokens_is_char_over_four() -> None:
+    assert _native.estimate_tokens("") == 0
+    assert _native.estimate_tokens("abcd") == 1
+    assert _native.estimate_tokens("a" * 40) == 10
+
+
+def test_estimate_cost_scales_with_model_tier() -> None:
+    opus = _native.estimate_cost_usd("claude-opus-4-8", 1_000_000, 0)
+    haiku = _native.estimate_cost_usd("claude-haiku-4-5", 1_000_000, 0)
+    assert opus > haiku > 0
+
+
+def test_registry_catalog_and_selected_tokens() -> None:
+    reg = _native.ToolRegistry()
+    reg.register(
+        "read_file",
+        "read_file",
+        "Read a file from disk and return its contents.",
+        {"properties": {"x": {"type": "string"}}},
+        {},
+    )
+    reg.register("send_email", "send_email", "Send an email message to a recipient.", {}, {})
+    full = reg.catalog_tokens()
+    one = reg.tokens_for(["read_file"])
+    assert full > one > 0
+    # an unknown id contributes nothing
+    assert reg.tokens_for(["nope"]) == 0
+
+
+# -- catalog observe --------------------------------------------------------
+
+
+def _build_catalog(observe: bool) -> ToolCatalog:
+    catalog = ToolCatalog(trace=TraceSinkConfig(kind="memory", session_id="s"), observe=observe)
+    catalog.register(_exec_tool("read_file", "Read a file from disk and return its contents."))
+    catalog.register(_exec_tool("send_email", "Send an email message to a recipient address."))
+    catalog.register(_exec_tool("list_dir", "List the entries of a directory on local disk."))
+    return catalog
+
+
+def test_search_records_savings_in_memory_when_observing() -> None:
+    catalog = _build_catalog(observe=True)
+    catalog.drain_trace_events()  # clear registration churn
+
+    catalog.search("read a file from disk", top_k=1)
+
+    # Savings is in-memory only — nothing is emitted to the trace stream.
+    types = {e["type"] for e in catalog.drain_trace_events()}
+    assert "tokens_saved" not in types
+
+    savings = catalog.last_savings
+    assert savings is not None
+    assert savings["full_catalog_tokens"] > savings["selected_tokens"]
+    assert savings["top_k"] == 1
+    assert savings["tokens_saved"] > 0
+
+
+def test_search_records_nothing_extra_when_not_observing() -> None:
+    catalog = _build_catalog(observe=False)
+    catalog.drain_trace_events()
+    catalog.search("read a file", top_k=1)
+    types = {e["type"] for e in catalog.drain_trace_events()}
+    assert "tokens_saved" not in types
+    # only the ordinary search event is recorded
+    assert types == {"search"}
+    assert catalog.last_savings is None