From 923ebe12af75bef27c5f010678569fd3a2a0c2d2 Mon Sep 17 00:00:00 2001 From: Hyu Date: Thu, 1 Oct 2026 18:10:42 +0800 Subject: [PATCH] feat(telemetry): attach called platform APIs to their step node on execution traces Every host step now opens a stage scope, so observations recorded inside it (including cross-task plugin/RPC calls resolved through the execution registry) carry the owning step as parent; the pipeline lane records its own pipeline/run node so acks and replies issued outside the runner still hang off a node. --- src/langbot/pkg/agent/runner/orchestrator.py | 245 ++++---- .../pkg/agent/runner/platform_tools.py | 102 ++-- src/langbot/pkg/pipeline/pipelinemgr.py | 96 ++- .../pkg/pipeline/process/handlers/chat.py | 96 ++- src/langbot/pkg/platform/botmgr.py | 157 +++-- src/langbot/pkg/telemetry/execution.py | 452 +++++++++----- src/langbot/pkg/telemetry/platform.py | 26 +- src/langbot/pkg/telemetry/trace.py | 166 +++++- src/langbot/templates/config.yaml | 4 + .../unit_tests/telemetry/test_chat_emitter.py | 110 ++++ .../telemetry/test_pipeline_lane.py | 272 +++++++++ tests/unit_tests/telemetry/test_trace.py | 561 ++++++++++++++++-- 12 files changed, 1798 insertions(+), 489 deletions(-) create mode 100644 tests/unit_tests/telemetry/test_chat_emitter.py create mode 100644 tests/unit_tests/telemetry/test_pipeline_lane.py diff --git a/src/langbot/pkg/agent/runner/orchestrator.py b/src/langbot/pkg/agent/runner/orchestrator.py index e207e743a..c7c9443eb 100644 --- a/src/langbot/pkg/agent/runner/orchestrator.py +++ b/src/langbot/pkg/agent/runner/orchestrator.py @@ -38,6 +38,7 @@ from .session_registry import AgentRunSessionRegistry, get_session_registry from .state_scope import build_state_context from ...provider.tools.loaders import skill as skill_loader from ...telemetry import trace as trace_mod +from ...telemetry.execution import bind_trace as bind_execution_trace from ...telemetry.execution import close_trace as close_execution_trace from ...telemetry.execution import record as record_execution @@ -213,13 +214,16 @@ class AgentRunOrchestrator: terminal_reason: str | None = None terminal_usage: dict[str, typing.Any] | None = None execution_outcome = 'unknown' + execution_error = '' # A run reached without a platform ingress (WebUI debug, service API) # owns its own chain; a run inside an ingress reuses that chain. trace_binding: trace_mod.TraceBinding | None = None run_token: typing.Any = None + # Workflow step node for this run; '' until the step scope is opened. + node = '' try: - trace_binding = trace_mod.bind() + trace_binding = bind_execution_trace(self.ap, run_id) run_token = trace_mod.set_run(run_id) await self.journal.create_run( event=event, @@ -261,139 +265,140 @@ class AgentRunOrchestrator: event_log_id=event_log_id, ) - async with contextlib.aclosing(self.invoker.invoke(descriptor, context)) as results: - async for result_dict in results: - result_dict = dict(result_dict) - sequence = result_dict.get('sequence') - if sequence is not None: - try: - sequence_int = int(sequence) - except (TypeError, ValueError): - self.ap.logger.warning( - f'Runner {descriptor.id} returned invalid result sequence: {sequence}' - ) - sequence_int = last_sequence + 1 - result_dict['sequence'] = sequence_int - else: - if sequence_int in seen_sequences: + with trace_mod.stage_scope() as node: + async with contextlib.aclosing(self.invoker.invoke(descriptor, context)) as results: + async for result_dict in results: + result_dict = dict(result_dict) + sequence = result_dict.get('sequence') + if sequence is not None: + try: + sequence_int = int(sequence) + except (TypeError, ValueError): self.ap.logger.warning( - f'Runner {descriptor.id} returned duplicate result sequence ' - f'{sequence_int} for run {run_id}; dropping duplicate' - ) - continue - if sequence_int <= 0: - self.ap.logger.warning( - f'Runner {descriptor.id} returned non-positive result sequence ' - f'{sequence_int} for run {run_id}' + f'Runner {descriptor.id} returned invalid result sequence: {sequence}' ) sequence_int = last_sequence + 1 result_dict['sequence'] = sequence_int - elif last_sequence and sequence_int != last_sequence + 1: - self.ap.logger.warning( - f'Runner {descriptor.id} result sequence gap or out-of-order ' - f'for run {run_id}: previous={last_sequence}, current={sequence_int}' - ) + else: + if sequence_int in seen_sequences: + self.ap.logger.warning( + f'Runner {descriptor.id} returned duplicate result sequence ' + f'{sequence_int} for run {run_id}; dropping duplicate' + ) + continue + if sequence_int <= 0: + self.ap.logger.warning( + f'Runner {descriptor.id} returned non-positive result sequence ' + f'{sequence_int} for run {run_id}' + ) + sequence_int = last_sequence + 1 + result_dict['sequence'] = sequence_int + elif last_sequence and sequence_int != last_sequence + 1: + self.ap.logger.warning( + f'Runner {descriptor.id} result sequence gap or out-of-order ' + f'for run {run_id}: previous={last_sequence}, current={sequence_int}' + ) + seen_sequences.add(sequence_int) + last_sequence = max(last_sequence, sequence_int) + else: + sequence_int = last_sequence + 1 + result_dict['sequence'] = sequence_int seen_sequences.add(sequence_int) - last_sequence = max(last_sequence, sequence_int) - else: - sequence_int = last_sequence + 1 - result_dict['sequence'] = sequence_int - seen_sequences.add(sequence_int) - last_sequence = sequence_int + last_sequence = sequence_int - result_type = result_dict.get('type') - if result_type and not self.result_normalizer.validate_payload( - result_type, - result_dict.get('data', {}), - descriptor, - ): - continue - - await self.journal.append_run_result( - result_dict=result_dict, - run_id=run_id, - sequence=sequence_int, - ) - - # Trusted Host observers receive validated events before message-only normalization. - result_observer = (adapter_context or {}).get('_result_observer') - if result_observer is not None: - await result_observer(result_dict) - - if result_type == 'state.updated': - await self.journal.handle_state_updated_event( - result_dict, - event, - binding, + result_type = result_dict.get('type') + if result_type and not self.result_normalizer.validate_payload( + result_type, + result_dict.get('data', {}), descriptor, - run_id=run_id, - ) - await self.result_normalizer.normalize(result_dict, descriptor) - continue - - if result_type == 'action.requested': - if await self.interaction_manager.handle_result( - result_dict=result_dict, - event=event, - binding=binding, - descriptor=descriptor, - run_id=run_id, - adapter_context=adapter_context, ): continue - if result_type == 'run.completed': - terminal_status = 'completed' - terminal_reason = ( - result_dict.get('data', {}).get('finish_reason') - if isinstance(result_dict.get('data'), dict) - else None - ) - usage = result_dict.get('usage') - if isinstance(usage, dict): - terminal_usage = usage - elif result_type == 'run.failed': - terminal_status = 'failed' - data = result_dict.get('data') if isinstance(result_dict.get('data'), dict) else {} - terminal_reason = data.get('error') or data.get('code') - usage = result_dict.get('usage') - if isinstance(usage, dict): - terminal_usage = usage - - has_completed_message = result_type == 'message.completed' or ( - result_type == 'run.completed' - and isinstance(result_dict.get('data'), dict) - and bool(result_dict['data'].get('message')) - ) - if has_completed_message and event.conversation_id and not assistant_transcript_written: - await self.journal.write_assistant_transcript( + await self.journal.append_run_result( result_dict=result_dict, - event=event, run_id=run_id, - runner_id=descriptor.id, + sequence=sequence_int, ) - assistant_transcript_written = True - result = await self.result_normalizer.normalize(result_dict, descriptor) - file_ids = result_dict.get('data', {}).get('file_ids', []) - if file_ids: - if result_type != 'message.completed' or result is None: - raise ValueError('Files must be attached to a completed message') - if binding.processor_type != 'pipeline': - raise ValueError('Agent files must be sent explicitly through the reply API') - from ...box.runner import exported_message, binding_for + # Trusted Host observers receive validated events before message-only normalization. + result_observer = (adapter_context or {}).get('_result_observer') + if result_observer is not None: + await result_observer(result_dict) - async with binding_for(execution_query).lock: - result.attachments = exported_message(execution_query, file_ids, consume=True) + if result_type == 'state.updated': + await self.journal.handle_state_updated_event( + result_dict, + event, + binding, + descriptor, + run_id=run_id, + ) + await self.result_normalizer.normalize(result_dict, descriptor) + continue - if result is not None: - yield result + if result_type == 'action.requested': + if await self.interaction_manager.handle_result( + result_dict=result_dict, + event=event, + binding=binding, + descriptor=descriptor, + run_id=run_id, + adapter_context=adapter_context, + ): + continue - run_snapshot = await self.journal.get_run(run_id) - if run_snapshot and run_snapshot.get('cancel_requested_at') is not None: - terminal_status = 'cancelled' - terminal_reason = run_snapshot.get('status_reason') or 'cancel_requested' - break + if result_type == 'run.completed': + terminal_status = 'completed' + terminal_reason = ( + result_dict.get('data', {}).get('finish_reason') + if isinstance(result_dict.get('data'), dict) + else None + ) + usage = result_dict.get('usage') + if isinstance(usage, dict): + terminal_usage = usage + elif result_type == 'run.failed': + terminal_status = 'failed' + data = result_dict.get('data') if isinstance(result_dict.get('data'), dict) else {} + terminal_reason = data.get('error') or data.get('code') + usage = result_dict.get('usage') + if isinstance(usage, dict): + terminal_usage = usage + + has_completed_message = result_type == 'message.completed' or ( + result_type == 'run.completed' + and isinstance(result_dict.get('data'), dict) + and bool(result_dict['data'].get('message')) + ) + if has_completed_message and event.conversation_id and not assistant_transcript_written: + await self.journal.write_assistant_transcript( + result_dict=result_dict, + event=event, + run_id=run_id, + runner_id=descriptor.id, + ) + assistant_transcript_written = True + + result = await self.result_normalizer.normalize(result_dict, descriptor) + file_ids = result_dict.get('data', {}).get('file_ids', []) + if file_ids: + if result_type != 'message.completed' or result is None: + raise ValueError('Files must be attached to a completed message') + if binding.processor_type != 'pipeline': + raise ValueError('Agent files must be sent explicitly through the reply API') + from ...box.runner import exported_message, binding_for + + async with binding_for(execution_query).lock: + result.attachments = exported_message(execution_query, file_ids, consume=True) + + if result is not None: + yield result + + run_snapshot = await self.journal.get_run(run_id) + if run_snapshot and run_snapshot.get('cancel_requested_at') is not None: + terminal_status = 'cancelled' + terminal_reason = run_snapshot.get('status_reason') or 'cancel_requested' + break await self.journal.finalize_run( run_id=run_id, status=terminal_status or 'completed', @@ -403,11 +408,15 @@ class AgentRunOrchestrator: execution_outcome = {'completed': 'success', 'failed': 'failed', 'cancelled': 'cancelled'}.get( terminal_status or '', 'unknown' ) + execution_error = '' if execution_outcome == 'success' else (terminal_reason or execution_outcome) except asyncio.CancelledError: execution_outcome = 'cancelled' + execution_error = 'cancelled' raise except Exception as exc: execution_outcome = 'timeout' if self._is_deadline_exhausted(context) else 'failed' + # A chain that broke must always carry a reason, even for a bare exception. + execution_error = str(exc) or type(exc).__name__ failed_usage = terminal_usage await self.journal.finalize_run( run_id=run_id, @@ -426,6 +435,8 @@ class AgentRunOrchestrator: runner=descriptor.id, outcome=execution_outcome, synthetic=event.source == 'webui', + error=execution_error, + node=node, ) trace_mod.reset_run(run_token) if trace_binding is not None and trace_mod.unbind_root(trace_binding): diff --git a/src/langbot/pkg/agent/runner/platform_tools.py b/src/langbot/pkg/agent/runner/platform_tools.py index 5599c4953..70795f304 100644 --- a/src/langbot/pkg/agent/runner/platform_tools.py +++ b/src/langbot/pkg/agent/runner/platform_tools.py @@ -684,49 +684,71 @@ async def execute_platform_tool( normalized = _normalize_platform_params(definition, parameters) if definition.scope == 'event': normalized = _event_params(definition, context, normalized) - # This flag is frozen by the Host from the synthetic debug envelope, not tool arguments. - if delivery.get('surface') == 'webui' and (delivery.get('platform_capabilities') or {}).get('debug_mock') is True: - result = _execute_mock_platform_tool(definition, context, normalized) - if message_chain is not None: - result['parameters']['message'] = message_chain.model_dump(mode='json') - from ...telemetry.execution import record + # Plugin/RPC actions run outside the ingress context, so bind the owning + # execution id here: nested adapter observations then join its trace. + from ...telemetry.execution import record, reset_execution_id, set_execution_id - record( - ap, - execution_context, - family='platform_api', - operation=definition.api, - mode=authorization.get('processor_type', 'none'), - synthetic=True, - outcome='success', - ) - return result - bot_id = authorization.get('bot_id') - if not bot_id: - raise ValueError('This run is not associated with a platform bot') - bot = await ap.platform_mgr.get_bot_by_uuid(execution_context, bot_id) - if bot is None: - raise ValueError(f'Bot {bot_id} is not running') - if definition.api not in set(bot.adapter.get_supported_apis() or []): - raise ValueError(f'Platform API {definition.api} is no longer supported by bot {bot_id}') - api_func = getattr(bot.adapter, definition.api, None) - if not callable(api_func): - raise ValueError(f'Platform API {definition.api} is declared but not implemented') - if definition.api == 'send_message': - normalized = { - 'target_type': _require_string(normalized, 'target_type'), - 'target_id': _require_string(normalized, 'target_id'), - 'message': message_chain - if message_chain is not None - else platform_message.MessageChain([platform_message.Plain(text=_require_string(normalized, 'text'))]), - } - from ...telemetry.platform import processing_mode - - token = processing_mode.set(authorization.get('processor_type', 'none')) + execution_id = str(session.get('run_id') or '').strip() + execution_token = set_execution_id(execution_id) try: - return await api_func(**normalized) + # This flag is frozen by the Host from the synthetic debug envelope, not tool arguments. + mock = ( + delivery.get('surface') == 'webui' + and (delivery.get('platform_capabilities') or {}).get('debug_mock') is True + ) + if mock: + outcome = 'unknown' + error_detail = '' + try: + result = _execute_mock_platform_tool(definition, context, normalized) + if message_chain is not None: + result['parameters']['message'] = message_chain.model_dump(mode='json') + outcome = 'success' + except Exception as exc: + outcome = 'failed' + error_detail = str(exc) + raise + finally: + record( + ap, + execution_context, + family='platform_api', + operation=definition.api, + mode=authorization.get('processor_type', 'none'), + synthetic=True, + outcome=outcome, + error=error_detail, + execution_id=execution_id or None, + ) + return result + bot_id = authorization.get('bot_id') + if not bot_id: + raise ValueError('This run is not associated with a platform bot') + bot = await ap.platform_mgr.get_bot_by_uuid(execution_context, bot_id) + if bot is None: + raise ValueError(f'Bot {bot_id} is not running') + if definition.api not in set(bot.adapter.get_supported_apis() or []): + raise ValueError(f'Platform API {definition.api} is no longer supported by bot {bot_id}') + api_func = getattr(bot.adapter, definition.api, None) + if not callable(api_func): + raise ValueError(f'Platform API {definition.api} is declared but not implemented') + if definition.api == 'send_message': + normalized = { + 'target_type': _require_string(normalized, 'target_type'), + 'target_id': _require_string(normalized, 'target_id'), + 'message': message_chain + if message_chain is not None + else platform_message.MessageChain([platform_message.Plain(text=_require_string(normalized, 'text'))]), + } + from ...telemetry.platform import processing_mode + + token = processing_mode.set(authorization.get('processor_type', 'none')) + try: + return await api_func(**normalized) + finally: + processing_mode.reset(token) finally: - processing_mode.reset(token) + reset_execution_id(execution_token) def _execute_mock_platform_tool( diff --git a/src/langbot/pkg/pipeline/pipelinemgr.py b/src/langbot/pkg/pipeline/pipelinemgr.py index a737594d4..c98b94c8a 100644 --- a/src/langbot/pkg/pipeline/pipelinemgr.py +++ b/src/langbot/pkg/pipeline/pipelinemgr.py @@ -3,6 +3,7 @@ from __future__ import annotations import dataclasses import typing import traceback +import asyncio import sqlalchemy @@ -257,6 +258,9 @@ class RuntimePipeline: self.ap.logger.error(result.error_notice) # Mark query as having error query.variables['_monitoring_has_error'] = True + # The lane reports failures as a value instead of raising, so record the + # reason here: without it the uploaded trace would not explain the break. + self._record_lane_failure(query, str(result.error_notice)) # Record error to monitoring system try: await self._assert_execution_active(query) @@ -368,14 +372,59 @@ class RuntimePipeline: async def process_query(self, query: pipeline_query.Query): from ..telemetry.execution import ingress + from ..telemetry.execution import record as record_execution from ..telemetry.platform import processing_mode + from ..telemetry.trace import stage_scope token = processing_mode.set('pipeline') + # The Workspace-scoped opaque query uuid is the execution identity Space + # shows for this lane; it is a stable UUID for every pooled query. + execution_id = str(getattr(query, 'query_uuid', '') or '').strip() or str(query.query_id) try: # Callers without a platform event (Webchat, HTTP API) still get one # trace for the whole Pipeline lane; nested calls reuse the trace. - with ingress(self.ap, 'pipeline_done'): - return await self._process_query(query) + with ingress(self.ap, 'pipeline_done', getattr(self, 'execution_context', None), execution_id=execution_id): + # The lane is its own workflow step: acknowledgements and replies it + # sends directly, and the runner it drives, hang under this node. + # When the lane runs under a platform route the node nests there. + with stage_scope() as node: + lane_outcome = 'success' + lane_error = '' + try: + return await self._process_query(query) + except asyncio.CancelledError: + lane_outcome = 'cancelled' + lane_error = 'cancelled' + raise + except BaseException as exc: + lane_outcome = 'failed' + lane_error = str(exc) or type(exc).__name__ + raise + finally: + try: + variables = getattr(query, 'variables', None) or {} + lane_has_error = bool(variables.get('_monitoring_has_error')) + except Exception: + lane_has_error = False + if lane_outcome == 'success' and lane_has_error: + # The lane reported the failure as a value, not an exception. + lane_outcome = 'failed' + config = getattr(query, 'pipeline_config', None) + try: + runner_id = (RunnerConfigResolver.resolve_runner_id(config) or '') if config else '' + except Exception: + runner_id = '' + record_execution( + self.ap, + getattr(self, 'execution_context', None), + family='pipeline', + operation='run', + mode='pipeline', + runner=runner_id, + outcome=lane_outcome, + error=lane_error, + node=node, + ) finally: processing_mode.reset(token) @@ -490,6 +539,9 @@ class RuntimePipeline: inst_name = query.current_stage_name if query.current_stage_name else 'unknown' self.ap.logger.error(f'Error processing query {query.query_id} stage={inst_name} : {e}') self.ap.logger.error(f'Traceback: {traceback.format_exc()}') + # The lane itself broke: land the reason on the execution trace so the + # chain is uploaded even though the error never became a StageProcessResult. + self._record_lane_failure(query, str(e) or type(e).__name__) # Record query error try: @@ -513,6 +565,46 @@ class RuntimePipeline: self.ap.logger.debug(f'Query {query.query_id} processed') await self.ap.query_pool.remove_query(query) + def _record_lane_failure(self, query: pipeline_query.Query, reason: str) -> None: + """Land one failed ``runner/execute`` stage on the owning execution trace.""" + try: + from ..telemetry.execution import record as record_execution + from ..telemetry.trace import current as current_trace + from ..telemetry.trace import stage_scope + + detail = reason or 'pipeline_lane_failed' + state = current_trace() + # The runner orchestrator already lands its own failed ``runner/execute`` + # node when the break happens inside a runner; in that case only carry the + # reason over instead of repeating the node on the trace. + already_recorded = bool( + state is not None + and any( + stage.get('family') == 'runner' and stage.get('outcome') not in ('success', 'skipped') + for stage in state.stages + ) + ) + if not already_recorded: + # The lane failure is its own workflow step, hanging under whatever + # step routed into it (or a root when nothing did). + with stage_scope() as node: + record_execution( + self.ap, + getattr(self, 'execution_context', None), + family='runner', + operation='execute', + mode='pipeline', + runner=(RunnerConfigResolver.resolve_runner_id(query.pipeline_config) or ''), + outcome='failed', + error=detail, + node=node, + ) + state = current_trace() + if state is not None: + state.mark_failure(detail) + except Exception: + pass + class PipelineManager: """流水线管理器""" diff --git a/src/langbot/pkg/pipeline/process/handlers/chat.py b/src/langbot/pkg/pipeline/process/handlers/chat.py index 2fc7b0f82..2cd681fa1 100644 --- a/src/langbot/pkg/pipeline/process/handlers/chat.py +++ b/src/langbot/pkg/pipeline/process/handlers/chat.py @@ -3,8 +3,6 @@ from __future__ import annotations import uuid import typing import traceback -import time -from datetime import datetime from .. import handler @@ -14,9 +12,7 @@ from ... import plugin_diagnostics import langbot_plugin.api.entities.events as events from ....agent.runner.config_resolver import RunnerConfigResolver from ....agent.runner import config_schema -from ....utils import constants, runner as runner_utils -from ....telemetry import features as telemetry_features -from ....telemetry.identity import workspace_identity +from ....utils import runner as runner_utils import langbot_plugin.api.entities.builtin.provider.session as provider_session import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query import langbot_plugin.api.entities.builtin.provider.message as provider_message @@ -115,9 +111,6 @@ class ChatMessageHandler(handler.MessageHandler): text_length = 0 runner = None try: - # Mark start time for telemetry - start_ts = time.time() - try_claim_steering = getattr( self.ap.agent_run_orchestrator, 'try_claim_steering_from_query', @@ -258,69 +251,48 @@ class ChatMessageHandler(handler.MessageHandler): debug_notice=traceback.format_exc(), ) finally: - # Telemetry reporting + # Telemetry: the per-execution record is built when the owning + # trace closes, so only attach the execution-scoped fields this + # lane knows here. try: - end_ts = time.time() - duration_ms = None - if 'start_ts' in locals(): - duration_ms = int((end_ts - start_ts) * 1000) + from ....telemetry.trace import current as current_trace + state = current_trace() + if state is not None: + adapter_name = query.adapter.__class__.__name__ if hasattr(query, 'adapter') else '' + runner_name = self.ap.agent_run_orchestrator.resolve_runner_id_for_telemetry(query) + state.adapter = state.adapter or adapter_name + state.runner = state.runner or (runner_name or '') + state.runner_category = state.runner_category or runner_utils.get_runner_category_from_runner( + runner_name, None, query.pipeline_config + ) + model_name = '' + try: + if getattr(query, 'use_llm_model_uuid', None): + m = await self.ap.model_mgr.get_model_by_uuid( + get_query_execution_context(query), + query.use_llm_model_uuid, + ) + if m and getattr(m, 'model_entity', None): + model_name = getattr(m.model_entity, 'name', '') or '' + except Exception: + model_name = '' + state.model_name = state.model_name or model_name + if state.pipeline_plugins is None: + state.pipeline_plugins = query.variables.get('_pipeline_bound_plugins', None) + except Exception as ex: + self.ap.logger.warning(f'Failed to attach execution telemetry fields: {ex}') + + # Trigger survey events on successful non-WebSocket responses + try: adapter_name = query.adapter.__class__.__name__ if hasattr(query, 'adapter') else None - - # Use orchestrator to resolve runner ID for telemetry - runner_name = self.ap.agent_run_orchestrator.resolve_runner_id_for_telemetry(query) - - # Model name if available - model_name = None - try: - if getattr(query, 'use_llm_model_uuid', None): - m = await self.ap.model_mgr.get_model_by_uuid( - get_query_execution_context(query), - query.use_llm_model_uuid, - ) - if m and getattr(m, 'model_entity', None): - model_name = getattr(m.model_entity, 'name', None) - except Exception: - model_name = None - - pipeline_plugins = query.variables.get('_pipeline_bound_plugins', None) - - runner_category = runner_utils.get_runner_category_from_runner( - runner_name, None, query.pipeline_config - ) - - # Feature usage collected during query processing (tool calls, - # knowledge base usage, sandbox executions, activated skills, ...) - features = telemetry_features.collect_features(query) - - payload = { - 'event_type': 'query', - 'query_id': query.query_id, - 'adapter': adapter_name, - 'runner': runner_name, - 'runner_category': runner_category, - 'duration_ms': duration_ms, - 'model_name': model_name, - 'version': constants.semantic_version, - **workspace_identity(get_query_execution_context(query)), - 'runtime_instance_id': constants.instance_id, - 'edition': constants.edition, - 'pipeline_plugins': pipeline_plugins, - 'features': features, - 'error': locals().get('error_info', None), - 'timestamp': datetime.utcnow().isoformat(), - } - - await self.ap.telemetry.start_send_task(payload) - - # Trigger survey events on successful non-WebSocket responses if not locals().get('error_info') and adapter_name and 'WebSocket' not in adapter_name: if self.ap.survey: await self.ap.survey.trigger_event('first_bot_response_success') # Counts toward the bot_response_success_100 milestone event await self.ap.survey.record_bot_response_success() except Exception as ex: - self.ap.logger.warning(f'Failed to send telemetry: {ex}') + self.ap.logger.warning(f'Failed to trigger survey event: {ex}') async def _ensure_conversation_for_history( self, diff --git a/src/langbot/pkg/platform/botmgr.py b/src/langbot/pkg/platform/botmgr.py index d4430f55d..7b942c54d 100644 --- a/src/langbot/pkg/platform/botmgr.py +++ b/src/langbot/pkg/platform/botmgr.py @@ -401,15 +401,19 @@ class RuntimeBot: from ..telemetry.execution import record with trace_mod.scope(route_ref=self._route_ref(binding, target_type, target_uuid)): - record( - getattr(self, 'ap', None), - getattr(self, 'execution_context', None), - family='event_route', - operation=event_type, - adapter=type(getattr(self, 'adapter', None)).__name__, - mode=target_type if target_type in {'pipeline', 'agent', 'event_processor'} else 'none', - outcome={'delivered': 'success', 'failed': 'failed'}.get(status, 'skipped'), - ) + # The routing decision is its own workflow step; the Pipeline or + # processor it delivers to hangs under this node. + with trace_mod.stage_scope() as node: + record( + getattr(self, 'ap', None), + getattr(self, 'execution_context', None), + family='event_route', + operation=event_type, + adapter=type(getattr(self, 'adapter', None)).__name__, + mode=target_type if target_type in {'pipeline', 'agent', 'event_processor'} else 'none', + outcome={'delivered': 'success', 'failed': 'failed'}.get(status, 'skipped'), + node=node, + ) return metadata def get_pipeline_target_for_event_type(self, event_type: str = 'message.received') -> str | None: @@ -736,12 +740,19 @@ class RuntimeBot: self, event: platform_events.EBAEvent, adapter: abstract_platform_adapter.AbstractMessagePlatformAdapter, + execution_id: str | None = None, ) -> AgentEventEnvelope: event_type = getattr(event, 'type', None) or event.__class__.__name__ event_time = getattr(event, 'timestamp', None) or time.time() - event_id = ( - getattr(event, 'message_id', None) or getattr(event, 'feedback_id', None) or f'{event_type}:{uuid.uuid4()}' - ) + if execution_id: + # Reuse the ingress execution identity so the run view, the journaled + # event and the uploaded chain all share one id. + envelope_event_id = str(execution_id) + prefix = f'platform:{self.bot_entity.uuid}:' + event_id = envelope_event_id[len(prefix) :] if envelope_event_id.startswith(prefix) else envelope_event_id + else: + event_id = self._platform_event_raw_id(event) + envelope_event_id = f'platform:{self.bot_entity.uuid}:{event_id}' target_type, target_id, target_metadata = self._infer_reply_target(event) supported_apis = self._get_adapter_supported_apis(adapter) @@ -773,7 +784,7 @@ class RuntimeBot: delivery_data['interactions'] = interaction_capabilities return AgentEventEnvelope( - event_id=f'platform:{self.bot_entity.uuid}:{event_id}', + event_id=envelope_event_id, event_type=event_type, event_time=int(event_time) if isinstance(event_time, (int, float)) else None, source='platform', @@ -863,6 +874,21 @@ class RuntimeBot: processor_id=agent.get('uuid'), ) + def _platform_event_execution_id(self, event: platform_events.EBAEvent) -> str: + """Space-visible execution identity for one inbound platform event. + + Matches the ``AgentEventEnvelope.event_id`` the same event produces, so + the stored chain and the Space run view share one identity. + """ + return f'platform:{self.bot_entity.uuid}:{self._platform_event_raw_id(event)}' + + @staticmethod + def _platform_event_raw_id(event: platform_events.EBAEvent) -> str: + event_type = getattr(event, 'type', None) or event.__class__.__name__ + return str( + getattr(event, 'message_id', None) or getattr(event, 'feedback_id', None) or f'{event_type}:{uuid.uuid4()}' + ) + async def _handle_platform_event( self, event: platform_events.EBAEvent, @@ -870,51 +896,85 @@ class RuntimeBot: ) -> None: # One inbound event owns one execution trace; every stage recorded while # it is handled (routing, runner, platform API calls) joins that trace. + from ..telemetry import trace as trace_mod from ..telemetry.execution import ingress event.bot_uuid = self.bot_entity.uuid - with ingress(getattr(self, 'ap', None), 'event_done'): - await self._handle_platform_event_body(event, adapter) + execution_id = self._platform_event_execution_id(event) + with ingress(getattr(self, 'ap', None), 'event_done', getattr(self, 'execution_context', None), execution_id): + # The inbound event is the root workflow step: everything recorded + # while handling it hangs under this node. + with trace_mod.stage_scope() as node: + await self._handle_platform_event_body(event, adapter, execution_id, node) - async def _handle_platform_event_body( - self, - event: platform_events.EBAEvent, - adapter: abstract_platform_adapter.AbstractMessagePlatformAdapter, - ) -> None: + @staticmethod + def _summarize_event_delivery(results: list[typing.Any]) -> tuple[str, str]: + outcome = 'success' + error_detail = '' + for result in results: + if isinstance(result, BaseException): + outcome = 'failed' + error_detail = error_detail or f'{type(result).__name__}: {result}' + return outcome, error_detail + + def _record_platform_event(self, event, adapter, outcome: str, error_detail: str, node: str = '') -> None: from ..telemetry.execution import record record( getattr(self, 'ap', None), getattr(self, 'execution_context', None), family='platform_event', - operation=event.type, + operation=str(getattr(event, 'type', None) or event.__class__.__name__), adapter=adapter.__class__.__name__, - outcome='success', + outcome=outcome, + error=error_detail, + node=node, ) - await self._record_adapter_event(event, adapter) - primary = ( - self._handle_interaction_submission(event, adapter) - if isinstance(event, platform_events.PlatformSpecificEvent) and event.action == 'interaction.submitted' - else self._dispatch_eba_event_to_processor(event, adapter) - ) - subscriptions = self._get_event_bindings_from_value(getattr(self.bot_entity, 'plugin_processors', [])) - seen = set() - tasks = [primary] - for subscription in subscriptions: - processor_uuid = subscription.get('processor_uuid') - if not processor_uuid or processor_uuid in seen or not subscription.get('enabled', True): - continue - seen.add(processor_uuid) - tasks.append(self._dispatch_plugin_subscription(event, adapter, processor_uuid)) - # Start all deliveries together. A slow or failed subscriber cannot block - # another subscriber or the primary route from receiving the event. - results = await asyncio.gather(*tasks, return_exceptions=True) + async def _handle_platform_event_body( + self, + event: platform_events.EBAEvent, + adapter: abstract_platform_adapter.AbstractMessagePlatformAdapter, + execution_id: str | None = None, + node: str = '', + ) -> None: + results: list[typing.Any] = [] + try: + await self._record_adapter_event(event, adapter) + + primary = ( + self._handle_interaction_submission(event, adapter, execution_id) + if isinstance(event, platform_events.PlatformSpecificEvent) and event.action == 'interaction.submitted' + else self._dispatch_eba_event_to_processor(event, adapter, execution_id=execution_id) + ) + subscriptions = self._get_event_bindings_from_value(getattr(self.bot_entity, 'plugin_processors', [])) + seen = set() + tasks = [primary] + for subscription in subscriptions: + processor_uuid = subscription.get('processor_uuid') + if not processor_uuid or processor_uuid in seen or not subscription.get('enabled', True): + continue + seen.add(processor_uuid) + tasks.append(self._dispatch_plugin_subscription(event, adapter, processor_uuid, execution_id)) + # Start all deliveries together. A slow or failed subscriber cannot block + # another subscriber or the primary route from receiving the event. + results = await asyncio.gather(*tasks, return_exceptions=True) + except BaseException as exc: + self._record_platform_event(event, adapter, 'failed', f'{type(exc).__name__}: {exc}', node) + raise + + # Recorded after handling so the inbound stage carries the real outcome + # instead of a hardcoded success. + outcome, error_detail = self._summarize_event_delivery(results) for result in results: if isinstance(result, BaseException): - await self.logger.error(f'Event delivery failed: {result}') + try: + await self.logger.error(f'Event delivery failed: {result}') + except Exception: + pass + self._record_platform_event(event, adapter, outcome, error_detail, node) - async def _dispatch_plugin_subscription(self, event, adapter, processor_uuid): + async def _dispatch_plugin_subscription(self, event, adapter, processor_uuid, execution_id=None): event_type = event.type event_binding = { 'id': f'plugin_processor:{processor_uuid}', @@ -934,7 +994,9 @@ class RuntimeBot: if not patterns or not self._agent_supports_event_type(patterns, event_type): return agent = {**agent, 'supported_event_patterns': patterns} - return await self._dispatch_eba_event_to_processor(event, adapter, event_binding, agent) + return await self._dispatch_eba_event_to_processor( + event, adapter, event_binding, agent, execution_id=execution_id + ) except Exception as exc: return await self._record_event_route_trace( event_type=event_type, @@ -954,6 +1016,7 @@ class RuntimeBot: adapter: abstract_platform_adapter.AbstractMessagePlatformAdapter, event_binding: dict | None = None, agent: dict | None = None, + execution_id: str | None = None, ) -> dict[str, typing.Any]: event_type = getattr(event, 'type', None) or event.__class__.__name__ @@ -1101,7 +1164,7 @@ class RuntimeBot: text=f'Event {event_type} target agent has no runner: {target_uuid}', ) - envelope = self._eba_event_to_agent_envelope(event, adapter) + envelope = self._eba_event_to_agent_envelope(event, adapter, execution_id) if target_type == 'event_processor': envelope.data = event.model_dump(mode='json', exclude={'source_platform_object', 'legacy_event'}) from ..telemetry import trace as trace_mod @@ -1318,6 +1381,7 @@ class RuntimeBot: self, event: platform_events.PlatformSpecificEvent, adapter: abstract_platform_adapter.AbstractMessagePlatformAdapter, + execution_id: str | None = None, ) -> None: """Consume a platform interaction callback and resume its original processor.""" data = event.data if isinstance(event.data, dict) else {} @@ -1346,7 +1410,7 @@ class RuntimeBot: await self.ap.agent_run_orchestrator.interaction_manager.acknowledge_submission(record, adapter) if record['processor_type'] == 'agent': - await self._resume_agent_interaction(record, event, adapter, actor_id) + await self._resume_agent_interaction(record, event, adapter, actor_id, execution_id) return if record['processor_type'] != 'pipeline': raise ValueError(f'Unsupported interaction processor type: {record["processor_type"]}') @@ -1428,6 +1492,7 @@ class RuntimeBot: event: platform_events.PlatformSpecificEvent, adapter: abstract_platform_adapter.AbstractMessagePlatformAdapter, actor_id: str | None, + execution_id: str | None = None, ) -> None: """Resume the exact independent Agent that produced an interaction.""" agent = await self.ap.agent_service.get_agent( @@ -1480,7 +1545,7 @@ class RuntimeBot: delivery_data['interactions'] = interaction_capabilities envelope = AgentEventEnvelope( - event_id=f'interaction:{record["id"]}:{uuid.uuid4()}', + event_id=execution_id or f'interaction:{record["id"]}:{uuid.uuid4()}', event_type='interaction.submitted', event_time=int(event.timestamp) if event.timestamp else int(time.time()), source='platform', diff --git a/src/langbot/pkg/telemetry/execution.py b/src/langbot/pkg/telemetry/execution.py index ff5ef28ff..f4af0defa 100644 --- a/src/langbot/pkg/telemetry/execution.py +++ b/src/langbot/pkg/telemetry/execution.py @@ -1,34 +1,33 @@ -"""Bounded, content-free execution counters for the existing telemetry sender. +"""Content-free execution records for the existing telemetry sender. This module reports observations only. Coverage catalogs and acceptance rules -belong to Space. Aggregation keys include both immutable execution identities. +belong to Space. -Two shapes share one payload type: +One complete execution is one record: the execution identity (Pipeline query, +inbound platform event or Agent run), its bounded ordered stages and its +terminal outcome. Nothing is aggregated or counted across executions — every +record carries the id of the execution it belongs to, so Space can fetch the +whole chain by primary identity. Records are buffered whole and flushed as +``{"records": [...]}`` on the configured cadence and on shutdown; an in-flight +trace is never flushed half-written. -* window counters, aggregated by ``(family, operation, mode, adapter, runner, - outcome, synthetic)`` — unchanged coverage reporting; -* per-event traces, one bounded payload per traced event, keyed by - ``query_id = trace_id`` so Space can fetch a whole chain by primary identity. - -Traces are additive: they never alter the counters, and both are bounded in -memory, batch size and send concurrency. +Memory is bounded in stages per trace, buffered traces, buffered records, the +cross-task execution registry and flush batch size. """ from __future__ import annotations import asyncio import contextlib +import contextvars +import json import time from datetime import datetime, timezone -from uuid import uuid4 from . import trace as trace_mod from .identity import workspace_identity from .trace import TraceState -MAX_KEYS = 512 -MAX_BATCH = 32 -FLUSH_SECONDS = 60 MODES = frozenset({'pipeline', 'agent', 'event_processor', 'none'}) OUTCOMES = frozenset({'success', 'failed', 'cancelled', 'timeout', 'skipped', 'unknown'}) @@ -39,15 +38,70 @@ TRACE_MODES = frozenset({'off', 'failures', 'sampled', 'all'}) DEFAULT_TRACE_MODE = 'sampled' DEFAULT_TRACE_SAMPLE = 20 +# Record bounds: buffered records, one flush batch, flush cadence, registry. +MAX_BUFFERED_RECORDS = 512 +MAX_RECORDS_PER_FLUSH = 128 +MAX_FLUSH_BYTES = 200 * 1024 +DEFAULT_FLUSH_SECONDS = 180 +MAX_REGISTRY_KEYS = 256 + +# Set by cross-task call sites (plugin/RPC actions) that run without the +# ingress context but know which execution they belong to. +_execution_id: contextvars.ContextVar[str] = contextvars.ContextVar('telemetry_execution_id', default='') + + +def set_execution_id(execution_id: str) -> contextvars.Token: + return _execution_id.set(str(execution_id or '').strip()) + + +def reset_execution_id(token: contextvars.Token | None) -> None: + if token is None: + return + try: + _execution_id.reset(token) + except (ValueError, RuntimeError): + # A token from another context must never break execution. + pass + + +def current_execution_id() -> str: + return _execution_id.get() + + +def _bounded_error(value, limit: int = 400) -> str: + """Bound and de-fang one failure detail before it leaves the process. + + Error text is operator-facing: keep it single-line, printable and short so a + noisy failure cannot smuggle control characters or bloat the payload. + """ + if value is None: + return '' + text = ''.join(ch if ch.isprintable() else ' ' for ch in str(value)) + text = ' '.join(text.split()) + return text[:limit] + + +def _duration_ms(started_at: str, ended_at: str) -> int: + """Wall-clock duration of one execution; 0 when the interval is unusable.""" + try: + delta = datetime.fromisoformat(ended_at) - datetime.fromisoformat(started_at) + return max(int(delta.total_seconds() * 1000), 0) + except Exception: + return 0 + class ExecutionCounters: def __init__(self, manager): self.manager = manager - self.pending: dict[tuple, dict] = {} + self.records: list[dict] = [] self.task: asyncio.Task | None = None self.dropped = 0 self.traces: dict[str, TraceState] = {} self.trace_deadlines: dict[str, float] = {} + # Execution identity -> in-flight trace, for stages recorded by tasks + # that no longer share the ingress context (plugin/RPC platform calls). + self.registry: dict[str, TraceState] = {} + self.registry_deadlines: dict[str, float] = {} self.dropped_traces = 0 # ------------------------------------------------------------------ config @@ -63,6 +117,13 @@ class ExecutionCounters: sample = DEFAULT_TRACE_SAMPLE return sample if 1 <= sample <= 100000 else DEFAULT_TRACE_SAMPLE + def flush_seconds(self) -> int: + try: + seconds = int(self.manager.telemetry_config.get('telemetry_flush_seconds', DEFAULT_FLUSH_SECONDS)) + except (TypeError, ValueError): + seconds = DEFAULT_FLUSH_SECONDS + return seconds if 1 <= seconds <= 86400 else DEFAULT_FLUSH_SECONDS + # --------------------------------------------------------------- recording def record( @@ -76,67 +137,91 @@ class ExecutionCounters: runner: str = '', outcome: str = 'unknown', synthetic: bool = False, + error: str = '', + execution_id: str | None = None, + node: str = '', ): try: cfg = self.manager.telemetry_config if not cfg or cfg.get('disable_telemetry', False) or not cfg.get('url'): return - if family not in {'platform_event', 'event_route', 'runner', 'platform_api'}: + if family not in {'platform_event', 'event_route', 'pipeline', 'runner', 'platform_api'}: return if mode not in MODES or outcome not in OUTCOMES: return # Only code-defined identifiers are accepted; never pass user values. - if any(not isinstance(v, str) or len(v) > 160 for v in (operation, adapter, runner)): + if any(not isinstance(v, str) or len(v) > 160 for v in (operation, adapter, runner, node)): return - identity = workspace_identity(context) state = trace_mod.current() - if state is not None: - self._record_trace_stage( - state, - identity=identity, - family=family, - operation=operation, - mode=mode, - adapter=adapter, - runner=runner, - outcome=outcome, - synthetic=synthetic, - ) - key = ( - identity['instance_id'], - identity['workspace_uuid'], - family, - operation, - mode, - adapter, - runner, - outcome, - bool(synthetic), + if state is None: + # Cross-task work has no inherited trace; resolve the owning + # execution through the registry instead. + lookup = str(execution_id or '').strip() or current_execution_id() + if lookup: + state = self.registry.get(lookup) + if state is None: + # Every observation belongs to exactly one execution chain. + return + try: + identity = workspace_identity(context) + except Exception: + # A cross-task caller may only carry the Workspace id; reuse the + # identity already bound to the owning trace. + identity = state.identity or {} + if not identity.get('instance_id') or not identity.get('workspace_uuid'): + return + self._record_trace_stage( + state, + identity=identity, + family=family, + operation=operation, + mode=mode, + adapter=adapter, + runner=runner, + outcome=outcome, + synthetic=synthetic, + error=_bounded_error(error), + node=node, ) - row = self.pending.get(key) - if row is None: - if len(self.pending) >= MAX_KEYS: - self.dropped += 1 - return - row = {'count': 0, 'first_seen': datetime.now(timezone.utc).isoformat()} - self.pending[key] = row - row['count'] = min(row['count'] + 1, 2147483647) - row['last_seen'] = datetime.now(timezone.utc).isoformat() self._ensure_loop() except Exception: # Observability must never change execution behavior. return def _record_trace_stage( - self, state: TraceState, *, identity, family, operation, mode, adapter, runner, outcome, synthetic + self, + state: TraceState, + *, + identity, + family, + operation, + mode, + adapter, + runner, + outcome, + synthetic, + error: str = '', + node: str = '', ) -> None: if state.abandoned: return + if trace_mod.current() is state: + # Recorded by the task that owns the step: its step stack is exact. + # A step's own record sits under its enclosing step, never itself. + parent = trace_mod.current_parent() + if node and parent == node: + parent = trace_mod.enclosing_parent() + else: + # A cross-task observation (plugin/RPC) cannot see the step + # contextvar; attach it to the innermost step the owning task has + # open on the trace. + parent = state.open_nodes[-1] if state.open_nodes else '' if state.trace_id not in self.traces: # Never buffer traces this configuration would discard anyway. if self.trace_mode() == 'off' or len(self.traces) >= MAX_TRACES: self.dropped += 1 state.abandoned = True + self._unregister_state(state) return self.traces[state.trace_id] = state self.trace_deadlines[state.trace_id] = time.monotonic() + TRACE_TTL_SECONDS @@ -149,50 +234,79 @@ class ExecutionCounters: runner=runner, outcome=outcome, synthetic=bool(synthetic), + error=error, + node=node, + parent=parent, ) def _ensure_loop(self) -> None: if self.task is None or self.task.done(): self.task = asyncio.create_task(self._loop()) - # ------------------------------------------------------------------ traces + # --------------------------------------------------------------- registry - def close_trace(self, state: TraceState, reason: str = 'event_done') -> None: - """Emit one trace payload when it matches the configured sample.""" + def register_trace(self, state: TraceState, execution_id) -> None: + """Alias one more execution identity onto an in-flight trace.""" try: - if self.traces.pop(state.trace_id, None) is None: + exec_id = str(execution_id or '').strip() + if not exec_id or state.abandoned: return - self.trace_deadlines.pop(state.trace_id, None) - if not state.stages or not self._trace_emitted(state): - return - payload = self._build_trace_payload(state, reason) - if payload is not None: - self._dispatch_payload(payload) + if exec_id not in self.registry and len(self.registry) >= MAX_REGISTRY_KEYS: + oldest = next(iter(self.registry)) + self.registry.pop(oldest, None) + self.registry_deadlines.pop(oldest, None) + self.registry[exec_id] = state + self.registry_deadlines[exec_id] = time.monotonic() + TRACE_TTL_SECONDS except Exception: return - def _dispatch_payload(self, payload: dict) -> None: - """Hand one built payload to the telemetry manager from this sync context. + def _unregister_state(self, state: TraceState) -> None: + for exec_id in [key for key, value in self.registry.items() if value is state]: + self.registry.pop(exec_id, None) + self.registry_deadlines.pop(exec_id, None) - TelemetryManager.start_send_task is a coroutine, so calling it without - scheduling dropped every trace closed here. Stand-ins used by tests and - manual tools schedule synchronously, hence the coroutine check. - """ - result = self.manager.start_send_task(payload) - if not asyncio.iscoroutine(result): - return + def _sweep_registry(self, now: float) -> None: + """Drop aliases whose owning execution never closed (leaked bindings).""" + for exec_id in [key for key, deadline in self.registry_deadlines.items() if deadline <= now]: + self.registry.pop(exec_id, None) + self.registry_deadlines.pop(exec_id, None) + + # ------------------------------------------------------------------ traces + + def close_trace(self, state: TraceState, reason: str = 'event_done', error: str = '') -> None: + """Buffer one complete execution record when it matches the sample.""" try: - asyncio.get_running_loop().create_task(result) - except RuntimeError: - result.close() - logger = getattr(getattr(self.manager, 'ap', None), 'logger', None) - if logger is not None: - logger.debug('Execution trace payload dropped: no running event loop') + if state.closed: + return + state.closed = True + registered = self.traces.pop(state.trace_id, None) is not None + self.trace_deadlines.pop(state.trace_id, None) + self._unregister_state(state) + if not registered and not state.failure_reason: + return + if (not state.stages and not state.failure_reason) or not self._trace_emitted(state): + return + record = self._build_record(state, reason, error) + if record is not None: + self._enqueue_record(record) + except Exception: + return + + def _enqueue_record(self, record: dict) -> None: + if len(self.records) >= MAX_BUFFERED_RECORDS: + self.records.pop(0) + self.dropped += 1 + self.records.append(record) + self._ensure_loop() def _trace_emitted(self, state: TraceState) -> bool: mode = self.trace_mode() if mode == 'off': return False + # A chain that failed must always be uploaded, even when no stage was + # recorded before it broke: the reason is the whole point of the trace. + if state.failure_reason: + return True if mode == 'all': return True # A failed, cancelled or timed-out stage is always worth keeping. @@ -208,31 +322,69 @@ class ExecutionCounters: except ValueError: return False - def _build_trace_payload(self, state: TraceState, reason: str) -> dict | None: + def _build_record(self, state: TraceState, reason: str, error: str = '') -> dict | None: from ..utils import constants identity = state.identity if not identity.get('instance_id') or not identity.get('workspace_uuid'): return None - observations = [{**stage, 'trace_id': state.trace_id, 'count': 1} for stage in state.stages] + ended_at = datetime.now(timezone.utc).isoformat() + stages = state.stages + observations = [] + for stage in stages: + observation = { + 'seq': stage['seq'], + 'family': stage['family'], + 'operation': stage['operation'], + 'mode': stage['mode'], + 'adapter': stage['adapter'], + 'runner': stage['runner'], + 'outcome': stage['outcome'], + 'synthetic': stage['synthetic'], + 'error': stage['error'], + 'route_ref': stage['route_ref'], + 'run_id': stage['run_id'], + 'first_seen': stage['first_seen'], + 'last_seen': stage['last_seen'], + 'trace_id': state.trace_id, + } + # Workflow position is optional: records outside any step omit both + # keys entirely, so their payloads stay byte-identical. + if stage.get('node'): + observation['node'] = stage['node'] + if stage.get('parent'): + observation['parent'] = stage['parent'] + observations.append(observation) return { 'event_type': 'feature_execution', - # One payload per trace: lookup uses the indexed query_id column. + # The execution identity: one record == one chain == one Space row. 'query_id': state.trace_id, 'instance_id': identity['instance_id'], 'workspace_uuid': identity['workspace_uuid'], 'version': constants.semantic_version, 'edition': constants.edition, - 'timestamp': datetime.now(timezone.utc).isoformat(), + 'timestamp': ended_at, + 'trusted': True, + # First failure detail in this chain, so Space can show why it broke. + 'error': _bounded_error(error) + or state.failure_reason + or next((stage['error'] for stage in stages if stage['error']), ''), + # Legacy-compatible columns Space already maps; best effort. + 'duration_ms': _duration_ms(state.started_at, ended_at), + 'model_name': state.model_name or '', + 'adapter': state.adapter or next((stage['adapter'] for stage in stages if stage['adapter']), ''), + 'runner': state.runner or next((stage['runner'] for stage in stages if stage['runner']), ''), + 'runner_category': state.runner_category or '', + 'pipeline_plugins': state.pipeline_plugins, 'features': { 'schema': 1, 'observations': observations, 'trace': { 'closed_by': reason, 'started_at': state.started_at, - 'ended_at': datetime.now(timezone.utc).isoformat(), - 'route_ref': next((s['route_ref'] for s in state.stages if s['route_ref']), ''), - 'run_id': next((s['run_id'] for s in state.stages if s['run_id']), ''), + 'ended_at': ended_at, + 'route_ref': next((stage['route_ref'] for stage in stages if stage['route_ref']), ''), + 'run_id': next((stage['run_id'] for stage in stages if stage['run_id']), ''), 'outcome': state.outcome(), 'dropped_stages': state.dropped_stages, }, @@ -248,56 +400,36 @@ class ExecutionCounters: # ----------------------------------------------------------------- flushing async def _loop(self): + interval = self.flush_seconds() last_flush = time.monotonic() - while self.pending or self.traces: - # Traces expire on a short TTL; counters keep their minute cadence. - await asyncio.sleep(1 if self.traces else FLUSH_SECONDS) + while self.records or self.traces or self.registry: + # Traces expire on a short TTL; records keep their own cadence. + await asyncio.sleep(1 if (self.traces or self.registry) else interval) now = time.monotonic() self._sweep_traces(now) - if self.pending and now - last_flush >= FLUSH_SECONDS: + self._sweep_registry(now) + if self.records and now - last_flush >= interval: last_flush = now await self.flush() self.task = None async def flush(self): - from ..utils import constants - - # Remove only one bounded batch; subsequent windows drain the remainder. - # Drain one tenant per minute, at most one request, round-robin by insertion. - if not self.pending: + if not self.records: return - tenant = next(iter(self.pending))[:2] - keys = [key for key in self.pending if key[:2] == tenant][:MAX_BATCH] - groups: dict[tuple[str, str], list[dict]] = {} - for key in keys: - row = self.pending.pop(key) - instance, workspace, family, operation, mode, adapter, runner, outcome, synthetic = key - groups.setdefault((instance, workspace), []).append( - { - **row, - 'family': family, - 'operation': operation, - 'mode': mode, - 'adapter': adapter, - 'runner': runner, - 'outcome': outcome, - 'synthetic': synthetic, - } - ) - for (instance, workspace), observations in groups.items(): - payload = { - 'event_type': 'feature_execution', - 'query_id': str(uuid4()), - 'instance_id': instance, - 'workspace_uuid': workspace, - 'version': constants.semantic_version, - 'edition': constants.edition, - 'timestamp': datetime.now(timezone.utc).isoformat(), - 'features': {'schema': 1, 'observations': observations}, - } - if not await self.manager.send(payload): - await asyncio.sleep(1) - await self.manager.send(payload) + batch: list[dict] = [] + size = 0 + while self.records and len(batch) < MAX_RECORDS_PER_FLUSH: + encoded = len(json.dumps(self.records[0], ensure_ascii=False, default=str)) + if batch and size + encoded > MAX_FLUSH_BYTES: + break + batch.append(self.records.pop(0)) + size += encoded + if not batch: + return + payload = {'records': batch} + if not await self.manager.send(payload): + await asyncio.sleep(1) + await self.manager.send(payload) async def shutdown(self): if self.task is not None: @@ -305,47 +437,93 @@ class ExecutionCounters: await asyncio.gather(self.task, return_exceptions=True) self.task = None try: - await asyncio.wait_for(self.flush(), timeout=2) + deadline = time.monotonic() + 5 + while self.records and time.monotonic() < deadline: + remaining = len(self.records) + await asyncio.wait_for(self.flush(), timeout=2) + if len(self.records) >= remaining: + break except (Exception, asyncio.CancelledError): pass - self.pending.clear() - # Traces of in-flight events cannot be completed during shutdown. + self.records.clear() + # Executions still in flight cannot be completed during shutdown. self.dropped += len(self.traces) self.dropped_traces += len(self.traces) self.traces.clear() self.trace_deadlines.clear() + self.registry.clear() + self.registry_deadlines.clear() + + +def _execution_counters(ap): + counters = getattr(getattr(ap, 'telemetry', None), 'execution', None) + return counters if isinstance(counters, ExecutionCounters) else None def record(ap, context, **observation): """Best-effort bridge usable with optional telemetry and test doubles.""" try: - counters = getattr(getattr(ap, 'telemetry', None), 'execution', None) - if isinstance(counters, ExecutionCounters): + counters = _execution_counters(ap) + if counters is not None: counters.record(context, **observation) except Exception: pass -def close_trace(ap, state, reason: str = 'event_done') -> None: +def close_trace(ap, state, reason: str = 'event_done', error: str = '') -> None: """Best-effort trace close usable with optional telemetry and test doubles.""" try: - counters = getattr(getattr(ap, 'telemetry', None), 'execution', None) - if isinstance(counters, ExecutionCounters): - counters.close_trace(state, reason) + counters = _execution_counters(ap) + if counters is not None: + counters.close_trace(state, reason, error) except Exception: pass +def bind_trace(ap, execution_id: str | None = None) -> trace_mod.TraceBinding: + """Bind an execution trace and alias its identity for cross-task stages.""" + binding = trace_mod.bind(execution_id) + if execution_id: + try: + counters = _execution_counters(ap) + if counters is not None: + counters.register_trace(binding.state, execution_id) + except Exception: + pass + return binding + + @contextlib.contextmanager -def ingress(ap, reason: str = 'event_done'): - """Trace one ingress boundary; only the owner of the trace closes it. +def ingress(ap, reason: str = 'event_done', context=None, execution_id: str | None = None): + """Trace one execution boundary; only the owner of the trace closes it. Nested boundaries (an event routed into a Pipeline, a Runner invoked from a - dispatch) reuse the in-flight trace instead of starting a second one. + dispatch) reuse the in-flight trace instead of starting a second one, and + only alias their own execution id onto it. """ - binding = trace_mod.bind() + binding = trace_mod.bind(execution_id) + if binding.created and context is not None: + try: + binding.state.identity = workspace_identity(context) + except Exception: + pass + if execution_id and not binding.state.abandoned: + try: + counters = _execution_counters(ap) + if counters is not None: + counters.register_trace(binding.state, execution_id) + except Exception: + pass + failure = '' try: yield binding + except BaseException as exc: + # The boundary itself failed: keep the reason on the trace so the uploaded + # payload explains the failure instead of reporting a bare outcome. + if binding.created: + failure = f'{type(exc).__name__}: {exc}' + binding.state.mark_failure(failure) + raise finally: if trace_mod.unbind_root(binding): - close_trace(ap, binding.state, reason) + close_trace(ap, binding.state, reason, failure) diff --git a/src/langbot/pkg/telemetry/platform.py b/src/langbot/pkg/telemetry/platform.py index a7e636bcf..934f36e10 100644 --- a/src/langbot/pkg/telemetry/platform.py +++ b/src/langbot/pkg/telemetry/platform.py @@ -11,6 +11,21 @@ from .execution import record processing_mode: ContextVar[str] = ContextVar('telemetry_processing_mode', default='none') +def result_failure_detail(result) -> str: + """Pull a bounded human-readable reason out of a failed platform API result.""" + raw = getattr(result, 'raw', result) + if isinstance(raw, dict): + for key in ('error', 'message', 'msg', 'detail', 'reason'): + value = raw.get(key) + if isinstance(value, str) and value.strip(): + return value + if isinstance(value, dict): + for inner in ('message', 'msg', 'detail'): + if isinstance(value.get(inner), str) and value[inner].strip(): + return value[inner] + return '' + + def result_outcome(result): # Empty returns do not prove a remote operation succeeded. if result is None: @@ -69,18 +84,24 @@ def observe_adapter(ap, context, adapter): if action in declared: observed_operation = action outcome = 'unknown' + error_detail = '' try: result = await method(*args, **kwargs) outcome = result_outcome(result) + if outcome != 'success': + error_detail = result_failure_detail(result) return result except asyncio.CancelledError: outcome = 'cancelled' + error_detail = 'cancelled' raise - except TimeoutError: + except TimeoutError as exc: outcome = 'timeout' + error_detail = str(exc) or 'timeout' raise - except Exception: + except Exception as exc: outcome = 'failed' + error_detail = str(exc) raise finally: record( @@ -91,6 +112,7 @@ def observe_adapter(ap, context, adapter): adapter=adapter.__class__.__name__, mode=processing_mode.get(), outcome=outcome, + error=error_detail, ) return wrapped diff --git a/src/langbot/pkg/telemetry/trace.py b/src/langbot/pkg/telemetry/trace.py index a239c707e..445bb470e 100644 --- a/src/langbot/pkg/telemetry/trace.py +++ b/src/langbot/pkg/telemetry/trace.py @@ -1,9 +1,11 @@ -"""Content-free per-event execution traces. +"""Content-free per-execution traces. -One trace identity is minted per inbound platform event and survives until the -ingress handler returns. Stage records appended under it describe how that one -event was routed and processed, using only code-defined identifiers. Nothing -here is aggregated: a trace is a bounded, ordered sequence for one event. +One trace identity is one execution identity: the id a Pipeline query, inbound +platform event or Agent run is known by, so a stored chain can be joined with +the Space view of that same execution. Stage records appended under it describe +how that one execution was routed and processed, using only code-defined +identifiers. Nothing here is aggregated: a trace is a bounded, ordered sequence +for one execution. Three bounds keep memory independent of traffic volume: @@ -12,7 +14,9 @@ Three bounds keep memory independent of traffic volume: * a wall-clock TTL, enforced by the sender's sweep. The identity itself lives in a ContextVar so that asynchronous work spawned -while handling one event inherits it, mirroring ``telemetry.platform``. +while handling one execution inherits it, mirroring ``telemetry.platform``. +Cross-task stages (plugin/RPC work with no inherited context) are resolved by +the owner's execution-id registry instead. """ from __future__ import annotations @@ -28,6 +32,10 @@ MAX_STAGES = 32 _current: contextvars.ContextVar['TraceState | None'] = contextvars.ContextVar('telemetry_trace', default=None) _route: contextvars.ContextVar[str] = contextvars.ContextVar('telemetry_trace_route', default='') _run: contextvars.ContextVar[str] = contextvars.ContextVar('telemetry_trace_run', default='') +# Open workflow step nodes, innermost last. Records emitted inside a step carry +# the innermost node as their ``parent``; the step's own record carries the one +# below it (its enclosing step), never itself. +_node: contextvars.ContextVar[tuple[str, ...]] = contextvars.ContextVar('telemetry_trace_node', default=()) def _now() -> str: @@ -35,29 +43,55 @@ def _now() -> str: class TraceState: - """Bounded stage buffer for exactly one event.""" + """Bounded stage buffer for exactly one execution.""" __slots__ = ( 'trace_id', + 'execution_id', 'started_at', 'stages', 'dropped_stages', 'synthetic', 'sequence', + 'node_sequence', + 'open_nodes', 'identity', 'abandoned', + 'failure_reason', + 'closed', + 'adapter', + 'runner', + 'runner_category', + 'model_name', + 'pipeline_plugins', ) - def __init__(self) -> None: - self.trace_id = str(uuid4()) + def __init__(self, execution_id: str | None = None) -> None: + # The execution identity is the trace identity so a stored chain can be + # joined to the Space view of that same Pipeline query / event / run. + self.execution_id = str(execution_id).strip() if execution_id else '' + self.trace_id = self.execution_id or str(uuid4()) self.started_at = _now() self.stages: list[dict[str, typing.Any]] = [] self.dropped_stages = 0 self.synthetic = False self.sequence = 0 + # Node ids number the workflow steps of this one execution ("n1", "n2"...). + self.node_sequence = 0 + # Mirror of the open step stack, readable from any task: cross-task + # observations (plugin/RPC) resolve their parent through this. + self.open_nodes: tuple[str, ...] = () self.identity: dict[str, str] = {} # Set when the sender refused to buffer this trace: stop appending. self.abandoned = False + self.failure_reason = '' + self.closed = False + # Execution-scoped record fields attached by the owning lane when known. + self.adapter = '' + self.runner = '' + self.runner_category = '' + self.model_name = '' + self.pipeline_plugins: typing.Any = None def append( self, @@ -69,34 +103,62 @@ class TraceState: runner: str, outcome: str, synthetic: bool, + error: str = '', + node: str = '', + parent: str = '', ) -> None: if len(self.stages) >= MAX_STAGES: self.dropped_stages += 1 return seen = _now() - self.stages.append( - { - 'family': family, - 'operation': operation, - 'mode': mode, - 'adapter': adapter, - 'runner': runner, - 'outcome': outcome, - 'synthetic': bool(synthetic), - 'seq': self.sequence, - 'route_ref': _route.get(), - 'run_id': _run.get(), - 'first_seen': seen, - 'last_seen': seen, - } - ) + entry = { + 'family': family, + 'operation': operation, + 'mode': mode, + 'adapter': adapter, + 'runner': runner, + 'outcome': outcome, + 'synthetic': bool(synthetic), + 'seq': self.sequence, + 'route_ref': _route.get(), + 'run_id': _run.get(), + 'first_seen': seen, + 'last_seen': seen, + 'error': error, + } + # Only workflow steps carry these; legacy records stay byte-identical. + if node: + entry['node'] = node + if parent: + entry['parent'] = parent + self.stages.append(entry) self.sequence += 1 if synthetic: self.synthetic = True + def allocate_node(self) -> str: + """Mint the next step-node id for this execution ("n1", "n2", ...).""" + self.node_sequence += 1 + return f'n{self.node_sequence}' + + def mark_failure(self, reason: str) -> None: + """Remember why this chain broke so the payload can explain itself.""" + if not self.failure_reason: + self.failure_reason = reason + def outcome(self) -> str: - """Terminal outcome of the last recorded stage, for space-side display.""" - return self.stages[-1]['outcome'] if self.stages else 'unknown' + """Terminal outcome of this chain, for space-side display. + + A chain that carries a failure reason never reports ``success``: the + reason is the point of the trace. + """ + if self.stages: + last = self.stages[-1]['outcome'] + else: + last = 'unknown' + if self.failure_reason and last in ('success', 'skipped', 'unknown'): + return 'failed' + return last class TraceBinding(typing.NamedTuple): @@ -105,12 +167,16 @@ class TraceBinding(typing.NamedTuple): token: typing.Any -def bind() -> TraceBinding: - """Start a trace unless one is already in flight in this context.""" +def bind(execution_id: str | None = None) -> TraceBinding: + """Start a trace unless one is already in flight in this context. + + The execution id becomes the trace id only when this call starts the trace; + a nested boundary keeps the enclosing execution's identity. + """ existing = _current.get() if existing is not None: return TraceBinding(existing, False, None) - state = TraceState() + state = TraceState(execution_id) return TraceBinding(state, True, _current.set(state)) @@ -136,6 +202,46 @@ def current() -> TraceState | None: return _current.get() +def current_parent() -> str: + """The step node that records emitted right now attach to ('' when none).""" + stack = _node.get() + return stack[-1] if stack else '' + + +def enclosing_parent() -> str: + """The step enclosing the innermost open one; '' when it is a root step.""" + stack = _node.get() + return stack[-2] if len(stack) > 1 else '' + + +@contextlib.contextmanager +def stage_scope() -> typing.Iterator[str]: + """Open one workflow step node for the duration of the block. + + Allocates a node id unique within the execution, publishes it as the current + parent so that nested observations recorded inside the block attach to it, + and yields that id so the step's own records can be stamped with ``node=``. + Outside any trace nothing is allocated and the empty id is yielded. + """ + state = _current.get() + if state is None: + yield '' + return + node = state.allocate_node() + previous_open = state.open_nodes + state.open_nodes = previous_open + (node,) + token = _node.set(_node.get() + (node,)) + try: + yield node + finally: + state.open_nodes = previous_open + try: + _node.reset(token) + except (ValueError, RuntimeError): + # A token from another context must never break execution. + pass + + def set_run(run_id: str) -> typing.Any: return _run.set(run_id) diff --git a/src/langbot/templates/config.yaml b/src/langbot/templates/config.yaml index 19c46ae35..e7cb87990 100644 --- a/src/langbot/templates/config.yaml +++ b/src/langbot/templates/config.yaml @@ -434,3 +434,7 @@ space: # (N is execution_trace_sample) and always keeps WebUI debug runs. execution_trace: sampled execution_trace_sample: 20 + # How often buffered execution records are uploaded, in seconds. + # Records of complete executions are batched as {"records": [...]}; a + # still-running execution is never uploaded half-written. + telemetry_flush_seconds: 180 diff --git a/tests/unit_tests/telemetry/test_chat_emitter.py b/tests/unit_tests/telemetry/test_chat_emitter.py new file mode 100644 index 000000000..0c66d3f83 --- /dev/null +++ b/tests/unit_tests/telemetry/test_chat_emitter.py @@ -0,0 +1,110 @@ +"""The Chat handler must not emit the legacy per-query telemetry payload. + +Per-execution telemetry is produced by the trace that owns the Pipeline lane; +the handler only annotates that trace with execution-scoped record fields. +""" + +from __future__ import annotations + +import types +from importlib import import_module +from unittest.mock import AsyncMock, Mock + +import langbot_plugin.api.entities.builtin.provider.session as provider_session + + +def get_modules(): + # Import the application package first: the handler package is wired into + # it during initialization, so importing the handler directly would re-enter + # the partially initialized pipeline package. + import_module('langbot.pkg.core.app') + return ( + import_module('langbot.pkg.pipeline.process.handlers.chat'), + import_module('langbot.pkg.telemetry.trace'), + ) + + +class FakeEvent: + """Stand-in for the plugin event models, which need a real Query.""" + + def __init__(self, **kwargs): + self.__dict__.update(kwargs) + self.event_name = 'person_normal_message_received' + + +class FakeEventContext: + event = types.SimpleNamespace(user_message_alter=None) + + def is_prevented_default(self) -> bool: + return False + + +async def empty_run_from_query(query): + if False: # pragma: no cover - async generator with no results + yield + + +def build_ap(): + ap = Mock() + ap.logger = Mock() + ap.plugin_connector = Mock() + ap.plugin_connector.emit_event = AsyncMock(return_value=FakeEventContext()) + ap.agent_run_orchestrator = Mock() + ap.agent_run_orchestrator.try_claim_steering_from_query = AsyncMock(return_value=False) + ap.agent_run_orchestrator.run_from_query = empty_run_from_query + ap.agent_run_orchestrator.resolve_runner_id_for_telemetry = Mock(return_value='runner-1') + ap.survey = Mock() + ap.survey.trigger_event = AsyncMock() + ap.survey.record_bot_response_success = AsyncMock() + ap.telemetry = Mock() + ap.telemetry.start_send_task = AsyncMock() + ap.telemetry.send = AsyncMock() + return ap + + +def build_query(): + query = Mock() + query.query_id = 7 + query.query_uuid = 'exec-uuid' + query.launcher_type = provider_session.LauncherTypes.PERSON + query.launcher_id = 'user-1' + query.sender_id = 'user-1' + query.message_chain = 'hello' + query.message_event = object() + query.variables = {'_pipeline_bound_plugins': ['plugin:a']} + query.pipeline_config = {'ai': {}, 'output': {'misc': {}}} + query.session = Mock(using_conversation=object()) + query.use_llm_model_uuid = None + query.adapter = Mock() + query.adapter.is_stream_output_supported = AsyncMock(return_value=False) + query.resp_messages = [] + return query + + +async def test_normal_query_sends_no_legacy_payload_and_annotates_the_trace(monkeypatch): + chat, trace_mod = get_modules() + monkeypatch.setattr(chat.events, 'PersonNormalMessageReceived', FakeEvent) + monkeypatch.setattr(chat.events, 'GroupNormalMessageReceived', FakeEvent) + + ap = build_ap() + handler = chat.ChatMessageHandler(ap) + query = build_query() + + binding = trace_mod.bind('exec-uuid') + try: + async for _ in handler.handle(query): + pass + finally: + trace_mod.unbind_root(binding) + + # The legacy `event_type='query'` emitter is gone. + ap.telemetry.start_send_task.assert_not_called() + ap.telemetry.send.assert_not_called() + # The survey trigger and its milestone bookkeeping still fire. + ap.survey.trigger_event.assert_awaited_once_with('first_bot_response_success') + ap.survey.record_bot_response_success.assert_awaited_once() + # Execution-scoped fields are attached to the owning trace instead. + assert binding.state.runner == 'runner-1' + assert binding.state.runner_category == 'unknown' + assert binding.state.pipeline_plugins == ['plugin:a'] + assert binding.state.model_name == '' diff --git a/tests/unit_tests/telemetry/test_pipeline_lane.py b/tests/unit_tests/telemetry/test_pipeline_lane.py new file mode 100644 index 000000000..f091f2bb8 --- /dev/null +++ b/tests/unit_tests/telemetry/test_pipeline_lane.py @@ -0,0 +1,272 @@ +"""The Pipeline lane must land its failure on the owning execution trace.""" + +from __future__ import annotations + +import asyncio +import types +from importlib import import_module + +import pytest + + +def get_modules(): + # Import the application package first; the pipeline package is wired into it. + import_module('langbot.pkg.core.app') + return ( + import_module('langbot.pkg.pipeline.pipelinemgr'), + import_module('langbot.pkg.telemetry.execution'), + ) + + +def get_trace(): + import_module('langbot.pkg.core.app') + return import_module('langbot.pkg.telemetry.trace') + + +class FakeManager: + def __init__(self): + self.telemetry_config = {'url': 'https://space.example.test', 'execution_trace': 'all'} + self.sent: list[dict] = [] + + async def send(self, payload: dict) -> bool: + self.sent.append(payload) + return True + + +CONTEXT = types.SimpleNamespace(instance_uuid='instance-1', workspace_uuid='workspace-1') + + +def build_pipeline(pipelinemgr, ap): + pipeline = object.__new__(pipelinemgr.RuntimePipeline) + pipeline.ap = ap + pipeline.execution_context = CONTEXT + return pipeline + + +def test_lane_failure_marks_the_owned_trace_without_any_prior_stage(): + pipelinemgr, execution = get_modules() + manager = FakeManager() + counters = execution.ExecutionCounters(manager) + ap = types.SimpleNamespace(telemetry=types.SimpleNamespace(execution=counters)) + pipeline = build_pipeline(pipelinemgr, ap) + query = types.SimpleNamespace(pipeline_config={}) + + with execution.ingress(ap, 'pipeline_done', CONTEXT, execution_id='query-7'): + pipeline._record_lane_failure(query, 'sandbox exited 127') + + assert len(counters.records) == 1 + record = counters.records[0] + assert record['query_id'] == 'query-7' + assert record['error'] == 'sandbox exited 127' + assert record['features']['trace']['outcome'] == 'failed' + observations = record['features']['observations'] + assert [(row['family'], row['operation'], row['outcome']) for row in observations] == [ + ('runner', 'execute', 'failed') + ] + + +def test_lane_failure_does_not_duplicate_the_runner_node_the_orchestrator_recorded(): + """A break inside a runner is already a node; the lane only carries the reason over.""" + pipelinemgr, execution = get_modules() + manager = FakeManager() + counters = execution.ExecutionCounters(manager) + ap = types.SimpleNamespace(telemetry=types.SimpleNamespace(execution=counters)) + pipeline = build_pipeline(pipelinemgr, ap) + query = types.SimpleNamespace(pipeline_config={}) + + with execution.ingress(ap, 'pipeline_done', CONTEXT, execution_id='query-8'): + execution.record( + ap, + CONTEXT, + family='runner', + operation='execute', + mode='pipeline', + runner='plugin:langbot-team/LocalAgent/default', + outcome='failed', + error='No authorized model for local-agent', + ) + pipeline._record_lane_failure(query, 'No authorized model for local-agent') + + observations = [ + (row['family'], row['operation'], row['outcome']) + for record in counters.records + for row in record['features']['observations'] + ] + assert observations == [('runner', 'execute', 'failed')] + assert counters.records[-1]['features']['trace']['outcome'] == 'failed' + + +def build_counters(): + manager = FakeManager() + counters = import_module('langbot.pkg.telemetry.execution').ExecutionCounters(manager) + ap = types.SimpleNamespace(telemetry=types.SimpleNamespace(execution=counters)) + return counters, ap + + +def build_lane(pipelinemgr, ap): + # ``process_query`` lives on the RuntimePipeline lane, not on PipelineManager. + lane = object.__new__(pipelinemgr.RuntimePipeline) + lane.ap = ap + lane.execution_context = CONTEXT + return lane + + +def build_query(**overrides): + query = types.SimpleNamespace(query_uuid='exec-lane', query_id=1, pipeline_config={}, variables={}) + for key, value in overrides.items(): + setattr(query, key, value) + return query + + +def lane_shapes(counters): + return [ + (row['family'], row['operation'], row.get('node'), row.get('parent')) + for record in counters.records + for row in record['features']['observations'] + ] + + +async def stop_counters(counters): + """Cancel the telemetry flush task so no loop is left with pending work.""" + task = counters.task + if task is not None and not task.done(): + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + counters.task = None + + +def test_lane_step_owns_the_apis_and_runner_it_calls(): + pipelinemgr, execution = get_modules() + trace = get_trace() + counters, ap = build_counters() + lane = build_lane(pipelinemgr, ap) + query = build_query() + seen = {} + + async def fake_process(_query): + execution.record( + ap, CONTEXT, family='platform_api', operation='send_message', + adapter='AiocqhttpAdapter', mode='pipeline', outcome='success', + ) + with trace.stage_scope() as run_node: + seen['run'] = run_node + execution.record( + ap, CONTEXT, family='runner', operation='execute', mode='pipeline', outcome='success', node=run_node + ) + # A reply issued after the runner still belongs to the lane. + execution.record( + ap, CONTEXT, family='platform_api', operation='reply_message', + adapter='AiocqhttpAdapter', mode='pipeline', outcome='success', + ) + + lane._process_query = fake_process + + async def scenario(): + await lane.process_query(query) + await stop_counters(counters) + + asyncio.run(scenario()) + + shapes = lane_shapes(counters) + lane_node = shapes[-1][2] + assert lane_node == 'n1' + assert shapes == [ + ('platform_api', 'send_message', None, lane_node), + ('runner', 'execute', seen['run'], lane_node), + ('platform_api', 'reply_message', None, lane_node), + # The lane step is recorded last; a standalone lane is a root. + ('pipeline', 'run', lane_node, None), + ] + + +def test_lane_node_nests_under_the_platform_route_that_called_it(): + pipelinemgr, execution = get_modules() + trace = get_trace() + counters, ap = build_counters() + lane = build_lane(pipelinemgr, ap) + query = build_query(query_uuid='exec-routed') + + async def fake_process(_query): + execution.record( + ap, CONTEXT, family='platform_api', operation='send_message', mode='pipeline', outcome='success' + ) + + lane._process_query = fake_process + + async def scenario(): + # A platform event owns the trace; the route step wraps the Pipeline lane. + with execution.ingress(ap, 'event_done', CONTEXT, execution_id='exec-routed'): + with trace.stage_scope() as event_node: + execution.record( + ap, CONTEXT, family='platform_event', operation='message.received', + adapter='AiocqhttpAdapter', outcome='success', node=event_node, + ) + with trace.stage_scope() as route_node: + execution.record( + ap, CONTEXT, family='event_route', operation='message.received', + adapter='AiocqhttpAdapter', outcome='success', node=route_node, + ) + await lane.process_query(query) + await stop_counters(counters) + return event_node, route_node + + event_node, route_node = asyncio.run(scenario()) + + assert (event_node, route_node) == ('n1', 'n2') + assert lane_shapes(counters) == [ + ('platform_event', 'message.received', event_node, None), + ('event_route', 'message.received', route_node, event_node), + # The API is called by the lane, so it hangs on the lane node. + ('platform_api', 'send_message', None, 'n3'), + ('pipeline', 'run', 'n3', route_node), + ] + + +def test_broken_lane_records_a_failed_step_with_its_reason(): + pipelinemgr, execution = get_modules() + counters, ap = build_counters() + lane = build_lane(pipelinemgr, ap) + + async def fake_process(_query): + raise RuntimeError('lane exploded') + + lane._process_query = fake_process + + async def scenario(): + try: + await lane.process_query(build_query()) + finally: + await stop_counters(counters) + + with pytest.raises(RuntimeError, match='lane exploded'): + asyncio.run(scenario()) + + family, operation, node, parent = lane_shapes(counters)[-1] + row = counters.records[0]['features']['observations'][-1] + assert (family, operation, node, parent) == ('pipeline', 'run', 'n1', None) + assert row['outcome'] == 'failed' + assert row['error'] == 'lane exploded' + + +def test_lane_reported_failure_marks_the_step_failed(): + pipelinemgr, execution = get_modules() + counters, ap = build_counters() + lane = build_lane(pipelinemgr, ap) + query = build_query(variables={'_monitoring_has_error': True}) + + async def fake_process(_query): + return None + + lane._process_query = fake_process + + async def scenario(): + await lane.process_query(query) + await stop_counters(counters) + + asyncio.run(scenario()) + + row = counters.records[0]['features']['observations'][-1] + assert (row['family'], row['operation'], row['outcome']) == ('pipeline', 'run', 'failed') diff --git a/tests/unit_tests/telemetry/test_trace.py b/tests/unit_tests/telemetry/test_trace.py index f19ab2668..283cdac65 100644 --- a/tests/unit_tests/telemetry/test_trace.py +++ b/tests/unit_tests/telemetry/test_trace.py @@ -1,10 +1,13 @@ -"""Unit tests for per-event execution traces (pkg/telemetry/trace.py + execution.py).""" +"""Unit tests for per-execution execution traces (pkg/telemetry/trace.py + execution.py).""" from __future__ import annotations import asyncio +import contextvars +import json import time import types +from datetime import datetime, timedelta, timezone from importlib import import_module @@ -48,6 +51,11 @@ def trace_config(**overrides): return config +def records_sent(manager): + """Flatten the buffered-record batches a FakeManager received.""" + return [record for payload in manager.sent for record in payload.get('records', [])] + + CONTEXT = types.SimpleNamespace(instance_uuid='instance-1', workspace_uuid='workspace-1') STAGE = { @@ -85,6 +93,27 @@ class TestTraceIdentity: # A repeated unbind of the same binding must never raise. assert trace.unbind_root(outer) is True + def test_trace_id_is_the_execution_id(self): + trace, _ = get_modules() + binding = trace.bind('exec-1234') + try: + assert binding.created is True + assert binding.state.trace_id == 'exec-1234' + assert binding.state.execution_id == 'exec-1234' + finally: + trace.unbind_root(binding) + + def test_nested_boundary_keeps_the_enclosing_execution_id(self): + trace, _ = get_modules() + outer = trace.bind('event-1') + inner = trace.bind('query-9') + try: + assert inner.created is False + assert inner.state.trace_id == 'event-1' + finally: + trace.unbind_root(inner) + trace.unbind_root(outer) + def test_route_and_run_are_only_pinned_inside_the_block(self): trace, _ = get_modules() binding = trace.bind() @@ -174,23 +203,174 @@ class TestTraceRecording: assert bindings[-1].state.abandoned is True assert bindings[-1].state.stages == [] - def test_counters_keep_aggregating_while_tracing(self): + def test_observation_without_an_owning_execution_is_dropped(self): + trace, _ = get_modules() + _, counters = make_counters() + # Clustering is gone: an observation that belongs to no execution is + # discarded instead of being aggregated into a synthetic bucket. + counters.record(CONTEXT, **STAGE) + assert counters.records == [] + assert counters.traces == {} + + +class TestTraceNodeScope: + """Workflow step nodes: a step's own record and the observations it calls.""" + + @staticmethod + def _step(**overrides): + return { + 'family': 'runner', + 'operation': 'execute', + 'mode': 'agent', + 'runner': 'runner-1', + 'outcome': 'success', + **overrides, + } + + def test_nested_observation_carries_the_step_node_as_parent(self): trace, _ = get_modules() _, counters = make_counters() binding = trace.bind() try: - for _ in range(3): - counters.record(CONTEXT, **STAGE) + with trace.stage_scope() as step: + counters.record(CONTEXT, node=step, **self._step()) + counters.record(CONTEXT, **{**STAGE, 'family': 'platform_api', 'operation': 'send_message'}) finally: trace.unbind_root(binding) - assert list(counters.pending.values())[0]['count'] == 3 + + step_row, api_row = binding.state.stages + assert step_row['node'] == step + assert 'parent' not in step_row # a root step carries no parent + assert api_row['parent'] == step + assert 'node' not in api_row # observations are not step nodes + + def test_nested_step_records_its_enclosing_step_as_parent(self): + trace, _ = get_modules() + _, counters = make_counters() + binding = trace.bind() + try: + with trace.stage_scope() as outer: + counters.record(CONTEXT, node=outer, **self._step(mode='pipeline')) + with trace.stage_scope() as inner: + counters.record(CONTEXT, node=inner, **self._step()) + counters.record(CONTEXT, **{**STAGE, 'family': 'platform_api', 'operation': 'send_message'}) + finally: + trace.unbind_root(binding) + + outer_row, inner_row, api_row = binding.state.stages + assert (outer_row['node'], outer_row.get('parent')) == (outer, None) + # The step's own record sits under its enclosing step, never under itself. + assert (inner_row['node'], inner_row.get('parent')) == (inner, outer) + assert api_row.get('parent') == inner and 'node' not in api_row + + def test_sequential_steps_get_distinct_ids_and_roots_have_no_parent(self): + trace, _ = get_modules() + _, counters = make_counters() + binding = trace.bind() + try: + with trace.stage_scope() as first: + counters.record(CONTEXT, node=first, **self._step()) + with trace.stage_scope() as second: + counters.record(CONTEXT, node=second, **self._step()) + finally: + trace.unbind_root(binding) + + assert (first, second) == ('n1', 'n2') + assert [row['node'] for row in binding.state.stages] == [first, second] + assert all('parent' not in row for row in binding.state.stages) + + def test_cross_task_observation_attaches_to_the_open_step(self): + trace, execution = get_modules() + _, counters = make_counters() + binding = execution.bind_trace(make_ap(counters), 'exec-cross') + try: + with trace.stage_scope() as lane: + counters.record( + CONTEXT, family='pipeline', operation='run', mode='pipeline', outcome='success', node=lane + ) + # A plugin/RPC observation arrives on a task that never had the + # trace context; it is resolved through the execution registry and + # must still hang off the step the owning task has open. + foreign = contextvars.Context() + foreign.run(execution.set_execution_id, 'exec-cross') + foreign.run( + counters.record, + CONTEXT, + family='platform_api', + operation='send_message', + mode='pipeline', + outcome='success', + ) + finally: + trace.unbind_root(binding) + + rows = binding.state.stages + assert rows[0]['node'] == lane and 'parent' not in rows[0] + assert rows[1].get('parent') == lane and 'node' not in rows[1] + + def test_cross_task_observation_without_an_open_step_has_no_parent(self): + trace, execution = get_modules() + _, counters = make_counters() + binding = execution.bind_trace(make_ap(counters), 'exec-bare') + try: + foreign = contextvars.Context() + foreign.run(execution.set_execution_id, 'exec-bare') + foreign.run( + counters.record, + CONTEXT, + family='platform_api', + operation='send_message', + mode='pipeline', + outcome='success', + ) + finally: + trace.unbind_root(binding) + + row = binding.state.stages[0] + assert 'node' not in row and 'parent' not in row + + def test_scope_outside_any_trace_is_inert(self): + trace, _ = get_modules() + with trace.stage_scope() as node: + assert node == '' + assert trace.current_parent() == '' + + def test_records_outside_any_scope_serialize_without_node_keys(self): + trace, execution = get_modules() + manager, counters = make_counters(trace_config()) + binding = trace.bind('exec-legacy') + try: + counters.record(CONTEXT, **STAGE) + counters.close_trace(binding.state, 'event_done') + finally: + trace.unbind_root(binding) + + stage = binding.state.stages[0] + assert set(stage) == { + 'family', + 'operation', + 'mode', + 'adapter', + 'runner', + 'outcome', + 'synthetic', + 'seq', + 'route_ref', + 'run_id', + 'first_seen', + 'last_seen', + 'error', + } + observation = counters.records[0]['features']['observations'][0] + assert 'node' not in json.dumps(observation, sort_keys=True) + assert 'parent' not in json.dumps(observation, sort_keys=True) class TestTraceDelivery: - def test_close_emits_one_payload_per_trace(self): + def test_close_buffers_one_record_per_trace(self): trace, _ = get_modules() manager, counters = make_counters(trace_config()) - binding = trace.bind() + binding = trace.bind('exec-1') try: counters.record(CONTEXT, **STAGE) counters.record( @@ -201,18 +381,22 @@ class TestTraceDelivery: finally: trace.unbind_root(binding) - assert len(manager.sent) == 1 - payload = manager.sent[0] - assert payload['event_type'] == 'feature_execution' - assert payload['query_id'] == binding.state.trace_id - assert len(payload['query_id']) == 36 - assert payload['instance_id'] == 'instance-1' - assert payload['workspace_uuid'] == 'workspace-1' - features = payload['features'] + # Closing buffers; nothing is sent before the flush cadence. + assert manager.sent == [] + assert len(counters.records) == 1 + assert counters.traces == {} + + record = counters.records[0] + assert record['event_type'] == 'feature_execution' + assert record['query_id'] == 'exec-1' + assert record['instance_id'] == 'instance-1' + assert record['workspace_uuid'] == 'workspace-1' + assert record['trusted'] is True + features = record['features'] assert features['schema'] == 1 - assert [row['count'] for row in features['observations']] == [1, 1] + assert 'count' not in json.dumps(record) assert [row['seq'] for row in features['observations']] == [0, 1] - assert [row['trace_id'] for row in features['observations']] == [binding.state.trace_id] * 2 + assert [row['trace_id'] for row in features['observations']] == ['exec-1'] * 2 assert features['trace'] == { 'closed_by': 'event_done', 'started_at': binding.state.started_at, @@ -222,7 +406,6 @@ class TestTraceDelivery: 'outcome': 'success', 'dropped_stages': 0, } - assert counters.traces == {} def test_sampling_decision_is_deterministic_and_keeps_failures(self): trace, _ = get_modules() @@ -242,27 +425,28 @@ class TestTraceDelivery: def test_failures_mode_keeps_failed_and_debug_traces_only(self): trace, _ = get_modules() - manager, counters = make_counters(trace_config(execution_trace='failures', execution_trace_sample=1000000)) + _, counters = make_counters(trace_config(execution_trace='failures', execution_trace_sample=1000000)) - failed = trace.bind() + failed = trace.bind('exec-failed') counters.record(CONTEXT, **{**STAGE, 'outcome': 'failed'}) counters.close_trace(failed.state, 'event_done') trace.unbind_root(failed) - debug = trace.bind() + debug = trace.bind('exec-debug') counters.record(CONTEXT, **{**STAGE, 'synthetic': True}) counters.close_trace(debug.state, 'event_done') trace.unbind_root(debug) - sampled_out = trace.bind() + sampled_out = trace.bind('exec-sampled') counters.record(CONTEXT, **STAGE) counters.close_trace(sampled_out.state, 'event_done') trace.unbind_root(sampled_out) - assert [payload['features']['observations'][0]['outcome'] for payload in manager.sent] == ['failed', 'success'] - assert manager.sent[1]['features']['observations'][0]['synthetic'] is True + assert [record['query_id'] for record in counters.records] == ['exec-failed', 'exec-debug'] + assert counters.records[0]['features']['observations'][0]['outcome'] == 'failed' + assert counters.records[1]['features']['observations'][0]['synthetic'] is True - def test_off_mode_and_opt_out_emit_nothing(self): + def test_off_mode_and_opt_out_buffer_nothing(self): trace, _ = get_modules() manager, counters = make_counters(trace_config(execution_trace='off')) binding = trace.bind() @@ -272,6 +456,7 @@ class TestTraceDelivery: finally: trace.unbind_root(binding) assert manager.sent == [] + assert counters.records == [] disabled_manager, disabled = make_counters( {'url': 'https://space.example.test', 'disable_telemetry': True, 'execution_trace': 'all'} @@ -283,89 +468,359 @@ class TestTraceDelivery: finally: trace.unbind_root(binding) assert disabled_manager.sent == [] - assert disabled.pending == {} + assert disabled.records == [] assert disabled.traces == {} def test_ttl_sweep_closes_stale_trace(self): trace, _ = get_modules() manager, counters = make_counters(trace_config()) - binding = trace.bind() + binding = trace.bind('exec-ttl') try: counters.record(CONTEXT, **{**STAGE, 'outcome': 'failed'}) counters.trace_deadlines[binding.state.trace_id] = 0 counters._sweep_traces(time.monotonic()) finally: trace.unbind_root(binding) - assert manager.sent[0]['features']['trace']['closed_by'] == 'ttl' + assert counters.records[0]['features']['trace']['closed_by'] == 'ttl' assert counters.traces == {} assert counters.trace_deadlines == {} + assert manager.sent == [] async def test_shutdown_drops_open_traces(self): trace, _ = get_modules() manager, counters = make_counters(trace_config()) - binding = trace.bind() + binding = trace.bind('exec-open') try: counters.record(CONTEXT, **{**STAGE, 'outcome': 'failed'}) await counters.shutdown() finally: trace.unbind_root(binding) - # Shutdown may still flush window counters; it must not claim a trace. - assert [payload for payload in manager.sent if 'trace' in payload['features']] == [] + # Shutdown must not claim a trace that never completed. + assert records_sent(manager) == [] + assert counters.records == [] assert counters.traces == {} + assert counters.registry == {} assert counters.dropped_traces == 1 - async def test_loop_sweeps_and_sends_expired_trace(self): + async def test_loop_sweeps_and_flushes_expired_trace(self): trace, _ = get_modules() - manager, counters = make_counters(trace_config()) - binding = trace.bind() + manager, counters = make_counters(trace_config(telemetry_flush_seconds=1)) + binding = trace.bind('exec-ttl-loop') try: counters.record(CONTEXT, **{**STAGE, 'outcome': 'failed'}) counters.trace_deadlines[binding.state.trace_id] = 0 - await asyncio.sleep(1.2) - assert [payload['features']['trace']['closed_by'] for payload in manager.sent] == ['ttl'] + await asyncio.sleep(1.3) + assert [record['features']['trace']['closed_by'] for record in records_sent(manager)] == ['ttl'] assert counters.traces == {} finally: await counters.shutdown() trace.unbind_root(binding) + async def test_flush_sends_one_batch_of_complete_records(self): + _, execution = get_modules() + manager, counters = make_counters(trace_config()) + ap = make_ap(counters) + for index in range(2): + with execution.ingress(ap, 'event_done', CONTEXT, execution_id=f'exec-{index}'): + counters.record(CONTEXT, **STAGE) + assert len(counters.records) == 2 + + await counters.flush() + assert len(manager.sent) == 1 + payload = manager.sent[0] + assert set(payload) == {'records'} + assert [record['query_id'] for record in payload['records']] == ['exec-0', 'exec-1'] + assert 'count' not in json.dumps(payload) + + async def test_flush_interval_is_honoured(self): + _, execution = get_modules() + manager, counters = make_counters(trace_config(telemetry_flush_seconds=1)) + ap = make_ap(counters) + with execution.ingress(ap, 'event_done', CONTEXT, execution_id='exec-timed'): + counters.record(CONTEXT, **STAGE) + assert counters.flush_seconds() == 1 + assert manager.sent == [] + await asyncio.sleep(1.3) + assert [record['query_id'] for record in records_sent(manager)] == ['exec-timed'] + await counters.shutdown() + + def test_flush_interval_config_is_bounded(self): + _, counters = make_counters(trace_config()) + assert counters.flush_seconds() == 180 + counters.manager.telemetry_config = trace_config(telemetry_flush_seconds='nope') + assert counters.flush_seconds() == 180 + counters.manager.telemetry_config = trace_config(telemetry_flush_seconds=0) + assert counters.flush_seconds() == 180 + counters.manager.telemetry_config = trace_config(telemetry_flush_seconds=300) + assert counters.flush_seconds() == 300 + + def test_buffered_record_cap_drops_oldest(self): + trace, execution = get_modules() + _, counters = make_counters(trace_config()) + original = execution.MAX_BUFFERED_RECORDS + execution.MAX_BUFFERED_RECORDS = 3 + try: + for index in range(5): + binding = trace.bind(f'exec-{index}') + counters.record(CONTEXT, **STAGE) + counters.close_trace(binding.state, 'event_done') + trace.unbind_root(binding) + finally: + execution.MAX_BUFFERED_RECORDS = original + assert [record['query_id'] for record in counters.records] == ['exec-2', 'exec-3', 'exec-4'] + assert counters.dropped == 2 + class TestIngress: def test_ingress_closes_only_the_trace_it_started(self): _, execution = get_modules() manager, counters = make_counters(trace_config()) ap = make_ap(counters) - with execution.ingress(ap, 'event_done'): - with execution.ingress(ap, 'pipeline_done'): + with execution.ingress(ap, 'event_done', execution_id='exec-outer'): + with execution.ingress(ap, 'pipeline_done', execution_id='exec-inner'): counters.record(CONTEXT, **{**STAGE, 'outcome': 'failed'}) - assert len(manager.sent) == 1 - assert manager.sent[0]['features']['trace']['closed_by'] == 'event_done' + assert len(counters.records) == 1 + assert counters.records[0]['query_id'] == 'exec-outer' + assert counters.records[0]['features']['trace']['closed_by'] == 'event_done' + assert manager.sent == [] def test_ingress_without_telemetry_is_a_no_op(self): _, execution = get_modules() - with execution.ingress(types.SimpleNamespace(), 'event_done'): + with execution.ingress(types.SimpleNamespace(), 'event_done', execution_id='exec-x'): pass + def test_ingress_registers_every_execution_alias(self): + _, execution = get_modules() + _, counters = make_counters(trace_config()) + ap = make_ap(counters) + with execution.ingress(ap, 'event_done', execution_id='event-1') as outer: + with execution.ingress(ap, 'pipeline_done', execution_id='query-1') as inner: + assert inner.state is outer.state + assert counters.registry == {} + # Closing the owning trace unregisters every alias it held. + assert counters.registry_deadlines == {} -class AsyncSendManager(FakeManager): - """Stand-in whose start_send_task is a coroutine, like TelemetryManager.""" - async def start_send_task(self, payload: dict) -> None: - self.sent.append(payload) +class TestCrossTaskAttribution: + def test_registry_joins_stages_from_a_task_without_a_trace(self): + trace, execution = get_modules() + _, counters = make_counters(trace_config()) + ap = make_ap(counters) + binding = execution.bind_trace(ap, 'run-1') + trace.unbind_root(binding) + # The plugin/RPC task has no inherited trace context. + assert trace.current() is None + + counters.record(CONTEXT, **{**STAGE, 'family': 'platform_api', 'operation': 'send_message'}) + counters.record( + CONTEXT, **{**STAGE, 'family': 'platform_api', 'operation': 'send_message'}, execution_id='run-1' + ) + assert len(binding.state.stages) == 1 + assert binding.state.stages[0]['operation'] == 'send_message' + + counters.close_trace(binding.state, 'event_done') + assert counters.records[0]['query_id'] == 'run-1' + + def test_registry_aliases_a_nested_execution(self): + _, execution = get_modules() + _, counters = make_counters(trace_config()) + ap = make_ap(counters) + with execution.ingress(ap, 'event_done', CONTEXT, execution_id='event-1') as binding: + with execution.ingress(ap, 'pipeline_done', CONTEXT, execution_id='query-1'): + pass + assert set(counters.registry) == {'event-1', 'query-1'} + counters.record(CONTEXT, **STAGE, execution_id='query-1') + assert counters.records[0]['query_id'] == 'event-1' + assert len(binding.state.stages) == 1 + + def test_contextvar_execution_id_is_used_when_no_kwarg(self): + trace, execution = get_modules() + _, counters = make_counters(trace_config()) + ap = make_ap(counters) + binding = execution.bind_trace(ap, 'run-2') + trace.unbind_root(binding) + token = execution.set_execution_id('run-2') + try: + counters.record(CONTEXT, **STAGE) + finally: + execution.reset_execution_id(token) + assert len(binding.state.stages) == 1 + + def test_registry_alias_is_swept_when_never_closed(self): + trace, execution = get_modules() + _, counters = make_counters(trace_config()) + ap = make_ap(counters) + binding = execution.bind_trace(ap, 'run-3') + trace.unbind_root(binding) + counters.registry_deadlines['run-3'] = 0 + counters._sweep_registry(time.monotonic()) + assert counters.registry == {} + assert counters.registry_deadlines == {} class TestTraceDispatch: - async def test_close_trace_schedules_the_coroutine_send(self): - trace, execution = get_modules() + async def test_flush_awaits_the_manager_send(self): + _, execution = get_modules() manager = AsyncSendManager(trace_config()) counters = execution.ExecutionCounters(manager) - binding = trace.bind() + ap = make_ap(counters) + with execution.ingress(ap, 'event_done', CONTEXT, execution_id='exec-async'): + counters.record(CONTEXT, **STAGE) + await counters.flush() + assert [record['query_id'] for record in records_sent(manager)] == ['exec-async'] + assert manager.sent[0]['records'][0]['event_type'] == 'feature_execution' + + +class AsyncSendManager(FakeManager): + """Stand-in whose send is a coroutine, like TelemetryManager.""" + + async def send(self, payload: dict) -> bool: + self.sent.append(payload) + return True + + +class TestTraceFailureDetail: + def test_failed_stage_carries_the_reason(self): + trace, _ = get_modules() + _, counters = make_counters(trace_config()) + binding = trace.bind('exec-fail') + try: + counters.record(CONTEXT, **{**STAGE, 'outcome': 'failed'}, error='boom: sandbox exited 127') + counters.close_trace(binding.state, 'event_done') + finally: + trace.unbind_root(binding) + record = counters.records[0] + assert record['error'] == 'boom: sandbox exited 127' + assert record['features']['observations'][0]['error'] == 'boom: sandbox exited 127' + + def test_successful_stage_reports_no_error(self): + trace, _ = get_modules() + _, counters = make_counters(trace_config()) + binding = trace.bind('exec-ok') try: counters.record(CONTEXT, **STAGE) counters.close_trace(binding.state, 'event_done') - # The payload is only delivered once the scheduled task runs. - assert manager.sent == [] - await asyncio.sleep(0) finally: trace.unbind_root(binding) - assert len(manager.sent) == 1 - assert manager.sent[0]['event_type'] == 'feature_execution' + record = counters.records[0] + assert record['error'] == '' + assert record['features']['observations'][0]['error'] == '' + + def test_error_detail_is_bounded_and_single_line(self): + trace, _ = get_modules() + _, counters = make_counters(trace_config()) + binding = trace.bind('exec-noisy') + try: + counters.record(CONTEXT, **{**STAGE, 'outcome': 'failed'}, error='x' * 1000 + '\n\tboom') + counters.close_trace(binding.state, 'event_done') + finally: + trace.unbind_root(binding) + detail = counters.records[0]['error'] + assert len(detail) <= 400 + assert '\n' not in detail and '\t' not in detail + + +class TestIngressFailureDetail: + def test_ingress_records_the_failure_reason(self): + _, execution = get_modules() + _, counters = make_counters(trace_config()) + ap = make_ap(counters) + raised = False + try: + with execution.ingress(ap, 'event_done', execution_id='exec-boom'): + counters.record(CONTEXT, **STAGE) + raise RuntimeError('dispatch exploded') + except RuntimeError: + raised = True + assert raised is True + record = counters.records[0] + assert record['error'].startswith('RuntimeError: dispatch exploded') + assert record['features']['trace']['outcome'] == 'failed' + assert record['features']['trace']['closed_by'] == 'event_done' + + def test_successful_ingress_still_reports_success(self): + _, execution = get_modules() + _, counters = make_counters(trace_config()) + ap = make_ap(counters) + with execution.ingress(ap, 'event_done', execution_id='exec-fine'): + counters.record(CONTEXT, **STAGE) + record = counters.records[0] + assert record['error'] == '' + assert record['features']['trace']['outcome'] == 'success' + + +class TestFailureOnlyTrace: + def test_failure_without_stages_still_uploads(self): + _, execution = get_modules() + # sampled-out config: only failures are guaranteed to be kept + _, counters = make_counters({'url': 'https://space.example.test', 'execution_trace_sample': 1}) + ap = make_ap(counters) + try: + with execution.ingress(ap, 'pipeline_done', CONTEXT, execution_id='exec-empty'): + raise RuntimeError('pipeline exploded before any stage') + except RuntimeError: + pass + assert len(counters.records) == 1 + record = counters.records[0] + assert record['error'].startswith('RuntimeError: pipeline exploded') + assert record['features']['trace']['outcome'] == 'failed' + assert record['features']['observations'] == [] + + def test_cancelled_and_timeout_chains_report_their_outcome(self): + trace, _ = get_modules() + _, counters = make_counters(trace_config()) + for outcome, reason in (('cancelled', 'run cancelled by operator'), ('timeout', 'deadline exhausted')): + binding = trace.bind(f'exec-{outcome}') + try: + counters.record(CONTEXT, **{**STAGE, 'outcome': outcome}, error=reason) + counters.close_trace(binding.state, 'runner_done') + finally: + trace.unbind_root(binding) + assert len(counters.records) == 1 + record = counters.records.pop(0) + assert record['query_id'] == f'exec-{outcome}' + assert record['features']['trace']['outcome'] == outcome + assert record['error'] == reason + + +class TestLegacyRecordFields: + def test_record_carries_legacy_top_level_columns(self): + trace, _ = get_modules() + _, counters = make_counters(trace_config()) + binding = trace.bind('exec-legacy') + binding.state.started_at = (datetime.now(timezone.utc) - timedelta(seconds=2)).isoformat() + binding.state.model_name = 'gpt-4o' + binding.state.pipeline_plugins = ['plugin:a'] + binding.state.runner_category = 'cloud' + try: + counters.record(CONTEXT, **STAGE) + counters.record( + CONTEXT, **{**STAGE, 'family': 'runner', 'operation': 'execute', 'mode': 'agent', 'runner': 'runner-1'} + ) + counters.close_trace(binding.state, 'event_done') + finally: + trace.unbind_root(binding) + record = counters.records[0] + assert record['model_name'] == 'gpt-4o' + assert record['pipeline_plugins'] == ['plugin:a'] + assert record['runner_category'] == 'cloud' + assert record['adapter'] == 'AiocqhttpAdapter' + assert record['runner'] == 'runner-1' + assert record['duration_ms'] >= 1000 + assert record['timestamp'] == record['features']['trace']['ended_at'] + + def test_legacy_string_fields_are_never_null(self): + trace, _ = get_modules() + _, counters = make_counters(trace_config()) + binding = trace.bind('exec-blank') + try: + counters.record(CONTEXT, **STAGE) + counters.close_trace(binding.state, 'event_done') + finally: + trace.unbind_root(binding) + record = counters.records[0] + assert record['model_name'] == '' + assert record['runner'] == '' + assert record['runner_category'] == '' + assert isinstance(record['duration_ms'], int) and record['duration_ms'] >= 0 + assert record['pipeline_plugins'] is None