fix(telemetry): avoid global plugin enumeration in workspace snapshots and support model adapters

src/langbot/pkg/telemetry/heartbeat.py

@@ -189,10 +189,10 @@ async def build_heartbeat_payload(
             cloud_counter=lambda: len(ap.platform_mgr._bots_by_key),
         )
 
-    # Plugin count (from plugin runtime)
+    # Scoped Cloud counts are already available; never issue tenantless RPCs.
     try:
         plugin_connector = getattr(ap, 'plugin_connector', None)
-        if plugin_connector is not None:
+        if workspace_resource is None and plugin_connector is not None:
             plugins = await plugin_connector.list_plugins()
             features['plugin_count'] = len(plugins)
     except Exception:

src/langbot/pkg/telemetry/platform.py

@@ -117,5 +117,7 @@ def observe_adapter(ap, context, adapter):
 
             return wrapped
 
-        setattr(adapter, name, make_wrapper(original, name))
-    setattr(adapter, '_execution_observed', True)
+        # Adapters are Pydantic models: method instrumentation is not a model
+        # field assignment. Keep wrappers instance-local, never on the class.
+        object.__setattr__(adapter, name, make_wrapper(original, name))
+    object.__setattr__(adapter, '_execution_observed', True)

tests/unit_tests/telemetry/test_heartbeat.py

@@ -183,6 +183,7 @@ class TestBuildHeartbeatPayload:
         assert by_workspace['workspace-b']['execution_generation'] == 9
         assert by_workspace['workspace-b']['adapters'] == ['WorkspaceBAdapter']
         assert 'workspace_resources' not in by_workspace['workspace-a']
+        ap.plugin_connector.list_plugins.assert_not_awaited()
         ap.persistence_mgr.execute_async.assert_not_awaited()
         ap.workspace_service.list_active_execution_bindings.assert_awaited_once()
 

tests/unit_tests/telemetry/test_platform_model.py

@@ -0,0 +1,27 @@
+from types import SimpleNamespace
+
+import pytest
+from pydantic import BaseModel
+
+from langbot.pkg.telemetry.platform import observe_adapter
+
+
+class ModelAdapter(BaseModel):
+    def get_supported_apis(self):
+        return ['send_message']
+
+    async def send_message(self):
+        return {'message_id': 'ok'}
+
+
+@pytest.mark.asyncio
+async def test_model_adapter_instrumentation_is_instance_local():
+    first, second = ModelAdapter(), ModelAdapter()
+    ap = SimpleNamespace(telemetry=None)
+    observe_adapter(ap, None, first)
+    wrapped = first.send_message
+    observe_adapter(ap, None, first)
+    assert first.send_message is wrapped
+    assert not getattr(second, '_execution_observed', False)
+    assert await first.send_message() == {'message_id': 'ok'}
+    assert await second.send_message() == {'message_id': 'ok'}