diff --git a/src/langbot/pkg/core/app.py b/src/langbot/pkg/core/app.py index 5a4120240..7da4a0a15 100644 --- a/src/langbot/pkg/core/app.py +++ b/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', ())), } diff --git a/src/langbot/pkg/plugin/connector.py b/src/langbot/pkg/plugin/connector.py index dc9aa9bec..6071f6cfb 100644 --- a/src/langbot/pkg/plugin/connector.py +++ b/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 diff --git a/src/langbot/pkg/plugin/runtime_ops.py b/src/langbot/pkg/plugin/runtime_ops.py new file mode 100644 index 000000000..29f1a5de9 --- /dev/null +++ b/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 diff --git a/src/langbot/templates/config.yaml b/src/langbot/templates/config.yaml index e7cb87990..9e5a0f7ac 100644 --- a/src/langbot/templates/config.yaml +++ b/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 diff --git a/tests/unit_tests/plugin/test_runtime_ops_reporter.py b/tests/unit_tests/plugin/test_runtime_ops_reporter.py new file mode 100644 index 000000000..51e9c7861 --- /dev/null +++ b/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