feat(plugin): push plugin runtime ops samples to the Space control plane

The shared Cloud deployment runs one Plugin Runtime for every Workspace, but
nothing reported its health: the instance only exposed two aggregate numbers on
its own health probe, and the per-installation failure registry, the desired
state registry and the reconcile outcome lived in memory with no reader.

Add RuntimeOpsReporter, which every 30 seconds reads the Runtime ``/healthz``
probe over HTTP (derived from ``plugin.runtime_ws_url``), merges it with the
connector's desired-state, failure and reconcile summaries, and posts one
sample to the Space control plane on the existing authenticated direction.
Delivery is best-effort and bounded: a single in-flight request, a ten second
timeout, dropped samples counted in ``runtime_ops`` on the instance health
probe, and a failed probe recorded as a drop so the dashboard shows staleness
instead of grading a stale sample as healthy.

Cloud-only: a non-cloud deployment never enables the reporter, and the channel
is independent of product telemetry so ``space.disable_telemetry`` cannot blind
the ops dashboard.

src/langbot/pkg/core/app.py

@@ -304,6 +304,11 @@ class Application:
         if callable(database_snapshot):
             database_stats = database_snapshot()
 
+        runtime_ops_stats = {}
+        runtime_ops_reporter = getattr(self.plugin_connector, 'runtime_ops_reporter', None)
+        if runtime_ops_reporter is not None:
+            runtime_ops_stats = runtime_ops_reporter.get_stats()
+
         return {
             'asyncio_tasks': asyncio_tasks,
             'event_loop': self.event_loop_monitor.snapshot(),
@@ -314,6 +319,7 @@ class Application:
             'query_pool': query_pool_stats,
             'models': model_stats,
             'runtimes': runtime_stats,
+            'runtime_ops': runtime_ops_stats,
             'telemetry_tasks': len(getattr(self.telemetry, 'send_tasks', ())),
         }
 

src/langbot/pkg/plugin/connector.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 import asyncio
 import contextlib
 import contextvars
+import datetime
 import hashlib
 import json
 import math
@@ -21,6 +22,7 @@ from langbot_plugin.api.entities.builtin.pipeline.query import provider_session
 
 from ..core import app
 from . import handler
+from . import runtime_ops
 from .errors import (
     PluginRuntimeNotConnectedError,
     PluginInstallationFailedError,
@@ -97,6 +99,12 @@ _HEARTBEAT_FAILURE_THRESHOLD = 3
 _RECONNECT_MAX_DELAY_SEC = 60.0
 
 
+def _utc_now_iso() -> str:
+    """Return the current UTC time as a second-precision RFC 3339 string."""
+
+    return datetime.datetime.now(datetime.timezone.utc).isoformat(timespec='seconds').replace('+00:00', 'Z')
+
+
 async def _read_httpx_response_limited(
     response: httpx.Response,
     *,
@@ -239,6 +247,14 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
         self._reconnect_task: asyncio.Task | None = None
         self._generation = 0
         self._connected = asyncio.Event()
+        self.runtime_ops_reporter = runtime_ops.RuntimeOpsReporter(ap)
+        self._reconcile_summary: dict[str, Any] = {
+            'last_started_at': None,
+            'last_duration_ms': None,
+            'last_ok': None,
+            'failed_installations': 0,
+            'missing_artifacts': 0,
+        }
 
     @staticmethod
     def _build_runtime_id() -> str:
@@ -674,6 +690,21 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
                 self._installation_failures.pop(installation_uuid, None)
         self._installation_failures.update(failures)
 
+    def _record_reconcile_outcome(self, started_at: str, duration_ms: float, result: dict[str, Any]) -> None:
+        """Expose the last reconcile outcome for the runtime ops sample."""
+
+        failed = result.get('failed_installations')
+        missing = result.get('missing_artifacts')
+        failed_installations = len(failed) if isinstance(failed, list) else 0
+        missing_artifacts = len(missing) if isinstance(missing, list) else 0
+        self._reconcile_summary = {
+            'last_started_at': started_at,
+            'last_duration_ms': int(max(duration_ms, 0.0)),
+            'last_ok': failed_installations == 0 and missing_artifacts == 0,
+            'failed_installations': failed_installations,
+            'missing_artifacts': missing_artifacts,
+        }
+
     async def _repair_reconcile_missing_artifacts(
         self,
         desired_states: dict[str, PluginInstallationDesiredState],
@@ -754,12 +785,19 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
         reconcile_timeout_seconds = max(
             300.0, self._runtime_connect_timeout(self.ap.instance_config.data.get('plugin', {}))
         )
+        reconcile_started_at = _utc_now_iso()
+        reconcile_started_monotonic = time.monotonic()
         result = await runtime_handler.reconcile_plugin_installations(
             tuple(self._known_desired_states.values()),
             timeout=reconcile_timeout_seconds,
         )
         await self._repair_reconcile_missing_artifacts(self._known_desired_states, result)
         self._record_reconcile_failures(self._known_desired_states, result)
+        self._record_reconcile_outcome(
+            reconcile_started_at,
+            (time.monotonic() - reconcile_started_monotonic) * 1000.0,
+            result,
+        )
 
     async def reconcile_projected_workspaces(
         self,
@@ -775,6 +813,7 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
 
         runtime_handler = self._runtime_handler()
         started_at = time.monotonic()
+        started_at_iso = _utc_now_iso()
         async with self._state_lock:
             all_states: dict[str, PluginInstallationDesiredState] = {}
             workspace_installations: dict[str, set[str]] = {}
@@ -801,6 +840,11 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
             )
             await self._repair_reconcile_missing_artifacts(all_states, result)
             self._record_reconcile_failures(all_states, result)
+            self._record_reconcile_outcome(
+                started_at_iso,
+                (time.monotonic() - started_at) * 1000.0,
+                result,
+            )
             for installation_uuid, previous in tuple(self._known_desired_states.items()):
                 if installation_uuid not in all_states:
                     runtime_handler.unregister_installation_binding(previous.binding)
@@ -1068,6 +1112,8 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
             if self.heartbeat_task is None or self.heartbeat_task.done():
                 self.heartbeat_task = asyncio.create_task(self.heartbeat_loop())
 
+            self.runtime_ops_reporter.start()
+
     def schedule_reconnect(self) -> None:
         if self._closing or not self.is_enable_plugin:
             return
@@ -1130,6 +1176,7 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
             await asyncio.gather(self.heartbeat_task, return_exceptions=True)
             self.heartbeat_task = None
         await self._stop_transport()
+        await self.runtime_ops_reporter.stop()
         await self._close_managed_subprocess()
 
     @staticmethod

src/langbot/pkg/plugin/runtime_ops.py

@@ -0,0 +1,389 @@
+"""Cloud Edition runtime ops sample reporter.
+
+Aggregates this instance's plugin Runtime health (probed from the Runtime
+``/healthz`` endpoint) together with the connector's own desired-state and
+failure registries, then pushes one compact sample to the Space control
+plane where the internal ops dashboard stores and renders it.
+
+This is a Cloud-only ops channel, deliberately separate from product
+telemetry so ``space.disable_telemetry`` cannot blind operations. Sending is
+best-effort: a failed probe or upload never raises out of the reporter.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import datetime
+import math
+import os
+import typing
+from urllib.parse import urlsplit, urlunsplit
+
+import httpx
+
+from ..utils import constants, httpclient
+
+_CONTROL_PLANE_TOKEN_ENV = 'LANGBOT_SPACE_CONTROL_PLANE_TOKEN'
+_RUNTIME_OPS_PATH = '/api/v1/cloud/v2/internal/runtime-ops'
+_DEFAULT_INTERVAL_SECONDS = 30.0
+_DEFAULT_TIMEOUT_SECONDS = 10.0
+_DEFAULT_MAX_FAILURES = 5
+_MAX_FAILURE_MESSAGE_CHARS = 512
+
+
+def _utc_now_iso() -> str:
+    """Return the current UTC time as a second-precision RFC 3339 string."""
+
+    return datetime.datetime.now(datetime.timezone.utc).isoformat(timespec='seconds').replace('+00:00', 'Z')
+
+
+def _coerce_bool(value: typing.Any, default: bool) -> bool:
+    return value if isinstance(value, bool) else default
+
+
+def _coerce_positive_float(value: typing.Any, default: float) -> float:
+    if isinstance(value, bool) or not isinstance(value, (int, float)):
+        return default
+    value = float(value)
+    if not math.isfinite(value) or value <= 0:
+        return default
+    return value
+
+
+def _coerce_positive_int(value: typing.Any, default: int) -> int:
+    if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
+        return default
+    return value
+
+
+def _sanitize_failure_message(value: typing.Any) -> str:
+    """Flatten and bound one failure message for the wire.
+
+    Mirrors the connector's own sanitization so ops never receives control
+    characters, unbounded payloads, or newline-injected log content.
+    """
+
+    candidate = str(value or '').replace('\r', ' ').replace('\n', ' ').strip()
+    return candidate[:_MAX_FAILURE_MESSAGE_CHARS]
+
+
+class RuntimeOpsReporter:
+    """Collect and push one runtime ops sample per interval.
+
+    The reporter never raises out of its send loop; failures are counted and
+    surfaced through :meth:`get_stats`.
+    """
+
+    def __init__(self, ap: typing.Any):
+        self.ap = ap
+        instance_data = getattr(getattr(ap, 'instance_config', None), 'data', None)
+        if not isinstance(instance_data, dict):
+            instance_data = {}
+        plugin_config = instance_data.get('plugin', {})
+        if not isinstance(plugin_config, dict):
+            plugin_config = {}
+        config = plugin_config.get('runtime_ops', {})
+        if not isinstance(config, dict):
+            config = {}
+        self.config = config
+
+        deployment_mode = getattr(getattr(ap, 'deployment', None), 'mode', 'oss')
+        cloud_edition = deployment_mode == 'cloud'
+        # Cloud-only feature: an explicit flag may turn it off, but a
+        # non-cloud deployment is never enabled.
+        self.enabled = cloud_edition and _coerce_bool(config.get('enabled'), True)
+        self.interval_seconds = _coerce_positive_float(config.get('interval_seconds'), _DEFAULT_INTERVAL_SECONDS)
+        self.timeout_seconds = _coerce_positive_float(config.get('timeout_seconds'), _DEFAULT_TIMEOUT_SECONDS)
+        self.max_failures = _coerce_positive_int(config.get('max_failures'), _DEFAULT_MAX_FAILURES)
+
+        self.health_url = self._derive_health_url(plugin_config.get('runtime_ws_url'))
+        self.space_url = str(instance_data.get('space', {}).get('url', '') or '').rstrip('/')
+        self._token = str(os.environ.get(_CONTROL_PLANE_TOKEN_ENV, '') or '').strip()
+
+        if self.enabled and not self.health_url:
+            self._log_info(
+                'Plugin runtime ops reporter is disabled: no plugin runtime WebSocket URL is configured'
+            )
+            self.enabled = False
+        if self.enabled and not self.space_url:
+            self._log_info('Plugin runtime ops reporter is disabled: no Space URL is configured')
+            self.enabled = False
+
+        self._task: asyncio.Task | None = None
+        self._client: httpx.AsyncClient | None = None
+        self._sent_total = 0
+        self._dropped_total = 0
+        self._last_error: str | None = None
+
+    def _log_info(self, message: str) -> None:
+        logger = getattr(self.ap, 'logger', None)
+        info = getattr(logger, 'info', None)
+        if callable(info):
+            info(message)
+
+    @staticmethod
+    def _derive_health_url(runtime_ws_url: typing.Any) -> str | None:
+        """Derive the Runtime ``/healthz`` URL from the control WebSocket URL."""
+
+        if not isinstance(runtime_ws_url, str):
+            return None
+        candidate = runtime_ws_url.strip()
+        if not candidate:
+            return None
+        try:
+            parts = urlsplit(candidate)
+            port = parts.port
+        except ValueError:
+            return None
+        if parts.scheme not in ('ws', 'wss') or not parts.hostname:
+            return None
+        scheme = 'https' if parts.scheme == 'wss' else 'http'
+        netloc = parts.hostname if port is None else f'{parts.hostname}:{port}'
+        return urlunsplit((scheme, netloc, '/healthz', '', ''))
+
+    def start(self) -> None:
+        """Start the background sample loop once the control link is up."""
+
+        if not self.enabled:
+            return
+        if self._task is not None and not self._task.done():
+            return
+        self._task = asyncio.create_task(self._run())
+
+    async def stop(self) -> None:
+        """Cancel the sample loop and release the HTTP client."""
+
+        task = self._task
+        self._task = None
+        if task is not None and task is not asyncio.current_task():
+            task.cancel()
+            await asyncio.gather(task, return_exceptions=True)
+        client = self._client
+        self._client = None
+        if client is not None and not client.is_closed:
+            with contextlib.suppress(Exception):
+                await client.aclose()
+
+    def get_stats(self) -> dict[str, typing.Any]:
+        """Return in-memory counters for the instance liveness surface."""
+
+        return {
+            'sent_total': self._sent_total,
+            'dropped_total': self._dropped_total,
+            'last_error': self._last_error,
+        }
+
+    @contextlib.asynccontextmanager
+    async def _client_context(self):
+        if self._client is None or self._client.is_closed:
+            self._client = httpx.AsyncClient(
+                timeout=httpx.Timeout(self.timeout_seconds),
+                event_hooks=httpclient.httpx_response_limit_hooks(),
+            )
+        yield self._client
+
+    async def _run(self) -> None:
+        while True:
+            try:
+                await self._collect_and_send()
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:  # never let the ops channel crash the connector
+                self._last_error = _sanitize_failure_message(f'sample cycle failed: {exc}')
+            await asyncio.sleep(self.interval_seconds)
+
+    async def _collect_and_send(self) -> None:
+        health = await self._fetch_health()
+        if health is None:
+            self._dropped_total += 1
+            return
+        await self._post(self.build_payload(health))
+
+    async def _fetch_health(self) -> dict[str, typing.Any] | None:
+        """Probe the Runtime ``/healthz`` endpoint, returning its verbatim JSON."""
+
+        try:
+            async with self._client_context() as client:
+                response = await asyncio.wait_for(
+                    client.get(self.health_url),
+                    timeout=self.timeout_seconds,
+                )
+            if response.status_code >= 400:
+                self._last_error = f'health probe returned status {response.status_code}'
+                return None
+            payload = await httpclient.parse_json_response(response)
+            if not isinstance(payload, dict):
+                self._last_error = 'health probe returned a non-object payload'
+                return None
+            self._last_error = None
+            return payload
+        except asyncio.TimeoutError:
+            self._last_error = 'health probe timed out'
+            return None
+        except Exception as exc:
+            self._last_error = _sanitize_failure_message(f'health probe failed: {exc}')
+            return None
+
+    async def _post(self, payload: dict[str, typing.Any]) -> None:
+        """Best-effort POST of one sample; never raises."""
+
+        url = f'{self.space_url}{_RUNTIME_OPS_PATH}'
+        headers = {'Authorization': f'Bearer {self._token}'}
+        try:
+            async with self._client_context() as client:
+                response = await asyncio.wait_for(
+                    client.post(url, json=payload, headers=headers),
+                    timeout=self.timeout_seconds,
+                )
+            if response.status_code >= 400:
+                body = await httpclient.response_text(response, max_chars=200)
+                self._dropped_total += 1
+                self._last_error = f'POST {url} returned status {response.status_code} - {body}'
+                return
+            self._sent_total += 1
+            self._last_error = None
+        except asyncio.TimeoutError:
+            self._dropped_total += 1
+            self._last_error = f'POST {url} timed out'
+        except Exception as exc:
+            self._dropped_total += 1
+            self._last_error = _sanitize_failure_message(f'POST {url} failed: {exc}')
+
+    def build_payload(self, health: dict[str, typing.Any]) -> dict[str, typing.Any]:
+        """Assemble one ops sample in the frozen Core -> Space contract shape."""
+
+        connector = getattr(self.ap, 'plugin_connector', None)
+        instance_data = getattr(getattr(self.ap, 'instance_config', None), 'data', None)
+        if not isinstance(instance_data, dict):
+            instance_data = {}
+        system_config = instance_data.get('system', {})
+        if not isinstance(system_config, dict):
+            system_config = {}
+        instance_uuid = str(system_config.get('instance_id', '') or '') or str(constants.instance_id or '')
+        edition = str(system_config.get('edition', 'oss') or 'oss')
+        return {
+            'event': 'runtime_ops_sample',
+            'kind': 'aggregate',
+            'instance_uuid': instance_uuid,
+            'generated_at': _utc_now_iso(),
+            'core': {
+                'version': constants.semantic_version,
+                'edition': edition,
+                'connected': self._runtime_connected(connector),
+                'runtime_profile': getattr(connector, 'runtime_profile', 'oss_dev'),
+                'worker_policy': self._worker_policy(connector),
+                'desired_states': self._desired_states(connector),
+                'reconcile': self._reconcile_summary(connector),
+            },
+            'health': health,
+            'failures': self._failures(connector),
+        }
+
+    @staticmethod
+    def _runtime_connected(connector: typing.Any) -> bool:
+        available = getattr(connector, '_runtime_available', None)
+        if not callable(available):
+            return False
+        try:
+            return bool(available())
+        except Exception:
+            return False
+
+    @staticmethod
+    def _worker_policy(connector: typing.Any) -> dict[str, typing.Any] | None:
+        policy = getattr(connector, 'worker_policy', None)
+        if policy is None:
+            return None
+        model_dump = getattr(policy, 'model_dump', None)
+        if callable(model_dump):
+            return model_dump()
+        if isinstance(policy, dict):
+            return dict(policy)
+        return None
+
+    @staticmethod
+    def _desired_states(connector: typing.Any) -> dict[str, typing.Any]:
+        states = getattr(connector, '_known_desired_states', None)
+        if not isinstance(states, dict):
+            states = {}
+        enabled = 0
+        by_mode: dict[str, int] = {}
+        for state in states.values():
+            if bool(getattr(state, 'enabled', False)):
+                enabled += 1
+            mode = getattr(getattr(state, 'execution_mode', None), 'value', None)
+            mode = str(mode) if isinstance(mode, str) and mode else 'dedicated'
+            by_mode[mode] = by_mode.get(mode, 0) + 1
+        return {'total': len(states), 'enabled': enabled, 'by_mode': by_mode}
+
+    @staticmethod
+    def _reconcile_summary(connector: typing.Any) -> dict[str, typing.Any]:
+        summary = getattr(connector, '_reconcile_summary', None)
+        if isinstance(summary, dict):
+            return dict(summary)
+        return {
+            'last_started_at': None,
+            'last_duration_ms': None,
+            'last_ok': None,
+            'failed_installations': 0,
+            'missing_artifacts': 0,
+        }
+
+    def _failures(self, connector: typing.Any) -> list[dict[str, typing.Any]]:
+        records = getattr(connector, '_installation_failures', None)
+        if not isinstance(records, dict):
+            return []
+        states = getattr(connector, '_known_desired_states', None)
+        if not isinstance(states, dict):
+            states = {}
+        identities = self._plugin_identities(connector)
+        observed_at = _utc_now_iso()
+        failures: list[dict[str, typing.Any]] = []
+        for installation_uuid in list(records)[: self.max_failures]:
+            record = records.get(installation_uuid)
+            if not isinstance(record, dict):
+                continue
+            state = states.get(installation_uuid)
+            binding = getattr(state, 'binding', None)
+            mode = getattr(getattr(state, 'execution_mode', None), 'value', None)
+            mode = str(mode) if isinstance(mode, str) and mode else 'dedicated'
+            plugin_author, plugin_name = identities.get(str(installation_uuid), ('', ''))
+            failures.append(
+                {
+                    'workspace_uuid': str(getattr(binding, 'workspace_uuid', '') or ''),
+                    'installation_uuid': str(installation_uuid),
+                    'plugin_author': plugin_author,
+                    'plugin_name': plugin_name,
+                    'execution_mode': mode,
+                    'error_code': str(record.get('error_code', '') or ''),
+                    'message': _sanitize_failure_message(record.get('message')),
+                    'observed_at': observed_at,
+                }
+            )
+        return failures
+
+    @staticmethod
+    def _plugin_identities(connector: typing.Any) -> dict[str, tuple[str, str]]:
+        """Read the connector handler's installation identity registry.
+
+        The Runtime connection handler binds every desired installation to
+        its plugin identity; the connector's failure records themselves only
+        carry the installation UUID, so the reporter joins them here.
+        """
+
+        handler = getattr(connector, 'handler', None)
+        bindings = getattr(handler, '_installation_bindings', None)
+        if not isinstance(bindings, dict):
+            return {}
+        identities: dict[str, tuple[str, str]] = {}
+        for installation_uuid, entry in bindings.items():
+            try:
+                identity = entry[1]
+            except (TypeError, IndexError, KeyError):
+                continue
+            identities[str(installation_uuid)] = (
+                str(getattr(identity, 'plugin_author', '') or ''),
+                str(getattr(identity, 'plugin_name', '') or ''),
+            )
+        return identities

src/langbot/templates/config.yaml

@@ -293,6 +293,20 @@ plugin:
     binary_storage:
         # Max bytes for a single plugin binary storage value
         max_value_bytes: 10485760
+    runtime_ops:
+        # Cloud-only ops telemetry: aggregates this instance's plugin Runtime
+        # health, desired installations and reconcile failures and pushes them
+        # to the Space control plane for the internal ops dashboard. This is
+        # separate from product telemetry (space.disable_telemetry) and has no
+        # effect outside the cloud edition. Defaults to on for the cloud
+        # edition; set false to stop pushing samples.
+        enabled: true
+        # Seconds between samples pushed to Space.
+        interval_seconds: 30
+        # Per-request timeout for the Runtime /healthz probe and the upload.
+        timeout_seconds: 10
+        # Maximum number of failed installations included per sample.
+        max_failures: 5
 runner:
     # Host-level admin permissions for trusted control plugins. These plugins
     # can use existing plugin action handlers to inspect or manage Runner

tests/unit_tests/plugin/test_runtime_ops_reporter.py

@@ -0,0 +1,294 @@
+"""Tests for the Cloud plugin runtime ops sample reporter.
+
+Covers the frozen Core -> Space sample contract (C2):
+- Runtime /healthz URL derivation from the control WebSocket URL
+- Disablement without a runtime WebSocket URL / outside the cloud edition
+- Sample assembly from connector internals (desired states, failures, health)
+- Best-effort upload semantics (non-2xx and transport failures never raise)
+- The pushed payload never carries the runtime control token
+"""
+
+from __future__ import annotations
+
+import json
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import httpx
+import pytest
+
+from langbot.pkg.plugin.runtime_ops import RuntimeOpsReporter
+from langbot_plugin.entities.io.context import PluginExecutionMode
+from langbot_plugin.runtime.security import PLUGIN_RUNTIME_CONTROL_TOKEN_ENV
+
+_CONTROL_PLANE_TOKEN_ENV = 'LANGBOT_SPACE_CONTROL_PLANE_TOKEN'
+
+_DEFAULT_WS_URL = 'ws://langbot_plugin_runtime:5400/control/ws'
+
+
+def make_policy() -> SimpleNamespace:
+    return SimpleNamespace(model_dump=lambda: {'max_workers': 32, 'max_cpus': 0.25})
+
+
+def make_connector(
+    *,
+    desired_states: dict | None = None,
+    failures: dict | None = None,
+    identities: dict | None = None,
+    connected: bool = True,
+) -> SimpleNamespace:
+    return SimpleNamespace(
+        runtime_profile='shared',
+        worker_policy=make_policy(),
+        _known_desired_states=desired_states if desired_states is not None else {},
+        _installation_failures=failures if failures is not None else {},
+        _reconcile_summary={
+            'last_started_at': '2026-10-01T07:11:00Z',
+            'last_duration_ms': 1234,
+            'last_ok': True,
+            'failed_installations': 0,
+            'missing_artifacts': 0,
+        },
+        handler=SimpleNamespace(_installation_bindings=identities if identities is not None else {}),
+        _runtime_available=lambda: connected,
+    )
+
+
+def make_app(
+    *,
+    runtime_ws_url: str | None = _DEFAULT_WS_URL,
+    runtime_ops: dict | None = None,
+    space_url: str = 'https://space.example',
+    deployment_mode: str = 'cloud',
+    edition: str = 'cloud',
+    connector: SimpleNamespace | None = None,
+) -> SimpleNamespace:
+    plugin: dict = {}
+    if runtime_ws_url is not None:
+        plugin['runtime_ws_url'] = runtime_ws_url
+    if runtime_ops is not None:
+        plugin['runtime_ops'] = runtime_ops
+    return SimpleNamespace(
+        logger=Mock(),
+        instance_config=SimpleNamespace(
+            data={
+                'plugin': plugin,
+                'space': {'url': space_url},
+                'system': {'instance_id': 'inst-1', 'edition': edition},
+            }
+        ),
+        deployment=SimpleNamespace(mode=deployment_mode),
+        plugin_connector=connector if connector is not None else make_connector(),
+    )
+
+
+def make_desired_state(workspace_uuid: str, mode: PluginExecutionMode, *, enabled: bool = True) -> SimpleNamespace:
+    return SimpleNamespace(
+        enabled=enabled,
+        execution_mode=mode,
+        binding=SimpleNamespace(workspace_uuid=workspace_uuid),
+    )
+
+
+class TestHealthUrlDerivation:
+    @pytest.mark.parametrize(
+        ('ws_url', 'expected'),
+        [
+            ('ws://langbot_plugin_runtime:5400/control/ws', 'http://langbot_plugin_runtime:5400/healthz'),
+            ('wss://runtime.example/control/ws', 'https://runtime.example/healthz'),
+            ('wss://runtime.example:8443/control/ws', 'https://runtime.example:8443/healthz'),
+            ('ws://127.0.0.1:5400', 'http://127.0.0.1:5400/healthz'),
+            ('  ws://runtime.example:5400/control/ws  ', 'http://runtime.example:5400/healthz'),
+        ],
+    )
+    def test_derives_health_url(self, ws_url: str, expected: str):
+        reporter = RuntimeOpsReporter(make_app(runtime_ws_url=ws_url))
+
+        assert reporter.health_url == expected
+
+    @pytest.mark.parametrize('ws_url', ['', None, 'http://runtime.example:5400', 'not a url'])
+    def test_rejects_non_websocket_url(self, ws_url):
+        reporter = RuntimeOpsReporter(make_app(runtime_ws_url=ws_url))
+
+        assert reporter.health_url is None
+
+
+class TestEnablement:
+    def test_disabled_without_runtime_websocket_url(self):
+        reporter = RuntimeOpsReporter(make_app(runtime_ws_url=None))
+
+        assert reporter.enabled is False
+
+    def test_disabled_on_non_cloud_deployment(self):
+        reporter = RuntimeOpsReporter(make_app(deployment_mode='oss'))
+
+        assert reporter.enabled is False
+
+    def test_explicit_disable_on_cloud(self):
+        reporter = RuntimeOpsReporter(make_app(runtime_ops={'enabled': False}))
+
+        assert reporter.enabled is False
+
+    def test_enabled_by_default_on_cloud(self):
+        reporter = RuntimeOpsReporter(make_app())
+
+        assert reporter.enabled is True
+        assert reporter.interval_seconds == 30.0
+        assert reporter.timeout_seconds == 10.0
+        assert reporter.max_failures == 5
+
+
+class TestSampleAssembly:
+    def test_build_payload_matches_contract(self):
+        desired = {
+            'inst-dedicated': make_desired_state('ws-1', PluginExecutionMode.DEDICATED),
+            'inst-shared-off': make_desired_state(
+                'ws-2', PluginExecutionMode.SHARED_CERTIFIED, enabled=False
+            ),
+            'inst-shared-on': make_desired_state('ws-3', PluginExecutionMode.SHARED_CERTIFIED),
+        }
+        failures = {
+            'inst-dedicated': {
+                'installation_uuid': 'inst-dedicated',
+                'error_code': 'worker_launch_failed',
+                'message': 'boom\nstack line',
+            },
+            'inst-shared-off': {
+                'installation_uuid': 'inst-shared-off',
+                'error_code': 'dependency_prepare_failed',
+                'message': 'prepare failed',
+            },
+        }
+        identities = {
+            'inst-dedicated': (
+                None,
+                SimpleNamespace(plugin_author='acme', plugin_name='dedicated-plugin'),
+            ),
+        }
+        connector = make_connector(desired_states=desired, failures=failures, identities=identities)
+        reporter = RuntimeOpsReporter(make_app(connector=connector))
+        health = {'live': True, 'runtime': {'version': '0.7.6'}, 'shared_pool': {'workers': 12}}
+
+        payload = reporter.build_payload(health)
+
+        assert set(payload) == {'event', 'kind', 'instance_uuid', 'generated_at', 'core', 'health', 'failures'}
+        assert payload['event'] == 'runtime_ops_sample'
+        assert payload['kind'] == 'aggregate'
+        assert payload['instance_uuid'] == 'inst-1'
+        assert payload['generated_at'].endswith('Z')
+        # The health blob is forwarded verbatim.
+        assert payload['health'] is health
+
+        core = payload['core']
+        assert core['edition'] == 'cloud'
+        assert core['connected'] is True
+        assert core['runtime_profile'] == 'shared'
+        assert core['worker_policy'] == {'max_workers': 32, 'max_cpus': 0.25}
+        assert core['desired_states'] == {
+            'total': 3,
+            'enabled': 2,
+            'by_mode': {'dedicated': 1, 'shared-runtime-v1': 2},
+        }
+        assert core['reconcile'] == {
+            'last_started_at': '2026-10-01T07:11:00Z',
+            'last_duration_ms': 1234,
+            'last_ok': True,
+            'failed_installations': 0,
+            'missing_artifacts': 0,
+        }
+
+        assert len(payload['failures']) == 2
+        first = payload['failures'][0]
+        assert first['workspace_uuid'] == 'ws-1'
+        assert first['installation_uuid'] == 'inst-dedicated'
+        assert first['plugin_author'] == 'acme'
+        assert first['plugin_name'] == 'dedicated-plugin'
+        assert first['execution_mode'] == 'dedicated'
+        assert first['error_code'] == 'worker_launch_failed'
+        assert first['message'] == 'boom stack line'
+        assert first['observed_at'].endswith('Z')
+        second = payload['failures'][1]
+        assert second['workspace_uuid'] == 'ws-2'
+        assert second['plugin_author'] == ''
+        assert second['plugin_name'] == ''
+        assert second['execution_mode'] == 'shared-runtime-v1'
+
+    def test_failures_capped_by_max_failures(self):
+        failures = {
+            f'inst-{index}': {
+                'installation_uuid': f'inst-{index}',
+                'error_code': 'worker_launch_failed',
+                'message': 'boom',
+            }
+            for index in range(8)
+        }
+        connector = make_connector(failures=failures)
+        reporter = RuntimeOpsReporter(make_app(runtime_ops={'max_failures': 3}, connector=connector))
+
+        payload = reporter.build_payload({'live': True})
+
+        assert len(payload['failures']) == 3
+
+    def test_disconnected_connector_reports_not_connected(self):
+        connector = make_connector(connected=False)
+        reporter = RuntimeOpsReporter(make_app(connector=connector))
+
+        payload = reporter.build_payload({'live': False})
+
+        assert payload['core']['connected'] is False
+
+
+class TestUpload:
+    @pytest.mark.asyncio
+    async def test_successful_post_increments_sent(self):
+        reporter = RuntimeOpsReporter(make_app())
+        reporter._client = httpx.AsyncClient(
+            transport=httpx.MockTransport(lambda request: httpx.Response(200, json={'code': 0}))
+        )
+
+        await reporter._post({'event': 'runtime_ops_sample'})
+
+        assert reporter.get_stats() == {'sent_total': 1, 'dropped_total': 0, 'last_error': None}
+        await reporter.stop()
+
+    @pytest.mark.asyncio
+    async def test_non_2xx_increments_dropped_without_raising(self):
+        reporter = RuntimeOpsReporter(make_app())
+        reporter._client = httpx.AsyncClient(
+            transport=httpx.MockTransport(lambda request: httpx.Response(503, text='unavailable'))
+        )
+
+        await reporter._post({'event': 'runtime_ops_sample'})
+
+        stats = reporter.get_stats()
+        assert stats['sent_total'] == 0
+        assert stats['dropped_total'] == 1
+        assert '503' in stats['last_error']
+        await reporter.stop()
+
+    @pytest.mark.asyncio
+    async def test_transport_exception_increments_dropped_without_raising(self):
+        def raise_connect_error(request: httpx.Request) -> httpx.Response:
+            raise httpx.ConnectError('connection refused')
+
+        reporter = RuntimeOpsReporter(make_app())
+        reporter._client = httpx.AsyncClient(transport=httpx.MockTransport(raise_connect_error))
+
+        await reporter._post({'event': 'runtime_ops_sample'})
+
+        stats = reporter.get_stats()
+        assert stats['dropped_total'] == 1
+        assert 'connection refused' in stats['last_error']
+        await reporter.stop()
+
+
+def test_payload_never_contains_a_control_token(monkeypatch):
+    monkeypatch.setenv(PLUGIN_RUNTIME_CONTROL_TOKEN_ENV, 'runtime-control-secret')
+    monkeypatch.setenv(_CONTROL_PLANE_TOKEN_ENV, 'space-control-plane-secret')
+    reporter = RuntimeOpsReporter(make_app())
+
+    payload = reporter.build_payload({'live': True})
+    serialized = json.dumps(payload)
+
+    assert 'runtime-control-secret' not in serialized
+    assert 'space-control-plane-secret' not in serialized